| |
| """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": "<the entity exactly as listed above>",\n' |
| ' "anchor_status": "<the documented value or status>",\n' |
| ' "absent_entity": "<the missing thing, with a/an>",\n' |
| ' "absent_regex": "<regex, case-insensitive, matching any mention>",\n' |
| ' "key": "<slug>"\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": "<terse chart fact>",\n' |
| ' "anchor_entity": "<short>",\n' |
| ' "anchor_value": "<the value or status>",\n' |
| ' "absent_entity": "<the missing test, with a/an>",\n' |
| ' "absent_regex": "<regex matching any mention of it>",\n' |
| ' "topic": "<slug>"}') |
| 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() |
|
|
