← Back to DevBytes

Error Recovery Patterns with AutoGen: Complete Guide

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:

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:

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

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.

— Ad —

Google AdSense will appear here after approval

← Back to all articles