Building Reliable Multi-Agent Systems: Error Handling and Fallbacks
Building Reliable Multi-Agent Systems: Error Handling and Fallbacks
Multi-agent systems promise compound intelligence — specialized agents collaborating to solve complex problems. But compound intelligence brings compound failure modes. A chain of 5 agents, each with 95% reliability, has only a 77% chance of completing without any failure. Error handling isn’t optional in multi-agent systems — it’s the architecture.
Failure Modes in Multi-Agent Systems
Understanding what can go wrong is the first step to building resilience:
1. Agent Output Failures
- Malformed output: Agent returns invalid JSON, missing required fields, or wrong data types
- Hallucinated content: Agent generates plausible but incorrect information
- Off-task drift: Agent ignores instructions and produces irrelevant content
- Truncated output: Response hit the token limit mid-generation
2. Coordination Failures
- Deadlock: Two agents waiting on each other’s output
- Infinite loops: Agent A keeps delegating back to Agent B and vice versa
- Stale context: Agent receives outdated information from a predecessor
- Cascade failures: One agent’s small error compounds downstream
3. Infrastructure Failures
- LLM API timeouts: The underlying model API is slow or unavailable
- Context overflow: Combined agent outputs exceed the context window
- Rate limiting: Too many concurrent API calls trigger throttling
- Tool failures: External API calls (search, database, file system) fail
Error Handling Patterns
Pattern 1: Validation Gates
Implement structured output validation between every agent handoff:
class ValidationGate:
def __init__(self, schema: BaseModel):
self.schema = schema
def validate(self, agent_output: str) -> ValidationResult:
try:
data = json.loads(agent_output)
validated = self.schema(**data)
return ValidationResult(valid=True, data=validated)
except json.JSONDecodeError:
return ValidationResult(
valid=False,
error="Invalid JSON",
recovery="retry_with_fix_prompt"
)
except ValidationError as e:
return ValidationResult(
valid=False,
error=str(e),
recovery="provide_schema_example"
)
# Usage between agents
gate = ValidationGate(ResearchResult)
result = gate.validate(researcher_agent.output)
if not result.valid:
orchestrator.request_fix(researcher_agent, result.error)Pattern 2: Retry with Escalation
Don’t just retry — escalate the fix strategy:
class RetryWithEscalation:
def __init__(self, max_retries=3):
self.max_retries = max_retries
def execute(self, agent, task, context):
strategies = [
lambda: agent.run(task, context), # Plain retry
lambda: agent.run(task, context + self.add_examples()), # Add examples
lambda: agent.run(self.simplify(task), context), # Simplify task
lambda: self.delegate_to_fallback_agent(task, context), # Fallback agent
]
for attempt in range(self.max_retries):
try:
result = strategies[min(attempt, len(strategies)-1)]()
if self.validate(result):
return result
except Exception as e:
context = self.add_error_context(context, e)
return self.human_escalation(task, context)
def add_error_context(self, context, error):
return context + f"nnPrevious attempt failed with: {error}. Please avoid this issue."
def human_escalation(self, task, context):
return OrchestratorResult(
status="needs_human",
task=task,
context=context,
partial_results=self.gather_partial()
)Pattern 3: Circuit Breaker
Prevent cascade failures by isolating failing agents:
class AgentCircuitBreaker:
def __init__(self, failure_threshold=3, recovery_timeout=60):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
self.last_failure_time = None
def call(self, agent, task):
if self.state == "OPEN":
if self._timeout_elapsed():
self.state = "HALF_OPEN"
else:
return self.fallback_response(agent, task)
try:
result = agent.execute(task)
self._on_success()
return result
except Exception as e:
self._on_failure()
if self.state == "OPEN":
return self.fallback_response(agent, task)
raise
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
def _on_success(self):
self.failure_count = 0
self.state = "CLOSED"Pattern 4: Graceful Degradation
When an agent fails, the system should still deliver partial value:
class GracefulOrchestrator:
def execute_pipeline(self, task, agents):
results = {}
for agent in agents:
try:
results[agent.name] = agent.run(
task,
upstream_results=results
)
except AgentFailure as e:
# Log the failure but continue with remaining agents
self.log_failure(agent, e)
results[agent.name] = PartialResult(
status="degraded",
error=str(e),
partial_output=e.partial_output
)
# Compile best-effort result from whatever succeeded
return self.synthesize_partial_results(results)
def synthesize_partial_results(self, results):
successful = {k: v for k, v in results.items()
if v.status == "complete"}
degraded = {k: v for k, v in results.items()
if v.status == "degraded"}
if not successful:
return FinalResult(
status="failed",
message="All agents failed. Human intervention required.",
partial_data=results
)
return FinalResult(
status="partial" if degraded else "complete",
content=self.aggregate(successed),
warnings=[f"{k} produced degraded output" for k in degraded]
)Pattern 5: Human-in-the-Loop Checkpoints
For high-stakes decisions, insert human approval points:
class HumanCheckpoint:
def __init__(self, threshold_risk="high"):
self.threshold = threshold_risk
def should_escalate(self, agent_output, task_context):
risk = self.assess_risk(agent_output, task_context)
if risk.level >= self.threshold:
return EscalationRequest(
task=task_context,
agent_output=agent_output,
risk_factors=risk.factors,
options=[
"approve",
"modify_and_proceed",
"retry_with_changes",
"abort"
]
)
return None
def assess_risk(self, output, context):
factors = []
if output.confidence = 2 else "medium" if factors else "low",
factors=factors
)Architecture Blueprint: Resilient Multi-Agent System
Input Task
↓
Task Decomposer (breaks task into subtasks)
↓
Agent Router (assigns subtasks to specialized agents)
↓
┌─────────────────────────────────────────┐
│ Agent Pool (with Circuit Breakers) │
│ ┌──────────┐ ┌──────────┐ ┌─────────┐ │
│ │ Researcher│ │ Analyzer │ │ Writer │ │
│ └──────────┘ └──────────┘ └─────────┘ │
│ ┌──────────┐ ┌──────────┐ │
│ │ Reviewer │ │FALLBACK │ │
│ └──────────┘ └──────────┘ │
└─────────────────────────────────────────┘
↓
Validation Gates (between each handoff)
↓
Result Synthesizer (aggregates outputs)
↓
Quality Gate (final validation)
↓
Output (with confidence score)
Metrics to Track
Monitor these metrics to catch reliability issues early:
- Agent success rate: % of tasks completed without retry (target: >90%)
- Median retry count: How many retries before success (target: <1.5)
- Circuit breaker trips: Frequency of circuit breaker activations
- Cascade failure rate: % of tasks where one failure caused downstream failures
- Human escalation rate: % of tasks requiring human intervention (target: <5%)
- End-to-end latency: P50, P95, P99 completion times
- Partial completion rate: % of tasks that delivered partial value despite failures
Conclusion
Reliability in multi-agent systems doesn’t come from making each agent perfect — it comes from designing the system to handle inevitable failures gracefully. Validation gates, retry escalation, circuit breakers, and graceful degradation transform a fragile chain of agents into a resilient pipeline.
Start with validation gates and retry logic for your most critical agent handoffs. Add circuit breakers as you scale to more agents. And always, always have a human escalation path for the edge cases your automated recovery can’t handle.
Schreibe einen Kommentar