M27: Certification Exam Prep

mkdir cert-prep-lab && cd cert-prep-lab
python -m venv venv && source venv/bin/activate   # Windows: venv\Scripts\activate
pip install anthropic
export ANTHROPIC_API_KEY=your-key-here             # Windows: set ANTHROPIC_API_KEY=your-key-here

import anthropic
import json
from dataclasses import dataclass

@dataclass
class ValidationError:
    field: str
    expected: str
    actual: str

def validate_extraction(data: dict) -> list[ValidationError]:
    """Validate extracted data against business rules.
    Returns list of validation errors (empty = valid)."""
    errors = []

    # CPT codes must be exactly 5 digits
    cpt = data.get("cpt_code", "")
    if not (cpt.isdigit() and len(cpt) == 5):
        errors.append(ValidationError(
            field="cpt_code",
            expected="5-digit numeric code (e.g., '27447')",
            actual=repr(cpt),
        ))

    # ICD-10 codes: letter + 2+ digits + optional decimal
    import re
    icd = data.get("icd10_code", "")
    if not re.match(r'^[A-Z]\d{2}(\.\d{1,4})?$', icd):
        errors.append(ValidationError(
            field="icd10_code",
            expected="ICD-10 format: letter + 2 digits + optional decimal (e.g., 'M17.11')",
            actual=repr(icd),
        ))

    # Units must be positive integer
    units = data.get("units")
    if not isinstance(units, int) or units < 1:
        errors.append(ValidationError(
            field="units",
            expected="positive integer >= 1",
            actual=repr(units),
        ))

    return errors

def extract_with_retry(
    client: anthropic.Anthropic,
    document: str,
    max_retries: int = 3,
) -> dict:
    """Extract structured data with validation-retry loop.

    Key pattern: on validation failure, append SPECIFIC error
    details — not generic "try again". This gives Claude the
    information it needs to self-correct.
    """
    # Define the extraction tool — Claude must return this schema
    tools = [{
        "name": "extract_medical_data",
        "description": "Extract structured medical data from a document",
        "input_schema": {
            "type": "object",
            "properties": {
                "cpt_code": {
                    "type": "string",
                    "description": "5-digit CPT procedure code"
                },
                "icd10_code": {
                    "type": "string",
                    "description": "ICD-10 diagnosis code (e.g., M17.11)"
                },
                "units": {
                    "type": "integer",
                    "description": "Number of procedure units"
                },
            },
            "required": ["cpt_code", "icd10_code", "units"],
        },
    }]

    messages = [{
        "role": "user",
        "content": f"Extract medical data from this document:\n\n{document}",
    }]

    # Track per-field failure counts — if same field fails 3 times,
    # it's not a self-correction issue, it's a data issue.
    field_failures: dict[str, int] = {}

    for attempt in range(max_retries + 1):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            tools=tools,
            # Force the specific tool on first attempt
            tool_choice=(
                {"type": "tool", "name": "extract_medical_data"}
                if attempt == 0
                else {"type": "auto"}
            ),
            messages=messages,
        )

        # Find the tool_use block in the response
        tool_use = next(
            (b for b in response.content if b.type == "tool_use"),
            None,
        )
        if not tool_use:
            break  # Claude decided not to use the tool

        extracted = tool_use.input
        errors = validate_extraction(extracted)

        if not errors:
            return {"success": True, "data": extracted, "attempts": attempt + 1}

        # Check for stuck fields — same field failing repeatedly
        for err in errors:
            field_failures[err.field] = field_failures.get(err.field, 0) + 1
            if field_failures[err.field] >= 3:
                return {
                    "success": False,
                    "data": extracted,
                    "reason": f"Field '{err.field}' failed 3 times — flagging for human review",
                    "attempts": attempt + 1,
                }

        if attempt < max_retries:
            # THE KEY PATTERN: append SPECIFIC errors, not "try again"
            error_details = "\n".join(
                f"- Field '{e.field}': expected {e.expected}, got {e.actual}"
                for e in errors
            )
            # Add the tool result + error feedback to conversation
            messages.append({"role": "assistant", "content": response.content})
            messages.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": tool_use.id,
                    "content": f"Validation failed:\n{error_details}\n\nPlease correct these specific fields.",
                    "is_error": True,
                }],
            })

    return {"success": False, "data": extracted, "reason": "Max retries exceeded", "attempts": max_retries + 1}

# Add to bottom of validation_retry.py:
if __name__ == "__main__":
    client = anthropic.Anthropic()
    test_doc = """
    Patient: Jane Smith, DOB 1965-03-15
    Procedure: Total knee replacement (right knee)
    Diagnosis: Primary osteoarthritis, right knee
    Units: 1
    """
    result = extract_with_retry(client, test_doc)
    print(json.dumps(result, indent=2))

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class Confidence(Enum):
    HIGH = "high"       # Official source (state filing DB)
    MEDIUM = "medium"   # Third-party aggregator
    LOW = "low"         # User-submitted / unverified

class ClaimStatus(Enum):
    WELL_ESTABLISHED = "well_established"  # Multiple sources agree
    CONTESTED = "contested"                # Sources disagree
    SINGLE_SOURCE = "single_source"        # Only one source

@dataclass
class ProvenanceRecord:
    """Track the origin and reliability of each claim.

    The exam tests this in Scenario 3: when subagents return
    conflicting data, provenance lets the coordinator decide
    which claim to trust.
    """
    claim: str                    # The actual data/assertion
    source: str                   # Which agent/tool/API
    confidence: Confidence        # Source reliability
    timestamp: datetime           # When data was retrieved
    agent_id: str                 # Which subagent in pipeline
    source_type: str = ""         # "official_filing" vs "aggregator"

@dataclass
class ProvenanceTracker:
    """Aggregate claims from multiple subagents and resolve conflicts."""
    claims: dict[str, list[ProvenanceRecord]] = field(default_factory=dict)

    def add_claim(self, topic: str, record: ProvenanceRecord):
        """Add a claim from a subagent."""
        if topic not in self.claims:
            self.claims[topic] = []
        self.claims[topic].append(record)

    def resolve(self, topic: str) -> dict:
        """Resolve conflicting claims using provenance rules:
        1. Official sources trump aggregators
        2. Fresher data trumps stale data
        3. If still contested, flag for human review
        """
        records = self.claims.get(topic, [])
        if not records:
            return {"status": "no_data", "claim": None}
        if len(records) == 1:
            r = records[0]
            return {
                "status": ClaimStatus.SINGLE_SOURCE.value,
                "claim": r.claim,
                "source": r.source,
                "confidence": r.confidence.value,
                "warning": "Single source — verify independently",
            }

        # Check for agreement
        unique_claims = set(r.claim for r in records)
        if len(unique_claims) == 1:
            confidence_rank = {"low": 0, "medium": 1, "high": 2}
            best = max(records, key=lambda r: confidence_rank[r.confidence.value])
            return {
                "status": ClaimStatus.WELL_ESTABLISHED.value,
                "claim": best.claim,
                "sources": [r.source for r in records],
                "confidence": "high",
            }

        # Conflict — resolve by source reliability, then freshness
        # Sort descending: highest confidence first, then most recent timestamp
        ranked = sorted(
            records,
            key=lambda r: (
                ["low", "medium", "high"].index(r.confidence.value),
                r.timestamp,
            ),
            reverse=True,
        )
        best = ranked[0]
        return {
            "status": ClaimStatus.CONTESTED.value,
            "claim": best.claim,
            "source": best.source,
            "confidence": best.confidence.value,
            "alternatives": [
                {"claim": r.claim, "source": r.source}
                for r in ranked[1:]
            ],
            "warning": "Sources disagree — flagged for human review",
        }

# Add to bottom of provenance.py:
if __name__ == "__main__":
    tracker = ProvenanceTracker()

    # Subagent A: official state filing database
    tracker.add_claim("acme_lien", ProvenanceRecord(
        claim="Acme Corp has an active UCC lien — Filing #2024-NY-0047821",
        source="ny_state_filing_db",
        confidence=Confidence.HIGH,
        timestamp=datetime(2026, 4, 2, 14, 30),
        agent_id="subagent-filings-ny",
        source_type="official_filing",
    ))

    # Subagent B: third-party aggregator disagrees
    tracker.add_claim("acme_lien", ProvenanceRecord(
        claim="No active liens found for Acme Corp",
        source="third_party_aggregator",
        confidence=Confidence.MEDIUM,
        timestamp=datetime(2026, 4, 1, 9, 0),
        agent_id="subagent-aggregator",
        source_type="aggregator",
    ))

    import json
    result = tracker.resolve("acme_lien")
    print(json.dumps(result, indent=2, default=str))