You've built a RAG system. It works — mostly. You point it at a vector store of internal documentation, wire up a retrieval chain, and watch it answer questions about your company's HR policies or product catalog. Then your stakeholders start asking the questions that break everything: "What does our Q3 sales data say about customer churn, and does that match what the support team has been logging in Jira?" Now your single-agent RAG pipeline is juggling a PostgreSQL database, a Jira API, a PDF archive of customer contracts, and a Confluence wiki — and it's doing all of it badly, because a monolithic retriever with a single embedding space wasn't designed for this.
The solution isn't to stuff more retrieval logic into one agent. It's to stop pretending that one agent should be responsible for everything. Multi-agent RAG architectures distribute retrieval and reasoning across specialized agents — each expert in a specific data source, schema, or reasoning pattern — and coordinate them through an orchestration layer that decides who does what, when, and with whose output. This is the architectural leap that separates toy RAG demos from production intelligence systems.
By the end of this lesson, you'll be able to design and implement a multi-agent RAG system from first principles: defining agent contracts, building a working orchestrator, handling inter-agent dependencies, managing failure gracefully, and reasoning clearly about the trade-offs that separate good architectures from fragile ones.
What you'll learn:
This lesson assumes you're comfortable with:
asyncio, await)Before designing the solution, let's be precise about what actually fails when you push a single-agent RAG system beyond its comfort zone.
Embedding space homogeneity. A single vector store assumes that all your content can be meaningfully embedded in one shared semantic space. This works fine when your documents are all prose in the same domain. It collapses when you're co-embedding SQL query results (highly structured), legal contract clauses (formal, precise language), and Slack messages (informal, abbreviated, context-dependent). The cosine similarity scores become meaningless comparisons — you're measuring distance between apples and parking meters.
Retrieval strategy mismatch. Different data sources require fundamentally different retrieval strategies. A time-series database of sales metrics needs aggregation queries with temporal filters. A code repository needs keyword + AST-aware search. A document archive benefits from hierarchical chunking with parent-document retrieval. A single retriever forced to serve all these cases will either be over-engineered to the point of unmaintainability or will default to the lowest common denominator — pure semantic similarity — which is wrong for most of them.
Context contamination. When you retrieve 20 chunks from five different data sources and jam them all into one context window, you've created a coherence problem. The LLM now has to simultaneously reason about a Jira ticket from 2021, a paragraph from an onboarding PDF, three rows from a CRM export, and a Confluence page that contradicts the PDF. The model isn't equipped to weight these sources appropriately without explicit structure, and you lose the ability to trace which source drove which claim in the output.
Debugging opacity. When something goes wrong — and it will — a monolithic RAG pipeline gives you one chain of events to inspect. If the wrong document got retrieved, if the reranker deprioritized the most relevant chunk, if the LLM hallucinated a detail that none of your sources contain, you have limited visibility into which step broke the chain. Multi-agent architectures make failure modes explicit and localizable.
The first design decision in a multi-agent RAG system is defining what agents exist and what they're responsible for. This isn't a technical question yet — it's a domain modeling question.
The cleanest separation I've found in practice is between data source agents (specialists in retrieving and interpreting content from one specific source) and reasoning agents (specialists in synthesis, comparison, and conclusion-drawing across inputs from multiple data source agents).
A data source agent owns exactly one retrieval context. It has its own index, its own retrieval strategy, its own domain-specific prompting, and its own output schema. A reasoning agent never touches a data source directly — it receives structured outputs from data source agents and performs higher-order reasoning.
User Query
│
▼
┌─────────────┐
│ Orchestrator │ ← Decides which agents to invoke and in what order
└──────┬──────┘
│
┌────┴────────────────────┐
│ │
▼ ▼
[SQL Agent] [Document Agent] ← Data source agents
│ │
└──────────┬──────────────┘
│
▼
[Synthesis Agent] ← Reasoning agent
│
▼
Final Answer
This separation matters because it creates a clean interface boundary. Your orchestrator doesn't need to know how the SQL Agent queries PostgreSQL or how the Document Agent does parent-document retrieval. It only needs to know what question to ask each agent and what shape the answer will come back in.
An agent contract specifies three things: what inputs the agent accepts, what outputs it guarantees, and what it will not do. That last part is underrated. An agent with a clear negative scope is much easier to compose than one with open-ended responsibilities.
Here's a practical contract definition using Python dataclasses:
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
class ConfidenceLevel(Enum):
HIGH = "high" # Direct match, high-quality sources
MEDIUM = "medium" # Inferred or partially matched
LOW = "low" # Best-effort, unreliable sources
NO_DATA = "no_data" # Agent could not find relevant information
@dataclass
class AgentQuery:
question: str
context: Optional[str] = None # Prior conversation context
filters: dict = field(default_factory=dict) # Source-specific filters
max_tokens: int = 1024
@dataclass
class AgentResponse:
agent_id: str
answer: str
sources: list[dict] # Citations: {id, title, url, excerpt}
confidence: ConfidenceLevel
raw_retrieved: list[str] # The actual retrieved chunks, for debugging
metadata: dict = field(default_factory=dict)
error: Optional[str] = None
@dataclass
class AgentContract:
agent_id: str
description: str
data_sources: list[str] # Which sources this agent owns
capabilities: list[str] # What question types it handles
limitations: list[str] # Explicit out-of-scope items
output_schema: type # Should return AgentResponse
Notice that AgentResponse includes raw_retrieved — the actual chunks before they were synthesized into an answer. This is non-negotiable for debugging. When your system produces a wrong answer, you need to see whether the failure was in retrieval (wrong chunks) or reasoning (right chunks, wrong conclusion).
Let's build three concrete agents that demonstrate the range of retrieval strategies you'll encounter in practice.
For a corpus of PDFs — think legal documents, technical specs, research papers — the standard chunk-and-embed approach loses critical context. A sentence that reads "This provision supersedes all prior agreements" means nothing without the surrounding clause. Parent-document retrieval solves this by indexing small chunks for semantic search but returning larger parent chunks for the LLM.
import asyncio
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.storage import InMemoryStore
from langchain.retrievers import ParentDocumentRetriever
class DocumentAgent:
def __init__(self, doc_store_path: str, agent_id: str = "document_agent"):
self.agent_id = agent_id
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
# Child splitter: small chunks for dense retrieval
child_splitter = RecursiveCharacterTextSplitter(
chunk_size=400,
chunk_overlap=50
)
# Parent splitter: larger chunks returned to the LLM
parent_splitter = RecursiveCharacterTextSplitter(
chunk_size=2000,
chunk_overlap=200
)
vectorstore = Chroma(
collection_name="document_children",
embedding_function=self.embeddings,
persist_directory=doc_store_path
)
docstore = InMemoryStore()
self.retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=docstore,
child_splitter=child_splitter,
parent_splitter=parent_splitter,
)
async def query(self, agent_query: AgentQuery) -> AgentResponse:
try:
# Retrieve parent documents via child chunk matching
docs = await asyncio.to_thread(
self.retriever.invoke, agent_query.question
)
if not docs:
return AgentResponse(
agent_id=self.agent_id,
answer="No relevant documents found.",
sources=[],
confidence=ConfidenceLevel.NO_DATA,
raw_retrieved=[]
)
raw_chunks = [doc.page_content for doc in docs]
sources = [
{
"id": doc.metadata.get("source", "unknown"),
"title": doc.metadata.get("title", "Untitled"),
"page": doc.metadata.get("page", None),
"excerpt": doc.page_content[:200]
}
for doc in docs
]
# Domain-specific synthesis prompt
context = "\n\n---\n\n".join(raw_chunks)
synthesis_prompt = f"""You are analyzing legal and technical documents.
Based ONLY on the following document excerpts, answer the question.
If the documents don't contain sufficient information, say so explicitly.
Do not infer or extrapolate beyond what is written.
Documents:
{context}
Question: {agent_query.question}
Provide a precise answer with direct quotes where appropriate."""
response = await asyncio.to_thread(
self.llm.invoke, synthesis_prompt
)
return AgentResponse(
agent_id=self.agent_id,
answer=response.content,
sources=sources,
confidence=ConfidenceLevel.HIGH if len(docs) >= 2 else ConfidenceLevel.MEDIUM,
raw_retrieved=raw_chunks
)
except Exception as e:
return AgentResponse(
agent_id=self.agent_id,
answer="",
sources=[],
confidence=ConfidenceLevel.NO_DATA,
raw_retrieved=[],
error=str(e)
)
SQL data sources need a completely different approach. You're not doing semantic retrieval over rows — you're generating a query, executing it, and interpreting structured results. The key challenge here is schema awareness: the agent needs to know what tables and columns exist before it can write correct SQL.
import sqlalchemy
from langchain_community.utilities import SQLDatabase
from langchain_openai import ChatOpenAI
from langchain.chains import create_sql_query_chain
from langchain_community.tools.sql_database.tool import QuerySQLDataBaseTool
class SQLAgent:
def __init__(self, connection_string: str, agent_id: str = "sql_agent"):
self.agent_id = agent_id
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
# LangChain's SQLDatabase handles schema introspection automatically
self.db = SQLDatabase.from_uri(
connection_string,
# Only expose tables this agent should know about
include_tables=["sales_orders", "customers", "products", "returns"]
)
self.query_chain = create_sql_query_chain(self.llm, self.db)
self.executor = QuerySQLDataBaseTool(db=self.db)
async def query(self, agent_query: AgentQuery) -> AgentResponse:
try:
# Step 1: Generate SQL from natural language
sql_query = await asyncio.to_thread(
self.query_chain.invoke,
{"question": agent_query.question}
)
# Strip markdown fencing if the LLM included it
sql_query = sql_query.strip().strip("```sql").strip("```").strip()
# Step 2: Execute the query
raw_result = await asyncio.to_thread(
self.executor.invoke, sql_query
)
# Step 3: Interpret the results in natural language
interpretation_prompt = f"""You are a data analyst interpreting SQL query results.
Question asked: {agent_query.question}
SQL executed: {sql_query}
Query result: {raw_result}
Provide a clear, precise interpretation of these results.
Include specific numbers and percentages where relevant.
If the result is empty, explicitly state that no data matched the query criteria."""
response = await asyncio.to_thread(
self.llm.invoke, interpretation_prompt
)
return AgentResponse(
agent_id=self.agent_id,
answer=response.content,
sources=[{"id": "sql_db", "title": "Sales Database",
"query": sql_query, "raw_result": str(raw_result)[:500]}],
confidence=ConfidenceLevel.HIGH,
raw_retrieved=[str(raw_result)],
metadata={"sql_query": sql_query}
)
except sqlalchemy.exc.OperationalError as e:
# SQL execution failed — return structured error
return AgentResponse(
agent_id=self.agent_id,
answer="",
sources=[],
confidence=ConfidenceLevel.NO_DATA,
raw_retrieved=[],
error=f"SQL execution error: {str(e)}"
)
Warning: Never let a SQL agent run against a production database with write access. Use a read-only connection string and a read-only database user. Even with
temperature=0, LLMs occasionally generateUPDATEorDELETEstatements when asked ambiguous questions like "fix the data for customer X."
Some of your most valuable data lives behind APIs — Jira, Salesforce, GitHub, internal microservices. These agents need to handle rate limiting, pagination, and the fact that API responses aren't documents — they're structured data that needs transformation before it's meaningful in a RAG context.
import httpx
import json
from datetime import datetime, timedelta
from functools import lru_cache
class JiraAgent:
def __init__(self, jira_base_url: str, api_token: str,
agent_id: str = "jira_agent"):
self.agent_id = agent_id
self.base_url = jira_base_url
self.headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json"
}
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
self._cache: dict = {}
self._cache_ttl = timedelta(minutes=15)
def _build_jql_query(self, natural_language_question: str) -> str:
"""Convert natural language to JQL using the LLM."""
jql_prompt = f"""Convert this question into a Jira Query Language (JQL) query.
Question: {natural_language_question}
Return ONLY the JQL query, no explanation. Examples:
- "bugs in the authentication service from last month" →
project = AUTH AND issuetype = Bug AND created >= -30d
- "open tickets assigned to the platform team" →
project = PLATFORM AND status != Done AND assignee is not EMPTY
JQL query:"""
response = self.llm.invoke(jql_prompt)
return response.content.strip()
async def _fetch_issues(self, jql: str, max_results: int = 20) -> list[dict]:
cache_key = f"{jql}:{max_results}"
if cache_key in self._cache:
entry = self._cache[cache_key]
if datetime.now() - entry["timestamp"] < self._cache_ttl:
return entry["data"]
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/rest/api/3/search",
headers=self.headers,
params={"jql": jql, "maxResults": max_results,
"fields": "summary,status,assignee,description,priority,created,updated"}
)
response.raise_for_status()
issues = response.json().get("issues", [])
self._cache[cache_key] = {"data": issues, "timestamp": datetime.now()}
return issues
def _format_issues_for_context(self, issues: list[dict]) -> str:
"""Transform raw Jira API response into readable context for the LLM."""
formatted = []
for issue in issues:
fields = issue.get("fields", {})
description = fields.get("description", {})
# Jira returns description as Atlassian Document Format (ADF)
# We need to extract plain text
description_text = self._extract_adf_text(description)
formatted.append(f"""
Issue: {issue['key']}
Summary: {fields.get('summary', 'N/A')}
Status: {fields.get('status', {}).get('name', 'N/A')}
Priority: {fields.get('priority', {}).get('name', 'N/A')}
Assignee: {fields.get('assignee', {}).get('displayName', 'Unassigned') if fields.get('assignee') else 'Unassigned'}
Created: {fields.get('created', 'N/A')[:10]}
Description: {description_text[:500]}
""")
return "\n---\n".join(formatted)
def _extract_adf_text(self, adf_content) -> str:
"""Recursively extract plain text from Atlassian Document Format."""
if not adf_content:
return ""
if isinstance(adf_content, str):
return adf_content
if isinstance(adf_content, dict):
if adf_content.get("type") == "text":
return adf_content.get("text", "")
content = adf_content.get("content", [])
return " ".join(self._extract_adf_text(item) for item in content)
if isinstance(adf_content, list):
return " ".join(self._extract_adf_text(item) for item in adf_content)
return ""
async def query(self, agent_query: AgentQuery) -> AgentResponse:
try:
jql = self._build_jql_query(agent_query.question)
issues = await self._fetch_issues(jql)
if not issues:
return AgentResponse(
agent_id=self.agent_id,
answer=f"No Jira issues found matching the query criteria (JQL: {jql})",
sources=[],
confidence=ConfidenceLevel.NO_DATA,
raw_retrieved=[],
metadata={"jql_used": jql}
)
formatted_context = self._format_issues_for_context(issues)
synthesis_prompt = f"""Based on the following Jira issues, answer the question.
{formatted_context}
Question: {agent_query.question}
Provide a summary that directly addresses the question, referencing specific issue keys where relevant."""
response = await asyncio.to_thread(self.llm.invoke, synthesis_prompt)
return AgentResponse(
agent_id=self.agent_id,
answer=response.content,
sources=[{"id": i["key"], "title": i["fields"]["summary"],
"url": f"{self.base_url}/browse/{i['key']}"}
for i in issues],
confidence=ConfidenceLevel.HIGH,
raw_retrieved=[json.dumps(issues[:3])],
metadata={"jql_used": jql, "issues_retrieved": len(issues)}
)
except httpx.HTTPStatusError as e:
return AgentResponse(
agent_id=self.agent_id, answer="", sources=[],
confidence=ConfidenceLevel.NO_DATA, raw_retrieved=[],
error=f"Jira API error {e.response.status_code}: {e.response.text[:200]}"
)
The orchestrator is where multi-agent systems earn their complexity budget. It has three responsibilities: deciding which agents to invoke, managing the execution order and dependencies between them, and synthesizing their outputs into a coherent final answer.
The orchestrator's first task is understanding what the user is actually asking and mapping that to the right combination of agents. This is itself a classification problem — and crucially, it needs to recognize multi-source questions that require multiple agents.
from typing import Any
import json
@dataclass
class OrchestratorPlan:
"""Represents the orchestrator's execution plan for a query."""
routing_reason: str
parallel_agents: list[str] # These can run simultaneously
sequential_steps: list[list[str]] # Each inner list runs in parallel,
# outer list is sequential
synthesis_required: bool
synthesis_instruction: str
class MultiAgentOrchestrator:
def __init__(self, agents: dict[str, Any]):
self.agents = agents # agent_id -> agent instance
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
self.execution_log: list[dict] = []
async def _plan_execution(self, query: str) -> OrchestratorPlan:
"""Use the LLM to decompose the query into an execution plan."""
agent_descriptions = "\n".join([
f"- {agent_id}: {agent.agent_id}"
for agent_id, agent in self.agents.items()
])
planning_prompt = f"""You are an orchestrator for a multi-agent information retrieval system.
Available agents:
- document_agent: Retrieves from legal documents, contracts, technical PDFs, and policy documentation
- sql_agent: Queries the sales database (orders, customers, products, returns)
- jira_agent: Retrieves from Jira for bug reports, feature tickets, and engineering tasks
User query: {query}
Analyze this query and create an execution plan. Return a JSON object with this exact structure:
{{
"routing_reason": "explanation of why these agents are needed",
"agents_needed": ["agent_id_1", "agent_id_2"],
"can_parallelize": true/false,
"synthesis_required": true/false,
"synthesis_instruction": "how to combine the results if synthesis needed"
}}
Rules:
- Only include agents whose data is actually relevant to the query
- can_parallelize is true when agents are independent (not using each other's outputs)
- synthesis_required is true when the final answer requires combining information from multiple agents
- synthesis_instruction should describe HOW to resolve conflicts or combine perspectives"""
response = await asyncio.to_thread(self.llm.invoke, planning_prompt)
try:
plan_data = json.loads(response.content)
return OrchestratorPlan(
routing_reason=plan_data["routing_reason"],
parallel_agents=plan_data["agents_needed"] if plan_data["can_parallelize"] else [],
sequential_steps=[[a] for a in plan_data["agents_needed"]] if not plan_data["can_parallelize"] else [plan_data["agents_needed"]],
synthesis_required=plan_data["synthesis_required"],
synthesis_instruction=plan_data["synthesis_instruction"]
)
except (json.JSONDecodeError, KeyError) as e:
# Fallback: route to all agents in parallel
return OrchestratorPlan(
routing_reason="Planning failed, using fallback routing",
parallel_agents=list(self.agents.keys()),
sequential_steps=[list(self.agents.keys())],
synthesis_required=True,
synthesis_instruction="Combine all available information to answer the question."
)
async def _execute_parallel(
self,
agent_ids: list[str],
query: AgentQuery
) -> dict[str, AgentResponse]:
"""Execute multiple agents concurrently."""
tasks = {
agent_id: self.agents[agent_id].query(query)
for agent_id in agent_ids
if agent_id in self.agents
}
results = await asyncio.gather(
*tasks.values(),
return_exceptions=True
)
response_map = {}
for agent_id, result in zip(tasks.keys(), results):
if isinstance(result, Exception):
response_map[agent_id] = AgentResponse(
agent_id=agent_id,
answer="",
sources=[],
confidence=ConfidenceLevel.NO_DATA,
raw_retrieved=[],
error=str(result)
)
else:
response_map[agent_id] = result
return response_map
async def _synthesize(
self,
original_query: str,
agent_responses: dict[str, AgentResponse],
synthesis_instruction: str
) -> str:
"""Combine outputs from multiple agents into a coherent final answer."""
# Filter out NO_DATA responses for synthesis
useful_responses = {
agent_id: resp
for agent_id, resp in agent_responses.items()
if resp.confidence != ConfidenceLevel.NO_DATA and not resp.error
}
if not useful_responses:
return "I was unable to find relevant information across any of the available data sources."
# Build synthesis context with clear source attribution
synthesis_context = ""
for agent_id, resp in useful_responses.items():
synthesis_context += f"""
=== {agent_id.upper()} (Confidence: {resp.confidence.value}) ===
{resp.answer}
Sources: {', '.join(s.get('title', s.get('id', 'unknown')) for s in resp.sources[:3])}
"""
synthesis_prompt = f"""You are synthesizing information from multiple specialized data agents.
Original question: {original_query}
Synthesis instruction: {synthesis_instruction}
Agent responses:
{synthesis_context}
Instructions:
1. Synthesize a coherent answer that directly addresses the question
2. Where agents provide complementary information, combine it naturally
3. Where agents CONTRADICT each other, explicitly acknowledge the discrepancy and indicate which source is more authoritative for that specific claim
4. Cite which agent/source supports each significant claim using [source_name] notation
5. If any agents returned no useful data, mention what information is missing
Final synthesized answer:"""
response = await asyncio.to_thread(self.llm.invoke, synthesis_prompt)
return response.content
async def run(self, user_query: str) -> dict:
"""Main entry point: plan, execute, and synthesize."""
plan = await self._plan_execution(user_query)
all_responses: dict[str, AgentResponse] = {}
agent_query = AgentQuery(question=user_query)
# Execute according to plan (parallel steps within sequential stages)
for step_agents in plan.sequential_steps:
step_responses = await self._execute_parallel(step_agents, agent_query)
all_responses.update(step_responses)
# For sequential steps: inject prior results as context for next step
if len(plan.sequential_steps) > 1:
prior_context = "\n".join([
f"{aid}: {resp.answer[:300]}"
for aid, resp in all_responses.items()
if resp.confidence != ConfidenceLevel.NO_DATA
])
agent_query = AgentQuery(
question=user_query,
context=prior_context
)
# Synthesize if needed
if plan.synthesis_required and len(all_responses) > 1:
final_answer = await self._synthesize(
user_query, all_responses, plan.synthesis_instruction
)
else:
# Single agent or no synthesis needed
best_response = max(
all_responses.values(),
key=lambda r: {"high": 3, "medium": 2, "low": 1, "no_data": 0}[r.confidence.value]
)
final_answer = best_response.answer
# Log execution for debugging
self.execution_log.append({
"query": user_query,
"plan": plan,
"responses": {k: {"confidence": v.confidence.value, "error": v.error}
for k, v in all_responses.items()},
"agents_invoked": list(all_responses.keys())
})
return {
"answer": final_answer,
"agent_responses": all_responses,
"execution_plan": plan,
"sources": [
source
for resp in all_responses.values()
for source in resp.sources
]
}
The code above gets you to a working system. Now let's talk about the edge cases and architectural challenges that will actually determine whether it succeeds in production.
When your SQL Agent says Q3 churn was 12% and your Document Agent finds a board report claiming it was 8%, you have a genuine conflict. Your synthesis prompt needs to do more than average them — it needs to reason about why they differ (different time ranges? different churn definitions? different customer segments?) and surface that ambiguity rather than resolve it silently.
One practical technique: give each agent a data_freshness field in its response metadata, and train your synthesis prompt to prefer more recent data for factual claims while preserving older data for historical context.
# Add to AgentResponse
@dataclass
class AgentResponse:
# ... existing fields ...
data_freshness: Optional[str] = None # ISO timestamp of most recent source
data_scope: Optional[str] = None # Description of what this data covers
Then in synthesis, include this in the conflict resolution logic:
conflict_guidance = """
When two sources conflict:
1. Note the conflict explicitly: "Source A reports X while Source B reports Y"
2. Check data_freshness: prefer more recent data for current-state claims
3. Check data_scope: sources may cover different populations or time periods
4. If scope explains the conflict, explain that: "This discrepancy likely reflects..."
5. If scope does NOT explain the conflict, flag it: "These sources appear to contradict
each other on the same claim. Human verification recommended."
"""
When you're running five agents and each returns 1500 tokens of context, you're at 7500 tokens before you've written a single word of synthesis prompt. At scale, you need to be deliberate about your context budget.
The right approach is progressive compression: retrieve more, then compress what gets passed to synthesis.
async def _compress_agent_response(
self,
agent_id: str,
response: AgentResponse,
target_tokens: int
) -> str:
"""Compress an agent response to a target token count while preserving key facts."""
# Rough token estimation: 1 token ≈ 4 characters
estimated_tokens = len(response.answer) // 4
if estimated_tokens <= target_tokens:
return response.answer
compression_prompt = f"""Compress the following information to approximately {target_tokens} tokens.
CRITICAL: Preserve all specific numbers, dates, names, and factual claims.
Remove: explanatory prose, repetition, qualifications that don't affect the core facts.
Original ({estimated_tokens} estimated tokens):
{response.answer}
Compressed version:"""
compressed = await asyncio.to_thread(self.llm.invoke, compression_prompt)
return f"[Compressed from {agent_id}]: {compressed.content}"
Apply this before synthesis when total context would exceed your target:
SYNTHESIS_CONTEXT_BUDGET = 6000 # tokens
per_agent_budget = SYNTHESIS_CONTEXT_BUDGET // len(useful_responses)
for agent_id, resp in useful_responses.items():
compressed = await self._compress_agent_response(agent_id, resp, per_agent_budget)
# use compressed instead of resp.answer in synthesis
In production, external agents fail. The Jira API has rate limits. Your vector store occasionally times out. Your synthesis LLM call occasionally returns malformed JSON. You need circuit breakers.
import time
from collections import deque
class CircuitBreaker:
def __init__(self, agent_id: str, failure_threshold: int = 3,
reset_timeout: float = 60.0):
self.agent_id = agent_id
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self.failures = deque(maxlen=failure_threshold)
self.state = "closed" # closed = normal, open = failing, half-open = testing
self.last_failure_time: float = 0
def record_failure(self):
self.failures.append(time.time())
if len(self.failures) >= self.failure_threshold:
self.state = "open"
self.last_failure_time = time.time()
def record_success(self):
self.failures.clear()
self.state = "closed"
def is_available(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
if time.time() - self.last_failure_time > self.reset_timeout:
self.state = "half-open"
return True
return False
return True # half-open: allow one test request
Wrap agent calls with circuit breaker logic in the orchestrator:
async def _safe_agent_call(
self,
agent_id: str,
agent_query: AgentQuery
) -> AgentResponse:
breaker = self.circuit_breakers.get(agent_id)
if breaker and not breaker.is_available():
return AgentResponse(
agent_id=agent_id,
answer="",
sources=[],
confidence=ConfidenceLevel.NO_DATA,
raw_retrieved=[],
error=f"Agent {agent_id} is temporarily unavailable (circuit open)"
)
try:
response = await asyncio.wait_for(
self.agents[agent_id].query(agent_query),
timeout=30.0 # Hard timeout per agent
)
if breaker:
breaker.record_success()
return response
except asyncio.TimeoutError:
if breaker:
breaker.record_failure()
return AgentResponse(
agent_id=agent_id, answer="", sources=[],
confidence=ConfidenceLevel.NO_DATA, raw_retrieved=[],
error=f"Agent {agent_id} timed out after 30 seconds"
)
Key insight: A multi-agent system that degrades gracefully — returning partial answers with explicit acknowledgment of what's missing — is dramatically more useful than one that fails completely when any single component errors. Always design for partial success.
Evaluation is where multi-agent systems get genuinely tricky, because failures are often emergent: each individual agent might be performing well, but their combined output is wrong or misleading.
Evaluate each agent independently first. For document and API agents, use RAG-specific metrics:
RAGAS is the standard library for this. Wire it up per-agent:
from ragas import evaluate
from ragas.metrics import faithfulness, context_precision, context_recall
from datasets import Dataset
def evaluate_agent(agent_responses: list[dict]) -> dict:
"""
agent_responses: list of {question, answer, contexts, ground_truth}
"""
dataset = Dataset.from_list(agent_responses)
results = evaluate(
dataset,
metrics=[faithfulness, context_precision, context_recall]
)
return results
For system-level evaluation, you need to test the orchestrator's routing decisions separately from the synthesis quality. Create a routing evaluation set:
routing_test_cases = [
{
"query": "How many enterprise customers churned in Q3?",
"expected_agents": ["sql_agent"],
"should_not_invoke": ["document_agent", "jira_agent"]
},
{
"query": "What does our data retention policy say, and are there any compliance tickets open for it?",
"expected_agents": ["document_agent", "jira_agent"],
"should_not_invoke": ["sql_agent"]
},
{
"query": "Which product had the most returns last quarter and are there any bugs related to it?",
"expected_agents": ["sql_agent", "jira_agent"],
"should_not_invoke": []
}
]
Test routing accuracy separately from answer quality. It's common to find that routing is 95% accurate but synthesis is degrading quality — or vice versa. You can't optimize what you can't measure independently.
Build a two-agent RAG system that answers questions about a software product by combining information from its documentation (a set of Markdown files) and its issue tracker (a CSV file simulating Jira exports).
Part 1: Set up the Document Agent
DocumentAgent using the pattern above, backed by a local Chroma instance.Part 2: Set up the Issues Agent
issue_id, title, status, priority, description, created_date.CSVAgent that loads this file into a pandas DataFrame, uses the LLM to generate a pandas query from natural language, executes it, and formats the result.Part 3: Wire the Orchestrator
MultiAgentOrchestrator with both agents registered.Stretch challenge: Add routing evaluation. For 10 test queries where you know which agent(s) should be invoked, measure the orchestrator's routing accuracy. Then try to improve it by improving the planning prompt.
Mistake 1: Giving agents overlapping responsibilities
If both your Document Agent and your SQL Agent are allowed to answer questions about pricing, you'll get contradictory outputs from the same source of truth split across two systems. Define agent boundaries at the data layer, not the question layer.
Mistake 2: Not handling the "I don't know" case properly
Agents that return generic "I couldn't find relevant information" responses are worse than useless — they're misleading. A well-designed agent should return ConfidenceLevel.NO_DATA with a specific explanation: what it searched, how it searched, and why it believes the information isn't there. This helps the orchestrator decide whether to retry with different parameters.
Mistake 3: Chaining LLM calls for routing without validation
When your orchestrator uses an LLM to plan execution, that plan can be malformed. Always validate that planned agents actually exist in your agent registry before executing. An agent_id hallucinated by the planner will cause a KeyError at exactly the worst possible moment.
# Always validate before executing
valid_agents = [aid for aid in plan.agents_needed if aid in self.agents]
if len(valid_agents) < len(plan.agents_needed):
invalid = set(plan.agents_needed) - set(valid_agents)
logger.warning(f"Planner hallucinated non-existent agents: {invalid}")
Mistake 4: Synchronous agents in an async orchestrator
If your agents make synchronous HTTP calls or database queries, running them in an asyncio.gather() won't actually parallelize them — they'll block the event loop. Always wrap synchronous I/O in asyncio.to_thread().
Mistake 5: Losing provenance in synthesis
The synthesis step, if not carefully prompted, will merge information from multiple agents into smooth prose that obscures which claim came from which source. This makes the output unfalsifiable. Require explicit source attribution in your synthesis prompt and verify it in testing.
Mistake 6: Over-engineering the orchestrator
The first version of your orchestrator should be simple — almost embarrassingly so. A hard-coded routing table based on keyword classification will outperform LLM-based routing on latency and reliability for well-defined query taxonomies. Add LLM-based routing only when your query space genuinely requires semantic understanding to route correctly. The cost of LLM-based routing is two extra round trips on every query.
Central orchestrator vs. peer-to-peer agents: The architecture above uses a central orchestrator. Alternatively, agents can communicate peer-to-peer, where each agent can request information from others. Peer-to-peer enables more flexible workflows (an agent can decide mid-task that it needs information from another) but creates debugging nightmares and makes it easy to create infinite loops. Start central.
Shared context store vs. message passing: Instead of passing context between agents as strings, you can maintain a shared context store (a dictionary in memory, or Redis for distributed systems) that all agents read from and write to. This is cleaner for complex workflows but introduces concurrency issues — two agents writing the same key simultaneously will produce nondeterministic behavior.
Stateless vs. stateful agents: Stateless agents (no memory between calls) are simpler to reason about and easier to scale horizontally. Stateful agents (maintaining conversation history, learned preferences, accumulated context) enable more sophisticated multi-turn interactions but require careful state management. For most production RAG systems, keep agents stateless and manage state at the orchestrator level.
Multi-agent RAG architecture solves the fundamental mismatch between the heterogeneity of real enterprise data and the homogeneity assumptions baked into single-agent retrieval pipelines. By giving each data source a specialized agent with its own retrieval strategy, output schema, and domain-specific prompting, you get systems that scale gracefully as your data landscape grows.
The key architectural decisions to internalize:
Agent contracts are your API. Define inputs, outputs, and scope before writing any agent code. Undefined scope is the most common source of agent overlap and routing confusion.
Separate data source agents from reasoning agents. Keep retrieval and synthesis as distinct concerns. Your SQL Agent should return structured data; it should not generate a three-paragraph executive summary.
Design for partial failure. Every external data source will be unavailable sometimes. An orchestrator that returns a partially-sourced answer with clear attribution is more valuable than one that fails completely.
Evaluate at every layer. Routing accuracy, per-agent retrieval quality, and synthesis faithfulness are independent failure modes that require independent evaluation.
Start simple, measure, then add complexity. The pattern of starting with hard-coded routing and upgrading to LLM-based routing only when you have evidence you need it applies throughout this architecture.
Where to go next:
The architecture in this lesson will handle a wide range of production scenarios. When you hit its limits, they'll be legible — you'll know exactly which component to evolve, because you built each one to be independently observable and replaceable.
Learning Path: RAG & AI Agents