LangChain：故障处理和降级策略

LangChain 中的 RunnableWithFallbacks 是一个用于错误处理和降级策略的核心组件，它允许为Runnable任务设置备选方案，确保在主要组件失败时能够优雅切换到备用方案。

概述

RunnableWithFallbacks 是 LangChain 中用于为可运行对象设置“备用执行逻辑”，增强应用鲁棒性的核心组件。它允许为一个主要的 Runnable（如 LLM、链或工具）配置一个或多个备用的 Runnable。当主要的 Runnable 执行失败时，系统会自动按顺序尝试备用的 Runnable，直到成功或所有选项都失败为止。

这种机制对于构建高可用性的 LLM 应用至关重要，可以有效应对以下常见问题：

从类层次来看，RunnableWithFallbacks 本身也是 Runnable 的子类，这意味着它实现了所有标准执行方法，可以像普通 Runnable 一样被调用、组合和链式使用。

它主要解决以下实际问题：

• API 层面：模型 API 超时、请求频率超限（Rate Limit）。
• 功能层面：工具调用失败、模型不支持特定输出格式、模型无法处理特定任务（如超长上下文）。
• 环境层面：网络波动、第三方服务临时下线。

与重试机制的区别：重试是对同一模型重复尝试，而 Fallbacks 是切换到不同的模型或链。

核心方法

RunnableWithFallbacks 利用Runnable基类提供的 with_fallbacks方法，来构建一个带有回退能力的RunnableWithFallbacks对象 ，这是为 Runnable 添加回退策略的推荐方式。

with_fallbacks方法

def with_fallbacks(
    self,
    fallbacks: Sequence[Runnable[Input, Output]],
    *,
    exceptions_to_handle: Optional[Tuple[Type[Exception], ...]] = None,
    exception_key: Optional[str] = None
) -> RunnableWithFallbacks[Input, Output]

参数说明：

• fallbacks：备选 Runnable 对象列表，按顺序依次尝试
• exceptions_to_handle：(可选) 触发回退的异常类型元组，用于指定哪些异常会触发回退。默认为 None，表示捕获所有异常。
• exception_key: (可选) 一个字符串键。如果设置，主 Runnable 产生的异常信息会作为输入的一部分传递给备用 Runnable，方便调试或日志记录。

执行流程：

1. 优先调用主 Runnable 的 invoke()（或 stream()、batch() 等）
2. 若抛出匹配 exceptions_to_handle 的异常，则按顺序尝试备选列表中的每个 Runnable
3. 返回第一个成功的结果，或抛出最后一个失败的异常

重要提示：许多 LLM 包装器默认会捕获错误并重试。在使用回退时，通常需要关闭内置重试，否则第一个包装器会持续重试而不会失败。

典型示例

模型回退

以下示例展示如何在 LLM 调用中配置主备模型回退策略：

# 安装依赖（若未安装）
%pip install --upgrade --quiet langchain-openai langchain-anthropic

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser

# 1. 定义主模型和备用模型
# 主模型：GPT-4，性能强但可能更贵或不稳定
primary_model = ChatOpenAI(model="gpt-4", request_timeout=10) # 设置较短超时以快速失败

# 备用模型1：GPT-3.5-Turbo，成本低、速度快
fallback_model_1 = ChatOpenAI(model="gpt-3.5-turbo")

# 备用模型2：Anthropic的Claude，作为不同供应商的备选
fallback_model_2 = ChatAnthropic(model="claude-3-haiku-20240307")

# 2. 使用 with_fallbacks 方法配置回退链
# 回退顺序为: primary_model -> fallback_model_1 -> fallback_model_2
model_with_fallbacks = primary_model.with_fallbacks([fallback_model_1, fallback_model_2])

# 3. 调用模型
# 如果 GPT-4 调用失败，会自动尝试 GPT-3.5-Turbo，如果还失败，则尝试 Claude
response = model_with_fallbacks.invoke("用一句话解释什么是 LangChain。")
print(response.content)

关键逻辑：

• 只需调用一次 model_with_fallback.invoke(...)，LangChain 会自动捕获主模型错误并切换到备用模型
• 若主模型（GPT-4）调用失败（如 API 限流、超时），系统会自动触发备用模型（Claude-instant）

链级回退

回退机制不仅限于单个模型，还可以应用于整个链（Chain）。这允许主链和备用链使用不同的提示词模板、模型甚至输出解析器。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 1. 定义主链
primary_prompt = ChatPromptTemplate.from_template("详细解释 {topic} 的核心概念。")
primary_llm = ChatOpenAI(model="gpt-4")
primary_chain = primary_prompt | primary_llm | StrOutputParser()

# 2. 定义备用链
# 备用链可以使用更简单的提示词和更便宜的模型
fallback_prompt = ChatPromptTemplate.from_template("简要描述 {topic}。")
fallback_llm = ChatOpenAI(model="gpt-3.5-turbo")
fallback_chain = fallback_prompt | fallback_llm | StrOutputParser()

# 3. 配置链级回退
chain_with_fallback = primary_chain.with_fallbacks([fallback_chain])

# 4. 调用链
# 如果主链（GPT-4 + 详细提示）失败，会自动切换到备用链（GPT-3.5 + 简要提示）
result = chain_with_fallback.invoke({"topic": "量子计算"})
print(result)

多级回退

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

# 主模型
primary_llm = init_chat_model("qwen-plus", model_provider="dashscope")

# 多个备用模型
backup_llm1 = init_chat_model("qwen-turbo", model_provider="dashscope")
backup_llm2 = init_chat_model("deepseek-chat", model_provider="deepseek")

# 自定义回退处理函数
def custom_fallback(input_data):
    return "抱歉，当前服务暂时不可用，请稍后重试。"

# 创建多级回退链
prompt = ChatPromptTemplate.from_template("回答：{question}")
robust_chain = (
    prompt 
    | primary_llm.with_fallbacks([
        backup_llm1, 
        backup_llm2, 
        RunnableLambda(custom_fallback)
    ]) 
    | StrOutputParser()
)

result = robust_chain.invoke({"question": "什么是深度学习?"})
print(result)

模拟失败

from langchain_core.runnables import RunnableWithFallback, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 模拟一个可能失败的 LLM（50% 概率失败）
class FailingChatModel(ChatOpenAI):
    def invoke(self, input, config=None):
        import random
        if random.random() < 0.5:
            raise ValueError("模拟 LLM API 调用失败！")
        return super().invoke(input, config)

# 主 LLM 链
main_chain = (
    ChatPromptTemplate.from_template("请详细解释 {topic} 的核心概念。")
    | FailingChatModel(temperature=0.7)   # 使用可能失败的 LLM
    | StrOutputParser()
)

# 兜底的 LLM 链（使用更稳定的模型）
fallback_chain = (
    ChatPromptTemplate.from_template("简要描述 {topic}。")
    | ChatOpenAI(model="gpt-3.5-turbo", temperature=0.5)
    | StrOutputParser()
)

# 组合成带回退的链
chain_with_fallback = main_chain.with_fallbacks([fallback_chain])

# 调用
result = chain_with_fallback.invoke({"topic": "LangChain"})
print(result)

此示例中，若主链因主模型随机失败而出错，系统会自动切换至备用链执行。

输出解析失败的回退

一个巧妙的用法是处理输出格式不正确的情况。例如，先用一个较快的模型尝试生成特定格式（如 JSON）的输出，如果解析失败，再回退到能力更强的模型。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

# 定义期望的输出结构
class AnswerWithSources(BaseModel):
    answer: str = Field(..., description="对问题的回答")
    sources: list[str] = Field(..., description="回答的来源列表")

# 1. 定义主链：先用 GPT-3.5 尝试
prompt = ChatPromptTemplate.from_template("根据以下信息回答问题：{question}。信息：{context}")
fast_llm_chain = prompt | ChatOpenAI(model="gpt-3.5-turbo") | JsonOutputParser(pydantic_object=AnswerWithSources)

# 2. 定义备用链：如果 GPT-3.5 输出格式不对，则用 GPT-4
powerful_llm_chain = prompt | ChatOpenAI(model="gpt-4") | JsonOutputParser(pydantic_object=AnswerWithSources)

# 3. 配置回退
# 如果 fast_llm_chain 因解析失败而抛出异常，则自动切换到 powerful_llm_chain
robust_chain = fast_llm_chain.with_fallbacks([powerful_llm_chain])

# 4. 调用
context_info = "LangChain 是一个用于开发由语言模型驱动的应用程序的框架。"
question_info = "LangChain 是什么？"
response = robust_chain.invoke({"question": question_info, "context": context_info})
print(response) # 输出一个符合 AnswerWithSources 结构的字典

Agent模型回退

从 LangChain v1.2 开始，新增了 ModelFallbackMiddleware，可在 Agent 整体模型调用的中间件层回退，与with_fallbacks是不同层级的机制，可以按需选择或结合使用。

模型回退示例

from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# 主模型：OpenAI GPT-4o
primary_model = ChatOpenAI(model="gpt-4o")

# 备选模型（按顺序依次尝试）
fallback_models = [
    ChatOpenAI(model="gpt-4o-mini"),           # 第一备选：更轻量的 OpenAI 模型
    ChatAnthropic(model="claude-3-sonnet-20240229"),  # 第二备选：Anthropic Claude
]

# 创建回退中间件（注意：只传入备选模型，主模型单独指定）
fallback_middleware = ModelFallbackMiddleware(
    fallback_models=fallback_models
)

# 创建带回退的 Agent
agent = create_agent(
    model=primary_model,           # 主模型
    middleware=[fallback_middleware],
    tools=[],                      # 工具列表，此处为空
)

# 调用 Agent（若 gpt-4o 失败，自动依次尝试 gpt-4o-mini，再尝试 claude）
result = agent.invoke({
    "messages": [{"role": "user", "content": "Hello"}]
})
print(result)

此中间件极大简化了 Agent 场景下的回退配置，推荐在构建生产级 Agent 时使用。

关键点说明

1. 中间件导入：ModelFallbackMiddleware 位于 langchain.agents.middleware 中，这是 LangChain 1.0 引入的生产级中间件机制。
2. 备选顺序：当主模型调用失败时，会自动按 fallback_models 列表顺序依次尝试，直到有一个模型成功或全部失败。
3. 适用场景：模型回退中间件特别适用于构建高可用智能体、跨供应商冗余（同时使用 OpenAI 和 Anthropic）、以及成本优化（主模型用高性能模型，失败时降级到低成本模型）等场景。

与with_retry的区别

理解 with_fallbacks 和 with_retry 的区别很重要：

• with_retry: 在同一个 Runnable 上重复尝试。适用于处理临时性网络错误或服务短暂不可用的情况。
• with_fallbacks: 切换到不同的 Runnable。适用于主方案完全不可用，需要一个完全不同的备选方案的情况。

在实际应用中，两者可以结合使用，例如先对主模型进行几次重试，如果仍然失败，则回退到备用模型。将 with_retry（重试机制）和 with_fallbacks（回退机制）结合使用，是构建生产级高可用 LLM 应用的最佳实践。

关键参数详解

组件	方法	关键参数	作用	建议配置
重试	with_retry()	stop_after_attempt	最大重试次数。	3次。太多会导致用户等待时间过长。
		wait_exponential_jitter	等待策略（指数退避+随机抖动）。	开启。防止网络拥塞时所有请求同时再次撞墙。
回退	with_fallbacks()	fallbacks	备选方案列表。	至少配置一个轻量级模型（如 GPT-3.5 或 Claude Haiku）。
		exceptions_to_handle	指定触发回退的异常类型。	默认为所有异常。生产环境建议指定如 APIError, Timeout。

这种组合通常遵循 “先重试，后降级” 的策略：

1. 第一道防线 (with_retry)：应对临时性故障（如网络抖动、瞬间超时）。如果失败，不要马上放弃，而是等待片刻后重试几次。
2. 第二道防线 (with_fallbacks)：应对持久性故障（如 API 彻底宕机、速率限制）。如果重试多次仍然失败，说明主服务不可用，此时自动切换到备用服务（如更便宜的模型或不同供应商的模型）。

核心逻辑图解

用户请求 >主模型 (GPT-4)

1. 调用失败

2. 重试机制 (等待 1s -> 重试 -> 失败 -> 等待 2s -> 重试…)

3. 重试次数耗尽，依然失败

4. 回退机制 > 备用模型 (GPT-3.5) > 返回结果

重试+回退示例

这个示例模拟了一个“智能问答”场景。我们将对主模型配置重试策略，并将其包装在回退链中。

import time
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

# 1. 定义模型
# 主模型：GPT-4 (假设它有时不稳定或响应慢)
primary_model = ChatOpenAI(model="gpt-4", request_timeout=2) 

# 备用模型：GPT-3.5-Turbo (更稳定、更便宜)
fallback_model = ChatOpenAI(model="gpt-3.5-turbo")

# 2. 构建基础链
prompt = ChatPromptTemplate.from_template("请简要解释：{question}")
output_parser = StrOutputParser()

# 主链：Prompt -> 主模型 -> 解析器
primary_chain = prompt | primary_model | output_parser

# 备用链：Prompt -> 备用模型 -> 解析器
fallback_chain = prompt | fallback_model | output_parser

# 3. 定义重试逻辑 (第一道防线)
# 这里我们模拟一个重试逻辑。
# 在生产中，你可以使用 .with_retry(stop_after_attempt=3, wait_exponential_jitter=True)
def retry_wrapper(func):
    def wrapper(*args, **kwargs):
        max_retries = 3
        for i in range(max_retries):
            try:
                print(f"尝试调用主模型 (第 {i+1}/{max_retries} 次)...")
                return func(*args, **kwargs)
            except Exception as e:
                if i == max_retries - 1: raise e # 最后一次尝试失败，抛出异常触发回退
                print(f"调用失败: {e}，准备重试...")
                time.sleep(1) # 简单等待，实际可用指数退避
    return wrapper

# 将重试逻辑应用到主链上
# 注意：LangChain 原生 .with_retry() 返回一个新的 Runnable，可以直接被 with_fallbacks 捕获
retryable_primary_chain = primary_chain.with_retry(
    stop_after_attempt=3, 
    wait_exponential_jitter=True
)

# 4. 定义回退逻辑 (第二道防线)
# 如果 retryable_primary_chain 彻底失败，则执行 fallback_chain
resilient_chain = retryable_primary_chain.with_fallbacks([fallback_chain])

# 5. 模拟调用
def ask_question(q):
    print(f"\n👤 用户提问: {q}")
    try:
        # 这里我们模拟主模型必定抛出超时错误，以演示完整流程
        # 在实际运行中，如果主模型正常，则不会触发重试和回退
        import openai
        # 模拟一个超时异常，触发重试，重试耗尽后触发回退
        raise openai.APITimeoutError(request="Mock Request") 
        
        # 正常情况直接调用：
        # response = resilient_chain.invoke({"question": q})
        # print(f"✅ 最终回复: {response}")
        
    except Exception as e:
        # 只有当所有回退方案都失败时，才会到达这里
        print(f"❌ 所有方案均失败: {e}")
        return "服务不可用"

# --- 执行流程演示 ---
# 1. 调用 ask_question -> 触发主模型 -> 抛出异常
# 2. with_retry 捕获异常 -> 重试 3 次 -> 依然失败
# 3. with_fallbacks 捕获最终异常 -> 切换到 fallback_chain -> 成功返回
ask_question("什么是 LangChain？")

顺序很重要

**必须先 with_retry 再 with_fallbacks**。

• 错误写法

  primary.with_fallbacks([fallback]).with_retry()

  • 后果：如果主模型失败，系统会切换到备用模型。如果备用模型也失败，重试机制会再次尝试“主模型->备用模型”的整个过程。这会导致不必要的延迟和资源浪费。

• 正确写法

  primary.with_retry().with_fallbacks([fallback])

  • 后果：系统会尽全力尝试主模型（重试）。只有当主模型彻底“无可救药”时，才放弃并切换到备用模型。这是最经济且用户体验最好的方式。

通过这种组合，应用既能容忍瞬间的网络波动（通过重试），又能应对长时间的服务宕机（通过回退），从而实现真正的高可用性。

与RunnableBranch的协同

将 RunnableWithFallbacks（容错回退）与 RunnableBranch（条件分支）结合使用，是构建企业级高可用 LLM 应用的黄金模式。

这种组合能实现“智能路由 + 兜底保障”的双重保险：

1. RunnableBranch：负责“分流”，根据用户意图（如：是写代码还是闲聊？是简单问题还是复杂问题？）选择最合适的处理路径。
2. RunnableWithFallbacks：负责“兜底”，确保每一条路径在遇到 API 故障、超时或报错时，都能自动切换到备用方案，保证服务不中断。

路由+回退示例

场景：智能客服系统的“分级处理 + 故障降级”

业务需求：

• 分流逻辑
  • 技术问题（包含“报错”、“代码”、“故障”）：路由到高性能模型（如 GPT-4），因为需要更强的推理能力。
  • 常规问题（包含“价格”、“地址”、“你好”）：路由到低成本模型（如 GPT-3.5），以节省成本。
• 容错逻辑
  • 无论哪条路径，如果主模型调用失败（如网络波动），都必须自动降级到本地轻量级模型或备用模型，绝对不能让用户看到报错信息。

完整代码示例：

from langchain_core.runnables import RunnableBranch, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 1. 定义模型 (模拟生产环境配置)
# 主模型：高性能但可能较慢/贵
primary_model = ChatOpenAI(model="gpt-4", request_timeout=5) 
# 备用模型：成本低、速度快，作为兜底
fallback_model = ChatOpenAI(model="gpt-3.5-turbo")

# 2. 定义处理链 (每条链都独立配置了回退机制)

# --- 路径 A: 技术问题处理链 ---
# 提示词：强调专业性和深度
tech_prompt = ChatPromptTemplate.from_template("作为技术专家，请专业解答：{question}")
# 核心逻辑：先尝试主模型，失败则回退到备用模型
tech_chain = (tech_prompt | primary_model | StrOutputParser()).with_fallbacks(
    fallbacks=[tech_prompt | fallback_model | StrOutputParser()]
)

# --- 路径 B: 常规问题处理链 ---
# 提示词：强调简洁和友好
general_prompt = ChatPromptTemplate.from_template("作为客服，请简洁友好地回答：{question}")
# 核心逻辑：同样配置回退，但使用的是常规提示词
general_chain = (general_prompt | primary_model | StrOutputParser()).with_fallbacks(
    fallbacks=[general_prompt | fallback_model | StrOutputParser()]
)

# 3. 定义分支条件函数
def is_tech_question(inputs):
    # 简单的关键词匹配，实际生产中可用更复杂的分类器
    tech_keywords = ["报错", "代码", "异常", "故障", "API"]
    return any(keyword in inputs["question"] for keyword in tech_keywords)

# 4. 构建 RunnableBranch (智能路由器)
# 结构：[(条件函数, 对应的链), ..., 默认链]
branch = RunnableBranch(
    (is_tech_question, tech_chain),   # 如果是技术问题 -> 走 tech_chain (带回退)
    general_chain                     # 否则 -> 走 general_chain (带回退)
)

# 5. 模拟调用函数
def invoke_service(user_input):
    try:
        # 传入字典以支持多参数
        result = branch.invoke({"question": user_input})
        return f"回复成功: {result}"
    except Exception as e:
        return f"系统严重错误: {e}"

# --- 测试场景 ---

# 场景 1: 技术问题 (命中 tech_chain)
# 假设 GPT-4 此时正常
print(f"用户: 我的代码报错了怎么办？")
# 流程: 分支判断为技术问题 -> 调用 GPT-4 -> 成功

# 场景 2: 常规问题 (命中 general_chain)
# 假设 GPT-4 此时网络超时 (模拟故障)
print(f"用户: 你们公司在哪里？")
# 流程: 分支判断为常规问题 -> 调用 GPT-4 (失败/超时) -> 自动触发回退 -> 调用 GPT-3.5 -> 成功

核心逻辑解析：

这个示例展示了两个维度的健壮性设计：

维度	组件	作用	结果
横向分流	RunnableBranch	根据内容 (is_tech_question) 决定走哪条路。	避免了“杀鸡用牛刀”，优化了成本和响应速度。
纵向兜底	with_fallbacks	附加在每一条具体的链 (tech_chain) 上。	即使路由正确，如果模型挂了，也能自动切换到备用模型，用户无感知。

嵌套使用示例

还可以将 RunnableBranch 作为一个整体，再包裹一层 with_fallbacks。这适用于“无论什么问题，只要主系统挂了，就全部切到备用系统”的场景。

# 定义一个“全功能主系统”
main_system = RunnableBranch(
    (is_tech_question, complex_tech_chain),
    simple_general_chain
)

# 定义一个“极简备用系统” (比如只返回固定话术或调用本地小模型)
backup_system = RunnableLambda(lambda x: "系统繁忙，请稍后再试，或联系人工客服。")

# 整体回退：如果主系统里的任何分支（无论是技术还是常规）报错，都走备用系统
resilient_app = main_system.with_fallbacks([backup_system])

在实际生产中，“分支决定路径，回退保障存活”。通过这种协同，可以构建出既聪明（能区分问题类型）又皮实（抗造、不易崩溃）的 LLM 应用架构。

流式输出场景回退

在 LangChain 的流式输出（Streaming）场景中，with_fallbacks 的处理逻辑与标准调用（Invoke）有显著不同，且存在核心限制。

当使用流式输出时，回退机制仅会在流创建初始阶段发生失败时触发。一旦流已经开始输出数据（即第一个 Token 发送给客户端），如果后续发生错误，LangChain 无法自动回退并“撤回”已发送的内容来切换模型。

在流式模式下，with_fallbacks 的触发时机非常苛刻：

1. 流创建阶段失败（可回退）：
   如果主模型在建立连接或发送第一个 Token 之前就抛出了异常（例如：鉴权失败、连接超时、DNS 解析错误），with_fallbacks 可以成功捕获异常并切换到备用模型。
2. 流传输阶段失败（不可回退）：
   如果主模型已经成功发送了部分内容（例如：“LangChain 是一个…”），然后在传输中途报错（例如：网络中断、速率限制），回退机制通常无法生效。
   • 原因：HTTP 响应流一旦开始，Header 和部分内容已经发送给客户端。你无法“撤回”已发送的字节流来替换成备用模型的回答。
   • 结果：客户端会收到一段不完整的回答，随后连接断开或抛出异常。

最佳实践与常见陷阱

推荐做法

1. 合理排序备选列表：将成本更低、响应更快的模型放在备选列表的前面
2. 关闭内置重试：使用回退前，关闭主模型的内置重试，避免主模型持续重试而非失败
3. 区分异常类型：使用 exceptions_to_handle 参数，只对预期错误触发回退（如限流、超时）
4. 本地与云端组合：本地模型（如 Ollama）失败时切换到云端模型，兼顾成本与可靠性

常见陷阱

• 过度依赖回退：回退是应急方案，不应替代主模型的稳定性优化
• 忽略流式场景限制：流已经开始后产生的错误不会触发回退，需另行处理
• 备选列表过长：过多备选会增加延迟，影响用户体验

总结

维度	说明
定位	Runnable 的备用执行逻辑，提升系统鲁棒性
核心方法	with_fallbacks(fallbacks, exceptions_to_handle)
推荐创建方式	在主 Runnable 上调用 .with_fallbacks([...])
流式限制	仅在流创建初始阶段失败时触发回退
Agent 场景	v1.2+ 推荐使用 modelFallbackMiddleware