TL;DR:

  • AI agents fail differently to conventional software — tool timeouts, partial results, hallucinated tool calls, and context saturation all require specific recovery patterns that try/catch alone won’t solve
  • Retry with exponential backoff handles transient failures; circuit breakers prevent cascading failures across tool chains; fallback chains route around unavailable services
  • Checkpointing agent state at task boundaries lets you resume long workflows from the last successful step rather than restarting from scratch
  • Graceful degradation means defining a usable reduced-capability response for every critical path — not just hoping the failure mode is safe

Production AI agents are not just scripts that happen to call an LLM. They’re stateful, long-running processes that interact with external tools, APIs, and databases — all of which fail. And they fail in ways that expose the limits of standard software error handling.

A regular web service that gets a database timeout can retry the query and return a 500 if it keeps failing. An AI agent that gets a tool timeout midway through a multi-step task has a harder problem: partial state has been written, the context window contains evidence of what happened, and the agent needs to decide whether to retry, reroute, or declare partial completion. None of those decisions map cleanly to a try/catch block.

This guide covers the patterns that make agents resilient in production, starting from simple retry logic and building up to checkpoint-based recovery for long-running workflows.

The Failure Modes That Matter

Before you can design recovery, you need to know what you’re recovering from. Agent failures cluster into a few categories:

Transient tool failures — the tool was unavailable for a moment but will work if tried again. HTTP 429 (rate limit), 503 (service temporarily unavailable), and connection timeouts are all in this category. These are the easiest to handle and the most common.

Partial tool results — the tool returned something, but not everything. A file listing tool that returns results for some directories but errors on others. A web scraper that retrieves page content but can’t extract structured data. The agent needs to decide whether to proceed with incomplete information or block until the gap is filled.

LLM errors — the language model itself returns an error, times out, or returns a malformed response. These are rare from major providers but happen, particularly at high load or when requests exceed context limits.

Context saturation — the conversation context fills up during a long task. The agent can no longer see early tool results or instructions. This causes errors that look like reasoning failures but are actually an architecture problem.

Cascading failures — one failing tool causes the agent to make different decisions, which trigger other tool calls, which also fail. By the time the cascade stops, the agent has consumed significant resources and the error is hard to diagnose.

State divergence — the agent’s internal model of what it has done diverges from what actually happened, usually because it lost track of which tool calls succeeded.

Retry Logic: The Foundation

Most transient failures resolve on retry. The question is how to retry without making things worse.

Basic exponential backoff — wait longer between each attempt. Start with 1 second, then 2, then 4, then 8. Cap the maximum wait at something reasonable (30-60 seconds) and cap the number of retries (3-5 times). Add jitter — a random fraction of the wait time — to prevent synchronized retry storms when many agents are recovering simultaneously.

import asyncio
import random

async def retry_with_backoff(fn, max_retries=4, base_delay=1.0, max_delay=30.0):
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except TransientError as e:
            if attempt == max_retries:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = delay * 0.25 * random.random()
            await asyncio.sleep(delay + jitter)

Retry only on retryable errors — not all errors should be retried. A 404 (resource not found) won’t resolve on retry. An authentication error (401) won’t either. Only retry errors that indicate a temporary state: rate limits (429), service unavailability (503), timeouts, and connection errors. Retrying non-retryable errors wastes time and can cause side effects if the tool call has a write side.

Communicate retry state to the agent — when a retry succeeds after failures, include that context in what the agent sees. “Fetched data from API (required 3 attempts due to rate limiting)” is more useful to the agent than just the raw data, because it tells the agent something about the reliability of that data source for this task.

Circuit Breakers

Retrying aggressively against a tool that is genuinely down doesn’t just waste time — it can make the downstream problem worse (hammering an already-overloaded API) and it burns the agent’s context window with failure logs.

Circuit breakers solve this by tracking failure rates over time and “opening” the circuit (stopping all calls) when failure rate exceeds a threshold. During the open period, calls fail immediately rather than waiting for timeouts. After a cooldown, the circuit transitions to a “half-open” state that allows a test call through. If the test succeeds, the circuit closes.

from enum import Enum
from datetime import datetime, timedelta

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing fast
    HALF_OPEN = "half_open" # Testing recovery

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.last_failure_time = None

    async def call(self, fn):
        if self.state == CircuitState.OPEN:
            if datetime.now() - self.last_failure_time > timedelta(seconds=self.recovery_timeout):
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitOpenError("Circuit is open -- tool temporarily unavailable")

        try:
            result = await fn()
            if self.state == CircuitState.HALF_OPEN:
                self.reset()
            return result
        except Exception as e:
            self.record_failure()
            raise

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

    def reset(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED

In an agent context, each tool should have its own circuit breaker, and the agent should know the circuit state before deciding which tools to call. An agent that knows a tool is currently circuit-broken can skip it or route around it, rather than discovering the failure mid-task.

Fallback Chains

Some tool failures can be routed around entirely if you have alternatives. A fallback chain tries the primary tool, and on failure, tries a sequence of alternatives in order.

For an agent that needs to retrieve webpage content, the fallback chain might look like:

  1. Primary: Direct HTTP fetch with content extraction
  2. Fallback 1: Cached version from a content store
  3. Fallback 2: Summarized version from an indexer
  4. Fallback 3: Proceed without the content, noting the gap

Each fallback is lower fidelity than the one before, but the agent can still complete the task with degraded data rather than failing entirely.

The critical thing with fallback chains is logging which level was used and why. An agent that silently fell back to a cached version from three weeks ago will produce different output than one that used fresh content. If that difference matters to the task, the agent needs to communicate it.

async def fetch_with_fallback(url: str, agent_context: dict):
    # Primary: live fetch
    try:
        return await live_fetch(url), "live"
    except (TimeoutError, ConnectionError):
        pass

    # Fallback 1: cached content
    cached = await cache.get(url)
    if cached and cached.age_hours < 24:
        agent_context["degraded_sources"].append(f"{url} (cached {cached.age_hours:.0f}h ago)")
        return cached.content, "cache"

    # Fallback 2: indexed summary
    indexed = await search_index.get_summary(url)
    if indexed:
        agent_context["degraded_sources"].append(f"{url} (indexed summary only)")
        return indexed, "summary"

    # Fallback 3: mark as unavailable
    agent_context["unavailable_sources"].append(url)
    return None, "unavailable"

Checkpoint-Based Recovery

For long-running agents — tasks that take minutes or hours — retrying from the beginning after a failure is often unacceptable. The task has consumed real resources (API calls, tool usage, time) and some of it produced valid results. Checkpoint-based recovery stores the agent’s progress at defined points and resumes from the last checkpoint on failure.

A checkpoint should capture:

  • Completed steps and their outputs — what has been done and what the results were
  • Current task state — what step the agent is on, what it’s trying to accomplish
  • Accumulated context — any important information gathered during the run
  • Pending work — what still needs to be done
import json
from pathlib import Path

class AgentCheckpointer:
    def __init__(self, task_id: str, checkpoint_dir: str = "/tmp/checkpoints"):
        self.task_id = task_id
        self.checkpoint_path = Path(checkpoint_dir) / f"{task_id}.json"
        self.state = self.load_or_init()

    def load_or_init(self):
        if self.checkpoint_path.exists():
            return json.loads(self.checkpoint_path.read_text())
        return {"completed_steps": [], "results": {}, "pending_steps": [], "context": {}}

    def checkpoint(self, step_name: str, result: any, remaining_steps: list):
        self.state["completed_steps"].append(step_name)
        self.state["results"][step_name] = result
        self.state["pending_steps"] = remaining_steps
        self.checkpoint_path.write_text(json.dumps(self.state, indent=2))

    def is_completed(self, step_name: str) -> bool:
        return step_name in self.state["completed_steps"]

    def get_result(self, step_name: str):
        return self.state["results"].get(step_name)

    def cleanup(self):
        self.checkpoint_path.unlink(missing_ok=True)

Using the checkpointer in a multi-step agent:

async def run_research_task(task_id: str, query: str):
    cp = AgentCheckpointer(task_id)

    if not cp.is_completed("web_search"):
        results = await web_search(query)
        cp.checkpoint("web_search", results, ["summarize", "extract_facts", "write_report"])
    else:
        results = cp.get_result("web_search")

    if not cp.is_completed("summarize"):
        summary = await llm_summarize(results)
        cp.checkpoint("summarize", summary, ["extract_facts", "write_report"])
    else:
        summary = cp.get_result("summarize")

    # ... continue for remaining steps

    cp.cleanup()  # Remove checkpoint on successful completion

A task that fails at the “write_report” step resumes there, with all the search and summarization results already available. Only the failed step reruns.

Handling Context Saturation

Context saturation is an architecture problem that error handling can only partially mitigate. When the agent’s context fills up with tool results and intermediate outputs, its ability to follow original instructions and make coherent decisions degrades.

Short-term recovery options:

  • Summarize and compress — when context reaches a threshold, have the agent summarize what it has learned so far and discard raw tool outputs from earlier steps
  • Sliding window — keep only the most recent N tool results in context; store older ones in a retrieval store the agent can query explicitly
  • Task decomposition — split the task into sub-agents, each handling a bounded portion, with a coordinator managing the outputs

Long-term, context saturation usually means the task scope is too large for a single agent run. Redesign the task to fit within available context, or build a multi-agent architecture where no single agent needs to hold everything in context simultaneously.

Defining Degraded Modes

The last and most important pattern is intentional: define what a degraded but still useful response looks like for every critical agent task.

For each agent capability, ask: if the primary path fails, what can we still deliver? Options include:

  • Reduced scope — complete the task but with fewer results (“I found 5 sources rather than 10 due to API limits”)
  • Lower confidence — complete the task but flag uncertainty (“Based on cached data from 6 hours ago”)
  • Human handoff — flag the task for human review rather than returning a potentially wrong automated result
  • Partial completion — return what was completed and explicitly mark what couldn’t be done

Build these modes explicitly, not as afterthoughts. An agent that always returns a result, even a degraded one, with clear metadata about what degraded it, is far more useful and auditable than one that fails silently or returns hallucinated content when its tools are unavailable.

The frame shift is: instead of “what happens if this fails,” design for “what should be delivered in every possible outcome.” Production agents are reliable not because they never fail, but because their failure modes are as well-designed as their success modes.