我最近被问得最多的一句话是ReAct 模式跑得好好的为什么要改成 Workflow问的人多半刚把大模型接进工具调用正沉浸在“模型自己会规划”的兴奋里。可一到生产环境问题就来了同一个任务有时候三秒完成有时候卡三分钟中间一步 API 超时整个循环就得重来日志里只能看到模型输出了几段话却说不清楚业务到底走到哪一步。这篇文章要聊的就是从 ReAct 到 Workflow 的转变思路。我们会把一个大模型实战项目从“让模型自由循环”改成“事件驱动的云端工作流”用云端 API 把任务接收、状态流转、工具调用、结果回传全部串起来。这个话题适合已经在用大模型 API 做工具调用的人也适合正准备把智能体从 Demo 推向生产的同学。你可以从我踩过的坑里直接抄作业省掉那些来回试错的成本。1. 先捋清楚ReAct 和 Workflow 到底在解决什么问题1.1 ReAct 模式的底层逻辑ReAct 是 Reasoning Acting 的缩写核心是让模型在一个循环里交替进行推理和行动。简单说模型先想“我要查什么”然后调用一个工具看到工具返回的结果再想“下一步怎么办”直到它认为任务完成。这个模式用生活类比特别好懂你出门前不确定天气于是打开手机查天气 App看到“下午有雨”然后决定带伞。查天气是行动根据结果决定带伞是推理这就是一个完整的 ReAct 循环。用伪代码表示大概是这样的task_done False while not task_done: thought llm.reason(current_state, observation) if thought.is_final: task_done True break action thought.action observation call_tool(action) current_state current_state observation这个循环的优势是灵活。面对开放任务时模型可以根据中间结果动态调整计划不需要提前把所有步骤写死。比如“帮我整理一份产品竞品分析”模型可能先去查官网再搜新闻再总结差异点路径不固定ReAct 能兜住。但问题也藏在这个循环里。第一步模型每轮都要把历史上下文重新塞给 APIToken 消耗增长快第二步任何一个工具调用失败模型可能陷入“想新办法—再失败—再想”的循环没有明确的重试机制第三步整个循环没有持久化状态任务跑到一半崩了就什么痕迹都留不下来。1.2 从 ReAct 到 Workflow为什么需要工作流我后来复盘了手上几个项目发现真正放到生产环境的大模型任务真正需要“完全自由决策”的只占一小部分。更多场景其实是固定的步骤接收数据、做预处理、调用模型、检查结果、写入存储、发通知。这些步骤的顺序稳定偶尔某个环节需要模型发挥推理能力但整体流程是确定的。与其让模型在每一步都重新“发明流程”不如把流程固化成工作流。工作流和 ReAct 最大的区别在于ReAct 把“下一步做什么”的权利交给模型Workflow 把“流程怎么走”的权利交给编排器。模型只负责工作流里需要理解语义的节点比如写摘要、判断情绪、抽取关键字段而那些不需要语义理解的节点比如格式校验、状态更新、结果通知直接用普通代码实现成本低且稳定。这是一个很重要的认知转变。你不需要在 ReAct 和 Workflow 之间二选一更好的做法是让它们协作工作流负责流程骨架ReAct 节点负责动态推理。比如文档审阅任务工作流固定为“接收文档 → 模型摘要 → 规则校验 → 通知结果”其中“模型摘要”这个节点内部可以再用 ReAct 模式决定是否要查询额外的参考文档。从 ReAct 到 Workflow不是说模型能力变弱了而是把智能用在刀刃上。1.3 事件驱动从“主动问”到“被通知”有了工作流之后还有一个更实际的问题工作流什么时候启动最简单的方式是定时轮询每隔几秒去查一下有没有新任务。但轮询有两个毛病一是响应有延迟二是明明没有新事件也得空转消耗资源。事件驱动换个思路系统在事件发生的那一刻主动通知你的服务你的工作流收到通知后立刻启动。这是从“主动问”到“被通知”的变化。举个例子某团队以前做文档审阅数据库里加一个状态字段定时任务每五分钟扫一次发现有新文档就处理。改成事件驱动后文档一上传对象存储直接发出一个事件云端 API 网关收到事件后触发工作流整个过程从分钟级延迟变成秒级延迟。事件驱动不是新概念消息队列、Webhook、触发器都是常见实现。我们需要关心的核心是事件源、触发器、队列、消费者。事件源产生事件触发器把事件转成消息队列做缓冲和削峰消费者执行实际工作流。云端 API 的价值在于这些组件都不需要你自己搭建直接用现成的服务组合起来这也是“基于云端 API 构建”最直接的好处。2. 基于云端 API 的事件驱动架构组件与设计权衡2.1 云端 API 的基本组件我先按自己在项目里的实际使用把搭建事件驱动智能体需要的云端组件分成四类。组件类型负责的事设计要点模型 API文本生成、工具调用、语义判断关注延迟、限流、上下文长度事件接入层接收外部事件、触发流程使用 Webhook 或消息队列保证可靠投递存储保存任务状态、中间结果、日志选择 KV 或对象存储按访问频度分层计算执行层执行工作流节点、调用模型和外部工具用云函数或任务服务保持节点无状态回调通道将处理结果返回给业务方支持重试和幂等防止回调丢失这里最重要的一条设计原则是计算节点必须无状态。如果一个节点把状态保存在自己的内存里一旦节点重启状态就全丢了。所以任务状态要落在存储里节点之间通过任务 ID 关联数据。云端 API 的执行单元天然适合无状态设计因为每个事件进来都可能由一个全新实例处理。我在早期设计时踩过一个坑把中间结果直接放在云函数的内存变量里结果下一个事件被同一个实例处理时上一个人的状态还留在内存里导致串数据。后来把所有中间结果统一写到存储云函数只做“读输入 → 处理 → 写输出”这件事问题立刻消失了。2.2 事件模型的定义事件驱动架构的基础是事件本身。如果事件定义得随意后面所有节点都要跟着改。一个规范的事件体至少要包含这些字段事件 ID全局唯一用于幂等去重。事件类型比如 document.uploaded、task.completed需要能对应到明确业务含义。时间戳事件发生时间最好统一成 UTC 时间戳。载荷事件附带的数据比如文件地址、任务参数。元数据来源、环境、调用的关联 ID方便排查。具体格式可以用这样的 JSON{ event_id: evt_20250115_0001, event_type: document.uploaded, occurred_at: 2025-01-15T08:30:00Z, payload: { doc_id: doc_abc123, file_url: https://storage.example.com/docs/doc_abc123.pdf, uploader: alice }, metadata: { source: internal_console, trace_id: trace_00x } }一个常见的错误是把大文件内容直接塞进事件体。比如有人会把整个文档的全文放进 payload结果消息队列被大消息卡住处理耗时暴增。正确做法是事件体里只放文件引用和元信息消费者拿到文件地址后再去对象存储拉取具体内容。事件体是“通知”不是“搬运工”。2.3 同步还是异步两条路线怎么选有了事件还要决定工作流怎么回报结果。最常见的问题是调用方需不需要同步等结果同步模式适合短任务。用户发一个请求云端 API 处理完后直接返回结果整个链路是“请求 → 处理 → 响应”。比如用模型给一段短文本做分类几百毫秒能完成同步就够了。这个方案好处是简单坏处是只要模型稍微慢一点用户就得一直等着HTTP 连接也容易超时。异步模式适合长任务。用户发请求后服务端先把任务记录下来然后立刻返回一个任务 ID处理过程在后台跑处理完成后再通过回调或用户主动查询来获得结果。文档审阅、视频分析、批量生成这类任务都应采用异步方案。我个人的判断标准是如果任务预期耗时超过五秒或者涉及多个外部 API 调用直接选异步。异步还有一个额外好处可以把任务推进度和中间状态暴露给用户体验比转一个圈干等好得多。混合模式也常用对外提供同步 API服务端收到请求后转成事件投递到队列消费者处理完通过回调通知。这样调用方的接入体验是同步的但内部处理又是异步的两头的好处都占了。2.4 状态管理与幂等性设计异步工作流一定要有状态机。每个任务从创建到完成至少要经历这几个状态pending事件已进入任务待处理。running正在执行某个节点。success全部节点完成结果已写入。failed出现无法自动恢复的错误。retrying执行失败但还允许重试。状态要持久化到存储而不是放在内存里。云端 API 场景下每个节点执行前都要去存储里读最新的状态避免并发节点互相覆盖。我的做法是给每个任务加一个版本号更新状态时带上版本号做条件更新防止两个节点同时改状态导致丢失。幂等性是事件驱动里最容易忽略的点。消息队列的“至少一次投递”意味着同一个事件可能会被消费两次。如果消费者不做幂等一个文档可能被生成两次摘要、发两次通知。解决办法是让消费者先根据事件 ID 查询是否处理过如果处理过直接返回旧结果。这种去重表在云端 KV 存储里特别容易实现成本也很低。3. 落地实战把 ReAct 循环改造成事件驱动工作流3.1 从需求反推一个文档审阅智能体我们用一个虚拟项目来走完整流程。某团队需要自动审阅上传的文档包括生成文档摘要、做敏感信息检查、判断是否允许发送通知。最初他们直接用 ReAct 模式让模型在一个循环里去调用“摘要 API”“检查 API”“通知 API”。结果发现三个问题第一模型偶尔会漏掉某个工具调用特别是中间某次工具返回超时时模型容易在“重试”里打转。第二每次运行都把整个文档内容作为上下文Token 成本高。第三执行过程几乎没有日志出了问题很难定位。改造方案很直接把整个流程固化为一个四步工作流。接收文档上传事件。调用模型生成摘要。调用规则 API 做敏感信息检查。汇总结果并发送通知。你可能会问这不是把原本智能的东西变得更死板了吗其实不会。因为在这个业务里流程本身是固定的真正需要智能判断的只有“生成摘要”和“判断哪些敏感信息需要标记”这两步。其余步骤用固定代码反而更可靠。这就是从 ReAct 到 Workflow 的核心取舍把固定的流程交给编排器把开放的理解交给模型。3.2 拆解 ReAct 循环为工作流节点把原来的 ReAct 循环拆成节点时有一个原则一个节点只做一件事并且明确标注这个节点是否需要模型。当时我把节点拆成以下五个。节点 A接收事件做格式校验。节点 B调用模型生成摘要。节点 C调用规则 API 做敏感词与合规检查。节点 D聚合摘要与检查结果生成最终结论。节点 E通过回调通道发送结果通知。其中节点 B 和 D 可能会用到模型。节点 C 用规则引擎就够了因为敏感信息检查需要确定性规则匹配比模型更可靠。节点 A 和 E 是完全的工程逻辑用云函数写普通代码即可。这样的拆法还有一个好处每个节点都可以独立测试。如果换一个模型只需要改节点 B不影响其他节点。如果业务里新增“错别字检查”直接在节点 C 后面加一个新节点事件的执行顺序用配置改一下就行不用重写整个循环。3.3 定义云端 API 契约在写代码之前先把接口定义清楚。我们至少需要两个对外接口事件接收接口POST /webhook/document-uploaded外部系统在上传文档后调用。回调通知接口POST /callback/notify工作流处理完成后把结果推送给业务方。事件接收接口的请求体就是我们前面定义过的事件 JSON。这个接口收到合法请求后不直接开始处理而是把事件写入任务表然后返回 202 Accepted并在响应体里带回任务 ID。{ task_id: task_20250115_0001, status: pending, message: 任务已进入处理队列 }这样做的好处是调用方不需要等待整个工作流完成只需要拿住任务 ID后续可以主动查询进度也可以等待回调通知。回调通知接口的请求体则要约定好至少要包括任务 ID、最终状态、结果数据。{ task_id: task_20250115_0001, status: success, summary: 文档主要介绍了一款新产品的技术架构..., sensitive_flags: [地址, 手机号], notified: true }接口一定要支持幂等重试。意思是同一个 task_id 的回调推两次业务方不应收到两次通知。实现上可以在回调接口里根据 task_id 去重。3.4 实现编排从事件到完成有了契约就可以写处理流程了。我用类云函数的风格展示核心逻辑重点讲流程而不是具体平台。第一步是事件接入。当外部系统向 Webhook 接口发送事件时我们需要做签名校验、解析事件、写入任务表然后把任务 ID 推送到消息队列。def webhook_handler(event, context): body parse_json(event[body]) verify_signature(event[headers], body) task_id create_task(body[event_id], body[event_type], statuspending) send_to_queue(document_tasks, task_id) return {statusCode: 202, body: json.dumps({task_id: task_id})}第二步是消费队列。消费者从队列里取出任务 ID然后按照工作流定义依次执行节点。这里的关键是每个节点执行前都从存储加载任务输入执行后把输出保存再推进到下一个节点。def process_task(task_id): task load_task(task_id) if task.status pending: task.status running save_task(task) doc load_document(task.doc_id) summary llm_summarize(doc.text) # 节点 B flags rule_check(doc.text) # 节点 C conclusion llm_conclude(summary, flags) # 节点 D notify(task_id, conclusion) # 节点 E task.status success save_task(task)当然真实场景里不能把节点串在一个同步函数里否则又回到了同步阻塞。更合理的做法是每个节点一个消费者节点 A 完成后向“摘要队列”投递一个消息摘要节点消费后继续向“检查队列”投递消息形成链式调用。模型节点单独抽出来很重要。因为你可能需要对模型调用做单独的重试、限流和超时控制如果模型调用和工作流主逻辑混在一起任何一次模型抖动都会拖垮整个流程。我为模型节点设计了一个隔离的执行器里面做重试和指数退避其他节点根本不需要知道模型 API 的细节。3.5 错误处理与重试策略错误处理是事件驱动智能体最容易翻车的地方。我的做法是把错误分两类。可重试错误包括模型 API 超时、限流返回 429、临时网络错误、依赖的服务 5xx。这类错误要进入重试队列并使用指数退避策略。第一次失败等 1 秒重试第二次等 2 秒第三次等 4 秒同时加一点随机抖动防止所有重试同时压过来。不可重试错误包括事件格式错误、参数缺失、权限校验失败、业务规则不允许。这类错误直接标记任务为 failed并写入日志不需要浪费重试资源。还要有死信队列。如果一个消息重试了十次仍然失败就把它投递到死信队列由人工或者专门的补偿任务去检查。初期项目里死信队列不显眼但越早上越好否则消息丢了都不知道去哪查。幂等执行同样要落到底层。每个消费者在开始执行节点前先检查当前节点是否已经执行过。实现方式很简单任务表里存一个当前节点的执行记录同一节点重复执行时发现记录已存在就直接复用旧结果不做二次调用。4. 实战中常见的坑与排查技巧实录4.1 回调丢失与重试风暴第一次上线回调通知时我遇到一个很典型的坑。任务处理完成代码里发起了一个回调请求结果回调地址临时不可用请求失败了。由于代码没做重试任务状态已经更新成 success但业务方永远没有收到通知。后来用户来问“为什么没消息”我才发现回调失败被忽略了。解决办法是三重保险回调发送失败时把回调消息重新投递到队列队列自带重试。回调接收方做幂等同一任务重复回调不会造成重复通知。提供主动查询接口GET /tasks/{id}调用方即使收不到回调也能主动拉取结果。这个组合下来回调丢失的概率会低很多。不过要注意重试风暴如果回调接收方持续不可用重试队列会不断堆积。必须设置最大重试次数超过次数进入死信队列。4.2 上下文膨胀与 Token 失控把 ReAct 循环改成工作流之后Token 费用下降了很多但一开始改造得不够彻底。有一个节点需要模型判断“文档摘要是否足够简洁”我把完整文档文本和摘要一起传给模型结果每次调用都消耗大量 Token。后来总结了一个经验工作流节点之间只传摘要和引用不传全文。比如摘要节点输出的是“摘要文本 文档地址”检查节点需要全文时再根据文档地址按需拉取。这里的“按需拉取”很重要因为并不是每个节点都需要完整数据。你还要注意模型输入里的历史消息不要无限增长。在 ReAct 模式里循环每转一圈所有中间步骤都会进入上下文到工作流里每个节点只带自己的输入和必要的参考信息上下文体积就可控了。这算是从 ReAct 到 Workflow 最直观的成本收益。4.3 并发与限流事件驱动的并发问题也很有意思。某段时间业务量突增消息队列里瞬间积累了上千个任务。消费者来不及消化模型 API 被频繁调用触发限流紧接着一堆任务开始报错重试局面一下就乱了。解决并发问题要分两层。队列消费者要设置最大并发数比如同时最多处理 10 个任务其余任务在队列里排队。模型调用要有自己的信号量或限流器控制每秒最大请求数和 API 的限流阈值匹配。我习惯在压测环境先测一下模型 API 的实际 QPS 上限然后给消费者并发数设置一个保守值留出 20% 的余量。要不然线上一旦流量上来限流比业务逻辑更先击垮系统。4.4 可观测性事件追踪到底在哪一步改造成工作流之后排查问题的思路也要跟着变。以前 ReAct 模式出问题就看模型输出现在节点多了任何一个节点都可能出问题。如果日志里只有每个节点的一句话完全不够用。我的做法是给每个任务生成一个 trace_id从 Webhook 接收事件开始所有节点日志都带上这个 trace_id。日志记录每个节点的输入摘要、输出摘要、耗时和错误信息。这样顺着 trace_id 一查就能看到任务卡在哪个节点是该节点的 API 超时还是数据问题。下面是一张我自己常用的排查速查表。症状可能原因排查方式任务一直 pending队列消费者没拉起检查队列消息堆积确认消费者进程任务一直 running某个节点执行耗时过长或死锁按 trace_id 查节点日志看最近活跃节点回调多次重复回调幂等没做好检查回调去重表按 task_id 去重模型 API 限流报错并发超过上限查看限流指标降低消费者并发数任务 failed 但原因未知异常日志未记录堆栈把异常堆栈和上下文写入日志存储这张表帮我省了很多时间。事件驱动智能体调试的关键不是看代码逻辑而是看事件在系统里流动的轨迹。只要每个节点都能留下足够清晰的日志问题定位基本就是按图索骥。我个人在几次项目里最大的体会是从 ReAct 到 Workflow不是要消除模型的能力而是要把模型放在最擅长的位置。固定流程交给工作流自由推理留给模型节点事件驱动负责把它们串起来。最后分享一个小技巧动手写代码之前先画出事件流和任务状态流转图哪怕只写在一张纸上也比直接开写代码稳妥。这个步骤能逼你想清楚每个事件的入口、每个节点的输入输出、每个失败路径的补偿方式想清楚这些代码只是一个执行过程。 
长河君 · 2026/10/20 6:47:32 · 阅读 1.2k 从ReAct到Workflow：事件驱动架构实现生产级智能体
我最近被问得最多的一句话是ReAct 模式跑得好好的为什么要改成 Workflow问的人多半刚把大模型接进工具调用正沉浸在“模型自己会规划”的兴奋里。可一到生产环境问题就来了同一个任务有时候三秒完成有时候卡三分钟中间一步 API 超时整个循环就得重来日志里只能看到模型输出了几段话却说不清楚业务到底走到哪一步。这篇文章要聊的就是从 ReAct 到 Workflow 的转变思路。我们会把一个大模型实战项目从“让模型自由循环”改成“事件驱动的云端工作流”用云端 API 把任务接收、状态流转、工具调用、结果回传全部串起来。这个话题适合已经在用大模型 API 做工具调用的人也适合正准备把智能体从 Demo 推向生产的同学。你可以从我踩过的坑里直接抄作业省掉那些来回试错的成本。1. 先捋清楚ReAct 和 Workflow 到底在解决什么问题1.1 ReAct 模式的底层逻辑ReAct 是 Reasoning Acting 的缩写核心是让模型在一个循环里交替进行推理和行动。简单说模型先想“我要查什么”然后调用一个工具看到工具返回的结果再想“下一步怎么办”直到它认为任务完成。这个模式用生活类比特别好懂你出门前不确定天气于是打开手机查天气 App看到“下午有雨”然后决定带伞。查天气是行动根据结果决定带伞是推理这就是一个完整的 ReAct 循环。用伪代码表示大概是这样的task_done False while not task_done: thought llm.reason(current_state, observation) if thought.is_final: task_done True break action thought.action observation call_tool(action) current_state current_state observation这个循环的优势是灵活。面对开放任务时模型可以根据中间结果动态调整计划不需要提前把所有步骤写死。比如“帮我整理一份产品竞品分析”模型可能先去查官网再搜新闻再总结差异点路径不固定ReAct 能兜住。但问题也藏在这个循环里。第一步模型每轮都要把历史上下文重新塞给 APIToken 消耗增长快第二步任何一个工具调用失败模型可能陷入“想新办法—再失败—再想”的循环没有明确的重试机制第三步整个循环没有持久化状态任务跑到一半崩了就什么痕迹都留不下来。1.2 从 ReAct 到 Workflow为什么需要工作流我后来复盘了手上几个项目发现真正放到生产环境的大模型任务真正需要“完全自由决策”的只占一小部分。更多场景其实是固定的步骤接收数据、做预处理、调用模型、检查结果、写入存储、发通知。这些步骤的顺序稳定偶尔某个环节需要模型发挥推理能力但整体流程是确定的。与其让模型在每一步都重新“发明流程”不如把流程固化成工作流。工作流和 ReAct 最大的区别在于ReAct 把“下一步做什么”的权利交给模型Workflow 把“流程怎么走”的权利交给编排器。模型只负责工作流里需要理解语义的节点比如写摘要、判断情绪、抽取关键字段而那些不需要语义理解的节点比如格式校验、状态更新、结果通知直接用普通代码实现成本低且稳定。这是一个很重要的认知转变。你不需要在 ReAct 和 Workflow 之间二选一更好的做法是让它们协作工作流负责流程骨架ReAct 节点负责动态推理。比如文档审阅任务工作流固定为“接收文档 → 模型摘要 → 规则校验 → 通知结果”其中“模型摘要”这个节点内部可以再用 ReAct 模式决定是否要查询额外的参考文档。从 ReAct 到 Workflow不是说模型能力变弱了而是把智能用在刀刃上。1.3 事件驱动从“主动问”到“被通知”有了工作流之后还有一个更实际的问题工作流什么时候启动最简单的方式是定时轮询每隔几秒去查一下有没有新任务。但轮询有两个毛病一是响应有延迟二是明明没有新事件也得空转消耗资源。事件驱动换个思路系统在事件发生的那一刻主动通知你的服务你的工作流收到通知后立刻启动。这是从“主动问”到“被通知”的变化。举个例子某团队以前做文档审阅数据库里加一个状态字段定时任务每五分钟扫一次发现有新文档就处理。改成事件驱动后文档一上传对象存储直接发出一个事件云端 API 网关收到事件后触发工作流整个过程从分钟级延迟变成秒级延迟。事件驱动不是新概念消息队列、Webhook、触发器都是常见实现。我们需要关心的核心是事件源、触发器、队列、消费者。事件源产生事件触发器把事件转成消息队列做缓冲和削峰消费者执行实际工作流。云端 API 的价值在于这些组件都不需要你自己搭建直接用现成的服务组合起来这也是“基于云端 API 构建”最直接的好处。2. 基于云端 API 的事件驱动架构组件与设计权衡2.1 云端 API 的基本组件我先按自己在项目里的实际使用把搭建事件驱动智能体需要的云端组件分成四类。组件类型负责的事设计要点模型 API文本生成、工具调用、语义判断关注延迟、限流、上下文长度事件接入层接收外部事件、触发流程使用 Webhook 或消息队列保证可靠投递存储保存任务状态、中间结果、日志选择 KV 或对象存储按访问频度分层计算执行层执行工作流节点、调用模型和外部工具用云函数或任务服务保持节点无状态回调通道将处理结果返回给业务方支持重试和幂等防止回调丢失这里最重要的一条设计原则是计算节点必须无状态。如果一个节点把状态保存在自己的内存里一旦节点重启状态就全丢了。所以任务状态要落在存储里节点之间通过任务 ID 关联数据。云端 API 的执行单元天然适合无状态设计因为每个事件进来都可能由一个全新实例处理。我在早期设计时踩过一个坑把中间结果直接放在云函数的内存变量里结果下一个事件被同一个实例处理时上一个人的状态还留在内存里导致串数据。后来把所有中间结果统一写到存储云函数只做“读输入 → 处理 → 写输出”这件事问题立刻消失了。2.2 事件模型的定义事件驱动架构的基础是事件本身。如果事件定义得随意后面所有节点都要跟着改。一个规范的事件体至少要包含这些字段事件 ID全局唯一用于幂等去重。事件类型比如 document.uploaded、task.completed需要能对应到明确业务含义。时间戳事件发生时间最好统一成 UTC 时间戳。载荷事件附带的数据比如文件地址、任务参数。元数据来源、环境、调用的关联 ID方便排查。具体格式可以用这样的 JSON{ event_id: evt_20250115_0001, event_type: document.uploaded, occurred_at: 2025-01-15T08:30:00Z, payload: { doc_id: doc_abc123, file_url: https://storage.example.com/docs/doc_abc123.pdf, uploader: alice }, metadata: { source: internal_console, trace_id: trace_00x } }一个常见的错误是把大文件内容直接塞进事件体。比如有人会把整个文档的全文放进 payload结果消息队列被大消息卡住处理耗时暴增。正确做法是事件体里只放文件引用和元信息消费者拿到文件地址后再去对象存储拉取具体内容。事件体是“通知”不是“搬运工”。2.3 同步还是异步两条路线怎么选有了事件还要决定工作流怎么回报结果。最常见的问题是调用方需不需要同步等结果同步模式适合短任务。用户发一个请求云端 API 处理完后直接返回结果整个链路是“请求 → 处理 → 响应”。比如用模型给一段短文本做分类几百毫秒能完成同步就够了。这个方案好处是简单坏处是只要模型稍微慢一点用户就得一直等着HTTP 连接也容易超时。异步模式适合长任务。用户发请求后服务端先把任务记录下来然后立刻返回一个任务 ID处理过程在后台跑处理完成后再通过回调或用户主动查询来获得结果。文档审阅、视频分析、批量生成这类任务都应采用异步方案。我个人的判断标准是如果任务预期耗时超过五秒或者涉及多个外部 API 调用直接选异步。异步还有一个额外好处可以把任务推进度和中间状态暴露给用户体验比转一个圈干等好得多。混合模式也常用对外提供同步 API服务端收到请求后转成事件投递到队列消费者处理完通过回调通知。这样调用方的接入体验是同步的但内部处理又是异步的两头的好处都占了。2.4 状态管理与幂等性设计异步工作流一定要有状态机。每个任务从创建到完成至少要经历这几个状态pending事件已进入任务待处理。running正在执行某个节点。success全部节点完成结果已写入。failed出现无法自动恢复的错误。retrying执行失败但还允许重试。状态要持久化到存储而不是放在内存里。云端 API 场景下每个节点执行前都要去存储里读最新的状态避免并发节点互相覆盖。我的做法是给每个任务加一个版本号更新状态时带上版本号做条件更新防止两个节点同时改状态导致丢失。幂等性是事件驱动里最容易忽略的点。消息队列的“至少一次投递”意味着同一个事件可能会被消费两次。如果消费者不做幂等一个文档可能被生成两次摘要、发两次通知。解决办法是让消费者先根据事件 ID 查询是否处理过如果处理过直接返回旧结果。这种去重表在云端 KV 存储里特别容易实现成本也很低。3. 落地实战把 ReAct 循环改造成事件驱动工作流3.1 从需求反推一个文档审阅智能体我们用一个虚拟项目来走完整流程。某团队需要自动审阅上传的文档包括生成文档摘要、做敏感信息检查、判断是否允许发送通知。最初他们直接用 ReAct 模式让模型在一个循环里去调用“摘要 API”“检查 API”“通知 API”。结果发现三个问题第一模型偶尔会漏掉某个工具调用特别是中间某次工具返回超时时模型容易在“重试”里打转。第二每次运行都把整个文档内容作为上下文Token 成本高。第三执行过程几乎没有日志出了问题很难定位。改造方案很直接把整个流程固化为一个四步工作流。接收文档上传事件。调用模型生成摘要。调用规则 API 做敏感信息检查。汇总结果并发送通知。你可能会问这不是把原本智能的东西变得更死板了吗其实不会。因为在这个业务里流程本身是固定的真正需要智能判断的只有“生成摘要”和“判断哪些敏感信息需要标记”这两步。其余步骤用固定代码反而更可靠。这就是从 ReAct 到 Workflow 的核心取舍把固定的流程交给编排器把开放的理解交给模型。3.2 拆解 ReAct 循环为工作流节点把原来的 ReAct 循环拆成节点时有一个原则一个节点只做一件事并且明确标注这个节点是否需要模型。当时我把节点拆成以下五个。节点 A接收事件做格式校验。节点 B调用模型生成摘要。节点 C调用规则 API 做敏感词与合规检查。节点 D聚合摘要与检查结果生成最终结论。节点 E通过回调通道发送结果通知。其中节点 B 和 D 可能会用到模型。节点 C 用规则引擎就够了因为敏感信息检查需要确定性规则匹配比模型更可靠。节点 A 和 E 是完全的工程逻辑用云函数写普通代码即可。这样的拆法还有一个好处每个节点都可以独立测试。如果换一个模型只需要改节点 B不影响其他节点。如果业务里新增“错别字检查”直接在节点 C 后面加一个新节点事件的执行顺序用配置改一下就行不用重写整个循环。3.3 定义云端 API 契约在写代码之前先把接口定义清楚。我们至少需要两个对外接口事件接收接口POST /webhook/document-uploaded外部系统在上传文档后调用。回调通知接口POST /callback/notify工作流处理完成后把结果推送给业务方。事件接收接口的请求体就是我们前面定义过的事件 JSON。这个接口收到合法请求后不直接开始处理而是把事件写入任务表然后返回 202 Accepted并在响应体里带回任务 ID。{ task_id: task_20250115_0001, status: pending, message: 任务已进入处理队列 }这样做的好处是调用方不需要等待整个工作流完成只需要拿住任务 ID后续可以主动查询进度也可以等待回调通知。回调通知接口的请求体则要约定好至少要包括任务 ID、最终状态、结果数据。{ task_id: task_20250115_0001, status: success, summary: 文档主要介绍了一款新产品的技术架构..., sensitive_flags: [地址, 手机号], notified: true }接口一定要支持幂等重试。意思是同一个 task_id 的回调推两次业务方不应收到两次通知。实现上可以在回调接口里根据 task_id 去重。3.4 实现编排从事件到完成有了契约就可以写处理流程了。我用类云函数的风格展示核心逻辑重点讲流程而不是具体平台。第一步是事件接入。当外部系统向 Webhook 接口发送事件时我们需要做签名校验、解析事件、写入任务表然后把任务 ID 推送到消息队列。def webhook_handler(event, context): body parse_json(event[body]) verify_signature(event[headers], body) task_id create_task(body[event_id], body[event_type], statuspending) send_to_queue(document_tasks, task_id) return {statusCode: 202, body: json.dumps({task_id: task_id})}第二步是消费队列。消费者从队列里取出任务 ID然后按照工作流定义依次执行节点。这里的关键是每个节点执行前都从存储加载任务输入执行后把输出保存再推进到下一个节点。def process_task(task_id): task load_task(task_id) if task.status pending: task.status running save_task(task) doc load_document(task.doc_id) summary llm_summarize(doc.text) # 节点 B flags rule_check(doc.text) # 节点 C conclusion llm_conclude(summary, flags) # 节点 D notify(task_id, conclusion) # 节点 E task.status success save_task(task)当然真实场景里不能把节点串在一个同步函数里否则又回到了同步阻塞。更合理的做法是每个节点一个消费者节点 A 完成后向“摘要队列”投递一个消息摘要节点消费后继续向“检查队列”投递消息形成链式调用。模型节点单独抽出来很重要。因为你可能需要对模型调用做单独的重试、限流和超时控制如果模型调用和工作流主逻辑混在一起任何一次模型抖动都会拖垮整个流程。我为模型节点设计了一个隔离的执行器里面做重试和指数退避其他节点根本不需要知道模型 API 的细节。3.5 错误处理与重试策略错误处理是事件驱动智能体最容易翻车的地方。我的做法是把错误分两类。可重试错误包括模型 API 超时、限流返回 429、临时网络错误、依赖的服务 5xx。这类错误要进入重试队列并使用指数退避策略。第一次失败等 1 秒重试第二次等 2 秒第三次等 4 秒同时加一点随机抖动防止所有重试同时压过来。不可重试错误包括事件格式错误、参数缺失、权限校验失败、业务规则不允许。这类错误直接标记任务为 failed并写入日志不需要浪费重试资源。还要有死信队列。如果一个消息重试了十次仍然失败就把它投递到死信队列由人工或者专门的补偿任务去检查。初期项目里死信队列不显眼但越早上越好否则消息丢了都不知道去哪查。幂等执行同样要落到底层。每个消费者在开始执行节点前先检查当前节点是否已经执行过。实现方式很简单任务表里存一个当前节点的执行记录同一节点重复执行时发现记录已存在就直接复用旧结果不做二次调用。4. 实战中常见的坑与排查技巧实录4.1 回调丢失与重试风暴第一次上线回调通知时我遇到一个很典型的坑。任务处理完成代码里发起了一个回调请求结果回调地址临时不可用请求失败了。由于代码没做重试任务状态已经更新成 success但业务方永远没有收到通知。后来用户来问“为什么没消息”我才发现回调失败被忽略了。解决办法是三重保险回调发送失败时把回调消息重新投递到队列队列自带重试。回调接收方做幂等同一任务重复回调不会造成重复通知。提供主动查询接口GET /tasks/{id}调用方即使收不到回调也能主动拉取结果。这个组合下来回调丢失的概率会低很多。不过要注意重试风暴如果回调接收方持续不可用重试队列会不断堆积。必须设置最大重试次数超过次数进入死信队列。4.2 上下文膨胀与 Token 失控把 ReAct 循环改成工作流之后Token 费用下降了很多但一开始改造得不够彻底。有一个节点需要模型判断“文档摘要是否足够简洁”我把完整文档文本和摘要一起传给模型结果每次调用都消耗大量 Token。后来总结了一个经验工作流节点之间只传摘要和引用不传全文。比如摘要节点输出的是“摘要文本 文档地址”检查节点需要全文时再根据文档地址按需拉取。这里的“按需拉取”很重要因为并不是每个节点都需要完整数据。你还要注意模型输入里的历史消息不要无限增长。在 ReAct 模式里循环每转一圈所有中间步骤都会进入上下文到工作流里每个节点只带自己的输入和必要的参考信息上下文体积就可控了。这算是从 ReAct 到 Workflow 最直观的成本收益。4.3 并发与限流事件驱动的并发问题也很有意思。某段时间业务量突增消息队列里瞬间积累了上千个任务。消费者来不及消化模型 API 被频繁调用触发限流紧接着一堆任务开始报错重试局面一下就乱了。解决并发问题要分两层。队列消费者要设置最大并发数比如同时最多处理 10 个任务其余任务在队列里排队。模型调用要有自己的信号量或限流器控制每秒最大请求数和 API 的限流阈值匹配。我习惯在压测环境先测一下模型 API 的实际 QPS 上限然后给消费者并发数设置一个保守值留出 20% 的余量。要不然线上一旦流量上来限流比业务逻辑更先击垮系统。4.4 可观测性事件追踪到底在哪一步改造成工作流之后排查问题的思路也要跟着变。以前 ReAct 模式出问题就看模型输出现在节点多了任何一个节点都可能出问题。如果日志里只有每个节点的一句话完全不够用。我的做法是给每个任务生成一个 trace_id从 Webhook 接收事件开始所有节点日志都带上这个 trace_id。日志记录每个节点的输入摘要、输出摘要、耗时和错误信息。这样顺着 trace_id 一查就能看到任务卡在哪个节点是该节点的 API 超时还是数据问题。下面是一张我自己常用的排查速查表。症状可能原因排查方式任务一直 pending队列消费者没拉起检查队列消息堆积确认消费者进程任务一直 running某个节点执行耗时过长或死锁按 trace_id 查节点日志看最近活跃节点回调多次重复回调幂等没做好检查回调去重表按 task_id 去重模型 API 限流报错并发超过上限查看限流指标降低消费者并发数任务 failed 但原因未知异常日志未记录堆栈把异常堆栈和上下文写入日志存储这张表帮我省了很多时间。事件驱动智能体调试的关键不是看代码逻辑而是看事件在系统里流动的轨迹。只要每个节点都能留下足够清晰的日志问题定位基本就是按图索骥。我个人在几次项目里最大的体会是从 ReAct 到 Workflow不是要消除模型的能力而是要把模型放在最擅长的位置。固定流程交给工作流自由推理留给模型节点事件驱动负责把它们串起来。最后分享一个小技巧动手写代码之前先画出事件流和任务状态流转图哪怕只写在一张纸上也比直接开写代码稳妥。这个步骤能逼你想清楚每个事件的入口、每个节点的输入输出、每个失败路径的补偿方式想清楚这些代码只是一个执行过程。