系统平稳运行一个月后,周二下午,采购部门的小李提交了一份采购申请。
老张登录系统,看到流程如下:
``` 采购申请(小李提交) │ ├── 财务审批(财务 Agent 处理)── ✅ 通过 │ ├── 法务审核(法务 Agent 处理)── ⏸️ 等待 │ │ │ └── 法务 Agent 不知道财务已经通过,需要人工通知 │ ├── IT 配置(IT Agent 处理)── ⏸️ 等待 │ │ │ └── IT Agent 不知道前两步完成了,需要人工通知 │ └── 完成(预计 3 天) ```问题显而易见:每个部门的 Agent 能力都很强,但跨部门协作仍需人工通知。老张统计发现:平均每个流程涉及 3 个部门协作,每次跨部门交接都需要人工介入,一个流程从提交到完成平均耗时 3 天,其中大部分时间都浪费在“等待通知”上。
王总看到数据,问了一句:“能不能让这些 Agent 自己协调?”
这个问题,让老张陷入了沉思。
---老张的困境:单一 Agent 的局限性
老张分析了当前的系统架构:每个 Agent 相互独立,互不知晓对方的工作状态。他尝试创建一个“超级 Agent”,将财务、法务、IT 的所有工具整合在一起。
结果,问题立刻暴露了。
问题一:上下文爆炸。超级 Agent 必须掌握所有领域的知识:财务审批的标准是什么?法务审核需要哪些文件?IT 配置有什么依赖?系统提示词(System Prompt)越来越长,Agent 开始“迷失”——它无法确定该用哪一个标准进行决策。
问题二:权限混乱。财务 Agent 只能访问财务系统,法务 Agent 只能访问法务系统,这是合规的基本要求。但超级 Agent 拥有所有工具,意味着它在进行财务审批时也能访问法务数据——这直接违反了合规性原则。
问题三:无法并行执行。采购流程中,财务审批和法务审核本来可以同时进行。但单个 Agent 只能串行执行,无法同时处理多个任务,只能一步一步来。
老张意识到:单一 Agent 并非万能解决方案,复杂场景需要多 Agent 协同工作。
---Orchestrator-Workers 模式:协调者与专家团队
老张开始研究多 Agent 协作模式,参考了 Anthropic 官方文档和开源项目实践,发现了 Orchestrator-Workers(协调者-执行者) 模式。
核心思想很简洁:与其让一个超级 Agent 包揽所有工作,不如设置一个“协调者”负责任务分配,多个“专家”负责具体执行。
``` ┌─────────────────────────┐ │ Orchestrator │ │ (采购流程协调者) │ │ │ │ 职责: │ │ - 分析任务 │ │ - 分配子任务 │ │ - 整合结果 │ └───────────┬─────────────┘ │ ┌───────────┼───────────┐ │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │财务Agent │ │法务Agent │ │IT Agent │ │(财务专家)│ │(法务专家)│ │(IT专家) │ │ │ │ │ │ │ │工具: │ │工具: │ │工具: │ │-查预算 │ │-查合同 │ │-配置账号 │ │-审批 │ │-审核 │ │-开通权限 │ └─────┬────┘ └─────┬────┘ └─────┬────┘ │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │财务系统 │ │合同数据库│ │IT 系统 │ │(工具作用域)│(工具作用域)│(工具作用域)│ └──────────┘ └──────────┘ └──────────┘ ```对比单 Agent 与 Orchestrator-Workers 模式,差异一目了然。在上下文层面,单个 Agent 承载所有知识 vs 每个专家只了解自己领域;在权限上,所有工具混杂 vs 每个专家只有自己的工具;在执行上,串行执行 vs 专家可以并行工作;在可维护性上,改动一处影响全局 vs 修改一个专家不影响其他维度。结论明确:简单场景使用单 Agent,复杂场景采用 Orchestrator-Workers。
老张决定,就采用这个模式。
---Orchestrator:协调者的工作方式
Orchestrator 的职责概括为四步:分析用户请求,确定需要哪些专家,分配子任务,最后收集并整合结果。
具体示例:用户提交“处理采购申请 PROJ-2025-001”。Orchestrator 收到请求后,首先分析:这是一个采购流程,涉及财务、法务、IT 三个领域,其中财务和法务可以并行处理,IT 依赖前两者的结果。然后,它分配任务:财务 Agent 审批预算,法务 Agent 审核合同,两者并行执行。待两个结果都返回后,再将 IT 配置任务交给 IT Agent。最后,整合所有结果,生成完整的处理报告。
Orchestrator 的实现思路非常直观。核心逻辑是:先分析任务,得到并行和串行任务列表,再通过异步并发执行并行任务,然后顺序执行串行任务,最后整合所有结果。在 Prompt(提示词)设计上,需要明确告知协调者:你是一个采购流程协调者,你需要分析任务、制定计划、分配任务、整合结果。同时,列出可调用的专家列表和期望的输出格式。
---专家 Agent:角色分工与工具权限隔离
每个专家 Agent 只掌握自己领域的知识,只允许访问自己领域的工具,这是整个架构安全性和可维护性的基础。
财务专家仅负责检查预算是否充足、核对采购金额是否合理,只能访问财务系统,可调用的工具包括查询预算和审批。法务专家仅负责检查合同条款是否合规、核实供应商资质,只能访问法务系统,工具包括查询合同和核实供应商。IT 专家仅负责配置系统账号、开通相关权限,只能访问 IT 系统,工具包括创建账号和开通权限。
这种工具作用域隔离的价值,在多部门协作场景中尤为突出。如果没有隔离,财务 Agent 可以访问所有系统,既不符合合规要求也难以维护。而采用工具作用域隔离后,每个 Agent 仅能访问自己领域的系统,合规性得到保障,修改一个专家也不会影响其他模块。
---并行与串行:任务调度策略
在采购流程中,有些任务可以并行执行,有些则必须串行。
财务审批和法务审核互不依赖,完全可以同时进行。从时间轴上看,财务 Agent 和法务 Agent 可以并行运行,总耗时为 max(财务时间, 法务时间),假设各需10分钟,则总共只需10分钟。而 IT 配置必须等前两者都通过后才能开始,属于串行任务。财务和法务完成后,IT Agent 再启动,总耗时为财务法务时间加上 IT 时间,假设 IT 需要5分钟,则总共15分钟。
如果所有任务都串行执行:财务→法务→IT,总耗时为三者之和,大约20分钟。采用并行+串行混合执行,总耗时从20分钟优化到15分钟,效率提升25%。
调度策略的实现核心是使用异步并发处理并行任务,同时确保串行任务按依赖顺序执行。如果任一并行任务被驳回,流程立即终止,避免后续无效操作。
---Outcome 验证:结果确认机制
Orchestrator 分配任务后,如何确保专家 Agent 的输出符合预期?老张设计了 Outcome 验证循环。
验证流程大致如下:Orchestrator 向专家分配任务,专家执行后生成结果,系统对结果进行验证。若验证通过,则进入整合阶段;若未通过,则触发重试或人工介入。重试次数用尽后,仍由人工最终处理。
验证规则需要根据具体场景定制。以财务审批为例,结果必须包含审批结论;如果通过,必须附带审批金额;如果驳回,必须附带驳回原因。法务审核也类似,必须包含合规结论;如果有风险条款,必须列出明细。通过这种结构化验证,可以捕获90%的输出问题,大幅减少人工介入次数。
---完整系统架构
老张将所有组件整合在一起,形成了完整的多 Agent 协作系统。
整个架构从用户请求层开始,请求进入 Orchestrator。Orchestrator 作为核心协调者,负责分析任务、制定计划、分配子任务、验证结果、整合输出。其下挂接三个专家 Agent:财务 Agent、法务 Agent、IT Agent。每个专家拥有自己的系统提示词(System Prompt)、工具列表、工具作用域和验证规则。专家 Agent 的下层是各自对应的业务系统,彼此隔离访问。最终,所有结果汇聚回 Orchestrator,由其生成最终的完整报告。
底层组件清单如下:
| 组件 | 职责 | 关键技术 | |------|------|---------| | Orchestrator | 分析任务、分配子任务、整合结果 | 任务调度、结果验证 | | 财务 Agent | 预算审批、财务核对 | 工具作用域隔离 | | 法务 Agent | 合同审核、风险评估 | 工具作用域隔离 | | IT Agent | 系统配置、权限开通 | 工具作用域隔离 | | 验证循环 | 确保输出符合预期 | Outcome 验证 | ---实际效果:从 3 天到 15 分钟
系统上线后,采购流程的效率实现了质的飞跃。
之前:采购申请提交后,等待财务审批需要人工通知,花 1 天;再等待法务审核,又是人工通知,花 1 天;再等待 IT 配置,还是人工通知,花 1 天。整个流程跑下来,总耗时 3 天。大部分时间都是在“等待通知”中白白流逝。
之后:采购申请提交,财务 Agent 和法务 Agent 并行处理,总共只需要 10 分钟;接着 IT Agent 串行处理,需要 5 分钟。总耗时压缩到了 15 分钟。
效率对比非常直观:平均耗时从 3 天(含人工协调等待)降至 15 分钟,提升幅度高达 99%;人工介入频率从每次跨部门交接都需人工,降至仅异常情况才需介入,减少 90%;错误率也从 5% 降至 0.5%,下降 90%;可追溯性从依赖人工记录变为完整日志,实现 100% 追踪覆盖。
王总看到数据,很满意:“这个系统,可以推广到其他流程。”
---老张的经验总结
回顾整个进阶之路,老张总结了多 Agent 协作的核心原则。
职责分离:每个 Agent 只专注于自己擅长的领域。工具隔离:每个 Agent 仅能访问自己领域的工具。并行优先:能并行的任务尽量并行执行。验证循环:结果需验证,不通过则重试或人工介入。协调者统筹:由 Orchestrator 统一负责分析与整合。
在关键技术点上,Orchestrator-Workers 模式解决了单 Agent 能力有限的问题,适用于复杂多领域任务。专家 Agent 分工解决了上下文爆炸和权限混乱的问题,适用于多部门协作场景。工具作用域隔离解决了权限安全和合规要求的问题,是企业级部署的必备条件。并行任务调度用于提升效率,适用于独立子任务场景。Outcome 验证则用于保证输出质量,适用于高可靠性要求。
从单 Agent 到多 Agent 协作,不仅仅是数量上的增加,更是架构层面的升级。
回顾整个系列:入门篇,老张学到了用 Agent SDK 构建一个能运行的 Demo;进阶篇,他为 Agent 添加了记忆、审计、定制化能力;生产篇,他使 Agent 在生产环境稳定运行;扩展篇,他实现了多个 Agent 协作。现在,老张的系统已经是一个完整的企业级 Agent 平台了。
---系列总结:老张的 Agent 成长之路
老张回顾这三个月,总结了一张清晰的进阶路线表。
入门阶段,解决怎么快速做一个 Demo 的问题,核心技术是 Claude Agent SDK、query()、WebSearch。进阶阶段,解决怎么让 Demo 变成可靠产品的问题,核心技术是 CLAUDE.md、Hooks、Output Styles、Subagent。生产阶段,解决怎么让产品稳定运行的问题,核心技术是 Managed Agents、Session、Environment、Human-in-the-loop。扩展阶段,解决怎么让多个 Agent 协作的问题,核心技术是 Orchestrator-Workers、工具作用域、验证循环。
Agent 开发的心法其实很简单:从简单开始,先用 query() 将系统跑起来,验证可行性。然后逐步迭代,每一步解决一个具体问题,不要试图一步到位。保持生产化思维,Demo 只是起点,稳定性、安全性、可维护性才是最终目标。最后,协作优先,复杂场景下避免让一个 Agent 承担所有责任。
老张的最后一句话,或许是最好的总结:“Agent 不是万能的,但用好 Agent,可以解放很多重复劳动。关键是要知道什么时候用单 Agent,什么时候用多 Agent,什么时候不用 Agent。”
---附录:关键技术实现思路
以下伪代码展示了 Orchestrator-Workers 模式的核心实现,用于阐释核心思路。
Orchestrator 实现
```python class Orchestrator: def __init__(self, workers: dict): self.workers = workers async def process(self, request: str): # 分析任务 plan = await self.analyze(request) # plan = { # "parallel_tasks": [ # {"worker": "finance_agent", "task": "审批预算"}, # {"worker": "legal_agent", "task": "审核合同"} # ], # "serial_tasks": [ # {"worker": "it_agent", "task": "配置系统"} # ] # } # 并行执行 parallel_results = await asyncio.gather(*[ self.dispatch(t["worker"], t["task"]) for t in plan["parallel_tasks"] ]) # 验证并行结果 for result in parallel_results: if not self.validate(result): return {"status": "failed", "reason": "并行任务验证失败"} # 串行执行 serial_results = [] for t in plan["serial_tasks"]: result = await self.dispatch(t["worker"], t["task"]) if not self.validate(result): return {"status": "failed", "reason": "串行任务验证失败"} serial_results.append(result) # 整合结果 return self.synthesize(parallel_results + serial_results) async def dispatch(self, worker_name: str, task: str): worker = self.workers[worker_name] return await worker.execute(task) def validate(self, result): # 验证结果是否符合预期 return result.get("status") == "success" def synthesize(self, results): # 整合所有结果 return { "status": "success", "details": results } ```专家 Agent 实现
```python # 财务专家 finance_agent = Agent( name="finance_agent", system_prompt=""" 你是一个财务审批专家,只能访问财务系统。 审批通过时,返回 {"status": "success", "decision": "approved", "amount": xxx} 审批驳回时,返回 {"status": "success", "decision": "rejected", "reason": "xxx"} """, tools=[check_budget, approve_request], allowed_resources=["finance_db"], ) # 法务专家 legal_agent = Agent( name="legal_agent", system_prompt=""" 你是一个法务审核专家,只能访问法务系统。 审核通过时,返回 {"status": "success", "compliance": "passed"} 审核不通过时,返回 {"status": "success", "compliance": "failed", "risks": [...]} """, tools=[check_contract, verify_vendor], allowed_resources=["contract_db"], ) # IT 专家 it_agent = Agent( name="it_agent", system_prompt=""" 你是一个 IT 配置专家,只能访问 IT 系统。 配置完成时,返回 {"status": "success", "account_id": "xxx", "permissions": [...]} """, tools=[create_account, grant_permission], allowed_resources=["it_system"], ) ```工具作用域隔离
```python # 确保 Agent 只能访问自己的工具 class Agent: def __init__(self, tools, allowed_resources): self.tools = tools self.allowed_resources = allowed_resources async def execute(self, task): # 执行前检查工具权限 for tool in self.tools: if tool.resource not in self.allowed_resources: raise PermissionError(f"无权访问 {tool.resource}") # 执行任务 return await self._execute_internal(task) ``` ---老张的 Agent 进阶之路至此告一段落。从两个小时的 Demo 到企业级多 Agent 协作平台,他经历了三个月。但他深知,这只是一个开始。Agent 技术仍在快速发展,新的模式、新的工具不断涌现。保持学习,持续迭代——这就是 Agent 开发的正确态度。
