Introduction to Error Recovery Patterns with AutoGen
AutoGen, Microsoft's multi-agent conversation framework, empowers developers to build sophisticated LLM-powered applications where agents collaborate to solve complex problems. However, when multiple agents interact with external tools, APIs, and language models, failures are inevitable. Network timeouts, malformed tool outputs, rate limits, and unexpected LLM responses can all derail an otherwise productive conversation. This is where error recovery patterns come in — a set of strategies that allow your AutoGen agents to detect, handle, and recover from failures gracefully without crashing the entire workflow.
What Are Error Recovery Patterns?
Error recovery patterns are reusable strategies for detecting failures during agent execution and responding in ways that allow the workflow to continue or fail safely. In the context of AutoGen, these patterns address several common failure modes:
- LLM API failures — rate limits, authentication errors, or transient network issues when calling the model provider.
- Tool execution failures — exceptions raised inside functions registered as tools, such as invalid inputs or unavailable services.
- Invalid or malformed outputs — the model returns content that doesn't match the expected schema or format.
- Conversation deadlocks — agents get stuck in loops or fail to converge on a solution.
- Resource exhaustion — token limits exceeded or maximum conversation rounds reached.
Each of these failure modes requires a different recovery approach. A robust AutoGen application combines multiple patterns to handle the full spectrum of possible errors.
Why Error Recovery Matters
Without explicit error handling, a single failed tool call or API timeout can terminate an entire multi-agent conversation, wasting compute resources and losing valuable context. In production environments, this translates directly to poor user experience, increased costs from retries, and unreliable behavior. Error recovery patterns matter because they:
- Improve reliability — workflows continue functioning even when individual components fail.
- Reduce costs — intelligent retry and fallback strategies avoid unnecessary re-computation.
- Preserve context — failed operations don't destroy the conversation history that agents have built up.
- Enable observability — structured error handling makes it easier to log, monitor, and debug failures.
- Support graceful degradation — when full recovery isn't possible, the system can still return a useful partial result.
Setting Up Your Environment
Before diving into the patterns, make sure you have AutoGen installed and configured. The examples below use AutoGen version 0.4.x (the agentic framework with async support).
pip install "autogen-agentchat" "autogen-ext[openai]"
Set up your API key as an environment variable:
import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
Pattern 1: Retry with Exponential Backoff
The most fundamental error recovery pattern is retrying failed operations with increasing delays between attempts. This is especially effective for transient failures like rate limits or temporary network issues. AutoGen's ChatCompletionClient supports retry configuration, but you can also wrap individual tool calls with custom retry logic.
import asyncio
import logging
from autogen_ext.models.openai import OpenAIChatCompletionClient
logger = logging.getLogger(__name__)
async def call_with_retry(func, *args, max_retries=3, base_delay=1.0, **kwargs):
"""
Call an async function with exponential backoff retry.
Args:
func: The async function to call.
max_retries: Maximum number of retry attempts.
base_delay: Initial delay in seconds before first retry.
Returns:
The result of the function call.
Raises:
The last exception if all retries fail.
"""
last_exception = None
for attempt in range(max_retries + 1):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt == max_retries:
logger.error(f"All {max_retries} retries exhausted. Last error: {e}")
raise
delay = base_delay * (2 ** attempt)
logger.warning(
f"Attempt {attempt + 1} failed with: {e}. "
f"Retrying in {delay} seconds..."
)
await asyncio.sleep(delay)
raise last_exception
# Configure the model client with built-in retry
model_client = OpenAIChatCompletionClient(
model="gpt-4o",
max_retries=3,
retry_timeout=30,
)
This helper function can wrap any async operation, including tool calls and model invocations. The exponential backoff ensures you don't overwhelm a struggling service while still giving it time to recover.
Pattern 2: Safe Tool Execution Wrappers
When agents call external tools, those tools can fail for many reasons. Instead of letting exceptions propagate and crash the conversation, wrap your tool functions to catch errors and return structured error messages that the agent can understand and act upon.
import json
import functools
from typing import Any, Callable
def safe_tool(func: Callable) -> Callable:
"""
Decorator that wraps a tool function to catch exceptions
and return a structured error message instead of crashing.
The agent receives the error as a normal tool response,
allowing it to reason about the failure and try alternatives.
"""
@functools.wraps(func)
async def wrapper(*args, **kwargs) -> str:
try:
result = await func(*args, **kwargs)
return json.dumps({"status": "success", "data": result})
except ValueError as e:
return json.dumps({
"status": "error",
"error_type": "validation",
"message": str(e),
"suggestion": "Check the input parameters and try again."
})
except TimeoutError as e:
return json.dumps({
"status": "error",
"error_type": "timeout",
"message": str(e),
"suggestion": "The service is slow. Try a simpler query."
})
except Exception as e:
return json.dumps({
"status": "error",
"error_type": "unexpected",
"message": str(e),
"suggestion": "An unexpected error occurred. Try a different approach."
})
return wrapper
# Example usage: wrapping a database query tool
@safe_tool
async def query_database(sql: str) -> dict:
"""Execute a SQL query and return results."""
if not sql.strip():
raise ValueError("SQL query cannot be empty")
# Simulate a database call
if "DROP" in sql.upper():
raise ValueError("Destructive operations are not allowed")
return {"rows": [{"id": 1, "name": "example"}], "count": 1}
By returning errors as structured JSON, the agent can parse the error type and suggestion, then decide whether to retry with different parameters, ask the user for clarification, or try an alternative approach entirely.
Pattern 3: Fallback Model Chains
Sometimes the primary model is unavailable or returns poor results. A fallback chain lets you try a cheaper or more reliable model when the primary one fails. This pattern is particularly useful for cost optimization — use an expensive model first, but fall back to a cheaper one if it's unavailable.
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ChatCompletionClient
from typing import List, Optional
class FallbackModelClient(ChatCompletionClient):
"""
A model client that tries multiple underlying clients in order,
falling back to the next one if the current one fails.
"""
def __init__(self, clients: List[ChatCompletionClient]):
self._clients = clients
self._current_index = 0
async def create(self, messages, **kwargs):
last_error = None
for i, client in enumerate(self._clients):
try:
result = await client.create(messages, **kwargs)
if i != self._current_index:
logger.info(f"Fell back to client {i}, updating primary")
self._current_index = i
return result
except Exception as e:
logger.warning(f"Client {i} failed: {e}")
last_error = e
raise last_error
def _create_args(self):
return self._clients[self._current_index]._create_args()
async def close(self):
for client in self._clients:
await client.close()
# Build a fallback chain: GPT-4o -> GPT-4o-mini -> GPT-3.5-turbo
fallback_client = FallbackModelClient([
OpenAIChatCompletionClient(model="gpt-4o", max_retries=1),
OpenAIChatCompletionClient(model="gpt-4o-mini", max_retries=1),
OpenAIChatCompletionClient(model="gpt-3.5-turbo", max_retries=2),
])
This pattern ensures your application remains available even when the primary model provider experiences issues, at the cost of potentially lower quality responses from fallback models.
Pattern 4: Circuit Breaker for External Services
When an external service is down, repeatedly retrying calls wastes resources and can make the situation worse. A circuit breaker pattern monitors failure rates and temporarily stops calling a failing service, giving it time to recover before trying again.
import time
from enum import Enum
from typing import Optional
class CircuitState(Enum):
CLOSED = "closed" # Normal operation, calls go through
OPEN = "open" # Service is failing, calls are blocked
HALF_OPEN = "half_open" # Testing if service has recovered
class CircuitBreaker:
"""
Circuit breaker that tracks failures and blocks calls
when a service is unhealthy.
Args:
failure_threshold: Number of failures before opening the circuit.
recovery_timeout: Seconds to wait before trying again (half-open).
success_threshold: Consecutive successes in half-open to close circuit.
"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
success_threshold: int = 2,
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.success_threshold = success_threshold
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
self._last_failure_time: Optional[float] = None
@property
def state(self) -> CircuitState:
if self._state == CircuitState.OPEN:
if self._last_failure_time and \
time.time() - self._last_failure_time > self.recovery_timeout:
self._state = CircuitState.HALF_OPEN
self._success_count = 0
logger.info("Circuit breaker entering HALF_OPEN state")
return self._state
def record_success(self):
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.success_threshold:
self._state = CircuitState.CLOSED
self._failure_count = 0
logger.info("Circuit breaker CLOSED - service recovered")
elif self._state == CircuitState.CLOSED:
self._failure_count = 0
def record_failure(self):
self._last_failure_time = time.time()
if self._state == CircuitState.HALF_OPEN:
self._state = CircuitState.OPEN
logger.warning("Circuit breaker OPEN - service still failing")
elif self._state == CircuitState.CLOSED:
self._failure_count += 1
if self._failure_count >= self.failure_threshold:
self._state = CircuitState.OPEN
logger.warning(
f"Circuit breaker OPEN after {self._failure_count} failures"
)
def can_execute(self) -> bool:
return self.state != CircuitState.OPEN
# Usage with a tool
api_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
@safe_tool
async def call_external_api(endpoint: str, params: dict) -> dict:
"""Call an external API with circuit breaker protection."""
if not api_breaker.can_execute():
raise RuntimeError(
"External API is currently unavailable (circuit open). "
"Please try again later or use cached data."
)
try:
# Simulate API call
result = await make_http_request(endpoint, params)
api_breaker.record_success()
return result
except Exception as e:
api_breaker.record_failure()
raise
The circuit breaker prevents cascading failures by stopping calls to a broken service, then gradually testing recovery through the half-open state before fully restoring normal operation.
Pattern 5: Conversation-Level Error Recovery with Reflection
Sometimes the error isn't a crash but a logical failure — the agent produces incorrect or unhelpful output. A reflection pattern lets an agent evaluate its own output and retry if the quality is insufficient. This is particularly powerful when combined with a dedicated critic agent.
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_agentchat.messages import TextMessage
async def task_with_reflection(
task: str,
model_client,
max_reflection_rounds: int = 3,
) -> str:
"""
Execute a task with a worker agent and a critic agent.
The critic evaluates the output and requests revisions
if the quality is insufficient.
The critic signals satisfaction by including "APPROVED" in its response.
"""
worker = AssistantAgent(
name="worker",
model_client=model_client,
system_message=(
"You are a task execution agent. Complete the assigned task "
"carefully. If the critic provides feedback, revise your work "
"accordingly. Be concise and accurate."
),
)
critic = AssistantAgent(
name="critic",
model_client=model_client,
system_message=(
"You are a quality critic. Evaluate the worker's output for "
"correctness, completeness, and clarity. If the output is "
"acceptable, respond with 'APPROVED' followed by a brief reason. "
"If not, provide specific, actionable feedback for improvement. "
"Do NOT do the task yourself."
),
)
termination = TextMentionTermination("APPROVED") | MaxMessageTermination(
max_messages=max_reflection_rounds * 2 + 1
)
team = RoundRobinGroupChat(
participants=[worker, critic],
termination_condition=termination,
)
result = await team.run(task=task)
# Extract the final approved output
messages = result.messages
worker_outputs = [
m for m in messages
if isinstance(m, TextMessage) and m.source == "worker"
]
if worker_outputs:
return worker_outputs[-1].content
return "Task could not be completed within the reflection limit."
# Run a task with reflection-based error recovery
async def main():
from autogen_ext.models.openai import OpenAIChatCompletionClient
client = OpenAIChatCompletionClient(model="gpt-4o")
result = await task_with_reflection(
task="Write a Python function that safely parses a JSON string "
"and returns a default value on failure. Include docstrings "
"and type hints.",
model_client=client,
max_reflection_rounds=3,
)
print("Final output:", result)
await client.close()
asyncio.run(main())
This pattern effectively turns logical errors into recoverable conditions. Instead of accepting a flawed first attempt, the system iterates until the output meets quality standards or the maximum rounds are exhausted.
Pattern 6: Timeout and Resource Guardrails
Agents can sometimes get stuck in long-running operations or infinite loops. Timeout guardrails ensure that no single operation can block the entire system indefinitely. AutoGen provides termination conditions, but you should also add timeouts at the tool and task levels.
import asyncio
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination
async def run_with_timeout(team, task: str, timeout_seconds: float = 120):
"""
Run a team task with an overall timeout.
If the timeout is exceeded, returns a partial result with
whatever messages were produced before the timeout.
"""
try:
result = await asyncio.wait_for(
team.run(task=task),
timeout=timeout_seconds,
)
return {
"status": "completed",
"messages": result.messages,
"message_count": len(result.messages),
}
except asyncio.TimeoutError:
logger.warning(
f"Task timed out after {timeout_seconds}s. "
"Returning partial results."
)
return {
"status": "timeout",
"messages": [],
"message_count": 0,
"error": f"Task exceeded {timeout_seconds} second timeout",
}
# Tool-level timeout wrapper
async def tool_with_timeout(func, *args, timeout=30, **kwargs):
"""Wrap a tool call with a timeout to prevent hanging."""
try:
return await asyncio.wait_for(
func(*args, **kwargs),
timeout=timeout,
)
except asyncio.TimeoutError:
return {
"status": "error",
"error_type": "timeout",
"message": f"Tool '{func.__name__}' timed out after {timeout}s",
"suggestion": "Try simplifying the request or increasing the timeout."
}
Combining task-level and tool-level timeouts creates a defense in depth against runaway operations. The task timeout catches loops between agents, while tool timeouts catch individual operations that hang on external dependencies.
Pattern 7: Checkpointing and State Recovery
For long-running multi-agent workflows, losing all progress to a single failure is costly. Checkpointing saves conversation state at key points so you can resume from the last successful checkpoint rather than starting over.
import json
import os
from datetime import datetime
from autogen_agentchat.messages import BaseChatMessage
class ConversationCheckpoint:
"""
Saves and restores conversation state to disk.
Allows recovery from the last checkpoint after a failure.
"""
def __init__(self, checkpoint_dir: str = "checkpoints"):
self.checkpoint_dir = checkpoint_dir
os.makedirs(checkpoint_dir, exist_ok=True)
def save(self, session_id: str, messages: list, metadata: dict = None):
"""Save current conversation state to a checkpoint file."""
checkpoint = {
"session_id": session_id,
"timestamp": datetime.now().isoformat(),
"messages": [
{
"source": m.source,
"content": m.content if hasattr(m, 'content') else str(m),
"type": type(m).__name__,
}
for m in messages
],
"metadata": metadata or {},
}
path = os.path.join(self.checkpoint_dir, f"{session_id}.json")
with open(path, "w") as f:
json.dump(checkpoint, f, indent=2)
logger.info(f"Checkpoint saved: {path} ({len(messages)} messages)")
def load(self, session_id: str) -> dict:
"""Load the most recent checkpoint for a session."""
path = os.path.join(self.checkpoint_dir, f"{session_id}.json")
if not os.path.exists(path):
return None
with open(path, "r") as f:
return json.load(f)
def exists(self, session_id: str) -> bool:
"""Check if a checkpoint exists for the given session."""
return os.path.exists(
os.path.join(self.checkpoint_dir, f"{session_id}.json")
)
# Integration with a team run
async def run_with_checkpoints(
team,
task: str,
session_id: str,
checkpoint_every: int = 5,
):
"""
Run a team task with periodic checkpointing.
If a previous checkpoint exists, resume from there.
"""
checkpoint_mgr = ConversationCheckpoint()
# Check for existing checkpoint
if checkpoint_mgr.exists(session_id):
saved = checkpoint_mgr.load(session_id)
logger.info(
f"Resuming from checkpoint ({len(saved['messages'])} messages)"
)
# Reconstruct context from checkpoint
context = "\n".join(
f"{m['source']}: {m['content']}" for m in saved["messages"]
)
task = f"Previous conversation:\n{context}\n\nContinue the task: {task}"
# Run with checkpointing
result = await team.run(task=task)
# Save final state
checkpoint_mgr.save(
session_id=session_id,
messages=result.messages,
metadata={"task": task, "status": "completed"},
)
return result
Checkpointing is especially valuable for workflows that involve expensive operations like web scraping, data analysis, or code generation. By saving state periodically, you can resume from the last good state rather than repeating the entire workflow.
Pattern 8: Comprehensive Error Recovery Agent
The final pattern combines several of the above strategies into a single resilient agent setup. This example shows how to build an agent with safe tools, retry logic, and a reflection loop all working together.
import asyncio
import logging
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import (
TextMentionTermination,
MaxMessageTermination,
)
from autogen_ext.models.openai import OpenAIChatCompletionClient
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# --- Safe tools with retry and circuit breaker ---
api_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
@safe_tool
async def fetch_weather(city: str) -> dict:
"""Fetch weather data for a city."""
if not api_breaker.can_execute():
raise RuntimeError("Weather service is temporarily unavailable")
try:
# Simulate API call with possible failure
if city.lower() == "error":
raise ConnectionError("Simulated API failure")
result = await call_with_retry(
simulate_weather_api, city, max_retries=2, base_delay=0.5
)
api_breaker.record_success()
return result
except Exception as e:
api_breaker.record_failure()
raise
async def simulate_weather_api(city: str) -> dict:
"""Simulated weather API for demonstration."""
await asyncio.sleep(0.1)
return {"city": city, "temp": 22, "condition": "sunny"}
# --- Build the resilient agent ---
async def build_resilient_agent():
model_client = OpenAIChatCompletionClient(
model="gpt-4o",
max_retries=3,
)
agent = AssistantAgent(
name="weather_assistant",
model_client=model_client,
tools=[fetch_weather],
system_message=(
"You are a weather assistant. Use the fetch_weather tool to "
"get weather data. If a tool returns an error, analyze the "
"error message and suggestion, then either retry with "
"different parameters or inform the user about the issue. "
"Always be helpful and transparent about any problems."
),
)
return agent, model_client
# --- Run with full error recovery ---
async def main():
agent, client = await build_resilient_agent()
try:
# Normal request
result = await agent.run(task="What's the weather in Paris?")
print("Result 1:", result.messages[-1].content)
# Request that triggers error recovery
result = await agent.run(task="What's the weather in Error?")
print("Result 2:", result.messages[-1].content)
except Exception as e:
logger.error(f"Unhandled error in main: {e}")
finally:
await client.close()
asyncio.run(main())
This comprehensive example demonstrates how the patterns compose: the safe tool wrapper catches exceptions, the circuit breaker prevents cascading failures, the retry logic handles transient issues, and the agent's system prompt instructs it to reason about errors and respond appropriately.
Best Practices
- Layer your error handling. Don't rely on a single pattern. Combine retry logic at the tool level, circuit breakers at the service level, timeouts at the task level, and reflection at the agent level for defense in depth.
- Always log errors with context. Include the agent name, tool name, input parameters, and conversation state in your error logs. This makes debugging multi-agent failures much easier.
- Return errors as data, not exceptions. When tools fail, return structured error messages that agents can parse and reason about. This lets the agent decide how to respond rather than crashing the conversation.
- Set reasonable limits. Always configure maximum retries, maximum conversation rounds, and timeouts. Without limits, a stuck agent can run indefinitely, consuming resources and costs.
- Test failure scenarios explicitly. Write tests that simulate API failures, timeouts, and malformed inputs. Error recovery code that's never tested in failure conditions will fail when you need it most.
- Use idempotent operations. When retrying tool calls, ensure the operation is idempotent or can safely be repeated. Non-idempotent operations like payment processing require more careful retry logic.
- Monitor and alert on error rates. In production, track error rates per tool, per agent, and per model. Sudden spikes can indicate upstream issues before they become visible to users.
- Graceful degradation over complete failure. When full recovery isn't possible, return the best partial result available along with an explanation of what went wrong. Users prefer a useful partial answer over a crash.
- Keep system prompts updated with error handling instructions. Explicitly tell agents how to behave when tools fail. Without guidance, agents may hallucinate responses or repeat the same failing call.
- Close resources properly. Always use try/finally blocks or async context managers to close model clients and clean up resources, even when errors occur.
Conclusion
Error recovery is not an optional add-on for production AutoGen applications — it is a core architectural concern. By combining retry logic with exponential backoff, safe tool wrappers, fallback model chains, circuit breakers, reflection loops, timeouts, and checkpointing, you can build multi-agent systems that remain reliable even when individual components fail. The key insight is that errors in multi-agent systems are inevitable, but how your system responds to them determines whether users experience a frustrating crash or a seamless recovery. Start with the simplest patterns — safe tool wrappers and retry logic — and progressively add more sophisticated strategies like circuit breakers and checkpointing as your application's complexity and reliability requirements grow. With these patterns in place, your AutoGen agents will be well-equipped to handle the unpredictable realities of production environments while delivering consistent, dependable results.