advancedadvancedtokenscontext-windowoptimizationcost
Chapter 5 of 8
Preventing Token Overload in AI Agent Loops
Token counting, conversation pruning, summarization, and cost optimization for Claude, GPT-4, and Gemini.
Token management is the single most overlooked bottleneck in production agent loops. Every tool in this space -- from Claude Code to Aider, Cursor to OpenHands -- must grapple with the same fundamental problem: naive agent loops accumulate conversation history linearly, and because every new request re-sends the entire history, token costs grow quadratically with each iteration.
Token budgets, not compute time, are the real bottleneck in production agent loops. This tutorial covers practical strategies for counting tokens, pruning conversations, summarizing history, and controlling costs -- grounded in real tools and documented case studies.
Related: For the fundamentals of loop mechanics, see the Agent Loop wiki page. For persisting loop state across iterations, see State Persistence.
Understanding Token Budgets
Every LLM has a maximum context window -- the total number of tokens it can process in a single request. This window is shared across all input: your system prompt, the full conversation history, the current user input, and the reserved space for the model's output.
Each loop iteration adds tokens to the conversation history. After just a few iterations of an agent loop that processes tool results, file contents, and multi-step reasoning, you can exhaust even a 200K-token context window.
Here is how a token budget is typically allocated:
┌─────────────────────────────────────────────────────────────────────┐
│ MODEL CONTEXT WINDOW (200K tokens) │
├────────────────┬──────────────────────┬──────────┬─────────┬────────┤
│ System Prompt │ Conversation History │ Current │ Reserved│ Safety │
│ (~500 tokens) │ (GROWS each iter) │ Input │ Output │ Margin │
│ │ │(~200 tok)│(~1000 t)│(~200 t)│
│ │ Iter 1: ~2000 tokens │ │
│ │ Iter 5: ~10000 tokens │ │
│ │ Iter 10: ~50000 tokens ← danger zone │ │
│ │ Iter 20: ~190000 tokens ← OVERFLOW │ │
├────────────────┴──────────────────────┴──────────┴─────────┴────────┤
│ System: 500 │ History: variable │ Input: 200 │ Out: 1000│ M: 200│
└─────────────────────────────────────────────────────────────────────┘
The two failure modes:
- Context overflow: When
System Prompt + History + Current Input + Reserved Outputexceeds the model's context window, the API returns an error. Your loop crashes. - Cost explosion: Even if you fit within the window, each iteration re-sends the entire growing history. At iteration 10, you are paying to process 50K tokens of input. At iteration 20, you are paying for 190K tokens -- and the model's output adds even more for the next iteration.
Model Context Limits (Real Numbers)
Different models have dramatically different context windows and pricing. Here are the current numbers for the most common models used in agent loops, based on Anthropic's official pricing at $3.00/1M input tokens for Claude 3.5 Sonnet and current published rates:
| Model | Context Window | Input Price | Output Price | Notes |
|---|---|---|---|---|
| Claude Sonnet 4 / Opus 4 | 200K tokens | $3 - $15 /M tokens | $15 - $75 /M tokens | Standard pricing; Opus is 5x the cost of Sonnet |
| Claude Sonnet 4.5 (beta, >200K) | 200K+ tokens | $6 /M input | $22.50 /M output | Beta extended-context pricing; 2x premium over standard |
| Claude Opus 4.6 / Sonnet 4.6 | 1M tokens | Varies | Varies | Newer models; extended window support |
| GPT-4o | 128K tokens | $2.50 /M tokens | $10 /M tokens | Smallest context window of the group |
| Gemini 2.0 Flash | 1M tokens | $0.075 /M tokens | $0.30 /M tokens | Dramatically cheaper; ~40x cheaper than Claude Opus for input |
Key takeaways:
- Gemini 2.0 Flash is roughly 40x cheaper than Claude Opus 4 for input tokens ($0.075 vs $15 per million tokens).
- Claude Code sessions crossing 200K tokens still carry a 2x premium on all tokens, even for newer models supporting 1M contexts.
- GPT-4o has the smallest context window at 128K, making it the most prone to overflow in long-running loops.
- Newer Claude models (Opus 4.6, Sonnet 4.6) support 1M-token contexts, but cost premiums apply.
Strategy 1: Accurate Token Counting
Token counting is the foundation of all budget management. If you cannot measure your token usage, you cannot control it.
For OpenAI Models: tiktoken
tiktoken is OpenAI's fast open-source BPE tokenizer for Python. It is the most accurate way to count tokens for OpenAI models.
import tiktoken
def count_tokens_openai(text: str, model: str = "gpt-4o") -> int:
"""Count tokens for OpenAI models using tiktoken."""
encoding = tiktoken.encoding_for_model(model)
return len(encoding.encode(text))
# Example usage
tokens = count_tokens_openai("Hello, how are you today?")
print(f"Tokens: {tokens}")
For Anthropic/Claude: Estimation and API Usage
Anthropic does not expose a public tokenizer. You have three options:
- API response field: After each request, check
response.usage.input_tokensfor the actual count. - Heuristic estimation:
len(text) // 4is a rough approximation (English text averages ~4 characters per token). - LiteLLM: Use the unified library for cross-provider counting (see below).
Claude Code (github.com/anthropics/claude-code, docs at code.claude.com/docs) automatically exposes token counts in its CLI output, which you can use to monitor usage in real time during interactive sessions.
Unified Counting with LiteLLM
LiteLLM provides a single interface for token counting across OpenAI, Anthropic, and Gemini:
import litellm
# Works across providers
tokens = litellm.token_counter(model="claude-sonnet-4-20250514", messages=messages)
tokens = litellm.token_counter(model="gpt-4o", messages=messages)
tokens = litellm.token_counter(model="gemini/gemini-2.0-flash", messages=messages)
This is particularly useful in multi-agent frameworks like LangGraph or CrewAI, where different agents in the same workflow may call different model providers.
TokenBudget Class
Here is a practical TokenBudget class that calculates remaining budget, counts tokens, and reports status:
import tiktoken
from dataclasses import dataclass, field
from typing import Optional
MODEL_LIMITS = {
"claude-sonnet-4-20250514": 200_000,
"claude-opus-4-20250514": 200_000,
"claude-sonnet-4.5-20250514": 200_000,
"gpt-4o": 128_000,
"gpt-4o-2024-08-06": 128_000,
"gemini-2.0-flash": 1_000_000,
}
# Approximate pricing per million tokens (Anthropic official pricing)
MODEL_PRICING = {
"claude-sonnet-4-20250514": {"input": 3.0, "output": 15.0},
"claude-opus-4-20250514": {"input": 15.0, "output": 75.0},
"claude-sonnet-4.5-20250514":{"input": 6.0, "output": 22.5},
"gpt-4o": {"input": 2.5, "output": 10.0},
"gemini-2.0-flash": {"input": 0.075, "output": 0.30},
}
@dataclass
class TokenBudget:
model: str
system_prompt_tokens: int = 500
reserved_output_tokens: int = 1000
safety_margin_tokens: int = 200
total_input_tokens: int = 0
@property
def context_limit(self) -> int:
return MODEL_LIMITS.get(self.model, 200_000)
@property
def available_for_history(self) -> int:
return (
self.context_limit
- self.system_prompt_tokens
- self.reserved_output_tokens
- self.safety_margin_tokens
)
@property
def history_budget_remaining(self) -> int:
return self.available_for_history - self.total_input_tokens
@property
def utilization_pct(self) -> float:
return (self.total_input_tokens / self.available_for_history) * 100
def count_tokens(self, text: str) -> int:
"""Count tokens using the appropriate method for the model."""
if "gpt" in self.model:
encoding = tiktoken.encoding_for_model(self.model)
return len(encoding.encode(text))
# Heuristic for Anthropic / Gemini (no public tokenizer)
return len(text) // 4
def add_tokens(self, count: int):
self.total_input_tokens += count
def is_over_budget(self) -> bool:
return self.total_input_tokens >= self.available_for_history
def report(self) -> str:
limit = self.context_limit
used = self.total_input_tokens
remaining = self.history_budget_remaining
pct = self.utilization_pct
status = "OK" if pct < 70 else "WARNING" if pct < 90 else "CRITICAL"
return (
f"[{status}] {self.model}: {used:,}/{self.available_for_history:,} "
f"tokens ({pct:.1f}%) | {remaining:,} remaining"
)
# Usage
budget = TokenBudget(model="claude-sonnet-4-20250514")
budget.add_tokens(50_000)
print(budget.report())
# [WARNING] claude-sonnet-4-20250514: 50,000/198,300 tokens (25.2%) | 148,300 remaining
Strategy 2: Conversation Pruning
Here are three approaches, implemented in a single ConversationPruner class:
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
class PruneStrategy(Enum):
OLDEST_FIRST = "oldest_first"
SLIDING_WINDOW = "sliding_window"
IMPORTANCE_BASED = "importance_based"
@dataclass
class Message:
role: str # "user" or "assistant"
content: str
important: bool = False # Used by importance-based pruning
token_count: int = 0
def __post_init__(self):
if self.token_count == 0:
self.token_count = len(self.content) // 4
class ConversationPruner:
"""Prune conversation history to fit within a token budget."""
def __init__(self, budget: TokenBudget):
self.budget = budget
def prune(
self,
messages: list[Message],
strategy: PruneStrategy = PruneStrategy.OLDEST_FIRST,
keep_recent: int = 5,
) -> list[Message]:
total_tokens = sum(m.token_count for m in messages)
if total_tokens <= self.budget.history_budget_remaining:
return messages # Nothing to prune
if strategy == PruneStrategy.OLDEST_FIRST:
return self._prune_oldest_first(messages)
elif strategy == PruneStrategy.SLIDING_WINDOW:
return self._prune_sliding_window(messages, keep_recent)
elif strategy == PruneStrategy.IMPORTANCE_BASED:
return self._prune_importance_based(messages, keep_recent)
else:
raise ValueError(f"Unknown strategy: {strategy}")
def _prune_oldest_first(self, messages: list[Message]) -> list[Message]:
"""Remove oldest messages until history fits within budget.
Pros: Simple, predictable.
Cons: Loses early context that may contain critical instructions or decisions.
"""
result = list(messages)
while sum(m.token_count for m in result) > self.budget.history_budget_remaining:
if len(result) <= 1:
break
removed = result.pop(0)
print(f"[prune] Removed oldest message ({removed.token_count} tokens)")
return result
def _prune_sliding_window(
self, messages: list[Message], keep_recent: int = 5
) -> list[Message]:
"""Keep only the last N exchanges (messages).
Pros: Very predictable token usage, easy to reason about.
Cons: Fixed window may be too small for complex tasks or too large for simple ones.
"""
return messages[-keep_recent:]
def _prune_importance_based(
self, messages: list[Message], keep_recent: int = 3
) -> list[Message]:
"""Keep messages marked as 'important' plus the most recent N messages.
Pros: Preserves critical context (e.g., task definitions, key decisions).
Cons: Requires upfront marking of important messages.
"""
budget_tokens = self.budget.history_budget_remaining
# Always keep recent messages
recent = messages[-keep_recent:]
remaining = [m for m in messages[:-keep_recent]]
# Keep important messages from older history
important = [m for m in remaining if m.important]
other = [m for m in remaining if not m.important]
result = []
total = 0
for msg in important + other:
if total + msg.token_count > budget_tokens - sum(m.token_count for m in recent):
break
result.append(msg)
total += msg.token_count
return result + recent
# Usage examples
budget = TokenBudget(model="claude-sonnet-4-20250514")
pruner = ConversationPruner(budget)
messages = [
Message(role="user", content="Fix the login bug in auth.py", important=True),
Message(role="assistant", content="I'll examine the auth.py file..."),
Message(role="assistant", content="Found the issue: the token validation is checking the wrong header."),
Message(role="user", content="Apply the fix"),
Message(role="assistant", content="Done. The fix has been applied."),
Message(role="user", content="Run the tests"),
Message(role="assistant", content="Running tests... all 47 tests pass."),
]
# Oldest-first: removes from the beginning until it fits
pruned = pruner.prune(messages, PruneStrategy.OLDEST_FIRST)
print(f"Oldest-first: {len(pruned)} messages remaining")
# Sliding window: keeps exactly the last 5
pruned = pruner.prune(messages, PruneStrategy.SLIDING_WINDOW, keep_recent=5)
print(f"Sliding window: {len(pruned)} messages remaining")
# Importance-based: keeps important messages + last 3
pruned = pruner.prune(messages, PruneStrategy.IMPORTANCE_BASED, keep_recent=3)
print(f"Importance-based: {len(pruned)} messages remaining")
Trade-off summary:
| Strategy | Predictability | Context Preservation | Implementation Complexity |
|---|---|---|---|
| Oldest-first | Low (loses early context) | Poor | Low |
| Sliding window | High (fixed N messages) | Medium (recent context only) | Low |
| Importance-based | Medium (depends on marking) | Good (keeps critical messages) | Medium |
Real-world relevance: Aider (github.com/paul-gauthier/aider, 30K+ GitHub stars) implements a sliding-window approach by default -- it keeps the repo map and recent changes in context but aggressively prunes older conversation turns. The Aider CLI command aider --model claude-3.5-sonnet manages this automatically, which is one reason it remains efficient even across long editing sessions.
Strategy 3: Summarization
Summarization replaces older conversation turns with a compressed version that preserves key information while dramatically reducing tokens. This is more effective than simple pruning because you retain context rather than discarding it entirely.
How It Works
- Keep the last 3-5 conversation turns verbatim (these need full fidelity).
- Summarize everything before that into a condensed summary.
- When the summary itself grows too large, summarize the summary (chain summarization).
Use a cheaper/faster model (e.g., Claude Haiku or Gemini 2.0 Flash) for summarization to keep costs down.
import anthropic
client = anthropic.Anthropic()
SUMMARIZE_PROMPT = """Summarize the following conversation history concisely.
Preserve: task objectives, key decisions made, code changes applied,
error messages encountered, and any constraints or requirements mentioned.
Output a structured summary in under 200 words."""
class ConversationSummarizer:
"""Summarize older conversation turns to reduce token usage."""
def __init__(
self,
summary_model: str = "claude-3-5-haiku-20241022",
keep_recent: int = 4,
max_summary_tokens: int = 500,
):
self.summary_model = summary_model
self.keep_recent = keep_recent
self.max_summary_tokens = max_summary_tokens
self.accumulated_summary: str = ""
def summarize(
self, messages: list[Message], budget: TokenBudget
) -> list[Message]:
"""Replace old history with a summary if over budget."""
total_tokens = sum(m.token_count for m in messages)
if total_tokens <= budget.history_budget_remaining:
return messages
# Split into recent (keep verbatim) and old (to summarize)
recent = messages[-self.keep_recent:]
old = messages[:-self.keep_recent]
if not old:
return messages
# Build text to summarize
old_text = "\n".join(
f"{m.role}: {m.content}" for m in old
)
# If we already have a summary, prepend it
if self.accumulated_summary:
old_text = f"Previous summary:\n{self.accumulated_summary}\n\nNew messages:\n{old_text}"
# Summarize using a cheap model
response = client.messages.create(
model=self.summary_model,
max_tokens=self.max_summary_tokens,
messages=[
{"role": "user", "content": f"{SUMMARIZE_PROMPT}\n\n{old_text}"}
],
)
new_summary = response.content[0].text
self.accumulated_summary = new_summary
summary_tokens = len(new_summary) // 4
print(f"[summarize] Compressed {sum(m.token_count for m in old)} tokens "
f"into ~{summary_tokens} tokens ({summary_tokens / max(sum(m.token_count for m in old), 1) * 100:.0f}% reduction)")
# Reconstruct: summary message + recent verbatim messages
summary_msg = Message(
role="system",
content=f"[Conversation summary - previous turns condensed]\n{new_summary}",
token_count=summary_tokens,
)
return [summary_msg] + recent
# Usage
summarizer = ConversationSummarizer(summary_model="claude-3-5-haiku-20241022")
budget = TokenBudget(model="claude-sonnet-4-20250514")
# After several iterations, when history is getting long:
messages = summarizer.summarize(messages, budget)
Strategy 4: Cost Optimization
Tracking and controlling spending is as important as managing context windows. Without cost visibility, a runaway loop can drain your API budget in minutes.
import time
from dataclasses import dataclass, field
@dataclass
class CostRecord:
iteration: int
input_tokens: int
output_tokens: int
input_cost: float
output_cost: float
total_cost: float
timestamp: float = field(default_factory=time.time)
class CostTracker:
"""Track per-iteration costs and enforce a budget cap."""
def __init__(
self,
model: str,
daily_budget_usd: float = 50.0,
run_budget_usd: float = 10.0,
):
self.model = model
self.daily_budget = daily_budget_usd
self.run_budget = run_budget_usd
self.pricing = MODEL_PRICING.get(model, {"input": 3.0, "output": 15.0})
self.records: list[CostRecord] = []
self.total_cost_usd: float = 0.0
self.iteration: int = 0
def record_usage(
self, input_tokens: int, output_tokens: int
) -> CostRecord:
"""Record token usage for the current iteration."""
input_cost = (input_tokens / 1_000_000) * self.pricing["input"]
output_cost = (output_tokens / 1_000_000) * self.pricing["output"]
total = input_cost + output_cost
self.iteration += 1
record = CostRecord(
iteration=self.iteration,
input_tokens=input_tokens,
output_tokens=output_tokens,
input_cost=round(input_cost, 6),
output_cost=round(output_cost, 6),
total_cost=round(total, 6),
)
self.records.append(record)
self.total_cost_usd += total
return record
def is_over_run_budget(self) -> bool:
return self.total_cost_usd >= self.run_budget
def check_and_enforce(self, input_tokens: int, output_tokens: int) -> bool:
"""Record usage and return True if we should stop (over budget)."""
record = self.record_usage(input_tokens, output_tokens)
print(
f"[cost] Iter {record.iteration}: "
f"{record.input_tokens:,} in + {record.output_tokens:,} out = "
f"${record.total_cost:.4f} | "
f"Running total: ${self.total_cost_usd:.4f} / "
f"${self.run_budget:.2f} budget"
)
if self.is_over_run_budget():
print(
f"[COST LIMIT] Run budget of ${self.run_budget:.2f} exceeded. "
f"Stopping loop."
)
return True
return False
def get_token_growth_rate(self) -> Optional[float]:
"""Calculate the token growth rate across iterations.
Returns the ratio of input tokens in the last iteration vs the first.
If the rate is >2.0, your history is growing too fast.
"""
if len(self.records) < 2:
return None
first = self.records[0].input_tokens
last = self.records[-1].input_tokens
return last / first if first > 0 else None
def report(self) -> str:
growth = self.get_token_growth_rate()
lines = [
f"Cost Report for {self.model}",
f"{'=' * 45}",
f"Iterations: {self.iteration}",
f"Total cost: ${self.total_cost_usd:.4f}",
f"Run budget: ${self.run_budget:.2f} "
f"({'EXCEEDED' if self.is_over_run_budget() else 'OK'})",
]
if growth is not None:
status = "OK" if growth < 2.0 else "WARNING"
lines.append(f"Token growth rate: {growth:.1f}x [{status}]")
return "\n".join(lines)
# Usage in a loop
tracker = CostTracker(model="claude-sonnet-4-20250514", run_budget_usd=5.0)
for iteration in range(1, 100):
# ... run your loop iteration ...
simulated_input_tokens = 5000 * iteration # Growing history
simulated_output_tokens = 500
should_stop = tracker.check_and_enforce(
simulated_input_tokens, simulated_output_tokens
)
if should_stop:
break
print(tracker.report())
Cost Optimization Tips
- Set a daily cost cap and enforce it before each iteration. Never rely on post-hoc monitoring.
- Use cheaper models for validation/summarization tasks. Claude Haiku or Gemini 2.0 Flash for summarization; reserve Sonnet/Opus for complex reasoning steps.
- Cache repeated prompts. If your loop sends the same system prompt or file content across iterations, use prompt caching (Anthropic and OpenAI both support this) to avoid re-processing.
- Monitor token growth rate. If input tokens are doubling each iteration, your history is growing unboundedly -- something is wrong with your loop design.
- Track costs per iteration with running totals. This makes cost anomalies immediately visible.
The Layered Model Strategy
In practice, this maps to tools like:
- Aider (
aider --model claude-3.5-sonnet) for code editing with automatic repo-map compression - Continue.dev (continue.dev) which lets you configure different models for different tasks in its VS Code/JetBrains plugin
- Codex CLI (github.com/openai/codex) which supports model selection per invocation
Strategy 5: Architecture-Level Optimization
Beyond per-iteration management, you can design your loop architecture to minimize token usage from the start.
Shorter System Prompts
Be specific, not verbose. A 500-token system prompt is often more effective than a 2000-token one that the model must re-read every iteration.
# Bad: 2000+ tokens of verbose instructions
SYSTEM_PROMPT_VERBOSE = """
You are an AI coding assistant that helps with software development tasks.
You should carefully analyze the code, think step by step, consider edge cases,
review best practices, check for security vulnerabilities, ensure test coverage,
follow the single responsibility principle, use meaningful variable names,
add docstrings, handle errors gracefully, log important events, ...
"""
# Good: 200 tokens, specific instructions
SYSTEM_PROMPT_CONCISE = """
You are a code fix agent. Given a file path and error description:
1. Read the relevant file
2. Identify the root cause
3. Apply the minimal fix
4. Report what changed and why
Return only the fix -- no explanations of your reasoning process.
"""
Pass Only Relevant Content
Do not dump entire codebases into context. Extract only the relevant files and functions:
def get_relevant_context(file_path: str, error_line: int, context_lines: int = 10) -> str:
"""Read only the relevant portion of a file around an error."""
with open(file_path) as f:
lines = f.readlines()
start = max(0, error_line - context_lines)
end = min(len(lines), error_line + context_lines)
return "".join(f"{i+1:4d}: {line}" for i, line in enumerate(lines[start:end], start))
Truncate Large Tool Results
Before feeding tool outputs back into the conversation, truncate them:
MAX_TOOL_OUTPUT_TOKENS = 2000
def truncate_tool_output(output: str, max_chars: int = MAX_TOOL_OUTPUT_TOKENS * 4) -> str:
if len(output) <= max_chars:
return output
return output[:max_chars] + f"\n\n[... truncated, {len(output)} total characters]"
Open Loop Pattern
Instead of accumulating everything in a single conversation, process in batches:
def run_open_loop(task: str, max_subtasks: int = 5):
"""Process a task in isolated sub-loops rather than one growing conversation."""
subtasks = decompose_task(task)
for i, subtask in enumerate(subtasks[:max_subtasks]):
# Each subtask gets a fresh conversation
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": subtask},
]
result = call_llm(messages)
# Only pass forward the result, not the full history
subtask_results.append(result)
# Final synthesis with just the subtask results
return synthesize_results(subtask_results)
Semantic Chunking
Instead of passing entire documents, chunk them by semantic relevance and include only what is needed:
def get_relevant_chunks(document: str, query: str, chunk_size: int = 500, max_chunks: int = 3) -> str:
"""Extract only the most relevant chunks from a document."""
chunks = [document[i:i+chunk_size] for i in range(0, len(document), chunk_size)]
# Score chunks by simple keyword overlap (in production, use embeddings)
query_terms = set(query.lower().split())
scored = [(sum(1 for t in query_terms if t in chunk.lower()), chunk) for chunk in chunks]
scored.sort(key=lambda x: x[0], reverse=True)
return "\n---\n".join(chunk for _, chunk in scored[:max_chunks])
Putting It All Together
Here is a combined loop wrapper that integrates token counting, conversation pruning, summarization, and cost tracking:
import anthropic
client = anthropic.Anthropic()
SYSTEM_PROMPT = "You are a code analysis agent. Analyze files and report issues."
def run_budget_aware_loop(
task: str,
model: str = "claude-sonnet-4-20250514",
max_iterations: int = 20,
run_budget_usd: float = 5.0,
prune_strategy: PruneStrategy = PruneStrategy.IMPORTANCE_BASED,
keep_recent: int = 4,
) -> str:
"""Run an agent loop with full token and cost management.
Integrates: token counting, conversation pruning, summarization,
and cost tracking in a single loop.
"""
budget = TokenBudget(model=model)
tracker = CostTracker(model=model, run_budget_usd=run_budget_usd)
pruner = ConversationPruner(budget)
summarizer = ConversationSummarizer(
summary_model="claude-3-5-haiku-20241022",
keep_recent=keep_recent,
)
messages: list[Message] = [
Message(role="user", content=task, important=True),
]
budget.add_tokens(budget.count_tokens(task))
for iteration in range(1, max_iterations + 1):
print(f"\n{'=' * 60}")
print(f"Iteration {iteration}/{max_iterations}")
print(budget.report())
# 1. Check if we need to summarize (before pruning)
messages = summarizer.summarize(messages, budget)
budget.total_input_tokens = sum(m.token_count for m in messages)
# 2. Check if we need to prune (after summarization)
if budget.is_over_budget():
messages = pruner.prune(messages, prune_strategy, keep_recent)
budget.total_input_tokens = sum(m.token_count for m in messages)
print(budget.report())
# 3. Call the LLM
try:
api_messages = [
{"role": "system", "content": SYSTEM_PROMPT}
] + [
{"role": m.role, "content": m.content} for m in messages
]
response = client.messages.create(
model=model,
max_tokens=budget.reserved_output_tokens,
messages=api_messages,
)
except anthropic.APIError as e:
print(f"[ERROR] API call failed: {e}")
# Emergency prune on error
messages = pruner.prune(messages, PruneStrategy.SLIDING_WINDOW, keep_recent=2)
budget.total_input_tokens = sum(m.token_count for m in messages)
continue
# 4. Extract response
assistant_text = response.content[0].text
# 5. Track costs
input_tokens = response.usage.input_tokens
output_tokens = response.usage.output_tokens
should_stop = tracker.check_and_enforce(input_tokens, output_tokens)
if should_stop:
break
# 6. Add response to history
assistant_msg = Message(
role="assistant",
content=assistant_text,
token_count=output_tokens,
)
messages.append(assistant_msg)
budget.add_tokens(output_tokens)
# 7. Check for task completion
if "<task_complete>" in assistant_text.lower():
print("[LOOP] Task marked as complete.")
break
# Final report
print(f"\n{tracker.report()}")
return messages[-1].content if messages else "No result"
# Run the loop
result = run_budget_aware_loop(
task="Analyze the auth module for security vulnerabilities",
model="claude-sonnet-4-20250514",
max_iterations=15,
run_budget_usd=3.0,
)
Token Monitoring Tools
These are the real tools available for monitoring token usage:
| Tool | Type | Provider | Notes |
|---|---|---|---|
| tiktoken | Python library | OpenAI | Fast BPE tokenizer. pip install tiktoken |
| LiteLLM | Python library | Multi-provider | Unified token counting across OpenAI, Anthropic, Gemini |
| tiktoken-cli | CLI tool | OpenAI | Count tokens in files from the command line |
| tiktoken MCP Server | MCP integration | OpenAI | Use within Claude Code, Cursor, VS Code for real-time counting |
Anthropic API response.usage | API field | Anthropic | Accurate per-request token counting (input + output) |
Quick tiktoken CLI Usage
# Install
pip install tiktoken-cli
# Count tokens in a file
tiktoken-cli count myfile.py
# Count a string
tiktoken-cli count --text "Hello, world"
Quick tiktoken MCP Server Setup
For use within Claude Code, Cursor, or Cline (github.com/cline/cline, VS Code plugin with MCP protocol support):
{
"mcpServers": {
"tiktoken": {
"command": "npx",
"args": ["-y", "@anthropic-ai/tiktoken-mcp"]
}
}
}
Cline in particular benefits from MCP-based token counting since it runs long autonomous sessions where context accumulation is a real risk.
Best Practices Summary
Next Steps
- Loop Timeout Solutions -- handling loops that take too long to complete, including timeout strategies and async patterns.
- State Persistence -- persisting loop state across iterations and restarts without losing progress.
- Production Loop Standards -- the full set of standards for running agent loops reliably in production environments.
- Custom Loop Script (Python) -- building a custom agent loop in Python with full control over token management and iteration logic.