148 lines
5.1 KiB
Python
148 lines
5.1 KiB
Python
"""Langfuse observability — singleton client and prompt helpers.
|
|
|
|
If LANGFUSE_SECRET_KEY / LANGFUSE_PUBLIC_KEY are not set,
|
|
all helpers are no-ops so the app works without Langfuse configured.
|
|
|
|
Usage
|
|
-----
|
|
Tracing::
|
|
|
|
from app.core.langfuse_client import get_langfuse
|
|
|
|
lf = get_langfuse()
|
|
if lf:
|
|
with lf.start_as_current_observation(as_type="span", name="my-agent") as span:
|
|
span.update(input=user_message)
|
|
# ... do work ...
|
|
span.update(output=result)
|
|
lf.flush()
|
|
|
|
Prompt management::
|
|
|
|
from app.core.langfuse_client import get_prompt_or_fallback
|
|
|
|
text, prompt_obj = get_prompt_or_fallback("home_system", FALLBACK_PROMPT)
|
|
# Use text as the system prompt; pass prompt_obj to generations for linking.
|
|
|
|
Linking a prompt to a generation::
|
|
|
|
with lf.start_as_current_observation(
|
|
as_type="generation",
|
|
name="llm-call",
|
|
model="gpt-4o",
|
|
prompt=prompt_obj, # links generation → prompt version in the UI
|
|
input=messages,
|
|
) as gen:
|
|
response = await llm.ainvoke(messages)
|
|
gen.update(output=response.content, usage=_usage(response))
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_client: Any = None
|
|
_initialized: bool = False
|
|
|
|
|
|
def get_langfuse() -> Any | None:
|
|
"""Return the Langfuse singleton, or ``None`` when not configured."""
|
|
global _client, _initialized
|
|
if _initialized:
|
|
return _client
|
|
_initialized = True
|
|
|
|
from app.config.settings import settings # local import to avoid circular deps
|
|
|
|
if not settings.LANGFUSE_SECRET_KEY or not settings.LANGFUSE_PUBLIC_KEY:
|
|
logger.debug("langfuse: not configured — observability disabled")
|
|
return None
|
|
|
|
try:
|
|
from langfuse import Langfuse
|
|
|
|
_client = Langfuse(
|
|
secret_key=settings.LANGFUSE_SECRET_KEY,
|
|
public_key=settings.LANGFUSE_PUBLIC_KEY,
|
|
host=settings.LANGFUSE_BASE_URL,
|
|
)
|
|
logger.info("langfuse: client initialized host=%s", settings.LANGFUSE_BASE_URL)
|
|
except Exception as exc:
|
|
logger.warning("langfuse: failed to initialize: %s", exc)
|
|
_client = None
|
|
|
|
return _client
|
|
|
|
|
|
def get_prompt_or_fallback(name: str, fallback: str) -> tuple[str, Any]:
|
|
"""Fetch a text prompt from Langfuse; fall back to ``fallback`` on any error.
|
|
|
|
Returns ``(raw_template, prompt_obj_or_None)``.
|
|
|
|
* ``raw_template`` — the uncompiled template string. Do NOT call ``.format()``
|
|
on it directly; use :func:`compile_prompt` instead so the correct variable
|
|
syntax is applied (``{{var}}`` for Langfuse, ``{var}`` for the fallback).
|
|
* ``prompt_obj`` — the Langfuse prompt object, or ``None`` when Langfuse is
|
|
unavailable / the fetch failed. Pass this to generation observations so
|
|
Langfuse links the generation to the exact prompt version in the UI.
|
|
"""
|
|
lf = get_langfuse()
|
|
if lf is None:
|
|
return fallback, None
|
|
|
|
try:
|
|
prompt = lf.get_prompt(name, label="production", fallback=fallback)
|
|
# For text-type prompts .prompt holds the raw template string.
|
|
raw = prompt.prompt if hasattr(prompt, "prompt") and isinstance(prompt.prompt, str) else fallback
|
|
return raw, prompt
|
|
except Exception as exc:
|
|
logger.warning("langfuse: get_prompt %r failed: %s — using fallback", name, exc)
|
|
return fallback, None
|
|
|
|
|
|
def compile_prompt(template: str, prompt_obj: Any, **variables: Any) -> str:
|
|
"""Compile *template* with *variables*, choosing the right syntax.
|
|
|
|
* When *prompt_obj* is a real Langfuse prompt object, calls
|
|
``prompt_obj.compile(**variables)`` which handles ``{{variable}}``
|
|
substitution as defined in the Langfuse UI.
|
|
* When *prompt_obj* is ``None`` (Langfuse unavailable or fetch failed),
|
|
falls back to ``template.format(**variables)`` which handles the
|
|
``{variable}`` syntax used in the hardcoded fallback strings.
|
|
|
|
This keeps callers oblivious to which syntax is in use.
|
|
"""
|
|
if prompt_obj is not None:
|
|
try:
|
|
compiled = prompt_obj.compile(**variables)
|
|
# compile() returns a string for text prompts.
|
|
if isinstance(compiled, str):
|
|
return compiled
|
|
# Chat prompts return a list of dicts — join text parts.
|
|
if isinstance(compiled, list):
|
|
return "\n".join(
|
|
m.get("content", "") for m in compiled if isinstance(m, dict)
|
|
)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"langfuse: compile failed for prompt %r: %s — falling back to .format()",
|
|
getattr(prompt_obj, "name", "?"),
|
|
exc,
|
|
)
|
|
return template.format(**variables)
|
|
|
|
|
|
def extract_usage(response: Any) -> dict[str, int]:
|
|
"""Extract token usage from a LangChain AI message into Langfuse format."""
|
|
meta = getattr(response, "usage_metadata", None)
|
|
if not meta:
|
|
return {}
|
|
return {
|
|
"input": int(meta.get("input_tokens", 0)),
|
|
"output": int(meta.get("output_tokens", 0)),
|
|
"total": int(meta.get("total_tokens", 0)),
|
|
}
|