別再怪罪 LLM：AI 智能體故障其實是分散式系統失效

Originally published on tamiz.pro.

The Illusion of a Single Brain

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.

Table of Contents

• 1. Anatomy of an Agent as a Distributed System
• 2. Failure Modes That Masquerade as Model Bugs
• 3. The Retry Trap: When Idempotency Matters
• 4. Partial Failure and the Saga Pattern
• 5. Observability: You Can't Debug What You Can't See
• 6. Testing Strategies for Non-Deterministic Systems
• 7. Production Guardrails: Circuit Breakers, Timeouts, and Fallbacks
• 8. Case Study: The "Stuck Loop" That Was a Timeout Cascade
• Frequently Asked Questions

1. Anatomy of an Agent as a Distributed System

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:

• Network boundary: HTTP/gRPC calls with latency, timeouts, retries
• Rate limits: Token buckets, concurrent request caps, quota exhaustion
• Partial degradation: Elevated latency without errors, degraded quality responses
• State inconsistency: Caches stale, indexes lag, transactions roll back
• Version skew: Model versions change, tool APIs evolve, schemas drift

The orchestrator is the only component you fully control. Everything else is a remote service you don't own.

1.1 The Latency Budget Reality

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.

# ponytail: naive sequential execution, parallelize independent calls if latency matters
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
# Total: ~6s sequential, ~3s with parallelization

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.

2. Failure Modes That Masquerade as Model Bugs

2.1 The Timeout Truncation Problem

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.

# What the model actually produced (streaming):
"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—"

# What your orchestrator receives (gateway timeout at 60s):
"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"

# Your code sees: incomplete response → treats as model error → retries

The model did reason correctly. The infrastructure ate the tail.

2.2 The Silent Model Swap

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.

# Detect this by logging model fingerprint headers
# OpenAI returns: x-request-id, x-ratelimit-*, and sometimes model fingerprint
curl -v https://api.openai.com/v1/chat/completions ... 2>&1 | grep -i fingerprint
# x-model-fingerprint: fp_abc123  ← log this every request

3. The Retry Trap: When Idempotency Matters

Agents retry. A lot. Network blips, 429s, 5xx errors, timeouts—all trigger retries. But most agent operations are not idempotent.

# DANGEROUS: Non-idempotent operations in a retry loop
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()

# execute_plan might:
# - Charge a credit card (twice)
# - Send an email (twice)
# - Mutate database state (twice)
# - Create a Jira ticket (twice)

3.1 Idempotency Keys for Every Tool Call

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":
        # Deterministic key: execution_id + step + tool + args hash
        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)

# Tool executor checks idempotency store before executing
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

3.2 The "Thinking" Retry Anti-Pattern

Retrying the entire agent loop because one tool failed is wasteful and dangerous. It re-does reasoning, re-calls other tools, and burns tokens.

# BETTER: Granular retry at tool level, not agent level
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

        # Retry ONLY this step, with its own budget
        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)

4. Partial Failure and the Saga Pattern

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:
                # Compensate in reverse order
                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:
                # Log but continue compensating other steps
                logger.error(f"Compensation failed for {step.name}: {comp_err}")
                await alert_oncall(f"Saga {self.execution_id} compensation failed at {step.name}")

4.1 Concrete Compensation Examples

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.

5. Observability: You Can't Debug What You Can't See

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.

5.1 The Minimum Viable Trace

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

5.2 What to Instrument (The Golden Signals for Agents)

5.3 Structured Logging for Post-Mortems

{
  "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."

6. Testing Strategies for Non-Deterministic Systems

You cannot unit-test an LLM call like a pure function. But you can test the system around it.

6.1 Deterministic Replay with Recorded Traces

Record production traces (inputs, outputs, timings, errors). Replay them against code changes.

# test_replay.py
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 structural equivalence, not exact string match
    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

6.2 Chaos Engineering for Agents

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)

# Run in CI: 1000 iterations with 5% random failure injection
# Assert: no uncompensated mutations, no lost user data, graceful degradation

6.3 Evaluation as Regression Testing

Treat your eval set as a regression test suite, not a one-time benchmark.

# evals/regression_suite.yaml
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"]

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)

7. Production Guardrails: Circuit Breakers, Timeouts, and Fallbacks

Distributed systems survive partial degradation through explicit boundaries. Your agent needs the same.

7.1 Per-Provider Circuit Breakers

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()

7.2 Adaptive Timeouts Based on Percentiles

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

# Per-operation adaptive timeouts
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

7.3 Graceful Degradation Ladder

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:
        # Level 0: Full capability
        if not self.degraded:
            return await self._full_run(input, context)

        # Level 1: No web search (slow, unreliable)
        if self.degraded_level >= 1:
            context.available_tools.discard("web_search")

        # Level 2: No code execution (sandbox overhead)
        if self.degraded_level >= 2:
            context.available_tools.discard("python")

        # Level 3: Single-turn only, no planning loop
        if self.degraded_level >= 3:
            return await self._single_turn_fallback(input, context)

        # Level 4: Canned responses only
        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)

8. Case Study: The "Stuck Loop" That Was a Timeout Cascade

The Symptom

"Agent gets stuck repeating 'Let me search for that...' three times then gives generic answer. Model quality issue?"

The Investigation

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.

The Fix (Three Lines)

# 1. Increase gateway timeout to 60s (match search p99 + buffer)
# 2. Make web_search idempotent (cache results for 5min)
# 3. Retry ONLY the failed tool, not the whole plan

# Before (broken):
async def run_agent():
    for attempt in range(3):
        try:
            return await full_agent_loop()
        except ToolTimeout:
            continue

# After (fixed):
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.

Frequently Asked Questions

Q: "My agent is simple—just one LLM call. Is this overkill?"

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.

Q: "How do I convince my team to invest in this infrastructure?"

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.

Q: "What's the minimum viable observability stack?"

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.

Q: "Do I need saga compensation for read-only tools?"

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.

Idempotency and idempotency keys

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.

# Agent side: generate and attach a key
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},
)

# Service side: check cache before executing
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
    # ... actual payment processing ...
    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.

Liveness and timeouts

The third pillar is liveness enforcement. Distributed systems fail—network partitions, GC pauses, 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)

# Circuit breaker state (persisted across restarts)
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.

Orchestration vs. choreography

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.

Observability: the debugging superpower

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.

Putting it all together: a minimal skeleton

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

# --- Mock infrastructure ---

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"


# --- Agent core ---

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."""
        # Attach idempotency key if present
        key = payload.pop("idempotency_key", None)
        if key:
            cached = self.idempotency.get(key)
            if cached:
                self.metrics["calls"] += 1
                return cached

        # Enforce timeout
        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

        # Cache result for idempotency
        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."""
        # Simulate flaky external service
        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."""
        # Step 1: idempotent lookup
        lookup_key = f"workflow-{input_value}"
        cached = self.idempotency.get(lookup_key)
        if cached:
            self.metrics["retries"] += 1
            return cached

        # Step 2: perform external action with timeout + circuit breaker
        payload = {"value": input_value, "idempotency_key": lookup_key}
        result = await self.call_external(payload)

        # Step 3: store result for future idempotent repeats
        self.idempotency.set(lookup_key, result, ttl=3600)

        # Emit metrics (in real code, push to Prometheus/Datadog)
        self._emit_metrics()

        return result

    def _emit_metrics(self):
        # Placeholder for metric emission
        pass


# --- Example usage ---

async def main():
    agent = DistributedAgent()

    # First call—may succeed or fail
    print("Attempt 1:", await agent.run_workflow("hello-world"))

    # Second call with same input—idempotent cache hit
    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.

Conclusion

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.

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.