#!/usr/bin/env python3 """Axis 5 (abstention): LLM discovery with a deterministic absence check. Stage 1: the LLM reads the patient's conditions and proposes (anchor, expected but absent companion) pairs. Stage 2: the absent entity's pattern must have zero hits over the whole event chain and evidence chain, otherwise the pair is rejected. Golds come from GOLD_PLAIN. synthesize() authors synthetic questions. Stage G additionally checks the absent entity against the dialogues (pipeline.validation). python -m pipeline.questions.axis5 discovery -> _discover_axis5.json """ from __future__ import annotations import json import re from pathlib import Path from pipeline.common import BASE, FIELDS, Progress, has_text, model, patient_context, patient_id, patient_line from pipeline.llm import ChatClient PATIENT = patient_id() ENGINE = model("generation") AXIS = 5 _ARTICLE = re.compile(r"^(?:an?|the)\s+", re.I) GOLD_PLAIN = ("Not documented — {absent_entity} was never recorded anywhere in this patient's chart. " "The closest documented fact is {anchor_entity}: {anchor_value} ({anchor_department}), " "first documented {anchor_valid_from_date}.") def regex_covers(rx, absent_entity): """Whether the probe's regex matches the absent entity's own name: the name without its article and each alternative given in parentheses, e.g. 'a dialysis adequacy measurement (Kt/V or URR)'.""" bare = _ARTICLE.sub("", str(absent_entity or "")).strip() names = [re.sub(r"\s*\([^)]*\)", "", bare).strip()] for inner in re.findall(r"\(([^)]+)\)", bare): names += [x.strip() for x in re.split(r"\s+or\s+|,\s*", inner) if x.strip()] try: return all(re.search(rx, n, re.I) for n in names if n) except re.error: return False DISCOVER_SYS = ( "You design abstention probes for a medical-memory benchmark. A probe is a question about something\n" "that a clinician would reasonably expect to find in this patient's chart, given what is\n" "documented, but that is in fact never recorded. The gold answer is 'not documented'. Return only\n" "JSON.") def discover_pairs(cli, tl): """Read the patient's conditions and propose (anchor, absent entity) pairs.""" rows = [f"- {t.get('entity')!r} ({t.get('aspect')}, {t.get('department')}): " f"{t.get('current_status')!r}" for t in tl if t.get("aspect") in ("disease_state", "medication", "test_result", "procedure_or_followup", "constraint_or_safety")] anchors = "\n".join(rows[:120]) usr = ( "Documented conditions and treatments, the anchors available:\n" + anchors + "\n\n" "Propose at least 20 abstention probes. Each pairs a documented anchor from the list above with a\n" "companion test, treatment or assessment that a clinician would expect to follow from that anchor\n" "but that may be absent. The pairing must make the absence notable, such as a genuine care gap or\n" "an expected work-up, and never trivial. Prefer anchors specific to this patient's disease.\n\n" "Many of your guesses will turn out to be present in the chart and are then discarded. So propose\n" "generously, and prefer specific companions, such as a named scoring scale, a specific\n" "confirmatory assay or a specific staging study, over broad categories that a rich chart probably\n" "contains somewhere. Aim for about 20 to 30, so that enough survive verification.\n\n" "For each probe give the anchor, a companion entity that you believe is absent, and a broad regex\n" "that matches any mention of that companion, including synonyms and abbreviations. The regex must\n" "also match the companion's own name as you write it in absent_entity. It is checked against the\n" "whole chart, and the probe is dropped if it matches anything.\n\n" "Return JSON in this form:\n" '{"probes": [{\n' ' "anchor_entity": "",\n' ' "anchor_status": "",\n' ' "absent_entity": "",\n' ' "absent_regex": "",\n' ' "key": ""\n' "}]}") return cli.complete_json(DISCOVER_SYS, usr).get("probes") or [] def discover() -> None: """Discover the axis-5 candidates of this patient -> _discover_axis5.json.""" cli = ChatClient(model=ENGINE, reasoning_effort="medium", max_completion_tokens=9000) tl = [json.loads(l) for l in open(f"{BASE}/evidence_chain.jsonl") if l.strip()] chain = [json.loads(l) for l in open(f"{BASE}/event_chain.jsonl") if l.strip()] by_v = {r["visit_number"]: r for r in chain} plast = max(r["date"] for r in chain) hay = "\n".join(str(a) for r in chain for f in FIELDS for a in (r.get(f) or [])) evidence_hay = "\n".join(f"{t.get('entity')} {t.get('current_status')} " + " ".join(str(s.get('value')) for s in (t.get('segments') or [])) for t in tl) def anchor_of(name): for t in tl: if t.get("entity") == name and t.get("segments"): vf = t["segments"][0].get("valid_from") or {} return (vf.get("visit_number"), vf.get("date"), str(t["segments"][0].get("value") or t.get("current_status") or ""), t.get("department")) hits = [t for t in tl if name and name.lower() in str(t.get("entity", "")).lower() and t.get("segments")] if len(hits) == 1: vf = hits[0]["segments"][0].get("valid_from") or {} return (vf.get("visit_number"), vf.get("date"), str(hits[0]["segments"][0].get("value") or hits[0].get("current_status") or ""), hits[0].get("department")) return None with Progress("axis5 discover (LLM)", total=1, out=f"{BASE}/progress_axis5_discover.json") as pg: probes = discover_pairs(cli, tl) pg.tick(ok=bool(probes)) candidates, rejected = [], [] for p in probes: rx = p.get("absent_regex") try: n_chain = len(re.findall(rx, hay, re.I)) n_evidence = len(re.findall(rx, evidence_hay, re.I)) except re.error: rejected.append((p.get("key"), "bad regex")) continue if n_chain or n_evidence: rejected.append((p.get("key"), f"present: chain={n_chain} evidence={n_evidence}")) continue if not regex_covers(rx, p.get("absent_entity", "")): rejected.append((p.get("key"), "regex does not match the absent entity's own name")) continue a = anchor_of(p.get("anchor_entity")) if not a or a[0] not in by_v: rejected.append((p.get("key"), "anchor not found")) continue host, adate, aval, adept = a bare = _ARTICLE.sub("", p.get("absent_entity", "")) ops = {"absent_entity": p.get("absent_entity"), "absent_entity_bare": bare, "anchor_entity": p.get("anchor_entity"), "anchor_value": aval or p.get("anchor_status"), "anchor_department": adept, "anchor_valid_from_date": adate, "absent_regex": rx, "key": p.get("key"), "n_support": 1} gold = GOLD_PLAIN.format(**ops) candidates.append({ "qid": f"{PATIENT}/axis{AXIS}/{len(candidates):03d}", "patient_id": PATIENT, "axis": "5_abstention", "question": f"What was the result of the patient's most recent {bare}?", "answer": gold, "answer_type": "abstain", "grounding": "record", "key": p.get("key"), "evidence_visits": [host], "gt_visit": host, "gt_date": adate, "question_appearance_window": {"from": adate, "to": plast}, "derivation": { "evidence": [{"visit": host, "date": adate, "field": "evidence_chain", "atom": aval}], "steps": [f"Anchor documented v{host}: {p.get('anchor_entity')} = {ops['anchor_value']!r}.", f"A chart documenting {p.get('anchor_entity')} would be expected to also document " f"{p.get('absent_entity')}.", f"/{rx}/ matches 0 atoms across the whole chain and evidence_chain.", "Therefore the correct answer is that it was never documented."], "gold": gold, "rivals": [], "operands": ops, "derived": True}, "provenance": f"LLM-discovered abstention ({ENGINE}); absence verified, 0 hits over the event and " f"evidence chains", "axis_specific": {"absent_entity": p.get("absent_entity"), "anchor_entity": p.get("anchor_entity")}, }) out = {"axis": AXIS, "patient_id": PATIENT, "n_record": len(candidates), "candidates": candidates} Path(f"{BASE}/_discover_axis{AXIS}.json").write_text(json.dumps(out, ensure_ascii=False, indent=1)) print(f"\npatient {PATIENT} engine={ENGINE}") print(f" probes proposed (LLM) : {len(probes)}") print(f" rejected (present / bad) : {len(rejected)}") for k, why in rejected[:6]: print(f" {str(k):<28} {why}") print(f" CANDIDATES (absence verified): {len(candidates)} -> {BASE}/_discover_axis{AXIS}.json") def synthesize(cli, chain, by_v, forbidden, n, batch): """Author an anchor whose expected companion is absent from the record, verifying the absence against the record's event chain.""" hay = "\n".join(str(x) for r in chain for f in FIELDS for x in (r.get(f) or [])) out, vs = [], sorted(by_v) for i in range(n): host = vs[int(len(vs) * (0.2 + 0.5 * i / max(1, n)))] dept = by_v[host]["department"] host_date = by_v[host]["date"] patient, context = patient_line(), patient_context(chain) forbidden_list = "\n".join(f"- {e}" for e in forbidden[:50]) r = cli.complete_json( "You design one abstention probe for a medical-memory benchmark. Return only JSON.", f"{patient}\n{context}\n" f"Visit: v{host} · {host_date} · {dept}\n\n" "Give (a) an anchor that this department would plausibly document, and (b) a companion test that\n" "a clinician would then expect but that this chart must never have recorded. The absence is the\n" "answer, so the pairing must make the absence notable, not trivial. The regex must also match the\n" "missing test's own name as you write it in absent_entity.\n\n" "Do not use these, which existing answer keys already track:\n" + forbidden_list + "\n\nReturn JSON in this form:\n" '{"anchor_text": "",\n' ' "anchor_entity": "",\n' ' "anchor_value": "",\n' ' "absent_entity": "",\n' ' "absent_regex": "",\n' ' "topic": ""}') missing = [k for k in ("anchor_text", "anchor_entity", "anchor_value", "absent_entity", "absent_regex") if not has_text(r, k)] if missing: print(f" axis5[{i}] rejected: the reply lacks {', '.join(missing)}", flush=True) continue try: hits = len(re.findall(r["absent_regex"], hay, re.I)) except re.error: hits = 1 if hits: print(f" axis5[{i}] rejected: /{r['absent_regex']}/ matches {hits}x; not absent", flush=True) continue if not regex_covers(r["absent_regex"], r["absent_entity"]): print(f" axis5[{i}] rejected: /{r['absent_regex']}/ does not match {r['absent_entity']!r}", flush=True) continue ops = {"absent_entity": r["absent_entity"], "absent_entity_bare": _ARTICLE.sub("", r["absent_entity"]), "anchor_entity": r["anchor_entity"], "anchor_value": r["anchor_value"], "anchor_department": dept, "anchor_valid_from_date": by_v[host]["date"], "absent_regex": r["absent_regex"], "key": r.get("topic"), "n_support": 1, "hits_chain": 0, "hits_evidence": 0} atom = {"atom_id": f"{batch}/{len(out):03d}", "batch_id": batch, "axis": "5", "visit_number": host, "date": by_v[host]["date"], "department": dept, "field": "diagnosis", "text": r["anchor_text"], "entities": [r["anchor_entity"]], "topic": r.get("topic"), "anchors_axis": "5", "anchors_on": []} gold = GOLD_PLAIN.format(**ops) out.append({"atom": atom, "q": { "qid": f"{PATIENT}/axis5/S{len(out):03d}", "patient_id": PATIENT, "axis": "5_abstention", "question": f"What was the result of the patient's most recent {ops['absent_entity_bare']}?", "answer": gold, "answer_type": "abstain", "grounding": f"synthetic:{batch}", "evidence_visits": [host], "gt_visit": host, "gt_date": by_v[host]["date"], "question_appearance_window": {"from": by_v[host]["date"], "to": by_v[vs[-1]]["date"]}, "derivation": {"evidence": [{"visit": host, "date": by_v[host]["date"], "field": "diagnosis", "atom": r["anchor_text"]}], "steps": [f"Anchor documented v{host}: {r['anchor_text']!r}.", f"A chart documenting {r['anchor_entity']} would be expected to also " f"document {r['absent_entity']}.", f"/{r['absent_regex']}/ matches 0 atoms across the whole chain.", "Therefore the correct answer is that it was never documented."], "gold": gold, "rivals": [], "operands": ops, "derived": True}, "provenance": f"synthetic anchor ({batch}); absence verified, 0 hits in the record's event chain", "axis_specific": {"absent_entity": r["absent_entity"], "anchor_entity": r["anchor_entity"], "depends_on_atoms": [atom["atom_id"]]}, }}) return out if __name__ == "__main__": discover()