"""Run and record the PDF-derived 60-turn production chat battery. This is intentionally a transparent HTTP recorder, not an LLM judge. Each case has observable invariants (decision, relation/section, candidate bound, citations and drug provenance). The JSONL output keeps every full response for subsequent human review against the rendered PDF pages. """ from __future__ import annotations import argparse import json import time import urllib.error import urllib.request from pathlib import Path from typing import Any def _read_cases(path: Path) -> list[dict[str, Any]]: return [ json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip() ] def _post(url: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]: 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: return json.loads(response.read().decode("utf-8")) def _normalise_response(raw: dict[str, Any]) -> dict[str, Any]: message = raw.get("message") if not isinstance(message, dict): return raw return { "trace_id": message.get("traceId"), "decision": message.get("decision"), "reason": message.get("reason"), "answer": message.get("content"), "resolved_drug_id": message.get("resolvedDrugId"), "citations": [ { "chunk_id": item.get("chunkId"), "drug_name": item.get("drugName"), "section_key": item.get("sectionType"), "evidence_text": item.get("snippet", ""), "physical_page": item.get("physicalPage"), "printed_page_start": (item.get("sourcePageRange") or [None])[0], } for item in message.get("citations", []) ], "candidate_assessments": [ { "drug_id": item.get("drugId"), "drug_name": item.get("drugName"), "status": item.get("status"), "indication_source_ids": item.get("indicationSourceIds", []), "safety_source_ids": item.get("safetySourceIds", []), } for item in message.get("candidateAssessments", []) ], } def _drug_id_from_chunk(chunk_id: str | None) -> str | None: return chunk_id.split("__", 1)[0] if chunk_id and "__" in chunk_id else None def _check(case: dict[str, Any], response: dict[str, Any]) -> list[str]: failures: list[str] = [] decision = response.get("decision") allowed = case.get("decision_any", [case.get("decision")]) if decision not in allowed: failures.append(f"decision={decision!r}, expected={allowed!r}") if case.get("reason") and response.get("reason") != case["reason"]: failures.append(f"reason={response.get('reason')!r}") citations = response.get("citations") or [] assessments = response.get("candidate_assessments") or [] if case.get("no_citations") and citations: failures.append("expected no citations") if case.get("must_have_citations") and not citations: failures.append("missing citations") if case.get("must_have_citations_if_answerable") and decision == "answerable" and not citations: failures.append("answerable without citations") if case.get("condition_mode") == "general": bad = [item.get("section_key") for item in citations if item.get("section_key") != "chi_dinh"] if bad: failures.append(f"general reverse lookup cited non-indication sections: {bad}") if case.get("condition_mode") == "patient" and not assessments: failures.append("patient query missing candidate assessments") if case.get("require_patient_assessment") and not assessments: failures.append("missing patient assessment") if case.get("require_patient_assessment_if_answerable") and decision == "answerable" and not assessments: failures.append("answerable follow-up lost patient assessment") candidate_ids = {item.get("drug_id") for item in assessments if item.get("drug_id")} resolved = response.get("resolved_drug_id") or "" resolved_ids = {part.strip() for part in resolved.split(",") if part.strip()} cited_ids = { drug_id for drug_id in (_drug_id_from_chunk(item.get("chunk_id")) for item in citations) if drug_id } observed_ids = candidate_ids | resolved_ids | cited_ids expected_any = set(case.get("expected_any_drug_ids", [])) if expected_any and not observed_ids.intersection(expected_any): failures.append(f"none of expected drugs observed: {sorted(expected_any)}") if candidate_ids and not cited_ids.issubset(candidate_ids): failures.append(f"citation drug outside candidate set: {sorted(cited_ids - candidate_ids)}") if case.get("max_drugs") and len(candidate_ids or cited_ids) > case["max_drugs"]: failures.append(f"too many drugs: {len(candidate_ids or cited_ids)}") interaction_terms = [term.casefold() for term in case.get("interaction_terms", [])] interaction_hits = [ item for item in citations if item.get("section_key") == "tuong_tac_thuoc" ] for hit in interaction_hits: text = str(hit.get("evidence_text", "")).casefold() if interaction_terms and not any(term in text for term in interaction_terms): failures.append("interaction citation does not mention a current medication") answer = str(response.get("answer") or "").casefold() forbidden = ("first-line", "đầu tay", "lựa chọn tốt nhất", "phác đồ chuẩn") found = [term for term in forbidden if term in answer] if found: failures.append(f"unsupported guideline language: {found}") return failures def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--base-url", required=True) parser.add_argument("--target", choices=("ai", "web"), default="web") parser.add_argument( "--cases", type=Path, default=Path(__file__).resolve().parents[1] / "evals/production_manual_60.jsonl", ) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--timeout", type=float, default=70.0) parser.add_argument("--start", type=int, default=1, help="1-based first case") parser.add_argument("--ids", help="comma-separated case ids") parser.add_argument("--run-id", default=str(int(time.time()))) parser.add_argument("--limit", type=int) args = parser.parse_args() cases = _read_cases(args.cases) if args.ids: wanted = {item.strip() for item in args.ids.split(",") if item.strip()} cases = [case for case in cases if case["id"] in wanted] cases = cases[max(0, args.start - 1):] if args.limit: cases = cases[: args.limit] endpoint = args.base_url.rstrip("/") + ( "/v1/rag/query" if args.target == "ai" else "/api/chat" ) args.output.parent.mkdir(parents=True, exist_ok=True) passed = 0 started_all = time.monotonic() with args.output.open("w", encoding="utf-8") as handle: for index, case in enumerate(cases, start=1): base_conversation = case.get("conversation_id") or f"manual-{case['id']}" conversation_id = f"{base_conversation}-{args.run_id}" payload = ( { "query": case["query"], "subject_scope": "human", "intent": "fact_lookup", "conversation_id": conversation_id, } if args.target == "ai" else {"content": case["query"], "conversationId": conversation_id} ) started = time.monotonic() try: raw = _post(endpoint, payload, args.timeout) response = _normalise_response(raw) failures = _check(case, response) error = None except (OSError, urllib.error.HTTPError, ValueError) as exc: response = {} failures = [f"request error: {exc}"] error = repr(exc) elapsed = round(time.monotonic() - started, 3) ok = not failures passed += int(ok) record = { "case": case, "passed": ok, "failures": failures, "elapsed_seconds": elapsed, "error": error, "response": response, } handle.write(json.dumps(record, ensure_ascii=False) + "\n") handle.flush() print( f"[{index:02d}/{len(cases)}] {case['id']} " f"{'PASS' if ok else 'FAIL'} {elapsed:.1f}s " f"{'; '.join(failures)}", flush=True, ) elapsed_all = time.monotonic() - started_all print(f"SUMMARY {passed}/{len(cases)} passed in {elapsed_all:.1f}s") if __name__ == "__main__": main()