Managing Token Limits in Multi-Turn Agentic Loops

Token Accumulation Patterns in Agentic Loops

Multi-turn agentic loops accumulate tokens at a rate that surprises most developers. A typical agent loop adds 400–1,200 tokens per turn: the user message or tool result (200–800 tokens) plus the assistant response (200–600 tokens). At this rate, a 32K context window fills in 25–80 turns. A 128K window fills in 100–320 turns. For tasks that run hundreds of turns, even the largest available context windows are insufficient without active management.

Accumulation by Context Component

• Conversation history (fastest growing): Every turn adds to this. Doubles every 20–50 turns depending on verbosity. Must be actively managed.
• Tool call results (bursty growth): Each tool call appends its full output. A single web search can add 2,000–8,000 tokens. Manage per-tool output size.
• System prompt (stable): Fixed at session start. Grows only when capabilities are added dynamically. Keep static or use lazy loading.
• Working memory (moderate growth): Scratch pad information the agent accumulates. Can be explicitly managed by the agent itself if instructed to compress its notes.
• RAG injections (per-turn additions): Each retrieval adds document chunks. Without explicit management, retrieval context compounds across turns.

Sliding Window Strategy

The sliding window is the simplest and most reliable token management strategy for multi-turn agents. It maintains a fixed-size window of N recent turns plus a compressed summary of everything before the window. As the agent advances, the window slides forward, with the oldest turn in the window being compressed and added to the summary.

Window Size Calibration

The right window size depends on your task's reference horizon, how far back does the agent typically need to look for relevant context? For question-answering tasks with short follow-ups, 5–8 turns is sufficient. For complex multi-step tasks with back-references, 10–15 turns. For tasks where any turn might reference any earlier turn (code review, document editing), use full context with compression rather than a sliding window.

Sliding Window Implementation

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class SlidingWindowManager:
    window_size: int = 8
    compressor: "LLMCompressor"  # async callable
    _window: list[dict] = field(default_factory=list)
    _rolling_summary: Optional[str] = None
    _summary_tokens: int = 0

    async def add_turn(
        self,
        role: str,
        content: str,
        token_counter,
    ) -> list[dict]:
        """Add a new turn and return the current context for LLM call."""
        self._window.append({"role": role, "content": content})

        if len(self._window) > self.window_size:
            # Oldest turn falls out of window, compress it into summary
            evicted = self._window.pop(0)
            await self._update_summary(evicted, token_counter)

        return self._build_context()

    async def _update_summary(
        self, evicted_turn: dict, token_counter
    ) -> None:
        """Incorporate an evicted turn into the rolling summary."""
        turn_text = f"{evicted_turn['role'].upper()}: {evicted_turn['content']}"

        if self._rolling_summary is None:
            # First summary, just compress the evicted turn
            self._rolling_summary = await self.compressor.compress(
                turn_text, target_tokens=200
            )
        else:
            # Merge evicted turn into existing summary
            merged = f"{self._rolling_summary}\n\nNew information:\n{turn_text}"
            self._rolling_summary = await self.compressor.compress(
                merged, target_tokens=self._summary_tokens + 100
            )

        self._summary_tokens = token_counter(self._rolling_summary)

    def _build_context(self) -> list[dict]:
        messages = []
        if self._rolling_summary:
            messages.append({
                "role": "system",
                "content": f"[CONTEXT SUMMARY - Earlier conversation]\n{self._rolling_summary}",
            })
        messages.extend(self._window)
        return messages

Window Strategy Tradeoffs

• Too small a window (< 4 turns): Agent frequently loses short-term context. Asks for information just provided, repeats completed steps. Noticeable quality degradation.
• Too large a window (> 20 turns): Minimal compression benefit. Context fills up nearly as fast as without windowing. Defeats the purpose.
• Asymmetric windows: Keep more turns from the beginning (task setup) and end (recent work), compress the middle. Better matches typical reference patterns in task-oriented agents.

Checkpoint Summarization Pattern

Checkpoint summarization differs from sliding window in that it's triggered at milestone events rather than at a fixed turn interval. When the agent completes a meaningful sub-task, a checkpoint is created: the context up to this point is summarized and stored, and the next phase starts with a clean context seeded from the checkpoint summary.

Checkpoint Triggers

• Phase completion: Agent transitions from research to planning, or planning to execution. Natural point to create a compressed summary of findings.
• Sub-task completion: Agent finishes a discrete deliverable (writes a document, executes a query batch, completes a test run). Checkpoint captures the result and discards the work context.
• Budget threshold: Context utilization reaches 75%. Forced checkpoint regardless of task phase.
• Explicit summarize instruction: Agent receives a 'checkpoint now' instruction in the system prompt at regular intervals. Simplest to implement.

Checkpoint Content Structure

from dataclasses import dataclass
import json
from datetime import datetime

@dataclass
class ContextCheckpoint:
    session_id: str
    checkpoint_id: str
    created_at: str
    turn_number: int
    task_phase: str
    summary: str             # compressed narrative of history to this point
    key_facts: list[str]     # exact preserved values (numbers, names, constraints)
    completed_subtasks: list[str]   # what has been accomplished
    pending_subtasks: list[str]     # what remains to do
    decisions_made: list[dict]      # explicit decisions with rationale
    constraints: list[str]          # hard constraints still in effect
    active_tool_states: dict        # current state of stateful tools

    def to_system_message(self) -> str:
        """Convert checkpoint to a compact system message for next phase."""
        return f"""[SESSION CHECKPOINT, Turn {self.turn_number}]
Phase: {self.task_phase}

SUMMARY:
{self.summary}

KEY FACTS (exact values preserved):
{chr(10).join(f"- {f}" for f in self.key_facts)}

COMPLETED:
{chr(10).join(f"- {s}" for s in self.completed_subtasks)}

REMAINING:
{chr(10).join(f"- {s}" for s in self.pending_subtasks)}

ACTIVE CONSTRAINTS:
{chr(10).join(f"- {c}" for c in self.constraints)}"""

    def save(self, storage) -> None:
        key = f"checkpoint:{self.session_id}:{self.checkpoint_id}"
        storage.set(key, json.dumps(self.__dict__), ex=86400)  # 24h TTL

Conversation Compaction Techniques

Conversation compaction is the process of replacing a full conversation history with a compressed representation that preserves the semantic content needed for task completion. Unlike checkpointing (which creates a new start point), compaction maintains the appearance of continuous conversation while reducing token count.

Compaction Strategy 1: Batch Summarization

Group the oldest N turns and compress them into a single summary message. Insert the summary at the start of the conversation with a special role annotation like [SUMMARY]. The model treats this as prior context. Effective compression ratio: 4:1 to 6:1 for typical conversational turns.

Compaction Strategy 2: Semantic Deduplication

Identify turns that contain the same information (common in verbose agentic loops where the agent repeats itself). Embed each turn and cluster similar ones. Keep only the most information-dense turn from each cluster. This is especially effective for tool result turns where the same tool is called multiple times with similar outputs.

Compaction Strategy 3: Role-Selective Compaction

async def role_selective_compact(
    messages: list[dict],
    max_tokens: int,
    token_counter,
    llm_compressor,
) -> list[dict]:
    """
    Compact selectively based on message role.
    - System messages: keep unchanged (protected)
    - Tool results: compress to key findings only
    - Assistant reasoning: compress to conclusion only
    - User messages: keep unchanged (user intent)
    """
    total = sum(token_counter(m["content"]) for m in messages)
    if total <= max_tokens:
        return messages

    compacted = []
    for msg in messages:
        role = msg.get("role", "")
        content = msg["content"]
        token_count = token_counter(content)

        if role == "system":
            compacted.append(msg)  # always keep

        elif role == "user":
            compacted.append(msg)  # always keep user intent

        elif role == "tool":
            if token_count > 300:
                # Compress to key findings
                compressed = await llm_compressor.extract_key_findings(
                    content, max_tokens=150
                )
                compacted.append({**msg, "content": compressed})
            else:
                compacted.append(msg)

        elif role == "assistant":
            if token_count > 400:
                # Keep conclusion, compress reasoning
                compressed = await llm_compressor.extract_conclusion(
                    content, max_tokens=120
                )
                compacted.append({**msg, "content": compressed})
            else:
                compacted.append(msg)

    return compacted

Token Management in Multi-Agent Loops

Multi-agent systems have an additional complexity: messages pass between agents, and each agent accumulates its own context. A message that is a summary for the orchestrator might be a full detailed report for the worker. Token management must be designed at the system level, not just per-agent.

Cross-Agent Message Size Contracts

Define explicit token budgets for inter-agent messages. The orchestrator sends task assignments that should not exceed 500 tokens. Workers return results that should not exceed 1,000 tokens. Intermediate reasoning should be compressed before transmission. These contracts prevent a single verbose worker from exhausting the orchestrator's context budget.

Hierarchical Summarization for Agent Chains

• Worker to orchestrator: Workers compress their full output to a structured summary before returning to the orchestrator. Include: conclusion, key facts, any exceptions.
• Orchestrator to parent: Orchestrators compress their multi-worker synthesis before passing to any parent agent. Keep only the consolidated finding.
• Error propagation: Error context should be summarized but not lost. Pass error type, cause, and resolution attempt. Omit stack traces in inter-agent messages.
• Token budget headers: Include the token count in each inter-agent message so the receiver can plan context budget before loading the message.

Tool Call Token Overhead

Tool calls add token overhead in two places: the tool call specification itself (sent in the assistant message) and the tool result (returned in the tool message). Both must be managed as part of the overall context budget.

Tool Output Size Management

import json
from typing import Any

MAX_TOOL_OUTPUT_TOKENS = {
    "web_search": 800,
    "database_query": 600,
    "file_read": 1200,
    "api_call": 500,
    "code_execution": 400,
}

async def managed_tool_call(
    tool_name: str,
    tool_fn,
    tool_args: dict,
    token_counter,
    compressor,
) -> str:
    """Execute tool and compress output if it exceeds the budget."""
    raw_result = await tool_fn(**tool_args)
    result_str = json.dumps(raw_result) if not isinstance(raw_result, str) else raw_result

    max_tokens = MAX_TOOL_OUTPUT_TOKENS.get(tool_name, 600)
    actual_tokens = token_counter(result_str)

    if actual_tokens <= max_tokens:
        return result_str

    # Compress oversized result
    compressed = await compressor.compress_tool_output(
        tool_name=tool_name,
        content=result_str,
        target_tokens=max_tokens,
        original_tokens=actual_tokens,
    )
    return compressed + f" [compressed from {actual_tokens} to {token_counter(compressed)} tokens]"

Tool Call Result Lifecycle

• Injection (turn N): Full tool output added to context. This is when maximum token count is incurred.
• Reference window (turns N to N+K): Tool result referenced by agent reasoning. Should be kept in full during this window.
• Summary window (turns N+K to N+M): Tool result still potentially relevant but no longer being actively reasoned about. Compress to key findings.
• Archive window (turns N+M onwards): Tool result no longer relevant. Can be removed from context entirely.
• Adaptive management: Track whether each tool result is referenced in subsequent turns. Set K and M based on actual reference patterns in your task type.

Long-Running Agent Session Architecture

For agents that run hours or days on a single task, the context management problem is qualitatively different from a standard multi-turn conversation. You need an architecture designed for persistence, recovery, and efficient state management at scale.

External Memory Architecture

Long-running agents should use external memory as the primary store, with the context window as a cache. The agent works from a minimal in-context state, retrieves from external memory as needed, and persists findings back to external memory after each significant step.

External Memory Components

• Task state store (Redis/DynamoDB): Current phase, completed steps, pending steps, decisions made. Small and highly structured. Retrieved in full each turn.
• Knowledge store (vector DB): Facts, findings, and retrieved information accumulated during the task. Retrieved semantically as needed.
• Tool result cache (S3/blob storage): Full tool outputs compressed and stored. Referenced by ID in context; loaded only when needed.
• Conversation archive (PostgreSQL): Full conversation history compressed and stored. Retrieved via checkpoint summaries, not raw history.
• Working notes (Redis): Agent's scratch pad. Small, frequently updated, loaded in full each turn.

Recovery and Session Resumption

A robust long-running agent must be able to recover from interruptions. The session should be resumable from the last checkpoint with minimal context loss.

Resumption Implementation

async def resume_agent_session(
    session_id: str,
    storage,
    llm_fn,
    tools: list,
) -> "AgentSession":
    """Resume an agent session from its latest checkpoint."""
    # Load latest checkpoint
    checkpoint_key = f"checkpoint:{session_id}:latest"
    checkpoint_data = storage.get(checkpoint_key)
    if not checkpoint_data:
        raise ValueError(f"No checkpoint found for session {session_id}")

    import json
    checkpoint = ContextCheckpoint(**json.loads(checkpoint_data))

    # Reconstruct initial context from checkpoint
    system_message = checkpoint.to_system_message()

    # Load any in-progress tool calls
    pending_tools = storage.get(f"pending_tools:{session_id}") or []

    # Build the resumption prompt
    resumption_prompt = (
        f"You are resuming a task that was interrupted at turn {checkpoint.turn_number}.\n"
        f"Review the checkpoint summary above and continue from where you left off.\n"
        f"Current phase: {checkpoint.task_phase}\n"
        f"Remaining tasks: {', '.join(checkpoint.pending_subtasks)}"
    )

    return AgentSession(
        session_id=session_id,
        system_prompt=system_message,
        initial_user_message=resumption_prompt,
        tools=tools,
        checkpoint=checkpoint,
        llm_fn=llm_fn,
    )

Production Implementation Patterns

Combining all the above techniques into a production-grade token management system requires careful integration. Here is the reference architecture that covers all cases.

Token Management Middleware Pattern

Implement token management as middleware that wraps the LLM call. The agent code makes a standard LLM call; the middleware intercepts it, applies the appropriate management strategy, and returns the result. This keeps token management logic separate from task logic.

Management Strategy Selection

• < 5 turns: No management needed. Pass full history.
• 5–15 turns, budget < 70%: Light pruning only. Remove clearly redundant tool results.
• 5–15 turns, budget 70–85%: Sliding window kicks in. Compress turns beyond window_size.
• > 15 turns, budget > 85%: Checkpoint summarization. Create a new start point from compressed history.
• > 50 turns: External memory architecture. Stop growing in-context history; use semantic retrieval from external store.

Benchmarking Your Token Management Strategy

Token management is a system that makes quality-cost tradeoffs. Benchmarking validates that your strategy stays above the minimum quality threshold while achieving its cost and token reduction targets.

Benchmark Metrics

• Task completion rate: Percentage of tasks completed successfully end-to-end. Should not drop by more than 3% compared to no-management baseline.
• Answer accuracy at turn N: Run factual accuracy probes at turns 5, 10, 20, 50. Measures whether important context is preserved at scale.
• Tokens per task: Average total input tokens to complete a task. Your management system should reduce this by 30–60% vs unmanaged.
• Context overhead ratio: Tokens spent on management operations (compression calls) vs tokens saved. Target: >3:1 savings ratio.
• Recovery success rate: When an agent session is resumed from a checkpoint, percentage of cases where it successfully continues without requesting already-provided information.