Recipe 8.5: Python Implementation Example

pip install boto3

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

import boto3
from botocore.config import Config
from boto3.dynamodb.conditions import Key, Attr

# Structured logging. In production, use JSON-formatted output for
# CloudWatch Logs Insights queries. Never log PHI field values.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles burst throttling from Comprehend Medical.
BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 5, "mode": "adaptive"})

# AWS clients
comprehend_medical = boto3.client("comprehendmedical", config=BOTO3_RETRY_CONFIG)
s3_client = boto3.client("s3", config=BOTO3_RETRY_CONFIG)
dynamodb = boto3.resource("dynamodb", config=BOTO3_RETRY_CONFIG)

# Table and bucket names. Replace with your actual resource names.
RESULTS_BUCKET = "clinical-nlp-results"
RECOMMENDATIONS_TABLE = "problem-list-recommendations"
PROBLEMS_TABLE = "patient-problems"

# Confidence threshold for accepting a problem extraction.
# Below this, we still extract but flag it for human review.
EXTRACTION_CONFIDENCE_THRESHOLD = 0.75

# Section header classifications.
# These determine the default assertion for problems found in each section.
# A problem in "ASSESSMENT" defaults to PRESENT.
# A problem in "FAMILY HISTORY" defaults to FAMILY_HISTORY.
ACTIVE_SECTION_HEADERS = [
    "assessment", "assessment and plan", "a/p", "active problems",
    "problem list", "diagnoses", "impression", "current problems",
    "hospital course", "hpi", "history of present illness"
]

PMH_SECTION_HEADERS = [
    "past medical history", "pmh", "medical history",
    "past history", "prior diagnoses", "significant history"
]

FAMILY_SECTION_HEADERS = [
    "family history", "fh", "family hx", "familial history"
]

RESOLVED_SECTION_HEADERS = [
    "resolved problems", "inactive problems", "past problems",
    "resolved", "inactive"
]

# Resolution markers: words/phrases that suggest a condition is no longer active.
# Used as a heuristic when section context alone is ambiguous.
RESOLUTION_MARKERS = [
    "resolved", "s/p", "status post", "history of", "h/o",
    "previous", "former", "prior", "no longer", "discontinued",
    "in remission", "cured"
]

import re

def detect_sections(note_text: str) -> list[dict]:
    """
    Split a clinical note into sections based on header patterns.

    Clinical notes use various header conventions:
    - ALL CAPS followed by colon: "ASSESSMENT:"
    - Title case followed by colon: "Assessment and Plan:"
    - Wrapped in asterisks: "** Family History **"

    Each returned section includes its category (ACTIVE, PMH, FAMILY, etc.)
    and the text content under that header.
    """
    # Pattern matches common clinical note header formats.
    # Looks for lines that are mostly a label followed by a colon,
    # or all-caps lines that serve as section dividers.
    header_pattern = re.compile(
        r"^(?:"
        r"([A-Z][A-Za-z /&]+):[ ]*$"     # "Assessment and Plan:" on its own line
        r"|([A-Z][A-Z /&]{2,}):?"         # "ASSESSMENT:" or "ASSESSMENT" (all caps)
        r"|\*\*\s*(.+?)\s*\*\*"           # "** Family History **"
        r")",
        re.MULTILINE
    )

    sections = []
    last_end = 0
    last_category = "UNKNOWN"

    for match in header_pattern.finditer(note_text):
        # Save the text between the previous header and this one.
        if last_end > 0 or match.start() > 0:
            section_text = note_text[last_end:match.start()].strip()
            if section_text:
                sections.append({
                    "category": last_category,
                    "text": section_text,
                    "start_offset": last_end
                })

        # Determine the header text from whichever capture group matched.
        header_text = match.group(1) or match.group(2) or match.group(3)
        header_text = header_text.strip().lower()

        # Classify this header.
        last_category = classify_section_header(header_text)
        last_end = match.end()

    # Don't forget the final section after the last header.
    remaining = note_text[last_end:].strip()
    if remaining:
        sections.append({
            "category": last_category,
            "text": remaining,
            "start_offset": last_end
        })

    # If no headers were found at all, treat the entire note as UNKNOWN.
    if not sections:
        sections.append({
            "category": "UNKNOWN",
            "text": note_text.strip(),
            "start_offset": 0
        })

    return sections

def classify_section_header(header_lower: str) -> str:
    """
    Map a lowercase header string to a section category.
    The category determines default assertion status for problems found there.
    """
    if header_lower in ACTIVE_SECTION_HEADERS:
        return "ACTIVE"
    elif header_lower in PMH_SECTION_HEADERS:
        return "PMH"
    elif header_lower in FAMILY_SECTION_HEADERS:
        return "FAMILY"
    elif header_lower in RESOLVED_SECTION_HEADERS:
        return "RESOLVED"
    else:
        return "UNKNOWN"

def extract_problems(sections: list[dict]) -> list[dict]:
    """
    Run NER on each relevant section to extract MEDICAL_CONDITION entities.

    We skip sections like allergies or medications that won't contain
    problem list entries. For each extracted condition, we carry forward
    the section category so assertion classification knows where it came from.
    """
    # We only process sections that might contain problem mentions.
    relevant_categories = {"ACTIVE", "PMH", "FAMILY", "RESOLVED", "UNKNOWN"}
    all_problems = []

    for section in sections:
        if section["category"] not in relevant_categories:
            continue

        # Comprehend Medical has a 20,000 character limit per request.
        # Discharge summaries and operative notes can exceed this.
        # Chunk at sentence boundaries to preserve negation scope.
        text = section["text"]
        chunks = _chunk_text(text, max_chars=20000)

        for chunk_offset, chunk_text in chunks:
            if not chunk_text.strip():
                continue

            # Call DetectEntitiesV2. This returns all medical entity types:
            # MEDICAL_CONDITION, MEDICATION, ANATOMY, TEST_TREATMENT_PROCEDURE, etc.
            # We only care about MEDICAL_CONDITION for problem list extraction.
            response = comprehend_medical.detect_entities_v2(Text=chunk_text)

            for entity in response.get("Entities", []):
                if entity["Category"] != "MEDICAL_CONDITION":
                    continue

                # Build our problem record with everything we need downstream.
                # Offset correction: add both section start and chunk offset.
                problem = {
                    "text": entity["Text"],
                    "confidence": entity["Score"],
                    "begin_offset": entity["BeginOffset"] + section["start_offset"] + chunk_offset,
                    "end_offset": entity["EndOffset"] + section["start_offset"] + chunk_offset,
                    "section": section["category"],
                    "traits": entity.get("Traits", []),
                    "attributes": entity.get("Attributes", []),
                    # entity_type helps distinguish DIAGNOSIS vs SIGN vs SYMPTOM
                    "entity_type": next(
                        (t["Name"] for t in entity.get("Traits", [])
                         if t["Name"] in ("DIAGNOSIS", "SIGN", "SYMPTOM")),
                        "DIAGNOSIS"  # default if no type trait present
                    )
                }
                all_problems.append(problem)

    logger.info("Extracted %d MEDICAL_CONDITION entities", len(all_problems))
    return all_problems


def _chunk_text(text: str, max_chars: int = 20000) -> list[tuple[int, str]]:
    """
    Split text into chunks that respect the Comprehend Medical character limit.

    Chunks at sentence boundaries (period followed by space or newline) to avoid
    splitting mid-sentence, which would break negation scope. Returns a list of
    (offset, chunk_text) tuples so callers can correct entity offsets.
    """
    if len(text) <= max_chars:
        return [(0, text)]

    chunks = []
    start = 0
    while start < len(text):
        end = start + max_chars
        if end >= len(text):
            chunks.append((start, text[start:]))
            break

        # Find the last sentence boundary before the limit.
        # Look for ". " or ".\n" working backwards from the cut point.
        boundary = text.rfind(". ", start, end)
        if boundary == -1:
            boundary = text.rfind(".\n", start, end)
        if boundary == -1 or boundary <= start:
            # No sentence boundary found; fall back to cutting at max_chars.
            boundary = end
        else:
            boundary += 2  # include the period and space

        chunks.append((start, text[start:boundary]))
        start = boundary

    return chunks

def classify_assertions(problems: list[dict]) -> list[dict]:
    """
    Determine assertion status for each extracted problem.

    Assertion status = is this problem actually active for this patient?
    Possible values: PRESENT, NEGATED, HISTORICAL, FAMILY_HISTORY,
    HYPOTHETICAL, RESOLVED.

    We use a two-layer approach:
    1. Section context sets the baseline (a problem in "Family History" is FAMILY_HISTORY)
    2. Entity-level traits can override (NEGATION trait overrides section context)
    """
    classified = []

    for problem in problems:
        # Layer 1: Section context provides the default assertion.
        assertion = _section_to_assertion(problem["section"])

        # Layer 2: Entity-level traits can override.
        for trait in problem["traits"]:
            trait_name = trait["Name"]
            trait_score = trait["Score"]

            # Only trust high-confidence traits for overrides.
            if trait_score < 0.80:
                continue

            if trait_name == "NEGATION":
                # "No diabetes" in the Assessment section means NOT active.
                # Negation overrides section context.
                assertion = "NEGATED"
                break  # negation is definitive

            if trait_name == "HYPOTHETICAL":
                # "If she develops CKD" is not an active problem.
                assertion = "HYPOTHETICAL"

        # Layer 3: Resolution marker heuristic for ambiguous cases.
        # If the problem text itself contains resolution language and we
        # haven't already classified it as negated/hypothetical, downgrade.
        if assertion == "PRESENT":
            if _has_resolution_markers(problem["text"]):
                assertion = "HISTORICAL"

        problem_with_assertion = {**problem, "assertion": assertion}
        classified.append(problem_with_assertion)

    # Log assertion distribution for monitoring.
    assertion_counts = {}
    for p in classified:
        assertion_counts[p["assertion"]] = assertion_counts.get(p["assertion"], 0) + 1
    logger.info("Assertion distribution: %s", assertion_counts)

    return classified

def _section_to_assertion(section_category: str) -> str:
    """Map section category to default assertion status."""
    mapping = {
        "ACTIVE": "PRESENT",
        "PMH": "HISTORICAL",
        "FAMILY": "FAMILY_HISTORY",
        "RESOLVED": "RESOLVED",
        "UNKNOWN": "PRESENT"  # conservative default: assume active
    }
    return mapping.get(section_category, "PRESENT")

def _has_resolution_markers(text: str) -> bool:
    """Check if the problem text contains language suggesting resolution."""
    # Simple substring matching. Production systems use more sophisticated
    # context-aware resolution detection (e.g., checking whether the marker
    # modifies the condition or describes its current status).
    text_lower = text.lower()
    return any(marker in text_lower for marker in RESOLUTION_MARKERS)

def normalize_problems(classified_problems: list[dict]) -> list[dict]:
    """
    Map extracted problem text to SNOMED CT and ICD-10-CM codes.

    We only normalize problems that might go on the problem list.
    Negated and hypothetical problems don't need codes (they won't be added
    to any structured list), but we still include them in the output for
    completeness and auditability.
    """
    # Only these assertion types get normalized.
    normalize_assertions = {"PRESENT", "HISTORICAL", "FAMILY_HISTORY"}
    normalized = []

    for problem in classified_problems:
        if problem["assertion"] not in normalize_assertions:
            # Still include in output, just without codes.
            normalized.append({**problem, "icd10": None, "snomed": None})
            continue

        # The text we send for normalization. Comprehend Medical's inference
        # APIs work best with the exact extracted span plus a bit of context,
        # but even just the entity text produces reasonable results.
        text_for_inference = problem["text"]

        # Call ICD-10-CM inference.
        icd10_codes = _infer_icd10(text_for_inference)

        # Call SNOMED CT inference.
        snomed_codes = _infer_snomed(text_for_inference)

        normalized.append({
            **problem,
            "icd10": icd10_codes,
            "snomed": snomed_codes
        })

    return normalized

def _infer_icd10(text: str) -> list[dict]:
    """
    Call InferICD10CM and return top 3 candidates.

    The API returns entities with ICD10CMConcepts ranked by confidence.
    We take the top 3 because the top-1 isn't always correct, especially
    for abbreviations or partial mentions.
    """
    response = comprehend_medical.infer_icd10_cm(Text=text)

    # InferICD10CM returns entities, each with a list of ICD10CMConcepts.
    # We want the concepts from the first (most relevant) entity.
    entities = response.get("Entities", [])
    if not entities:
        return []

    concepts = entities[0].get("ICD10CMConcepts", [])
    # Return top 3, with the fields downstream needs.
    return [
        {
            "Code": c["Code"],
            "Description": c["Description"],
            "Score": round(c["Score"], 3)
        }
        for c in concepts[:3]
    ]

def _infer_snomed(text: str) -> list[dict]:
    """
    Call InferSNOMEDCT and return top 3 candidates.

    Same pattern as ICD-10 inference. SNOMED codes are preferred for
    clinical problem lists because SNOMED has richer hierarchy and
    more clinical granularity than ICD-10.
    """
    response = comprehend_medical.infer_snomedct(Text=text)

    entities = response.get("Entities", [])
    if not entities:
        return []

    concepts = entities[0].get("SNOMEDCTConcepts", [])
    return [
        {
            "Code": c["Code"],
            "Description": c["Description"],
            "Score": round(c["Score"], 3)
        }
        for c in concepts[:3]
    ]

def reconcile_problems(
    patient_id: str,
    extracted_problems: list[dict],
    note_id: str
) -> list[dict]:
    """
    Compare extracted active problems against the existing problem list.

    This produces RECOMMENDATIONS, not changes. Problem list maintenance
    is a clinical act that requires physician review. We're surfacing
    candidates, not making decisions.
    """
    # Fetch the patient's current active problem list from DynamoDB.
    current_list = _get_current_problem_list(patient_id)
    current_codes = {p["snomed_code"] for p in current_list}

    # Separate extracted problems by assertion type.
    active_extracted = [
        p for p in extracted_problems
        if p["assertion"] == "PRESENT" and p["snomed"]
    ]
    resolved_signals = [
        p for p in extracted_problems
        if p["assertion"] in ("HISTORICAL", "RESOLVED") and p["snomed"]
    ]

    recommendations = []

    # 1. Find problems active in notes but missing from the problem list.
    for problem in active_extracted:
        # Check all top-3 SNOMED candidates, not just top-1.
        # The same condition can normalize to different codes depending on context.
        # Checking only the top candidate produces false ADD_CANDIDATE recommendations
        # for problems already on the list under a different synonym code.
        extracted_codes = {c["Code"] for c in problem["snomed"]}
        if extracted_codes & current_codes:
            # At least one candidate matches an existing problem. Skip.
            continue

        recommendations.append({
            "type": "ADD_CANDIDATE",
            "problem_text": problem["text"],
            "snomed": problem["snomed"][0],
            "icd10": problem["icd10"][0] if problem["icd10"] else None,
            "confidence": problem["confidence"],
            "source_note": note_id,
            "rationale": (
                f"Mentioned as active in {problem['section']} section "
                f"but not on current problem list"
            )
        })

    # 2. Find problems on the list that notes suggest are resolved.
    resolved_codes = {
        p["snomed"][0]["Code"] for p in resolved_signals
    }
    for existing in current_list:
        if existing["snomed_code"] in resolved_codes:
            recommendations.append({
                "type": "RESOLVE_CANDIDATE",
                "problem_text": existing["problem_text"],
                "snomed": {"Code": existing["snomed_code"]},
                "source_note": note_id,
                "rationale": "Mentioned with resolution markers in recent note"
            })

    # Note: The main recipe's pseudocode also includes specificity upgrade detection
    # (checking SNOMED hierarchy relationships with is_child_of). That requires a
    # SNOMED ontology service or lookup table, which is beyond the scope of this example.

    logger.info(
        "Reconciliation: %d recommendations for patient %s",
        len(recommendations), patient_id
    )
    return recommendations

def _get_current_problem_list(patient_id: str) -> list[dict]:
    """
    Query DynamoDB for the patient's active problems.

    In production, you'd have a properly designed table with patient_id
    as partition key and problem entries as items. This is a simplified
    query pattern.
    """
    table = dynamodb.Table(PROBLEMS_TABLE)

    response = table.query(
        KeyConditionExpression=Key("patient_id").eq(patient_id),
        FilterExpression=Attr("status").eq("ACTIVE")
    )

    return response.get("Items", [])

import hashlib


def store_results(
    patient_id: str,
    extracted_problems: list[dict],
    recommendations: list[dict],
    note_id: str
) -> dict:
    """
    Persist extraction results and recommendations.

    Two destinations:
    - S3: Full extraction record for audit trail and reprocessing
    - DynamoDB: Actionable recommendations for clinician review queue

    Recommendation IDs are deterministic (derived from note_id + snomed_code + type)
    so reprocessing the same note produces the same IDs. Conditional writes prevent
    duplicates, making the pipeline idempotent.
    """
    timestamp = datetime.datetime.now(timezone.utc).isoformat()

    # Build the full results record.
    results_record = {
        "patient_id": patient_id,
        "note_id": note_id,
        "extraction_timestamp": timestamp,
        "problems_extracted": _serialize_problems(extracted_problems),
        "recommendations": recommendations
    }

    # Write to S3 for audit trail.
    s3_key = f"results/{patient_id}/{note_id}.json"
    s3_client.put_object(
        Bucket=RESULTS_BUCKET,
        Key=s3_key,
        Body=json.dumps(results_record, default=str),
        ContentType="application/json",
        ServerSideEncryption="aws:kms"
        # In production, also specify SSEKMSKeyId with your CMK ARN.
    )
    logger.info("Stored extraction results to s3://%s/%s", RESULTS_BUCKET, s3_key)

    # Write recommendations to DynamoDB for clinician review.
    # Deterministic IDs + conditional writes = idempotent reprocessing.
    if recommendations:
        rec_table = dynamodb.Table(RECOMMENDATIONS_TABLE)
        for rec in recommendations:
            # Deterministic ID: same note + same code + same type always
            # produces the same recommendation_id. This prevents duplicates
            # when a note is reprocessed (e.g., after a pipeline bug fix).
            snomed_code = rec["snomed"]["Code"] if rec.get("snomed") else "NONE"
            rec_id = hashlib.sha256(
                f"{note_id}:{snomed_code}:{rec['type']}".encode()
            ).hexdigest()[:32]

            try:
                rec_table.put_item(
                    Item={
                        "patient_id": patient_id,
                        "recommendation_id": rec_id,
                        "type": rec["type"],
                        "problem_text": rec["problem_text"],
                        "snomed_code": snomed_code,
                        "icd10_code": rec.get("icd10", {}).get("Code", "N/A") if rec.get("icd10") else "N/A",
                        "confidence": Decimal(str(round(rec.get("confidence", 0.0), 3))),
                        "source_note": note_id,
                        "rationale": rec["rationale"],
                        "status": "PENDING_REVIEW",
                        "created_at": timestamp
                    },
                    # Conditional write: only insert if this recommendation doesn't
                    # already exist. Prevents duplicates on reprocessing.
                    ConditionExpression="attribute_not_exists(recommendation_id)"
                )
            except rec_table.meta.client.exceptions.ConditionalCheckFailedException:
                # Already exists from a previous run. That's fine, skip it.
                logger.debug("Recommendation %s already exists, skipping", rec_id)

        logger.info("Stored %d recommendations to DynamoDB", len(recommendations))

    return results_record

def _serialize_problems(problems: list[dict]) -> list[dict]:
    """
    Prepare problem records for JSON serialization.

    Strips internal fields (traits, attributes) that are useful during
    processing but noisy in the final output. Keeps what a reviewer needs.
    """
    serialized = []
    for p in problems:
        serialized.append({
            "text": p["text"],
            "assertion": p["assertion"],
            "section": p["section"],
            "confidence": round(p["confidence"], 3),
            "snomed": p.get("snomed"),
            "icd10": p.get("icd10")
        })
    return serialized

def process_note_for_problems(
    patient_id: str,
    note_text: str,
    note_id: str
) -> dict:
    """
    Run the full problem list extraction pipeline on a single clinical note.

    This is the main entry point. The pipeline:
    1. Detects sections in the note
    2. Extracts MEDICAL_CONDITION entities from relevant sections
    3. Classifies assertion status (active, negated, historical, etc.)
    4. Normalizes active problems to SNOMED CT and ICD-10 codes
    5. Reconciles against the existing problem list
    6. Stores results and generates recommendations

    Returns the full extraction record with recommendations.
    """
    logger.info("Processing note %s for patient %s", note_id, patient_id)

    # Step 1: Split note into sections.
    logger.info("Step 1: Detecting sections")
    sections = detect_sections(note_text)
    logger.info("  Found %d sections", len(sections))

    # Step 2: Extract clinical problems from each relevant section.
    logger.info("Step 2: Extracting clinical problems")
    raw_problems = extract_problems(sections)
    logger.info("  Extracted %d problem mentions", len(raw_problems))

    # Step 3: Classify assertion status for each problem.
    logger.info("Step 3: Classifying assertions")
    classified = classify_assertions(raw_problems)

    # Step 4: Normalize active/historical problems to standard codes.
    logger.info("Step 4: Normalizing to SNOMED CT and ICD-10")
    normalized = normalize_problems(classified)

    # Step 5: Reconcile against existing problem list.
    logger.info("Step 5: Reconciling against current problem list")
    recommendations = reconcile_problems(patient_id, normalized, note_id)

    # Step 6: Store everything.
    logger.info("Step 6: Storing results")
    result = store_results(patient_id, normalized, recommendations, note_id)

    logger.info(
        "Done. Extracted %d problems, %d active, %d recommendations.",
        len(normalized),
        sum(1 for p in normalized if p["assertion"] == "PRESENT"),
        len(recommendations)
    )
    return result

# --- Example usage with synthetic data ---

if __name__ == "__main__":
    # Synthetic clinical note. This mimics a typical progress note with
    # multiple sections and various assertion types.
    SAMPLE_NOTE = """
HISTORY OF PRESENT ILLNESS:
62-year-old male with type 2 diabetes and hypertension presents for routine
follow-up. Reports good medication adherence. Denies chest pain or shortness
of breath. Blood glucose has been well-controlled per home monitoring.

PAST MEDICAL HISTORY:
Pneumonia (2023, resolved)
Appendectomy (2015)
History of smoking, quit 2018

FAMILY HISTORY:
Mother with breast cancer, diagnosed age 58
Father with coronary artery disease

ASSESSMENT AND PLAN:
1. Type 2 diabetes - well controlled, A1c 6.8%. Continue metformin.
2. Hypertension - at goal on lisinopril 20mg daily.
3. Chronic kidney disease stage 3 - stable, Cr 1.6. Continue monitoring.
4. Depression - started on sertraline 3 months ago, reports improvement.
   PHQ-9 score improved from 14 to 6. Continue current dose.
5. Obesity - BMI 32. Discussed dietary modifications.
"""

    result = process_note_for_problems(
        patient_id="PAT-2847103",
        note_text=SAMPLE_NOTE,
        note_id="NOTE-20260301-0482"
    )

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