多Agent串行流水线设计：节点抽象、重试与断点续跑实践

很多人第一次接触多 Agent 编排时，习惯把所有的工具调用、模型对话、条件判断都塞进一个巨大的循环里。跑通 Demo 没问题，一旦放到真实业务场景，模型偶尔抽风、第三方接口超时、中间步骤参数错位，任何一个环节出问题，整个任务就得从头开始。之前在一个模拟项目里做自动化报告生成，串了五六个 Agent，每次跑到第三个环节接口超时，整条链路就废了，前面识别、清洗、分类的结果全部丢失，重跑一遍不仅烧 token，时间成本也扛不住。

后来我把整个流程改造成了串行流水线：每个 Agent 只负责一个节点，节点之间通过结构化的数据对象传递，每个节点可以独立重试，整条链路可以从任意节点续跑。这套方案上线之后，任务失败率明显下降，排查问题也简单了很多——哪个节点挂了，直接看那个节点的日志就行，不用再从头到尾捋一遍。

这篇就把这套多 Agent 串行流水线的设计思路和落地细节完整拆开，从节点抽象、输入输出契约、重试策略到断点续跑机制，一步步讲清楚。适合已经在做 Agent 编排、但被任务可靠性和可维护性困扰的开发者，也适合正准备从单 Agent 转向多 Agent 协作的团队参考。

1. 为什么多轮编排需要流水线：从编排痛点到节点链抽象

先聊清楚一个问题：为什么不能像写普通脚本那样，从上到下把 Agent 调用写完就行？如果你只是本地玩一玩，倒也无所谓。但只要你涉及外部依赖——模型 API、数据库、文件系统、第三方服务——你就一定会遇到三个问题。

首先是任务中断的恢复成本太高。一次任务可能涉及多个步骤，每个步骤都是一次模型调用，一次调用就是几百上千 token。假设一个流程有八个节点，跑到第六个节点时网络抖动导致请求超时，整个任务回滚到起点重新执行。前面的 token 全部白费，如果流程中还涉及外部系统的副作用操作，比如写数据库、发通知，重复执行还可能造成数据重复。

其次是失败原因很难定位。多 Agent 协作时，节点之间的数据传递往往通过内存变量或全局状态来实现，一旦结果不对，你很难说是哪个 Agent 产生了错误输出，是模型理解偏差，还是上游数据在传递过程中被意外修改。没有清晰的边界，排查问题基本靠猜。

第三是并发或重试时的状态污染。多个任务同时跑，或者同一个任务手动重试，如果状态没有做隔离，两个任务的中间结果互相覆盖，那问题就更复杂了。

流水线解决的就是这三个问题。核心思路其实很简单：把任务拆成一个个职责单一、边界明确的节点，每个节点内部可以包含一个 Agent 以及它需要的工具调用，节点之间不直接通信，而是通过一个统一的上下文对象来传递数据。每个节点是一个独立的工作单元，可以单独重试、单独替换、单独测试。

这里有一个关键的设计转变：从"一个 Agent 干完所有事"变成"一个节点干一件事，多个节点协作完成任务"。节点里面的 Agent 不再需要理解整个任务，它只需要理解自己负责的那一段。这也意味着你不再依赖一个上下文超长的 Prompt 来约束模型行为，每个节点的输入输出都收敛到一个小范围内，效果稳定性会明显好很多。

我们来看一个具体的例子。假设我们要做一个"行业研报自动生成"任务，传统的单 Agent 方案是：给模型一个长 Prompt，让它自己完成信息检索、数据整理、报告撰写、格式排版。结果就是模型经常漏掉某些检索步骤，或者把检索到的数据在报告中张冠李戴。

用流水线重构之后，任务被拆成这样几个节点：

• 信息检索节点：根据主题生成查询词，调用搜索工具，返回原始资料列表
• 内容清洗节点：从原始资料中提取关键数据，去重，格式化，输出结构化字段
• 大纲生成节点：基于清洗后的内容生成文章大纲，输出大纲 JSON
• 正文撰写节点：按照大纲逐段撰写，输出 Markdown 正文
• 格式排版节点：对正文进行排版优化，输出最终文档

每个节点只负责一件事，输入输出都有固定格式，任何一个节点失败，只需要重跑这个节点。

有的朋友可能会问：节点拆得越细，模型调用次数就越多，成本不是更高吗？确实，节点多了之后单次任务的调用次数会上升，但你仔细算一笔账：节点细了之后，重试的单位从"整条链路"变成"单个节点"。一个包含十次模型调用的任务，每个节点失败概率如果是 5%，整条链路无重试成功率只有 59.8%；但有了节点级重试，每个节点失败后最多重试 3 次，链路最终成功率能到 97% 以上。更重要的是，你省掉了"从头再来"的重复调用开销。算下来总体成本反而更低。

2. 节点链设计的关键一步：把 Agent 装进标准化的执行单元

确定了要用流水线架构之后，下一个问题就是：节点长什么样？怎么定义一个节点？这一步是整个方案的地基，定义得好不好，直接决定了后面重试、续跑、监控这些能力能不能实现。

2.1 节点抽象：输入、执行、输出的统一契约

我在落地这套方案时，每个节点都实现成一个函数式组件，有明确的输入 schema 和输出 schema。核心结构大概是这样的：

class PipelineNode: def __init__(self, name, input_schema, output_schema, handler): self.name = name self.input_schema = input_schema self.output_schema = output_schema self.handler = handler def execute(self, context): # 从上下文中取出本节点需要的输入字段 input_data = self.input_schema.filter(context) # 调用具体的 Agent 逻辑 result = self.handler(input_data) # 校验输出是否符合预期 self.output_schema.validate(result) # 将输出写回上下文 context.update(self.name, result) return result def retry(self, context, max_attempts=3): for attempt in range(max_attempts): try: return self.execute(context) except Exception as e: logger.warning(f"节点 {self.name} 第 {attempt + 1} 次执行失败: {e}") raise RuntimeError(f"节点 {self.name} 重试 {max_attempts} 次后仍然失败")

每个节点有三样东西：名字、输入输出契约、执行函数。执行函数内部可以是任何逻辑——调一个模型、跑一段代码、调一个外部 API——但对外部来说，它就是一个黑盒，输入一个结构化的数据，输出一个结构化的结果。

输入输出契约用 schema 来定义，我习惯用 Pydantic 的 BaseModel 来实现，这样既能做运行时校验，又能自动生成文档。比如内容清洗节点的输出 schema 可能长这样：

class CleaningOutput(BaseModel): source_id: str title: str key_points: List[str] raw_text: str cleaned_length: int quality_score: float

其中 quality_score 是清洗节点自己对数据质量的评分，下游节点可以依据这个分数决定是否使用这段数据。这个字段是实践中加上的，因为单纯的文本清洗结果很难用规则判断好坏，让大模型在输出时附带一个置信度，下游节点可以做阈值判断，能有效拦截低质量数据进入后续环节，避免模型"一本正经地胡说八道"带来的连锁反应。

2.2 上下文对象：节点间数据传递的唯一通道

节点之间不直接持有彼此的引用，所有数据都通过一个上下文对象传递。上下文对象本质上就是一个带有版本管理能力的字典，每个节点写入的数据都会记录来源节点和执行批次。

这样做有三个好处。第一，数据流清晰。任何时候你想知道某个字段是谁产生的，只需要看上下文的元信息，比在代码里到处搜索变量赋值要高效得多。第二，支持并行扩展。虽然串行模式下节点依次执行，但上下文对象的数据隔离设计意味着未来某些无依赖节点可以并行执行，不需要改动节点内部逻辑。第三，便于持久化。上下文对象是纯数据，可以序列化成 JSON 存到数据库或文件里，这就是后面续跑机制的基础。

2.3 编排器：负责连接节点和处理失败

节点定义好了，还需要一个编排器来把它们串起来。我把编排器设计得尽量简单，它只做三件事：

• 按顺序执行节点列表
• 记录每个节点的执行状态
• 在节点失败时抛出异常，由上层决定是重试还是暂停

class PipelineOrchestrator: def __init__(self, nodes, context_store): self.nodes = nodes self.context_store = context_store def run(self, initial_context, start_node=None, max_retries=3): context = initial_context.copy() start_index = self._find_node_index(start_node) if start_node else 0 for i in range(start_index, len(self.nodes)): node = self.nodes[i] try: node.retry(context, max_attempts=max_retries) self.context_store.save(context, node.name, status="success") except Exception as e: self.context_store.save(context, node.name, status="failed", error=str(e)) raise PipelineExecutionError( f"节点 {node.name} 执行失败，当前进度: {i}/{len(self.nodes)}" ) return context

注意这个start_node参数，它是断点续跑的关键入口。只要上下文对象被持久化了，下次重跑时就可以从失败的节点开始，而不用从头执行。

3. 节点的失败处理与重试策略：错误分类、退避重试、幂等控制

重试是流水线最核心的能力，但重试不是简单地在 except 里再调一次函数。设计不好的重试机制，可能带来额外的副作用，甚至掩盖真正的问题。在处理重试之前，我们得先对错误做一个分类。

3.1 错误分类：决定哪些值得重试

我把错误分成三类：

瞬时错误重试是基本操作，但要注意退避策略。如果用固定间隔重试，短时间内的密集重试很容易继续触发限流。我用的策略是"指数退避 + 抖动"，第一次等 1 秒，第二次等 2 秒，第三次等 4 秒，然后加上一个随机偏移量：

import random import time def exponential_backoff(attempt, base_delay=1, max_delay=60): delay = min(base_delay * (2 ** attempt), max_delay) # 加一个 0.5 倍的随机抖动，防止多个节点同时重试造成惊群 delay = delay * (0.5 + random.random()) return delay

数据错误不重试是原则。如果输入数据本身有问题，重试一百次结果都一样，白白浪费资源。这时候应该做的是快速失败，把问题暴露出来，让上游环节或者人工介入。

模型错误最有意思。模型输出有时候是偶发的，换个 temperature 再来一次可能就好了；但也有可能是 Prompt 本身设计有缺陷，重试永远会碰到相同的问题。我的处理方式是：模型输出校验失败后重试一次，如果第二次还是失败，就把当前节点的完整输入、输出和失败原因存下来，然后让整个流水线进入"暂停"状态，等待人工介入。这样做虽然损失了一点自动化率，但避免了在错误方向上空转。

3.2 重试状态机：让每次重试有据可查

重试不仅仅是循环调用，还需要一个状态机来管理每次尝试的状态。我给每个节点的每次执行都记录一个状态条目，包含：尝试序号、开始时间、结束时间、耗时、状态（success/failed）、错误信息、输出结果摘要。这些状态条目最终形成一个执行轨迹，所有节点的状态组合在一起，就是整条流水线的一次完整执行记录。

有了这个执行轨迹，你不仅能看到哪个节点失败，还能看到它重试了哪些次，每次失败的原因是什么。之前排查一个线上问题，节点"大纲生成"连续失败了 3 次，第一次是因为 API 超时，第二次是因为输出 JSON 格式不合法，第三次是因为关键字段缺失，每次原因都不一样。如果没有执行轨迹，你根本不知道问题其实包含三个关联的子问题，只会盲目地调整个别参数。

3.3 幂等控制：防止重复执行产生副作用

重试天然要求节点具备幂等性——同一个输入执行两次，效果和执行一次完全相同。对于纯计算型节点，幂等性天然满足，模型调用只要你用相同的输入和参数，生成的输出大体一致（除了随机性）。问题出在包含外部副作用的节点，比如写数据库、发邮件、调用支付接口。

应对思路有两个方向。一个是从节点设计层面规避：把有副作用的操作从节点中拆出去，节点只负责生成"要执行的操作描述"，由一个统一的外部执行器去真正落库。另一个是给每个节点执行加一个 request_id 作为幂等键，外部系统根据这个键判断请求是否已经处理过。具体用哪种，取决于你的外部系统是否支持幂等写入。但无论如何，在设计节点时就把幂等性纳入考虑，否则重试机制越强大，潜在的风险也越大。

4. 断点续跑机制：任务状态持久化与从失败节点恢复

重试解决的是小概率的瞬时失败，但现实中还有一种更麻烦的情况：任务跑到一半，进程挂了，服务器重启了，或者模型 API 连续几个小时不可用。这时候你不能指望靠重试硬扛，你需要的是——把任务状态保存起来，等条件恢复了接着跑。这就引入了断点续跑机制。

4.1 状态快照：把内存中的上下文变成可恢复的数据

续跑的前提是状态能够持久化。我在设计上下文对象的时候，就让它天然支持序列化和反序列化：

class PipelineContext: def __init__(self, initial_data=None): self.data = initial_data or {} self.metadata = { "created_at": time.time(), "updated_at": time.time(), "node_history": [], "version": 1 } def to_dict(self): return {"data": self.data, "metadata": self.metadata} @classmethod def from_dict(cls, raw): ctx = cls() ctx.data = raw["data"] ctx.metadata = raw["metadata"] return ctx

持久化的策略是在每个节点成功后保存一次快照。快照包含三部分：当前上下文数据、已执行节点列表、各节点的执行轨迹。保存位置可以是本地文件、Redis 或者数据库，看你的部署环境。我用的是数据库存储，因为线上是多实例部署，本地文件每台机器都不一样，没法共享状态。

4.2 从失败节点恢复：续跑的执行流程

续跑的流程说起来很简单：加载最近一次成功的快照，从失败节点的下一个节点开始继续执行。但有一个细节想提醒大家，快照里的上下文数据必须是"干净"的。什么算干净？就是没有包含失败节点写入的部分。所以在保存快照时，要确保只保存节点成功之后的数据，不要让失败节点的半成品数据污染快照。

下面是续跑的完整流程：

1. 加载指定任务 ID 的最新快照
2. 解析快照中的 node_history，找到最后一个成功的节点
3. 从该节点的下一个节点开始执行
4. 中途再次失败，则重复整个过程

这里有个值得注意的小问题，那就是"最后一个成功的节点"到底在哪个位置。我见过有人简单地存一个数字下标，但节点的增删会破坏这个下标的有效性。更稳妥的方式是存节点名，而不是下标。节点名加节点列表的哈希，启动时先校验哈希，能快速发现代码版本变更导致的节点链不一致。

4.3 暂停与人工介入：把自动化流程变得可操控

有时候失败并不是"重试就能解决"的，模型输出反复不合格，或者上游数据源发生了结构性变化，这时候需要人工介入。我在编排器中增加了一个暂停机制：

class PipelineOrchestrator: def run(self, initial_context, start_node=None, max_retries=3): context = initial_context.copy() start_index = self._find_node_index(start_node) if start_node else 0 for i in range(start_index, len(self.nodes)): node = self.nodes[i] if context.metadata.get("paused_at") == node.name and context.metadata.get("paused"): # 当前节点处于暂停状态，等待人工确认 raise PipelinePausedError(f"节点 {node.name} 等待人工介入") try: node.retry(context, max_attempts=max_retries) self.context_store.save(context, node.name, status="success") except Exception as e: self.context_store.save(context, node.name, status="failed", error=str(e)) raise PipelineExecutionError( f"节点 {node.name} 执行失败，当前进度: {i}/{len(self.nodes)}" ) return context

暂停机制的核心是允许任务停在某个节点前，人工检查或修改上下文数据之后，再继续执行。比如有一个"生成财务指标"节点，模型算出来的数字明显有误，人工介入后可以直接把正确的数值更新到上下文里，然后从该节点重新执行。这比"报错、改代码、从头跑"要高效得多，也让流水线在关键节点上能够保持"人审"的灵活性。

4.4 任务管理：让每个任务在数据库里有一行可追踪的记录

断点续跑持续跑的前提，是每一个任务在系统里有一个明确的记录。我给任务分配一个 task_id，然后维护一个任务状态表：

• 任务基本信息：task_id、流水线名称、创建时间、启动方式
• 当前状态：running / paused / success / failed
• 当前进度：最后一个成功节点名、当前执行节点名
• 执行统计：总节点数、已执行节点数、重试次数、总耗时
• 错误信息：最近一次失败的异常堆栈

有了这张表，你可以随时查看所有任务的执行状态，也可以写一个简单的管理后台，列出失败任务并支持一键续跑。

5. 状态持久化与运行监控：把执行过程变成可以复盘的数据

流水线跑起来只是第一步，真正让它变得可靠的是监控和复盘能力。如果执行过程是黑盒，那出了问题只能干瞪眼；如果执行过程有完整的日志和数据，大多数问题都可以通过分析数据来定位。

5.1 执行轨迹：记录每个节点每一次运行的完整信息

我推荐每个任务都保存一份完整的执行轨迹，而不仅是最终结果。这个轨迹是一个列表，每个节点每次执行都对应一个条目。

execution_trace = [ { "node": "信息检索节点", "attempt": 1, "status": "success", "start_time": "2025-01-15T10:00:01", "end_time": "2025-01-15T10:00:12", "duration_ms": 11000, "token_usage": {"prompt_tokens": 1200, "completion_tokens": 800}, "output_summary": "检索到 15 条资料，来自 3 个数据源" }, { "node": "内容清洗节点", "attempt": 1, "status": "failed", "error": "API 超时（超过 30 秒未响应）", "start_time": "2025-01-15T10:00:13", "end_time": "2025-01-15T10:00:45" }, { "node": "内容清洗节点", "attempt": 2, "status": "success", "start_time": "2025-01-15T10:01:00", "end_time": "2025-01-15T10:01:08", "duration_ms": 8000, "output_summary": "清洗完成，保留 12 条有效资料" } ]

有了这个执行轨迹，你可以做几件事。第一，定位问题：哪个节点耗时最长、失败次数最多，一目了然。第二，成本核算：把每次调用的 token 使用量累加，可以精确算出每个任务、每个节点的成本。第三，优化依据：如果某个节点频繁失败，要么调整 Prompt，要么更换模型，要么拆分节点，决策都有了数据支撑。

5.2 运行监控面板：让任务状态可视化

代码写完后，我搭了一个简单的监控面板。不需要很复杂，几个核心指标就够用：当前运行中的任务数、成功率、平均执行时长、各节点失败率、各节点平均耗时。这个面板对团队协作特别有用——之前流程跑挂了只能靠同事口头通知，现在谁都能打开面板看到任务状态，减少了很多沟通成本。

建议在节点级别也做失败率统计。如果某个节点的失败率明显高于其他节点，说明它要么是依赖的外部服务不稳定，要么是节点内部的 Prompt 设计有问题。我遇到过的情况是"正文撰写节点"失败率特别高，一查日志发现是因为模型输出的 Markdown 经常带有多余的前置说明文字，导致后续的分段解析逻辑报错。这属于典型的 Prompt 设计缺陷，在单 Agent 模式下很难发现，但在流水线的统计视角下一眼就能看出来。

5.3 上下文版本管理：复盘时能看到每一步的状态

为了支持人审和复盘，我给上下文对象加了简单的版本管理。每个节点执行完成后，保存一份上下文数据的只读副本，带时间戳和版本号。这样一旦发现问题，你可以回溯任意时间点的上下文状态，看数据是什么时候开始偏离预期的。

这种机制的收益在排查"数据悄悄变脏"的问题时特别明显。有一次任务最终报告里出现了一个异常数据点，我们顺着上下文版本一路倒推，发现在第二个节点处理时，上游传入的日期字段被错误地做了时区转换，导致后续所有时间相关的计算都偏移了一个小时。没有版本管理，这种问题几乎不可能定位。

6. 一个完整的串行流水线实例：从行业资料到结构化报告

前面把机制讲得差不多了，这里用一个完整示例把整套流程串起来。假设我们要做一个"行业情报日报"的自动生成任务，流程包含五个节点：资料采集、分类打标、摘要提取、要点整合、日报生成。

6.1 节点配置与编排代码

每个节点的 handler 内部可以是调模型，也可以是调工具，外部看起来都是统一的输入输出。下面是一个完整到可以直接跑的示例：

from pydantic import BaseModel from typing import List # ---------- 节点输入输出定义 ---------- class ArticleItem(BaseModel): article_id: str title: str url: str source: str published_at: str class ClassifiedItem(BaseModel): article_id: str category: str tags: List[str] relevance_score: float # ---------- 节点实现 ---------- class CollectNode: """资料采集节点：从多个数据源拉取当天的文章列表""" def __init__(self, fetcher): self.fetcher = fetcher def handler(self, input_data: dict) -> List[ArticleItem]: articles = self.fetcher.fetch_articles(input_data["query"]) return [ArticleItem(**a) for a in articles if a.get("title")] class ClassifyNode: """分类打标节点：判断每篇文章的领域分类""" def __init__(self, llm_client): self.llm = llm_client def handler(self, input_data: dict) -> List[ClassifiedItem]: # 批量分类，减少模型调用次数 items = input_data["articles"] classified = self.llm.classify_batch(items, categories=["AI", "金融", "新能源", "医疗"]) return classified # ---------- 编排 ---------- def build_daily_report_pipeline(): nodes = [ PipelineNode( name="资料采集", input_schema=SimpleSchema(["query"]), output_schema=SimpleSchema(["articles"]), handler=CollectNode(fetcher).handler ), PipelineNode( name="分类打标", input_schema=SimpleSchema(["articles"]), output_schema=SimpleSchema(["classified_items"]), handler=ClassifyNode(llm_client).handler ), # ... 之后的节点类似，这里省略 ] return PipelineOrchestrator(nodes=nodes, context_store=DBContextStore()) if __name__ == "__main__": ctx = PipelineContext({"query": "人工智能行业动态"}) orchestrator = build_daily_report_pipeline() result = orchestrator.run(ctx) print(result.to_dict())

6.2 触发重试与续跑的完整日志

下面是这次执行产生的关键日志序列，注意看节点失败之后系统是如何处理的：

2025-01-15T10:00:01 INFO 节点 "资料采集" 开始执行 2025-01-15T10:00:12 INFO 节点 "资料采集" 执行成功，耗时 11s，产出 15 条文章 2025-01-15T10:00:13 INFO 节点 "分类打标" 开始执行 2025-01-15T10:00:40 ERROR 节点 "分类打标" 执行失败：第 1 次尝试，模型输出 JSON 格式非法 2025-01-15T10:00:45 INFO 节点 "分类打标" 第 2 次尝试开始（退避延时 1.32s） 2025-01-15T10:01:10 ERROR 节点 "分类打标" 执行失败：第 2 次尝试，API 连接超时 2025-01-15T10:01:12 INFO 节点 "分类打标" 第 3 次尝试开始（退避延时 2.87s） 2025-01-15T10:01:20 INFO 节点 "分类打标" 执行成功，耗时 8s，产出 15 条分类结果

6.3 实际效果与收益

这套流水线上线跑了一个多月，整体表现相当稳定。从统计看，单次任务的平均执行时长从之前的 8 分钟降到了 3 分钟以内，不是说模型变快了，而是省掉了大量"从头重跑"的浪费。任务最终成功率稳定在 98% 以上，剩余 2% 主要是外部数据源接口持续不可用或输入数据质量太差，这类问题重试解决不了，需要数据侧修复。

7. 上线前后最容易踩的坑：字段缺失、不可重试节点、状态持久化失败

最后分享几个实操中踩过的坑，也是我每次给别人讲这套方案时必然会提到的点。

第一个坑：字段缺失在节点边界才暴露。多 Agent 串行流水线里，数据传递依赖字段名的一致性，任何一个节点的输出 schema 和下一个节点的输入 schema 不匹配，都会在执行时报错。Pydantic 的校验能帮你兜底，但更重要的还是节点设计时的字段命名规范。我的建议是，给每个字段起名时尽量用"名词 + 语义"的组合，比如cleaned_articles、classified_items，不要用data、result这种太通用的名字。这样即使节点多了，字段的语义也不会混淆。另外，在写节点 handler 时，尽量用input_data.get("xxx")而不是input_data["xxx"]，再加上一个显式的缺失字段检查，报错信息就清晰很多。

第二个坑：把不可重试的节点设计成可重试。有些节点的执行逻辑天然不可重试，典型的有两种。一种是"扣减型"操作，比如消耗积分、请求外部服务扣款，重试可能导致重复扣减。另一种是"追加型"操作，比如向数据库插入记录，重试会产生重复行。面对这类节点，我的处理方式很明确：不在节点内部做重试，把这个节点的重试能力关闭，它一旦失败就进入暂停状态，由上层逻辑判断如何处理。宁可让流程卡住，也不能让它带着副作用重试。

第三个坑：状态持久化本身的失败。续跑机制依赖状态保存，但状态保存这个动作本身也可能失败。比如数据库连接超时、磁盘空间不足、JSON 序列化报错。如果你不做额外的保护，节点执行成功了但状态保存失败，下次续跑时你以为"它跑到了第五个节点"，实际恢复的时候却是"第二个节点"，中间的结果全部丢失。我的做法是节点结果先写入内存，状态保存失败时整条流水线立刻终止并上报，防止继续执行导致数据不一致。

第四个坑：上下文对象越攒越大。节点多了之后，上下文里可能堆了很多只在中间环节需要的临时数据，比如搜索结果原始 HTML、中间格式的草稿文本。这些数据加在一起可能非常臃肿，不仅拖慢序列化和存储速度，还增加了每次模型调用的前文长度（如果你把整个上下文传给模型的话）。我的建议是：每个节点在输出 schema 中明确声明哪些字段是"下游必需"的，哪些是"仅本节点使用"的临时数据。编排器在节点成功后，自动从上下文中清理掉所有不再被下游需要的临时字段。这相当于给上下文做了一次垃圾回收，对控制 token 成本非常有效。

8. 进阶方向：动态分支、并行节点与人工审批的融合

串行流水线解决了任务可靠性的问题，但它的架构决定了它是一条直线——A 做完做 B，B 做完做 C。真实业务里其实有很多场景需要更加灵活的控制流，比如根据中间结果决定走哪个分支、多个独立节点并行执行、关键节点需要人工审批。这些内容已经超出了串行流水线的范围，但我把探索思路整理在这里，给有需要的朋友一些方向参考。

第一个方向是动态分支。目前节点列表是静态的，不可能根据执行结果动态增删节点。但实现上其实并不复杂：在编排器中增加一个"决策节点"，它的 handler 输出一个next_node字段，编排器根据这个字段决定下一步跳转到哪个节点。本质上是用一个小的状态机替换了线性的 for 循环。

第二个方向是并行节点。串行流水线的设计保证每个节点只依赖上游数据，这意味着某些无依赖的节点天然可以并行执行。我目前的编排器还只支持串行，但数据流设计已经为并行做好了准备——需要改动的是编排器内部，把"按顺序执行"变成"按依赖关系执行"，这个方案可以参考有向无环图的调度思路来扩展。

第三个方向是人工审批节点。前面提到了暂停机制，可以再往前一步，设计一个专门的"审批节点"：节点执行完成后，不自动进入下一个节点，而是把当前上下文发送给一个审批接口，由人工确认后返回放行或修改的指令。这特别适合内容审核、财务对账这类需要人审的业务场景，而且和现有的暂停状态、上下文快照机制可以很好地衔接。

对于串行流水线本身，最后想强调的一点是：它最大的价值不是用了多少 Agent，而是让每一个 Agent 的执行过程变得可控、可查、可恢复。模型输出的不确定性无法消除，但我们可以通过架构设计，把这些不确定性约束在一个个独立的小节点里，让它们对整体流程的影响降到最低。如果你正在做多 Agent 应用，还在为"跑一半就挂、挂了就得重来"发愁，我很建议从这个方案入手，把它当成你的任务编排底座。