Multi-Agent Orchestration: Building RAG Systems Where Specialized Agents Collaborate Across Data Sources

Introduction

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:

• How to decompose a complex RAG problem into specialized agent roles with clear contracts
• How to build an orchestrator that routes queries, manages agent dependencies, and synthesizes results
• How to implement inter-agent communication patterns including sequential, parallel, and conditional routing
• How to handle partial failures, conflicting evidence across data sources, and context window management in multi-agent pipelines
• How to evaluate and debug multi-agent systems where failure modes are non-obvious and emergent

---

Prerequisites

This lesson assumes you're comfortable with:

• Building single-agent RAG pipelines with LangChain, LlamaIndex, or raw API calls
• Embedding models, vector stores, and semantic retrieval fundamentals
• Python async programming (asyncio, await)
• Basic familiarity with LLM function/tool calling APIs
• At least one production deployment of a retrieval-augmented system

---

Why Single-Agent RAG Breaks at Scale

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.

---

Designing the Agent Taxonomy

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.

Data Source Agents vs. Reasoning Agents

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.

Defining Agent Contracts

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).

---

Building Specialized Data Source Agents

Let's build three concrete agents that demonstrate the range of retrieval strategies you'll encounter in practice.

The Document Agent: Hierarchical Retrieval with Parent Documents

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)
            )

The SQL Agent: Text-to-SQL with Schema-Aware Retrieval

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)}"
            )

The API Agent: External Data Sources with Caching

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: Routing, Dependency Management, and Synthesis

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.

Query Analysis and Agent Selection

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
            ]
        }

---

Handling the Hard Problems

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.

Conflicting Evidence Across Sources

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."
"""

Context Window Budget Management

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

Circuit Breaking and Graceful Degradation

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"
        )

---

Evaluating Multi-Agent Systems

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.

Component-Level Evaluation

Evaluate each agent independently first. For document and API agents, use RAG-specific metrics:

• Context Precision: Are the retrieved chunks actually relevant to the query?
• Context Recall: Are all the chunks needed to answer the query present in the retrieved set?
• Faithfulness: Does the agent's answer contain claims not supported by its retrieved context?

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

System-Level Evaluation

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.

---

Hands-On Exercise

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

1. Create a folder with 5-10 Markdown files containing mock product documentation (installation guides, API reference, configuration options).
2. Build a DocumentAgent using the pattern above, backed by a local Chroma instance.
3. Test it in isolation: does it correctly answer "What are the supported authentication methods?" only from the documentation, without hallucination?

Part 2: Set up the Issues Agent

1. Create a CSV with columns: issue_id, title, status, priority, description, created_date.
2. Build a 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.
3. Test in isolation: does "How many open critical bugs exist?" return an accurate count?

Part 3: Wire the Orchestrator

4. Build the MultiAgentOrchestrator with both agents registered.
5. Test these queries, which span both agents:
   • "What does the docs say about rate limiting, and are there any open bugs related to rate limiting?"
   • "Is the OAuth2 authentication feature complete, or are there still open issues?"

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.

---

Common Mistakes & Troubleshooting

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.

---

Architecture Trade-offs Worth Discussing

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.

---

Summary & Next Steps

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:

1. 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.

2. 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.

3. 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.

4. Evaluate at every layer. Routing accuracy, per-agent retrieval quality, and synthesis faithfulness are independent failure modes that require independent evaluation.

5. 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:

• LangGraph for more sophisticated orchestration patterns, including cycles, conditional routing, and human-in-the-loop approval steps
• Hierarchical agent systems: orchestrators that spawn sub-orchestrators for particularly complex queries
• Agentic memory: giving your orchestrator persistent memory of past query patterns to improve routing accuracy over time
• Multi-modal agents: extending this architecture to handle image, audio, and structured data within the same orchestration framework
• Evaluation frameworks: building automated regression testing suites for multi-agent systems using tools like RAGAS, TruLens, and Langfuse

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.