Introduction to Error Recovery Patterns with CrewAI
Building multi-agent systems with CrewAI is powerful, but production-grade applications demand resilience. When agents call external APIs, parse unstructured data, or chain complex reasoning steps, failures are inevitable. Error recovery patterns are the strategies and code structures you implement to detect, handle, and recover from these failures gracefully — without crashing your entire crew or producing garbage output.
In this guide, we'll explore practical error recovery patterns tailored specifically for CrewAI workflows. You'll learn how to wrap agent tasks with retry logic, validate intermediate outputs, implement fallback agents, and build self-healing crews that can adapt when things go wrong.
Why Error Recovery Matters in CrewAI
CrewAI orchestrates multiple LLM-powered agents that collaborate to complete tasks. Each agent relies on an LLM provider, optional tools, and structured prompts. Any of these components can fail:
- LLM API failures: Rate limits, timeouts, or transient network errors from OpenAI, Anthropic, or local models.
- Tool execution errors: A web scraper returns empty HTML, a database query times out, or a file is missing.
- Output format issues: An agent returns free-form text when your downstream task expects JSON.
- Reasoning failures: The agent hallucinates, loops, or produces an incomplete answer.
- Cascading failures: One agent's bad output poisons every downstream agent in the crew.
Without recovery patterns, a single failure can halt the entire pipeline or, worse, silently propagate incorrect results. Robust error recovery transforms fragile demos into reliable production systems.
Pattern 1: Retry with Exponential Backoff
The simplest and most effective pattern is retrying failed operations with exponential backoff. This handles transient errors like rate limits and network blips. CrewAI tasks delegate to underlying LLM calls, so wrapping the crew execution in a retry loop is a good first line of defense.
Basic Retry Wrapper
import time
import logging
from crewai import Crew, Task, Agent
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def run_crew_with_retry(crew: Crew, max_retries: int = 3, base_delay: float = 2.0):
"""Execute a Crew with exponential backoff retry logic."""
last_exception = None
for attempt in range(1, max_retries + 1):
try:
logger.info(f"Crew execution attempt {attempt}/{max_retries}")
result = crew.kickoff()
return result
except Exception as e:
last_exception = e
delay = base_delay * (2 ** (attempt - 1))
logger.warning(f"Attempt {attempt} failed: {e}. Retrying in {delay}s...")
if attempt < max_retries:
time.sleep(delay)
raise RuntimeError(f"Crew failed after {max_retries} attempts") from last_exception
# Usage
researcher = Agent(
role="Research Analyst",
goal="Find accurate information about the topic",
backstory="An expert researcher with attention to detail.",
verbose=True
)
research_task = Task(
description="Research the latest trends in renewable energy.",
expected_output="A summary report with key findings.",
agent=researcher
)
crew = Crew(agents=[researcher], tasks=[research_task])
final_result = run_crew_with_retry(crew, max_retries=4, base_delay=1.5)
print(final_result)
This pattern catches any exception raised during crew execution — including LLM API errors — and retries with increasing delays. The exponential backoff gives rate-limited APIs time to recover.
Pattern 2: Output Validation and Self-Correction
Sometimes the crew completes without throwing an exception, but the output is malformed or incomplete. A validation layer lets you detect bad output and re-prompt the agent to fix it.
Validating Structured Output
import json
from crewai import Crew, Task, Agent
def validate_json_output(output: str) -> dict:
"""Attempt to parse and validate crew output as JSON."""
try:
data = json.loads(output)
except json.JSONDecodeError as e:
raise ValueError(f"Output is not valid JSON: {e}")
if not isinstance(data, dict):
raise ValueError("Output must be a JSON object")
required_keys = {"title", "summary", "sources"}
missing = required_keys - data.keys()
if missing:
raise ValueError(f"Missing required keys: {missing}")
return data
def run_with_validation(crew: Crew, validator, max_attempts: int = 3):
"""Run crew, validate output, and re-prompt on failure."""
context = ""
for attempt in range(1, max_attempts + 1):
if context:
# Inject feedback into the task description for self-correction
for task in crew.tasks:
task.description += f"\n\nPrevious attempt failed validation: {context}. Please fix and retry."
result = crew.kickoff()
try:
return validator(str(result))
except ValueError as e:
context = str(e)
print(f"Validation attempt {attempt} failed: {context}")
raise RuntimeError(f"Output failed validation after {max_attempts} attempts")
# Usage with a JSON-producing agent
analyst = Agent(
role="Data Analyst",
goal="Produce structured JSON reports",
backstory="A meticulous analyst who always returns valid JSON.",
verbose=True
)
task = Task(
description="Analyze Q3 sales data and return a JSON object with 'title', 'summary', and 'sources' keys.",
expected_output="A JSON object with title, summary, and sources.",
agent=analyst
)
crew = Crew(agents=[analyst], tasks=[task])
report = run_with_validation(crew, validate_json_output)
print(report)
This pattern is especially useful when downstream systems consume the crew's output programmatically. The validator acts as a gatekeeper, ensuring only well-formed data passes through.
Pattern 3: Fallback Agents and Model Degradation
Different LLMs have different strengths, latency profiles, and failure modes. A fallback pattern tries a primary agent first, and if it fails, delegates to a backup agent using a different model or simpler prompt.
Implementing a Fallback Crew
from crewai import Crew, Task, Agent
from functools import wraps
def with_fallback(primary_agent, fallback_agent, task_template: str):
"""Create a task that tries the primary agent, falling back on failure."""
def execute_task(inputs: dict = None):
primary_task = Task(
description=task_template,
expected_output="A complete and accurate response.",
agent=primary_agent
)
try:
crew = Crew(agents=[primary_agent], tasks=[primary_task])
result = crew.kickoff(inputs=inputs)
if not str(result).strip():
raise ValueError("Empty output from primary agent")
return str(result), "primary"
except Exception as e:
print(f"Primary agent failed ({e}), switching to fallback...")
fallback_task = Task(
description=f"{task_template}\n\nNote: The primary agent failed. Provide a simpler but correct response.",
expected_output="A correct response, even if less detailed.",
agent=fallback_agent
)
crew = Crew(agents=[fallback_agent], tasks=[fallback_task])
result = crew.kickoff(inputs=inputs)
return str(result), "fallback"
return execute_task
# Define agents with different models
primary_agent = Agent(
role="Senior Strategist",
goal="Provide detailed strategic analysis",
backstory="A seasoned strategist with deep industry knowledge.",
llm="gpt-4o",
verbose=True
)
fallback_agent = Agent(
role="Junior Analyst",
goal="Provide reliable baseline analysis",
backstory="A dependable analyst who focuses on accuracy.",
llm="gpt-4o-mini",
verbose=True
)
task_template = "Analyze the competitive landscape for {company} and identify three key opportunities."
execute = with_fallback(primary_agent, fallback_agent, task_template)
result, source = execute(inputs={"company": "Acme Corp"})
print(f"Result from {source}:\n{result}")
The fallback pattern is powerful because it isolates failures. If the premium model is unavailable or produces poor output, the crew degrades gracefully to a cheaper, more reliable model instead of failing entirely.
Pattern 4: Tool-Level Error Handling
CrewAI agents use tools to interact with the outside world. Tool failures are a common source of crew crashes. Wrap your custom tools with error handling so they return informative error messages instead of raising exceptions.
Resilient Custom Tool
from crewai.tools import tool
import requests
@tool("Fetch Web Page")
def fetch_web_page(url: str) -> str:
"""Fetch the content of a web page given its URL.
Args:
url: The URL of the page to fetch.
"""
try:
response = requests.get(url, timeout=10, headers={"User-Agent": "CrewAI-Bot/1.0"})
response.raise_for_status()
if not response.text.strip():
return "ERROR: Page returned empty content."
return response.text[:5000] # Truncate to avoid token overflow
except requests.Timeout:
return "ERROR: Request timed out. Try a different URL or retry later."
except requests.ConnectionError:
return "ERROR: Could not connect to the URL. Check if the site is reachable."
except requests.HTTPError as e:
return f"ERROR: HTTP {response.status_code} - {e}"
except Exception as e:
return f"ERROR: Unexpected error fetching page: {e}"
# Agent using the resilient tool
researcher = Agent(
role="Web Researcher",
goal="Extract useful information from web pages",
backstory="A researcher skilled at navigating the web.",
tools=[fetch_web_page],
verbose=True
)
task = Task(
description="Fetch https://example.com and summarize its main content.",
expected_output="A concise summary of the page content.",
agent=researcher
)
crew = Crew(agents=[researcher], tasks=[task])
print(crew.kickoff())
By returning error strings instead of raising exceptions, the tool gives the agent a chance to reason about the failure and try an alternative approach — like fetching a different URL or adjusting its query.
Pattern 5: Circuit Breaker for Repeated Failures
If an agent or tool fails repeatedly, continuing to retry wastes resources and time. A circuit breaker pattern tracks failure counts and temporarily stops calling the failing component after a threshold is reached.
import time
from datetime import datetime, timedelta
class CircuitBreaker:
"""Simple circuit breaker for agent or tool calls."""
def __init__(self, failure_threshold: int = 3, reset_timeout: float = 60.0):
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "closed" # closed, open, half-open
def record_failure(self):
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.failure_count >= self.failure_threshold:
self.state = "open"
print(f"Circuit breaker OPENED after {self.failure_count} failures")
def record_success(self):
self.failure_count = 0
self.state = "closed"
def can_execute(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
if datetime.now() - self.last_failure_time > timedelta(seconds=self.reset_timeout):
self.state = "half-open"
print("Circuit breaker entering HALF-OPEN state")
return True
return False
return True # half-open allows a test call
# Usage wrapper
breaker = CircuitBreaker(failure_threshold=3, reset_timeout=30.0)
def execute_with_breaker(crew, inputs=None):
if not breaker.can_execute():
raise RuntimeError("Circuit breaker is open. Agent temporarily unavailable.")
try:
result = crew.kickoff(inputs=inputs)
breaker.record_success()
return result
except Exception as e:
breaker.record_failure()
raise
The circuit breaker prevents cascading failures and gives external systems time to recover. It's especially useful in long-running crews that process many items in a loop.
Pattern 6: Checkpointing Long-Running Crews
For crews with many sequential tasks, losing all progress to a late-stage failure is painful. Checkpointing saves intermediate results so you can resume from the last successful task.
import json
import os
from crewai import Crew, Task, Agent
def run_with_checkpoints(crew: Crew, checkpoint_file: str = "crew_checkpoint.json"):
"""Run crew tasks sequentially, saving after each successful task."""
completed = {}
if os.path.exists(checkpoint_file):
with open(checkpoint_file, "r") as f:
completed = json.load(f)
print(f"Resuming from checkpoint: {len(completed)} tasks already done")
for i, task in enumerate(crew.tasks):
task_key = f"task_{i}"
if task_key in completed:
print(f"Skipping completed task {i}: {task.description[:50]}...")
task.output = completed[task_key]
continue
try:
# Run a single-task crew for this task
single_crew = Crew(agents=[task.agent], tasks=[task])
result = single_crew.kickoff()
completed[task_key] = str(result)
with open(checkpoint_file, "w") as f:
json.dump(completed, f, indent=2)
print(f"Checkpoint saved after task {i}")
except Exception as e:
print(f"Task {i} failed: {e}. Checkpoint preserved at task {i - 1}.")
raise
return completed
# Usage
writer = Agent(role="Writer", goal="Write content", backstory="A skilled writer.")
editor = Agent(role="Editor", goal="Edit content", backstory="A meticulous editor.")
tasks = [
Task(description="Write a draft article about AI.", expected_output="A draft article.", agent=writer),
Task(description="Edit and polish the draft article.", expected_output="A polished article.", agent=editor),
]
crew = Crew(agents=[writer, editor], tasks=tasks)
results = run_with_checkpoints(crew)
print(results)
Checkpointing is invaluable for crews that take minutes or hours to run. If task 8 of 10 fails, you fix the issue and resume from task 8 without re-running tasks 1 through 7.
Best Practices for Error Recovery in CrewAI
- Layer your defenses: Combine retry logic, output validation, and fallback agents. No single pattern covers all failure modes.
- Log everything: Use Python's logging module to record failures, retries, and fallbacks. This data is essential for debugging production issues.
- Set timeouts: Always configure timeouts on LLM calls and tool executions. A hung agent is worse than a failed one.
- Keep prompts explicit about output format: The more specific your
expected_output, the less likely agents produce unusable results. - Test failure paths: Deliberately break tools and APIs in staging to verify your recovery patterns actually work.
- Avoid infinite loops: Always cap retries and self-correction attempts. An agent stuck in a retry loop burns tokens and time.
- Use structured output when available: CrewAI supports Pydantic-based output schemas. Leverage them to reduce format errors.
- Monitor token usage: Retries and fallbacks consume extra tokens. Track usage to avoid surprise costs.
- Design idempotent tasks: If a task might re-run, ensure re-execution doesn't cause side effects like duplicate database entries.
- Graceful degradation over hard failure: When full recovery isn't possible, return partial results with clear metadata about what succeeded and what didn't.
Conclusion
Error recovery is what separates experimental CrewAI prototypes from production-ready multi-agent systems. By combining retry logic with exponential backoff, output validation, fallback agents, resilient tools, circuit breakers, and checkpointing, you can build crews that withstand the chaos of real-world LLM APIs and external dependencies. Start with the retry wrapper for immediate gains, then layer in validation and fallback patterns as your workflows grow in complexity. The key insight is that failures are not exceptional — they are expected — and your architecture should treat them as a normal part of agent orchestration. With these patterns in place, your CrewAI applications will be resilient, debuggable, and ready for production traffic.