轻量Workflow设计：用最少结构约束Agent决策边界

最近在内部项目里做多步骤 Agent 任务时我把原先那套“通用编排框架”推倒重来换成了一个只有几百行代码的轻量 Workflow 核心结果整个系统反而稳了不少。这篇文章想把“面向 Agent 的轻量 Workflow 设计”这件事完整拆开讲一讲它到底解决什么问题、应该轻在哪里、代码怎么写、以及我在实际落地中踩过哪些坑。如果你正在做 Agent 相关应用觉得“流程越来越难控、出问题不好查”这篇值得读到底。1. 先搞清楚Agent 编排里的“轻量”到底是什么意思1.1 传统工作流和 Agent 工作流的本质差别先说结论传统工作流是把“已知流程”固化成代码Agent 工作流是在“未知路径”上做有边界的决策。这俩的逻辑完全相反。传统场景里一笔审批、一次订单处理路径是确定的你完全可以画出一张精确到每个节点的流程图然后交给引擎执行。流程里没有“不确定性”只有“异常分支”。开发者的核心工作是覆盖分支条件。Agent 场景不一样。大模型本身是概率系统同一个 prompt 同一个输入两次输出可能就不同。你没法提前枚举所有可能的执行路径也没法在编码阶段预置“所有情况下的下一个步骤”。这时候最怕的就是用传统工作流引擎的思路去做 Agent 编排——试图把每一步都钉死反而失去了模型推理的意义但完全让模型自由发挥又会在生产环境里失控。所以“面向 Agent 的轻量 Workflow”要做的事情是用最小的结构约束模型的决策边界而不是替模型把所有决策都做掉。它轻在“约束少、规则少、代码少”但依然能保证主干稳定、失败可恢复、过程可追踪。1.2 为什么重型编排方案会拖垮团队我自己最早用的方案是市面上常见的图编排框架节点、边、状态机、并行分支、人工审批界面都有。看起来很好实际用起来问题不少抽象层多一个节点要套三四层概念类型、配置、执行器、监听器写一个简单调用要查一堆借口。隐式机制多重试策略、超时、缓存、并发控制都由框架在内部处理出问题时你根本不知道它是怎么定的。排错成本高线上链路一旦出问题需要同时看引擎日志、业务日志、模型调用日志还得把这三者的 trace id 对上才能还原现场。团队心智负担大新同学上手要学框架概念而不是学业务这不是什么好信号。后来我把方案换成了自己维护的几百行“轻量编排”只保留五件事步骤定义、顺序执行、条件跳转、失败处理、日志追踪。其他东西全部从代码层去掉。结果反而更稳因为每一个行为都是显式写出来的出问题看代码和日志就能定位。1.3 轻量方案适合什么场景不适合什么场景根据我的实践经验以下情况最适合用轻量 Workflow一个 Agent 任务的主链路步骤数量在 3~10 个之间。比如“收集信息 → 整理分析 → 生成报告 → 质量校验 → 输出结果”。分支决策点明确且有限不超过 5 个。比如“校验通过就进入下一步不通过就重试一次重试还不行就降级”。团队没有专门的平台工程师来维护一个复杂编排系统。你想先把一条核心业务链路跑通、跑稳再考虑复杂化。不适合的场景也很清楚如果流程确实需要上百个节点、多人并行编辑、复杂的子流程嵌套那轻量方案会撑不住。这时候该上重型框架还是得上但大多数团队的 Agent 应用根本没到那个规模提前引入复杂框架只是在给未来制造麻烦。2. 轻量 Workflow 设计的三个关键取舍2.1 流程骨架显式化节点决策交给模型轻量不代表“让模型自由发挥”。我实际的设计原则是——主干显式节点隐式。什么意思Workflow 的骨架步骤顺序、依赖关系、跳转条件、错误处理必须由开发者显式写出来这些是确定的、可测试的规则不能交给模型去推理。而每个节点内部的思考过程、工具选择、输出内容则可以交给大模型去自由发挥。举个例子一个“招聘简历初筛”的 Agent骨架解析简历 → 提取结构化信息 → 按 JD 打分 → 输出结论。节点内简历里哪些内容更重要、打分时怎么权衡这些可以由模型自行判断。如果反过来让模型自己决定“我第一步要不要解析简历、第二步要不要打分”那结果就是每次跑的路径都不一样根本没法做质量保证。流程骨架是确定性资产必须攥在自己手里。2.2 节点粒度控制在“一句话能讲清任务”的尺度节点粒度是我改完第一版之后体会最深的部分。刚开始我把节点做得很大一个节点让模型“完成信息收集与分析总结”结果输出不稳定后来我把节点切得很细每一步只做一个动作结果 Token 消耗翻倍延迟增加上下文也更长。最后我找到的判断标准很简单这个节点能否用一句自然语言讲清楚任务且输入输出都是原子的。举几个例子合适粒度“从简历文本中提取姓名、电话、工作年限”输入是文本输出是结构化字段。不合适粒度“理解简历并根据岗位要求给出综合评估”这其实混合了“提取”和“评估”两个动作应当拆开。也不合适粒度“提取姓名”单步动作太小没必要单独调一次大模型。每个节点内部可以做一些小循环但节点之间保持清晰边界。这样既不会因为步骤太细导致多次模型调用也不会因为步骤太粗导致输出不可控。2.3 失败处理必须是显式的“一等公民”在 Agent 编排里流程正常路径谁都能画明白难的永远是失败之后怎么办。我的设计习惯是每个步骤定义时必须写清楚三种失败行为。我用一个结构体或者配置片段来约束核心就四类策略策略含义适用场景retry重试本次步骤最多 N 次可选退避间隔网络超时、限流、临时不可用skip跳过本步骤使用预先给定的默认值继续非关键步骤失败如生成摘要fallback调用备选步骤或备选模型继续主模型不可用、工具调用返回空raise直接终止整个工作流并返回错误核心步骤失败后续没有继续意义我把失败策略从引擎内部拆到步骤定义里有两个好处第一每个步骤的失败行为在代码里就能看到不需要去翻框架源码第二大模型不会被拉进来做“错误分类决策”错误处理走的是确定性代码稳定可靠。3. 落地实现一个能满足 80% 场景的轻量编排核心这部分直接给能抄作业的实现。我习惯用 Python 来做核心逻辑因为团队里大部分人都会读第三方库生态也够用。3.1 先用数据模型把结构定下来核心离不开三个东西步骤Step、上下文Context、结果Result。一个步骤必须有一个执行函数和一个失败策略。执行函数的输入是当前上下文输出是新的结果数据失败策略决定异常怎么处理。# workflow.py from dataclasses import dataclass, field from typing import Any, Callable, Awaitable import asyncio StepFunc Callable[[dict], Awaitable[dict]] dataclass class Step: name: str execute: StepFunc retries: int 2 timeout: float 30.0 on_error: str raise # raise | skip | fallback | retry fallback_value: Any None # on_errorskip 时使用 dataclass class Workflow: steps: list[Step] max_iterations: int 10 async def run(self, initial_ctx: dict) - dict: ctx dict(initial_ctx) i 0 while self.steps: if i self.max_iterations: raise RuntimeError(workflow exceeded max_iterations) step self._pick_next(ctx) if step is None: break result await self._run_step(step, ctx) ctx[step.name] result i 1 return ctx def _pick_next(self, ctx: dict) - Step | None: # 默认按顺序取第一个实际可以结合 step.when 条件判断 for step in self.steps: if step.name not in ctx: return step return None async def _run_step(self, step: Step, ctx: dict) - dict: last_err None for attempt in range(step.retries 1): try: return await asyncio.wait_for(step.execute(ctx), timeoutstep.timeout) except Exception as e: last_err e if step.on_error retry: await asyncio.sleep(0.5 * (attempt 1)) continue if step.on_error skip: return step.fallback_value if step.on_error fallback: # 这里可以换成你的备选执行逻辑 return {fallback: True, value: step.fallback_value} raise raise last_err这段代码没用什么魔法做的事情只有四件取下一个步骤、执行、处理超时、按策略处理异常。但它已经足够支撑大多数线性 Agent 链路。3.2 用配置分离“流程定义”和“执行代码”纯代码编排有个问题每次改流程都要改代码、走发布流程。我会再套一层轻量的配置描述把流程定义变成纯数据。这里用 YAML 做示例因为在团队协作里比 JSON 更好读# demo_flow.yaml name: resume_screen max_iterations: 10 steps: - id: extract_info type: llm model: gpt-4o-mini retries: 2 timeout: 30 on_error: raise prompt_template: 从以下简历文本中提取结构化信息{input_text} output_schema: resume_schema - id: score_candidate type: llm model: gpt-4o retries: 1 timeout: 40 on_error: fallback fallback_value: { score: 0 } prompt_template: 基于以下客观信息对候选人匹配度打分{extract_info} output_schema: score_schema - id: notify_result type: tool tool: send_email retries: 3 timeout: 10 on_error: skip fallback_value: { sent: false } when: {score_candidate.score} 60这个配置的好处是非工程师也能看懂流程大概做了什么也能直接拿去讨论业务逻辑。我把配置里的type限定为两类值llm和tool。llm节点负责调用大模型并做结构化解析tool节点负责执行确定的工具逻辑发送通知、写数据库、调接口两类节点都返回 dict统一进上下文。3.3 条件跳转和“when”字段的执行逻辑上面代码里的_pick_next目前只是按顺序找没有执行过的步骤但实际流程经常需要条件跳转。我的做法是在步骤配置里加一个可选的when表达式只在条件满足时才把该步骤加入待执行序列。条件表达式我做得非常简单支持字符串格式、数字比较还有简单的and、or。不引入复杂规则引擎因为一旦条件复杂你就应该回去写代码而不是继续堆配置。# evaluate_when.py def evaluate_when(expr: str | None, ctx: dict) - bool: if expr is None: return True parts expr.split() # 支持 60, done, in [1,2,3] 等简单表达 # 实际实现中最好用一个安全的环境来执行 return bool(eval(expr, {__builtins__: {}}, ctx)) # 仅作示例生产环境应限制这个字段很实用。比如“当上一步打分低于阈值时直接走人工流程不继续调用后续模型”这种逻辑用when写在配置里比散落在代码里好维护多了。需要提醒一句eval直接执行有风险我的实际做法是把支持的操作符解析成安全的 AST 求值器配置里只允许出现白名单内的变量名。配置文件虽然是内部使用但线上环境安全不能图省事。3.4 LLM 节点和工具节点必须分开封装一开始我把call_llm和业务逻辑混在一个函数里后来发现两个问题模型输出不稳定导致业务代码跟着乱了以及工具调用失败时不知道应该重试模型还是重试工具。现在我的封装原则是llm节点只做两件事拼 prompt、调模型、解析结构化输出。业务规则不写在里面。tool节点是纯函数输入 dict、输出 dict任何副作用发邮件、写库都在节点内部完成。两类的返回值都必须是 JSON 可序列化的不允许出现“自定义对象直接塞进上下文”的情况。LLM 节点封装代码大概长这样# llm_node.py import json from typing import Callable async def create_llm_step( name: str, llm_func: Callable[[str], str], prompt_template: str, output_schema: dict, retries: int 2, on_error: str retry, ): def make_prompt(ctx: dict) - str: return prompt_template.format(**ctx) async def execute(ctx: dict) - dict: prompt make_prompt(ctx) raw llm_func(prompt) return parse_and_validate(raw, output_schema) return Step(namename, executeexecute, retriesretries, on_erroron_error)结构化输出很重要。大模型返回字符串直接塞给下游格式稍微一变解析就崩了。我要求所有 LLM 节点必须输出符合 JSON Schema 的数据解析失败就重试重试还失败就按失败策略处理。这样上一步的输出必然是下一步能认的输入步骤间才不会因为字段缺失而炸掉。4. 上下文管理、错误处理与可观测性一个都不能少4.1 上下文预算防止 Agent 被自己“撑死”轻量 Workflow 的上下文和 ChatBot 不同它不只是对话历史还包括步骤产出、工具返回值、中间日志。最容易踩的坑是每个步骤都把完整上一步的原始输出塞进 prompt几轮之后上下文爆了。我的处理方式很直接上下文只保留三类信息。用户原始意图精炼后的一句话描述占用小。结构化中间产物每个步骤产出的 JSON 结果只保留必要字段。步骤摘要大模型或确定性函数对冗长输出的压缩摘要而不是原文。举个例子“信息收集”这一步可能抓回来 50 条网页内容我不会把它全量放进下一步 prompt而是先做一个摘要把核心要点整理成 500 字以内的结构化描述再传给下一步。这看起来多调了一次模型但实际上省了后续每一步的 Token且整体响应更稳。4.2 错误分级与重试策略错误处理不能“一视同仁”。我按照实际表现把错误分成三类每一类有专门应对方法。错误类型典型表现应对方式瞬时错误网络超时、API 限流、暂时性 5xx重试 2~3 次指数退避不轻易终止格式错误模型返回内容不符合 JSON Schema用更强的 prompt 约束重试 1 次再失败走 fallback业务错误数据不符合业务条件、关键信息缺失不重试直接 raise 或转人工重试没有意义有个细节容易被忽略模型返回格式错误时重试时不要用同一个 prompt。我会在重试的 prompt 里附带上一次的错误信息和错误片段让模型知道“你刚才哪里不对”。单纯让它再来一次往往还是错同样的问题。4.3 可观测性先有 trace再谈复盘Agent 链路一个显著特点是“每一步都不可预测”所以必须有可观测性。我要求每个工作流运行都生成一份结构化的 trace至少包含全局唯一 workflow_id每个步骤的输入摘要哈希或裁剪后的关键字段执行耗时、模型名、Token 使用量、重试次数每一步的状态success / skipped / failed失败时的错误信息与堆栈具体落地我会在每个步骤执行前后记录日志统一格式类似{ workflow_id: wf_12345, step: extract_info, status: success, duration_ms: 2300, tokens: 1200, attempt: 2, error: null }有了这份 trace出现问题时可以直接提问“哪个步骤最慢哪一步消耗 Token 最多哪一步重试了失败原因是模型问题还是工具问题”这比对着终端日志一行行翻高效得多。我见过太多团队把 Agent 链路的排查变成“猜谜游戏”根因就是没把 trace 做进去。5. 踩坑记录常见问题和排查方法5.1 模型“自作主张”跳过了关键步骤有个常见场景你要求模型“先执行信息校验再生成结果”但模型可能直接跳过校验就开始生成。这类问题不是模型坏而是流程约束不够强。我的解决办法是“前置断言 必选步骤”。每个步骤执行前都会检查它依赖的步骤是否已经执行并产出有效数据如果前置数据缺失工作流直接报错而不是让模型自由发挥。步骤定义里也会区分mandatory: true和optional: true核心步骤必须是 mandatory不允许跳过。5.2 步骤间数据结构不兼容第一版代码里上游步骤输出一个candidate字段下游步骤期望resume字段。模型输出总是飘忽不定偶尔字段名对不上下游就报 KeyError。后来我给每个步骤的输出都定义了 JSON Schema在步骤完成后立即做校验不合法就当失败走重试。这相当于给数据流加了一层“合同”上下游各认各的契约这一点钱不能省。一个团队写测试用例也可以直接基于这些 Schema 生成也方便后续扩展。5.3 循环失控工作流停不下来某一次线上运行Agent 在“改写 → 校验 → 再改写 → 再校验”之间来回了好几轮Token 烧了很多才被人发现。原因是没有给循环写终止条件。我现在对任何可能出现循环的设计都做三重保险执行器层面max_iterations限制总执行轮数超了直接终止。节点层面每个 LLM 节点的 Prompt 里明确写“最多修改 N 次”。数据层面工具返回的内容有长度上限避免超大输入导致模型反复尝试。那之后我再没遇到过“停不下来”的事故。不要把终止条件寄托在模型自觉上承载流程控制的只能是代码。5.4 链路日志太难看懂老版本日志里全是零散的 print 语句和模型返回原文根本没法串联起来。后来我统一引入 workflow_id 和 step_id所有日志都带上这两个标识再配合上面说的 trace 结构排查效率瞬间就上来了。一个小建议模型返回内容不要直接打完整原文到日志截断或者只记录摘要否则日志服务被大文本塞满查问题也费劲。又想要原文又想日志干净可以额外把原始数据存到对象存储里日志里只留一个引用地址。6. 最后的实操心得轻量 Workflow 给我最大的感受是它的价值不在代码量少而在让团队对“Agent 能做什么、不能做什么”达成共识。骨架是确定的节点内部是自由的失败是有兜底的过程是可追踪的一旦这四点做到了再复杂的 Agent 也乱不到哪里去。我每次设计新的 Agent 任务时会先画一张图但只画两条路径——主路径和失败路径。主路径列出 3~10 个步骤失败路径标记每个步骤出错了怎么办。这张图能画出来轻量 Workflow 的骨架也就出来了画不出来说明你对业务理解还不够深应该先回到业务梳理而不是急着写代码。最后分享一个自己的小习惯把 Agent 当成年纪很小但学习能力很强的实习生来设计系统。流程上你要告诉它下一步做什么骨架过程上你要允许它有自己的方法节点内自由出错时你要有预案失败策略事后你要能复盘可观测性。这三件事做到位你会明显感到手头那套“面向 Agent 的轻量 Workflow”已经从“能用”变成一个真正“好用”的底层设计。