Pydantic AI 实时语音(Realtime)故障排查全指南:音频、轮次、打断、重连与工具卡顿问题详解 📅 发布时间:2026/9/14 14:52:00 👁 浏览次数: Pydantic AI 实时语音Realtime故障排查全指南音频、轮次、打断、重连与工具卡顿问题详解【免费下载链接】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 session场景下的症状优先symptom-first排障指南。围绕 docs/realtime/troubleshooting.md 归纳的常见问题——无声/无语音、模型不回复、重复回答、自我打断、问候语丢失、工具卡顿、重连丢上下文、Gemini 会话上限——逐一给出修复思路并结合 turns.md、audio.md、lifecycle.md 等文档及 pydantic_ai_slim/pydantic_ai/realtime/ 源码深入解释每个问题背后的底层机制。读完你将掌握如何用音频采样率、commit_audio()/create_response()、barge-in 处理、ReconnectPolicy与state_restored等核心 API 定位并修复实时语音应用中的典型故障。本文所有结论均以当前仓库实现为准。若你遇到的问题不在本文列表中可先查阅通用的 troubleshooting.md、在 help.md 中获取社区支持渠道或到项目 issue 区反馈。排查思路先看症状再对机制实时会话realtime session与普通请求-响应式模型调用不同它是一条持久化的双向连接语音的采集、播放、打断、重连、工具执行都发生在同一条链路上任何一个环节错位都可能表现为“听起来很奇怪”的症状。troubleshooting 文档建议的排查方式是按症状没有声音、模型不回答、重复回答……反向定位到对应的行为机制而不是逐个检查配置项。下表是全文的问题速查症状常见根因对应机制章节没有声音或听不到有效语音采集/播放采样率不一致或格式不是 mono PCM16音频线缆契约模型从不回复推按通话push-to-talk模式下缺少commit_audio()或create_response()推按通话模型重复说同一句话send(...)之后又调用create_response()触发两次回复文本轮次模型打断自己麦克风听到扬声器回音打断自己问候语不播放或访客一说话模型就回两次音频通路打开时扬声器回音/麦克风瞬态信号取消了问候语问候语丢失工具看起来卡住本地工具并发执行但提供方在等待工具结果时暂停了语音工具看起来卡住重连后丢失上下文重连时state_restored为 false重连丢上下文Gemini 达到会话上限未配置reconnect策略Gemini 会话上限没有声音或听不到有效语音实时链路中传输的是裸音频采样raw audio samples没有任何容器或编解码器包装发送端用send_audio()上传带符号 16 位小端序单声道 PCMsigned 16-bit little-endian mono PCM接收端用stream_audio()拿到的也是同一格式。因此“没有声音”最常见的两个原因分别是格式不匹配和采样率不匹配格式错误麦克风采集或扬声器播放的不是 mono PCM16例如是立体声、浮点采样或带 WAV 头的容器数据会话会直接解析成噪声或静音。采样率不一致采集端必须使用session.audio_input_sample_rate对应的采样率录制播放端必须使用session.audio_output_sample_rate对应的采样率播放——不要假设两者相等。在 源码实现 中这两个采样率来自模型的 profile缺省时回退到DEFAULT_AUDIO_SAMPLE_RATE24000 Hz。以 Gemini Live 为例输入为 16 kHz、输出为 24 kHz见 gemini.md如果按同一采样率采集和播放必然出现“音调不对”或“没有有效语音”。实操建议从 100 ms 的输入块开始平衡交互节奏与每块开销再根据你的传输通道微调具体模型的采样率与约束以各提供方页面为准OpenAI、Azure OpenAI、Google Gemini、xAI。完整的麦克风/扬声器回路含有界缓冲、播放进度核算、干净关闭可参考 examples/realtime-voice.md。模型从不回复如果应用处于推按通话push-to-talk模式——即用turn_detectionFalse关闭了自动语音活动检测VAD——那么发送音频只是把数据送进输入缓冲区不会触发任何回复。必须按顺序执行三步send_audio(...)发送音频commit_audio()结束用户轮次user turncreate_response()主动请求模型回复。from pydantic_ai import Agent from pydantic_ai.realtime.openai import OpenAIRealtimeModel, OpenAIRealtimeModelSettings agent Agent() model OpenAIRealtimeModel( gpt-realtime, settingsOpenAIRealtimeModelSettings(turn_detectionFalse) ) async def main(): async with agent.realtime(model).session() as session: await session.send_audio(b...) await session.commit_audio() await session.create_response()机制说明关闭自动检测后commit_audio()只负责“定稿”用户的输入模型端并没有收到回复指令只有显式调用create_response()才会让模型开口。这一点在 turns.md 中被反复强调也是“推按通话没声音”的经典根因——遗漏了commit_audio()或create_response()。另外可用clear_audio()丢弃尚未提交的输入。需要注意commit_audio()、clear_audio()、create_response()属于“手动轮次控制”能力只有 profile 声明supports_manual_turn_control的模型才支持OpenAI、Azure OpenAI、xAI 等。Gemini 不暴露手动轮次动词对 Gemini 设置turn_detectionFalse会在连接前直接抛出UserError详见 gemini.md。模型重复说同一句话session.send(...)发送字符串时本身就构成一个完整的用户轮次并请求模型回复见 turns.md。如果在它后面再调用create_response()就等于为同一段输入请求了第二次回复模型自然会把同一句话再说一遍。async def send_turns(session): await session.send(Greet the visitor.) # 已请求回复 # 只想补充上下文、不想要回复时用 respondFalse await session.send(The visitor is called Ada., respondFalse)排查要点检查调用链中是否出现“send()之后紧跟create_response()”的模式。若只是想给后续语音/文本轮次补充上下文而不要求回复使用respondFalse若想针对某段输入包括图片请求回复使用respondTrue。图片默认仅作上下文context-only要求对图片做出回复需要模型支持手动轮次控制。模型打断自己“模型说着说着自己停了/被打断”通常不是模型端的问题而是麦克风听到了扬声器的输出服务端 VAD 检测到“新的用户语音”于是中断了正在播放的回复。修复分为两步回声消除在设备层或 WebRTC 层为麦克风/扬声器回路增加回声消除echo cancellation避免扬声器输出被采集回去。真实打断时的本地清理在真正的 barge-in用户插话发生时及时停止本地已缓冲、用户再也听不到的音频播放。关于 barge-in 的完整处理见 turns.md当播放循环以设备节奏消费 session 唯一的stream_audio()迭代器时可打开handle_barge_inTrue由会话自动丢弃用户听不到的缓冲音频、将提供方侧转录截断到实际播放位置并取消当前回复也可自行监听事件并调用interrupt(played_bytes...)或interrupt(played_ms...)。后者在 源码 中与handle_barge_inTrue走同一套“flush-attribute-truncate-cancel”处理路径。问候语丢失典型场景代理先开口播放问候语但访客一说话问候语立刻被取消甚至出现“模型回复了两次”。troubleshooting 文档给出的根因是音频通路audio path打开期间扬声器回音或麦克风瞬态信号被 VAD 误判为用户语音从而取消了正在播放的问候语。服务端 VAD 默认启用interrupt_response任何被检测到的语音都会取消进行中的问候语。修复方法见 turns.md保持麦克风关闭直到问候语播放完毕再打开麦克风开始发送音频避免上述竞态不要用固定sleep来判断“问候语播完了”——应等待问候语对应的已定稿SpeechPart它会在生成完成后到达然后让播放循环排空后再开麦。import asyncio from collections.abc import AsyncIterator from pydantic_ai import Agent from pydantic_ai.messages import SpeechPart agent Agent(instructionsYou are a welcoming museum guide.) async def play_audio(chunks: AsyncIterator[bytes]) - None: async for chunk in chunks: ... # 将 PCM16 chunk 写入扬声器/音频输出流 async def wait_for_assistant_speech(parts: AsyncIterator[SpeechPart]) - None: async for part in parts: if part.speaker assistant: return async def main(): async with agent.realtime(openai:gpt-realtime).session() as session: playback asyncio.create_task(play_audio(session.stream_audio())) greeted asyncio.create_task(wait_for_assistant_speech(session.stream_transcripts())) await session.send(Greet the visitor.) await greeted ... # 等扬声器排空再打开麦克风开始发送音频 await playback # 会话关闭后音频视图结束确认根因的手段遍历事件流在第一个响应上查找RealtimeResponseInterruptedEventGemini 在服务端打断模型输出时发出并留意任何RealtimeSessionErrorEvent也可以检查 Logfire 追踪 确认是否有过早的“用户语音开始”事件。工具看起来卡住症状是模型已经调用了工具但语音长时间没有进展看起来像“卡住”。原因在于本地工具是并发执行的local tool runs concurrently但提供方可能在此期间暂停语音输出等待工具结果返回后再继续说话。这并非死锁而是提供方刻意为之的节奏。处理建议见 tools.md展示工具生命周期事件tool lifecycle events让用户/调用方直观看到工具正在执行而不是“卡死”审查工具是否真的耗时长、是否需要异步化或拆分成更快返回的步骤在支持异步工具调用的模型上如 Gemini 原生音频模型的google_async_tool_callsTrue可以让模型在工具执行期间继续说话但注意“快的工具结果”可能打断刚开口的语音并留下空的中断轮次见 gemini.md。重连丢上下文启用重连策略后连接掉线会自动重拨。此时必须检查RealtimeSessionReconnectEvent的state_restored字段见 lifecycle.mdstate_restoredFalse重连没有把对话完整地带过来某个轮次被截断。此时应主动开启一个全新对话fresh conversation避免在残缺的上下文中继续。state_restoredTrue但当前的语句utterance丢失了说明丢失的是重连时正在进行中的媒体in-flight media它不在被恢复的已完成轮次历史里——也就是说恢复机制只保证“已完成的轮次”存活正在说的那句话不保证。背后的机制分两类源码注释见 messages.py 中RealtimeSessionReconnectEvent的定义本地回放OpenAI、Azure OpenAI它们没有跨连接的服务器状态Pydantic AI 会把本地消息历史回放到新会话中。已完成的转录轮次保留进行中的音频不保留若掉线时恰好有回复在途会话会先把它结算为“被中断的响应”并取消进行中的工具调用再发出重连事件此时state_restoredFalse。原生会话恢复Gemini Live、xAI Grok Voice使用进程内in-memory的服务端会话句柄配置了reconnect策略时自动启用。xAI 下被掉线接住的回复在新连接上继续输出state_restored保持TrueGemini 在发出句柄前掉线则报告False并且把被截断的回复标记为中断响应后保持安静直到下一次输入。注意这些句柄只存在于内存中无法持久化到另一个进程。Gemini 达到会话上限Gemini 等提供方对单条连接时长设有上限。当会话因达到上限被断开时如果没有配置重连策略应用就会直接失败。解决办法是给reconnect设置一个ReconnectPolicyfrom pydantic_ai import Agent agent Agent() realtime agent.realtime( openai:gpt-realtime, model_settings{reconnect: {max_attempts: 5}}, )要点见 lifecycle.md 与 gemini.mdmax_attempts单次掉线允许的重拨次数默认3max_reconnects整个会话生命周期内允许的成功重连总次数默认50——这个“慷慨”的默认值正是为“提供方按时长上限断线、长会话在边界处合法续期”设计的如 OpenAI 60 分钟上限Gemini 的会话恢复session resumption会随重连策略自动启用重连使用掉线后最新的内存中服务器句柄并发出state_restoredTrue若在配置了策略的同时显式设置google_enable_session_resumptionFalse会抛出UserError而不是静默丢失对话已知限制Gemini 会在上限前发出GoAway但当前 Pydantic AI 只在连接真正掉线后才重连因此长通话可能在轮次中途出现短暂掉线见 gemini.md。附录排障时的关键诊断入口以下接口与事件是排查上述所有症状时最常用的“探针”均可从当前仓库源码中直接查看实现诊断目标使用/查找对象源码位置音频采样率session.audio_input_sample_rate/session.audio_output_sample_ratepydantic_ai_slim/pydantic_ai/realtime/_session.py播放进度session.played_audio_bytes需恰好一个stream_audio()消费者pydantic_ai_slim/pydantic_ai/realtime/_session.py用户开始说话RealtimeInputSpeechStartEventOpenAI/Azure/xAI 发出Gemini 不发出pydantic_ai_slim/pydantic_ai/messages.py响应被中断RealtimeResponseInterruptedEventGemini 发出pydantic_ai_slim/pydantic_ai/messages.py重连是否恢复上下文RealtimeSessionReconnectEvent.state_restoredpydantic_ai_slim/pydantic_ai/messages.py会话级错误RealtimeSessionErrorEventpydantic_ai_slim/pydantic_ai/messages.py模型能力开关RealtimeModelProfile的supports_*/emits_*标志pydantic_ai_slim/pydantic_ai/realtime/profiles.py重连策略ReconnectPolicymax_attempts/max_reconnectspydantic_ai_slim/pydantic_ai/realtime/settings.py排障时请始终遵循两个原则先确认症状对应的机制再动手改配置例如“没声音”先查采样率而不是先改 VAD用 profile 标志而不是 provider 名称分支——不同提供方对语音开始事件、中断事件的报告方式不同emits_input_speech_events只被 OpenAI 协议类提供方声明Gemini 则通过RealtimeResponseInterruptedEvent报告按标志判断才能写出可移植的实时语音代码。若问题仍未解决请结合 lifecycle.md 中的异常层级UserError、ModelHTTPError、RealtimeError、UsageLimitExceeded与 Logfire 可观测性 进一步定位。【免费下载链接】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),仅供参考