"""Quick-read v3: two-pass approach + Anthropic extended thinking. Pass 1: forced per-lead ST/T enumeration with thinking enabled. The model builds an unbiased catalog of every visible finding before naming any diagnosis. Pass 2: synthesis. Sees the original image + pass-1 output, produces the formal reading. Thinking enabled. Total time ~60-150s. Designed to be invoked from a background worker, not synchronously from an HTTP-edge-timeout-bounded request. """ from __future__ import annotations import base64 import os import time from pathlib import Path import anthropic from fastapi import HTTPException from pydantic import BaseModel from .config import config PASS1_PROMPT = """You are an expert cardiac electrophysiologist examining a 12-lead ECG. DO NOT yet attempt to diagnose. Your only job in this pass is to systematically catalog every visible finding so a synthesis step later can interpret them. For EACH of the 13 leads visible (I, II, III, aVR, aVL, aVF, V1, V2, V3, V4, V5, V6, and the rhythm strip — typically lead II at the bottom), describe what you see in plain language: 1. **P wave** — present? visible morphology? upright/inverted? 2. **PR interval** — short/normal/long? 3. **QRS** — narrow/wide? notching? deep S? tall R? Q wave presence + width + depth (Q ≥40 ms wide or ≥25% R height = pathologic)? 4. **ST segment** — elevation (mm), depression (mm), morphology (concave/convex/flat/downsloping/upsloping), or isoelectric? 5. **T wave** — upright, flat, inverted, biphasic, hyperacute (tall + symmetric)? 6. **Rhythm strip features** — regularity, P-QRS relationship, any ectopy? Also note overall: - Heart rate (count R-R intervals in the rhythm strip) - QRS axis (estimate quadrant from leads I and aVF) - Image quality issues (rotation, low contrast, partial obstruction) Format as a structured per-lead table or bulleted list. Be quantitative whenever possible (mm for ST shifts, ms for intervals). Do NOT name a diagnosis. Do NOT use the words "STEMI", "NSTEMI", "MI", "ischemia" yet — those are interpretive labels for the synthesis pass. Output plain markdown, no JSON, no code fences. End with: "---END PASS 1---" on its own line.""" PASS2_PROMPT_TEMPLATE = """You are an expert cardiac electrophysiologist. You previously examined a 12-lead ECG and produced this objective per-lead catalog: ═══════════════════════════════════════════════════════════════════ PASS 1 OUTPUT (your prior cataloging step) ═══════════════════════════════════════════════════════════════════ {pass1_text} ═══════════════════════════════════════════════════════════════════ END PASS 1 OUTPUT ═══════════════════════════════════════════════════════════════════ Now look at the ECG image again WITH the pass-1 catalog in mind and produce the formal reading. Required structure (plain markdown, no JSON, no code fences): **Diagnosis** — ONE sentence. If STEMI/OMI criteria are met anywhere, lead with that explicitly and name the territory. Otherwise state the dominant clinical pattern. **Rate** — bpm. **Rhythm** — primary rhythm + regularity. **Axis** — quadrant + estimated degrees. **Intervals** — PR / QRS / QT(c) in ms. **ST/T summary table** — Group by territory (inferior, lateral, anteroseptal, anterior, lateral precordial, posterior). For each: what was elevated, depressed, or otherwise abnormal, with mm where measurable. **Reciprocal changes** — Explicitly check every ST elevation against its anatomical opposite. State yes/no per pairing. **Q waves** — Pathologic? Which leads? Significance? **Final synthesis** — 2-3 sentences integrating the findings into a clinical picture. **Why — three teaching pearls** — each pearl must reference specific leads or features from your pass-1 catalog and explain the reasoning chain. CRITICAL: - Use the pass-1 catalog as your evidence base. Do not invent findings not present in pass 1. - If ST elevation in any territory was noted in pass 1, you MUST address whether it meets STEMI/OMI criteria (and check for reciprocal changes). - If pass 1 noted "no ST elevation anywhere", confidently call this non-ischemic. - Do NOT default to "nonspecific ST-T changes" — name a pattern.""" class QuickReadRequest(BaseModel): image_url: str class QuickReadResponse(BaseModel): markdown: str elapsed_ms: int model: str def _read_image_bytes(image_url: str) -> tuple[bytes, str]: if image_url.startswith("file://"): path = Path(image_url[7:]) elif image_url.startswith("/"): path = Path(image_url) else: raise ValueError(f"unsupported image_url scheme: {image_url}") if not path.exists(): raise FileNotFoundError(str(path)) ext = path.suffix.lower() media_type = { ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp", }.get(ext, "image/png") return path.read_bytes(), media_type def _extract_text(response) -> str: """Concatenate text blocks from an Anthropic response, skipping thinking blocks.""" parts = [] for block in response.content: # Skip thinking blocks — they're internal reasoning, not output if getattr(block, "type", None) == "text": parts.append(block.text) return "".join(parts).strip() def register_quick_read(app): @app.post("/v1/quick-read", response_model=QuickReadResponse) def quick_read(req: QuickReadRequest): if not config.anthropic_api_key: raise HTTPException(status_code=503, detail="ANTHROPIC_API_KEY not configured") path_str = req.image_url if path_str.startswith("/uploads/"): path_str = os.path.join(config.upload_dir, path_str[len("/uploads/"):]) try: data, media_type = _read_image_bytes(path_str) except FileNotFoundError as e: raise HTTPException(status_code=404, detail=f"image not found: {e}") except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) img_b64 = base64.standard_b64encode(data).decode("ascii") image_content = { "type": "image", "source": {"type": "base64", "media_type": media_type, "data": img_b64}, } client = anthropic.Anthropic( api_key=config.anthropic_api_key, timeout=300, # generous per-call timeout ) model = os.environ.get("EKG_TUTOR_INTERPRET_MODEL", "claude-sonnet-4-6") # Extended thinking budget (tokens). 4k is reasonable for ECG; bumping # higher costs more without meaningfully better reads in early testing. thinking_budget = int(os.environ.get("EKG_TUTOR_INTERPRET_THINKING_BUDGET", "4000")) thinking_config = {"type": "enabled", "budget_tokens": thinking_budget} start = time.monotonic() # ─── PASS 1 — per-lead catalog ───────────────────────────────────── try: pass1 = client.messages.create( model=model, max_tokens=8192, temperature=1.0, # thinking requires temperature=1 thinking=thinking_config, messages=[ { "role": "user", "content": [image_content, {"type": "text", "text": PASS1_PROMPT}], } ], ) except anthropic.APIError as e: raise HTTPException(status_code=502, detail=f"anthropic pass-1 error: {e}") pass1_text = _extract_text(pass1) if not pass1_text: raise HTTPException(status_code=502, detail="empty pass-1 response") # ─── PASS 2 — synthesis ───────────────────────────────────────────── pass2_prompt = PASS2_PROMPT_TEMPLATE.format(pass1_text=pass1_text) try: pass2 = client.messages.create( model=model, max_tokens=4096, temperature=1.0, thinking=thinking_config, messages=[ { "role": "user", "content": [image_content, {"type": "text", "text": pass2_prompt}], } ], ) except anthropic.APIError as e: raise HTTPException(status_code=502, detail=f"anthropic pass-2 error: {e}") synthesis = _extract_text(pass2) if not synthesis: raise HTTPException(status_code=502, detail="empty pass-2 response") # Assemble the final markdown with both passes preserved for transparency markdown = ( synthesis + "\n\n---\n\n" + "
Show per-lead analysis (pass 1)\n\n" + pass1_text.replace("---END PASS 1---", "").strip() + "\n\n
" ) elapsed_ms = int((time.monotonic() - start) * 1000) return QuickReadResponse( markdown=markdown, elapsed_ms=elapsed_ms, model=pass2.model, )