ADK Python 模型容错实战：用 FallbackModel 构建跨模型故障转移的可靠 Agent
【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python
FallbackModel是 ADK（Agent Development Kit）Python 提供的一个BaseLlm包装器：它持有一组有序的模型列表，当主模型调用失败（如限流 429、服务不可用 503）时，自动把同一请求转交给下一个模型，让 LLM 提供商的故障不再传导为整个 Agent 的中断。本文以官方指南 docs/guides/models/fallback_model/index.md 为核心，结合 源码实现 与 单元测试，完整讲解其配置方式、故障转移判定规则、请求回滚机制、流式与 Live 连接的边界行为以及已知限制，帮助你在生产环境把"提供商坏了"变成"换个模型继续跑"。
为什么需要 FallbackModel：把提供商的坏消息挡在调用链之外
LLM 提供商限流（rate-limit）和宕机是常态。没有恢复路径时，一个 429 或 503 会从模型调用处抛出，穿透整个 Flow，直接终结一次 invocation——提供商的"糟糕一分钟"就成了 Agent 的"一次中断"。
FallbackModel给失败一个去处：它持有多个模型，按顺序依次尝试，返回第一个成功的响应。关键设计在于它自身就是一个BaseLlm（见 base_llm.py），因此 ADK 其余部分无需任何感知：
LlmAgent.model直接接受它；- Flow 照常通过
generate_content_async调用它； - 真正服务请求的那个委托模型（delegate）会以其名字出现在请求和 trace 上。
从设计哲学看，它是刻意窄化的：它不做单模型重试，也不按成本或任务路由——模型失败就直接放弃，换下一个。相邻问题（重试、路由）由模型层已有的机制解决，下文 "与其他容错层次的分工" 一节会说明它们如何各司其职。
快速上手：两种配置方式
方式一：直接给模型名字列表
把想要的模型和后备模型放进models：
from google.adk.agents import LlmAgent from google.adk.models import FallbackModel agent = LlmAgent( name='reliable_agent', model=FallbackModel(models=['gemini-3.1-pro-preview', 'gemini-3.5-flash']), instruction='You are a helpful assistant.', )如果gemini-3.1-pro-preview返回 429，同一个请求会被转给gemini-3.5-flash，Agent 看到的是一个正常响应。
方式二：混入模型实例，给后备模型独立配置
条目也可以是模型实例，这正是给后备模型配独立参数的途径：
from google.adk.models import FallbackModel from google.adk.models.google_llm import Gemini from google.genai import types FallbackModel( models=[ 'gemini-3.1-pro-preview', Gemini( model='gemini-3.5-flash', retry_options=types.HttpRetryOptions(attempts=3), ), ], )在这个例子里，后备模型Gemini自带retry_options，意味着轮到它时，genai SDK 会在 HTTP 层先按自己的重试策略尝试，再决定是否把错误交回给FallbackModel。
工作原理：从解析到回滚的完整链路
延迟解析与缓存
models的每个条目都会被解析为一个BaseLlm：实例原样使用；字符串通过LLMRegistry.new_llm解析一次并缓存（见 registry.py）。解析被推迟到首次使用时，与LlmAgent.model的做法一致，因此构造FallbackModel永远不会触发提供商包的导入——给一个Claude后备并不会拉入anthropic包，除非真的走到那个后备。代价是：拼错的后备模型名要到第一次真正需要后备时才会报错，而不是在定义 Agent 时。
这一点在源码_delegate方法（src/google/adk/models/_fallback_model.py#L325-L331）中清晰可见，测试 test_names_are_not_resolved_at_construction 也专门验证了构造时不解析、未知名字延迟暴露的行为。
模型名重定向：委托者是谁，请求就指向谁
委托模型的名称会在调用前写入LlmRequest.model（源码见 src/google/adk/models/_fallback_model.py#L386），因为模型是从请求上读取名字，而不是从自身读取。没有这一步，后备模型会被塞进主模型的名字，从而调用错误的模型。测试 test_request_model_points_at_the_delegate 验证了主模型失败后，请求最终保留的是实际服务者的名字（backup）。
请求就地编辑与失败回滚
模型在发送前会就地修改请求：追加用户轮次、预处理工具；Live 连接还会把语音配置、系统指令、工具和 HTTP 选项写上去。因此，一个模型失败后，必须先把它的改动回滚再尝试下一个模型，否则后备模型会继承调用者从未要求的设置。更隐蔽的问题是：模型只在"自己拥有"某些配置时才写它们，若后备模型没有自己的语音配置，它就会用主模型的声音说话。
回滚对两种路径（普通轮次和 Live 连接）都生效。其快照实现是_RequestSnapshot（src/google/adk/models/_fallback_model.py#L87-L130），只捕获四类内容：
contents（内容列表）config（GenerateContentConfig生成配置）live_connect_config（Live 连接配置）- 请求自身的簿记（bookkeeping，即
_SNAPSHOT_PRIVATE指定的私有属性）
之所以不能整份深拷贝请求，是因为tools_dict持有 live 工具对象，MCP 工具内部还握有一个无法复制的threading.Lock。工具是模型只读、从不编辑的注册表，因此被排除在快照之外。测试 test_falls_back_with_a_tool_that_cannot_be_copied 专门用带锁的工具验证了这一点：回滚不会尝试复制它们，委托看到的仍是同一个实例。
成功模型保留自己的改动——这正是 trace 上展示的内容（测试 test_the_model_that_succeeds_keeps_its_edits 验证）。测试文件中还通过test_a_failed_model_does_not_leak_its_edits_to_the_next与test_every_private_attribute_is_accounted_for等用例，把"哪些字段被恢复、哪些被刻意跳过"固化为回归约束，防止LlmRequest后续新增字段悄悄泄漏到下一次尝试。
什么样的失败才会触发切换
只有携带retriable_status_codes之一的状态码才会切换到下一个模型。状态码从提供商抛出的各种错误形态中提取（_status_code实现见 src/google/adk/models/_fallback_model.py#L179-L203）：
| 错误来源 | 状态码读取位置 |
|---|---|
google.genai的APIError | code字段（google-genai 把所有 4xx 折叠为ClientError、5xx 折叠为ServerError，状态码在code上） |
| litellm / OpenAI / Anthropic 错误 | status_code字段 |
httpx错误（如ApigeeLlm抛出） | response.status_code |
没有状态码的错误——连接重置、回调里的 bug——从未到达服务端，不足以构成换一家模型尝试的理由，因此原样向外传播（测试 test_error_without_status_propagates）。
被刻意排除的"带状态"错误
有些错误形态虽带状态码，却被有意排除：
- litellm 的误报 500：
APIConnectionError和APIResponseValidationError都被 litellm 硬编码为 status 500，但前者从未到达服务，后者说明响应已到达却在客户端检查失败。两者都被识别并当作"无状态"处理（实现见 src/google/adk/models/_fallback_model.py#L141-L176）。若按面值对待，会导致把服务可能已计费并执行过的提示词重发一遍。测试 test_litellm_misreported_500_does_not_fall_back 验证了这一点。 - 408 不在默认集合中：与 ADK 的重试列表不同，默认集合刻意不含 408，因为超时并不能说明请求是否已被处理。切换到另一个模型是比重试同一个模型更重的承诺，代价可能是同一提示词被付费执行两次。litellm 甚至把客户端超时也报成 408。若某个提供商的 408 确定表示请求被丢弃，可自行加回（见下文配置）。
配置选项
| 选项 | 类型 | 默认值 | 说明 |
|---|---|---|---|
models | list[str \| BaseLlm] | 必填 | 按顺序尝试的模型列表，第一项是主模型。 |
retriable_status_codes | frozenset[int] | {429, 500, 502, 503, 504} | 触发切换到下一个模型的状态码集合。 |
models 的约束
models至少需要一个条目，空列表在构造时即被拒绝（Pydantic 的min_length=1，见 src/google/adk/models/_fallback_model.py#L276，测试 test_empty_models_is_rejected 验证）。第一项是主模型：
capabilities报告的是主模型的能力；model属性由主模型派生而来。
model继承自BaseLlm，不能直接设置——直接传model=会被 Pydantic 校验拒绝（测试 test_setting_model_directly_is_rejected），源码_derive_model_name_from_primary（src/google/adk/models/_fallback_model.py#L311-L323）会抛出说明性错误，因为直接给的名字只会被报告、不会真正选择任何模型。要配置的就是models列表。
收窄或放宽 retriable_status_codes
默认集合是"ADK 重试的状态码集合减去 408"——ADK 自身的重试集合见 evaluation/_retry_options_utils.py#L27-L34，包含 408、429、500、502、503、504。可以收窄为只对限流做故障转移，或为某个用别的方式表达过载的提供商放宽：
from google.adk.models import FallbackModel FallbackModel( models=['gemini-3.1-pro-preview', 'gemini-3.5-flash'], retriable_status_codes=FallbackModel.DEFAULT_STATUS_CODES | {529}, )DEFAULT_STATUS_CODES是公开的类属性（源码定义见 src/google/adk/models/_fallback_model.py#L257-L263），可直接取用并扩展。测试 test_default_status_codes_is_reachable_from_the_class 与 test_default_status_codes_membership 固定了这一集合的内容，测试 test_a_timeout_does_not_fall_back_by_default 验证 408 默认不触发切换。
所有模型都失败之后：在 ADK 既有错误处理处兜底
当每个模型都失败时，最后一个提供商抛出的错误会原样向外传播（源码 src/google/adk/models/_fallback_model.py#L412-L416 的注释说明这是为了让LlmAgent.on_model_error_callback能接手）。若想用一条回复而不是终结本次 invocation，就在 ADK 本来就处理模型错误的地方处理它——LlmAgent.on_model_error_callback或等价的插件钩子：
from google.adk.agents import LlmAgent from google.adk.agents.callback_context import CallbackContext from google.adk.models import FallbackModel from google.adk.models.llm_request import LlmRequest from google.adk.models.llm_response import LlmResponse from google.genai import types def on_model_error( callback_context: CallbackContext, llm_request: LlmRequest, error: Exception, ) -> LlmResponse: return LlmResponse( content=types.Content( role='model', parts=[types.Part(text='Every model is unavailable; try again.')], ) ) agent = LlmAgent( name='reliable_agent', model=FallbackModel(models=['gemini-3.1-pro-preview', 'gemini-3.5-flash']), on_model_error_callback=on_model_error, )这个钩子并非本类专属，因此它同样覆盖 FallbackModel 吸收不了的失败：不可重试的状态码，以及轮次已经开始流式输出后的失败。
与其他容错层次的分工
FallbackModel刻意只做"跨模型故障转移"，相邻问题由其他层次负责，三层互不干扰：
1. LiteLLM 提供商的 fallback。如果所有想用的模型都能通过 LiteLLM 触达，LiteLlm本身就有此能力：
LiteLlm(model=..., fallbacks=[...])该列表被透传给 litellm，由提供商内部完成失败转移。FallbackModel是跨模型类的方案——Gemini主模型配Claude后备，或任何BaseLlm子类——且两者可组合：一个配置了fallbacks的LiteLlm实例可以作为这里的条目之一。仓库样例 contributing/samples/models/litellm_with_fallback_models/agent.py 展示了LiteLlm(model='gemini/gemini-2.5-pro', fallbacks=['anthropic/claude-sonnet-4-5-20250929', 'openai/gpt-4o'])的用法，并配合before_model_callback/after_model_callback观察模型选择的变化——注意该样例用的是 LiteLLM 自带 fallback，而非FallbackModel类本身。
2. 单模型重试。重试是独立一层，留在模型自己身上：Gemini和ApigeeLlm接受retry_options，由 genai SDK 在 HTTP 层应用。FallbackModel对每个模型恰好尝试一次，这样单个 429 不会被两个互不知晓的层次重复重试。
3. 服务端路由。第三层通过模型名触达：model-optimizer-*条目在 Vertex AI 上做服务端路由，它本身可以作为这里的第一个条目，后面再跟一个客户端后备：
FallbackModel(models=['model-optimizer-exp-04-09', 'gemini-3.5-flash'])流式输出：什么时点之后不再切换
流式对故障转移施加了约束。一旦模型产出了该轮次的第一个响应，这一轮就归它所有，之后的失败直接传播而不是切换：调用者已经持有前面发出的 chunk，启动第二个模型会把两个模型的输出拼接进同一轮次（源码注释见 src/google/adk/models/_fallback_model.py#L404-L408）。非流式调用只产出一次，因此几乎总是在那个时点之前失败，可以自由切换。测试 test_streaming_failure_after_first_chunk_does_not_fall_back 用"先产出两个 chunk 再抛 429"的场景验证了后备模型不会被叫来收尾。
此外，包装器通过Aclosing传递"提前放弃"的语义：若调用者提前停止消费（回调抛异常、客户端断开），委托模型的流——及其下的提供商连接——会被立即关闭，而不是留给事件循环的终结器。测试 test_abandoning_the_stream_closes_the_delegate 验证了这一点。
Live 连接：同一条规则，在连接边界上
connect按顺序尝试每个模型，产出第一个成功建立连接的那个：
- 连接尚未打开时：没有任何数据穿越连接，把尝试交给另一个模型不会损失任何东西，因此可以故障转移。失败的尝试在尝试下一个模型前被回滚（与轮次相同）。
- 连接已打开后：会话归该模型所有——后备模型无法恢复一个已在进行的双向会话——之后的失败原样到达调用者。
- 收尾保证：无论正常退出还是
async with体抛异常，已打开的连接都会被关闭（AsyncExitStack实现，见 src/google/adk/models/_fallback_model.py#L463）。测试 test_connect_closes_the_connection_when_the_body_raises 验证了异常路径上的清理。
Live 的"就地编辑回滚"同样严格：Gemini.connect会写入speech_config、system_instruction、tools、thinking_config、safety_settings和http_options，其中一些只在模型持有它们时才写，所以没有回滚的话，没有自己语音的后备模型就会用主模型的声音说话。测试 test_a_failed_connect_does_not_leak_its_edits_to_the_next 用_VoiceLlm验证了失败连接不泄漏语音配置。
限制与边界行为
实验性状态。FallbackModel默认开启（特性注册见 src/google/adk/features/_feature_registry.py#L157-L159，FALLBACK_MODEL为 EXPERIMENTAL 且 default_on=True），首次构造时警告一次，但 API 仍可能变化。设置环境变量ADK_DISABLE_FALLBACK_MODEL可关闭它，此时构造会直接抛错。
"不可达"不等于"可切换"。提供商不可达（而非带错误应答）不会触发故障转移——见上文"什么样的失败才会触发切换"。后备模型只能救"服务在线但拒绝干活"的情况。
Live 会话断线回到所属模型，不故障转移。会话恢复句柄只对签发它的模型有意义；若该模型持续宕机，重连会持续失败，而不是悄悄启动另一个模型的会话。所有者是针对 live flow 为本次运行构建的请求记录的，因此跟随的是会话而非模型名——两个条目可以同名（同一模型背后的两个 key 或区域），仍能被区分。相关测试包括 test_reconnect_follows_the_session_not_the_name、test_reconnect_works_for_two_entries_with_the_same_name。
跨运行恢复的局限。通过RunConfig.session_resumption把句柄带进新的run 时，该句柄属于这个模型从未见过的请求，唯一可依据的是 flow 写上的名字——即 Agent 自己的名字。这样的 run 被钉在主模型上，首次连接没有故障转移；如果会话实际由后备模型持有，句柄会被交给从未签发它的模型。两个报告相同模型名的条目在这种情况下无法区分，重连会抛错而不是猜测（_candidate_indexes的 ValueError 逻辑见 src/google/adk/models/_fallback_model.py#L567-L593）。
包装会隐藏具体类型。live flow 只为 Vertex AI 上的Gemini设置session_resumption.transparent，而FallbackModel不是Gemini，因此该默认值不会应用。
capabilities 始终来自主模型。即使后备模型服务了请求，请求也是在任何调用之前构建的，即为主模型构建的。因此请让各条目在能力上尽量接近，使同一请求能适配所有条目（源码 src/google/adk/models/_fallback_model.py#L340-L349）。
相关样例
官方指南指出目前尚无FallbackModel专属样例，最接近的现有样例是 contributing/samples/models/litellm_with_fallback_models/agent.py，它使用的是 LiteLLM 自带的提供商级 fallback 而非本类。若要亲自验证FallbackModel的行为，可参照单元测试 tests/unittests/models/test_fallback_model.py——其中_FakeLlm、_rate_limited等测试桩完整覆盖了"主模型成功不动后备"、"429 切换"、"多模型逐级切换"、"非可重试状态码传播"、"流式半途失败不切换"等关键路径，是理解本类语义最直接的活教材。
【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
