218 lines
9.0 KiB
Python
218 lines
9.0 KiB
Python
"""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()
|