源码级别解析 · 源码解析 · 分布式协作
2026-04-30 | 每日技术深度解读
支持 Python 和 .NET 实现
支持 OpenAI, Anthropic, Azure 等模型提供商
基于插件架构，易于扩展
# 安装 AutoGen
pip install pyautogen
import autogen
from autogen import AssistantAgent, UserProxyAgent
# 创建智能体
assistant = AssistantAgent(
"assistant",
llm_config={"config_list": [{"model": "gpt-4"}]}
)
user_proxy = UserProxyAgent(
"user_proxy",
human_input_mode="NEVER",
code_execution_config={"work_dir": ".", "use_docker": False}
)
# 开始对话
chat_result = user_proxy.initiate_chat(
assistant,
message="帮我分析一下代码质量"
)这是 AutoGen 的基本使用模式，智能体之间可以进行自然对话
每种智能体都有特定的职责和功能
class AssistantAgent(ConversableAgent):
"""The default assistant agent.
Args:
name (str): agent name.
system_message (str): system message for the OpenAI API.
**kwargs: other kwargs.
"""
def __init__(self, name, system_message=None, **kwargs):
super().__init__(name, system_message=system_message, **kwargs)
def generate_reply(self, messages=None, **kwargs):
"""Generate a reply using the OpenAI API."""
response = self.client.create(messages=messages, **kwargs)
return response.choices[0].message['content']AssistantAgent 是 ConversableAgent 的子类，专注于回答问题和生成内容
工具使用 JSON Schema 进行定义和验证
def calculator(expression):
"""计算数学表达式"""
try:
result = eval(expression)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {str(e)}"
# 注册工具
user_proxy.register_for_execution()(calculator)
# 智能体可以使用工具
user_proxy.initiate_chat(
assistant,
message="计算 15 * 23 + 7 的结果"
)工具需要注册到智能体才能被调用
GroupChat 管理多个智能体的对话
from autogen import GroupChat, GroupChatManager
# 创建多个智能体
coder = AssistantAgent("coder")
reviewer = AssistantAgent("reviewer")
product_manager = AssistantAgent("product_manager")
# 创建组聊
group_chat = GroupChat(
agents=[coder, reviewer, product_manager],
messages=[],
max_round=10
)
# 创建聊天管理器
group_chat_manager = GroupChatManager(
groupchat=group_chat,
llm_config={"config_list": [{"model": "gpt-4"}]}
)
# 开始多智能体协作
user_proxy.initiate_chat(group_chat_manager, message="编写一个待办事项应用")GroupChat 可以管理多个智能体的复杂对话
配置支持嵌套结构和环境变量
# config.yaml
llm_config:
config_list:
- model: "gpt-4-turbo"
api_key: "${OPENAI_API_KEY}"
base_url: "https://api.openai.com/v1"
- model: "claude-3-sonnet"
api_key: "${ANTHROPIC_API_KEY}"
base_url: "https://api.anthropic.com"
temperature: 0.7
max_tokens: 4000
tool_config:
work_dir: "/tmp/autogen_work"
use_docker: true
timeout: 300配置文件支持环境变量和默认值
对话历史以 JSON 格式存储
支持自定义错误处理策略
支持可视化调试界面
import logging
# 配置日志
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# 启用详细日志
autogen.logging.set_verbosity(logging.DEBUG)
# 创建带调试的智能体
assistant = AssistantAgent(
"debug_assistant",
llm_config={
"config_list": [{"model": "gpt-4"}],
"temperature": 0.7
},
debug=True
)启用调试模式可以获得更详细的执行信息
基于 Python 的面向对象设计
class CustomAssistant(AssistantAgent):
"""自定义智能体示例"""
def __init__(self, name, **kwargs):
super().__init__(name, **kwargs)
self.custom_data = {}
def process_message(self, message):
"""自定义消息处理"""
# 添加自定义前缀
processed = f"[CUSTOM] {message}"
return processed
def get_custom_data(self):
"""获取自定义数据"""
return self.custom_data
# 使用自定义智能体
custom_assistant = CustomAssistant(
"custom_assistant",
llm_config={"config_list": [{"model": "gpt-4"}]}
)可以通过继承基类创建自定义智能体
适用于需要智能协作的场景
支持分布式部署
# 性能优化配置
performance_config = {
"cache": {
"enabled": True,
"max_size": 1000,
"ttl": 3600
},
"parallel": {
"max_workers": 4,
"async_mode": True
},
"load_balancer": {
"strategy": "round_robin",
"health_check": True
}
}
# 应用配置
assistant = AssistantAgent(
"optimized_assistant",
llm_config={
"config_list": [{"model": "gpt-4"}],
**performance_config
}
)性能配置可以显著提升系统效率
支持企业级安全要求
支持容器化和微服务架构
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安装依赖
COPY requirements.txt .
RUN pip install -r requirements.txt
# 复制应用代码
COPY . .
# 设置环境变量
ENV PYTHONPATH=/app
# 启动应用
CMD ["python", "app.py"]
# requirements.txt
autogen>=0.2.0
openai>=1.0.0
anthropic>=0.3.0Docker 容器化部署简化了环境配置
支持 Prometheus 和 Grafana 集成
微软官方维护，长期支持
持续迭代和改进
遵循最佳实践可以提高开发效率
适合初学者和高级开发者
AutoGen 是构建多智能体应用的理想选择
更多问题请参考官方文档
感谢阅读！
访问 https://atcfu.com/ai-articles/autogen-multi-agent/ 回顾本文