最近半年我一直在做 Agent 相关的项目，发现一个很有意思的现象：理念上大家都觉得 Agent 应该有自主规划和决策能力，可真落地上线时，又都在想方设法限制它的自由发挥。比如让模型自己决定调用什么函数、按什么顺序执行，听起来很美，但模型的一次意外输出就可能让整个流程崩掉，排查起来还特别痛苦。后来我逐渐把重心从"大而全的智能编排"转到了"面向 Agent 的轻量 Workflow 设计"上，简单说就是：保留 Agent 的灵活性，但把它限定在一个可控、可观测、可恢复的流程骨架里。这篇文章就聊聊我在这个方向上的设计思路、踩过的坑，以及一份可以直接抄作业的最小实现。

如果你也在做 Agent 应用，或者单纯被复杂的流程编排平台折磨过，这篇文章应该能给你一些参考。我会从核心抽象讲起，然后给出一个约 150 行的 Python 参考实现，最后复盘哪些设计在真实项目中真正经住了考验，哪些被我自己推翻了。

1. 为什么非要一个"轻量"的 Agent 工作流

1.1 从一次线上事故说起

当时我在维护一个基于大模型的内容处理系统，架构很简单：接收文章 -> 让模型做摘要 -> 提取关键词 -> 入库。第一版代码就是顺序调用几个函数，中间夹着 prompt 调用，跑了两个月没出过大事。直到有一天，模型在摘要步骤里突然开始"自由发挥"，输出了带 Markdown 表格的长文，直接把下游解析炸了。更麻烦的是，整个流程是一次性脚本，没有断点、没有重试、没有中间状态落盘，我只能翻日志手工补数据，花了几个小时。

那之后我意识到一个问题：Agent 类的应用，不能只有"调用模型"这一步，它需要一个流程骨架。但这个骨架又不能太重型，否则开发效率会严重下降。

1.2 重型编排平台与轻量 Workflow 的边界

市面上的工作流引擎大致分两类。一类是集成式平台，自带可视化画布、调度器、监控面板、分布式执行，功能确实全，但对于一个小团队、几个微服务、若干模型调用来说，引入它意味着额外维护一套基础设施，还要学一套自己的 DSL。另一类是库级别的编排工具，比如各种 Python 的 workflow 库，它们提供节点、管线、状态存储，相对灵活，但很多设计面向的是数据处理管线和批任务，对 Agent 场景里的"模型调用"、"工具调用"、"人工确认"这些动作支持得不够自然。

我想要的轻量 Workflow 长这样：

• 核心只包含节点、边、运行时上下文，不引入分布式概念；
• 节点的执行逻辑就是普通函数，模型调用也只是函数之一；
• 单机可跑通，但状态结构要清晰，方便以后迁移到队列或远程执行；
• 对 Agent 的"不可预测输出"有专门的校验和恢复机制，而不是假设它一定会成功。

这样一个东西对已有系统侵入性很小，团队里任何会写函数的人都能上手。

1.3 "轻量"不等于"简陋"

很多人一听轻量，就以为是把 if-else 循环包一层。如果只追求代码少，那直接用 Python 脚本更省事。轻量的核心价值在于三件事：可观测、可恢复、可组合。

• 可观测：每个节点的开始时间、结束时间、输入输出的摘要、失败原因，至少要有日志或回调钩子能看到；
• 可恢复：某个节点失败后，可以单独重跑这个节点，而不用重启整个流程；
• 可组合：节点可以复用，流程之间可以互相调用，新流程可以拼积木一样搭出来。

要做到这三点，其实不需要多少代码。真正难的是把"状态管理"和"执行控制"分开，让流程定义成为一种可读的数据结构，而不是埋在一堆 if 里。

我后来总结了一个判断标准：如果新来一个同学，看一遍流程定义文件，能在一分钟内说出这个 Agent 从输入到输出经历了哪些步骤，那这个 Workflow 就是合格的。如果做不到，说明抽象还不够清晰。

2. 最小可用的核心抽象：节点、边、状态与上下文

2.1 像电路图一样定义流程

轻量 Workflow 设计最核心的抽象只有三个：节点、边、上下文。节点是执行单元，边是执行顺序，上下文则是在整个流程中传递数据的"背包"。

拿一个典型的 RAG Agent 流程来举例：

• 意图识别节点
• 检索节点（调用工具或搜索接口）
• 重排节点
• 生成节点
• 校验节点

在代码层面，我不建议一开始就引入复杂的分支结构。更实用的方法是让每个节点自己决定"我是否要执行"以及"我的输出将影响哪条边"。这本质上是把决策权下放给 Agent，但执行框架保持确定性。

节点接口可以设计成：

class Node: id: str def run(self, ctx: Context) -> NodeResult: ...

每个节点从上下文读取所需数据，执行自己的逻辑，然后把结果写回上下文。边不承载数据，只决定流转路径。这样最大的好处是，任意两个节点之间解耦，单个节点可以独立测试。

2.2 上下文作为唯一数据源

上下文的实现方式直接影响整个系统的复杂度。我见过有人用全局单例存上下文，这在小项目里很省事，但一旦流程并发或嵌套，就等着被 bug 淹没吧。更稳妥的做法是把上下文设计成一个普通的字典结构，在流程实例化时创建，随流程生命周期销毁。

上下文里应该放什么？我的原则是：只放跨节点需要共享的数据，以及便于追踪的元数据。某个节点内部临时算出来的中间结果，如果不被下游使用，就不要塞进上下文。否则上下文会变成一个垃圾场，谁知道哪个节点依赖了哪个字段，流程维护就变成了考古。

状态与上下文的关系也值得说清楚。状态是流程维度的概念，表示当前进度；上下文是数据维度的概念，表示当前数据。我用一个简单约定：

• 流程状态：未开始、运行中、成功、失败、已暂停
• 节点状态：未执行、执行中、成功、失败、被跳过、被重试

运行时只维护这两类状态，不关心节点内部逻辑。

2.3 线图、环图和 DAG 的选择

轻量 Workflow 最常见的执行结构是 DAG（有向无环图），因为好理解、好分析、好并发。但 Agent 场景往往有限循环的需求，比如"校验不通过就重新生成，最多重试三次"。我建议把循环当作节点内部逻辑来处理，而不是画成图上的环。

举个例子，生成节点的内部逻辑可以是：

1. 调用模型生成内容
2. 校验结构
3. 若不通过且重试次数未达上限，回到步骤 1

这样从 Workflow 的视角看，生成节点就是一个独立的、可能内部自循环的节点，整个图依然是无环的。这个设计带来的直接好处是：可视化容易、断点恢复容易、并发控制简单。如果你真的需要跨节点的复杂循环，那说明流程还没被拆解干净。

我整理了一张对比表，方便大家选型：

多数情况下，混合使用最舒服：图保持 DAG，局部不确定性放到节点内部。

3. 让 Agent 遵循流程：工具协议、结构化中间产物与确定性边界

3.1 别让 Agent 直接控制流程分支

刚开始做 Agent 工作流时，我总是试图让模型输出一个"决策结果"来决定走哪条边。比如让模型从"检索""计算""回复"里选一个。这种做法在小规模演示里没问题，但真实场景里模型的输出会被用户 prompt 干扰、会被上下文污染。一旦模型选错了边，后面每一步都在错误前提下运行，而且很难定位。

后来我改了思路：流程的控制权永远在代码手上，Agent 只负责制造价值，不负责导航。比如流程需要判断"用户的问题是否需要搜索"，我不会让模型来选，而是先用一个简单的规则判断是否含某些信号词，或者交给一个 NLU 节点做分类，但分类结果最多决定"是否跳过检索节点"，而不是跳到任意节点。

这里的关键词是确定性边界：

• 流程结构确定：有哪些节点、默认顺序如何、跳过条件是什么；
• 节点内部不确定：模型的输出可能有多种形式，但必须经过解析、校验、归一化后再写回上下文；
• 异常出口确定：任何节点失败都有固定的失败传播路径，不会因为某个模型输出怪异就卡死。

3.2 用工具协议统一模型与 Workflow 的接口

Agent 最擅长的"决策"其实是决定"要调用哪个工具、传什么参数"。这恰恰是可以通过 Tool Protocol 来标准化的。一个工具的输入输出都定义成严格的 JSON Schema，模型的调用动作也遵循同一个协议。这样 Workflow 在调用模型时，可以让模型输出的工具调用指令直接映射到节点内部逻辑。

举个例子，一个任务拆分节点内部：

tools = [ { "name": "search_knowledge_base", "description": "在知识库中搜索与问题相关的内容", "parameters": { "type": "object", "properties": { "query": {"type": "string"} } } }, { "name": "call_calculator", "description": "执行数学计算", "parameters": { "type": "object", "properties": { "expression": {"type": "string"} } } } ]

模型输出一个 JSON 数组，Workflow 解析后查找对应工具并执行。这个执行动作可以放在一个节点内，也可以由节点内部代码发起。关键在于：工具注册、执行、结果回填这套逻辑应该是统一的，而不是每个节点各写各的。

我个人建议在轻量 Workflow 中把工具调用当作"一等公民"，给每个工具加上超时控制、错误码、重试策略，因为大模型的工具调用一不小心就会让节点挂掉。

3.3 中间产物必须结构化

有一类 bug 特别难排查：模型输出的内容本身是正确的自然语言，但下游节点把它当作了某个字段。比如提取关键词节点输出了"关键词：AI、Workflow"，下游节点却把整串字符串当作"AI、Workflow"塞进了标签列表。

为了减少这类问题，我要求所有模型产出在节点内部必须经过解析和校验。轻量 Workflow 应该内置一个 ResultValidator 接口，比如要求输出必须是 JSON、必须包含指定字段、字段类型必须正确。校验不通过时，可以在节点内部做一个轻量的格式修复，或者触发重试。

这里举一个变通案例：摘要节点要求输出纯文本，但模型偶尔会输出带 Markdown 标题的文本。我在摘要节点内部增加了一个规范化函数，把 Markdown 的#去掉、把冗余换行压缩。这样一个简单的防御措施，让下游的数据质量稳定了很多。

4. 在轻量工作流中管理失败：重试、回退与人工介入

4.1 失败不可怕，可怕的是没有兜底

Agent 工作流最大的不确定来源有三个：模型输出的不可控、远程工具调用的超时与异常、数据质量引起的下游异常。三者里任何一件发生了，流程都可能中断。轻量 Workflow 虽然没有重量级平台的弹性，但至少要具备三个兜底能力：

1. 节点级重试：对瞬时错误（超时、限流）自动重试；
2. 回退分支：对业务失败（校验不通过、数据缺失）执行预先定义的替代逻辑；
3. 人工介入：当自动恢复无法解决时，将流程挂起并通知人类处理。

这三个能力实现起来都不难，难点在于"什么算重试、什么算回退、什么算人工"的判定标准要写在节点定义里，而且要有迹可循。

我的经验是，不要让 Workflow 引擎去猜，最好把重试策略显式注释在节点上。比如：

@node( retry=RetryPolicy(max_attempts=3, backoff=2.0), fallback="rule_based_extraction", require_confirm=False ) def llm_extraction(ctx): ...

这样看代码时，一眼就能知道这个节点有什么防护措施。

4.2 有限重试与指数退避

大模型服务经常有偶发超时或返回 429 限流，直接放弃太可惜，无限重试又会拖垮系统。我常用的策略是：最多重试 3 次，每次间隔按 1 秒、2 秒、4 秒递增，超出后触发回退。

指数退避的关键是记录重试次数，这可以放在节点结果里：

class NodeResult: node_id: str status: str attempt: int output_meta: dict

有了 attempt 字段，日志里就能清楚看到某次失败发生在第几次重试。与此同时，重试时要不要重新调用模型？这取决于节点是否幂等。如果模型调用不幂等，重试前最好把上一次的输入快照也存到上下文里，防止数据在第一次执行时被污染。

4.3 人工确认环节要用事件驱动

有些 Agent 流程天然需要人类介入，比如"生成推广文案后，必须审核通过才能发送"。在轻量设计中，我不会在 Python 进程里等待人工输入，而是把节点状态置为"等待确认"，然后发出一个事件，退出执行循环。

等到外部系统回调确认后，再恢复上下文并继续执行。这听起来像状态机，实际上就是一个带事件回调的普通节点。实现上可以用一个简单的 Callback 存储：

ctx.pause_until(node_id, callback_token)

这种设计特别适合接即时通讯工具或管理后台的审批按钮。跑批任务时，人工环节常常是流程里最不可控的部分，把它事件化之后，整个流程在"挂起-恢复"上反而更稳定。

4.4 失败传播要清晰

我曾经踩过一个坑：某个节点失败后，我把它标记为"可选节点"，但下游节点根本不知道上游失败了，拿着空字段继续跑，最后造成了一系列连锁反应。后来我在引擎里加了一条规则：

• 如果一个节点失败，它的所有下游节点默认不会执行，除非在定义时显式标注can_run_on_failed=True。

这一条规则虽然简单，但立刻阻止了大多数连锁异常。失败的传播路径也变清晰了：要么提前终止，要么走回退分支，绝不允许带着半截数据继续乱跑。

5. 从零开始的内置引擎：约 150 行的 Node Runner 参考实现

5.1 核心数据模型

聊了这么多理念，最终还是要落地。下面这套实现是我在两次重写后保留下来的最短版本，不含任何外部依赖，可以直接塞进项目里用。先看数据模型：

from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional @dataclass class Context: data: Dict[str, Any] = field(default_factory=dict) attempt_map: Dict[str, int] = field(default_factory=dict) status: str = "ready" flow_id: Optional[str] = None @dataclass class NodeResult: node_id: str status: str output: Any = None meta: Dict[str, Any] = field(default_factory=dict)

Context 承担两个职责：节点间的数据传递和节点状态记忆。我没把它和运行时执行器耦合，这样单测时可以直接构造 Context 来测试某个节点。

5.2 节点的定义方式

我用了 Python 装饰器来注册节点，这样可以避免继承类带来的模板代码。每个节点函数接收 Context，返回 NodeResult：

def node(name: str, retry: int = 1, fallback: Optional[Callable] = None): def deco(fn): fn.node_name = name fn.retry = retry fn.fallback = fallback return fn return deco @node("parse_input", retry=1) def parse_input(ctx: Context) -> NodeResult: raw = ctx.data.get("raw", "") return NodeResult(node_id="parse_input", status="success", output={"text": raw.strip()})

装饰器的好处是，任何已有函数都可以变成工作流节点，不需要为框架改变业务代码风格。团队里写业务的人也不需要学习新的类体系，只需要记住ctx.data的约定就行。

5.3 执行器的核心循环

执行器是我整个轻量 Workflow 中最核心的部分，大概只有几十行。关键逻辑是：按顺序执行节点列表，检查节点结果，失败时按策略重试或回退，更新上下文，最后返回整体状态。

class Workflow: def __init__(self, name: str): self.name = name self.nodes = [] def add(self, fn): self.nodes.append(fn) return self def run(self, initial_data: Dict[str, Any]) -> Context: ctx = Context(data=initial_data, flow_id=uuid4().hex) for fn in self.nodes: name = fn.node_name attempt = 0 last_result = None while attempt < fn.retry: attempt += 1 ctx.attempt_map[name] = attempt try: last_result = fn(ctx) if last_result.status == "success": ctx.data[name] = last_result.output break except Exception as e: logger.warning("node %s failed: %s", name, e) last_result = NodeResult(node_id=name, status="error", meta={"error": str(e)}) if fn.fallback is not None: fb_result = fn.fallback(ctx) ctx.data[name] = fb_result.output last_result = fb_result break if attempt >= fn.retry: ctx.status = f"failed:{name}" return ctx if all(n.status == "success" for n in self.nodes): # 简化：这里可进一步记录 pass ctx.status = "success" return ctx

实际的执行逻辑里，我增加了条件跳过节点的支持：

@dataclass class FlowSpec: node_fn: Callable skip_if: Optional[Callable] = None def should_run(self, ctx: Context) -> bool: return self.skip_if is None or not self.skip_if(ctx)

这一步让流程具备了"Agent 决策节点"的能力：比如某个节点输出一个布尔值，下一个节点可以根据这个值决定是否跳过。

5.4 内置的日志与埋点钩子

轻量不代表黑盒，我在执行器里加了两个钩子：

• before_node(node_fn, ctx)：节点执行前调用，记录节点名、上下文数据的 key 列表；
• after_node(node_fn, ctx, result)：节点执行后调用，记录耗时、成功与否、输出的摘要。

有了这两个钩子，我可以在不嵌套业务逻辑的情况下实现日志、监控、甚至 LLM 调用计费。

class LoggingHook: def before_node(self, fn, ctx): logger.info("start %s, keys=%s", fn.node_name, list(ctx.data.keys())) def after_node(self, fn, ctx, result): logger.info("finish %s, status=%s, output_summary=%s", fn.node_name, result.status, str(result.output)[:120])

把钩子挂到 Workflow 实例上：

wf = Workflow("content_pipeline") wf.hooks.append(LoggingHook())

到这一步，这套轻量引擎已经能支撑不少 Agent 场景了。后面如果不满足于单机执行，可以把add的节点替换成远程函数或队列任务，核心抽象不用变。

6. 真实业务场景中的取舍与复盘：有哪些设计我后来又推翻了

6.1 刚开始做"万能可视化"，后来全删了

第一次设计时，我特别想在 Workflow 上套一个可视化拖拽画布，觉得这样才够"产品级"。后来发现，团队里真正频繁修改流程的是后端工程师，他们最顺手的是直接看 YAML 或 Python 定义。一旦引入可视化，就得维护节点类型注册表、画布组件、连线合法性校验，复杂度翻倍，而实际收益很小。

现在我的建议是：前 80% 需求用代码或配置文件定义流程就够了，可视化留到流程需要给非技术角色展示或审批时再考虑。

6.2 YAML 配置 vs 代码定义：哪种更合适

一开始我们倾向把流程定义写到 YAML 里，希望业务人员也能改流程。现实是：业务人员看不懂 YAML 的节点依赖和回退策略，工程师又觉得 YAML 缺乏类型检查和调试断点，两头不讨好。

我现在的折中方案是：

• 简单线性流程用 YAML 定义，方便查看和审计；
• 涉及分支、循环、人工审批的复杂流程用 Python 代码定义，因为复杂逻辑根本不适合声明式表达；
• 两者都输出同一个 FlowSpec 对象，运行时无差别。

flow: name: simple_summary nodes: - parse_input - llm_summary - validate_output

如果某个节点失败需要复杂的回退逻辑，我会把它重构成节点内部的代码，而不是在 YAML 里表达"如果 A 失败且 B 可用则走 C"这种规则。声明式配置最大的陷阱就是试图表达它不擅长的复杂控制逻辑。

6.3 上下文里放什么，决定了维护体验

有一段时间，为了让节点函数好写，我把所有历史输出都留在 ctx.data 里。这样下游节点确实能取到任何数据，但代价是：调试时要翻一大堆冗余字段，节点之间的隐式依赖倍增。后来我引入了"上下文契约"：

• 每个节点声明自己需要读哪些 key、写哪些 key；
• 执行器在执行前校验所需 key 是否存在；
• 上下文只保留被后续节点需要的 key，执行完成后可清理。

这有点像给函数的参数和返回值加上类型声明。虽然增加了一点模板代码，但对团队协作帮助巨大，新同学再也不会因为 ctx.data 里的魔法字段而找不着北。

6.4 关于可测试性：这是轻量 Workflow 最赚的一笔投资

我把节点函数设计成纯函数风格后，单元测试写起来特别爽。因为每个节点只依赖传入的 Context，我可以直接在测试里构造一个最小上下文，然后断言输出。

def test_parse_input(): ctx = Context(data={"raw": "hello world"}) result = parse_input(ctx) assert result.output["text"] == "hello world"

复杂流程的测试也简单：按依赖顺序跑一遍节点，最后断言 ctx.status 和关键节点输出。这种可测试性在大型平台型工作流引擎里反而不容易获得，因为它们把调度和业务逻辑绑得太深。

6.5 被模型调用"骗"过之后，我加的最后一层防御

最后一个让我印象深刻的教训是：模型产出即使通过校验，也未必是业务上需要的。比如一个情感分析节点输出了"负面"和置信度 0.98，但它读的文本已经被上游节点错误截断，分析结果再准也没用。

所以我给所有关键模型节点加了一个 input_fingerprint（输入指纹）的概念：把节点读到的上游关键字段的哈希值记录到 NodeResult 里。一旦发现某个节点的输出异常，可以顺着指纹追溯到上游哪个字段被破坏。这个机制加上后，定位问题的时间从小时级降到了分钟级。

轻量 Workflow 从来不是一次性把系统做重，而是持续做减法后再做加固。现在回看，我用这套思路维护的项目，代码量、调试时间、上线信心都明显优于之前用临时脚本拼 Agent 的时期。如果你也在组织自己的 Agent 流程，不妨先从一个带着重试机制的线性节点开始，跑顺了再加分支、加回退、加人工审批，最后需要的往往只比这多一点点，而不是一张庞大的编排网。