File size: 3,854 Bytes
547ce6e | 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 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """Langfuse tracing, wrapped so it can never take the service down.
Observability is the first thing to fail in production — an expired key, a network
partition, a provider outage — and the last thing that should be allowed to fail a user's
request because of it. Every entry point here is therefore total: if Langfuse is not
configured, or its SDK raises for any reason, the wrapper degrades to a no-op and the
pipeline proceeds. Nothing in the answer path depends on a trace being written.
What is traced, per the spec, is *every model call*: the Haiku query gate and the Sonnet
generation, each with its inputs, outputs, model name and token usage — which is what
makes "why did it answer that?" answerable from the retrieval set rather than guessed.
"""
from __future__ import annotations
import logging
from collections.abc import Iterator
from contextlib import contextmanager
from typing import Any, Protocol
from app.core.settings import Settings, get_settings
logger = logging.getLogger(__name__)
_CLIENT: Any | None = None
_INITIALISED = False
class Span(Protocol):
"""The subset of the Langfuse span surface this codebase uses."""
def update(self, **kwargs: Any) -> Any: ...
class _NullSpan:
"""Stand-in used whenever tracing is unavailable."""
def update(self, **kwargs: Any) -> None:
"""Accept and discard everything: this stands in for a span that is not there."""
del kwargs
NULL_SPAN = _NullSpan()
def get_langfuse(settings: Settings | None = None) -> Any | None:
"""Return a configured Langfuse client, or ``None`` when tracing is off."""
global _CLIENT, _INITIALISED # noqa: PLW0603 - process-wide singleton
if _INITIALISED:
return _CLIENT
cfg = settings or get_settings()
_INITIALISED = True
if not cfg.langfuse_enabled:
logger.info("Langfuse not configured; tracing disabled")
_CLIENT = None
return None
try:
from langfuse import Langfuse
_CLIENT = Langfuse(
public_key=cfg.langfuse_public_key,
secret_key=cfg.langfuse_secret_key,
host=cfg.langfuse_host,
)
logger.info("Langfuse tracing enabled (%s)", cfg.langfuse_host)
except Exception:
logger.warning("Langfuse initialisation failed; continuing untraced", exc_info=True)
_CLIENT = None
return _CLIENT
@contextmanager
def trace_span(name: str, **attributes: Any) -> Iterator[Span]:
"""Trace a pipeline stage. Yields a span that is safe to call unconditionally."""
client = get_langfuse()
if client is None:
yield NULL_SPAN
return
try:
with client.start_as_current_span(name=name, **attributes) as span:
yield span
except Exception:
logger.debug("trace_span(%s) failed; continuing untraced", name, exc_info=True)
yield NULL_SPAN
@contextmanager
def trace_generation(name: str, **attributes: Any) -> Iterator[Span]:
"""Trace a model call, recording model name, input, output and usage."""
client = get_langfuse()
if client is None:
yield NULL_SPAN
return
try:
with client.start_as_current_generation(name=name, **attributes) as span:
yield span
except Exception:
logger.debug("trace_generation(%s) failed; continuing untraced", name, exc_info=True)
yield NULL_SPAN
def flush() -> None:
"""Best-effort flush of pending traces, called on shutdown."""
client = get_langfuse()
if client is None:
return
try:
client.flush()
except Exception:
logger.debug("Langfuse flush failed", exc_info=True)
def reset() -> None:
"""Forget the cached client. Used by tests."""
global _CLIENT, _INITIALISED # noqa: PLW0603 - mirrors get_langfuse
_CLIENT = None
_INITIALISED = False
|