autogen-development（自体发育）

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", ...)
]