Context Window Management

Optimizing the finite context window for agent performance.

The Constraint

Context Window = Fixed Token Budget
├── System Prompt (always)
├── Task/Goal (always)
├── Available Tools (always)
├── Relevant Memory (variable)
├── Recent History (variable)
├── Retrieved Context (variable)
└── Generation Budget (reserved)

Key insight: Every token in context costs money, latency, and attention quality.

Budget Allocation

Static Allocation

class ContextBudget:
    def __init__(self, max_tokens: int):
        self.max = max_tokens
        self.allocation = {
            "system": 0.05,      # 5%
            "tools": 0.10,       # 10%
            "memory": 0.25,      # 25%
            "history": 0.20,     # 20%
            "retrieval": 0.30,   # 30%
            "generation": 0.10   # 10% reserved
        }

    def tokens_for(self, category: str) -> int:
        return int(self.max * self.allocation[category])

Dynamic Allocation (Adaptive)

class AdaptiveBudget:
    def allocate(self, task: Task, context: Context) -> dict[str, int]:
        # Complex tasks → more retrieval
        if task.complexity > 0.7:
            retrieval = 0.40
        else:
            retrieval = 0.20

        # Long history → compress
        if context.history_tokens > self.max * 0.3:
            history = 0.10
        else:
            history = 0.20

        return self._balance({
            "system": fixed,
            "tools": fixed,
            "memory": adaptive,
            "history": history,
            "retrieval": retrieval,
            "generation": reserved
        })

Compression Strategies

1. Summarization

async def summarize_history(messages: list[Message], budget: int) -> str:
    # Hierarchical: recent full, older summarized
    recent = messages[-10:]  # Keep last 10 full
    older = messages[:-10]

    if older:
        summary = await llm.summarize(older, max_tokens=budget)
        return f"[Summary: {summary}]\n\n" + "\n".join(recent)
    return "\n".join(recent)

2. Extractive Compression

def extract_key_info(messages: list[Message], query: str, k: int) -> list[Message]:
    # Retrieve most relevant messages for current query
    embeddings = embed([m.content for m in messages])
    query_emb = embed(query)
    scores = cosine_similarity(query_emb, embeddings)
    top_indices = np.argsort(scores)[-k:]
    return [messages[i] for i in top_indices]

3. Structured Compression

def compress_to_structured(messages: list[Message]) -> StructuredContext:
    return StructuredContext(
        decisions=extract_decisions(messages),
        facts=extract_facts(messages),
        open_questions=extract_questions(messages),
        pending_actions=extract_actions(messages),
        errors=extract_errors(messages)
    )

4. Token-Aware Truncation

def smart_truncate(messages: list[Message], budget: int) -> list[Message]:
    # Priority: system > task > tools > recent > memory > history
    priority_order = [
        ("system", messages.system),
        ("task", messages.task),
        ("tools", messages.tools),
        ("recent", messages.recent[-5:]),
        ("memory", messages.memory),
        ("history", messages.history)
    ]

    result = []
    remaining = budget
    for category, items in priority_order:
        for item in items:
            if item.tokens <= remaining:
                result.append(item)
                remaining -= item.tokens
            else:
                break
    return result

Context Construction Pipeline

class ContextBuilder:
    def __init__(self, budget: ContextBudget):
        self.budget = budget

    async def build(self, task: Task, state: AgentState) -> list[Message]:
        parts = []

        # 1. System (always)
        parts.append(SystemMessage(state.system_prompt))

        # 2. Task (always)
        parts.append(UserMessage(f"Goal: {task.description}"))

        # 3. Tools (always)
        parts.append(SystemMessage(format_tools(state.tools)))

        # 4. Relevant memory (semantic search)
        memory_tokens = self.budget.tokens_for("memory")
        memories = await state.memory.search(task.query, k=5, token_budget=memory_tokens)
        parts.extend(memories)

        # 5. Retrieved context (RAG)
        retrieval_tokens = self.budget.tokens_for("retrieval")
        docs = await state.rag.retrieve(task.query, token_budget=retrieval_tokens)
        parts.extend(docs)

        # 6. Recent history
        history_tokens = self.budget.tokens_for("history")
        recent = state.history.get_recent(token_budget=history_tokens)
        parts.extend(recent)

        # 7. Validate total
        total = sum(m.tokens for m in parts)
        if total > self.budget.max:
            parts = self.compress(parts, self.budget.max)

        return parts

Advanced Techniques

1. Attention Sinks (Preserve First Tokens)

# Research: First few tokens critical for attention
def preserve_sinks(messages, sink_size=4):
    sinks = messages[:sink_size]  # System, task, tools
    rest = compress(messages[sink_size:])
    return sinks + rest

2. Recursive Summarization

async def recursive_summarize(messages, target_tokens):
    if total_tokens(messages) <= target_tokens:
        return messages

    # Summarize in chunks
    chunk_size = target_tokens // 2
    chunks = chunk_messages(messages, chunk_size)
    summaries = await asyncio.gather(*[llm.summarize(c) for c in chunks])
    return await recursive_summarize(summaries, target_tokens)

3. Retrieval-Augmented Context

# Instead of stuffing all history, retrieve relevant
def rag_context(history, query, k=5):
    relevant = semantic_search(history, query, k=k)
    return format_as_context(relevant)

Model-Specific Optimization

Monitoring & Metrics

class ContextMetrics:
    def __init__(self):
        self.metrics = {
            "total_tokens": [],
            "utilization": [],
            "compression_ratio": [],
            "retrieval_relevance": [],
            "generation_quality": []
        }

    def record(self, context: Context, response: Response):
        self.metrics["total_tokens"].append(context.total_tokens)
        self.metrics["utilization"].append(context.total_tokens / context.max_tokens)
        self.metrics["compression_ratio"].append(context.original_tokens / context.final_tokens)
        self.metrics["retrieval_relevance"].append(evaluate_relevance(context.retrieved))
        self.metrics["generation_quality"].append(evaluate_quality(response))

Anti-Patterns

Related Notes

• RAG Fundamentals — Retrieval as context source
• Advanced RAG — Adaptive retrieval
• Memory Systems — Memory as context source
• Multimodal Context — Multimodal context budgeting
• Inference Optimization — KV cache, context optimization
• LangGraph — Context in state
• Cost Optimization — Token cost management
• Research Directions — Infinite context research

Sources: "Lost in the Middle" (Liu et al.), "Attention Sinks" (Xiao et al.), "Recursive Summarization" (Wu et al.), LongLoRA, LongContextReorder, LlamaIndex context management, Anthropic prompt engineering, OpenAI context optimization guides.