Pydantic AI SelectModel 能力详解：按步骤动态选择模型与优先级机制-开发者社区

Pydantic AI SelectModel 能力详解：按步骤动态选择模型与优先级机制

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai

本文围绕 Pydantic AI 内置能力 SelectModel 展开。读完你将掌握：如何用 SelectModel 让 Agent 根据运行依赖、消息历史或用量在每个模型请求步骤动态挑选模型、无需在Agent构造器中提供模型；理解ModelSelectionContext中各字段的含义与取值时机；厘清能力模型选择与run(model=...)、spec=模型、构造器模型之间的优先级；以及如何在自定义能力中通过get_model()实现等价的模型选择逻辑和它的使用限制。

什么是 SelectModel

SelectModel 是 Pydantic AI 能力体系（capability）中的一个内置能力，它的作用是从**运行依赖（deps）、消息历史（message history）、累计用量（usage）或当前步骤（current step）**中挑选一个模型。选择器最早在 **run 初始化阶段（run setup）**就会被求值，这意味着 Agent 构造器里不再需要传模型参数——模型可以完全由选择逻辑在运行时决定。

从源码看，SelectModel 本身是一个非常薄的封装：它是一个 dataclass，唯一字段是selector，其get_model()方法直接把这个选择器返回给能力框架：

@dataclass class SelectModel(AbstractCapability[AgentDepsT]): """Select a model before each logical model request step. The selector receives a [`ModelSelectionContext`][pydantic_ai.models.ModelSelectionContext] containing the run dependencies, message history, accumulated usage, and lower-precedence model. It may be synchronous or asynchronous and return either a model instance or model ID. """ selector: ModelSelector[AgentDepsT] def get_model(self) -> ModelSelector[AgentDepsT]: return self.selector

也就是说，SelectModel是把“模型选择”这一单一职责封装成开箱即用的能力，供Agent(..., capabilities=[SelectModel(...)])直接挂载，而不必为选模型去子类化AbstractCapability。

完整示例：按任务复杂度选择模型

以下示例完整继承了官方文档中的用法：用Literal['standard', 'complex']标注依赖中的任务复杂度，选择器据此在两个模型之间切换，且Agent 构造时没有提供任何模型：

from dataclasses import dataclass from typing import Literal from pydantic_ai import Agent, ModelSelectionContext from pydantic_ai.capabilities import SelectModel @dataclass class Deps: """Dependencies that influence model selection.""" task_complexity: Literal['standard', 'complex'] def select_model(ctx: ModelSelectionContext[Deps]) -> str: """Use the larger model for complex tasks.""" return 'openai:gpt-5.6-sol' if ctx.deps.task_complexity == 'complex' else 'openai:gpt-5.6-luna' agent = Agent(deps_type=Deps, capabilities=[SelectModel(select_model)])

几个关键点：

选择器可以是同步或异步函数，返回模型 ID 字符串（如'openai:gpt-5.6-sol'，随后走常规的模型推断与解析流程）或直接返回Model实例（跳过 ID 解析，直接使用该实例）。
选择器在每个新的“逻辑模型请求步骤”（logical model request step）之前求值。同一次 run 内的多个步骤（工具调用后的下一轮请求）都会重新走一遍选择逻辑。仓库测试 tests/test_agent.py 中验证了这一点：一个记录ctx.run_step的 selector 在三次模型请求后记录到selected_steps == [1, 2, 3]，即逐步求值。
同一模型 ID 会被复用：当选择器在多个步骤返回相同的模型 ID 时，已解析出的 model/provider 实例会在本次 run 内被复用，不会重复构造。同一步骤内的 provider 端续传轮询（continuation polling）始终锁定（pinned）在该步骤选中的模型上，不会中途换模型。
如果选择需要查询外部系统（如按租户查网关配置），就把选择器写成async def，在异步选择器里做 I/O，而不是在同步代码里阻塞。

理解 ModelSelectionContext

选择器收到的ModelSelectionContext与RunContext是不同的对象——因为完整的RunContext本身需要依赖“正在被选中的模型”，而选模型恰恰发生在RunContext能构建出来之前。它是一个 frozen dataclass，包含以下字段：

字段	类型	含义
`agent`	`AbstractAgent`	正在被解析模型的 Agent（继承自`ModelResolutionContext`）
`deps`	依赖类型	本次 run 传入的运行依赖，示例中即`Deps`
`model`	`Model \\| None`	低优先级的回退模型：在第一个步骤时是较低优先级的模型（如构造器模型或`None`），之后则是上一个步骤使用的模型
`run_step`	`int`	正在选择的请求步骤号，从`1`开始
`messages`	`list[ModelMessage]`	本请求步骤之前可用的消息历史
`usage`	`RunUsage`	本请求步骤之前 run 已累计的用量

从源码实现可以确认messages的精确定义：agent 图的_select_model函数在构造上下文时显式传入messages=list(ctx.state.message_history[:-1])——当前步骤的ModelRequest在追加进历史之后又被剔除，保证选择器看到的是“处理该请求之前”的历史，与 bootstrap 选择时的语义保持一致，且不让选择器通过该列表间接改动图状态：

async def _select_model(ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, Any]]) -> None: selector = ctx.deps.model_selector if selector is None or ctx.deps.model_selected_for_step == ctx.state.run_step: return agent = ctx.deps.agent assert agent is not None selection_ctx = models.ModelSelectionContext( agent=agent, deps=ctx.deps.user_deps, model=ctx.deps.model, run_step=ctx.state.run_step, # The current request has already been appended, but selection describes the model # that will handle it. Expose the history available before this request step, matching # bootstrap selection, and do not let selectors mutate graph state through the context. messages=list(ctx.state.message_history[:-1]), usage=ctx.state.usage, ) model, model_id = await ctx.deps.evaluate_model_selector(selector, selection_ctx) await ctx.deps.enter_model(model) ctx.deps.model = model ctx.deps.model_id = model_id ctx.deps.model_selected_for_step = ctx.state.run_step

（参见 pydantic_ai_slim/pydantic_ai/_agent_graph.py。）这段代码还说明了**“同 ID 复用”的机制**：model_selected_for_step记录了为哪个步骤完成过选择，只有当它落后于当前run_step时才重新求值，否则直接复用。

基于这些字段，你可以写出比示例更精细的策略，例如：run_step用于“前几步用小模型做路由、后续步骤换大模型”；usage用于“累计 token 超过阈值后升级到更强模型”；messages用于“历史中出现大量代码块时选择代码能力更强的模型”；model字段则用于“默认沿用低优先级模型，仅满足某条件时才覆盖”。

选择优先级：谁决定最终模型

能力提供的模型并非无条件生效。从自定义能力文档的模型选择章节可以确认完整的优先级链（从高到低）：

run() / iter() 的 model= 参数 › run 级 spec= 模型 › 能力 get_model() 选择 › Agent 构造器模型

在调用点显式传agent.run('hi', model='openai:gpt-5.6-luna')会完全跳过能力侧的模型选择；
run 级spec=中的模型高于能力选择；
能力get_model()的选择高于构造器模型；
另外，override(model=...)返回的派生 Agent 的模型仍高于上述全部来源。

所以SelectModel的定位是：在调用点没有显式指定模型时，为本次 run 提供“动态的构造器模型”。它与模型 ID 解析能力 Resolve Model ID 是互补关系——SelectModel决定“选哪个模型 ID”，resolve_model_id()决定“这个 ID 如何变成带正确凭证/注册表配置的 Model 实例”。解析结果在单次 run 内按模型 ID 缓存：若后续步骤的选择器再次返回同一个字符串，框架直接复用已解析的 model、provider 与 client，而不会重新调用 resolver；若希望在后续步骤刻意得到不同实例，应选择另一个 ID 或直接返回Model实例。

在自定义能力中实现模型选择

当模型选择只是你更大的自定义能力的一部分时，不需要SelectModel，直接子类化AbstractCapability并覆盖get_model()即可。get_model()返回一个ModelSelector——从源码的类型别名看（pydantic_ai_slim/pydantic_ai/capabilities/abstract.py）：

ModelSelector: TypeAlias = 'Callable[[ModelSelectionContext[AgentDepsT]], ModelSelection | Awaitable[ModelSelection]]'

即一个接收ModelSelectionContext、返回模型选择结果（或其 awaitable）的同步/异步可调用对象。完整示例（沿用官方文档中按依赖选模型的写法）：

from __future__ import annotations from dataclasses import dataclass from typing import Literal from pydantic_ai import Agent, ModelSelectionContext from pydantic_ai.capabilities import AbstractCapability, ModelSelector @dataclass class Deps: """Dependencies that influence model selection.""" task_complexity: Literal['standard', 'complex'] class AdaptiveModel(AbstractCapability[Deps]): """Select a model for each request step.""" def get_model(self) -> ModelSelector[Deps]: return self.select_model def select_model(self, ctx: ModelSelectionContext[Deps]) -> str: return 'openai:gpt-5.6-sol' if ctx.deps.task_complexity == 'complex' else 'openai:gpt-5.6-luna' agent = Agent(deps_type=Deps, capabilities=[AdaptiveModel()])

与SelectModel行为上的差别在于get_model()的返回值语义：

直接返回模型或模型 ID：每 run 只解析一次（static selection）；
返回选择器（callable）：如上文示例，在每个逻辑模型请求步骤前求值（per-step selection）。SelectModel恰好是“始终走 per-step 选择”的封装。

注意get_model()本身是同步的配置方法，应保持轻量；需要 I/O 请放在（异步）选择器里。

选择生命周期与使用限制

以下几个边界条件来自 custom.md 的 Model selection lifecycle and limitations 章节，使用动态选模型时应提前了解：

Bootstrap 时机的划分。for_agent()绑定之后、for_run()之前，框架会用能力树做 bootstrap 模型解析——因为“解析出第一个模型”本身就是构造完整RunContext的前提。若for_run()返回了一个携带不同选择器的替换能力，则从步骤一开始用新选择器重新选择；若for_run()原样返回能力本身，则 bootstrap 选出的模型被步骤一直接复用。
模型选择与解析是急切（eager）钩子。延迟加载（deferred）的能力即使后来被加载，也不会贡献模型选择或解析；CapabilityFunc或只有for_run()才引入模型的能力，必须依赖一个已存在的 bootstrap 模型——它可以替换它，但不能为一个“无模型 Agent”冷启动模型。
与 durable execution 的兼容限制。动态模型选择目前不被持久执行（durable execution）能力支持：durable run 需要执行前注册模型 ID，并在重放/跨 run 恢复时重建出同一个被选中的模型。因此使用 durable execution 时应传入显式注册过的模型；从一个“由选择器选出模型”的 run 恢复挂起的 provider 请求到另一个普通 run 时，同样需要显式模型。
与 FallbackModel 的分工。FallbackModel 解决的是“请求失败后换模型重试”，与SelectModel的“每步主动选模型”互补而非替代：如果需求是失败重试，应选择配置FallbackModel返回给get_model()，而不是在选择器里自己写重试逻辑。