LangChain源码解析25:with_retry与with_fallbacks如何接住失败

模型调用放进生产之后，最先暴露出来的往往不是“模型不会回答”，而是网络超时、限流、Provider 临时故障、结构化输出解析失败，以及某个工具执行到一半突然报错。

如果所有失败都直接抛给业务层，调用方很快会堆满重复的重试、切模型和兜底代码；如果所有异常都被无脑重试，又可能把一个有副作用的操作执行三遍。

LangChain 没有把可靠性写死在每一个模型或工具里，而是把它们做成两个可组合的 Runnable wrapper：

• with_retry()
  ：同一个 Runnable 失败后，按策略再次尝试；
• with_fallbacks()
  ：当前 Runnable 失败后，按顺序切换到备用 Runnable。

重试解决“这一次可能只是暂时失败”，fallback 解决“这条执行路径可能已经不值得继续”。两者都是策略，不是万能的异常吞咽器。

一、失败不是一种状态，先分清楚再决定动作

在写任何 with_retry() 之前，先给异常分型会更可靠。

这里最容易犯的错误，是把“调用抛异常”直接等同于“应该再次调用”。异常只说明当前尝试没有完成，不说明下一次调用一定安全。

尤其是工具和外部写操作：请求可能已经在服务端成功，只是响应在返回途中丢失。此时 retry 不是恢复，而可能是重复提交。

二、两个方法都返回新 Runnable，不会修改原对象

Runnable 的容错 API 仍然遵循声明式组合的思路：

retriable_model = model.with_retry(
    retry_if_exception_type=(TimeoutError, ConnectionError),
    stop_after_attempt=3,
)

resilient_model = retriable_model.with_fallbacks(
    [backup_model],
    exceptions_to_handle=(TimeoutError, ConnectionError),
)

原始 model 没有被改写。调用方得到的是一层包一层的执行结构：

RunnableWithFallbacks
  -> RunnableRetry
      -> original model

这件事看起来简单，却带来三个重要效果：

1. 同一个模型可以在不同链路上使用不同可靠性策略；
2. 策略可以继续和 prompt、parser、middleware、tracing 组合；
3. 序列化和调试时，容错层仍然是可见的 Runnable，而不是模型对象里一组隐藏开关。

包装顺序本身就是语义

下面两种写法并不等价：

# 先把主模型重试三次，仍失败才切备用模型
route_a = model.with_retry(stop_after_attempt=3).with_fallbacks([backup])

# 主模型失败后先切备用；整个“主模型 + fallback”都失败才重试整条路由
route_b = model.with_fallbacks([backup]).with_retry(stop_after_attempt=3)

route_a 表达的是“主路径要有耐心，备用路径是最后一道门”；route_b 表达的是“每一轮都重新评估整条 Provider 路由”。选择哪一种，取决于失败成本和 Provider 的幂等性。

三、RunnableRetry 的核心是 Tenacity，而不是一个 while 循环

RunnableRetry 继承 RunnableBindingBase。它把原 Runnable 作为 bound，再把 retry 参数作为自己的配置：

RunnableRetry
  - retry_exception_types
  - wait_exponential_jitter
  - exponential_jitter_params
  - max_attempt_number
  - bound

with_retry() 对外暴露的参数名称是：

runnable.with_retry(
    retry_if_exception_type=(ValueError,),
    wait_exponential_jitter=True,
    exponential_jitter_params={
        "initial": 0.5,
        "max": 8.0,
        "exp_base": 2.0,
        "jitter": 1.0,
    },
    stop_after_attempt=3,
)

内部会把这些参数转换成 Tenacity 的：

• retry_if_exception_type(...)
  ：哪些异常进入重试；
• stop_after_attempt(...)
  ：最多几次尝试；
• wait_exponential_jitter(...)
  ：指数退避加随机抖动；
• reraise=True
  ：单次调用耗尽后重新抛出最后一次异常。

stop_after_attempt=3 是三次总尝试

它不是“第一次调用之外再重试三次”，而是最多执行三次：

第 1 次：原始尝试
第 2 次：第一次 retry
第 3 次：第二次 retry
结束：成功，或抛出最后异常

如果把这个数字理解错，延迟和成本都会比预期高一倍。

默认策略不代表适合所有业务

基类默认捕获 Exception，默认开启指数抖动，默认最多三次。但生产配置通常应该缩小异常范围：

model.with_retry(
    retry_if_exception_type=(TimeoutError, ConnectionError),
    stop_after_attempt=2,
    exponential_jitter_params={"initial": 0.2, "max": 3.0},
)

Provider SDK 自己的限流或服务端异常类型，也应该放入明确的白名单，而不是用一个过宽的基类覆盖所有错误。

四、一次 invoke 的重试链路：外层生命周期，内层多次尝试

RunnableRetry.invoke() 不直接写业务函数，而是进入 _call_with_config()，让 retry wrapper 自己拥有一层 callback 生命周期。

可以把它简化成下面的结构：

def invoke(input, config=None):
    return self._call_with_config(self._invoke, input, config)

def _invoke(input, run_manager, config):
    for attempt in Retrying(..., reraise=True):
        with attempt:
            output = bound.invoke(
                input,
                patch_config_for_attempt(config, run_manager, attempt),
            )
        if attempt_succeeded(attempt):
            attempt.set_result(output)
    return output

这里存在两层不同的运行单元：

RunnableRetry root
  -> original runnable, attempt 1
  -> original runnable, attempt 2
  -> original runnable, attempt 3

外层 wrapper 负责记录整条策略的结果，内层 child run 代表每一次真正执行原 Runnable 的尝试。

重试尝试会接上 child callback

_patch_config() 会从当前 run_manager 取得 child callback。第一次尝试不额外加 retry tag，后续尝试会带上类似：

retry:attempt:2
retry:attempt:3

这让 tracing 和事件流能够区分：一次模型调用是慢，还是实际上调用了三次。

重试不是把同一个调用偷偷执行多次，而是把每次尝试放回 Runnable 的生命周期里。

五、为什么 retry scope 应该尽量小

源码文档明确建议：如果一条链由 prompt、model 和 parser 组成，通常只重试最可能暂时失败的模型，而不是把整条链全部包起来。

# 更窄的重试范围
chain = prompt | model.with_retry()

# 更宽的重试范围，可能重复 prompt、工具或解析副作用
chain = (prompt | model | parser).with_retry()

宽范围重试会重复什么

• 已经完成的本地预处理；
• 可能有副作用的工具调用；
• 已经成功但解析失败的模型请求；
• 计费、日志或外部写操作；
• 随机性不同的整条 Agent 子流程。

宽范围不是绝对错误。在你明确希望“整个事务重新执行”，并且每一步都幂等时，它可能正是正确选择。但它应该是一个有意识的事务边界，而不是为了省代码随手包在最外层。

retry 只解决异常，不验证结果质量

如果模型成功返回了一段格式合法但业务上错误的文本，RunnableRetry 不会认为它失败。质量校验需要 parser、validator、middleware 或 Agent 路由显式抛出可处理的异常，retry 才有机会介入。

这也是 retry 与“重新生成直到满意”的区别：前者看异常契约，后者需要业务评价函数。

六、batch 重试的关键：已经成功的输入不会再跑

RunnableRetry._batch() 没有简单地对整个 inputs 列表重复调用。它维护一个 results_map：

第 1 轮：处理 [0, 1, 2, 3]
        0、2 成功，1、3 失败

第 2 轮：只处理 [1, 3]
        1 成功，3 失败

第 3 轮：只处理 [3]

最终再按原始索引组装结果。因此，成功项不会因为别的输入失败而重复请求，且输出顺序仍与输入顺序一致。

这两个性质都很重要：前者减少成本和副作用，后者保证 batch API 的位置语义。

一个容易误读的边界：batch 不是每个元素独立套 Tenacity

源码的批量算法会把未成功的输入组成一个 pending batch，并让底层 batch 返回异常对象。只要这一轮出现异常，就取其中第一个异常驱动外层 retry 判断。

因此，混合异常类型时不能简单理解为：

输入 0 按 ValueError 的策略重试
输入 1 按 RuntimeError 的策略不重试

真实行为更接近：

pending batch 中的第一个异常
  -> 决定这一轮是否进入下一次 retry
进入 retry 后，所有仍失败的 pending 输入会再次提交

如果业务需要对不同错误做完全独立的重试预算，应该先按错误类型拆分批次，或者在更上层自己建立逐项策略，不能把 RunnableRetry.batch() 当成 N 个互不相关的 retry machine。

return_exceptions 只改变交付方式

return_exceptions=False 时，最终未解决的异常会按普通调用抛出；True 时，结果列表中保留异常对象。

它不会让 retry 变成“忽略错误”，只是决定错误是抛出还是作为 batch 结果交给调用方继续处理。

七、RunnableRetry 为什么不重试 stream 和 transform

retry.py 里有一条非常直接的注释：

stream() and transform() are not retried because retrying a stream
is not very intuitive.

原因不是技术上做不到，而是流式调用已经把部分结果交给了消费者。

假设模型已经输出：

“根据目前资料，结论是……”

此时网络断开，再从头 retry 并把另一份结果接到同一条流后面，会造成重复前缀、顺序错乱和无法判断的最终文本。

所以 with_retry() 主要覆盖 invoke、ainvoke、batch 和 abatch；直接使用 stream 时，不能把它当成 token 级自动重试。

如果产品确实需要流式断线恢复，就要另设计协议：保存请求 ID、已消费位置、可重放 token 或完整响应，而不是简单复用普通 retry wrapper。

八、RunnableWithFallbacks 是一条有顺序的候选链

with_fallbacks([a, b, c]) 在逻辑上展开成：

primary -> a -> b -> c

它为整条 fallback wrapper 创建一个 root run，然后把每个候选 Runnable 作为 child 执行。候选成功就结束；候选失败且异常类型在 exceptions_to_handle 中，就继续下一个。

model = primary.with_fallbacks(
    [secondary, local_model],
    exceptions_to_handle=(TimeoutError, ConnectionError),
)

这不是随机负载均衡，也不是并发竞速。顺序本身就是优先级：主模型通常拥有最好的能力，后面的模型承担可用性或成本兜底。

未声明处理的异常会立即穿透

默认处理的是 Exception，但可以把范围缩小。任何不属于 exceptions_to_handle 的异常会直接向上抛出，不会继续尝试 fallback。

例如把 schema 编程错误误配置成只处理网络异常，那么它会在主模型处立即失败；这是好事，因为切换 Provider 并不能修复错误的输入契约。

异步取消也不应被普通 fallback 吞掉。取消信号属于控制流的一部分，应该尽快传播，而不是让备用模型继续消耗请求。

九、exception_key：把上一个错误交给备用路径

默认情况下，fallback 只知道“上一个 Runnable 失败了”，不知道具体异常。设置 exception_key 后，处理过的异常会被写入输入字典，再交给下一个 Runnable：

def explain_failure(inputs: dict[str, Any]) -> str:
    error = inputs["error"]
    return f"fallback because: {error}"

safe = primary.with_fallbacks(
    [RunnableLambda(explain_failure)],
    exception_key="error",
)

它有两个硬约束：

1. 主 Runnable 和所有 fallback 都必须接受字典输入；
2. wrapper 会把异常写入输入字典，调用方如果复用原字典，需要注意这个可见的输入变化。

first error 与 last error 分工不同

单次 invoke 中，wrapper 会同时保存：

• first_error
  ：整条候选链最先发生的错误；
• last_error
  ：交给下一候选的最新错误。

如果所有候选都失败，最终抛出的重点是第一处被处理的错误；而 exception_key 传给下一个候选的是上一候选的最新错误。

这两者并不矛盾：对外保留最初根因，有助于定位主路径为什么失效；对内把最近一次失败交给备用路径，能让它知道刚刚尝试过什么。

十、fallback 的 batch 逻辑是逐输入推进的

和 retry 的 batch 不同，RunnableWithFallbacks.batch() 会为每个输入维护自己的状态：

输入 0：primary 成功 -> 完成
输入 1：primary 失败 -> fallback A 成功 -> 完成
输入 2：primary 失败 -> fallback A 失败 -> fallback B 成功

每轮只把仍未完成的输入交给下一个候选，并把已经成功的结果固定下来。这样同一个 batch 中可以同时存在不同的 fallback 层级。

它还会维护原始索引，最后按输入顺序返回。abatch() 采用同样的 per-input 状态，只是候选批次调用走异步路径。

batch 中的非处理异常

如果某个输入返回的异常不属于 exceptions_to_handle：

• return_exceptions=False
  ：按普通 batch 语义抛出；
• return_exceptions=True
  ：该输入从 fallback 候选队列移出，并把异常保留在结果中。

这保证了“可切换的 Provider 故障”和“不能由换 Provider 修复的编程错误”不会混在一起。

十一、stream fallback 有一个不可逆的首块边界

RunnableWithFallbacks.stream() 的实现很有代表性。它不会一上来就把主流全部消费完，而是先尝试从候选流中取出第一块：

stream = runnable.stream(input)
chunk = next(stream)

如果在第一块之前就失败，才会切换到下一个 fallback；一旦第一块已经成功取出并交给调用方，候选就已经“提交”了，后续流中途失败会直接向上抛出，不再切换。

尚未产生 chunk
  -> 主流失败 -> 可以 fallback

已经产生 chunk
  -> 后续失败 -> 不能无缝切换

这条边界保护了输出的连续性。否则消费者可能先看到主模型的半句话，再接上备用模型从头开始的另一句话。

为什么需要先拿一块再交付

如果 wrapper 先把整个主流缓存起来，等确认主流成功后再输出，就失去了 streaming 的低延迟价值；如果第一块都没确认就同时启动 fallback，又会造成重复请求。

“取首块作为提交点”是在延迟、重复请求和输出一致性之间做出的折中。

十二、模型 fallback 如何保持 tools 与结构化输出一致

RunnableWithFallbacks 不只用于普通字符串 Runnable，也常用于多个聊天模型之间的切换。这里有一个容易忽略的实现：当你对 fallback wrapper 调用一个“会返回 Runnable 的方法”时，wrapper 会把这个方法同步应用到主 Runnable 和所有 fallback。

model = primary.with_fallbacks([backup])
model_with_tools = model.bind_tools(tools)

内部效果相当于：

primary.bind_tools(tools).with_fallbacks(
    [backup.bind_tools(tools)]
)

这样备用模型不会因为忘记绑定工具而在运行到 fallback 时突然收到不同输入。

这个行为依赖 _returns_runnable() 对方法返回类型的判断。像 bind_tools()、with_structured_output() 这类返回 Runnable 的声明式方法会被传播；普通属性读取则只从主 Runnable 读取。

因此：

• model.model_name
   代表主模型的属性；
• model.bind_tools(tools)
   会同步改造候选链；
• 如果 fallback 的方法签名或能力不兼容，问题应在构造阶段尽早暴露。

“能切换”不等于“语义天然一致”

备用 Provider 仍然必须满足：

• 接受相同的输入类型；
• 输出同一类结果或下游可接受的类型；
• 工具 schema、tool call 和结构化输出能被同一套 Agent 消费；
• 流式 chunk 的聚合语义不会破坏上层。

如果两个模型返回的结构不同，应该先加 parser/adapter 把它们收束，而不是希望 fallback wrapper 自动完成类型转换。

十三、wrapper 的组合顺序决定故障树

把常见组合列出来，会更容易看清语义差异：

可以用一棵故障树来思考：

请求
  -> primary attempt 1
      -> primary attempt 2
          -> backup attempt 1
              -> hard failure

每多包一层，可能的请求次数、延迟上限和 tracing 节点都会增加。可靠性配置不能只看“成功率”，还要算最坏路径的成本。

十四、重试和 fallback 不会自动解决幂等性

这是最需要在生产代码 review 里单独拎出来的一点。

对只读模型调用

重复请求通常主要增加成本和延迟，风险相对可控，但仍要考虑随机输出、上下文窗口和 rate limit。

对工具调用

工具可能写数据库、发送邮件、创建订单或修改文件。一次超时后再次调用，不能假设第一次一定没有发生。

更稳妥的做法包括：

1. 为外部写操作设计幂等键；
2. 在工具层记录 request ID 和提交状态；
3. 只对明确可重试的异常开启 retry；
4. 对不确定状态转人工确认，而不是盲目 fallback；
5. 把 retry scope 放在真正可重试的网络边界，而不是整个 Agent loop。

对结构化输出

模型调用已经成功但 parser 失败时，重新调用可能得到不同结果。是否重试应该由 parser 错误类型和业务容忍度决定，不要把所有 ValueError 都视为 Provider 临时故障。

十五、用 callback 和 tracing 看清“到底重了几次、切了几家”

容错如果不可见，就很难调优。

RunnableRetry 的每次 attempt 会建立 child callback；RunnableWithFallbacks 的每个候选也会建立 child run。结合前面 tracing 章节的父子关系，可以还原：

resilient root
  -> primary attempt 1    [成功 / 失败]
  -> primary attempt 2    [成功 / 失败]
  -> fallback provider    [成功 / 失败]

生产至少应该观察：

• 单次请求的 retry 次数；
• fallback 触发率和每家 Provider 的成功率；
• retry 等待时间与总延迟；
• 哪种异常最常触发切换；
• batch 中有多少输入进入第二轮；
• stream 在首块前失败还是首块后失败；
• fallback 后的输出是否需要人工校验。

如果只统计最终成功率，会把“经常重试后才成功”和“第一次就成功”混成同一个指标，无法发现成本和延迟已经恶化。

十六、可靠性策略的一份落地清单

在给 Runnable 加容错之前，可以按下面顺序检查：

1. 这类异常是暂时性、Provider 特有，还是输入/业务错误？
2. retry 的异常类型是否足够窄，是否误包含取消和编程错误？
3. stop_after_attempt
    是否包含第一次尝试，最坏延迟是多少？
4. 指数退避和 jitter 是否适合当前流量与限流窗口？
5. retry scope 是否只包住真正可能暂时失败的 Runnable？
6. 被重试的操作是否幂等，是否会重复写入或扣费？
7. batch 是否允许 pending 输入重复提交，混合异常是否会影响 retry 判断？
8. stream 是否需要断线恢复协议，而不是普通 retry？
9. fallback 的输入输出、工具和 structured output 是否兼容？
10. 是否需要 exception_key 把原因交给备用逻辑？
11. fallback 顺序和 wrapper 顺序是否符合故障树预期？
12. callback/tracing 是否能看出每次 attempt 和候选切换？
13. return_exceptions
     的选择是否和调用方的批量处理方式一致？

十七、真正成熟的容错，是让失败变得可预测

with_retry() 和 with_fallbacks() 的价值，不是让系统永远不报错，而是把失败从一段散落在业务代码里的 try/except，提升为可组合、可观察、可测试的执行策略。

它们分别守住不同边界：

RunnableRetry
  同一条路径再次尝试

RunnableWithFallbacks
  换一条已声明的路径

batch 用索引和 pending 集合控制重复执行，stream 用首块提交点保护输出连续性，exception filter 决定哪些错误有资格改变路径，callback 则把每次尝试重新接回统一生命周期。

最终要记住的不是某个默认参数，而是这条判断：

可靠性不是“失败后做什么”的孤立代码，而是输入契约、执行范围、异常分类、幂等性和观测结构共同组成的协议。

系列链接

第 1 篇：LangChain源码解析01：先看懂Agent工程骨架

第 2 篇：LangChain源码解析02：Runnable把一切串起来

第 3 篇：LangChain源码解析03：RunnableConfig如何追踪到底

第 4 篇：LangChain源码解析04：Message不只是字符串

第 5 篇：LangChain源码解析05：Tool如何从函数变成契约

第 6 篇：LangChain源码解析06：Prompt和Parser守住两端

第 7 篇：LangChain源码解析07：BaseChatModel如何统一模型调用

第 8 篇：LangChain源码解析08：init_chat_model如何动态切换模型

第 9 篇：LangChain源码解析09：create_agent如何编译Agent运行图

第 10 篇：LangChain源码解析10：Agent条件边如何决定下一步

第 11 篇：LangChain源码解析11：middleware如何接管Agent执行链

第 12 篇：LangChain源码解析12：结构化输出如何在两种策略间切换

第 13 篇：LangChain源码解析13：ToolRuntime如何把上下文注入工具

第 14 篇：LangChain源码解析14：stream_events如何汇合Agent多路事件

第 15 篇：LangChain源码解析15：Agent为什么最终变成StateGraph

第 16 篇：LangChain源码解析16：AgentExecutor如何驱动经典Agent循环

第 17 篇：LangChain源码解析17：ChatOpenAI如何统一两套API

第 18 篇：LangChain源码解析18：ChatAnthropic如何编排思考与工具

第 19 篇：LangChain源码解析19：一套测试如何验收所有模型

第 20 篇：LangChain源码解析20：长文档如何切成可检索的块

第 21 篇：LangChain源码解析21：模型能力如何驱动运行时决策

第 22 篇：LangChain源码解析22：一次Agent调用如何变成运行树

第 23 篇：LangChain源码解析23：新Provider如何通过框架验收

第 24 篇：LangChain源码解析24：LangChain到底设计对了什么

源码参考： GitHub: https://github.com/langchain-ai/langchain

当重试、fallback、限流和流式输出同时存在时，系统应该由哪一层决定下一次请求还能不能发出去？