Swarm Orchestration

import asyncio
from typing import List
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    # Create specialized agents
    planner = AssistantAgent(
        "planner",
        model_client=model_client,
        description="Expert at breaking down complex problems into steps.",
        system_message="""You are a strategic planner.
        Break down complex problems into clear, actionable steps.
        Don't implement - just plan.""",
    )

    researcher = AssistantAgent(
        "researcher",
        model_client=model_client,
        description="Expert at gathering and analyzing information.",
        system_message="""You are a researcher.
        Gather relevant information and provide detailed analysis.
        Focus on facts and data.""",
    )

    implementer = AssistantAgent(
        "implementer",
        model_client=model_client,
        description="Expert at implementing solutions and writing code.",
        system_message="""You are an implementer.
        Write clean, efficient code to solve problems.
        Follow best practices.""",
    )

    reviewer = AssistantAgent(
        "reviewer",
        model_client=model_client,
        description="Expert at reviewing work and ensuring quality.",
        system_message="""You are a quality reviewer.
        Review solutions for correctness, efficiency, and best practices.
        Provide constructive feedback.""",
    )

    # Create swarm with selector
    team = SelectorGroupChat(
        participants=[planner, researcher, implementer, reviewer],
        model_client=model_client,
        selector_prompt="""Select the most appropriate agent based on the current task:
        - planner: For breaking down problems
        - researcher: For gathering information
        - implementer: For writing code
        - reviewer: For quality checks
        
        Consider the conversation history and select the agent that best fits the next step.""",
    )

    # Run collaborative task
    await Console(
        team.run_stream(
            task="""Create a Python function to analyze sentiment in customer reviews.
            The function should handle edge cases and be production-ready."""
        )
    )

    await model_client.close()


asyncio.run(main())

python swarm_pattern.py

---------- planner ----------
I'll break this down into steps:
1. Define the function interface
2. Research sentiment analysis approaches
3. Implement the core logic
4. Add error handling
5. Review for production readiness

---------- researcher ----------
For sentiment analysis, we should consider:
- Using TextBlob or VADER for simple cases
- Handling empty strings, special characters
- Supporting multiple languages if needed

---------- implementer ----------
```python
from textblob import TextBlob
from typing import Optional

def analyze_sentiment(review: str) -> dict:
    """Analyze sentiment of customer review."""
    if not review or not review.strip():
        return {"polarity": 0.0, "sentiment": "neutral", "confidence": 0.0}
    
    analysis = TextBlob(review)
    polarity = analysis.sentiment.polarity
    
    if polarity > 0.1:
        sentiment = "positive"
    elif polarity < -0.1:
        sentiment = "negative"
    else:
        sentiment = "neutral"
    
    return {
        "polarity": polarity,
        "sentiment": sentiment,
        "confidence": abs(polarity)
    }

## How It Works

1. **Agent Descriptions**: Each agent has a clear role and expertise
2. **Selector**: Uses an LLM to choose the most appropriate agent
3. **Context-Aware**: Selection considers conversation history
4. **Dynamic Flow**: No fixed order - agents are called as needed

## Advanced Swarm with Handoffs

```python
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import Swarm
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    # Customer service swarm
    triage_agent = AssistantAgent(
        "triage",
        model_client=model_client,
        description="Routes customer inquiries to the right specialist.",
        system_message="""You triage customer inquiries.
        Determine if they need: billing, technical, or general support.
        Hand off to the appropriate specialist.""",
        handoffs=["billing_specialist", "tech_specialist", "general_support"],
    )

    billing_agent = AssistantAgent(
        "billing_specialist",
        model_client=model_client,
        description="Handles billing and payment issues.",
        system_message="You handle billing inquiries. Be clear and professional.",
    )

    tech_agent = AssistantAgent(
        "tech_specialist",
        model_client=model_client,
        description="Handles technical support issues.",
        system_message="You provide technical support. Give step-by-step guidance.",
    )

    general_agent = AssistantAgent(
        "general_support",
        model_client=model_client,
        description="Handles general questions and information.",
        system_message="You handle general inquiries. Be helpful and informative.",
    )

    # Create swarm with handoff capability
    team = Swarm(
        participants=[triage_agent, billing_agent, tech_agent, general_agent],
        termination_condition=TextMentionTermination("TERMINATE"),
    )

    # Process customer inquiry
    await Console(
        team.run_stream(
            task="My account was charged twice this month. How do I get a refund?"
        )
    )

    await model_client.close()


asyncio.run(main())

agent = AssistantAgent(
    "specialist",
    description="Expert at X. Call when: [specific scenarios]. Don't call for: [other scenarios].",
    ...
)

from autogen_agentchat.conditions import MaxMessageTermination

team = SelectorGroupChat(
    participants=[...],
    termination_condition=MaxMessageTermination(max_messages=20),
)