diff --git a/.github/scripts/set_langfuse_keys.py b/.github/scripts/set_langfuse_keys.py new file mode 100644 index 0000000..decac27 --- /dev/null +++ b/.github/scripts/set_langfuse_keys.py @@ -0,0 +1,179 @@ +"""Put the Langfuse project API keys into the live Application's inline values. + +`values-production.yaml` (tracked) carries only `aiService.config.langfuseBaseUrl`, +which is not a secret. The two keys are, so they live here the same way +`secret.jwtSecret` and `secret.grafanaAdminPassword` do: inline on the ArgoCD +Application, injected from GitHub Secrets by this script, never in Git. + +Edits go through the API, not the ArgoCD UI. Editing that text box saved the +block as a FOLDED scalar once and took production down for ~16 hours by +swallowing a key into a comment -- see the header of +infra/argocd/applications/medical-chatbot-app.yaml. Reading the object, editing +the string and PUTting it back keeps the structure intact. + +Verification drives POST /api/chat, not pod health: ai-service adds the Langfuse +exporter at startup, so a bad value would surface as a crashed or silently +degraded service, and health checks stayed green through that same outage. +Chat outranks tracing -- if chat stops answering, this reverts itself. + +Required env: ARGOCD_PRACTICE_URL, ARGOCD_PRACTICE_PASSWORD, +LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY. +""" + +from __future__ import annotations + +import json +import os +import sys +import time +import urllib.error +import urllib.request + +APP_NAME = "medical-chatbot-app" +SITE = "https://realvuxbaro.me" +ROLLOUT_WAIT = 20 +ROLLOUT_ROUNDS = 12 +PUBLIC_FIELD = "langfusePublicKey" +SECRET_FIELD = "langfuseSecretKey" + + +def call(base, method, path, token=None, body=None): + req = urllib.request.Request( + f"{base}{path}", + data=json.dumps(body).encode() if body is not None else None, + method=method, + headers={"Content-Type": "application/json"}, + ) + if token: + req.add_header("Authorization", f"Bearer {token}") + with urllib.request.urlopen(req, timeout=30) as resp: + raw = resp.read() + return json.loads(raw) if raw else {} + + +def set_secret_keys(values: str, public_key: str, secret_key: str) -> str: + """Set both keys under the top-level `secret:` block, and nowhere else. + + Replaces them if already present (so re-running after a key rotation is + safe), otherwise appends them to the existing `secret:` block. Refuses to + guess if there is no `secret:` block at all -- that would mean the inline + values are not the shape this expects, which is exactly when blindly + appending causes an outage. + """ + wanted = {PUBLIC_FIELD: public_key, SECRET_FIELD: secret_key} + lines = values.splitlines() + out: list[str] = [] + in_secret = False + seen = set() + secret_end = -1 + indent = " " + + for line in lines: + stripped = line.strip() + is_top_level = bool(line) and not line[0].isspace() + if is_top_level: + if in_secret: + secret_end = len(out) # first line after the block + in_secret = stripped == "secret:" + elif in_secret and stripped: + indent = line[: len(line) - len(line.lstrip())] + field = stripped.split(":", 1)[0] + if field in wanted: + seen.add(field) + out.append(f"{indent}{field}: {wanted[field]}") + continue + out.append(line) + + if in_secret: + secret_end = len(out) + if secret_end < 0: + raise SystemExit( + "No top-level `secret:` block in the inline values -- refusing to guess " + "where the keys belong. Inspect with inspect-argocd-app.yml first." + ) + + missing = [f"{indent}{name}: {value}" for name, value in wanted.items() if name not in seen] + if missing: + out[secret_end:secret_end] = missing + return "\n".join(out) + "\n" + + +def chat_works() -> bool: + """Drive the real user path. True only on a genuine grounded answer.""" + req = urllib.request.Request( + f"{SITE}/api/chat", + data=json.dumps({"content": "Chống chỉ định của Metformin là gì?"}).encode(), + method="POST", + headers={"Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(req, timeout=120) as resp: + body = resp.read().decode(errors="replace") + ok = resp.status == 200 and "disclaimer" in body and "gặp sự cố" not in body + print(f" chat: HTTP {resp.status}, grounded={ok}") + return ok + except Exception as exc: # noqa: BLE001 - any failure is a failed check + print(f" chat: FAILED -- {exc}") + return False + + +def put_values(base, token, values: str) -> None: + app = call(base, "GET", f"/api/v1/applications/{APP_NAME}", token=token) + app["spec"]["source"]["helm"]["values"] = values + call(base, "PUT", f"/api/v1/applications/{APP_NAME}", token=token, body=app) + try: + call(base, "POST", f"/api/v1/applications/{APP_NAME}/sync", token=token, body={}) + except urllib.error.HTTPError as exc: + if exc.code != 400: # 400 = already syncing + raise + + +def main() -> int: + base = os.environ["ARGOCD_PRACTICE_URL"].rstrip("/") + password = os.environ["ARGOCD_PRACTICE_PASSWORD"] + public_key = os.environ["LANGFUSE_PUBLIC_KEY"] + secret_key = os.environ["LANGFUSE_SECRET_KEY"] + if not (public_key and secret_key): + print("Both LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY must be set.", file=sys.stderr) + return 1 + + token = call(base, "POST", "/api/v1/session", + body={"username": "admin", "password": password})["token"] + + print("=== baseline ===") + if not chat_works(): + print("Chat is already broken before any change -- refusing to touch anything.", file=sys.stderr) + return 1 + + app = call(base, "GET", f"/api/v1/applications/{APP_NAME}", token=token) + before = app["spec"]["source"]["helm"].get("values", "") + after = set_secret_keys(before, public_key, secret_key) + if after == before: + print("Inline values already carry these exact keys -- nothing to do.") + return 0 + + print("=== setting Langfuse keys ===") + put_values(base, token, after) + + for i in range(ROLLOUT_ROUNDS): + time.sleep(ROLLOUT_WAIT) + status = call(base, "GET", f"/api/v1/applications/{APP_NAME}", token=token).get("status", {}) + health = status.get("health", {}).get("status") + print(f"poll {i + 1}/{ROLLOUT_ROUNDS}: sync={status.get('sync', {}).get('status')} health={health}") + if health == "Healthy" and i >= 2: + break + + print("=== verifying chat still answers ===") + if chat_works(): + print("Langfuse keys set; chat verified working.") + return 0 + + print("Chat broke -- reverting.", file=sys.stderr) + put_values(base, token, before) + time.sleep(ROLLOUT_WAIT * 2) + print(f"reverted; chat_ok_after_revert={chat_works()}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.github/workflows/set-langfuse-keys.yml b/.github/workflows/set-langfuse-keys.yml new file mode 100644 index 0000000..62c07b4 --- /dev/null +++ b/.github/workflows/set-langfuse-keys.yml @@ -0,0 +1,23 @@ +name: Set Langfuse keys (self-verifying) + +# Injects the Langfuse project API keys into the live Application's inline +# values, then proves chat still answers by driving POST /api/chat. Reverts +# automatically if it does not -- chat outranks tracing. + +on: + workflow_dispatch: {} + +jobs: + set-keys: + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - uses: actions/checkout@v4 + - name: Set keys, verify, self-revert on failure + env: + ARGOCD_PRACTICE_URL: ${{ secrets.ARGOCD_PRACTICE_URL }} + ARGOCD_PRACTICE_PASSWORD: ${{ secrets.ARGOCD_PRACTICE_PASSWORD }} + LANGFUSE_PUBLIC_KEY: ${{ secrets.LANGFUSE_PUBLIC_KEY }} + LANGFUSE_SECRET_KEY: ${{ secrets.LANGFUSE_SECRET_KEY }} + run: python3 .github/scripts/set_langfuse_keys.py diff --git a/apps/ai-service/adapters/langfuse_scores.py b/apps/ai-service/adapters/langfuse_scores.py new file mode 100644 index 0000000..df0f3fb --- /dev/null +++ b/apps/ai-service/adapters/langfuse_scores.py @@ -0,0 +1,82 @@ +"""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 diff --git a/apps/ai-service/adapters/postgres.py b/apps/ai-service/adapters/postgres.py index 9d3b655..a1f7e29 100644 --- a/apps/ai-service/adapters/postgres.py +++ b/apps/ai-service/adapters/postgres.py @@ -191,6 +191,28 @@ class PostgresTraceRepository: raise FeedbackTraceNotFound(trace_id) return str(row[0]) + def otel_trace_id_for(self, trace_id: str) -> str | None: + """The OpenTelemetry id of a persisted answer, for attaching a score. + + Separate from `save_feedback` rather than folded into its RETURNING + clause: that method's `-> str` shape is part of the `TraceWriter` + protocol the router depends on, and widening it to carry an unrelated + id would make every implementation and test carry it too. Feedback is + rare enough that a second short query costs nothing. + + None when the row has no OTel id -- true for every answer produced + while tracing was off, which is most of the history predating + 2026-08-21. Callers skip posting rather than inventing one. + """ + import psycopg + + with psycopg.connect(self._dsn, connect_timeout=5) as connection: + row = connection.execute( + "SELECT otel_trace_id FROM rag_retrieval_trace WHERE trace_id = %s", + (trace_id,), + ).fetchone() + return str(row[0]) if row and row[0] else None + class PostgresConversationStore: """Durable, cross-worker replacement for `RagAgent`'s in-process diff --git a/apps/ai-service/config.py b/apps/ai-service/config.py index 152e1a4..820fe7c 100644 --- a/apps/ai-service/config.py +++ b/apps/ai-service/config.py @@ -65,6 +65,14 @@ class Settings(BaseSettings): otel_service_name: str = "ai-service" otel_exporter_otlp_endpoint: str = "http://localhost:4318/v1/traces" otel_sample_ratio: float = Field(default=1.0, ge=0.0, le=1.0) + # Langfuse ingests the SAME spans over OTLP as a second exporter, so it + # adds an LLM-shaped view (per-trace answers, eval scores) without + # replacing Tempo or touching any instrumentation. All three must be set + # for the exporter to be added; keys are secrets and live only in .env / + # the cluster Secret, never in values.yaml. + langfuse_base_url: str = "" + langfuse_public_key: str = "" + langfuse_secret_key: str = "" entities_path: Path = _default_entities_path() # F-08: a per-turn budget across RagAgent's sequential Bedrock calls # (understand, generate, one entailment check on the live agent path). diff --git a/apps/ai-service/main.py b/apps/ai-service/main.py index fc388c7..aba1422 100644 --- a/apps/ai-service/main.py +++ b/apps/ai-service/main.py @@ -1,11 +1,13 @@ from __future__ import annotations +from contextlib import nullcontext from hmac import compare_digest from time import perf_counter from typing import Any from fastapi import FastAPI, Request, Response +from adapters.langfuse_scores import LangfuseScoreClient from adapters.postgres import PostgresTraceRepository from bootstrap import build_runtime from config import Settings, get_settings @@ -38,6 +40,10 @@ def create_app( app.state.conversational = conversational app.state.trace_writer = trace_writer app.state.metrics = metrics + # None unless Langfuse is fully configured; `/v1/rag/feedback` then skips + # mirroring the user's verdict there. Built from the same settings the + # span exporter reads, so the two can never point at different projects. + app.state.langfuse_scores = LangfuseScoreClient.from_settings(configured) # The raw Qdrant retriever, not routed through RagAgent — Feature-List # #4/#23's section-list and verbatim-section-text endpoints are plain # payload-filtered reads with no LLM/generation step, so nothing about @@ -51,8 +57,19 @@ def create_app( method = request.method.upper() started = perf_counter() status_code = 500 + # Kubernetes probes and the Prometheus scrape run every few seconds + # forever and carry no information a trace could show. Left traced they + # were ~90% of all spans within a day, burying the RAG turns anyone + # actually opens a trace viewer to read, and costing storage and + # ingestion for nothing. Their Prometheus counters below are untouched, + # which is the right place to watch probe health anyway. + span_context = ( + nullcontext(None) + if route in _UNTRACED_ROUTES + else request_span(method, route, request.headers) + ) with correlation_context(request.headers.get("x-correlation-id")) as correlation_id: - with request_span(method, route, request.headers) as span: + with span_context as span: try: response = await call_next(request) status_code = response.status_code @@ -126,6 +143,9 @@ def create_app( return app +_UNTRACED_ROUTES = frozenset({"/health", "/ready", "/metrics"}) + + def _route_label(path: str) -> str: known = { "/health", "/ready", "/metrics", "/v1/rag/query", "/v1/rag/suggest", diff --git a/apps/ai-service/rag/telemetry.py b/apps/ai-service/rag/telemetry.py index 1af4e19..004ea7e 100644 --- a/apps/ai-service/rag/telemetry.py +++ b/apps/ai-service/rag/telemetry.py @@ -7,6 +7,7 @@ thread-pool endpoint. """ from __future__ import annotations +import base64 import re import uuid from contextlib import contextmanager @@ -75,11 +76,37 @@ def configure_telemetry(settings: Any, metrics: Metrics | None = None) -> bool: ) exporter = OTLPSpanExporter(endpoint=settings.otel_exporter_otlp_endpoint) provider.add_span_processor(BatchSpanProcessor(exporter)) + for extra in _langfuse_exporters(settings, OTLPSpanExporter): + provider.add_span_processor(BatchSpanProcessor(extra)) otel_trace.set_tracer_provider(provider) _tracer = provider.get_tracer("duocthu.ai-service") return True +def _langfuse_exporters(settings: Any, exporter_cls: Any) -> list[Any]: + """Zero or one extra OTLP exporter pointed at Langfuse. + + Langfuse accepts standard OTLP/HTTP on `/api/public/otel/v1/traces`, + authenticated with HTTP Basic using the project's public key as the + username and the secret key as the password. Returning a list keeps the + caller unchanged when Langfuse is not configured -- which is the default, + and the case for every local run without a `.env`. + """ + base_url = str(getattr(settings, "langfuse_base_url", "") or "").rstrip("/") + public_key = str(getattr(settings, "langfuse_public_key", "") or "") + secret_key = str(getattr(settings, "langfuse_secret_key", "") or "") + if not (base_url and public_key and secret_key): + return [] + + token = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode() + return [ + exporter_cls( + endpoint=f"{base_url}/api/public/otel/v1/traces", + headers={"Authorization": f"Basic {token}"}, + ) + ] + + def normalize_correlation_id(candidate: str | None) -> str: value = (candidate or "").strip() return value if _SAFE_CORRELATION.fullmatch(value) else str(uuid.uuid4()) diff --git a/apps/ai-service/routers/rag.py b/apps/ai-service/routers/rag.py index 011e516..a6095e3 100644 --- a/apps/ai-service/routers/rag.py +++ b/apps/ai-service/routers/rag.py @@ -161,9 +161,24 @@ def _section_retriever(request: Request) -> SectionListRetriever: router = APIRouter(prefix="/v1/rag", tags=["rag"]) +def _clip(text: str | None, limit: int = 2000) -> str: + """Bounded text for a span attribute. + + Span attributes are not a place for unbounded content: every answer would + otherwise be sent twice per turn (Tempo and Langfuse) and stored forever. + 2000 characters is past the end of a normal formulary answer, so this + almost never truncates -- it exists for the pathological case. + """ + value = (text or "").strip() + return value if len(value) <= limit else value[: limit - 1] + "…" + + + + @router.post("/feedback", response_model=RagFeedbackResponse) def save_feedback( payload: RagFeedbackRequest, + request: Request, traces: Annotated[TraceWriter, Depends(_trace_writer)], ) -> RagFeedbackResponse: comment = payload.comment.strip() if payload.comment else None @@ -178,9 +193,53 @@ def save_feedback( raise HTTPException(status_code=404, detail="trace_not_found") from exc except Exception as exc: raise HTTPException(status_code=503, detail="feedback_store_unavailable") from exc + + _mirror_feedback_to_langfuse(request, traces, str(payload.trace_id), payload.rating, comment) return RagFeedbackResponse(feedback_id=feedback_id) +def _mirror_feedback_to_langfuse( + request: Request, + traces: TraceWriter, + trace_id: str, + rating: str, + comment: str | None, +) -> None: + """Attach the user's verdict to its Langfuse trace, or quietly do nothing. + + This is the only quality signal in the system that comes from a human + instead of a model judging a model, so it is worth having next to the + Ragas scores rather than only in Postgres. + + Runs AFTER the Postgres write and cannot affect its outcome: the durable + record is what matters, and a Langfuse outage must never turn a working + thumbs-up into a 503. Everything here is guarded -- a missing client, an + answer from before tracing was on, a database hiccup on the lookup, or a + refused post all end the same way: silently. + + 1.0/0.0 rather than a label so it averages: the mean of `user_feedback` + over a period reads directly as "share of answers users found helpful". + """ + client = getattr(request.app.state, "langfuse_scores", None) + if client is None: + return + lookup = getattr(traces, "otel_trace_id_for", None) + if lookup is None: + return + try: + otel_trace_id = lookup(trace_id) + except Exception: # noqa: BLE001 - feedback is already safely stored + return + if not otel_trace_id: + return + client.post_score( + otel_trace_id=otel_trace_id, + name="user_feedback", + value=1.0 if rating == "helpful" else 0.0, + comment=comment, + ) + + class HistoryItem(BaseModel): trace_id: str query: str @@ -614,6 +673,28 @@ def query_rag( "duocthu.citation_count": len(citations), "duocthu.generated": generated, "duocthu.persisted_trace_id": trace_id, + # Langfuse-mapped names (see its OpenTelemetry attribute mapping). + # Without these a trace shows stage timings but neither the + # question nor the answer, which is most of what makes a trace + # worth opening -- and scores attached to it unreadable, since + # "faithfulness 0.42" means nothing without the answer it judged. + # + # NOTE this is a deliberate change of posture: these spans also go + # to Tempo, and the README previously stated that full prompts and + # responses are not stored. Both destinations are self-hosted on + # this cluster and the corpus is a published formulary, so the + # exposure is bounded -- but it is a change, not an oversight, and + # the text is truncated rather than unbounded. + "langfuse.trace.name": f"rag.{decision}", + "langfuse.observation.input": _clip(payload.query), + "langfuse.observation.output": _clip(answer), + "langfuse.trace.metadata.decision": decision, + "langfuse.trace.metadata.reason": reason, + "langfuse.trace.metadata.resolved_drug_id": resolved_drug_id or "", + # Groups a multi-turn conversation into one Langfuse session, so a + # clarify turn and the answer that follows it read together + # instead of as two unrelated traces. + "langfuse.session.id": payload.conversation_id or "", } ) with stage("response"): diff --git a/apps/ai-service/scripts/run_all_evals.py b/apps/ai-service/scripts/run_all_evals.py index 41c4483..1985abc 100644 --- a/apps/ai-service/scripts/run_all_evals.py +++ b/apps/ai-service/scripts/run_all_evals.py @@ -25,6 +25,7 @@ import importlib.util import json import sys import time +import urllib.request from pathlib import Path from typing import Any @@ -44,6 +45,34 @@ def _load_battery_module(): return module +def _post_capturing_trace( + url: str, payload: dict[str, Any], timeout: float +) -> tuple[dict[str, Any], str | None]: + """`run_manual_battery._post`, but also returning the OpenTelemetry trace id. + + Langfuse attaches a score to a trace by its OTel trace id (32 hex chars), + NOT by the `traceId` in the response body -- that one is ai-service's own + Postgres row id, a UUID, and Langfuse has never heard of it. ai-service + sets the OTel id on the `X-Trace-ID` response header and the web BFF + forwards it, so the eval runner can record it per case and the Ragas + scorer can post scores that land on the right trace. + + Duplicating the POST rather than widening `_post`'s return type: three + other call sites depend on that signature, and this script deliberately + reuses the battery's *checker* -- forking the definition of "correct" is + the thing worth avoiding, not four lines of urllib. + """ + request = urllib.request.Request( + url, + data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), + headers={"Content-Type": "application/json; charset=utf-8"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + body = json.loads(response.read().decode("utf-8")) + return body, response.headers.get("X-Trace-ID") + + def run_suite( battery: Any, name: str, @@ -63,8 +92,9 @@ def run_suite( conversation_id = f"{base}-{run_id}" payload = {"content": case["query"], "conversationId": conversation_id} started = time.monotonic() + otel_trace_id = None try: - raw = battery._post(endpoint, payload, timeout) + raw, otel_trace_id = _post_capturing_trace(endpoint, payload, timeout) response = battery._normalise_response(raw) failures = battery._check(case, response) error = None @@ -78,6 +108,10 @@ def run_suite( { "suite": name, "case": case, + # None when the request failed outright, or when the + # deployment has tracing off -- the scorer treats a + # missing id as "score locally, do not push". + "otel_trace_id": otel_trace_id, "passed": ok, "failures": failures, "elapsed_seconds": elapsed, diff --git a/apps/ai-service/scripts/score_evals_ragas.py b/apps/ai-service/scripts/score_evals_ragas.py index 72d0f10..8349a1f 100644 --- a/apps/ai-service/scripts/score_evals_ragas.py +++ b/apps/ai-service/scripts/score_evals_ragas.py @@ -49,7 +49,11 @@ from __future__ import annotations import argparse import asyncio +import base64 import json +import os +import urllib.error +import urllib.request import warnings from pathlib import Path from typing import Any @@ -79,13 +83,86 @@ def _contexts_for(citations: list[dict[str, Any]]) -> list[str]: return contexts +class LangfuseScores: + """Posts one score per metric per case to Langfuse, or does nothing. + + Scores attach to a trace by its OpenTelemetry id, which `run_all_evals.py` + records as `otel_trace_id` from the `X-Trace-ID` response header. A row + without one (request failed, or the deployment had tracing off) is skipped + rather than posted against a guessed id -- a score on the wrong trace is + worse than no score, because nothing later distinguishes it from a real one. + + Failures here never abort scoring: the local .jsonl is the source of truth + and a Langfuse outage must not cost a whole Bedrock-funded run. + """ + + def __init__(self, base_url: str, public_key: str, secret_key: str, run_name: str): + 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.run_name = run_name + self.posted = 0 + self.failed = 0 + self.no_trace = 0 + + @classmethod + def from_env(cls, run_name: str) -> "LangfuseScores | None": + base_url = os.environ.get("LANGFUSE_BASE_URL", "").strip() + public_key = os.environ.get("LANGFUSE_PUBLIC_KEY", "").strip() + secret_key = os.environ.get("LANGFUSE_SECRET_KEY", "").strip() + if not (base_url and public_key and secret_key): + return None + return cls(base_url, public_key, secret_key, run_name) + + def post(self, trace_id: str | None, case_id: str, metric: str, value: float) -> None: + if not trace_id: + self.no_trace += 1 + return + body = { + "traceId": trace_id, + "name": metric, + "value": value, + "dataType": "NUMERIC", + # The run name groups one eval run's scores so two runs of the same + # suite stay comparable instead of averaging into each other. + "comment": f"ragas/{self.run_name} case={case_id}", + } + 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=20): + self.posted += 1 + except Exception as exc: # noqa: BLE001 - reported in the summary + self.failed += 1 + if self.failed <= 3: # a broken endpoint says it once, not 90 times + print(f" ! langfuse score post failed: {exc!r}", flush=True) + + async def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--input", type=Path, required=True, help="run_all_evals.py output") parser.add_argument("--output", type=Path, required=True) parser.add_argument("--limit", type=int) + parser.add_argument( + "--run-name", + default="manual", + help="labels this run's scores in Langfuse so runs stay comparable", + ) args = parser.parse_args() + langfuse = LangfuseScores.from_env(args.run_name) + print( + f"langfuse: posting scores to {langfuse.endpoint}" + if langfuse + else "langfuse: not configured (set LANGFUSE_BASE_URL/_PUBLIC_KEY/_SECRET_KEY " + "to push scores); scoring locally only", + flush=True, + ) + from langchain_aws import BedrockEmbeddings, ChatBedrockConverse from ragas import SingleTurnSample from ragas.embeddings import LangchainEmbeddingsWrapper @@ -138,13 +215,21 @@ async def main() -> int: response=answer, retrieved_contexts=contexts, ) - result: dict[str, Any] = {"id": case_id, "query": case.get("query", "")} + otel_trace_id = row.get("otel_trace_id") + result: dict[str, Any] = { + "id": case_id, + "query": case.get("query", ""), + "otel_trace_id": otel_trace_id, + } for name, metric in metrics.items(): try: result[name] = float(await metric.single_turn_ascore(sample)) except Exception as exc: # noqa: BLE001 - recorded, not hidden result[name] = None result[f"{name}_error"] = repr(exc)[:200] + else: + if langfuse: + langfuse.post(otel_trace_id, case_id, name, result[name]) scored.append(result) handle.write(json.dumps(result, ensure_ascii=False) + "\n") handle.flush() @@ -168,6 +253,12 @@ async def main() -> int: ) else: print(f" {name:<18} no successful scores") + + if langfuse: + print( + f" langfuse posted={langfuse.posted} " + f"failed={langfuse.failed} no_trace_id={langfuse.no_trace}" + ) return 0 diff --git a/apps/ai-service/tests/test_observability.py b/apps/ai-service/tests/test_observability.py index 38dacf0..4e6d87b 100644 --- a/apps/ai-service/tests/test_observability.py +++ b/apps/ai-service/tests/test_observability.py @@ -120,3 +120,104 @@ def test_otel_server_and_stage_spans_share_trace_and_emit_metric_exemplar(): assert "duocthu_stage_duration_seconds_bucket" in rendered assert 'trace_id="' in rendered import pytest + + +def test_langfuse_exporter_added_only_when_fully_configured(): + """Langfuse is opt-in: all three settings, or no extra exporter at all. + + A half-configured Langfuse (say a base URL pasted in but no keys yet) + must not produce an exporter that would fail auth on every batch. + """ + from rag.telemetry import _langfuse_exporters + + captured = [] + + def fake_exporter(**kwargs): + captured.append(kwargs) + return object() + + # Explicit empty values, not Settings() defaults: Settings reads the + # developer's real .env, which on a configured machine already carries + # live Langfuse keys and would make this assertion pass or fail by + # accident depending on whose machine runs it. + unset = Settings( + langfuse_base_url="", langfuse_public_key="", langfuse_secret_key="" + ) + assert _langfuse_exporters(unset, fake_exporter) == [] + assert ( + _langfuse_exporters( + Settings( + langfuse_base_url="https://langfuse.example", + langfuse_public_key="", + langfuse_secret_key="", + ), + fake_exporter, + ) + == [] + ) + assert captured == [] + + result = _langfuse_exporters( + Settings( + langfuse_base_url="https://langfuse.example/", + langfuse_public_key="pk-lf-public", + langfuse_secret_key="sk-lf-secret", + ), + fake_exporter, + ) + + assert len(result) == 1 + assert len(captured) == 1 + # Trailing slash stripped, so the path is not doubled. + assert captured[0]["endpoint"] == "https://langfuse.example/api/public/otel/v1/traces" + # Basic auth is public-key-as-username, secret-key-as-password. + import base64 + + expected = base64.b64encode(b"pk-lf-public:sk-lf-secret").decode() + assert captured[0]["headers"] == {"Authorization": f"Basic {expected}"} + + +def test_langfuse_score_client_is_all_or_nothing(): + """Same partial-config rule as the span exporter, for the same reason.""" + from adapters.langfuse_scores import LangfuseScoreClient + + unset = Settings( + langfuse_base_url="", langfuse_public_key="", langfuse_secret_key="" + ) + assert LangfuseScoreClient.from_settings(unset) is None + assert ( + LangfuseScoreClient.from_settings( + Settings( + langfuse_base_url="https://langfuse.example", + langfuse_public_key="pk-lf-x", + langfuse_secret_key="", + ) + ) + is None + ) + + client = LangfuseScoreClient.from_settings( + Settings( + langfuse_base_url="https://langfuse.example/", + langfuse_public_key="pk-lf-public", + langfuse_secret_key="sk-lf-secret", + ) + ) + assert client is not None + # Trailing slash stripped, so the path is not doubled. + assert client.endpoint == "https://langfuse.example/api/public/scores" + + +def test_langfuse_score_post_never_raises(monkeypatch): + """A Langfuse outage must not propagate into the feedback request path.""" + from adapters import langfuse_scores + + client = langfuse_scores.LangfuseScoreClient( + "https://langfuse.example", "pk", "sk" + ) + + def boom(*args, **kwargs): + raise OSError("connection refused") + + monkeypatch.setattr(langfuse_scores.urllib.request, "urlopen", boom) + assert client.post_score(otel_trace_id="a" * 32, name="user_feedback", value=1.0) is False diff --git a/infra/helm/langfuse/values-production.yaml b/infra/helm/langfuse/values-production.yaml index 5f92f21..e7ea6ab 100644 --- a/infra/helm/langfuse/values-production.yaml +++ b/infra/helm/langfuse/values-production.yaml @@ -20,9 +20,14 @@ # install when left unset. No manual secret entry needed for this file. langfuse: + # 512Mi crashed both web and worker with "JavaScript heap out of memory" + # right after a clean Postgres+ClickHouse connect and Next.js "Ready" -- + # Node sizes its heap off the container memory limit, and 512Mi wasn't + # enough headroom above the actual app. Bumped once, empirically, after + # confirming live on 2026-08-21; revisit if it still OOMs under real load. resources: - requests: { cpu: 100m, memory: 256Mi } - limits: { cpu: "1", memory: 512Mi } + requests: { cpu: 200m, memory: 512Mi } + limits: { cpu: "1", memory: 2Gi } ingress: enabled: true diff --git a/infra/helm/medical-chatbot/templates/ai-service.yaml b/infra/helm/medical-chatbot/templates/ai-service.yaml index 274266a..700a588 100644 --- a/infra/helm/medical-chatbot/templates/ai-service.yaml +++ b/infra/helm/medical-chatbot/templates/ai-service.yaml @@ -22,6 +22,9 @@ data: OTEL_SERVICE_NAME: ai-service OTEL_EXPORTER_OTLP_ENDPOINT: {{ printf "http://%s-otel-collector:4318/v1/traces" (include "medical-chatbot.fullname" .) | quote }} OTEL_SAMPLE_RATIO: {{ .Values.aiService.config.otelSampleRatio | quote }} + {{- if .Values.aiService.config.langfuseBaseUrl }} + LANGFUSE_BASE_URL: {{ .Values.aiService.config.langfuseBaseUrl | quote }} + {{- end }} MAX_WALL_CLOCK_MS: {{ .Values.aiService.config.maxWallClockMs | quote }} MAX_LLM_CALLS_PER_TURN: {{ .Values.aiService.config.maxLlmCallsPerTurn | quote }} --- @@ -81,6 +84,25 @@ spec: secretKeyRef: name: {{ include "medical-chatbot.secretName" . }} key: postgres-dsn + {{- if .Values.aiService.config.langfuseBaseUrl }} + # Langfuse keys are real credentials, so they come from the + # Secret, never the ConfigMap above (which is world-readable to + # anyone with namespace get access). Both are marked optional so + # a cluster that sets the base URL before creating the Secret + # starts anyway, with the exporter simply not added. + - name: LANGFUSE_PUBLIC_KEY + valueFrom: + secretKeyRef: + name: {{ include "medical-chatbot.secretName" . }} + key: langfuse-public-key + optional: true + - name: LANGFUSE_SECRET_KEY + valueFrom: + secretKeyRef: + name: {{ include "medical-chatbot.secretName" . }} + key: langfuse-secret-key + optional: true + {{- end }} {{- if .Values.aws.region }} - name: AWS_REGION value: {{ .Values.aws.region | quote }} diff --git a/infra/helm/medical-chatbot/templates/secret.yaml b/infra/helm/medical-chatbot/templates/secret.yaml index 4feac77..a5a8a32 100644 --- a/infra/helm/medical-chatbot/templates/secret.yaml +++ b/infra/helm/medical-chatbot/templates/secret.yaml @@ -14,6 +14,13 @@ stringData: aws-access-key-id: {{ .Values.aws.staticCredentials.accessKeyId | quote }} aws-secret-access-key: {{ .Values.aws.staticCredentials.secretAccessKey | quote }} {{- end }} + {{- if .Values.secret.langfusePublicKey }} + {{/* Both keys or neither: a public key without its secret authenticates + nothing, and ai-service already treats a partial config as "not + configured" (see rag/telemetry.py's _langfuse_exporters). */}} + langfuse-public-key: {{ .Values.secret.langfusePublicKey | quote }} + langfuse-secret-key: {{ required "secret.langfuseSecretKey is required when secret.langfusePublicKey is set" .Values.secret.langfuseSecretKey | quote }} + {{- end }} {{- if or .Values.authService.enabled .Values.apiGateway.enabled }} {{/* Required (not defaulted) once either service is turned on — same fail-closed posture as the image-tag guard above: a guessable or empty diff --git a/infra/helm/medical-chatbot/values-production.yaml b/infra/helm/medical-chatbot/values-production.yaml index 61a8df4..1624773 100644 --- a/infra/helm/medical-chatbot/values-production.yaml +++ b/infra/helm/medical-chatbot/values-production.yaml @@ -52,6 +52,22 @@ aiService: answerProvider: bedrock-converse answerModelId: qwen.qwen3-next-80b-a3b rerankEnabled: true + # Second trace destination alongside Tempo, for the LLM-shaped view + # (per-trace answers + eval scores). Self-hosted on this same cluster by + # the separate `langfuse` ArgoCD Application. The keys it needs are + # secrets and stay inline on the Application, like jwtSecret does — + # setting this URL alone changes nothing until they exist. + # + # The in-cluster Service, NOT https://langfuse.realvuxbaro.me. That public + # name resolves to the Elastic IP of the very node these Pods run on, so a + # Pod reaching it has to hairpin out and back through the node's own + # public address — which silently never completes here. Spans were created + # and dropped with no error in ai-service's log and nothing arriving on + # Langfuse's otel-ingestion-queue; an identical span sent from a laptop + # over the public URL ingested fine, which is what isolated it to the + # cluster-internal hop. Cross-namespace is fine: Services resolve + # cluster-wide, unlike Secrets. + langfuseBaseUrl: http://langfuse-web.medical-chatbot-data.svc.cluster.local:3000 # Auth is live in production (2026-08-19). These stayed false here for a day # while the live Application carried `enabled: true` inline, so Git and the diff --git a/infra/helm/medical-chatbot/values.yaml b/infra/helm/medical-chatbot/values.yaml index 0c4e1bc..a6b3520 100644 --- a/infra/helm/medical-chatbot/values.yaml +++ b/infra/helm/medical-chatbot/values.yaml @@ -17,6 +17,11 @@ secret: # Set when postgres runs in a different Application/release (the # data/app split pattern) — overrides the default in-release host. postgresHost: "" + # Langfuse project API keys. Real credentials -- set these inline on the + # ArgoCD Application or a pre-created Secret, never in a tracked values + # file. Leaving them empty simply means no Langfuse export. + langfusePublicKey: "" + langfuseSecretKey: "" grafanaAdminPassword: change-me # Required (chart render fails without it) once authService or apiGateway # is enabled — signs/verifies every JWT. Must be the same value both @@ -76,6 +81,11 @@ aiService: metricsEnabled: true otelEnabled: true otelSampleRatio: 1.0 + # Empty by default: Langfuse is an optional second trace destination. + # Setting this alone is inert -- the keys live in the Secret (see + # `secret.langfusePublicKey`), and ai-service adds the exporter only when + # URL + both keys are present. + langfuseBaseUrl: "" maxWallClockMs: 40000 maxLlmCallsPerTurn: 8 migration: