RAG Agents: Orchestrating Multi-Agent Systems

A simple RAG agent follows a linear pipeline: retrieval, generation, response. A multi-agent system orchestrates several specialized agents that collaborate to handle complex tasks. This guide explores architectures and patterns for building these systems.

Why Multi-Agents?

Limitations of Monolithic RAG

A classic RAG struggles with certain tasks:

The Multi-Agent Approach

Divide and conquer:

Complex question
       │
       ▼
┌─────────────────┐
│  ORCHESTRATOR   │ ← Decomposes and coordinates
└────────┬────────┘
         │
    ┌────┼────┬────────────┐
    ▼    ▼    ▼            ▼
┌──────┐┌──────┐┌──────┐┌──────┐
│Agent ││Agent ││Agent ││Agent │
│Search││Reason││Verify││Action│
└──────┘└──────┘└──────┘└──────┘
    │    │    │            │
    └────┴────┴────────────┘
                │
                ▼
         Final response

Orchestration Architectures

Pattern 1: Router

The orchestrator routes to the specialized agent.

DEVELOPERpython
from enum import Enum
from typing import Callable, Dict

class AgentType(Enum):
    FAQ = "faq"
    TECHNICAL = "technical"
    SALES = "sales"
    ESCALATION = "escalation"

class RouterOrchestrator:
    def __init__(self, agents: Dict[AgentType, Callable], classifier):
        self.agents = agents
        self.classifier = classifier

    async def process(self, query: str, context: dict = None) -> dict:
        """
        Route request to appropriate agent
        """
        # 1. Classify request
        classification = await self.classifier.classify(query)
        agent_type = AgentType(classification["agent"])

        # 2. Call agent
        agent = self.agents.get(agent_type)
        if not agent:
            agent = self.agents[AgentType.ESCALATION]

        result = await agent(query, context)

        return {
            "response": result["answer"],
            "agent_used": agent_type.value,
            "confidence": classification["confidence"],
            "sources": result.get("sources", [])
        }

class IntentClassifier:
    def __init__(self, llm):
        self.llm = llm

    async def classify(self, query: str) -> dict:
        prompt = f"""
        Classify this user request.

        Categories:
        - faq: General questions, policies, information
        - technical: Technical issues, bugs, configurations
        - sales: Pricing, purchases, subscriptions
        - escalation: Complaints, emergencies, sensitive requests

        Request: {query}

        Respond in JSON: {{"agent": "...", "confidence": 0.0-1.0}}
        """

        result = await self.llm.generate(prompt, temperature=0)
        return self._parse_json(result)

Pattern 2: Sequential Pipeline

Agents execute in sequence, each enriching the context.

DEVELOPERpython
from dataclasses import dataclass
from typing import List

@dataclass
class AgentResult:
    agent_name: str
    output: dict
    success: bool
    error: str = None

class SequentialPipeline:
    def __init__(self, agents: List[tuple]):
        """
        agents: List of (name, agent_callable)
        """
        self.agents = agents

    async def process(self, query: str, initial_context: dict = None) -> dict:
        """
        Execute agents in sequence
        """
        context = initial_context or {}
        context["original_query"] = query
        results = []

        for agent_name, agent in self.agents:
            try:
                result = await agent(query, context)

                agent_result = AgentResult(
                    agent_name=agent_name,
                    output=result,
                    success=True
                )

                # Enrich context for next agent
                context[f"{agent_name}_result"] = result
                context["last_result"] = result

            except Exception as e:
                agent_result = AgentResult(
                    agent_name=agent_name,
                    output={},
                    success=False,
                    error=str(e)
                )

            results.append(agent_result)

            # Stop if agent fails (optional)
            if not agent_result.success and self.stop_on_failure:
                break

        return {
            "final_result": context.get("last_result"),
            "pipeline_results": results,
            "context": context
        }

# Example pipeline
pipeline = SequentialPipeline([
    ("query_analyzer", QueryAnalyzerAgent()),
    ("retriever", RetrievalAgent()),
    ("fact_checker", FactCheckAgent()),
    ("generator", GenerationAgent()),
    ("citation_adder", CitationAgent())
])

Pattern 3: Parallel with Fusion

Multiple agents work in parallel, then results are fused.

DEVELOPERpython
import asyncio
from typing import List, Callable

class ParallelOrchestrator:
    def __init__(
        self,
        agents: List[tuple],
        fusion_agent: Callable
    ):
        self.agents = agents  # (name, agent, weight)
        self.fusion = fusion_agent

    async def process(self, query: str, context: dict = None) -> dict:
        """
        Execute agents in parallel then fuse
        """
        # Launch all agents in parallel
        tasks = [
            self._run_agent(name, agent, query, context)
            for name, agent, _ in self.agents
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Collect valid results
        valid_results = []
        for i, result in enumerate(results):
            name, _, weight = self.agents[i]
            if not isinstance(result, Exception):
                valid_results.append({
                    "agent": name,
                    "result": result,
                    "weight": weight
                })

        # Fuse results
        fused = await self.fusion(query, valid_results)

        return {
            "response": fused["answer"],
            "contributing_agents": [r["agent"] for r in valid_results],
            "fusion_confidence": fused.get("confidence")
        }

class FusionAgent:
    def __init__(self, llm):
        self.llm = llm

    async def __call__(self, query: str, results: List[dict]) -> dict:
        """
        Fuse responses from multiple agents
        """
        responses_text = "\n\n".join([
            f"Agent {r['agent']} (weight {r['weight']}):\n{r['result'].get('answer', 'N/A')}"
            for r in results
        ])

        prompt = f"""
        Synthesize these responses from different agents into a single coherent response.

        Original question: {query}

        Agent responses:
        {responses_text}

        Rules:
        1. Prioritize agents with higher weight
        2. In case of contradiction, indicate both viewpoints
        3. Cite sources if available
        4. Generate a single coherent response

        Synthesized response:
        """

        answer = await self.llm.generate(prompt, temperature=0.3)

        return {
            "answer": answer,
            "confidence": self._calculate_confidence(results)
        }

Pattern 4: ReAct (Reasoning + Acting)

The agent reasons, acts, observes, and iterates.

DEVELOPERpython
from typing import Dict, Any

class ReActAgent:
    def __init__(self, llm, tools: Dict[str, Callable], max_iterations: int = 5):
        self.llm = llm
        self.tools = tools
        self.max_iterations = max_iterations

    async def process(self, query: str) -> dict:
        """
        Execute ReAct pattern: Thought -> Action -> Observation
        """
        history = []
        final_answer = None

        for i in range(self.max_iterations):
            # Generate next thought/action
            step = await self._generate_step(query, history)

            if step["type"] == "thought":
                history.append({"thought": step["content"]})

            elif step["type"] == "action":
                # Execute action
                tool_name = step["tool"]
                tool_input = step["input"]

                if tool_name in self.tools:
                    observation = await self.tools[tool_name](tool_input)
                else:
                    observation = f"Tool '{tool_name}' not available"

                history.append({
                    "action": f"{tool_name}({tool_input})",
                    "observation": observation
                })

            elif step["type"] == "answer":
                final_answer = step["content"]
                break

        return {
            "answer": final_answer,
            "reasoning_trace": history,
            "iterations": i + 1
        }

# Tools for ReAct agent
tools = {
    "search_kb": lambda q: rag_search(q),
    "calculate": lambda expr: eval(expr),
    "get_current_date": lambda _: datetime.now().isoformat(),
    "check_inventory": lambda product_id: inventory_api.check(product_id)
}

Specialized Agents

Search Agent

DEVELOPERpython
class SearchAgent:
    def __init__(self, retrievers: Dict[str, Retriever], llm):
        self.retrievers = retrievers
        self.llm = llm

    async def __call__(self, query: str, context: dict) -> dict:
        """
        Agent specialized in multi-source search
        """
        # 1. Determine relevant sources
        sources = await self._select_sources(query)

        # 2. Search each source
        all_results = []
        for source_name in sources:
            retriever = self.retrievers.get(source_name)
            if retriever:
                results = await retriever.search(query)
                for r in results:
                    r["source"] = source_name
                all_results.extend(results)

        # 3. Re-rank results
        ranked = await self._rerank(query, all_results)

        return {
            "documents": ranked[:10],
            "sources_used": sources,
            "total_found": len(all_results)
        }

Verification Agent

DEVELOPERpython
class VerificationAgent:
    def __init__(self, llm, fact_db):
        self.llm = llm
        self.fact_db = fact_db

    async def __call__(self, query: str, context: dict) -> dict:
        """
        Verify claims against reliable sources
        """
        # Get response to verify
        answer = context.get("last_result", {}).get("answer", "")

        # 1. Extract claims
        claims = await self._extract_claims(answer)

        # 2. Verify each claim
        verifications = []
        for claim in claims:
            verification = await self._verify_claim(claim)
            verifications.append(verification)

        # 3. Determine overall score
        verified_count = sum(1 for v in verifications if v["verified"])
        total = len(verifications)

        return {
            "claims": verifications,
            "verification_score": verified_count / total if total > 0 else 1.0,
            "needs_correction": any(not v["verified"] for v in verifications)
        }

Action Agent

DEVELOPERpython
class ActionAgent:
    def __init__(self, action_registry: Dict[str, Callable], llm):
        self.actions = action_registry
        self.llm = llm

    async def __call__(self, query: str, context: dict) -> dict:
        """
        Execute actions based on request
        """
        # 1. Determine required actions
        action_plan = await self._plan_actions(query, context)

        # 2. Execute actions
        results = []
        for action in action_plan:
            if action["name"] in self.actions:
                try:
                    result = await self.actions[action["name"]](**action["params"])
                    results.append({
                        "action": action["name"],
                        "success": True,
                        "result": result
                    })
                except Exception as e:
                    results.append({
                        "action": action["name"],
                        "success": False,
                        "error": str(e)
                    })

        return {
            "actions_executed": results,
            "all_successful": all(r["success"] for r in results)
        }

# Action registry
action_registry = {
    "create_ticket": create_support_ticket,
    "send_email": send_email,
    "update_order": update_order_status,
    "schedule_callback": schedule_callback,
    "apply_discount": apply_discount_code
}

Failure Handling

Retry with Backoff

DEVELOPERpython
import asyncio
from functools import wraps

def with_retry(max_attempts: int = 3, backoff: float = 1.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        wait_time = backoff * (2 ** attempt)
                        await asyncio.sleep(wait_time)

            raise last_exception

        return wrapper
    return decorator

Graceful Fallback

DEVELOPERpython
class ResilientOrchestrator:
    def __init__(self, primary_agents, fallback_agent):
        self.primary = primary_agents
        self.fallback = fallback_agent

    async def process(self, query: str, context: dict = None) -> dict:
        """
        Execute with fallback on failure
        """
        try:
            # Try primary pipeline
            result = await self._run_primary(query, context)

            if self._is_valid_result(result):
                return result

        except Exception as e:
            logger.warning(f"Primary pipeline failed: {e}")

        # Fallback to simple agent
        return await self.fallback(query, context)

Monitoring and Observability

DEVELOPERpython
import time
from dataclasses import dataclass

@dataclass
class AgentTrace:
    agent_name: str
    start_time: float
    end_time: float
    input_query: str
    output: dict
    success: bool
    error: str = None

class TracingOrchestrator:
    def __init__(self, orchestrator, trace_store):
        self.orchestrator = orchestrator
        self.traces = trace_store

    async def process(self, query: str, context: dict = None) -> dict:
        """
        Execute with full tracing
        """
        trace_id = str(uuid.uuid4())
        traces = []

        # Wrapper to capture traces
        original_agents = self.orchestrator.agents.copy()

        for name, agent in original_agents.items():
            self.orchestrator.agents[name] = self._wrap_agent(name, agent, traces)

        try:
            result = await self.orchestrator.process(query, context)

            # Save traces
            await self.traces.save(trace_id, traces)

            result["trace_id"] = trace_id
            result["execution_time_ms"] = sum(
                (t.end_time - t.start_time) * 1000 for t in traces
            )

            return result

        finally:
            # Restore original agents
            self.orchestrator.agents = original_agents

Best Practices

1. Atomic Agents

Each agent should have a single, clear responsibility.

2. Timeouts Everywhere

Define timeouts to avoid blocking.

3. Structured Logging

Log each step for debugging.

4. Unit Tests per Agent

Test each agent independently before integration.

5. Circuit Breaker

Temporarily disable failing agents.

Learn More

• Conversational RAG - Memory and context
• LLM Generation - Optimize responses
• RAG Evaluation - Measure quality

---

Multi-Agent Orchestration with Ailog

Building a robust multi-agent architecture requires deep expertise. With Ailog, benefit from pre-built agent infrastructure:

• Specialized agents: search, verification, action
• Configurable orchestration without code
• Integrated monitoring with complete traces
• Automatic fallbacks and error handling
• Scalability for heavy loads

Try Ailog for free and deploy multi-agent assistants in a few clicks.