ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

AutoGen多智能体编排实战:从对话设计到生产落地

AutoGen多智能体编排实战:从对话设计到生产落地 做LLM应用开发绕不过去的一个问题当你手上有需要多角色协作才能完成的任务时怎么编排对话流程。我去年开始系统使用AutoGen框架解决这类问题一路从pyautogen 0.2追到现在的0.4、0.5版本测试环境和生产环境都跑过踩了不少坑。这篇内容对应我学习笔记里的第6.2节聊聊我在实际项目中会用得到的AutoGen能力、选型判断、容易出错的地方以及怎么让它稳定跑在业务里。先说结论AutoGen不是把多个LLM调包拼在一起它提供的是“对话即编排”的范式。你把不同角色定义成Agent把任务丢进对话流让Agent之间通过消息互相驱动而不是你写几百行if/else控制调用顺序。这个思路看懂了很多所谓“多智能体难落地”的问题其实都能拆开。这篇文章适合三类人一是刚接触Agent框架、正在对比选型的开发者二是已经在用别的框架、想看看AutoGen差异的工程师三是在生产环境里跑过AutoGen但遇到坑、想找排查思路的朋友。我会把从零搭建到实际部署的完整路径写出来包括不是官方文档里能找到的那些细节。1. 为什么Agent框架值得单独学编排问题是多智能体开发的真正门槛很多人一开始做LLM应用都是从单次调用开始。一个Prompt进去一个答案出来这不叫Agent。当你需要让大模型完成“理解需求—拆解步骤—调用工具—自我检查—修正输出”这样一串动作时问题就从“单个Prompt写得好不好”变成了“状态怎么流转”。1.1 手写对话编排为什么会在第9轮崩掉我自己最早做多Agent协作所有逻辑都手写。核心是一个while循环里面放一大串if/else判断当前该谁说话、要不要调工具、要不要退出。代码写出来倒也能跑但有两个硬伤。第一个硬伤是消息历史的管理。每轮对话结束后我得把这一轮的消息追加到一个list里然后下次请求全部传进去。一旦超过一定轮次Token占用暴涨。再加上不同角色带不同System Prompt整个上下文拼起来很容易超出模型窗口。当时我的解决办法是粗暴截断截完发现Agent“失忆”了前面说的结论后面不认。第二个硬伤是发言权控制。现实场景里谁先说话、谁在什么时候打断、什么时候终止还是有规则的。我在代码里用一堆状态标记控制发言顺序调试的时候要把每个状态打印出来看。有一次线上出问题我看日志发现Agent居然在互相抬杠一直循环了十几轮才被我的超时机制掐断。那一刻我就知道手写编排这条路不能走远。AutoGen解决这两个问题的思路是把“对话流”本身当成控制流。你不必手动维护消息队列也不必写复杂的轮转逻辑。你定义Agent的system_message、llm_config、工具列表然后调用initiate_chat框架内部会处理消息的往返和终止判断。1.2 AutoGen和LangChain、LangGraph的核心差异市面上Agent编排的工具不少我选型时对比过这几个这里直接给结论。框架编排核心抽象适合场景不适合场景LangChainChain / Tool单轮或简单序列调用工具多角色多轮复杂交互LangGraphGraph / State明确状态机流转、严格可控对话式协作需要灵活发言AutoGenAgent / Chat多角色对话、自协作、任务拆解需要强流程控制、确定性高的场景这个表是我实际用下来总结的。LangGraph适合那种状态明确、流转固定的任务流比如审核流程、订单状态机每一步做什么很清楚。AutoGen则适合任务本身需要“商量着来”的场景比如让一个研究员Agent和代码执行Agent配合写分析报告过程中需要来回质疑和补充。不过要说明白AutoGen不是不能用确定性流程。它有GroupChat的speaker_selection_method可配成auto、manual或自定义函数。你可以在需要控制的地方手动指定下一个说话者在需要自由讨论的地方放开让模型自己选。这个柔性是它比LangGraph更灵活的地方但也意味着你对流程的控制需要主动配置否则Agent自由发挥起来也挺让人头疼。2. 搭建AutoGen环境与首个双Agent对话AutoGen的安装非常直接pip install autogen-agentchat就够用了。但我得专门说下版本问题因为网上能找到的教程有大量是0.2时代的写法照着写在新版本里是跑不起来的。2.1 版本陷阱0.2、0.4、0.5到底该用哪个AutoGen演进到0.4之后核心包的命名和API做了很大调整。旧版本的包名叫pyautogen导入方式通常是from autogen import ConversableAgent。新版本拆成了autogen-agentchat、autogen-core等多个命名空间导入方式也变了。我的建议是直接用0.4及以上版本不要用旧教程的代码硬套。原因有两个一是新版本对异步支持更完善生产环境跑并发不容易被卡住二是新版本的功能边界更清晰autogen-agentchat专注对话编排autogen-ext放各种扩展工具。0.5版本在0.4基础上增加了更多事件处理能力。安装的时候有两点容易出问题。第一autogen-agentchat会自动拉取核心依赖但如果你机器上已经有其他版本的pydantic可能会冲突。建议在虚拟环境里单独装。第二如果需要使用AutoGen的代码执行功能代码写进文件再运行需要额外装autogen-ext并选择对应的执行器比如Docker执行器或本地子进程执行器。2.2 第一个代码示例让两个Agent完成一波代码评审这里给一个最简单的双Agent示例把概念讲清楚。场景是代码评审一个开发Agent提交代码描述一个评审Agent负责找问题。from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.messages import TextMessage from autogen_agentchat.ui import Console from autogen_ext.models.openai import OpenAIChatCompletionClient # 配置模型客户端这里以 OpenAI 兼容接口为例 model_client OpenAIChatCompletionClient( modelqwen2.5:72b, # 也可以是 gpt-4o / deepseek-chat 等 api_keyYOUR_API_KEY, base_urlhttps://your-api-endpoint.com/v1, # 兼容 OpenAI 协议即可 model_info{ max_tokens: 8000, vision: False, # 如果模型不支持视觉这里必须设 False }, ) # 定义开发 Agent developer AssistantAgent( namedeveloper, description负责描述代码改动及其意图, model_clientmodel_client, system_message你是一名资深开发工程师请清晰地描述你的代码改动和设计思路。, ) # 定义评审 Agent reviewer AssistantAgent( namereviewer, description负责审查代码改动指出潜在问题, model_clientmodel_client, system_message你是一名严格的代码评审专家请逐项检查改动并提出修改建议。, )上面只是定义了两个角色真正让它们跑起来还需要发起对话。在0.4时代对话触发方式有一点绕你可以让 developer 先发一条 TextMessage然后 reviewer 回复这样来回来去会比较繁琐且不好控制退出。更常用的方式是直接用initiate_chatresult await developer.initiate_chat( recipientreviewer, message我修改了用户认证模块把token过期时间从30分钟改成15分钟并在刷新逻辑里增加了一个异常捕获。, max_turns4, ) # 打印对话过程 await Console(result) # result 里包含了 Agent 之间的所有消息max_turns是控制对轮流次上限的这里限制4轮。实际使用中如果不设上限两个Agent可能会无限聊下去最终消耗大量Token。生产环境我建议必设。2.3 两个关键认知模型选型和收敛控制第一个认知是模型决定Agent的“讨论质量”。我用过小模型和旗舰模型跑同一个评审场景区别非常明显。小模型在Agent多轮对话里容易“随声附和”评审Agent会顺着开发Agent说“没问题”而不是真正挑毛病。旗舰模型能保持角色立场但Token消耗高。实际业务里要看预算和任务复杂度做平衡没有绝对最优。第二个认知是收敛比对话精彩更重要。很多教程展示的是Agent能聊多久但生产环境里我关心的是“能不能在预定轮次内结束”。上面代码里的max_turns还有后面要讲的GroupChat的max_round都是用来保证收敛的。我一开始不敢限制怕聊不充分结果每次都在无意义的往复上烧钱。后来调整策略先设较低上限看结果再逐步放开。实践下来大多数任务在合理上限内都能完成不能完成的往往不是轮次不够而是任务描述或角色定义有问题。3. 从两方对话到群聊协作GroupChat机制解析双Agent只是热身。真实业务里经常需要三个以上角色协作比如产品经理、前端工程师、后端工程师一起讨论需求拆解或者运维、开发、测试一起复盘线上故障。AutoGen的群聊机制对应GroupChat和GroupChatManager。3.1 群聊的核心机制发言权由谁决定GroupChat维护一个Agent列表和所有消息。GroupChatManager是控制器它决定每一轮让哪个Agent说话。默认的speaker_selection_method是auto即让模型根据当前对话内容自动选择下一个发言人。这个设计很巧妙。你不需要把所有跳转逻辑写死只需要给每个Agent写清楚职责描述即description字段管理器的模型会自己判断“这个话题该谁接”。我把它理解成一个会议室里没有主持人但是大家心里都有个默契知道什么问题该谁说话。不过auto模式有两个实际缺陷。一是选错人。模型偶尔把一个话题分配给不相关的Agent导致回答文不对题。二是容易重复选同一个人比如产品经理Agent一直抢话。解决办法是把每个Agent的description写细因为管理器在选择发言人时主要依赖这个字段。另一个模式是manual即由人类在控制台中指名下一个发言人。这个适合调试阶段或者有严格人工审核的场景。还有一种更进阶的玩法自定义选择函数。你可以在函数里写业务规则比如“如果话题包含数据库强制让DBA发言”然后传给GroupChat的speaker_selection_method。我实际项目里混合用了auto加自定义函数既保证灵活性又给关键环节加上硬约束。3.2 群聊最容易踩的坑循环对话和Token失控群聊模式最典型的问题是循环。我遇到过三个Agent在群里来回说“同意”“1”或者两个角色互相甩锅十几轮停不下来。排查的时候发现源头往往是某个Agent的system_message没有写“什么时候该结束讨论”导致它永远在输出新内容。针对性解决方案有三个给GroupChat设置max_round从机制上卡死总轮数。在需要收敛的角色System Prompt里明确要求成员在任务完成时输出特定结束词。通过GroupChatManager的llm_config禁用某些Agent的主动发言权让它们只在被点名时说话。另外要留意Token消耗。群聊模式下所有Agent共享一份消息历史每次调用管理层都需要把全部历史发给模型。轮数一多Token量上升极其快。我做过一次实测五个Agent群聊20轮输入Token轻松到10万以上。生产环境如果没有预算管控建议每轮都检查result里的Token使用量超出阈值就提前终止。3.3 什么时候不该用GroupChat群聊不是万能的。我有一个项目最开始把所有Agent都放进一个GroupChat里结果发现真正干活的时候大部分Agent都在陪跑。后来我把任务拆成了多个小组每组两到三个Agent用不同的会话处理不同子任务最后再汇总。综合下来效果更好单次查询的Token消耗大约下降了40%。还有个细节GroupChat里面的消息历史是可以手动注入的。你可以把之前会话的结果作为新的初始消息加入群聊实现跨会话的“带记忆协作”。这个技巧在做阶段性任务接力时很管用。4. 让Agent真正学会用工具Function Calling的注册与最佳实践Agent不能只聊天最终要落地就必须调用真实工具。AutoGen的函数调用机制本质上是把大模型的Function Call能力和Agent消息循环结合起来。4.1 工具注册的完整示例在AutoGen 0.4里面工具注册的推荐方式是register_for_llm和register_for_execution。前者是告诉模型“有这样一个函数可以用参数是什么”后者是告诉框架“模型说要调这个函数时真的去执行它”。下面给一个简单但完整示例场景是让一个助理Agent学会查询数据库表结构。import json from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.conditions import MaxTurnsTermination, TextMentionTermination from autogen_agentchat.teams import RoundRobinGroupChat from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_core.tools import FunctionTool # 自建函数模拟查询数据库表结构 def get_table_schema(table_name: str) - str: 返回指定表的建表语句摘要用于协助编写SQL。表名示例orders, users table_schemas { orders: CREATE TABLE orders (id INT, user_id INT, amount DECIMAL, created_at DATE);, users: CREATE TABLE users (id INT, name VARCHAR, email VARCHAR);, } return json.dumps({table: table_name, schema: table_schemas.get(table_name, table not found)}) # 包装成 FunctionTool schema_tool FunctionTool( get_table_schema, description查询数据库表结构信息输入表名返回建表语句。, ) # 注册给模型调用 await schema_tool.register_for_llm() model_client OpenAIChatCompletionClient( modelgpt-4o-mini, api_keyYOUR_API_KEY, model_info{max_tokens: 4000, vision: False}, ) sql_assistant AssistantAgent( namesql_assistant, model_clientmodel_client, tools[schema_tool], system_message你是SQL助手需要查询表结构后再编写SQL。, ) # 定义终止条件提到SQL完成或达到最大轮次 termination TextMentionTermination(SQL完成) | MaxTurnsTermination(5) # 用 RoundRobinGroupChat 让用户消息先走Agent 再回复 team RoundRobinGroupChat([sql_assistant], termination_conditiontermination) async for message in team.run_stream(task请查询orders表的表结构然后生成一条统计订单总额的SQL。): print(message)这里有几个细节值得重点说。第一FunctionTool的description字段非常关键。这个描述是给大模型看的不是你给人看的。写得模糊模型就不知道该不该调用、什么时候调用。我见过许多糟糕实践description只写“获取表结构”模型在需要写SQL的时候完全没意识到可以调用。正确写法要包含函数能力、适用条件和调用时机。第二函数返回必须是字符串。框架执行完函数后会把它变成一条消息放回对话。如果你返回的是复杂对象要么转成JSON字符串要么实现自定义序列化。一个常见失误是忘记转字符串结果消息列表里出现“对象无法解析”的报错。第三注册工具的顺序不能乱。register_for_llm()必须在Agent定义前完成否则Agent看不到工具定义。4.2 工具返回的内容会影响后续对话质量工具返回字符串时最好同时包含“结构化结果”和“模型可阅读的说明”。例如返回JSON时可以加一句“以上为表结构请基于该结构编写SQL”。有些模型的Function Call执行结果不会自动做二次理解需要后续的Agent消息来承接。还有一个容易被忽略的点工具执行出错怎么办。AutoGen不会因为函数内部抛异常就终止整个会话但你需要让异常信息以可读形式返回给模型。比如查表不存在时返回“表不存在请确认表名”而不是抛一个Python异常。这样模型才能自行调整下一步。5. 实战一个带记忆和工具调用的代码变更评审Agent前几节讲了基础能力这里把它们拼起来做一个稍微完整、可以直接复用的场景。我在公司里落地过类似方案用来做小型代码评审的预检查效果不错。5.1 场景设定与Agent角色任务开发Agent提交一个代码变更描述评审Agent需要调用静态代码扫描工具这里用一个模拟函数代替再结合扫描结果输出评审意见。评审过程中还需要参考历史评审记录避免提出和上次重复的问题。角色拆成三个dev_agent开发角色负责描述变更内容。scanner_tool审查工具负责对变更文件做“扫描”返回问题列表。review_agent评审角色汇总扫描结果和本次变更上下文给出最终意见。第三个Agent需要记忆能力。AutoGen的默认记忆范围只在单次会话内。如果你希望它记住“上回提过什么意见”有两条路一是把历史内容塞进本次对话的上下文适合历史量小二是外接记忆服务适合长期项目。5.2 实现记忆的正确姿势我用了一个比较轻量级的方案把历史评审结论写进一个JSON文件每次启动新会话时读取注入到前端Agent的System Prompt里。这样做的优点是零依赖缺点是历史量大了会撑爆上下文。后续如果要扩展可以把存储换成向量数据库然后用检索的方式只注入相关的历史片段。这里的关键不是用什么存储而是“注入什么内容”。如果你把所有历史问题一股脑塞进去模型会变得保守新代码检查维度会被稀释。正确做法是只注入“已指出的问题类型”和“对应修复结果”而不是原始对话原文。附一个简化后的代码流程import json, asyncio from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.conditions import TextMentionTermination, MaxTurnsTermination from autogen_agentchat.teams import RoundRobinGroupChat from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_core.tools import FunctionTool # 模拟静态扫描工具 def run_scan_rule(rule_name: str, file_content: str) - str: fake_rules { no_sql_injection: 存在字符串拼接SQL的风险建议使用参数化查询。, no_hardcode_secret: 检测到疑似硬编码密钥请移除并改用环境变量。, } # 这里简单模拟包含关键词则报出风险 if file_content.find(SELECT) -1 and rule_name no_sql_injection: return json.dumps({rule: rule_name, risk: fake_rules[rule_name]}) return json.dumps({rule: rule_name, risk: pass}) scan_tool FunctionTool(run_scan_rule, description对文件内容执行静态校验规则规则名可选no_sql_injection 或 no_hardcode_secret。) await scan_tool.register_for_llm() model_client OpenAIChatCompletionClient( modelgpt-4o, api_keyYOUR_API_KEY, model_info{max_tokens: 6000, vision: False}, ) review_agent AssistantAgent( namereview_agent, model_clientmodel_client, tools[scan_tool], system_message( 你是资深代码评审专家。先调用静态扫描工具检查风险再结合变更描述给出评审意见。 若扫描结果包含pass则只做常规逻辑审查。输出格式问题列表修改建议。 ), ) termination TextMentionTermination(评审完成) | MaxTurnsTermination(6) team RoundRobinGroupChat([review_agent], termination_conditiontermination) async def run_review(change_desc: str, file_content: str): task f变更描述{change_desc}\n文件内容{file_content}\n请开始评审评审完成后输出评审完成。 async for msg in team.run_stream(tasktask): print(msg)跑起来之后review_agent会先调用扫描工具获取结果再结合变更描述返回意见。我实际使用中遇到过一个问题Agent在拿到工具的pass结果后就不再继续做“常规逻辑审查”了直接说“没问题”。这是System Prompt写得不够强导致的后来我加了约束“如果扫描结果全是pass必须额外指出代码可读性或潜在异常风险”。再跑就正常了。5.3 评测视角如何判断我的Agent比之前好这里提一个很多人忽略的环节怎么量化Agent表现。没有评测你只能靠肉眼感觉“好像变聪明了”这对生产迭代是不够的。我给这个评审Agent配了两个量化指标有效问题率评审意见中被人工确认为真问题的比例。漏报率在有标注的样本集里Agent漏掉的关键问题数。这样调整Prompt和工具描述后我可以对比前后变化而不是停留在感觉层面。我之前也看到有人用deepeval这类评测框架做自动化评估和AutoGen本身搭配起来使用。具体做法是把对话结果导出为测试集再跑LLM作为评审者给Agent输出打分本质上和上面两个指标逻辑一致。大模型开发到后面评测一定得先于优化。6. 我在生产服务中踩过的坑和调整方案把AutoGen从Demo推到生产环境中间有不少坑。有些是框架本身的有些是业务设计问题。这里挑几个值得展开的。6.1 并发压力下的模型客户端复用第一个坑是模型客户端的复用。AutoGen官方示例里经常直接OpenAIChatCompletionClient(...)创建一个实例给Agent用。但在Web服务里如果每个用户请求都new一个客户端连接池会被打爆还可能出现限流。我的做法是把model_client提升为模块级单例所有Agent共享。这样请求会自动复用连接也方便统一配置超时和重试。注意这样做有一个前提不同Agent如果使用不同的模型就需要分别建客户端实例不能强行共用一个。6.2 缺少完整trace导致线上问题无法排查第二个大坑是链路追踪。AutoGen的对话过程比较长如果中间某个Agent返回异常内容或者函数调用失败你去看日志可能只有一行报错完全不知道是哪个环节的问题。后来我在每次对话里挂了事件监听把agent、message、tool_call的关键字段记录下来。比如用ConversableAgent的register_reply包装或者直接钩子函数。只要每个步骤有结构化日志线上遇到问题就能快速定位到具体角色和具体消息。这个成本很低但收益极高。没有这段日志在复杂群聊里排查问题就是大海捞针。6.3 轮次和Token成本的双重失控第三个坑是成本。我在第五节说Token控制很重要这里展开讲。AutoGen默认不会主动告诉你“这次花费了多少钱”你需要自己估算。OpenAI类模型的计费逻辑很简单输入Token数乘以单价加输出Token数乘以单价。但多Agent对话的输入Token是累计的因为每轮都要把历史全部带上。我做过一个粗算一个五个Agent的群聊跑20轮假设平均每轮上下文8万Token那光输入就有160万Token。按主流模型价格算单次任务成本可能到几元甚至十几元。这个成本放到高频接口场景里是不可接受的。解决办法有三步设max_turns/max_round上限。任务简单时减少Agent数量不要为对话而对话。对上下文做抽取摘要把早期对话压缩成摘要消息而不是永远带全量历史。特别是第三点在长会话场景里几乎是必选项。AutoGen的消息历史可以手动编辑所以我会在每5轮之后调用一次extract_text生成摘要把旧轮次换成一条“Summary”消息。对话质量会略降但Token量可以压缩一半以上。6.4 有些场景放弃Agent反而是更优解最后说一个反直觉的建议。不是所有任务都适合Agent。我在业务里见过有人非要把“一句话翻译”也包成Agent对话结果多花了好几个来回还容易出错。那种不需要多步骤拆解、不依赖多角色协作的任务直接用单次Prompt带工具调用效果更稳、成本更低。我自己现在接任务的判断标准很简单如果目标任务能在两句话内说清楚并且不需要根据中间结果做多轮决策就用单次调用如果任务天然包含多个角色视角、且需要多步工具调用和结果回填才上Agent。把这两个边界划清楚项目不会为了用框架而用框架。7. 后续可以这样扩展如果你看完这篇想把AutoGen用到自己的项目里我建议按这个顺序往下试。第一步把你手头最频繁的单个任务改造成双Agent流程感受一下消息驱动的编排到底比手写循环好在哪里。第二步接入一个真正的外部工具别用模拟函数让Agent调真实的API或数据库查询体验Function Calling的全链路。第三步再做群聊场景并且严格控制轮次和成本。如果做长了可以考虑AutoGen的分布式扩展能力。目前autogen-core支持基于Actor模型的事件驱动架构理论上可以跨进程运行不同Agent。我曾经用它把“任务拆解Agent”和“工具执行Agent”分开部署互不阻塞。不过这套设计对基础设施的要求更高不建议一上来就上。最后分享一个小技巧写Agent的System Prompt时不要把它当成普通Prompt写而是当成“岗位说明书”。写清楚这个角色的职责边界、输入输出格式、遇到什么情况该做什么遇到什么情况不该做。岗位说明书写得越细Agent在群聊里越不会跑偏。我在那套评审Agent里完善System Prompt之后发言总算不跑题了。
返回列表