Skip to content

Utils

GLLM Agents Utils.

This module contains utility functions and classes for the GLLM Agents package.

Authors

Christian Trisno Sen Long Chen (christian.t.s.l.chen@gdplabs.id)

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.

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_log_format(custom_log_format)
manager.set_date_format(custom_date_format)

Example to add a custom handler:

manager = LoggerManager()

handler = logging.FileHandler("app.log")
manager.add_handler(handler)

Output format example:

[16/04/2025 15:08:18.323 GDPLabsGenAILogger INFO] Loading prompt_builder catalog for chatbot `general-purpose`

__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.

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.

Parameters:

Name Type Description Default
log_format str

The log format to set.

 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).

embed_references_in_content(content, references)

Embed references into content for A2A protocol.

Parameters:

Name Type Description Default
content str

The original content string

 required
references list[Any]

List of reference objects to embed

 required

Returns:

Type Description
str

JSON string with content and references embedded, or original content if references fail

serialize_references_for_a2a(references)

Serialize references into a JSON string for A2A protocol.

Parameters:

Name Type Description Default
references list[Any]

List of reference objects to serialize

 required

Returns:

Type Description
str | None

JSON string with serialized references, or None if no valid references

validate_references(references)

Deduplicate reference data from agent state.

Since the reducer function (add_references_chunks) already filters for valid Chunk objects, this function focuses on deduplication by content.

Parameters:

Name Type Description Default
references list[Any]

List of reference data from agent state (expected to be Chunk objects).

 required

Returns:

Type Description
list[Chunk]

List of deduplicated Chunk objects by content.