By the end of this lesson, you'll recognize the seven most common anti-patterns in multi-agent system design and know how to refactor each one. These patterns emerge repeatedly in production systems and cause failures that are hard to diagnose after the fact.
After this lesson, you'll have a pattern library of what not to do — each with a concrete fix. Use it as a code review checklist before any agent or skill reaches production.
The AutoGen paper (Wu et al., 2023) demonstrates that multi-agent systems fail not from novel complexity but from well-known software engineering mistakes amplified by non-determinism. A monolithic prompt that works 90% of the time in testing fails 40% of the time in production because the input distribution shifts. A hard-coded endpoint that works for months breaks silently when the service migrates. These are preventable failures.
Each anti-pattern below follows a consistent structure: the Problem (what goes wrong and why), the Solution (the correct pattern), and a code example showing the fix.
A single skill handles multiple unrelated tasks — code review, deployment, documentation generation, and ticket management all in one file. The prompt becomes a wall of conflicting instructions. The model struggles to determine which behavior applies. Testing becomes impossible because each test case must account for all other behaviors. Changes to one capability break another.
Apply the Single Responsibility Principle. Each skill does one thing well. Compose complex workflows from focused skills via an orchestrator.
# ❌ Anti-pattern: Monolithic skill trying to do everything
class UniversalAgentSkill:
"""Handles code review, deploys, writes docs, and manages tickets."""
def execute(self, task):
if "review" in task.description:
return self._do_code_review(task)
elif "deploy" in task.description:
return self._do_deployment(task)
elif "document" in task.description:
return self._do_documentation(task)
elif "ticket" in task.description:
return self._do_ticket_management(task)
else:
# Guess what the user wants based on vibes
return self._try_everything(task)
# ✅ Solution: Focused, composable skills
class CodeReviewSkill:
"""Reviews code changes for correctness, style, and security."""
def execute(self, diff: str, context: ReviewContext) -> ReviewResult:
prompt = self.build_review_prompt(diff, context)
return self.model.generate(prompt, schema=ReviewResult)
class DeploymentSkill:
"""Executes deployment pipelines with safety checks."""
def execute(self, package: str, environment: str) -> DeployResult:
self.validate_environment(environment)
return self.pipeline.deploy(package, environment)
class Orchestrator:
"""Composes focused skills into workflows."""
def handle_release(self, package: str):
review = self.code_review.execute(get_diff(package))
if review.approved:
return self.deployment.execute(package, "production")
Endpoints, credentials, model names, and thresholds embedded directly in source code. When the model version changes, you edit code. When an endpoint migrates, you hunt through files. Credentials in source get committed to version control. Different environments (dev, staging, production) require code changes instead of configuration changes.
Externalize all configuration. Use environment variables for secrets, configuration files for tunable parameters, and dependency injection for service endpoints.
# ❌ Anti-pattern: Hard-coded everything
class MyAgent:
def __init__(self):
self.api_key = "sk-abc123secretkey456"
self.endpoint = "https://api.openai.com/v1/chat/completions"
self.model = "gpt-4-0125-preview"
self.max_tokens = 4096
self.temperature = 0.7
self.retry_count = 3
self.timeout = 30
# ✅ Solution: Externalized, injectable configuration
from dataclasses import dataclass
from os import environ
@dataclass(frozen=True)
class AgentConfig:
"""All tunable parameters externalized."""
api_key: str
endpoint: str
model: str
max_tokens: int = 4096
temperature: float = 0.7
retry_count: int = 3
timeout_seconds: int = 30
@classmethod
def from_environment(cls) -> "AgentConfig":
return cls(
api_key=environ["AGENT_API_KEY"],
endpoint=environ.get("AGENT_ENDPOINT", "https://api.example.com/v1"),
model=environ.get("AGENT_MODEL", "claude-sonnet-4-20250514"),
max_tokens=int(environ.get("AGENT_MAX_TOKENS", "4096")),
)
class MyAgent:
def __init__(self, config: AgentConfig):
self.config = config # injected, testable, no secrets in source
Vague system prompts like "You are a helpful assistant" provide no behavioral constraints. The agent's output varies wildly between invocations. It hallucinates tool calls, invents capabilities, and produces inconsistent formatting. Without specific instructions, the model falls back to generic completion behavior rather than task-specific expertise.
Write precise prompts with explicit constraints: define the role, specify the output format, list what the agent must and must not do, and provide examples of correct behavior.
# ❌ Anti-pattern: Vague, unconstrained prompt
SYSTEM_PROMPT = "You are a helpful assistant that reviews code."
# ✅ Solution: Precise, constrained prompt with examples
SYSTEM_PROMPT = """You are a code review agent for Python backend services.
ROLE: Identify bugs, security issues, and style violations in diffs.
OUTPUT FORMAT: Return a JSON array of findings. Each finding has:
- file: string (path)
- line: integer (line number in the diff)
- severity: "critical" | "warning" | "suggestion"
- category: "bug" | "security" | "style" | "performance"
- message: string (one sentence explaining the issue)
- fix: string (the corrected code, or null if no fix is obvious)
RULES:
- Only comment on changed lines (+ lines in the diff).
- Never suggest changes to unchanged code.
- Flag SQL injection, path traversal, and credential exposure as critical.
- Limit findings to 10 most important issues.
- If the diff is clean, return an empty array [].
EXAMPLE INPUT:
+ password = request.args.get("pw")
+ db.execute(f"SELECT * FROM users WHERE pass='{password}'")
EXAMPLE OUTPUT:
[{"file": "auth.py", "line": 42, "severity": "critical",
"category": "security", "message": "SQL injection via string interpolation.",
"fix": "db.execute('SELECT * FROM users WHERE pass=?', (password,))"}]
"""
Errors are caught and silently swallowed, or not caught at all. The agent continues processing with corrupted state. Downstream consumers receive partial or incorrect results with no indication that something went wrong. Debugging requires reproducing the exact input because no diagnostic information was preserved.
Handle errors explicitly at every boundary. Log diagnostic context. Propagate failures with actionable information. Distinguish between retryable and terminal failures.
# ❌ Anti-pattern: Silent failure
class DataAgent:
def fetch_and_analyze(self, query):
try:
data = self.api.search(query)
except Exception:
data = [] # silently return empty — caller never knows it failed
try:
analysis = self.model.analyze(data)
except Exception:
analysis = "No analysis available." # looks like a valid response
return analysis # caller can't distinguish success from total failure
# ✅ Solution: Explicit error handling with context
from dataclasses import dataclass
from enum import Enum
class FailureType(Enum):
RETRYABLE = "retryable" # network timeout, rate limit
TERMINAL = "terminal" # invalid input, auth failure
DEGRADED = "degraded" # partial success
@dataclass
class AgentResult:
success: bool
data: any = None
error: str | None = None
failure_type: FailureType | None = None
partial_results: list | None = None
class DataAgent:
def fetch_and_analyze(self, query: str) -> AgentResult:
# Fetch with explicit error classification
try:
data = self.api.search(query)
except TimeoutError:
logger.warning("API timeout for query=%s", query)
return AgentResult(
success=False,
error=f"Search timed out after {self.config.timeout_seconds}s",
failure_type=FailureType.RETRYABLE,
)
except AuthenticationError as e:
logger.error("Auth failure: %s", e)
return AgentResult(
success=False,
error="Authentication failed — check credentials",
failure_type=FailureType.TERMINAL,
)
if not data:
return AgentResult(success=True, data="No results found for query.")
# Analyze with graceful degradation
try:
analysis = self.model.analyze(data)
return AgentResult(success=True, data=analysis)
except TokenLimitError:
# Partial success: return raw data without analysis
logger.warning("Token limit exceeded — returning raw results")
return AgentResult(
success=True,
data="Analysis unavailable (input too large). Raw results attached.",
failure_type=FailureType.DEGRADED,
partial_results=data[:10],
)
Prompts grow unchecked as conversation history accumulates or large documents are injected. The model silently truncates context, losing critical instructions or data. Alternatively, the API returns an error and the task fails completely. Developers don't notice because short test inputs fit within limits — the failure only manifests with production-scale inputs.
Implement token budgeting at the architecture level. Reserve space for system prompts, measure inputs before sending, and use chunking or summarization when inputs exceed limits.
# ❌ Anti-pattern: No token awareness
class NaiveAgent:
def run(self, user_input: str, history: list[dict]) -> str:
messages = [
{"role": "system", "content": self.system_prompt},
*history, # unbounded — grows forever
{"role": "user", "content": user_input}, # could be 200k tokens
]
return self.model.generate(messages) # silently truncated or error
# ✅ Solution: Token-budgeted context management
class TokenBudgetedAgent:
MODEL_CONTEXT_WINDOW = 128_000
RESPONSE_RESERVE = 4_096 # tokens reserved for model output
SYSTEM_PROMPT_BUDGET = 2_000
def run(self, user_input: str, history: list[dict]) -> str:
available = self.MODEL_CONTEXT_WINDOW - self.RESPONSE_RESERVE
# System prompt gets fixed budget
system_tokens = count_tokens(self.system_prompt)
if system_tokens > self.SYSTEM_PROMPT_BUDGET:
raise ConfigurationError(
f"System prompt uses {system_tokens} tokens, "
f"budget is {self.SYSTEM_PROMPT_BUDGET}"
)
available -= system_tokens
# User input: chunk if too large
input_tokens = count_tokens(user_input)
if input_tokens > available * 0.6:
user_input = self._chunk_and_summarize(user_input, available * 0.6)
available -= count_tokens(user_input)
# History: keep recent messages within remaining budget
pruned_history = self._prune_history(history, max_tokens=available)
messages = [
{"role": "system", "content": self.system_prompt},
*pruned_history,
{"role": "user", "content": user_input},
]
total = count_tokens(messages)
assert total + self.RESPONSE_RESERVE <= self.MODEL_CONTEXT_WINDOW
return self.model.generate(messages)
def _prune_history(self, history: list[dict], max_tokens: int) -> list[dict]:
"""Keep most recent messages that fit within budget."""
result = []
used = 0
for msg in reversed(history):
msg_tokens = count_tokens(msg["content"])
if used + msg_tokens > max_tokens:
break
result.insert(0, msg)
used += msg_tokens
return result
Tools are called without retry logic, timeout handling, or input validation. A single API hiccup crashes the entire agent workflow. Rate limits are hit repeatedly because there's no backoff. Responses aren't validated — malformed JSON from an API silently corrupts downstream processing. Tool failures don't propagate useful diagnostic information.
Wrap every external tool call with retry logic, circuit breakers, input validation, and structured error responses.
# ❌ Anti-pattern: Brittle tool integration
class SearchTool:
def search(self, query: str) -> list[dict]:
response = requests.get(
"https://api.search.com/v1/search",
params={"q": query},
)
return response.json()["results"] # crashes on non-200, non-JSON, missing key
# ✅ Solution: Resilient tool wrapper
import time
from functools import wraps
class ResilientSearchTool:
MAX_RETRIES = 3
BACKOFF_BASE = 2 # exponential: 2s, 4s, 8s
TIMEOUT_SECONDS = 10
def search(self, query: str) -> ToolResult:
# Input validation
if not query or not query.strip():
return ToolResult(success=False, error="Empty search query")
if len(query) > 1000:
query = query[:1000] # truncate, don't crash
# Retry with exponential backoff
last_error = None
for attempt in range(self.MAX_RETRIES):
try:
response = requests.get(
self.config.search_endpoint,
params={"q": query},
timeout=self.TIMEOUT_SECONDS,
headers={"Authorization": f"Bearer {self.config.api_key}"},
)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
time.sleep(retry_after)
continue
response.raise_for_status()
# Validate response structure
data = response.json()
if "results" not in data:
return ToolResult(
success=False,
error=f"Unexpected response structure: {list(data.keys())}",
)
return ToolResult(success=True, data=data["results"])
except requests.Timeout:
last_error = f"Timeout after {self.TIMEOUT_SECONDS}s"
except requests.ConnectionError as e:
last_error = f"Connection failed: {e}"
except json.JSONDecodeError:
last_error = "Response was not valid JSON"
# Exponential backoff between retries
if attempt < self.MAX_RETRIES - 1:
sleep_time = self.BACKOFF_BASE ** (attempt + 1)
time.sleep(sleep_time)
return ToolResult(
success=False,
error=f"All {self.MAX_RETRIES} attempts failed. Last error: {last_error}",
retryable=True,
)
Agents are deployed based on manual spot-checks. "It worked when I tried it" substitutes for a test suite. Prompt changes go live without regression testing. No one knows the agent's accuracy baseline, so gradual degradation goes undetected. When failures are reported, there's no way to reproduce them because inputs weren't logged and there's no evaluation framework.
Implement a three-layer testing strategy: unit tests for individual components, integration tests for end-to-end workflows, and continuous evaluation that tracks quality metrics over time.
# ❌ Anti-pattern: "Testing" by manual trial
def deploy_agent():
# "Tested" by running it once with a sample input
agent = build_agent()
print(agent.run("test query")) # looks good? ship it.
upload_to_production(agent)
# ✅ Solution: Three-layer testing before any deployment
import pytest
from agent.evaluation import EvaluationSuite, QualityGate
# Layer 1: Unit tests — deterministic component checks
class TestPromptBuilder:
def test_system_prompt_under_token_limit(self):
prompt = build_system_prompt()
assert count_tokens(prompt) < 2000
def test_output_parser_handles_malformed_json(self):
result = parse_agent_output("not json {{{")
assert result.status == "parse_error"
assert result.raw_output == "not json {{{"
# Layer 2: Integration tests — end-to-end workflows
class TestAgentWorkflow:
@pytest.fixture
def agent(self):
return build_agent(config=AgentConfig.from_environment())
def test_complete_task_end_to_end(self, agent):
result = agent.run("Summarize the Q3 earnings report")
assert result.success
assert 100 < len(result.output) < 2000
assert result.tokens_used < 5000
def test_graceful_degradation_on_api_failure(self, agent, mock_api):
mock_api.configure(status=503)
result = agent.run("Summarize the Q3 earnings report")
assert result.success is False
assert "unavailable" in result.error.lower()
# Layer 3: Continuous evaluation — quality tracking over time
def test_quality_gate():
suite = EvaluationSuite.load("regression_tests.yaml")
results = suite.run(agent)
gate = QualityGate(
min_accuracy=0.92,
max_latency_p95_ms=5000,
max_error_rate=0.02,
)
violations = gate.check(results)
assert not violations, f"Quality gate failed: {violations}"
The seven patterns above are technical. But the most dangerous anti-pattern is organizational: Shadow AI — teams deploying unregistered, ungoverned agents that bypass review processes.
Shadow AI emerges when governance processes are too slow or too burdensome. A developer builds an agent that "just works" for their team, shares it informally, and suddenly it's processing customer data without security review, logging, or access controls. The governance guide identifies this as the primary organizational risk in enterprise AI deployment.
Shadow AI thrives when: (1) the official platform is too restrictive or slow to onboard, (2) there's no lightweight path for experimental agents, (3) teams aren't educated on why governance exists, or (4) there's no visibility into what agents are running across the organization.
# Shadow AI prevention through a lightweight registry
# The registry makes it EASIER to be compliant than non-compliant
@dataclass
class AgentRegistration:
"""Minimum metadata required to register any agent."""
name: str
owner: str # team or individual
purpose: str # one-sentence description
data_classification: str # public, internal, confidential, restricted
model_provider: str # which LLM provider
tools_used: list[str] # external integrations
review_status: str # draft, reviewed, approved
class AgentRegistry:
"""Central registry — the alternative to shadow AI."""
def register(self, agent: AgentRegistration) -> str:
"""Register takes < 2 minutes. Returns agent_id."""
# Auto-approve low-risk agents (no customer data, no external tools)
if self._is_low_risk(agent):
agent.review_status = "auto_approved"
return self._save(agent)
# Flag for human review only when genuinely risky
agent.review_status = "pending_review"
self._notify_reviewers(agent)
return self._save(agent)
def _is_low_risk(self, agent: AgentRegistration) -> bool:
return (
agent.data_classification in ("public", "internal")
and len(agent.tools_used) == 0
and agent.model_provider in APPROVED_PROVIDERS
)
The principle: make the governed path the path of least resistance. If registration takes two minutes and auto-approves low-risk agents, developers have no incentive to go around it.
| # | Anti-Pattern | Symptom | Fix |
|---|---|---|---|
| 1 | Monolithic Skills | One skill handles unrelated tasks; changes break unrelated features | Single Responsibility + orchestration |
| 2 | Hard-Coded Config | Secrets in source; environment changes require code edits | Environment variables + injectable config |
| 3 | Overly Generic Prompts | Wildly inconsistent outputs; hallucinated capabilities | Precise constraints + output schema + examples |
| 4 | Missing Error Handling | Silent failures; corrupted downstream state | Classify errors (retryable vs terminal) + propagate context |
| 5 | Ignoring Token Limits | Silent truncation; works in testing, fails on real inputs | Token budgeting + chunking + history pruning |
| 6 | Poor Tool Integration | Single API hiccup crashes workflow; rate limits hammered | Retry + backoff + input validation + circuit breakers |
| 7 | Lack of Testing | Gradual degradation undetected; can't reproduce failures | Unit + integration + continuous evaluation |
| 8 | Shadow AI (org) | Ungoverned agents processing sensitive data | Lightweight registry + auto-approval for low-risk |
An agent works perfectly in development but produces truncated, incoherent responses when processing customer support transcripts in production. Which anti-pattern is the likely cause?
A team builds a useful internal agent and shares it via Slack. Within a month, 50 people are using it to process customer data, but it was never security-reviewed. What organizational anti-pattern does this represent?
An agent's search tool returns an empty result set when the API times out, and downstream processing treats this as "no results found" rather than a failure. Which anti-pattern is this?
Audit one of your existing agents or skills against the anti-pattern checklist above. For each anti-pattern, answer: does this apply to my system? If yes, write a one-line fix description and prioritize by blast radius — which failure would affect the most users or corrupt the most data? Fix the highest-priority item first.
Wu et al., AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation (2023). The paper's multi-agent architecture patterns reveal how compositional design prevents monolithic failures, and how conversation-based orchestration replaces brittle hard-coded workflows.