autogen-development

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

AutoGen Multi-Agent Development

AutoGen多智能体开发

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.
您是Microsoft AutoGen方面的专家,这是一个使用Python构建多智能体AI系统的框架,专注于智能体编排、工具集成和可扩展AI应用。

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
  • 撰写简洁、技术准确的响应,并附带正确的Python示例
  • 使用async/await模式实现智能体通信
  • 实现适当的错误处理和日志记录
  • 遵循事件驱动架构模式
  • 为所有函数签名使用类型提示

Setup and Installation

安装与设置

Environment Setup

环境设置

python
undefined
python
undefined

Install AutoGen

Install AutoGen

pip install autogen-agentchat autogen-ext

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
undefined
from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.teams import RoundRobinGroupChat from autogen_ext.models.openai import OpenAIChatCompletionClient
undefined

Model Configuration

模型配置

python
import os
python
import os

Configure the model client

Configure the model client

model_client = OpenAIChatCompletionClient( model="gpt-4o", api_key=os.environ.get("OPENAI_API_KEY") )
undefined
model_client = OpenAIChatCompletionClient( model="gpt-4o", api_key=os.environ.get("OPENAI_API_KEY") )
undefined

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
AutoGen提供多种智能体类型:
  • AssistantAgent:用于对话和任务完成的AI驱动智能体
  • UserProxyAgent:代表人类用户,可执行代码
  • GroupChat:编排多智能体对话
  • ConversableAgent:自定义智能体的基类

Creating Agents

创建智能体

Basic Assistant Agent

基础Assistant Agent

python
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."""
)
python
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

集成工具的智能体

python
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)}"
python
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

Create tools

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

Create agent with tools

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." )
undefined
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." )
undefined

Multi-Agent Conversations

多智能体对话

Two-Agent Chat

双智能体对话

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

Create agents

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

Create termination condition

termination = TextMentionTermination("TASK_COMPLETE")
termination = TextMentionTermination("TASK_COMPLETE")

Create group chat

Create group chat

team = RoundRobinGroupChat( [researcher, writer], termination_condition=termination )
team = RoundRobinGroupChat( [researcher, writer], termination_condition=termination )

Run the conversation

Run the conversation

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

Group Chat with Multiple Agents

多智能体群组对话

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

Create specialized agents

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

Selector-based group chat

team = SelectorGroupChat( [planner, coder, reviewer], model_client=model_client, termination_condition=MaxMessageTermination(20) )
undefined
team = SelectorGroupChat( [planner, coder, reviewer], model_client=model_client, termination_condition=MaxMessageTermination(20) )
undefined

Code Execution

代码执行

Setting Up Code Execution

代码执行设置

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

Create code executor

Create code executor

code_executor = LocalCommandLineCodeExecutor( work_dir="./workspace", timeout=60 )
code_executor = LocalCommandLineCodeExecutor( work_dir="./workspace", timeout=60 )

Agent that can execute code

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.""" )
undefined
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.""" )
undefined

Docker-Based Execution

基于Docker的执行

python
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
python
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor

Secure code execution in Docker

Secure code execution in Docker

docker_executor = DockerCommandLineCodeExecutor( image="python:3.11-slim", timeout=120, work_dir="./workspace" )
undefined
docker_executor = DockerCommandLineCodeExecutor( image="python:3.11-slim", timeout=120, work_dir="./workspace" )
undefined

Conversation Patterns

对话模式

Sequential Workflow

顺序工作流

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

Define agents for each step

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

Create swarm for handoff-based workflow

team = Swarm([analyst, developer, tester])
undefined
team = Swarm([analyst, developer, tester])
undefined

Hierarchical Structure

分层结构

python
undefined
python
undefined

Manager agent that coordinates others

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

Worker agents

workers = [ AssistantAgent(name="researcher", model_client=model_client, ...), AssistantAgent(name="analyst", model_client=model_client, ...), AssistantAgent(name="writer", model_client=model_client, ...) ]
undefined
workers = [ AssistantAgent(name="researcher", model_client=model_client, ...), AssistantAgent(name="analyst", model_client=model_client, ...), AssistantAgent(name="writer", model_client=model_client, ...) ]
undefined

Memory and State

内存与状态

Conversation Memory

对话内存

python
from autogen_agentchat.messages import TextMessage
python
from autogen_agentchat.messages import TextMessage

Agents maintain conversation history automatically

Agents maintain conversation history automatically

Access through the team's message history

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}")
undefined
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}")
undefined

Event-Driven Architecture

事件驱动架构

Custom Event Handling

自定义事件处理

python
from autogen_core import Event
python
from autogen_core import Event

Subscribe to events

Subscribe to events

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

Events enable reactive patterns

Events enable reactive patterns

- Agent activation

- Agent activation

- Tool execution

- Tool execution

- Error handling

- Error handling

- State changes

- State changes

undefined
undefined

Error Handling

错误处理

Robust Agent Design

健壮的智能体设计

python
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
python
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
  • 绝不直接执行不可信代码
  • 使用Docker执行代码
  • 验证工具输入
  • 实现速率限制
  • 记录所有智能体操作

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)
  • autogen-agentchat
  • autogen-core
  • autogen-ext
  • openai(或其他大语言模型提供商)
  • python-dotenv
  • docker(用于安全代码执行)

Common Patterns

常见模式

Research and Writing

研究与写作

python
undefined
python
undefined

Pattern: Research -> Analyze -> Write -> Review

Pattern: Research -> Analyze -> Write -> Review

agents = [ AssistantAgent(name="researcher", ...), AssistantAgent(name="analyst", ...), AssistantAgent(name="writer", ...), AssistantAgent(name="reviewer", ...) ]
undefined
agents = [ AssistantAgent(name="researcher", ...), AssistantAgent(name="analyst", ...), AssistantAgent(name="writer", ...), AssistantAgent(name="reviewer", ...) ]
undefined

Code Generation

代码生成

python
undefined
python
undefined

Pattern: Plan -> Code -> Test -> Review

Pattern: Plan -> Code -> Test -> Review

agents = [ AssistantAgent(name="architect", ...), AssistantAgent(name="developer", code_executor=executor, ...), AssistantAgent(name="tester", ...), AssistantAgent(name="reviewer", ...) ]
undefined
agents = [ AssistantAgent(name="architect", ...), AssistantAgent(name="developer", code_executor=executor, ...), AssistantAgent(name="tester", ...), AssistantAgent(name="reviewer", ...) ]
undefined

Data Analysis

数据分析

python
undefined
python
undefined

Pattern: Extract -> Transform -> Analyze -> Report

Pattern: Extract -> Transform -> Analyze -> Report

agents = [ AssistantAgent(name="data_engineer", ...), AssistantAgent(name="analyst", tools=[calc_tools], ...), AssistantAgent(name="reporter", ...) ]
undefined
agents = [ AssistantAgent(name="data_engineer", ...), AssistantAgent(name="analyst", tools=[calc_tools], ...), AssistantAgent(name="reporter", ...) ]
undefined