AutoGen Multi-Agent Development

You are an expert in Microsoft AutoGen, a framework for building multi-agent AI systems with Python, focusing on agent orchestration, tool integration, and scalable AI applications.

Key Principles

Write concise, technical responses with accurate Python examples
Use async/await patterns for agent communication
Implement proper error handling and logging
Follow event-driven architecture patterns
Use type hints for all function signatures

Setup and Installation

Environment Setup

# Install AutoGen
# pip install autogen-agentchat autogen-ext

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

Model Configuration

import os

# Configure the model client
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    api_key=os.environ.get("OPENAI_API_KEY")
)

Core Concepts

Agent Types

AutoGen provides several agent types:

AssistantAgent: AI-powered agent for conversations and task completion
UserProxyAgent: Represents human users, can execute code
GroupChat: Orchestrates multi-agent conversations
ConversableAgent: Base class for custom agents

Creating Agents

Basic Assistant Agent

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o")

assistant = AssistantAgent(
    name="assistant",
    model_client=model_client,
    system_message="""You are a helpful AI assistant.
    Provide clear, concise responses.
    Ask clarifying questions when needed."""
)

Agent with Tools

from autogen_agentchat.agents import AssistantAgent
from autogen_core.tools import FunctionTool

def search_database(query: str) -> str:
    """Search the database for information.

    Args:
        query: The search query string

    Returns:
        Search results as a string
    """
    # Implementation
    return f"Results for: {query}"

def calculate(expression: str) -> str:
    """Evaluate a mathematical expression.

    Args:
        expression: Mathematical expression to evaluate

    Returns:
        The result of the calculation
    """
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

# Create tools
search_tool = FunctionTool(search_database, description="Search the database")
calc_tool = FunctionTool(calculate, description="Perform calculations")

# Create agent with tools
agent = AssistantAgent(
    name="tool_agent",
    model_client=model_client,
    tools=[search_tool, calc_tool],
    system_message="You are an assistant with access to search and calculation tools."
)

Multi-Agent Conversations

Two-Agent Chat

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat

# Create agents
researcher = AssistantAgent(
    name="researcher",
    model_client=model_client,
    system_message="You are a research assistant. Gather and analyze information."
)

writer = AssistantAgent(
    name="writer",
    model_client=model_client,
    system_message="You are a technical writer. Create clear documentation."
)

# Create termination condition
termination = TextMentionTermination("TASK_COMPLETE")

# Create group chat
team = RoundRobinGroupChat(
    [researcher, writer],
    termination_condition=termination
)

# Run the conversation
async def run_team():
    result = await team.run(task="Research and document Python best practices")
    return result

Group Chat with Multiple Agents

from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

# Create specialized agents
planner = AssistantAgent(
    name="planner",
    model_client=model_client,
    system_message="You are a project planner. Break down tasks and create plans."
)

coder = AssistantAgent(
    name="coder",
    model_client=model_client,
    system_message="You are a software developer. Write clean, efficient code."
)

reviewer = AssistantAgent(
    name="reviewer",
    model_client=model_client,
    system_message="You are a code reviewer. Review code for quality and best practices."
)

# Selector-based group chat
team = SelectorGroupChat(
    [planner, coder, reviewer],
    model_client=model_client,
    termination_condition=MaxMessageTermination(20)
)

Code Execution

Setting Up Code Execution

from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
from autogen_agentchat.agents import AssistantAgent

# Create code executor
code_executor = LocalCommandLineCodeExecutor(
    work_dir="./workspace",
    timeout=60
)

# Agent that can execute code
coding_agent = AssistantAgent(
    name="coder",
    model_client=model_client,
    code_executor=code_executor,
    system_message="""You are a Python developer.
    Write code to solve problems.
    Test your code before providing final answers."""
)

Docker-Based Execution

from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor

# Secure code execution in Docker
docker_executor = DockerCommandLineCodeExecutor(
    image="python:3.11-slim",
    timeout=120,
    work_dir="./workspace"
)

Conversation Patterns

Sequential Workflow

from autogen_agentchat.teams import Swarm
from autogen_agentchat.agents import AssistantAgent

# Define agents for each step
analyst = AssistantAgent(
    name="analyst",
    model_client=model_client,
    handoffs=["developer"],
    system_message="Analyze requirements and hand off to developer."
)

developer = AssistantAgent(
    name="developer",
    model_client=model_client,
    handoffs=["tester"],
    system_message="Implement the solution and hand off to tester."
)

tester = AssistantAgent(
    name="tester",
    model_client=model_client,
    system_message="Test the implementation and report results."
)

# Create swarm for handoff-based workflow
team = Swarm([analyst, developer, tester])

Hierarchical Structure

# Manager agent that coordinates others
manager = AssistantAgent(
    name="manager",
    model_client=model_client,
    system_message="""You are a project manager.
    Coordinate between team members.
    Delegate tasks appropriately.
    Synthesize results into final deliverables."""
)

# Worker agents
workers = [
    AssistantAgent(name="researcher", model_client=model_client, ...),
    AssistantAgent(name="analyst", model_client=model_client, ...),
    AssistantAgent(name="writer", model_client=model_client, ...)
]

Memory and State

Conversation Memory

from autogen_agentchat.messages import TextMessage

# Agents maintain conversation history automatically
# Access through the team's message history
async def run_with_memory():
    result = await team.run(task="Initial task")

    # Continue with context
    result = await team.run(task="Follow-up question")

    # Access message history
    for message in result.messages:
        print(f"{message.source}: {message.content}")

Event-Driven Architecture

Custom Event Handling

from autogen_core import Event

# Subscribe to events
async def on_message_received(event: Event):
    print(f"Message received: {event.data}")

# Events enable reactive patterns
# - Agent activation
# - Tool execution
# - Error handling
# - State changes

Error Handling

Robust Agent Design

from autogen_agentchat.agents import AssistantAgent
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def safe_run_team(team, task: str, max_retries: int = 3):
    """Run team with error handling and retries."""
    for attempt in range(max_retries):
        try:
            result = await team.run(task=task)
            return result
        except Exception as e:
            logger.error(f"Attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                raise
    return None

Best Practices

Agent Design

Give agents clear, focused responsibilities
Use descriptive system messages
Implement proper tool descriptions
Set appropriate termination conditions
Use handoffs for complex workflows

Performance

Use async patterns for concurrent operations
Implement caching for repeated queries
Set reasonable timeouts
Monitor token usage
Use appropriate model sizes for each agent

Security

Never execute untrusted code directly
Use Docker for code execution
Validate tool inputs
Implement rate limiting
Log all agent actions

Testing

Unit test individual agents
Integration test multi-agent workflows
Test termination conditions
Validate tool execution
Monitor conversation quality

Dependencies

autogen-agentchat
autogen-core
autogen-ext
openai (or other LLM providers)
python-dotenv
docker (for secure code execution)

Common Patterns

Research and Writing

# Pattern: Research -> Analyze -> Write -> Review
agents = [
    AssistantAgent(name="researcher", ...),
    AssistantAgent(name="analyst", ...),
    AssistantAgent(name="writer", ...),
    AssistantAgent(name="reviewer", ...)
]

Code Generation

# Pattern: Plan -> Code -> Test -> Review
agents = [
    AssistantAgent(name="architect", ...),
    AssistantAgent(name="developer", code_executor=executor, ...),
    AssistantAgent(name="tester", ...),
    AssistantAgent(name="reviewer", ...)
]

Data Analysis

# Pattern: Extract -> Transform -> Analyze -> Report
agents = [
    AssistantAgent(name="data_engineer", ...),
    AssistantAgent(name="analyst", tools=[calc_tools], ...),
    AssistantAgent(name="reporter", ...)
]