"""Run every eval dataset in `evals/` against a deployment, in one command. Before this, only `production_manual_60.jsonl` had a working runner. The other two datasets sat unreferenced by any code: one's intended consumer was a metrics dataclass with no entry point, the other's (`rag/run_eval.py`) had been unable to load the corpus since the ingestion artifact format changed. So two thirds of the checked-in eval coverage was never actually checked, and nothing in the repo said so. Reuses `run_manual_battery.py`'s checker rather than reimplementing it -- a second copy of the invariants would drift from the first, and the point of this script is to widen coverage, not to fork the definition of "correct". `evals/adapters.py` holds the per-dataset schema mapping and the reasoning behind each choice. Usage: python3 scripts/run_all_evals.py --base-url https://realvuxbaro.me \\ --output-dir /tmp/evals """ from __future__ import annotations import argparse import importlib.util import json import sys import time import urllib.request from pathlib import Path from typing import Any SERVICE_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(SERVICE_ROOT)) from evals.adapters import SUITES # noqa: E402 def _load_battery_module(): """Import the sibling script by path -- `scripts/` is not a package.""" path = Path(__file__).resolve().parent / "run_manual_battery.py" spec = importlib.util.spec_from_file_location("run_manual_battery", path) module = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(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, cases: list[dict[str, Any]], endpoint: str, timeout: float, run_id: str, output: Path, ) -> tuple[int, int]: passed = 0 with output.open("w", encoding="utf-8") as handle: for index, case in enumerate(cases, start=1): # Split out rather than nested: reusing the outer quote inside an # f-string is a 3.12+ syntax, and ruff targets 3.11 here. default_conversation = f"{name}-{case['id']}" base = case.get("conversation_id") or default_conversation conversation_id = f"{base}-{run_id}" payload = {"content": case["query"], "conversationId": conversation_id} started = time.monotonic() otel_trace_id = None try: raw, otel_trace_id = _post_capturing_trace(endpoint, payload, timeout) response = battery._normalise_response(raw) failures = battery._check(case, response) error = None except Exception as exc: # noqa: BLE001 - recorded, not swallowed response, failures, error = {}, [f"request error: {exc}"], repr(exc) elapsed = round(time.monotonic() - started, 3) ok = not failures passed += int(ok) handle.write( json.dumps( { "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, "error": error, "response": response, }, ensure_ascii=False, ) + "\n" ) handle.flush() print( f" [{index:02d}/{len(cases)}] {case['id']} " f"{'PASS' if ok else 'FAIL'} {elapsed:.1f}s {'; '.join(failures)}", flush=True, ) return passed, len(cases) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--base-url", required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--timeout", type=float, default=120.0) parser.add_argument("--suites", help="comma-separated subset of: " + ", ".join(SUITES)) parser.add_argument("--run-id", default=str(int(time.time()))) args = parser.parse_args() wanted = ( [s.strip() for s in args.suites.split(",") if s.strip()] if args.suites else list(SUITES) ) unknown = [s for s in wanted if s not in SUITES] if unknown: print(f"unknown suite(s): {unknown}; known: {list(SUITES)}", file=sys.stderr) return 2 battery = _load_battery_module() endpoint = args.base_url.rstrip("/") + "/api/chat" args.output_dir.mkdir(parents=True, exist_ok=True) totals: list[tuple[str, int, int]] = [] for name in wanted: cases = SUITES[name]() print(f"\n=== {name} ({len(cases)} cases) ===", flush=True) passed, total = run_suite( battery, name, cases, endpoint, args.timeout, args.run_id, args.output_dir / f"{name}.jsonl", ) totals.append((name, passed, total)) print("\n=== SUMMARY ===") grand_passed = grand_total = 0 for name, passed, total in totals: print(f" {name:<16} {passed}/{total}") grand_passed += passed grand_total += total print(f" {'TOTAL':<16} {grand_passed}/{grand_total}") return 0 if grand_passed == grand_total else 1 if __name__ == "__main__": raise SystemExit(main())