Pydantic AI Realtime 模块架构与开发指南Codec 连接、会话循环与 Provider 适配规范【免费下载链接】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 Realtime语音到语音模块的架构级技术指南以仓库内 pydantic_ai_slim/pydantic_ai/realtime/AGENTS.md 为主体结合 realtime 目录 下的源码实现展开。读者将掌握 Realtime 模块的四层架构RealtimeModel→RealtimeConnection→RealtimeSession→ 共享消息词汇表、策略必须收敛在共享核心的安全设计原则、Provider 适配器OpenAI / Azure / xAI / Gemini的接入规范以及历史保真与 Cassette 测试方法论从而既能在应用层正确使用语音会话也能为新增 Provider 或修改协议行为提供开发依据。架构总览一次实时会话是如何组装起来的Pydantic AI 的实时语音能力与普通请求-响应式Agent.run()走的是完全不同的数据通路但共享同一套消息模型与工具执行核心。从源码结构看pydantic_ai_slim/pydantic_ai/realtime/ 目录的职责划分如下文件职责model.pyRealtimeModel抽象基类ABC与infer_realtime_model()模型推断入口codec.py连接层RealtimeConnectionABC与低层编解码词汇表codec 事件、输入类型_session.pyRealtimeSession包装连接、翻译事件、构建历史、自动执行工具settings.pyRealtimeModelSettings等跨 Provider 共享设置词汇表profiles.pyRealtimeModelProfile能力描述类型与合并工具openai.py / azure.py / google.py / xai.py各具体 Provider 的实时模型实现_openai_protocol.pyOpenAI 协议编解码被 Azure 与 xAI 复用一次会话的核心调用链是RealtimeModel.connect()打开一条 Provider 专属的RealtimeConnection即 codec 词汇表中的连接对象见 codec.pyRealtimeSession_session.py包装这条连接把 codec 事件翻译成pydantic_ai.messages中的共享消息/部件part词汇表会话一边构建普通ModelMessage历史一边运行工具循环——整个过程中工具执行走的是与 Agent 图完全相同的ToolManager核心。用 docs/realtime/overview.md 中的拓扑来表达device ↔ media bridge ↔ RealtimeSession ↔ provider ├── typed tools └── message history (your backend)这套Provider 无关的布局刻意与请求-响应侧对称model.py持有RealtimeModelABC 与infer_realtime_model对应普通模型的Model与infer_modelsettings.py持有设置词汇表对应 settings.pyprofiles.py持有RealtimeModelProfile类型对应普通模型的 profile 体系codec.py则是实时侧独有的连接层。这保证了实时会话与文本 Agent 之间共享的不仅是概念更是同一套数据与执行原语。策略必须收敛在共享核心绝不重新实现这是整个 realtime 模块最重要的一条安全设计原则。RealtimeSession执行工具时并不自己写一套会话版工具循环而是直接复用 Agent 图所用的同一个ToolManager核心具体来说就是其上的三个方法validate_tool_call—— 校验工具调用execute_tool_call—— 执行工具调用build_tool_return_part—— 把执行结果构建成历史部件。正因如此hooks、重试retries、usage 统计在实时会话中与普通运行由构造保证行为一致by construction而不是靠复制代码再祈祷不漂移。指南特别强调Agent 图在分发dispatch前施加的策略——声明式工具类型unapproved→ApprovalRequired、external→CallDeferred——是在ToolManager.handle_call内部强制执行的。会话层不得在其之上重新实现或过滤。理由是直接的一个重新实现、且与核心漂移的策略层是安全漏洞而非风格问题——真实的实时审批绕过approval-bypass漏洞恰好就是这样产生的。从 _session.py 的_unsettled_call_return可以看到会话对这两种策略结果的落地方式当一个工具调用无法正常收敛ApprovalRequired或CallDeferred且无 handler 现场解决时会话把它变成一个outcomefailed的ToolReturnPart内容是模型可以朗读的解释性文本如 requires approval and cannot be completed during a realtime session而不是把拒绝记录成一次成功的工具运行——因为all_messages()交给Agent.run()后一次被错误标记为成功的运行会误导审计。这里的关键点是会话只是表达核心的判定而非重新裁决。此外当实时行为与标准运行不同没有图节点、没有before_model_request、延迟结果现场内联解决时这种差异必须是有意为之的并且要求记录在 docs/realtime/ 用户文档中由 tests/realtime/ 下的 parity 测试覆盖。没有文档、没有测试的差异在评审中被视为缺陷。Provider 适配器typed 模型解析、共享协议与能力归一化用 typed SDK 模型解析帧禁止 ad-hoc dict 访问Provider 适配器解析线上的帧时必须通过类型化的 SDK 模型而不是随手用dict下标访问。具体约定OpenAI 协议的事件类型来自 OpenAI SDKopenai.types.realtime见 openai.py 的导入Gemini 的事件类型来自google-genai。以 openai.py 为例websockets与openai均为硬性依赖缺失时以带指引的ImportError提示安装openai-realtime可选组——即pip install pydantic-ai-slim[openai-realtime]。Azure 与 xAI 复用 OpenAI codecAzure 与 xAI 复用 OpenAI 的编解码实现_openai_protocol.py。这意味着协议 bug 应修在_openai_protocol.py一处让 OpenAI、Azure、xAI 三方共同受益真正属于某 Provider 的差异才放进该 Provider 自己的模块。这条规则避免了修好 OpenAI 却漏掉 Azure的典型分裂。能力差异走 Profile 标志禁用 isinstance / 名字判断Provider 之间的能力差异必须通过RealtimeModelProfile的标志位来表达解析顺序为resolved defaults → provider → user profile严禁在使用点用isinstance或 Provider 名字字符串做分支。凡是依据某个 profile 标志分支的改动都需要一个把该标志两侧都钉死的测试。从 model.py 的profile属性实现可以看到完整的四级解析DEFAULT_REALTIME_PROFILEprofiles.py——为每个键提供基础值Provider 的realtime_model_profile(model_name)结果——Provider 专属默认来自 genai-prices 数据的最佳努力context_window除非 Provider 或用户 profile 显式设置过包括显式设为None用户的profile参数——部分 dict 合并覆盖或传入(resolved) - profile可调用对象实现整体替换后者可撤销 Provider 的错误声明。解析完成后supported_native_tools还会与当前模型类实际实现的原生工具集合取交集确保解析出的 profile 是什么可用的唯一事实来源。RealtimeModelProfileprofiles.py中值得注意的典型标志包括标志含义supports_image_input是否接受图像/视频帧输入supports_manual_turn_control是否支持手动轮转push-to-talksupports_interruption是否支持服务端打断响应barge-insupports_output_truncation是否支持把音频输出截断到用户实际听到的位置区别于打断xAI 支持打断但不支持截断supports_text_output是否支持纯文本输出默认为TrueGemini Live 与 xAI 为Falsesupports_webrtc是否支持浏览器 WebRTC 信令与侧带sideband会话仅 OpenAI 与 Azure OpenAIsupports_thinking是否支持推理/思考配置OpenAIgpt-realtime-2*、Gemini native-audio、xAI Grok Voiceemits_input_speech_events是否上报用户开始/停止说话事件注意是emits_而非supports_描述的是流中出现的事件Gemini Live 不发此类事件audio_input_sample_rate/audio_output_sample_ratePCM 音频采样率默认 24000 HzDEFAULT_AUDIO_SAMPLE_RATE一个事件一个含义在 codec 中归一化一个事件在每个 Provider 上只有一种含义是一条硬性规范。如果不同 Provider 对同一帧的含义理解不一致比如语音开始、轮转结束的判定必须在 codec 层归一化会话层不得按 Provider 分支。特别地RealtimeTurnCompleteEvent由会话合成绝不从线路上直接读取。这条规则与上一节形成互补Provider 差异要么进 profile 标志要么在 codec 归一化会话层始终保持 Provider 无关。错误包装约定连接/握手的失败包装进ModelAPIError/ModelHTTPError。RealtimeError正是ModelAPIError的子类model.py用于会话未能打开或已经结束这类与请求-响应 API 不可达同类的失败WebSocket 升级被拒绝时带有 HTTP 状态码则按普通请求一样抛ModelHTTPError。会话过程中可恢复的 Provider 错误变成RealtimeSessionErrorEvent并保持会话仍可用——这与一错即亡的连接层失败严格区分。历史保真会话历史永远是可以回放的合法输入RealtimeSession累积的历史必须是Agent.run(message_history...)的合法输入。这条契约由 _session.py 的实现方式保证助理语音被翻译成携带SpeechPartspeakerassistant的PartStartEvent/PartDeltaEvent/PartEndEvent在轮转结束时定稿为一个ModelResponse用户语音成为同样的 part 事件speakeruser定稿为一个ModelRequest工具调用变成ToolCallPartstart/end执行开始/结束对应FunctionToolCallEvent/FunctionToolResultEvent携带归一化的ToolReturnPart或RetryPromptPart。关键保证在于关闭与重连丢失时的结算settle语义不完整的回复被记录为被中断的响应interrupted responses运行中的工具被给予已取消的返回cancelled returns历史绝不能以悬空的ToolCallPart结尾——因为请求-响应 API 要求工具结果紧邻其调用_pending_tool_returns机制保证FunctionToolResultEvent即使晚到其结果 part 在all_messages()中也被放回携带调用的响应之后。工具结果在实时通道上是纯字符串的因此结构化的ToolReturnPart只在真正发送时被渲染成扁平形状见 codec 的 ToolResultOpenAI 协议连接会把附加用户内容作为后续会话条目发送Gemini 则在工具响应中包含文本回退。若 Provider 取消了在途调用会话会记录一条合成的中断返回以保持历史合法但不会把那个被遗弃的结果发给 Provider。测试方法论Cassette 录制、Parity 网络与脚本化示例实时模块的测试体系围绕三个约定展开实时 WebSocket Cassette存放在 tests/realtime/cassettes/原始帧、密钥已脱敏secrets scrubbed、音频已截断。录制命令为uv run --env-file .env pytest --record-moderewrite test跨 Provider 矩阵测试是 Parity 网络新增 Provider 行为时必须扩展它以此钉死同一事件在各方含义一致的规范。文档示例运行于脚本化的MockRealtimeConnectiontests/test_examples.py。默认脚本只说一句助理台词Hello from the realtime assistant.一旦 Agent 定义了名为check_availability的工具就会触发预留给 quickstart 的脚本化对话——因此其它地方不要使用check_availability这个工具名否则会意外触发保留剧本。这条约定同样反映在 docs/realtime/AGENTS.md 的文档页规则中。文档页章程Page Charters为了不让同一个概念在多处重复解释而漂移docs/realtime/AGENTS.md 为每页定义了专属主题规则只在属主页面陈述一次其它页面链接引用overview.md—— 门户页Provider 矩阵与限制表每个限制行都链接到跟踪 issueaudio.md—— 媒体 I/O音频、图像、转录events.md—— 事件词汇表及其与标准运行事件的重叠轮转边界规则在此turns.md、tools.md只讲工具、capabilities.md每个 hook 的支持情况、history.md、deployment.md前端传输、lifecycle.md只讲连接生命周期、observability.md、troubleshooting.md症状优先的索引各页 Edge cases 不得重复它四个 Provider 页安装、模型名、设置、怪癖的权威来源。文档页的通用规则还包括示例统一使用字符串模型形式agent.realtime(openai:gpt-realtime)、gateway/openai:gpt-realtime仅当演示模型级配置时才导入模型类安装块使用按 Provider 的 extrasopenai-realtime、google-realtime、xai-realtime完整示例依赖async with退出自动关闭会话除非示例本身讲的就是close()。从开发规范到应用视角关键 API 速览理解上述架构后应用层的关键入口可以快速对应到源码模型推断infer_realtime_model()model.py支持openai、azure、xai、googleGemini Developer API、google-cloudVertex AI以及gateway/openai/gateway/google网关路由非法格式缺provider:model分隔符抛UserError。连接层词汇RealtimeConnection.send接受RealtimeInput联合文本轮次、TextContext、BinaryAudio、BinaryImage与轮转控制动词CommitAudio/ClearAudio/CreateResponse/CancelResponse/TruncateOutput/ToolResult迭代连接得到RealtimeCodecEvent联合AudioDelta、OutputTranscript、InputTranscript、ToolCall、ResponseDone、SessionUsage等。会话层RealtimeSession 把上述 codec 事件翻译为用户可见的RealtimeEventPartStartEvent/PartDeltaEvent/PartEndEvent、FunctionToolCallEvent/FunctionToolResultEvent、RealtimeTurnCompleteEvent等并提供send()/send_audio()、stream_audio()、stream_transcripts()、interrupt()、commit_audio()等双工 API。共享设置RealtimeModelSettings 涵盖max_tokens、parallel_tool_calls、tool_choice、input_transcription_model、output_modality、thinking、turn_detection、handshake_timeout默认 30.0 秒、reconnectReconnectPolicymax_attempts默认 3、max_reconnects默认 50、base_delay0.5 秒指数退避、max_delay30.0 秒、默认开启抖动AudioRetention控制历史中音频的保留程度transcript_only/input_audio/output_audio/all默认为transcript_only保留的音频以 WAV 容器存储于SpeechPart.audio而实时增量始终是裸 PCM。一个值得注意的 fail-fast 设计是output_modalitytext对 profile 报告supports_text_outputFalse的模型Gemini Live、xAI会在连接前就抛UserError而不是让 Provider 握手失败Gemini 会答1007不支持 TEXT 模态或更糟地静默输出语音xAI——这与本指南策略在共享核心中裁决、Profile 是唯一事实来源的设计一脉相承。总结Pydantic AI Realtime 模块的工程核心可以浓缩为四条原则架构对称实时侧与请求-响应侧共享消息模型与工具核心、策略收敛安全相关的工具判定只在ToolManager.handle_call中裁决会话层绝不重实现、差异归一化Provider 差异要么进 profile 标志、要么在 codec 中归一化会话层保持 Provider 无关、历史可回放会话历史永远构成Agent.run(message_history...)的合法输入。对于想深入理解或扩展实时能力的开发者建议从 codec.py 与 _session.py 这对连接-会话核心入手再对照 openai.py 与 _openai_protocol.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），仅供参考