Context Window Management#

Total Context Budget
├── System prompt (stable, cached)
├── Conversation history (grows with multi-turn interaction)
├── Retrieved context / RAG chunks
├── User message (current turn)
└── Output reservation (space for the response)

Budget Constraint:
  system_prompt + history + retrieved + user_message + output_reservation
  ≤ max_context_window

Context position quality map:
┌─────────────────────────────────────────────────────┐
│ HIGH ATTENTION    ██████████░░░░░░░░░░░████████████ │
│ (early tokens)   (degraded middle)   (late tokens)  │
└─────────────────────────────────────────────────────┘

"""
Progressive conversation history summarization for HMS clinical consultation.
Context: A care coordinator is having an extended conversation with the
clinical AI about a complex patient's care plan. After 10 turns, early
turns are summarized to preserve context budget.
"""
import anthropic
from dataclasses import dataclass, field

@dataclass
class ConversationMessage:
    role: str  # "user" or "assistant"
    content: str
    turn_index: int

@dataclass
class ManagedConversation:
    messages: list[ConversationMessage] = field(default_factory=list)
    summary: str = ""
    max_recent_turns: int = 10
    summary_trigger_turns: int = 15  # Summarize when history exceeds this


def should_summarize(conversation: ManagedConversation) -> bool:
    return len(conversation.messages) >= conversation.summary_trigger_turns


def summarize_early_turns(
    conversation: ManagedConversation,
    client: anthropic.Anthropic,
) -> str:
    """
    Summarize the earliest turns of the conversation to free context budget.
    Retains the most recent max_recent_turns turns verbatim.
    """
    turns_to_summarize = conversation.messages[:-conversation.max_recent_turns]
    
    if not turns_to_summarize:
        return conversation.summary

    conversation_text = "\n".join(
        f"{msg.role.upper()}: {msg.content}"
        for msg in turns_to_summarize
    )

    # Include previous summary if it exists
    prior_context = ""
    if conversation.summary:
        prior_context = f"Prior conversation summary:\n{conversation.summary}\n\n"

    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # Small model for summarization
        max_tokens=512,
        system=(
            "You are summarizing a clinical consultation conversation. "
            "Capture: clinical questions asked, key clinical facts discussed, "
            "decisions made, and any pending items. Be concise. Clinical accuracy required."
        ),
        messages=[{
            "role": "user",
            "content": f"{prior_context}Summarize this conversation:\n\n{conversation_text}",
        }],
    )

    return response.content[0].text


def build_context_for_llm(
    conversation: ManagedConversation,
    current_user_message: str,
) -> list[dict]:
    """Build the messages array for the LLM API call."""
    messages = []

    # Include summary as a system-level context if it exists
    if conversation.summary:
        messages.append({
            "role": "user",
            "content": f"[Conversation context from earlier turns]:\n{conversation.summary}",
        })
        messages.append({
            "role": "assistant",
            "content": "Understood. I have the context from our earlier discussion.",
        })

    # Include recent turns verbatim
    recent = conversation.messages[-conversation.max_recent_turns:]
    for msg in recent:
        messages.append({"role": msg.role, "content": msg.content})

    # Add current message
    messages.append({"role": "user", "content": current_user_message})

    return messages

"""
Map-reduce pattern for processing a long clinical document.
Context: HMS scenario — analyzing a 200-page Joint Commission survey
report to extract all action items and deficiency findings.
"""
import anthropic
from math import ceil

client = anthropic.Anthropic()


def chunk_document(text: str, max_tokens_per_chunk: int = 50_000) -> list[str]:
    """Split document into chunks that fit within context budget."""
    # Approximate: 1 token ≈ 4 characters for clinical text
    chars_per_chunk = max_tokens_per_chunk * 4
    chunks = []
    for i in range(0, len(text), chars_per_chunk):
        chunks.append(text[i : i + chars_per_chunk])
    return chunks


def map_extract_findings(chunk: str, chunk_index: int, total_chunks: int) -> list[dict]:
    """
    MAP phase: Extract action items from a single document chunk.
    Each chunk is processed independently — enables parallel processing.
    """
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=(
            "Extract all action items, deficiency findings, and required corrections "
            "from this document chunk. Return as JSON array: "
            '[{"finding": "...", "category": "...", "priority": "immediate|30-day|90-day", '
            '"responsible_department": "..."}]. '
            "If no findings, return empty array []."
        ),
        messages=[{
            "role": "user",
            "content": (
                f"Document chunk {chunk_index + 1} of {total_chunks}:\n\n{chunk}"
            ),
        }],
    )
    import json
    return json.loads(response.content[0].text)


def reduce_findings(all_findings: list[list[dict]]) -> dict:
    """
    REDUCE phase: Synthesize findings from all chunks into a consolidated report.
    """
    flat_findings = [f for chunk_findings in all_findings for f in chunk_findings]

    if not flat_findings:
        return {"total_findings": 0, "immediate_action": [], "findings": []}

    import json
    findings_json = json.dumps(flat_findings, indent=2)

    response = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=4096,
        system=(
            "You are synthesizing quality improvement findings from a survey report. "
            "Deduplicate findings, organize by priority and department, and produce "
            "a structured action plan in JSON format."
        ),
        messages=[{
            "role": "user",
            "content": f"Synthesize these findings into a consolidated action plan:\n\n{findings_json}",
        }],
    )

    return json.loads(response.content[0].text)


def process_long_document(document_text: str) -> dict:
    """Process a document that exceeds the context window using map-reduce."""
    chunks = chunk_document(document_text)
    
    # MAP phase (can be parallelized in production)
    chunk_findings = [
        map_extract_findings(chunk, i, len(chunks))
        for i, chunk in enumerate(chunks)
    ]

    # REDUCE phase
    return reduce_findings(chunk_findings)

"""
Hierarchical summarization for HMS patient record processing.
Context: Generating a comprehensive patient summary from 5 years
of clinical notes (>500 pages) for a new specialist consultation.
"""

def hierarchical_summarize(
    sections: list[str],
    client: anthropic.Anthropic,
    intermediate_max_tokens: int = 1024,
    final_max_tokens: int = 2048,
    target_specialty: str = "cardiology",
) -> str:
    """
    Two-level hierarchical summarization:
    Level 1: Summarize each section independently
    Level 2: Synthesize section summaries into final clinical summary

    target_specialty focuses the summary on clinically relevant details
    for the receiving specialist.
    """
    section_summaries = []
    for i, section in enumerate(sections):
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",  # Fast, cheap for intermediate step
            max_tokens=intermediate_max_tokens,
            system=(
                f"Summarize this clinical note section for a {target_specialty} specialist. "
                "Focus on: diagnoses, relevant findings, medications, procedures, "
                "and clinical course. Preserve dates and clinical values precisely."
            ),
            messages=[{"role": "user", "content": section}],
        )
        section_summaries.append(response.content[0].text)

    # Final synthesis
    combined = "\n\n---\n\n".join(
        f"Period {i+1} summary:\n{s}" for i, s in enumerate(section_summaries)
    )
    final_response = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=final_max_tokens,
        system=(
            f"Create a comprehensive clinical summary for a {target_specialty} consultation. "
            "Synthesize the periodic summaries into a coherent clinical narrative. "
            "Highlight trends, significant events, and items most relevant to the consultation."
        ),
        messages=[{"role": "user", "content": combined}],
    )
    return final_response.content[0].text

Illustrative ratio (verify current pricing in official vendor documentation):
- 2K context call:   2,000 tokens × rate = X
- 20K context call: 20,000 tokens × rate = 10X
- 200K context call: 200,000 tokens × rate = 100X