☰
面向Agent的轻量Workflow设计:状态流转、Router与工程落地实践
2026/10/17 11:35:08 网站建设 项目流程

最近在复盘一个 Agent 项目的时候,有个感受特别深:现在一提 Agent 工作流,大家第一反应就是上重型框架。编排引擎、缓存层、分布式协调、可视化画布全都配齐,结果写出来的流程又重又僵,模型稍微说一句偏离预期的话,整个状态机就崩给你看。

反而是自己动手做的那套“面向 Agent 的轻量 Workflow 设计”,一路用下来最顺手。核心思路不复杂:不用大引擎,不做全局持久化,不搞中心化控制器,只用一套极简的状态流转框架,让 Agent 模型自己决定怎么走,我们只负责兜底和约束边界。

这篇文章就把这套设计的完整思路拆出来,包括核心抽象怎么定、状态怎么组织、模型在什么位置做决策、以及我实际踩过的几个坑。适合那些已经搭过一两个 Agent Demo、想往工程化走一步但又不愿意直接上重型框架的开发者参考。

1. 为什么传统流程编排不适合 Agent 场景

先聊一个底层问题:传统工作流引擎和 Agent Workflow 的出发点完全不一样。

传统流程引擎的核心假设是:所有步骤都是确定的、可枚举的、需要被精确控制的。比如订单处理流程,下单、支付、仓库出库、物流发货,每一步的走向都事先画好,连异常分支也是提前定义好的。

Agent 工作流的核心假设恰好相反:步骤可能是动态的,模型会根据用户输入生成不同的路径。你没法预知用户是直接问“什么是多态”,还是会说“帮我把这段代码重构一下,顺便加测试”。如果把这些全都画进流程图里,那就回到了老的专家系统老路上去了——写规则的人累死,规则还永远覆盖不全。

我早期做 Agent 工作时,就是这个状态。每个业务场景都去画状态机,把“用户可能说的一切”都枚举一遍,结果图越画越大,代码越写越僵。后来想明白了一件事:Agent 时代的流程设计,不应该是“把路径写死”,而是“给模型一个边界,让它在边界内自由活动”。

这带来的一个直接转变是:工作流里不再有唯一的“控制器”。传统流程引擎里有一个 controller 在决定下一步执行哪个节点,而在轻量设计里,这个“决策权”被下放给了模型。

1.1 轻量 Workflow 的目标到底是什么

我个人对轻量 Workflow 的定义有三条硬指标:

  1. 不依赖外部服务。不需要专门的流程引擎、不需要独立的数据库存流程实例。一套纯代码实现,一个类加一个配置文件就够了。

  2. 状态清晰可见。任何时刻,整个流程处于什么位置、模型已经做了什么决策、积累了哪些中间结果,都能一眼看明白。

  3. 模型决策空间受控但不僵死。模型的跳转自由度受到约束,但这些约束不是靠写死 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): Promise<NodeResult>; canRun(ctx: WorkflowContext): boolean; } interface NodeResult { status: 'success' | 'failed' | 'skip'; updates: Partial<WorkflowState>; 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 对象、不能塞数据库连接、不能塞函数引用。原因很简单:

  1. 方便调试。任意时刻把 State 打印出来,一眼就能看到流程走到了哪里、积累了哪些信息。
  2. 方便回溯。如果流程出问题了,把出错那个时点的 State 保存下来,可以离线复现。
  3. 方便接入日志和观测系统。纯 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 ): Promise<string> { 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: Map<string, WorkflowNode>, maxSteps: number = 10 ): Promise<WorkflowState> { 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 加了两层限制:

  1. memory 数据有上限。比如retrieved_chunks最多保留前 5 段,tool_results最多保留最近 3 个,历史记录只保留最近两轮。

  2. 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 是好东西,但它不是银弹。我用下来的体会是,当出现下面这些信号时,就该考虑迁移到更重的平台或者框架了:

  1. 节点数超过 20 个。节点一多,Router 的选择空间就变大,模型决策准确率会明显下降。这时候就需要可视化的编排界面、更结构化的 DAG 定义来兜底。

  2. 需要多人协作维护流程定义。如果你需要让非工程师也能画流程、改流程,那么轻量的纯代码方案就不太合适了——图形化编排工具是更优解。

  3. 需要持久化流程实例。如果你需要保存每个流程的运行状态,支持中途暂停继续、支持跨服务调用,那就必须引入存储层。轻量方案里那个随请求生灭的 State 就不够用了。

  4. 状态流转复杂度过高。如果你发现某个节点需要判断几十个条件才能决定下一步,而简单代码又写不清楚,就说明业务本身需要一个规则引擎,不再是轻量 Workflow 能覆盖的。

  5. 需要考虑并发和分布式执行。多个流程并行跑、流程之间有依赖关系,这些场景下集中式调度是不可少的,轻量方案支撑不起来。

我的经验是:从轻量方案起步,遇到上面这些信号再迁移,比一上来就上重框架成本低得多,而且你把业务逻辑真正摸透了,迁移到重框架的时候也知道哪些节点需要拆、哪些状态需要保留。

6. 最后再分享一个小技巧

写到这里,我想起一个特别实用的技巧,也是我最近才总结出来的:在 Router 的 prompt 里加入“你已知的信息”和“你还需要的信息”这两个小节。

我发现,很多模型跳错节点,不是因为能力不行,而是因为它不清楚自己已经掌握了什么。你只要把一个简单的信息清单放在它面前:

当前已掌握的信息:

  • 用户输入:xxx
  • 检索到的资料:2 段,主题为 xxx、xxx
  • 工具返回结果:无

根据以上信息,判断是否还需要新的信息,或可以直接回答用户问题。

加上这段之后,Router 的决策准确率肉眼可见地提升了。核心原因是:模型是个“上下文驱动”的系统,你把决策所需的信息摆得越清楚,它做决策的质量就越高。这个技巧成本极低,收益却很稳定,做 Agent 工作流设计的朋友很值得试一试。

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

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

立即咨询