Understanding AI Agent Production Costs in 2026
By 2026, AI agents have moved from experimental prototypes to production-grade systems handling millions of tasks daily. These agents—autonomous software entities that perceive, reason, and act—carry a price tag that extends far beyond simple API calls. Understanding the true cost means accounting for compute cycles, token consumption, tool invocations, memory retrieval, guardrail evaluations, and the cascading infrastructure that keeps agents reliable at scale.
What Constitutes the "True Cost"
The surface-level cost most developers track is the LLM provider's per-token billing. In 2026, this is table stakes. The true cost includes:
- LLM inference tokens — both input (context, system prompts, tool definitions, memory) and output (reasoning chains, tool calls, final responses)
- Embedding and vector search operations — every RAG retrieval or memory lookup consumes embedding compute and database I/O
- Tool execution costs — API calls to external services, code interpreters, browser automation, and database queries each carry their own billing meters
- Guardrail and safety evaluations — input/output content moderation, PII detection, and policy compliance checks often require additional model calls
- Observability infrastructure — tracing, logging, and metric storage for debugging agent loops at production throughput
- Orchestration overhead — the framework itself (LangGraph, custom state machines, event buses) consumes compute cycles and memory
- Retry and error recovery waste — failed tool calls, malformed outputs, and guardrail rejections trigger expensive re-executions
Why It Matters
A single agent conversation can easily consume 50,000+ tokens across multiple LLM calls, tool round-trips, and memory lookups. At $3–$15 per million tokens (depending on the model tier), a seemingly simple task can cost $0.15–$0.75. Multiply this by millions of monthly interactions, and the difference between a well-optimized agent and a naive implementation can mean hundreds of thousands of dollars per month. Beyond raw cost, inefficient agents have higher latency, worse user experience, and increased carbon footprint—all of which translate to business impact in 2026's competitive landscape.
Cost Modeling: The Four-Layer Framework
To systematically track costs, model your agent stack in four layers. Here is a practical cost calculator implemented as a Python dataclass that aggregates costs across all layers:
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
from enum import Enum
class ModelTier(Enum):
FAST = "fast" # e.g., GPT-6o-mini, Claude 4 Haiku
BALANCED = "balanced" # e.g., GPT-6o, Claude 4 Sonnet
REASONING = "reasoning" # e.g., o4, Claude 4 Opus
# 2026 approximate pricing per 1M tokens (input/output blended average)
MODEL_PRICING = {
ModelTier.FAST: 0.25, # $0.25/1M tokens
ModelTier.BALANCED: 2.50, # $2.50/1M tokens
ModelTier.REASONING: 12.00, # $12.00/1M tokens
}
TOOL_PRICING = {
"web_search": 0.005, # $0.005 per call
"code_execution": 0.02, # $0.02 per execution
"database_query": 0.001, # $0.001 per query
"browser_action": 0.03, # $0.03 per browser step
}
@dataclass
class AgentCostBreakdown:
"""Tracks cost across all four layers of an agent run."""
trace_id: str
timestamp: datetime = field(default_factory=datetime.now)
# Layer 1: LLM inference
llm_calls: List[Dict] = field(default_factory=list)
total_llm_tokens: int = 0
total_llm_cost: float = 0.0
# Layer 2: Embedding & retrieval
embedding_operations: int = 0
embedding_cost: float = 0.0
vector_search_operations: int = 0
vector_search_cost: float = 0.0
# Layer 3: Tool execution
tool_calls: List[Dict] = field(default_factory=list)
total_tool_cost: float = 0.0
# Layer 4: Guardrails & safety
guardrail_evaluations: int = 0
guardrail_cost: float = 0.0
# Overhead
retry_count: int = 0
retry_wasted_cost: float = 0.0
def record_llm_call(self, tier: ModelTier, input_tokens: int, output_tokens: int):
cost = ((input_tokens + output_tokens) / 1_000_000) * MODEL_PRICING[tier]
self.llm_calls.append({
"tier": tier.value,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": round(cost, 6),
})
self.total_llm_tokens += input_tokens + output_tokens
self.total_llm_cost += cost
def record_tool_call(self, tool_name: str, success: bool):
price = TOOL_PRICING.get(tool_name, 0.01)
self.tool_calls.append({"tool": tool_name, "success": success, "cost": price})
self.total_tool_cost += price
if not success:
self.retry_count += 1
def record_retry_waste(self, wasted_cost: float):
self.retry_wasted_cost += wasted_cost
def record_guardrail_eval(self, cost: float = 0.001):
self.guardrail_evaluations += 1
self.guardrail_cost += cost
def record_embedding(self, tokens: int):
# Embedding cost ~$0.02 per 1M tokens
cost = (tokens / 1_000_000) * 0.02
self.embedding_operations += 1
self.embedding_cost += cost
def record_vector_search(self):
# Approximate I/O cost per search
self.vector_search_operations += 1
self.vector_search_cost += 0.0001
def grand_total(self) -> float:
return (
self.total_llm_cost +
self.embedding_cost +
self.vector_search_cost +
self.total_tool_cost +
self.guardrail_cost +
self.retry_wasted_cost
)
def summary(self) -> Dict:
return {
"trace_id": self.trace_id,
"grand_total": round(self.grand_total(), 6),
"llm_cost": round(self.total_llm_cost, 6),
"llm_tokens": self.total_llm_tokens,
"tool_cost": round(self.total_tool_cost, 6),
"retrieval_cost": round(self.embedding_cost + self.vector_search_cost, 6),
"guardrail_cost": round(self.guardrail_cost, 6),
"retry_waste": round(self.retry_wasted_cost, 6),
"retry_count": self.retry_count,
}Building a Cost-Aware Agent Loop
The key to controlling costs is making the agent itself cost-aware. Instead of blindly looping through reasoning steps, implement a token budget that the agent must respect. Here is a production-grade agent loop with built-in cost limits:
import asyncio
import time
from typing import Callable, Awaitable
class TokenBudgetExhausted(Exception):
"""Raised when the agent exceeds its allocated token budget."""
pass
class CostAwareAgent:
"""
An agent that tracks and enforces cost budgets across its execution loop.
Uses a hierarchical budget: total run budget, per-step budget, and tool budget.
"""
def __init__(
self,
total_token_budget: int = 100_000,
per_step_token_limit: int = 20_000,
tool_cost_budget: float = 0.50,
model_tier: ModelTier = ModelTier.BALANCED,
):
self.total_token_budget = total_token_budget
self.per_step_token_limit = per_step_token_limit
self.tool_cost_budget = tool_cost_budget
self.model_tier = model_tier
self.tokens_consumed = 0
self.tool_cost_consumed = 0.0
self.step_count = 0
self.max_steps = 15 # Safety valve
async def run(
self,
task: str,
tools: list,
llm_call: Callable[[str, list], Awaitable[dict]],
) -> dict:
"""
Executes the agent loop with cost enforcement.
Args:
task: The user's objective
tools: Available tool definitions
llm_call: Async function that takes (system_prompt, messages) and returns
a dict with 'content', 'tool_calls', and 'token_counts'
"""
cost_breakdown = AgentCostBreakdown(trace_id=f"run_{int(time.time())}")
messages = [{"role": "user", "content": task}]
final_response = None
system_prompt = self._build_cost_aware_system_prompt()
while self.step_count < self.max_steps:
# Check total budget before each step
if self.tokens_consumed >= self.total_token_budget:
raise TokenBudgetExhausted(
f"Total token budget exhausted: {self.tokens_consumed}/{self.total_token_budget}"
)
self.step_count += 1
# Pre-call: estimate input token count and enforce per-step limit
estimated_input = self._estimate_tokens(str(messages) + system_prompt)
if estimated_input > self.per_step_token_limit:
# Compress context: drop oldest tool results, summarize
messages = self._compress_context(messages)
estimated_input = self._estimate_tokens(str(messages) + system_prompt)
# Execute LLM call
full_messages = [{"role": "system", "content": system_prompt}] + messages
response = await llm_call(system_prompt, messages)
# Record costs
input_tokens = response.get("input_tokens", estimated_input)
output_tokens = response.get("output_tokens", 0)
self.tokens_consumed += input_tokens + output_tokens
cost_breakdown.record_llm_call(
self.model_tier, input_tokens, output_tokens
)
# Process tool calls if present
tool_calls = response.get("tool_calls", [])
if tool_calls:
assistant_msg = {
"role": "assistant",
"content": response.get("content", ""),
"tool_calls": tool_calls,
}
messages.append(assistant_msg)
for tc in tool_calls:
tool_name = tc.get("name", "unknown")
# Check tool budget
tool_price = TOOL_PRICING.get(tool_name, 0.01)
if self.tool_cost_consumed + tool_price > self.tool_cost_budget:
# Skip expensive tool, ask LLM to find alternative
messages.append({
"role": "tool",
"tool_call_id": tc.get("id", ""),
"content": f"Tool '{tool_name}' skipped: tool budget limit reached.",
})
cost_breakdown.record_tool_call(tool_name, success=False)
continue
self.tool_cost_consumed += tool_price
# Execute tool (wrapped with timeout and error handling)
try:
result = await asyncio.wait_for(
self._execute_tool(tool_name, tc.get("arguments", {})),
timeout=30.0,
)
success = True
except Exception as e:
result = f"Tool execution failed: {str(e)}"
success = False
cost_breakdown.record_retry_waste(
(input_tokens + output_tokens) / 1_000_000 * MODEL_PRICING[self.model_tier]
)
cost_breakdown.record_tool_call(tool_name, success=success)
messages.append({
"role": "tool",
"tool_call_id": tc.get("id", ""),
"content": str(result)[:8000], # Truncate to control next input size
})
else:
# No tool calls — agent is done
final_response = response.get("content", "")
messages.append({"role": "assistant", "content": final_response})
break
if final_response is None:
final_response = "Agent reached maximum steps without a final response."
return {
"response": final_response,
"cost_summary": cost_breakdown.summary(),
"steps": self.step_count,
"tokens_consumed": self.tokens_consumed,
"tool_cost_consumed": round(self.tool_cost_consumed, 4),
}
def _build_cost_aware_system_prompt(self) -> str:
return f"""You are a cost-efficient AI agent. Your token budget for this entire run is {self.total_token_budget:,} tokens.
Current consumption: {self.tokens_consumed:,} tokens used ({self.tokens_consumed/self.total_token_budget*100:.1f}%).
Tool budget remaining: ${self.tool_cost_budget - self.tool_cost_consumed:.3f}.
Guidelines:
- Be concise. Every token costs money and latency.
- Prefer completing the task in as few steps as possible.
- If a tool is expensive (browser_action, code_execution), consider whether a simpler approach works.
- If you're near budget limits, deliver your best partial answer rather than continuing indefinitely.
- Do not repeat tool calls that have already returned sufficient information."""
def _estimate_tokens(self, text: str) -> int:
"""Rough token estimator: ~4 chars per token for English text."""
return len(text) // 4
def _compress_context(self, messages: list) -> list:
"""Compress older messages to stay within per-step token limits."""
if len(messages) <= 2:
return messages
# Keep system-like context and last 4 messages, summarize middle
preserved = messages[:1] + [{
"role": "system",
"content": "[Compressed context: earlier steps summarized. Proceed with latest information.]"
}] + messages[-4:]
return preserved
async def _execute_tool(self, tool_name: str, arguments: dict) -> str:
"""Stub for actual tool execution — replace with your real implementations."""
# In production, this would call your tool executor, API gateway, or sandbox
await asyncio.sleep(0.05) # Simulate I/O
return f"Result from {tool_name} with args {arguments}"Real-Time Cost Observability
Tracking costs after the fact isn't enough. In 2026, production agents require real-time cost dashboards that alert when spending anomalies occur. Here's how to wire the cost breakdown into an observability pipeline using OpenTelemetry-style instrumentation:
import threading
import time
from collections import defaultdict
from datetime import datetime, timedelta
class CostMonitor:
"""
Thread-safe real-time cost aggregator for production agent fleets.
Tracks costs by agent type, model tier, customer, and time window.
"""
def __init__(self, alert_threshold_per_minute: float = 5.0):
self._lock = threading.Lock()
self.alert_threshold = alert_threshold_per_minute
self.alerts = []
# Rolling windows for cost aggregation
self.minute_buckets: defaultdict = defaultdict(float)
self.hour_buckets: defaultdict = defaultdict(float)
self.daily_buckets: defaultdict = defaultdict(float)
# Per-dimension breakdowns
self.by_agent_type: defaultdict = defaultdict(lambda: defaultdict(float))
self.by_model_tier: defaultdict = defaultdict(lambda: defaultdict(float))
self.by_customer: defaultdict = defaultdict(lambda: defaultdict(float))
self.total_costs: defaultdict = defaultdict(float)
def record(self, cost_breakdown: AgentCostBreakdown, agent_type: str, customer_id: str):
"""Ingest a completed agent run's cost breakdown."""
now = datetime.now()
total = cost_breakdown.grand_total()
with self._lock:
# Time-windowed buckets
minute_key = now.strftime("%Y-%m-%dT%H:%M")
hour_key = now.strftime("%Y-%m-%dT%H")
day_key = now.strftime("%Y-%m-%d")
self.minute_buckets[minute_key] += total
self.hour_buckets[hour_key] += total
self.daily_buckets[day_key] += total
# Dimensional breakdowns
self.by_agent_type[agent_type]["total"] += total
self.by_agent_type[agent_type]["tokens"] += cost_breakdown.total_llm_tokens
self.by_agent_type[agent_type]["tool_cost"] += cost_breakdown.total_tool_cost
self.by_agent_type[agent_type]["retry_waste"] += cost_breakdown.retry_wasted_cost
for call in cost_breakdown.llm_calls:
self.by_model_tier[call["tier"]]["cost"] += call["cost"]
self.by_model_tier[call["tier"]]["calls"] += 1
self.by_customer[customer_id]["total"] += total
self.by_customer[customer_id]["runs"] += 1
self.total_costs["all_time"] += total
# Check alert thresholds
recent_minutes = sum(
v for k, v in self.minute_buckets.items()
if datetime.strptime(k, "%Y-%m-%dT%H:%M") > now - timedelta(minutes=1)
)
if recent_minutes > self.alert_threshold:
self.alerts.append({
"timestamp": now.isoformat(),
"message": f"Cost spike: ${recent_minutes:.2f} in the last minute (threshold: ${self.alert_threshold})",
"breakdown_by_agent": dict(self.by_agent_type),
})
def get_cost_report(self) -> dict:
"""Generate a comprehensive cost report."""
with self._lock:
return {
"all_time_total": round(self.total_costs["all_time"], 4),
"current_minute_spend": round(
sum(v for k, v in self.minute_buckets.items()
if datetime.strptime(k, "%Y-%m-%dT%H:%M") > datetime.now() - timedelta(minutes=1)),
4,
),
"current_hour_spend": round(
sum(v for k, v in self.hour_buckets.items()
if datetime.strptime(k, "%Y-%m-%dT%H") > datetime.now() - timedelta(hours=1)),
4,
),
"today_spend": round(
sum(v for k, v in self.daily_buckets.items()
if k == datetime.now().strftime("%Y-%m-%d")),
4,
),
"by_agent_type": {
k: {mk: round(mv, 4) if isinstance(mv, float) else mv
for mk, mv in v.items()}
for k, v in self.by_agent_type.items()
},
"by_model_tier": {
k: {mk: round(mv, 4) if isinstance(mv, float) else mv
for mk, mv in v.items()}
for k, v in self.by_model_tier.items()
},
"active_alerts": self.alerts[-5:],
}
def project_monthly_cost(self) -> float:
"""Linear projection based on today's spend rate."""
today_spend = sum(
v for k, v in self.daily_buckets.items()
if k == datetime.now().strftime("%Y-%m-%d")
)
# Estimate based on current hour if early in the day
hour_spend = sum(
v for k, v in self.hour_buckets.items()
if datetime.strptime(k, "%Y-%m-%dT%H") > datetime.now() - timedelta(hours=1)
)
current_hour = datetime.now().hour
if current_hour < 6:
# Too early to project from today, use hourly rate
return hour_spend * 24 * 30
else:
hours_elapsed = current_hour
daily_rate = today_spend / max(hours_elapsed, 1) * 24
return daily_rate * 30Strategic Cost Optimization Patterns
The most impactful cost reductions come from architectural decisions, not just smaller prompts. Here are the patterns that matter most in 2026:
Pattern 1: Tiered Model Routing
Not every agent step requires a reasoning model. Implement a router that classifies the complexity of each step and delegates to the cheapest capable model:
class ModelRouter:
"""
Routes agent steps to the cheapest model tier capable of handling them.
Uses a fast classifier (itself a cheap model) to make routing decisions.
"""
def __init__(self):
self.step_patterns = {
"tool_selection": ModelTier.FAST, # Choosing which tool to call
"parameter_extraction": ModelTier.FAST, # Extracting tool arguments
"result_summarization": ModelTier.FAST, # Summarizing tool results
"complex_reasoning": ModelTier.BALANCED, # Multi-step planning
"error_recovery": ModelTier.BALANCED, # Diagnosing failures
"final_synthesis": ModelTier.BALANCED, # Crafting final response
"ambiguous_analysis": ModelTier.REASONING, # High-stakes decisions
}
self.default_tier = ModelTier.BALANCED
async def classify_step(self, messages: list, available_tools: list) -> ModelTier:
"""
Uses a fast, cheap model to classify what kind of step this is.
Returns the appropriate model tier.
"""
classification_prompt = f"""Classify the next agent step based on the conversation.
Options: tool_selection, parameter_extraction, result_summarization, complex_reasoning,
error_recovery, final_synthesis, ambiguous_analysis.
Conversation (last 3 messages):
{messages[-3:] if len(messages) > 3 else messages}
Available tools: {[t.get('name') for t in available_tools]}
Respond with ONLY the classification label."""
# This classification call itself should be fast and cheap
# In production, use a tiny model or even regex heuristics
classification = await self._fast_classify(classification_prompt)
return self.step_patterns.get(
classification.strip().lower(), self.default_tier
)
async def _fast_classify(self, prompt: str) -> str:
"""Stub for a fast classification call (e.g., GPT-6o-mini with 1 token output)."""
# In production: call a minimal model with max_tokens=5
return "tool_selection"Pattern 2: Context Window Economics
In 2026, models support 1M+ token context windows, but filling them is expensive. Every token in the context window costs compute and money. Implement aggressive context pruning:
class ContextManager:
"""
Manages agent context to minimize token waste.
Implements sliding window with summarization, tool result compression,
and dead information removal.
"""
def __init__(self, max_context_tokens: int = 40_000, summary_trigger_ratio: float = 0.7):
self.max_context_tokens = max_context_tokens
self.summary_trigger = int(max_context_tokens * summary_trigger_ratio)
self.conversation_summary = ""
def optimize_context(self, messages: list) -> list:
"""
Returns an optimized message list that fits within the token budget.
Preserves semantic value while dropping redundant information.
"""
estimated_tokens = sum(len(str(m)) // 4 for m in messages)
if estimated_tokens <= self.max_context_tokens:
return messages
# Strategy: keep system messages, last N messages, and a rolling summary
system_messages = [m for m in messages if m.get("role") == "system"]
non_system = [m for m in messages if m.get("role") != "system"]
# Always keep the most recent user message and last assistant response
recent_messages = []
for m in reversed(non_system):
recent_messages.insert(0, m)
if len(recent_messages) >= 6:
break
# Summarize older messages if they contain meaningful content
older_messages = non_system[:-len(recent_messages)] if len(recent_messages) < len(non_system) else []
if older_messages:
self._update_summary(older_messages)
optimized = system_messages + [{
"role": "system",
"content": f"[Prior conversation summary: {self.conversation_summary}]"
}] + recent_messages
# Final safety trim
final_tokens = sum(len(str(m)) // 4 for m in optimized)
if final_tokens > self.max_context_tokens:
# Hard truncate: drop oldest non-system messages
while sum(len(str(m)) // 4 for m in optimized) > self.max_context_tokens and len(optimized) > 3:
# Remove the summary first if it's large, then oldest messages
for i, m in enumerate(optimized):
if m.get("role") == "system" and "Prior conversation summary" in str(m.get("content", "")):
optimized[i]["content"] = "[Summary truncated for space]"
break
return optimized
def _update_summary(self, messages: list):
"""Update the rolling summary with key facts from older messages."""
# In production, this would call a cheap model to extract key facts
extracted_facts = []
for m in messages:
content = str(m.get("content", ""))
if len(content) > 100:
extracted_facts.append(content[:200] + "...")
self.conversation_summary = "; ".join(extracted_facts[-5:])[:1000]Pattern 3: Batching and Caching
Many agent operations are repetitive across runs. Caching embeddings, tool results, and even LLM responses for identical queries can yield 30–60% cost reduction:
import hashlib
import json
from typing import Optional
class AgentCache:
"""
Multi-level cache for agent operations.
Caches: embeddings, tool results, LLM responses for deterministic prompts.
"""
def __init__(self, max_size: int = 10_000):
self.embedding_cache: dict = {} # text hash -> embedding vector
self.tool_result_cache: dict = {} # (tool_name, args_hash) -> result
self.prompt_cache: dict = {} # prompt_hash -> response (for temperature=0 calls)
self.max_size = max_size
self.hit_count = 0
self.miss_count = 0
def _hash(self, *args) -> str:
"""Create a deterministic hash for cache keys."""
serialized = json.dumps(args, sort_keys=True, default=str)
return hashlib.sha256(serialized.encode()).hexdigest()[:16]
def get_embedding(self, text: str) -> Optional[list]:
key = self._hash(text)
result = self.embedding_cache.get(key)
if result:
self.hit_count += 1
else:
self.miss_count += 1
return result
def set_embedding(self, text: str, vector: list):
if len(self.embedding_cache) >= self.max_size:
# Evict oldest entry (simple FIFO for illustration)
oldest_key = next(iter(self.embedding_cache))
del self.embedding_cache[oldest_key]
self.embedding_cache[self._hash(text)] = vector
def get_tool_result(self, tool_name: str, arguments: dict) -> Optional[str]:
key = self._hash(tool_name, arguments)
result = self.tool_result_cache.get(key)
if result:
self.hit_count += 1
else:
self.miss_count += 1
return result
def set_tool_result(self, tool_name: str, arguments: dict, result: str):
if len(self.tool_result_cache) >= self.max_size:
oldest_key = next(iter(self.tool_result_cache))
del self.tool_result_cache[oldest_key]
self.tool_result_cache[self._hash(tool_name, arguments)] = result
def get_prompt_response(self, system_prompt: str, messages: list, temperature: float) -> Optional[str]:
# Only cache temperature=0 calls (deterministic)
if temperature > 0:
return None
key = self._hash(system_prompt, [str(m) for m in messages[-3:]])
result = self.prompt_cache.get(key)
if result:
self.hit_count += 1
else:
self.miss_count += 1
return result
def set_prompt_response(self, system_prompt: str, messages: list, temperature: float, response: str):
if temperature > 0:
return
key = self._hash(system_prompt, [str(m) for m in messages[-3:]])
if len(self.prompt_cache) >= self.max_size:
oldest_key = next(iter(self.prompt_cache))
del self.prompt_cache[oldest_key]
self.prompt_cache[key] = response
def stats(self) -> dict:
total = self.hit_count + self.miss_count
return {
"hit_rate": round(self.hit_count / max(total, 1), 3),
"hits": self.hit_count,
"misses": self.miss_count,
"estimated_savings": f"${self.hit_count * 0.002:.2f}", # rough estimate
}Best Practices for Production Cost Control
- Set hard budgets, not soft limits. Implement circuit breakers that terminate agent loops when costs exceed thresholds. A runaway agent looping through tool calls can burn thousands of dollars before anyone notices.
- Use the cheapest model that works for each step. In 2026, the cost gap between tiers is enormous—$0.25 vs $12.00 per million tokens. Route 80%+ of your steps to fast models and reserve reasoning models for genuinely hard problems.
- Measure retry waste explicitly. Failed tool calls and guardrail rejections aren't just frustrating—they're expensive. Track retry cost as a first-class metric and optimize your tool implementations to succeed on the first attempt.
- Cache aggressively at every layer. Embeddings, tool results for idempotent operations, and deterministic LLM responses all benefit from caching. A 40% cache hit rate can translate to tens of thousands in monthly savings.
- Implement cost attribution. Tag every agent run with customer ID, agent type, and model tier. Without attribution, cost spikes are impossible to diagnose. With it, you can implement chargebacks and identify expensive customers or use cases.
- Monitor in real time with alerts. Cost anomalies happen fast—a buggy tool integration can double your spend in minutes. Per-minute cost monitoring with automated alerts is essential.
- Compress context relentlessly. Long context windows are a trap. Every extra token in the prompt increases both cost and latency. Use rolling summaries, drop stale tool results, and trim verbose outputs before feeding them back into the LLM.
- Batch where possible, but don't over-batch. Combining multiple operations into single LLM calls reduces round-trips and total tokens, but overly large prompts can degrade output quality. Find the sweet spot—typically 3–5 operations per batch.
Putting It All Together: The Cost-Optimized Agent Factory
Here is a factory function that wires all the components together into a production-ready, cost-optimized agent:
async def create_cost_optimized_agent(
task: str,
customer_id: str,
agent_type: str = "general_assistant",
total_token_budget: int = 80_000,
tool_cost_budget: float = 0.40,
) -> dict:
"""
Factory that assembles a fully instrumented, cost-optimized agent run.
Returns both the agent's response and a detailed cost report.
"""
# Initialize all components
router = ModelRouter()
context_manager = ContextManager(max_context_tokens=40_000)
cache = AgentCache()
monitor = CostMonitor(alert_threshold_per_minute=5.0)
cost_breakdown = AgentCostBreakdown(
trace_id=f"{agent_type}_{customer_id}_{int(time.time())}"
)
# Define available tools (in production, this would be a registry)
tools = [
{"name": "web_search", "description": "Search the web for information"},
{"name": "code_execution", "description": "Execute Python code in sandbox"},
{"name": "database_query", "description": "Query internal database"},
]
messages = [{"role": "user", "content": task}]
tokens_consumed = 0
tool_cost_consumed = 0.0
step_count = 0
max