Pydantic AI Realtime 会话中的 Capabilities 与 Hooks生命周期映射、事件流与延迟加载完整指南【免费下载链接】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 的实时speech-to-speech会话建立在与普通 Agent 运行完全相同的 capabilities 与 hooks 机制之上一个挂在 Agent 上、或通过realtime(capabilities...)传入的 capability其生命周期会被映射到一条持续存在的连接上。本文基于仓库文档 docs/realtime/capabilities.md完整拆解 capability 各阶段在会话中的行为哪些 hook 会执行、哪些不会、run hooks 的会话级语义、事件流包装、RunContext在会话中的取值以及延迟加载 capability 在实时场景下的边界与限制并佐以源码级证据帮助你为语音 Agent 正确设计可复用的能力扩展。一、背景Capability 与会话的关系在进入实时会话之前先明确 capability 的基本概念。在 Pydantic AI 中capability 是可复用、可组合的 Agent 行为单元不再需要把指令、模型设置、工具集、历史处理器等参数逐个穿进Agent构造函数而是把相关行为打包成一个 capability通过capabilities参数传入见 docs/capabilities/overview.md。在实时场景下这一点同样成立一个挂在 Agent 上、或通过realtime(capabilities...)传入的 capability会参与实时会话其生命周期映射到一条持久连接上。第三方 capabilities 的加载方式与普通运行完全一致不需要为实时会话做任何特殊处理。从 Agent 入口源码 可以看到Agent.realtime()的完整签名agent.realtime( model, # openai:gpt-realtime或一个 RealtimeModel 实例 deps..., # 依赖与 run()/iter() 相同 model_settings..., # RealtimeModelSettings instructions..., # 与 Agent 的 instructions 合并 toolsets..., # 会话的额外工具集 capabilities..., # 会话的额外 capabilities usage..., usage_limits..., message_history..., # 用于 seed 会话的历史对话 )该方法绑定 Agent 配置与实时模型返回AgentRealtime通过session()打开会话时会复用相同的 instructions、tools、capabilities 和 run context无需重复传入。二、Capability 在会话中的阶段映射实时会话没有请求-响应图request-response graph因此 capability 的各阶段被映射到会话生命周期中的不同位置。原文档给出的完整映射表如下Capability 阶段会话中的行为for_agent、for_run、get_instructions在建立连接前的 setup 阶段运行动态指令在连接时只求值一次。get_toolset、get_wrapper_toolset、prepare_tools在连接前贡献、包装、准备本地工具。get_native_tools在连接前贡献 native tools动态 native-tool 函数基于连接时的上下文解析一次与动态指令类似。工具校验/执行 hooks在每次本地函数工具调用前后运行。handle_deferred_tool_calls内联解析延迟请求详见延迟与需审批工具。图节点、模型请求、输出处理 hooks不运行会话中不存在 Agent 图或输出处理阶段。这与 abstract.py 中realtime()的文档注释 完全一致Capabilities runfor_runonce when the session connects; their instructions, toolsets, and native tools are applied. Tool hooks (prepare_toolsandbefore/after/wrap/on_errorfortool_validateandtool_execute) run for each tool call. Run hooks (before_run,after_run,wrap_run,on_run_error) run once around the session and event-stream hooks wrap the session iterator; graph, model-request, and output-stage hooks do not run.关键设计点动态指令只求值一次普通运行中动态指令agent.instructions修饰的函数在每个请求步骤都会重新求值而实时会话中没有“每请求重建”的概念动态指令在连接建立时求值一次整个会话复用该结果。动态 native-tool 函数同理在连接时基于当时的RunContext解析一次。工具在连接前固定连接建立时工具的集合即被锁定后文延迟加载部分会说明这一限制的后果。三、工具校验与执行 hooks与普通运行完全一致实时会话中所有常规的工具校验 hooks与工具执行 hooks——两个阶段的before、after、wrap、on_error——都会在每次本地函数工具调用周围运行行为与标准运行完全一致包括重试逻辑校验 hooksbefore_tool_validate、after_tool_validate、wrap_tool_validate、on_tool_validate_error在模型 JSON 参数被解析和校验时触发接收callToolCallPart与tool_defToolDefinition参数执行 hooksbefore_tool_execute、after_tool_execute、wrap_tool_execute、on_tool_execute_error在工具函数实际运行时触发args始终是校验后的dict[str, Any]仅对函数工具触发内部输出工具不参与会话没有输出阶段与标准运行相同工具 hook 支持tools参数按名称过滤目标工具也支持timeout。不运行的是任何与请求-响应图绑定的部分节点 hooks——会话没有图节点模型请求 hooks如before_model_request——会话没有按请求划分的边界输出校验 hooks与输出处理 hooks——会话没有输出阶段。从会话实现 pydantic_ai_slim/pydantic_ai/realtime/_session.py 可以看到工具调用被翻译为共享的FunctionToolCallEvent/FunctionToolResultEvent事件流该文件头部的类型别名RealtimeEvent严格是AgentStreamEvent的子集因此为文本 Agent 编写的工具日志、事件处理代码可以在会话中原样复用。四、Run hooks一次会话 一次 runbefore_run、after_run、wrap_run、on_run_error这四类 run hooks 在会话周围执行一次——实时会话本身就是一次 run——并拥有与iter()相同的 close-boundary 恢复wrap_run的错误恢复与结果转换after_run的结果变换语义。hooks.md 文档 也明确呼应了这一点A realtime session is a run: the same four hooks fire once around the session, withwrap_runrecovery andafter_runresult transformation applied when the session closes.在Hookscapability 中注册方式为hooks.on.before_run/after_run/run对应wrap_run/run_error对应on_run_error也可通过构造函数 kwargs 传入。五、事件流wrap_run_event_stream与ProcessEventStreamwrap_run_event_stream包装面向消费者的会话迭代器。它可以观察或转换共享的AgentStreamEvent成员以及仅实时会话才有的RealtimeEvent成员详见事件参考而不会改变历史记录或工具执行。两类事件在同一个流中流动共享事件AgentStreamEvent成员与标准流式运行相同PartStartEvent、PartDeltaEvent、PartEndEvent、FunctionToolCallEvent、FunctionToolResultEvent、DeferredToolRequestsEvent、DeferredToolResultsEvent实时专属事件RealtimeEvent成员RealtimeInputSpeechStartEvent、RealtimeInputSpeechEndEvent、RealtimeResponseInterruptedEvent、RealtimeOutputSpeechStartEvent/RealtimeOutputSpeechEndEvent、RealtimeTurnCompleteEvent、RealtimeSessionReconnectEvent、RealtimeSessionErrorEvent、RealtimeInputTranscriptionErrorEvent。要点没有event_stream_handler参数realtime()不接受event_stream_handler。如果你需要 handler 风格的消费者应挂载ProcessEventStreamcapability——它正是通过这同一个流工作。对Hooks而言run_event_stream对应wrap_run_event_stream与event两个 hook 在实时会话中都会触发且实时专属事件也流经同一管道见 docs/hooks.md。六、模型设置与RunContext的会话语义模型设置用RealtimeModelSettings而非普通model_settingsget_model_settings()可能在 capability setup 阶段运行但普通模型设置不会配置实时模型。实时会话拥有自己独立的设置类型RealtimeModelSettings扮演普通运行中 model run settings 的角色。传入方式from pydantic_ai import Agent from pydantic_ai.realtime import RealtimeModelSettings agent Agent(instructionsYou are a helpful voice assistant.) realtime agent.realtime( openai:gpt-realtime, model_settingsRealtimeModelSettings(output_modalityaudio) )RealtimeModelSettings是TypedDict包含以下跨提供商的共享设置从 settings.py 整理设置项说明支持方max_tokens每次响应停止前生成的最大 token 数OpenAI、Azure OpenAI、Gemini、xAIparallel_tool_calls是否允许并行工具调用OpenAI、Azure OpenAI、xAItool_choice控制模型可使用哪些函数工具none与 allow-list 在所有提供商上通过限制会话创建时广告的工具来强制实施OpenAI、Azure OpenAI、Gemini仅none与 allow-list、xAIinput_transcription_model用于转写用户音频输入的模型auto默认使用提供商推荐的实时转写模型传具体 id 固定之None关闭转写OpenAI、Azure OpenAI、Gemini仅None、xAIoutput_modality模型生成的单一模态默认audioOpenAI、Azure OpenAIGemini Live 与 xAI 始终生成音频thinking启用或配置推理/思考镜像请求-响应模型的thinking设置OpenAIgpt-realtime-2*、Gemini native-audio 模型、xAI reasoning Grok Voice 模型turn_detection自动语音活动检测VAD/ 轮次控制True或缺省为开启提供商默认False关闭push-to-talkTurnDetectiondict 提供跨提供商旋钮全部handshake_timeout等待实时协议握手事件的秒数默认30.0OpenAI、Azure OpenAI、xAIreconnectReconnectPolicy连接中断时透明恢复默认无策略时连接意外关闭是致命的全部TurnDetection提供sensitivitylow/medium/high映射 OpenAI/Azure/xAI 的 server-VAD threshold 与 Gemini 的起止灵敏度、prefix_padding_ms、silence_duration_ms三个跨提供商旋钮更细粒度控制使用提供商前缀的逃生口openai_turn_detection、xai_turn_detection、google_vad设置后完全覆盖该字段。ReconnectPolicy提供max_attempts单次断线重拨次数默认 3、max_reconnects会话生命周期总成功重连数默认 50、base_delay默认 0.5 秒指数退避翻倍、max_delay默认 30.0、jitter默认True。设置策略后xAI 与 Gemini 启用原生会话恢复Gemini 可通过google_enable_session_resumptionFalse显式退出但该组合在连接时抛UserError。值得注意的一个刻意例外普通运行中不支持的共享设置会被静默忽略但output_modalitytext用在 profile 报告supports_text_outputFalse的模型Gemini Live、xAI上时会在连接前抛出UserError——静默地用语音回答比不启动更糟。RunContext在会话中的取值在会话 hooks 与工具内部RunContext反映的是会话状态RunContext字段实时会话中的值ctx.model_settings会话连接时合并后的RealtimeModelSettingsctx.realtime整个 run 均为True包括在连接建立之前运行的for_run与指令函数ctx.realtime_session连接建立后的实时RealtimeSession注意ctx.realtime_session在before_run、指令函数、以及wrap_run的 handler 前段中仍然是None——这些都在连接建立之前运行。七、Seed 的历史不会被处理历史处理类 capabilities不会在message_history被 seed 进会话之前转换它。如果需要对历史做过滤或脱敏必须在打开会话之前预处理历史对应的限制追踪见 overview.md 的 Limitations 表。也就是说ProcessHistory这类包装历史处理器的 capability其message_history变换逻辑在实时 seed 场景下不会生效属于seeded history is not processed的明确边界。八、延迟加载deferred capability loading边界与报错延迟加载的 capability 在会话中的加载方式与普通运行一致capability catalog 是会话 instructions 的一部分调用load_capability工具返回所加载 capability 的 instructions 作为结果——这一点在所有提供商上都成立。但会话有一个普通运行没有的硬限制无法在对话中途广告新工具——连接的工具在连接打开时就固定了对应 #7288。因此如果打开会话时传入一个defer_loadingTrue且会贡献工具或 native tools 的 capability会在连接前抛出UserError——接受它意味着静默地提供比请求更少的能力因此直接拒绝。从会话实现看_session.py的_build_session_tool_return会话在打开时就拒绝defer_loadingTrue的工具和贡献工具的延迟 capability所以它持有的任何东西都不可能被中途揭示reveal对揭示请求统一走_reject_unloaded_capability_reveals拒绝。未来展望实时场景的每轮/每次交换per-turn/exchangehooks 预期会拓宽这个边界见 #7190 与 #7191。在它们落地之前需要在会话中提供工具的能力请使用非延迟立即加载方式。九、与标准 Agent 运行的对照总览标准运行特性实时会话中的表现函数工具与工具 hooks✓ —— 校验、重试、执行 hooks 与标准运行一致Run hooksbefore_run、after_run、wrap_run、on_run_error✓ —— 在会话周围各执行一次Capabilities含第三方✓ —— 在连接时解析一次事件流✓ —— 迭代会话或挂载ProcessEventStreamoutput_type与输出校验器✗ —— 在通话中委托给文本 Agent图节点与模型请求 hooks如before_model_request✗ —— 没有 Agent 图Seed 时的历史处理器✗ —— 需在打开会话前预处理event_stream_handler参数✗ —— 使用ProcessEventStream十、实践建议小结能力设计需要工具、指令或模型设置的复用逻辑用 capability 打包后挂到 Agent 或realtime(capabilities...)工具校验/执行 hooks 与 run hooks 在会话中完整生效可以放心依赖。不要依赖图相关 hooks会话没有节点、没有按请求边界、没有输出阶段before_model_request、节点 hooks、输出 hooks 都不会触发。动态内容在连接时定稿动态指令与动态 native-tool 函数只求值一次跨会话变化的状态应在打开会话前准备好。设置走RealtimeModelSettings普通model_settings配置不了实时模型会话内部通过ctx.model_settings读取合并后的实时设置ctx.realtime全程为True连接建立前ctx.realtime_session为None。历史提前清洗seed 之前自行完成过滤/脱敏历史处理 capabilities 不会介入。延迟加载慎用纯指令类延迟 capability 可以在会话中加载会贡献工具或 native tools 的defer_loadingTruecapability 会在连接前被UserError拒绝。如需进一步深入可继续阅读仓库内的 Realtime 事件参考、Realtime 概览与提供商支持、Hooks 完整参考以及实现源码 realtime/_session.py 与 realtime/settings.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），仅供参考