File size: 2,797 Bytes
35c0d38 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """Observability via Langfuse (v4).
Initializes the Langfuse client from our settings and re-exports the pieces the
rest of the app uses to trace itself:
- ``observe`` : decorator that turns a function into a traced span
(use as_type="generation" for model calls,
as_type="tool" for tool calls).
- ``trace_attributes`` : context manager that stamps trace-level fields
(session_id, tags, metadata) onto every span
created inside it.
- ``annotate_span`` : add metadata to the currently-running span.
- ``flush`` / ``current_trace_url`` : flush buffered events; fetch the URL of
the trace currently in scope.
If Langfuse keys are not configured, every export degrades to a safe no-op so
the app and eval still run untraced.
"""
from __future__ import annotations
from contextlib import contextmanager, nullcontext
from src.config import settings
# Tracing is only active when both Langfuse keys are present.
enabled = bool(settings.langfuse_public_key and settings.langfuse_secret_key)
if enabled:
from langfuse import Langfuse, get_client
from langfuse import observe as observe # re-export
from langfuse import propagate_attributes as _propagate
# Configure the singleton client explicitly. We pass keys from settings
# because they live in .env (read by pydantic) and are NOT exported to
# os.environ, which is the only place the Langfuse SDK would look otherwise.
Langfuse(
public_key=settings.langfuse_public_key,
secret_key=settings.langfuse_secret_key,
host=settings.langfuse_host,
)
_client = get_client()
def trace_attributes(**kwargs):
"""Stamp trace-level attributes (session_id, tags, metadata, ...)."""
return _propagate(**kwargs)
def annotate_span(**kwargs) -> None:
"""Attach metadata/name/level to the current span."""
_client.update_current_span(**kwargs)
def flush() -> None:
_client.flush()
def current_trace_url() -> str | None:
try:
return _client.get_trace_url()
except Exception: # noqa: BLE001
return None
else:
def observe(func=None, **_kwargs): # type: ignore[misc]
"""No-op stand-in for langfuse.observe (supports @observe and @observe(...))."""
if callable(func):
return func
def _decorator(f):
return f
return _decorator
def trace_attributes(**_kwargs):
return nullcontext()
def annotate_span(**_kwargs) -> None:
pass
def flush() -> None:
pass
def current_trace_url() -> str | None:
return None
|