Agent Context Management

Why This Matters

Context is an agent's working memory—its ability to "remember" what happened in the conversation and use that information intelligently. Without proper context management, agents either forget crucial details, waste tokens on irrelevant information, or exceed context limits and crash. Mastering context management is the difference between a helpful assistant and a forgetful chatbot.

Real-World Analogy: Think of your brain during a conversation. You don't remember every word you've ever heard—you keep recent details in focus, summarize older information, and sometimes forget irrelevant tangents. Context management does the same for AI agents: it decides what to keep, what to compress, and what to discard.

Context Management Architecture

Context Window Manager

import tiktoken
import time
import logging
from dataclasses import dataclass, field
from typing import Any, Optional
from collections import deque

logger = logging.getLogger(__name__)


@dataclass
class Message:
    role: str
    content: str
    timestamp: float = field(default_factory=time.time)
    tokens: int = 0
    importance: float = 0.5
    metadata: dict = field(default_factory=dict)

    def __post_init__(self):
        if self.tokens == 0:
            self.tokens = len(self.content.split()) * 1.3


class ContextWindowManager:
    def __init__(self, max_tokens: int = 4096, model: str = "gpt-4"):
        self.max_tokens = max_tokens
        self.system_prompt_tokens = 0
        self.reserved_tokens = 512
        self.encoding = tiktoken.encoding_for_model(model)
        self.messages: deque[Message] = deque()
        self.total_tokens = 0

    def set_system_prompt(self, prompt: str):
        self.system_prompt_tokens = len(self.encoding.encode(prompt))

    def add_message(self, message: Message):
        self.messages.append(message)
        self.total_tokens += message.tokens
        self._trim_if_needed()

    def _trim_if_needed(self):
        available = self.max_tokens - self.system_prompt_tokens - self.reserved_tokens
        while self.total_tokens > available and len(self.messages) > 1:
            removed = self.messages.popleft()
            self.total_tokens -= removed.tokens
            logger.debug(f"Evicted message: {removed.role} ({removed.tokens} tokens)")

    def get_context(self) -> list[dict]:
        context = []
        if self.system_prompt_tokens > 0:
            context.append({
                "role": "system",
                "content": self._get_system_prompt(),
            })
        for msg in self.messages:
            context.append({
                "role": msg.role,
                "content": msg.content,
            })
        return context

    def _get_system_prompt(self) -> str:
        return ""

    def get_token_usage(self) -> dict:
        available = self.max_tokens - self.system_prompt_tokens - self.reserved_tokens
        return {
            "total": self.max_tokens,
            "system": self.system_prompt_tokens,
            "used": self.total_tokens,
            "available": max(0, available - self.total_tokens),
            "utilization": self.total_tokens / max(available, 1),
        }

    def clear(self):
        self.messages.clear()
        self.total_tokens = 0

Sliding Window Strategy

from collections import deque
from dataclasses import dataclass
from typing import Any, Callable, Optional
import hashlib
import time
import logging

logger = logging.getLogger(__name__)


@dataclass
class WindowMessage:
    role: str
    content: str
    message_id: str = ""
    tokens: int = 0
    pinned: bool = False

    def __post_init__(self):
        if not self.message_id:
            self.message_id = hashlib.md5(
                f"{self.content}{time.time()}".encode()
            ).hexdigest()[:8]
        if self.tokens == 0:
            self.tokens = len(self.content.split()) * 1.3


class SlidingWindowContext:
    def __init__(self, window_size: int = 20, max_tokens: int = 4096):
        self.window_size = window_size
        self.max_tokens = max_tokens
        self.messages: deque[WindowMessage] = deque(maxlen=window_size)
        self.pinned_messages: list[WindowMessage] = []
        self.total_tokens = 0

    def add(self, message: WindowMessage):
        self.messages.append(message)
        self.total_tokens += message.tokens
        self._enforce_token_limit()

    def pin(self, message: WindowMessage):
        message.pinned = True
        self.pinned_messages.append(message)
        self.total_tokens += message.tokens

    def unpin(self, message_id: str):
        self.pinned_messages = [
            m for m in self.pinned_messages if m.message_id != message_id
        ]
        self._recalculate_tokens()

    def _enforce_token_limit(self):
        while self.total_tokens > self.max_tokens and self.messages:
            removed = self.messages.popleft()
            self.total_tokens -= removed.tokens
            logger.debug(f"Evicted: {removed.message_id}")

    def _recalculate_tokens(self):
        self.total_tokens = sum(m.tokens for m in self.messages) + sum(
            m.tokens for m in self.pinned_messages
        )

    def get_context(self) -> list[dict]:
        context = []
        for msg in self.pinned_messages:
            context.append({"role": msg.role, "content": msg.content})
        for msg in self.messages:
            context.append({"role": msg.role, "content": msg.content})
        return context

    def get_window_stats(self) -> dict:
        return {
            "window_size": self.window_size,
            "current_size": len(self.messages),
            "pinned_count": len(self.pinned_messages),
            "total_tokens": self.total_tokens,
            "utilization": self.total_tokens / self.max_tokens,
        }

Context Compression

from dataclasses import dataclass
from typing import Any, Callable, Optional
import re
import logging

logger = logging.getLogger(__name__)


@dataclass
class CompressionResult:
    original_tokens: int
    compressed_tokens: int
    summary: str
    key_points: list[str]
    compression_ratio: float


class ContextCompressor:
    def __init__(self, max_tokens: int = 1000):
        self.max_tokens = max_tokens
        self.extraction_strategies: list[Callable] = [
            self._extract_key_sentences,
            self._extract_entities,
            self._extract_actions,
        ]

    async def compress(self, text: str, style: str = "balanced") -> CompressionResult:
        original_tokens = int(len(text.split()) * 1.3)
        
        if style == "aggressive":
            summary = await self._aggressive_compress(text)
        elif style == "conservative":
            summary = await self._conservative_compress(text)
        else:
            summary = await self._balanced_compress(text)

        key_points = self._extract_key_points(text)
        compressed_tokens = int(len(summary.split()) * 1.3)

        return CompressionResult(
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            summary=summary,
            key_points=key_points,
            compression_ratio=compressed_tokens / max(original_tokens, 1),
        )

    async def _balanced_compress(self, text: str) -> str:
        sentences = self._split_sentences(text)
        if len(sentences) <= 3:
            return text
        important = self._rank_sentences(sentences)[:max(3, len(sentences) // 2)]
        return " ".join(important)

    async def _aggressive_compress(self, text: str) -> str:
        sentences = self._split_sentences(text)
        important = self._rank_sentences(sentences)[:2]
        return " ".join(important)

    async def _conservative_compress(self, text: str) -> str:
        sentences = self._split_sentences(text)
        return " ".join(sentences[:len(sentences) // 2 + 1])

    def _split_sentences(self, text: str) -> list[str]:
        return re.split(r'[.!?]+', text)

    def _rank_sentences(self, sentences: list[str]) -> list[str]:
        scored = []
        for s in sentences:
            score = 0
            if any(w in s.lower() for w in ["important", "critical", "must", "key"]):
                score += 2
            if any(w in s.lower() for w in ["therefore", "consequently", "result"]):
                score += 1
            score += len(s.split()) * 0.1
            scored.append((s.strip(), score))
        scored.sort(key=lambda x: x[1], reverse=True)
        return [s for s, _ in scored]

    def _extract_key_points(self, text: str) -> list[str]:
        points = []
        for strategy in self.extraction_strategies:
            points.extend(strategy(text))
        return list(set(points))[:5]

    def _extract_key_sentences(self, text: str) -> list[str]:
        return [s.strip() for s in self._split_sentences(text) if len(s.split()) > 5][:3]

    def _extract_entities(self, text: str) -> list[str]:
        return re.findall(r'\b[A-Z][a-z]+ (?:of|for) [A-Z][a-z]+\b', text)

    def _extract_actions(self, text: str) -> list[str]:
        return re.findall(r'(?:must|should|need to|will) \w+ \w+', text)

Performance Considerations

Strategy	Memory	Quality	Speed	Cost	Best For
Full Context	Unbounded	Highest	Slow	High	Short conversations
Sliding Window	Bounded	Medium	Fast	Low	Chat applications
Compression	Bounded	High	Medium	Medium	Long documents
Hierarchical	Bounded	High	Medium	High	Multi-level detail

Security Considerations

Token Limit Enforcement: Always enforce token limits to prevent context overflow attacks
Content Sanitization: Sanitize user inputs before adding to context to prevent injection
Memory Bounds: Set maximum context sizes to prevent resource exhaustion
Audit Logging: Log context truncation events for debugging and compliance

Mathematical Foundation

Context Utilization Ratio:

Compression Ratio:

Optimal Window Size (for sliding window):

Where is information loss and is computational cost.

Interview Questions

1. What is the difference between sliding window and compression strategies?

Answer: Sliding window keeps the most recent N messages, discarding old ones entirely—simple and constant memory but may lose important historical context. Compression retains information from all messages by summarizing or extracting key points—preserves more context but requires computation and may lose nuance. Sliding window is best for chat applications; compression is best for document analysis. Hybrid approaches use sliding window for recent messages and compression for older context.

2. How do you handle token limits when context exceeds the window?

Answer: Priority-based eviction: 1) System prompt is never evicted, 2) Pinned/important messages have highest priority, 3) Recent messages have higher priority than old, 4) Compress before evicting when possible, 5) Maintain a summary buffer for evicted content. Implementation: track message importance scores, use a priority queue for eviction, and maintain a compression cache. Always inform the user when context is truncated.

3. What is context window pollution and how do you prevent it?

Answer: Context window pollution occurs when irrelevant or redundant information fills the context, reducing the model's ability to focus on important content. Causes: verbose responses, repeated information, tool outputs, irrelevant metadata. Prevention: 1) Summarize tool outputs immediately, 2) Deduplicate repeated information, 3) Remove irrelevant context dynamically, 4) Use importance scoring to prioritize content.

4. How does dynamic token allocation work?

Answer: Dynamic allocation adjusts the proportion of tokens reserved for history vs. current context based on conversation stage and complexity: Early conversation: more history to maintain context. Complex tasks: more current context for reasoning. Simple queries: less context needed. Implementation: track conversation metrics (turns, topic shifts, complexity), adjust allocation ratios, and monitor model performance to tune parameters.

5. What are the tradeoffs of hierarchical context management?

Answer: Hierarchical context maintains multiple levels of detail: recent full messages, compressed summaries of older context, and high-level topic summaries. Benefits: preserves long-term context while staying within token limits. Tradeoffs: increased complexity, potential information loss during compression, and overhead from maintaining multiple levels. Best for: long-running agents, document analysis, multi-session conversations.

6. How would you implement context-aware compression?

Answer: Use the current task to guide compression: 1) Analyze the user's query to identify relevant information, 2) Score each piece of context by relevance to the current task, 3) Compress low-relevance content more aggressively, 4) Preserve high-relevance content in full, 5) Use semantic similarity to match context to query, 6) Implement attention mechanisms to weight important content.

7. How do you handle multi-turn context dependencies?

Answer: Track dependencies between messages using a context graph: 1) Link responses to the messages they reference, 2) Maintain a reference count for each context piece, 3) Never evict content with active references, 4) Compress referenced content but preserve the reference link, 5) Implement lazy evaluation—only include referenced context when needed.

8. What metrics should you monitor for context management?

Answer: Key metrics: 1) Token utilization — percentage of context window used, 2) Compression ratio — information preserved after compression, 3) Context accuracy — model performance with managed context vs. full context, 4) Latency impact — overhead from compression/management, 5) Memory efficiency — bytes per token of managed context, 6) Task success rate — end-to-end performance.

Common Pitfalls

Pitfall	Solution
Losing critical historical context	Implement importance scoring and pinning
Compression loses important details	Use multiple extraction strategies and validate
Token limit exceeded unexpectedly	Track tokens per message and enforce limits
Context window pollution	Summarize tool outputs and deduplicate
High latency from compression	Cache compression results and use async
Inconsistent context across turns	Maintain context graph with references
Memory leaks from unmanaged context	Implement TTL and automatic cleanup
Poor task performance	Monitor context quality and adjust strategies

Summary with Key Takeaways

Context windows have fixed token limits requiring careful management of system prompts, history, and response space
Sliding window provides constant memory with simple FIFO eviction; best for chat applications
Compression preserves more context through summarization; best for document analysis
Dynamic allocation adjusts token budget based on conversation stage and complexity
Hierarchical context maintains multiple detail levels for long-running agents
Context-aware compression uses the current task to guide what to preserve
Token tracking is essential to prevent unexpected overflows and optimize utilization
Monitoring utilization, compression quality, and task performance ensures effective context management

KnowledgeCheck

What is the primary advantage of a sliding window strategy?
- a) Preserves all historical context
- b) Constant memory usage
- c) Provides highest quality responses
- d) Requires no computation
What is context window pollution?
- a) The context window becoming too large
- b) Irrelevant information filling the context window
- c) The model hallucinating context
- d) Token counting errors
How does compression differ from sliding window?
- a) Compression discards old messages entirely
- b) Compression retains information through summarization
- c) Compression is faster than sliding window
- d) Compression uses less memory
What should have the highest priority in context management?
- a) Recent messages
- b) System prompt
- c) User messages
- d) Tool outputs
What is dynamic token allocation?
- a) Fixed token limits for all components
- b) Adjusting token budget based on conversation needs
- c) Randomly allocating tokens
- d) Using the maximum possible tokens
Why track token utilization metrics?
- a) To increase context window size
- b) To prevent unexpected overflows and optimize usage
- c) To reduce model latency
- d) To simplify the codebase

Answers: 1-b, 2-b, 3-b, 4-b, 5-b, 6-b

Agent Context Management: Windows, Compression & Sliding Window