Testing Multi-Agent Systems

"Testing one agent is hard. Testing five agents talking to each other is a combinatorial adventure."

Eval, Thoroughly Tested AI Agent

1. The Testing Challenge

Testing multi-agent systems is harder than testing traditional software because of non-determinism (the same input can produce different outputs), emergent behavior (agents interact in unexpected ways), and environmental dependencies (tools, APIs, and data sources can change between test runs). A test suite that passes today might fail tomorrow because the LLM produced a slightly different reasoning trace, triggering a different tool call sequence. Standard unit testing approaches are necessary but insufficient; testing agents requires additional strategies tailored to these challenges.

The testing pyramid for agent systems has four levels. Unit tests verify individual components: tool implementations, input validators, state management logic. These are deterministic and fast. Integration tests verify that components work together: the agent can call tools correctly, tools return results in the expected format, and state transitions work as designed. Scenario tests run the complete agent on predefined tasks and check for acceptable outcomes (not exact matches). Chaos tests inject failures into the system to verify that error handling, fallbacks, and graceful degradation work correctly.

For the non-deterministic layers (scenario tests), use outcome-based assertions rather than exact-match assertions. Instead of checking that the agent produced a specific string, check that the output contains the required information, that tool calls were made in a valid order, that the final answer is factually correct, and that the agent stayed within its budget. This makes tests robust to the natural variation in LLM outputs while still catching genuine failures.

2. Contract Testing for Multi-Agent Systems

In a multi-agent system, each agent depends on the outputs of other agents. If Agent A changes its output format, Agent B (which consumes that output) may break. Contract testing verifies that each agent's inputs and outputs conform to agreed-upon schemas, catching integration issues before they reach production. The "contract" is a formal specification of what each agent expects to receive and what it promises to produce.

from pydantic import BaseModel
from typing import List

# Define the contract between the Research Agent and the Writing Agent
class ResearchOutput(BaseModel):
 """Contract: what the Research Agent must produce."""
 topic: str
 findings: List[dict] # Each finding has 'source', 'content', 'relevance'
 gaps: List[str] # Topics that need more research
 confidence: float # 0.0 to 1.0

class WritingInput(BaseModel):
 """Contract: what the Writing Agent expects to receive."""
 topic: str
 findings: List[dict]
 tone: str # "formal", "casual", "technical"
 max_length: int # words

def test_research_output_matches_writing_input():
 """Verify that the Research Agent's output satisfies the Writing Agent's input contract."""
 # Run the Research Agent on a test task
 research_result = research_agent.run("Summarize recent advances in RAG")

 # Validate against the contract
 output = ResearchOutput(**research_result)
 assert len(output.findings) >= 1, "Must produce at least one finding"
 assert 0 <= output.confidence <= 1, "Confidence must be between 0 and 1"

 # Verify it can be transformed into the Writing Agent's expected input
 writing_input = WritingInput(
 topic=output.topic,
 findings=output.findings,
 tone="technical",
 max_length=2000,
 )
 assert writing_input # Pydantic validation passed

Contract testing is especially important for multi-agent systems because agents evolve independently. If the Research Agent's developer changes the output format from a list of dictionaries to a flat string, the Writing Agent breaks silently because it receives valid text but not the structure it expects. Contract tests catch this at the boundary before it manifests as a subtle quality degradation in production. This is the same principle that drives API versioning in microservice architectures: the interface between components must be explicitly defined and tested, independent of each component's internal implementation.

3. Chaos Engineering for Agents

Chaos engineering deliberately introduces failures into the system to verify that it handles them correctly. For agent systems, chaos tests inject: LLM API failures (timeouts, rate limits, garbage responses), tool failures (services returning errors, slow responses, incorrect data), data corruption (tools returning malformed JSON, unexpected data types), and resource exhaustion (memory limits, token budget depletion). Each injected failure tests the system's resilience and reveals gaps in error handling.

The approach is systematic: define a steady state (the agent successfully completes a reference task), introduce a failure, and verify that the system either recovers to the steady state or degrades gracefully. Each chaos test should have a clear hypothesis: "If the database tool fails, the agent should fall back to cached data and note the limitation in its response." Running chaos tests regularly, especially before major deployments, builds confidence that the system is resilient to real-world failures.

import random
from unittest.mock import patch

class ChaosInjector:
 """Inject failures into agent tool calls for chaos testing."""

 def __init__(self, failure_rate: float = 0.3):
 self.failure_rate = failure_rate
 self.injected_failures = []

 def maybe_fail(self, tool_name: str, original_func):
 """Wrap a tool function with random failure injection."""
 async def chaos_wrapper(*args, **kwargs):
 if random.random() < self.failure_rate:
 failure_type = random.choice([
 "timeout", "rate_limit", "server_error", "malformed_response"
 ])
 self.injected_failures.append((tool_name, failure_type))

 if failure_type == "timeout":
 raise TimeoutError(f"{tool_name} timed out")
 elif failure_type == "rate_limit":
 raise RateLimitError(f"{tool_name} rate limited")
 elif failure_type == "server_error":
 raise APIError(f"{tool_name} returned 500")
 elif failure_type == "malformed_response":
 return "{{invalid json"

 return await original_func(*args, **kwargs)
 return chaos_wrapper

def test_agent_resilience():
 """Chaos test: agent should handle random tool failures gracefully."""
 chaos = ChaosInjector(failure_rate=0.3)

 # Wrap all tools with chaos injection
 chaotic_tools = {
 name: chaos.maybe_fail(name, tool.execute)
 for name, tool in agent.tools.items()
 }

 # Run the agent on a reference task
 result = agent.run(
 "Analyze last month's sales data",
 tools=chaotic_tools,
 )

 # Verify graceful degradation
 assert result is not None, "Agent should produce some output even with failures"
 assert "error" not in result.lower() or "unavailable" in result.lower(), \
 "Error messages should be user-friendly"
 print(f"Injected {len(chaos.injected_failures)} failures: {chaos.injected_failures}")
 print(f"Agent output: {result[:200]}...")

Lab: Chaos Test a Multi-Agent Pipeline

pip install torch transformers numpy

# TODO: Implement setup code here

# Complete solution for this lab exercise
# TODO: Full implementation here

Exercises