1. 从一个真实需求说起为什么要做自动化工作流 Agent我在过去一年多的时间里陆续帮几个团队落地过自动化工作流的项目。最开始大家的诉求都很朴素——把重复的、跨系统的、需要人工盯着的事情交给机器去做。但真正动手之后你会发现单纯的脚本编排、定时任务、甚至传统的 RPA都只能解决“流程固定、输入稳定、异常可控”的那一类问题。一旦流程里出现需要判断、需要查资料、需要根据上下文动态决定下一步的场景传统方案就开始力不从心。这就是自动化工作流 Agent 要解决的核心问题。它不是简单的“把几个 API 串起来”而是让一个具备推理能力的智能体在编排框架的约束下自主决定调用哪些工具、按什么顺序调用、遇到异常怎么回退。关键词里的自动化工作流、Agent、MCP、多智能体、编排其实正好勾勒出了这个案例的五个技术支柱工作流是骨架Agent 是大脑MCP 是手脚工具接入协议多智能体是分工协作编排是调度中枢。这篇文章适合三类人看。第一类是有一定开发基础、想从零搭一个 Agent 工作流的工程师第二类是在业务侧被重复流程折磨、想搞清楚 Agent 到底能干什么的产品或运营同学第三类是对 MCP、多智能体这些概念听过但没真正跑通过想找一个完整案例照着复现的学习者。我会尽量把每一步的“为什么”讲清楚而不是只丢一段代码让你抄。需要先说明一点这个案例我采用的是“单主控 Agent 多子 Agent MCP 工具层”的架构这是目前工程上比较稳妥、也比较好调试的一种组合。市面上也有纯多智能体平权协作的方案但那种在真实业务里调试成本很高后面我会专门讲为什么。2. 整体架构设计与方案选型拆解2.1 为什么是“编排 Agent”而不是纯 Agent很多人一上来就想做一个“全自主”的 Agent给它一个目标让它自己规划、自己执行、自己反思。理想很丰满但实际跑起来你会发现两个致命问题一是不可控同样的输入两次跑出来的路径可能完全不同业务方没法接受二是不可观测出了问题你根本不知道它卡在哪一步。所以我采用的是编排为主、Agent 为辅的思路。编排层负责定义整个工作流的骨架——有哪些阶段、阶段之间的依赖关系、每个阶段的输入输出契约、失败重试策略。Agent 则负责填充骨架里的“智能节点”也就是那些需要推理和动态决策的环节。这样既保留了流程的可控性和可观测性又让需要智能的地方真正智能起来。打个比方编排就像是一条生产线的传送带和工位划分Agent 则是站在工位上的工人。传送带决定了物料怎么流、在哪停工人决定这个工位上具体怎么干活。你不会让工人自己决定整条生产线怎么排布但你也不会把工人换成只会做固定动作的机械臂。2.2 MCP 在架构里扮演什么角色MCP 这个词最近热度很高但很多人对它的理解还停留在“又一个工具调用协议”。我的理解是MCP 的价值在于把工具接入这件事标准化了。在没有 MCP 之前你每接一个外部能力读数据库、调搜索、操作文件、访问某个 SaaS都要写一套适配代码参数格式、错误处理、鉴权方式各不相同。Agent 想用这些工具就得为每个工具单独写提示词和解析逻辑。MCP 把这些统一成了“资源Resource 工具Tool 提示Prompt”三件套。Agent 只需要知道“有一个工具叫 xxx它接受这些参数返回这种结构”剩下的接入细节由 MCP Server 屏蔽掉。这意味着你换一个同类工具Agent 侧几乎不用改。在这个案例里我把所有外部能力都封装成了 MCP Server主控 Agent 通过 MCP 客户端统一调用。提示MCP 不是银弹。对于非常简单、一次性的工具调用直接写函数可能更快。MCP 的收益在工具数量多、需要复用、需要跨 Agent 共享的时候才明显。2.3 多智能体的分工原则多智能体最容易踩的坑就是“为了多而多”。我见过有人把一个大 Agent 硬拆成五个结果五个 Agent 之间来回传话token 消耗翻了三倍效果还不如一个。我的分工原则是按能力边界拆不按流程步骤拆。具体到这个案例我拆了三个子 Agent一个负责信息检索与整理Retriever Agent一个负责内容生成与加工Writer Agent一个负责校验与纠错Reviewer Agent。它们各自有独立的系统提示词、独立的工具集、独立的上下文窗口。主控 Agent 只负责决定“现在该谁上场、给它什么输入、拿到输出后下一步干什么”。这样拆的好处是每个子 Agent 的职责单一提示词可以写得很聚焦调试的时候也容易定位问题。坏处是主控 Agent 的调度逻辑会复杂一些需要处理好子 Agent 之间的数据传递格式。2.4 方案对比几种常见架构的取舍架构方案可控性调试难度适用场景我的评价纯脚本编排极高低流程固定、无判断简单场景够用遇到动态决策就废单 Agent 全自主低高探索性任务演示好看生产难用编排 单 Agent高中大部分业务流性价比最高推荐起步编排 多 Agent中高中高复杂多阶段任务本案例采用需控制 Agent 数量纯多 Agent 平权低极高研究性质生产环境慎用这张表是我踩过坑之后总结的。新手我建议从“编排 单 Agent”起步跑通了再往多 Agent 演进。直接上多 Agent很容易在调度逻辑里迷失。3. 核心细节解析与实操要点3.1 工作流的状态管理怎么做工作流跑起来之后最核心的问题就是状态。每一步的输入、输出、中间结果、错误信息都需要有个地方存。我的做法是定义一个统一的WorkflowState结构所有节点读写都通过它。from dataclasses import dataclass, field from typing import Any dataclass class WorkflowState: task_id: str goal: str context: dict field(default_factorydict) artifacts: dict field(default_factorydict) history: list field(default_factorylist) errors: list field(default_factorylist) current_stage: str init这个结构看起来简单但有几个设计考量。context存的是全局共享的上下文比如用户原始需求、配置参数artifacts存的是各阶段产出的中间结果比如检索到的文档、生成的草稿history是执行轨迹用于回溯和调试errors单独拎出来方便做失败分析和重试。注意不要把大对象比如几 MB 的文档全文直接塞进 state 然后到处传。我的做法是 artifacts 里只存引用比如文件路径或对象存储的 key真正的内容按需读取。否则上下文窗口很快就被撑爆。3.2 子 Agent 的提示词怎么写才稳子 Agent 的提示词是这个案例里最花时间的部分。我总结了一个“四段式”结构角色定义、能力边界、输入输出契约、异常处理。角色定义要具体到“你是谁、你擅长什么”不要写“你是一个有用的助手”这种废话。能力边界要明确告诉它“你能做什么、不能做什么”尤其是不能做什么这能大幅减少它乱调工具的情况。输入输出契约要规定格式最好给出示例。异常处理要告诉它“信息不足时怎么办、工具报错时怎么办”。以 Retriever Agent 为例它的提示词大致是这样的你是信息检索专家擅长从多个来源定位并整理与目标相关的资料。 你的能力 - 调用 search 工具进行关键词检索 - 调用 fetch 工具获取指定 URL 的正文 - 对检索结果做去重、相关性排序、摘要 你不能 - 编造未检索到的信息 - 对检索结果做主观评价 输入一个明确的信息需求描述 输出JSON 格式包含 sources 数组和 summary 字段 如果检索结果为空返回 {sources: [], summary: 未找到相关信息} 不要尝试用你的知识补充。这套结构跑下来子 Agent 的稳定性明显比“一句话提示词”高很多。3.3 MCP 工具的封装规范把外部能力封装成 MCP Server 时我遵循几个规范。第一工具名用动词开头语义清晰比如search_web、read_file、query_db不要用tool1、helper这种。第二参数用 JSON Schema 严格定义必填项和可选项分清楚类型写明确。第三返回值统一结构成功返回{ok: true, data: ...}失败返回{ok: false, error: ...}这样 Agent 侧处理起来逻辑一致。{ name: search_web, description: 根据关键词检索网络信息返回标题、摘要和链接列表, inputSchema: { type: object, properties: { query: {type: string, description: 检索关键词}, limit: {type: integer, default: 5, description: 返回结果数量上限} }, required: [query] } }提示工具描述description是给 Agent 看的不是给人看的。要写得让 Agent 一眼明白“什么时候该用这个工具”。我见过有人把描述写成技术文档Agent 根本不知道啥时候调。3.4 编排层的调度逻辑编排层的核心是一个状态机。每个阶段定义三个东西进入条件、执行逻辑、退出条件。主控 Agent 在每个阶段开始时根据当前 state 决定调用哪个子 Agent 或哪个工具。STAGES [retrieve, draft, review, finalize] def run_workflow(state: WorkflowState): while state.current_stage ! done: stage state.current_stage if stage retrieve: state retriever_agent.run(state) state.current_stage draft elif stage draft: state writer_agent.run(state) state.current_stage review elif stage review: state reviewer_agent.run(state) if state.context.get(need_revision): state.current_stage draft else: state.current_stage finalize elif stage finalize: state finalize(state) state.current_stage done return state这里有个关键设计review 阶段可以回退到 draft形成循环。但要设置最大循环次数否则可能死循环。我一般设 3 次超过就强制进入 finalize 并标记“未通过审核”。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把基础环境搭起来。我用的是 Python 3.11主要依赖包括 Agent 框架、MCP 客户端库、以及几个工具库。python -m venv venv source venv/bin/activate pip install mcp anthropic pydantic httpx选 Python 是因为生态成熟、调试方便。如果你追求性能Rust 也有 Agent 框架但开发效率会低不少除非你的场景对延迟极其敏感否则没必要。4.2 搭建第一个 MCP Server先做一个最简单的 MCP Server提供一个search_web工具。这里用官方 SDK 的写法from mcp.server import Server from mcp.types import Tool, TextContent import httpx app Server(demo-tools) app.list_tools() async def list_tools(): return [ Tool( namesearch_web, description根据关键词检索网络信息, inputSchema{ type: object, properties: {query: {type: string}}, required: [query] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name search_web: query arguments[query] async with httpx.AsyncClient() as client: resp await client.get(https://api.example.com/search, params{q: query}) data resp.json() return [TextContent(typetext, textstr(data))]跑起来之后用 MCP 客户端连上去列出工具调用一次确认链路通了。这一步别嫌麻烦工具层不通后面全白搭。4.3 主控 Agent 的实现主控 Agent 的职责是调度不是干活。它的提示词要写清楚“你有哪几个子 Agent 可以调用、每个子 Agent 负责什么、什么情况下调谁”。MAIN_PROMPT 你是工作流调度器。你有三个子 Agent 可以调用 1. retriever负责信息检索输入是信息需求输出是资料列表 2. writer负责内容生成输入是资料列表和目标输出是草稿 3. reviewer负责校验输入是草稿输出是审核意见 你的任务是根据当前工作流状态决定下一步调用哪个子 Agent。 只输出 JSON{next: retriever|writer|reviewer|done, reason: ...} 主控 Agent 本身不直接调工具它只做决策。真正的工具调用发生在子 Agent 内部。这样分层的好处是主控的逻辑非常轻token 消耗小而且容易测试。4.4 子 Agent 与 MCP 工具的对接子 Agent 需要能调用 MCP 工具。我的做法是给每个子 Agent 配一个 MCP 客户端在它的执行循环里把可用工具列表注入到提示词里然后解析它的工具调用请求。async def run_agent(agent_config, state, mcp_client): tools await mcp_client.list_tools() messages build_messages(agent_config, state, tools) while True: response await llm.chat(messages) if response.stop_reason tool_use: tool_name response.tool_name tool_args response.tool_args result await mcp_client.call_tool(tool_name, tool_args) messages.append({role: tool, content: result}) else: break return parse_output(response.content)这个循环是 Agent 的核心。它不断问模型“你要调工具吗”要就调调完把结果喂回去直到模型说“我不调了这是我的最终输出”。4.5 完整跑通一次工作流把上面几块拼起来跑一次完整流程。输入一个任务比如“整理一份关于自动化工作流 Agent 的技术综述”。观察日志看它怎么走retriever 先检索writer 根据检索结果写草稿reviewer 审核如果审核不通过就回退重写。我实测下来一个中等复杂度的任务整个流程大概消耗 3 到 5 万 token耗时 1 到 3 分钟。这个成本在可接受范围内。如果发现某一步特别慢或特别贵就针对性优化——通常是提示词太长或者工具返回的数据太大。注意第一次跑通不代表稳定。我建议至少跑 20 次不同的输入统计成功率和平均耗时才能判断这套工作流是否真的可用。5. 常见问题与排查技巧实录5.1 Agent 不调工具直接编答案这是最常见的问题。原因通常是提示词里没强调“必须基于工具返回的信息”或者工具描述写得太模糊Agent 觉得“我自己知道不用调”。解决办法有三个。第一在系统提示词里明确写“禁止使用你的内部知识回答所有事实必须来自工具返回”。第二把工具描述写得更具体让 Agent 清楚这个工具能解决什么问题。第三在输出格式里要求它标注信息来源没有来源的答案直接判为无效。5.2 工具调用参数格式错误Agent 有时候会把参数拼错比如该传字符串的传了数字该传数组的传了单个值。这通常是 JSON Schema 定义不够严格或者提示词里没给示例。我的做法是在工具描述里附上一个调用示例并且在 Agent 侧做参数校验格式不对就返回错误信息让它重试。重试两次还不对就降级处理或报错。5.3 多 Agent 之间数据传递丢失子 Agent 之间传数据时最容易丢字段。比如 retriever 返回的 sources 数组writer 拿到后只用了 summary把 sources 丢了导致 reviewer 没法核对来源。解决办法是定义严格的数据契约每个子 Agent 的输入输出都用 Pydantic 模型校验。传之前校验一次收之后校验一次不通过就报错别让它悄悄丢。5.4 工作流陷入死循环review 和 draft 之间来回跳跳了十几次还没收敛。这通常是因为 reviewer 的审核标准太严或者 writer 一直改不对。我的处理是设置最大循环次数一般 3 次超过就强制退出并标记。同时分析日志看是审核标准问题还是生成质量问题针对性调整。5.5 常见问题速查表问题现象可能原因排查方向解决手段Agent 不调工具提示词未强制、工具描述模糊看提示词和工具定义强化约束、补充示例参数格式错误Schema 不严、无示例看调用日志加校验、加重试数据传递丢失无契约、字段未校验看各阶段输入输出用 Pydantic 校验死循环审核标准严、生成质量差看循环次数和内容设上限、调标准响应慢提示词长、返回数据大看 token 消耗精简提示词、截断数据成本高循环多、模型选大看调用次数换小模型、减少循环5.6 几个我踩过的坑第一个坑是过早引入多 Agent。我一开始就拆了五个 Agent结果调度逻辑写了一堆效果还不如单 Agent。后来砍到三个反而更稳。教训是能一个 Agent 解决的别拆两个。第二个坑是工具返回数据不截断。有个工具返回了几万字的文档直接塞进上下文token 瞬间爆掉。后来我在 MCP Server 侧就做了截断和摘要只返回关键部分。第三个坑是忽略错误处理。早期版本工具报错就直接崩整个工作流挂掉。后来加了重试和降级工具报错时 Agent 可以选择换一个工具或跳过鲁棒性好了很多。第四个坑是没有可观测性。出问题的时候两眼一抹黑不知道卡在哪。后来加了完整的日志和 trace每一步的输入输出都记下来排查效率提升巨大。6. 性能与成本优化的实战经验6.1 模型选型的取舍不是所有节点都需要用最强的模型。我的做法是分级主控 Agent 用中等模型决策不复杂retriever 用便宜模型主要是调工具和整理writer 用强模型生成质量关键reviewer 用中等模型审核相对简单。这样搭配下来成本比全用强模型低一半以上效果几乎没差别。关键是你要清楚每个节点的核心诉求是什么。6.2 上下文窗口的管理上下文是稀缺资源。我的策略是只保留最近 N 轮对话更早的做摘要压缩工具返回的大数据只保留摘要原文存到外部子 Agent 之间传递只传必要字段不传整个 state。def compress_history(history, max_turns5): if len(history) max_turns: return history old history[:-max_turns] recent history[-max_turns:] summary summarize(old) return [{role: system, content: f历史摘要{summary}}] recent6.3 并发场景下的处理如果工作流要扛并发有几个点要注意。第一MCP 客户端要支持连接池别每次调用都新建连接。第二子 Agent 之间如果无依赖可以并行执行。第三状态存储要用支持并发的后端别用本地文件。我实测过单机跑 10 个并发工作流只要工具层不拖后腿基本没问题。再往上就要考虑分布式了那是另一个话题。7. 后续可以怎么扩展这套架构跑通之后扩展方向其实很多。比如把 MCP 工具层做成可插拔的不同业务场景挂不同的工具集比如把子 Agent 做成可配置的通过配置文件定义有哪些 Agent、各自用什么提示词和工具比如加一个“学习”环节把每次执行的成功案例存下来作为后续的参考。我个人在实际操作中的体会是Agent 工作流这个东西架构设计占三成提示词工程占三成剩下四成全是调试和踩坑。别指望一次写对做好反复迭代的准备。另外可观测性一定要从第一天就做不然出了问题你会非常痛苦。最后分享一个小技巧每次改动提示词或工具定义后用同一批测试用例跑一遍对比成功率和耗时。这样你能清楚知道这次改动是变好了还是变差了而不是凭感觉。我维护了一个 30 条的测试集每次改动都跑省了很多返工的时间。 
长河君 · 2026/10/6 5:14:33 · 阅读 1.2k 从零搭建自动化工作流Agent：MCP与多智能体编排实战
1. 从一个真实需求说起为什么要做自动化工作流 Agent我在过去一年多的时间里陆续帮几个团队落地过自动化工作流的项目。最开始大家的诉求都很朴素——把重复的、跨系统的、需要人工盯着的事情交给机器去做。但真正动手之后你会发现单纯的脚本编排、定时任务、甚至传统的 RPA都只能解决“流程固定、输入稳定、异常可控”的那一类问题。一旦流程里出现需要判断、需要查资料、需要根据上下文动态决定下一步的场景传统方案就开始力不从心。这就是自动化工作流 Agent 要解决的核心问题。它不是简单的“把几个 API 串起来”而是让一个具备推理能力的智能体在编排框架的约束下自主决定调用哪些工具、按什么顺序调用、遇到异常怎么回退。关键词里的自动化工作流、Agent、MCP、多智能体、编排其实正好勾勒出了这个案例的五个技术支柱工作流是骨架Agent 是大脑MCP 是手脚工具接入协议多智能体是分工协作编排是调度中枢。这篇文章适合三类人看。第一类是有一定开发基础、想从零搭一个 Agent 工作流的工程师第二类是在业务侧被重复流程折磨、想搞清楚 Agent 到底能干什么的产品或运营同学第三类是对 MCP、多智能体这些概念听过但没真正跑通过想找一个完整案例照着复现的学习者。我会尽量把每一步的“为什么”讲清楚而不是只丢一段代码让你抄。需要先说明一点这个案例我采用的是“单主控 Agent 多子 Agent MCP 工具层”的架构这是目前工程上比较稳妥、也比较好调试的一种组合。市面上也有纯多智能体平权协作的方案但那种在真实业务里调试成本很高后面我会专门讲为什么。2. 整体架构设计与方案选型拆解2.1 为什么是“编排 Agent”而不是纯 Agent很多人一上来就想做一个“全自主”的 Agent给它一个目标让它自己规划、自己执行、自己反思。理想很丰满但实际跑起来你会发现两个致命问题一是不可控同样的输入两次跑出来的路径可能完全不同业务方没法接受二是不可观测出了问题你根本不知道它卡在哪一步。所以我采用的是编排为主、Agent 为辅的思路。编排层负责定义整个工作流的骨架——有哪些阶段、阶段之间的依赖关系、每个阶段的输入输出契约、失败重试策略。Agent 则负责填充骨架里的“智能节点”也就是那些需要推理和动态决策的环节。这样既保留了流程的可控性和可观测性又让需要智能的地方真正智能起来。打个比方编排就像是一条生产线的传送带和工位划分Agent 则是站在工位上的工人。传送带决定了物料怎么流、在哪停工人决定这个工位上具体怎么干活。你不会让工人自己决定整条生产线怎么排布但你也不会把工人换成只会做固定动作的机械臂。2.2 MCP 在架构里扮演什么角色MCP 这个词最近热度很高但很多人对它的理解还停留在“又一个工具调用协议”。我的理解是MCP 的价值在于把工具接入这件事标准化了。在没有 MCP 之前你每接一个外部能力读数据库、调搜索、操作文件、访问某个 SaaS都要写一套适配代码参数格式、错误处理、鉴权方式各不相同。Agent 想用这些工具就得为每个工具单独写提示词和解析逻辑。MCP 把这些统一成了“资源Resource 工具Tool 提示Prompt”三件套。Agent 只需要知道“有一个工具叫 xxx它接受这些参数返回这种结构”剩下的接入细节由 MCP Server 屏蔽掉。这意味着你换一个同类工具Agent 侧几乎不用改。在这个案例里我把所有外部能力都封装成了 MCP Server主控 Agent 通过 MCP 客户端统一调用。提示MCP 不是银弹。对于非常简单、一次性的工具调用直接写函数可能更快。MCP 的收益在工具数量多、需要复用、需要跨 Agent 共享的时候才明显。2.3 多智能体的分工原则多智能体最容易踩的坑就是“为了多而多”。我见过有人把一个大 Agent 硬拆成五个结果五个 Agent 之间来回传话token 消耗翻了三倍效果还不如一个。我的分工原则是按能力边界拆不按流程步骤拆。具体到这个案例我拆了三个子 Agent一个负责信息检索与整理Retriever Agent一个负责内容生成与加工Writer Agent一个负责校验与纠错Reviewer Agent。它们各自有独立的系统提示词、独立的工具集、独立的上下文窗口。主控 Agent 只负责决定“现在该谁上场、给它什么输入、拿到输出后下一步干什么”。这样拆的好处是每个子 Agent 的职责单一提示词可以写得很聚焦调试的时候也容易定位问题。坏处是主控 Agent 的调度逻辑会复杂一些需要处理好子 Agent 之间的数据传递格式。2.4 方案对比几种常见架构的取舍架构方案可控性调试难度适用场景我的评价纯脚本编排极高低流程固定、无判断简单场景够用遇到动态决策就废单 Agent 全自主低高探索性任务演示好看生产难用编排 单 Agent高中大部分业务流性价比最高推荐起步编排 多 Agent中高中高复杂多阶段任务本案例采用需控制 Agent 数量纯多 Agent 平权低极高研究性质生产环境慎用这张表是我踩过坑之后总结的。新手我建议从“编排 单 Agent”起步跑通了再往多 Agent 演进。直接上多 Agent很容易在调度逻辑里迷失。3. 核心细节解析与实操要点3.1 工作流的状态管理怎么做工作流跑起来之后最核心的问题就是状态。每一步的输入、输出、中间结果、错误信息都需要有个地方存。我的做法是定义一个统一的WorkflowState结构所有节点读写都通过它。from dataclasses import dataclass, field from typing import Any dataclass class WorkflowState: task_id: str goal: str context: dict field(default_factorydict) artifacts: dict field(default_factorydict) history: list field(default_factorylist) errors: list field(default_factorylist) current_stage: str init这个结构看起来简单但有几个设计考量。context存的是全局共享的上下文比如用户原始需求、配置参数artifacts存的是各阶段产出的中间结果比如检索到的文档、生成的草稿history是执行轨迹用于回溯和调试errors单独拎出来方便做失败分析和重试。注意不要把大对象比如几 MB 的文档全文直接塞进 state 然后到处传。我的做法是 artifacts 里只存引用比如文件路径或对象存储的 key真正的内容按需读取。否则上下文窗口很快就被撑爆。3.2 子 Agent 的提示词怎么写才稳子 Agent 的提示词是这个案例里最花时间的部分。我总结了一个“四段式”结构角色定义、能力边界、输入输出契约、异常处理。角色定义要具体到“你是谁、你擅长什么”不要写“你是一个有用的助手”这种废话。能力边界要明确告诉它“你能做什么、不能做什么”尤其是不能做什么这能大幅减少它乱调工具的情况。输入输出契约要规定格式最好给出示例。异常处理要告诉它“信息不足时怎么办、工具报错时怎么办”。以 Retriever Agent 为例它的提示词大致是这样的你是信息检索专家擅长从多个来源定位并整理与目标相关的资料。 你的能力 - 调用 search 工具进行关键词检索 - 调用 fetch 工具获取指定 URL 的正文 - 对检索结果做去重、相关性排序、摘要 你不能 - 编造未检索到的信息 - 对检索结果做主观评价 输入一个明确的信息需求描述 输出JSON 格式包含 sources 数组和 summary 字段 如果检索结果为空返回 {sources: [], summary: 未找到相关信息} 不要尝试用你的知识补充。这套结构跑下来子 Agent 的稳定性明显比“一句话提示词”高很多。3.3 MCP 工具的封装规范把外部能力封装成 MCP Server 时我遵循几个规范。第一工具名用动词开头语义清晰比如search_web、read_file、query_db不要用tool1、helper这种。第二参数用 JSON Schema 严格定义必填项和可选项分清楚类型写明确。第三返回值统一结构成功返回{ok: true, data: ...}失败返回{ok: false, error: ...}这样 Agent 侧处理起来逻辑一致。{ name: search_web, description: 根据关键词检索网络信息返回标题、摘要和链接列表, inputSchema: { type: object, properties: { query: {type: string, description: 检索关键词}, limit: {type: integer, default: 5, description: 返回结果数量上限} }, required: [query] } }提示工具描述description是给 Agent 看的不是给人看的。要写得让 Agent 一眼明白“什么时候该用这个工具”。我见过有人把描述写成技术文档Agent 根本不知道啥时候调。3.4 编排层的调度逻辑编排层的核心是一个状态机。每个阶段定义三个东西进入条件、执行逻辑、退出条件。主控 Agent 在每个阶段开始时根据当前 state 决定调用哪个子 Agent 或哪个工具。STAGES [retrieve, draft, review, finalize] def run_workflow(state: WorkflowState): while state.current_stage ! done: stage state.current_stage if stage retrieve: state retriever_agent.run(state) state.current_stage draft elif stage draft: state writer_agent.run(state) state.current_stage review elif stage review: state reviewer_agent.run(state) if state.context.get(need_revision): state.current_stage draft else: state.current_stage finalize elif stage finalize: state finalize(state) state.current_stage done return state这里有个关键设计review 阶段可以回退到 draft形成循环。但要设置最大循环次数否则可能死循环。我一般设 3 次超过就强制进入 finalize 并标记“未通过审核”。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把基础环境搭起来。我用的是 Python 3.11主要依赖包括 Agent 框架、MCP 客户端库、以及几个工具库。python -m venv venv source venv/bin/activate pip install mcp anthropic pydantic httpx选 Python 是因为生态成熟、调试方便。如果你追求性能Rust 也有 Agent 框架但开发效率会低不少除非你的场景对延迟极其敏感否则没必要。4.2 搭建第一个 MCP Server先做一个最简单的 MCP Server提供一个search_web工具。这里用官方 SDK 的写法from mcp.server import Server from mcp.types import Tool, TextContent import httpx app Server(demo-tools) app.list_tools() async def list_tools(): return [ Tool( namesearch_web, description根据关键词检索网络信息, inputSchema{ type: object, properties: {query: {type: string}}, required: [query] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name search_web: query arguments[query] async with httpx.AsyncClient() as client: resp await client.get(https://api.example.com/search, params{q: query}) data resp.json() return [TextContent(typetext, textstr(data))]跑起来之后用 MCP 客户端连上去列出工具调用一次确认链路通了。这一步别嫌麻烦工具层不通后面全白搭。4.3 主控 Agent 的实现主控 Agent 的职责是调度不是干活。它的提示词要写清楚“你有哪几个子 Agent 可以调用、每个子 Agent 负责什么、什么情况下调谁”。MAIN_PROMPT 你是工作流调度器。你有三个子 Agent 可以调用 1. retriever负责信息检索输入是信息需求输出是资料列表 2. writer负责内容生成输入是资料列表和目标输出是草稿 3. reviewer负责校验输入是草稿输出是审核意见 你的任务是根据当前工作流状态决定下一步调用哪个子 Agent。 只输出 JSON{next: retriever|writer|reviewer|done, reason: ...} 主控 Agent 本身不直接调工具它只做决策。真正的工具调用发生在子 Agent 内部。这样分层的好处是主控的逻辑非常轻token 消耗小而且容易测试。4.4 子 Agent 与 MCP 工具的对接子 Agent 需要能调用 MCP 工具。我的做法是给每个子 Agent 配一个 MCP 客户端在它的执行循环里把可用工具列表注入到提示词里然后解析它的工具调用请求。async def run_agent(agent_config, state, mcp_client): tools await mcp_client.list_tools() messages build_messages(agent_config, state, tools) while True: response await llm.chat(messages) if response.stop_reason tool_use: tool_name response.tool_name tool_args response.tool_args result await mcp_client.call_tool(tool_name, tool_args) messages.append({role: tool, content: result}) else: break return parse_output(response.content)这个循环是 Agent 的核心。它不断问模型“你要调工具吗”要就调调完把结果喂回去直到模型说“我不调了这是我的最终输出”。4.5 完整跑通一次工作流把上面几块拼起来跑一次完整流程。输入一个任务比如“整理一份关于自动化工作流 Agent 的技术综述”。观察日志看它怎么走retriever 先检索writer 根据检索结果写草稿reviewer 审核如果审核不通过就回退重写。我实测下来一个中等复杂度的任务整个流程大概消耗 3 到 5 万 token耗时 1 到 3 分钟。这个成本在可接受范围内。如果发现某一步特别慢或特别贵就针对性优化——通常是提示词太长或者工具返回的数据太大。注意第一次跑通不代表稳定。我建议至少跑 20 次不同的输入统计成功率和平均耗时才能判断这套工作流是否真的可用。5. 常见问题与排查技巧实录5.1 Agent 不调工具直接编答案这是最常见的问题。原因通常是提示词里没强调“必须基于工具返回的信息”或者工具描述写得太模糊Agent 觉得“我自己知道不用调”。解决办法有三个。第一在系统提示词里明确写“禁止使用你的内部知识回答所有事实必须来自工具返回”。第二把工具描述写得更具体让 Agent 清楚这个工具能解决什么问题。第三在输出格式里要求它标注信息来源没有来源的答案直接判为无效。5.2 工具调用参数格式错误Agent 有时候会把参数拼错比如该传字符串的传了数字该传数组的传了单个值。这通常是 JSON Schema 定义不够严格或者提示词里没给示例。我的做法是在工具描述里附上一个调用示例并且在 Agent 侧做参数校验格式不对就返回错误信息让它重试。重试两次还不对就降级处理或报错。5.3 多 Agent 之间数据传递丢失子 Agent 之间传数据时最容易丢字段。比如 retriever 返回的 sources 数组writer 拿到后只用了 summary把 sources 丢了导致 reviewer 没法核对来源。解决办法是定义严格的数据契约每个子 Agent 的输入输出都用 Pydantic 模型校验。传之前校验一次收之后校验一次不通过就报错别让它悄悄丢。5.4 工作流陷入死循环review 和 draft 之间来回跳跳了十几次还没收敛。这通常是因为 reviewer 的审核标准太严或者 writer 一直改不对。我的处理是设置最大循环次数一般 3 次超过就强制退出并标记。同时分析日志看是审核标准问题还是生成质量问题针对性调整。5.5 常见问题速查表问题现象可能原因排查方向解决手段Agent 不调工具提示词未强制、工具描述模糊看提示词和工具定义强化约束、补充示例参数格式错误Schema 不严、无示例看调用日志加校验、加重试数据传递丢失无契约、字段未校验看各阶段输入输出用 Pydantic 校验死循环审核标准严、生成质量差看循环次数和内容设上限、调标准响应慢提示词长、返回数据大看 token 消耗精简提示词、截断数据成本高循环多、模型选大看调用次数换小模型、减少循环5.6 几个我踩过的坑第一个坑是过早引入多 Agent。我一开始就拆了五个 Agent结果调度逻辑写了一堆效果还不如单 Agent。后来砍到三个反而更稳。教训是能一个 Agent 解决的别拆两个。第二个坑是工具返回数据不截断。有个工具返回了几万字的文档直接塞进上下文token 瞬间爆掉。后来我在 MCP Server 侧就做了截断和摘要只返回关键部分。第三个坑是忽略错误处理。早期版本工具报错就直接崩整个工作流挂掉。后来加了重试和降级工具报错时 Agent 可以选择换一个工具或跳过鲁棒性好了很多。第四个坑是没有可观测性。出问题的时候两眼一抹黑不知道卡在哪。后来加了完整的日志和 trace每一步的输入输出都记下来排查效率提升巨大。6. 性能与成本优化的实战经验6.1 模型选型的取舍不是所有节点都需要用最强的模型。我的做法是分级主控 Agent 用中等模型决策不复杂retriever 用便宜模型主要是调工具和整理writer 用强模型生成质量关键reviewer 用中等模型审核相对简单。这样搭配下来成本比全用强模型低一半以上效果几乎没差别。关键是你要清楚每个节点的核心诉求是什么。6.2 上下文窗口的管理上下文是稀缺资源。我的策略是只保留最近 N 轮对话更早的做摘要压缩工具返回的大数据只保留摘要原文存到外部子 Agent 之间传递只传必要字段不传整个 state。def compress_history(history, max_turns5): if len(history) max_turns: return history old history[:-max_turns] recent history[-max_turns:] summary summarize(old) return [{role: system, content: f历史摘要{summary}}] recent6.3 并发场景下的处理如果工作流要扛并发有几个点要注意。第一MCP 客户端要支持连接池别每次调用都新建连接。第二子 Agent 之间如果无依赖可以并行执行。第三状态存储要用支持并发的后端别用本地文件。我实测过单机跑 10 个并发工作流只要工具层不拖后腿基本没问题。再往上就要考虑分布式了那是另一个话题。7. 后续可以怎么扩展这套架构跑通之后扩展方向其实很多。比如把 MCP 工具层做成可插拔的不同业务场景挂不同的工具集比如把子 Agent 做成可配置的通过配置文件定义有哪些 Agent、各自用什么提示词和工具比如加一个“学习”环节把每次执行的成功案例存下来作为后续的参考。我个人在实际操作中的体会是Agent 工作流这个东西架构设计占三成提示词工程占三成剩下四成全是调试和踩坑。别指望一次写对做好反复迭代的准备。另外可观测性一定要从第一天就做不然出了问题你会非常痛苦。最后分享一个小技巧每次改动提示词或工具定义后用同一批测试用例跑一遍对比成功率和耗时。这样你能清楚知道这次改动是变好了还是变差了而不是凭感觉。我维护了一个 30 条的测试集每次改动都跑省了很多返工的时间。