langgraph_agent_toolkit.core.observability.base module

class langgraph_agent_toolkit.core.observability.base.BaseObservabilityPlatform(remote_first=False)

Bases: ABC

Base class for observability platforms.

Provides shared utilities without disk I/O. This prevents blocking in asynchronous contexts.

Initialize the observability platform.

Parameters:

remote_first (bool) – Prioritize remote prompts over the local cache.

__init__(remote_first=False)

Initialize the observability platform.

Parameters:

remote_first (bool) – Prioritize remote prompts over the local cache.

property remote_first: bool

property required_vars: List[str]

validate_environment()

Validate that required environment variables are set.

Return type:

bool

static requires_env_vars(func)

Validate environment variables before calling a method.

Parameters:

func (Callable[[...], T])

Return type:

Callable[[…], T]

abstractmethod get_callback_handler(**kwargs)

Get the callback handler for the observability platform.

Return type:

Any

abstractmethod before_shutdown()

Perform any necessary cleanup before shutdown.

Return type:

None

abstractmethod record_feedback(run_id, key, score, **kwargs)

Record feedback for a run.

Parameters:

• run_id (str)

• key (str)

• score (float)

Return type:

None

abstractmethod push_prompt(name, prompt_template, metadata=None, force_create_new_version=True)

Push a prompt to the observability platform.

Parameters:

• name (str)

• prompt_template (str | List[ChatMessageDict])

• metadata (Dict[str, Any] | None)

• force_create_new_version (bool)

Return type:

None

abstractmethod pull_prompt(name, template_format='f-string', **kwargs)

Pull a prompt from the observability platform.

Parameters:

• name (str)

• template_format (Literal['f-string', 'mustache', 'jinja2'])

Return type:

ChatPromptTemplate | str | dict | None

async apull_prompt(name, template_format='f-string', **kwargs)

Asynchronously run pull_prompt in a thread pool.

Parameters:

• name (str)

• template_format (Literal['f-string', 'mustache', 'jinja2'])

Return type:

ChatPromptTemplate | str | dict | None

abstractmethod delete_prompt(name)

Delete a prompt from the observability platform.

Parameters:

name (str)

Return type:

None

trace_context(run_id, **kwargs)

Create an execution trace context.

Subclasses provide platform-specific implementations.

Parameters:

• run_id (str) – Run ID to use as trace ID.

• **kwargs – Context parameters, such as user_id and input.

Yields:

None or a platform-specific context object.

update_trace(trace, **attributes)

Set attributes on a trace from trace_context.

Call this after the run to record output. Platforms without a trace object do nothing.

Parameters:

• trace (Any)

• attributes (Any)

Return type:

None