Appearance
5.6 Compulsory Operational Safety: Rate Limiters, Cost Caps & Circuit Breakers
Canonical Curriculum Reference:
docs/plan/05_phase5_harness_engineering_mcp.md
Textbook Reference: AI Agents and Applications (Manning Publications), Chapter 14, Section 14.2 ("Guardrails", pp. 365–373) & Section 14.3 (pp. 374–376).
🛑 1. Unbounded Consumption: The Production Threat Model
Autonomous agents possess the ability to spawn self-directed API calls, query external tools, and retry failing workflows. Left ungoverned, a single runaway agent loop or malicious tenant can cause:
- Financial Denial-of-Service (Denial-of-Wallet): Incurring thousands of dollars in commercial LLM token bills within minutes.
- Provider Quota Starvation: Consuming the entire team's Tier 4 API rate limits, crashing production web apps sharing the same credentials.
- Cascading Tool Failures: Overwhelming fragile downstream internal APIs (e.g. staging databases or ERP endpoints) with hundreds of concurrent tool queries.
To mitigate these threats, the AI Harness must implement strict, non-bypassable operational guardrails.
⏱️ 2. The Multi-Dimensional AIRateLimiter
Standard web rate limiters count raw HTTP requests. In AI systems, counting requests alone is insufficient: a single request might consume 15 tokens or 128,000 tokens.
A production AIRateLimiter must enforce three simultaneous boundaries:
- Requests Per Minute (RPM): Throttles concurrency.
- Tokens Per Minute (TPM): Throttles token volume to stay within cloud provider rate limits.
- Daily Budget Cap (USD): Enforces an absolute financial hard stop per user/tenant.
Incoming LLM Request
|
v
+-----------------------------------------------------------+
| AIRateLimiter |
| |
| 1. Check RPM (Sliding Window): Requests <= 20/min? |
| [FAIL] --> Raise RateLimitExceededException |
| |
| 2. Check TPM (Sliding Window): Estimated Tokens <= 50k? |
| [FAIL] --> Raise TokenQuotaExceededException |
| |
| 3. Check Financial Cap: Daily Spend <= $10.00? |
| [FAIL] --> Raise BudgetExceededException (Hard Stop) |
+-----------------------------------------------------------+
|
[ALL PASS]
v
Forward Call to LLM Model⚡ 3. The Tool Circuit Breaker & Error Classification
When external tools (e.g. an MCP weather server or internal customer database) experience an outage, an agent without a circuit breaker will retry relentlessly, inflating latency and exacerbating the outage.
Tool Error Classification
The harness must classify every tool error into one of three distinct categories:
- Transient (Retryable): Network socket timeout, HTTP 429 rate limit, 503 service unavailable
Retry with exponential backoff. - Deterministic (Non-Retryable): Schema validation failure, 404 not found, invalid parameter format
Feed error back to model immediately without retry. - Security / Fatal: 401 unauthorized, 403 forbidden, PII violation
Terminate execution immediately and alert security operations.
python
from enum import StrEnum
class ErrorSeverity(StrEnum):
TRANSIENT = "TRANSIENT" # Eligible for retry
DETERMINISTIC = "DETERMINISTIC"# Non-retryable; feed to model context
FATAL = "FATAL" # Abort workflow immediately💻 4. Production Implementation: Safety Guardrails
python
import time
from collections import deque
from typing import Final
from pydantic import BaseModel, ConfigDict, Field
class RateLimitStatus(BaseModel):
model_config = ConfigDict(frozen=True)
allowed: bool
current_rpm: int
current_tpm: int
daily_spend_usd: float
rejection_reason: str | None = None
class AIRateLimiter:
"""Enforces multi-dimensional rate and financial budget boundaries."""
def __init__(
self,
max_rpm: int = 20,
max_tpm: int = 60_000,
daily_budget_usd: float = 10.0,
cost_per_1k_tokens: float = 0.002
) -> None:
self._max_rpm: Final[int] = max_rpm
self._max_tpm: Final[int] = max_tpm
self._daily_budget: Final[float] = daily_budget_usd
self._cost_rate: Final[float] = cost_per_1k_tokens
self._request_timestamps: deque[float] = deque()
self._token_records: deque[tuple[float, int]] = deque()
self._total_spent_today: float = 0.0
def evaluate_request(self, estimated_tokens: int) -> RateLimitStatus:
"""Evaluates whether an outgoing LLM call is within safe operational limits."""
now = time.monotonic()
one_minute_ago = now - 60.0
# 1. Clean sliding windows
while self._request_timestamps and self._request_timestamps[0] < one_minute_ago:
self._request_timestamps.popleft()
while self._token_records and self._token_records[0][0] < one_minute_ago:
self._token_records.popleft()
# 2. Check RPM
current_rpm = len(self._request_timestamps)
if current_rpm >= self._max_rpm:
return RateLimitStatus(
allowed=False,
current_rpm=current_rpm,
current_tpm=sum(t[1] for t in self._token_records),
daily_spend_usd=self._total_spent_today,
rejection_reason=f"Exceeded max RPM limit of {self._max_rpm}"
)
# 3. Check TPM
current_tpm = sum(t[1] for t in self._token_records)
if current_tpm + estimated_tokens > self._max_tpm:
return RateLimitStatus(
allowed=False,
current_rpm=current_rpm,
current_tpm=current_tpm,
daily_spend_usd=self._total_spent_today,
rejection_reason=f"Exceeded max TPM limit of {self._max_tpm}"
)
# 4. Check Daily Spend Cap
estimated_cost = (estimated_tokens / 1000.0) * self._cost_rate
if self._total_spent_today + estimated_cost > self._daily_budget:
return RateLimitStatus(
allowed=False,
current_rpm=current_rpm,
current_tpm=current_tpm,
daily_spend_usd=self._total_spent_today,
rejection_reason=f"Exceeded daily budget cap of ${self._daily_budget:.2f}"
)
# Record usage
self._request_timestamps.append(now)
self._token_records.append((now, estimated_tokens))
self._total_spent_today += estimated_cost
return RateLimitStatus(
allowed=True,
current_rpm=len(self._request_timestamps),
current_tpm=sum(t[1] for t in self._token_records),
daily_spend_usd=self._total_spent_today
)
class CircuitBreakerOpenException(Exception):
"""Raised when an external tool is tripped into OPEN state."""
pass
class ToolCircuitBreaker:
"""Protects external microservices and tools from cascading failure loops."""
def __init__(self, failure_threshold: int = 3, reset_timeout_seconds: float = 30.0) -> None:
self._threshold: Final[int] = failure_threshold
self._timeout: Final[float] = reset_timeout_seconds
self._failure_count: int = 0
self._last_failure_time: float = 0.0
self._state: str = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def record_success(self) -> None:
self._failure_count = 0
self._state = "CLOSED"
def record_failure(self) -> None:
self._failure_count += 1
self._last_failure_time = time.monotonic()
if self._failure_count >= self._threshold:
self._state = "OPEN"
def check_execution_allowed(self) -> None:
if self._state == "OPEN":
if time.monotonic() - self._last_failure_time > self._timeout:
self._state = "HALF_OPEN"
return
raise CircuitBreakerOpenException(
f"Tool circuit breaker is OPEN ({self._failure_count} consecutive failures). "
f"Fast-failing request to protect downstream service."
)🧪 5. Automated Verification with Pytest
python
import pytest
def test_ai_rate_limiter_enforces_budget_cap():
limiter = AIRateLimiter(daily_budget_usd=0.01, cost_per_1k_tokens=0.005)
# 1. First call costs $0.005 (Allowed)
res1 = limiter.evaluate_request(estimated_tokens=1000)
assert res1.allowed is True
# 2. Second call costs $0.005 (Total = $0.010, Allowed)
res2 = limiter.evaluate_request(estimated_tokens=1000)
assert res2.allowed is True
# 3. Third call exceeds $0.010 budget (Rejected)
res3 = limiter.evaluate_request(estimated_tokens=500)
assert res3.allowed is False
assert "Exceeded daily budget cap" in res3.rejection_reason
def test_tool_circuit_breaker_trips():
breaker = ToolCircuitBreaker(failure_threshold=2, reset_timeout_seconds=10.0)
breaker.record_failure()
breaker.check_execution_allowed() # 1 failure: Still CLOSED
breaker.record_failure() # 2 failures: Trips to OPEN
with pytest.raises(CircuitBreakerOpenException):
breaker.check_execution_allowed()