做企业级智能体有一段时间了最大的体感是很多团队卡住不是因为模型不够聪明而是架构没跟上。把智能体接到对话框里做一个问答助手这事儿现在已经不太稀奇了难的是让它真得像一名员工一样接住一个目标自己拆任务、调工具、做汇报、对结果负责。这也是我最近把大部分精力放在双 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多 Agent3个及以上上下文管理全局与局部信息混在一起容易干扰全局上下文归 Planner局部上下文归 Worker边界清晰每个 Agent 都维护自己的上下文极易不一致权限控制要么太大要么太小很难两头兼顾Worker 持有最小权限Planner 不带工具权限矩阵复杂配置和维护成本高故障隔离一个环节崩了整个流程全部回滚Worker 单个任务失败Planner 可以重新规划链路长任意节点失败都可能引起连锁反应开发成本最低适合 Demo中等通信协议只需定义一次较高要处理 Agent 间协商、广播、发现收敛难度高单循环既规划又执行低每个 Agent 的心智负担单一中高多个 Agent 之间的协商循环很难收敛适用场景简单问答、单一工具调用大多数真实业务流程复杂模拟、研究性质的多角色协作结论很直接企业级生产环境里双 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_factorylist) completed: set field(default_factoryset) 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 fP-{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_idplan_id, missionmission) for item in plan_data[nodes]: plan.nodes.append(Node( iditem[id], nameitem[name], intentitem[intent], depsitem[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_asciiFalse)} ]) 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}每个工具都声明自己支持哪些intentWorker 在调用大模型之前先做一次硬过滤把可选范围缩小。比如退货流程的查询订单这一步可选工具被过滤到只剩查询订单接口大模型哪怕再笨也选不错工具。第三段消息总线与主流程。这里用最简单的方式让 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( namequery_order, description查询订单状态, supported_intents[query_order], handlerorder_service.query_by_id ) tools {query_order: query_order_tool} planner Planner(llmllm_client, busbus) worker Worker(llmllm_client, toolstools, agent_idworker-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 架构实现了一个售后工单自动处理系统业务目标很直接客户提交退货退款申请后系统自动核验身份、查订单、判政策、填审批单、发通知。完整的处理流程拆成了五个子任务核实客户身份从工单系统读客户信息和订单系统里的下单人比对。这一步是纯查询低风险。查询订单状态确认订单已发货且签收时间在退货政策期限内。这步需要调用订单系统也是低风险。判断退货政策匹配把订单信息和退货政策库做匹配输出允许退货 / 拒绝退货 / 需要人工审核三种结论。这步会读取政策库中风险因为结论直接影响后续动作。提交退款审批如果政策匹配通过生成审批单提交到财务审批系统。这一步是高风险操作默认走人审网关不直接放行。通知客户结果审批通过后通过消息网关给客户发短信和站内信。这是中风险操作因为对外发的文案需要用统一模板。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 调用查询接口失败返回结果给 PlannerPlanner 重新调度同一节点Worker 再调一次循环了十几次Token 烧光了还没成功。根因分析Planner 的调度逻辑里没有失败上限只要节点状态不是 done就无限重新调度。解决方案加失败计数和失败策略。节点失败一次后备注失败原因然后放回待执行队列连续失败达到 3 次节点状态置为blocked整个任务挂起并通知管理员介入。千万不要让 Agent永不放弃真实业务里重复失败十次只会让系统变得更糟。5.3 多轮任务状态丢失现象系统运行到一半进程崩溃重启重新恢复后 Worker 不记得之前执行到哪一步了。排查方法确认状态存储的位置。如果你只用内存变量存 DAG进程一挂就全丢。解决方案把 DAG 序列化到 Redis 或者数据库里每次节点状态变更就落一次盘。节点数量少状态变更频率也不高这个写操作的开销完全可以忽略。我们在生产中就是把 DAG 存到 Redis Hash 里key 是mission_idfield 是节点 idvalue 是状态 JSON。恢复流程先读出 DAG标记所有running状态改成pending再重新调度简单有效。5.4 敏感信息通过工具参数泄露现象Worker 在调用某个内部系统时因为接口报错把 SQL 语句或者内部 token 夹杂在报错信息里返回了而这些报错信息被完整记录进了日志系统。对策两层防线。第一层工具封装层要做异常清洗捕获异常后只提取用户友好的错误摘要不把原始异常堆栈直接抛给大模型。第二层在消息总线和日志出口加敏感信息过滤插件用正则匹配掉常见的 token、密钥、手机号格式。数字劳动力系统一定是会被审计的安全设计从一开始就要进去。5.5 快速排查速查表症状优先排查位置常用修复手段Planner 拆出的任务和用户目标不一致Planner 的 System Prompt 和计划校验器规则收紧 Prompt 约束增加硬规则校验Worker 频繁选错工具工具的supported_intents和描述控制在选前做规则过滤缩小候选集任务进度推进缓慢任务总线堆积情况Worker 并发数增加 Worker 实例或多个节点并发消费节点失败后无限重试Planner 调度逻辑里的失败计数加入失败阈值和阻塞机制端到端链路偶发超时单个工具响应时间、大模型调用延迟给工具调用加超时熔断独立超时参数多轮会话后结果漂移Planner 的全局上下文长度定期压缩历史信息只保留关键事实列表每次排查问题我都建议团队保留一份现场记录——Agent 当时的输入、工具候选集、最终选择、错误信息。这不是为了追责是为了日后优化 Prompt 时手里有真实样本。没有这些样本调 Prompt 全靠猜效率低得可怕。写在最后的几条体感架构设计最忌讳的就是在脑子里觉得差不多能跑就开干。我这套双 Agent 方案看着结构简单真正落地时四处漏风的场景见过太多了。开始动手之前先把三个问题想清楚Planner 和 Worker 之间的消息协议是不是足够结构化、失败机制是不是有兜底、人审节点是不是覆盖了所有高风险操作。这三点想清楚整个系统就已经稳定了一半。另外一个经验是不要指望规划器的 Prompt 一次性写对它是整个系统里被改动最频繁的组件。我们每次因为业务调整修改了流程规则都会同步改 Planner 的提示词和计划校验器。版本记录一定要做好不然新规则上线后 Planner 不按预期拆任务排查起来特别痛苦。最后想说一句数字劳动力这个方向是实打实的趋势但通往生产环境的路上没有任何捷径。老老实实做权限隔离、做状态持久化、做人审兜底把每个环节的细节抠到位这套架构才真正撑得起企业级三个字。模板代码可以直接拿去改但排查和优化的耐心只能自己攒。