langgraph_agent_toolkit.core.observability.langfuse module

class langgraph_agent_toolkit.core.observability.langfuse.LangfuseObservability(remote_first=False)[source][source]

Bases: BaseObservabilityPlatform

Langfuse observability platform.

Initialize LangfuseObservability.

Parameters:

remote_first (bool) – Prioritize remote prompts when True.

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

Initialize LangfuseObservability.

Parameters:

remote_first (bool) – Prioritize remote prompts when True.

validate_environment()[source][source]

Accept the current URL variable and its legacy alias.

Return type:

bool

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

Get the LangChain Langfuse callback handler.

Create a separate handler for each run. SDK v2 handlers keep mutable trace state.

Return type:

LangchainCallbackHandler

before_shutdown()[source][source]

Flush an initialized client without creating a new client.

Return type:

None

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

Attach feedback to the same trace ID as the run.

Parameters:

  • run_id (str)

  • key (str)

  • score (float)

Return type:

None

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

Push a prompt to Langfuse.

Parameters:

  • name (str) – Name of the prompt

  • prompt_template (str | List[ChatMessageDict]) – The prompt template (string or list of message dicts)

  • metadata (Dict[str, Any] | None) – Optional metadata including ‘labels’

  • force_create_new_version (bool) – If True, always create a new version

Return type:

None

pull_prompt(name, return_with_prompt_object=False, cache_ttl_seconds=settings.LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS, template_format='f-string', label=None, version=None, **kwargs)[source][source]

Pull a prompt from Langfuse.

Parameters:

  • name (str) – Name of the prompt

  • return_with_prompt_object (bool) – If True, return tuple of (prompt, langfuse_prompt)

  • cache_ttl_seconds (int | None) – Cache TTL for the prompt

  • template_format (Literal['f-string', 'mustache', 'jinja2']) – Format for the template

  • label (str | None) – Optional label to fetch specific version

  • version (int | None) – Optional version number to fetch

  • **kwargs – Additional kwargs (prompt_label, prompt_version as aliases)

Returns:

ChatPromptTemplate or tuple of (ChatPromptTemplate, langfuse_prompt)

Return type:

ChatPromptTemplate | str | dict | None | Tuple[ChatPromptTemplate | str | dict | None, Any]

async apull_prompt(name, return_with_prompt_object=False, cache_ttl_seconds=settings.LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS, template_format='f-string', label=None, version=None, **kwargs)[source][source]

Asynchronously run pull_prompt in a thread pool.

Parameters:

  • name (str)

  • return_with_prompt_object (bool)

  • cache_ttl_seconds (int | None)

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

  • label (str | None)

  • version (int | None)

Return type:

ChatPromptTemplate | str | dict | None | Tuple[ChatPromptTemplate | str | dict | None, Any]

delete_prompt(name)[source][source]

Delete all prompt versions when the installed SDK supports deletion.

Parameters:

name (str)

Return type:

None

trace_context(run_id, **kwargs)[source]

Open the root trace and propagate user and session attributes.

Parameters:

run_id (str)

update_trace(trace, **attributes)[source][source]

Record final output on the root trace before it closes.

Return type:

None