# 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
```python
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)
```python
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
```python
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
```python
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
```python
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
```python
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

```python
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)
```python
# 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
```python
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
```python
# 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

| Model | Context | Strategy |
|---|---|---|
| **GPT-4o** | 128k | Standard budgeting |
| **Claude 3.5** | 200k | Generous retrieval |
| **Gemini 1.5** | 1M/2M | Near-unlimited, use for corpus |
| **o1** | 200k | Reserve for reasoning |
| **Llama 3.1** | 128k | Standard |
| **Local (8k-32k)** | Small | Aggressive compression |

---

## Monitoring & Metrics

```python
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

| Anti-Pattern | Problem | Fix |
|---|---|---|
| **Stuff everything** | OOM, poor attention, high cost | Budget + prioritize |
| **No compression** | Context grows unbounded | Recursive summarization |
| **Fixed retrieval k** | Irrelevant docs waste tokens | Adaptive k by query |
| **Ignoring generation budget** | Truncated outputs | Reserve tokens |
| **Same budget for all tasks** | Waste on simple, starve complex | Dynamic allocation |

---

## Related Notes

- [[ContextEngineering/RAG-Fundamentals.md]] — Retrieval as context source
- [[ContextEngineering/Advanced-RAG.md]] — Adaptive retrieval
- [[ContextEngineering/Memory-Systems.md]] — Memory as context source
- [[ContextEngineering/Multimodal-Context.md]] — Multimodal context budgeting
- [[ModelArchitectures/Inference-Optimization.md]] — KV cache, context optimization
- [[Frameworks/LangGraph.md]] — Context in state
- [[Production/Cost-Optimization.md]] — Token cost management
- [[Frontiers/Research-Directions.md]] — 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.*