AgentRouter

Overview

The AgentRouter is an embedding-based routing system that intelligently matches tasks to the most appropriate specialized agent using cosine similarity on embeddings. It uses LiteLLM’s embedding models to generate vector representations of both agents and tasks, enabling semantic matching for optimal agent selection. When a task is submitted, the router:

Generates an embedding vector for the task
Calculates cosine similarity between the task embedding and all agent embeddings
Returns the agent with the highest similarity score
Optionally updates agent embeddings with interaction history for improved matching over time

Installation

pip install -U swarms

Class Definition

from swarms.structs.agent_router import AgentRouter

Attributes

str

default:"text-embedding-ada-002"

The embedding model to use for generating embeddings. Supports various models like text-embedding-3-small, text-embedding-3-large, cohere/embed-english-v3.0, huggingface/microsoft/codebert-base, etc.

int

default:"1"

Number of agents to return in queries (currently supports returning the best match)

Optional[str]

default:"None"

API key for the embedding service. If not provided, will use environment variables.

Optional[str]

default:"None"

Custom API base URL for the embedding service.

Optional[List[AgentType]]

default:"None"

List of agents to initialize the router with. Each agent should have name, description, and system_prompt attributes.

Methods

add_agent()

Add a single agent to the router. The agent will be embedded using its name, description, and system prompt.

def add_agent(self, agent: AgentType) -> None

Parameters:

agent (AgentType): The agent to add. Must have name, description, and system_prompt attributes.

Raises:

Exception: If there’s an error generating the embedding or adding the agent.

add_agents()

Add multiple agents to the router at once.

def add_agents(self, agents: List[Union[AgentType, Callable, Any]]) -> None

Parameters:

agents (List[Union[AgentType, Callable, Any]]): List of agents to add.

find_best_agent()

Find the best matching agent for a given task using cosine similarity on embeddings.

def find_best_agent(self, task: str, *args, **kwargs) -> Optional[AgentType]

Parameters:

task (str): The task description to match against agents.

Returns: The best matching agent if found, None otherwise.

run()

Convenience method that calls find_best_agent. Run the agent router on a given task.

def run(self, task: str) -> Optional[AgentType]

Parameters:

task (str): The task description to match against agents.

Returns: The best matching agent if found, None otherwise.

update_agent_history()

Update the agent’s embedding in the router with its interaction history. This allows the router to learn from past interactions and improve matching over time.

def update_agent_history(self, agent_name: str) -> None

Parameters:

agent_name (str): The name of the agent to update.

This method updates the agent’s embedding to include its conversation history, which can improve future routing decisions based on what the agent has learned or discussed.

Usage Examples

Basic Medical Use Case

from swarms import Agent
from swarms.structs.agent_router import AgentRouter

# Initialize the router
agent_router = AgentRouter(
    embedding_model="text-embedding-ada-002",
    n_agents=1,
    agents=[
        Agent(
            agent_name="Symptom Checker",
            agent_description="Expert agent for initial triage and identifying possible causes based on symptom input.",
            system_prompt=(
                "You are a medical symptom checker agent. Ask clarifying questions "
                "about the patient's symptoms, duration, severity, and related risk factors. "
                "Provide a list of possible conditions and next diagnostic steps, but do not make a final diagnosis."
            ),
        ),
        Agent(
            agent_name="Diagnosis Synthesizer",
            agent_description="Agent specializing in synthesizing diagnostic possibilities from patient information and medical history.",
            system_prompt=(
                "You are a medical diagnosis assistant. Analyze the patient's reported symptoms, medical history, and any test results. "
                "Provide a differential diagnosis, and highlight the most likely conditions a physician should consider."
            ),
        ),
        Agent(
            agent_name="Lab Interpretation Expert",
            agent_description="Specializes in interpreting laboratory and imaging results for diagnostic support.",
            system_prompt=(
                "You are a medical lab and imaging interpretation agent. Take the patient's test results, imaging findings, and vitals, "
                "and interpret them in context of their symptoms. Suggest relevant follow-up diagnostics or considerations for the physician."
            ),
        ),
    ],
)

# Route a task to the best agent
result = agent_router.run(
    "I have a headache, fever, and cough. What could be wrong?"
)

# Use the selected agent
if result:
    print(f"Selected agent: {result.agent_name}")
    response = result.run("I have a headache, fever, and cough. What could be wrong?")
    print(response)

    # Update agent history after interaction
    agent_router.update_agent_history(result.name)

Finance Analysis Use Case

from swarms import Agent
from swarms.structs.agent_router import AgentRouter

# Define specialized finance agents
finance_agents = [
    Agent(
        agent_name="Market Analyst",
        agent_description="Analyzes market trends and provides trading insights",
        system_prompt="You are a financial market analyst specializing in market data analysis and trading insights."
    ),
    Agent(
        agent_name="Risk Assessor",
        agent_description="Evaluates financial risks and compliance requirements",
        system_prompt="You are a risk assessment specialist focusing on financial risk analysis and compliance."
    ),
    Agent(
        agent_name="Investment Strategist",
        agent_description="Provides investment strategies and portfolio management",
        system_prompt="You are an investment strategy specialist developing long-term financial planning strategies."
    )
]

# Initialize router
finance_router = AgentRouter(
    embedding_model="text-embedding-ada-002",
    agents=finance_agents
)

# Route tasks
market_task = finance_router.run("Analyze current market conditions for technology sector")
if market_task:
    market_analysis = market_task.run("Analyze current market conditions for technology sector")

risk_task = finance_router.run("Assess the risk profile for a cryptocurrency investment")
if risk_task:
    risk_assessment = risk_task.run("Assess the risk profile for a cryptocurrency investment")

Custom Embedding Model

from swarms.structs.agent_router import AgentRouter

# Use a different embedding model
router = AgentRouter(
    embedding_model="text-embedding-3-large",  # OpenAI's newer model
    n_agents=1,
    api_key="your-api-key",  # Optional: specify API key
)

# Or use Cohere embeddings
cohere_router = AgentRouter(
    embedding_model="cohere/embed-english-v3.0",
    api_key="your-cohere-api-key"
)

Dynamic Agent Addition

from swarms import Agent
from swarms.structs.agent_router import AgentRouter

# Initialize empty router
router = AgentRouter(embedding_model="text-embedding-ada-002")

# Add agents dynamically
agent1 = Agent(
    agent_name="Data Extractor",
    agent_description="Extracts structured data from documents",
    system_prompt="You are a data extraction specialist..."
)
router.add_agent(agent1)

agent2 = Agent(
    agent_name="Document Summarizer",
    agent_description="Creates concise summaries of documents",
    system_prompt="You are a document summarization expert..."
)
router.add_agent(agent2)

# Now use the router
best_agent = router.run("Extract key information from this contract")

Best Practices

Agent Descriptions

Provide clear, specific descriptions for agents to improve matching accuracy:

# Good: Specific description
Agent(
    agent_name="Medical Diagnostician",
    agent_description="Specializes in analyzing patient symptoms, medical history, and test results to provide differential diagnoses for common medical conditions.",
    system_prompt="..."
)

# Poor: Vague description
Agent(
    agent_name="Doctor",
    agent_description="Helps with medical stuff",
    system_prompt="..."
)

Embedding Model Selection

Provider	Model Names	Recommended Use
OpenAI	`text-embedding-ada-002`, `text-embedding-3-small`, `text-embedding-3-large`	Ada-002 is default, cost-effective; 3-small/large offer higher accuracy
Cohere	`cohere/embed-english-v3.0`	Excellent for English text
HuggingFace	`huggingface/microsoft/codebert-base`	Best for code-related tasks

Error Handling

Always handle cases where no agent is found:

best_agent = router.run(task)

if best_agent:
    result = best_agent.run(task)
    print(f"Task completed by {best_agent.name}: {result}")
else:
    print("No suitable agent found for this task")

Performance Considerations

Embedding Generation: The first time agents are added, embeddings are generated which can take a few seconds per agent
API Rate Limits: Be aware of rate limits when using embedding APIs, especially when adding many agents
Caching: The router doesn’t cache task embeddings — consider caching results for repeated tasks
Batch Processing: For processing multiple tasks, consider batching or using concurrent execution

Troubleshooting

No Agent Found

If find_best_agent returns None:

Check that agents have been added: len(router.agents) > 0
Verify agent descriptions are clear and specific
Ensure the task description is detailed enough to match an agent
Check logs for embedding generation errors

Low Similarity Scores

If agents are being matched but with low similarity:

Improve agent descriptions to be more specific
Enhance system prompts with more relevant keywords
Consider using a different embedding model
Update agent history after interactions to improve matching

Source Code

View the source code on GitHub

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

Overview

Installation

Class Definition

Attributes

Methods

add_agent()

add_agents()

find_best_agent()

run()

update_agent_history()

Usage Examples

Basic Medical Use Case

Finance Analysis Use Case

Custom Embedding Model

Dynamic Agent Addition

Best Practices

Agent Descriptions

Embedding Model Selection

Error Handling

Performance Considerations

Troubleshooting

No Agent Found

Low Similarity Scores

Source Code

​Overview

​Installation

​Class Definition

​Attributes

​Methods

​add_agent()

​add_agents()

​find_best_agent()

​run()

​update_agent_history()

​Usage Examples

​Basic Medical Use Case

​Finance Analysis Use Case

​Custom Embedding Model

​Dynamic Agent Addition

​Best Practices

​Agent Descriptions

​Embedding Model Selection

​Error Handling

​Performance Considerations

​Troubleshooting

​No Agent Found

​Low Similarity Scores

​Source Code

Overview

Installation

Class Definition

Attributes

Methods

add_agent()

add_agents()

find_best_agent()

run()

update_agent_history()

Usage Examples

Basic Medical Use Case

Finance Analysis Use Case

Custom Embedding Model

Dynamic Agent Addition

Best Practices

Agent Descriptions

Embedding Model Selection

Error Handling

Performance Considerations

Troubleshooting

No Agent Found

Low Similarity Scores

Source Code