ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

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

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()实现等价的模型选择逻辑和它的使用限制。什么是 SelectModelSelectModel 是 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_typeDeps, 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包含以下字段字段类型含义agentAbstractAgent正在被解析模型的 Agent继承自ModelResolutionContextdeps依赖类型本次 run 传入的运行依赖示例中即DepsmodelModel \| None低优先级的回退模型在第一个步骤时是较低优先级的模型如构造器模型或None之后则是上一个步骤使用的模型run_stepint正在选择的请求步骤号从1开始messageslist[ModelMessage]本请求步骤之前可用的消息历史usageRunUsage本请求步骤之前 run 已累计的用量从源码实现可以确认messages的精确定义agent 图的_select_model函数在构造上下文时显式传入messageslist(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( agentagent, depsctx.deps.user_deps, modelctx.deps.model, run_stepctx.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. messageslist(ctx.state.message_history[:-1]), usagectx.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, modelopenai: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.pyModelSelector: 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_typeDeps, 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()而不是在选择器里自己写重试逻辑。小结与延伸阅读SelectModel以最小的封装成本提供了 Pydantic AI 中最灵活的模型接入方式之一Agent 可以完全不绑定构造器模型模型在每次请求步骤前由你的选择函数基于 deps、历史与用量决定框架负责实例复用、步骤级锁定与优先级仲裁。相关文档与源码入口能力总览docs/capabilities/overview.md自定义能力含get_model()、解析与优先级全节docs/capabilities/custom.md应用自定义模型 ID 解析docs/capabilities/resolve-model-id.md能力源码pydantic_ai_slim/pydantic_ai/capabilities/select_model.py选择上下文定义pydantic_ai_slim/pydantic_ai/models/init.py逐步选择实现pydantic_ai_slim/pydantic_ai/_agent_graph.py逐步求值测试tests/test_agent.py【免费下载链接】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创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表