"""Score recorded eval responses with Ragas, using Bedrock as the judge. `run_all_evals.py` answers "did the service break a rule" -- decision, citation presence, drug provenance. It cannot answer "was the answer any good", so a change to retrieval or prompting can degrade quality while every invariant still passes. This fills that gap. Three metrics, chosen because the datasets carry no reference answers and any metric needing one (context recall, answer correctness) would be unmeasurable: faithfulness -- is every claim in the answer supported by the cited evidence? The hallucination check. context_precision -- were the retrieved chunks actually relevant, or did useful evidence arrive buried in noise? A retrieval check, which faithfulness alone cannot see: an answer can be perfectly faithful to one good chunk that arrived alongside nine useless ones. answer_relevancy -- does the answer address the question asked? Catches a grounded, well-cited answer to a different question. CONTEXT MUST CARRY THE DRUG NAME. Each citation's `evidence_text` is the raw section prose, which frequently never repeats the drug it belongs to ("Tăng huyết áp (dùng đơn trị liệu...)"). The service knows the drug from a separate field; a judge handed the bare text does not. Scoring a multi-drug answer that way on 2026-08-18 produced faithfulness 0.251 -- every claim marked unsupported because no context could be attributed to any drug -- and the same run scored 1.000 once `[drug_name]` was prefixed. That was a defect in the measurement, not the service, and it is exactly the kind of error that gets reported as a model regression. Hence `_contexts_for`. Only `answerable` turns are scored: an abstain or a clarify has no claims to be faithful to, and scoring them would drag the mean around with values that mean nothing. Runs against recorded output, so it never re-queries production and can be re-run offline as often as needed. Requires ragas + langchain-aws, which conflict with this service's own pinned dependencies -- install them in a separate virtualenv and run this with that interpreter. See evals/README-ragas.md. Usage: /python scripts/score_evals_ragas.py \\ --input /tmp/evals/production60.jsonl \\ --output /tmp/evals/production60.ragas.jsonl """ 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 warnings.filterwarnings("ignore", category=DeprecationWarning) ANSWER_MODEL = "qwen.qwen3-next-80b-a3b" # NOT cohere-v4, which production uses for retrieval: langchain-aws cannot # parse v4's response envelope and raises a bare KeyError(0). v3-multilingual # is the right substitute anyway -- this embedder only measures how close the # answer sits to the question, never touching the indexed corpus, so it does # not need to match the retrieval model. Multilingual matters more here, the # corpus and questions both being Vietnamese. EMBED_MODEL = "cohere.embed-multilingual-v3" REGION = "us-east-1" def _contexts_for(citations: list[dict[str, Any]]) -> list[str]: """Label every context with its drug -- see the module docstring.""" contexts = [] for citation in citations: text = (citation.get("evidence_text") or "").strip() if not text: continue name = citation.get("drug_name") or citation.get("drug_id") or "" contexts.append(f"[{name}] {text}" if name else text) 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 from ragas.llms import LangchainLLMWrapper from ragas.metrics import ( Faithfulness, LLMContextPrecisionWithoutReference, ResponseRelevancy, ) judge = LangchainLLMWrapper( ChatBedrockConverse(model=ANSWER_MODEL, region_name=REGION, temperature=0) ) embedder = LangchainEmbeddingsWrapper( BedrockEmbeddings(model_id=EMBED_MODEL, region_name=REGION) ) metrics = { "faithfulness": Faithfulness(llm=judge), "context_precision": LLMContextPrecisionWithoutReference(llm=judge), "answer_relevancy": ResponseRelevancy(llm=judge, embeddings=embedder), } rows = [ json.loads(line) for line in args.input.read_text(encoding="utf-8").splitlines() if line.strip() ] if args.limit: rows = rows[: args.limit] scored: list[dict[str, Any]] = [] skipped = 0 args.output.parent.mkdir(parents=True, exist_ok=True) with args.output.open("w", encoding="utf-8") as handle: for index, row in enumerate(rows, start=1): case, response = row.get("case", {}), row.get("response") or {} case_id = case.get("id", f"row{index}") decision = response.get("decision") answer = (response.get("answer") or "").strip() contexts = _contexts_for(response.get("citations") or []) if decision != "answerable" or not answer or not contexts: skipped += 1 print(f"[{index:02d}/{len(rows)}] {case_id} SKIP ({decision})", flush=True) continue sample = SingleTurnSample( user_input=case.get("query", ""), response=answer, retrieved_contexts=contexts, ) 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() print( f"[{index:02d}/{len(rows)}] {case_id} " + " ".join( f"{n}={result[n]:.2f}" if result[n] is not None else f"{n}=ERR" for n in metrics ), flush=True, ) print(f"\n=== {args.input.name}: scored {len(scored)}, skipped {skipped} ===") for name in metrics: values = [r[name] for r in scored if r.get(name) is not None] if values: worst = min(values) print( f" {name:<18} mean={sum(values) / len(values):.3f} " f"min={worst:.3f} n={len(values)}" ) 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 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))