Originally published on tamiz.pro.
You ship an AI agent. It works in staging. Production hits and suddenly it's "hallucinating," "ignoring instructions," or "getting stuck in loops." The team blames the model. They swap providers, tweak prompts, add few-shot examples. The bugs persist.
Here's the uncomfortable truth: your AI agent is not a model. It's a distributed system.
Every agentβwhether a simple ReAct loop or a multi-agent swarmβdecomposes into multiple independent processes communicating over unreliable networks: the LLM API, tool executors (code interpreters, browsers, API clients), vector databases, memory stores, orchestration logic, and often human-in-the-loop checkpoints. Each component fails independently. Each introduces latency variance. Each has its own failure modes that look exactly like "model stupidity" when observed from the outside.
This article maps the anatomy of an AI agent to distributed systems primitives, shows why classic LLM debugging fails, and gives you the operational toolkit to treat agents like the distributed systems they are.
Draw the architecture of any production agent and you'll find these components:
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β ORCHESTRATOR (Your Code) β
β βββββββββββ βββββββββββ βββββββββββ βββββββββββ β
β β Planner ββ β Executor ββ β Critic ββ β Memory β β
β ββββββ¬βββββ ββββββ¬βββββ ββββββ¬βββββ ββββββ¬βββββ β
βββββββββΌβββββββββββββΌβββββββββββββΌβββββββββββββΌβββββββββββββββ
β β β β
βΌ βΌ βΌ βΌ
βββββββββββββββββ βββββββββββ ββββββββββββ ββββββββββββββ
β LLM Provider β β Tools β β Vector β β State β
β (API Gateway)β β (Code, β β Store β β Store β
β β β Browse,β β (Pineconeβ β (Redis, β
β - Rate limitsβ β API) β β , etc.) β β Postgres) β
β - Latency β β β β β β β
β - Errors β β - Sand- β β - Index β β - TTL β
β - Quotas β β boxes β β lag β β - Locks β
βββββββββββββββββ βββββββββββ ββββββββββββ ββββββββββββββ
Each box is a separate failure domain with its own:
The orchestrator is the only component you fully control. Everything else is a remote service you don't own.
A typical agent turn involves 3β10 LLM calls plus tool invocations. At 500msβ3s per LLM call and 100msβ5s per tool, a single user request spans 5β60 seconds of wall-clock time across 10β50 network hops.
async def agent_turn(user_input: str, ctx: AgentContext) -> AgentResponse:
plan = await llm.plan(user_input, ctx.history) # 1.2s
tool_results = await execute_tools(plan.tools) # 2.8s (parallel)
critique = await llm.critique(plan, tool_results) # 0.9s
memory.update(ctx.session_id, critique) # 0.1s
return await llm.synthesize(plan, tool_results, critique) # 1.1s
Distributed systems rule: latency adds up, and the tail latency dominates. The 99th percentile of your agent's response time is determined by the slowest component's 99th percentile, compounded across hops.
| Observed Symptom | Actual Distributed Systems Cause | Why It Looks Like an LLM Bug |
|---|---|---|
| Agent repeats itself | Tool call timed out, orchestrator retries with same prompt | "Model is stuck in a loop" |
| Incomplete output | LLM stream cut off by gateway timeout | "Model stopped reasoning" |
| Hallucinated facts | Vector store returned stale/empty results | "Model doesn't know" |
| Ignores instructions | System prompt truncated by context window management | "Model doesn't follow prompts" |
| Nonsensical tool args | Tool schema validation failed silently, fell back to string parsing | "Model can't use tools" |
| Inconsistent behavior | Race condition between memory reads/writes across replicas | "Model is unpredictable" |
| Silent degradation | Provider silently routes to smaller/faster model variant | "Model got dumber" |
Most LLM APIs stream tokens. Your gateway (Cloudflare, AWS ALB, nginx) has a default idle timeout of 60s. A complex reasoning trace with tool calls can exceed this.
"I need to check the database... [tool: query_db] ...
The result shows 42 users. Now I'll calculate... [tool: python] ...
The answer is 42 * 1.5 = 63. So the final answer isβ"
"I need to check the database... [tool: query_db] ...
The result shows 42 users. Now I'll calculate... [tool: python] ...
The answer is 42 * 1.5 = 63. So the final answer is"
The model did reason correctly. The infrastructure ate the tail.
Major providers route traffic across model variants (e.g., gpt-4o β gpt-4o-mini for cost optimization) without changing the model identifier. You observe quality drift. You blame "prompt sensitivity." The reality: you're calling a different model.
curl -v https://api.openai.com/v1/chat/completions ... 2>&1 | grep -i fingerprint
Agents retry. A lot. Network blips, 429s, 5xx errors, timeoutsβall trigger retries. But most agent operations are not idempotent.
async def run_agent(task: Task) -> Result:
for attempt in range(3):
try:
plan = await llm.plan(task) # Idempotent (read-only)
result = await execute_plan(plan) # NOT idempotent!
return result
except TransientError:
continue
raise MaxRetriesExceeded()
Every tool invocation that mutates state must carry an idempotency key derived from the agent's execution context.
import hashlib
import json
from dataclasses import dataclass
from typing import Any
@dataclass
class ToolInvocation:
tool_name: str
args: dict[str, Any]
idempotency_key: str
attempt: int = 0
@classmethod
def create(cls, tool_name: str, args: dict, execution_id: str, step: int) -> "ToolInvocation":
content = f"{execution_id}:{step}:{tool_name}:{json.dumps(args, sort_keys=True)}"
key = hashlib.sha256(content.encode()).hexdigest()[:32]
return cls(tool_name=tool_name, args=args, idempotency_key=key)
async def execute_tool(invocation: ToolInvocation) -> ToolResult:
existing = await idempotency_store.get(invocation.idempotency_key)
if existing:
return ToolResult.from_cached(existing) # Return previous result
result = await actual_tool_call(invocation.tool_name, invocation.args)
await idempotency_store.set(invocation.idempotency_key, result, ttl=86400)
return result
Retrying the entire agent loop because one tool failed is wasteful and dangerous. It re-does reasoning, re-calls other tools, and burns tokens.
async def execute_plan(plan: Plan, ctx: ExecutionContext) -> PlanResult:
results = {}
for step in plan.steps:
if step.depends_on and not all(results[d].success for d in step.depends_on):
results[step.id] = StepResult(skipped=True, reason="dependency_failed")
continue
results[step.id] = await execute_step_with_retry(
step,
max_attempts=3,
timeout=step.timeout or 30,
idempotency_key=ctx.make_key(step.id)
)
return PlanResult(results)
An agent turn is a long-running transaction across multiple services. When step 3 of 5 fails, steps 1β2 may have already mutated state. You need compensating actionsβthe Saga pattern.
@dataclass
class SagaStep:
name: str
execute: Callable[[], Awaitable[StepResult]]
compensate: Callable[[StepResult], Awaitable[None]] # Undo action
idempotency_key: str
class AgentSaga:
def __init__(self, execution_id: str):
self.execution_id = execution_id
self.completed_steps: list[SagaStep] = []
self.step_results: dict[str, StepResult] = {}
async def run(self, steps: list[SagaStep]) -> SagaResult:
for step in steps:
try:
result = await step.execute()
self.step_results[step.name] = result
self.completed_steps.append(step)
except Exception as e:
await self._compensate()
return SagaResult(success=False, failed_step=step.name, error=e)
return SagaResult(success=True, results=self.step_results)
async def _compensate(self):
for step in reversed(self.completed_steps):
try:
await step.compensate(self.step_results[step.name])
except Exception as comp_err:
logger.error(f"Compensation failed for {step.name}: {comp_err}")
await alert_oncall(f"Saga {self.execution_id} compensation failed at {step.name}")
| Tool Action | Compensation |
|---|---|
send_email |
Send follow-up "Please disregard previous email" |
create_jira_ticket |
Transition ticket to "Cancelled" |
charge_customer |
Initiate refund via payment provider |
write_database |
Delete inserted rows / revert update |
call_external_api |
Call cancellation endpoint if available |
create_file |
Delete file |
Critical: Compensations themselves can fail. They must be idempotent and retried independently. Store saga state in durable storage (PostgreSQL, not memory) so crashes don't leave dangling partial state.
Most agent observability stops at "input β output." That's like debugging a microservice by only reading the API gateway logs. You need distributed tracing across every hop.
Every agent execution gets a trace_id. Every sub-operation gets a span_id with parent linkage.
from contextvars import ContextVar
import uuid
current_trace: ContextVar[str] = ContextVar("current_trace", default="")
current_span: ContextVar[str] = ContextVar("current_span", default="")
class TracedAgent:
async def run(self, input: str) -> AgentOutput:
trace_id = str(uuid.uuid4())
current_trace.set(trace_id)
with span(trace_id, "agent.run", {"input": input}) as root_span:
plan = await self._plan_with_trace(input, root_span)
tool_results = await self._execute_with_trace(plan, root_span)
output = await self._synthesize_with_trace(plan, tool_results, root_span)
root_span.set_attribute("output", output)
return output
async def _plan_with_trace(self, input: str, parent_span) -> Plan:
with span(parent_span.trace_id, "llm.plan", {"model": self.model}, parent_span.span_id) as span:
span.set_attribute("prompt_tokens", estimate_tokens(input))
result = await self.llm.plan(input)
span.set_attribute("completion_tokens", estimate_tokens(result))
span.set_attribute("latency_ms", span.duration_ms)
return result
| Signal | Where to Capture | Alert Threshold |
|---|---|---|
| Latency (p50, p95, p99) | Per LLM call, per tool, per agent turn | p99 > 30s |
| Error rate | Per provider, per tool, per error type | > 1% 5xx |
| Token usage | Input/output per call, per session | Budget exceeded |
| Retry rate | Per operation type | > 10% retries |
| Partial failure rate | Saga compensation triggered | > 0.1% |
| Context window pressure | Tokens used / max tokens per call | > 80% |
| Tool success rate | Per tool, per error category | < 99% |
{
"timestamp": "2024-01-15T10:23:45.123Z",
"level": "INFO",
"trace_id": "abc-123",
"span_id": "span-456",
"component": "tool_executor",
"tool": "web_search",
"args": {"query": "quarterly revenue", "max_results": 5},
"idempotency_key": "abc123...",
"attempt": 2,
"latency_ms": 2340,
"status": "success",
"result_summary": {"results_count": 5, "tokens": 1200}
}
Query this in production: "Show me all traces where web_search took >5s and the subsequent LLM call produced <100 tokens."
You cannot unit-test an LLM call like a pure function. But you can test the system around it.
Record production traces (inputs, outputs, timings, errors). Replay them against code changes.
import pytest
from agent.tracing import TraceRecord
from agent.orchestrator import AgentOrchestrator
@pytest.mark.parametrize("trace", load_production_traces("2024-01", sample=1000))
async def test_agent_replay(trace: TraceRecord):
"""Replay a real production trace through current code."""
orchestrator = AgentOrchestrator(
llm=FakeLLM(responses=trace.llm_responses), # Deterministic replay
tools=FakeTools(responses=trace.tool_responses),
memory=FakeMemory(initial_state=trace.initial_memory)
)
result = await orchestrator.run(trace.input)
assert result.tool_calls == trace.expected_tool_calls
assert result.final_answer.semantic_equiv(trace.expected_answer)
assert result.latency_ms < trace.original_latency_ms * 1.5 # No regression
Inject failures at the orchestration layer to verify compensations, retries, and fallbacks.
class ChaosAgentWrapper:
"""Wraps an agent and injects failures per configuration."""
def __init__(self, agent, chaos_config: ChaosConfig):
self.agent = agent
self.chaos = chaos_config
async def run(self, input: str) -> AgentOutput:
if self.chaos.should_fail("llm_timeout"):
raise TimeoutError("Injected LLM timeout")
if self.chaos.should_fail("tool_error"):
raise ToolError("Injected tool failure")
if self.chaos.should_delay("network_latency"):
await asyncio.sleep(self.chaos.latency_ms / 1000)
return await self.agent.run(input)
Treat your eval set as a regression test suite, not a one-time benchmark.
cases:
- id: "finance_calc_001"
input: "What's 15% of $2,340?"
expected_tools: ["calculator"]
expected_answer_pattern: "351"
max_latency_ms: 5000
tags: ["math", "single_tool"]
- id: "multi_hop_research_001"
input: "Compare Q3 revenue of Apple vs Microsoft"
expected_tools: ["web_search", "calculator", "web_search"]
expected_answer_contains: ["Apple", "Microsoft", "revenue", "billion"]
max_latency_ms: 30000
tags: ["multi_hop", "research"]
php
async def run_regression_suite(agent, suite_path: str) -> EvaluationReport:
results = []
for case in load_suite(suite_path):
trace = await agent.run_with_trace(case.input)
results.append(EvalResult(
case_id=case.id,
passed=check_case(case, trace),
latency_ms=trace.total_latency_ms,
tokens_used=trace.total_tokens,
trace_id=trace.trace_id
))
return EvaluationReport(results)
Distributed systems survive partial degradation through explicit boundaries. Your agent needs the same.
from enum import Enum
import time
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing fast
HALF_OPEN = "half_open" # Testing recovery
class ProviderCircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 30.0,
half_open_max_calls: int = 3
):
self.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time = 0
self.half_open_calls = 0
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
async def call(self, func: Callable, *args, **kwargs):
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
else:
raise CircuitOpenError("Circuit breaker open")
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls >= self.half_open_max_calls:
raise CircuitOpenError("Half-open limit reached")
self.half_open_calls += 1
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
Wire it per-provider:
llm_clients = {
"openai": CircuitBreakerClient(OpenAIClient(), ProviderCircuitBreaker()),
"anthropic": CircuitBreakerClient(AnthropicClient(), ProviderCircuitBreaker()),
"local": CircuitBreakerClient(LocalLLMClient(), ProviderCircuitBreaker(failure_threshold=10)),
}
async def call_llm_with_fallback(messages, preferred="openai"):
for provider in [preferred, "anthropic", "local"]:
try:
return await llm_clients[provider].complete(messages)
except (CircuitOpenError, RateLimitError, TimeoutError):
continue
raise AllProvidersUnavailable()
Fixed timeouts are wrong. Set timeouts at p99 + buffer per operation type.
class AdaptiveTimeout:
def __init__(self, percentile: float = 0.99, buffer_multiplier: float = 1.5):
self.percentile = percentile
self.buffer = buffer_multiplier
self.latencies: list[float] = [] # Rolling window
self.max_samples = 1000
def record(self, latency_ms: float):
self.latencies.append(latency_ms)
if len(self.latencies) > self.max_samples:
self.latencies.pop(0)
def get_timeout(self) -> float:
if len(self.latencies) < 10:
return 30000 # Default 30s
sorted_lat = sorted(self.latencies)
idx = int(len(sorted_lat) * self.percentile)
return sorted_lat[idx] * self.buffer
timeouts = {
"llm.plan": AdaptiveTimeout(0.99, 1.5),
"llm.critique": AdaptiveTimeout(0.99, 1.5),
"tool.web_search": AdaptiveTimeout(0.95, 2.0), # More variable
"tool.code_exec": AdaptiveTimeout(0.99, 1.2),
"memory.read": AdaptiveTimeout(0.999, 1.1),
}
async def execute_with_timeout(op_name: str, coro):
timeout = timeouts[op_name].get_timeout()
try:
start = time.time()
result = await asyncio.wait_for(coro, timeout=timeout / 1000)
timeouts[op_name].record((time.time() - start) * 1000)
return result
except asyncio.TimeoutError:
timeouts[op_name].record(timeout) # Record timeout as latency
raise
When things go wrong, degrade functionality, not correctness.
class DegradedAgent:
"""Agent that progressively sheds capabilities under stress."""
async def run(self, input: str, context: AgentContext) -> AgentOutput:
if not self.degraded:
return await self._full_run(input, context)
if self.degraded_level >= 1:
context.available_tools.discard("web_search")
if self.degraded_level >= 2:
context.available_tools.discard("python")
if self.degraded_level >= 3:
return await self._single_turn_fallback(input, context)
return await self._canned_fallback(input)
def update_health(self, metrics: HealthMetrics):
"""Called periodically by health monitor."""
if metrics.error_rate > 0.2 or metrics.p99_latency > 60000:
self.degraded_level = min(self.degraded_level + 1, 4)
elif metrics.error_rate < 0.01 and metrics.p99_latency < 15000:
self.degraded_level = max(self.degraded_level - 1, 0)
"Agent gets stuck repeating 'Let me search for that...' three times then gives generic answer. Model quality issue?"
Trace analysis revealed:
Trace: agent_7x9k2
ββ llm.plan (1.2s) β
ββ tool.web_search "quarterly revenue" (28.3s) β TIMEOUT (30s gateway)
ββ llm.plan RETRY (1.1s) β "Let me search for that..."
ββ tool.web_search "quarterly revenue" (29.1s) β TIMEOUT
ββ llm.plan RETRY (1.3s) β "Let me search for that..."
ββ tool.web_search "quarterly revenue" (31.2s) β TIMEOUT
ββ llm.synthesize (0.8s) β Generic answer (no tool results)
Root cause: The search API's p99 latency was 28s. The gateway timeout was 30s. The orchestrator retried the entire plan on any tool failure. Three retries = 90s of user-facing latency.
async def run_agent():
for attempt in range(3):
try:
return await full_agent_loop()
except ToolTimeout:
continue
async def run_agent():
plan = await llm.plan() # No retry needed, idempotent
results = {}
for step in plan.steps:
results[step.id] = await execute_with_retry(
step, max_attempts=2, timeout=60 # Per-step config
)
return await llm.synthesize(plan, results)
Result: Loop eliminated. Latency p99 dropped from 90s β 35s. User complaints dropped to zero.
A: If it's truly one call with no tools, no memory, no retriesβthen yes, it's just an LLM call. But the moment you add: a second LLM call, a tool, a retry loop, or session memoryβyou have a distributed system. The complexity threshold is lower than most teams realize.
A: Start with one trace that proves the point. Find a "model bug" ticket, add distributed tracing, show the actual failure was a timeout/cascade/partial failure. The evidence speaks louder than architecture diagrams.
A: Structured JSON logs with trace_id/ span_id + a log aggregator (Datadog, Loki, CloudWatch) + basic dashboards for the 7 golden signals in section 5.2. You don't need Jaeger/Tempo to startβcorrelation IDs in logs get you 80% of the value.
A: No. Compensation is only for mutating operations (writes, charges, sends, creates). Search, calculation, read-only API calls need no compensationβbut they do need idempotency keys if you retry them, to avoid duplicate costs/latency.
This article is adapted from patterns used in production agent systems at scale. The code snippets are simplified for clarityβproduction versions include metrics emission, structured error types, and integration with your observability stack.
The second pillar is idempotency. LLMs are inherently non-deterministic; the same prompt can yield different completions across calls. When an agent retries a failed action, duplicate side-effects become a critical problem. The standard pattern is to require an idempotency key from the caller, which the service stores and returns unchanged on repeat executions.
Consider a checkout flow where an agent initiates a payment, the LLM times out, and retries. Without idempotency, the customer is charged twice. With it, the payment service recognizes the key and returns the original transaction status.
payment_id = "pay_12345"
idempotency_key = f"checkout-{payment_id}-{int(time.time())}"
result = await agent.call_tool(
"payments_process",
{"amount": 999, "currency": "USD", "idempotency_key": idempotency_key},
)
python
async def process_payment(data: dict):
key = data.pop("idempotency_key", None)
cached = idempotency_store.get(key)
if cached:
return cached # return original result unchanged
idempotency_store.set(key, result, ttl=86400)
return result
ponytail: in-process cache only; production uses distributed ID generators (Redis, DynamoDB) with global TTL coordination.
The third pillar is liveness enforcement. Distributed systems failβnetwork partitions, GC s, thread starvation. An agent without timeouts will hang indefinitely, consuming resources and cascading failures downstream. Every external call must have a hard timeout, and the agent must have a circuit-breaker pattern to stop hammering unhealthy services.
async with asyncio.timeout(5): # 5-second hard limit
result = await agent.call_tool("some_remote_service", payload)
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=30):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failures = 0
self.last_failure = None
self.state = "closed" # closed/open/half-open
async def call(self, fn, *args, **kwargs):
if self.state == "open":
if time.time() - self.last_failure > self.recovery_timeout:
self.state = "half-open"
else:
raise CircuitOpenError("service temporarily unavailable")
try:
result = await fn(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
self.failures = 0
self.state = "closed"
def _on_failure(self):
self.failures += 1
self.last_failure = time.time()
if self.failures >= self.failure_threshold:
self.state = "open"
ponytail: simple in-memory breaker; production persists state and uses exponential backoff with jitter.
The fourth pillar is system architecture choice. In a choreographed system, agents react to events without central coordinationβmessage queues, topics, eventual consistency. In an orchestrated system, a conductor or state machine drives the flow. For AI agents, pure choreography often leads to undetected stuck states; a lightweight orchestrator (Temporal, LangGraph, or even a DB-driven state table) makes liveness and idempotency guarantees explicit.
Consider the difference: in a choreographed flow, an agent receives a "payment_completed" event but the UI never renders because the agent crashed before updating state. An orchestrated flow tracks each step's completion explicitly and can replay or compensate on failure.
All of this hinges on observability. You cannot fix what you cannot measure. Distributed systems already require tracing, metrics, and logs; AI agents add non-determinism on top. Structure your agent outputs so they integrate with your existing stack.
Metrics to emit per call:
agent_call_duration_seconds`` agent_call_success_total (increment on success, decrement on failure)agent_call_error_type (labels: timeout, validation, external_service, llm_token_limit)agent_retry_count (histogram or counter)
Structured logs should include:
request_id (correlates across all service calls)llm_model and temperature
tokens_used and cost_estimate
external_service_latency`` idempotency_key (for duplicate detection debugging)
Traces should span the entire agent workflow: user prompt β tool selection β tool execution β response generation. If a tool fails, the trace shows exactly which step and what input caused it.
Below is a minimal, runnable skeleton that demonstrates the patterns discussed. It is not a production-grade framework but illustrates the core ideas: timeout, idempotency key, circuit breaker, and structured metrics.
import asyncio
import time
import random
from typing import Optional, Dict, Any
class InMemoryIdempotencyStore:
def __init__(self):
self._store: Dict[str, Any] = {}
def get(self, key: str) -> Optional[Any]:
return self._store.get(key)
def set(self, key: str, value: Any, ttl: int = 3600):
self._store[key] = value
class SimpleCircuitBreaker:
def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30.0):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failures = 0
self.last_failure: Optional[float] = None
self.state = "closed" # closed / open / half-open
async def call(self, fn, *args, **kwargs):
if self.state == "open":
if time.time() - self.last_failure > self.recovery_timeout:
self.state = "half-open"
else:
raise RuntimeError("CircuitOpen: service unavailable")
try:
result = await fn(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
self.failures = 0
self.state = "closed"
def _on_failure(self):
self.failures += 1
self.last_failure = time.time()
if self.failures >= self.failure_threshold:
self.state = "open"
class DistributedAgent:
def __init__(self):
self.idempotency = InMemoryIdempotencyStore()
self.circuit = SimpleCircuitBreaker()
self.metrics = {"calls": 0, "errors": 0, "retries": 0}
async def call_external(self, payload: dict) -> dict:
"""Wrap external calls with timeout, idempotency, and circuit breaking."""
key = payload.pop("idempotency_key", None)
if key:
cached = self.idempotency.get(key)
if cached:
self.metrics["calls"] += 1
return cached
try:
async with asyncio.timeout(5):
result = await self.circuit.call(self._do_call, payload)
except asyncio.TimeoutError:
self.metrics["errors"] += 1
raise RuntimeError("Call timed out after 5s") from None
if key:
self.idempotency.set(key, result, ttl=3600)
self.metrics["calls"] += 1
return result
async def _do_call(self, payload: dict) -> dict:
"""Actual external callβhere simulated with random latency."""
await asyncio.sleep(random.uniform(0.1, 2.0))
if random.random() < 0.3:
raise RuntimeError("Simulated external failure")
return {"status": "ok", "data": payload.get("value")}
async def run_workflow(self, input_value: str) -> dict:
"""Example workflow that uses all four pillars."""
lookup_key = f"workflow-{input_value}"
cached = self.idempotency.get(lookup_key)
if cached:
self.metrics["retries"] += 1
return cached
payload = {"value": input_value, "idempotency_key": lookup_key}
result = await self.call_external(payload)
self.idempotency.set(lookup_key, result, ttl=3600)
self._emit_metrics()
return result
def _emit_metrics(self):
pass
async def main():
agent = DistributedAgent()
print("Attempt 1:", await agent.run_workflow("hello-world"))
print("Attempt 2 (cached):", await agent.run_workflow("hello-world"))
if __name__ == "__main__":
asyncio.run(main())
ponytail: in-memory store and breaker for demo only; replace with Redis/DynamoDB and persistent breaker state for production.
AI agent bugs are almost never purely an LLM quality problem. They are distributed systems problems: network calls that fail, services that become unavailable, retries that create duplicates, and state that gets out of sync. By applying standard distributed-systems patternsβidempotency, timeouts, circuit breakers, and explicit orchestrationβyou turn unpredictable LLM behavior into deterministic, observable, and recoverable workflows.
The LLM remains the source of non-determinism in content generation, but the surrounding infrastructure can absorb, retry, and compensate for that non-determinism without bringing the whole system down. Treat your agent like any other distributed service: instrument it, guard its calls, and design for failure from the start.