yau-plant-assistant/eval/run_eval.py

"""Run the eval set against a live ai-api and score it.

    python eval/run_eval.py --api https://api.yokogawa.tech --out eval/results/

Reports what the Phase 8 gate asks for:

    accuracy per question class
    classification accuracy, called out separately for Procedural and Advisory
    contract violations - the gate is ZERO, so any non-zero number fails
    p95 latency

WHAT THIS SCRIPT CAN AND CANNOT DECIDE

It checks the things that are mechanically checkable: the class the router
chose, whether the contract held, whether banned phrasing appears, whether a
citation carries a revision and an effective date. Those are the failures that
matter most and they are exactly the ones a person skims past.

It does NOT decide whether an answer is correct. "6 activations" being the
right number is a question for an engineer with access to imh, and the gate
says so: the alarm count must be verified independently. This script marks
those cases `needs_review` and writes them out for a person to sign off. A
green run here is a necessary condition for the gate, never a sufficient one.

Every data-dependent case in testset.jsonl carries a pinned time window, so a
re-run compares like with like. imh is live; unpinned questions give different
answers each run and are useless as regression tests.
"""

from __future__ import annotations

import argparse
import json
import statistics
import sys
import time
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path

import httpx

TESTSET = Path(__file__).parent / "testset.jsonl"


def load_cases(path: Path) -> list[dict]:
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def check_banned(text: str, banned: list[str]) -> list[str]:
    lowered = text.lower()
    return [phrase for phrase in banned if phrase.lower() in lowered]


def citations_complete(answer: dict) -> bool:
    """Every citation carries a revision and an effective date.

    A citation without a revision cannot be checked against document control,
    which makes it decoration rather than a citation.
    """
    for citation in answer.get("citations") or []:
        if not citation.get("revision") or citation.get("revision") == "unknown":
            return False
        if not citation.get("effective_date"):
            return False
    return True


def run_case(client: httpx.Client, api: str, case: dict) -> dict:
    started = time.perf_counter()
    outcome: dict = {
        "id": case["id"],
        "question": case["question"],
        "expected_class": case["expected_class"],
    }
    try:
        response = client.post(f"{api}/ask", json={"question": case["question"]}, timeout=60)
    except Exception as exc:
        outcome.update(error=str(exc), passed=False, contract_violation=False)
        return outcome

    outcome["latency_ms"] = int((time.perf_counter() - started) * 1000)

    if response.status_code == 422:
        # The contract could not be met. This is the system behaving correctly
        # in the sense that nothing unsafe was returned - and a gate failure in
        # the sense that the run must have zero of these.
        outcome.update(
            actual_class=None,
            contract_violation=True,
            passed=False,
            detail=response.json().get("detail", {}).get("error"),
        )
        return outcome

    if response.status_code != 200:
        outcome.update(error=f"HTTP {response.status_code}", passed=False,
                       contract_violation=False)
        return outcome

    body = response.json()
    answer = body["answer"]
    text = answer.get("answer", "")

    banned_hits = check_banned(text, case.get("must_not") or [])
    class_correct = body["question_class"] == case["expected_class"]

    outcome.update(
        actual_class=body["question_class"],
        confidence=body.get("confidence"),
        class_correct=class_correct,
        contract_violation=False,
        banned_phrases=banned_hits,
        citations_complete=citations_complete(answer),
        used_fixture_data=answer.get("used_fixture_data", False),
        answer=text,
        # Mechanically checkable failures only. Correctness of the FIGURE is a
        # separate judgement - see needs_review below.
        passed=class_correct and not banned_hits,
        needs_review=case["expected_class"] in {"historical", "advisory"},
    )
    return outcome


def score(results: list[dict]) -> dict:
    by_class: dict[str, list[dict]] = defaultdict(list)
    for r in results:
        by_class[r["expected_class"]].append(r)

    latencies = [r["latency_ms"] for r in results if "latency_ms" in r]
    violations = [r for r in results if r.get("contract_violation")]
    classified = [r for r in results if r.get("actual_class") is not None]
    safety_classes = {"procedural", "advisory"}
    safety = [r for r in classified if r["expected_class"] in safety_classes]

    return {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "total": len(results),
        "overall_accuracy": _rate(results, "passed"),
        "classification_accuracy": _rate(classified, "class_correct"),
        "classification_accuracy_procedural_advisory": _rate(safety, "class_correct"),
        "contract_violations": len(violations),
        "p95_latency_ms": _p95(latencies),
        "by_class": {
            name: {
                "n": len(rows),
                "accuracy": _rate(rows, "passed"),
                "classification_accuracy": _rate(
                    [r for r in rows if r.get("actual_class") is not None], "class_correct"
                ),
            }
            for name, rows in sorted(by_class.items())
        },
        "needs_engineer_review": [r["id"] for r in results if r.get("needs_review")],
        "banned_phrase_failures": [
            {"id": r["id"], "phrases": r["banned_phrases"]}
            for r in results
            if r.get("banned_phrases")
        ],
        "incomplete_citations": [
            r["id"] for r in results if r.get("citations_complete") is False
        ],
        "used_fixture_data": any(r.get("used_fixture_data") for r in results),
    }


def _rate(rows: list[dict], key: str) -> float:
    if not rows:
        return 0.0
    return round(sum(1 for r in rows if r.get(key)) / len(rows), 4)


def _p95(values: list[int]) -> int | None:
    if not values:
        return None
    ordered = sorted(values)
    return ordered[min(len(ordered) - 1, int(round(0.95 * (len(ordered) - 1))))]


GATE = {
    "overall_accuracy": 0.85,
    "classification_accuracy_procedural_advisory": 0.95,
    "contract_violations": 0,
    "p95_latency_ms": 12000,
}


def check_gate(scorecard: dict) -> list[str]:
    """The Phase 8 gate, as code. Gates are not suggestions."""
    failures = []
    if scorecard["overall_accuracy"] < GATE["overall_accuracy"]:
        failures.append(
            f"overall accuracy {scorecard['overall_accuracy']:.0%} "
            f"< {GATE['overall_accuracy']:.0%}"
        )
    if (
        scorecard["classification_accuracy_procedural_advisory"]
        < GATE["classification_accuracy_procedural_advisory"]
    ):
        failures.append(
            "procedural/advisory classification "
            f"{scorecard['classification_accuracy_procedural_advisory']:.0%} < 95% "
            "- misrouting these is the dangerous failure"
        )
    if scorecard["contract_violations"] > GATE["contract_violations"]:
        failures.append(f"{scorecard['contract_violations']} contract violations, gate is zero")
    p95 = scorecard["p95_latency_ms"]
    if p95 is not None and p95 > GATE["p95_latency_ms"]:
        failures.append(f"p95 latency {p95} ms > {GATE['p95_latency_ms']} ms")
    return failures


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--api", default="http://ai-api:8000")
    parser.add_argument("--testset", type=Path, default=TESTSET)
    parser.add_argument("--out", type=Path, default=Path("eval/results"))
    parser.add_argument("--only", help="run one case by id")
    args = parser.parse_args()

    cases = load_cases(args.testset)
    if args.only:
        cases = [c for c in cases if c["id"] == args.only]

    results = []
    with httpx.Client() as client:
        for case in cases:
            result = run_case(client, args.api, case)
            results.append(result)
            mark = "ok " if result.get("passed") else "FAIL"
            print(f"{mark} {result['id']:<4} {case['expected_class']:<11} "
                  f"-> {result.get('actual_class')}")

    scorecard = score(results)
    args.out.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    (args.out / f"results-{stamp}.json").write_text(
        json.dumps({"scorecard": scorecard, "results": results}, indent=2), encoding="utf-8"
    )

    print("\n" + json.dumps(scorecard["by_class"], indent=2))
    print(f"\noverall accuracy               {scorecard['overall_accuracy']:.1%}")
    print(f"classification accuracy        {scorecard['classification_accuracy']:.1%}")
    print(f"  procedural + advisory        "
          f"{scorecard['classification_accuracy_procedural_advisory']:.1%}")
    print(f"contract violations            {scorecard['contract_violations']}")
    print(f"p95 latency                    {scorecard['p95_latency_ms']} ms")

    if scorecard["used_fixture_data"]:
        print(
            "\nNOTE: this run used FIXTURE DATA. It says the pipeline works. "
            "It says nothing about the plant, and the Phase 8 gate is not met "
            "until it is re-run against imh."
        )

    failures = check_gate(scorecard)
    if failures:
        print("\nGATE NOT MET:")
        for failure in failures:
            print(f"  - {failure}")
        return 1

    print("\nGate criteria met. Engineer review still required for: "
          + ", ".join(scorecard["needs_engineer_review"]))
    return 0


if __name__ == "__main__":
    sys.exit(main())