前阵子有人拿一段Agent代码给我看几十个条件判断层层嵌套穿插着三次模型调用和两次工具请求改一个分支要先把整条链路读三遍。这种代码在Demo阶段还能跑一旦开始加需求就会迅速变成“只有当事人能改”的祖传代码。这不是写代码的人能力不行而是缺少一层“流程控制”的抽象。说白了Agent应用跑起来以后最需要的不是更聪明的模型调用而是一套能把步骤、分支、状态串起来的工作流机制。我这里的做法是“面向 Agent 的轻量 Workflow 设计”不依赖重型工作流引擎用很小的代码量搞定节点编排、条件路由和上下文传递。适合正在做Agent原型、RAG流水线、工具链自动化的独立开发者也适合中小团队在正式框架落地前快速验证流程。1. 为什么“轻量”成了硬要求1.1 重型引擎并不适合所有场景一说Workflow大家第一时间想到的是各种带可视化界面的分布式调度平台。这类系统确实强大有持久化、有重试队列、有分布式执行器但代价也不小你得先部署一套服务再学它的概念模型还要设计任务如何序列化、如何恢复。对大部分Agent场景来说这是典型的杀鸡用牛刀。Agent场景有一个特殊性流程往往依赖模型输出才能决定下一步方向。模型返回什么JSON、置信度多少、工具调用是否成功这些在写流程前根本没法完全确定。这意味着工作流必须支持非常“动态”的路由而不是执行前就画死的DAG。传统工作流引擎擅长的是“一批任务按顺序跑完”Agent工作流擅长的是“跑完这一步根据结果临时决定怎么走”。两者的核心逻辑并不一样。1.2 轻量Workflow的设计边界我说的“轻量”有明确边界。第一不需要可视化编排界面用代码描述流程就够了。第二不需要分布式执行和持久化队列单进程内存调度就能覆盖绝大多数场景。第三不需要庞大的插件体系每个节点就是一个Python函数或类拿来即用。这不是功能缺失而是刻意取舍。轻量意味着把最核心的“节点上下文路由”做成一套极薄的抽象让开发者在写业务逻辑时不受框架约束。实测下来这套设计能覆盖这些场景LLM多步推理、工具调用组合、RAG检索后处理、客服工单自动分类、内容审核多级分流、定时报告生成。凡是流程节点在几个到十几个之间、需要按事件动态跳转的都属于它的舒适区。对比维度轻量方案重型引擎部署成本无随应用内嵌独立服务存储学习成本半天数天到数周动态路由天然支持需额外设计恢复与重试需自行实现内置适用阶段原型、中小流程大规模、高可靠2. 轻量Workflow的抽象与设计2.1 节点、上下文与路由——三个最小概念任何工作流系统不管多复杂最终都能抽象成三个词节点、上下文、路由。节点Node是一段最小可执行逻辑。它负责做一件事调一次LLM、查一次数据库、调用一个工具、生成一段回复。节点不关心前后是谁只接收上下文对象执行自己的逻辑然后把控制权交回给引擎。上下文Context是贯穿整个工作流的共享状态。所有节点从里面拿自己需要的参数再把结果放回去。设计上我用一个普通的字典对象承载为了调试方便加上了操作记录和时间线。这个设计几乎零成本但对全链路的数据流转至关重要。路由Route是决定流程下一步走向的机制。每个节点执行完毕后返回一个字符串代表下一节点名称如果不返回就顺着默认边继续走。这比传统的“前置条件后置节点”配置简单得多而且天然支持动态路由——模型说“意图是下单”节点就返回“下单处理节点”。from dataclasses import dataclass from typing import Any, Dict, Optional dataclass class WorkflowContext: data: Dict[str, Any] None timeline: list None def __post_init__(self): self.data self.data or {} self.timeline self.timeline or [] def set(self, key: str, value: Any): self.data[key] value self.timeline.append((len(self.timeline), key, str(value)[:80])) return self def get(self, key: str, defaultNone): return self.data.get(key, default) class BaseNode: name: str base def execute(self, ctx: WorkflowContext) - Optional[str]: raise NotImplementedError2.2 共享上下文为什么不用数据流管道设计工作流的时候很多人第一反应是定义每个节点的输入输出类型做成一条严格的数据流管道。这个思路本身没问题但在Agent场景里会非常累。模型输出的字段是不稳定的你不可能每次都定义好“返回值2”是字符串还是整数是必然存在还是可能缺省。共享上下文相当于一条传送带每个工位只从传送带上取自己需要的零件再把成品放回去而不是每个工位之间单独递箱子。好处是节点之间解耦增加一个节点不需要改其他节点的接口定义。代价是数据访问变得自由容易埋雷——一个节点不小心覆盖了另一个节点的字段排查起来会怀疑人生。我的改进做法是约定每个节点写入的数据都带上自己的节点名前缀。比如意图识别节点写入intent_router.resultFAQ查询节点写入faq_search.result。这避免了九成以上的字段冲突而且执行日志里一眼能看到是谁写了什么。2.3 命令式编排与声明式配置轻量方案不等于拒绝配置化。我见过不少团队希望用JSON描述流程这样产品和运营也能看懂整体走向。这个需求合理。所以我在实现里做了两层。核心引擎支持命令式构建也就是代码里add_node、add_route直接连同时附带一个从字典描述加载流程的工具函数。def build_flow(desc: dict): wf Workflow(namedesc.get(name, default)) node_defs desc[nodes] for nd in node_defs: node_cls NODE_REGISTRY.get(nd[type]) if not node_cls: raise ValueError(funknown node type: {nd[type]}) wf.add_node(node_cls(namend.get(name), **nd.get(params, {}))) wf.set_entry(desc[entry]) for edge in desc.get(edges, []): wf.add_route(edge[from], edge[to]) return wf这样的设计让流程定义很容易版本化管理也方便在不同环境之间共享。我个人的习惯是业务开发阶段用代码直接搭等流程稳定之后再花十分钟转成JSON描述作为系统配置固化下来。3. 从零实现一个能用的极简编排器3.1 环境准备与依赖整个实现只需要Python 3.9以上标准库不需要装任何第三方框架。如果为了好看用了dataclass这也是标准库自带。所以这套逻辑几乎可以复制到任何项目里不受技术栈绑架。依赖极少的背后是刻意压低抽象层。节点不继承重型基类不做依赖注入容器不搞插件发现机制。每个节点就是一个有execute方法的对象引擎只负责按路由调用。这一点非常重要因为Agent领域迭代速度太快今天用的模型调用方式下周可能就变了。如果工作流框架绑死了调用协议升级成本会落在每一个节点上。3.2 节点基类与上下文实现节点基类设计要非常克制。我只要求子类实现execute方法并约定返回值为下一个节点名或者None。为了支持更复杂的生命周期我加了一个on_enter钩子但默认实现为空方法不影响主体逻辑。class BaseNode: name: str base def __init__(self, name: Optional[str] None, **kwargs): if name: self.name name def on_enter(self, ctx: WorkflowContext): pass def execute(self, ctx: WorkflowContext) - Optional[str]: raise NotImplementedError上下文对象在前面已经给出这里补充一个关键设计每调用一次ctx.set()就会记录一条时间线条目。这个日志不是可有可无的点缀而是我排查Agent行为最重要的依据。模型输出不稳定的时候你必须知道到底是哪个节点写入了错误数据、流程在哪一步走偏的。没有这条时间线所有问题都只能靠打日志重跑复现效率极其低下。3.3 工作流引擎主体引擎核心是一个循环当前节点、步数计数、绑定路由。每步先执行节点拿到返回的next节点名找不到就查默认路由然后移动指针。为了防止流程死循环我加入了最大步数保护默认200步。class Workflow: def __init__(self, name: str default, max_steps: int 200): self.name name self.nodes {} self.routes {} self.entry: Optional[str] None self.max_steps max_steps def add_node(self, node: BaseNode, route_to: Optional[str] None): self.nodes[node.name] node if route_to: self.routes[node.name] route_to if self.entry is None: self.entry node.name return self def set_entry(self, name: str): self.entry name return self def add_route(self, from_node: str, to_node: str): self.routes[from_node] to_node return self def run(self, init_data: Optional[dict] None): if not self.entry: raise RuntimeError(workflow entry not set) ctx WorkflowContext() if init_data: for k, v in init_data.items(): ctx.set(k, v) current self.entry step 0 ctx.set(_flow, {current: current, history: []}) while current is not None: step 1 if step self.max_steps: raise RuntimeError(fworkflow exceeded max_steps{self.max_steps}) node self.nodes.get(current) if node is None: ctx.set(_error, funknown node: {current}) break node.on_enter(ctx) ctx.data[_flow][history].append(current) try: next_node node.execute(ctx) except Exception as exc: ctx.set(_error, f{current} execute failed: {exc}) ctx.set(_last_node, current) break if next_node is None: next_node self.routes.get(current) if next_node is not None and next_node not in self.nodes: ctx.set(_error, froute to unknown node: {next_node}) break current next_node ctx.data[_flow][current] current ctx.set(_status, success if _error not in ctx.data else failed) return ctx这段代码就是整套编排器的骨架不到六十行。单看内容可能觉得很简单但它的核心价值在于“约定”。所有Agent流程最终都收敛成这种形式不管节点内部是调模型、查向量库还是打外部接口对引擎来说只有一种交互方式——执行、返回、路由。3.4 重试与错误处理策略基础引擎没有内置重试因为重试策略高度依赖节点类型。LLM调用超时后的重试通常要配合指数退避数据库查询失败不能盲目重试工具调用则要看是否幂等。把重试逻辑内置在引擎里等于替所有节点做决定这个决定大概率是错的。我的做法是把重试做成一个装饰器按需挂在节点方法上。比如某节点需要最多重试三次、间隔递增直接在execute方法上标注。这样既保留了引擎的轻量又不牺牲可靠性。import time def retry(max_attempts: int 3, base_interval: float 0.5): def decorator(func): def wrapper(*args, **kwargs): for attempt in range(max_attempts): try: return func(*args, **kwargs) except Exception: if attempt max_attempts - 1: raise time.sleep(base_interval * (2 ** attempt)) return wrapper return decorator4. 实战演示客服工单Agent中的条件工作流4.1 场景与流程定义用一个真实高频的客服场景来演示。用户给客服机器人发了一条消息系统需要判断消息属于哪类意图再决定走哪条处理流程。步骤涉及意图识别、创建工单、查FAQ、转人工、生成回复。流程定义intake接收原始消息。intent_router调用LLM识别意图返回枚举值。create_ticket意图为下单/退款时执行写入工单。faq_search意图为咨询时执行检索FAQ库。human_handoff意图为投诉或意图置信度过低时转人工。final_reply汇总以上结果生成最终回复。这个流程包含了一个条件分支和两条主链路很适合验证工作流引擎的动态路由能力。4.2 节点实现与运行效果每个节点都实现成独立的类只依赖上下文。下面给出关键节点class IntentRouter(BaseNode): def __init__(self, name: str intent_router): super().__init__(name) def execute(self, ctx: WorkflowContext) - Optional[str]: raw_msg ctx.get(raw_message) # 此处替换为真实LLM调用模型返回JSON例如 {intent: faq} # demo中直接模拟一个分类结果 intent faq ctx.set(intent_router.result, intent) if intent human: return human_handoff if intent in (create_order, refund): return create_ticket return faq_search class FaqSearch(BaseNode): def execute(self, ctx: WorkflowContext) - Optional[str]: intent_result ctx.get(intent_router.result) ctx.set(faq_search.result, f根据意图[{intent_result}]检索到的FAQ内容) return final_reply class CreateTicket(BaseNode): def execute(self, ctx: WorkflowContext) - Optional[str]: msg ctx.get(raw_message) ticket_id TK- str(len(ctx.get(_flow.history, [])) 1000) ctx.set(create_ticket.ticket_id, ticket_id) ctx.set(create_ticket.detail, msg) return final_reply class HumanHandoff(BaseNode): def execute(self, ctx: WorkflowContext) - Optional[str]: ctx.set(human_handoff.channel, oncall_queue) return final_reply class FinalReply(BaseNode): def execute(self, ctx: WorkflowContext) - Optional[str]: parts [] if ctx.get(create_ticket.ticket_id): parts.append(工单已创建) if ctx.get(faq_search.result): parts.append(ctx.get(faq_search.result)) if ctx.get(human_handoff.channel): parts.append(已为你转接人工) ctx.set(final_reply.text, / .join(parts)) return None组装流程并运行wf Workflow(customer_service_agent) wf.add_node(IntakeNode()) wf.add_node(IntentRouter()) wf.add_node(CreateTicket(), route_tofinal_reply) wf.add_node(FaqSearch(), route_tofinal_reply) wf.add_node(HumanHandoff(), route_tofinal_reply) wf.add_node(FinalReply()) wf.set_entry(intake) ctx wf.run({raw_message: 我想查一下发货时间}) print(ctx.get(final_reply.text)) # 输出根据意图[faq]检索到的FAQ内容整个流程没有出现任何一处业务代码去判断“下一步该调谁”节点之间互相不认识。新增一个“转人工”的节点不影响现有节点删除一个分支也只需要改一行路由配置。这就是抽象带来的最直接收益。4.3 声明式版本把流程变成配置如果要把上面这套流程交给非开发人员评审可以用配置化形式呈现flow_desc { name: customer_service_agent, entry: intake, nodes: [ {name: intake, type: intake}, {name: intent_router, type: intent_router}, {name: create_ticket, type: create_ticket, route_to: final_reply}, {name: faq_search, type: faq_search, route_to: final_reply}, {name: human_handoff, type: human_handoff, route_to: final_reply}, {name: final_reply, type: final_reply}, ], } wf build_flow(flow_desc)配置化的价值在于流程的骨架一目了然改分支走向时不需要翻代码。团队里总有同事只关心“意图识别失败后走哪边”这种问题给他看JSON比给代码更直接。5. 实操中的坑与优化记录5.1 上下文覆盖隐蔽的数据污染轻量方案的上下文设计自由度很高这让编码很舒服但也容易出问题。我踩过最典型的坑是两个节点都把结果写在result字段里后面的节点读到的永远是最新那次写入的值而前面的结果被静默覆盖了。执行日志里能看到两次写入但排查时依然需要不少时间。后来我立了一条规矩所有节点写入前必须带节点名前缀。上面代码里的intent_router.result、faq_search.result就是这种风格。对已经稳定运行的系统还可以在保存时自动加命名空间不过约定优先代码层面不留隐患最重要。5.2 动态路由依赖模型输出低置信度问题用LLM做意图识别时模型给出的分类不一定可靠。如果路由逻辑完全依赖模型返回的文本一旦出现幻觉或未知类别流程直接走错分支。我的应对方法是在节点内部增加置信度校验低于阈值时返回“human_handoff”宁可转人工也不要瞎处理。具体做法是让LLM输出带分数的JSON。{intent: faq, confidence: 0.38}这种形式路由节点先判断分数再决定下一跳。这是Agent工作流区别于传统工作流的一个关键细节——模型输出质量必须纳入路由设计。5.3 死循环防护与幂等保护动态路由意味着流程执行路径不可能全部在编码阶段预测。如果两个节点互相引用对方当下一跳或者某个节点在异常时不断抛错就有可能形成死循环。引擎层的max_steps是第一道保险。但实际经验告诉我光有步数限制还不够需要在节点层面加入调用频率保护。我给工作流加了节点调用次数统计。每个节点如果连续执行超过指定次数引擎会直接终止并把流程标记为失败。这在实际运行中拦截过不止一次由模型反复输出同一结果导致的无限循环。# 在Workflow中添加节点调用统计 self.node_calls {} # run循环内部每次执行前 self.node_calls[current] self.node_calls.get(current, 0) 1 if self.node_calls[current] 10: ctx.set(_error, fnode {current} invoked too many times) break5.4 可观测性执行轨迹是排查利器轻量方案没有独立的可视化界面并不代表无法观测。相反上下文里的timeline和history字段足够回答大多数问题流程走到哪里、哪个节点写了什么值、为什么最终走了某条分支。我还会在每次run结束之后打印一段简化轨迹print( - .join(ctx.get(_flow.history))) # 输出示例intake - intent_router - faq_search - final_reply如果你在部署环境里跑把这条轨迹接入日志系统跟模型输入输出一起存下来几乎可以回放每一次Agent行为。这比早期堆printf调试高效太多。5.5 什么时候该换重型方案轻量方案不是万能的。当流程节点数量到了二三十个、需要跨进程恢复执行、要支持条件等待和定时触发、或者多个任务并行跑的时候我的建议是切换到更成熟的框架。判断标准不是“未来的需求可能会很大”而是当前真实的复杂度。为了应对可能存在的大规模提前引入重型框架只会让开发速度变慢不确定性更高。我个人实际使用中的体会是Agent应用迭代极快今天设计的流程下个月几乎必然大改。轻量Workflow带来的最大价值不是执行效率而是让流程结构以最低成本保持清晰。很多事情等真的复杂到需要重型引擎那天再迁移也不迟而且因为有清晰的节点抽象迁移本身并不会伤筋动骨。