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

Bases: ABC

Base class for observability platforms.

This is a lightweight base class that provides common utilities for observability platforms. It does NOT perform any disk I/O operations to avoid blocking in async contexts.

Initialize the observability platform.

Parameters:

remote_first (bool) – If True, prioritize remote prompts over local cache.

__init__(remote_first=False)[source][source]

Initialize the observability platform.

Parameters:

remote_first (bool) – If True, prioritize remote prompts over local cache.

property remote_first: bool
property required_vars: List[str]
validate_environment()[source][source]

Validate that required environment variables are set.

Return type:

bool

static requires_env_vars(func)[source][source]

Validate environment variables before calling a method.

Parameters:

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

Return type:

Callable[[…], T]

abstractmethod get_callback_handler(**kwargs)[source][source]

Get the callback handler for the observability platform.

Return type:

Any

abstractmethod before_shutdown()[source][source]

Perform any necessary cleanup before shutdown.

Return type:

None

abstractmethod record_feedback(run_id, key, score, **kwargs)[source][source]

Record feedback for a run.

Parameters:
Return type:

None

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

Push a prompt to the observability platform.

Parameters:
Return type:

None

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

Pull a prompt from the observability platform.

Parameters:

  • name (str)

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

Return type:

ChatPromptTemplate | str | dict | None

abstractmethod delete_prompt(name)[source][source]

Delete a prompt from the observability platform.

Parameters:

name (str)

Return type:

None

trace_context(run_id, **kwargs)[source][source]

Create a trace context for the execution.

Override in subclasses for platform-specific implementation.

Parameters:

  • run_id (str) – The run ID to use as trace ID

  • **kwargs – Additional context parameters (user_id, input, etc.)

Yields:

None (or platform-specific context object)