Multi-Agent Patterns

User Query → Orchestrator Agent → [Research Agent | Product Agent | Trip Agent]
                ↓
          Final Response

from strands import Agent
from strands.models.litellm import LiteLLMModel

ORCHESTRATOR_SYSTEM_PROMPT = """
You are a master assistant that routes complex queries to specialized agents.
Based on the user's request, determine the best tool to use.

- For research questions → Use the `research_assistant`
- For product recommendations → Use the `product_recommendation_assistant`
- For travel planning → Use the `trip_planning_assistant`
- For simple queries → Answer directly without using a tool
"""

def create_orchestrator_agent() -> Agent:
    model = LiteLLMModel(
        client_args={"api_key": os.getenv("NEBIUS_API_KEY")},
        model_id="nebius/moonshotai/Kimi-K2-Instruct",
    )
    
    orchestrator = Agent(
        model=model,
        system_prompt=ORCHESTRATOR_SYSTEM_PROMPT,
        tools=[
            research_assistant,
            product_recommendation_assistant,
            trip_planning_assistant,
        ],
    )
    return orchestrator

# Specialized agents for different SEO tasks
search_insights_agent = Agent(
    name="Search Insights Agent",
    model=Nebius(id="moonshotai/Kimi-K2-Instruct"),
    tools=[search_google_ai_mode, search_google_ai_overview],
    description="Expert SEO researcher for SERP data gathering"
)

serp_analysis_agent = Agent(
    name="SERP Analysis Agent",
    description="Expert at understanding and structuring SERP data",
    output_schema=SearchInsights,
)

content_strategist_agent = Agent(
    name="Content Strategist Agent",
    description="Senior content strategist for SEO optimization"
)

seo_editor_agent = Agent(
    name="SEO Editor Agent",
    description="Expert editor for keyword integration",
    output_schema=SectionEdits,
)

Researcher ⇄ Coder ⇄ Reviewer ⇄ Architect
    ↓         ↓        ↓         ↓
         Autonomous Handoffs

from strands.multiagent import Swarm

# Create specialized agents
researcher = Agent(
    name="researcher",
    model=model,
    system_prompt="You are a research specialist..."
)

coder = Agent(
    name="coder",
    model=model,
    system_prompt="You are a coding specialist..."
)

reviewer = Agent(
    name="reviewer",
    model=model,
    system_prompt="You are a code review specialist..."
)

architect = Agent(
    name="architect",
    model=model,
    system_prompt="You are a system architecture specialist..."
)

# Create a swarm with autonomous handoffs
swarm = Swarm(
    [coder, researcher, reviewer, architect],
    entry_point=researcher,  # Start with researcher
    max_handoffs=20,
    max_iterations=20,
    execution_timeout=900.0,  # 15 minutes
    node_timeout=300.0,  # 5 minutes per agent
    repetitive_handoff_detection_window=8,
    repetitive_handoff_min_unique_agents=3,
)

# Execute with autonomous agent selection
result = swarm("Design and implement a simple REST API for a todo app")

         Researcher
           /    \
          ↓      ↓
    Analyst  Fact Checker
          \      /
           ↓    ↓
       Report Writer

from strands.multiagent import GraphBuilder

# Create specialized agents
researcher = Agent(
    name="researcher",
    system_prompt="Research specialist for gathering data"
)

analyst = Agent(
    name="analyst",
    system_prompt="Data analysis specialist"
)

fact_checker = Agent(
    name="fact_checker",
    system_prompt="Fact checking specialist"
)

report_writer = Agent(
    name="report_writer",
    system_prompt="Report writing specialist"
)

# Build the graph with explicit dependencies
builder = GraphBuilder()

# Add nodes
builder.add_node(researcher, "research")
builder.add_node(analyst, "analysis")
builder.add_node(fact_checker, "fact_check")
builder.add_node(report_writer, "report")

# Define dependencies (edges)
builder.add_edge("research", "analysis")
builder.add_edge("research", "fact_check")
builder.add_edge("analysis", "report")
builder.add_edge("fact_check", "report")

# Set entry point
builder.set_entry_point("research")
builder.set_execution_timeout(600)  # 10 minute timeout

# Build and execute
graph = builder.build()
result = graph("Research AI impact on healthcare and create a report")

Searcher → Analyst → Writer → Final Report

from agno.workflow import Workflow
from agno.agent import Agent

class DeepResearcherAgent(Workflow):
    """
    Multi-stage research workflow:
    1. Gathers information from the web
    2. Analyzes and synthesizes findings
    3. Produces a clear, structured report
    """
    
    # Stage 1: Research and gather information
    searcher: Agent = Agent(
        tools=[ScrapeGraphTools(api_key=os.getenv("SGAI_API_KEY"))],
        model=Nebius(id="deepseek-ai/DeepSeek-V3-0324"),
        description="Expert at finding and extracting web information",
        instructions="""
        1. Search for authoritative sources on the topic
        2. Extract key facts, statistics, expert opinions
        3. Cover multiple perspectives
        4. Mention References & Sources
        """
    )
    
    # Stage 2: Analyze research findings
    analyst: Agent = Agent(
        model=Nebius(id="deepseek-ai/DeepSeek-V3-0324"),
        description="Critical thinker who synthesizes research",
        instructions="""
        1. Identify key themes, trends, contradictions
        2. Highlight important findings and implications
        3. Suggest areas for further investigation
        4. Extract ONLY actual reference links provided
        5. Don't hallucinate or make up information
        """
    )
    
    # Stage 3: Write final report
    writer: Agent = Agent(
        model=Nebius(id="deepseek-ai/DeepSeek-V3-0324"),
        description="Professional technical writer",
        instructions="""
        1. Write engaging introduction with context
        2. Organize findings into logical sections
        3. Use bullet points, tables for clarity
        4. Include actionable recommendations
        5. Add References section if analyst provided links
        6. Use ONLY actual links from the analyst
        """
    )
    
    def run(self, topic: str) -> Iterator[RunResponse]:
        """Execute the research pipeline"""
        # Step 1: Research
        research_content = self.searcher.run(topic)
        
        # Step 2: Analysis
        analysis = self.analyst.run(research_content.content)
        
        # Step 3: Report Writing (streaming)
        report = self.writer.run(analysis.content, stream=True)
        yield from report

agent = DeepResearcherAgent()
response = agent.run(topic="AI agent frameworks comparison")

for chunk in response:
    if chunk.content:
        print(chunk.content, end="")

# ✅ Good: Sequential workflow for research pipeline
class ResearchWorkflow(Workflow):
    searcher: Agent
    analyst: Agent
    writer: Agent

# ❌ Bad: Swarm for simple sequential tasks (unnecessary complexity)

# ✅ Good: Clear limits and timeouts
swarm = Swarm(
    agents=[...],
    max_handoffs=20,
    execution_timeout=900.0,
    repetitive_handoff_detection_window=8
)

# ❌ Bad: No limits (risk of infinite loops)
swarm = Swarm(agents=[...])  # Missing safety limits

# ✅ Good: Typed outputs for agent communication
class SearchInsights(BaseModel):
    primary_keywords: str
    related_keywords: str
    search_intent: str

analyst = Agent(
    output_schema=SearchInsights,  # Structured output
    ...
)

# ❌ Bad: Unstructured string outputs between agents
analyst = Agent(...)  # Hard to parse output

# ✅ Good: Focused, single-responsibility agents
researcher = Agent(
    description="Expert at finding information from the web",
    instructions="Search for sources and extract facts"
)

# ❌ Bad: Vague, multi-purpose agent
agent = Agent(description="Do everything")  # Too broad