☰
从零搭建自动化工作流Agent:MCP与多智能体编排实战
2026/10/6 5:14:27 网站建设 项目流程

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_factory=dict) artifacts: dict = field(default_factory=dict) history: list = field(default_factory=list) errors: list = field(default_factory=list) 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( name="search_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(type="text", text=str(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_turns=5): if len(history) <= max_turns: return history old = history[:-max_turns] recent = history[-max_turns:] summary = summarize(old) return [{"role": "system", "content": f"历史摘要:{summary}"}] + recent

6.3 并发场景下的处理

如果工作流要扛并发,有几个点要注意。第一,MCP 客户端要支持连接池,别每次调用都新建连接。第二,子 Agent 之间如果无依赖,可以并行执行。第三,状态存储要用支持并发的后端,别用本地文件。

我实测过,单机跑 10 个并发工作流,只要工具层不拖后腿,基本没问题。再往上就要考虑分布式了,那是另一个话题。

7. 后续可以怎么扩展

这套架构跑通之后,扩展方向其实很多。比如把 MCP 工具层做成可插拔的,不同业务场景挂不同的工具集;比如把子 Agent 做成可配置的,通过配置文件定义有哪些 Agent、各自用什么提示词和工具;比如加一个“学习”环节,把每次执行的成功案例存下来,作为后续的参考。

我个人在实际操作中的体会是,Agent 工作流这个东西,架构设计占三成,提示词工程占三成,剩下四成全是调试和踩坑。别指望一次写对,做好反复迭代的准备。另外,可观测性一定要从第一天就做,不然出了问题你会非常痛苦。

最后分享一个小技巧:每次改动提示词或工具定义后,用同一批测试用例跑一遍,对比成功率和耗时。这样你能清楚知道这次改动是变好了还是变差了,而不是凭感觉。我维护了一个 30 条的测试集,每次改动都跑,省了很多返工的时间。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询