examples/medical_diagnosis.py
| 1 | """7-step medical diagnosis reasoning trace with inclusion proof for the decision. |
| 2 | |
| 3 | Shows how selective disclosure works: we can prove that the model's DECISION |
| 4 | step was part of a signed trace without revealing the other 6 steps. |
| 5 | """ |
| 6 | |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | from quantumshield.identity.agent import AgentIdentity |
| 10 | |
| 11 | from pqc_reasoning_ledger import ( |
| 12 | ReasoningProver, |
| 13 | ReasoningRecorder, |
| 14 | TraceVerifier, |
| 15 | ) |
| 16 | |
| 17 | |
| 18 | def main() -> None: |
| 19 | print("=" * 72) |
| 20 | print("PQC Reasoning Ledger - Medical Diagnosis") |
| 21 | print("=" * 72) |
| 22 | |
| 23 | identity = AgentIdentity.create("clinical-decision-signer") |
| 24 | rec = ReasoningRecorder(identity) |
| 25 | rec.begin_trace( |
| 26 | model_did="did:pqaid:clinical-reasoner", |
| 27 | model_version="3.2.1", |
| 28 | task="differential-diagnosis", |
| 29 | domain="medical", |
| 30 | session_id="enc-2026-04-20-9127", |
| 31 | ) |
| 32 | |
| 33 | print("\n[1] Recording 7-step diagnostic reasoning...\n") |
| 34 | |
| 35 | rec.record_observation( |
| 36 | "Patient (M, 58) presents with substernal chest pain radiating to left " |
| 37 | "arm, onset 45 min ago, associated diaphoresis." |
| 38 | ) |
| 39 | rec.record_observation( |
| 40 | "Vitals: BP 148/92, HR 102, SpO2 96% RA, afebrile. ECG: 1 mm ST " |
| 41 | "depression in V4-V6, no Q waves." |
| 42 | ) |
| 43 | rec.record_retrieval( |
| 44 | "Retrieved: ACC/AHA 2021 Guideline for the Evaluation and Diagnosis of " |
| 45 | "Chest Pain; HEART score criteria." |
| 46 | ) |
| 47 | rec.record_hypothesis( |
| 48 | "Differential: NSTEMI > unstable angina > GERD/esophageal spasm > " |
| 49 | "aortic dissection (less likely, no asymmetric pulses).", |
| 50 | confidence=0.78, |
| 51 | ) |
| 52 | rec.record_deduction( |
| 53 | "HEART score: History 2 + ECG 1 + Age 1 + Risk 1 + Troponin pending = " |
| 54 | "at least 5 (moderate-high risk). Cannot rule out ACS.", |
| 55 | confidence=0.82, |
| 56 | ) |
| 57 | rec.record_self_critique( |
| 58 | "Troponin not yet resulted; decision cannot be deferred pending labs " |
| 59 | "because 45-minute window already elapsed. Proceed with ACS workup." |
| 60 | ) |
| 61 | decision = rec.record_decision( |
| 62 | "IMMEDIATE ACTION: obtain serial troponins (0, 3h), start aspirin 325 mg " |
| 63 | "chewed, admit to telemetry, cardiology consult. Re-evaluate for cath " |
| 64 | "lab activation if troponin elevated or ECG evolves.", |
| 65 | confidence=0.91, |
| 66 | ) |
| 67 | |
| 68 | print(f" recorded {len(rec.trace.steps)} steps") |
| 69 | print(f" decision step_id: {decision.step_id}") |
| 70 | |
| 71 | print("\n[2] Sealing trace...\n") |
| 72 | sealed = rec.seal() |
| 73 | print(f" trace_id: {sealed.metadata.trace_id}") |
| 74 | print(f" merkle_root: {sealed.merkle_root[:32]}...") |
| 75 | print(f" algorithm: {sealed.algorithm}") |
| 76 | |
| 77 | print("\n[3] Verifying sealed trace...\n") |
| 78 | result = TraceVerifier.verify(sealed) |
| 79 | print(f" fully_verified: {result.fully_verified}") |
| 80 | |
| 81 | print("\n[4] Producing inclusion proof for the DECISION step only...\n") |
| 82 | proof = ReasoningProver.prove_step(sealed, decision.step_id) |
| 83 | print(f" step proved: {proof.step.kind.value} at index {proof.proof.index}") |
| 84 | print(f" proof siblings: {len(proof.proof.siblings)} hashes") |
| 85 | print(f" tree_size: {proof.proof.tree_size}") |
| 86 | print(f" root agrees: {proof.proof.root == sealed.merkle_root}") |
| 87 | |
| 88 | ok = ReasoningProver.verify_proof(proof) |
| 89 | status = "[OK]" if ok else "[FAIL]" |
| 90 | print(f"\n {status} decision step proven to be member of signed trace") |
| 91 | print(" without revealing the other 6 steps.") |
| 92 | |
| 93 | |
| 94 | if __name__ == "__main__": |
| 95 | main() |
| 96 | |