做AI应用这些年我越来越觉得“单智能体包打天下”这个思路在真实业务约束下并不可靠。最近我搭了一套内部代号叫agency-agents的模拟项目核心就是让多个智能体像一个小团队一样分工协作。它解决的场景很典型一次任务里既要做资料搜集又要写方案还要做质量审核与其把这些逻辑全塞进一个超大提示词不如拆成几个各有专职的小智能体通过统一的消息通道互相配合。这篇文章会把设计思路、核心代码结构、实际跑通流程以及我踩过的坑都写出来对正在研究多智能体协作的同学来说应该能省下不少时间。1. 整体设计与思路拆解1.1 为什么要拆成多个智能体这事情得从一次真实需求说起。我当时接到的需求是做一个“行业趋势调研”输出一份带数据引用的报告草案。最早我试着让一个智能体一口气完成先让它搜索行业新闻再让它整理数据再让它写报告最后让它检查引用格式。结果很惨模型不是丢数据就是把编造的内容当作搜索结果最离谱的一次是它为了凑字数虚构了两家公司合作的消息。问题不是某个模型能力不行而是单个智能体在同一轮上下文里要承担的角色太多推理注意力被稀释了。人的组织里为什么需要分工因为每次决策都有一个认知边界。智能体也一样当“检索、提炼、写作、校验”互相穿插时模型很难在每一跳都保持高质量。所以我把任务拆成四个角色检索员、分析师、作者、校对员。每个角色是独立的智能体有自己的系统提示词、可调用的工具、输入输出格式约定。这样每个智能体只需要做一件事提示词能写得更具体模型也更容易遵循。你可能会问这不就是几个独立的API调用拼起来吗是但不全是。关键是智能体之间有明确的消息语义和状态流转不是散乱的多次调用。这一步从“函数调用”到“团队协作”的转变才是多智能体框架真正有优势的地方。这个设计模式覆盖的应用范围很广内容生产、营销策划、数据分析、软件研发辅助都能用。只要任务本身可以拆成有先后依赖或者并行处理的子步骤而且每个子步骤都有清晰输入输出就值得用多智能体方案。尤其适合那些以前靠堆提示词、反复抽卡才能勉强跑通的复杂任务换成多智能体之后可控性会明显提升。1.2 方案选型集中调度还是自主协作多智能体协作有几种常见实现方式。我最早想走自主协作模式所有智能体都挂在同一个消息总线上彼此订阅和发布事件A发布了结果B觉得跟自己有关就自动接手。听着灵活实际调试起来非常痛苦因为你根本不知道一条消息到底触发了多少个智能体也不知道某个环节为什么没人处理。后来我换成集中调度模式也叫Orchestrator模式。所有智能体自己不决定下一步做什么而是由一个调度器读任务图按依赖关系依次或并行调用。这样流程是显式的失败了能快速定位是哪一步超时了也能在调度器这一层统一处理。对比两种方案有几点体会自主协作适合探索性强、执行路径不确定的场景但需要很强的消息契约和容错设计。集中调度适合大部分业务流水线尤其是我们这种“任务边界清楚、步骤前后依赖”的场景。如果团队以后要做成通用平台可以设计成组装式底层支持事件总线上层支持有向无环图调度。我在项目里只先实现了后者避免一开始就背上分布式复杂度。选集中调度的另一个原因是成本可控。自主协作模式里智能体之间会反复“商量”每次商量都是一次大模型调用消耗的是真金白银。用调度器控制流程一次任务跑多少步、调用多少次模型是可预测的。影响范围上集中调度模式更适合需要预算控制、服务等级协议和审计要求的团队。比如给客户交付项目你总得告诉对方“这次任务总共调用了多少次模型、耗时多少”集中调度天然能把这份账单直接打出来。自主协作模式在这件事上会非常吃力因为调用次数和路径都不可控。1.3 系统架构总览从任务入口到存储的五层划分这个项目整体分成五层。最外层是入口接收一个任务描述通常是自然语言比如“调研某新兴市场智慧零售的落地情况并输出一份中文简报”。第二层是调度器会做两件事把任务拆成小步骤按依赖关系排成执行计划。第三层是智能体池里面注册了各个智能体对象每个对象都持有自己的系统提示词和工具集。第四层是工具层封装了外部检索API、向量数据库、文件读写、翻译等能力只对智能体暴露统一接口。第五层是存储用来保存每一步的原始输入、输出、耗时、token用量方便回放和审计。消息在这几层之间流转时统一走一个消息对象不搞自定义结构。哪怕两个智能体是同一个开发者写的也别为了省事直接传字典标准化的消息结构在后面做日志和重试时会省非常多事情。从实现角度看调度器不直接持有调用链而是保存一个“步骤定义列表”每个步骤包括 agent_name、input_key、output_key、retry_count、timeout。数据在步骤之间通过一个共享上下文对象读取和写入。这样既清晰又能在任意一步结束后把整个上下文序列化出来做检查。我在实际搭建时先小步跑通一条链路再慢慢加功能。刚开始只有一个入口脚本连存储都省了等链路稳定后才把存储、日志、审计补上。这个顺序很重要如果你一上来就把架构搭得很重迭代和调参都变得非常慢。2. 核心细节解析与实操要点2.1 智能体角色怎么定义角色定义是整个项目的基石。我总结下来的经验是角色信息至少要包含五个字段缺一个后面都会出问题。以检索员为例nameagent-researcher用于日志和调度。role一句话描述职责比如“负责抽取用户问题中的检索关键词并调用搜索工具”。system_prompt详细说明它是谁、它要做什么、它不能做什么、输出格式必须是什么。tools允许调用的工具白名单比如 search_web、read_url但不能调用 write_doc。result_schema一个JSON Schema或一个Pydantic模型用于输出校验。很多半路出家做智能体的开发者容易忽略 result_schema让模型自己自由发挥。自由发挥的结果就是解析报错或者下游智能体根本看不懂上游文本。真正稳的做法是给每个智能体定义严格的输出Schema并且要求输出纯JSON同时开启强制JSON模式。这里还有一个容易踩的细节system_prompt 里不要写“你是团队里的助手”这种模糊话术要写“当且仅当你需要查找最新资料时调用 search_web禁止自行编造数据。如果你没有找到有效信息请明确输出 empty_result 状态”。给模型一个“没有结果时该怎么办”的出口比单纯叮嘱它别瞎编更有用。角色定义这件事直接影响整个链路的下限。如果角色定义模糊后面加再多的校验和重试都只是补救因为脏数据源头没堵住。我建议每个角色分配好后先用一个最简单的测试任务验证输出格式满意了再交给调度器编排不要一上来就串全链路。2.2 消息协议与通信规则智能体之间通信如果靠自然语言文本直接拼很快就乱。我在项目里定义了一个统一消息类字段如下message_id全局唯一ID用于追踪和幂等。sender发送方 agent_name。receiver接收方 agent_name或者是 orchestrator。task本次调用的任务描述可能是原始任务的一部分也可能是经过解析后的结构化指令。payload上下文数据通常是字典存放上游输出。statuspending、running、succeeded、failed。error出错时的异常信息或错误码。token_usage记录本次调用消耗的输入/输出token数。这个结构看起来简单但把三个事情解决了一是消息可追踪出了问题直接把 message_id 拿去日志过滤器查几秒钟就能看到整条链路二是消息可重试如果下游超时调度器可以用同一 message_id 重启步骤而不会重复执行副作用三是消息可审计每一步谁处理的、花了多少token、结果是什么都一目了然。通信规则上我明确规定所有智能体不能相互之间直接发消息只能通过调度器转发。这是刻意牺牲一点灵活性换来的可控性。如果某天你真需要两个智能体私下同步状态可以在调度器里加一个“会议室”机制把它们的对话内容记录成一组消息本质上还是被调度器包住。这套通信规则带来的直接好处是新接一个智能体非常快你不需要知道其他智能体是怎么写的只需要知道消息协议长什么样。对一个小团队来说这大大降低了协作开发的沟通成本。我曾经试过让两个智能体直接对话结果双方天然形成了一个消息风暴日志根本看不过来从那以后我彻底倒向集中转发。2.3 任务分解与结果聚合任务分解不是拍脑袋我是按“输出物反推”来做的。先定义整个任务的最终产物是什么再往回拆要产出最终报告需要章节要写章节需要数据和案例要拿数据和案例需要检索和分析。每一步都是对下一步输出物的具体化。比如“某海外新兴市场智慧零售简报”这个任务我拆成这样检索员搜索“新兴市场 智慧零售 落地案例”整理出十个候选资料输出带链接和摘要的列表。分析师读取这十个候选筛选去重提取三个关键趋势输出结构化结论。作者读取趋势结论按照预设模板编写简报正文。校对员核对正文中的事实陈述是否有来源支持输出修订意见。调度器把修订意见合并给作者再跑一次产出终稿。结果聚合放在调度器里做。每个步骤的输出先写入共享上下文某个字段依赖多个上游时调度器只取依赖的key聚合不把整个上下文传给模型。这能显著降低模型输入空间减少干扰。聚合时还有一个细节要处理“上游输出为空”的情况。我写了一个 result_selector优先取 statussucceeded 且结果非空的字段如果所有上游都失败整个任务直接短路返回失败报告不往下游传空气。很多人容易漏掉这个边界导致下游拿到一个空字典之后照样硬跑最终输出一堆无意义内容。任务分解的粒度也要控制。粒度太粗单个智能体还是压力很大粒度太细调度开销和token浪费又会抵消收益。我的参考线是每个智能体完成的任务不超过一个独立业务环节输出信息不超过两三百个结构化字段。超过这个范围我宁可多拆一个角色。3. 实操过程与核心环节实现3.1 运行环境准备与依赖安装一页纸清单我先说下我这个项目的运行环境操作系统是常见的Linux容器Python版本3.10以上核心依赖包括 pydantic用来做输出Schema校验、requests调用外部API、orjson高性能JSON序列化。如果你用兼容大模型接口还需要一个API Key放在环境变量里不要写进代码。安装命令也不复杂pip install pydantic requests orjson python-dotenv然后创建一个.env文件LLM_API_BASEhttps://your-endpoint.example.com/v1 LLM_API_KEYsk-xxxxxxxx LLM_MODELmodel-name注意那个API地址你们内部服务或者第三方兼容服务都可以但所有智能体共用同一个模型接入层。想让不同智能体用不同模型只需要在Agent初始化时传入不同的model字段。这里有一条实操经验API Key和模型名称不要写死在代码里也不要写死在系统提示词里。因为智能体角色定义经常要改改配置文件总比重发代码要安全。我见过同事把模型名写在提示词里换模型的时候还得逐字核对提示词非常容易漏。3.2 Agent基类实现一个可继承的骨架项目里所有智能体都继承同一个基类。基类做了三件事格式化系统提示词、调用大模型接口、解析并校验输出。没有把业务逻辑塞在基类里具体工具怎么用交给子类。这样新加一个智能体只要几分钟。# agent_base.py import json import os import requests from abc import ABC, abstractmethod from typing import Any, Dict class Agent(ABC): def __init__(self, name: str, role: str, system_prompt: str, tools: list None, result_schema: Any None, model: str None): self.name name self.role role self.system_prompt system_prompt self.tools tools or [] self.result_schema result_schema self.model model or os.getenv(LLM_MODEL) def build_user_message(self, task: str, context: Dict[str, Any]) - str: # 具体子类可以覆盖默认把任务和相关上下文序列化成JSON entire {task: task, context: context} return json.dumps(entire, ensure_asciiFalse) def call_llm(self, messages: list) - str: endpoint os.getenv(LLM_API_BASE) api_key os.getenv(LLM_API_KEY) resp requests.post( f{endpoint}/chat/completions, headers{Authorization: fBearer {api_key}}, json{model: self.model, messages: messages, temperature: 0.3}, timeout60, ) resp.raise_for_status() return resp.json()[choices][0][message][content] def validate_result(self, text: str) - Dict[str, Any]: if not self.result_schema: return {raw_text: text} data json.loads(text) return self.result_schema(**data).model_dump() def run(self, task: str, context: Dict[str, Any]) - Dict[str, Any]: user_content self.build_user_message(task, context) messages [ {role: system, content: self.system_prompt}, {role: user, content: user_content}, ] raw_output self.call_llm(messages) return self.validate_result(raw_output)基类里我把call_llm做成了标准HTTP调用。为什么要用 pydantic 来做输出Schema校验因为模型偶尔会输出json格式不够规范比如尾逗号、单引号json.loads会直接报错。用pydantic可以先加载再校验字段类型和必填都能兜住。这里还有个经验不要只在最后一步做输出校验而要在每个智能体执行完都做。宁可让下游拿到一个“校验失败”的错误也不要拿到一份看起来像那么回事、但字段缺失的脏数据。脏数据会静默地污染整条链路错误则能被迅速发现和阻断。3.3 调度器实现从顺序流到自愈机制调度器是项目里最需要花心思的部分。我只实现了顺序流和有向无环图两种模式顺序流已经够用了。调度器持有一个 context 字典按步骤列表逐个执行。# orchestrator.py from typing import Dict, Any, List class Orchestrator: def __init__(self, agents: Dict[str, Agent]): self.agents agents self.context: Dict[str, Any] {} def add_context(self, key: str, value: Any) - None: self.context[key] value def run_step(self, step: Dict[str, str]) - Dict[str, Any]: agent self.agents[step[agent]] task step[task] input_context { key: self.context[key] for key in step.get(input_keys, []) } return agent.run(task, input_context) def run(self, steps: List[Dict[str, str]], initial_context: Dict[str, Any] None) - Dict[str, Any]: if initial_context: self.context.update(initial_context) for i, step in enumerate(steps): output_key step.get(output_key, fstep_{i}) result self.run_step(step) self.context[output_key] result print(f[orchestrator] step {i} done, agent{step[agent]}) return self.context这个简化版能跑但真实场景里还要加三样东西。第一是超时控制给每个步骤设一个最大等待时间大模型调用超过60秒就按失败处理。第二是重试失败时根据错误类型判断是重试还是短路网络抖动可以重试输出校验失败也建议重试一次但系统提示词写错导致的原则性问题重试也没用。第三是上下文序列化每完成一步就把 context 写到本地JSON文件崩溃之后能恢复现场。我强烈建议你在调度器里加一个 max_steps 限制防止有人误配了带环的步骤定义任务在Agent之间无限循环账单直接爆炸。我设的是总步数不超过20步。在自愈机制上我做过一个小demo如果校对员返回了问题列表调度器不会直接把任务标成失败而是追加一个“修订步骤”把问题列表合并给作者重新生成一版。这个追加逻辑的触发条件和数据传递同样写成分离的步骤定义而不是在代码里硬编码。这样整个流程依然可观察、可审计。3.4 跑通一个文本生成链路现在把三个人物实例化出来检索员、作者、校对员。用最简单的方式串起一条链路。# main.py researcher Agent( nameresearcher, role检索汇总最新资料, system_prompt( 你是检索员。当且仅当需要查找最新资料时调用search_web工具。 你必须输出JSON{\results\: [{\title\: str, \url\: str, \summary\: str}], \status\: str}。 如果没有有效资料status输出empty不要编造。 ), tools[search_web], result_schemaSearchResultsSchema, ) writer Agent( namewriter, role基于资料撰写简报, system_prompt( 你是作者。根据上游提供的results写一份600字以内的中文简报。 输出JSON{\content\: str, \sources\: [str]}。 ), result_schemaWriterOutputSchema, ) editor Agent( nameeditor, role校对简报事实和格式, system_prompt( 你是校对员。检查content中每个核心论断是否能在sources中找到对应支撑。 输出JSON{\pass\: bool, \issues\: [str]}。 ), result_schemaEditorOutputSchema, ) orchestrator Orchestrator(agents{ researcher: researcher, writer: writer, editor: editor, }) steps [ {agent: researcher, task: 新兴市场智慧零售落地案例, output_key: research}, {agent: writer, task: 基于research撰写简报, input_keys: [research], output_key: draft}, {agent: editor, task: 校对draft, input_keys: [draft], output_key: review}, ] final_context orchestrator.run(steps, initial_context{topic: 智慧零售}) print(final_context[review])执行完你会在日志里看到三步依次完成。上下文对象里最后会有 research、draft、review 三个结果。如果校对员返回 passfalse调度器可以追加一个步骤把 issues 合并给作者再改一版这个逻辑我一般写在run外面作为自愈机制。这里要提醒一下实际项目里不要把三个步骤都放在同一个进程里。我后面把它们拆成了三个独立服务用消息队列传递结果这样任一环节崩溃不会拖垮整个流水线。不过在小规模跑通阶段先用同一个进程把逻辑跑顺比一上来就搞分布式实用得多。如果你想把这条链路延伸成生产级服务需要补的东西也就清晰了消息队列、任务队列、数据库存储、监控面板、调用链追踪。每一步都不难难的是在链路变长之后还能快速定位问题所以我在很早的时候就把结构化日志和message_id设计进去了。4. 常见问题与排查技巧实录4.1 多个智能体来回踢皮球这是我第一次跑通多智能体时最常遇到的问题。两个智能体互相要求对方补充信息来回十几轮任务就是结束不了。后来定位到原因是系统提示词里给智能体留了“重新表述问题”的选项它俩就顺着这个口子无休止地确认。解决方法是两条一是明确禁止智能体反向提问每个智能体只接受调度器分发的任务任务要求模糊就直接输出 insufficient_info 状态不许把问题抛回去二是在调度器层面限制每个步骤的执行次数超过就强制失败。这个现象在自主协作架构里尤其常见因为消息可以直接互相触达没有中间人做裁决。哪怕你用的是集中调度也要小心步骤定义里意外出现环。我每次修改步骤列表后都会跑一个简单的环检测脚本确保执行计划是无环的。4.2 上下文窗口溢出多智能体协作最容易把整个上下文传给每个步骤。我的分析师初版 prompt 里直接序列化了全部历史输出导致后几个智能体的输入token量远超标。这个问题的根本原因是对“哪些上下文是当前步骤真正需要的”没有一个清晰的判断。我的做法是给每个步骤声明 input_keys只把需要的字段传给智能体。比如校对员只需要 draft 里的 content 和 sources根本不需要 research。如果你发现输入token还是很大可以进一步让前一个智能体输出“摘要版”而不是把原始全文传给下一位。用摘要传递信息整个链条的token消耗是线性增长的而不是指数爆炸。这里还有个容易被低估的点即使输入token没有超过模型限制过大的上下文也会明显降低输出质量。模型会优先注意开头结尾的信息中间的历史数据常常被忽略。所以按需传上下文不只是为了省钱更是为了保证下游智能体的推理质量。4.3 输出格式解析报错模型虽然被要求输出JSON但还是经常出错。最常见的是多行字符串没转义以及模型自作聪明在JSON外面加了markdown代码块标记。我在 validate_result 里做了两件防御text text.strip() if text.startswith(): first_line text.find(\n) text text[first_line:].strip() text text.rsplit(, 1)[0].strip()然后才是json.loads。另外如果输出的JSON里字段类型不对pydantic的 ValidationError 会明确告诉我哪个字段出问题方便快速调整提示词示例。这个问题的深层原因是模型对“输出纯JSON”的理解不稳定。只靠提示词约束不够还要在上游智能体发出结果后立刻校验一次。凡是校验不通过的就触发一次重试并且把上次错误信息写回提示词让模型知道哪里出问题了。重试两次还不行再判断是提示词写错还是任务本身有问题。4.4 日志与链路追踪技巧在这类项目里日志不是可选项是保命项。我的每个步骤都打印一个结构化日志行包含 message_id、agent、step、status、耗时、token用量、错误摘要。查询的时候用 grep 过滤 message_id就能还原整条链路。我把常用的排查问题整理成了一个表方便大家对照现象可能原因排查/解决方案智能体之间循环对话提示词允许反向提问或调度步骤存在环提示词禁止提问调度器加max_steps下游经常解析失败上游输出Schema未强制校验每个智能体输出都过pydantic校验token消耗超出预期上下文传得太全没有按需传字段声明input_keys只传必要字段同一个bug反复出现没有记录message_id无法定位链路日志加上message_id统一过滤任务偶尔完全失败某个步骤超时后续步骤拿到None加超时和重试错误信息写入context还有一点是提示词版本管理。我每次修改某个智能体的 system_prompt都会把这个prompt连同当时的测试用例一起存到一个prompts目录里。智能体优化本质上是prompt迭代没有版本管理你就不知道哪次改动导致效果回退而且这种回退在系统里很难通过代码测试发现。如果你用的是Git仓库建议把提示词也纳入版本管理。每次改动提示词都单独提一个分支跑一遍回归用例再合入。这比直接在生产环境里改提示词要稳得多。我后来能快速定位各种问题很大程度就是靠这层提示词版本库。最后说点个人体会。跑这个项目给我的最大收获不是“多智能体有多强”而是“限制会让系统更可靠”。你每允许智能体自由发挥一步就要准备一个兜底策略你每着急删掉一条校验规则后面就会用一次现场事故补回来。如果让我重新设计我还是会先把调度逻辑、输出Schema和日志体系做扎实再谈花哨的自主协作。这也算是我踩了不少坑之后的一条实用心得吧。
做AI应用这些年我越来越觉得“单智能体包打天下”这个思路在真实业务约束下并不可靠。最近我搭了一套内部代号叫agency-agents的模拟项目核心就是让多个智能体像一个小团队一样分工协作。它解决的场景很典型一次任务里既要做资料搜集又要写方案还要做质量审核与其把这些逻辑全塞进一个超大提示词不如拆成几个各有专职的小智能体通过统一的消息通道互相配合。这篇文章会把设计思路、核心代码结构、实际跑通流程以及我踩过的坑都写出来对正在研究多智能体协作的同学来说应该能省下不少时间。1. 整体设计与思路拆解1.1 为什么要拆成多个智能体这事情得从一次真实需求说起。我当时接到的需求是做一个“行业趋势调研”输出一份带数据引用的报告草案。最早我试着让一个智能体一口气完成先让它搜索行业新闻再让它整理数据再让它写报告最后让它检查引用格式。结果很惨模型不是丢数据就是把编造的内容当作搜索结果最离谱的一次是它为了凑字数虚构了两家公司合作的消息。问题不是某个模型能力不行而是单个智能体在同一轮上下文里要承担的角色太多推理注意力被稀释了。人的组织里为什么需要分工因为每次决策都有一个认知边界。智能体也一样当“检索、提炼、写作、校验”互相穿插时模型很难在每一跳都保持高质量。所以我把任务拆成四个角色检索员、分析师、作者、校对员。每个角色是独立的智能体有自己的系统提示词、可调用的工具、输入输出格式约定。这样每个智能体只需要做一件事提示词能写得更具体模型也更容易遵循。你可能会问这不就是几个独立的API调用拼起来吗是但不全是。关键是智能体之间有明确的消息语义和状态流转不是散乱的多次调用。这一步从“函数调用”到“团队协作”的转变才是多智能体框架真正有优势的地方。这个设计模式覆盖的应用范围很广内容生产、营销策划、数据分析、软件研发辅助都能用。只要任务本身可以拆成有先后依赖或者并行处理的子步骤而且每个子步骤都有清晰输入输出就值得用多智能体方案。尤其适合那些以前靠堆提示词、反复抽卡才能勉强跑通的复杂任务换成多智能体之后可控性会明显提升。1.2 方案选型集中调度还是自主协作多智能体协作有几种常见实现方式。我最早想走自主协作模式所有智能体都挂在同一个消息总线上彼此订阅和发布事件A发布了结果B觉得跟自己有关就自动接手。听着灵活实际调试起来非常痛苦因为你根本不知道一条消息到底触发了多少个智能体也不知道某个环节为什么没人处理。后来我换成集中调度模式也叫Orchestrator模式。所有智能体自己不决定下一步做什么而是由一个调度器读任务图按依赖关系依次或并行调用。这样流程是显式的失败了能快速定位是哪一步超时了也能在调度器这一层统一处理。对比两种方案有几点体会自主协作适合探索性强、执行路径不确定的场景但需要很强的消息契约和容错设计。集中调度适合大部分业务流水线尤其是我们这种“任务边界清楚、步骤前后依赖”的场景。如果团队以后要做成通用平台可以设计成组装式底层支持事件总线上层支持有向无环图调度。我在项目里只先实现了后者避免一开始就背上分布式复杂度。选集中调度的另一个原因是成本可控。自主协作模式里智能体之间会反复“商量”每次商量都是一次大模型调用消耗的是真金白银。用调度器控制流程一次任务跑多少步、调用多少次模型是可预测的。影响范围上集中调度模式更适合需要预算控制、服务等级协议和审计要求的团队。比如给客户交付项目你总得告诉对方“这次任务总共调用了多少次模型、耗时多少”集中调度天然能把这份账单直接打出来。自主协作模式在这件事上会非常吃力因为调用次数和路径都不可控。1.3 系统架构总览从任务入口到存储的五层划分这个项目整体分成五层。最外层是入口接收一个任务描述通常是自然语言比如“调研某新兴市场智慧零售的落地情况并输出一份中文简报”。第二层是调度器会做两件事把任务拆成小步骤按依赖关系排成执行计划。第三层是智能体池里面注册了各个智能体对象每个对象都持有自己的系统提示词和工具集。第四层是工具层封装了外部检索API、向量数据库、文件读写、翻译等能力只对智能体暴露统一接口。第五层是存储用来保存每一步的原始输入、输出、耗时、token用量方便回放和审计。消息在这几层之间流转时统一走一个消息对象不搞自定义结构。哪怕两个智能体是同一个开发者写的也别为了省事直接传字典标准化的消息结构在后面做日志和重试时会省非常多事情。从实现角度看调度器不直接持有调用链而是保存一个“步骤定义列表”每个步骤包括 agent_name、input_key、output_key、retry_count、timeout。数据在步骤之间通过一个共享上下文对象读取和写入。这样既清晰又能在任意一步结束后把整个上下文序列化出来做检查。我在实际搭建时先小步跑通一条链路再慢慢加功能。刚开始只有一个入口脚本连存储都省了等链路稳定后才把存储、日志、审计补上。这个顺序很重要如果你一上来就把架构搭得很重迭代和调参都变得非常慢。2. 核心细节解析与实操要点2.1 智能体角色怎么定义角色定义是整个项目的基石。我总结下来的经验是角色信息至少要包含五个字段缺一个后面都会出问题。以检索员为例nameagent-researcher用于日志和调度。role一句话描述职责比如“负责抽取用户问题中的检索关键词并调用搜索工具”。system_prompt详细说明它是谁、它要做什么、它不能做什么、输出格式必须是什么。tools允许调用的工具白名单比如 search_web、read_url但不能调用 write_doc。result_schema一个JSON Schema或一个Pydantic模型用于输出校验。很多半路出家做智能体的开发者容易忽略 result_schema让模型自己自由发挥。自由发挥的结果就是解析报错或者下游智能体根本看不懂上游文本。真正稳的做法是给每个智能体定义严格的输出Schema并且要求输出纯JSON同时开启强制JSON模式。这里还有一个容易踩的细节system_prompt 里不要写“你是团队里的助手”这种模糊话术要写“当且仅当你需要查找最新资料时调用 search_web禁止自行编造数据。如果你没有找到有效信息请明确输出 empty_result 状态”。给模型一个“没有结果时该怎么办”的出口比单纯叮嘱它别瞎编更有用。角色定义这件事直接影响整个链路的下限。如果角色定义模糊后面加再多的校验和重试都只是补救因为脏数据源头没堵住。我建议每个角色分配好后先用一个最简单的测试任务验证输出格式满意了再交给调度器编排不要一上来就串全链路。2.2 消息协议与通信规则智能体之间通信如果靠自然语言文本直接拼很快就乱。我在项目里定义了一个统一消息类字段如下message_id全局唯一ID用于追踪和幂等。sender发送方 agent_name。receiver接收方 agent_name或者是 orchestrator。task本次调用的任务描述可能是原始任务的一部分也可能是经过解析后的结构化指令。payload上下文数据通常是字典存放上游输出。statuspending、running、succeeded、failed。error出错时的异常信息或错误码。token_usage记录本次调用消耗的输入/输出token数。这个结构看起来简单但把三个事情解决了一是消息可追踪出了问题直接把 message_id 拿去日志过滤器查几秒钟就能看到整条链路二是消息可重试如果下游超时调度器可以用同一 message_id 重启步骤而不会重复执行副作用三是消息可审计每一步谁处理的、花了多少token、结果是什么都一目了然。通信规则上我明确规定所有智能体不能相互之间直接发消息只能通过调度器转发。这是刻意牺牲一点灵活性换来的可控性。如果某天你真需要两个智能体私下同步状态可以在调度器里加一个“会议室”机制把它们的对话内容记录成一组消息本质上还是被调度器包住。这套通信规则带来的直接好处是新接一个智能体非常快你不需要知道其他智能体是怎么写的只需要知道消息协议长什么样。对一个小团队来说这大大降低了协作开发的沟通成本。我曾经试过让两个智能体直接对话结果双方天然形成了一个消息风暴日志根本看不过来从那以后我彻底倒向集中转发。2.3 任务分解与结果聚合任务分解不是拍脑袋我是按“输出物反推”来做的。先定义整个任务的最终产物是什么再往回拆要产出最终报告需要章节要写章节需要数据和案例要拿数据和案例需要检索和分析。每一步都是对下一步输出物的具体化。比如“某海外新兴市场智慧零售简报”这个任务我拆成这样检索员搜索“新兴市场 智慧零售 落地案例”整理出十个候选资料输出带链接和摘要的列表。分析师读取这十个候选筛选去重提取三个关键趋势输出结构化结论。作者读取趋势结论按照预设模板编写简报正文。校对员核对正文中的事实陈述是否有来源支持输出修订意见。调度器把修订意见合并给作者再跑一次产出终稿。结果聚合放在调度器里做。每个步骤的输出先写入共享上下文某个字段依赖多个上游时调度器只取依赖的key聚合不把整个上下文传给模型。这能显著降低模型输入空间减少干扰。聚合时还有一个细节要处理“上游输出为空”的情况。我写了一个 result_selector优先取 statussucceeded 且结果非空的字段如果所有上游都失败整个任务直接短路返回失败报告不往下游传空气。很多人容易漏掉这个边界导致下游拿到一个空字典之后照样硬跑最终输出一堆无意义内容。任务分解的粒度也要控制。粒度太粗单个智能体还是压力很大粒度太细调度开销和token浪费又会抵消收益。我的参考线是每个智能体完成的任务不超过一个独立业务环节输出信息不超过两三百个结构化字段。超过这个范围我宁可多拆一个角色。3. 实操过程与核心环节实现3.1 运行环境准备与依赖安装一页纸清单我先说下我这个项目的运行环境操作系统是常见的Linux容器Python版本3.10以上核心依赖包括 pydantic用来做输出Schema校验、requests调用外部API、orjson高性能JSON序列化。如果你用兼容大模型接口还需要一个API Key放在环境变量里不要写进代码。安装命令也不复杂pip install pydantic requests orjson python-dotenv然后创建一个.env文件LLM_API_BASEhttps://your-endpoint.example.com/v1 LLM_API_KEYsk-xxxxxxxx LLM_MODELmodel-name注意那个API地址你们内部服务或者第三方兼容服务都可以但所有智能体共用同一个模型接入层。想让不同智能体用不同模型只需要在Agent初始化时传入不同的model字段。这里有一条实操经验API Key和模型名称不要写死在代码里也不要写死在系统提示词里。因为智能体角色定义经常要改改配置文件总比重发代码要安全。我见过同事把模型名写在提示词里换模型的时候还得逐字核对提示词非常容易漏。3.2 Agent基类实现一个可继承的骨架项目里所有智能体都继承同一个基类。基类做了三件事格式化系统提示词、调用大模型接口、解析并校验输出。没有把业务逻辑塞在基类里具体工具怎么用交给子类。这样新加一个智能体只要几分钟。# agent_base.py import json import os import requests from abc import ABC, abstractmethod from typing import Any, Dict class Agent(ABC): def __init__(self, name: str, role: str, system_prompt: str, tools: list None, result_schema: Any None, model: str None): self.name name self.role role self.system_prompt system_prompt self.tools tools or [] self.result_schema result_schema self.model model or os.getenv(LLM_MODEL) def build_user_message(self, task: str, context: Dict[str, Any]) - str: # 具体子类可以覆盖默认把任务和相关上下文序列化成JSON entire {task: task, context: context} return json.dumps(entire, ensure_asciiFalse) def call_llm(self, messages: list) - str: endpoint os.getenv(LLM_API_BASE) api_key os.getenv(LLM_API_KEY) resp requests.post( f{endpoint}/chat/completions, headers{Authorization: fBearer {api_key}}, json{model: self.model, messages: messages, temperature: 0.3}, timeout60, ) resp.raise_for_status() return resp.json()[choices][0][message][content] def validate_result(self, text: str) - Dict[str, Any]: if not self.result_schema: return {raw_text: text} data json.loads(text) return self.result_schema(**data).model_dump() def run(self, task: str, context: Dict[str, Any]) - Dict[str, Any]: user_content self.build_user_message(task, context) messages [ {role: system, content: self.system_prompt}, {role: user, content: user_content}, ] raw_output self.call_llm(messages) return self.validate_result(raw_output)基类里我把call_llm做成了标准HTTP调用。为什么要用 pydantic 来做输出Schema校验因为模型偶尔会输出json格式不够规范比如尾逗号、单引号json.loads会直接报错。用pydantic可以先加载再校验字段类型和必填都能兜住。这里还有个经验不要只在最后一步做输出校验而要在每个智能体执行完都做。宁可让下游拿到一个“校验失败”的错误也不要拿到一份看起来像那么回事、但字段缺失的脏数据。脏数据会静默地污染整条链路错误则能被迅速发现和阻断。3.3 调度器实现从顺序流到自愈机制调度器是项目里最需要花心思的部分。我只实现了顺序流和有向无环图两种模式顺序流已经够用了。调度器持有一个 context 字典按步骤列表逐个执行。# orchestrator.py from typing import Dict, Any, List class Orchestrator: def __init__(self, agents: Dict[str, Agent]): self.agents agents self.context: Dict[str, Any] {} def add_context(self, key: str, value: Any) - None: self.context[key] value def run_step(self, step: Dict[str, str]) - Dict[str, Any]: agent self.agents[step[agent]] task step[task] input_context { key: self.context[key] for key in step.get(input_keys, []) } return agent.run(task, input_context) def run(self, steps: List[Dict[str, str]], initial_context: Dict[str, Any] None) - Dict[str, Any]: if initial_context: self.context.update(initial_context) for i, step in enumerate(steps): output_key step.get(output_key, fstep_{i}) result self.run_step(step) self.context[output_key] result print(f[orchestrator] step {i} done, agent{step[agent]}) return self.context这个简化版能跑但真实场景里还要加三样东西。第一是超时控制给每个步骤设一个最大等待时间大模型调用超过60秒就按失败处理。第二是重试失败时根据错误类型判断是重试还是短路网络抖动可以重试输出校验失败也建议重试一次但系统提示词写错导致的原则性问题重试也没用。第三是上下文序列化每完成一步就把 context 写到本地JSON文件崩溃之后能恢复现场。我强烈建议你在调度器里加一个 max_steps 限制防止有人误配了带环的步骤定义任务在Agent之间无限循环账单直接爆炸。我设的是总步数不超过20步。在自愈机制上我做过一个小demo如果校对员返回了问题列表调度器不会直接把任务标成失败而是追加一个“修订步骤”把问题列表合并给作者重新生成一版。这个追加逻辑的触发条件和数据传递同样写成分离的步骤定义而不是在代码里硬编码。这样整个流程依然可观察、可审计。3.4 跑通一个文本生成链路现在把三个人物实例化出来检索员、作者、校对员。用最简单的方式串起一条链路。# main.py researcher Agent( nameresearcher, role检索汇总最新资料, system_prompt( 你是检索员。当且仅当需要查找最新资料时调用search_web工具。 你必须输出JSON{\results\: [{\title\: str, \url\: str, \summary\: str}], \status\: str}。 如果没有有效资料status输出empty不要编造。 ), tools[search_web], result_schemaSearchResultsSchema, ) writer Agent( namewriter, role基于资料撰写简报, system_prompt( 你是作者。根据上游提供的results写一份600字以内的中文简报。 输出JSON{\content\: str, \sources\: [str]}。 ), result_schemaWriterOutputSchema, ) editor Agent( nameeditor, role校对简报事实和格式, system_prompt( 你是校对员。检查content中每个核心论断是否能在sources中找到对应支撑。 输出JSON{\pass\: bool, \issues\: [str]}。 ), result_schemaEditorOutputSchema, ) orchestrator Orchestrator(agents{ researcher: researcher, writer: writer, editor: editor, }) steps [ {agent: researcher, task: 新兴市场智慧零售落地案例, output_key: research}, {agent: writer, task: 基于research撰写简报, input_keys: [research], output_key: draft}, {agent: editor, task: 校对draft, input_keys: [draft], output_key: review}, ] final_context orchestrator.run(steps, initial_context{topic: 智慧零售}) print(final_context[review])执行完你会在日志里看到三步依次完成。上下文对象里最后会有 research、draft、review 三个结果。如果校对员返回 passfalse调度器可以追加一个步骤把 issues 合并给作者再改一版这个逻辑我一般写在run外面作为自愈机制。这里要提醒一下实际项目里不要把三个步骤都放在同一个进程里。我后面把它们拆成了三个独立服务用消息队列传递结果这样任一环节崩溃不会拖垮整个流水线。不过在小规模跑通阶段先用同一个进程把逻辑跑顺比一上来就搞分布式实用得多。如果你想把这条链路延伸成生产级服务需要补的东西也就清晰了消息队列、任务队列、数据库存储、监控面板、调用链追踪。每一步都不难难的是在链路变长之后还能快速定位问题所以我在很早的时候就把结构化日志和message_id设计进去了。4. 常见问题与排查技巧实录4.1 多个智能体来回踢皮球这是我第一次跑通多智能体时最常遇到的问题。两个智能体互相要求对方补充信息来回十几轮任务就是结束不了。后来定位到原因是系统提示词里给智能体留了“重新表述问题”的选项它俩就顺着这个口子无休止地确认。解决方法是两条一是明确禁止智能体反向提问每个智能体只接受调度器分发的任务任务要求模糊就直接输出 insufficient_info 状态不许把问题抛回去二是在调度器层面限制每个步骤的执行次数超过就强制失败。这个现象在自主协作架构里尤其常见因为消息可以直接互相触达没有中间人做裁决。哪怕你用的是集中调度也要小心步骤定义里意外出现环。我每次修改步骤列表后都会跑一个简单的环检测脚本确保执行计划是无环的。4.2 上下文窗口溢出多智能体协作最容易把整个上下文传给每个步骤。我的分析师初版 prompt 里直接序列化了全部历史输出导致后几个智能体的输入token量远超标。这个问题的根本原因是对“哪些上下文是当前步骤真正需要的”没有一个清晰的判断。我的做法是给每个步骤声明 input_keys只把需要的字段传给智能体。比如校对员只需要 draft 里的 content 和 sources根本不需要 research。如果你发现输入token还是很大可以进一步让前一个智能体输出“摘要版”而不是把原始全文传给下一位。用摘要传递信息整个链条的token消耗是线性增长的而不是指数爆炸。这里还有个容易被低估的点即使输入token没有超过模型限制过大的上下文也会明显降低输出质量。模型会优先注意开头结尾的信息中间的历史数据常常被忽略。所以按需传上下文不只是为了省钱更是为了保证下游智能体的推理质量。4.3 输出格式解析报错模型虽然被要求输出JSON但还是经常出错。最常见的是多行字符串没转义以及模型自作聪明在JSON外面加了markdown代码块标记。我在 validate_result 里做了两件防御text text.strip() if text.startswith(): first_line text.find(\n) text text[first_line:].strip() text text.rsplit(, 1)[0].strip()然后才是json.loads。另外如果输出的JSON里字段类型不对pydantic的 ValidationError 会明确告诉我哪个字段出问题方便快速调整提示词示例。这个问题的深层原因是模型对“输出纯JSON”的理解不稳定。只靠提示词约束不够还要在上游智能体发出结果后立刻校验一次。凡是校验不通过的就触发一次重试并且把上次错误信息写回提示词让模型知道哪里出问题了。重试两次还不行再判断是提示词写错还是任务本身有问题。4.4 日志与链路追踪技巧在这类项目里日志不是可选项是保命项。我的每个步骤都打印一个结构化日志行包含 message_id、agent、step、status、耗时、token用量、错误摘要。查询的时候用 grep 过滤 message_id就能还原整条链路。我把常用的排查问题整理成了一个表方便大家对照现象可能原因排查/解决方案智能体之间循环对话提示词允许反向提问或调度步骤存在环提示词禁止提问调度器加max_steps下游经常解析失败上游输出Schema未强制校验每个智能体输出都过pydantic校验token消耗超出预期上下文传得太全没有按需传字段声明input_keys只传必要字段同一个bug反复出现没有记录message_id无法定位链路日志加上message_id统一过滤任务偶尔完全失败某个步骤超时后续步骤拿到None加超时和重试错误信息写入context还有一点是提示词版本管理。我每次修改某个智能体的 system_prompt都会把这个prompt连同当时的测试用例一起存到一个prompts目录里。智能体优化本质上是prompt迭代没有版本管理你就不知道哪次改动导致效果回退而且这种回退在系统里很难通过代码测试发现。如果你用的是Git仓库建议把提示词也纳入版本管理。每次改动提示词都单独提一个分支跑一遍回归用例再合入。这比直接在生产环境里改提示词要稳得多。我后来能快速定位各种问题很大程度就是靠这层提示词版本库。最后说点个人体会。跑这个项目给我的最大收获不是“多智能体有多强”而是“限制会让系统更可靠”。你每允许智能体自由发挥一步就要准备一个兜底策略你每着急删掉一条校验规则后面就会用一次现场事故补回来。如果让我重新设计我还是会先把调度逻辑、输出Schema和日志体系做扎实再谈花哨的自主协作。这也算是我踩了不少坑之后的一条实用心得吧。
