多模型Fallback

线上调用 GPT 时可能遇到限流、区域故障或单次超时。手写 try/except 换模型很快变成嵌套分支。with_fallbacks 把「主 Runnable 失败 → 依次尝试备用 Runnable」声明在链上，与 LCEL 管道自然组合。

段末注释：Fallback（降级）= 主路径失败时自动切换到备用实现，此处指 ChatModel 或整条链的备用实例。

1. 一句话定位

维度	内容
角色	能力层可靠性：模型/链级别的故障转移
输入 → 输出	与主 Runnable 相同（如 messages → AIMessage）
典型调用入口	`primary.with_fallbacks([backup1, backup2])`
与 LangGraph	节点内可用 with_fallbacks；复杂重试策略（指数退避、环）用 LangGraph

2. 实现逻辑

1. 定义 primary = ChatOpenAI(...) 或 prompt | model
2. fallback = ChatAnthropic(...) 或另一 region 的 OpenAI
3. robust = primary.with_fallbacks([fallback], exceptions_to_handle=(...))
4. robust.invoke(input) → 先调 primary
5. 若抛出指定异常 → 调 fallback.invoke(同一 input)
6. 全部失败 → 抛出最后一个异常

字段级变形：输入 messages 不变；成功时输出仍为 AIMessage，仅 响应来源模型 不同（可在 callback metadata 中区分）。

3. 原理说明

3.1 Runnable.with_fallbacks

定义在 langchain_core.runnables.base.Runnable。fallback 列表有序：第一个失败才试第二个。

with_fallbacks(fallbacks, *, exceptions_to_handle=(Exception,), exception_key=None)（方法）
功能：主 Runnable 抛指定异常时按序换备用实现。返回新 Runnable，不改原对象。
默认值：exceptions_to_handle=(Exception,)（过宽，生产应收窄）；exception_key=None。
最小维度：fallbacks 为 Sequence，len ≥ 1；空列表无降级。

robust = primary.with_fallbacks(
    [backup],
    exceptions_to_handle=(TimeoutError, OSError),
)
ai = robust.invoke([HumanMessage("ping")])

3.2 与 retry 的区别

with_retry(*, stop_after_attempt=3, wait_exponential_jitter=True, ...)（方法）
功能：对同一 Runnable 失败重试。默认约 3 次、指数退避（具体关键字视版本）。
最小维度：stop_after_attempt ≥ 2 才构成重试（=1 等于不重试）。

1	flaky.with_retry(stop_after_attempt=3).invoke(msg)

with_fallbacks 换不同实现（不同 model / 厂商 / max_tokens）。可叠：先 retry 再 fallback。

3.3 链级 fallback

可对整条 LCEL：chain.with_fallbacks([simple_chain])。主链 RAG+大模型，备用链小模型直答。
最小维度：主链与备用链 输入输出类型须兼容（同为 list[BaseMessage] → AIMessage，或同为 dict → str）。

3.4 路由（Router）对比

RunnableBranch（类）
功能：按输入条件选支路，不是失败驱动。
默认值：须给 (condition, runnable) 列表 + 默认支路。
最小维度：至少 1 条条件支路 + 1 个 default。

from langchain_core.runnables import RunnableBranch, RunnableLambda
branch = RunnableBranch(
    (lambda x: x["lang"] == "en", en_chain),
    zh_chain,  # default
)

4. 最小可运行示例

1	pip install -U langchain-openai

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from openai import APIConnectionError, APITimeoutError

def ollama_chat(model: str = "qwen3.5:9b", **kwargs) -> ChatOpenAI:
    """指向本地 Ollama 的 ChatOpenAI。

    输入：model 为 Ollama 模型名；kwargs 覆盖 timeout、base_url 等。
    输出：ChatOpenAI 实例。
    处理：走 OpenAI 兼容 /v1；需 `ollama serve` 且已 pull 该模型。
    """
    return ChatOpenAI(
        model=model,
        api_key="ollama",
        base_url="http://localhost:11434/v1",
        temperature=0,
        **kwargs,
    )

# 主路径故意连错端口，触发连接失败 → 备用 llama3.1
primary = ollama_chat(base_url="http://127.0.0.1:9", timeout=5)
backup = ollama_chat("llama3.1:8b")
robust = primary.with_fallbacks(
    [backup],
    exceptions_to_handle=(APIConnectionError, APITimeoutError, TimeoutError, OSError),
)

msg = [HumanMessage(content="只回复一个词：pong")]
print(robust.invoke(msg).content)
# 预期形态：含 pong（来自备用 llama3.1，主路径连不上）

# 主备都健康时：先走 qwen，失败才 llama
healthy = ollama_chat("qwen3.5:9b").with_fallbacks([backup])
print(healthy.invoke(msg).content)

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点	配置指导
`fallbacks`	`list[Runnable]`，必填	主路径失败后按序改走的备用 Runnable	链越长尾延迟越高；空列表无降级	1～2 个	不宜过长
`exceptions_to_handle`	tuple[type]，默认 `(Exception,)`	哪些异常才触发降级	过宽会吞业务 bug；过窄降级不生效	`(TimeoutError,)` + 厂商 APIError	勿捕获 ValueError
`exception_key`	str，可选，默认 None	把捕获的异常写入输出 dict 的哪个键	打开便于调试，会改变输出形状	少用	仅调试开
主/备 `model`	str，构造器	主路径与备用路径各自用的模型 ID	备模型弱则质量降、可用性升	备模型略弱但稳定	备模型须支持同一输出形态（如 tool_calls）
主/备 `timeout`	float	单次请求超时，主备可分别设	主过短会过早降级；备过短仍失败	主 30s，备 60s	备模型可略放宽
`max_retries`	int（model 上）	同一模型 SDK 重试，发生在 fallback 之前	与 fallback 叠用会放大延迟	与 fallback 分层	先 retry 再 fallback 或二选一

5. 易踩坑

备用模型不支持 tool_calls：主模型走工具链，降级后 schema 不兼容，Agent 静默失败。
exceptions_to_handle 过宽：捕获 Exception 会吞掉编程错误，难以排查。
fallback 顺序与成本：把最贵模型放第一顺位 fallback，故障时费用激增。

小结

with_fallbacks 在 Runnable 层声明主备切换，适合限流与区域故障。
与 with_retry 互补：前者换实现，后者同实现重试。
生产须收窄 exceptions_to_handle，并保证备模型能力匹配（尤其工具调用）。
按内容路由用 Branch，不是 fallback。