1. 为什么我放弃重框架转头自己写轻量 Workflow大概从去年开始身边讨论 Agent 的人一下子多了起来。各种编排框架、可视化流程平台一个接一个冒出来功能确实强大节点拖拖拽拽就能拼出一个多智能体协作的复杂链路。但说实话等我自己真正上手做项目时那种大而全的方案反而让我有点喘不过气。如果你也遇到过下面这些情况我相信你能懂我在说什么框架本身的抽象概念特别多文档几百页光是搞懂“状态机”“会话窗口”“上下文总线”这些概念就得花掉大半天从项目初始化到跑通一条最简单的 Agent 调用链中间要跨越一堆配置项和插件依赖稍不留神就报错最难受的是当你只是想快速验证一个“查资料 - 总结 - 发通知”的三步流程时框架的启动开销反而比业务流程本身还重。我当时的处境是手上有几个业务场景需要接入大模型能力但不希望被某个特定框架锁死也不想为了一个轻量需求扛着一整套重型武器跑。于是我开始思考一个问题能不能用最朴素的方式为 Agent 设计一套“刚刚好够用”的轻量 Workflow这里的“轻量”指的不是功能残缺而是指核心抽象足够少、依赖足够可控、跑起来足够快。我最终自己动手搭了一套并顺利跑通了几个真实场景。这篇文章把设计过程和踩过的坑整理出来希望对正在纠结“要不要自研工作流”的你有点参考价值。2. 工作流设计之前先想清楚 Agent 和 Workflow 的关系很多人一说起 Agent 和 Workflow第一反应是二选一要么纯让 Agent 自己自由发挥要么把一切流程固化成死板的流水线。实际上真实生产项目里两者是配合关系不是对立关系。2.1 Agent 负责“怎么想”Workflow 负责“怎么走”我用一个类比来解释。假设你要开一家餐厅Workflow 相当于后厨的动线设计点单进来后传给哪个档口、炒菜和装盘怎么衔接、传菜口在哪这些都是提前规划好的目的是让整条链路稳定、可控、不出错。而 Agent 就像一个经验丰富的厨师长遇到“客人说不要香菜但菜单上没写这个选项”这种非标准需求时需要临场判断、灵活应变。回到技术场景Agent 的能力在于理解复杂指令、拆解模糊任务、调用工具做出决策。但如果你让它在一个完全开放的空间里自由行动结果往往不可控——它可能绕来绕去可能会漏掉必要的校验步骤也可能突然跑偏去做无关的事。所以我们需要 Workflow 来划定边界、固定主路径确保关键的业务步骤一定被执行、执行顺序一定正确。反过来说纯固定的 Workflow 又显得太僵硬。业务是变化的API 参数可能要调整用户输入的方式千奇百怪你不可能把所有分支都预先写死。这时候就要在 Workflow 的关键节点上嵌入 Agent 的智能判断能力让它在多个候选动作中做选择、在异常情况下做兜底处理。2.2 轻量 Workflow 的三个核心要素结合我的实践一个面向 Agent 的轻量 Workflow 只需要具备三个核心要素其他都是可选项。第一明确的执行入口和出口。Workflow 必须有清晰的输入输出边界不能和外部系统的数据流搅在一起。输入就是一次任务请求输出就是一个确定的结果对象。第二节点是最小执行单元。每个节点只做一件事比如“调用搜索工具”“调用大模型生成摘要”“发送企业微信通知”。节点之间通过统一的上下文对象传递数据。第三可控的流转机制。这里的“可控”包括默认顺序执行的模式、基于条件的分支跳转、出现异常时可以重试或降级。我见过的一些团队会在 Workflow 里塞进事件总线、分布式锁、持久化队列拜托那是任务调度系统的范畴了。我的主张是先跑通核心闭环再按需增加复杂度。轻量的意义正在于让新成员接手代码时能在半小时内读懂全貌。2.3 什么时候你应该考虑自研什么时候不建议这不是广告而是真诚的劝退。如果你只是想要一个快速的 Demo现有开源框架完全能覆盖没必要自己造轮子。自研 Workflow 的价值主要体现在这几种情况一是你对框架的抽象层不满意希望流程逻辑以最直观的代码方式呈现而不是把流程定义藏在配置文件和数据库表里。二是你需要深度定制比如某些节点的超时策略、重试算法、上下文的序列化方式跟框架的默认行为差异太大改造框架的成本大于自己维护一套轻量实现。三是你的场景特别简单简单到使用框架反而显得大材小用团队学习和理解成本不值得付出。当初支撑我下定决心的还有一个判断标准未来可能的改动到底集中在哪些地方。我发现自己 80% 的改动都会落在“节点内部的处理逻辑”上而很少去调整“节点之间的连接关系”。这意味着我把精力重心放在节点实现层是更划算的选择。3. 设计一套轻量 Workflow 的完整思路基于上面的认识我开始设计自己的轻量 Workflow。目标很明确让一个刚接触项目的人看完一个工作流定义文件后就能大致说出这个流程在做什么。3.1 先界定边界这个 Workflow 不做什么设计之初最难的不是想清楚要做什么而是想清楚不做什么。我给自己定了三条禁令。不用外部存储来保存运行状态。也就是说工作流运行过程中不会把每一步的执行状态写进数据库。运行期间的上下文只保存在内存中任务结束后就销毁。如果需要审计日志我在业务侧额外记录关键节点的出入参即可。不做跨机器的分布式编排。所有节点跑在同一个进程内用多线程处理并发而不是把节点拆成微服务通过网络调用。这条约束对整个架构影响巨大。不引入可视化拖拽编辑器。我知道很多人觉得可视化是标配但对轻量场景来说代码本身就是最精确的流程描述。用 YAML 或 JSON 描述流程虽然直观但一旦需要表达复杂条件逻辑会变得非常拧巴。代码定义流程配合入参说明文档已经足够清晰。3.2 核心抽象Node 与 Context整个 Workflow 模型简化为两个核心抽象。节点Node是一个接口有一个统一的方法签名大致长这样class Node: def run(self, ctx: Context) - Context: passContext 是一个字典风格的上下文对象贯穿整个流程。每个节点从 Context 里读取自己需要的输入处理完后把结果写回 Context然后流程调度器决定下一个该执行哪个节点。整个模块只需要三样东西节点的实现、流程的编排定义、Context 的传递。没有服务中心没有注册中心没有复杂的生命周期管理。3.3 流程编排最朴素的图结构流程编排的表达方式我参考了有向无环图DAG的思想但不追求通用性和完备性。一个工作流定义包含两个部分节点集和边集。workflow { nodes: [ {id: fetch_data, type: search_api}, {id: generate_summary, type: llm_call}, {id: send_notify, type: webhook_sender} ], edges: [ {from: fetch_data, to: generate_summary}, {from: generate_summary, to: send_notify} ] }这里用字典而不是用类是为了让流程定义可以直接 JSON 序列化。我甚至考虑过把流程定义存成 JSON 文件来加载运行后来发现直接用 Python 数据结构定义反而更方便因为可以直接引用函数对象不用做字符串到类的映射解析。3.4 条件分支与错误处理够用就好很多简单工作流不需要复杂分支但在实际生产场景里完全线性的流程是不存在的。比如搜索结果为空时应该跳过总结节点直接返回一个提示调用大模型超时可能需要选择备选模型重试。我的设计思路是用一个前置条件函数condition来控制边是否能走。每条边除了携带目标节点 ID还可以带一个条件函数返回布尔值决定是否继续。edges [ {from: fetch_data, to: generate_summary, condition: lambda ctx: ctx.get(has_data)}, {from: fetch_data, to: return_empty, condition: lambda ctx: not ctx.get(has_data)} ]这个设计的简洁之处在于条件判断就是普通函数逻辑和业务代码完全融合不存在额外的规则引擎或 DSL 语言。对于轻量场景函数即规则比什么都直观。4. 实操30 分钟手写一个能跑通的 Agent 工作流理论说太多容易飘我直接带你走一遍我实际写过的代码。场景选一个最常见的用户提问 - 检索本地知识库 - 生成回答 - 记录日志。这样一个 4 节点的流程正好能看出轻量设计在不同环节的取舍。4.1 第一步定义上下文对象上下文是整个工作流的血管所有数据都靠它传输。我用一个简单的类来封装其实内部就是一个 dict 加了个访问限制class Context: def __init__(self, initial_data: dict None): self._data initial_data or {} def get(self, key: str, defaultNone): return self._data.get(key, default) def set(self, key: str, value): self._data[key] value def to_dict(self): return self._data.copy()没有复杂的模式校验没有类型强制。初期靠团队约定和代码评审来保证 key 的一致性跑几天后你就自然知道哪些 key 是核心数据需要固化成常量。我后来把常用 key 提取成了枚举避免字符串散落到各个节点里。4.2 第二步实现基础节点节点是真正干活的单元。成本最低的做法是把“调用大模型”“调用工具 API”这些通用能力封装成基类业务相关逻辑写子类。class LLMNode(Node): def __init__(self, modelgpt-4o, temperature0.3): self.model model self.temperature temperature def run(self, ctx: Context) - Context: prompt self.build_prompt(ctx) response call_llm_api(prompt, modelself.model) ctx.set(llm_output, response) return ctx def build_prompt(self, ctx: Context) - str: # 子类覆盖此方法拼装具体的提示词 raise NotImplementedError实现一个具体的总结节点时你只需要继承这个基类然后实现 build_prompt。看起来像模板方法模式其实本质上是把可变的和不变的逻辑分开。可变的提示词构造逻辑留在子类里不可变的 API 调用和超时重试逻辑收拢到基类里。4.3 第三步实现调度器调度器是整个工作流的心脏。它接收一个入口节点 ID 和一个初始 Context然后沿着边逐个执行节点。我最初的实现版本只有不到 50 行代码核心逻辑就是一个 while 循环加一个 for 循环class WorkflowEngine: def __init__(self, nodes: dict, edges: list): self.nodes nodes self.edges edges def run(self, entry_node_id: str, ctx: Context) - Context: current_id entry_node_id visited set() while current_id: if current_id in visited: raise LoopError(floop detected: {current_id}) visited.add(current_id) node self.nodes[current_id] ctx node.run(ctx) # 找从这个节点出发的所有边 next_nodes [] for edge in self.edges: if edge[from] current_id: cond edge.get(condition) if cond is None or cond(ctx): next_nodes.append(edge[to]) if not next_nodes: current_id None elif len(next_nodes) 1: current_id next_nodes[0] else: # 多分支场景暂时只支持顺序执行所有分支 # 实际生产里这里可以改为并行执行或选择主分支 current_id next_nodes[0] self._branch_stack next_nodes[1:] # 简化起见这个版本不做分支栈处理 raise NotImplementedError(parallel branching not implemented) return ctx这个版本故意简化了对多分支的处理。实际项目中分支往往是互斥的if/else 结构所以我后续改造成了一个简单的决策表而不是让调度器维护复杂的状态栈。4.4 第四步组装并跑通全流程把节点和边拼装到一起就得到了一个可以运行的工作流实例nodes { parse_query: QueryParseNode(), search_kb: KnowledgeBaseSearchNode(), generate_answer: AnswerGenerationNode(), write_log: LogWriteNode(), } edges [ {from: parse_query, to: search_kb}, {from: search_kb, to: generate_answer, condition: lambda ctx: ctx.get(search_hit, False)}, {from: search_kb, to: write_log, condition: lambda ctx: not ctx.get(search_hit, False)}, {from: generate_answer, to: write_log}, ] engine WorkflowEngine(nodes, edges) result engine.run(parse_query, Context({query: 什么是轻量工作流})) print(result.to_dict())整个流程跑起来后我最大的感受是这种代码即流程的组织方式调试体验非常顺畅。我可以直接在节点里打断点直接查看 Context 里当前的数据不需要在可视化的黑盒界面上猜来猜去。对于“把流程弄清楚”这个诉求没有比原生代码更直接的方式。5. 我把 Agent 的灵活性嵌进 Workflow 的关键一步这个设计里有一处转折点让我想了很久如果所有节点都是实现好的固定逻辑那么 Agent 在哪里Agent 的智能性、自主决策能力表现在哪里5.1 Agent 节点把“模型决策”封装成普通节点我把 Agent 描述为一种特殊的节点类型。它不是固定的 if-else 逻辑而是调用大模型让模型根据当前 Context 里的信息从预先定义好的候选动作中选出下一步要做什么。这样做的好处是Workflow 的骨架仍然稳定可控但每个节点内部可以足够智能。我不用推翻流程结构去迁就 Agent 的自由发挥而是把 Agent 放在了一个“节点”的盒子里让它在这个盒子的边界内做决策。具体实现上一个 AgentNode 通常包含这么几样内容候选动作列表对应后端实现好的几个工具函数、系统提示词说明当前任务背景和决策规则、决策结果解析逻辑由于模型返回的可能是自然语言需要提取出结构化指令。5.2 一个真实样例客服工单分类与流转举个例子。假设我在做一个客服系统的工单自动分流流程。传统 Workflow 可能是这样的检查工单文本 - 使用正则匹配关键词 - 依据关键词判断类型 - 转给对应处理组。这套逻辑的问题很明显用户描述千奇百怪关键词匹配的规则永远补不完。我把分类这一步换成了 Agent 节点情况立刻好了很多。class TicketClassifyNode(AgentNode): def __init__(self, classes): super().__init__() self.classes classes def build_prompt(self, ctx): ticket_text ctx.get(ticket_text) return f 你是工单分类助手。请阅读用户提交的内容判断它属于以下类别之一 {, .join(self.classes)} 用户内容{ticket_text} 仅输出类别名称不要输出其他内容。 对应地Workflow 的边会根据分类结果走到不同分支edges [ {from: classify, to: handle_refund, condition: lambda ctx: ctx.get(category) 退款}, {from: classify, to: handle_technical, condition: lambda ctx: ctx.get(category) 技术故障}, {from: classify, to: manual_review, condition: lambda ctx: ctx.get(category) 其他} ]结构还是那个结构但节点内部的“分类规则”不再是死板的字符串匹配而是变成了模型的语义理解能力。这让我想到一个更本质的问题轻量 Workflow 系统稳定性的边界在哪里当边界内的决策点交给 Agent 时系统的可预测性是否会下降我的答案是会但可以通过约束候选集、要求输出格式、增加人工确认环节来对冲。系统永远不该完全信任模型的输出Workflow 的职责恰恰是让模型只能做有限的选择且在输出不合法时走兜底分支。5.3 让模型输出结构化结果的细节技巧这里单独提醒一句Agent 节点想让结果稳定可用关键不在于提示词写得有多花哨而在于输出解析是不是足够宽松。我经历过不少次模型输出“退款”带了个句号或者输出“技术故障\n\n以上是我的判断”导致解析失败。解决方案很简单解析时用包含匹配而不是精确匹配。先把模型输出去掉空白和标点然后遍历候选类别只要候选字符串出现在输出文本里就算命中。某些场景下甚至可以要求模型输出 JSON但那样又增加了解析失败的可能性。权衡之后我在非必要场景下更倾向于简短的纯文本输出。6. 实操中躲不开的三个深坑与解决办法设计是一回事运行起来是另一回事。下面这几个问题是我在项目跑了一阵子之后真实踩到的如果你打算动手写轻量 Workflow提前知道能省很多时间。6.1 上下文里的数据污染问题Context 从入口传到出口所有节点都能读写这意味着任何一个节点都可能不小心覆盖其他节点写的数据。比如 A 节点设置了 key resultB 节点不理解这个 key 的含义也往上面写等 C 节点读的时候取到的就不是 A 的结果了。我为此设计了两条约定 第一节点的输出统一使用“节点ID.result”这种命名空间避免全局扁平化的 key 冲突。 第二在节点 run 方法的开头断言输入中必需的 key 存在不存在就快速失败而不是带着空数据往下跑。这两条约定用代码实现起来很简单。命名空间靠约定断言靠写公共工具函数。成本很低但确实避免过好几次冤案。我记得有一次线上问题查了半天最后发现是某个节点无意间改了一个同名 key 被其他节点读取排查的痛苦在前期做好规范后完全可以避免。6.2 超时和重试不能一股脑罩在所有节点上调用外部 API 和大模型接口时超时与重试是必须的。但每个节点的超时策略差异很大搜索接口 5 秒没返回基本没戏大模型接口可能 30 秒是常态发送企微通知如果 3 秒没成功可能说明网络有问题。如果工作流引擎统一设置超时时间就会出现要么大模型任务频繁失败、要么通知任务白白等待的情况。我在实践里把超时和重试策略作为每个节点自身的配置参数由节点内部负责引擎不插手。这样设计还有个额外好处单测可以方便地 mock 节点不需要真的等待超时才能验证分支逻辑。class BaseNode(Node): def __init__(self, timeout: float 10.0, max_retries: int 1): self.timeout timeout self.max_retries max_retries def run(self, ctx: Context) - Context: for attempt in range(self.max_retries 1): try: return self.execute(ctx) except Exception as e: if attempt self.max_retries: raise6.3 边界情况Agent 决策结果与 Workflow 分支不匹配这是最让我头疼的一种情况。Agent 节点输出了分类结果但 Workflow 里并没有定义这个分类对应的分支边于是流程走到尽头Context 里的数据没有后续处理整个任务静默失败。尤其当分类清单由大模型动态生成时这种情况更容易发生。我最后的处理方案是先固化一个顶层分类清单模型的输出必须严格属于清单中的一项。清单本身不开放给模型随意扩充最多是增加一个“其他”兜底项。“其他”分支再进一步转发到人工处理这样一来流程永远不会因为没有边而中断。7. 轻量 Workflow 后续还能往哪些方向扩展写到这里估计你已经发现这套轻量设计只是给自己的工作流打了个地基。它未来想承载更多复杂场景时有几个明确可走的方向。一个方向是增加并行节点执行机制。有些场景下多个互补关系的工具调用并不存在前后依赖比如同时检索多个数据源、同时对一段文本做多个角度的分析。并行能显著缩短整体响应时间。我后续的做法是在边定义里增加一个 group 标记同一个 group 的节点分发到线程池并发执行group 结束后再合并结果进入后续节点。这个平行扩展对原调度器的侵入不算大因为只有节点的调度时机变了节点本身的实现不用改。另一个方向是引入人工审核的等待机制。Agent 可以做决策但涉及转账、发文、外呼这类高风险动作始终会希望有一个人工确认的环节。此时 Workflow 不能一路跑到底而是该在某个节点上暂停等待外部回调再继续。我给引擎加了一个 wait_event 方法在需要停留的节点上打标记任务先序列化成待恢复状态等外部系统确认后重新注入执行。这个能力对生产环境非常重要值得后续单独写一篇展开讲。还有一个方向是完善可观测性和审计追踪。轻量方案初期可以不存状态但一旦流程不仅要跑通还要跑对就需要把每次运行的入参出参、节点耗时、异常信息记录下来。最简单的做法是在引擎层面切一个钩子每执行完一个节点就往日志结构里追加一条记录最终随着结果对象一同返回。这个钩子的成本相比加数据库依赖要小得多却能在排查问题时给关键线索。8. 这套设计的成本账和适用红线聊到最后我掏心窝子说一段。自研轻量 Workflow 并不是免费的午餐它也有机会成本。框架能帮你解决的事情自己写不一定能省事甚至可能更折腾。先说隐性成本。你省下来的学习代价会转移成维护代价。虽然核心引擎只有一两百行代码但周边配套——节点实现、条件函数、错误处理、日志——会随着业务迭代越积越多逐渐变成一笔需要持续维护的资产。这个账要算清楚。再说适用边界。我认为轻量 Workflow 最舒服的区间是单个进程内 5 到 30 个节点、执行频率中等、单次运行时间控制在数分钟内。如果节点数量到几百个、任务需要长时间挂载、多个服务之间需要高可靠的消息传递这套方案就明显吃力了。到那个阶段应该及时切换到成熟的分布式任务编排系统硬扛反而是技术债。从我个人的实践经验来看先动手写好一个轻量内核再基于它去理解复杂框架的设计是条性价比很高的路。你能亲手感受到上下文传递、条件分支、节点复用的痛点在哪里等再去读那些框架的文档时也会更容易理解它为什么要做那些抽象。这就好比你先骑过自行车再学摩托摔的跟头少很多。如果条件允许我建议把流程定义和节点实现分开保存流程定义用数据格式描述节点实现用普通代码承载。只要守住这条分层原则后续无论切换到哪种框架迁移成本都能控制在重写一份流程配置的范围内。预判到这一点你设计时会更从容。