面向Agent的轻量Workflow核心设计与工程实践

1. 项目概述与核心需求解析1.1 这个 Workflow 要给谁用、解决什么问题先说结论面向 Agent 的轻量 Workflow本质上是一套给大模型智能体准备的“最小执行轨道”。它不追求覆盖所有业务流程场景不做复杂的 BPMN 建模也不搞分布式调度那一套就是让 Agent 在完成多步骤任务时有一个清晰、可控、可回溯的执行路径。为什么需要这个东西我自己在实操里频繁撞到一个问题直接让 Agent 自由发挥完成复杂任务比如“整理资料并输出一份行业分析报告”它经常会做一半就迷失方向。要么步骤顺序混乱先写了结论再去收集数据要么某个环节失败后完全不知道该回退到哪一步最要命的是任务一旦拆分到五步以上Agent 的上下文就开始打架前一步的结果在后一步被重复加工或者干脆被遗忘。这不是模型能力不够是缺少一个结构化的执行约束。你需要一个轻量级的 Workflow 层做的事就三件第一明确任务步骤第二维护每一步的输入输出第三在步骤之间传递上下文。至于步骤内部怎么发挥Agent 爱怎么想怎么想那是它的事。这就是“面向 Agent”的核心区别——传统工作流把每一步的动作也固化了而我们这里只固化流程不固化动作动作由 Agent 动态决策。适合谁来参考如果你正在做 AI Agent 类应用无论是自动化脚本、智能客服、个人助理还是内容生产工具只要遇到“多步骤任务不可控”的问题这篇文章的思路可以直接帮你落地一套轻量执行框架。我自己是在一个内部工具项目中实践这套设计的当时目标很简单让 Agent 能按固定流程完成资料检索、数据抽取、报告生成三个环节并且每一步都能被人工干预和复审。1.2 轻量到底轻在哪里“轻量”这两个字我在项目里体会很深。市面上成熟的工作流引擎比如各类 BPM 中间件功能确实全流程图编辑器、定时触发、人工审批节点、子流程嵌套、复杂网关路由一堆能力。但问题是Agent 任务很多时候根本用不上这些反而是这些重能力引出了新麻烦。我举一个实际的反面案例。初期我尝试过用一套完整的工作流框架配置了一个只有六个节点的流程。结果发现光是学习它的节点类型和网关语义就花了两天写配置文件花了半天跑起来之后Agent 回调流程引擎的接口还经常因为鉴权配置不对而失败。那套系统的设计假设是“流程是稳定的、参与者是固定的”可 Agent 场景恰恰相反流程经常要调整比如临时加一步数据清洗或者删掉一个决策分支参与者是模型模型的行为又有不确定性。这一冲突导致我用传统工作流引擎的成本远超收益。所以后来我总结出轻量的三个标准启动轻不需要独立部署服务可以在你的应用进程内直接运行一个类、几个函数就能完成核心逻辑。改动轻调整流程结构时改代码配置即可不需要重新建模、审批、发布。心智轻整套概念模型不超过五个节点、边、上下文、执行器、钩子掌握这五个概念就能上手写流程。1.3 设计目标给 Agent 一条轨道而不是一堆枷锁这个项目在设计之初我给自己定的目标是轨道约束要足够强确保流程不跑偏但单点自由度要足够高确保 Agent 的推理能力不被扼杀。说白了就是在“完全自由”和“完全固化”之间找一个务实的平衡点。具体到设计目标我拆成了四条每条都对应实际踩过的坑确定性优先同一个流程定义无论跑多少次节点的执行顺序必须是确定的。分支判断可以依赖 Agent 的决策但决策之后走哪条路要明确。状态可观测任何时刻你都要能回答“当前执行到哪个节点了”“上下文里积累了哪些数据”。这一点在后面做错误排查时帮了大忙。失败可恢复单个节点失败不应该导致整个流程崩溃。好的设计是能让失败节点重试甚至跳过同时保留失败现场。运行时可干预流程执行过程中人可以暂停、调整参数、重新触发某个节点。这在 Agent 场景里尤其重要因为模型输出质量波动大人工复核节点必须能插进去。这套设计目标听起来不算宏大但每一条在实际编码中都对应着具体的实现取舍下面几节我会逐个展开。2. 整体设计与核心抽象2.1 节点即函数把 Workflow 的抽象层做薄我见过不少 Agent 框架把 Workflow 节点包装得特别复杂有输入 Schema、输出 Schema、重试策略配置、超时配置、回调钩子甚至还要配置依赖注入。结果就是写一个节点要写几十行样板代码。我的选择是节点就是一个异步函数其余一切皆可选。在 Python 里我的节点定义长这样async def extract_data(context: WorkflowContext) - WorkflowContext: # 从上下文中获取输入 raw_text context.get(raw_text) # 调用 LLM 抽取结构化信息 result await llm_extract(raw_text) # 把结果写回上下文 context.set(extracted_data, result) return context你可能注意到这个函数直接把整个 context 传进来又传出去。是的我没有设计独立的输入输出参数模型而是让节点直接读写全局上下文对象。这个选择有得有失失是节点之间的数据依赖关系不显式代码可读性会差一点得是写节点极其快临时加一个节点只需要写业务逻辑不用写数据适配代码。在实际项目中我还会给节点的函数对象附加两个可选元数据字段一个是node_name用于日志和可视化另一个是retry_count用于控制失败重试次数。其余如超时、并发控制都通过统一的执行器配置而不是分散在每个节点里。这么设计之后一个节点就从“一个配置实体”简化成了“一个函数加两个装饰器参数”。2.2 有向无环图还是线性链根据任务的真实复杂度来选轻量 Workflow 的拓扑结构我评估过两种主流方案线性链Chain和有向无环图DAG。各有各的适用场景但我的实践结论是大部分 Agent 任务用线性链就足够了DAG 只在少数并行场景才需要。先说线性链。它的好处是极其直观执行顺序就是定义顺序分支逻辑用手写 if 就能解决。比如流程是“先检索后分析”你写成两步链第五行代码做完。开发者读代码时脑子里能生成一条直线调试时从第一个节点追到最后一个节点没有任何分叉路径需要跟踪。但线性链扛不住两类场景一类是并行执行比如同时检索两个不同数据库结果汇总后再进入下一步另一类是条件分支比如根据输入类型走不同处理逻辑。这两类如果硬用线性链写代码会变成一团剪不断理还乱的 if-else。这时候 DAG 就来了。我实现的是一个极简 DAG只支持“多入边依赖”和“多出边分支”。节点在注册时需要声明依赖哪些上游节点执行器用拓扑排序决定执行顺序。核心数据结构是一个邻接表class Workflow: def __init__(self): self.nodes {} self.edges {} # node_name - list of downstream node names self.dependencies {} # node_name - list of upstream node names拓扑排序算法用的是经典的 Kahn 算法每次挑选入度为零的节点执行。def _topological_sort(self) - list: indegree {name: len(deps) for name, deps in self.dependencies.items()} queue [name for name, deg in indegree.items() if deg 0] result [] while queue: node queue.pop(0) result.append(node) for downstream in self.edges.get(node, []): indegree[downstream] - 1 if indegree[downstream] 0: queue.append(downstream) if len(result) ! len(self.nodes): raise ValueError(检测到循环依赖请在流程定义中移除环路) return result这里有个经验点即便你只需要线性链把拓扑结构设计成 DAG 也花不了多少额外成本但能换来未来的扩展空间。我常用的一句话是用 DAG 的底层承载线性链的简洁用不上时假装它是线性链就行。2.3 上下文对象Agent 的共享工作台上下文Context是这个设计里最核心的概念没有之一。它相当于 Agent 在执行过程中的“共享工作台”所有节点的输入都从这里取所有节点的产出都往这里写。我把它设计成一个简单的字典包装器支持 get、set、delete、merge 四个基础操作外加一个访问历史记录。为什么强调访问历史这是我在调试时踩了大坑后加的功能。Agent 节点的输入输出都是自然语言或结构化文本体积大且内容相似如果只记录最终上下文出了问题根本不知道是哪个节点写脏的。加了访问历史之后每个键的写入时间、写入节点、旧值、新值都有了记录排查问题效率翻倍。class WorkflowContext: def __init__(self): self._data {} self._history [] def set(self, key, value, node_nameunknown): old_value self._data.get(key) self._data[key] value self._history.append({ key: key, node: node_name, old: old_value, new: value, ts: time.time() }) def get(self, key, defaultNone): return self._data.get(key, default) def merge(self, data: dict, node_nameunknown): for k, v in data.items(): self.set(k, v, node_name)命名规范上我也定了一条节点写入的键名必须带节点名前缀比如extract_data.result。理由是避免两个节点写同一个键导致隐性覆盖。这个规范在五个节点以内看不出价值一旦节点超过八个就非常救命。2.4 执行引擎串行靠状态机并行靠队列执行器是整个 Workflow 的心脏。我的实现分两层底层是一个简单的状态机负责维护节点生命周期上层是一个调度器负责从拓扑排序结果中取出可执行节点分发执行。节点生命周期我定义了四个状态pending、running、success、failed。初始时所有节点都是 pending执行器循环扫描把“依赖全部成功”的 pending 节点取出放入 running节点执行完毕后根据返回结果置为 success 或 failed。全部节点 success 流程结束任一 failed 触发重试或终止。async def run(self, context): execution_order self._topological_sort() for node_name in execution_order: node self.nodes[node_name] context.set(_current_node, node_name, node_name) try: result await self._run_with_retry(node, context) if result is False: raise NodeExecutionError(f节点 {node_name} 返回失败) node.status success except Exception as e: if not self._handle_failure(node, context, e): raise return context并行调度的实现自然引入了多路复用。当 DAG 中存在多个入度为零的节点时我用asyncio.gather并发执行它们。这里有个非常关键的性能优化点并行节点之间的上下文写入冲突。我的解决办法很简单——每个并发节点写自己的键前缀最后汇总节点再 merge。这个模式避免了锁机制性能又高逻辑又清晰。3. 与 LLM 的集成让 Agent 动起来3.1 工具调用与函数注册机制Workflow 要驱动 Agent就得让 Agent 具备调用外部工具的能力。我的做法是建立一个工具注册表把函数、描述、参数 Schema 三者绑定然后在节点里给 LLM 提供这个注册表让它自主决策该调用哪个工具。class ToolRegistry: def __init__(self): self._tools {} def register(self, name, description, parameters_schema, func): self._tools[name] { description: description, parameters_schema: parameters_schema, func: func } def list_tools(self): return [{name: k, description: v[description], parameters_schema: v[parameters_schema]} for k, v in self._tools.items()]这里有一个我之前反复踩的坑给 LLM 的工具描述写得太随意。后来我意识到大部分工具调用失败都是因为参数不匹配比如 LLM 传了一个不存在的字段名或者把字符串传给了整数参数。修复办法是在register里强制校验函数签名调用前用 Schema 校验并尝试做一次类型转换。async def call(self, name, **kwargs): tool self._tools.get(name) if not tool: raise ToolNotFoundError(f工具 {name} 不存在) # Schema 校验与类型修正 validated self._validate_and_coerce( tool[parameters_schema], kwargs ) return await tool[func](**validated)底层原理上LLM 工具调用就是把函数描述转成 JSON Schema让模型“假装”理解再通过生成结构化参数来触发真实函数。这个机制本身并不复杂但工程上坑很多。比如模型返回的 JSON 里可能夹带 Markdown 代码块标记或者同时选择了多个工具。我的处理是写一个健壮的解析函数先剥离代码块标记再尝试直接解析最后用正则提取 JSON 片段按优先级一次一次尝试。3.2 决策节点设计让模型在流程岔路口做选择流程中的一个重要节点类型是决策节点。它不执行具体业务动作而是根据上下文内容判断下一步该走哪个分支。实现思路是提前定义好分支名称和每个分支的选项说明让 LLM 从候选里选择一个再把它映射到具体的下游节点。async def make_decision(context, question, options): prompt f 根据当前上下文回答以下问题{question} 选项 {chr(10).join(f- {opt} for opt in options)} 只输出选项本身不要输出其他内容。 response await llm_complete(prompt, context.get(_current_window, )) cleaned response.strip().strip() for opt in options: if cleaned opt or cleaned.startswith(opt): return opt # 兜底匹配失败走第一个选项 logger.warning(f决策解析失败输出了: {cleaned}回退到首个选项) return options[0]这个兜底逻辑不是偷懒而是经过实践后的务实选择。LLM 在开放问题时返回格式错误的概率不低尤其当上下文很长、模型注意力分布不稳定的时候。与其让整个流程崩溃不如让它回退到默认分支至少保证流程可以继续跑同时把异常记录到日志里提醒人工审查。这里有一个值得强调的设计原则决策节点不要做太细粒度判断。比如让模型决定“温度参数取 0.1 还是 0.2”就是浪费适合交给模型判断的是“这个用户输入是咨询还是投诉”“这一步的结果是否满足质量标准”这类宏观方向性问题。细粒度的判断应该用规则大方向的判断才用模型。3.3 结果校验与自省回路Agent 流程最容易被忽视的环节就是输出质量校验。模型生成的文本大概率结构完整、言辞流畅但这不等于内容正确。为了建立基本的质量防线我实现了两种校验器一是规则校验器用来检查硬性条件比如字段是否缺失、长度是否超出限制、是否包含敏感词。这类校验机快准狠成本为零适合前置过滤。二是模型校验器用来检查软性条件比如“抽取的金额字段与原文是否一致”“总结是否覆盖了所有要点”。这类校验基于模型自省self-reflection用一次额外的 LLM 调用做二次确认。async def verify_by_llm(context, result, requirements): prompt f 请验证以下结果是否满足所有要求 要求{requirements} 结果{json.dumps(result, ensure_asciiFalse)} 请只回答 PASS 或 FAIL。 verdict await llm_complete(prompt) if PASS in verdict.upper(): return True, PASS return False, verdict校验失败后走自省回路把失败原因打包发给一个修复节点让 LLM 根据原因修正上一次的结果修正完再回到校验节点重新验证。这就是一个典型的“执行-校验-修复”循环也是 Agent 流程从“线性跑一遍”升级到“质量自适应”的关键一步。实际的循环上限我配置成三次超过三次直接置为失败防止模型陷入自说自话的死循环。4. 实操阶段一个可落地的 Agent 流水线4.1 场景设定从客户留言到结构化工单光讲抽象设计不够我拿一个真实改造过的场景做全程演示。背景是一个内部服务 Team每天收到大量非结构化的客户留言历史做法是人工阅读、分类、填工单系统。我们做的工作是把这段流程交给 Agent Workflow 自动跑人工只负责最终确认。流程共五步设计如下register_ticket接收原始留言文本生成工单号写入上下文。classify_intent让 LLM 判断留言类型咨询、故障、投诉、建议。extract_entities根据类型抽取关键实体。故障类抽设备型号和错误码投诉类抽涉及部门和问题描述。priority_assess根据抽取结果和紧急度关键词评估优先级 P0/P1/P2。human_confirm把整理好的结构化工单交给人工确认人工可一键通过或退回修改。这正是线性链的典型用法。每一步的依赖关系简单明确不需要 DAG 并行甚至不需要条件分支。如果按传统工作流引擎的建模思路这五步要做成五个服务、配五个接口、画一张流程图而在我这套轻量设计里就是五个函数加一个线性链注册。4.2 流程编排代码详解下面这段代码呈现了完整可运行的流程定义省去了真正的 LLM 调用实现用模拟函数替代便于你聚焦在编排逻辑上。import asyncio from typing import Optional class LLMClient: async def complete(self, prompt: str) - str: # 实际场景替换为具体模型调用注意设置超时与重试 await asyncio.sleep(0.5) return fault class WorkflowContext: def __init__(self): self._data {} self._history [] def set(self, key, value, node_nameunknown): old self._data.get(key) self._data[key] value self._history.append({key: key, node: node_name, old: old, new: value}) def get(self, key, defaultNone): return self._data.get(key, default) class WorkflowNode: def __init__(self, name, fn, retries2): self.name name self.fn fn self.retries retries self.status pending class Workflow: def __init__(self): self.nodes {} self.sequence [] def add_node(self, node): self.nodes[node.name] node self.sequence.append(node.name) async def run(self, context): for name in self.sequence: node self.nodes[name] context.set(_current_node, name, name) last_error None for attempt in range(node.retries 1): try: result await node.fn(context) if result is False: raise RuntimeError(node returned False) node.status success break except Exception as e: last_error e logger.warning(f{name} 第 {attempt1} 次失败{e}) if attempt node.retries: node.status failed raise context.set(f{name}.node_status, node.status, name) return context节点函数的基本样例如下。实际项目中每个节点内部都会调用 LLM 或外部工具但是节点的输入输出约定保持统一async def classify_intent(ctx): raw_text ctx.get(raw_text) prompt f将以下留言分类为咨询、故障、投诉、建议。只输出一个词。\\n留言{raw_text} intent await llm.complete(prompt) ctx.set(intent, intent.strip(), classify_intent) return True如果你把上面这段跑一遍再打印context._data会看到每一步的产出都牢牢地存在上下文对象里。这个就是我们唯一定义的数据协议上下文对象就是节点间通信的全部。4.3 人工确认节点与非阻塞干预第五步 human_confirm 比较特殊它不能自动通过必须等人工响应。在设计上我用了回调与轮询两种方式回调适合嵌在已有的 IM 通知体系里轮询适合作为独立命令行工具运行。我们在内部工具里做得更简单——这个节点会发送一条企微通知给值班人点“通过”就调用确认接口点“退回”就带上原因并触发一个重做节点。这个设计表面上只是加了一个节点实际上改变了整个流程的可靠性模型。因为有人工兜底前四步的模型错误不会直接变成线上的坏数据而是被人在最后一步截住。我强烈建议所有 Agent 流水线在早期阶段不要追求全自动而是在关键结果落地前保留一道人工确认。5. 常见问题与排查技巧实录5.1 上下文越走越大把模型“撑”坏了这是我遇到最多的性能问题。Agent 流程跑多轮之后上下文里堆积了大量中间结果尤其是一次性把原始文本、抽取结果、修复版本、校验历史全塞进去LLM 的输入窗口迟早爆掉。表现是突然有一轮响应变慢然后报请求过大错误。排查方法用WorkflowContext._history做一次统计找出哪个键占了最大空间。修复方案有两个一是节点写回时只保留必要字段二是设计一个“上下文压缩节点”定期把旧的中间结果做一次摘要替代原始内容继续往下传。我很多流程里都会加一个summarize_context节点跑在长文本处理步骤之前这样既保留关键信息又控制体积。5.2 LLM 输出格式五花八门解析稳定不了模型输出的格式波动是 Agent 工程里最搞心态的问题没有之一。同一个 Prompt 这次老老实实输出 “故障”下次可能以 “故障。原因如下...” 开头。我在决策节点和抽取节点上都栽过跟头。我的三层兜底策略按优先级排序字符串精确匹配尝试用完整句对齐候选选项。包含匹配检查候选选项是否出现在输出文本中。正则提取用re.search提取特定模式比如引号内的内容。如果三层都失败就记录原始输出到日志然后返回默认值。这套策略不完美但实测把决策成功率从 90% 提升到了 98.5%。剩下的 1.5%交给人工兜底。5.3 重试机制反而变成了故障放大器重试是双刃剑。早期我天真地给所有节点配置了 3 次重试结果一个模型调用超时每个节点都重试三次整个流程跑了四倍时间还把外部 API 的限流打满了。后来的修正方案是区分错误类型超时类错误自动重试业务逻辑错误不重试立刻终止流程。class TransientError(Exception): pass class PermanentError(Exception): pass async def _run_with_retry(self, node, context): for attempt in range(node.retries 1): try: return await node.fn(context) except TransientError: if attempt node.retries: await asyncio.sleep(2 ** attempt) # 指数退避 continue raise except PermanentError: raise5.4 状态不同步并行节点把数据写乱了并行分支的上下文写入冲突是最隐蔽的 bug。两个节点同时写同一个键后完成的那一方覆盖先完成的而你不知道哪次会后完成。前面我提过用键前缀隔离这里展开说一下。假设有retrieve_a和retrieve_b两个并行检索节点它们分别写retrieve_a.result和retrieve_b.result汇总节点读取这两个键做融合。这样即使两个节点执行顺序不稳定也不会互相污染数据。关键是汇总节点的实现要容错比如ctx.get(f{node}.result)时万一某个键不存在要有默认值。5.5 死循环自省回路失控自省修复环节最常见的问题是陷入死循环。模型每次 “修复” 后的结果跟上一版差不多但校验器总是不通过于是一次又一次地调用直到打爆 token 配额。现在我的修复循环做了三重保险最大重试次数3 次、时间戳死线从进入修复环开始算超过 60 秒退出、以及修复内容差异检测如果两次修复结果的相似度超过 90%直接判定修复无效。6. 再进一步扩展性、持久化与可观测性6.1 持久化与回放让 Agent 流程可复盘Agent 流程是状态型系统一旦开始执行它的中间状态就应该能落盘。我用了最朴素也最实用的方案每次节点执行完成后把整个上下文和节点状态序列化到硬盘或数据库里。def save_snapshot(context, snapshot_path): snapshot { data: context._data, history: context._history, timestamp: time.time() } with open(snapshot_path, w, encodingutf-8) as f: json.dump(snapshot, f, ensure_asciiFalse, indent2)回放功能就是读取这个快照把执行过程在日志里重新打印一遍。这个能力在复盘一次失败的 Agent 任务时特别有用你能清清楚楚看到模型在哪个节点做了错误判断上下文到底在哪一步被污染。没有这个能力排一次错平均要半小时以上有了它五到十分钟就能定位问题。6.2 日志与追踪不要小看一条结构化日志早期我的日志就是随手 print后来发现排查 Agent 问题完全不够用。我改进后的日志格式是 JSON 化的每条日志至少带上节点名、执行编号、耗时、状态、上下文快照的容量。这样无论是 grep 还是接入日志分析平台都能快速筛选出异常点。比如我常做的一个操作搜索status:failed然后按时间线看它前后的日志就能还原整个故障现场。6.3 配置化把流程定义从代码中剥离节点是函数流程定义是数据这两者分开之后就有了配置化的空间。我的流程定义存在 YAML 文件里代码负责把节点函数按名字注册到执行器。这样一来调整流程顺序、增减节点只需修改 YAML 后重启应用不需要改代码、走发布流程。对于 Agent 流程这种迭代频率极高的场景这个特性重要程度排在前面能省掉大量无意义的开发排期。workflow: name: ticket_pipeline nodes: - name: register_ticket module: nodes.ticket_ops function: register_ticket - name: classify_intent module: nodes.intent function: classify_intent - name: human_confirm module: nodes.human function: human_confirm retries: 06.4 不要在没想清楚时引入消息队列和外部存储这是最后想提醒的一点。很多开发者一听到 Workflow就联想到消息队列、分布式任务、独立存储觉得一套流程引擎天生就该这样。但面向 Agent 的轻量 Workflow最大的价值恰恰在于简单在于跑在应用进程内在于一个字典搞定上下文。当你的流程超过 20 个节点、需要跨服务调用、或者有高并发要求时再考虑迁移到消息队列也不迟到时候你的流程代码因为节点和上下文都做了合理抽象迁移成本也会低很多。过早引入分布式组件只会让项目的上手门槛和运维成本同时增加却解决不了 Agent 流程当前真正的问题——可控性。先把可控性做好后续再谈规模。