pydantic-ai Realtime 事件流完全指南掌握 RealtimeSession 的 16 类事件与轮次边界【免费下载链接】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体系中RealtimeSession不只是一个发送/接收音频的管道它还是一个完整的事件流生产者。对会话做异步迭代async for event in session即可拿到整个会话的全景事件流内容分片content parts、工具活动tool activity、轮次边界turn boundaries、自动重连reconnects与可恢复错误recoverable errors。本文以 docs/realtime/events.md 为主线结合 pydantic_ai_slim/pydantic_ai/realtime/_session.py 与 pydantic_ai_slim/pydantic_ai/messages.py 的源码实现完整讲解这 16 类事件的语义、共享事件与 Realtime 专属事件的边界、轮次边界的判定方法以及直接消费原始音频事件的高级用法。读完本文你将能够基于事件流编写控制流、构建字幕/播放逻辑并理解stream_audio()/stream_transcripts()视图与底层事件流的关系。1. 事件流是什么一次迭代掌控整个会话RealtimeSession实现了异步迭代协议迭代它得到的正是会话的事件流。它在源码中的定位是把底层 codec 事件翻译成pydantic_ai.messages中共享的消息/分片事件词汇表同时把对话进行中的普通ModelMessage历史累积起来见 pydantic_ai_slim/pydantic_ai/realtime/_session.py 的RealtimeSession类文档。关键设计点在于单一数据源高层视图stream_audio()和stream_transcripts()都是从这同一条事件流派生的受限视图绝大多数应用应该迭代会话以掌控流程而把媒体播放/字幕渲染交给这两个视图。两种事件来源合一事件流中的事件分为两类——来自pydantic_ai.messages的共享流事件AgentStreamEvent成员和只有会话才会发射的RealtimeEvent成员说话检测、打断、轮次完成、重连、可恢复错误。一个能力capability的事件流钩子能看到这两类事件在同一条流中流动详见 Capabilities and hooks。惰性启动__aiter__只有在首次迭代时才启动 receive pump且迭代不会销毁 pump——提前break不会影响资源生命周期__aexit__会在连接和工具集关闭前排空一切源码。从源码层面看会话在 _session.py 中显式声明了RealtimeEvent联合类型它是AgentStreamEvent的一个严格子集内容以共享的PartStartEvent/PartDeltaEvent/PartEndEvent流式传输携带SpeechPart与ToolCallPart工具执行以FunctionToolCallEvent/FunctionToolResultEvent表达内联延迟解析以DeferredToolRequestsEvent/DeferredToolResultsEvent表达其余为 realtime 控制面事件turn complete、speech start/end、interrupted、reconnect、error 等。import asyncio from pydantic_ai import Agent from pydantic_ai.realtime import RealtimeSession async def main() - None: async with Agent(openai:gpt-4o-realtime).realtime( openai:gpt-4o-realtime, ) as session: # 事件流就是async for event in session async for event in session: # 按类型分派处理见第 2 节的事件参考表 print(event) asyncio.run(main())注意__aiter__要求先以async with进入会话且一个会话同一时刻只能被一个迭代器消费否则会抛出UserError见 _session.py。测试代码也遵循同样的使用模式例如 tests/realtime/test_session.py 中的drain_events(session)就是[event async for event in session]。2. 事件参考16 类事件的完整语义表下表完整覆盖事件流中可能出现的事件类型原文档核心内容事件含义PartStartEvent一个语音、文本或工具分片part开始了。若收到相同index的多个PartStartEvent新的事件应完全替换旧事件。PartDeltaEvent增量式的语音音频/转写或文本内容。携带index与delta一个ModelResponsePartDelta。PartEndEvent一个已定稿的分片被保留的语音音频出现在这里而不是在 part start。携带完整的partModelResponsePart。FunctionToolCallEvent一个本地函数工具开始执行。它是ToolCallEvent的子类event_kindfunction_tool_call。FunctionToolResultEvent一个本地函数工具完成或返回了重试提示retry prompt。携带将作为UserPromptPart发送给模型的content。DeferredToolRequestsEvent一个内联能力处理器HandleDeferredToolCallshandler解决了延迟请求。每个被延迟的调用也会发射自己的FunctionToolCallEvent该事件额外携带批量的DeferredToolRequests便于流消费者判断哪些调用在等待交互。DeferredToolResultsEvent内联的延迟结果已就绪可进入正常的工具处理流程。被解析的调用随后走常规工具执行管线每个结果都会发射一个FunctionToolResultEvent。RealtimeInputSpeechStartEvent提供方检测到用户开始说话——前提是模型 profile 声明了emits_input_speech_events。非常适合用于 barge-in打断收到该事件就停止播放任何缓冲中的模型音频。OpenAI、Azure OpenAI 和 xAI 会报告说话开始Gemini Live 不报告。RealtimeInputSpeechEndEvent提供方检测到用户说话结束同样依赖emits_input_speech_events。适合作为处理中指示器用户的轮次已结束模型即将回应。RealtimeResponseInterruptedEvent提供方报告模型响应被打断。它在响应终止符之前到达是冲刷缓冲模型音频的时机。Gemini Live 在听到用户说话时会服务端打断并报告该事件其他提供方则报告RealtimeInputSpeechStartEvent并把取消交给interrupt()因此从不报告此事件。RealtimeInputTranscriptionErrorEvent一个用户轮次无法被转写但会话仍可继续使用。它携带message、type、code和定位用户的item_id/content_index是可恢复事件。RealtimeOutputSpeechStartEvent/RealtimeOutputSpeechEndEvent模型变得可闻 / 停止可闻。这两个事件只在 WebRTC sideband 上发射——那里由提供方持有音频播放权。注意这是关于播放而非生成的事件提供方生成音频远快于播放因此它可能在音频生成后很久才到达OutputSpeechEndEvent才是说完话的真实终点。RealtimeTurnCompleteEvent模型回复结束且没有工具仍在执行。由会话在无工具调用运行中且无响应在途时合成用户输入转写可能在此事件之后才完成。RealtimeSessionReconnectEvent连接被自动重新建立。每次重连都会恢复会话配置instructions、工具、语音等对话状态则要么由提供方原生恢复Gemini Live 开启时、xAI Grok Voice要么由会话把本地历史重放进新的服务端会话OpenAI/Azure OpenAI。携带state_restored字段指示是否完整延续对话。RealtimeSessionErrorEvent一个可恢复的提供方错误发生会话仍可使用。携带message、type、code与recoverable标志协议error可恢复连接断开则不可恢复。2.1 事件在源码中的翻译路径这些事件并非凭空产生。会话的 pump 循环在_handle_pump_event中接收底层 codec 事件AudioDelta、OutputTranscript、InputTranscript、ResponseDone、ToolCall等其中工具调用和用量统计先被剥离处理其余交由_translate_event翻译成共享事件并顺带构建历史_session.py。例如AudioDelta/OutputTranscript/InputTranscript变成携带SpeechPart的PartStartEvent/PartDeltaEvent/PartEndEventResponseDone触发_handle_turn_complete最终产出RealtimeTurnCompleteEventToolCall变成ToolCallPartstart/end加上执行期间的FunctionToolCallEvent/FunctionToolResultEventRealtimeInputSpeechStartEvent、RealtimeSessionReconnectEvent、RealtimeOutputSpeechStartEvent/EndEvent属于控制面事件原样透传RealtimeSessionErrorEvent可恢复时作为事件上抛给消费者用于可观测性不可恢复时直接抛出RealtimeError终止会话。翻译函数以assert_never收尾保证将来新增的变体在类型检查阶段就会被发现。3. 共享事件与 Realtime 专属事件的边界事件流中的前七行PartStartEvent到DeferredToolResultsEvent是AgentStreamEvent的成员来自pydantic_ai.messages——与标准流式运行产出的事件完全相同。这意味着为文本 Agent 编写的事件处理代码渲染分片、记录工具调用等可以原封不动地用于实时会话无需任何改动。其余带Realtime*前缀的事件则是RealtimeEvent成员只有会话才会发射说话检测speech detection、打断interruption、轮次完成turn completion、重连reconnection和可恢复错误在请求-响应式运行request-response run中没有对应物属于实时会话的专属语义。对能力的事件流钩子而言两类事件流过同一条流都会可见Agent.realtime会把wrap_event_stream包装器应用到迭代器上包装器词汇表是AgentStreamEvent超集但应用于实时流时必须产出RealtimeEvent子集见 _session.py。4. 轮次边界用 RealtimeTurnCompleteEvent 判断这一轮结束了在实时对话中判断模型是否已经回答完毕远比想象中复杂。模型可以先说话、再调用工具、然后再说话——因此收到语音或收到工具结果都不意味着本轮结束。官方推荐的做法是把RealtimeTurnCompleteEvent作为一次交换exchange的边界。它在会话确认无工具调用仍在运行、且没有响应在途时由会话合成是唯一可靠的本轮结束信号。async for event in session: if isinstance(event, RealtimeTurnCompleteEvent): # 本轮真正结束模型回复完毕且没有工具还在执行 break # 或在此推进对话状态机这一用法在测试中也有直接体现tests/realtime/test_session.py 用isinstance(event, RealtimeTurnCompleteEvent)作为循环终止条件之后再await asyncio.sleep(0)数次让后台工具任务收尾。两个需要留意的补充语义来自源码事件类文档转写可能晚到输入转写可以在RealtimeTurnCompleteEvent之后才完成messages.py。播放可能晚到在 WebRTC sideband 上提供方的播放可以持续到RealtimeOutputSpeechEndEvent才结束。因此如果要驱动正在说话指示器应使用OutputSpeechStart/End这对事件而不是用 turn completemessages.py。另外从 _session.py 的注释可以看到只有交换边界被标记每个 response 本身已经是chatspan而模型真正完成的轮次边界没有自己的 span所以RealtimeTurnCompleteEvent是追踪中唯一标记轮次边界的事件在被打断的边界上事件显示文本还会说明被打断。5. 读取原始音频事件流媒体与底层事件的关系5.1 首选方式stream_audio() 受限视图音频流本质上就是这些事件——stream_audio()是对语音分片增量speech part deltas的受限视图bounded view绝大多数应用应当使用它而不是直接解析底层事件。它只包含实时的模型音频、按播放顺序排列绝不重复来自已定稿语音分片的保留音频源码。它的几个关键实现细节对编写播放循环很有用订阅即开始订阅在调用stream_audio()时就注册调用与消费者首次迭代之间产生的音频会被缓冲而不是丢失每个迭代器有 32 个数据块的缓冲_AUDIO_TAP_SIZE 32。落后不阻塞如果消费者跟不上或从不开始消费最旧的数据块会被丢弃drop-oldest保证音频播放永远不会拖垮工具执行、轮次跟踪或主事件流。设备时钟对账设备节奏的消费者喂入played_audio_bytes在 barge-in 时interrupt(played_bytes...)会丢弃用户永远不会听到的缓冲块。WebRTC sideband 不可用在 sideband 上浏览器持有音频路径stream_audio()会抛出UserError应改消费浏览器的远端媒体轨道源码。同理stream_transcripts()是另一个受限视图默认产出每个已完成轮次的SpeechPart含speaker与完整transcript传deltaTrue则产出TranscriptUpdate携带新文本、该轮次到目前为止的完整转写、说话人以及标识轮次的index两者使用独立的订阅互不挤占源码。TranscriptUpdate的完整字段定义见 _session.py。5.2 高级替代直接从 PartDeltaEvent 读取 SpeechPartDelta.audio_chunk作为高级用法也可以绕过视图直接播放原始PartDeltaEvent中携带的SpeechPartDelta.audio_chunk。适用场景是你需要把音频与其它事件如工具调用、转写在同一条流中按序处理而不想额外订阅stream_audio()。from pydantic_ai.messages import PartDeltaEvent, SpeechPartDelta async for event in session: if isinstance(event, PartDeltaEvent) and isinstance(event.delta, SpeechPartDelta): chunk event.delta.audio_chunk # 原始 PCM 音频块可送入声卡/WebSocket if chunk: await player.write(chunk)需要注意的两个不要重复播放陷阱模型音频无论是否启用历史保留都会完整到达也就是说即使audio_retention关闭了输出保留事件流中的音频块也是完整的可以直接播放。启用输出音频保留时最终 SpeechPart 会再包含一次整轮音频的 WAV 快照用于历史保留。此时不要两处都播放否则本轮会被播放两遍。正确做法是播放时只用实时事件流或stream_audio()把保留的 WAV 快照留给历史/回放场景。从源码看实时SpeechPartDelta.audio_chunk是原始 PCM而保留历史使用 WAV 容器以便自描述、可移植到经典模型适配器_session.py二者在会话内部本就是两套数据。6. 总结事件流驱动的实时应用骨架把上述内容组合起来一个典型的实时语音应用控制流如下async with Agent(...).realtime(...) as session:进入会话async for event in session:迭代事件流按类型分派PartStart/Delta/EndEvent→ 渲染分片 / 增量文本文本输出或交给视图FunctionToolCallEvent/FunctionToolResultEvent→ 更新工具状态 UI / 记录日志RealtimeInputSpeechStartEvent→ barge-in停止播放缓冲音频RealtimeInputSpeechEndEvent→ 显示处理中指示器RealtimeTurnCompleteEvent→ 推进对话状态机本轮真正结束RealtimeOutputSpeechStart/EndEventWebRTC sideband→ 驱动正在说话指示器RealtimeSessionReconnectEvent→ 依据state_restored决定是否提示用户会话连续性RealtimeSessionErrorEvent/RealtimeInputTranscriptionErrorEvent→ 可恢复错误处理会话继续可用媒体播放与字幕分别交给stream_audio()/stream_transcripts()视图两者都是同一事件流的受限投影绝不会与主循环竞争或阻塞。由于前七类事件与标准AgentStreamEvent完全同源文本 Agent 的事件处理代码可以无缝复用这也是 pydantic-ai 实时能力typed end to end的体现——同一套事件词汇贯穿流式文本与实时语音两种运行模式。相关资源本文主线文档docs/realtime/events.md会话与事件翻译实现pydantic_ai_slim/pydantic_ai/realtime/_session.py事件类定义pydantic_ai_slim/pydantic_ai/messages.py视图文档Audio, images, and transcripts、Retaining audio事件流钩子与能力Event stream hooks、The event streamWebRTC sideband 部署docs/realtime/deployment.md#browser-webrtc-server-sideband会话行为测试tests/realtime/test_session.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），仅供参考 
长河君 · 2026/9/14 18:07:47 · 阅读 1.2k pydantic-ai Realtime 事件流完全指南：掌握 RealtimeSession 的 16 类事件与轮次边界
pydantic-ai Realtime 事件流完全指南掌握 RealtimeSession 的 16 类事件与轮次边界【免费下载链接】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体系中RealtimeSession不只是一个发送/接收音频的管道它还是一个完整的事件流生产者。对会话做异步迭代async for event in session即可拿到整个会话的全景事件流内容分片content parts、工具活动tool activity、轮次边界turn boundaries、自动重连reconnects与可恢复错误recoverable errors。本文以 docs/realtime/events.md 为主线结合 pydantic_ai_slim/pydantic_ai/realtime/_session.py 与 pydantic_ai_slim/pydantic_ai/messages.py 的源码实现完整讲解这 16 类事件的语义、共享事件与 Realtime 专属事件的边界、轮次边界的判定方法以及直接消费原始音频事件的高级用法。读完本文你将能够基于事件流编写控制流、构建字幕/播放逻辑并理解stream_audio()/stream_transcripts()视图与底层事件流的关系。1. 事件流是什么一次迭代掌控整个会话RealtimeSession实现了异步迭代协议迭代它得到的正是会话的事件流。它在源码中的定位是把底层 codec 事件翻译成pydantic_ai.messages中共享的消息/分片事件词汇表同时把对话进行中的普通ModelMessage历史累积起来见 pydantic_ai_slim/pydantic_ai/realtime/_session.py 的RealtimeSession类文档。关键设计点在于单一数据源高层视图stream_audio()和stream_transcripts()都是从这同一条事件流派生的受限视图绝大多数应用应该迭代会话以掌控流程而把媒体播放/字幕渲染交给这两个视图。两种事件来源合一事件流中的事件分为两类——来自pydantic_ai.messages的共享流事件AgentStreamEvent成员和只有会话才会发射的RealtimeEvent成员说话检测、打断、轮次完成、重连、可恢复错误。一个能力capability的事件流钩子能看到这两类事件在同一条流中流动详见 Capabilities and hooks。惰性启动__aiter__只有在首次迭代时才启动 receive pump且迭代不会销毁 pump——提前break不会影响资源生命周期__aexit__会在连接和工具集关闭前排空一切源码。从源码层面看会话在 _session.py 中显式声明了RealtimeEvent联合类型它是AgentStreamEvent的一个严格子集内容以共享的PartStartEvent/PartDeltaEvent/PartEndEvent流式传输携带SpeechPart与ToolCallPart工具执行以FunctionToolCallEvent/FunctionToolResultEvent表达内联延迟解析以DeferredToolRequestsEvent/DeferredToolResultsEvent表达其余为 realtime 控制面事件turn complete、speech start/end、interrupted、reconnect、error 等。import asyncio from pydantic_ai import Agent from pydantic_ai.realtime import RealtimeSession async def main() - None: async with Agent(openai:gpt-4o-realtime).realtime( openai:gpt-4o-realtime, ) as session: # 事件流就是async for event in session async for event in session: # 按类型分派处理见第 2 节的事件参考表 print(event) asyncio.run(main())注意__aiter__要求先以async with进入会话且一个会话同一时刻只能被一个迭代器消费否则会抛出UserError见 _session.py。测试代码也遵循同样的使用模式例如 tests/realtime/test_session.py 中的drain_events(session)就是[event async for event in session]。2. 事件参考16 类事件的完整语义表下表完整覆盖事件流中可能出现的事件类型原文档核心内容事件含义PartStartEvent一个语音、文本或工具分片part开始了。若收到相同index的多个PartStartEvent新的事件应完全替换旧事件。PartDeltaEvent增量式的语音音频/转写或文本内容。携带index与delta一个ModelResponsePartDelta。PartEndEvent一个已定稿的分片被保留的语音音频出现在这里而不是在 part start。携带完整的partModelResponsePart。FunctionToolCallEvent一个本地函数工具开始执行。它是ToolCallEvent的子类event_kindfunction_tool_call。FunctionToolResultEvent一个本地函数工具完成或返回了重试提示retry prompt。携带将作为UserPromptPart发送给模型的content。DeferredToolRequestsEvent一个内联能力处理器HandleDeferredToolCallshandler解决了延迟请求。每个被延迟的调用也会发射自己的FunctionToolCallEvent该事件额外携带批量的DeferredToolRequests便于流消费者判断哪些调用在等待交互。DeferredToolResultsEvent内联的延迟结果已就绪可进入正常的工具处理流程。被解析的调用随后走常规工具执行管线每个结果都会发射一个FunctionToolResultEvent。RealtimeInputSpeechStartEvent提供方检测到用户开始说话——前提是模型 profile 声明了emits_input_speech_events。非常适合用于 barge-in打断收到该事件就停止播放任何缓冲中的模型音频。OpenAI、Azure OpenAI 和 xAI 会报告说话开始Gemini Live 不报告。RealtimeInputSpeechEndEvent提供方检测到用户说话结束同样依赖emits_input_speech_events。适合作为处理中指示器用户的轮次已结束模型即将回应。RealtimeResponseInterruptedEvent提供方报告模型响应被打断。它在响应终止符之前到达是冲刷缓冲模型音频的时机。Gemini Live 在听到用户说话时会服务端打断并报告该事件其他提供方则报告RealtimeInputSpeechStartEvent并把取消交给interrupt()因此从不报告此事件。RealtimeInputTranscriptionErrorEvent一个用户轮次无法被转写但会话仍可继续使用。它携带message、type、code和定位用户的item_id/content_index是可恢复事件。RealtimeOutputSpeechStartEvent/RealtimeOutputSpeechEndEvent模型变得可闻 / 停止可闻。这两个事件只在 WebRTC sideband 上发射——那里由提供方持有音频播放权。注意这是关于播放而非生成的事件提供方生成音频远快于播放因此它可能在音频生成后很久才到达OutputSpeechEndEvent才是说完话的真实终点。RealtimeTurnCompleteEvent模型回复结束且没有工具仍在执行。由会话在无工具调用运行中且无响应在途时合成用户输入转写可能在此事件之后才完成。RealtimeSessionReconnectEvent连接被自动重新建立。每次重连都会恢复会话配置instructions、工具、语音等对话状态则要么由提供方原生恢复Gemini Live 开启时、xAI Grok Voice要么由会话把本地历史重放进新的服务端会话OpenAI/Azure OpenAI。携带state_restored字段指示是否完整延续对话。RealtimeSessionErrorEvent一个可恢复的提供方错误发生会话仍可使用。携带message、type、code与recoverable标志协议error可恢复连接断开则不可恢复。2.1 事件在源码中的翻译路径这些事件并非凭空产生。会话的 pump 循环在_handle_pump_event中接收底层 codec 事件AudioDelta、OutputTranscript、InputTranscript、ResponseDone、ToolCall等其中工具调用和用量统计先被剥离处理其余交由_translate_event翻译成共享事件并顺带构建历史_session.py。例如AudioDelta/OutputTranscript/InputTranscript变成携带SpeechPart的PartStartEvent/PartDeltaEvent/PartEndEventResponseDone触发_handle_turn_complete最终产出RealtimeTurnCompleteEventToolCall变成ToolCallPartstart/end加上执行期间的FunctionToolCallEvent/FunctionToolResultEventRealtimeInputSpeechStartEvent、RealtimeSessionReconnectEvent、RealtimeOutputSpeechStartEvent/EndEvent属于控制面事件原样透传RealtimeSessionErrorEvent可恢复时作为事件上抛给消费者用于可观测性不可恢复时直接抛出RealtimeError终止会话。翻译函数以assert_never收尾保证将来新增的变体在类型检查阶段就会被发现。3. 共享事件与 Realtime 专属事件的边界事件流中的前七行PartStartEvent到DeferredToolResultsEvent是AgentStreamEvent的成员来自pydantic_ai.messages——与标准流式运行产出的事件完全相同。这意味着为文本 Agent 编写的事件处理代码渲染分片、记录工具调用等可以原封不动地用于实时会话无需任何改动。其余带Realtime*前缀的事件则是RealtimeEvent成员只有会话才会发射说话检测speech detection、打断interruption、轮次完成turn completion、重连reconnection和可恢复错误在请求-响应式运行request-response run中没有对应物属于实时会话的专属语义。对能力的事件流钩子而言两类事件流过同一条流都会可见Agent.realtime会把wrap_event_stream包装器应用到迭代器上包装器词汇表是AgentStreamEvent超集但应用于实时流时必须产出RealtimeEvent子集见 _session.py。4. 轮次边界用 RealtimeTurnCompleteEvent 判断这一轮结束了在实时对话中判断模型是否已经回答完毕远比想象中复杂。模型可以先说话、再调用工具、然后再说话——因此收到语音或收到工具结果都不意味着本轮结束。官方推荐的做法是把RealtimeTurnCompleteEvent作为一次交换exchange的边界。它在会话确认无工具调用仍在运行、且没有响应在途时由会话合成是唯一可靠的本轮结束信号。async for event in session: if isinstance(event, RealtimeTurnCompleteEvent): # 本轮真正结束模型回复完毕且没有工具还在执行 break # 或在此推进对话状态机这一用法在测试中也有直接体现tests/realtime/test_session.py 用isinstance(event, RealtimeTurnCompleteEvent)作为循环终止条件之后再await asyncio.sleep(0)数次让后台工具任务收尾。两个需要留意的补充语义来自源码事件类文档转写可能晚到输入转写可以在RealtimeTurnCompleteEvent之后才完成messages.py。播放可能晚到在 WebRTC sideband 上提供方的播放可以持续到RealtimeOutputSpeechEndEvent才结束。因此如果要驱动正在说话指示器应使用OutputSpeechStart/End这对事件而不是用 turn completemessages.py。另外从 _session.py 的注释可以看到只有交换边界被标记每个 response 本身已经是chatspan而模型真正完成的轮次边界没有自己的 span所以RealtimeTurnCompleteEvent是追踪中唯一标记轮次边界的事件在被打断的边界上事件显示文本还会说明被打断。5. 读取原始音频事件流媒体与底层事件的关系5.1 首选方式stream_audio() 受限视图音频流本质上就是这些事件——stream_audio()是对语音分片增量speech part deltas的受限视图bounded view绝大多数应用应当使用它而不是直接解析底层事件。它只包含实时的模型音频、按播放顺序排列绝不重复来自已定稿语音分片的保留音频源码。它的几个关键实现细节对编写播放循环很有用订阅即开始订阅在调用stream_audio()时就注册调用与消费者首次迭代之间产生的音频会被缓冲而不是丢失每个迭代器有 32 个数据块的缓冲_AUDIO_TAP_SIZE 32。落后不阻塞如果消费者跟不上或从不开始消费最旧的数据块会被丢弃drop-oldest保证音频播放永远不会拖垮工具执行、轮次跟踪或主事件流。设备时钟对账设备节奏的消费者喂入played_audio_bytes在 barge-in 时interrupt(played_bytes...)会丢弃用户永远不会听到的缓冲块。WebRTC sideband 不可用在 sideband 上浏览器持有音频路径stream_audio()会抛出UserError应改消费浏览器的远端媒体轨道源码。同理stream_transcripts()是另一个受限视图默认产出每个已完成轮次的SpeechPart含speaker与完整transcript传deltaTrue则产出TranscriptUpdate携带新文本、该轮次到目前为止的完整转写、说话人以及标识轮次的index两者使用独立的订阅互不挤占源码。TranscriptUpdate的完整字段定义见 _session.py。5.2 高级替代直接从 PartDeltaEvent 读取 SpeechPartDelta.audio_chunk作为高级用法也可以绕过视图直接播放原始PartDeltaEvent中携带的SpeechPartDelta.audio_chunk。适用场景是你需要把音频与其它事件如工具调用、转写在同一条流中按序处理而不想额外订阅stream_audio()。from pydantic_ai.messages import PartDeltaEvent, SpeechPartDelta async for event in session: if isinstance(event, PartDeltaEvent) and isinstance(event.delta, SpeechPartDelta): chunk event.delta.audio_chunk # 原始 PCM 音频块可送入声卡/WebSocket if chunk: await player.write(chunk)需要注意的两个不要重复播放陷阱模型音频无论是否启用历史保留都会完整到达也就是说即使audio_retention关闭了输出保留事件流中的音频块也是完整的可以直接播放。启用输出音频保留时最终 SpeechPart 会再包含一次整轮音频的 WAV 快照用于历史保留。此时不要两处都播放否则本轮会被播放两遍。正确做法是播放时只用实时事件流或stream_audio()把保留的 WAV 快照留给历史/回放场景。从源码看实时SpeechPartDelta.audio_chunk是原始 PCM而保留历史使用 WAV 容器以便自描述、可移植到经典模型适配器_session.py二者在会话内部本就是两套数据。6. 总结事件流驱动的实时应用骨架把上述内容组合起来一个典型的实时语音应用控制流如下async with Agent(...).realtime(...) as session:进入会话async for event in session:迭代事件流按类型分派PartStart/Delta/EndEvent→ 渲染分片 / 增量文本文本输出或交给视图FunctionToolCallEvent/FunctionToolResultEvent→ 更新工具状态 UI / 记录日志RealtimeInputSpeechStartEvent→ barge-in停止播放缓冲音频RealtimeInputSpeechEndEvent→ 显示处理中指示器RealtimeTurnCompleteEvent→ 推进对话状态机本轮真正结束RealtimeOutputSpeechStart/EndEventWebRTC sideband→ 驱动正在说话指示器RealtimeSessionReconnectEvent→ 依据state_restored决定是否提示用户会话连续性RealtimeSessionErrorEvent/RealtimeInputTranscriptionErrorEvent→ 可恢复错误处理会话继续可用媒体播放与字幕分别交给stream_audio()/stream_transcripts()视图两者都是同一事件流的受限投影绝不会与主循环竞争或阻塞。由于前七类事件与标准AgentStreamEvent完全同源文本 Agent 的事件处理代码可以无缝复用这也是 pydantic-ai 实时能力typed end to end的体现——同一套事件词汇贯穿流式文本与实时语音两种运行模式。相关资源本文主线文档docs/realtime/events.md会话与事件翻译实现pydantic_ai_slim/pydantic_ai/realtime/_session.py事件类定义pydantic_ai_slim/pydantic_ai/messages.py视图文档Audio, images, and transcripts、Retaining audio事件流钩子与能力Event stream hooks、The event streamWebRTC sideband 部署docs/realtime/deployment.md#browser-webrtc-server-sideband会话行为测试tests/realtime/test_session.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），仅供参考