Recipe 8.8: Python Implementation Example

pip install boto3

import json
import logging
import re
import datetime
from datetime import timezone
from decimal import Decimal

import boto3
from botocore.config import Config

# Structured logging. Never log PHI field values (entity text, patient IDs).
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Retry configuration for AWS API calls under load.
BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 3, "mode": "adaptive"})

# Module-level clients reused across Lambda invocations.
comprehend_medical_client = boto3.client("comprehendmedical", config=BOTO3_RETRY_CONFIG)
sagemaker_runtime_client = boto3.client("sagemaker-runtime", config=BOTO3_RETRY_CONFIG)
dynamodb = boto3.resource("dynamodb")

# ---------- Assertion Configuration ----------

# The seven assertion classes from the i2b2/VA 2010 taxonomy.
ASSERTION_CLASSES = [
    "present", "absent", "possible", "conditional",
    "historical", "family", "hypothetical"
]

# Confidence thresholds: two-tier system for routing entities.
#   - Below EXCLUDE_THRESHOLD: entity excluded from downstream until reviewed.
#   - Between EXCLUDE and INCLUDE: included with low_confidence flag, queued for review.
#   - Above INCLUDE_THRESHOLD: included without qualification.
EXCLUDE_THRESHOLD = 0.70
INCLUDE_THRESHOLD = 0.85

# Context window: characters before and after the entity to extract.
# ~2-3 sentences of surrounding text. Enough for negation cues without noise.
CONTEXT_WINDOW_CHARS = 300

# Entity extraction confidence: skip low-confidence entities from Comprehend Medical.
ENTITY_CONFIDENCE_THRESHOLD = 0.80

# SageMaker endpoint name for the custom assertion classifier.
# Replace with your deployed endpoint name.
SAGEMAKER_ENDPOINT_NAME = "clinical-assertion-classifier"

# DynamoDB table for annotated entities.
ANNOTATED_ENTITIES_TABLE = "assertion-annotated-entities"

# Separate restricted-access audit table for context snippets (PHI).
AUDIT_TABLE = "assertion-audit"

# Records retention: 7 years in seconds (default; adjust per institutional policy).
RETENTION_SECONDS = 7 * 365 * 24 * 60 * 60

# ---------- Rule-Based Assertion Patterns ----------

# Negation cues: words/phrases that signal the entity is absent.
# These are checked in the text BEFORE the entity (pre-negation).
NEGATION_CUES_PRE = [
    "no evidence of", "no signs of", "no symptoms of",
    "denies", "denied", "denying",
    "negative for", "without",
    "rules out", "ruled out", "rule out",
    "absence of", "free of", "unremarkable for",
    "no ", "not "
]

# Post-entity negation cues (less common but real).
NEGATION_CUES_POST = [
    "absent", "negative", "not found", "not seen",
    "not present", "not detected", "ruled out"
]

# Pseudo-negation: patterns that LOOK like negation but aren't.
# "not only" means the condition IS present (and more).
# "no longer" means it WAS present (historical, not absent).
PSEUDO_NEGATION = [
    "not only", "no longer", "no increase",
    "no change", "no improvement", "not ruled out"
]

# Historical cues: signals the condition existed in the past.
HISTORICAL_CUES = [
    "history of", "h/o", "s/p", "status post",
    "previously", "prior", "in the past",
    "resolved", "former", "remote history",
    "childhood history of", "past history of"
]

# Hedging cues: signals the clinician is uncertain (possible/probable).
HEDGING_CUES = [
    "possible", "probable", "likely", "suspected",
    "concerning for", "suspicious for", "consistent with",
    "cannot rule out", "cannot exclude", "differential includes",
    "questionable", "equivocal", "uncertain"
]

# Conditional cues: the entity depends on a future condition.
CONDITIONAL_CUES = [
    "if ", "in the event of", "should ",
    "in case of", "when ", "upon ",
    "contingent on", "provided that"
]

# Hypothetical cues: planning or speculative context.
HYPOTHETICAL_CUES = [
    "will consider", "may start", "would recommend",
    "plan to", "consider starting", "may develop",
    "could lead to", "potential for", "risk of developing"
]

# Family cues: entity pertains to a family member, not the patient.
FAMILY_CUES = [
    "mother", "father", "brother", "sister",
    "maternal", "paternal", "family history",
    "fhx", "family hx", "grandfather", "grandmother",
    "aunt", "uncle", "cousin", "sibling"
]

# Section headers that carry strong assertion signals.
# Mapping from normalized section name to the default assertion for entities
# found in that section. This is a strong prior, not absolute.
SECTION_ASSERTION_MAP = {
    "family history": "family",
    "fhx": "family",
    "family hx": "family",
    "past medical history": "historical",
    "pmh": "historical",
    "past surgical history": "historical",
    "psh": "historical",
    "surgical history": "historical",
}

# Maximum characters for entity text in the context window before we
# skip the rules and go straight to the model.
MAX_RULE_ENTITY_LENGTH = 100

def extract_entities(note_text: str) -> list[dict]:
    """
    Extract clinical entities from note text using Comprehend Medical.

    DetectEntitiesV2 identifies medical conditions, medications, tests,
    procedures, and anatomy mentions. Each entity comes with:
      - Character offsets (BeginOffset, EndOffset) in the original text
      - Category and Type labels
      - Confidence score
      - Traits (including basic negation detection)

    We filter to entities above ENTITY_CONFIDENCE_THRESHOLD to reduce noise.
    The built-in Traits include NEGATION and other signals, but they only
    cover present vs. absent. Our assertion classifier extends this to the
    full 7-class taxonomy.

    Args:
        note_text: raw clinical note text (up to 20,000 characters for DetectEntitiesV2)

    Returns:
        List of entity dicts with text, category, offsets, traits, and score.
    """
    # DetectEntitiesV2 accepts up to 20,000 UTF-8 characters per request.
    # For longer notes, you'd split at sentence boundaries and process in chunks.
    if len(note_text) > 20000:
        note_text = note_text[:20000]
        logger.warning("Note text truncated to 20,000 chars for DetectEntitiesV2")

    response = comprehend_medical_client.detect_entities_v2(Text=note_text)

    entities = []
    for entity in response.get("Entities", []):
        score = entity.get("Score", 0.0)
        if score < ENTITY_CONFIDENCE_THRESHOLD:
            continue

        # Extract trait names for quick lookup downstream.
        # Traits include NEGATION, SIGN, SYMPTOM, DIAGNOSIS, etc.
        trait_names = [
            t["Name"] for t in entity.get("Traits", [])
            if t.get("Score", 0.0) >= 0.75
        ]

        entities.append({
            "text": entity["Text"],
            "category": entity["Category"],
            "type": entity.get("Type", ""),
            "begin_offset": entity["BeginOffset"],
            "end_offset": entity["EndOffset"],
            "score": round(score, 3),
            "traits": trait_names,
        })

    logger.info("Extracted %d entities above confidence threshold", len(entities))
    return entities

def detect_section_header(note_text: str, entity_offset: int) -> str | None:
    """
    Attempt to detect the section header governing a given text position.

    Clinical notes are structured into sections (Assessment, HPI, PMH, etc.)
    but section headers are inconsistent across EHR templates and institutions.
    We look backwards from the entity position for common header patterns.

    This is deliberately simple. Production systems use ML-based section
    detection or rely on EHR-provided section metadata.
    """
    # Look backwards up to 500 chars for a line that looks like a section header.
    # Common patterns: "FAMILY HISTORY:", "Past Medical History", "Assessment and Plan"
    search_start = max(0, entity_offset - 500)
    preceding_text = note_text[search_start:entity_offset]

    # Split into lines and search backwards for a header-like line.
    lines = preceding_text.split("\n")
    for line in reversed(lines):
        stripped = line.strip()
        if not stripped:
            continue
        # Heuristic: section headers are short, often uppercase or end with ":"
        if len(stripped) < 50 and (stripped.endswith(":") or stripped.isupper()):
            return stripped.rstrip(":").strip().lower()

    return None

def extract_context_windows(note_text: str, entities: list[dict]) -> list[dict]:
    """
    Build context windows around each entity for assertion classification.

    Each context window includes:
      - The surrounding text (CONTEXT_WINDOW_CHARS before and after)
      - The entity's position within that window
      - The detected section header (if any)

    The classifier uses this window to find negation cues, hedging language,
    family history indicators, and other assertion-relevant signals.

    Args:
        note_text: the full clinical note text
        entities: list of entity dicts from extract_entities

    Returns:
        List of context dicts, one per entity, ready for classification.
    """
    entity_contexts = []

    for entity in entities:
        begin = entity["begin_offset"]
        end = entity["end_offset"]

        # Calculate window boundaries, clamping to document edges.
        window_start = max(0, begin - CONTEXT_WINDOW_CHARS)
        window_end = min(len(note_text), end + CONTEXT_WINDOW_CHARS)

        context_text = note_text[window_start:window_end]

        # Entity position within the extracted context window.
        entity_start_in_context = begin - window_start
        entity_end_in_context = end - window_start

        # Detect the section header for this region of the note.
        section_header = detect_section_header(note_text, begin)

        entity_contexts.append({
            "entity": entity,
            "context_text": context_text,
            "entity_start_in_context": entity_start_in_context,
            "entity_end_in_context": entity_end_in_context,
            "section_header": section_header,
        })

    return entity_contexts

def apply_assertion_rules(ctx: dict) -> dict | None:
    """
    Rule-based assertion detection for high-confidence cases.

    This handles about 60% of entities in typical clinical text:
      - Clear negations ("denies", "no evidence of", "negative for")
      - Section-based assertions (Family History section -> family)
      - Obvious historical markers ("history of", "s/p", "previously")
      - Clear hedging ("possible", "suspected", "cannot rule out")

    Returns a result dict if rules can confidently classify, or None if
    the entity needs ML model classification.

    Args:
        ctx: context dict from extract_context_windows

    Returns:
        Dict with assertion and confidence, or None if rules can't decide.
    """
    section = ctx["section_header"]
    entity_start = ctx["entity_start_in_context"]
    entity_end = ctx["entity_end_in_context"]
    context = ctx["context_text"]

    # Text before and after the entity in the context window.
    text_before = context[:entity_start].lower()
    text_after = context[entity_end:].lower()

    # --- Check 1: Section header (strongest signal) ---
    if section and section in SECTION_ASSERTION_MAP:
        return {
            "assertion": SECTION_ASSERTION_MAP[section],
            "confidence": 0.95 if "family" in section else 0.90,
            "method": "rule_based",
            "rule": f"section_header:{section}",
        }

    # --- Check 2: Family cues in surrounding text ---
    for cue in FAMILY_CUES:
        if cue in text_before[-80:]:
            return {
                "assertion": "family",
                "confidence": 0.91,
                "method": "rule_based",
                "rule": f"family_cue:{cue}",
            }

    # --- Check 3: Pre-entity negation ---
    # Only look within 60 chars before the entity (scope approximation).
    near_text_before = text_before[-60:]

    # First check for pseudo-negation (looks like negation but isn't).
    for pseudo in PSEUDO_NEGATION:
        if pseudo in near_text_before:
            # Pseudo-negation detected. Don't classify as absent.
            # Fall through to ML model for nuanced classification.
            return None

    for cue in NEGATION_CUES_PRE:
        if cue in near_text_before:
            return {
                "assertion": "absent",
                "confidence": 0.92,
                "method": "rule_based",
                "rule": f"negation_pre:{cue}",
            }

    # Post-entity negation (less common).
    near_text_after = text_after[:40]
    for cue in NEGATION_CUES_POST:
        if cue in near_text_after:
            return {
                "assertion": "absent",
                "confidence": 0.89,
                "method": "rule_based",
                "rule": f"negation_post:{cue}",
            }

    # --- Check 4: Historical cues ---
    for cue in HISTORICAL_CUES:
        if cue in near_text_before:
            return {
                "assertion": "historical",
                "confidence": 0.88,
                "method": "rule_based",
                "rule": f"historical:{cue}",
            }

    # --- Check 5: Conditional cues ---
    for cue in CONDITIONAL_CUES:
        if cue in near_text_before:
            return {
                "assertion": "conditional",
                "confidence": 0.86,
                "method": "rule_based",
                "rule": f"conditional:{cue}",
            }

    # --- Check 6: Hypothetical cues ---
    for cue in HYPOTHETICAL_CUES:
        if cue in text_before[-100:]:
            return {
                "assertion": "hypothetical",
                "confidence": 0.85,
                "method": "rule_based",
                "rule": f"hypothetical:{cue}",
            }

    # --- Check 7: Hedging cues ---
    for cue in HEDGING_CUES:
        if cue in near_text_before:
            return {
                "assertion": "possible",
                "confidence": 0.87,
                "method": "rule_based",
                "rule": f"hedging:{cue}",
            }

    # Rules couldn't confidently classify. Pass to ML model.
    return None

def classify_with_model(ctx: dict) -> dict:
    """
    Call the SageMaker-hosted assertion classifier for ambiguous cases.

    The model expects input formatted with [ENTITY] markers around the
    target span, plus the section header prepended as context. This tells
    the model which concept to classify vs. what's just surrounding text.

    Example input:
      "Section: Assessment\nPatient has [ENTITY]early-stage CKD[/ENTITY] based on labs"

    The model returns a JSON response with the predicted assertion class
    and confidence score.

    Args:
        ctx: context dict from extract_context_windows

    Returns:
        Dict with assertion, confidence, and method="ml_model".
    """
    context_text = ctx["context_text"]
    start = ctx["entity_start_in_context"]
    end = ctx["entity_end_in_context"]
    section = ctx["section_header"]

    # Format input: insert entity markers so the model knows the target span.
    marked_text = (
        context_text[:start]
        + "[ENTITY]"
        + context_text[start:end]
        + "[/ENTITY]"
        + context_text[end:]
    )

    # Prepend section header if available.
    if section:
        model_input = f"Section: {section}\n{marked_text}"
    else:
        model_input = marked_text

    # Call the SageMaker endpoint.
    payload = json.dumps({"text": model_input})

    response = sagemaker_runtime_client.invoke_endpoint(
        EndpointName=SAGEMAKER_ENDPOINT_NAME,
        ContentType="application/json",
        Body=payload,
    )

    result = json.loads(response["Body"].read().decode("utf-8"))

    return {
        "assertion": result["assertion_class"],
        "confidence": result["confidence"],
        "method": "ml_model",
    }

def classify_assertions(entity_contexts: list[dict]) -> list[dict]:
    """
    Two-pass assertion classification: rules first, ML model for the rest.

    Pass 1 (rules): handles obvious negations, section-based assertions,
    and clear temporal/hedging cues. Fast, deterministic, no network call.

    Pass 2 (ML model): invoked only for entities the rules couldn't
    confidently classify. Calls the SageMaker endpoint.

    Args:
        entity_contexts: list of context dicts from extract_context_windows

    Returns:
        List of classification result dicts with entity, assertion, confidence, and method.
    """
    results = []
    rule_count = 0
    model_count = 0

    for ctx in entity_contexts:
        # Pass 1: Try rules.
        rule_result = apply_assertion_rules(ctx)

        if rule_result and rule_result["confidence"] >= INCLUDE_THRESHOLD:
            # Propagate section_header into the entity dict so conflict
            # resolution (Step 4) can use it for priority scoring.
            entity_with_section = {**ctx["entity"], "section_header": ctx["section_header"]}
            results.append({
                "entity": entity_with_section,
                "assertion": rule_result["assertion"],
                "confidence": rule_result["confidence"],
                "method": rule_result["method"],
                "rule": rule_result.get("rule"),
            })
            rule_count += 1
            continue

        # Pass 2: ML model for ambiguous cases.
        try:
            model_result = classify_with_model(ctx)
            entity_with_section = {**ctx["entity"], "section_header": ctx["section_header"]}
            results.append({
                "entity": entity_with_section,
                "assertion": model_result["assertion"],
                "confidence": model_result["confidence"],
                "method": model_result["method"],
            })
            model_count += 1
        except Exception as e:
            # If the model endpoint is unavailable, fall back to "present"
            # with low confidence (will be flagged for review).
            logger.warning("Model invocation failed for entity '%s': %s",
                           ctx["entity"]["text"], str(e))
            entity_with_section = {**ctx["entity"], "section_header": ctx["section_header"]}
            results.append({
                "entity": entity_with_section,
                "assertion": "present",
                "confidence": 0.50,
                "method": "fallback",
            })

    logger.info("Classification complete: %d by rules, %d by model", rule_count, model_count)
    return results

# Section priority for conflict resolution.
# Higher number = more likely to reflect current clinical truth.
SECTION_PRIORITY = {
    "assessment": 5, "assessment and plan": 5, "a/p": 5, "plan": 5,
    "hpi": 4, "history of present illness": 4,
    "ros": 3, "review of systems": 3,
    "pmh": 2, "past medical history": 2,
    "fhx": 1, "family history": 1, "family hx": 1,
}

def resolve_assertion_conflicts(classified_entities: list[dict]) -> list[dict]:
    """
    Resolve conflicts when the same concept has multiple assertion mentions.

    Example: "diabetes" in PMH (historical) AND in Assessment (present).
    The Assessment mention wins because it reflects today's clinical state.

    We group by normalized entity text and pick the highest-priority mention.
    All mentions are preserved in the audit trail.

    Args:
        classified_entities: list from classify_assertions

    Returns:
        Deduplicated list with one assertion per unique concept.
        Conflict-resolved entries include all_mentions for audit.
    """
    # Group entities by normalized text (lowercase, stripped).
    # Production systems use concept normalization (CUI mapping via UMLS)
    # rather than raw text matching. Raw text conflates distinct entities
    # that happen to share surface forms.
    groups = {}
    for item in classified_entities:
        key = item["entity"]["text"].lower().strip()
        if key not in groups:
            groups[key] = []
        groups[key].append(item)

    resolved = []

    for concept, mentions in groups.items():
        if len(mentions) == 1:
            resolved.append(mentions[0])
            continue

        # Multiple mentions: pick highest priority.
        def priority_score(item):
            # Use section priority if available, otherwise default to 3.
            section = item["entity"].get("section_header") or ""
            section_score = SECTION_PRIORITY.get(section, 3)
            # Tie-break: later in the note = higher priority.
            position_score = item["entity"]["begin_offset"] / 10000.0
            return section_score + position_score

        best = max(mentions, key=priority_score)
        best["all_mentions"] = mentions
        best["conflict_resolved"] = True
        resolved.append(best)

    conflicts = sum(1 for r in resolved if r.get("conflict_resolved"))
    if conflicts:
        logger.info("Resolved %d assertion conflicts", conflicts)

    return resolved

def store_annotated_entities(
    patient_id: str,
    note_id: str,
    note_date: str,
    resolved_entities: list[dict],
) -> dict:
    """
    Write assertion-annotated entities to DynamoDB.

    Each entity record includes:
      - Patient/note linkage for provenance
      - The assertion classification and confidence
      - Two-tier routing flags:
        * excluded_pending_review: confidence < EXCLUDE_THRESHOLD (not sent downstream)
        * low_confidence: between thresholds (included with flag, queued for review)
      - A TTL epoch for automatic records retention expiration
      - Context snippet stored in a SEPARATE restricted-access audit table

    Also writes a note-level summary showing assertion distribution.

    Args:
        patient_id: patient identifier
        note_id: unique note identifier
        note_date: ISO date string for the clinical note
        resolved_entities: deduplicated list from resolve_assertion_conflicts

    Returns:
        Summary dict with counts by assertion class.
    """
    table = dynamodb.Table(ANNOTATED_ENTITIES_TABLE)
    audit_table = dynamodb.Table(AUDIT_TABLE)

    # Count assertions by class for the summary.
    assertion_counts = {cls: 0 for cls in ASSERTION_CLASSES}
    review_count = 0
    excluded_count = 0

    # TTL: current time + institutional retention period.
    ttl_epoch = int(datetime.datetime.now(timezone.utc).timestamp()) + RETENTION_SECONDS

    for idx, result in enumerate(resolved_entities):
        needs_review = result["confidence"] < INCLUDE_THRESHOLD
        excluded = result["confidence"] < EXCLUDE_THRESHOLD
        if needs_review:
            review_count += 1
        if excluded:
            excluded_count += 1

        assertion = result["assertion"]
        if assertion in assertion_counts:
            assertion_counts[assertion] += 1

        entity_key = f"{note_id}#{idx:04d}"

        record = {
            "patient_id": patient_id,
            "note_id_entity_idx": entity_key,
            "note_id": note_id,
            "note_date": note_date,
            "entity_text": result["entity"]["text"],
            "entity_category": result["entity"]["category"],
            "assertion_status": assertion,
            "confidence": Decimal(str(round(result["confidence"], 3))),
            "method": result["method"],
            "needs_review": needs_review,
            "low_confidence": needs_review and not excluded,
            "excluded_pending_review": excluded,
            "processed_at": datetime.datetime.now(timezone.utc).isoformat(),
            "ttl_epoch": ttl_epoch,
        }

        table.put_item(Item=record)

        # Context snippet in separate restricted-access audit table.
        # Only the review workflow and audit processes have access to this table.
        audit_table.put_item(Item={
            "note_id_entity_idx": entity_key,
            "context_snippet": result.get("entity", {}).get("context_text", ""),
            "ttl_epoch": ttl_epoch,
        })

    summary = {
        "note_id": note_id,
        "patient_id": patient_id,
        "note_date": note_date,
        "total_entities": len(resolved_entities),
        "assertion_counts": assertion_counts,
        "needs_review_count": review_count,
        "excluded_count": excluded_count,
    }

    logger.info(
        "Stored %d annotated entities (%d for review, %d excluded)",
        len(resolved_entities), review_count, excluded_count
    )
    return summary

def classify_note_assertions(
    note_text: str,
    patient_id: str,
    note_id: str,
    note_date: str,
) -> dict:
    """
    Run the full assertion classification pipeline on one clinical note.

    Steps:
      1. Extract entities with Comprehend Medical
      2. Build context windows around each entity
      3. Classify assertions (rules + ML model)
      4. Resolve conflicts for repeated concepts
      5. Store annotated entities in DynamoDB

    Args:
        note_text: raw clinical note text
        patient_id: patient identifier
        note_id: unique note identifier
        note_date: ISO date string

    Returns:
        Summary dict with assertion counts and review statistics.
    """
    # Step 1: Entity extraction.
    logger.info("Step 1: Extracting entities from note %s", note_id)
    entities = extract_entities(note_text)

    if not entities:
        logger.info("No entities found in note %s", note_id)
        return {"note_id": note_id, "total_entities": 0}

    # Step 2: Context window extraction.
    logger.info("Step 2: Extracting context windows for %d entities", len(entities))
    entity_contexts = extract_context_windows(note_text, entities)

    # Step 3: Assertion classification (hybrid rules + ML).
    logger.info("Step 3: Classifying assertion status")
    classified = classify_assertions(entity_contexts)

    # Step 4: Conflict resolution.
    logger.info("Step 4: Resolving assertion conflicts")
    resolved = resolve_assertion_conflicts(classified)

    # Step 5: Store results.
    logger.info("Step 5: Storing annotated entities")
    summary = store_annotated_entities(patient_id, note_id, note_date, resolved)

    logger.info("Done. %d entities classified for note %s", summary["total_entities"], note_id)
    return summary

# ---------- Demo with synthetic clinical data ----------

if __name__ == "__main__":

    # Synthetic clinical note exercising all seven assertion classes.
    # This is NOT real patient data. It's designed to test the assertion logic.
    SAMPLE_NOTE = """
HISTORY OF PRESENT ILLNESS:
62-year-old male presenting with fatigue and polyuria. Patient has type 2 diabetes
managed with metformin. Denies chest pain or shortness of breath. Reports occasional
dizziness when standing quickly.

PAST MEDICAL HISTORY:
History of MI in 2019, treated with PCI. Status post appendectomy (childhood).
Previously treated for H. pylori (resolved 2021).

FAMILY HISTORY:
Mother had breast cancer diagnosed at age 58. Father died of colon cancer at 72.
Maternal grandmother had rheumatoid arthritis.

REVIEW OF SYSTEMS:
No fever, no chills, no weight loss. Denies hematuria. Reports mild nocturia x3.

ASSESSMENT AND PLAN:
1. Type 2 diabetes, poorly controlled. HbA1c 8.9%. Will increase metformin to 1000mg BID.
2. Possible early CKD based on elevated creatinine. Repeat labs in 3 months.
3. If GFR drops below 30, will initiate nephrology referral.
4. Hypertension, stable on current regimen.
5. Cannot rule out sleep apnea given fatigue and BMI. Consider sleep study.
"""

    result = classify_note_assertions(
        note_text=SAMPLE_NOTE,
        patient_id="pat-00291",
        note_id="note-2026-03-15-00847",
        note_date="2026-03-15",
    )

    print(json.dumps(result, indent=2, default=str))

{
  "note_id": "note-2026-03-15-00847",
  "patient_id": "pat-00291",
  "note_date": "2026-03-15",
  "total_entities": 14,
  "assertion_counts": {
    "present": 4,
    "absent": 3,
    "possible": 2,
    "conditional": 1,
    "historical": 3,
    "family": 3,
    "hypothetical": 1
  },
  "needs_review_count": 2
}

import os

def lambda_handler(event: dict, context) -> dict:
    """
    Lambda handler for real-time assertion classification.

    Triggered by SQS messages containing note finalization events.
    The SQS queue provides automatic retries (maxReceiveCount=3) and
    routes persistently failing notes to a dead-letter queue (DLQ).
    A CloudWatch alarm on DLQ depth > 0 alerts when notes are stuck.

    Each SQS record contains:
      - note_text or an S3 reference to the note
      - patient_id
      - note_id
      - note_date

    Returns the assertion summary for monitoring and downstream routing.
    """
    # SQS event contains one or more records (batch size configured on the trigger).
    results = []

    for sqs_record in event.get("Records", [event]):
        # Parse the message body (JSON-encoded note event).
        if "body" in sqs_record:
            record = json.loads(sqs_record["body"])
        else:
            record = sqs_record.get("detail", sqs_record)

        note_text = record.get("note_text")
        patient_id = record["patient_id"]
        note_id = record["note_id"]
        note_date = record["note_date"]

        # If note_text not inline, fetch from S3.
        if not note_text and "s3_bucket" in record:
            s3 = boto3.client("s3", config=BOTO3_RETRY_CONFIG)
            obj = s3.get_object(
                Bucket=record["s3_bucket"],
                Key=record["s3_key"],
            )
            note_text = obj["Body"].read().decode("utf-8")

        if not note_text:
            logger.error("No note text available for note %s", note_id)
            results.append({"error": "no_note_text", "note_id": note_id})
            continue

        summary = classify_note_assertions(note_text, patient_id, note_id, note_date)

        # Route notes with entities needing review to the review SQS queue.
        if summary.get("needs_review_count", 0) > 0:
            logger.info("Note %s has %d entities for review", note_id, summary["needs_review_count"])
            # In production: send to SQS review queue (RBAC-protected).
            # sqs_client.send_message(
            #     QueueUrl=os.environ["REVIEW_QUEUE_URL"],
            #     MessageBody=json.dumps({"note_id": note_id, "patient_id": patient_id}),
            # )

        results.append(summary)

    return results[0] if len(results) == 1 else results