Pydantic AI 实时摄像头智能体实战浏览器画面直播 语音对话的完整搭建与原理剖析【免费下载链接】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_camera示例文档源码位于 examples/pydantic_ai_examples/realtime_camera/app.py 与 index.html讲解如何构建一个「看着摄像头说话」的实时语音智能体它把麦克风音频和每秒一帧的摄像头画面流进 realtime 会话再把模型的语音回复播放出来并实时显示字幕。读完本文你将掌握provider 无关的实时会话接入、基于模型 profile 的 PCM 采样率协商、图像输入BinaryContent、turn_coverageall_input的实时视觉、Watch 主动播报、原生 Web 搜索引用、以及通过普通函数工具把绘图委托给第二个 Agent 的多智能体协作模式。示例概览它到底演示了什么该摄像头智能体把「看」和「说」组合成一套完整的演示覆盖了实时语音应用里最常见的几个技术点provider 无关的 realtime 会话采样率来自所选模型的 profile——Gemini 输入用 16 kHzOpenAI 与 Azure OpenAI 输入用 24 kHz无需手工指定图像输入通过BinaryContent发送 JPEG 帧模拟近似实时的视频流实时视觉turn_coverageall_input让每一帧摄像头画面都留在模型上下文中配合一个Watch开关实现主动播报普通函数工具 第二 Agentredraw_diagram工具把草图描述委托给独立的绘图Agent生成干净的 HTML 图表Web 搜索通过WebSearch能力启用返回可点击的引用来源模型选择器与 provider 感知的配置面板语音、输出模态、VAD 灵敏度、Gemini 专属设置一应俱全。正如 realtime 概述 所述realtime 会话复用了与标准文本 Agent 相同的工具、依赖、指令、消息历史、能力、用量限制与可观测性——本次调用过程中 Agent 可以像文本对话一样查订单、查可用性而通话本身会变成普通消息历史便于后续交给Agent.run()做摘要或结构化跟进。运行示例1. 准备凭据把所选模型的凭据写入仓库根目录的.env。默认模型是 Google Gemini所以最小配置只需要GOOGLE_API_KEYyour-google-api-key如果你要在模型选择器中切到 OpenAI 或 Azure OpenAI则对应设置OPENAI_API_KEY或AZURE_OPENAI_*系列变量。草稿重绘工具默认委托给独立的绘图 Agent——默认模型为google:gemini-3.5-flash复用同一个GOOGLE_API_KEY。可以用CAMERA_DRAW_MODEL换成任意凭据可覆盖的provider:model或用CAMERA_DRAWfalse完全禁用绘图其余功能不受影响。2. 安装依赖并启动按 examples 使用说明 安装依赖并设置环境变量后启动本地服务python/uv-run -m pydantic_ai_examples.realtime_camera.app服务会监听127.0.0.1:8000见 app.py 中的uvicorn.run(app, host127.0.0.1, port8000)。浏览器打开 http://localhost:8000点击Start允许摄像头与麦克风访问。localhost本身就是浏览器的安全上下文无需额外配置。3. 模型与采样率默认模型为google:gemini-3.1-flash-live-preview可通过CAMERA_REALTIME_MODEL修改或在页面设置面板中按会话切换到任意 Google、OpenAI、Azure OpenAI 的provider:modelxAI 的 realtime 不支持摄像头图像输入因此未列入选择器。采样率的协商是整套桥接的关键浏览器发送的是裸 PCM本身不携带采样率信息所以服务端会在麦克风采集开始前先把所选模型 profile 推导出的输入/输出采样率通过 JSON 通道下发见下文「桥接原理」。Gemini 输入 16 kHzOpenAI 与 Azure 输入 24 kHz——这些数值全部来自model.audio_input_sample_rate与model.audio_output_sample_rate代码里没有任何硬编码。警告请保持示例在本地运行WebSocket 使用服务端的 provider 凭据且没有任何用户认证。示例会校验浏览器 Origin 是否与它被提供的主机一致见 app.py 的_same_origin但这只是开发期防护不是生产级访问控制。 不要把服务暴露到 Cloudflare quick tunnel、ngrok 或公共反向代理之后。若要在其他设备上使用请部署在你自己可控的网络中置于认证与 TLS 之后并加上与你的环境匹配的用户级配额和速率限制。环境变量速查表以下是源码 app.py 中实际读取的全部CAMERA_*配置环境变量默认值作用CAMERA_REALTIME_MODELgoogle:gemini-3.1-flash-live-preview默认实时模型CAMERA_REALTIME_VOICE空用 provider 默认语音指定语音留空可避免 Gemini/OpenAI 语音名互相冲突CAMERA_TURN_COVERAGEall_inputGemini 的 turn coverageactivity_only/all_input/all_videoCAMERA_PROACTIVEfalseGemini native-audio 模型可保持沉默proactive audioCAMERA_AFFECTIVEfalseGemini 情感感知对话affective dialogCAMERA_WATCH_PROMPT见源码默认指令Watch 模式的提示词CAMERA_DRAWtrue是否注册redraw_diagram工具CAMERA_DRAW_MODELgoogle:gemini-3.5-flash绘图 Agent 的模型CAMERA_WEB_SEARCHtrue是否启用 Web 搜索能力模型支持时CAMERA_ALLOWED_ORIGINS空逗号分隔的额外允许 Origin如https://myapp.example.com另外GOOGLE_GENAI_USE_VERTEXAI、GOOGLE_CLOUD_PROJECT、GOOGLE_CLOUD_LOCATION用于切换 Vertex AILOGFIRE_TOKEN用于 Logfire 可观测性。布尔值解析支持1/true/yes/on任意大小写见_truthy函数app.py。Watch 模式让模型主动播报画面变化摄像头画面只提供视觉上下文本身不会触发模型回合。Watch 模式在模型空闲时周期性地发送一段短文本回合提示它报告视觉变化且不会打断正在进行的语音。export CAMERA_WATCH_PROMPTLook at the current camera view. In a few words, say whats changed since you last spoke; if nothing notable changed, stay silent.默认提示词即源码中的 WATCH_PROMPT。前端实现位于 index.html每WATCH_MS3500 ms检查一次仅当界面状态为listening模型空闲时才发送{type:nudge}消息——若在思考或说话时发送会打断进行中的回复相当于 barge-in导致音频被截断。服务端收到nudge后将其转换为WATCH_PROMPT文本发送进会话见_dispatch_textapp.py。Gemini native-audio 模型可以主动决定「没什么可说的」从而保持沉默export CAMERA_PROACTIVEtrue export CAMERA_AFFECTIVEtrueCAMERA_TURN_COVERAGE默认是all_input它在 Gemini Developer API 和 Vertex AI 上都可用更新的all_video在 Vertex 上尚不支持它把每一帧摄像头画面都保留在模型上下文中——这正是智能体持续「看着」现场场景的依据。需要注意Watch 模式开启期间会持续消耗 token。搜索与引用CAMERA_WEB_SEARCHtrue默认开启时只要所选模型的 profile 支持原生搜索示例就会为该会话添加WebSearch能力。判断逻辑在_web_search_supportedapp.pydef _web_search_supported(model: RealtimeModel) - bool: return WebSearchTool in model.profile.get(supported_native_tools, frozenset())也就是说切换模型后若不支持原生搜索能力会被静默摘除而不是让会话报错同时_instructions也会相应地在系统提示中加入「需要时搜索网页」的引导app.py避免告诉模型一个它没有的工具。原生工具返回事件会被转换成引用「chips」服务端从NativeToolReturnPart.content中提取{url, title}列表见_grounding_sourcesapp.py浏览器端只接受 HTTP(S) 协议的来源 URL最多显示前 6 条15 秒后自动淡出index.html。重绘草图把绘图委托给第二个 AgentCAMERA_DRAWtrue默认时实时 Agent 可以调用redraw_diagram工具。它把可见草图的详细文字描述交给一个独立的Agent名为diagram_drawer后者产出自包含的 HTML浏览器把这个 HTML 显示在 opaque-origin 的沙箱 iframe 中sandbox属性阻断脚本与同源访问CSP 进一步阻断网络加载并保留 PNG 导出按钮。关键实现细节app.py绘图 Agent 是懒加载的lru_cache 函数内创建app.py只有真正用到时才需要绘图模型的凭据model_settings{max_tokens: 16_384}显式抬高输出上限——复杂的图表 HTML 可能超出 provider 默认的max_tokens比如 Anthropic 的默认值偏低会把页面截断工具被取消时如用户中途打断、provider 放弃回合会清除浏览器的加载遮罩再重新抛出避免残留 loading 状态默认绘图模型选择「快而小」的google:gemini-3.5-flash因为用户正在实时通话中等待重绘延迟主要由 HTML 输出 token 主导更大的模型主要增加的是思考时间而非质量。绘图模型可独立配置export CAMERA_DRAW_MODELanthropic:claude-haiku-4-5绘图与 Web 搜索在所选实时模型同时支持时可以共存——工具是并发执行的所以重绘不会打断语音对话。前端接收drawing_started/drawing/drawing_error三类消息先显示加载遮罩收到 HTML 后在srcdoc中注入额外的 CSP 头default-src none; style-src unsafe-inline; img-src data:并用 SVGforeignObject方案把渲染结果光栅化为 PNG 下载index.html。Vertex AI不用 API Key 也能跑 Gemini当组织不允许使用 Gemini API key 时改用 Application Default Credentialsgcloud auth application-default login export GOOGLE_GENAI_USE_VERTEXAItrue export GOOGLE_CLOUD_PROJECTyour-project export GOOGLE_CLOUD_LOCATIONus-central1源码中_build_model会检测GOOGLE_GENAI_USE_VERTEXAI对google:开头的模型改用GoogleRealtimeModel(..., providergoogle-cloud)构建app.py。all_input的 turn coverage 在两条路径上都能正常工作。桥接原理双泵并发架构浏览器与 provider 之间由_run_session中的两个并发 pump 连接app.pybrowser ── PCM16 JPEG/text ──▶ FastAPI /ws ──▶ RealtimeSession browser ◀── PCM16 JSON events ──────────────── RealtimeSession握手顺序很关键麦克风采集开始前服务端先通过 JSON 通道发送session_config携带 profile 推导的音频输入/输出采样率app.py浏览器只有收到它之后才创建AudioContext以对应采样率并开始采集。随后入站 pumppump_inbound循环接收浏览器 WebSocket 帧。二进制帧PCM16 音频直接session.send_audio(chunk)文本帧交给_dispatch_text分派——imagebase64 解码为BinaryContent、text直接输入、nudgeWatch 触发事件 pumppump_events迭代async for event in session。SpeechPartDelta中的音频块以裸二进制帧回传transcript增量按说话人speaker流式进入字幕气泡其余一次性事件barge-in、grounding 引用、回合结束经_json_message转为 JSON 消息。任一端结束都会取消对方所在的 task grouptg.cancel_scope.cancel()从而干净地关闭会话。WebSocket 发送用anyio.Lock串行化——因为工具的emit可能与事件 pump 竞争发送app.py。值得注意的细节音频输入使用 16 位小端单声道裸 PCM无容器无编码send_audio接受单个 chunk 或异步迭代器事件 pump 中SpeechPartDelta的转录增量自带speaker所以用户和助手两侧的字幕可以同时流式更新而不需要回绑到PartStartEventapp.py。前端要点无构建工具的单页index.html 是一个零构建的浏览器单页关键参数集中在脚本头部index.htmlconst FRAME_MS 1000; // 约每秒 1 帧 const FRAME_W 512; // 流式帧缩放到 512px 宽控制带宽与成本 const WATCH_MS 3500; // Watch 模式下的检查间隔画面采集startFrames从getUserMedia视频流定时截帧缩放到FRAME_W后以 JPEG质量 0.6base64 发送{type:image}音频采集startMicScriptProcessor把 Float32 转成 Int16 PCM 后直接sock.send(i16.buffer)回声消除与降噪在getUserMedia约束中开启播放playPCMInt16 转 Float32按outputRate创建AudioBufferSource排队播放barge-in 时flushPlayback清空已排队的音频设置面板模型、语音、输出模态audio/text、Gemini 专属的 turn coverage / 起止灵敏度 / proactive / affective应用后通过 URL query 参数重建 WebSocket 连接/ws?...服务端注入默认值FastAPI 的/路由把环境变量默认值 JSON 转义后替换__DEFAULTS__占位符面板初始值因此与服务端一致app.py。output_modalitytext时模型回复以TextPartDelta形式流式进入同一个字幕气泡app.py无需改动任何前端逻辑。安全边界与可观测性示例明确把自身定位为本地开发演示_same_origin提供三重放行——回环 Origin 匹配Host直接本地使用、匹配X-Forwarded-Host可信反向代理因为浏览器 WebSocket API 无法发送自定义头、或命中CAMERA_ALLOWED_ORIGINS白名单app.py。不满足时以1008关闭连接并记录 Logfire 警告。可观测性方面app.py 在load_dotenv()之后配置 Logfiresend_to_logfireif-token-present保证没有LOGFIRE_TOKEN时什么都不发送、示例照常工作logfire.instrument_pydantic_ai()与logfire.instrument_fastapi(app)分别对实时会话、模型回合、工具调用和 HTTP 请求做埋点会话内 Agent 被命名为camera_assistant以便与diagram_drawer区分。重连策略在_build_model中统一设置ReconnectPolicy(max_attempts5)app.py。小结realtime_camera示例展示了 Pydantic AI realtime 能力的完整闭环从浏览器媒体采集、PCM 采样率协商、图像输入、turn coverage 实时视觉到函数工具并发执行、能力WebSearch按模型 profile 动态裁剪、多 Agent 委托再到事件驱动的浏览器渲染与安全边界设计。若要进一步深入可以继续阅读 realtime 事件模型、回合与打断、音频与图像输入 以及 连接生命周期完整的服务端与前端实现分别在 app.py 与 index.html配套说明见 realtime_camera README。【免费下载链接】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 实时摄像头智能体实战：浏览器画面直播 + 语音对话的完整搭建与原理剖析
Pydantic AI 实时摄像头智能体实战浏览器画面直播 语音对话的完整搭建与原理剖析【免费下载链接】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_camera示例文档源码位于 examples/pydantic_ai_examples/realtime_camera/app.py 与 index.html讲解如何构建一个「看着摄像头说话」的实时语音智能体它把麦克风音频和每秒一帧的摄像头画面流进 realtime 会话再把模型的语音回复播放出来并实时显示字幕。读完本文你将掌握provider 无关的实时会话接入、基于模型 profile 的 PCM 采样率协商、图像输入BinaryContent、turn_coverageall_input的实时视觉、Watch 主动播报、原生 Web 搜索引用、以及通过普通函数工具把绘图委托给第二个 Agent 的多智能体协作模式。示例概览它到底演示了什么该摄像头智能体把「看」和「说」组合成一套完整的演示覆盖了实时语音应用里最常见的几个技术点provider 无关的 realtime 会话采样率来自所选模型的 profile——Gemini 输入用 16 kHzOpenAI 与 Azure OpenAI 输入用 24 kHz无需手工指定图像输入通过BinaryContent发送 JPEG 帧模拟近似实时的视频流实时视觉turn_coverageall_input让每一帧摄像头画面都留在模型上下文中配合一个Watch开关实现主动播报普通函数工具 第二 Agentredraw_diagram工具把草图描述委托给独立的绘图Agent生成干净的 HTML 图表Web 搜索通过WebSearch能力启用返回可点击的引用来源模型选择器与 provider 感知的配置面板语音、输出模态、VAD 灵敏度、Gemini 专属设置一应俱全。正如 realtime 概述 所述realtime 会话复用了与标准文本 Agent 相同的工具、依赖、指令、消息历史、能力、用量限制与可观测性——本次调用过程中 Agent 可以像文本对话一样查订单、查可用性而通话本身会变成普通消息历史便于后续交给Agent.run()做摘要或结构化跟进。运行示例1. 准备凭据把所选模型的凭据写入仓库根目录的.env。默认模型是 Google Gemini所以最小配置只需要GOOGLE_API_KEYyour-google-api-key如果你要在模型选择器中切到 OpenAI 或 Azure OpenAI则对应设置OPENAI_API_KEY或AZURE_OPENAI_*系列变量。草稿重绘工具默认委托给独立的绘图 Agent——默认模型为google:gemini-3.5-flash复用同一个GOOGLE_API_KEY。可以用CAMERA_DRAW_MODEL换成任意凭据可覆盖的provider:model或用CAMERA_DRAWfalse完全禁用绘图其余功能不受影响。2. 安装依赖并启动按 examples 使用说明 安装依赖并设置环境变量后启动本地服务python/uv-run -m pydantic_ai_examples.realtime_camera.app服务会监听127.0.0.1:8000见 app.py 中的uvicorn.run(app, host127.0.0.1, port8000)。浏览器打开 http://localhost:8000点击Start允许摄像头与麦克风访问。localhost本身就是浏览器的安全上下文无需额外配置。3. 模型与采样率默认模型为google:gemini-3.1-flash-live-preview可通过CAMERA_REALTIME_MODEL修改或在页面设置面板中按会话切换到任意 Google、OpenAI、Azure OpenAI 的provider:modelxAI 的 realtime 不支持摄像头图像输入因此未列入选择器。采样率的协商是整套桥接的关键浏览器发送的是裸 PCM本身不携带采样率信息所以服务端会在麦克风采集开始前先把所选模型 profile 推导出的输入/输出采样率通过 JSON 通道下发见下文「桥接原理」。Gemini 输入 16 kHzOpenAI 与 Azure 输入 24 kHz——这些数值全部来自model.audio_input_sample_rate与model.audio_output_sample_rate代码里没有任何硬编码。警告请保持示例在本地运行WebSocket 使用服务端的 provider 凭据且没有任何用户认证。示例会校验浏览器 Origin 是否与它被提供的主机一致见 app.py 的_same_origin但这只是开发期防护不是生产级访问控制。 不要把服务暴露到 Cloudflare quick tunnel、ngrok 或公共反向代理之后。若要在其他设备上使用请部署在你自己可控的网络中置于认证与 TLS 之后并加上与你的环境匹配的用户级配额和速率限制。环境变量速查表以下是源码 app.py 中实际读取的全部CAMERA_*配置环境变量默认值作用CAMERA_REALTIME_MODELgoogle:gemini-3.1-flash-live-preview默认实时模型CAMERA_REALTIME_VOICE空用 provider 默认语音指定语音留空可避免 Gemini/OpenAI 语音名互相冲突CAMERA_TURN_COVERAGEall_inputGemini 的 turn coverageactivity_only/all_input/all_videoCAMERA_PROACTIVEfalseGemini native-audio 模型可保持沉默proactive audioCAMERA_AFFECTIVEfalseGemini 情感感知对话affective dialogCAMERA_WATCH_PROMPT见源码默认指令Watch 模式的提示词CAMERA_DRAWtrue是否注册redraw_diagram工具CAMERA_DRAW_MODELgoogle:gemini-3.5-flash绘图 Agent 的模型CAMERA_WEB_SEARCHtrue是否启用 Web 搜索能力模型支持时CAMERA_ALLOWED_ORIGINS空逗号分隔的额外允许 Origin如https://myapp.example.com另外GOOGLE_GENAI_USE_VERTEXAI、GOOGLE_CLOUD_PROJECT、GOOGLE_CLOUD_LOCATION用于切换 Vertex AILOGFIRE_TOKEN用于 Logfire 可观测性。布尔值解析支持1/true/yes/on任意大小写见_truthy函数app.py。Watch 模式让模型主动播报画面变化摄像头画面只提供视觉上下文本身不会触发模型回合。Watch 模式在模型空闲时周期性地发送一段短文本回合提示它报告视觉变化且不会打断正在进行的语音。export CAMERA_WATCH_PROMPTLook at the current camera view. In a few words, say whats changed since you last spoke; if nothing notable changed, stay silent.默认提示词即源码中的 WATCH_PROMPT。前端实现位于 index.html每WATCH_MS3500 ms检查一次仅当界面状态为listening模型空闲时才发送{type:nudge}消息——若在思考或说话时发送会打断进行中的回复相当于 barge-in导致音频被截断。服务端收到nudge后将其转换为WATCH_PROMPT文本发送进会话见_dispatch_textapp.py。Gemini native-audio 模型可以主动决定「没什么可说的」从而保持沉默export CAMERA_PROACTIVEtrue export CAMERA_AFFECTIVEtrueCAMERA_TURN_COVERAGE默认是all_input它在 Gemini Developer API 和 Vertex AI 上都可用更新的all_video在 Vertex 上尚不支持它把每一帧摄像头画面都保留在模型上下文中——这正是智能体持续「看着」现场场景的依据。需要注意Watch 模式开启期间会持续消耗 token。搜索与引用CAMERA_WEB_SEARCHtrue默认开启时只要所选模型的 profile 支持原生搜索示例就会为该会话添加WebSearch能力。判断逻辑在_web_search_supportedapp.pydef _web_search_supported(model: RealtimeModel) - bool: return WebSearchTool in model.profile.get(supported_native_tools, frozenset())也就是说切换模型后若不支持原生搜索能力会被静默摘除而不是让会话报错同时_instructions也会相应地在系统提示中加入「需要时搜索网页」的引导app.py避免告诉模型一个它没有的工具。原生工具返回事件会被转换成引用「chips」服务端从NativeToolReturnPart.content中提取{url, title}列表见_grounding_sourcesapp.py浏览器端只接受 HTTP(S) 协议的来源 URL最多显示前 6 条15 秒后自动淡出index.html。重绘草图把绘图委托给第二个 AgentCAMERA_DRAWtrue默认时实时 Agent 可以调用redraw_diagram工具。它把可见草图的详细文字描述交给一个独立的Agent名为diagram_drawer后者产出自包含的 HTML浏览器把这个 HTML 显示在 opaque-origin 的沙箱 iframe 中sandbox属性阻断脚本与同源访问CSP 进一步阻断网络加载并保留 PNG 导出按钮。关键实现细节app.py绘图 Agent 是懒加载的lru_cache 函数内创建app.py只有真正用到时才需要绘图模型的凭据model_settings{max_tokens: 16_384}显式抬高输出上限——复杂的图表 HTML 可能超出 provider 默认的max_tokens比如 Anthropic 的默认值偏低会把页面截断工具被取消时如用户中途打断、provider 放弃回合会清除浏览器的加载遮罩再重新抛出避免残留 loading 状态默认绘图模型选择「快而小」的google:gemini-3.5-flash因为用户正在实时通话中等待重绘延迟主要由 HTML 输出 token 主导更大的模型主要增加的是思考时间而非质量。绘图模型可独立配置export CAMERA_DRAW_MODELanthropic:claude-haiku-4-5绘图与 Web 搜索在所选实时模型同时支持时可以共存——工具是并发执行的所以重绘不会打断语音对话。前端接收drawing_started/drawing/drawing_error三类消息先显示加载遮罩收到 HTML 后在srcdoc中注入额外的 CSP 头default-src none; style-src unsafe-inline; img-src data:并用 SVGforeignObject方案把渲染结果光栅化为 PNG 下载index.html。Vertex AI不用 API Key 也能跑 Gemini当组织不允许使用 Gemini API key 时改用 Application Default Credentialsgcloud auth application-default login export GOOGLE_GENAI_USE_VERTEXAItrue export GOOGLE_CLOUD_PROJECTyour-project export GOOGLE_CLOUD_LOCATIONus-central1源码中_build_model会检测GOOGLE_GENAI_USE_VERTEXAI对google:开头的模型改用GoogleRealtimeModel(..., providergoogle-cloud)构建app.py。all_input的 turn coverage 在两条路径上都能正常工作。桥接原理双泵并发架构浏览器与 provider 之间由_run_session中的两个并发 pump 连接app.pybrowser ── PCM16 JPEG/text ──▶ FastAPI /ws ──▶ RealtimeSession browser ◀── PCM16 JSON events ──────────────── RealtimeSession握手顺序很关键麦克风采集开始前服务端先通过 JSON 通道发送session_config携带 profile 推导的音频输入/输出采样率app.py浏览器只有收到它之后才创建AudioContext以对应采样率并开始采集。随后入站 pumppump_inbound循环接收浏览器 WebSocket 帧。二进制帧PCM16 音频直接session.send_audio(chunk)文本帧交给_dispatch_text分派——imagebase64 解码为BinaryContent、text直接输入、nudgeWatch 触发事件 pumppump_events迭代async for event in session。SpeechPartDelta中的音频块以裸二进制帧回传transcript增量按说话人speaker流式进入字幕气泡其余一次性事件barge-in、grounding 引用、回合结束经_json_message转为 JSON 消息。任一端结束都会取消对方所在的 task grouptg.cancel_scope.cancel()从而干净地关闭会话。WebSocket 发送用anyio.Lock串行化——因为工具的emit可能与事件 pump 竞争发送app.py。值得注意的细节音频输入使用 16 位小端单声道裸 PCM无容器无编码send_audio接受单个 chunk 或异步迭代器事件 pump 中SpeechPartDelta的转录增量自带speaker所以用户和助手两侧的字幕可以同时流式更新而不需要回绑到PartStartEventapp.py。前端要点无构建工具的单页index.html 是一个零构建的浏览器单页关键参数集中在脚本头部index.htmlconst FRAME_MS 1000; // 约每秒 1 帧 const FRAME_W 512; // 流式帧缩放到 512px 宽控制带宽与成本 const WATCH_MS 3500; // Watch 模式下的检查间隔画面采集startFrames从getUserMedia视频流定时截帧缩放到FRAME_W后以 JPEG质量 0.6base64 发送{type:image}音频采集startMicScriptProcessor把 Float32 转成 Int16 PCM 后直接sock.send(i16.buffer)回声消除与降噪在getUserMedia约束中开启播放playPCMInt16 转 Float32按outputRate创建AudioBufferSource排队播放barge-in 时flushPlayback清空已排队的音频设置面板模型、语音、输出模态audio/text、Gemini 专属的 turn coverage / 起止灵敏度 / proactive / affective应用后通过 URL query 参数重建 WebSocket 连接/ws?...服务端注入默认值FastAPI 的/路由把环境变量默认值 JSON 转义后替换__DEFAULTS__占位符面板初始值因此与服务端一致app.py。output_modalitytext时模型回复以TextPartDelta形式流式进入同一个字幕气泡app.py无需改动任何前端逻辑。安全边界与可观测性示例明确把自身定位为本地开发演示_same_origin提供三重放行——回环 Origin 匹配Host直接本地使用、匹配X-Forwarded-Host可信反向代理因为浏览器 WebSocket API 无法发送自定义头、或命中CAMERA_ALLOWED_ORIGINS白名单app.py。不满足时以1008关闭连接并记录 Logfire 警告。可观测性方面app.py 在load_dotenv()之后配置 Logfiresend_to_logfireif-token-present保证没有LOGFIRE_TOKEN时什么都不发送、示例照常工作logfire.instrument_pydantic_ai()与logfire.instrument_fastapi(app)分别对实时会话、模型回合、工具调用和 HTTP 请求做埋点会话内 Agent 被命名为camera_assistant以便与diagram_drawer区分。重连策略在_build_model中统一设置ReconnectPolicy(max_attempts5)app.py。小结realtime_camera示例展示了 Pydantic AI realtime 能力的完整闭环从浏览器媒体采集、PCM 采样率协商、图像输入、turn coverage 实时视觉到函数工具并发执行、能力WebSearch按模型 profile 动态裁剪、多 Agent 委托再到事件驱动的浏览器渲染与安全边界设计。若要进一步深入可以继续阅读 realtime 事件模型、回合与打断、音频与图像输入 以及 连接生命周期完整的服务端与前端实现分别在 app.py 与 index.html配套说明见 realtime_camera README。【免费下载链接】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），仅供参考