个人AI助手双Agent架构实战:规划与执行分离的完整方案

发布时间:2026/10/12 13:20:07
个人AI助手双Agent架构实战:规划与执行分离的完整方案
做到个人AI助手的第6阶段我最大的感触是这东西开始从“玩具”往“同事”的方向过渡了。前5个阶段我一直在搞定单 Agent 的模型接入、工具调用、记忆管理和任务状态机能聊天、能搜索、能写代码可一旦让同一个 Agent 连续处理“查资料→整理→写报告”这种多步骤任务它就开始频繁跑偏经常是查着查着忘了要干什么写报告写到一半又去补调研。阶段6我做的核心改造就是双 Agent 系统开发把“规划”和“执行”拆成两个 Agent一个负责想一个负责干。这篇文章记录我的完整设计、关键代码、参数调优和踩坑记录希望能给同样在折腾个人AI助手的朋友一份可以直接上手的参考。开头先说结论双 Agent 并不是简单地把两个模型接在一起而是要在角色职责、通信协议、上下文隔离和错误恢复这四个层面做设计。否则你得到的不是两个协同工作的助手而是两个互相扯皮的聊天机器人。我这套方案不是凭空拍脑袋想出来的而是在单 Agent 连续失败了很多次之后被逼着拆分出来的。下面你会看到我踩过的具体问题以及每个设计选择背后的理由。1. 为什么要做双 Agent从单兵作战到团队协作1.1 单 Agent 助手为什么越用越呆单 Agent 的核心问题不是“能力不够”而是“角色冲突”。同一个模型既要在对话里扮演一个善解人意的助手又要在执行任务时做严谨的任务拆解还要在调用工具时当一把稳定的调度器这几个角色的思维方式差异很大。我实际测试过程中最典型的现象有三个第一个是目标漂移。让同一个 Agent 先“搜索关于某开源库的资料”再“读其中三个链接”最后“写一份周报”。前两步还能正常跑到第三步生成周报的时候它经常会重新输出一段资料总结而不是周报。原因很简单Agent 的注意力被前两步的搜索结果给带跑了模型没有在每一步之间做目标校验。第二个是上下文污染。单 Agent 会把用户指令、工具返回结果、中间推理全部塞进同一个上下文窗口。搜索到的无关内容、报错日志、临时笔记都会影响后续生成。尤其是工具调用返回大量 JSON 时Agent 很容易在对话里学着 JSON 的语气“装模作样”而不是继续推进主线任务。第三个是故障恢复能力差。单 Agent 在某个工具调用失败之后经常会重新尝试同样的事情或者干脆放弃任务。它没有“换条路走”的意识因为规划和执行在同一个上下文里模型分不清“现在应该调整计划”还是“现在应该继续执行”。这三点叠加起来单 Agent 在多步骤任务上的成功率会随步骤数量指数级下降。我实测一个5步任务单 Agent 一次跑通的概率不到30%其中一半以上是目标漂移导致的。1.2 双 Agent 解决的三个核心问题双 Agent 不是把模型调用次数从1变2而是把任务流程从“一个人从头干到尾”变成“一个负责拆解和验收、一个负责动手干活”。这对应解决了我上面说的三个问题第一角色分离让提示词彻底分工。规划 Agent 的 system prompt 只需要聚焦任务拆解、目标校验和质量验收执行 Agent 的 system prompt 只需要聚焦工具选择、参数填入和结果整理。两者的上下文互不干扰模型不需要在同一轮里面切换思维方式。第二上下文隔离让记忆不再乱套。规划 Agent 只维护任务目标和步骤列表执行 Agent 只维护当前步骤的中间结果。步骤完成后执行 Agent 把摘要交还给规划 Agent长任务也能保持上下文相对精简。第三失败路径清晰可恢复。执行 Agent 发现某工具调用失败它只需要向规划 Agent 返回一条结构化错误消息规划 Agent 根据错误类型决定换方案、重试还是终止。这就像团队里有明确的“上报机制”而不是每个人都在那儿自己硬扛。我用双 Agent 之后AI助手的5步复杂任务一次跑通率从不到30%提升到了70%以上剩下的30%里大部分也能通过一次重规划恢复。这个提升是很直观的。2. 整体架构设计角色拆分与通信协议2.1 角色定义规划者 Agent 与执行者 Agent我把整个系统拆成了两个角色名字很直接规划者Planner和执行者Worker。规划者接收用户的原始请求负责拆解任务步骤、制定验收标准、校验中间结果执行者接收单个步骤负责调用外部工具、收集信息、整理产出结果。这两个角色的 system prompt 设计差异很大。规划者的核心指令是只拆解任务不执行任何工具调用每次输出必须包含 steps 列表和 expected_output 描述收到 worker 返回的结果后必须判断结果是否符合预期不符合就打回重做。执行者的核心指令是只负责当前单个步骤不做全局规划工具调用必须返回结构化结果如果工具失败直接报告错误信息和可能原因不要自作聪明换工具。这里有一个很容易被忽略的点规划者不能碰工具执行者不能做计划。我一开始就是没把这条写死结果规划者偶尔会去“查阅”工具执行者有时候会自作主张调整步骤顺序。后来我把两条限制写进 system prompt 并且做成硬性策略才稳定下来。2.2 通信协议别让两个 Agent 鸡同鸭讲两个 Agent 之间传递消息我采用的是 JSON 结构化消息。原因很简单对话式文本容易把指令和结果混在一起Agent 解析起来全靠模型心情结构化消息能让模型以确定性的字段识读关键信息也方便我记录日志和做断点调试。消息格式我定义成了这样{ sender: planner, receiver: worker, message_type: task, task_id: task-001-step-2, payload: { instruction: 读取指定链接的文档提取核心要点, inputs: {url: https://example.com/doc} } }Worker 返回的结果消息也保持同样结构{ sender: worker, receiver: planner, message_type: result, task_id: task-001-step-2, payload: { status: success, summary: 文档共3部分核心要点是……, raw_path: /tmp/output/step2.json } }如果执行失败就把 message_type 改成 error并在 payload 里带上 error_code 和 retryable 字段。规划者看到 retryable 为 true 会生成一个新的执行方案retryable 为 false 就直接终止任务向用户说明原因。这个协议帮我解决了一个隐性问题文本消息里夹带的“语气”会影响模型判断。比如 worker 返回一大段话“已完成阅读并总结我认为这里可以优化”planner 看着像结果实际是观点。结构化协议把所有内容按字段拆开让两个 Agent 都只关心能确定对应的字段。2.3 为什么不拆成三个或更多 Agent我在设计阶段也考虑过要不要引入第三个监管者 Agent专门负责审计前两个 Agent 的输出。后来放弃了原因很实际模型生成成本增加30%以上但效果提升不到10%。两个 Agent 已经形成了“规划-执行-校验”的闭环校验逻辑放在规划者内部完成即可不需要第三套独立上下文。更多 Agent 数量带来的是两方面的浪费一是 token 消耗成倍上升二是 Agent 之间的通信组合复杂度呈平方级上升。对于个人AI助手这种业务双 Agent 算是一个甜点值。后来有人问我“要不要给每个工具绑一个 Agent”我直接说不要那是工具多到处理不过来时的工程需求不是个人助手的刚需。工具多的话更好的做法是让 worker 内部做工具路由而不是再起一个 Agent。3. 核心实现细节代码骨架与关键机制3.1 环境准备与依赖我这边实现双 Agent 系统用的语言是 Python版本 3.11。依赖没有用太重的框架主要用三类大模型 API 的调用封装、结构化日志、以及一个内置的 JSON Schema 校验库。由于个人项目需要轻量迭代我不建议一上来就套重型多 Agent 框架先自己手搓一个最小闭环理解每一步的流转之后真有需要再换框架也不迟。以下是我项目里的目录结构agent_assistant/ ├── agent.py # BaseAgent 基类 ├── planner.py # 规划者 Agent ├── worker.py # 执行者 Agent ├── orchestrator.py # 编排器主循环 ├── protocols.py # 消息结构定义 ├── tools/ # 工具集 │ ├── search.py │ └── page_reader.py └── config.yaml # 模型参数配置核心依赖只有一个消息结构定义文件。这个文件决定了两个 Agent 之间如何交流值得先写清楚。3.2 定义消息结构最小通信单元我建议用 dataclass 来定义消息体比 Python 字典更好用因为有类型标注和默认值约束。# protocols.py from dataclasses import dataclass, field from typing import Any, Dict, Optional dataclass class AgentMessage: sender: str receiver: str message_type: str # task / result / error task_id: str payload: Dict[str, Any] field(default_factorydict) error_code: Optional[str] None retryable: bool True created_at: str 实际运行中我会先给每条消息分配 task_id保证整个流程里的日志和状态追踪都可以通过 task_id 关联起来。没有 task_id两个 Agent 的对话就成为没有关联的流水账排查问题会很痛苦。消息校验方面我用了非常朴素的方式在 orchestrator 中校验 sender、receiver 和 message_type 是否符合状态机预期。比如 planner 发出的消息 receiver 必须是 workerworker 返回的消息 receiver 必须是 planner。如果不符合直接报错不会留给模型“自由发挥”。3.3 规划者 Agent从用户请求到步骤列表规划者的核心工作是两步接收用户请求输出结构化步骤列表接收 worker 的中间结果判断是否通过验收。我写规划者时没有把它做成一个只管拆解任务的一次性角色而是设计成一个有“短时记忆”的状态跟踪器。它会维护当前任务的目标、步骤列表、每步的验收标准和已完成步骤摘要。这样当 worker 返回结果时它可以拿“验收标准”和“实际产出”做对比而不是凭感觉。以下是规划者的核心代码简化版# planner.py from agent import BaseAgent from protocols import AgentMessage class PlannerAgent(BaseAgent): def create_plan(self, user_request: str) - AgentMessage: prompt ( 请将以下用户请求拆解为3-5个可执行步骤。\n 每个步骤必须包含: step_id, action, input_summary, acceptance_criteria。\n 只输出JSON数组不要输出额外解释:\n user_request ) plan_data self.llm(prompt, formatjson) return AgentMessage( senderplanner, receiverorchestrator, message_typeplan, task_idgenerate_task_id(), payload{goal: user_request, steps: plan_data[steps]} ) def verify_result(self, step_id: str, worker_result: str, acceptance_criteria: str) - bool: prompt ( 你是质量验收员。根据验收标准判断worker的产出是否达标。\n f验收标准: {acceptance_criteria}\n f实际产出: {worker_result}\n 只回答 PASS 或 FAIL并给出简短原因。 ) decision self.llm(prompt).strip().upper() return decision.startswith(PASS)这里有个细节验证结果的时候我把 worker 的原始产出和验收标准都放进规划者的上下文但不会放 worker 的完整对话历史。这样可以避免规划者被无关信息干扰也节省了 token。3.4 执行者 Agent从步骤指令到工具调用执行者的核心工作是接收规划者下达的单个步骤选择合适的工具执行工具调用整理结果返回给规划者。我实现执行者时特别关注“工具调用失败的上报机制”。工具调用不能只返回 success/failure 两个字还要返回错误码和可重试标志。比如搜索 API 超时这是一个可重试错误如果搜索词本身无效那是不可重试错误。规划者看到不可重试错误会直接换一个搜索策略而不是傻傻地重试三次。简化版代码如下# worker.py from agent import BaseAgent from protocols import AgentMessage from tools.search import search_web from tools.page_reader import read_page class WorkerAgent(BaseAgent): def __init__(self, name, tool_registry, llm): super().__init__(name, system_prompt你是执行者只负责执行单个步骤。, llmllm) self.tool_registry tool_registry # {search: search_web, read_page: read_page} def execute_step(self, task: AgentMessage) - AgentMessage: instruction task.payload[instruction] inputs task.payload.get(inputs, {}) prompt ( f请从可用工具中选择一个并生成调用参数。\n f指令: {instruction}\n f可用工具: {list(self.tool_registry.keys())}\n f输入参数: {inputs}\n 只输出JSON格式为 {tool: xxx, args: {...}}。 ) decision self.llm(prompt, formatjson) tool_name decision[tool] args decision[args] if tool_name not in self.tool_registry: return AgentMessage( senderworker, receiverplanner, message_typeerror, task_idtask.task_id, payload{error: 未知工具}, error_codeTOOL_NOT_FOUND, retryableFalse, ) try: result self.tool_registry[tool_name](**args) summary self.llm(f请将以下工具结果总结成不超过200字保留关键数据: {result}) return AgentMessage( senderworker, receiverplanner, message_typeresult, task_idtask.task_id, payload{summary: summary, raw_preview: str(result)[:1000]}, ) except Exception as e: return AgentMessage( senderworker, receiverplanner, message_typeerror, task_idtask.task_id, payload{error: str(e)}, error_codeTOOL_EXEC_FAILED, retryableTrue, )执行者的 system prompt 里我写了一句更有约束力的话如果你不清楚该调用哪个工具不要编造工具名直接返回错误。这能明显减少模型“一本正经地胡说八道”的情况。3.5 编排器主循环Plan-Execute-Verify双 Agent 之间需要一个调度中枢我把它放进了 orchestrator。它的职责不是参与生成而是控制消息流转、记录状态、执行重试策略。完整流程如下接收用户原始请求交给 PlannerAgent 生成计划。遍历计划中的每个 step逐个构造 AgentMessage 发给 WorkerAgent。WorkerAgent 执行并返回 result 或 error。PlannerAgent 收到 result 后做 verify。verify 通过继续下一个 stepverify 不通过让 PlannerAgent 重新生成针对当前步骤的指令最多重试2次。所有步骤完成后让 PlannerAgent 根据所有步骤摘要生成最终答案。代码骨架# orchestrator.py class Orchestrator: def __init__(self, planner: PlannerAgent, worker: WorkerAgent, max_retries2): self.planner planner self.worker worker self.max_retries max_retries def run(self, user_request: str): plan_msg self.planner.create_plan(user_request) steps plan_msg.payload[steps] step_results [] for step in steps: task_msg AgentMessage( senderplanner, receiverworker, message_typetask, task_idf{plan_msg.task_id}-{step[step_id]}, payload{instruction: step[action], inputs: step.get(inputs, {})} ) result self._run_with_retry(task_msg, step) if result is None: return {status: failed, step: step[step_id]} step_results.append({step: step[step_id], result: result.payload[summary]}) final self.planner.generate_final_report( goalplan_msg.payload[goal], step_resultsstep_results ) return {status: success, answer: final} def _run_with_retry(self, task_msg, step, retries0): if retries self.max_retries: return None result_msg self.worker.execute_step(task_msg) if result_msg.message_type error: if not result_msg.retryable: return None # 让 planner 基于错误信息重新生成指令 new_instruction self.planner.revise_instruction( original_instructionstep[action], errorresult_msg.payload[error] ) task_msg.payload[instruction] new_instruction return self._run_with_retry(task_msg, step, retries 1) accepted self.planner.verify_result( step_idstep[step_id], worker_resultresult_msg.payload[summary], acceptance_criteriastep[acceptance_criteria] ) if not accepted: new_instruction self.planner.revise_instruction( original_instructionstep[action], error验收未通过 ) task_msg.payload[instruction] new_instruction return self._run_with_retry(task_msg, step, retries 1) return result_msg这套主循环的价值在于把“Agent 的自主判断”约束在了一个可控范围内。Agent 可以自由决定怎么拆解、怎么执行但整个流程不会无限跑下去。每次重规划都要求重新生成新的指令并且带上前一轮失败的简短原因这样能有效避免在一个错误点打转。3.6 上下文共享与记忆隔离双 Agent 系统最容易踩的坑就是上下文共享方式不对。我把上下文分成三个层级全局任务上下文、规划者上下文、执行者上下文。全局任务上下文只保存任务目标、当前步骤ID、已完成步骤摘要这块信息两个 Agent 都可读。规划者上下文保存用户请求、完整计划、每一步的验收标准、worker 返回的摘要、以及最近一次失败原因。执行者上下文只保存当前步骤的指令、工具调用参数、工具原始结果和最终摘要。每个 Agent 的历史里不会出现对方的大量推理过程。我举一个具体例子。如果 worker 中间调用了10次搜索每次返回2千字搜索结果那这2万字不会进入 planner 的上下文。worker 只把最终摘要200字交给 planner。planner 不需要知道搜索的过程细节它只需要知道“查到了哪些核心要点”。这样的隔离让两个 Agent 的上下文窗口利用率高了很多。实现层面我每轮处理完 worker 结果后就把原始工具返回归档到本地日志文件只在内存中保留结构化摘要。这一点对于长任务尤其重要不然做个7步任务第一个步骤的原始输出会把后面的上下文挤爆。4. 实操过程与控制参数解析4.1 完整示例让它写一份调研周报为了让你直观看到双 Agent 怎么工作我把一个真实场景拆给你看。我在某天给助手布置了一个任务调研某个开源文档处理库并生成一份周报。用户请求进来后规划者输出的计划大致是step 1搜索该库的基本信息获取官网和 GitHub 仓库地址step 2读取核心文档提取主要模块和使用方式step 3搜索社区近期关于该库的讨论判断活跃度step 4结合前三步结果生成一份包含概述、核心功能、社区情况、使用建议的周报执行者拿到 step 1 后选择 search 工具填入关键词得到5条搜索结果。搜索接口返回的内容比较杂执行者会先用大模型把结果里真正相关的部分提炼出来再_输出一段摘要给规划者。规划者对照验收标准“是否包含官网和仓库地址”来判断发现少一个仓库地址于是让执行者重新搜索第二次执行者换了一个关键词拿到了仓库地址验收通过。整个流程跑完后规划者把4个步骤的摘要拼接起来生成了一份结构完整的周报。整个过程中我只在开始时输入了一句话后面全是两个 Agent 自动推进。中间有一次执行者搜索超时规划者换了搜索方式没有需要我介入。4.2 关键参数与计算temperature、最大轮次、超时双 Agent 系统里需要控制的参数比单 Agent 多最容易出问题的是下面几个第一是 temperature。规划者做任务拆解和结果验收时我把它设为0.2甚至0。因为拆解和验收都需要确定性不需要创造性。执行者调用工具时同样设置为0.2避免编造工具参数。只有最后生成最终周报这种开放性内容时我才会把 temperature 调到0.7让文字更自然。第二是最大轮次。我建议以“每个步骤最多重试2次、总步骤数不超过5个”为起始配置。按这个配置一次完整任务最多会触发15次模型调用每次按上下文4k token估算总消耗在60k token左右。如果你用的是非常小的上下文窗口比如8k就需要把步骤数降到3否则中期就会超限。这里可以直接给个计算过程。假设每个步骤平均消耗2k token包括工具结果和模型输出规划者上下文里还会保留已完成步骤摘要步长为10步时累计消耗会到20k以上。所以如果你的基础模型只有8k上下文那步骤数上限就应该控制在8步以内我不建议单任务超过8步否则就得上外置存储或者自动压缩历史。第三是超时设置。工具调用和模型生成都要设超时。我这边模型生成超时设为30秒工具调用根据类型不同从10秒到60秒不等。超时后 orchestrator 会重新把该步骤交给 worker 执行一次还不成就上报给规划者换方案。如果你不设置超时整个任务可能挂在一个长期不响应的外部工具上非常难受。4.3 冲突消解两个 Agent 意见不一致怎么办双 Agent 运行过程中必然出现不一致最常见的是“worker 认为自己完成了任务planner 认为没完成”。这其实是我故意设计的如果两边永远一致那系统就不需要双角色了。冲突消解我用了一个简单规则以规划者的验收结果为准。worker 的产出只是输入是否合格由规划者判断。如果 worker 多次提交的结果都不合格worker 需要向 planner 提供“当前提交为何不合格”的自我分析这个分析会和结果一起进入 planner 的上下文。这算是一个轻量的反馈机制而不是简单的重复劳动。我还加了计数器策略。当同一 worker 对同一 task_id 返回错误达到3次时orchestrator 会生成一条“任务卡壳报告”内容包含错误序列、worker 每次提交的摘要、规划者每次的验收意见然后发送给一个专门做总结的模型调用输出“卡壳原因分析”。这个过程不以对话形式出现而是作为一个独立工具调用注入到 planner 上下文里。实测这个机制能把烂尾任务比例再降低10%左右。5. 常见问题与排查技巧实录5.1 两个 Agent 陷入死循环双 Agent 最让人头疼的是死循环planner 要求改worker 改了但还是不达标再来再来……如果没设轮次上限这个循环能烧掉大量 token。我遇到过一次搜索任务worker 连续3次返回同样的结果只是换了一种措辞planner 每次都判断不合格结果重复调用接近10次。排查时我首先看的是 orchestrator 日志里的 message_type 和 task_id。如果同一个 task_id 反复出现且 payload 的内容几乎一样说明 worker 没有真正理解 planner 的修改意见。解决方式如下一是强硬限制重试次数不超上限就终止并返回错误二是让 worker 在重新执行前“复述一遍修改意见”把修改点结构化输出而不是直接去生成结果。这一步可以有效打断“改了个寂寞”的循环。5.2 工具调用失败后没有恢复路径早期版本里worker 遇到工具超时返回的错误只有文字描述“搜索失败”planner 只能傻乎乎地说“请再试一次”。后来我把错误信息改成了结构化字段error_code、retryable、suggestion。比如搜索工具超时返回{ error_code: SEARCH_TIMEOUT, retryable: true, suggestion: 可尝试缩短搜索词长度或使用备用搜索接口 }planner 看到 retryable 为 true 会重新规划步骤看到 retryable 为 false 会直接向用户说明原因不会再白费力气。这套做法让失败恢复从“模型猜”变成了“代码判断”稳定很多。5.3 上下文被撑爆token 超限双 Agent 的 token 消耗比单 Agent 高因为有两个上下文窗口在同时增长。我遇到过的最极端情况某次长任务执行到第6步时worker 突然报 context length exceeded整个 orchestrator 直接崩掉。排查后发现原因是 worker 的上下文里保留了大量工具调用的 raw result哪怕我已经做了摘要日志字段里仍有 raw_preview 字符串。解决方法是把 raw_preview 从1000字改成只保留首400字同时把中间结果归档到本地磁盘。更彻底的做法是开启自动摘要当 worker 历史超过一定长度时用一次模型调用把早期内容压缩成200字摘要再放到上下文开头。这个机制很实用适合给个人助手长任务兜底。5.4 两个 Agent 相互踢皮球“踢皮球”指的是 planner 让 worker 做worker 说做不了planner 又说你先做了再说两边来回几轮没有任何实质进展。这类问题通常不是模型能力不足而是角色职责边界没有写清楚。我的对策是在 system prompt 里明确worker 如果无法执行必须返回 error且 error 里必须包含“无法执行的原因”和“建议下一步动作”planner 收到 error 且 retryable 为 false 时禁止再原样下发任务而必须重新拆解或向用户求助。同时orchestrator 增加一条规则planner 连续两次给 worker 下发相同 instruction 且没有修改任何字段直接判定失败。这个硬规则能防止无意义踢皮球。5.5 踩坑经验速查表我把阶段6开发中反复遇到的坑整理成了表格后面迭代的时候直接对照着看省了很多重复排查的时间。症状根因快速处理方案两个 Agent 反复说同样的话上下文里没有携带上一次失败的关键差异每次下发新任务前要求 worker 复述修改意见任务跑到一半变成无关话题规划者上下文掺入了工具原始输出隔离上下文只让规划者看到结构化摘要token 消耗翻倍每个 Agent 都保留了对方的全量历史压缩历史或只保留必要字段worker 编造工具名称提示词约束不足在 worker 提示词里加入“未知工具必须报错”planner 对合格结果误判验收标准写得模糊拆解计划时要求“验收标准必须可量化”这个表格并不是万能灵药但覆盖了大多数双 Agent 项目会遇到的初期问题。每一条我都实际踩过不是凭空总结。另外我还想多说一句不要一上来就追求“漂亮”的架构。我第一次写双 Agent 系统时参考了网上各种复杂设计搞得自己代码里到处都是回调、事件总线、状态机结果运行起来反而更难排查。后来我把架构简化到“编排器调度两个角色”的朴素模型所有逻辑都线性可见问题就好处理多了。对于个人AI助手这样的场景简单可靠比花哨重要得多。双 Agent 系统的开发核心不是让两个模型互相聊天而是让它们像真实团队里的同事一样有明确分工、有统一协议、有验收标准还有出了问题时的上报路径。如果你正准备从单 Agent 升级到双 Agent建议先把这五件事想明白角色边界是什么、消息格式是什么、失败怎么处理、上下文怎么隔离、重试上限是多少。想明白了再写代码会比边写边试稳很多。