Documentation Index
Fetch the complete documentation index at: https://docs.swarms.world/llms.txt
Use this file to discover all available pages before exploring further.
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
Class Definition
from swarms.structs.agent_router import AgentRouter
Attributes
embedding_model
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.
Number of agents to return in queries (currently supports returning the best match)
api_key
Optional[str]
default:"None"
API key for the embedding service. If not provided, will use environment variables.
api_base
Optional[str]
default:"None"
Custom API base URL for the embedding service.
agents
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
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
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]
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]
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
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")
- 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
