双Agent架构实战：从单Agent瓶颈到企业级数字劳动力落地

做企业级智能体有一段时间了，最大的体感是：很多团队卡住不是因为模型不够聪明，而是架构没跟上。把智能体接到对话框里做一个问答助手，这事儿现在已经不太稀奇了；难的是让它真得像一名员工一样，接住一个目标，自己拆任务、调工具、做汇报、对结果负责。这也是我最近把大部分精力放在双 Agent 架构上的原因——一个规划器管思考，一个执行器管动手，让 AI 从辅助工具真正变成企业里的数字劳动力。

这篇文章想把我在实际项目里踩过的坑、验证过的设计取舍、以及一套可以直接抄作业的双 Agent 骨架完整讲清楚。不管你现在用的是大模型原生接口，还是接在成熟框架上，只要目标是让智能体承接真实业务流程，这篇都值得看完。我会从"为什么单 Agent 不够用"讲起，然后拆解双 Agent 的职责边界、关键细节、完整代码案例，最后是落地时的高频故障清单。

1. 先搞清楚：为什么企业里的单 Agent 总是不够用

1.1 从辅助工具到数字劳动力，差在哪里

"数字劳动力"这个词这两年很热，但很多人理解得过于简单。它不是把一个能问答的机器人挂在企业微信里就叫数字劳动力，也不是把一组 API 串起来跑个自动化脚本。我们团队在实际交付项目时，会用三个标准来判断一套智能体方案是否达到了数字劳动力的水准：

• 目标驱动：你交给它一个结果目标（比如"处理本月退货工单"），而不是一串操作指令。它需要自己决定先干什么、后干什么。
• 端到端闭环：从接收任务、拆解计划、调用工具、校验结果到输出汇报，整条链路是闭环的。中间卡住了，它能自己尝试修复。
• 可审计：每一步决策都能追溯到"当时为什么这么做"，关键节点有人工审批兜底。这其实是企业敢不敢放手让 AI 独立干活的前提。

传统的对话机器人只能做到"人问一句、它答一句"，本质上是一个接口封装；RPA 自动化只是把固定流程跑得更快，流程一变就废。而数字劳动力的核心差异在于自主决策能力——它面对的是没有 100% 预定义好的流程，需要临时规划路径。

举一个我们做过的真实场景：某公司的售后客服团队每天要处理几百条退货退款工单。早期方案是做一个"工单问答机器人"，用户问物流到哪了、退款几天到账，它查库回复，这确实省了一些人力。但问题是，工单处理本身还是由人工完成的——客服需要先核实身份、查订单、看是否满足退货政策、填审批单、再通知仓库。机器人只是把"查"这一步加速了。后来我们把目标改成了"让智能体独立完成整条退货流程"，这才引出了架构上的硬性需求：单 Agent 搞不定。

1.2 单 Agent 在企业场景的三个瓶颈

第一个瓶颈是上下文墙。一个 Agent 既要理解全局目标、梳理任务依赖，又要处理底层工具返回的各种异常信息，这些信息挤在同一个上下文窗口里会互相干扰。规划一个五步流程可能只需两三百个 token，但执行过程中工具返回的 JSON、报错日志轻松就能吃掉几千个 token。上下文一旦被细节塞满，Agent 就会"忘了"最初的目标，开始基于局部信息做决策——这在真实业务里非常危险。

第二个瓶颈是权限设计。企业级应用绕不开权限控制。如果单 Agent 拥有所有工具，它可能在某些情况下调用风险操作；如果只给它最小权限，它又会因为频繁触达权限边界而中断流程。我们在一个项目里就吃过亏：Agent 需要在客户管理后台里查客户等级、在订单系统里查历史订单、在工单系统里改状态。单 Agent 方案里，给它三个系统的全部读写权限不现实，但只给读权限又没法完成"修改工单状态"这一步。最后不得不写一堆规则代码去临时提升权限，架构变得又臭又硬。

第三个瓶颈是收敛困难。智能体是带循环的，不是线性函数。单 Agent 在一个循环里既要"想下一步做什么"（规划），又要"处理当前这步的结果"（执行），再遇到工具异常时，它的心智负担极重，非常容易陷入死循环。我们见过最离谱的一次，Agent 连续调了十几次同一个失败的查询接口，每次换个参数重试，白白烧掉大量 token，最后还是没跑完流程。

这三个瓶颈合在一起，指向同一个结论：在企业场景里，把规划和执行混在同一个 Agent 里是反模式。于是双 Agent 架构成了很自然的解法。

2. 双 Agent 架构的整体设计：一个动脑，一个动手

2.1 我的设计原型：规划器加执行器

这套架构的核心思路特别简单——模仿一个成熟团队里老板和员工的配合方式。

• Planner（规划器）：相当于老板，负责接收任务、拆解子任务、编排依赖关系、生成任务计划表，并在最终检查结果。
• Worker（执行器）：相当于员工，负责执行单个子任务，按需调用具体工具，并把执行结果反馈给 Planner。

两者之间通过一个任务总线（Task Bus）通信。这个总线可以是一个简单的内存队列，也可以是 Redis 或者消息队列，取决于你的部署规模。Planner 不直接接触工具，它只负责"想"，把任务拆成可执行的子任务后丢到总线上；Worker 只负责"做"，从总线取子任务、选择工具、执行、返回结果。整个链路是单向的，目标清晰、职责分明。

我见过很多团队一开始就上三四个 Agent 的大规模编排，结果把大量时间耗在了 Agent 之间的通信协议设计上。双 Agent 的优势恰恰在于只有一条通信链路，交互关系足够简单，容易排查问题，也容易被业务方理解。

2.2 为什么是两个，不是三个也不是一个

要回答这个，得看"规划"和"执行"这两种能力的本质差异。

规划任务需要的是全局上下文：企业目标、历史对话、约束条件、任务依赖关系；执行任务需要的是局部上下文：工具返回结果、参数校验、异常信息。这两类信息一个求广、一个求深，放一起必然互相干扰。所以拆成两个是"按认知粒度拆"的自然结果。

那为什么不再多拆几个？因为每多一个 Agent，通信成本是平方级上升的。三个 Agent 之间就有三条通信链路，四个就有六条，而且多 Agent 还会有"角色推诿"问题——A 觉得 B 该做，B 觉得 A 说了算，最后没人对结果负责。在我们的实践中，双 Agent 已经能覆盖绝大多数企业业务流程。如果某个任务特别复杂，没必要加第三个统筹 Agent，而是让 Worker 内部再挂子 Worker，颗粒度继续往下走。

2.3 三种架构的对比：看完就知道怎么选

我整理了一张对照表，是我们内部做技术选型时用的。

结论很直接：企业级生产环境里，双 Agent 是性价比最高的起点。不要一上来就追逐多 Agent 的花哨效果，先把两个角色之间的职责边界定义清楚，能解决 80% 以上问题。

3. 核心细节拆解：这几个地方决定了能不能落地

3.1 工具层的注册与权限隔离

双 Agent 架构里有个特别容易忽略的点：Planner 不持有工具，Worker 才持有工具。这个设计不是随便定的，它是权限安全的关键。

打个比方：老板不需要知道员工电脑的密码，他只需要分配任务"把这份合同打印出来"，具体用哪台打印机是员工的事。同样，Planner 只负责表达意图，比如"查询订单 T20240315 的物流状态"，而"调用哪个物流查询接口、参数怎么填、接口报错了怎么重试"全丢给 Worker。这样 Planner 的上下文永远保持干净，不会被工具细节污染；Worker 的权限也被限制在它需要的最小范围内。

我们在实际落地时，给每个 Worker 配了一张工具路由表。这张表不是简单的"工具名映射到函数"，而是包含了工具描述、参数 schema、以及权限等级。Worker 内部会先根据任务描述做一次工具匹配，把候选工具缩小到两三个，再调用大模型做精确选择。这一步能显著降低"瞎调工具"的概率。

建议把工具按风险等级分成三类：

• 低风险：信息查询类（查订单、查库存、查物流）。
• 中风险：状态修改类（改工单状态、发送通知），一般需要记录审计日志。
• 高风险：资金操作类（退款、转账、大额订单修改），必须走人工审批。

低风险工具 Worker 可以自主调用；中风险工具允许 Worker 发起，但结果要包含操作前后对账；高风险工具 Worker 只能发起"申请"，由人审网关决定是否放行。

3.2 上下文管理：共享与隔离的边界

这是双 Agent 架构最微妙的地方。我们最初犯过一个错：把 Planner 和 Worker 放在同一个共享上下文里跑，结果一次会话里塞满了几十个工具调用记录之后，Planner 开始"发疯"，把前面的历史订单数据当成新任务去处理。

后来我们定了两条铁律。

全局上下文归 Planner。它维护三样东西：原始任务目标、企业的硬性约束（比如退款不能超过订单金额、改价需要审批）、以及任务依赖关系图。每次 Worker 返回结果，Planner 只更新任务图的状态，不关心具体的工具调用日志。

局部上下文归 Worker。Worker 每次只处理一个子任务，它的上下文包含：当前任务描述、被授权的工具列表、以及最近几次工具调用的结果。子任务完成后，局部上下文直接清零，不留残余。这相当于员工做完一件事之后把工作台收拾干净，等下一个指令。

实际操作中，我们会在 Planner 的计划表结构里额外维护一个任务依赖 DAG：

{ "mission_id": "T20250317-001", "mission": "处理退货工单 RT-8842", "nodes": { "1": {"name": "核实客户身份", "status": "done", "deps": []}, "2": {"name": "查询订单状态", "status": "done", "deps": ["1"]}, "3": {"name": "判断是否满足退货政策", "status": "pending", "deps": ["2"]}, "4": {"name": "提交退款审批", "status": "pending", "deps": ["3"]}, "5": {"name": "通知客户结果", "status": "pending", "deps": ["4"]} } }

这个 DAG 是整个架构的"定海神针"。Worker 报过来一个结果，Planner 只需要知道"节点 2 完成了"，下一步该催节点 3。哪怕中途失败重试，所有状态都在这张图里，不会丢失。

3.3 人审网关：哪些环节必须停下来

企业级和玩具级最大的分水岭，就是有没有人审机制。我见过不少团队把 Agent 做得"太顺了"——全自动跑完一条流程，结果某一步操作错了，已经发出去了，想撤都撤不回来。

人审网关（Human Approval Gateway）插在 Planner 和 Worker 之间，本质上是一个拦截器。Worker 每次准备执行任务前，先把"意图声卡"（Intent Card）发给人审网关——内容包括要调哪个工具、参数是什么、涉及金额多少。网关根据预设规则判断：

• 规则放行：低风险操作，自动通过。
• 规则拒绝：违反业务约束的操作（比如退款金额超过订单原价），直接打回。
• 人工审批：命中高风险条件时，推送给负责人，在企微或飞书里点一下同意或拒绝。

在下面案例的退款工单流程里，退款金额低于 500 元自动过，500 到 5000 元推送给财务审核，高于 5000 元直接进人工客服渠道处理——智能体负责把信息整理得明明白白，但决定权必须留在人手里。

3.4 Token 成本控制与预算策略

双 Agent 架构比单 Agent 多了一个 Agent 的 token 开销，这是客观事实。但我们在实践中发现，如果协议设计得当，成本增加的幅度完全可控，甚至可能比单 Agent 反复失败重试更省钱。

关键是用结构化协议替代自然语言会话。Planner 和 Worker 之间不聊闲天，通信格式是固定的 JsonMessage，包含task_id、intent、params、status、result等字段。不需要让 Worker 理解一句自然语言"麻烦你帮我看看这个订单的物流情况然后顺便反馈一下"，而是直接给一个结构化的intent: query_logistics。省下来的 token 相当可观。

另外一个经验是给 Planner 设置"思考预算"。在计划阶段，允许 Planner 多花 token 做详细拆解；在执行阶段，Worker 的每次调用都限制输出长度，工具返回结果如果太长，做截断与摘要，只保留关键字段回传给 Planner。

4. 实操：从零搭一个双 Agent 数字劳动力

4.1 技术选型：什么时候不要选现成平台

这个章节给想动手实做的朋友。先说选型：我们最终没有用开箱即用的 Agent 平台，而是基于 LangGraph 做了一层轻量封装。原因是企业场景里常要私有化部署、要对接内部系统打通、要改审批流，现成平台的黑盒有时候让人很头疼。

轻量自研方案的选型是这样的：

• 大模型：根据需求选，一般用支持 function calling 的模型会省很多事。
• 框架：LangGraph 做状态编排，或者干脆自写一个有限状态机。如果任务流程不复杂，其实有限状态机就够。
• 消息总线：单机部署用asyncio.Queue，多节点部署用 Redis Streams。
• 工具封装：每个工具都是一个标准的Tool(name, description, parameters_schema, handler)结构。

从零搭的第一行代码先从 Planner 的数据结构开始，而不是先去写大模型调用。把任务 DAG 和消息格式定下来，后面整个系统就有了骨架，大模型只是骨架上的"肉"。

4.2 核心代码骨架串讲

为了不占太多篇幅，我挑最关键的三段逻辑来讲：Planner 的规划循环、Worker 的执行循环、以及两者之间的消息协议。

第一段，Planner 接收任务并拆解成子任务：

import json from dataclasses import dataclass, field from typing import Optional, List PLANNER_SYSTEM_PROMPT = """ 你是一名企业业务流程规划器。 你的任务是根据用户的目标，输出一份可执行的子任务计划。 规则： 1. 子任务之间要明确依赖关系。 2. 每个子任务必须附带意图描述，意图命名要清晰。 3. 只输出 JSON，不要输出解释。 输出格式： {"nodes": [{"id": 1, "name": "任务名", "intent": "意图名", "deps": []}]} """ @dataclass class Node: id: int name: str intent: str deps: List[int] status: str = "pending" @dataclass class Plan: mission_id: str mission: str nodes: List[Node] = field(default_factory=list) completed: set = field(default_factory=set) def next_ready(self): for node in self.nodes: if node.status == "pending" and all(d in self.completed for d in node.deps): return node return None class Planner: def __init__(self, llm, bus): self.llm = llm self.bus = bus self.plans = {} async def submit_mission(self, mission: str): plan_id = f"P-{time.time()}" resp = await self.llm.chat([ {"role": "system", "content": PLANNER_SYSTEM_PROMPT}, {"role": "user", "content": mission} ]) plan_data = json.loads(resp) plan = Plan(mission_id=plan_id, mission=mission) for item in plan_data["nodes"]: plan.nodes.append(Node( id=item["id"], name=item["name"], intent=item["intent"], deps=item["deps"] )) self.plans[plan_id] = plan self._schedule(plan) return plan_id async def _schedule(self, plan: Plan): node = plan.next_ready() while node: await self.bus.submit(plan.mission_id, node) node.status = "running" node = plan.next_ready() async def collect_result(self, msg): """接收 Worker 回传的结果，更新 DAG""" plan = self.plans[msg["mission_id"]] for node in plan.nodes: if node.id == msg["node_id"]: if msg["status"] == "success": node.status = "done" plan.completed.add(node.id) else: node.status = "failed" self._schedule(plan)

这段代码里最关键的是next_ready()方法——它只返回"依赖都已完成的待执行节点"。企业流程里先后顺序很重要，如果让大模型自由乱序执行，很容易出错，所以用 DAG 做硬约束。

第二段，Worker 从任务总线拉取子任务并调用工具：

WORKER_SYSTEM_PROMPT = """ 你是一名企业业务流程执行器。 每次只执行一个子任务。根据任务描述，从工具列表中选择最合适的工具。 如果任务无法完成，返回失败并说明原因。 只输出 JSON，格式： {"tool": "工具名", "arguments": {"参数名": "参数值"}} """ class Worker: def __init__(self, llm, tools: dict, agent_id: str): self.llm = llm self.tools = tools self.agent_id = agent_id async def run(self, task: dict): """task 结构：{mission_id, node_id, name, intent}""" tool_choices = self._filter_tools(task["intent"]) resp = await self.llm.chat([ {"role": "system", "content": WORKER_SYSTEM_PROMPT}, {"role": "user", "content": json.dumps({ "任务": task, "可选工具": tool_choices }, ensure_ascii=False)} ]) decision = json.loads(resp) tool = self.tools.get(decision["tool"]) if not tool: return {"status": "failed", "reason": f"工具 {decision['tool']} 不存在"} try: result = await tool.handler(**decision.get("arguments", {})) return { "mission_id": task["mission_id"], "node_id": task["node_id"], "status": "success", "result": result } except Exception as e: return { "mission_id": task["mission_id"], "node_id": task["node_id"], "status": "failed", "reason": str(e) } def _filter_tools(self, intent: str): """按意图过滤工具，只保留候选工具，减少大模型误选概率""" return {name: tool for name, tool in self.tools.items() if intent in tool.supported_intents}

每个工具都声明自己支持哪些intent，Worker 在调用大模型之前先做一次硬过滤，把可选范围缩小。比如退货流程的"查询订单"这一步，可选工具被过滤到只剩查询订单接口，大模型哪怕再笨也选不错工具。

第三段，消息总线与主流程。这里用最简单的方式，让 Planner 和 Worker 在同一个进程里协作：

import asyncio class TaskBus: def __init__(self): self.queue = asyncio.Queue() async def submit(self, mission_id, node): await self.queue.put(node) async def consume(self): return await self.queue.get() async def main(): bus = TaskBus() # 注册工具 query_order_tool = Tool( name="query_order", description="查询订单状态", supported_intents=["query_order"], handler=order_service.query_by_id ) tools = {"query_order": query_order_tool} planner = Planner(llm=llm_client, bus=bus) worker = Worker(llm=llm_client, tools=tools, agent_id="worker-1") # 提交任务 mission = "查询订单 T20240315 的物流状态" await planner.submit_mission(mission) # 执行循环 while True: task = await bus.consume() result = await worker.run(task) await planner.collect_result(result) if planner._is_finished(): break

这是最简版本，实际生产环境中还要加入超时控制、重试机制、审计日志。但核心链路已经通了：Planner 拆任务 → Bus 分发 → Worker 执行 → Planner 更新状态。

4.3 一个完整案例：售后工单自动处理

理论讲再多，不如看一个能跑的案例。我们用这套双 Agent 架构实现了一个售后工单自动处理系统，业务目标很直接：客户提交退货退款申请后，系统自动核验身份、查订单、判政策、填审批单、发通知。

完整的处理流程拆成了五个子任务：

1. 核实客户身份：从工单系统读客户信息，和订单系统里的下单人比对。这一步是纯查询，低风险。
2. 查询订单状态：确认订单已发货且签收时间在退货政策期限内。这步需要调用订单系统，也是低风险。
3. 判断退货政策匹配：把订单信息和退货政策库做匹配，输出"允许退货 / 拒绝退货 / 需要人工审核"三种结论。这步会读取政策库，中风险，因为结论直接影响后续动作。
4. 提交退款审批：如果政策匹配通过，生成审批单，提交到财务审批系统。这一步是高风险操作，默认走人审网关，不直接放行。
5. 通知客户结果：审批通过后，通过消息网关给客户发短信和站内信。这是中风险操作，因为对外发的文案需要用统一模板。

Planner 拿到"处理退货工单 RT-8842"这个目标后，生成的任务 DAG 如下：

节点1（核实身份） → 节点2（查询订单） → 节点3（政策匹配） → 节点4（退款审批） → 节点5（通知客户）

Worker 在执行节点 3 时，如果发现订单金额超过 5000 元或者退货原因是"非质量问题"，会把人审条件挂上去，流程在节点 4 停下来等审批人点击通过或驳回。我在前面人审网关部分说的规则，就是在这个案例里真正生效的。

这个系统上线两周后，处理效率提高了大概三倍。原来的客服团队要逐单查询、逐单填表，现在只需要处理那部分被拦截下来的人工审批单，以及少数 Agent 无法判定的边缘案例。这才是"AI 从辅助升级为数字劳动力"的真实含义——它不再是等指令接话的助手，而是主动完成一整条业务链路的数字员工。

5. 落地过程中的常见问题与排查记录

5.1 Agent 跑偏：规划与用户意图不一致

现象：Planner 把"退货"理解成了"退款"，跳过了退货入库核验，直接生成退款审批单。

排查方法：第一步查 Planner 的计划输出。在我们的代码里，计划会在提交时打印全量 DAG，直接看节点名称是否贴合原始目标；第二步检查 Planner 的 System Prompt，看是不是约束写得太宽松。

解决方案：在企业场景里，不要完全依赖大模型的语义理解来约束流程。最有效的做法是给 Planner 加"硬规则前置检查"——在生成计划后，用一段确定性代码校验关键节点是否存在。比如退货流程必须包含"退货入库确认"节点，如果缺少，直接打回重新规划。我把这种机制叫"计划校验器"，它比任何提示词技巧都靠谱。

5.2 工具调用失败后被无限重试

现象：Worker 调用查询接口失败，返回结果给 Planner，Planner 重新调度同一节点，Worker 再调一次，循环了十几次，Token 烧光了还没成功。

根因分析：Planner 的调度逻辑里没有失败上限，只要节点状态不是 done，就无限重新调度。

解决方案：加失败计数和失败策略。节点失败一次后，备注失败原因，然后放回待执行队列；连续失败达到 3 次，节点状态置为blocked，整个任务挂起并通知管理员介入。千万不要让 Agent"永不放弃"，真实业务里重复失败十次只会让系统变得更糟。

5.3 多轮任务状态丢失

现象：系统运行到一半进程崩溃重启，重新恢复后 Worker 不记得之前执行到哪一步了。

排查方法：确认状态存储的位置。如果你只用内存变量存 DAG，进程一挂就全丢。

解决方案：把 DAG 序列化到 Redis 或者数据库里，每次节点状态变更就落一次盘。节点数量少，状态变更频率也不高，这个写操作的开销完全可以忽略。我们在生产中就是把 DAG 存到 Redis Hash 里，key 是mission_id，field 是节点 id，value 是状态 JSON。恢复流程先读出 DAG，标记所有running状态改成pending，再重新调度，简单有效。

5.4 敏感信息通过工具参数泄露

现象：Worker 在调用某个内部系统时，因为接口报错把 SQL 语句或者内部 token 夹杂在报错信息里返回了，而这些报错信息被完整记录进了日志系统。

对策：两层防线。第一层，工具封装层要做异常清洗，捕获异常后只提取用户友好的错误摘要，不把原始异常堆栈直接抛给大模型。第二层，在消息总线和日志出口加敏感信息过滤插件，用正则匹配掉常见的 token、密钥、手机号格式。数字劳动力系统一定是会被审计的，安全设计从一开始就要进去。

5.5 快速排查速查表

每次排查问题，我都建议团队保留一份现场记录——Agent 当时的输入、工具候选集、最终选择、错误信息。这不是为了追责，是为了日后优化 Prompt 时手里有真实样本。没有这些样本，调 Prompt 全靠猜，效率低得可怕。

写在最后的几条体感

架构设计最忌讳的就是在脑子里觉得"差不多能跑"就开干。我这套双 Agent 方案，看着结构简单，真正落地时四处漏风的场景见过太多了。开始动手之前，先把三个问题想清楚：Planner 和 Worker 之间的消息协议是不是足够结构化、失败机制是不是有兜底、人审节点是不是覆盖了所有高风险操作。这三点想清楚，整个系统就已经稳定了一半。

另外一个经验是，不要指望规划器的 Prompt 一次性写对，它是整个系统里被改动最频繁的组件。我们每次因为业务调整修改了流程规则，都会同步改 Planner 的提示词和计划校验器。版本记录一定要做好，不然新规则上线后 Planner 不按预期拆任务，排查起来特别痛苦。

最后想说一句：数字劳动力这个方向是实打实的趋势，但通往生产环境的路上没有任何捷径。老老实实做权限隔离、做状态持久化、做人审兜底，把每个环节的细节抠到位，这套架构才真正撑得起"企业级"三个字。模板代码可以直接拿去改，但排查和优化的耐心只能自己攒。