Runnable组件容灾回退机制 with_fallback 深度解析降低程序错误率
一、前言
在某些场合中，对于 Runnable 组件的出错，并不想执行重试方案，而是执行特定的备份/回退方案，例如 OpenAI 的 LLM 大模型出现异常时，自动切换到 文心一言 的模型上，在 LangChain 中也提供了对应的回退机制—— with_fallback。
当构建高可用AI系统时，with_fallback 机制可实现：
- 服务降级：主服务异常时自动切换备用服务
- 混合云支持：公有云服务不可用时切换私有化部署模型
- 异常隔离：错误类型定向处理
二、参数配置详解
参数说明表
| 参数名 | 类型 | 必填 | 默认值 | 说明 |
|---|---|---|---|---|
fallbacks | List[Runnable] | ✅ | 无 | 按顺序尝试的回退组件链 |
exceptions_to_handle | Tuple[Exception] | ❌ | Exception | 触发回退的异常白名单 |
exception_key | str | ❌ | None | 将异常对象注入备用组件的参数键名 |
三. 资料推荐
- 🔗 官方文档参考
- 💡大模型中转API推荐
- ✨中转使用教程
四、运行流程图示
五、代码示例
例如，我们的回退备选方案是这样的，当 OpenAI 模型调用失败时，自动切换到文心一言执行任务：
import dotenv
from langchain_community.chat_models import QianfanChatEndpoint
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
dotenv.load_dotenv()
# 1.构建prompt与LLM，并将model切换为gpt-3.5-turbo-18k引发出错
prompt = ChatPromptTemplate.from_template("{query}")
llm = ChatOpenAI(model="gpt-3.5-turbo-18k").with_fallbacks([QianfanChatEndpoint()])
# 2.构建链应用
chain = prompt | llm | StrOutputParser()
# 3.调用链并输出结果
content = chain.invoke({"query": "你好，你是?"})
print(content)