1. 从单兵作战到团队协作为什么要把 Agent 组织起来很多人第一次接触 Agent 这个概念都是从一个简单的场景开始的装好一个框架配好 API Key写一段提示词然后看着它在终端里回答问题、调用工具、完成一个小任务。这个阶段很兴奋但也很短暂。因为很快你就会发现单个 Agent 的能力边界非常明显——它记不住太长的上下文它没法同时处理多个方向的任务它在遇到复杂问题时容易陷入死循环它甚至会在同一个错误上反复撞墙。我自己最开始也是这样装了三四个不同的 Agent 框架每个都跑通了 demo每个都能完成一些基础任务但真正想把它们用起来解决实际问题的时候就发现单打独斗的效率极低。比如我让一个 Agent 帮我整理一份技术调研报告它会先去搜索然后整理然后写摘要整个过程串行执行中间任何一步卡住后面全部停摆。更麻烦的是当任务涉及多个领域时一个 Agent 既要懂代码又要懂文案还要懂数据分析提示词写得再长也覆盖不全。这就是“把 Agent 组织起来干活”这个想法诞生的背景。不是让一个 Agent 变得更强而是让多个各有所长的 Agent 像一个小团队一样协作。有的负责规划有的负责执行有的负责检查有的负责汇总。每个 Agent 的职责单一提示词可以写得很精炼工具配置也可以很有针对性。整体上这个系统的鲁棒性和吞吐量都比单 Agent 方案高出一个量级。这篇文章主要面向已经装好至少一个 Agent 框架、跑通过基础对话和工具调用的读者。如果你还没到这一步建议先把单个 Agent 的安装、配置、工具调用跑通再来看多 Agent 协作的内容。因为多 Agent 系统的调试复杂度是单 Agent 的好几倍基础不牢会非常痛苦。我会从整体设计思路开始讲然后拆解核心细节接着给出完整的实操过程最后分享我在排查问题时积累的经验。整个过程中涉及的工具选型、参数配置、代码结构都会给出具体的理由和可复现的方案。2. 多 Agent 协作的整体设计与思路拆解2.1 为什么选择“角色分工”而不是“模型堆叠”很多人想到多 Agent第一反应是搞一个超级模型或者把多个模型的输出简单拼接。这两种思路我都试过效果都不理想。超级模型的问题在于成本和延迟而且单一模型再强也有知识盲区。简单拼接的问题在于没有真正的协作只是把多个回答放在一起信息冗余严重还容易互相矛盾。真正有效的多 Agent 系统核心在于角色分工。每个 Agent 有明确的职责边界、独立的系统提示词、专属的工具集以及清晰的输入输出规范。这就像一个小型创业团队有人负责产品规划有人负责技术实现有人负责质量检查有人负责对外沟通。每个人不需要是全才但组合起来能覆盖整个业务流程。我目前使用的架构是四层结构规划层、执行层、检查层、汇总层。规划层负责把用户需求拆解成可执行的任务列表执行层根据任务类型调用不同的专业 Agent检查层对执行结果做质量把关汇总层把多个执行结果整合成最终输出。这个结构不是固定的你可以根据任务复杂度增减层级但核心思想是“分工明确、接口清晰”。2.2 通信机制的选择消息队列还是共享状态多 Agent 系统最关键的工程决策之一就是 Agent 之间怎么通信。我试过三种方案直接函数调用、消息队列、共享状态。直接函数调用最简单Agent A 直接调用 Agent B 的函数同步等待结果。这种方式适合简单的串行流程但一旦某个 Agent 执行时间过长整个系统就会阻塞。而且错误处理很麻烦一个环节抛异常整个调用链就断了。消息队列方案更灵活每个 Agent 监听自己的队列收到消息就处理处理完把结果发到下一个队列。这种方式解耦彻底可以异步执行也方便做重试和监控。但引入消息队列意味着额外的运维成本对于单机环境来说有点重。共享状态方案是我目前最推荐的。所有 Agent 共享一个状态存储可以用内存字典、Redis、或者简单的 JSON 文件每个 Agent 从状态中读取自己需要的输入处理完后把输出写回状态。这种方式既解耦又轻量调试的时候可以直接查看状态快照非常直观。我现在的做法是用一个 Python 字典作为共享状态配合文件持久化。每个 Agent 是一个独立的函数或类接收状态对象返回更新后的状态。调度器负责按顺序或按条件调用各个 Agent。这种设计在单机环境下足够用迁移到分布式环境时也容易改造成消息队列模式。2.3 工具集的设计原则最小权限与最大复用每个 Agent 的工具集需要仔细设计。我踩过的坑是给每个 Agent 配了太多工具结果它不知道该用哪个经常调用错误的工具或者在一个简单任务上反复尝试不同的工具。后来我总结出一个原则每个 Agent 的工具集应该刚好覆盖它的职责范围不多不少。比如规划 Agent 只需要一个“任务拆解”工具和一个“优先级排序”工具不需要搜索、不需要代码执行、不需要文件读写。执行 Agent 根据具体类型配置工具代码执行 Agent 只配代码运行和文件读写搜索 Agent 只配搜索和网页抓取。另一个原则是工具复用。不要把同一个功能在不同 Agent 里重复实现而是抽成公共工具库每个 Agent 按需引入。这样维护成本低行为也一致。我现在的做法是建一个 tools 目录里面按功能分类存放工具函数每个 Agent 的配置文件里声明它需要哪些工具。2.4 调度策略串行、并行还是混合调度策略直接影响系统的吞吐量和响应时间。我试过全串行、全并行、以及混合模式。全串行最简单一个 Agent 执行完再执行下一个。适合有严格依赖关系的任务链但效率低。全并行适合独立任务比如同时搜索多个关键词但任务之间有依赖时就会出问题。混合模式是我目前的主力方案。规划层和执行层之间是串行的因为必须先有规划才能执行。执行层内部如果多个任务之间没有依赖关系就并行执行。检查层和汇总层是串行的因为需要等所有执行结果都出来才能检查。具体实现上我用 Python 的 concurrent.futures 线程池来管理并行任务。每个执行 Agent 是一个独立的线程主调度器等待所有线程完成后继续。这里要注意线程安全共享状态的读写需要加锁否则会出现数据竞争。3. 核心细节解析与实操要点3.1 Agent 角色定义与提示词工程每个 Agent 的角色定义决定了它的行为边界。我通常从三个维度来定义职责描述、输入输出规范、行为约束。职责描述要具体不能太宽泛。比如“你是一个规划助手”这种定义太模糊Agent 不知道自己该做什么。更好的定义是“你负责把用户需求拆解成 3 到 7 个可独立执行的子任务每个子任务必须包含明确的输入、输出和验收标准”。输入输出规范要明确格式。我一般要求 Agent 输出 JSON 格式包含固定的字段。比如规划 Agent 的输出格式是{ tasks: [ { id: task_1, type: search, description: 搜索 XXX 相关信息, input: {query: XXX}, dependencies: [] } ] }行为约束是防止 Agent 越界的关键。比如执行 Agent 不能修改任务列表检查 Agent 不能直接修改执行结果只能标记问题。这些约束要写在系统提示词里并且在代码层面做校验。提示词工程方面我建议每个 Agent 的提示词控制在 500 到 1000 字之间。太短了约束不够太长了模型容易忽略重点。关键约束要放在提示词的开头和结尾中间放具体说明。我习惯用“必须”、“禁止”、“只能”这类强约束词效果比“建议”、“尽量”好很多。3.2 共享状态的结构设计共享状态是整个系统的中枢设计好坏直接影响调试难度和扩展性。我目前用的结构包含四个部分任务元数据、执行上下文、中间结果、最终输出。任务元数据记录任务的原始需求、创建时间、当前状态。执行上下文记录每个 Agent 的执行状态、开始时间、结束时间、错误信息。中间结果按任务 ID 存储每个子任务的输出。最终输出是汇总后的结果。shared_state { meta: { task_id: xxx, user_request: 原始需求, status: running, created_at: 2024-01-01T00:00:00 }, context: { planner: {status: done, duration: 1.2}, executor_1: {status: running, duration: 3.5} }, intermediate: { task_1: {result: ..., status: done}, task_2: {result: ..., status: failed, error: ...} }, final: { output: ..., quality_score: 0.85 } }这个结构的好处是信息分层清晰调试的时候可以快速定位问题。比如某个任务失败了直接看 intermediate 里对应任务的状态和错误信息。某个 Agent 执行时间过长看 context 里的 duration。状态持久化我建议每个阶段结束后写一次文件格式用 JSON。这样即使程序崩溃也能从上次的状态恢复。文件命名带上时间戳和任务 ID方便追溯。3.3 错误处理与重试机制多 Agent 系统的错误处理比单 Agent 复杂得多因为错误可能发生在任何一层而且可能级联传播。我的策略是分层处理Agent 内部错误自己处理Agent 之间错误由调度器处理系统级错误由监控模块处理。Agent 内部错误包括 API 调用失败、工具执行异常、输出格式错误。这些错误在 Agent 内部捕获根据错误类型决定是重试还是返回失败。API 调用失败通常重试两到三次每次间隔递增。工具执行异常要看具体类型如果是参数错误就直接返回失败如果是临时故障就重试。输出格式错误最麻烦我一般让 Agent 重新生成一次如果还是格式错误就标记为失败。Agent 之间的错误由调度器处理。调度器检查每个 Agent 的返回状态如果失败根据任务依赖关系决定是否继续。如果失败的任务是其他任务的前置依赖就跳过依赖它的任务并在最终输出里说明。如果失败的任务是独立的就继续执行其他任务。重试机制我建议设置最大重试次数和超时时间。最大重试次数一般设 3 次超时时间根据任务类型定搜索类任务 30 秒代码执行类任务 60 秒汇总类任务 120 秒。超过限制就标记为失败不再重试。3.4 日志与可观测性设计多 Agent 系统的调试难度很高没有良好的日志几乎没法排查问题。我的日志设计包含三个层次Agent 级日志、调度级日志、系统级日志。Agent 级日志记录每个 Agent 的输入、输出、执行时间、错误信息。格式用结构化日志方便后续分析。我一般用 Python 的 logging 模块配置 JSON 格式的输出。调度级日志记录任务的分发、依赖关系的处理、并行任务的启动和完成。这些日志帮助理解系统的整体执行流程。系统级日志记录资源使用情况、API 调用次数、错误率等指标。这些日志用于监控系统健康度发现性能瓶颈。日志存储我建议按天分文件保留最近 7 天。关键错误日志单独存一份方便快速检索。如果条件允许可以接入简单的日志分析工具做错误聚合和趋势分析。4. 实操过程与核心环节实现4.1 环境准备与依赖安装我假设你已经有一个可用的 Python 环境版本 3.9 以上。先创建一个虚拟环境避免依赖冲突。python -m venv agent_env source agent_env/bin/activate # Linux/Mac # 或者 agent_env\Scripts\activate # Windows安装核心依赖。我用的框架是轻量级的主要依赖 requests 做 API 调用concurrent.futures 做并行调度pydantic 做数据校验。pip install requests pydantic python-dotenv如果你用的 Agent 框架有特定依赖按框架文档安装。我建议把依赖写进 requirements.txt方便复现。pip freeze requirements.txt环境变量配置用 .env 文件存放 API Key、模型名称、超时时间等。不要硬编码在代码里方便切换环境。API_KEYyour_key_here MODEL_NAMEyour_model MAX_RETRIES3 TIMEOUT604.2 基础 Agent 类的实现先定义一个基础 Agent 类封装通用的 API 调用、重试、日志逻辑。所有具体 Agent 继承这个类。import os import json import time import logging from abc import ABC, abstractmethod from typing import Dict, Any class BaseAgent(ABC): def __init__(self, name: str, config: Dict[str, Any]): self.name name self.config config self.logger logging.getLogger(fagent.{name}) self.max_retries int(os.getenv(MAX_RETRIES, 3)) self.timeout int(os.getenv(TIMEOUT, 60)) abstractmethod def build_prompt(self, state: Dict[str, Any]) - str: pass abstractmethod def parse_response(self, response: str) - Dict[str, Any]: pass def call_model(self, prompt: str) - str: for attempt in range(self.max_retries): try: # 这里替换成你实际使用的模型调用方式 response self._invoke_api(prompt) return response except Exception as e: self.logger.warning(fAttempt {attempt1} failed: {e}) if attempt self.max_retries - 1: raise time.sleep(2 ** attempt) return def _invoke_api(self, prompt: str) - str: # 具体 API 调用逻辑根据你的框架实现 pass def run(self, state: Dict[str, Any]) - Dict[str, Any]: start_time time.time() self.logger.info(f{self.name} started) try: prompt self.build_prompt(state) response self.call_model(prompt) result self.parse_response(response) state[context][self.name] { status: done, duration: time.time() - start_time } return result except Exception as e: self.logger.error(f{self.name} failed: {e}) state[context][self.name] { status: failed, duration: time.time() - start_time, error: str(e) } raise这个基类封装了重试、日志、状态更新。具体 Agent 只需要实现 build_prompt 和 parse_response 两个方法。4.3 规划 Agent 的实现规划 Agent 负责把用户需求拆解成任务列表。提示词要强调任务的可执行性和依赖关系。class PlannerAgent(BaseAgent): def build_prompt(self, state: Dict[str, Any]) - str: user_request state[meta][user_request] return f你是一个任务规划专家。请把以下需求拆解成 3 到 7 个可独立执行的子任务。 需求{user_request} 输出格式必须是 JSON包含 tasks 数组每个任务包含 - id: 任务唯一标识 - type: 任务类型search/code/analysis/summary - description: 任务描述 - input: 任务输入参数 - dependencies: 依赖的任务 id 列表 只输出 JSON不要有其他内容。 def parse_response(self, response: str) - Dict[str, Any]: # 清理可能的 markdown 代码块标记 cleaned response.strip().strip(json).strip().strip() data json.loads(cleaned) if tasks not in data: raise ValueError(Missing tasks field) return data这里要注意模型输出经常带有 markdown 代码块标记解析前要清理。另外要校验 tasks 字段是否存在避免后续流程出错。4.4 执行 Agent 的实现执行 Agent 根据任务类型有不同的实现。我以搜索 Agent 为例。class SearchAgent(BaseAgent): def build_prompt(self, state: Dict[str, Any]) - str: task state[current_task] return f你是一个信息搜索专家。请根据以下任务描述整理出关键信息。 任务{task[description]} 输入参数{json.dumps(task[input], ensure_asciiFalse)} 输出格式为 JSON包含 summary 和 sources 两个字段。 summary 是信息摘要sources 是来源列表。 def parse_response(self, response: str) - Dict[str, Any]: cleaned response.strip().strip(json).strip().strip() return json.loads(cleaned)代码执行 Agent 类似但需要配置文件读写和代码运行工具。这里要注意安全代码执行要在沙箱环境里限制文件访问范围。4.5 调度器的实现调度器负责按依赖关系执行任务管理并行和串行。import concurrent.futures class Orchestrator: def __init__(self, agents: Dict[str, BaseAgent]): self.agents agents self.logger logging.getLogger(orchestrator) def execute(self, state: Dict[str, Any]) - Dict[str, Any]: tasks state[intermediate][tasks] completed set() failed set() while len(completed) len(failed) len(tasks): ready [ t for t in tasks if t[id] not in completed and t[id] not in failed and all(d in completed for d in t[dependencies]) ] if not ready: # 没有可执行的任务说明有循环依赖或前置失败 break with concurrent.futures.ThreadPoolExecutor(max_workers4) as executor: futures {} for task in ready: agent self.agents[task[type]] state[current_task] task future executor.submit(agent.run, state) futures[future] task for future in concurrent.futures.as_completed(futures): task futures[future] try: result future.result() state[intermediate][task[id]] { result: result, status: done } completed.add(task[id]) except Exception as e: state[intermediate][task[id]] { status: failed, error: str(e) } failed.add(task[id]) return state这个调度器每次找出所有依赖已满足的任务并行执行。执行完后更新状态继续下一轮。注意这里用了线程池共享状态的读写需要加锁实际使用时要注意线程安全。4.6 汇总 Agent 的实现汇总 Agent 把所有执行结果整合成最终输出。class SummaryAgent(BaseAgent): def build_prompt(self, state: Dict[str, Any]) - str: results state[intermediate] results_text json.dumps(results, ensure_asciiFalse, indent2) return f你是一个报告汇总专家。请把以下多个子任务的结果整合成一份完整的报告。 子任务结果 {results_text} 要求 1. 按逻辑顺序组织内容 2. 标注每个结论的来源任务 3. 如果有任务失败在报告中说明 4. 输出格式为 Markdown def parse_response(self, response: str) - Dict[str, Any]: return {output: response}汇总 Agent 的提示词要强调逻辑组织和来源标注这样最终报告的可信度更高。4.7 完整流程串联把所有组件串起来主流程如下def main(user_request: str): state { meta: { task_id: ftask_{int(time.time())}, user_request: user_request, status: running }, context: {}, intermediate: {}, final: {} } # 初始化 Agent agents { planner: PlannerAgent(planner, {}), search: SearchAgent(search, {}), code: CodeAgent(code, {}), summary: SummaryAgent(summary, {}) } # 规划 planner agents[planner] plan_result planner.run(state) state[intermediate][tasks] plan_result[tasks] # 执行 orchestrator Orchestrator(agents) state orchestrator.execute(state) # 汇总 summary agents[summary] final_result summary.run(state) state[final] final_result state[meta][status] done # 持久化 with open(fstate_{state[meta][task_id]}.json, w) as f: json.dump(state, f, ensure_asciiFalse, indent2) return state[final][output]这个流程跑通后你就有了一个基础的多 Agent 协作系统。后续可以根据需要增加更多类型的 Agent比如检查 Agent、优化 Agent、翻译 Agent 等。5. 常见问题与排查技巧实录5.1 Agent 输出格式错误怎么办这是最常见的问题。模型有时候会忘记输出 JSON或者输出带 markdown 标记的 JSON或者字段名拼错。我的处理策略是三层防护。第一层是提示词约束明确要求只输出 JSON不要有其他内容。第二层是解析前清理去掉可能的代码块标记和多余空白。第三层是解析失败后重试让模型重新生成一次。如果重试后还是格式错误就标记该任务失败继续执行其他任务。不要让一个格式错误阻塞整个流程。我踩过的坑是提示词里写了“输出 JSON”但模型还是输出了带解释文字的 JSON。后来我在提示词里加了“只输出 JSON不要有任何解释文字”情况好了很多。另外用 pydantic 做字段校验能提前发现字段缺失或类型错误。5.2 任务依赖关系处理错误依赖关系处理错误通常表现为死循环或者任务被跳过。死循环的原因是依赖关系有环比如 A 依赖 BB 依赖 A。任务被跳过是因为前置任务失败但调度器没有正确处理。我的解决方案是在规划阶段就检测依赖关系是否有环。用拓扑排序算法检查如果有环就让规划 Agent 重新生成。调度器里也要加超时机制如果一轮调度没有产生任何完成的任务就跳出循环标记剩余任务为失败。另一个技巧是在任务 ID 里加入层级信息比如 task_1_1 表示第一个任务的第一个子任务。这样依赖关系更清晰也方便调试。5.3 并行执行时的资源竞争并行执行时多个 Agent 同时读写共享状态容易出现数据竞争。表现是状态数据不一致或者某些任务的输出丢失。我的解决方案是给共享状态的读写加锁。Python 的 threading.Lock 可以解决这个问题。但要注意锁的粒度太粗会影响并行效率太细容易死锁。我现在的做法是每个任务的结果单独存储不共享可变对象。调度器收集所有结果后再统一更新状态。这样避免了大部分竞争问题。另外要注意 API 调用的速率限制。并行调用太多会导致 API 返回 429 错误。我一般限制最大并行数为 4并且在 API 调用失败时做指数退避重试。5.4 常见问题速查表问题现象可能原因排查方法解决方案Agent 输出格式错误提示词约束不够查看原始输出加强提示词约束增加解析前清理任务死循环依赖关系有环检查依赖图拓扑排序检测重新规划任务被跳过前置任务失败查看失败任务日志调度器正确处理失败传播状态数据不一致并行读写竞争检查状态快照加锁或分离存储API 调用频繁失败速率限制查看错误码限制并行数指数退避执行时间过长任务粒度过大查看各任务耗时拆分任务设置超时最终输出质量差汇总提示词不够查看汇总输入优化汇总提示词增加检查层5.5 独家避坑技巧第一个技巧是给每个 Agent 的输出加一个“置信度”字段。模型自己评估对结果的把握程度汇总的时候可以根据置信度加权。这个技巧在搜索结果质量参差不齐的时候特别有用。第二个技巧是在规划阶段就让 Agent 预估每个任务的执行时间调度器根据预估时间动态调整并行数。预估时间长的任务优先启动避免最后卡在某个慢任务上。第三个技巧是保留每次执行的完整状态快照包括输入、输出、中间状态。这样出问题的时候可以精确复现而不是靠猜。我一般保留最近 10 次执行的快照超过就自动清理。第四个技巧是给 Agent 设置“最大思考轮数”。有些 Agent 会在一个问题上反复思考消耗大量 token。设置一个上限超过就强制输出当前最佳结果。第五个技巧是用小模型做规划大模型做执行。规划任务相对简单小模型足够用而且速度快、成本低。执行任务复杂需要大模型的能力。这样整体成本和延迟都能优化。5.6 性能优化的几个方向如果系统跑起来感觉慢可以从几个方向优化。首先是减少不必要的 Agent 调用有些简单任务不需要经过规划层直接执行就行。其次是缓存重复的搜索结果同样的查询不要重复调用 API。第三是并行化独立任务这个前面已经讲了。第四是压缩提示词去掉冗余的描述减少 token 消耗。第五是选择更快的模型做简单任务复杂任务再用大模型。我实测下来经过这些优化整体执行时间能减少 40% 左右。特别是缓存和并行化效果最明显。5.7 扩展方向这个基础框架跑通后可以往几个方向扩展。一是增加更多类型的 Agent比如翻译 Agent、格式化 Agent、审核 Agent。二是引入更复杂的调度策略比如基于优先级的调度、基于资源可用性的调度。三是接入外部工具比如数据库查询、API 调用、文件处理。四是做可视化界面实时展示每个 Agent 的状态和输出。我个人觉得最有价值的扩展方向是增加一个“学习”机制。每次执行完后把成功的任务拆解和提示词保存下来下次遇到类似任务时直接复用。这样系统会越用越聪明而不是每次从零开始。这个内容后续还可以这样扩展把单机多 Agent 系统改造成分布式系统每个 Agent 跑在不同的机器上通过消息队列通信。这样能处理更大规模的任务也能提高系统的容错性。不过分布式系统的复杂度会高很多建议先把单机版本跑稳再说。