Understanding AI Agent Resource Quotas
AI agents—autonomous software entities that can plan, reason, and execute multi-step tasks—are becoming increasingly powerful. They chain API calls, browse the web, query databases, and even write and execute code. Without guardrails, a single agent session can spiral into thousands of expensive LLM calls, exhaust your API budget, or trigger cascading failures across dependent services. Resource quotas are the programmatic boundaries you impose on an agent's consumption of compute, time, tokens, and money during a single session or task.
Think of resource quotas as the financial controls and circuit breakers for your autonomous systems. Just as cloud platforms enforce budget alerts and spending limits, your agent framework needs equivalent mechanisms baked directly into the orchestration layer.
Why Resource Quotas Matter
The cost implications of unconstrained agents are staggering. Consider a seemingly harmless prompt like "Research the latest developments in quantum computing and summarize the top five papers." An agent without quotas might:
- Execute 40+ sequential LLM calls as it searches, reads, summarizes, and fact-checks
- Process millions of input tokens across retrieved documents
- Run for 20 minutes while the user waits for a response
- Consume $15–$50 in API costs for a single query
- Trigger recursive loops when the agent's planning step keeps refining its approach
Resource quotas transform this unbounded risk into a predictable, controlled execution model. They protect your infrastructure budget, prevent one noisy agent from starving other services of resources, and give you the confidence to deploy agents in production environments where costs directly impact business margins.
Core Resource Dimensions to Control
Effective quota systems operate across multiple dimensions simultaneously. Here are the primary axes you need to monitor and limit:
1. Token Budgets
Track cumulative input and output tokens across all LLM calls within an agent session. This is the most direct proxy for API cost. Set both a soft warning threshold and a hard cutoff.
2. Monetary Cost Caps
Map token consumption to actual dollar amounts based on your model pricing tiers. A GPT-4 Turbo session costs roughly $0.01 per 1K input tokens and $0.03 per 1K output tokens. Calculate running costs in real time and halt execution when the budget is exhausted.
3. Step and Iteration Limits
Cap the total number of reasoning steps, tool calls, or loop iterations. Many agent architectures use a while-loop pattern (plan → act → observe → replan) that can theoretically run forever if the termination condition is never satisfied.
4. Time-to-Live (TTL) Timeouts
Enforce a wall-clock deadline. Even if token and step limits haven't been reached, a session that runs too long creates poor user experience and ties up server resources.
5. Concurrency and Rate Limits
At the system level, limit how many agent sessions run simultaneously and how frequently they can invoke downstream APIs. This prevents a fleet of agents from overwhelming your infrastructure.
Implementing a Resource Quota Manager
Let's build a complete quota management system in Python that you can integrate into any agent framework. We'll create a composable quota tracker that monitors tokens, cost, steps, and time, then wire it into a simplified agent loop.
The QuotaTracker Class
This class is the heart of the system. It maintains counters for every resource dimension and exposes a single method to check whether execution can continue. It's designed to be thread-safe for concurrent agent sessions and serializable for logging and debugging.
import time
import threading
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Dict, Any
import json
class QuotaStatus(Enum):
OK = "ok"
WARNING = "warning"
EXHAUSTED = "exhausted"
@dataclass
class QuotaConfig:
max_input_tokens: int = 100_000
max_output_tokens: int = 50_000
max_total_tokens: int = 150_000
max_cost_dollars: float = 5.0
max_steps: int = 25
max_wall_clock_seconds: int = 300
warning_threshold_ratio: float = 0.75
# Pricing per 1K tokens (customize per model)
input_cost_per_1k: float = 0.01 # GPT-4 Turbo input pricing
output_cost_per_1k: float = 0.03 # GPT-4 Turbo output pricing
@dataclass
class QuotaSnapshot:
input_tokens_used: int = 0
output_tokens_used: int = 0
total_tokens_used: int = 0
cost_accrued_dollars: float = 0.0
steps_completed: int = 0
start_time: float = field(default_factory=time.time)
status: QuotaStatus = QuotaStatus.OK
exhausted_reason: Optional[str] = None
class QuotaTracker:
"""
Tracks resource consumption for a single agent session.
Thread-safe for use in concurrent agent environments.
"""
def __init__(self, config: QuotaConfig):
self.config = config
self.snapshot = QuotaSnapshot()
self._lock = threading.Lock()
self._warning_issued = False
def record_llm_call(self, input_tokens: int, output_tokens: int) -> QuotaStatus:
"""
Update quotas after an LLM completion. Returns the current status
so the caller can decide whether to continue or abort.
"""
with self._lock:
self.snapshot.input_tokens_used += input_tokens
self.snapshot.output_tokens_used += output_tokens
self.snapshot.total_tokens_used += (input_tokens + output_tokens)
# Calculate cost
input_cost = (input_tokens / 1000.0) * self.config.input_cost_per_1k
output_cost = (output_tokens / 1000.0) * self.config.output_cost_per_1k
self.snapshot.cost_accrued_dollars += input_cost + output_cost
return self._evaluate_limits()
def record_step(self) -> QuotaStatus:
"""Increment the step counter after each reasoning/action cycle."""
with self._lock:
self.snapshot.steps_completed += 1
return self._evaluate_limits()
def check_time_only(self) -> QuotaStatus:
"""Check if the wall-clock deadline has been exceeded."""
with self._lock:
return self._evaluate_limits()
def current_snapshot(self) -> QuotaSnapshot:
"""Return a thread-safe copy of the current quota state."""
with self._lock:
# Create a copy to avoid leaking the internal mutable object
return QuotaSnapshot(
input_tokens_used=self.snapshot.input_tokens_used,
output_tokens_used=self.snapshot.output_tokens_used,
total_tokens_used=self.snapshot.total_tokens_used,
cost_accrued_dollars=self.snapshot.cost_accrued_dollars,
steps_completed=self.snapshot.steps_completed,
start_time=self.snapshot.start_time,
status=self.snapshot.status,
exhausted_reason=self.snapshot.exhausted_reason,
)
def can_proceed(self) -> bool:
"""Convenience method: returns True if the session is still within limits."""
status = self._evaluate_limits()
return status != QuotaStatus.EXHAUSTED
def _evaluate_limits(self) -> QuotaStatus:
"""Internal method that checks all limits and sets status accordingly.
Callers must hold self._lock before invoking this method."""
snap = self.snapshot
# Check wall-clock time
elapsed = time.time() - snap.start_time
if elapsed >= self.config.max_wall_clock_seconds:
snap.status = QuotaStatus.EXHAUSTED
snap.exhausted_reason = f"Wall-clock time exceeded: {elapsed:.1f}s >= {self.config.max_wall_clock_seconds}s"
return snap.status
# Check total tokens
if snap.total_tokens_used >= self.config.max_total_tokens:
snap.status = QuotaStatus.EXHAUSTED
snap.exhausted_reason = f"Total token limit reached: {snap.total_tokens_used} >= {self.config.max_total_tokens}"
return snap.status
# Check input tokens
if snap.input_tokens_used >= self.config.max_input_tokens:
snap.status = QuotaStatus.EXHAUSTED
snap.exhausted_reason = f"Input token limit reached: {snap.input_tokens_used} >= {self.config.max_input_tokens}"
return snap.status
# Check output tokens
if snap.output_tokens_used >= self.config.max_output_tokens:
snap.status = QuotaStatus.EXHAUSTED
snap.exhausted_reason = f"Output token limit reached: {snap.output_tokens_used} >= {self.config.max_output_tokens}"
return snap.status
# Check cost
if snap.cost_accrued_dollars >= self.config.max_cost_dollars:
snap.status = QuotaStatus.EXHAUSTED
snap.exhausted_reason = f"Cost limit reached: ${snap.cost_accrued_dollars:.4f} >= ${self.config.max_cost_dollars:.2f}"
return snap.status
# Check steps
if snap.steps_completed >= self.config.max_steps:
snap.status = QuotaStatus.EXHAUSTED
snap.exhausted_reason = f"Step limit reached: {snap.steps_completed} >= {self.config.max_steps}"
return snap.status
# Issue warnings when approaching limits
warning_ratio = self.config.warning_threshold_ratio
near_limit = (
snap.total_tokens_used >= self.config.max_total_tokens * warning_ratio or
snap.cost_accrued_dollars >= self.config.max_cost_dollars * warning_ratio or
snap.steps_completed >= self.config.max_steps * warning_ratio
)
if near_limit and not self._warning_issued:
self._warning_issued = True
snap.status = QuotaStatus.WARNING
return snap.status
snap.status = QuotaStatus.OK
return snap.status
def generate_report(self) -> str:
"""Produce a human-readable summary of resource consumption."""
snap = self.current_snapshot()
elapsed = time.time() - snap.start_time
lines = [
"═══════════════════════════════════════",
" AGENT RESOURCE QUOTA REPORT ",
"═══════════════════════════════════════",
f" Status: {snap.status.value.upper()}",
f" Wall-clock time: {elapsed:.1f}s / {self.config.max_wall_clock_seconds}s",
f" Input tokens: {snap.input_tokens_used:,} / {self.config.max_input_tokens:,}",
f" Output tokens: {snap.output_tokens_used:,} / {self.config.max_output_tokens:,}",
f" Total tokens: {snap.total_tokens_used:,} / {self.config.max_total_tokens:,}",
f" Cost accrued: ${snap.cost_accrued_dollars:.4f} / ${self.config.max_cost_dollars:.2f}",
f" Steps taken: {snap.steps_completed} / {self.config.max_steps}",
]
if snap.exhausted_reason:
lines.append(f" Exhausted: {snap.exhausted_reason}")
lines.append("═══════════════════════════════════════")
return "\n".join(lines)Integrating the Quota Tracker into an Agent Loop
Now we'll build a simplified agent orchestrator that demonstrates exactly where quota checks belong in a realistic agent execution loop. The pattern follows the standard ReAct (Reasoning + Acting) architecture that underpins LangChain, AutoGPT, and similar frameworks.
from typing import List, Dict, Any, Optional
import logging
logger = logging.getLogger("agent_orchestrator")
class QuotaExhaustedError(Exception):
"""Raised when the agent has exceeded its resource quota."""
def __init__(self, reason: str, snapshot: QuotaSnapshot):
self.reason = reason
self.snapshot = snapshot
super().__init__(f"Quota exhausted: {reason}")
class AgentOrchestrator:
"""
A simplified agent orchestrator that demonstrates quota-aware execution.
In production, this would integrate with actual LLM clients, tool executors,
and memory systems.
"""
def __init__(
self,
quota_config: QuotaConfig,
system_prompt: str,
tools: List[Dict[str, Any]],
llm_call_function, # Callable that takes messages and returns a response
):
self.tracker = QuotaTracker(quota_config)
self.system_prompt = system_prompt
self.tools = tools
self.llm_call = llm_call_function
self.messages: List[Dict[str, str]] = [
{"role": "system", "content": system_prompt}
]
self.final_answer: Optional[str] = None
def run(self, user_query: str) -> Dict[str, Any]:
"""
Execute the agent loop with quota enforcement at every stage.
Returns a structured result including the answer and quota report.
"""
self.messages.append({"role": "user", "content": user_query})
# Pre-flight quota check
if not self.tracker.can_proceed():
raise QuotaExhaustedError(
"Quota already exhausted before execution began.",
self.tracker.current_snapshot(),
)
for step_index in range(self.tracker.config.max_steps + 1):
# ── CHECKPOINT 1: Time-based quota check before each step ──
time_status = self.tracker.check_time_only()
if time_status == QuotaStatus.EXHAUSTED:
snap = self.tracker.current_snapshot()
logger.warning(self.tracker.generate_report())
return self._build_truncated_result(
"Agent timed out before completing the task.",
snap,
)
# ── CHECKPOINT 2: Step-based quota enforcement ──
step_status = self.tracker.record_step()
if step_status == QuotaStatus.EXHAUSTED:
snap = self.tracker.current_snapshot()
logger.warning(self.tracker.generate_report())
return self._build_truncated_result(
"Step limit reached before task completion.",
snap,
)
# Handle warnings mid-execution
if step_status == QuotaStatus.WARNING:
logger.warning(
f"Approaching quota limits at step {step_index}. "
"Consider reducing context size or simplifying the task."
)
# Optionally inject a system reminder to be concise
self.messages.append({
"role": "system",
"content": (
"SYSTEM NOTE: You are approaching your resource budget limit. "
"Be concise and prioritize the most critical remaining actions."
),
})
# ── LLM Reasoning Call ──
try:
response = self.llm_call(self.messages)
except Exception as e:
logger.error(f"LLM call failed at step {step_index}: {e}")
return self._build_truncated_result(
f"LLM call failed: {str(e)}",
self.tracker.current_snapshot(),
)
# Extract token counts from the API response metadata
# (In production, parse these from the actual API response object)
input_tokens = response.get("usage", {}).get("prompt_tokens", 0)
output_tokens = response.get("usage", {}).get("completion_tokens", 0)
# ── CHECKPOINT 3: Token and cost quota enforcement ──
llm_status = self.tracker.record_llm_call(input_tokens, output_tokens)
if llm_status == QuotaStatus.EXHAUSTED:
snap = self.tracker.current_snapshot()
logger.warning(self.tracker.generate_report())
return self._build_truncated_result(
"Token or cost limit reached during LLM call.",
snap,
)
assistant_message = response.get("choices", [{}])[0].get("message", {})
content = assistant_message.get("content", "")
self.messages.append({"role": "assistant", "content": content})
# ── Parse the response for tool calls or final answer ──
parsed = self._parse_response(content)
if parsed["type"] == "final_answer":
self.final_answer = parsed["content"]
logger.info(self.tracker.generate_report())
return {
"success": True,
"answer": self.final_answer,
"quota_report": self.tracker.current_snapshot(),
"total_steps": step_index + 1,
}
elif parsed["type"] == "tool_call":
# ── Execute the tool (also quota-aware) ──
tool_result = self._execute_tool(parsed["tool_name"], parsed["arguments"])
self.messages.append({
"role": "tool",
"content": json.dumps(tool_result),
"name": parsed["tool_name"],
})
# Loop continues to next reasoning step
else:
# Unparseable response; ask the model to clarify
self.messages.append({
"role": "user",
"content": "Please provide either a tool call or your final answer.",
})
# Fell through after max_steps iterations
snap = self.tracker.current_snapshot()
logger.warning(self.tracker.generate_report())
return self._build_truncated_result(
"Maximum steps reached without a final answer.",
snap,
)
def _parse_response(self, content: str) -> Dict[str, Any]:
"""
Parse the LLM response to determine if it's a tool call or final answer.
In production, use a structured output parser or function-calling format.
"""
# Simplified parsing logic for demonstration
if "FINAL_ANSWER:" in content:
answer = content.split("FINAL_ANSWER:", 1)[1].strip()
return {"type": "final_answer", "content": answer}
elif "TOOL_CALL:" in content:
# Expect format: TOOL_CALL: tool_name {"arg": "value"}
try:
tool_part = content.split("TOOL_CALL:", 1)[1].strip()
tool_name, args_str = tool_part.split(" ", 1)
arguments = json.loads(args_str)
return {"type": "tool_call", "tool_name": tool_name, "arguments": arguments}
except (ValueError, json.JSONDecodeError) as e:
return {"type": "unparseable", "error": str(e)}
return {"type": "unparseable"}
def _execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""
Execute a tool and return the result. In production, this would dispatch
to registered tool implementations with their own sub-quotas.
"""
# Simulate tool execution with a small delay to mimic real behavior
time.sleep(0.1)
# Find the tool definition
tool_def = next((t for t in self.tools if t["name"] == tool_name), None)
if tool_def is None:
return {"error": f"Unknown tool: {tool_name}"}
# In production, actually invoke the tool here
return {"tool": tool_name, "arguments": arguments, "result": "simulated_output"}
def _build_truncated_result(self, reason: str, snapshot: QuotaSnapshot) -> Dict[str, Any]:
"""Construct a graceful degradation response when quotas are exhausted."""
return {
"success": False,
"answer": f"Task incomplete: {reason}",
"partial_answer": self.final_answer,
"quota_report": snapshot,
"exhausted_reason": reason,
}System-Level Concurrency and Rate Limiting
Individual agent quotas are essential, but you also need system-wide controls. The following AgentExecutionPool acts as a global gatekeeper that limits concurrent agent sessions and enforces per-minute rate limits on downstream API calls. This prevents a single expensive agent or a burst of user requests from cascading into infrastructure overload.
import asyncio
import time
from collections import deque
from dataclasses import dataclass
from typing import Callable, Awaitable
@dataclass
class PoolConfig:
max_concurrent_sessions: int = 10
max_llm_calls_per_minute: int = 200
max_tool_calls_per_minute: int = 500
session_queue_timeout_seconds: int = 30
class AgentExecutionPool:
"""
System-wide concurrency and rate limiter for agent sessions.
Uses a sliding window algorithm for rate limiting and a semaphore
for concurrency control.
"""
def __init__(self, config: PoolConfig):
self.config = config
self._semaphore = asyncio.Semaphore(config.max_concurrent_sessions)
self._llm_call_times: deque = deque()
self._tool_call_times: deque = deque()
self._pool_lock = asyncio.Lock()
async def acquire_session_slot(self) -> bool:
"""
Try to acquire a slot for a new agent session.
Returns True if a slot was acquired within the timeout, False otherwise.
"""
try:
await asyncio.wait_for(
self._semaphore.acquire(),
timeout=self.config.session_queue_timeout_seconds,
)
return True
except asyncio.TimeoutError:
return False
def release_session_slot(self):
"""Release a session slot so another agent can start."""
self._semaphore.release()
async def check_llm_rate_limit(self) -> bool:
"""
Check if an LLM API call can proceed under the rate limit.
Implements a sliding window algorithm.
"""
async with self._pool_lock:
now = time.time()
window_start = now - 60.0 # 1-minute window
# Remove timestamps older than the window
while self._llm_call_times and self._llm_call_times[0] < window_start:
self._llm_call_times.popleft()
if len(self._llm_call_times) >= self.config.max_llm_calls_per_minute:
return False
self._llm_call_times.append(now)
return True
async def check_tool_rate_limit(self) -> bool:
"""Rate limit check for tool executions."""
async with self._pool_lock:
now = time.time()
window_start = now - 60.0
while self._tool_call_times and self._tool_call_times[0] < window_start:
self._tool_call_times.popleft()
if len(self._tool_call_times) >= self.config.max_tool_calls_per_minute:
return False
self._tool_call_times.append(now)
return True
async def run_agent_with_pool_limits(
self,
agent_runner: Callable[[], Awaitable[Dict[str, Any]]],
) -> Dict[str, Any]:
"""
Execute an agent session within the pool's concurrency limits.
Wraps the agent runner with pool-level acquisition and release.
"""
acquired = await self.acquire_session_slot()
if not acquired:
return {
"success": False,
"answer": "System is at capacity. Please try again later.",
"quota_report": None,
}
try:
return await agent_runner()
finally:
self.release_session_slot()
# Usage example with the pool
async def example_pool_usage():
pool_config = PoolConfig(
max_concurrent_sessions=5,
max_llm_calls_per_minute=100,
)
pool = AgentExecutionPool(pool_config)
async def my_agent_runner():
# This would instantiate an AgentOrchestrator with its own QuotaTracker
# and execute the agent loop
return {"success": True, "answer": "Example result"}
result = await pool.run_agent_with_pool_limits(my_agent_runner)
return resultDesigning a Quota-Aware Agent from Scratch
Let's consolidate everything into a complete, runnable example that ties together the quota tracker, agent orchestrator, and execution pool. This final example demonstrates the full lifecycle of a quota-aware agent session, including configuration, execution, graceful degradation, and detailed reporting.
import time
import json
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
def simulate_llm_call(messages, call_count, tracker):
"""
Simulate an LLM API response with realistic token counts.
In production, replace this with your actual LLM client (OpenAI, Anthropic, etc.).
"""
# Simulate increasing token usage as conversation grows
input_tokens = sum(len(m["content"]) // 4 for m in messages) # rough estimate
output_tokens = 150 + (call_count * 30) # responses tend to grow with context
# Simulate final answer after a few iterations
if call_count >= 3:
content = (
"Based on my research, the top quantum computing developments include "
"progress in error correction codes and new qubit architectures. "
"FINAL_ANSWER: Quantum error correction has advanced significantly "
"with the introduction of the surface code achieving thresholds above 1% "
"logical error rates. IBM and Google have both published results demonstrating "
"scalable logical qubits in 2024."
)
else:
content = (
f"TOOL_CALL: web_search {{\"query\": \"quantum computing breakthroughs {call_count}\"}}"
)
return {
"choices": [{"message": {"role": "assistant", "content": content}}],
"usage": {
"prompt_tokens": input_tokens,
"completion_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens,
},
}
# ── Configuration ──────────────────────────────────────────
quota_config = QuotaConfig(
max_input_tokens=50_000,
max_output_tokens=20_000,
max_total_tokens=70_000,
max_cost_dollars=2.00,
max_steps=8,
max_wall_clock_seconds=120,
warning_threshold_ratio=0.7,
input_cost_per_1k=0.01,
output_cost_per_1k=0.03,
)
tools = [
{"name": "web_search", "description": "Search the web for information"},
{"name": "calculator", "description": "Perform mathematical calculations"},
]
# ── Instantiate orchestrator ───────────────────────────────
orchestrator = AgentOrchestrator(
quota_config=quota_config,
system_prompt="You are a research assistant. Use tools when needed, then provide a final answer.",
tools=tools,
llm_call_function=lambda msgs: simulate_llm_call(
msgs,
orchestrator.tracker.snapshot.steps_completed,
orchestrator.tracker,
),
)
# ── Execute ─────────────────────────────────────────────────
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Starting agent session...")
start_time = time.time()
try:
result = orchestrator.run(
"What are the latest breakthroughs in quantum error correction?"
)
except QuotaExhaustedError as e:
print(f"\n❌ QUOTA EXHAUSTED: {e.reason}")
print(e.snapshot)
result = {
"success": False,
"answer": f"Agent stopped: {e.reason}",
"quota_report": e.snapshot,
}
elapsed = time.time() - start_time
# ── Display Results ────────────────────────────────────────
print("\n" + "="*50)
print(" EXECUTION COMPLETE")
print("="*50)
print(f" Wall time: {elapsed:.2f}s")
print(f" Success: {result.get('success', False)}")
print(f" Answer: {result.get('answer', 'No answer')[:200]}...")
print()
if result.get("quota_report"):
snap = result["quota_report"]
print(f" Input tokens used: {snap.input_tokens_used:,}")
print(f" Output tokens used: {snap.output_tokens_used:,}")
print(f" Total tokens used: {snap.total_tokens_used:,}")
print(f" Cost accrued: ${snap.cost_accrued_dollars:.4f}")
print(f" Steps completed: {snap.steps_completed}")
print(f" Final status: {snap.status.value}")
print("\n" + orchestrator.tracker.generate_report())Best Practices for Production Quota Systems
Building quota enforcement into your agents is not a one-time integration task—it requires ongoing tuning and operational discipline. Here are the patterns that experienced teams adopt:
1. Implement Graceful Degradation, Never Silent Failures
When a quota is exhausted, the agent must return a structured partial result—not an empty response or a cryptic exception. Include what was accomplished, what limit was hit, and actionable guidance for the user. The _build_truncated_result method in our orchestrator exemplifies this pattern. Users should receive something like: "I analyzed 12 of 20 requested documents before hitting my token budget. Here are the findings so far. Would you like me to continue with a new session?"
2. Budget Quotas Per User, Per Session, and Per Task Type
Not all agent tasks have equal value. A customer-facing summarization agent might warrant a $0.50 budget, while an internal code review agent could justify $5.00. Create quota profiles keyed by user tier, task complexity, or business priority. This prevents a free-tier user from consuming the budget allocated for enterprise workloads.
3. Monitor and Alert on Warning Thresholds, Not Just Hard Limits
The warning_threshold_ratio in our configuration exists for a reason. When consumption crosses 75% of any limit, emit structured log events, increment a metrics counter, and—for high-value sessions—page an on-call engineer. Warning thresholds catch anomalies before users experience failures. A sudden spike in token consumption often indicates a prompt injection attack or an infinite reasoning loop.
4. Persist Quota Snapshots for Audit and Billing
Every agent session should log its final QuotaSnapshot to your observability platform. These records are invaluable for cost attribution, capacity planning, and debugging runaway sessions. Serialize the snapshot as JSON and ship it to your logging system:
def log_quota_snapshot(snapshot: QuotaSnapshot, session_id: str, user_id: str):
"""Ship quota data to your observability platform."""
payload = {
"session_id": session_id,
"user_id": user_id,
"timestamp": datetime.utcnow().isoformat(),
"input_tokens": snapshot.input_tokens_used,
"output_tokens": snapshot.output_tokens_used,
"total_tokens": snapshot.total_tokens_used,
"cost_dollars": round(snapshot.cost_accrued_dollars, 6),
"steps": snapshot.steps_completed,
"wall_seconds": time.time() - snapshot.start_time,
"status": snapshot.status.value,
"exhausted_reason": snapshot.exhausted_reason,
}
# In production: logger.info("quota_snapshot", extra={"structured": payload})
# Or ship directly to Datadog, Prometheus, CloudWatch, etc.
print(json.dumps(payload, indent=2))5. Test Quota Exhaustion Scenarios Explicitly
Write integration tests that deliberately exhaust each quota dimension. Verify that the agent returns partial results, that alerts fire, and that no orphaned resources remain. Test with artificially low limits to accelerate validation:
def test_token_exhaustion_returns_partial_result():
"""Verify graceful handling when token budget is depleted."""
tiny_config = QuotaConfig(
max_total_tokens=500, # intentionally tiny
max_steps=3,
max_cost_dollars=0.05,
max_wall_clock_seconds=30,
)
orchestrator = AgentOrchestrator(
quota_config=tiny_config,
system_prompt="You are a helpful assistant.",
tools=[],
llm_call_function=lambda msgs: {
"choices": [{"message": {"content": "FINAL_ANSWER: Test complete."}}],
"usage": {"prompt_tokens": 400, "completion_tokens": 200},
},
)
result = orchestrator.run("Test query")
assert result["success"] == False
assert result["quota_report"].status == QuotaStatus.EXHAUSTED
assert "partial_answer" in result or "answer" in result
print("✅ Token exhaustion test passed: graceful degradation confirmed.")6. Combine Quotas with Semantic Guardrails
Resource quotas prevent runaway costs, but they don't prevent an agent from executing harmful actions within its budget. Pair quota enforcement with content safety checks, tool-calling allowlists, and human-in-the-loop approval for high-impact operations. The quota system handles the "how much" question; semantic guardrails handle the "what" question.
7. Expose Quota Status in Streaming Responses
For agents that stream their reasoning to users, include periodic quota status updates in the event stream. This builds user trust and helps users understand why an agent might suddenly become more concise or terminate early. A simple status event might read: {"type": "quota_update", "remaining_budget_pct": 22, "estimated_actions_left": 3}.
Handling Edge Cases and Failure Modes
Real-world agents encounter scenarios that stress quota systems in unexpected ways. Here are specific edge cases and how to handle them:
- Single enormous LLM call: A user uploads a