1. 为什么要把一个任务拆成 Pipeline 节点链1.1 从“一把梭”到“分步走”的认知转变最早做自动化任务编排的时候我的做法很粗暴一个函数从头写到尾中间调模型、查数据、生成文件、发通知全塞在一起。跑通了确实爽但只要中间任何一步出问题整个任务就得从头再来。最要命的是有些步骤本身要花不少时间重跑一次成本极高而且重跑过程中还可能产生重复的副作用比如重复写入、重复发送。后来我开始接触多 Agent 协作的场景发现问题的本质不是“Agent 不够聪明”而是任务的组织方式太原始。一个复杂任务天然就应该被拆成若干个有明确边界的阶段每个阶段有自己的输入、输出、成功条件和失败处理策略。这就像工厂流水线你不会让一个工人从零件锻造一直干到整车出厂而是分成冲压、焊接、涂装、总装几个工位每个工位只关心自己的活出了问题也只影响那一个环节。Pipeline 节点链的核心思路就是把任务拆成串行的节点每个节点是一个独立的执行单元节点之间有明确的数据契约任何一个节点失败都可以单独重试整个链条可以从断点续跑而不需要从头再来。这个思路听起来简单但真正落地的时候有很多细节需要想清楚。比如节点之间怎么传递数据重试的时候怎么保证不重复执行已经成功的节点如果一个节点依赖前面多个节点的输出怎么办这些问题我在实际项目中都踩过坑下面会逐一展开。1.2 哪些场景适合用 Pipeline 节点链不是所有任务都值得拆成 Pipeline。如果一个任务总耗时不到几秒钟中间也没有外部依赖那直接写一个函数就完了拆成 Pipeline 反而是过度设计。但如果你遇到以下几种情况Pipeline 节点链的价值就会非常明显任务耗时长整个流程跑下来要几分钟甚至几十分钟中间任何一步失败都意味着大量时间浪费。步骤之间有明确的阶段划分比如“数据采集 → 数据清洗 → 分析计算 → 结果生成 → 通知分发”每个阶段做的事情性质不同。部分步骤有外部依赖比如调用外部接口、读写数据库、操作文件系统这些步骤失败概率较高需要单独重试。需要可观测性你想知道每个阶段花了多长时间、成功还是失败、输出是什么而不是只有一个“整体成功/失败”的结果。需要断点续跑任务跑到一半因为某种原因中断了你希望能从断点继续而不是从头再来。我做过的一个典型场景是从多个数据源采集原始素材经过清洗和结构化处理然后调用模型生成分析报告最后把报告分发到不同的下游系统。整个流程跑一次大概要十几分钟其中数据采集和模型调用是最容易出问题的环节。用了 Pipeline 节点链之后如果模型调用失败了我只需要重试那一个节点前面的数据采集和清洗结果都还在不用重新跑。1.3 串行流水线与并行编排的区别这里要特别说明一下我选择的是串行流水线而不是并行编排。并行编排适合那些互相独立的子任务比如同时调用三个不同的模型生成三份结果然后汇总。但很多实际任务是有严格先后依赖的后一步的输入依赖前一步的输出这种情况下串行是唯一合理的选择。串行流水线的一个关键优势是状态管理更简单。每个节点只需要关心自己的输入和输出不需要考虑并发冲突、资源竞争、状态同步这些问题。当然代价是总耗时是各节点耗时之和但对于大多数 IO 密集型的任务来说串行带来的额外等待时间远小于重试和调试成本的节省。还有一个我特别看重的点串行流水线天然适合可续跑。因为节点是按顺序执行的每个节点的输出都可以持久化续跑的时候只需要从第一个未完成的节点开始就行。并行编排要做到可续跑就复杂得多你需要追踪每个并行分支的状态还要处理分支之间的依赖关系。2. 节点链的核心设计契约、状态与重试2.1 节点之间的数据契约怎么定Pipeline 节点链最容易出问题的地方就是节点之间的数据传递。我见过很多实现方式有的用全局变量有的用共享内存有的用消息队列。每种方式都有各自的适用场景但在串行流水线里我认为最稳妥的方式是显式的数据契约。所谓显式数据契约就是每个节点明确声明自己的输入类型和输出类型节点之间通过一个统一的上下文对象来传递数据。这个上下文对象可以理解为一个“任务黑板”每个节点从上面读取自己需要的输入执行完后把自己的输出写上去。from dataclasses import dataclass, field from typing import Any, Optional dataclass class PipelineContext: task_id: str data: dict field(default_factorydict) node_results: dict field(default_factorydict) errors: dict field(default_factorydict) def set_output(self, node_name: str, output: Any): self.node_results[node_name] output self.data[node_name] output def get_input(self, node_name: str) - Optional[Any]: return self.node_results.get(node_name)这样做的好处是每个节点的输入输出都是可追溯的调试的时候你可以清楚地看到每个节点产出了什么。而且因为数据是显式传递的续跑的时候只需要把之前保存的node_results恢复出来就行。注意上下文对象里的数据要尽量保持可序列化。如果你往里面塞了数据库连接、文件句柄这类不可序列化的对象续跑的时候就会出问题。我的做法是节点内部自己管理这些资源上下文里只放纯数据。2.2 节点状态机每个节点到底有几种状态一个节点从开始执行到最终结束中间会经历多个状态。如果状态定义不清晰重试和续跑的逻辑就会很混乱。我一般把节点状态定义为以下几种状态含义是否可重试续跑时的处理PENDING尚未开始执行是从该节点开始执行RUNNING正在执行中视情况需要判断是否为僵尸状态SUCCESS执行成功否跳过直接使用已有输出FAILED执行失败是根据重试策略决定是否重试SKIPPED被跳过否跳过不产生输出TIMEOUT执行超时是同 FAILED这里特别要处理的是 RUNNING 状态。如果任务在某个节点执行到一半的时候进程崩溃了那个节点会停留在 RUNNING 状态。续跑的时候你需要判断这个 RUNNING 是真的还在跑比如另一个进程正在执行还是一个僵尸状态。我的做法是给每个 RUNNING 状态加一个心跳时间戳如果超过一定时间没有更新就认为是僵尸状态可以安全地重新执行。2.3 重试策略什么时候重试、重试几次、间隔多久重试不是简单地“失败了就再来一次”。不同类型的失败需要不同的处理策略瞬时故障比如网络抖动、接口限流这类失败重试大概率能成功适合用指数退避策略重试间隔逐渐拉长。逻辑错误比如输入数据格式不对、参数配置错误这类失败重试多少次都是一样的结果应该直接标记为 FAILED不要浪费资源。资源不足比如内存不够、磁盘满了这类失败需要先解决资源问题再重试可以设置一个较长的重试间隔。import time from enum import Enum class RetryPolicy(Enum): EXPONENTIAL exponential FIXED fixed NONE none def should_retry(error: Exception, attempt: int, max_retries: int) - bool: if attempt max_retries: return False if isinstance(error, (ValueError, TypeError)): return False return True def get_retry_delay(policy: RetryPolicy, attempt: int, base: float 1.0) - float: if policy RetryPolicy.EXPONENTIAL: return min(base * (2 ** attempt), 60.0) elif policy RetryPolicy.FIXED: return base return 0.0我在实际项目里一般设置最大重试次数为 3 次指数退避的基数设为 2 秒。这样第一次重试等 2 秒第二次等 4 秒第三次等 8 秒。如果三次都失败基本可以判断不是瞬时故障了继续重试意义不大。实操心得重试的时候一定要记录每次失败的详细信息包括错误类型、错误消息、堆栈跟踪、当时的上下文数据。这些信息对于排查问题至关重要。我见过太多人只记录一个“失败了”然后排查的时候完全不知道发生了什么。3. 可续跑的持久化设计3.1 状态持久化存什么、存哪里、什么时候存可续跑的前提是状态持久化。你需要把每个节点的执行状态和输出结果保存到一个可靠的地方这样即使进程崩溃了重启之后也能恢复。存储介质的选择取决于你的部署环境。如果是单机运行用 SQLite 或者本地文件就够了。如果是分布式环境可能需要用 Redis 或者数据库。我在单机项目里最常用的是 SQLite因为它零配置、支持事务、查询方便。import sqlite3 import json from datetime import datetime class StateStore: def __init__(self, db_path: str pipeline_state.db): self.conn sqlite3.connect(db_path) self._init_tables() def _init_tables(self): self.conn.execute( CREATE TABLE IF NOT EXISTS node_states ( task_id TEXT, node_name TEXT, status TEXT, output TEXT, error TEXT, attempt INTEGER, updated_at TEXT, PRIMARY KEY (task_id, node_name) ) ) self.conn.commit() def save_node_state(self, task_id: str, node_name: str, status: str, outputNone, errorNone, attempt0): self.conn.execute( INSERT OR REPLACE INTO node_states (task_id, node_name, status, output, error, attempt, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?) , (task_id, node_name, status, json.dumps(output) if output else None, str(error) if error else None, attempt, datetime.now().isoformat())) self.conn.commit() def get_completed_nodes(self, task_id: str) - dict: cursor self.conn.execute( SELECT node_name, output FROM node_states WHERE task_id ? AND status SUCCESS , (task_id,)) return {row[0]: json.loads(row[1]) for row in cursor.fetchall()}持久化的时机也很关键。我的做法是节点开始执行前先写入 RUNNING 状态节点执行成功后立即写入 SUCCESS 和输出结果节点失败后写入 FAILED 和错误信息。这样即使在任何时刻崩溃状态都是可恢复的。3.2 续跑逻辑从哪个节点开始、怎么恢复上下文续跑的核心逻辑其实很简单加载已完成节点的输出找到第一个未完成的节点从那里开始继续执行。def resume_pipeline(pipeline, task_id: str, store: StateStore): completed store.get_completed_nodes(task_id) context PipelineContext(task_idtask_id) context.node_results completed for node in pipeline.nodes: if node.name in completed: continue result node.execute(context) store.save_node_state(task_id, node.name, SUCCESS, result) context.set_output(node.name, result) return context但实际落地的时候有几个细节需要注意节点顺序的稳定性Pipeline 的节点顺序必须是确定的不能因为代码改动导致节点顺序变化否则续跑的时候会错乱。我的做法是给每个节点一个固定的序号续跑的时候按序号排序。节点版本兼容如果节点逻辑改了之前保存的输出可能不兼容。这种情况下需要给节点加版本号版本不匹配时强制重新执行。上下文恢复的完整性有些节点可能依赖多个前置节点的输出恢复上下文的时候要确保所有依赖的输出都已经加载。3.3 幂等性重试和续跑的安全保障幂等性是重试和续跑的安全底线。如果一个节点不是幂等的重试就可能导致重复写入、重复发送、重复扣款等严重后果。保证幂等性的常见手段有几种唯一键去重每次执行时生成一个唯一标识写入之前先检查是否已经存在。状态检查执行前先检查目标状态是否已经达成如果已经达成则直接返回。事务包裹把相关操作放在一个事务里要么全部成功要么全部回滚。def idempotent_write(store, task_id: str, node_name: str, data: dict): existing store.get_node_output(task_id, node_name) if existing is not None: return existing result do_actual_write(data) store.save_node_state(task_id, node_name, SUCCESS, result) return result踩过的坑有一次我做一个文件生成的任务节点重试的时候没有检查文件是否已经存在结果生成了好几份重复文件。后来加了幂等检查才解决。所以任何有副作用的操作都一定要考虑幂等性。4. 多 Agent 协作下的节点设计4.1 Agent 与节点的映射关系在多 Agent 场景下每个节点背后可能是一个独立的 Agent。这个 Agent 有自己的系统提示词、工具集、模型配置。节点和 Agent 的映射关系可以是一对一也可以是多对一。一对一的情况最简单每个节点就是一个 Agent负责一个特定的子任务。比如“分析节点”对应一个分析 Agent“写作节点”对应一个写作 Agent。这种方式的优点是职责清晰每个 Agent 的提示词可以针对性地优化。多对一的情况是多个节点共享同一个 Agent 配置只是输入不同。比如“处理第一条数据”和“处理第二条数据”可以是两个节点但用的是同一个 Agent。这种方式适合批量处理场景。我一般倾向于一对一因为这样每个节点的行为更可预测调试的时候也更容易定位问题。如果发现某个 Agent 在某些节点上表现不好可以单独调整那个节点的提示词不会影响其他节点。4.2 节点间的数据格式约定多 Agent 协作时节点之间的数据格式约定尤为重要。因为不同 Agent 可能由不同的人维护如果数据格式不统一很容易出现“上游输出的是 JSON下游期望的是 Markdown”这种问题。我的做法是定义一个统一的节点输出格式dataclass class NodeOutput: content: str # 主要文本内容 metadata: dict # 元数据如耗时、模型名称、token 数 artifacts: list # 附件列表如生成的文件路径 status: str # 执行状态所有节点都必须返回这个格式下游节点从这个格式里取自己需要的数据。这样即使上游 Agent 换了实现只要输出格式不变下游就不受影响。4.3 错误传播与降级策略在多 Agent 流水线里一个节点失败不一定意味着整个任务失败。有些节点是关键的失败了整个任务就没法继续有些节点是辅助的失败了可以降级处理。我给每个节点定义一个critical属性标记它是否是关键节点。关键节点失败时整个 Pipeline 暂停等待人工介入或自动重试。非关键节点失败时可以走降级逻辑比如使用默认值、跳过该步骤、或者用备用方案替代。class PipelineNode: def __init__(self, name, execute_fn, criticalTrue, fallback_fnNone, max_retries3): self.name name self.execute_fn execute_fn self.critical critical self.fallback_fn fallback_fn self.max_retries max_retries def execute(self, context): for attempt in range(self.max_retries): try: return self.execute_fn(context) except Exception as e: if attempt self.max_retries - 1: if self.critical: raise elif self.fallback_fn: return self.fallback_fn(context) else: return None time.sleep(2 ** attempt)这种设计让整个流水线更有韧性。非关键节点的失败不会阻塞整个任务关键节点的失败也有明确的重试和告警机制。5. 实操从零搭建一条可续跑的 Pipeline5.1 项目结构与环境准备先说一下我的项目结构这个结构在多个项目里验证过比较清晰pipeline_demo/ ├── main.py # 入口负责组装和启动 Pipeline ├── pipeline/ │ ├── __init__.py │ ├── context.py # PipelineContext 定义 │ ├── node.py # PipelineNode 定义 │ ├── runner.py # PipelineRunner 执行器 │ └── store.py # StateStore 状态存储 ├── nodes/ │ ├── __init__.py │ ├── collect.py # 数据采集节点 │ ├── clean.py # 数据清洗节点 │ ├── analyze.py # 分析节点 │ └── report.py # 报告生成节点 └── requirements.txt环境方面Python 3.9 以上就够了核心依赖只有标准库。如果你要用模型能力可以加上对应 SDK。状态存储用 SQLite不需要额外安装。5.2 定义节点与组装 Pipeline先定义几个具体的节点。每个节点就是一个函数接收 context返回输出。# nodes/collect.py def collect_data(context): 模拟从数据源采集数据 source context.data.get(source, default) raw_data fetch_from_source(source) return {raw: raw_data, count: len(raw_data)} # nodes/clean.py def clean_data(context): 清洗上一步采集的数据 raw context.get_input(collect) if not raw: raise ValueError(缺少采集节点的输出) cleaned [normalize(item) for item in raw[raw]] return {cleaned: cleaned, count: len(cleaned)} # nodes/analyze.py def analyze_data(context): 对清洗后的数据进行分析 cleaned context.get_input(clean) if not cleaned: raise ValueError(缺少清洗节点的输出) result run_analysis(cleaned[cleaned]) return {analysis: result} # nodes/report.py def generate_report(context): 生成最终报告 analysis context.get_input(analyze) report format_report(analysis[analysis]) return {report: report}然后组装 Pipeline# main.py from pipeline.runner import PipelineRunner from pipeline.node import PipelineNode from pipeline.store import StateStore from nodes.collect import collect_data from nodes.clean import clean_data from nodes.analyze import analyze_data from nodes.report import generate_report def build_pipeline(): return [ PipelineNode(collect, collect_data, criticalTrue, max_retries3), PipelineNode(clean, clean_data, criticalTrue, max_retries2), PipelineNode(analyze, analyze_data, criticalTrue, max_retries3), PipelineNode(report, generate_report, criticalFalse, fallback_fnlambda ctx: {report: 报告生成失败使用默认模板}), ] if __name__ __main__: store StateStore(pipeline_state.db) runner PipelineRunner(build_pipeline(), store) result runner.run(task_idtask_001, initial_data{source: demo}) print(result.node_results)5.3 执行器实现串行调度与断点续跑执行器是整个 Pipeline 的核心负责按顺序调度节点、处理重试、保存状态。# pipeline/runner.py import time from .context import PipelineContext class PipelineRunner: def __init__(self, nodes, store): self.nodes nodes self.store store def run(self, task_id: str, initial_data: dict None): context PipelineContext(task_idtask_id) if initial_data: context.data.update(initial_data) completed self.store.get_completed_nodes(task_id) context.node_results completed for node in self.nodes: if node.name in completed: print(f[跳过] {node.name} 已完成) continue print(f[执行] {node.name} 开始) self.store.save_node_state(task_id, node.name, RUNNING) try: output node.execute(context) self.store.save_node_state(task_id, node.name, SUCCESS, output) context.set_output(node.name, output) print(f[完成] {node.name}) except Exception as e: self.store.save_node_state(task_id, node.name, FAILED, errore) print(f[失败] {node.name}: {e}) if node.critical: raise else: fallback node.fallback_fn(context) if node.fallback_fn else None context.set_output(node.name, fallback) return context这段代码的关键点在于每次执行节点前先检查是否已完成已完成则跳过执行前写入 RUNNING 状态成功后写入 SUCCESS 和输出失败后写入 FAILED 和错误信息。这样无论在哪一步中断重新运行run方法都能从断点继续。5.4 验证续跑效果验证续跑最简单的方法就是人为制造中断。你可以在某个节点里加一个raise Exception跑一次让它在中间失败然后去掉异常再跑一次观察是否从断点继续。# 第一次运行analyze 节点会失败 # 输出 # [执行] collect 开始 # [完成] collect # [执行] clean 开始 # [完成] clean # [执行] analyze 开始 # [失败] analyze: 模拟失败 # 第二次运行collect 和 clean 会被跳过 # 输出 # [跳过] collect 已完成 # [跳过] clean 已完成 # [执行] analyze 开始 # [完成] analyze # [执行] report 开始 # [完成] report实测下来这个续跑机制非常可靠。即使进程被强制杀掉重新启动后也能正确恢复。唯一需要注意的是如果节点执行到一半被杀掉那个节点会停留在 RUNNING 状态续跑的时候需要处理这种情况。6. 常见问题与排查技巧实录6.1 节点卡在 RUNNING 状态怎么办这是最常见的问题。原因通常是进程在节点执行过程中崩溃了状态没有来得及更新。解决方法是在续跑逻辑里加一个超时判断如果 RUNNING 状态的最后更新时间超过阈值比如 10 分钟就认为是僵尸状态可以安全地重新执行。def is_stale_running(store, task_id, node_name, timeout_seconds600): state store.get_node_state(task_id, node_name) if state[status] ! RUNNING: return False updated_at datetime.fromisoformat(state[updated_at]) return (datetime.now() - updated_at).total_seconds() timeout_seconds6.2 输出数据太大导致存储膨胀有些节点的输出可能非常大比如生成了几 MB 的文本。如果全部存到 SQLite 里数据库会迅速膨胀。我的做法是大文件存到文件系统数据库里只存文件路径。这样既保证了可追溯性又不会让数据库变得不可维护。6.3 节点顺序调整后续跑错乱如果你在 Pipeline 中间插入了一个新节点续跑的时候会出现问题新节点没有历史状态但它的位置在已完成节点之间。我的处理方式是给每个节点一个稳定的序号续跑时按序号排序新节点的序号如果小于已完成节点的最大序号就强制重新执行后续所有节点。6.4 常见问题速查表问题现象可能原因排查方法解决方案节点一直 RUNNING进程崩溃检查心跳时间戳加超时判断自动重置续跑后重复执行状态未持久化检查存储写入日志确保每次状态变更都落盘输出数据丢失序列化失败检查输出是否可 JSON 序列化大文件外置存储重试次数过多逻辑错误被误判为瞬时故障检查错误类型区分错误类型逻辑错误不重试节点顺序错乱节点列表被修改对比历史节点顺序使用稳定序号排序最后分享一个小技巧在开发阶段我会给每个节点加一个dry_run模式只打印将要执行的操作而不实际执行。这样可以在不产生副作用的情况下验证 Pipeline 的组装逻辑是否正确。等确认无误后再关闭 dry_run 正式执行。这个习惯帮我避免了很多次“跑了一半发现配置错了”的尴尬。 1. 为什么要把一个任务拆成 Pipeline 节点链1.1 从“一把梭”到“分步走”的认知转变最早做自动化任务编排的时候我的做法很粗暴一个函数从头写到尾中间调模型、查数据、生成文件、发通知全塞在一起。跑通了确实爽但只要中间任何一步出问题整个任务就得从头再来。最要命的是有些步骤本身要花不少时间重跑一次成本极高而且重跑过程中还可能产生重复的副作用比如重复写入、重复发送。后来我开始接触多 Agent 协作的场景发现问题的本质不是“Agent 不够聪明”而是任务的组织方式太原始。一个复杂任务天然就应该被拆成若干个有明确边界的阶段每个阶段有自己的输入、输出、成功条件和失败处理策略。这就像工厂流水线你不会让一个工人从零件锻造一直干到整车出厂而是分成冲压、焊接、涂装、总装几个工位每个工位只关心自己的活出了问题也只影响那一个环节。Pipeline 节点链的核心思路就是把任务拆成串行的节点每个节点是一个独立的执行单元节点之间有明确的数据契约任何一个节点失败都可以单独重试整个链条可以从断点续跑而不需要从头再来。这个思路听起来简单但真正落地的时候有很多细节需要想清楚。比如节点之间怎么传递数据重试的时候怎么保证不重复执行已经成功的节点如果一个节点依赖前面多个节点的输出怎么办这些问题我在实际项目中都踩过坑下面会逐一展开。1.2 哪些场景适合用 Pipeline 节点链不是所有任务都值得拆成 Pipeline。如果一个任务总耗时不到几秒钟中间也没有外部依赖那直接写一个函数就完了拆成 Pipeline 反而是过度设计。但如果你遇到以下几种情况Pipeline 节点链的价值就会非常明显任务耗时长整个流程跑下来要几分钟甚至几十分钟中间任何一步失败都意味着大量时间浪费。步骤之间有明确的阶段划分比如“数据采集 → 数据清洗 → 分析计算 → 结果生成 → 通知分发”每个阶段做的事情性质不同。部分步骤有外部依赖比如调用外部接口、读写数据库、操作文件系统这些步骤失败概率较高需要单独重试。需要可观测性你想知道每个阶段花了多长时间、成功还是失败、输出是什么而不是只有一个“整体成功/失败”的结果。需要断点续跑任务跑到一半因为某种原因中断了你希望能从断点继续而不是从头再来。我做过的一个典型场景是从多个数据源采集原始素材经过清洗和结构化处理然后调用模型生成分析报告最后把报告分发到不同的下游系统。整个流程跑一次大概要十几分钟其中数据采集和模型调用是最容易出问题的环节。用了 Pipeline 节点链之后如果模型调用失败了我只需要重试那一个节点前面的数据采集和清洗结果都还在不用重新跑。1.3 串行流水线与并行编排的区别这里要特别说明一下我选择的是串行流水线而不是并行编排。并行编排适合那些互相独立的子任务比如同时调用三个不同的模型生成三份结果然后汇总。但很多实际任务是有严格先后依赖的后一步的输入依赖前一步的输出这种情况下串行是唯一合理的选择。串行流水线的一个关键优势是状态管理更简单。每个节点只需要关心自己的输入和输出不需要考虑并发冲突、资源竞争、状态同步这些问题。当然代价是总耗时是各节点耗时之和但对于大多数 IO 密集型的任务来说串行带来的额外等待时间远小于重试和调试成本的节省。还有一个我特别看重的点串行流水线天然适合可续跑。因为节点是按顺序执行的每个节点的输出都可以持久化续跑的时候只需要从第一个未完成的节点开始就行。并行编排要做到可续跑就复杂得多你需要追踪每个并行分支的状态还要处理分支之间的依赖关系。2. 节点链的核心设计契约、状态与重试2.1 节点之间的数据契约怎么定Pipeline 节点链最容易出问题的地方就是节点之间的数据传递。我见过很多实现方式有的用全局变量有的用共享内存有的用消息队列。每种方式都有各自的适用场景但在串行流水线里我认为最稳妥的方式是显式的数据契约。所谓显式数据契约就是每个节点明确声明自己的输入类型和输出类型节点之间通过一个统一的上下文对象来传递数据。这个上下文对象可以理解为一个“任务黑板”每个节点从上面读取自己需要的输入执行完后把自己的输出写上去。from dataclasses import dataclass, field from typing import Any, Optional dataclass class PipelineContext: task_id: str data: dict field(default_factorydict) node_results: dict field(default_factorydict) errors: dict field(default_factorydict) def set_output(self, node_name: str, output: Any): self.node_results[node_name] output self.data[node_name] output def get_input(self, node_name: str) - Optional[Any]: return self.node_results.get(node_name)这样做的好处是每个节点的输入输出都是可追溯的调试的时候你可以清楚地看到每个节点产出了什么。而且因为数据是显式传递的续跑的时候只需要把之前保存的node_results恢复出来就行。注意上下文对象里的数据要尽量保持可序列化。如果你往里面塞了数据库连接、文件句柄这类不可序列化的对象续跑的时候就会出问题。我的做法是节点内部自己管理这些资源上下文里只放纯数据。2.2 节点状态机每个节点到底有几种状态一个节点从开始执行到最终结束中间会经历多个状态。如果状态定义不清晰重试和续跑的逻辑就会很混乱。我一般把节点状态定义为以下几种状态含义是否可重试续跑时的处理PENDING尚未开始执行是从该节点开始执行RUNNING正在执行中视情况需要判断是否为僵尸状态SUCCESS执行成功否跳过直接使用已有输出FAILED执行失败是根据重试策略决定是否重试SKIPPED被跳过否跳过不产生输出TIMEOUT执行超时是同 FAILED这里特别要处理的是 RUNNING 状态。如果任务在某个节点执行到一半的时候进程崩溃了那个节点会停留在 RUNNING 状态。续跑的时候你需要判断这个 RUNNING 是真的还在跑比如另一个进程正在执行还是一个僵尸状态。我的做法是给每个 RUNNING 状态加一个心跳时间戳如果超过一定时间没有更新就认为是僵尸状态可以安全地重新执行。2.3 重试策略什么时候重试、重试几次、间隔多久重试不是简单地“失败了就再来一次”。不同类型的失败需要不同的处理策略瞬时故障比如网络抖动、接口限流这类失败重试大概率能成功适合用指数退避策略重试间隔逐渐拉长。逻辑错误比如输入数据格式不对、参数配置错误这类失败重试多少次都是一样的结果应该直接标记为 FAILED不要浪费资源。资源不足比如内存不够、磁盘满了这类失败需要先解决资源问题再重试可以设置一个较长的重试间隔。import time from enum import Enum class RetryPolicy(Enum): EXPONENTIAL exponential FIXED fixed NONE none def should_retry(error: Exception, attempt: int, max_retries: int) - bool: if attempt max_retries: return False if isinstance(error, (ValueError, TypeError)): return False return True def get_retry_delay(policy: RetryPolicy, attempt: int, base: float 1.0) - float: if policy RetryPolicy.EXPONENTIAL: return min(base * (2 ** attempt), 60.0) elif policy RetryPolicy.FIXED: return base return 0.0我在实际项目里一般设置最大重试次数为 3 次指数退避的基数设为 2 秒。这样第一次重试等 2 秒第二次等 4 秒第三次等 8 秒。如果三次都失败基本可以判断不是瞬时故障了继续重试意义不大。实操心得重试的时候一定要记录每次失败的详细信息包括错误类型、错误消息、堆栈跟踪、当时的上下文数据。这些信息对于排查问题至关重要。我见过太多人只记录一个“失败了”然后排查的时候完全不知道发生了什么。3. 可续跑的持久化设计3.1 状态持久化存什么、存哪里、什么时候存可续跑的前提是状态持久化。你需要把每个节点的执行状态和输出结果保存到一个可靠的地方这样即使进程崩溃了重启之后也能恢复。存储介质的选择取决于你的部署环境。如果是单机运行用 SQLite 或者本地文件就够了。如果是分布式环境可能需要用 Redis 或者数据库。我在单机项目里最常用的是 SQLite因为它零配置、支持事务、查询方便。import sqlite3 import json from datetime import datetime class StateStore: def __init__(self, db_path: str pipeline_state.db): self.conn sqlite3.connect(db_path) self._init_tables() def _init_tables(self): self.conn.execute( CREATE TABLE IF NOT EXISTS node_states ( task_id TEXT, node_name TEXT, status TEXT, output TEXT, error TEXT, attempt INTEGER, updated_at TEXT, PRIMARY KEY (task_id, node_name) ) ) self.conn.commit() def save_node_state(self, task_id: str, node_name: str, status: str, outputNone, errorNone, attempt0): self.conn.execute( INSERT OR REPLACE INTO node_states (task_id, node_name, status, output, error, attempt, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?) , (task_id, node_name, status, json.dumps(output) if output else None, str(error) if error else None, attempt, datetime.now().isoformat())) self.conn.commit() def get_completed_nodes(self, task_id: str) - dict: cursor self.conn.execute( SELECT node_name, output FROM node_states WHERE task_id ? AND status SUCCESS , (task_id,)) return {row[0]: json.loads(row[1]) for row in cursor.fetchall()}持久化的时机也很关键。我的做法是节点开始执行前先写入 RUNNING 状态节点执行成功后立即写入 SUCCESS 和输出结果节点失败后写入 FAILED 和错误信息。这样即使在任何时刻崩溃状态都是可恢复的。3.2 续跑逻辑从哪个节点开始、怎么恢复上下文续跑的核心逻辑其实很简单加载已完成节点的输出找到第一个未完成的节点从那里开始继续执行。def resume_pipeline(pipeline, task_id: str, store: StateStore): completed store.get_completed_nodes(task_id) context PipelineContext(task_idtask_id) context.node_results completed for node in pipeline.nodes: if node.name in completed: continue result node.execute(context) store.save_node_state(task_id, node.name, SUCCESS, result) context.set_output(node.name, result) return context但实际落地的时候有几个细节需要注意节点顺序的稳定性Pipeline 的节点顺序必须是确定的不能因为代码改动导致节点顺序变化否则续跑的时候会错乱。我的做法是给每个节点一个固定的序号续跑的时候按序号排序。节点版本兼容如果节点逻辑改了之前保存的输出可能不兼容。这种情况下需要给节点加版本号版本不匹配时强制重新执行。上下文恢复的完整性有些节点可能依赖多个前置节点的输出恢复上下文的时候要确保所有依赖的输出都已经加载。3.3 幂等性重试和续跑的安全保障幂等性是重试和续跑的安全底线。如果一个节点不是幂等的重试就可能导致重复写入、重复发送、重复扣款等严重后果。保证幂等性的常见手段有几种唯一键去重每次执行时生成一个唯一标识写入之前先检查是否已经存在。状态检查执行前先检查目标状态是否已经达成如果已经达成则直接返回。事务包裹把相关操作放在一个事务里要么全部成功要么全部回滚。def idempotent_write(store, task_id: str, node_name: str, data: dict): existing store.get_node_output(task_id, node_name) if existing is not None: return existing result do_actual_write(data) store.save_node_state(task_id, node_name, SUCCESS, result) return result踩过的坑有一次我做一个文件生成的任务节点重试的时候没有检查文件是否已经存在结果生成了好几份重复文件。后来加了幂等检查才解决。所以任何有副作用的操作都一定要考虑幂等性。4. 多 Agent 协作下的节点设计4.1 Agent 与节点的映射关系在多 Agent 场景下每个节点背后可能是一个独立的 Agent。这个 Agent 有自己的系统提示词、工具集、模型配置。节点和 Agent 的映射关系可以是一对一也可以是多对一。一对一的情况最简单每个节点就是一个 Agent负责一个特定的子任务。比如“分析节点”对应一个分析 Agent“写作节点”对应一个写作 Agent。这种方式的优点是职责清晰每个 Agent 的提示词可以针对性地优化。多对一的情况是多个节点共享同一个 Agent 配置只是输入不同。比如“处理第一条数据”和“处理第二条数据”可以是两个节点但用的是同一个 Agent。这种方式适合批量处理场景。我一般倾向于一对一因为这样每个节点的行为更可预测调试的时候也更容易定位问题。如果发现某个 Agent 在某些节点上表现不好可以单独调整那个节点的提示词不会影响其他节点。4.2 节点间的数据格式约定多 Agent 协作时节点之间的数据格式约定尤为重要。因为不同 Agent 可能由不同的人维护如果数据格式不统一很容易出现“上游输出的是 JSON下游期望的是 Markdown”这种问题。我的做法是定义一个统一的节点输出格式dataclass class NodeOutput: content: str # 主要文本内容 metadata: dict # 元数据如耗时、模型名称、token 数 artifacts: list # 附件列表如生成的文件路径 status: str # 执行状态所有节点都必须返回这个格式下游节点从这个格式里取自己需要的数据。这样即使上游 Agent 换了实现只要输出格式不变下游就不受影响。4.3 错误传播与降级策略在多 Agent 流水线里一个节点失败不一定意味着整个任务失败。有些节点是关键的失败了整个任务就没法继续有些节点是辅助的失败了可以降级处理。我给每个节点定义一个critical属性标记它是否是关键节点。关键节点失败时整个 Pipeline 暂停等待人工介入或自动重试。非关键节点失败时可以走降级逻辑比如使用默认值、跳过该步骤、或者用备用方案替代。class PipelineNode: def __init__(self, name, execute_fn, criticalTrue, fallback_fnNone, max_retries3): self.name name self.execute_fn execute_fn self.critical critical self.fallback_fn fallback_fn self.max_retries max_retries def execute(self, context): for attempt in range(self.max_retries): try: return self.execute_fn(context) except Exception as e: if attempt self.max_retries - 1: if self.critical: raise elif self.fallback_fn: return self.fallback_fn(context) else: return None time.sleep(2 ** attempt)这种设计让整个流水线更有韧性。非关键节点的失败不会阻塞整个任务关键节点的失败也有明确的重试和告警机制。5. 实操从零搭建一条可续跑的 Pipeline5.1 项目结构与环境准备先说一下我的项目结构这个结构在多个项目里验证过比较清晰pipeline_demo/ ├── main.py # 入口负责组装和启动 Pipeline ├── pipeline/ │ ├── __init__.py │ ├── context.py # PipelineContext 定义 │ ├── node.py # PipelineNode 定义 │ ├── runner.py # PipelineRunner 执行器 │ └── store.py # StateStore 状态存储 ├── nodes/ │ ├── __init__.py │ ├── collect.py # 数据采集节点 │ ├── clean.py # 数据清洗节点 │ ├── analyze.py # 分析节点 │ └── report.py # 报告生成节点 └── requirements.txt环境方面Python 3.9 以上就够了核心依赖只有标准库。如果你要用模型能力可以加上对应 SDK。状态存储用 SQLite不需要额外安装。5.2 定义节点与组装 Pipeline先定义几个具体的节点。每个节点就是一个函数接收 context返回输出。# nodes/collect.py def collect_data(context): 模拟从数据源采集数据 source context.data.get(source, default) raw_data fetch_from_source(source) return {raw: raw_data, count: len(raw_data)} # nodes/clean.py def clean_data(context): 清洗上一步采集的数据 raw context.get_input(collect) if not raw: raise ValueError(缺少采集节点的输出) cleaned [normalize(item) for item in raw[raw]] return {cleaned: cleaned, count: len(cleaned)} # nodes/analyze.py def analyze_data(context): 对清洗后的数据进行分析 cleaned context.get_input(clean) if not cleaned: raise ValueError(缺少清洗节点的输出) result run_analysis(cleaned[cleaned]) return {analysis: result} # nodes/report.py def generate_report(context): 生成最终报告 analysis context.get_input(analyze) report format_report(analysis[analysis]) return {report: report}然后组装 Pipeline# main.py from pipeline.runner import PipelineRunner from pipeline.node import PipelineNode from pipeline.store import StateStore from nodes.collect import collect_data from nodes.clean import clean_data from nodes.analyze import analyze_data from nodes.report import generate_report def build_pipeline(): return [ PipelineNode(collect, collect_data, criticalTrue, max_retries3), PipelineNode(clean, clean_data, criticalTrue, max_retries2), PipelineNode(analyze, analyze_data, criticalTrue, max_retries3), PipelineNode(report, generate_report, criticalFalse, fallback_fnlambda ctx: {report: 报告生成失败使用默认模板}), ] if __name__ __main__: store StateStore(pipeline_state.db) runner PipelineRunner(build_pipeline(), store) result runner.run(task_idtask_001, initial_data{source: demo}) print(result.node_results)5.3 执行器实现串行调度与断点续跑执行器是整个 Pipeline 的核心负责按顺序调度节点、处理重试、保存状态。# pipeline/runner.py import time from .context import PipelineContext class PipelineRunner: def __init__(self, nodes, store): self.nodes nodes self.store store def run(self, task_id: str, initial_data: dict None): context PipelineContext(task_idtask_id) if initial_data: context.data.update(initial_data) completed self.store.get_completed_nodes(task_id) context.node_results completed for node in self.nodes: if node.name in completed: print(f[跳过] {node.name} 已完成) continue print(f[执行] {node.name} 开始) self.store.save_node_state(task_id, node.name, RUNNING) try: output node.execute(context) self.store.save_node_state(task_id, node.name, SUCCESS, output) context.set_output(node.name, output) print(f[完成] {node.name}) except Exception as e: self.store.save_node_state(task_id, node.name, FAILED, errore) print(f[失败] {node.name}: {e}) if node.critical: raise else: fallback node.fallback_fn(context) if node.fallback_fn else None context.set_output(node.name, fallback) return context这段代码的关键点在于每次执行节点前先检查是否已完成已完成则跳过执行前写入 RUNNING 状态成功后写入 SUCCESS 和输出失败后写入 FAILED 和错误信息。这样无论在哪一步中断重新运行run方法都能从断点继续。5.4 验证续跑效果验证续跑最简单的方法就是人为制造中断。你可以在某个节点里加一个raise Exception跑一次让它在中间失败然后去掉异常再跑一次观察是否从断点继续。# 第一次运行analyze 节点会失败 # 输出 # [执行] collect 开始 # [完成] collect # [执行] clean 开始 # [完成] clean # [执行] analyze 开始 # [失败] analyze: 模拟失败 # 第二次运行collect 和 clean 会被跳过 # 输出 # [跳过] collect 已完成 # [跳过] clean 已完成 # [执行] analyze 开始 # [完成] analyze # [执行] report 开始 # [完成] report实测下来这个续跑机制非常可靠。即使进程被强制杀掉重新启动后也能正确恢复。唯一需要注意的是如果节点执行到一半被杀掉那个节点会停留在 RUNNING 状态续跑的时候需要处理这种情况。6. 常见问题与排查技巧实录6.1 节点卡在 RUNNING 状态怎么办这是最常见的问题。原因通常是进程在节点执行过程中崩溃了状态没有来得及更新。解决方法是在续跑逻辑里加一个超时判断如果 RUNNING 状态的最后更新时间超过阈值比如 10 分钟就认为是僵尸状态可以安全地重新执行。def is_stale_running(store, task_id, node_name, timeout_seconds600): state store.get_node_state(task_id, node_name) if state[status] ! RUNNING: return False updated_at datetime.fromisoformat(state[updated_at]) return (datetime.now() - updated_at).total_seconds() timeout_seconds6.2 输出数据太大导致存储膨胀有些节点的输出可能非常大比如生成了几 MB 的文本。如果全部存到 SQLite 里数据库会迅速膨胀。我的做法是大文件存到文件系统数据库里只存文件路径。这样既保证了可追溯性又不会让数据库变得不可维护。6.3 节点顺序调整后续跑错乱如果你在 Pipeline 中间插入了一个新节点续跑的时候会出现问题新节点没有历史状态但它的位置在已完成节点之间。我的处理方式是给每个节点一个稳定的序号续跑时按序号排序新节点的序号如果小于已完成节点的最大序号就强制重新执行后续所有节点。6.4 常见问题速查表问题现象可能原因排查方法解决方案节点一直 RUNNING进程崩溃检查心跳时间戳加超时判断自动重置续跑后重复执行状态未持久化检查存储写入日志确保每次状态变更都落盘输出数据丢失序列化失败检查输出是否可 JSON 序列化大文件外置存储重试次数过多逻辑错误被误判为瞬时故障检查错误类型区分错误类型逻辑错误不重试节点顺序错乱节点列表被修改对比历史节点顺序使用稳定序号排序最后分享一个小技巧在开发阶段我会给每个节点加一个dry_run模式只打印将要执行的操作而不实际执行。这样可以在不产生副作用的情况下验证 Pipeline 的组装逻辑是否正确。等确认无误后再关闭 dry_run 正式执行。这个习惯帮我避免了很多次“跑了一半发现配置错了”的尴尬。