Router/Dispatcher Agent Pattern: Intent Classification & Routing

[yasha-dev1]

Overview

The Router/Dispatcher pattern uses a central agent to analyze user requests, classify intent, and route to specialized agents. This is a meta-agent pattern that enables building modular AI systems where each specialist agent handles a specific domain. The router performs dynamic task delegation using LLM-based classification, semantic similarity, or function calling.

How It Works

Architecture

User Query
    ↓
Router/Dispatcher Agent (LLM-based intent classification)
    ↓
    ├─→ Specialist Agent A (e.g., Billing)
    ├─→ Specialist Agent B (e.g., Technical Support)
    ├─→ Specialist Agent C (e.g., Sales)
    └─→ Specialist Agent D (e.g., General Assistant)

Routing Approaches

1. LLM-based Classification: Router uses LLM to classify query and select agent
2. Semantic Similarity: Embed query and compare to agent descriptions
3. Function Calling: Define agents as "tools" and let LLM route via function calling
4. Hybrid: Combine multiple strategies with fallback

Flow Example

User: "I was charged twice for my subscription, can you refund one?"

ROUTER AGENT:
  → Classify intent: "billing_issue"
  → Select specialist: BillingAgent
  → Route with context

BILLING AGENT (specialist):
  → search_orders(user_id) → [order1, order2]
  → detect_duplicate_charge() → True
  → initiate_refund(order2)
  → "I've processed a refund for the duplicate charge..."

RESPONSE: "I've processed a refund for the duplicate charge. You should see it in 5-7 business days."

Reference Implementations

• Google Cloud: Multi-Agent Coordinator Pattern - Official design patterns
• AWS: Routing & Dynamic Dispatch - Routing strategies
• Google Developer Blog: Multi-Agent Patterns in ADK - Implementation guide
• Azure: AI Agent Orchestration Patterns - Microsoft's approach
• InfoQ: Eight Essential Multi-Agent Design Patterns - Pattern overview

Proposed PyWorkflow Implementation

from pyworkflow import workflow, agent, start_child_workflow
from pyworkflow.agents import RouterAgent, SpecialistAgent

# Define specialist agents
@agent(
    pattern="tool_calling",
    model="claude-sonnet-3-7",
    tools=[search_orders, process_refund, update_subscription]
)
async def billing_agent(query: str):
    """Handle billing, payments, refunds, and subscription issues."""
    pass

@agent(
    pattern="rag",
    model="claude-sonnet-3-7",
    tools=[search_docs, search_tickets, create_ticket]
)
async def support_agent(query: str):
    """Handle technical support questions and troubleshooting."""
    pass

@agent(
    pattern="plan_execute",
    planner_model="claude-opus-4-6",
    executor_model="claude-sonnet-3-7",
    tools=[search_products, check_inventory, create_quote]
)
async def sales_agent(query: str):
    """Handle product inquiries, quotes, and sales questions."""
    pass

# Create router agent
@agent(
    pattern="router",
    model="claude-sonnet-3-7",
    specialists={
        "billing": billing_agent,
        "support": support_agent,
        "sales": sales_agent,
    },
    routing_strategy="llm",  # Options: "llm", "semantic", "function_calling", "hybrid"
    default_agent="support",  # Fallback if no clear match
)
async def customer_service_router(query: str):
    """
    Route customer queries to specialized agents.
    
    The router:
    1. Analyzes the user query
    2. Classifies intent (billing, support, sales, etc.)
    3. Routes to appropriate specialist agent
    4. Returns specialist's response
    """
    pass

# Use the router
result = await customer_service_router.run(
    "I was charged twice for my monthly subscription"
)
# → Routes to billing_agent
# → billing_agent processes refund
# → Returns result

print(result.answer)
print(result.routed_to)  # "billing"
print(result.specialist_trace)  # Full execution log from specialist

LLM-Based Routing Implementation

class RouterAgent:
    async def route(self, query: str) -> tuple[str, SpecialistAgent]:
        """Classify intent and select specialist using LLM."""
        
        system_prompt = f"""
        You are a routing assistant. Classify the user query into one of these categories:
        
        CATEGORIES:
        - billing: payments, refunds, subscriptions, invoices
        - support: technical issues, bugs, how-to questions
        - sales: product inquiries, pricing, quotes
        
        Respond with ONLY the category name (billing/support/sales).
        """
        
        response = await self.llm.generate(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": query}
            ]
        )
        
        intent = response.text.strip().lower()
        
        # Record routing decision
        await ctx.record_event(EventType.AGENT_ROUTING_DECISION, {
            "query": query,
            "classified_intent": intent,
            "routed_to": intent,
            "confidence": None,  # LLM doesn't provide confidence
            "strategy": "llm"
        })
        
        agent = self.specialists.get(intent, self.specialists[self.default_agent])
        return intent, agent

Semantic Similarity Routing

class RouterAgent:
    async def route_semantic(self, query: str) -> tuple[str, SpecialistAgent]:
        """Route using semantic similarity between query and agent descriptions."""
        
        # Embed user query
        query_embedding = await self.embedding_model.embed(query)
        
        # Compute similarity to each agent's description
        similarities = {}
        for name, agent in self.specialists.items():
            agent_embedding = await self.embedding_model.embed(agent.description)
            similarity = cosine_similarity(query_embedding, agent_embedding)
            similarities[name] = similarity
        
        # Select agent with highest similarity
        best_match = max(similarities, key=similarities.get)
        confidence = similarities[best_match]
        
        await ctx.record_event(EventType.AGENT_ROUTING_DECISION, {
            "query": query,
            "classified_intent": best_match,
            "routed_to": best_match,
            "confidence": confidence,
            "all_similarities": similarities,
            "strategy": "semantic"
        })
        
        return best_match, self.specialists[best_match]

Function Calling Routing

class RouterAgent:
    async def route_function_calling(self, query: str) -> tuple[str, SpecialistAgent]:
        """Route by defining agents as tools and using function calling."""
        
        # Define each specialist as a "tool"
        tools = [
            {
                "name": "billing_agent",
                "description": "Handle billing, payments, refunds, and subscription issues",
                "input_schema": {"type": "object", "properties": {"query": {"type": "string"}}}
            },
            {
                "name": "support_agent",
                "description": "Handle technical support questions and troubleshooting",
                "input_schema": {"type": "object", "properties": {"query": {"type": "string"}}}
            },
            ...
        ]
        
        response = await self.llm.generate(
            model=self.model,
            messages=[{"role": "user", "content": query}],
            tools=tools
        )
        
        # LLM selects which "tool" (agent) to call
        selected_agent_name = response.tool_calls[0].name.replace("_agent", "")
        
        await ctx.record_event(EventType.AGENT_ROUTING_DECISION, {
            "query": query,
            "classified_intent": selected_agent_name,
            "routed_to": selected_agent_name,
            "strategy": "function_calling"
        })
        
        return selected_agent_name, self.specialists[selected_agent_name]

Executing Specialist Agent

class RouterAgent:
    async def run(self, query: str):
        """Route to specialist and execute."""
        
        # 1. Classify and route
        intent, specialist_agent = await self.route(query)
        
        # 2. Execute specialist as child workflow
        result = await start_child_workflow(
            specialist_agent,
            query,
            wait_for_completion=True,  # Wait for specialist to finish
        )
        
        await ctx.record_event(EventType.AGENT_COMPLETED, {
            "routed_to": intent,
            "specialist_run_id": result.run_id,
            "answer": result.answer
        })
        
        return {
            "answer": result.answer,
            "routed_to": intent,
            "specialist_trace": result.events
        }

Event Types

Router agents record these events:

1. AGENT_STARTED - Router begins execution

   {
     "run_id": "abc123",
     "query": "...",
     "specialists": ["billing", "support", "sales"],
     "routing_strategy": "llm"
   }

2. AGENT_ROUTING_DECISION - Router classifies and routes

   {
     "query": "I was charged twice",
     "classified_intent": "billing",
     "routed_to": "billing",
     "confidence": 0.95,  # For semantic routing
     "strategy": "llm"
   }

3. AGENT_SPECIALIST_STARTED - Specialist agent begins

   {
     "specialist": "billing",
     "child_run_id": "child_xyz",
     "query": "..."
   }

4. AGENT_SPECIALIST_COMPLETED - Specialist finishes

   {
     "specialist": "billing",
     "child_run_id": "child_xyz",
     "result": "...",
     "duration_ms": 2500
   }

5. AGENT_ROUTING_FALLBACK - Fell back to default agent

   {
     "reason": "No clear match, confidence < 0.5",
     "fallback_to": "support"
   }

6. AGENT_COMPLETED / AGENT_FAILED

Implementation Details

Agent Registry

# pyworkflow/agents/registry.py
class AgentRegistry:
    """Global registry of specialist agents for routing."""
    
    _agents: dict[str, Agent] = {}
    
    @classmethod
    def register(cls, name: str, agent: Agent, description: str):
        """Register a specialist agent."""
        cls._agents[name] = {
            "agent": agent,
            "description": description,
            "embedding": None,  # Lazy-loaded for semantic routing
        }
    
    @classmethod
    def get(cls, name: str) -> Agent:
        """Get agent by name."""
        return cls._agents[name]["agent"]

# Auto-register when defining specialists
@agent(
    pattern="tool_calling",
    register_as="billing",  # Register in global registry
    description="Handle billing, payments, refunds, and subscription issues"
)
async def billing_agent(query: str):
    pass

Hybrid Routing Strategy

class RouterAgent:
    async def route_hybrid(self, query: str) -> tuple[str, SpecialistAgent]:
        """
        Hybrid routing:
        1. Try LLM-based classification
        2. If confidence low, fall back to semantic similarity
        3. If still uncertain, use default agent
        """
        
        # Try LLM classification
        llm_intent = await self._classify_with_llm(query)
        
        # Validate with semantic similarity
        query_embedding = await self.embedding_model.embed(query)
        semantic_scores = await self._compute_semantic_scores(query_embedding)
        
        llm_semantic_score = semantic_scores.get(llm_intent, 0)
        
        if llm_semantic_score > 0.7:
            # High confidence - trust LLM
            return llm_intent, self.specialists[llm_intent]
        
        elif max(semantic_scores.values()) > 0.8:
            # LLM uncertain but semantic is confident
            best_semantic = max(semantic_scores, key=semantic_scores.get)
            
            await ctx.record_event(EventType.AGENT_ROUTING_FALLBACK, {
                "reason": "LLM uncertain, using semantic match",
                "llm_intent": llm_intent,
                "semantic_intent": best_semantic
            })
            
            return best_semantic, self.specialists[best_semantic]
        
        else:
            # Both uncertain - use default
            await ctx.record_event(EventType.AGENT_ROUTING_FALLBACK, {
                "reason": "Both LLM and semantic uncertain",
                "fallback_to": self.default_agent
            })
            
            return self.default_agent, self.specialists[self.default_agent]

Trade-offs

Pros

• Modularity: Each specialist can be developed, tested, and scaled independently
• Scalability: Add new specialists without modifying router logic
• Cost optimization: Route simple queries to lighter models, complex to powerful ones
• Domain expertise: Specialists can use domain-specific tools and knowledge
• Maintainability: Clear separation of concerns
• Parallel development: Different teams can work on different specialists

Cons

• Routing overhead: Extra LLM call(s) before specialist execution
• Error propagation: Router misclassification leads to wrong specialist
• Complexity: More moving parts than single-agent system
• Context loss: Router may not pass full context to specialist

Comparison to Single-Agent

Aspect	Router/Dispatcher	Single Agent
Modularity	High (separate specialists)	Low (monolithic)
Routing Accuracy	85-95% (depends on strategy)	N/A
Latency	Higher (router + specialist)	Lower
Scalability	High (scale specialists independently)	Low
Maintenance	Easier (modular)	Harder (monolithic)

When to Use Router Pattern

Use Router when:

• System handles multiple distinct domains (billing, support, sales)
• Different specialists need different tools/models
• You want to scale components independently
• Team structure maps to specialist boundaries

Don't use when:

• All queries are similar domain
• Single agent can handle all tasks
• Routing overhead not justified (latency-sensitive)

Related Issues

• ReAct Agent Pattern: Reasoning + Acting Loop #153 - ReAct Agent (can be used as specialist)
• Tool-Calling Agent Pattern: Direct Function Calling #156 - Tool-Calling Agent (can be used as specialist)
• Plan-and-Execute Agent Pattern: Two-Phase Planning #160 - Plan-and-Execute Agent (can be used as specialist)
• #[ISSUE_5] - RAG Agent (can be used as specialist)

References

• Google Cloud: Multi-Agent Design Patterns
• AWS: Routing and Dynamic Dispatch Patterns
• Google Developers: Multi-Agent Patterns in ADK
• Azure: AI Agent Orchestration Patterns
• InfoQ: Eight Essential Multi-Agent Design Patterns (2026)
• Agentic Design Pattern Series: Routing
• Botpress: Ultimate Guide to AI Agent Routing (2026)

Implementation Checklist

• Create pyworkflow/agents/router.py with RouterAgent class
• Implement LLM-based routing strategy
• Implement semantic similarity routing (with embeddings)
• Implement function calling routing
• Implement hybrid routing strategy
• Create AgentRegistry for specialist management
• Integrate start_child_workflow() for specialist execution
• Add event types: AGENT_ROUTING_DECISION, AGENT_ROUTING_FALLBACK
• Add confidence thresholds and fallback logic
• Create @agent(pattern="router") decorator
• Add tests for all routing strategies
• Document router pattern in examples/
• Add integration test with multiple specialists
• Support dynamic specialist registration