LCEL 容错机制

要点

• 链路能跑起来不等于上线以后就稳，真实项目里需要给关键节点加容错层。
• LCEL 里最常用的两个容错工具是 withRetry() 和 withFallbacks()。
• withRetry() 处理临时性抖动，如超时、限流、偶发 500。
• withFallbacks() 处理这条路本身不可靠的场景，如主模型不可用、输出格式始终不稳定。
• 两个工具一起用时，顺序通常是：先 retry，再 fallback。
• Parser 失败也可以单独兜底，让链在格式偶发不稳定时还能继续走下去。

1. 背景：数据能流起来，还不够

前两篇讲了两件事：同一份输入可以并行处理，不同输入可以走不同分支。但链路能跑起来，不等于上线以后就稳。

在真实项目里，最容易被忽略的问题是：某一段链突然失败了怎么办。对直接面向用户的 Agent 产品来说，失败通常不止一种：

• 模型请求超时。
• provider 限流。
• 服务偶发抖动。
• 输出格式不对，parser 解析失败。
• 某个模型临时不可用。

如果整条链没有容错层，任何一段失败都会直接把异常抛给用户。这一篇要讲的，就是 LCEL 里最常用的两个容错工具：withRetry() 和 withFallbacks()。这次不只讲普通链，而是把它接回 Agent 场景里看。

2. withRetry()：处理临时性错误

withRetry() 适合处理这样一类问题：再试一次，可能就恢复了。比如：

• 网络超时。
• 429 限流。
• provider 偶发 500。

最简单的用法是直接加在某个 Runnable 上：

const safeModel = model.withRetry({
  stopAfterAttempt: 3,
});

stopAfterAttempt: 3 表示总共尝试 3 次。第一次算 1 次，后面再补 2 次重试。如果这 3 次都失败，错误才会真正抛出来。

在实际项目里，最常加 retry 的位置通常不是整条链，而是模型这一层。因为最容易抖动的，往往就是外部模型调用。

3. withFallbacks()：处理这条路本身不通

有些错误不是「再试一次可能就好」，而是这条路本身暂时不适合继续走。比如：

• 当前模型临时不可用。
• 某个 provider 正在抖。
• 输出格式始终不稳定。

这时候更需要的不是 retry，而是切到备用链。

const primaryChain = prompt.pipe(primaryModel).pipe(parser);
const backupChain = prompt.pipe(backupModel).pipe(parser);

const safeChain = primaryChain.withFallbacks([backupChain]);

执行顺序很直接：

1. 先跑主链。
2. 主链失败，再跑备用链。
3. 备用链也失败，才真正抛错。

如果有多个备用方案，也可以继续往后排：

const safeChain = primaryChain.withFallbacks([backupChain, finalFallbackChain]);

4. 顺序：先 retry，再 fallback

这两个工具经常一起用，但顺序不能乱。更符合预期的写法通常是：

const safeModel = primaryModel
  .withRetry({ stopAfterAttempt: 2 })
  .withFallbacks([backupModel]);

它表达的是：主模型先重试两次，还是失败，再切备用模型。如果顺序反过来：

const wrong = primaryModel
  .withFallbacks([backupModel])
  .withRetry({ stopAfterAttempt: 2 });

读起来就不是这个意思了。所以大多数时候，都优先记成一句话：先 retry，再 fallback。

5. Parser 失败，也可以单独兜底

还有一类失败，不是模型没回，而是模型回了，但格式不对。比如明明要求它只输出 JSON，它却前面多说了一句解释。这时候模型调用本身是成功的，但 JsonOutputParser 会失败。

最简单的办法有两种。

给 parser 所在链加 retry

const chain = prompt
  .pipe(model)
  .pipe(new JsonOutputParser())
  .withRetry({ stopAfterAttempt: 2 });

给 parser 准备一个更宽松的 fallback

import { AIMessage } from "@langchain/core/messages";
import { JsonOutputParser } from "@langchain/core/output_parsers";
import { RunnableLambda } from "@langchain/core/runnables";

const looseJsonParser = RunnableLambda.from((message: AIMessage | string) => {
  const text =
    typeof message === "string"
      ? message
      : typeof message.content === "string"
        ? message.content
        : message.content
            .map((item) =>
              typeof item === "string" ? item : JSON.stringify(item),
            )
            .join("\n");

  const match = text.match(/\{[\s\S]*\}/);
  if (match) {
    return JSON.parse(match[0]);
  }
  return { type: "unknown", confidence: 0 };
});

const safeParser = new JsonOutputParser().withFallbacks([looseJsonParser]);

这种做法不是为了让结果更漂亮，而是为了让链在格式偶发不稳定时还能继续走下去。

6. 把它接回 Agent：三层容错

这篇真正关键的不是 retry 和 fallback 本身，而是它们在 Agent 链路里该放在哪。更常见的做法是分三层：

1. 前置链失败，要不要给默认值。
2. 模型失败，要不要切备用模型。
3. 整条回复链都失败了，要不要给用户一条保底回复。

import { createAgent } from "langchain";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import {
  JsonOutputParser,
  StringOutputParser,
} from "@langchain/core/output_parsers";
import { RunnableLambda, RunnablePassthrough } from "@langchain/core/runnables";
import { ChatOpenAI } from "@langchain/openai";

const primaryModel = new ChatOpenAI({
  model: "deepseek-chat",
  apiKey: process.env.MODEL_API_KEY,
  configuration: {
    baseURL: process.env.MODEL_BASE_URL ?? "https://api.deepseek.com/v1",
  },
});

const backupModel = new ChatOpenAI({
  model: process.env.MODEL_FALLBACK_NAME ?? "deepseek-chat",
  apiKey: process.env.MODEL_API_KEY,
  configuration: {
    baseURL: process.env.MODEL_BASE_URL ?? "https://api.deepseek.com/v1",
  },
});

// 模型层容错：主模型先 retry，再切到备用模型。
const safeModel = primaryModel
  .withRetry({ stopAfterAttempt: 2 })
  .withFallbacks([backupModel.withRetry({ stopAfterAttempt: 2 })]);

// 前置分析链：负责给 Agent 补一个 type 字段。
const typeChain = ChatPromptTemplate.fromMessages([
  [
    "system",
    '分析用户消息类型，只返回 JSON：{"type":"bug|feature|doc", "confidence":0}',
  ],
  ["user", "{input}"],
])
  .pipe(safeModel)
  .pipe(new JsonOutputParser());

// 如果类型分析失败，就退到一个最小默认值，别让整条链直接中断。
const safeTypeChain = typeChain.withFallbacks([
  RunnableLambda.from(() => ({
    type: "unknown",
    confidence: 0,
  })),
]);

// 这条前置链负责保留原始输入，并补 type。
const preProcess = RunnablePassthrough.assign({
  type: safeTypeChain,
});

const agent = createAgent({
  model: safeModel,
  tools: [],
  systemPrompt: [
    "你是一个技术任务处理助手。",
    "如果 type=bug，优先给出排查步骤和临时止血方案。",
    "如果 type=unknown，就正常分析问题，不要编造分类结果。",
  ].join("\n"),
});

// 整条链最后的兜底：如果 Agent 回复这一步都失败了，至少返回固定文案。
const finalFallback = RunnableLambda.from(
  () =>
    "服务当前不太稳定，请稍后再试。你刚才的问题我已经收到，恢复后会继续处理。",
);

async function runAgent(input: string) {
  // 第 1 层：先跑前置链，尽量把分析结果补齐。
  const preProcessed = await preProcess.invoke({ input });

  // 第 2 层：再把整理后的消息交给 Agent。
  const replyChain = RunnableLambda.from(async () => {
    const result = await agent.invoke({
      messages: [
        {
          role: "user",
          content: [
            `input=${preProcessed.input}`,
            `type=${JSON.stringify(preProcessed.type)}`,
          ].join("\n"),
        },
      ],
    });
    return result.messages.at(-1)?.text ?? "";
  });

  // 第 3 层：如果回复链也失败，就落到最后一层固定文案。
  return replyChain.withFallbacks([finalFallback]).invoke({});
}

const reply = await runAgent(
  "支付接口偶尔返回 500，订单状态显示为未知，用户重复提交了三次。",
);
console.log(reply);

这段代码里，容错层是分开的：

• safeModel 处理模型抖动。
• safeTypeChain 处理前置分析失败。
• finalFallback 处理最后整条回复链失败。

这样写的好处是：不是所有失败都往同一层堆，而是每一层只兜自己那一类问题。

7. 一个简单的判断标准

如果你在想某个失败该放哪一层处理，可以直接按这个顺序判断：

这次失败是不是偶发抖动？

• 是，用 withRetry()。

这条路是不是暂时不可靠？

• 是，用 withFallbacks()。

这一层失败以后，后面的链还能不能继续？

• 能继续，就给默认值兜住。
• 不能继续，就给用户一个最终保底回复。

对直接面向用户的回复链来说，最后一层纯函数兜底通常都值得准备。它不一定聪明，但至少稳定。