LLM Context Window Management in Production: What Nobody Tells You

def get_context_messages(conversation_history: list, max_turns: int = 10) -> list:
    """Keep only the most recent N turns."""
    return conversation_history[-max_turns * 2:]  # *2 for user + assistant pairs

import anthropic
 
client = anthropic.Anthropic()
 
 
def count_tokens(messages: list, model: str = "claude-sonnet-5") -> int:
    """Count tokens for a message list using Anthropic's token counting API."""
    response = client.messages.count_tokens(
        model=model,
        messages=messages
    )
    return response.input_tokens
 
 
def trim_to_token_budget(
    messages: list,
    system_prompt: str,
    max_tokens: int = 150_000,
    model: str = "claude-sonnet-5"
) -> list:
    """
    Trim conversation history to stay within token budget.
    Always preserves the system prompt and the most recent user message.
    """
    system_tokens = count_tokens([{"role": "user", "content": system_prompt}], model)
    available = max_tokens - system_tokens - 2000  # 2K buffer for response
 
    # Start with all messages, remove oldest pairs until we fit
    trimmed = messages.copy()
    while len(trimmed) > 2:  # Keep at least the last exchange
        current_tokens = count_tokens(trimmed, model)
        if current_tokens <= available:
            break
        # Remove the oldest user+assistant pair
        trimmed = trimmed[2:]
 
    return trimmed

def summarize_conversation(messages: list, model: str = "claude-haiku-4-5-20251001") -> str:
    """
    Compress older conversation turns into a concise summary.
    Uses a faster/cheaper model for the summarization step.
    """
    client = anthropic.Anthropic()
 
    conversation_text = "\n".join([
        f"{m['role'].upper()}: {m['content']}"
        for m in messages
    ])
 
    response = client.messages.create(
        model=model,  # Use Haiku for cheap summarization
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Summarize this conversation history concisely.
Preserve: key decisions made, important facts stated, user preferences, any errors encountered.
Discard: pleasantries, repeated information, resolved issues.
 
CONVERSATION:
{conversation_text}
 
Provide a 3-5 sentence factual summary."""
        }]
    )
 
    return response.content[0].text
 
 
def manage_context_with_summary(
    messages: list,
    system_prompt: str,
    token_threshold: int = 80_000
) -> tuple[list, str]:
    """
    When context exceeds threshold, summarize old messages.
    Returns (trimmed_messages, updated_system_prompt_with_summary).
    """
    current_tokens = count_tokens(messages)
 
    if current_tokens < token_threshold:
        return messages, system_prompt
 
    # Split: keep last 6 turns fresh, summarize everything before
    recent_messages = messages[-12:]  # Last 6 exchanges
    old_messages = messages[:-12]
 
    if not old_messages:
        return messages, system_prompt
 
    summary = summarize_conversation(old_messages)
 
    # Inject summary into system prompt
    updated_system = f"""{system_prompt}
 
## Conversation History Summary
{summary}"""
 
    return recent_messages, updated_system

from anthropic import Anthropic
 
client = Anthropic()
 
 
def rag_query(
    user_question: str,
    document_chunks: list[str],
    top_k: int = 5
) -> str:
    """
    Simple semantic search + retrieval before sending to Claude.
    In production, replace simple_search with a proper vector store.
    """
    # Retrieve only relevant chunks
    relevant_chunks = simple_search(document_chunks, user_question, top_k)
 
    context = "\n\n---\n\n".join(relevant_chunks)
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1000,
        system=f"""Answer questions based on the provided context.
If the context doesn't contain the answer, say so clearly.
 
CONTEXT:
{context}""",
        messages=[{
            "role": "user",
            "content": user_question
        }]
    )
 
    return response.content[0].text
 
 
def simple_search(chunks: list[str], query: str, top_k: int) -> list[str]:
    """
    Placeholder — replace with embeddings + vector search in production.
    (ChromaDB, Pinecone, pgvector, etc.)
    """
    # Simple keyword overlap for illustration
    scored = [(chunk, sum(1 for word in query.lower().split()
                         if word in chunk.lower())) for chunk in chunks]
    scored.sort(key=lambda x: x[1], reverse=True)
    return [chunk for chunk, _ in scored[:top_k]]

import logging
 
logger = logging.getLogger(__name__)
 
 
def call_claude_with_monitoring(
    messages: list,
    system: str,
    model: str = "claude-sonnet-5"
) -> str:
    client = Anthropic()
 
    response = client.messages.create(
        model=model,
        max_tokens=2000,
        system=system,
        messages=messages
    )
 
    # Log for monitoring
    logger.info({
        "event": "llm_call",
        "model": model,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
        "total_tokens": response.usage.input_tokens + response.usage.output_tokens,
        "estimated_cost_usd": _estimate_cost(response.usage, model)
    })
 
    return response.content[0].text
 
 
def _estimate_cost(usage, model: str) -> float:
    """Rough cost estimation — check Anthropic pricing for current rates."""
    pricing = {
        "claude-sonnet-5": {"input": 3.0, "output": 15.0},   # per million tokens
        "claude-haiku-4-5-20251001": {"input": 0.25, "output": 1.25},
    }
    p = pricing.get(model, {"input": 3.0, "output": 15.0})
    return (usage.input_tokens * p["input"] + usage.output_tokens * p["output"]) / 1_000_000