最近在复盘一个 Agent 项目的时候有个感受特别深现在一提 Agent 工作流大家第一反应就是上重型框架。编排引擎、缓存层、分布式协调、可视化画布全都配齐结果写出来的流程又重又僵模型稍微说一句偏离预期的话整个状态机就崩给你看。反而是自己动手做的那套“面向 Agent 的轻量 Workflow 设计”一路用下来最顺手。核心思路不复杂不用大引擎不做全局持久化不搞中心化控制器只用一套极简的状态流转框架让 Agent 模型自己决定怎么走我们只负责兜底和约束边界。这篇文章就把这套设计的完整思路拆出来包括核心抽象怎么定、状态怎么组织、模型在什么位置做决策、以及我实际踩过的几个坑。适合那些已经搭过一两个 Agent Demo、想往工程化走一步但又不愿意直接上重型框架的开发者参考。1. 为什么传统流程编排不适合 Agent 场景先聊一个底层问题传统工作流引擎和 Agent Workflow 的出发点完全不一样。传统流程引擎的核心假设是所有步骤都是确定的、可枚举的、需要被精确控制的。比如订单处理流程下单、支付、仓库出库、物流发货每一步的走向都事先画好连异常分支也是提前定义好的。Agent 工作流的核心假设恰好相反步骤可能是动态的模型会根据用户输入生成不同的路径。你没法预知用户是直接问“什么是多态”还是会说“帮我把这段代码重构一下顺便加测试”。如果把这些全都画进流程图里那就回到了老的专家系统老路上去了——写规则的人累死规则还永远覆盖不全。我早期做 Agent 工作时就是这个状态。每个业务场景都去画状态机把“用户可能说的一切”都枚举一遍结果图越画越大代码越写越僵。后来想明白了一件事Agent 时代的流程设计不应该是“把路径写死”而是“给模型一个边界让它在边界内自由活动”。这带来的一个直接转变是工作流里不再有唯一的“控制器”。传统流程引擎里有一个 controller 在决定下一步执行哪个节点而在轻量设计里这个“决策权”被下放给了模型。1.1 轻量 Workflow 的目标到底是什么我个人对轻量 Workflow 的定义有三条硬指标不依赖外部服务。不需要专门的流程引擎、不需要独立的数据库存流程实例。一套纯代码实现一个类加一个配置文件就够了。状态清晰可见。任何时刻整个流程处于什么位置、模型已经做了什么决策、积累了哪些中间结果都能一眼看明白。模型决策空间受控但不僵死。模型的跳转自由度受到约束但这些约束不是靠写死 if-else 实现的而是靠设计良好的上下文、可选的下一步选项、清晰的终止条件来实现的。如果一套流程设计能满足这三条那它就可以叫“轻量 Workflow”哪怕它只有几百行代码。1.2 传统流程与 Agent Workflow 的核心差异对照我拿我踩过的例子做个对照可能更直观维度传统流程引擎面向 Agent 的轻量 Workflow路径定义预先画好 DAG节点顺序固定只定义节点和可选项路径由模型决策状态存储中心化数据库存储实例状态临时内存中的 JSON 对象随请求而生、随请求而灭异常处理预定义异常分支模型在 prompt 中感知异常并自行调整执行单元代码函数/微服务LLM 调用、工具调用、普通代码函数的混合体调试方式看流程图、查日志看状态切片、看每一步决策的 reason传统流程的核心在画图Agent Workflow 的核心在设计“节点的边界感”。2. 三个核心抽象Node、State、Router我的轻量 Workflow 设计里基本就三个核心概念Node节点、State状态、Router路由器。把这三个东西想清楚整个框架骨架就出来了。2.1 Node最小执行单元Node 是流程中一个可执行的最小单元。它可以是一个代码函数比如“调用向量数据库做检索”一个 LLM 调用比如“对用户问题做意图识别”一个工具调用比如“调用天气 API”一个纯逻辑步骤比如“对上游结果做格式校验”我的定义很简单interface WorkflowNode { id: string; name: string; type: step | router | end; execute(ctx: WorkflowContext): PromiseNodeResult; canRun(ctx: WorkflowContext): boolean; } interface NodeResult { status: success | failed | skip; updates: PartialWorkflowState; next?: string; // 只有 router 节点会设置 message?: string; // 给调试日志用的说明 }这里有个细节值得注意canRun是用来判断“这个节点要不要执行”的钩子。它不是流程引擎的复杂条件路由它就是一个简单函数。比如如果没有检索到相关内容answer_node就不该执行直接让 router 节点重新决策。如果用户没有要求调用工具tool_node就不执行。用代码表达这种“要不要执行”的判断比让模型去理解一个复杂状态要可靠得多。能用代码判断的就别让模型折腾这是轻量设计的第一原则。2.2 State一切中间结果的全景图State 是一个贯穿整个流程的 JSON 对象。它不是数据库实体不需要持久化每次用户请求进来的时候创建响应结束就销毁。我把 State 设计成三层结构{ request: { raw_input: 帮我把这个函数改成异步的, user_id: u_12345, preferences: {} }, memory: { history: [], retrieved_chunks: [], tool_results: [] }, output: { final_answer: , intermediate_steps: [] } }分这三层的逻辑是request用户请求的原始信息全程不变。memory整个流程中积累的中间产物。检索到的资料、工具返回的结果、模型前面步骤的输出都往这里塞。output最终输出路径上产生的数据包括最终答案和中间步骤的记录。我刻意把tool_results设计成数组而不是单个对象是因为 Agent 实际运行时经常需要调用多个工具。如果只存最后一个结果前面的结果就丢了模型在生成最终答案时可能缺失关键信息。还有一个设计细节State 必须是纯 JSON 可序列化的。这意味着不能往里面塞 Python 对象、不能塞数据库连接、不能塞函数引用。原因很简单方便调试。任意时刻把 State 打印出来一眼就能看到流程走到了哪里、积累了哪些信息。方便回溯。如果流程出问题了把出错那个时点的 State 保存下来可以离线复现。方便接入日志和观测系统。纯 JSON 直接打到日志平台不需要额外的解析逻辑。2.3 Router把“下一步怎么走”的选择权交给模型但给它边界Router 是整个设计里最核心、也最需要克制的地方。我的做法是这样的把 Router 设计成一种特殊的 Node。它内部调用一次大模型输入是当前 State 的摘要、上下文信息、以及可选的下一步选项列表输出是一个 JSON 对象包含next选哪个节点和reason为什么这么选。interface RouterConfig { id: string; description: string; options: Array{ id: string; description: string; when_to_choose?: string; }; fallback: string; max_retries?: number; }这里的关键设计在于options数组。它相当于告诉模型“你有这些路可以走”而不是让模型自己发明路径。举个例子。这是我在一个知识库问答 Agent 里用的 Router 配置{ id: main_router_01, description: 根据用户的输入和当前上下文决定下一步执行哪个节点, options: [ { id: rag_retrieval, description: 用户问题涉及专业知识需要从知识库中检索相关内容 }, { id: tool_call, description: 用户问题需要调用外部工具如查天气、算数学、查数据库才能回答 }, { id: direct_answer, description: 用户只是闲聊或者问题很简单不需要检索、不需要工具直接回答即可 } ], fallback: direct_answer }这个配置的精髓是我给了模型三个选择但模型不能自己发明第四个选择。它只能在rag_retrieval、tool_call、direct_answer三个里面挑一个。这样可以避免很多失控场景。之前我试过不给选项直接问模型“下一步做什么”结果模型经常自己编出一些不存在的节点名流程直接断裂。用了 options 之后这种问题几乎绝迹了。Router 的调用逻辑也很简单async function runRouter( routerConfig: RouterConfig, state: WorkflowState, llm: LLMClient ): Promisestring { const prompt buildRouterPrompt(routerConfig, state); const response await llm.chat(prompt, { response_format: { type: json_object } }); const parsed JSON.parse(response.content); // 校验如果模型输出的 next 不在 options 里走 fallback const validIds routerConfig.options.map(o o.id); if (!validIds.includes(parsed.next)) { return routerConfig.fallback; } return parsed.next; }注意那个校验逻辑。模型输出的next如果不在选项列表里直接回退到 fallback。这是最后一道防线防止模型瞎编路径。3. 状态流转的核心循环让决策显性化有了 Node、State、Router 三个概念之后整个 Workflow 的执行循环就非常简洁了。3.1 主循环实现我的实现里主循环大概长这样async function runWorkflow( entryNodeId: string, state: WorkflowState, nodes: Mapstring, WorkflowNode, maxSteps: number 10 ): PromiseWorkflowState { let currentNodeId entryNodeId; let stepCount 0; while (stepCount maxSteps) { const node nodes.get(currentNodeId); if (!node) throw new Error(Node not found: ${currentNodeId}); // 1. 判断节点是否应该执行 if (!node.canRun(state)) { currentNodeId findNextCandidate(nodes, state); continue; } // 2. 执行节点 const result await node.execute(state); // 3. 合并状态更新 Object.assign(state.memory, result.updates); // 4. 记录决策轨迹 state.output.intermediate_steps.push({ node: node.id, action: node.name, message: result.message ?? , result: result.updates }); // 5. 确定下一个节点 if (node instanceof RouterNode) { currentNodeId result.next ?? routerFallback(node.config); } else if (result.next) { // 普通节点也可以显式指定下一个节点 currentNodeId result.next; } else { currentNodeId findNextCandidate(nodes, state); } stepCount; } return state; }这个循环的核心要点是每一步决策都有记录。intermediate_steps数组记录了每一步的节点名、执行动作、做了哪些状态更新、以及如果是 Router模型给出的理由。这等于给整个 Agent 的运行过程打了一份完整的“黑匣子”日志。3.2 状态流转的实际例子我用一个“回答专业技术问题”的场景来演示状态怎么流转的。假设用户输入是“请解释一下什么是数据库索引以及为什么它能加速查询”初始 State{ request: { raw_input: 请解释一下什么是数据库索引以及为什么它能加速查询 }, memory: {}, output: { intermediate_steps: [] } }第 1 步进入main_router_01。模型查看上下文后判断这是一个专业知识问题需要检索知识库Router 输出{ next: rag_retrieval, reason: 用户问题涉及数据库索引的专业概念需要检索知识库确保回答准确 }State 更新后{ request: { raw_input: 请解释一下什么是数据库索引以及为什么它能加速查询 }, memory: {}, output: { intermediate_steps: [ { node: main_router_01, action: 判断下一步方向, message: 用户问题涉及数据库索引的专业概念需要检索知识库确保回答准确, result: { next: rag_retrieval } } ] } }第 2 步进入rag_retrieval节点调用向量数据库检索相关文档。假设检索到两段内容写入memory.retrieved_chunks。第 3 步回到主 Router这次的输入多了检索结果。模型看到已经有检索结果了判断可以直接生成答案跳转到answer_generation。第 4 步answer_generation节点读取memory.retrieved_chunks生成最终回答写入output.final_answer然后走到end节点流程结束。这个例子的关键点在于流程的走向不是预先写死的。如果用户上来就说“你好”Router 会直接跳到direct_answer根本不会触发后续的检索。同一个 Workflow 可以在不同输入下走出完全不同的路径这就是“面向 Agent”的本质。4. 我在实际项目中踩过的五个坑这套设计看着简单但真正落地的时候坑不少。我把最有代表性的五个列出来都是我踩过的。4.1 坑一给 Router 的自由度太大模型开始自创路径第一次做 Router 的时候我没有用 options 限制而是直接问模型“根据当前情况请你决定下一步应该执行什么操作。”结果模型非常“有创意”。它输出的next字段经常是我从来没定义过的节点名比如search_external、ask_user_for_clarification、plan_next_step。我的执行引擎拿到这些节点名自然是找不到对应的节点直接抛异常。后来加了 options 白名单并且在 prompt 里明确写了一句你必须从以下选项中选择下一步不允许创造新的选项。字符串校验那一层也加上之后这个问题才算彻底解决。而且我还学到了一个教训Fallback 节点必须存在。我一开始的 fallback 设置的是direct_answer后来发现不合适因为如果用户本来就需要检索知识库直接回答就是瞎编。后来我把 fallback 改成了“重新执行 Router”并且给 Router 设置了重试上限超过上限就终止流程并输出提示信息。4.2 坑二全局状态越滚越大上下文爆炸这点特别容易发生在连续对话场景。最开始我设计 State 的时候把每次检索到的 chunks 全部存进去模型生成回答时要把所有 chunks 重新读一遍。短对话还好多轮之后状态直接爆掉。我的解决办法是给 State 加了两层限制memory 数据有上限。比如retrieved_chunks最多保留前 5 段tool_results最多保留最近 3 个历史记录只保留最近两轮。Router 的输入只取摘要。Router 决策的时候不需要看到全文只需要知道“当前有哪些可用的信息”。所以我会提前把retrieved_chunks做摘要压缩只把每段的标题和第一句话传给 Router。这两步做完之后状态大小从动不动几千 token 降到了稳定的一两百 token 左右模型的决策稳定性反而提高了——因为上下文里没有那么多噪音了。4.3 坑三工具调用失败后没有兜底这个坑的触发场景很典型Agent 调用了天气查询工具结果工具挂了返回了一个超时错误。我当时的初始设计是工具节点执行失败就直接把错误抛给用户“抱歉暂时无法获取天气信息”。这体验当然不行。后来我加了一级“失败重试 降级回答”的机制工具返回超时重试一次再次超时把错误信息附加到 State 里再次调用 Router让模型决定下一步是换一个工具比如天气 API 挂了我还可以用另一个数据源直接告知用户“该功能暂时不可用”并给出替代建议。这一步的收益在于把工具失败的异常处理从代码逻辑转移到了模型决策逻辑。代码只负责捕获异常决定怎么走出下一步的是模型。4.4 坑四循环控制做得不好Agent 陷入死循环轻量 Workflow 最怕的就是模型在某个 Router 之间来回跳。比如在rag_retrieval和direct_answer之间反复横跳每次 Router 都说“需要检索”检索完又说“可以回答了”接着模型又改变主意说“还要再检索一次”直接卡死。我在主循环里加了一个maxSteps的硬性限制默认 10 步。超过 10 步就直接终止走兜底回答逻辑。同时我在 Router 的 prompt 里会强调“如果你已经拥有足够信息请直接选择生成回答”。但这还不够。我更推荐的做法是设置一个“路径记忆”机制在每次 Router 决策时把“已经走过的节点”列表传给模型让它知道“你已经做过一次检索了除非你有明确的理由否则不要重复执行同一步”。我的实现里通过在 prompt 中附加previous_steps摘要来达到这个效果。4.5 坑五整个流程黑盒化出问题根本没法排查最开始的版本里我把 State 的流转放在内存里出问题的时候只能靠打印日志来查。问题在于一次请求涉及多个节点、多次模型调用日志量巨大并且日志之间的关联关系靠时间戳根本拼不回来。后来我给每个请求生成一个trace_id并把intermediate_steps数组完整地打印到日志平台。每个节点的模型调用也统一在日志里打上node_id和step_index标记。这样排查问题时只要拿到一个出错的trace_id就能完整回放当时的决策链路{ trace_id: req_8f3a12, intermediate_steps: [ { node: main_router_01, action: 判断下一步方向, message: 需要检索, result: { next: rag_retrieval } }, { node: rag_retrieval, action: 向量检索, message: 命中 5 条, result: { chunk_count: 5 } }, { node: main_router_01, action: 判断下一步方向, message: 信息充足生成回答, result: { next: answer_generation } }, { node: answer_generation, action: 生成最终答案, message: 成功, result: { tokens: 612 } } ] }这套日志设计帮了大忙。现在基本不用问“当时发生了什么”直接查intermediate_steps就能还原现场。5. 什么时候该换重型方案轻量 Workflow 是好东西但它不是银弹。我用下来的体会是当出现下面这些信号时就该考虑迁移到更重的平台或者框架了节点数超过 20 个。节点一多Router 的选择空间就变大模型决策准确率会明显下降。这时候就需要可视化的编排界面、更结构化的 DAG 定义来兜底。需要多人协作维护流程定义。如果你需要让非工程师也能画流程、改流程那么轻量的纯代码方案就不太合适了——图形化编排工具是更优解。需要持久化流程实例。如果你需要保存每个流程的运行状态支持中途暂停继续、支持跨服务调用那就必须引入存储层。轻量方案里那个随请求生灭的 State 就不够用了。状态流转复杂度过高。如果你发现某个节点需要判断几十个条件才能决定下一步而简单代码又写不清楚就说明业务本身需要一个规则引擎不再是轻量 Workflow 能覆盖的。需要考虑并发和分布式执行。多个流程并行跑、流程之间有依赖关系这些场景下集中式调度是不可少的轻量方案支撑不起来。我的经验是从轻量方案起步遇到上面这些信号再迁移比一上来就上重框架成本低得多而且你把业务逻辑真正摸透了迁移到重框架的时候也知道哪些节点需要拆、哪些状态需要保留。6. 最后再分享一个小技巧写到这里我想起一个特别实用的技巧也是我最近才总结出来的在 Router 的 prompt 里加入“你已知的信息”和“你还需要的信息”这两个小节。我发现很多模型跳错节点不是因为能力不行而是因为它不清楚自己已经掌握了什么。你只要把一个简单的信息清单放在它面前当前已掌握的信息用户输入xxx检索到的资料2 段主题为 xxx、xxx工具返回结果无根据以上信息判断是否还需要新的信息或可以直接回答用户问题。加上这段之后Router 的决策准确率肉眼可见地提升了。核心原因是模型是个“上下文驱动”的系统你把决策所需的信息摆得越清楚它做决策的质量就越高。这个技巧成本极低收益却很稳定做 Agent 工作流设计的朋友很值得试一试。