SwarmRouter Examples¶

The SwarmRouter is a flexible routing system designed to manage different types of swarms for task execution. It provides a unified interface to interact with various swarm types, including AgentRearrange, MixtureOfAgents, SpreadSheetSwarm, SequentialWorkflow, and ConcurrentWorkflow.

Prerequisites¶

Python 3.7+
OpenAI API key or other supported LLM provider keys
Swarms library

Installation¶

pip3 install -U swarms

Environment Variables¶

WORKSPACE_DIR="agent_workspace"
OPENAI_API_KEY=""
ANTHROPIC_API_KEY=""
GROQ_API_KEY=""

Basic Usage¶

1. Initialize Specialized Agents¶

from swarms import Agent
from swarms.structs.swarm_router import SwarmRouter, SwarmType

# Initialize specialized agents
data_extractor_agent = Agent(
    agent_name="Data-Extractor",
    system_prompt="You are a data extraction specialist...",
    model_name="gpt-4o",
    max_loops=1,
)

summarizer_agent = Agent(
    agent_name="Document-Summarizer",
    system_prompt="You are a document summarization expert...",
    model_name="gpt-4o",
    max_loops=1,
)

financial_analyst_agent = Agent(
    agent_name="Financial-Analyst",
    system_prompt="You are a financial analysis specialist...",
    model_name="gpt-4o",
    max_loops=1,
)

2. Create SwarmRouter with Sequential Workflow¶

sequential_router = SwarmRouter(
    name="SequentialRouter",
    description="Process tasks in sequence",
    agents=[data_extractor_agent, summarizer_agent, financial_analyst_agent],
    swarm_type=SwarmType.SequentialWorkflow,
    max_loops=1
)

# Run a task
result = sequential_router.run("Analyze and summarize the quarterly financial report")

3. Create SwarmRouter with Concurrent Workflow¶

concurrent_router = SwarmRouter(
    name="ConcurrentRouter",
    description="Process tasks concurrently",
    agents=[data_extractor_agent, summarizer_agent, financial_analyst_agent],
    swarm_type=SwarmType.ConcurrentWorkflow,
    max_loops=1
)

# Run a task
result = concurrent_router.run("Evaluate multiple aspects of the company simultaneously")

4. Create SwarmRouter with AgentRearrange¶

rearrange_router = SwarmRouter(
    name="RearrangeRouter",
    description="Dynamically rearrange agents for optimal task processing",
    agents=[data_extractor_agent, summarizer_agent, financial_analyst_agent],
    swarm_type=SwarmType.AgentRearrange,
    flow=f"{data_extractor_agent.agent_name} -> {summarizer_agent.agent_name} -> {financial_analyst_agent.agent_name}",
    max_loops=1
)

# Run a task
result = rearrange_router.run("Process and analyze company documents")

5. Create SwarmRouter with MixtureOfAgents¶

mixture_router = SwarmRouter(
    name="MixtureRouter",
    description="Combine multiple expert agents",
    agents=[data_extractor_agent, summarizer_agent, financial_analyst_agent],
    swarm_type=SwarmType.MixtureOfAgents,
    max_loops=1
)

# Run a task
result = mixture_router.run("Provide comprehensive analysis of company performance")

Advanced Features¶

1. Error Handling and Logging¶

try:
    result = router.run("Complex analysis task")

    # Retrieve and print logs
    for log in router.get_logs():
        print(f"{log.timestamp} - {log.level}: {log.message}")
except Exception as e:
    print(f"Error occurred: {str(e)}")

2. Custom Configuration¶

router = SwarmRouter(
    name="CustomRouter",
    description="Custom router configuration",
    agents=[data_extractor_agent, summarizer_agent, financial_analyst_agent],
    swarm_type=SwarmType.SequentialWorkflow,
    max_loops=3,
    autosave=True,
    verbose=True,
    output_type="json"
)

Best Practices¶

Choose the appropriate swarm type based on your task requirements:¶

Swarm Type	Use Case
`SequentialWorkflow`	Tasks that need to be processed in order
`ConcurrentWorkflow`	Independent tasks that can be processed simultaneously
`AgentRearrange`	Tasks requiring dynamic agent organization
`MixtureOfAgents`	Complex tasks needing multiple expert perspectives

Configure agents appropriately:¶

Configuration Aspect	Description
Agent Names & Descriptions	Set meaningful and descriptive names that reflect the agent's role and purpose
System Prompts	Define clear, specific prompts that outline the agent's responsibilities and constraints
Model Parameters	Configure appropriate parameters like temperature, max_tokens, and other model-specific settings

Implement proper error handling:¶

Error Handling Practice	Description
Try-Except Blocks	Implement proper exception handling with try-except blocks
Log Monitoring	Regularly monitor and analyze system logs for potential issues
Edge Case Handling	Implement specific handling for edge cases and unexpected scenarios

Optimize performance:¶

Performance Optimization	Description
Concurrent Processing	Utilize parallel processing capabilities when tasks can be executed simultaneously
Max Loops Configuration	Set appropriate iteration limits based on task complexity and requirements
Resource Management	Continuously monitor and optimize system resource utilization

Example Implementation¶

Here's a complete example showing how to use SwarmRouter in a real-world scenario:

import os
from swarms import Agent
from swarms.structs.swarm_router import SwarmRouter, SwarmType

# Initialize specialized agents
research_agent = Agent(
    agent_name="ResearchAgent",
    system_prompt="You are a research specialist...",
    model_name="gpt-4o",
    max_loops=1
)

analysis_agent = Agent(
    agent_name="AnalysisAgent",
    system_prompt="You are an analysis expert...",
    model_name="gpt-4o",
    max_loops=1
)

summary_agent = Agent(
    agent_name="SummaryAgent",
    system_prompt="You are a summarization specialist...",
    model_name="gpt-4o",
    max_loops=1
)

# Create router with sequential workflow
router = SwarmRouter(
    name="ResearchAnalysisRouter",
    description="Process research and analysis tasks",
    agents=[research_agent, analysis_agent, summary_agent],
    swarm_type=SwarmType.SequentialWorkflow,
    max_loops=1,
    verbose=True
)

# Run complex task
try:
    result = router.run(
        "Research and analyze the impact of AI on healthcare, "
        "providing a comprehensive summary of findings."
    )
    print("Task Result:", result)

    # Print logs
    for log in router.get_logs():
        print(f"{log.timestamp} - {log.level}: {log.message}")

except Exception as e:
    print(f"Error processing task: {str(e)}")

This comprehensive guide demonstrates how to effectively use the SwarmRouter in various scenarios, making it easier to manage and orchestrate multiple agents for complex tasks.