多Agent串行流水线：从任务拆解到断点续跑的工程实践

先说一个很多人都会踩的坑你以为把任务丢给一个大模型加上上下文提示词它就能帮你把一个十步的活儿一口气干完。实际上大模型单次输出经常在中间某个环节悄悄跑偏或者直接因为超过了输出上限给你半截结果。一旦出问题就得从头再来前面的所有工作全部作废。这也是我后来决定重做一套“多 Agent 串行流水线”的原因——把任务拆成一串可重试、可续跑的 Pipeline 节点链每个节点只负责一个被明确定义的子任务Agent 在其中扮演该环节的执行者整条链路支持失败重试和断点续跑。这套方案的适用面很广。无论是让大模型批量处理几十份文档还是把一个“翻译 → 润色 → 校对”的流程稳定跑完都能用得上。对个人开发者和后端工程来说它不需要重型调度系统一个 SQLite 文件外加一个循环就能实现很可靠的恢复机制。我在自己的某个内部项目里用这套骨架跑了上百条多步骤任务最大的感受并不是“花了多少钱”而是“再也不用半夜爬起来盯着日志看它有没有从头再来一遍”。这篇文章就把这套架构的拆解思路、重试机制、断点续跑实现和实战问题完整写出来。1. 为什么要搞一套可重试、可续跑的多 Agent 流水线1.1 单次大模型调用解决复杂任务时的尴尬如果只是“请从这段文章里提取三个关键词”这种一步任务直接调用大模型当然没有任何问题。麻烦的是那种带有明显先后顺序的复合任务比如“读取原始报告 → 提取关键指标 → 生成摘要 → 整理成表格 → 写一段结论”。把整个流程塞进一次调用不是不能跑而是风险极高。模型在长上下文里容易“选择性失忆”前两步还在遵守格式要求第三步就开始自由发挥输出一个完全无法解析的结构。上下文一旦逼近窗口上限调用可能直接失败而且这个失败往往发生在最后阶段前面花费的 token 全部浪费。我做过一次很粗糙的实验让同一个模型完成一个五步骤的文本处理任务单次调用成功率勉强超过三成。失败的原因五花八门有的是 JSON 解析失败有的是输出被截断有的是中途开始胡编内容。更麻烦的是当你只拿到一个失败堆栈时几乎不可能知道问题到底出在“提示词说不清楚”“模型能力不够”还是“输入数据格式有问题”。这种黑盒式调试非常痛苦逼着我换一条路。1.2 串行流水线在 AI 任务里的价值串行流水线的思路其实是从 CI/CD 那里抄来的。软件发布时编译、测试、打包、部署是分阶段的每个阶段有独立的日志、重试策略和产物。AI 任务完全可以照搬把一个大任务拆成若干节点按顺序执行前一个节点的输出成为后一个节点的输入。每个节点可以是一个大模型调用也可以是一个普通函数、一次外部 API 请求。这样天然获得三个好处隔离性——某一步崩溃不会把已经算好的结果拖下水可观测性——任何一步的输入输出都能被审计可恢复性——失败后不必从头开始。我常用一个工厂流水线来类比。零件在传送带上经过 A、B、C 三个工位C 工位发现零件有问题只需要把零件退回 A 工位返工B、C 已完成的步骤根本不需要重做。AI Pipeline 也一样如果第二步的模型超时第三步已经生成的成果要继续保留等第二步补跑成功后再接着往下走。长耗时、高成本的任务尤其受益毕竟每次全量重跑都意味着真金白银的 token 消耗。1.3 这套方案适合谁、能解决什么问题如果你符合下面几种情况之一我很推荐尝试这套设计。第一种你经常用大模型处理多步骤文档任务比如“翻译 → 润色 → 风格统一”希望每一步失败时能单独重跑。第二种你的任务里有部分步骤依赖外部服务比如数据库查询、搜索 API、内部 HTTP 接口这些服务偶尔超时或限流需要重试。第三种你需要向别人说明“这个最终结果是怎么一步步得到的”这时每个节点独立的输入输出就是很好的审计记录。这套方案对个人项目和小团队很友好。它不需要引入分布式工作流引擎本地文件系统加少量代码就能实现断点续跑。等节点数量增加到几十甚至上百个再考虑把 checkpoint 迁移到 Redis 或数据库也不迟。核心能力“可重试、可续跑”与底层存储无关重点在于状态管理和节点契约的设计。2. 核心设计任务拆解与节点链模型2.1 任务拆解的原则粒度、边界、依赖关系任务拆解最常见的两个错误是拆得太粗和拆得太细。拆得太粗时一个节点里仍然塞了一大堆逻辑失败后重试的范围还是太大拆得太细时节点数量爆炸每个节点都要保存上下文的输入输出重复开销反而压过收益。我的判断标准是如果某个节点只依赖上一个节点的输出并且它的输出能被下一个节点直接使用那么粒度基本合适。如果节点内部需要维护额外状态比如多次累积计数、跨步骤的临时变量说明这里应该继续拆分或者重新设计。具体拆解时我会从最终目标反推先列出为了得到最终结果必须生成哪些中间物。举一个“生成产品周报”的例子第一步从数据库导出本周数据第二步把数据整理成指标表格第三步根据表格写总结第四步排版成 Markdown。为什么要单独拆“生成表格”和“写总结”因为总结环节可能要用更强的模型或者需要多次重试而表格生成属于机械转换失败概率小重试策略可以更激进。不同的节点各配各的参数这就是拆分带来的灵活性。节点之间只通过输入输出契约互动这个约束必须写进代码里。我在代码中禁止节点之间访问彼此的运行上下文所有数据都通过统一接口传递。这条规则能让流水线保持简单也保证了断点续跑时上下文重建完全可预测。2.2 节点与链路的抽象输入输出契约、Agent 封装为了统一管理我把每个节点定义成下面的数据结构from dataclasses import dataclass, field dataclass class PipelineNode: id: str agent_type: str # llm, function, api input_schema: dict output_schema: dict params: dict field(default_factorydict) max_retries: int 3这里的id是节点在流水线中的唯一标识agent_type决定执行方式。input_schema和output_schema用来描述这个节点期望收到什么、会输出什么。例如一个“生成摘要”节点输入可能是{instructions: str, text: str}输出可能是{summary: str, keywords: list}。params保存这个节点特有的参数比如模型名、温度、超时。max_retries控制重试次数。我还会把一条流水线定义成一个按顺序排列的PipelineNode列表。串行意味着节点的next_node_id隐含在列表顺序里不需要额外声明依赖关系。这样做的缺点是灵活性差但换来了极低的认知负担。在真实场景中绝大多数任务都是天然的线性流程先把它跑通比盲目追求 DAG 编排更重要。2.3 为什么选择串行而不是编排、并行有人会问市面上有那么多工作流编排工具支持并行分支、条件判断为什么偏要用这么原始的串行方式我的回答是对于“生成 → 校验 → 再生成”这类任务串行已经足够而且更不容易出错。翻译之后才润色润色之后才校对这是强顺序依赖。如果强行并行还要额外处理分支合并和结果一致性检查复杂度一下子拉高好几倍。但这不代表排斥并行。如果任务里有完全独立的子问题比如同时请求多个数据源我会先并行拉取再把所有采集结果合并成一个 JSON 灌回主链路。整个任务依然是一条主串行链并行只发生在“采集节点”内部。这样断点续跑时调度器只需要关注主链的节点状态不用处理复杂的并行分支恢复逻辑。3. 可重试机制的落地细节3.1 失败的分类与错误分级重试之前的头等大事是区分“值得重试的失败”和“不值得重试的失败”。如果不做区分重试机制反而会把问题放大比如网络超时重试两三次可能就成功了但输入数据为空这种确定性错误重试一万次也救不回来。我通常把错误分成三类基础设施类错误网络超时、服务限流、临时 5xx这类错误重试很有效。模型业务类错误JSON 解析失败、输出不符合 schema、内容质量不合格这类要具体分析有时重试能缓解有时要改提示词。确定性错误必填字段缺失、参数非法、数据不存在这类重试没有任何意义应该立刻终止。在代码里我会让所有 Agent 包装器把底层异常统一抛成自定义异常并带上retryable标志。执行器拿到异常后根据标志决定是否进入重试循环。这样分层之后重试策略看起来清晰很多。3.2 重试策略次数、退避、超时附参数示例我常用的重试策略是每个节点默认最多重试 3 次初始等待 1 秒之后按指数增长每次乘以 2同时加入 0 到 200 毫秒的随机抖动。超时根据节点耗时灵活设置简单 LLM 调用给 30 秒长文本分析给 120 秒外部 API 额外设置连接超时和读超时。这是我在 Python 里常用的装饰器写法import time import random from functools import wraps class RetryableError(Exception): pass class FatalError(Exception): pass def retry_policy(max_retries3, base_interval1.0, max_interval60.0, jitter0.2): def decorator(func): wraps(func) def wrapper(*args, **kwargs): attempt 0 while True: try: return func(*args, **kwargs) except RetryableError as e: attempt 1 if attempt max_retries: raise interval min(max_interval, base_interval * (2 ** attempt)) interval random.uniform(0, jitter) time.sleep(interval) return wrapper return decorator加抖动的目的很实际多个节点几乎同时失败时如果不加抖动下一轮重试还是会挤在同一个时间点再次触发服务端限流。指数退避的底数我选 2间隔增长不会太慢整个任务不会等太久。如果你对成本极度敏感可以把底数改为 3代价是用户等待时间变长。参数没有绝对标准建议先从小值开始观察重试成功率再调整。3.3 幂等性与状态处理让节点可以安全重跑重试机制有一个容易忽略的副作用节点重跑时可能重复执行有外部影响的操作。比如某个节点不仅调用大模型还会把结果写入数据库第一次重试前写入了一部分第二次重试又写一遍数据就重复了。为了解决这个问题我为每个节点执行单元引入了一个全局唯一的execution_id在节点开始前生成写入 checkpoint 状态然后才开始执行。具体流程是节点要执行任何外部写操作之前先检查该execution_id的状态。如果状态是 done说明之前已经成功过直接读取之前的输出返回如果状态是 running 但之前残留了部分写入则需要先清理再重新执行。大模型调用本身是相对纯净的“生成型”操作重复调用不会产生外部副作用但会多花 token因此我在策略上会把 LLM 节点的重试次数控制得很低避免无效烧钱。幂等性的核心不复杂但很容易被忽略。尤其是当你第一次写这套系统时会想当然地认为“失败就是干净地失败”现实往往是“失败发生在写入一半之后”。所以我会建议所有涉及外部写操作的节点都要把“写入前登记”和“写入后标记”当成强制要求。4. 断点续跑的 Pipeline 调度实现4.1 节点状态机的设计pending/running/done/failed/skipped断点续跑的前提是每个节点都有明确、可持久化的状态。状态机的设计尽量简单我只使用五种状态pending还没开始执行running正在执行中done已经成功完成输出已保存failed重试次数用尽最终失败skipped条件不满足或前置节点失败选择跳过状态迁移是严格单向的pending可以到running、failed或skippedrunning可以到done或failed。恢复时调度器只关心第一个不是done和skipped的节点。这里有一条铁律done状态必须和实际输出保持一致不允许出现状态是 done 但输出为空的情况。我在保存状态和输出时把它写在同一个事务里要么都成功要么都回滚。4.2 持久化 checkpoint什么信息必须记录checkpoint 我默认用 SQLite 存储因为它轻量、支持事务而且恢复时不需要全部加载到内存。表结构大致长这样CREATE TABLE pipeline_state ( pipeline_id TEXT NOT NULL, node_id TEXT NOT NULL, status TEXT NOT NULL, attempt_count INTEGER DEFAULT 0, input_ref TEXT, output_ref TEXT, error_message TEXT, updated_at TEXT, PRIMARY KEY (pipeline_id, node_id) );这里的input_ref和output_ref可以是指向具体内容的 ID也可以指向外部文件路径。我强烈建议不要直接把几万字的大模型输出全文塞进 checkpoint。之前我犯过这个错误一个二十节点的流水线跑完checkpoint 文件超过 200MB恢复一次要好几秒。后来改成输出超过 10KB 就存成外部文件checkpoint 里只记录路径恢复速度立刻正常。另一个经验是每个节点完成之后必须立刻把它的输出快照持久化后续节点读取输入时统一从快照加载而不是从内存变量拿数据。这个规则听起来像废话但在进程被杀、容器重启等场景下少了这一步就是断点续跑失效的根源。我现在所有节点读取输入都通过checkpoint.load_input(node_id)从不直接引用内存对象。4.3 调度器如何从断点恢复并跳过已完成节点调度器的核心逻辑比想象中简单。我把它写成一个循环遍历节点列表读取节点状态如果状态是done或skipped就跳过否则执行。真正复杂的是处理running状态如果上一次进程在执行中途被杀节点状态停留在running恢复时要先把它改回pending再交给重试机制处理。伪代码大致如下def run_pipeline(pipeline, checkpoint, resume_fromNone): start_index 0 if resume_from: start_index pipeline.index_of(resume_from) for node in pipeline.nodes[start_index:]: state checkpoint.get_node_state(node.id) if state and state.status in (done, skipped): continue if state and state.status running: checkpoint.mark_pending(node.id) input_data checkpoint.build_node_input(node.id, pipeline) try: result execute_node_with_retry(node, input_data, checkpoint) checkpoint.save_output(node.id, result) checkpoint.mark_done(node.id) except FatalError: checkpoint.mark_failed(node.id) raisebuild_node_input会从 checkpoint 中读取所有前置节点的输出然后按照当前节点的input_schema重组为字典。如果前置节点没有输出它直接抛错避免下游拿到残缺数据。整个恢复流程就是不断循环这个函数直到遇见一个failed节点或跑完全部节点。5. 实操一个可参考的多 Agent 串行 Pipeline 骨架5.1 整体架构与目录结构我实际使用的项目骨架长这样pipeline_runner/ ├── models.py # Pipeline、PipelineNode、ExecutionState 等数据结构 ├── checkpoint.py # SQLite 持久化 ├── retry.py # 重试策略 ├── executor.py # 节点执行与调度 ├── agent_wrappers.py # LLM、函数、API 的统一包装器 └── cli.py # 命令行入口这个结构不追求分布式只保证在单进程内实现断点续跑。checkpoint.py负责所有持久化executor.py负责调度循环agent_wrappers.py负责把不同类型的 Agent 封装成统一接口。如果以后需要多机并行只需要把checkpoint换成 Redis 或数据库然后让多个 worker 抢占pending节点即可核心状态机不用变。5.2 节点定义与 Agent 接入示例我写一个最简单的 LLM Agent 包装器核心思路就是“接收一个 dict返回一个 dict”。模型调用、超时、解析 JSON 都藏在包装器内部节点定义不关心这些细节。# agent_wrappers.py class LLMWrapper: def __init__(self, modeldefault-model, temperature0.2, timeout120): self.model model self.temperature temperature self.timeout timeout def run(self, input_data: dict) - dict: messages [ {role: system, content: input_data[instructions]}, {role: user, content: input_data[text]}, ] raw call_model( modelself.model, messagesmessages, temperatureself.temperature, timeoutself.timeout, ) return parse_json_response(raw)节点定义则专注于输入输出和参数node_summary PipelineNode( idsummary, agent_typellm, input_schema{instructions: str, text: str}, output_schema{summary: str, keywords: list}, params{model: fast-model, temperature: 0.3}, max_retries3, )这里最关键的是input_schema和output_schema。它们既是文档也是约束执行器在调用 Agent 前会校验传入字段是否齐全在 Agent 返回后校验字段类型是否正确。schema 一出问题马上就能定位到具体节点而不是含糊地说“模型不行”。我用 pydantic 做实际校验但简单的isinstance判断也够用。5.3 重试与断点续跑的完整流程执行器executor.py里的核心逻辑我已经在 4.3 展示过这里补上更完整的重试交互。每个节点执行前会先检查 checkpoint 里有没有残留的retry_wait状态如果上一次在等待重试时进程被杀恢复后会直接重新计算等待时间而不是再次执行模型调用。这个细节保证了重试语义不会因为中断而混乱。下面是我实际使用的execute_node_with_retry简化版def execute_node_with_retry(node, input_data, checkpoint): attempt 0 while True: try: agent create_agent(node.agent_type, node.params) result agent.run(input_data) checkpoint.record_attempt(node.id, attempt, statussuccess) return result except FatalError as e: checkpoint.record_attempt(node.id, attempt, statusfatal_error, errorstr(e)) raise except RetryableError as e: attempt 1 if attempt node.max_retries: checkpoint.record_attempt(node.id, attempt, statusfailed, errorstr(e)) raise wait_time compute_backoff(attempt, basenode.params.get(retry_base, 1)) checkpoint.record_attempt(node.id, attempt, statusretry_wait, errorstr(e)) time.sleep(wait_time)在每次time.sleep之前我都先把retry_wait状态记录下来。因为 sleep 期间进程可能被杀恢复时能明确知道当前处于“等待重试”阶段而不是“执行中”。我在实践中发现这种显式记录对排查问题非常有帮助。看 checkpoint 表格就能知道某节点是模型调用慢还是在等待退避还是真的执行失败了。5.4 极简命令行工具实现命令行入口让整个流水线变成一个可以独立运行的进程。我常用的命令是python cli.py run --pipeline configs/report_pipeline.json --checkpoint ./checks.db python cli.py resume --pipeline configs/report_pipeline.json --checkpoint ./checks.dbrun从头开始执行resume加载已有 checkpoint从第一个非 done 状态开始。--pipeline参数指向一个 JSON 文件描述节点列表--checkpoint指向 SQLite 文件名。CLI 启动时先加载配置初始化 checkpoint再调用executor.run_pipeline。这套骨架看起来简单但其实已经能支撑真实任务。我不少项目就是靠它批量跑了几百份材料中途断网、API 限流、进程被杀只要补一条resume命令流水线还能从断点继续。6. 常见问题与排查技巧实录6.1 重试导致重复副作用、费用飙升怎么避免一位朋友用了我的思路后反馈他的第四步是发送通知邮件模型输出内容后调用邮件 API 失败重试时重复发送了两封邮件。原因就是把发送邮件当成了普通节点直接重试没有做幂等处理。我在 3.3 里提到过execution_id这里再重申一遍所有有外部副作用的节点执行前必须先登记执行成功后必须标记。重试时如果发现该execution_id已经成功直接返回成功结果不再执行副作用。更好的做法是把“生成内容”和“执行副作用”拆成两个节点。生成邮件内容属于纯模型操作失败可以随便重试发送邮件属于外部副作用一旦成功就永久记录。这样既可以利用重试修复内容质量问题又不会重复发邮件。费用方面如果某个节点反复因模型输出格式问题失败建议先停机检查提示词而不是一味加大重试次数。重试不是万灵药它只对瞬时故障有效。6.2 checkpoint 文件太大或太碎恢复慢怎么办checkpoint 变成性能瓶颈通常是存了太多大文本。我最早遇到的问题是每个节点的完整输出都存成一个文件节点多之后文件碎片化严重恢复时需要遍历大量文件。后来我做了两件事一是把输出按大小分流小对象直接存在 SQLite BLOB 字段大对象存到outputs/目录checkpoint 只记录引用二是定期合并 checkpoint比如用一次新快照替换历史状态。SQLite 还有一个额外好处支持按行读取不需要反序列化整个文件。如果节点数量达到几百直接在pipeline_state表上加索引查询速度依然不错。如果未来checkpoint 落到 Redis可以考虑用哈希表按节点存储恢复时只拉取第一个未完成节点附近的少量状态效率和 SQLite 类似。6.3 节点之间上下文太长导致成本膨胀的优化这是非常隐蔽的问题。每个节点都拿到上游节点的完整输出然后拼进提示词越到后面上下文越长token 费用自然飙升。比如第一步生成了一份两万字的文档第二步只是提取标题却把全文重新传给模型一次这就很浪费。更麻烦的是上下文太长还可能引发模型“走神”降低输出质量。我建议的优化是两个方向并行。第一精确裁剪利用input_schema定义下游真正需要的字段执行器在build_node_input时只提取这些字段而不是把整个输出对象塞过去。第二增加摘要节点如果下游确实需要全文概览先对长文本做分段摘要再把摘要作为下游输入。这两个方向配合起来通常能削减 60% 以上的上下文体积我实测五节点流水线的成本能下降将近一半。6.4 边界情况任务中途被用户终止、版本变更兼容运行流水线时用户可能会按 CtrlC或者部署新版本后想继续旧 checkpoint。如果不在代码层面对这种情况做处理恢复时很容易出现状态错乱。我的做法有两点一是 checkpoint 表里加一个schema_version字段每次升级代码时手动递增。恢复时发现版本不匹配跑一个迁移函数把旧的节点输出结构转成新版本兼容的结构。无法迁移时提示用户选择强制重跑受影响节点。二是在执行循环外层捕获KeyboardInterrupt先把当前节点标记为running表示执行到一半。下次 resume 时会把running改回pending再继续。这样虽然不能保证完全无差错但至少能把断点定位到具体节点不会让整个任务从零再来。另外要提醒的是如果你改了某个已done节点的提示词或者 Agent 参数旧的 checkpoint 里已经保存了它的输出下次 resume 时会直接跳过你的修改不会生效。这是“断点续跑”的一种副作用。我解决的办法是每次调整节点配置后把对应节点的状态重置为pending或者用一个force参数强制重跑指定节点。我在实际使用这套架构之后最明显的感觉是心不慌了。以前遇到长任务失败只能心疼地看着它从头再烧一遍 token现在无论哪个节点中断我都能准确说出失败原因然后一条resume命令让它接着断点走。设计 Pipeline 的过程也是在训练自己把复杂的 AI 任务拆解成更小、更可控的单位。如果你也被“一步到位”式的大模型调用坑过不妨用这套思路试一次。