ADK LiveRequestQueue 使用指南构建 Gemini Live 双向流式输入的异步请求队列【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-pythonLiveRequestQueue是 ADKAgent Development Kit中面向 Gemini Live API 双向流式会话的输入缓冲队列负责承载文本内容、实时音频/视频 Blob 以及流控制信号并将客户端侧的输入生产与模型侧的消费解耦。本文以官方指南为核心结合仓库源码与测试系统讲解其数据模型、优先级规则、全部 API 用法及底层执行链路帮助你在实时语音对话、流式工具调用等场景中正确使用它。LiveRequestQueue 是什么在 ADK 中live 双向流式会话允许 Agent 与客户端持续交换文本和音频分片。LiveRequestQueue扮演 live 执行流BaseLlmFlow与GeminiLlmConnection的输入缓冲角色客户端可以随时向队列写入请求而模型消费端按固定优先级异步处理。它的核心能力可归纳为三类官方指南媒体流Media Streaming发送实时媒体 Blob如音频、视频分片回合内容Turn Content发送逐回合的types.Content流信号Stream Signals发送活动开始activity start、活动结束activity end与音频流终止audio stream end信号。类本身实现非常轻量live_request_queue.py内部仅持有一个asyncio.Queue[LiveRequest]和一个_closed布尔标记所有公开方法都是对put_nowait的封装。数据模型LiveRequest 及其处理优先级队列中流转的元素是LiveRequestPydantic 模型见 live_request_queue.py。其字段如下字段类型默认值含义contentOptional[types.Content]None逐回合模式发送给模型的内容blobOptional[types.Blob]None实时模式发送给模型的媒体 Blobactivity_startOptional[types.ActivityStart]None通知模型用户活动开始activity_endOptional[types.ActivityEnd]None通知模型用户活动结束audio_stream_endboolFalse通知模型音频流结束仅在启用 VAD 时使用closeboolFalse关闭队列的哨兵信号partialboolFalse该内容是一次不完整回合的部分更新不结束当前模型回合state_deltaOptional[dict[str, Any]]None应用到 session 的状态变更即使请求不带内容或属于 partial/function-response 回合也会生效当多个字段同时被设置时消费端按以下优先级处理最高优先在前来自类 docstring 与官方指南activity_start activity_end audio_stream_end blob content此外有一条例外规则state_delta一旦被设置无论其它字段如何始终会被应用。这一点在 send_to_model 的实现 中得到印证当state_delta存在且没有创建内容事件时会单独生成一个携带EventActions(state_delta...)的用户事件写入 session。快速开始基础流式示例官方指南给出了最直接的用法——创建队列、发送各类输入、关闭队列from google.adk.live import LiveRequestQueue from google.genai import types queue LiveRequestQueue() # 逐回合模式发送文本内容 queue.send_content(types.Content(parts[types.Part.from_text(textHello!)])) # 发送实时音频分片例如 PCM 音频字节 queue.send_realtime(types.Blob(dataaudio_bytes, mime_typeaudio/pcm)) # 通知音频输入流结束例如麦克风被关闭 queue.send_audio_stream_end() # live 会话结束时关闭队列 queue.close()上述代码中的每个方法调用在源码中都有明确对应live_request_queue.pysend_content(content, partialFalse)→put_nowait(LiveRequest(contentcontent, partialpartial))send_realtime(blob)→put_nowait(LiveRequest(blobblob))send_audio_stream_end()→put_nowait(LiveRequest(audio_stream_endTrue))close()→ 置_closed True并put_nowait(LiveRequest(closeTrue))注意close()在源码注释中被描述为“one-shot sentinel”一次性哨兵关闭信号一旦被消费队列中就不再存在关闭标记因此重连逻辑无法依赖它判断关闭状态而是依赖下面要讲的closed属性。流控制方法全览官方指南归纳了队列提供的全部辅助方法方法作用send_realtime(blob: types.Blob)入队一个实时媒体 Blobsend_content(content: types.Content, partial: bool False)入队逐回合内容send_activity_start()入队ActivityStart信号标记用户活动开始send_activity_end()入队ActivityEnd信号标记用户活动结束send_audio_stream_end()入队音频流结束信号LiveRequest(audio_stream_endTrue)close()入队关闭信号以终止队列处理其中send_activity_start/send_activity_end在实现中分别包装types.ActivityStart()与types.ActivityEnd()live_request_queue.py用于向模型标注用户输入如按住说话的开始与结束边界。直接入队queue.send(req)如果希望绕过辅助方法直接入队必须传入LiveRequest实例from google.adk.live import LiveRequest # 直接发送音频流结束信号 queue.send(LiveRequest(audio_stream_endTrue))从源码看send()是通用入队路径当请求携带closeTrue时它同样会置位_closed标记然后执行put_nowaitlive_request_queue.py。幂等读取closed 粘性标记LiveRequestQueue.closed是一个只读属性返回_closed状态。它在关闭后是**粘性sticky**的——即使关闭哨兵已被消费、队列已被清空closed依然保持True见属性 docstring 与 test_close_sets_closed_stickily 测试。这正是重连逻辑判断“是否客户端主动关闭”的依据。工作原理队列如何被消费LiveRequestQueue是纯生产者侧接口真正消费它的是 live 执行流中的send_to_model协程_live_llm_flow.py。其核心循环逻辑为await live_request_queue.get()阻塞取出下一个LiveRequest将请求复制分发给所有已注册的活动流式工具active_streaming_tools让流式工具也能实时感知输入按上文优先级分派close→ 关闭 LLM 连接并返回activity_start/activity_end→ 通过llm_connection.send_realtime(...)发送对应信号audio_stream_end→ 发送types.LiveClientRealtimeInput(audio_stream_endTrue)blob→ 先按run_config.save_live_blob缓存输入音频再send_realtime(blob)content→ 校验不含 function_call 后写入 session 事件并调用_send_content(content, partial...)发送给模型。在整个 live 执行流中LiveRequestQueue实例被挂载在InvocationContext.live_request_queue字段上invocation_context.py并通过require_live_request_queue()获取若缺失会抛出ValueError(Live model execution requires a LiveRequestQueue.)_live_llm_flow.py。LiveRequest 与 LiveClientRealtimeInput 的转换官方指南特别强调队列接受的是LiveRequest而不是LiveClientRealtimeInput。从GeminiLlmConnection.send_realtime的实现可以看到转换发生在连接层gemini_llm_connection.pyelif isinstance(input, types.LiveClientRealtimeInput): if input.audio_stream_end: await self._gemini_session.send_realtime_input(audio_stream_endTrue)即LiveRequest(audio_stream_endTrue)在队列消费时被翻译为LiveClientRealtimeInput(audio_stream_endTrue)再交给底层 Gemini session 发送。因此应用开发者不需要也不应该直接构造LiveClientRealtimeInput入队只需使用queue.send_audio_stream_end()或queue.send(LiveRequest(audio_stream_endTrue))。Audio Stream End 与 VAD 的正确取舍官方指南用一个专门小节澄清了最容易踩坑的语义差异开启 VADVoice Activity Detection时Gemini Live API 会自动检测用户语音的起止utterance应用不需要在每次说话结束时发送audio_stream_endsend_audio_stream_end()的用途仅当音频输入流本身终止时例如用户关闭或静音麦克风才调用。该信号通知后端“后续不再有音频分片”并触发缓冲音频的冲刷。[!NOTE] 在 VAD 模式下不要在每一轮对话结束后调用send_audio_stream_end()。否则每轮都会关闭音频流下一轮需要重新发送音频消息才能重新打开。进阶用法partial 与 state_deltapartialsend_content(..., partialTrue)表示这是一次部分回合更新不结束当前模型回合。消费端会跳过 partial 内容的 session 事件落盘但仍会把内容发送给模型_live_llm_flow.py。state_deltaLiveRequest(state_delta{...})可在不携带内容的情况下推进 session 状态适合在流式会话中同步上下文变量。生态中的其它入口除google.adk.live外仓库还提供了兼容性导出agents/live_request_queue.py 与 agents/__init__.py因此from google.adk.agents import LiveRequestQueue同样可用。此外LiveRequestQueue还被用于流式工具streaming tools工具函数若声明LiveRequestQueue类型参数即可持续向模型推送实时数据active_streaming_tool.py 与 function_tool.pyRunner 与 API Serverrun_live相关入口与 Web 服务端均会创建并注入LiveRequestQueuerunners.py、api_server.pyLiveKit 集成以LiveRequestQueue - run_live() - Event作为桥接契约将房间媒体轨道转发进队列见 livekit 集成说明。可参考的完整可运行示例位于 contributing/samples/live 目录例如live_bidi_streaming_single_agent、live_bidi_streaming_parallel_tools_agent等。测试验证仓库为队列行为提供了专门测试tests/unittests/live/test_live_request_queue.pytest_close_queueclose()恰好入队一个LiveRequest(closeTrue)test_send_content/test_send_realtime_blob辅助方法正确包装为对应LiveRequesttest_close_sets_closed_stickily关闭哨兵被消费后closed仍为Truetest_send_with_close_request_sets_closed经send()入队关闭请求同样会置位标记。这些测试直接印证了上述 API 契约是你在集成或二次开发时最可靠的参考。小结LiveRequestQueue用极简的接口一个asyncio.Queue加少量put_nowait封装撑起了 Gemini Live 双向流式会话的输入侧是理解 ADK live 执行链路的关键起点。掌握LiveRequest的字段与优先级、区分audio_stream_end与 VAD 的边界、理解LiveRequest/LiveClientRealtimeInput的转换层次你就能在自己的实时语音或多模态 Agent 中正确编排输入流。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ADK LiveRequestQueue 使用指南：构建 Gemini Live 双向流式输入的异步请求队列
ADK LiveRequestQueue 使用指南构建 Gemini Live 双向流式输入的异步请求队列【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-pythonLiveRequestQueue是 ADKAgent Development Kit中面向 Gemini Live API 双向流式会话的输入缓冲队列负责承载文本内容、实时音频/视频 Blob 以及流控制信号并将客户端侧的输入生产与模型侧的消费解耦。本文以官方指南为核心结合仓库源码与测试系统讲解其数据模型、优先级规则、全部 API 用法及底层执行链路帮助你在实时语音对话、流式工具调用等场景中正确使用它。LiveRequestQueue 是什么在 ADK 中live 双向流式会话允许 Agent 与客户端持续交换文本和音频分片。LiveRequestQueue扮演 live 执行流BaseLlmFlow与GeminiLlmConnection的输入缓冲角色客户端可以随时向队列写入请求而模型消费端按固定优先级异步处理。它的核心能力可归纳为三类官方指南媒体流Media Streaming发送实时媒体 Blob如音频、视频分片回合内容Turn Content发送逐回合的types.Content流信号Stream Signals发送活动开始activity start、活动结束activity end与音频流终止audio stream end信号。类本身实现非常轻量live_request_queue.py内部仅持有一个asyncio.Queue[LiveRequest]和一个_closed布尔标记所有公开方法都是对put_nowait的封装。数据模型LiveRequest 及其处理优先级队列中流转的元素是LiveRequestPydantic 模型见 live_request_queue.py。其字段如下字段类型默认值含义contentOptional[types.Content]None逐回合模式发送给模型的内容blobOptional[types.Blob]None实时模式发送给模型的媒体 Blobactivity_startOptional[types.ActivityStart]None通知模型用户活动开始activity_endOptional[types.ActivityEnd]None通知模型用户活动结束audio_stream_endboolFalse通知模型音频流结束仅在启用 VAD 时使用closeboolFalse关闭队列的哨兵信号partialboolFalse该内容是一次不完整回合的部分更新不结束当前模型回合state_deltaOptional[dict[str, Any]]None应用到 session 的状态变更即使请求不带内容或属于 partial/function-response 回合也会生效当多个字段同时被设置时消费端按以下优先级处理最高优先在前来自类 docstring 与官方指南activity_start activity_end audio_stream_end blob content此外有一条例外规则state_delta一旦被设置无论其它字段如何始终会被应用。这一点在 send_to_model 的实现 中得到印证当state_delta存在且没有创建内容事件时会单独生成一个携带EventActions(state_delta...)的用户事件写入 session。快速开始基础流式示例官方指南给出了最直接的用法——创建队列、发送各类输入、关闭队列from google.adk.live import LiveRequestQueue from google.genai import types queue LiveRequestQueue() # 逐回合模式发送文本内容 queue.send_content(types.Content(parts[types.Part.from_text(textHello!)])) # 发送实时音频分片例如 PCM 音频字节 queue.send_realtime(types.Blob(dataaudio_bytes, mime_typeaudio/pcm)) # 通知音频输入流结束例如麦克风被关闭 queue.send_audio_stream_end() # live 会话结束时关闭队列 queue.close()上述代码中的每个方法调用在源码中都有明确对应live_request_queue.pysend_content(content, partialFalse)→put_nowait(LiveRequest(contentcontent, partialpartial))send_realtime(blob)→put_nowait(LiveRequest(blobblob))send_audio_stream_end()→put_nowait(LiveRequest(audio_stream_endTrue))close()→ 置_closed True并put_nowait(LiveRequest(closeTrue))注意close()在源码注释中被描述为“one-shot sentinel”一次性哨兵关闭信号一旦被消费队列中就不再存在关闭标记因此重连逻辑无法依赖它判断关闭状态而是依赖下面要讲的closed属性。流控制方法全览官方指南归纳了队列提供的全部辅助方法方法作用send_realtime(blob: types.Blob)入队一个实时媒体 Blobsend_content(content: types.Content, partial: bool False)入队逐回合内容send_activity_start()入队ActivityStart信号标记用户活动开始send_activity_end()入队ActivityEnd信号标记用户活动结束send_audio_stream_end()入队音频流结束信号LiveRequest(audio_stream_endTrue)close()入队关闭信号以终止队列处理其中send_activity_start/send_activity_end在实现中分别包装types.ActivityStart()与types.ActivityEnd()live_request_queue.py用于向模型标注用户输入如按住说话的开始与结束边界。直接入队queue.send(req)如果希望绕过辅助方法直接入队必须传入LiveRequest实例from google.adk.live import LiveRequest # 直接发送音频流结束信号 queue.send(LiveRequest(audio_stream_endTrue))从源码看send()是通用入队路径当请求携带closeTrue时它同样会置位_closed标记然后执行put_nowaitlive_request_queue.py。幂等读取closed 粘性标记LiveRequestQueue.closed是一个只读属性返回_closed状态。它在关闭后是**粘性sticky**的——即使关闭哨兵已被消费、队列已被清空closed依然保持True见属性 docstring 与 test_close_sets_closed_stickily 测试。这正是重连逻辑判断“是否客户端主动关闭”的依据。工作原理队列如何被消费LiveRequestQueue是纯生产者侧接口真正消费它的是 live 执行流中的send_to_model协程_live_llm_flow.py。其核心循环逻辑为await live_request_queue.get()阻塞取出下一个LiveRequest将请求复制分发给所有已注册的活动流式工具active_streaming_tools让流式工具也能实时感知输入按上文优先级分派close→ 关闭 LLM 连接并返回activity_start/activity_end→ 通过llm_connection.send_realtime(...)发送对应信号audio_stream_end→ 发送types.LiveClientRealtimeInput(audio_stream_endTrue)blob→ 先按run_config.save_live_blob缓存输入音频再send_realtime(blob)content→ 校验不含 function_call 后写入 session 事件并调用_send_content(content, partial...)发送给模型。在整个 live 执行流中LiveRequestQueue实例被挂载在InvocationContext.live_request_queue字段上invocation_context.py并通过require_live_request_queue()获取若缺失会抛出ValueError(Live model execution requires a LiveRequestQueue.)_live_llm_flow.py。LiveRequest 与 LiveClientRealtimeInput 的转换官方指南特别强调队列接受的是LiveRequest而不是LiveClientRealtimeInput。从GeminiLlmConnection.send_realtime的实现可以看到转换发生在连接层gemini_llm_connection.pyelif isinstance(input, types.LiveClientRealtimeInput): if input.audio_stream_end: await self._gemini_session.send_realtime_input(audio_stream_endTrue)即LiveRequest(audio_stream_endTrue)在队列消费时被翻译为LiveClientRealtimeInput(audio_stream_endTrue)再交给底层 Gemini session 发送。因此应用开发者不需要也不应该直接构造LiveClientRealtimeInput入队只需使用queue.send_audio_stream_end()或queue.send(LiveRequest(audio_stream_endTrue))。Audio Stream End 与 VAD 的正确取舍官方指南用一个专门小节澄清了最容易踩坑的语义差异开启 VADVoice Activity Detection时Gemini Live API 会自动检测用户语音的起止utterance应用不需要在每次说话结束时发送audio_stream_endsend_audio_stream_end()的用途仅当音频输入流本身终止时例如用户关闭或静音麦克风才调用。该信号通知后端“后续不再有音频分片”并触发缓冲音频的冲刷。[!NOTE] 在 VAD 模式下不要在每一轮对话结束后调用send_audio_stream_end()。否则每轮都会关闭音频流下一轮需要重新发送音频消息才能重新打开。进阶用法partial 与 state_deltapartialsend_content(..., partialTrue)表示这是一次部分回合更新不结束当前模型回合。消费端会跳过 partial 内容的 session 事件落盘但仍会把内容发送给模型_live_llm_flow.py。state_deltaLiveRequest(state_delta{...})可在不携带内容的情况下推进 session 状态适合在流式会话中同步上下文变量。生态中的其它入口除google.adk.live外仓库还提供了兼容性导出agents/live_request_queue.py 与 agents/__init__.py因此from google.adk.agents import LiveRequestQueue同样可用。此外LiveRequestQueue还被用于流式工具streaming tools工具函数若声明LiveRequestQueue类型参数即可持续向模型推送实时数据active_streaming_tool.py 与 function_tool.pyRunner 与 API Serverrun_live相关入口与 Web 服务端均会创建并注入LiveRequestQueuerunners.py、api_server.pyLiveKit 集成以LiveRequestQueue - run_live() - Event作为桥接契约将房间媒体轨道转发进队列见 livekit 集成说明。可参考的完整可运行示例位于 contributing/samples/live 目录例如live_bidi_streaming_single_agent、live_bidi_streaming_parallel_tools_agent等。测试验证仓库为队列行为提供了专门测试tests/unittests/live/test_live_request_queue.pytest_close_queueclose()恰好入队一个LiveRequest(closeTrue)test_send_content/test_send_realtime_blob辅助方法正确包装为对应LiveRequesttest_close_sets_closed_stickily关闭哨兵被消费后closed仍为Truetest_send_with_close_request_sets_closed经send()入队关闭请求同样会置位标记。这些测试直接印证了上述 API 契约是你在集成或二次开发时最可靠的参考。小结LiveRequestQueue用极简的接口一个asyncio.Queue加少量put_nowait封装撑起了 Gemini Live 双向流式会话的输入侧是理解 ADK live 执行链路的关键起点。掌握LiveRequest的字段与优先级、区分audio_stream_end与 VAD 的边界、理解LiveRequest/LiveClientRealtimeInput的转换层次你就能在自己的实时语音或多模态 Agent 中正确编排输入流。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考