Pydantic AI Realtime WebRTC 语音智能体实战浏览器直连音频 服务端 Sideband 控制面架构【免费下载链接】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浏览器语音智能体通常面临两难选择让浏览器直连模型服务延迟最低但 API Key 面临泄露风险把所有音频都转发到后端中转密钥安全了却牺牲了实时性。Pydantic AI 官方示例examples/pydantic_ai_examples/realtime_webrtc给出了第三种、也是官方推荐的拓扑浏览器与 OpenAI 通过 WebRTC 直接交换音频最低延迟Pydantic AI Sideband 作为服务端控制面运行工具、构建对话历史并让 API Key 始终留在服务器。读完本文你将掌握这套架构的工作原理、完整可运行的部署步骤、关键环境变量与源码级实现细节以及如何把它扩展到手机 HTTPS 场景。为什么选择 Browser Server Sideband 拓扑在实时语音场景中音频路径上的每一跳都会增加可感知的延迟。本示例采用的核心思想是服务器绝不进入音频路径它只是控制平面。browser ──mic/speaker audio (WebRTC media)──▶ OpenAI Realtime ◀───────────────────────────────────── │ SDP offer (POST /offer) ▲ control WebSocket (call_id) ▼ │ FastAPI backend ──answer_webrtc_offer()──▶ OpenAI ──realtime_session(provider_session…)──┘ (relays SDP, gets call_id) (runs tools, builds history)流程分三层媒体面浏览器通过 WebRTC 与 OpenAI Realtime 直接建立媒体通道麦克风/扬声器音频不经过你的服务器信令面浏览器把 SDP offerPOST到后端/offer后端携带 API Key 与 OpenAI 完成信令协商取回 SDP answer 和call_id控制面后端用call_id挂接一条 Pydantic AI sideband 会话负责执行 Agent 的工具调用、维护对话历史、记录转录。这种拓扑带来的直接收益最低的音频延迟、API Key 不出服务器、Agent 能力工具、指令、历史依然完整可控。示例代码头部注释app.py明确指出该拓扑是浏览器语音智能体的推荐方案。快速运行三步起一个可对话的语音 Agent1. 准备环境变量在仓库根目录创建.env文件设置 OpenAI API KeyOPENAI_API_KEYsk-...可选配置项如下默认值均取自 app.py环境变量默认值说明OPENAI_API_KEY无OpenAI 密钥WebRTC 信令与 sideband 会话均由服务端持有它完成WEBRTC_REALTIME_MODELopenai:gpt-realtimeRealtime 模型标识经infer_realtime_model()解析Azure 场景需改为azure:deployment-nameWEBRTC_REALTIME_VOICEmarin语音音色通过OpenAIRealtimeModelSettings(openai_voiceVOICE)注入WEBRTC_TRANSCRIPTION_MODEL未设置Azure 输入转录部署名默认gpt-realtime-whisper名称不同时显式覆盖LOGFIRE_TOKEN未设置设置后启用 Logfire 可观测性不设置则if-token-present模式下不发送任何数据2. 启动服务uv run --all-packages uvicorn pydantic_ai_examples.realtime_webrtc.app:app3. 开始对话浏览器打开 http://localhost:8000localhost 属于安全上下文浏览器会放行麦克风权限点击Start call然后尝试“What time is it in Tokyo?” —— 触发服务端的lookup_time工具“Whats your refund policy?” —— 触发服务端的lookup_support_policy工具。页面左侧日志面板会实时显示你的转录You:、模型回复Assistant:以及模型发起的工具调用Model tool call:这些事件来自 WebRTC 数据通道上转发的oai-events见 index.html。架构拆解三条通道如何协同工作媒体面浏览器拥有音频传输前端 index.html 的start()完成以下动作navigator.mediaDevices.getUserMedia({ audio: true })采集麦克风创建RTCPeerConnection把音轨addTrack进连接ontrack把远端音频接到audio元素创建名为oai-events的 data channel用于接收并展示过滤后的提供方事件流createOffer()后setLocalDescription将 SDP offer 以Content-Type: application/sdp文本形式POST到/offer拿到返回的{ sdp, call_id }后setRemoteDescription完成握手进入Live — start talking状态。信令面/offer端点完成安全协商后端POST /offerapp.py是整条链路的枢纽对请求体做 UTF-8 解码与空值校验SDP 属于不可信信令输入畸形字节返回400而非500调用realtime.answer_webrtc_offer(sdp_offer)由服务端持有 API Key向 OpenAI 转交 offer拿到WebRTCAnswer后构造Call对象并登记到进程内CALLS字典先挂接 sideband 再返回 answer——因为浏览器只有拿到 answer 后才会开始说话必须确保工具在音频到来前已就绪。asyncio.wait_for(..., timeout10)限时 10 秒超时返回504sideband 挂接失败返回502避免把“死呼叫”的 answer 交给浏览器。控制面Sideband 会话运行 Agent 工具run_sideband()app.py是控制面的核心async with realtime.session(provider_sessioncall.provider_session) as session: call.attached.set() async for event in session: if isinstance(event, FunctionToolCallEvent): ... elif isinstance(event, FunctionToolResultEvent): ... elif isinstance(event, RealtimeTurnCompleteEvent): logfire.info(turn complete, messageslen(session.all_messages()))provider_sessioncall.provider_session让会话附着到已存在的 WebRTC 呼叫上不拥有音频传输会话迭代FunctionToolCallEvent/FunctionToolResultEvent/RealtimeTurnCompleteEvent等事件自动执行 Agent 已注册的工具并把结果回传给模型对话记录通过session.all_messages()保留在服务端。源码级原理Pydantic AI 如何支撑这套架构answer_webrtc_offer安全的服务端信令入口该方法定义于RealtimeModel基类model.py签名如下async def answer_webrtc_offer( self, sdp_offer: str, *, instructions: str | None None, tools: Sequence[ToolDefinition] | None None, model_settings: RealtimeModelSettings | None None, ) - WebRTCAnswer:它的定位是“安全信令路径”服务端代表浏览器完成 WebRTC 协商浏览器永远看不到任何令牌。返回值是WebRTCAnswermodel.py包含两个字段sdp提供方的 SDP answer送回浏览器作为 remote descriptionsession一个WebRTCSession交给 sideband 会话附着。WebRTCSessionmodel.py实现了RealtimeProviderSession协议——一个“传输无关”的契约sideband 只需知道provider_name校验附着模型与提供方匹配和session_id寻址控制面连接其中call_id是session_id在 OpenAI/Azure 线上协议名下的别名。需要注意并非所有 realtime 模型都支持 WebRTC。基类默认实现会抛出UserError提示“该模型不支持 WebRTC请先判断model.profile[supports_webrtc]或改用 WebSocket 传输”model.py。目前实现该能力的提供方是OpenAI 与 Azure OpenAI。AgentRealtime.sessionsideband 会话的挂接点agent.realtime(model).session(provider_sessioncall)返回一个异步生成器abstract.py其关键参数包括audio_retention会话在转录之外保留多少音频默认transcript_only丢弃音频字节。在 sideband 模式下必须保持该默认值handle_barge_in是否让会话自行处理打断的本地侧默认FalseWebRTC 场景下浏览器拥有播放路径应保持关闭需要时由自己调用interrupt()retain_images_every_n/retain_images_max会话期间图像在消息历史中的保留策略默认每 1 张保留、上限 100 张provider_session传入answer_webrtc_offer得到的WebRTCSession。设置后浏览器与提供方直连音频本会话只运行控制面指令、工具、转录、历史send_audio/commit_audio/clear_audio均不可用。这正是示例中 sideband 只跑工具与历史、不碰音频的根本原因——会话不拥有音频传输。测试佐证WebRTC 路径是被验证的一等公民仓库测试目录tests/realtime/下存在专门的test_webrtc.py测试文件与test_openai_ws.py、test_session.py等并列说明 WebRTC 信令 sideband 挂接是 Pydantic AI realtime 模块中被持续验证的正式能力而非示例独有的一次性代码。高级配置切换 Azure OpenAI示例不仅支持 OpenAI还支持 Azure OpenAI。切换到 Azure 需要设置WEBRTC_REALTIME_MODELazure:deployment-name设置AZURE_OPENAI_ENDPOINT与AZURE_OPENAI_API_KEY确保 Azure 资源中存在 realtime 部署对应azure:deployment-name段以及输入转录部署——转录部署默认名为gpt-realtime-whisper若你的部署命名不同用WEBRTC_TRANSCRIPTION_MODEL覆盖。关键点在于Azure 会针对你资源的deployments解析模型所以两个部署都必须真实存在sideband 会话才能完成转录与工具执行。在手机上使用HTTPS 与 Cloudflare 快速隧道麦克风只有在安全上下文localhost 或 HTTPS中才可用因此手机访问需要一个 HTTPS 入口。示例推荐无需注册账号的 Cloudflare 快速隧道cloudflared tunnel --url http://localhost:8000运行后打开打印出的https://....trycloudflare.com地址在手机上允许麦克风权限即可通话。挂断与清理本地资源释放优先前端stop()index.html体现了务实的容错设计先尝试POST /hangup/{call_id}通知后端取消 sideband 任务后端在 app.py 中取消任务并清理CALLS服务端挂断是尽力而为如果后端不可达finally块仍会关闭RTCPeerConnection、停止所有本地音轨并释放麦克风确保 WebRTC 连接与媒体资源不泄漏beforeunload时通过navigator.sendBeacon(/hangup/ callId)做最后的挂断尝试即使页面直接关闭也能发出请求。后端侧同样有兜底lifespan关闭时取消并等待所有残留的 sideband 任务app.py避免进程退出时悬挂后台 Agent 任务。可观测性用 Logfire 追踪每一次通话应用通过logfire.configure(send_to_logfireif-token-present, service_namerealtime-webrtc)与logfire.instrument_pydantic_ai()完成插桩app.py。在.env中设置LOGFIRE_TOKEN后即可在 Logfire 控制台看到realtime 会话的整体生命周期每个模型 turn 的事件流转RealtimeTurnCompleteEvent工具调用的名称与参数FunctionToolCallEvent及结果内容FunctionToolResultEvent。不设置 token 时不会发送任何遥测数据本地开发零负担。小结realtime_webrtc示例展示了 Pydantic AI 在实时语音领域的最佳实践拓扑媒体面直连、信令面由服务端代理、控制面由 Pydantic AI sideband 承担。三条通道各司其职让浏览器语音 Agent 同时获得低延迟、密钥安全与完整 Agent 能力。若想深入可以从 app.py 的answer_webrtc_offer调用点出发沿RealtimeModel.answer_webrtc_offer→WebRTCSession→AgentRealtime.session(provider_session...)的链路对照 model.py 与 abstract.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），仅供参考 Pydantic AI Realtime WebRTC 语音智能体实战：浏览器直连音频 + 服务端 Sideband 控制面架构
Pydantic AI Realtime WebRTC 语音智能体实战浏览器直连音频 服务端 Sideband 控制面架构【免费下载链接】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浏览器语音智能体通常面临两难选择让浏览器直连模型服务延迟最低但 API Key 面临泄露风险把所有音频都转发到后端中转密钥安全了却牺牲了实时性。Pydantic AI 官方示例examples/pydantic_ai_examples/realtime_webrtc给出了第三种、也是官方推荐的拓扑浏览器与 OpenAI 通过 WebRTC 直接交换音频最低延迟Pydantic AI Sideband 作为服务端控制面运行工具、构建对话历史并让 API Key 始终留在服务器。读完本文你将掌握这套架构的工作原理、完整可运行的部署步骤、关键环境变量与源码级实现细节以及如何把它扩展到手机 HTTPS 场景。为什么选择 Browser Server Sideband 拓扑在实时语音场景中音频路径上的每一跳都会增加可感知的延迟。本示例采用的核心思想是服务器绝不进入音频路径它只是控制平面。browser ──mic/speaker audio (WebRTC media)──▶ OpenAI Realtime ◀───────────────────────────────────── │ SDP offer (POST /offer) ▲ control WebSocket (call_id) ▼ │ FastAPI backend ──answer_webrtc_offer()──▶ OpenAI ──realtime_session(provider_session…)──┘ (relays SDP, gets call_id) (runs tools, builds history)流程分三层媒体面浏览器通过 WebRTC 与 OpenAI Realtime 直接建立媒体通道麦克风/扬声器音频不经过你的服务器信令面浏览器把 SDP offerPOST到后端/offer后端携带 API Key 与 OpenAI 完成信令协商取回 SDP answer 和call_id控制面后端用call_id挂接一条 Pydantic AI sideband 会话负责执行 Agent 的工具调用、维护对话历史、记录转录。这种拓扑带来的直接收益最低的音频延迟、API Key 不出服务器、Agent 能力工具、指令、历史依然完整可控。示例代码头部注释app.py明确指出该拓扑是浏览器语音智能体的推荐方案。快速运行三步起一个可对话的语音 Agent1. 准备环境变量在仓库根目录创建.env文件设置 OpenAI API KeyOPENAI_API_KEYsk-...可选配置项如下默认值均取自 app.py环境变量默认值说明OPENAI_API_KEY无OpenAI 密钥WebRTC 信令与 sideband 会话均由服务端持有它完成WEBRTC_REALTIME_MODELopenai:gpt-realtimeRealtime 模型标识经infer_realtime_model()解析Azure 场景需改为azure:deployment-nameWEBRTC_REALTIME_VOICEmarin语音音色通过OpenAIRealtimeModelSettings(openai_voiceVOICE)注入WEBRTC_TRANSCRIPTION_MODEL未设置Azure 输入转录部署名默认gpt-realtime-whisper名称不同时显式覆盖LOGFIRE_TOKEN未设置设置后启用 Logfire 可观测性不设置则if-token-present模式下不发送任何数据2. 启动服务uv run --all-packages uvicorn pydantic_ai_examples.realtime_webrtc.app:app3. 开始对话浏览器打开 http://localhost:8000localhost 属于安全上下文浏览器会放行麦克风权限点击Start call然后尝试“What time is it in Tokyo?” —— 触发服务端的lookup_time工具“Whats your refund policy?” —— 触发服务端的lookup_support_policy工具。页面左侧日志面板会实时显示你的转录You:、模型回复Assistant:以及模型发起的工具调用Model tool call:这些事件来自 WebRTC 数据通道上转发的oai-events见 index.html。架构拆解三条通道如何协同工作媒体面浏览器拥有音频传输前端 index.html 的start()完成以下动作navigator.mediaDevices.getUserMedia({ audio: true })采集麦克风创建RTCPeerConnection把音轨addTrack进连接ontrack把远端音频接到audio元素创建名为oai-events的 data channel用于接收并展示过滤后的提供方事件流createOffer()后setLocalDescription将 SDP offer 以Content-Type: application/sdp文本形式POST到/offer拿到返回的{ sdp, call_id }后setRemoteDescription完成握手进入Live — start talking状态。信令面/offer端点完成安全协商后端POST /offerapp.py是整条链路的枢纽对请求体做 UTF-8 解码与空值校验SDP 属于不可信信令输入畸形字节返回400而非500调用realtime.answer_webrtc_offer(sdp_offer)由服务端持有 API Key向 OpenAI 转交 offer拿到WebRTCAnswer后构造Call对象并登记到进程内CALLS字典先挂接 sideband 再返回 answer——因为浏览器只有拿到 answer 后才会开始说话必须确保工具在音频到来前已就绪。asyncio.wait_for(..., timeout10)限时 10 秒超时返回504sideband 挂接失败返回502避免把“死呼叫”的 answer 交给浏览器。控制面Sideband 会话运行 Agent 工具run_sideband()app.py是控制面的核心async with realtime.session(provider_sessioncall.provider_session) as session: call.attached.set() async for event in session: if isinstance(event, FunctionToolCallEvent): ... elif isinstance(event, FunctionToolResultEvent): ... elif isinstance(event, RealtimeTurnCompleteEvent): logfire.info(turn complete, messageslen(session.all_messages()))provider_sessioncall.provider_session让会话附着到已存在的 WebRTC 呼叫上不拥有音频传输会话迭代FunctionToolCallEvent/FunctionToolResultEvent/RealtimeTurnCompleteEvent等事件自动执行 Agent 已注册的工具并把结果回传给模型对话记录通过session.all_messages()保留在服务端。源码级原理Pydantic AI 如何支撑这套架构answer_webrtc_offer安全的服务端信令入口该方法定义于RealtimeModel基类model.py签名如下async def answer_webrtc_offer( self, sdp_offer: str, *, instructions: str | None None, tools: Sequence[ToolDefinition] | None None, model_settings: RealtimeModelSettings | None None, ) - WebRTCAnswer:它的定位是“安全信令路径”服务端代表浏览器完成 WebRTC 协商浏览器永远看不到任何令牌。返回值是WebRTCAnswermodel.py包含两个字段sdp提供方的 SDP answer送回浏览器作为 remote descriptionsession一个WebRTCSession交给 sideband 会话附着。WebRTCSessionmodel.py实现了RealtimeProviderSession协议——一个“传输无关”的契约sideband 只需知道provider_name校验附着模型与提供方匹配和session_id寻址控制面连接其中call_id是session_id在 OpenAI/Azure 线上协议名下的别名。需要注意并非所有 realtime 模型都支持 WebRTC。基类默认实现会抛出UserError提示“该模型不支持 WebRTC请先判断model.profile[supports_webrtc]或改用 WebSocket 传输”model.py。目前实现该能力的提供方是OpenAI 与 Azure OpenAI。AgentRealtime.sessionsideband 会话的挂接点agent.realtime(model).session(provider_sessioncall)返回一个异步生成器abstract.py其关键参数包括audio_retention会话在转录之外保留多少音频默认transcript_only丢弃音频字节。在 sideband 模式下必须保持该默认值handle_barge_in是否让会话自行处理打断的本地侧默认FalseWebRTC 场景下浏览器拥有播放路径应保持关闭需要时由自己调用interrupt()retain_images_every_n/retain_images_max会话期间图像在消息历史中的保留策略默认每 1 张保留、上限 100 张provider_session传入answer_webrtc_offer得到的WebRTCSession。设置后浏览器与提供方直连音频本会话只运行控制面指令、工具、转录、历史send_audio/commit_audio/clear_audio均不可用。这正是示例中 sideband 只跑工具与历史、不碰音频的根本原因——会话不拥有音频传输。测试佐证WebRTC 路径是被验证的一等公民仓库测试目录tests/realtime/下存在专门的test_webrtc.py测试文件与test_openai_ws.py、test_session.py等并列说明 WebRTC 信令 sideband 挂接是 Pydantic AI realtime 模块中被持续验证的正式能力而非示例独有的一次性代码。高级配置切换 Azure OpenAI示例不仅支持 OpenAI还支持 Azure OpenAI。切换到 Azure 需要设置WEBRTC_REALTIME_MODELazure:deployment-name设置AZURE_OPENAI_ENDPOINT与AZURE_OPENAI_API_KEY确保 Azure 资源中存在 realtime 部署对应azure:deployment-name段以及输入转录部署——转录部署默认名为gpt-realtime-whisper若你的部署命名不同用WEBRTC_TRANSCRIPTION_MODEL覆盖。关键点在于Azure 会针对你资源的deployments解析模型所以两个部署都必须真实存在sideband 会话才能完成转录与工具执行。在手机上使用HTTPS 与 Cloudflare 快速隧道麦克风只有在安全上下文localhost 或 HTTPS中才可用因此手机访问需要一个 HTTPS 入口。示例推荐无需注册账号的 Cloudflare 快速隧道cloudflared tunnel --url http://localhost:8000运行后打开打印出的https://....trycloudflare.com地址在手机上允许麦克风权限即可通话。挂断与清理本地资源释放优先前端stop()index.html体现了务实的容错设计先尝试POST /hangup/{call_id}通知后端取消 sideband 任务后端在 app.py 中取消任务并清理CALLS服务端挂断是尽力而为如果后端不可达finally块仍会关闭RTCPeerConnection、停止所有本地音轨并释放麦克风确保 WebRTC 连接与媒体资源不泄漏beforeunload时通过navigator.sendBeacon(/hangup/ callId)做最后的挂断尝试即使页面直接关闭也能发出请求。后端侧同样有兜底lifespan关闭时取消并等待所有残留的 sideband 任务app.py避免进程退出时悬挂后台 Agent 任务。可观测性用 Logfire 追踪每一次通话应用通过logfire.configure(send_to_logfireif-token-present, service_namerealtime-webrtc)与logfire.instrument_pydantic_ai()完成插桩app.py。在.env中设置LOGFIRE_TOKEN后即可在 Logfire 控制台看到realtime 会话的整体生命周期每个模型 turn 的事件流转RealtimeTurnCompleteEvent工具调用的名称与参数FunctionToolCallEvent及结果内容FunctionToolResultEvent。不设置 token 时不会发送任何遥测数据本地开发零负担。小结realtime_webrtc示例展示了 Pydantic AI 在实时语音领域的最佳实践拓扑媒体面直连、信令面由服务端代理、控制面由 Pydantic AI sideband 承担。三条通道各司其职让浏览器语音 Agent 同时获得低延迟、密钥安全与完整 Agent 能力。若想深入可以从 app.py 的answer_webrtc_offer调用点出发沿RealtimeModel.answer_webrtc_offer→WebRTCSession→AgentRealtime.session(provider_session...)的链路对照 model.py 与 abstract.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），仅供参考