企业官网建设流程全解析

做多Agent应用最头疼的问题，不是把每个Agent本身写好，而是多个Agent串起来之后，任务跑到一半卡住了。我今天想聊的“多Agent串行流水线”，本质上就是把这个复杂任务拆成一条Pipeline节点链：每个节点由一个Agent专职处理，节点之间用共享上下文传递结果，整条链支持单节点重试、断点续跑。我第一回把这套东西落地，是因为一个多步骤数据处理任务总在最后一步超时，每次从头重跑要花二十分钟；改成流水线之后，哪一步挂了就重跑哪一步，进程重启也能从失败节点继续往下走，整体耗时基本被压到原来的三分之一。这篇文章会从设计思路、状态管理、重试与续跑实现，一路讲到完整的可落地Demo，适合已经在用大模型接口做应用、但被任务编排搞得头疼的开发者，也适合刚接触Agent开发、想找一套稳定执行框架的朋友。

1. 为什么把任务拆成多Agent串行流水线

1.1 单次大Prompt的三大坑

先聊聊我为什么不直接用一个大Prompt把所有事情都交给模型做。单个Agent处理复杂任务，最明显的三个问题：

第一，结果不稳定。任务一旦超过两三个子步骤，模型很容易在中间逻辑上出错。比如你先让它做实体识别再做情感判断，它可能把两步的动作搅在一起，输出一个不伦不类的混合结果。你反复调Prompt能好一阵子，但换一批数据又翻车，这种随机性非常消耗耐心。

第二，失败成本高。一次请求从发出到返回可能要好几分钟，如果中间某一步格式错了，整次输出作废，重新提交时又要完整跑一遍。任务越重，重跑的时间和经济成本就越不可控。我遇到过一次批处理任务在倒数第二步格式错误，重跑时又碰上限流，前后浪费了小半天。

第三，不透明。你只能看到最终输出，完全不知道任务到底卡在哪一步。日志里只有一次Prompt和一次Response，出了问题只能靠猜，猜来猜去最后还是得开打印一个个试。

当时我做一个模拟项目X，输入是大批量的文本片段，需要先做实体识别，再做情感判断，最后拼装成结构化表格。一次性Prompt跑得又慢又烂，后来拆成三个独立Agent，反而稳定很多。这件事让我意识到，任务拆解本身就是一种可靠性设计，拆得合理，后续的所有故障处理都会简单一个量级。

1.2 串行流水线解决了什么

串行流水线的思路很直接：把任务拆成有先后依赖的多个步骤，每个步骤对应一个Agent，前一个节点输出结构化结果，后一个节点只消费它需要的那部分数据。节点之间靠一份共享的TaskContext传数据，而不是把整个对话历史滚动塞给下一个节点。

这样做的好处非常实际：

• 定位问题精准：哪一步失败，看节点状态就知道，不用再从最终结果反推。
• 单点重试成本低：失败节点单独重跑，已成功的节点一点都不动。
• 可续跑：进程崩了也没关系，只要把节点状态和结果落盘，下次启动直接从失败节点继续走。
• 上下文可控：每个Agent只看到它需要的那一小块数据，Prompt长度稳定，模型不容易“迷失”在无关信息里。

这里得先记住一句话：多Agent的价值不是“拆得越多越好”，而是“每个Agent只做最小必要粒度的事，同时链路状态能被完整记录和恢复”。串行只是当前最实用的起步方式，但它已经把一个LLM应用里的可靠性地基打好了。之后哪怕你想演变成并行或动态调度，这个状态管理的思路也一样能复用。

2. 流水线引擎的整体设计与核心抽象

2.1 三个核心角色：上下文、节点、引擎

动手写代码前，先把抽象定下来。我的流水线里只有三个核心角色：

• TaskContext：贯穿整个任务链的上下文对象，记录任务ID、原始输入、每个节点的输出、节点状态和重试次数。
• PipelineNode：一个可执行的Agent节点，只暴露execute(ctx)接口，内部可以是模型调用、规则处理或外部服务请求。
• PipelineEngine：负责编排节点顺序、执行重试、保存和恢复状态。

一开始别急着上复杂框架，把这三个东西定义清楚就够了。用Python的dataclass实现TaskContext非常顺手：

from dataclasses import dataclass, field, asdict from typing import Dict, Any, Optional import json import time import os @dataclass class TaskContext: task_id: str payload: Dict[str, Any] results: Dict[str, Any] = field(default_factory=dict) node_status: Dict[str, str] = field(default_factory=dict) retry_counts: Dict[str, int] = field(default_factory=dict) current_node: str = "" status: str = "pending" created_at: float = field(default_factory=time.time) updated_at: float = field(default_factory=time.time)

字段为什么要这么设计？node_status里存每个节点的当前状态，是为了让引擎在重启后能快速判断“哪个节点还没成功”。retry_counts单独记录重试次数，方便排查时看到底是哪个节点在反复失败，而不是和节点输出混在一起。results存的是节点已完成输出，续跑时成功节点不必重新调用模型，直接从results里取就行。

节点定义更简单，它只做一层很薄的封装：

class PipelineNode: def __init__(self, name: str, agent_func): self.name = name self.agent_func = agent_func def execute(self, ctx: TaskContext): return self.agent_func(ctx)

这里的agent_func就是真正会调用大模型接口的函数。你也可以把它包成一个类方法，内部处理Prompt拼接、输出解析和结果校验。重点是，节点只接收TaskContext，不直接访问数据库或上游服务，所有依赖都通过Context传递，这样引擎才能统一管理生命周期，也才能把重试和续跑做成通用能力。

2.2 为什么先做串行而不是并行或DAG

很多人会问：任务都拆成节点了，为什么不直接并行，或者做成有向无环图DAG？我的回答是：先做串行，因为串行把可靠性问题压缩到了最简单。

并行意味着多个节点同时执行，那你就要处理并发写Context、部分失败回滚、结果合并顺序等等一堆问题；DAG则要设计依赖解析和拓扑排序。对大多数LLM应用来说，任务天然是分步的：先抽取信息，再清洗，再格式化，每个步骤都强依赖前一步的输出。这时候串行不仅够用，还是最容易调试的。

串行还有一个隐藏优势：检查点天然清晰。每个节点跑完，就是一个稳定的中间快照。如果是并行，A节点成功了、B节点失败了，快照里会出现“半边成功”的尴尬状态，续跑时还得判断A的结果到底能不能复用。串行完全没这个问题，整条链的状态要么往前一步，要么停在原地。

串行的短板也很明显：总耗时等于所有节点耗时之和。如果某一步模型响应要30秒，十个节点就是五分钟。我的建议是，先把正确性和可恢复性跑通，再按热点节点做局部并行或批处理，不要一上来就追求架构上的“酷炫”。

2.3 节点之间传什么，不传什么

这是最容易翻车的地方。一开始我喜欢把整个Context JSON塞给下一个Agent，让它“根据上下文自由发挥”，结果模型被大量无关字段干扰，输出质量迅速下降，调用成本也跟着涨。

现在我的原则是：显式传输最小必要数据。每个节点的输入Schema单独定义，只包含上一步输出里它真正需要的字段。比如实体识别节点，只需要文本字段，不需要上一步的情感分数。如果硬塞，模型很可能把情感词也当成实体，或者在判断时被无关字段带偏。这个原则看起来很朴素，但真正影响的是稳定性和成本，越到后期越明显。

同样的道理，节点返回给Context的结果也应该是结构化dict，别让Agent自由发挥输出一整段话。强烈建议为每个节点定义一个输出Schema，执行完立刻做合法性校验。校验不通过就直接抛RetryableError，让重试机制兜底，而不是等到下一个节点解析时才报“字段缺失”。字段缺失这个错，在流水线里会一路传导，最后变成一个让人摸不着头脑的深层异常。

3. 重试机制：让每个节点都具备抗失败能力

3.1 节点状态机的生命周期

每个节点会经历四种状态：pending、running、success、failed。加上因为续跑被跳过的节点，我用skipped表示。具体含义如下：

• pending：还没开始。
• running：正在执行，如果此时进程崩溃，恢复时可以继续处理。
• success：执行成功，结果已经持久化。
• failed：超过最大重试次数，流水线在这里停止。

引擎每次执行节点前，先把状态置为running并落盘；执行成功后再置为success并落盘。这个“先标注再执行”的顺序非常重要，它保证了一个节点至少在语义上不会出现“未开始却已有结果”的矛盾。换句话说，看到running，它可能是执行中，也可能已经崩溃需要继续；看到success，它一定已经有可用结果。

实际代码里，状态更新和结果写入最好放成同一个原子操作。用本地文件存储时，我习惯先写临时文件再rename，避免进程中途崩溃留下半份坏JSON；如果业务量再大一点，可以换成一个支持事务的小型数据库，把整份Context作为一行记录，更新的原子性会更强。状态写入的时间点越早越安全，千万不要等Agent执行完再一次性写，那样崩溃恢复时根本不知道节点跑到哪里。

3.2 重试策略与退避算法

重试不是简单地循环调用。大模型接口最常见的错误有两类：一类是限流、超时，属于可重试错误；另一类是Prompt格式错误、参数非法，属于不可重试错误。可重试错误我单独建立一个异常类型，和普通业务异常区分开：

class RetryableError(Exception): pass class FatalError(Exception): pass

执行节点时，把Agent函数里的异常分类捕获。如果是RetryableError，按指数退避重试；如果是FatalError，直接中断，不再浪费请求。分类这件事看起来小，但能省一大笔成本，不然遇到参数错误还会反复撞墙，白白消耗额度。

重试参数我一般这样设置：max_retries=3，base_delay=2秒，指数退避。也就是第一次失败后等2秒，第二次等4秒，第三次等8秒，超过3次就不再自动继续。具体数值要根据上游服务的限流窗口调整，如果窗口是60秒，退避基数可能要调到10秒甚至更长。另外我还会加一点随机抖动，避免多个任务同时失败后一起重试，结果把限流打得更严重。

下面是带重试的节点执行函数：

def run_node_with_retry(node: PipelineNode, ctx: TaskContext, max_retries: int = 3, base_delay: float = 2.0): for attempt in range(max_retries): try: result = node.execute(ctx) except RetryableError as e: if attempt == max_retries - 1: ctx.node_status[node.name] = "failed" save_context(ctx) raise delay = base_delay * (2 ** attempt) ctx.retry_counts[node.name] = attempt + 1 save_context(ctx) time.sleep(delay) continue except FatalError: ctx.node_status[node.name] = "failed" save_context(ctx) raise ctx.results[node.name] = result ctx.node_status[node.name] = "success" save_context(ctx) return result

注意每次重试前都要save_context，把retry_counts更新。这能让你事后从状态文件里看到“这个节点到底试了几次”，排障时非常有用。保存状态这个动作看起来多点几次，但在续跑场景里，它就是唯一能依赖的“账本”。

3.3 幂等控制：别让重试变成重复执行

重试机制最大的隐性坑是副作用。比如节点里做了“写入一条数据库记录”或“发送一封邮件”，第一次请求其实已经成功，但因为响应超时被判定为失败，重试后又会执行一次，造成重复写库、重复发送。这个问题处理不好，重试机制反而变成数据污染的来源。

解决办法是给每个节点执行加上请求唯一ID，或者把操作设计成幂等的。更通用的做法是，节点内部不要直接执行有副作用的操作，而是先输出一个“动作描述”，由引擎统一去执行，执行前根据动作ID查一下是不是已经执行过。如果你只是想快速规避，也可以把写操作放在流水线的最后一步，并保证最后一步不做自动重试，主流程失败后人工确认再补跑。

我自己在模拟项目里是这样处理的：所有外部写操作都收口到一个叫“输出节点”的Agent里，这个节点不参与自动重试，只做状态登记。虽然牺牲了一点自动化，但至少不会产生脏数据。简单可靠，比花里胡哨的分布式事务方案值钱得多。

4. 续跑机制：从断点恢复，而不是从头再来

4.1 续跑的前提：全量状态落盘

续跑不是魔法，它的前提只有一个：每次状态变化都要持久化。进程可以丢，状态不能丢。所以save_context函数要保存完整的内容，包括任务ID、原始输入、每个节点的输出、节点状态、重试次数，以及当前正在处理的节点。

用本地JSON文件就够起步阶段使用。文件名直接用task_id，更新时整体覆盖，但一定要采用原子写。我第一次实现时直接覆盖原文件，结果有次进程被kill，留下一个写完一半的JSON，恢复逻辑彻底废掉。后来改成先写临时文件再替换，代码是这样的：

def save_context(ctx: TaskContext, path: str = "tasks"): os.makedirs(path, exist_ok=True) tmp_path = f"{path}/{ctx.task_id}.json.tmp" final_path = f"{path}/{ctx.task_id}.json" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(asdict(ctx), f, ensure_ascii=False, indent=2) os.replace(tmp_path, final_path) def load_context(task_id: str, path: str = "tasks") -> TaskContext: with open(f"{path}/{task_id}.json", encoding="utf-8") as f: data = json.load(f) return TaskContext(**data)

os.replace是原子操作，比“直接覆盖原文件”安全得多。这里有个隐含要求：TaskContext里所有字段都要能JSON序列化，payload和results里别放bytes、datetime这类对象，放进Context之前先转成基础类型。否则恢复时你会看到一堆诡异的反序列化异常，而且异常已经脱离节点上下文，很难定位。

4.2 续跑主循环：跳过已成功节点

有了持久化状态，续跑主循环就变得很自然：加载Context，找到第一个状态不是success的节点，从它开始继续执行。前面的成功节点直接跳过，因为它们的输出已经在results里。

def run_pipeline(ctx: TaskContext, nodes: list[PipelineNode]): start_idx = 0 if ctx.current_node: for i, node in enumerate(nodes): if node.name == ctx.current_node: start_idx = i break for node in nodes[start_idx:]: ctx.current_node = node.name if ctx.node_status.get(node.name) == "success": continue ctx.node_status[node.name] = "running" save_context(ctx) run_node_with_retry(node, ctx) ctx.current_node = "" ctx.status = "done" save_context(ctx)

注意current_node的意义。它在失败时指向失败节点，在续跑时告诉引擎“从哪个节点接着来”。当整个任务跑完后，我把它清空并置status为done，表示任务已经收尾。这套逻辑的简单之处在于：你不需要记录“最后一个成功的节点”，只需要记录“当前正在处理的节点”。“当前节点失败”和“当前节点未完成”在启动恢复时会指向同一个位置，语义很干净。

4.3 强制重跑与跳过修复

默认续跑策略是“成功节点不重跑”，但调试时你会有例外需求。比如某个节点虽然返回了success，但业务校验发现结果质量很差，你想单独重跑它；或者你改了某个节点的Prompt，希望该节点和它之后的所有节点重新执行；再比如节点代码本身有Bug，修复后想从该节点重放。这些情况都需要一个显式的强制重跑入口。

所以我都会给引擎加一个mode参数：resume是默认续跑；rerun_from指定从某节点开始强制重跑，该节点及之后的节点忽略已有success状态。实现时只需要在run_pipeline开头把目标节点及之后的node_status和results清掉，再走同样的循环即可。实际实现里要按顺序标记，让节点状态和结果保持同步，避免出现results里少一个字段但状态却是success的脏状态。

续跑能力的价值不只是机器崩了能恢复，它其实给你提供了一套“定点重放”的调试工具。哪个节点不顺眼，就只重跑哪个节点，不需要整条链陪跑。做Agent应用调试，这个能力几乎每天都会用到，尤其是遇到模型输出不稳定的时候。

5. 实操Demo：三段式节点链跑通“重试+续跑”

5.1 定义三个节点

用一个具体例子把前后逻辑串起来。模拟一个文本分析任务，输入一段原始文本，经过三步处理：

• 抽取节点：从文本里抽标题和关键词。
• 清洗节点：把抽取结果里的空字段、格式问题修掉。
• 格式化节点：把结果整理成固定JSON结构。

为了演示不依赖真实模型，我用确定性函数模拟Agent，并且让抽取节点第一次执行时故意抛一个RetryableError，用来展示重试生效；再让格式化节点支持一个修复开关，用来展示续跑。

def extraction_agent(ctx: TaskContext): raw = ctx.payload["raw_text"] if ctx.retry_counts.get("extract", 0) == 0: raise RetryableError("模拟模型限流") return {"title": raw.split("\n")[0], "keywords": ["流水线", "Agent"]} def cleaning_agent(ctx: TaskContext): extracted = ctx.results["extract"] title = extracted["title"].strip() keywords = list(dict.fromkeys([kw.strip() for kw in extracted["keywords"] if kw.strip()])) return {"title": title, "keywords": keywords} def formatting_agent(ctx: TaskContext): if not ctx.payload.get("format_repaired", False): raise FatalError("格式化节点遇到不可重试错误") cleaned = ctx.results["clean"] return {"document": {"title": cleaned["title"], "keywords": cleaned["keywords"]}}

抽取节点故意在第一次失败后重试成功，正好触发重试机制。真实场景下，这里就是“大模型偶发限流”的典型表现，把它单独拎出来演示，比看十页理论都直观。

然后组装节点链：

nodes = [ PipelineNode("extract", extraction_agent), PipelineNode("clean", cleaning_agent), PipelineNode("format", formatting_agent), ]

节点命名要稳定，我在前面说过这个坑，所以这里直接用extract、clean、format这种不会轻易改动的名字。

5.2 模拟一次完整执行

有了节点链，就可以构造TaskContext并执行主流程，代码如下：

ctx = TaskContext( task_id="demo-001", payload={"raw_text": "多Agent流水线实践\n这是一段用于测试的文本。"}, ) run_pipeline(ctx, nodes) print(ctx.results["format"])

正常情况输出：

{"document": {"title": "多Agent流水线实践", "keywords": ["流水线", "Agent"]}}

同时，tasks/demo-001.json里应该能看到三个节点都是success，extract的retry_counts是1。这说明状态记录不只在进程内存里存在，而是已经留给了下一次启动恢复。这时候你可以把这个文件当作一个完整的任务账本，里面记录了每一步谁做过、做成了什么结果、失败了几次。

5.3 模拟崩溃后续跑

再演示续跑。假如格式化节点第一次执行失败了，任务停止在format节点上。这时tasks/demo-001.json里extract和clean都是success，format是failed。我们手动做一次“修复操作”，把payload里的format_repaired字段设为true，然后重新加载Context调用run_pipeline，看看extract和clean会不会被重跑。

ctx = load_context("demo-001") ctx.payload["format_repaired"] = True save_context(ctx) run_pipeline(ctx, nodes) print(ctx.results["format"])

这里的关键是，extract和clean这两个节点的agent_func不会被调用。你可以自己在两个模拟函数里加print验证，运行日志只会出现类似下面这样的内容：

[engine] start from node: format [engine] node format completed

不会有extract和clean的执行记录。这就是续跑的核心体验：前面成功的节点原样保留，只从失败点继续推进，整个过程不需要用户重新提交原始输入，也不需要为已经做成功的步骤再次付费。

这个Demo里没有引入真实模型调用，但核心机制完全通用。你只需要把agent_func换成真正的模型请求函数，并让它按分类抛出我们定义的异常，重试和续跑就会按同样的路径工作。

6. 常见问题与排查技巧实录

6.1 问题速查表

这里整理了一份我实际使用中频率最高的问题对照表，排查时可以直接对着看：

这张表不是我临时编的，是踩完坑以后抄在便签上的。你看一遍基本能知道哪些问题值得提前预防，尤其“结果缓存没复用”这一点特别容易在流水线里被忽略。

6.2 我踩过的几个坑

第一个坑是节点名不稳定。一开始我用“1_抽取”这样的字符串当节点名，后来想调整执行顺序，改成“node_1_extract”，结果续跑时current_node对不上，后面所有节点状态全部作废。从那以后，节点名定义好就不再改动，顺序调整只靠列表顺序，不靠名字。这个约定看起来简单，但能避免一次非常难查的状态错乱问题。

第二个坑是把重试逻辑放在Agent内部。很多Demo里的Agent自己会重试，导致引擎完全不知道失败过，日志里也看不到重试次数。后来我把重试统一收归引擎，让Agent只负责单次执行并抛出分类异常。这样重试次数才可靠，续跑时也才能从真正失败的位置接上。如果重试被隐藏在Agent深处，你会看到整条链几乎不会失败，但成本却莫名其妙高出一大截。

第三个坑是保存Context时序列化失败。结果dict里混进了模型返回的非JSON对象，一执行save_context就崩。我的解决办法是定义一个统一的输出数据结构，Agent返回前先做一次强制转换，保证一切可序列化。不要偷懒把转换丢给引擎，引擎处理时错误信息早就脱离了节点上下文，排查起来非常费劲。

6.3 给新手的配置建议

如果你准备照着这套思路做一个最小可用版本，我建议从这些配置开始：节点数量先控制在3到5个，不要一开始就拆十几个；max_retries先设为2，跑通后再调大；文件存储路径放在临时目录，避免污染正式工程；每个Agent的执行超时时间必须设置，没有超时的LLM调用比显式失败更可怕。

我的经验是，先不要追求全自动，而是把“重试+续跑”这条主链路在人工可控的范围内跑出节奏感，再逐步放开。稳定性是靠监控和异常处理堆出来的，不是靠把代码写得看起来无所不能堆出来的。你先把状态文件打开看几次，真正理解了“每个节点什么时候写状态、什么时候写结果”，这套流水线就算入门了。

最后再分享一个我用着很顺手的小技巧：在每个节点执行前，往Context里塞一条begin_time，执行完后塞一条end_time。别小看这个时间戳，续跑发生时，你把整份Context打出来，就能看到每个节点实际跑了多久、重试间隔多久。定位慢节点或者排查限流问题，这个信息比什么日志都直观。我现在的所有Pipeline节点都默认带上这个字段，算是个人习惯，推荐你也试试。