In Why AI Agents Fail in Production, I wrote about the silent, architectural failures that make agents break under real load—structured output, constraints over roles, observable boundaries. That post was the diagnosis. This one is the prescription.
Traditional error handling—try/catch, retries, health checks—assumes deterministic code and obvious failures. AI agents are neither. A retry returns a different (possibly worse) answer. A hallucinated API call returns 200 OK with catastrophic side effects. By the time something visibly breaks, the root cause is three steps back.
When I say “AI agents,” I mean systems that call tools, mutate state, and trigger real side effects—not chatbots. Think database writes, API calls, infra changes, or workflow automation running unattended.
You need different patterns.
Five Patterns That Actually Work
After breaking things in production more times than I’d like to admit, these are the five patterns I now treat as non-negotiable. Together, they ensure failures are detected early, contained tightly, and surfaced deliberately—not silently propagated. The examples below use the Strands Agents SDK, but every pattern is framework-agnostic—the same ideas apply whether you’re using LangGraph, CrewAI, or raw function calling. They form a natural progression: detect failures (circuit breakers), prevent them (validation), contain partial failures (sagas), limit blast radius (budget guardrails), and know when to stop (escalation).
1. Circuit Breakers for LLM Calls
The classic circuit breaker pattern—closed, open, half-open—adapts well to AI agents, with one critical difference: you’re not just tracking HTTP failures. You’re tracking quality failures—any output that violates schema, fails a semantic invariant, or produces an unsafe action, even if the API call itself succeeded.
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import (
BeforeToolCallEvent,
AfterToolCallEvent,
)
class CircuitBreakerHook(HookProvider):
def __init__(self, failure_threshold=3, reset_timeout=60):
self.failures = 0
self.threshold = failure_threshold
self.reset_timeout = reset_timeout
self.state = "closed" # closed, open, half-open
self.last_failure_time = None
def register_hooks(self, registry: HookRegistry) -> None:
registry.add_callback(BeforeToolCallEvent, self.check_circuit)
registry.add_callback(AfterToolCallEvent, self.track_quality)
def check_circuit(self, event: BeforeToolCallEvent) -> None:
if self.state == "open":
if time.time() - self.last_failure_time > self.reset_timeout:
self.state = "half-open" # Allow one probe
else:
event.cancel_tool = True # Block execution
def track_quality(self, event: AfterToolCallEvent) -> None:
content = event.result.get("content", [])
result_text = content[0].get("text", "") if content else ""
has_error = "error" in result_text.lower()
if has_error or not self._passes_validation(result_text):
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.threshold:
self.state = "open"
else:
if self.state == "half-open":
self.state = "closed"
self.failures = 0
The key: when the circuit opens, stop. Don’t burn tokens on a model that’s producing garbage. Wait, then probe with a single request before resuming.
Using the Strands SDK’s HookProvider, the circuit breaker plugs directly into the agent lifecycle—BeforeToolCallEvent blocks execution when the circuit is open, and AfterToolCallEvent inspects results for quality failures. No wrapper functions, no monkey-patching. The hook fires on every tool call automatically.
I track validation failures, not just HTTP errors. If the agent produces three consecutive outputs that fail schema validation, the circuit opens—even though every API call “succeeded.”
2. Validate Before You Execute
In the previous post, I argued that constraints beat roles and structured output is the API contract for LLMs. This pattern is the concrete implementation of that principle.
Never let an agent’s output directly trigger a side effect. Always validate first.
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import BeforeToolCallEvent
class ValidationHook(HookProvider):
def register_hooks(self, registry: HookRegistry) -> None:
registry.add_callback(BeforeToolCallEvent, self.validate)
def validate(self, event: BeforeToolCallEvent) -> None:
tool_name = event.tool_use.get("name", "")
tool_input = event.tool_use.get("input", {})
# Sanity check: block dangerous operations
if (
tool_name == "delete_records"
and tool_input.get("count", 0) > 100
):
event.cancel_tool = True
# Boundary check: restrict to allowed targets
if tool_input.get("target") not in ALLOWED_TARGETS:
event.cancel_tool = True
With Strands’ BeforeToolCallEvent, you intercept every tool call before execution. The hook inspects event.tool_use — the tool name and inputs — and cancels via event.cancel_tool = True if anything fails validation. No separate validation function to remember to call; it fires automatically.
Three layers of validation:
- Schema — Is the output structurally correct? (Missing a required field, wrong type, malformed JSON.)
- Sanity — Does the action make sense? (Deleting 10,000 records? Probably not.)
- Boundary — Is the agent operating within its allowed scope? (Cross-tenant access, targeting a production table from a staging workflow.)
This builds directly on the tool design lesson from giving my agent full API access. When I built pdf-mcp, splitting one monolithic tool into eight focused tools eliminated most validation failures before they could happen. Constrain what the agent can do, and you prevent most errors at the source.
3. Idempotent Workflows (The Saga Pattern)
AI agents retry. Models have transient failures. Networks drop. If your agent workflow isn’t idempotent, retries create duplicate side effects.
Idempotency prevents duplicate effects; the saga pattern handles partial completion. You need both once agents can fail mid-workflow. Borrowing from the saga pattern in distributed systems, each step records its completion and defines a compensation action.
steps = [
Step("fetch_data", compensate=None), # Read-only, safe
Step("transform", compensate=None), # Pure function, safe
Step("write_to_db", compensate="delete_record"), # Reversible
Step("send_notification", compensate="send_correction"), # Compensatable
]
Classify every step:
- Read-only — Safe to retry freely
- Reversible — Can undo (delete what you created)
- Compensatable — Can’t undo, but can correct (send a follow-up notification)
- Final — Can’t undo at all (payment processed) — these need the most validation before execution, and should go through a human escalation flow (Pattern #5) so an irreversible action never fires without explicit approval
When an agent fails mid-workflow, you walk backwards through completed steps and run compensation. No orphaned records. No half-finished operations.
In Strands, the GraphBuilder multi-agent pattern provides a natural structure for this — each node is an agent, conditional edges route to compensation nodes on failure, and the graph handles execution order:
from strands import Agent
from strands.multiagent import GraphBuilder
from strands.multiagent.base import Status
order_agent = Agent(
name="order",
system_prompt="Create the order.",
callback_handler=None,
)
payment_agent = Agent(
name="payment",
system_prompt="Process payment.",
callback_handler=None,
)
fulfillment_agent = Agent(
name="fulfillment",
system_prompt="Ship the order.",
callback_handler=None,
)
rollback_agent = Agent(
name="rollback",
system_prompt="Cancel the order and notify the customer.",
callback_handler=None,
)
def payment_succeeded(state):
return (
"payment" in state.results
and state.results["payment"].status == Status.COMPLETED
)
def payment_failed(state):
return (
"payment" in state.results
and state.results["payment"].status == Status.FAILED
)
builder = GraphBuilder()
builder.add_node(order_agent, "create_order")
builder.add_node(payment_agent, "payment")
builder.add_node(fulfillment_agent, "fulfillment")
builder.add_node(rollback_agent, "compensate_order")
builder.add_edge("create_order", "payment")
builder.add_edge("payment", "fulfillment", condition=payment_succeeded)
builder.add_edge("payment", "compensate_order", condition=payment_failed)
builder.set_entry_point("create_order")
graph = builder.build()
If payment fails, the graph routes to compensate_order instead of fulfillment — no orphaned orders, no half-finished workflows.
4. Budget and Token Guardrails
An IDC survey found that 92% of organizations implementing agentic AI reported costs higher than expected. The primary driver? Runaway loops.
An agent that retries endlessly, or recursively calls tools, will happily burn through your API budget while producing nothing useful.
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import AfterInvocationEvent
class ExecutionGuardHook(HookProvider):
def __init__(self, max_tokens=100_000, max_cycles=20):
self.total_tokens = 0
self.total_cycles = 0
self.max_tokens = max_tokens
self.max_cycles = max_cycles
def register_hooks(self, registry: HookRegistry) -> None:
registry.add_callback(AfterInvocationEvent, self.check_budget)
def check_budget(self, event: AfterInvocationEvent) -> None:
if event.result is None:
return
usage = event.result.metrics.accumulated_usage
self.total_tokens += usage["totalTokens"]
self.total_cycles += event.result.metrics.cycle_count
if self.total_tokens > self.max_tokens:
raise BudgetExceeded("Token limit reached")
if self.total_cycles > self.max_cycles:
raise BudgetExceeded("Cycle limit reached — possible loop")
Hard limits, not hopes. The Strands SDK exposes result.metrics.accumulated_usage and result.metrics.cycle_count after each invocation — real token counts and reasoning cycles, not estimates. Set a ceiling for both, and the AfterInvocationEvent hook enforces it automatically.
This also catches the subtle failure where an agent enters a self-correction loop: generate, validate, fail, regenerate, validate, fail… Each cycle burns tokens with no progress. A step counter catches this immediately.
5. Know When to Ask for Help
The hardest pattern to get right: when should the agent stop and escalate to a human?
I use a simple three-tier framework:
| Risk Level | Confidence | Action |
|---|---|---|
| Low | High | Agent retries autonomously |
| Medium | Uncertain | Agent completes in draft/read-only mode, flags for review |
| High | Any | Agent stops immediately, escalates with context |
The key insight is risk level, not confidence alone. An agent that’s 90% sure about a read-only query can proceed. An agent that’s 90% sure about deleting production data should still ask.
Confidence isn’t model-reported — that’s unreliable. Instead, classify tools by blast radius upfront. The risk map is static and deterministic: read tools run freely, write tools get validation (Pattern #2), destructive tools trigger interrupt() which pauses execution and surfaces the full decision context to a human. The agent resumes only after explicit approval.
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import BeforeToolCallEvent
DANGEROUS_TOOLS = {
"delete_records",
"drop_table",
"revoke_access",
}
class EscalationHook(HookProvider):
def register_hooks(self, registry: HookRegistry) -> None:
registry.add_callback(BeforeToolCallEvent, self.check_risk)
def check_risk(self, event: BeforeToolCallEvent) -> None:
tool_name = event.tool_use.get("name", "")
if tool_name not in DANGEROUS_TOOLS:
return
approval = event.interrupt(
"high-risk-approval",
reason={
"tool": tool_name,
"inputs": event.tool_use.get("input", {}),
},
)
if approval.lower() != "y":
event.cancel_tool = "Human denied permission"
When the agent does escalate, include the full decision context: tool inputs, model output, validation failures, and the agent’s last reasoning step. An escalation without context just shifts the debugging burden to a human.
If you’re wondering how this composes with Pattern #2’s ValidationHook — both register on BeforeToolCallEvent, and Strands fires Before* hooks in registration order. Register validation first, escalation second: that way invalid inputs get rejected cheaply before the escalation hook ever prompts a human.
The Surprising Part
After implementing all five patterns, I expected the biggest win to come from circuit breakers or budget guardrails—the clever engineering patterns.
It didn’t.
The biggest improvement came from pattern #2: better tool design. When you constrain what an agent can do—smaller tools, clear boundaries, built-in validation—most errors never happen in the first place.
Error handling isn’t just about catching failures. It’s about designing systems where the most dangerous failures are structurally impossible.
What’s Next
These five patterns are a starting point, not a complete solution. I’m working through the rest of the production puzzle:
- Testing: How do you test non-deterministic agents without brittle mocks? I’ll cover this in an upcoming post on testing AI agents — unit tests, evals, and integration tests.
- Monitoring: These patterns catch failures, but you still need to see them. Next up: how to monitor AI agents in production.
- Cost: Budget guardrails are reactive. I’m also exploring proactive cost control — keeping LLM bills predictable before they spiral.
If you’re building AI agents that interact with the real world, start with these five patterns. They won’t prevent every failure—but they’ll make sure failures are loud, contained, and recoverable. For a hands-on walkthrough of the Strands SDK used throughout this post, see Strands Agents SDK: Building My First AI Agent.
Built from production failures—so yours can be less painful.
