duocthu/apps/ai-service/scripts/run_all_evals.py

"""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())