Router

The Router is a central component of the AgentOpera framework that intelligently routes user messages to the most appropriate agent based on intent classification. This document explains how the Router system works and how to extend it with custom intents and domain-specific agents.

Core Concepts

Intent Classification

The Router system uses an LLM-based intent classifier to determine the user's intent from their messages. Each intent is associated with a specific agent that is specialized to handle tasks related to that intent.

Agent Registry

The Agent Registry maintains a catalog of available agents and their capabilities. It serves as a lookup service that maps intents to appropriate agents and provides descriptions of agent capabilities.

Semantic Router Agent

The Semantic Router Agent orchestrates the routing process. It receives user messages, determines the appropriate agent through intent classification, and forwards the messages to the selected agent.

Router Architecture

The Router architecture in AgentOpera consists of three key components:

1. Intent Classifier: Analyzes user messages to identify their intent.

2. Agent Registry: Maintains a catalog of available agents and their capabilities.

3. Semantic Router Agent: Routes messages to the appropriate agent based on the classified intent.

Implementation Details

Intent Registry

The Intent Registry manages the mapping between intents and agents:

from typing import Dict, Union, Set, Optional

class IntentRegistry:
    def __init__(
        self,
        intent_descriptions: Dict[str, str],
        agent_intent_mapping: Dict[str, str],
        agent_descriptions: Dict[str, Union[str, list[str]]],
        schema_overrides: Optional[Dict[str, dict]] = None
    ):
        self.intent_descriptions = intent_descriptions
        self.agent_intent_mapping = agent_intent_mapping
        self.agent_descriptions = agent_descriptions
        self.schema_overrides = schema_overrides or {}
    
    def get_agent_for_intent(self, intent: str) -> str:
        """Returns agent_id associated with the given intent."""
        return self.agent_intent_mapping.get(intent, "chat")

    def get_tool_schemas(self) -> list[dict]:
        """Builds OpenAI-compatible tool definitions for all registered intents."""
        # Implementation details...

LLM Intent Classifier

The LLM Intent Classifier uses a language model to determine the user's intent:

class LLMIntentClassifier(SingleIntentClassifier):
    def __init__(self, intent_registry: IntentRegistry, model="Llama-3.1-8B-Instruct"):
        self.intent_registry = intent_registry
        self.model_name = model
        self.client = self._set_model_client(model)
        # Additional initialization...
    
    async def classify_intent(self, message: TextMessage | MultiModalMessage, timeout: Optional[float] = 80.0) -> str:
        """Returns the intent with the highest confidence score."""
        # Implementation details...

Semantic Router Agent

The Semantic Router Agent handles routing logic:

class SemanticRouterAgent(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__("Semantic Router Agent")
        self._name = name
        self._initialized = False
        # Additional initialization...
        
    @message_handler
    async def route_to_agent(self, message: TextMessage | MultiModalMessage, ctx: MessageContext) -> None:
        # Implementation details for routing messages...

Quick Start Guide for Adding a New Business Agent

To add a new domain-specific agent to the Router system:

1. Define your intent: Create a clear description of what your agent handles

   intent_description = "financial_planning_intent": "Help with personal financial planning and investment strategies"

2. Create your agent implementation: Implement a RoutedAgent subclass

   from agentopera.engine import RoutedAgent, MessageContext, message_handler
   from agentopera.chatflow.messages import TextMessage
   
   class FinancialPlanningAgent(RoutedAgent):
       def __init__(self) -> None:
           super().__init__("Financial Planning Agent")
       
       @message_handler
       async def handle_planning_request(self, message: TextMessage, ctx: MessageContext) -> None:
           # Process financial planning request and respond
           await self.publish_message(
               TextMessage(content="Here's your financial plan...", source="financial_agent"),
               ctx.message_channel,
               message_id=ctx.message_id
           )

3. Update the intent registry: Add your intent and agent mapping

   from agentopera.router.agent_registry import AgentFactory
   
   # Add to existing dictionaries
   intent_descriptions = {
       "financial_planning_intent": "Help with personal financial planning and investment strategies",
       "chat_intent": "General conversation and queries"
   }
   
   agent_intent_mapping = {
       "financial_planning_intent": "financial_agent",
       "chat_intent": "chat_agent"
   }
   
   agent_descriptions = {
       "financial_agent": "Agent that provides financial planning and investment advice",
       "chat_agent": "General chat assistant"
   }
   
   # Create or update intent registry
   intent_registry = AgentFactory.create_intent_registry(
       intent_descriptions,
       agent_intent_mapping,
       agent_descriptions
   )

4. Register your agent with the runtime: Make it available for routing

   from agentopera.engine import SingleThreadedAgentRuntime
   from agentopera.router.semantic_router_agent import SemanticRouterAgent
   
   # Set up runtime
   runtime = SingleThreadedAgentRuntime()
   
   # Register router agent
   await SemanticRouterAgent.register(runtime, "router", lambda: SemanticRouterAgent("router"))
   
   # Register your custom agent
   await FinancialPlanningAgent.register(
       runtime, 
       "financial_agent", 
       lambda: FinancialPlanningAgent()
   )
   
   # Start runtime
   runtime.start()

5. Test your agent: Send a message that should trigger your agent's intent

   from agentopera.engine import AgentId
   
   # Send a test message to the router (will be classified and routed to your agent)
   await runtime.send_message(
       TextMessage(content="I need help planning for retirement", source="user"),
       AgentId("router", "default")
   )