1. 从单兵作战到团队协同为什么我们需要重新思考智能体架构做过智能体项目的人大概都有这种体会一开始写个单Agent的Demo感觉无所不能接上大模型、挂几个工具、写个提示词模板就能跑通一个像模像样的任务。但一旦把场景放大到真实业务里——比如需要同时处理需求分析、代码生成、测试验证、文档撰写这几条线——单Agent立刻捉襟见肘。上下文窗口不够用、角色切换混乱、任务依赖理不清最后要么是提示词越写越长变成一锅粥要么是拆成多个独立脚本但彼此之间完全没法协作。Agency Agents这套多智能体协作系统就是冲着这个痛点来的。它的核心思路很直接既然一个Agent搞不定复杂任务那就组一个“开发团队”让不同角色的Agent各司其职通过一套明确的协作协议把任务串起来。你可以把它理解成一个虚拟的软件工作室——里面有负责拆解需求的项目经理、有写代码的工程师、有做代码审查的质检员、还有负责整理交付物的文档专员。每个角色都是一个独立的智能体实例拥有自己的系统提示词、工具集和记忆空间它们之间通过消息传递和任务队列来协同工作。这套系统适合谁看如果你正在做以下任何一件事这篇内容应该能给你一些可以直接抄作业的思路一是手头有复杂任务需要多个大模型调用串联但用单Agent的链式调用已经维护不动了二是想了解多智能体系统的架构设计特别是任务分配、状态管理和冲突解决这几个硬骨头怎么啃三是已经在用某些智能体框架但觉得角色划分太粗或者协作机制不够灵活想自己搭一套更贴合业务的东西。我接下来会从整体架构设计讲到每个核心模块的实现细节再到实际落地时踩过的坑和排查技巧。内容会尽量贴着实操走参数怎么定、消息格式怎么设计、状态怎么同步这些都会给到具体的方案和代码片段。你不需要有很深的分布式系统背景但最好对Python异步编程和大模型API调用有基本了解这样读起来会更顺畅。2. 整体架构设计角色分工、通信机制与状态管理2.1 为什么是“Agency”而不是“Pipeline”很多人第一反应会把这个系统理解成一条流水线需求进来经过分析Agent、编码Agent、测试Agent最后输出结果。Pipeline模式确实简单但它在真实场景里有个致命问题——任务不是线性的。编码过程中发现需求有歧义需要回头找分析Agent确认测试不通过要把缺陷单打回给编码Agent文档Agent可能需要等所有代码冻结之后才能动笔。这些反馈回路和依赖关系用Pipeline根本表达不了。Agency这个词的核心在于“代理关系”和“自主决策”。每个Agent不是被动等待上游输入而是主动从共享的任务池里领取自己能处理的任务处理完之后把结果和后续任务写回池子。这种设计借鉴了分布式任务队列的思路但加上了角色权限和领域知识的约束。打个比方Pipeline像工厂流水线Agency像一群自由职业者在一个共享工作台上协作——谁有空谁接活但每个人只接自己擅长的活。具体到架构上整个系统分成四层角色定义层每个Agent的角色描述、能力边界、可用工具、系统提示词模板。这一层决定了“你是谁、你能干什么”。通信层基于消息队列的异步通信机制支持点对点消息和广播消息。消息格式统一为结构化JSON包含发送者、接收者、消息类型、负载和上下文引用。任务调度层负责任务的创建、分配、状态流转和依赖管理。核心是一个带优先级和依赖检查的任务队列。状态存储层保存每个Agent的私有记忆、共享的全局状态、以及任务执行的历史记录。这一层保证了Agent重启后能恢复上下文。2.2 角色划分的粒度怎么定角色划分太粗等于没分太细通信开销爆炸。我试过几种粒度最后稳定下来的方案是“按交付物划分”。什么意思每个Agent负责一种明确的交付物类型。比如角色交付物核心工具典型任务需求分析师结构化需求文档文本解析、知识库检索拆解用户故事、识别歧义架构设计师技术方案与接口定义代码分析、图表生成模块划分、接口设计编码工程师可运行代码代码生成、静态检查功能实现、单元测试编写代码审查员审查报告与修改建议差异分析、规则引擎代码规范检查、逻辑漏洞识别测试工程师测试用例与执行报告测试框架、覆盖率分析用例设计、回归测试文档专员用户文档与API文档文档生成、格式转换使用说明、接口文档这个划分的好处是每个角色的输出都是可验证的——需求文档可以评审代码可以跑测试审查报告可以逐条核对。而且角色之间的依赖关系天然清晰编码依赖架构设计测试依赖编码文档依赖所有前置产物。注意不要按“技能”划分角色比如“Python专家”“前端专家”。这种划分在任务分配时会产生大量歧义——一个任务到底该给谁按交付物划分就没这个问题因为交付物类型是唯一的。2.3 通信机制消息队列还是直接调用早期版本我用的是Agent之间直接函数调用A调用B的方法B返回结果。简单直接但很快遇到三个问题一是同步阻塞A等B的时候什么都干不了二是错误传播B抛异常直接把A也带崩三是无法追踪调用链一长根本不知道哪一步出了问题。后来换成基于消息队列的异步通信核心设计如下# 消息结构定义 class AgentMessage: def __init__(self, sender, receiver, msg_type, payload, context_refNone): self.msg_id generate_uuid() self.sender sender # 发送者角色ID self.receiver receiver # 接收者角色IDNone表示广播 self.msg_type msg_type # task_assign / task_result / query / response self.payload payload # 具体内容结构化字典 self.context_ref context_ref # 关联的任务ID或会话ID self.timestamp now()消息队列用Redis的Stream或者RabbitMQ都行我选Redis Stream的原因是部署简单、支持消费者组、消息持久化够用。每个Agent启动时注册自己的消费者组只消费发给自己的消息和广播消息。这里有个关键设计消息不直接触发Agent执行而是先进入Agent的待处理队列由Agent的主循环决定什么时候处理。这样做的好处是Agent可以根据自己的当前负载和任务优先级来调度而不是被消息推着走。比如编码Agent正在处理一个复杂函数生成这时候收到一条低优先级的文档更新通知它可以先处理完手头的再响应。2.4 状态管理私有记忆与共享黑板多智能体系统最头疼的就是状态一致性。A改了需求文档B还在用旧版本编码最后交付物对不上。我的方案是“私有记忆共享黑板”的混合模式。私有记忆是每个Agent自己的上下文窗口和长期存储保存它处理过的任务、生成的中间结果、以及从共享黑板读取的快照。私有记忆用向量数据库加结构化存储向量库存语义相关的历史片段结构化存储存任务ID和状态。共享黑板是一个全局的键值存储所有Agent都可以读写。黑板上放的是“事实性”信息当前需求版本号、接口定义、已完成的模块列表、已知缺陷列表。任何Agent修改黑板上的内容都要发一条广播消息通知其他Agent。其他Agent收到通知后可以选择刷新自己的本地快照。# 共享黑板操作示例 class SharedBlackboard: def __init__(self, redis_client): self.redis redis_client def update(self, key, value, version, agent_id): # 乐观锁检查版本号 current self.redis.hget(blackboard, key) if current and json.loads(current)[version] version: raise VersionConflict(黑板内容已被其他Agent更新) new_value {value: value, version: version 1, updated_by: agent_id} self.redis.hset(blackboard, key, json.dumps(new_value)) # 广播变更通知 self.publish_change_notification(key, version 1, agent_id)版本冲突的处理策略是“后写者重试”。如果编码Agent想更新接口定义但发现版本号落后了它需要先读取最新版本合并自己的修改再尝试写入。如果合并有冲突比如两个Agent改了同一个接口的不同部分就升级为人工介入或者由架构设计师Agent来仲裁。3. 核心模块实现任务调度、角色引擎与协作协议3.1 任务调度器带依赖检查的优先级队列任务调度器是整个系统的心脏。它要解决三个问题任务怎么创建、任务怎么分配、任务之间的依赖怎么处理。任务创建有两种方式一是外部触发比如用户提交一个需求系统自动生成一个“需求分析”任务二是Agent在处理任务过程中动态生成子任务比如编码Agent发现需要先定义接口就创建一个“接口设计”任务并设置依赖关系。任务的数据结构如下class Task: def __init__(self, task_id, task_type, payload, priority5, depends_onNone): self.task_id task_id self.task_type task_type # 对应角色类型 self.payload payload # 任务输入 self.priority priority # 1-10数字越小优先级越高 self.depends_on depends_on or [] # 依赖的任务ID列表 self.status pending # pending / assigned / running / done / failed self.assigned_to None self.created_at now() self.updated_at now()调度循环的逻辑是每隔一个心跳周期我设的是2秒扫描所有pending状态的任务检查依赖是否全部完成。如果依赖满足就根据任务类型找到对应的空闲Agent分配任务并更新状态为assigned。Agent处理完后把状态改为done或failed并触发下一轮调度。优先级的设计有个细节不要用绝对优先级用相对优先级加老化机制。什么意思如果高优先级任务一直插队低优先级任务可能永远得不到执行。我的做法是每个任务有一个基础优先级然后每等待一个调度周期优先级提升一级。这样即使是最低优先级的任务等待足够长时间后也能被处理。实操心得依赖检查不要用递归查询性能会很差。我的做法是维护一个反向依赖表——每个任务记录“我完成后会解锁哪些任务”。任务完成时直接查这个表把解锁的任务标记为可调度。这样依赖检查是O(1)的。3.2 角色引擎提示词模板与工具绑定每个Agent的角色引擎由三部分组成系统提示词、工具集、执行循环。系统提示词不是随便写一段“你是一个程序员”就完事了。我总结了一个模板结构包含五个部分角色身份你是谁你的专业背景是什么。职责边界你负责什么不负责什么。这一条特别重要防止Agent越界干别人的活。输入输出规范你接收什么格式的输入必须输出什么格式的结果。协作规则什么时候需要向其他Agent发消息什么时候需要更新共享黑板。质量准则你的输出需要满足哪些质量标准。举个例子编码工程师的系统提示词片段你是开发团队中的编码工程师专注于将技术方案转化为可运行的代码。 你的职责是编写功能代码和单元测试不负责架构决策和需求变更。 输入架构设计师提供的接口定义和模块划分方案。 输出符合PEP8规范的Python代码文件包含docstring和类型注解。 协作规则完成编码后向代码审查员发送审查请求如果发现接口定义有歧义向架构设计师发起查询。 质量准则代码必须通过静态检查单元测试覆盖率不低于80%。工具绑定方面每个角色只挂载自己需要的工具。编码工程师挂载代码生成、文件读写、静态检查工具测试工程师挂载测试框架、覆盖率分析工具需求分析师挂载文本解析和知识库检索工具。这样做的好处是减少工具描述对上下文窗口的占用同时降低Agent误用工具的概率。执行循环是一个标准的“感知-决策-行动”循环async def run_loop(self): while self.running: # 感知检查消息队列和任务队列 messages await self.fetch_messages() tasks await self.fetch_assigned_tasks() # 决策根据当前状态决定下一步 if tasks: task self.select_task(tasks) plan await self.plan(task) elif messages: msg self.select_message(messages) plan await self.plan_for_message(msg) else: await asyncio.sleep(self.idle_interval) continue # 行动执行计划 result await self.execute(plan) # 反馈更新状态和发送消息 await self.update_task_status(task, result) await self.notify_dependents(task)3.3 协作协议请求-响应与发布-订阅的混合模式Agent之间的协作分两种场景一种是需要对方回复的比如编码Agent问架构师“这个接口的参数类型是什么”另一种是不需要回复的比如测试Agent广播“发现一个严重缺陷”。请求-响应模式用点对点消息实现。发送方创建一个请求消息附带一个回调地址通常是发送方的消息队列接收方处理完后把结果发到回调地址。这里有个超时机制如果发送方在指定时间内没收到响应就重试或者降级处理。超时时间我设的是30秒重试次数2次。发布-订阅模式用广播消息实现。任何Agent都可以往一个主题topic发消息订阅了该主题的Agent都会收到。主题按领域划分比如“需求变更”“接口更新”“缺陷通知”。订阅关系在Agent启动时注册运行时可动态调整。# 协作协议示例编码Agent请求架构师澄清接口 async def request_clarification(self, architect_id, question): request_msg AgentMessage( senderself.agent_id, receiverarchitect_id, msg_typequery, payload{question: question, context: self.current_task_id}, context_refself.current_task_id ) await self.message_bus.send(request_msg) # 等待响应带超时 response await self.message_bus.wait_for_response( request_msg.msg_id, timeout30 ) if response is None: # 超时降级基于现有信息做合理假设并记录假设 self.log_assumption(question, 超时未响应基于常见实践假设) return self.make_reasonable_assumption(question) return response.payload[answer]注意请求-响应模式要避免循环等待。A等BB等CC又在等A整个系统就死锁了。我的做法是给每个请求消息设置一个“跳数”计数器每经过一个Agent跳数加一超过阈值我设的是5跳就强制返回当前最佳结果不再继续等待。4. 落地实操从零搭建一个多智能体开发团队4.1 环境准备与依赖安装先说一下基础环境。我用的Python 3.10以上主要是需要asyncio的完整支持。Redis 7.0以上用于消息队列和共享黑板。向量数据库我选的Chroma轻量够用如果数据量大可以换Milvus或者Qdrant。依赖清单如下pip install redis4.5.0 pip install chromadb0.4.0 pip install openai1.0.0 # 或其他大模型SDK pip install pydantic2.0 # 用于消息和任务的数据校验 pip install structlog # 结构化日志排查问题必备Redis的配置需要注意两点一是开启Stream的持久化防止重启丢消息二是设置合理的内存淘汰策略我用的allkeys-lru因为消息和黑板数据都是可重建的。# redis.conf 关键配置 appendonly yes appendfsync everysec maxmemory 2gb maxmemory-policy allkeys-lru4.2 定义你的第一个Agent角色假设我们要搭建一个最小可用的开发团队先从需求分析师和编码工程师两个角色开始。需求分析师的配置文件# roles/analyst.py ANALYST_ROLE { role_id: analyst, name: 需求分析师, system_prompt: 你是开发团队中的需求分析师。 你的职责是将用户需求拆解为结构化的用户故事和验收标准。 你不负责技术方案设计和代码实现。 输入自然语言描述的需求。 输出JSON格式的用户故事列表每个故事包含id、描述、验收标准、优先级。 协作规则完成分析后创建架构设计任务并分配给架构设计师。 质量准则每个用户故事必须可独立测试验收标准必须具体可验证。, tools: [text_parser, knowledge_base_search], max_context_tokens: 8000, temperature: 0.3 }编码工程师的配置文件# roles/coder.py CODER_ROLE { role_id: coder, name: 编码工程师, system_prompt: 你是开发团队中的编码工程师。 你的职责是根据技术方案编写功能代码和单元测试。 你不负责需求变更和架构决策。 输入架构设计师提供的接口定义和模块划分。 输出Python代码文件包含docstring、类型注解和单元测试。 协作规则完成后向代码审查员发送审查请求发现接口歧义时向架构设计师查询。 质量准则代码通过pylint检查单元测试覆盖率不低于80%。, tools: [code_generator, file_io, static_checker], max_context_tokens: 16000, temperature: 0.2 }温度参数的选择有讲究。需求分析和编码这种需要精确输出的角色温度设低一点0.2-0.3减少随机性。如果是创意类角色比如文档专员可以设到0.5-0.7让输出更自然。4.3 启动调度器与Agent实例调度器和Agent是分开的进程通过Redis通信。这样做的好处是Agent可以独立重启和扩容调度器不会成为单点。调度器启动代码# scheduler.py import asyncio from task_queue import TaskQueue from blackboard import SharedBlackboard async def main(): redis_client await create_redis_client() task_queue TaskQueue(redis_client) blackboard SharedBlackboard(redis_client) # 注册角色与任务类型的映射 role_task_mapping { analyst: [requirement_analysis], architect: [architecture_design], coder: [code_implementation], reviewer: [code_review], tester: [test_design, test_execution], writer: [doc_writing] } scheduler Scheduler(task_queue, blackboard, role_task_mapping) await scheduler.run() if __name__ __main__: asyncio.run(main())Agent启动代码# agent_runner.py import asyncio from agent import Agent from roles.coder import CODER_ROLE async def main(): agent Agent( role_configCODER_ROLE, redis_urlredis://localhost:6379, llm_config{model: gpt-4, api_key: your-key} ) await agent.start() await agent.run_loop() if __name__ __main__: asyncio.run(main())启动顺序是先起Redis再起调度器最后起各个Agent。Agent启动后会向调度器注册自己的角色ID和消费者组调度器收到注册后开始给它分配任务。4.4 一个完整任务的流转实录假设用户提交了一个需求“做一个待办事项管理工具支持增删改查和优先级排序。”第一步需求分析。调度器创建requirement_analysis任务分配给analyst。analyst调用大模型拆解需求输出如下{ user_stories: [ {id: US-001, desc: 用户可以创建待办事项, criteria: 输入标题和优先级保存后出现在列表中, priority: 1}, {id: US-002, desc: 用户可以删除待办事项, criteria: 点击删除按钮事项从列表移除, priority: 2}, {id: US-003, desc: 用户可以修改待办事项, criteria: 点击编辑修改后保存生效, priority: 2}, {id: US-004, desc: 用户可以按优先级排序, criteria: 列表按优先级从高到低排列, priority: 3} ] }analyst把结果写入共享黑板并创建architecture_design任务依赖requirement_analysis完成。第二步架构设计。architect读取黑板上的用户故事设计数据模型和接口{ data_model: { TodoItem: { id: string, title: string, priority: int, created_at: datetime } }, interfaces: [ {name: create_todo, input: title, priority, output: TodoItem}, {name: delete_todo, input: id, output: bool}, {name: update_todo, input: id, title, priority, output: TodoItem}, {name: list_todos, input: sort_by, output: List[TodoItem]} ] }architect创建code_implementation任务分配给coder。第三步编码实现。coder根据接口定义生成代码# todo_service.py from dataclasses import dataclass from datetime import datetime from typing import List, Optional dataclass class TodoItem: id: str title: str priority: int created_at: datetime class TodoService: def __init__(self): self._items: dict[str, TodoItem] {} def create_todo(self, title: str, priority: int) - TodoItem: item_id generate_id() item TodoItem(iditem_id, titletitle, prioritypriority, created_atdatetime.now()) self._items[item_id] item return item def delete_todo(self, item_id: str) - bool: return self._items.pop(item_id, None) is not None def update_todo(self, item_id: str, title: str None, priority: int None) - Optional[TodoItem]: item self._items.get(item_id) if not item: return None if title is not None: item.title title if priority is not None: item.priority priority return item def list_todos(self, sort_by: str priority) - List[TodoItem]: items list(self._items.values()) if sort_by priority: items.sort(keylambda x: x.priority) return itemscoder同时生成单元测试然后创建code_review任务。第四步代码审查。reviewer检查代码规范、逻辑漏洞、边界条件。发现update_todo方法没有处理空字符串标题的情况生成审查意见并打回给coder。coder修改后重新提交reviewer通过。第五步测试执行。tester设计测试用例并执行覆盖增删改查和排序场景。全部通过后创建doc_writing任务。第六步文档撰写。writer根据代码和测试报告生成用户文档和API文档。整个流程走下来从需求到交付物大概需要15-20次大模型调用耗时取决于模型响应速度。我实测下来用中等规模的模型比如GPT-3.5级别整个流程在3-5分钟内完成。5. 常见问题与排查技巧实录5.1 Agent卡死或无限循环这是最常见的问题。表现是某个Agent一直在处理同一个任务或者两个Agent互相发消息停不下来。排查思路分三步先看日志确认是哪个Agent在循环再看消息队列确认是不是有消息被重复消费最后看任务状态确认是不是依赖检查出了问题。我遇到过的几个具体原因消息重复消费Redis Stream的消费者组在Agent重启后没有正确确认消息导致消息被重新投递。解决方法是Agent启动时先检查未确认的消息处理完再确认。依赖循环A任务依赖BB任务依赖A。调度器检测到循环依赖后应该直接标记为失败而不是无限等待。我在调度器里加了一个依赖图检测发现环就报警。提示词导致的行为循环Agent的系统提示词里写了“如果不确定就询问架构师”结果架构师也不确定又反问回来。解决方法是在提示词里加一条“如果对方也无法确定基于常见实践做合理假设并记录”。实操心得给每个Agent加一个“最大连续执行次数”限制比如连续处理20个任务后强制进入冷却期等待人工检查。这个阈值根据任务复杂度调整我一般设15-25。5.2 共享黑板数据不一致表现是不同Agent读到的黑板数据版本不同导致基于旧数据做决策。根本原因是读写没有原子性。我的解决方案是引入版本号和乐观锁前面已经提过。但还有一个隐蔽的问题Agent的本地缓存没有及时失效。Agent为了提高性能会把黑板数据缓存在本地内存里但广播通知可能因为网络延迟晚到。解决方法是给缓存设置一个较短的TTL我设的是5秒同时收到广播通知时立即失效对应缓存。另外关键决策前强制从黑板重新读取最新版本不依赖缓存。5.3 大模型输出格式不符合预期Agent之间的通信依赖结构化数据但大模型有时候会输出多余的解释文字或者格式错误的JSON。我的处理策略是三层防护第一层在提示词里明确要求“只输出JSON不要任何额外文字”第二层用Pydantic做输出校验格式不对直接抛异常第三层异常时触发重试重试时在提示词里加上“上次输出格式错误请严格按照JSON格式输出”。如果重试三次还是失败就降级为人工处理把原始输出和错误信息记录到日志里。from pydantic import BaseModel, ValidationError class UserStory(BaseModel): id: str desc: str criteria: str priority: int def parse_llm_output(raw_output: str) - list[UserStory]: # 尝试提取JSON部分 json_str extract_json(raw_output) try: data json.loads(json_str) return [UserStory(**item) for item in data[user_stories]] except (json.JSONDecodeError, ValidationError) as e: raise OutputFormatError(f输出格式错误: {e}, 原始输出: {raw_output[:200]})5.4 任务分配不均有的Agent忙死有的Agent闲死。原因是任务类型分布不均匀比如编码任务特别多但审查任务很少。解决方案有两个一是动态调整Agent数量给繁忙的角色启动多个实例共享同一个消费者组二是任务窃取空闲的Agent可以“借调”去处理其他类型的任务但需要临时加载对应的角色提示词和工具集。我一般先用方案一因为实现简单。方案二适合任务类型差异大但总量不大的场景。问题现象可能原因排查方法解决方案Agent卡死消息重复消费检查Redis未确认消息启动时清理未确认消息无限循环依赖成环检查任务依赖图调度器加环检测数据不一致缓存未失效对比本地缓存与黑板版本缩短TTL广播失效格式错误提示词不明确查看原始输出三层防护重试分配不均任务类型倾斜统计各角色任务量动态扩容或任务窃取5.5 性能优化减少不必要的模型调用多智能体系统跑起来之后大模型调用费用是主要成本。我做过统计一个中等复杂度的开发任务如果不做优化大概需要30-40次模型调用。优化后可以降到15-20次。优化手段包括缓存常见查询比如“Python中如何读取JSON文件”这种问题第一次调用后把结果缓存起来后续直接返回。批量处理把多个小任务合并成一个提示词一次调用完成。比如代码审查时把规范检查、逻辑检查、边界检查合并成一个提示词。规则引擎前置能用规则判断的不要用模型。比如代码格式检查直接用pylint跑一遍只把pylint无法判断的问题交给模型。模型分级简单任务用便宜的小模型复杂任务用大模型。比如需求拆解用大模型格式转换用小模型。注意缓存要设置合理的过期时间并且要区分上下文。同一个问题在不同项目背景下答案可能不同缓存键要包含项目ID和上下文摘要。6. 扩展方向与个人实践体会这套系统跑通之后我陆续做了一些扩展有些效果不错有些踩了坑。第一个扩展是引入人类审批节点。在关键决策点比如架构设计完成、代码审查通过插入人工确认步骤。实现方式是在任务状态里加一个awaiting_human状态调度器遇到这个状态就暂停后续任务等人工在管理界面点击确认后再继续。这个功能在真实项目里非常必要因为大模型有时候会做出看似合理但实际有问题的决策。第二个扩展是跨项目知识复用。把历史项目的需求文档、架构方案、代码片段存入向量数据库新项目启动时先检索相似项目作为参考。这个功能对提升输出质量帮助很大但要注意隐私隔离——不同项目的知识不能混在一起。第三个扩展是Agent自我评估。每个Agent完成任务后对自己的输出做一个置信度评分。低置信度的任务自动触发人工复核。置信度评分用一个小模型来做成本很低但效果不错。踩过的坑也有几个。一个是过度设计一开始想给每个Agent加复杂的记忆机制和反思循环结果系统变得极其难调试。后来砍掉了大部分花哨功能只保留最核心的任务队列和消息通信反而稳定了很多。另一个是忽视日志早期没有结构化日志出问题只能靠猜。后来用structlog把每个Agent的每个决策都记录下来排查效率提升了一个数量级。我个人在实际操作中的体会是多智能体系统的核心难点不在模型能力而在工程实现。任务怎么拆、消息怎么传、状态怎么同步、错误怎么恢复这些才是决定系统能不能跑起来的关键。模型选型反而没那么重要中等能力的模型配合好的工程架构效果远超大模型配烂架构。最后分享一个小技巧先用单Agent跑通全流程再拆成多Agent。很多人一上来就设计复杂的多Agent协作结果连基本流程都没跑通。我的做法是先写一个“全能Agent”把所有角色的提示词和工具都塞进去跑通一个完整任务。然后观察哪些环节是瓶颈再把瓶颈环节拆成独立Agent。这样拆出来的多Agent系统每个角色的边界都是经过验证的不会出现“为了拆而拆”的情况。