做 Agent 落地的同学最近应该都有感受单轮 Prompt 玩得再花也架不住真实业务要拆成好几个步骤来跑。我把设计重心从把提示词写到极致挪到了面向 Agent 的轻量 Workflow 设计上之后很多原本又乱又脆的流程突然变得可以维护、可以排查、可以复现了。这篇文章不是讲怎么搭一个重量级工作流引擎而是分享我实际用的这一套轻量设计思路。适合正在做工具调用、多步任务编排、自动化流程的开发者或者想摆脱一个 Prompt 干所有事这种思路的团队成员。我会尽量用大白话拆解把数据结构、调度逻辑、踩坑记录都摆出来你照着改就能用。1. 为什么 Agent 圈开始认真聊 Workflow1.1 Agent 不是万能串珠而是需要剧本的演员最早做 Agent 的时候大家习惯把所有逻辑塞进一个巨大的 System Prompt你是个助手请根据用户需求一步步思考需要调用工具时先分析再调用……听起来很聪明实际一跑就露馅。模型输出不稳定今天会按三步走明天非要跳步工具调用的参数格式偶发错误用户中途改需求你都不知道该从哪一步接着执行。后来我意识到一个朴素的道理大模型是聪明的执行者但不是可靠的状态管理器。它擅长的是在给定上下文里做判断、生成内容而不是记住现在已经执行到第几步、上一步拿到了什么结果。把这些状态和步骤管理交给程序让模型只负责它擅长的局部任务这就是 Workflow 存在的意义。所谓面向 Agent 的轻量 Workflow 设计我最朴素的理解是用代码把任务拆成一个个节点节点之间按依赖关系执行每个节点可以是调用大模型、执行函数、做条件判断甚至停下来等人工确认。Agent 还是那个 Agent但它的行动路径从自由发挥变成了在剧本框架内发挥。1.2 轻量 Workflow 和传统工作流引擎的边界很多企业里已经有成熟的工作流引擎比如基于 BPMN 的流程引擎或者各种云厂商的 workflow 服务。那些东西非常强大支持复杂的并发分叉、会签、历史版本、权限模型。但拿到 Agent 场景里你会发现问题太重了。传统工作流是为确定性业务流程设计的每个节点的行为几乎完全固定申请、审批、通知。Agent 工作流的节点行为却是概率性的大模型可能产生意外输出。你需要的是轻量、灵活、能够容忍模型不确定性的编排框架而不是一上来就是一堆 XML 定义、流程版本库、集群部署。我在某个模拟项目里甚至见过同事用商用工作流引擎编排 Agent结果光是把一个三十行的 JSON 流程定义导进去就在权限配置上折腾了半天。后来我们拆掉重来自己写了一个只有三个核心类、一百多行的调度器反而足够支撑全部业务。轻量的好处在于你真的能看懂每一步发生了什么出了问题可以直接改代码而不是去研究引擎的配置文档。1.3 什么时候该上 Workflow什么时候不需要不是所有 Agent 都需要 Workflow。我的判断标准很简单如果任务固定只有两三个步骤而且顺序永远不变直接在主程序里写死就行没必要抽象。如果任务超过五个步骤、有分支条件、需要人工介入、或者多个 Agent 要配合那就强烈建议上 Workflow。还有一种隐形需求复现性。单纯靠 Prompt 驱动的 Agent跑同一个任务十次可能有十个不同的过程这对测试和稳定性是灾难。引入轻量 Workflow 之后你可以把过程变成固定的骨架每次只是在关键节点上让模型做决策整个运行轨迹是可控的。很多团队最后选择改造不是觉得功能不够而是觉得复盘成本太高了。2. 设计核心先把执行模型定下来2.1 图模型选型DAG 还是状态机还是两者混合动笔写代码前先想清楚用哪种执行模型。最常见的三种有向无环图DAG节点是任务边是依赖关系一个节点所有依赖完成后才能执行。优点是适合并行、结构清晰缺点是不好表达循环不过 Agent 场景里大部分流程天然是 DAG。状态机把整个流程看作状态的迁移每个事件触发一次迁移。优点是异常控制强可以精确表达什么时候允许重试、什么时候回退到哪个状态缺点是流程复杂时状态爆炸维护累。两者混合用 DAG 表达整体结构在部分节点内部用状态机管理重试和人工确认。这是我最推荐的。实际写的时候我会把 Workflow 定义为一组节点和一组边边决定拓扑关系。至于循环比如生成内容质量不合格则重做我不会在图里画环而是把它建模成一个特殊节点内部自己循环完成后再把出口交给下游。这样既保住了 DAG 的简洁性又不丢失常见的循环重试语义。2.2 节点怎么描述从做什么到怎么做节点是整个设计的地基描述方式会直接影响扩展性。我习惯把每个节点拆成两部分声明declaration和实现implementation。声明是配置比如节点 ID、类型、依赖的输入字段、输出字段、参数。实现是真正的执行逻辑比如一段 Python 函数、一个 LLM 调用的封装、一个写入外部系统的操作。为什么这样拆因为想在日志、编排、测试的时候只看声明不关心实现。我通常用字典来定义节点大概长这样{ id: parse_requirement, type: llm, model: gpt-4o-mini, prompt_template: from_template_xxx, input: [user_request], output: structured_requirement, max_retries: 3 }这种描述方式的好处是节点之间通过字段名显式依赖上游输出对应下游输入完全是数据驱动。后面你要加一个新节点不需要改动调度器只要加一份声明和对应的执行函数就行。如果某个流程特别复杂声明式配置反而别扭我会允许节点直接用代码块包裹逻辑但外部接口仍然统一。原则是调度器永远只认节点接口不认内部实现。2.3 上下文传递避免把状态塞进模型提示词新手最爱犯的错就是把上下文全部塞进下一轮的 Prompt。比如在内容生成流程里先让 Agent 生成大纲再把大纲、历史对话、一堆中间结果全拼进提示词里让它写全文。这样做有三个问题token 成本指数级上升模型容易被冗余信息带偏你无法精确控制它到底用了哪些中间结果。我的做法是设计一个统一的上下文对象理解为整条流程的工作台。每个节点从工作台读取自己声明的输入字段执行完把结果写回工作台。模型节点的 Prompt 模板里只引用它真正需要的字段而不是把整个工作台丢进去。class WorkflowContext: def __init__(self, init_data): self.data dict(init_data) def get(self, field): return self.data.get(field) def set(self, field, value): self.data[field] value这样你审计流程的时候只要看每个节点的 input / output 声明就能确认数据链路。有人问如果某个节点需要访问全局信息怎么办我一般把全局信息跟流程级变量分开放进context.meta让模型节点只读取显式指定的入参减少意外。3. 轻量实现的关键细节3.1 调度内核用队列还是用递归调度器是执行模型落地的地方。对于轻量实现我不建议用递归因为节点多、嵌套深的时候调用栈不清晰异常也容易堆叠。推荐的做法是先对 DAG 做一次拓扑排序然后维护一个就绪队列——每次找出所有依赖都已完成且自身未执行的节点放进队列依次执行。如果节点没有依赖一开始就进队列每完成一个节点就检查它的下游节点是否需要被激活。周期重复直到所有节点完成。稀疏标记方法我写了个 schema 示例不过核心逻辑都是用队列简单可靠from collections import deque def run_workflow(workflow, init_data): ctx WorkflowContext(init_data) node_map {n[id]: n for n in workflow[nodes]} next_map build_next_map(workflow[edges]) dep_map {n[id]: set(dep) for n, dep in collect_dependencies(workflow[edges])} ready deque([nid for nid, deps in dep_map.items() if not deps]) completed set() while ready: nid ready.popleft() node node_map[nid] result execute_node(node, ctx) ctx.set(nid, result) completed.add(nid) for downstream in next_map.get(nid, []): dep_map[downstream].discard(nid) if not dep_map[downstream]: ready.append(downstream) return ctx这是我经验里最稳的调度器骨架。它没有处理优先级、并发、超时因为轻量设计不追求一次性把复杂度做满先把主流程跑通再说。优先级问题在队列里按类型排序即可并发问题后面单开章节聊。3.2 失败重试与分支回退Agent 工作流里模型调用失败太常见了。网络超时、JSON 解析失败、内容安全过滤拦截都可能让一个节点失败。失败不能直接终止整个流程要有重试重试也不是简单重跑一遍就完事还涉及退避、幂等和上下文回退。我的重试策略分两层第一层节点内部重试。比如模型调用连续失败三次第三次失败前每次等待时间按指数退避1s、2s、4s。第二层流程级处理。如果节点重试仍失败我会判断这个节点是否定义了对失败的回落方式。比如LLM 提取信息失败可以回落到规则正则函数调用失败可以跳到人工节点。def execute_node(node, ctx): last_exc None for attempt in range(node.get(max_retries, 1)): try: return invoke_node_implementation(node, ctx) except Exception as e: last_exc e time.sleep(2 ** attempt) return handle_node_failure(node, ctx, last_exc)分支回退更有意思。我以前接到一个需求先让 Agent 做意图判断如果判断为咨询就走问答分支判断为下单就走电商分支但半路发现信息不足要退回意图判断节点重新识别。这就需要在上下文里记录当前节点链。轻量方式是在 Workflow 里增加一个特殊边表示回退目标遇到失败时顺着这条边跳回去。状态机的优势在这里就体现出来了所以我前面强调混合模型你在图结构之外保留少量状态信息处理回退会舒服很多。3.3 人工审批节点的设计Agent 越聪明人在关键环节的确认越重要尤其涉及发消息、下单、删除数据这种不可逆操作。人工节点我通常设计成三类状态等待中、通过、拒绝。本质上它把流程挂起等待外部回调来唤醒。实现上不需要复杂。我会把节点执行状态持久化到一个存储里同时生成一个 token 作为审批凭据推送给人工处理端。处理端调用confirm(token, decision, comment)接口调度器再重新激活对应的节点。注意流程挂起期间工作流占用的资源需要释放不要让线程阻塞在那里干等而是用异步回调唤醒。在设计审批节点时还要考虑超时用户一直不审批怎么办。我给每个审批节点配一个timeout_policy可以是自动通过、自动拒绝、或者转交给替代人。这在大厂工作流引擎里属于常规能力但轻量实现里很多人会漏掉导致流程卡死。3.4 可观测性日志、追踪、运行快照最后一块核心设计是可观测性。Agent Workflow 比起传统接口不确定性很高没有日志你根本不知道模型为什么做出这个决定。我会在每个节点执行时记录四样信息节点开始时间、结束时间、耗时读取的输入字段和值节点输出如果是模型节点同时记录原始响应和解析后的结构化结果重试次数和失败原因。运行快照snapshot也很有用每隔一段时间或者每个节点完成时把当前WorkflowContext的完整内容保存一份。这样一旦流程出错你可以精准回复现场而不是靠猜测中间变量是什么。轻量实现里快照直接写成 JSON 存本地磁盘或者对象存储都行成本很低。提示不要偷懒只打日志不存快照。模型节点的中间输出有时候会被安全过滤或者被下游节点覆盖没有快照你将无从排查。4. 一套可直接借鉴的最小实现4.1 节点类型定义我实际项目里常用的节点类型不算多下面这张表是我在多个项目里沉淀下来的一套最小集合节点类型作用典型配置llm调用大模型输出结构化结果model、prompt_template、response_formatfunction执行普通函数比如查数据库、调接口runnable、input字段映射condition判断分支根据字段值决定走哪条边expression、brancheshuman人工审批/信息补充approver、timeout_policyloop内部循环执行子流程直到条件满足sub_workflow、max_iterationstransform纯数据变换比如格式化、拼接mapping规则节点类型不要贪多。每增加一种类型调试成本都跟着翻倍。先用最少的类型跑通等确实出现了新需求再加一种对应类型并重新审视是否能用现有类型组合实现。4.2 完整示例内容生产 Agent拿一个沾边的内容生产 Agent 举个例子。任务根据用户给的主题生成一篇带标题、摘要、结构和正文的文章正文写完需人工确认确认后再发布到指定系统。整个流程拆成六个节点主题解析llm 节点把用户主题解析成结构化需求目标读者、字数、风格标签资料整理function 节点根据结构化需求调用搜索接口拉回若干参考资料生成大纲llm 节点基于需求资料输出大纲数组大纲确认human 节点人工审阅大纲可以批准或者要求重写生成正文llm 节点输入大纲输出 Vue 风格的正文内容当然只是正文文体不是前端发布function 节点调用发布接口写入内容库。这几个节点用 DAG 串起来非常直观。代码就不全贴了数据流大概是workflow { nodes: [ {id: topic_parse, type: llm, input: [topic], output: requirement}, {id: material_collect, type: function, input: [requirement], output: materials}, {id: outline_generate, type: llm, input: [requirement, materials], output: outline}, {id: outline_confirm, type: human, input: [outline], output: confirm_result}, {id: article_generate, type: llm, input: [outline, confirm_result], output: article}, {id: publish, type: function, input: [article], output: publish_url} ], edges: [ (topic, topic_parse), (topic_parse, material_collect), (material_collect, outline_generate), (outline_generate, outline_confirm), (outline_confirm, article_generate), (article_generate, publish) ] }注意edges的第一条(topic, topic_parse)我是直接拿外部输入当源节点处理的这只是个记号真正实现时我会在调度器里把输入当作已完成的虚拟节点。4.3 接入真实 Agent 时的上下文约定接入真实模型时最容易遇到的问题就是上下文对象与 Prompt 模板如何拼接。我的约定如下每个 LLM 节点从上下文中读取声明好的字段把它格式化成 Prompt 模板的变量模板里只允许引用该节点声明过的输入字段模型返回结果如果是 JSON必须走一次结构校验器校验失败就重试或者走回落逻辑。def invoke_llm_node(node, ctx): variables {k: ctx.get(k) for k in node[input]} prompt render_template(node[prompt_template], variables) raw call_llm(node[model], prompt, response_formatnode.get(response_format)) if node.get(response_format) json: parsed safe_parse_json(raw) if parsed is None: raise ValueError(model returned invalid json) return parsed return raw很多人说模型返回 JSON 不稳定的问题我的经验是先在 Prompt 里要求只输出 JSON不要用 markdown 代码块包裹然后用safe_parse_json兜底解析失败时截取大括号部分重试。这条建议能解决大部分问题。4.4 并发执行的轻量方案如果某个流程里有两个节点互相没有依赖比如同时搜集 A 渠道和 B 渠道的资料轻量调度器怎么处理答案是用线程池。我一般会在架构里给调度器传入一个executorfrom concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers4) def run_workflow_parallel(workflow, init_data): ctx WorkflowContext(init_data) # ready 队列同一时刻可能包含多个节点 futures {executor.submit(execute_node, node, ctx): nid for nid, node in get_ready_nodes()} while futures: done, _ wait(futures, return_whenFIRST_COMPLETED) for f in done: nid futures.pop(f) try: result f.result() ctx.set(nid, result) activate_downstreams(nid, ready_queue) except Exception: handle_failure(nid) # 持续提交新的 ready 节点并发引入后问题会成倍增加数据竞态、结果写回顺序、共享上下文污染。我的建议是并发节点必须在声明里显式标注parallel: true并且彼此读写字段不能重叠在提交时校验一遍。宁可在最开始的时候限制并发度小一点也不要一上来开 32 线程后面排查起来头大。5. 常见问题与排查技巧实录5.1 循环依赖检测即使 DAG 是设计目标人总有手滑的时候。我遇到过一个案例调试流程时发现节点 A 等 B、B 等 C、C 又等 A调度器直接死循环了队列永远处理不完。后来我在流程加载时强制做一次拓扑排序排序失败直接报错列出环路成员。def topo_sort(nodes, edges): indeg {nid: 0 for nid in nodes} adj {nid: [] for nid in nodes} for src, dst in edges: indeg[dst] 1 adj[src].append(dst) queue deque([nid for nid, d in indeg.items() if d 0]) result [] while queue: cur queue.popleft() result.append(cur) for nxt in adj[cur]: indeg[nxt] - 1 if indeg[nxt] 0: queue.append(nxt) if len(result) ! len(nodes): raise Exception(cycle detected in workflow) return result这个检查放在工作流注册阶段启动即报错不要等到运行时才暴露。5.2 模型输出不稳定JSON 解析经常挂这个问题几乎每个做 Agent 的都碰到过。我的处理分三层第一层Prompt 里限定输出格式并给 example而不是只给 schema 描述。第二层解析器做容错比如剥离 markdown 代码块标记、自动修复缺失尾部括号这种能力可以自己写几十行也可以用第三方解析库。第三层解析失败按重试机制跑一次重试时在 Prompt 末尾追加一句提示你上一次返回的 JSON 格式无法解析请严格按要求的格式输出 JSON。实测下来三层叠加能把 JSON 解析成功率从不到八成拉到接近百分百。剩下一小部分就落到人工兜底或者规则兜底。5.3 上下文越跑越大token 成本飙高很多 Agent 工作流做成链式之后上下文对象会不断累积所有节点的输出。到最后一个节点它明明只需要一个字段却要把前面十步的中间结果全带上白白烧 token。我一般约定每个节点执行完后可以声明这个节点输出是否长期存活。一些临时字段可以在流程结束后清理或者只在工作台里保留最近 N 个字段。LLM 节点的输入必须显式字段引用不允许全部注入这种偷懒写法。5.4 并发节点的幂等与重复执行并发 失败重试会导致一个节点被重复执行。如果你的 node 实现里有副作用比如发送通知、写数据库重复执行就是事故。轻量设计里我会给有副作用的节点加一个执行 IDrun_id并且让外部系统支持幂等键。每次执行前检查幂等键如果已经处理过就直接返回旧结果。这一步很多初学者会漏掉觉得调接口而已没什么大不了的。等你真的因为一次网络超时重试给客户发了两遍通知就会乖乖补上幂等设计了。6. 几个容易踩的坑提前帮你排掉最后一个部分把我在实际项目里踩过的坑一次性列出来。不要把所有逻辑一股脑塞进 Workflow。有些简单的顺序调用硬套框架反而增加理解成本只有节点数量、分支复杂度到达临界点再上。不要在 Workflow 里写太重的前置逻辑。解析、校验这类操作尽量在流程入口前完成让每个节点职责单一。Prompt 模板版本也要纳入配置管理。模型节点的输出质量跟模板强相关我在模板文件里固定了version字段改动模板时同步更新方便回溯。人工节点挂起时的时间处理要注意跨天、夏令时等情况。监控看板一定要画不然轻量系统跑起来完全黑盒出问题根本找不到罪魁祸首。我自己的习惯是给每个工作流加一个简单的统计页各节点执行次数、平均耗时、失败次数、重试次数。别看数据简单很多时候模型某个输出 schema 频繁解析失败就是靠这些统计先暴露出来的。轻量不代表不要运维核心是最小必要的运维。最后分享一个小细节调度器启动前我会把要跑的流程定义整体做一次 dry-run也就是用假数据跑一遍检查节点之间的字段引用是否都存在、模型节点的响应 schema 是否合法。这个步骤跑完之后真正的流程上线心里会踏实很多。平时迭代新节点只要 dry-run 过基本不会出现到了生产环境才发现字段拼错的尴尬。面向 Agent 的轻量 Workflow 设计本质不是一件大工程它更像是给 Agent 画了一条条清晰的路径然后再给这条路装上护栏。你完全可以从今天开始用几个类、一个队列、几个字典搭出第一个可用的版本然后慢慢加分支、加人工节点、加并发。先把骨架立起来剩下来的填肉活儿就都简单了。