83 lines
3.2 KiB
Python
83 lines
3.2 KiB
Python
"""Post scores to a Langfuse project, or do nothing at all.
|
|
|
|
Traces already reach Langfuse as OTLP spans (see rag/telemetry.py). A *score*
|
|
is the other half: it attaches a number to one of those traces, which is what
|
|
turns Langfuse from a trace viewer into a place where answer quality is
|
|
visible over time.
|
|
|
|
Two producers exist:
|
|
- `scripts/score_evals_ragas.py` posts Ragas metrics for eval runs.
|
|
- This adapter posts the user's thumb up/down from `/v1/rag/feedback`,
|
|
which is the only signal in the whole system that comes from a human
|
|
rather than a model judging a model.
|
|
|
|
Deliberately best-effort. Feedback is stored in Postgres first and that write
|
|
is what must not fail; a Langfuse outage may not turn a working thumbs-up into
|
|
a 503 for the user. Every failure here is swallowed and counted, never raised.
|
|
|
|
Scores attach by OpenTelemetry trace id (32 hex), NOT by the `trace_id` UUID
|
|
that identifies the Postgres row -- Langfuse has never seen that one. The
|
|
caller is responsible for looking the OTel id up.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
|
|
class LangfuseScoreClient:
|
|
"""Minimal client: one endpoint, one verb, no dependency on the SDK."""
|
|
|
|
def __init__(self, base_url: str, public_key: str, secret_key: str, timeout: float = 5.0):
|
|
self.endpoint = f"{base_url.rstrip('/')}/api/public/scores"
|
|
token = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
|
|
self._auth = f"Basic {token}"
|
|
self._timeout = timeout
|
|
|
|
@classmethod
|
|
def from_settings(cls, settings: Any) -> "LangfuseScoreClient | None":
|
|
"""Return a client, or None when Langfuse is not fully configured.
|
|
|
|
Same all-or-nothing rule as the span exporter: a base URL without keys
|
|
would authenticate nothing, so treat a partial config as "off" rather
|
|
than posting requests that can only 401.
|
|
"""
|
|
base_url = str(getattr(settings, "langfuse_base_url", "") or "").strip()
|
|
public_key = str(getattr(settings, "langfuse_public_key", "") or "").strip()
|
|
secret_key = str(getattr(settings, "langfuse_secret_key", "") or "").strip()
|
|
if not (base_url and public_key and secret_key):
|
|
return None
|
|
return cls(base_url, public_key, secret_key)
|
|
|
|
def post_score(
|
|
self,
|
|
*,
|
|
otel_trace_id: str,
|
|
name: str,
|
|
value: float,
|
|
comment: str | None = None,
|
|
) -> bool:
|
|
"""True if Langfuse accepted the score. Never raises."""
|
|
body: dict[str, Any] = {
|
|
"traceId": otel_trace_id,
|
|
"name": name,
|
|
"value": value,
|
|
"dataType": "NUMERIC",
|
|
}
|
|
if comment:
|
|
body["comment"] = comment
|
|
request = urllib.request.Request(
|
|
self.endpoint,
|
|
data=json.dumps(body).encode("utf-8"),
|
|
headers={"Content-Type": "application/json", "Authorization": self._auth},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=self._timeout):
|
|
return True
|
|
except Exception: # noqa: BLE001 - see module docstring: never raise
|
|
return False
|