Utils
GLLM Agents Utils.
This module contains utility functions and classes for the GLLM Agents package.
LoggerManager
A singleton class to manage logging configuration.
This class ensures that the root logger is initialized only once and is used across the application. Supports both traditional colored output and standardized format with colors.
Example to get and use the logger:
manager = LoggerManager()
logger = manager.get_logger()
logger.info("This is an info message")
Example to set logging configuration:
manager = LoggerManager()
manager.set_level(logging.DEBUG)
manager.set_output_format("traditional") # or "standardized"
Example to add a custom handler:
manager = LoggerManager()
handler = logging.FileHandler("app.log")
manager.add_handler(handler)
Output format examples:
Traditional format:
[16/04/2025 15:08:18.323 GDPLabsGenAILogger INFO] Loading prompt_builder catalog for chatbot `general-purpose`
Standardized format:
{
"level": "INFO",
"timestamp": "2025-07-23T00:29:05.091Z",
"message": "Loading prompt_builder catalog for chatbot `general-purpose`"
}
Error standardized format:
{
"level": "ERROR",
"timestamp": "2025-07-23T00:29:05.091Z",
"message": "Failed to load configuration",
"error": {
"message": "Configuration file not found",
"code": "CONFIG_NOT_FOUND",
"stacktrace": "Traceback (most recent call last):..."
}
}
__new__()
Initialize the singleton instance.
add_handler(handler)
Add a custom handler to the root logger.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
handler |
Handler
|
The handler to add to the root logger. |
required |
get_logger(name=None)
Get a logger instance.
This method returns a logger instance that is a child of the root logger. If name is not provided, the root logger will be returned instead.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name |
str | None
|
The name of the child logger. If None, the root logger will be returned. Defaults to None. |
None
|
Returns:
| Type | Description |
|---|---|
Logger
|
logging.Logger: Configured logger instance. |
set_date_format(date_format)
Set date format for all loggers in the hierarchy (traditional mode only).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
date_format |
str
|
The date format to set. |
required |
set_level(level)
Set logging level for all loggers in the hierarchy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
level |
int
|
The logging level to set (e.g., logging.INFO, logging.DEBUG). |
required |
set_log_format(log_format)
Set logging format for all loggers in the hierarchy (traditional mode only).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
log_format |
str
|
The log format to set. |
required |
set_output_format(output_format)
Set the output format for all loggers in the hierarchy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
output_format |
str
|
The output format to set. Must be either "standardized" or "traditional". |
required |
add_references_chunks(left, right)
Reducer function to accumulate reference data from multiple tool calls.
This is a LangGraph reducer function that should be forgiving and handle edge cases gracefully. Non-Chunk items are filtered out.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
left |
list[Chunk]
|
Existing list of reference chunks (or None/non-list) |
required |
right |
list[Chunk]
|
New list of reference chunks to add (or None/non-list) |
required |
Returns:
| Type | Description |
|---|---|
list[Chunk]
|
Combined list of valid Chunk objects |
create_artifact_command(result, artifact_data, artifact_name, artifact_description='', mime_type=None, metadata_update=None)
Create a Command that updates artifacts (and optional metadata).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
result |
str
|
Message/result to show to the agent (for ToolMessage content). |
required |
artifact_data |
bytes | str
|
The binary data or base64 string for the artifact. |
required |
artifact_name |
str
|
The name for the artifact file. |
required |
artifact_description |
str
|
Description of the artifact. |
''
|
mime_type |
str | None
|
MIME type of the artifact. If None, auto-detected from filename. |
None
|
metadata_update |
dict[str, Any] | None
|
Optional metadata delta to merge into state metadata. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
Command |
Command
|
A LangGraph Command with update containing 'result', 'artifacts', and optional 'metadata'. |
create_artifact_response(result, artifact_data, artifact_name, artifact_description='', mime_type=None)
Create a standardized artifact response for tools.
This function creates a response that separates the agent-facing result from the user-facing artifact, following the established pattern for artifact generation in the agent system.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
result |
str
|
The message/result to show to the agent (clean, no file data). |
required |
artifact_data |
bytes | str
|
The binary data or base64 string for the artifact. |
required |
artifact_name |
str
|
The name for the artifact file. |
required |
artifact_description |
str
|
Description of the artifact. Defaults to "". |
''
|
mime_type |
str | None
|
MIME type of the artifact. If None, will be auto-detected from filename. |
None
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dictionary with 'result' and 'artifacts' keys (artifacts is always a list). |
Example
import io csv_data = "Name,Age\nAlice,30\nBob,25" response = create_artifact_response( ... result="Generated a 2-row CSV table", ... artifact_data=csv_data.encode('utf-8'), ... artifact_name="data.csv", ... artifact_description="Sample data table", ... mime_type="text/csv" ... ) assert "result" in response assert "artifacts" in response assert isinstance(response["artifacts"], list)
create_error_response(error_message)
Create a standardized error response for tools.
For error cases, we return a simple string that will be passed directly to the agent without any artifact processing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
error_message |
str
|
The error message to return. |
required |
Returns:
| Type | Description |
|---|---|
str
|
String with error information. |
create_multiple_artifacts_command(result, artifacts=None, metadata_update=None)
Create a Command that updates multiple artifacts (and optional metadata).
The 'artifacts' list accepts mixed item types: - Artifact: a typed artifact model (preferred). Will be converted via model_dump(). - dict: a prebuilt artifact dict ready for A2A. - DataSpec dict: raw spec to build an artifact with keys: {'data': bytes|str, 'artifact_name': str, 'artifact_description'?: str, 'mime_type'?: str}
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
result |
str
|
Message/result to show to the agent. |
required |
artifacts |
list[Artifact | dict[str, Any]] | None
|
List of items (Artifact | dict) to attach or build. |
None
|
metadata_update |
dict[str, Any] | None
|
Optional metadata delta to merge into state metadata. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
Command |
Command
|
A LangGraph Command with update containing 'result', 'artifacts', and optional 'metadata'. |
create_text_artifact_response(result, artifact_text, artifact_name, artifact_description='', mime_type=None)
Create a standardized artifact response for tools.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
result |
str
|
The message/result to show to the agent. |
required |
artifact_text |
str
|
The text content for the artifact. |
required |
artifact_name |
str
|
The name for the artifact file. |
required |
artifact_description |
str
|
Description of the artifact. |
''
|
mime_type |
str | None
|
MIME type of the artifact. If None, will be auto-detected from filename. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary with 'result' and 'artifacts' keys (artifacts is always a list). |
serialize_references_for_metadata(references)
Serialize references for inclusion in A2A metadata.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
references |
list[Any]
|
List of reference objects (typically Chunk objects). |
required |
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
List of serialized reference dictionaries. |
validate_references(references)
Validate and deduplicate reference data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
references |
list[Any]
|
List of reference data (expected to be Chunk objects). |
required |
Returns:
| Type | Description |
|---|---|
list[Chunk]
|
List of validated, deduplicated Chunk objects by content. |