Pydantic AI 浏览器实时语音 Agent：WebRTC 媒体直连 + 服务端 sideband 控制面的完整实现指南

Pydantic AI 浏览器实时语音 AgentWebRTC 媒体直连 服务端 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导读本文以 pydantic-ai 仓库中的 realtime-webrtc 示例 为主线讲解如何在浏览器中构建延迟最低的语音 Agent浏览器与 OpenAI / Azure OpenAI 的 Realtime 模型通过 WebRTC 直接交换麦克风与扬声器音频而 FastAPI 后端作为控制面control plane负责中继 SDP 协商、挂载 Pydantic AI sideband 会话并在服务端运行 Agent 工具、维护消息历史同时保证 API Key 永不落入客户端。读完本文你将掌握answer_webrtc_offersession(provider_session…)这套官方推荐的浏览器语音 Agent 拓扑并能够复现、配置与扩展这一完整可运行的示例。为什么选择浏览器直连 服务端 sideband拓扑在真实语音 Agent 场景中音频路径的延迟直接决定用户体验。若把浏览器麦克风音频先送到后端、再由后端转发给模型每一段往返都会叠加网络与转发开销。该示例采用的拓扑将媒体面与控制面彻底分离browser ──mic/speaker audio (WebRTC media)──▶ OpenAI / Azure OpenAI Realtime ◀───────────────────────────────────── │ SDP offer (POST /offer) ▲ control WebSocket (call_id) ▼ │ FastAPI backend ──answer_webrtc_offer()──▶ provider ──session(provider_session…)──┘ (relays the SDP, gets a call_id) (runs tools, builds history)浏览器 ↔ 模型WebRTC 媒体通道音频以最低延迟直达 provider后端从不介入音频流浏览器 ↔ 后端仅用于信令SDP offer/answer与挂断后端 ↔ 模型一条以call_id标识的控制 WebSocket sideband用于下发指令、运行工具、构建历史、记录用量。这与 部署指南 中推荐的浏览器路径完全一致也是官方文档中推荐浏览器语音 Agent 拓扑的理由媒体永远不经过你的服务器延迟保持最低而工具、历史、依赖与 provider 凭据全部保留在服务端。该示例同时演示了三大核心能力浏览器 WebRTC 服务端 sideband支持 OpenAIopenai:gpt-realtime与 Azure OpenAIazure:deployment-name两种 providerAgentRealtime.answer_webrtc_offer—— 在服务端中继浏览器的 SDP offer并把 Agent 的指令与工具定义烘焙进呼叫API Key 永远不会到达客户端agent.realtime(model).session(provider_session…)—— 通过呼叫的控制面运行 Agent 的工具详见 实时工具文档同时浏览器独占音频。环境准备与密钥配置首先需要一个具有 realtime 访问权限的OPENAI_API_KEY放在仓库根目录的.env文件中OPENAI_API_KEY...若改用Azure OpenAI则将WEBRTC_REALTIME_MODEL指向你的 realtime 部署WEBRTC_REALTIME_MODELazure:gpt-realtime AZURE_OPENAI_ENDPOINThttps://my-resource.openai.azure.com AZURE_OPENAI_API_KEY...注意Azure 需要一个输入转写input-transcription部署Azure 是按资源中的**部署deployments**解析模型的资源既需要上面azure:deployment-name指向的 realtime 部署也需要一个输入转写部署——因为 sideband 会话会把用户的每轮语音记录为转写文本见 部署指南的 sideband 说明。转写部署默认名为gpt-realtime-whisper如果你的转写部署名称不同请用WEBRTC_TRANSCRIPTION_MODEL指定其名称。环境变量一览变量默认值说明OPENAI_API_KEY无OpenAI realtime 访问密钥OpenAI 路径必填WEBRTC_REALTIME_MODELopenai:gpt-realtime实时模型标识Azure 路径写为azure:deployment-nameWEBRTC_REALTIME_VOICEmarin模型语音映射到OpenAIRealtimeModelSettings.openai_voiceWEBRTC_TRANSCRIPTION_MODELprovider 的auto选择输入转写部署名Azure 场景必须正确AZURE_OPENAI_ENDPOINT无Azure 资源端点Azure 路径必填AZURE_OPENAI_API_KEY无Azure 访问密钥Azure 路径必填LOGFIRE_TOKEN无设置后启用 Logfire 追踪可选见下文以上变量在 app.py 中通过os.getenv读取VOICE、模型标识、转写模型均支持覆盖其中模型标识交给infer_realtime_model()解析该函数定义于 realtime/model.py将provider:model形式的标识符解析为对应的RealtimeModel实例。启动示例安装依赖并设置好密钥后详见 示例安装说明启动服务uv run --all-packages uvicorn pydantic_ai_examples.realtime_webrtc.app:app打开 http://localhost:8000点击Start call允许麦克风权限然后尝试询问What time is it in Tokyo?或Whats your refund policy?—— 这两句话会触发服务端工具执行。手机访问需要安全上下文HTTPS浏览器只在localhost或 HTTPS 下授予麦克风权限。若要从其他设备打开示例请通过 Cloudflare quick tunnel 将本地服务暴露为 HTTPScloudflared tunnel --url http://localhost:8000然后使用打印出的https://....trycloudflare.com地址访问无需账号。服务端实现拆解信令、sideband 与工具整个服务端只有约 240 行代码全部位于 app.py。下面按职责拆解。1. Agent 与工具定义INSTRUCTIONS ( You are Roberto, a concise and friendly voice support assistant. Use lookup_time for time questions and lookup_support_policy for account or refund questions. Keep answers short and natural for speech. ) agent Agent(instructionsINSTRUCTIONS) agent.tool_plain def lookup_time(city: str) - str: ... agent.tool_plain def lookup_support_policy(topic: str) - str: ...Agent 使用普通函数工具agent.tool_plain一个查询城市当地时间内置伦敦、纽约、东京、悉尼、旧金山等示例城市一个返回预设的售后政策答案refund / return / password。指令中显式约束模型使用哪个工具并要求回答简短、自然、适合语音——这是语音 Agent 提示词设计的要点。realtime 会话的工具执行与标准 run 完全一致模型调用工具时 session 发出FunctionToolCallEvent参数校验通过后执行工具并返回结果FunctionToolResultEvent解析失败或抛出ModelRetry时产生重试提示详见 实时工具文档。2. 绑定实时模型与设置model infer_realtime_model(os.getenv(WEBRTC_REALTIME_MODEL, openai:gpt-realtime)) settings OpenAIRealtimeModelSettings(openai_voiceVOICE) if transcription_model : os.getenv(WEBRTC_TRANSCRIPTION_MODEL): settings[input_transcription_model] transcription_model realtime agent.realtime(model, model_settingssettings)infer_realtime_model按标识符解析模型默认openai:gpt-realtime语音通过OpenAIRealtimeModelSettings.openai_voice设置默认marin若设置WEBRTC_TRANSCRIPTION_MODEL则写入input_transcription_model让用户语音以转写形式进入历史sideband 会话需要输入转写才能在历史中保留用户语音见 部署指南 与 音频文档。3./offer端点中继 SDP 并先挂载 sidebandapp.post(/offer) async def offer(request: Request) - JSONResponse: sdp_offer (await request.body()).decode(utf-8) ... answer await realtime.answer_webrtc_offer(sdp_offer) call Call(answer_sdpanswer.sdp, provider_sessionanswer.session) CALLS[answer.session.call_id] call call.task asyncio.create_task(run_sideband(call)) ... return JSONResponse({sdp: call.answer_sdp, call_id: answer.session.call_id})关键流程读取浏览器 POST 的原始 SDP offer 文本调用realtime.answer_webrtc_offer(sdp_offer)该方法在 AgentRealtime 中实现解析 Agent 的指令、工具定义与设置并烘焙进呼叫再委托给 provider 的answer_webrtc_offerOpenAI 实现在 realtime/openai.pyAzure 实现在 realtime/azure.py两者都通过 provider 的 WebRTC calls API 创建呼叫返回值是WebRTCAnswer包含 provider 的 SDPanswer返回给浏览器完成握手与一个WebRTCSession句柄其call_id即 OpenAI/Azure 在Location头中返回的呼叫标识在返回 answer 之前先把 sideband 挂载起来asyncio.create_task(run_sideband(call))确保浏览器拿到 answer 开始说话时工具已经就绪若 10 秒内 sideband 未成功挂载则返回 504若挂载失败则返回 502避免answer 已返回但会话从未建立的假成功。4. sideband 会话运行工具与构建历史async def run_sideband(call: Call) - None: async with realtime.session(provider_sessioncall.provider_session) as session: call.attached.set() async for event in session: if isinstance(event, FunctionToolCallEvent): logfire.info(tool call, toolevent.part.tool_name, argsevent.part.args) elif isinstance(event, FunctionToolResultEvent): logfire.info(tool result, ...) elif isinstance(event, RealtimeTurnCompleteEvent): logfire.info(turn complete, messageslen(session.all_messages()))session(provider_session…)是 sideband 的核心见 AgentRealtime.session传入provider_session后该 session只运行控制面指令、工具、转写、历史浏览器的 WebRTC 连接独占音频。因此 sideband 会话中的send_audio()/commit_audio()/clear_audio()等方法不可用会直接报错且audio_retention必须保持默认的transcript_only——这一点在 部署指南 中有明确说明。工具调用由 session 自动执行FunctionToolCallEvent记录调用FunctionToolResultEvent记录结果RealtimeTurnCompleteEvent表示一轮结束此时session.all_messages()已持有完整对话历史。5./hangup端点与生命周期清理app.post(/hangup/{call_id}) async def hangup(call_id: str) - JSONResponse: call CALLS.get(call_id) if call is not None and call.task is not None: call.task.cancel() with suppress(asyncio.CancelledError): await call.task return JSONResponse({stopped: call is not None})浏览器挂断时调用/hangup/{call_id}服务端取消对应的 sideband 任务服务在 FastAPIlifespan关闭时会遍历CALLS取消所有残留任务app.py防止 provider 连接与后台任务泄漏若浏览器在拿到 answer 前断开/offer请求被取消服务端会主动取消 sideband 并从CALLS移除该呼叫因为客户端永远不会收到call_id、也就无法调用/hangupapp.py。浏览器端实现纯 HTML/JavaScript无构建步骤浏览器端是单个 index.html原生 Web API 即可完成全部工作采集麦克风navigator.mediaDevices.getUserMedia({ audio: true })建立 RTCPeerConnectionpc.addTrack()把麦克风轨道加入连接pc.ontrack把远端音频流接到audio autoplay播放信令协商pc.createOffer()→setLocalDescription→POST /offerContent-Type: application/sdpbody 为 SDP 文本→ 拿到{ sdp, call_id }→setRemoteDescription({ type: answer, sdp })。至此媒体通道建立完成事件展示创建名为oai-events的 data channel接收 provider 过滤后的事件流将用户转写 / 助手转写 / 模型工具调用实时渲染到页面日志区挂断清理点击 Stop 时POST /hangup/{call_id}尽力而为随后pc.close()、停止麦克风轨道、清空播放源页面卸载时用navigator.sendBeacon(/hangup/ callId)兜底通知服务端。前端刻意保持零依赖、零构建把整个 HTML 读入内存后由/路由直接返回app.py适合直接作为模板改造。可观测性Logfire 追踪示例内置了 Logfire 插桩app.pylogfire.configure(send_to_logfireif-token-present, service_namerealtime-webrtc) logfire.instrument_pydantic_ai()在.env中设置LOGFIRE_TOKEN后realtime 会话、模型轮次与工具调用会以 trace 形式呈现未设置 token 时不发送任何数据。sideband 中对工具调用、工具结果与轮次完成事件也分别打了结构化日志便于排查语音会话中的工具链路。从示例到生产注意事项安全边界浏览器是 provider 会话上的对等方可以发送 provider 原生控制事件。因此每个服务端工具都必须基于可信的deps授权而不是仅依赖模型收到的指令见 部署指南的安全说明。历史与隐私sideband 会话记录的是共享的 provider 会话浏览器可通过 data channel 读到其中的对话项。不要把机密历史 seed 进 sideband机密上下文应放在deps与工具逻辑中。打断barge-insideband 不拥有音频传输使用RealtimeOutputSpeechStartEvent/RealtimeOutputSpeechEndEvent做说话指示interrupt()会清空 provider 的对外 WebRTC 音频缓冲从而停止播放。Sideband 断线遵循 生命周期与重连规则一次干净的关闭会被视为浏览器挂断。Provider 限制answer_webrtc_offer目前由 OpenAI 与 Azure OpenAI 实时模型实现其他 provider 会抛出UserError可用supports_webrtcprofile 标志前置判断见 AgentRealtime 文档注释。Gemini Live 与 xAI 不支持该 sideband 通道应改用 浏览器 → 后端 WebSocket 中继 拓扑。参考文件索引示例文档docs/examples/realtime-webrtc.md服务端实现examples/pydantic_ai_examples/realtime_webrtc/app.py浏览器端实现examples/pydantic_ai_examples/realtime_webrtc/index.html拓扑原理与安全说明docs/realtime/deployment.md#browser-webrtc-server-sideband实时工具机制docs/realtime/tools.md核心 APIAgentRealtime、WebRTCSession/WebRTCAnswer、infer_realtime_model【免费下载链接】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），仅供参考