如果你在生产环境中调用大模型,一定遇到过这些问题:
- GPT-4 突然限流,请求直接报错
- DeepSeek 频繁降级,输出质量断崖式下降
- 某个模型返回格式不对,下游解析全部失败
单模型调用在生产环境中几乎必然会遇到可用性问题。最直觉的解决方案是自动降级——首选模型失败时,自动切换到备选模型。
但现有的降级方案(包括 LangChain 的 .with_fallbacks())在灵活性上有不少局限。所以我开发了 model-fallback-tool,一个通用的 Python 降级工具库。
设计原则
三个核心设计目标:
- 通用性——不依赖特定的大模型 SDK,适用于任何 Python 函数调用
- 灵活的降级策略——支持基于异常类型或返回值自定义降级判断
- 非侵入性——成功时直接返回原函数的返回值,无需修改现有业务逻辑
安装
pip install model-fallback-tool
# 或
uv add model-fallback-tool基础用法
最简单的场景:处理可能抛出异常的模型调用。
from model_fallback_tool import ModelFallback
def my_llm_call(prompt, model):
if model != "gpt-3.5-turbo":
raise Exception(f"Model {model} is not available")
return f"Response from {model}"
fallback_tool = ModelFallback(
candidate_models=["gpt-4", "gpt-3.5-turbo"]
)
# gpt-4 失败后自动切换到 gpt-3.5-turbo
result = fallback_tool.invoke(my_llm_call, prompt="Hello")自定义降级条件
默认情况下,任何异常都会触发降级。但在实际场景中,你可能需要更精细的控制:
from openai import OpenAI, RateLimitError
def should_fallback(error):
# 仅限流时降级,参数错误等直接抛出
return isinstance(error, RateLimitError)
fallback = ModelFallback(
candidate_models=["gpt-4", "gpt-3.5-turbo", "claude-3-opus"],
should_fallback=should_fallback
)降级判断函数 should_fallback 不仅能检查异常,还能检查返回值:
def should_fallback(result_or_error):
# 异常:输出格式错误时降级
if isinstance(result_or_error, OutputParserException):
return True
# 返回值:模型拒绝回答时降级
if isinstance(result_or_error, str) and "I cannot answer" in result_or_error:
return True
return False这意味着你可以根据模型输出内容的质量来决定是否切换——这是大多数降级工具做不到的。
与 LangChain LCEL 集成
如果你在 LangChain 生态中工作,ModelFallbackRunnable 可以像普通 Runnable 一样通过 | 操作符集成到 Chain 中:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from model_fallback_tool import ModelFallbackRunnable
# 批量创建备选模型实例
runnables = ModelFallbackRunnable.create_runnables_from_config(
runnable_class=ChatOpenAI,
candidate_models=["gpt-4", "gpt-3.5-turbo"],
temperature=0,
)
llm_with_fallback = ModelFallbackRunnable(runnables=runnables)
# 像普通 Runnable 一样使用
prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
chain = prompt | llm_with_fallback | StrOutputParser()
result = chain.invoke({"topic": "工程师"})结构化输出
如果需要 Pydantic 结构化输出,先绑定结构化再传入降级:
from pydantic import BaseModel
class Joke(BaseModel):
setup: str
punchline: str
llm_gpt4 = ChatOpenAI(model="gpt-4").with_structured_output(Joke)
llm_gpt35 = ChatOpenAI(model="gpt-3.5-turbo").with_structured_output(Joke)
structured_llm = ModelFallbackRunnable(runnables=[llm_gpt4, llm_gpt35])
joke = structured_llm.invoke("Tell me a joke") # 返回 Joke 对象与 DSPy 集成
DSPy 用户可以使用 FallbackLM 包装器:
import dspy
from model_fallback_tool import FallbackLM
fallback_lm = FallbackLM.from_config(
provider_class=dspy.LM,
models=["openai/gpt-4", "openai/gpt-3.5-turbo"],
)
dspy.settings.configure(lm=fallback_lm)
# 像往常一样使用 DSPy与 LangChain 原生 .with_fallbacks() 的区别
| 维度 | .with_fallbacks() | model-fallback-tool |
|---|---|---|
| 降级条件 | 仅基于异常 | 异常 + 返回值自定义 |
| 适用范围 | 仅 LangChain Runnable | 任何 Python 函数 |
| Agent 集成 | 需要额外处理 | 提供 Middleware 支持 |
| DSPy 支持 | 不支持 | 原生支持 |
| 配置便捷性 | 每个模型需手动实例化 | create_runnables_from_config 批量创建 |
总结
大模型调用的可靠性是生产环境中的核心问题。一个好的降级方案应该:
- 对业务代码无侵入——不需要为了降级改写已有逻辑
- 降级条件可定制——不只是"报错就切",而是能根据输出质量判断
- 生态兼容——OpenAI、LangChain、DSPy 都能用
如果你也苦于模型限流和降级带来的困扰,可以试试:
pip install model-fallback-tool项目地址:PyPI