从单智能体到多智能体AutoGen 让多个 AI 像真实团队一样协作完成任务。最终代码在最后前言前面学习了 ReAct、Plan-and-Solve、Reflection 等经典 Agent 工作流它们本质上都是单 Agent完成推理和决策。但在实际开发中一个复杂任务往往需要多个不同角色共同协作例如软件开发需要产品经理、程序员、测试工程师等共同完成。微软推出的AutoGen正是为了解决这个问题它将多个 Agent 组织成一个团队让它们通过自动对话完成复杂任务。本篇主要学习 AutoGen 0.7 版本的新架构并实现一个模拟的软件开发团队。为什么选择 AutoGenAutoGen 的核心思想可以总结成一句话Conversation Drives Collaboration用对话驱动协作它认为复杂任务 ≠ 一个 AI 完成而是多个拥有不同角色的 AI → 不断交流 → 最终完成目标。例如开发一个 Web 应用可以拆分为用户需求 │ ▼ ProductManager │ ▼ Engineer │ ▼ CodeReviewer │ ▼ UserProxy整个流程非常接近真实的软件开发团队。AutoGen 0.7 新架构相比旧版本AutoGen 0.7 最大的变化就是整体架构重构。主要有三个变化。① Core 与 AgentChat 分层新版框架拆分成两个模块autogen-core │ ▼ autogen-agentchat其中autogen-core负责模型调用、消息传递等底层能力。autogen-agentchat负责多 Agent 对话与协作。这种分层让框架更加模块化也更容易维护。② 全面采用异步Async新版全部采用async/await。原因很简单调用 LLM API 是整个 Agent 最耗时的操作。异步可以在等待模型回复时继续处理其它任务大幅提高多 Agent 协作效率。因此官方几乎所有示例都是async def run(): result await Console(team.run_stream(tasktask))相比同步方式更适合复杂智能体系统。③ Team 替代旧版 GroupChat以前使用 GroupChatManager 管理多个 Agent。新版统一采用 Team例如RoundRobinGroupChatSelectorGroupChatSwarm后续学习其中最简单也是最常见的是RoundRobinGroupChat。它会让所有 Agent 按顺序依次发言。AutoGen 核心组件1. AssistantAgentAssistantAgent 是最核心的智能体。它本质就是LLM System Prompt例如创建一个产品经理product_manager AssistantAgent( nameProductManager, model_clientmodel_client, system_message 你是一位经验丰富的产品经理 负责需求分析、项目规划。 完成分析后请说 请工程师开始实现。 )这里的system_message就决定了 Agent 的身份。工程师、代码审查员其实都是一样的创建方式只是 Prompt 不同。2. UserProxyAgentUserProxyAgent 是 AutoGen 中比较特殊的角色。它并不会像 Assistant 一样负责思考而是充当用户代理。主要职责接收用户需求执行工具或代码验证结果通知团队任务结束例如user_proxy UserProxyAgent( nameUserProxy, description 代表用户测试最终结果 完成后回复 TERMINATE。 )这里最终会发送TERMINATE结束整个团队协作。3. RoundRobinGroupChat多个 Agent 如何协作AutoGen 提供了RoundRobinGroupChat。team_chat RoundRobinGroupChat( participants[ product_manager, engineer, code_reviewer, user_proxy ], termination_conditionTextMentionTermination(TERMINATE), max_turns20, )这里几个参数十分重要participants参与协作的所有 Agent。termination_condition终止条件。max_turns最大对话轮数防止无限循环。整个流程会按照 participants 的顺序不断轮询。实战构建 AI 软件开发团队为了体验多 Agent 协作我实现了一个简单的软件开发团队。项目目标开发一个 Streamlit Web 应用实现实时显示 BTC 当前价格显示 24 小时涨跌幅支持刷新价格添加异常处理整个团队分工如下Agent职责ProductManager需求分析Engineer编写代码CodeReviewer审查代码UserProxy测试并结束任务启动团队完成 Agent 创建后只需要启动 Team 即可。result await Console( team_chat.run_stream(tasktask) )这里run_stream()会不断输出多个 Agent 的聊天过程我们可以实时观察整个协作流程。最终生成的应用最终生成的是一个 Streamlit 应用。通过 CoinGecko API 获取 BTC 最新价格response requests.get( https://api.coingecko.com/api/v3/simple/price )随后利用 Streamlit 展示数据st.metric( label当前比特币价格 (USD), valuef${current_price} )整个页面还增加了刷新按钮Loading 状态API 异常处理虽然功能简单但已经完整体验了一次多 Agent 自动协作开发流程。AutoGen 的优缺点优势✅ 多 Agent 分工明确更符合真实开发流程。✅ 通过 System Prompt 即可快速定义不同角色。✅ Team 能自动管理整个协作过程。✅ UserProxy 支持 Human in the Loop方便人工介入。不足❌ 对话存在随机性多个 Agent 有时会偏离预期。❌ Agent 越多调试越困难需要查看完整聊天记录。❌ Token 消耗较高运行成本明显高于单 Agent。国产模型适配官方示例默认使用 OpenAI。如果使用通义千问、DeepSeek等兼容 OpenAI API 的模型还需要配置模型能力信息例如model_client OpenAIChatCompletionClient( modelqwen3.7-plus, api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), model_infoModelInfo( visionFalse, function_callingTrue, json_outputTrue, familyqwen, context_window128000 ) )我这里使用的就是Qwen3.7-Plus按照这种方式配置后即可正常运行 AutoGen。学习总结AutoGen 给我最大的感受是它关注的不再是如何实现一个 Agent而是如何组织多个 Agent 协同工作。相比单智能体它更像一个真正的软件开发团队每个 Agent 都有自己的职责通过持续对话共同推进任务最终完成目标。后续我准备继续学习 AutoGen 的更多能力例如工具调用Tool Calling、SelectorGroupChat、Swarm 以及 MCP 集成进一步探索复杂多智能体系统的实现方式。最终代码autogen-software-team.pyimport os import asyncio from typing import List,Dict,Any from autogen_core.models import ModelInfo from dotenv import load_dotenv # 先测试一个版本使用 OpenAI 客户端 from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.agents import AssistantAgent, UserProxyAgent from autogen_agentchat.teams import RoundRobinGroupChat from autogen_agentchat.conditions import TextMentionTermination from autogen_agentchat.ui import Console # 加载环境变量 load_dotenv() def create_openai_model_client(): 创建并配置 OpenAI 模型客户端 return OpenAIChatCompletionClient( modelos.getenv(LLM_MODEL_ID,qwen3.7-plus), api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL,https://dashscope.aliyuncs.com/compatible-mode/v1), # 新增通义qwen3.7-plus模型规格信息 model_info ModelInfo( visionFalse, # 该模型不支持图像输入 function_callingTrue, # 支持工具调用Agent必备 json_outputTrue, familyqwen, context_window128000 # qwen3.7-plus上下文128k ) ) def create_product_manager(model_client): 创建产品经理智能体 system_message 你是一位经验丰富的产品经理专门负责软件产品的需求分析和项目规划。 你的核心职责包括 1. **需求分析**深入理解用户需求识别核心功能和边界条件 2. **技术规划**基于需求制定清晰的技术实现路径 3. **风险评估**识别潜在的技术风险和用户体验问题 4. **协调沟通**与工程师和其他团队成员进行有效沟通 当接到开发任务时请按以下结构进行分析 1. 需求理解与分析 2. 功能模块划分 3. 技术选型建议 4. 实现优先级排序 5. 验收标准定义 请简洁明了地回应并在分析完成后说请工程师开始实现。 return AssistantAgent( nameProductManager, model_clientmodel_client, system_messagesystem_message, ) def create_engineer(model_client): 创建软件工程师智能体 system_message 你是一位资深的软件工程师擅长 Python 开发和 Web 应用构建。 你的技术专长包括 1. **Python 编程**熟练掌握 Python 语法和最佳实践 2. **Web 开发**精通 Streamlit、Flask、Django 等框架 3. **API 集成**有丰富的第三方 API 集成经验 4. **错误处理**注重代码的健壮性和异常处理 当收到开发任务时请 1. 仔细分析技术需求 2. 选择合适的技术方案 3. 编写完整的代码实现 4. 添加必要的注释和说明 5. 考虑边界情况和异常处理 请提供完整的可运行代码并在完成后说请代码审查员检查。 return AssistantAgent( nameEngineer, model_clientmodel_client, system_messagesystem_message, ) def create_code_reviewer(model_client): 创建代码审查员智能体 system_message 你是一位经验丰富的代码审查专家专注于代码质量和最佳实践。 你的审查重点包括 1. **代码质量**检查代码的可读性、可维护性和性能 2. **安全性**识别潜在的安全漏洞和风险点 3. **最佳实践**确保代码遵循行业标准和最佳实践 4. **错误处理**验证异常处理的完整性和合理性 审查流程 1. 仔细阅读和理解代码逻辑 2. 检查代码规范和最佳实践 3. 识别潜在问题和改进点 4. 提供具体的修改建议 5. 评估代码的整体质量 请提供具体的审查意见完成后说代码审查完成请用户代理测试。 return AssistantAgent( nameCodeReviewer, model_clientmodel_client, system_messagesystem_message, ) def create_user_proxy(): 创建用户代理智能体 return UserProxyAgent( nameUserProxy, description用户代理负责以下职责 1. 代表用户提出开发需求 2. 执行最终的代码实现 3. 验证功能是否符合预期 4. 提供用户反馈和建议 完成测试后请回复 TERMINATE。, ) async def run_software_development_team(): 运行软件开发团队协作 print( 正在初始化模型客户端...) # 先使用标准的 OpenAI 客户端测试 model_client create_openai_model_client() print( 正在创建智能体团队...) # 创建智能体团队 product_manager create_product_manager(model_client) engineer create_engineer(model_client) code_reviewer create_code_reviewer(model_client) user_proxy create_user_proxy() # 添加终止条件 termination TextMentionTermination(TERMINATE) # 创建团队聊天 team_chat RoundRobinGroupChat( participants[ product_manager, engineer, code_reviewer, user_proxy ], termination_conditiontermination, max_turns20, # 增加最大轮次 ) # 定义开发任务 task 我们需要开发一个比特币价格显示应用具体要求如下 核心功能 - 实时显示比特币当前价格USD - 显示24小时价格变化趋势涨跌幅和涨跌额 - 提供价格刷新功能 技术要求 - 使用 Streamlit 框架创建 Web 应用 - 界面简洁美观用户友好 - 添加适当的错误处理和加载状态 请团队协作完成这个任务从需求分析到最终实现。 # 执行团队协作 print( 启动 AutoGen 软件开发团队协作...) print( * 60) # 使用 Console 来显示对话过程 result await Console(team_chat.run_stream(tasktask)) print(\n * 60) print(✅ 团队协作完成) return result # 主程序入口 if __name__ __main__: try: # 运行异步协作流程 result asyncio.run(run_software_development_team()) print(f\n 协作结果摘要) print(f- 参与智能体数量4个) print(f- 任务完成状态{成功 if result else 需要进一步处理}) except ValueError as e: print(f❌ 配置错误{e}) print(请检查 .env 文件中的配置是否正确) except Exception as e: print(f❌ 运行错误{e}) import traceback traceback.print_exc()output.pyimport streamlit as st import requests import json # 数据获取函数 def get_bitcoin_price(): try: # 获取 Bitcoin 的价格数据 response requests.get(https://api.coingecko.com/api/v3/simple/price?idsbitcoinvs_currenciesusdinclude_24hr_changetrue) data response.json() # 获取当前价格和24小时变化 current_price data[bitcoin][usd] price_change_percentage data[bitcoin][usd_24h_change] return current_price, price_change_percentage except requests.exceptions.RequestException as e: st.error(fError fetching data: {e}) return None, None # 初始化 Streamlit 应用 st.title(实时比特币价格) st.subheader(获取最新的比特币价格信息及其24小时价格变化趋势) # 添加刷新按钮 if st.button(刷新价格): st.experimental_rerun() # 显示加载状态 with st.spinner(加载中...): current_price, price_change_percentage get_bitcoin_price() # 显示数据 if current_price is not None: st.metric(label当前比特币价格 (USD), valuef${current_price}) if price_change_percentage is not None: st.metric(label24小时变化 (%), valuef{price_change_percentage:.2f}%) else: st.error(无法获取数据请稍后重试。)requirements.txt# AutoGen 软件开发团队案例依赖 (v0.7.4) autogen-agentchat autogen-ext[openai,azure] openai1.0.0 streamlit1.28.0 requests2.31.0 pandas2.0.0 plotly5.15.0 asyncio dotenv.env# .env file LLM_API_KEYYOUR_API_KEY LLM_MODEL_IDqwen3.7-plus LLM_BASE_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1