尧图网页设计 
ADK Runner 实时直播流run_live实战指南基于 Gemini Live API 的双向流式语音 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-pythonRunner.run_live是 ADKAgent Development Kit提供的实时执行模式用于与 Gemini Multimodal Live API 模型建立持久的双向流式会话。它通过LiveRequestQueue持续接收用户的 PCM 音频帧与文本分片将工具调用派发到后台任务以保持音频输出不被阻塞并向调用方实时产出模型Event流。阅读本篇后你将掌握如何用run_live构建可运行的实时语音/多模态 Agent、如何配置RunConfig控制音频持久化与响应模态以及如何编写不阻塞音频流的后台工具回调。为什么需要 run_live从回合制到流式标准的聊天模型遵循请求—响应回合制模式一次请求必须完整结束后模型才能给出整体回复。对于实时语音、会话式音频和流式多模态应用而言等待完整回合会带来不可接受的延迟。Gemini Live 模型需要持久、低延迟的 WebSocket 或 gRPC 连接要求能够一边持续接收 PCM 音频帧一边流式回传音频响应并执行工具。run_live子系统正是为此设计它将Runner连接到 Gemini Multimodal Live 端点调用方通过LiveRequestQueue注入实时的用户音频或文本分片Runner 维持活动的流式会话将非阻塞工具调用路由到后台执行任务不打断音频流并产出实时的模型内容Event。从 runners.py 的源码可以看到Runner.run_live是一个AsyncGenerator[Event, None]其核心实现委托给_live_runner_utils.run_live并通过aclosing保证调用方提前退出时后台任务被正确清理。需要特别注意的是并非所有产出的事件都会写入会话——这是 live 模式与普通run模式的关键差异之一详见下文事件分类与持久化策略。快速上手最小可运行的实时会话将LlmAgent挂载到App连接InMemoryRunner再用LiveRequestQueue驱动一次 live 会话from google.adk.agents import LlmAgent from google.adk.apps import App from google.adk.live import LiveRequestQueue from google.adk.runners import InMemoryRunner from google.genai import types root_agent LlmAgent( namevoice_agent, instructionYou are a voice assistant. Answer queries concisely in spoken English., ) app App(namevoice_app, root_agentroot_agent) runner InMemoryRunner(appapp) queue LiveRequestQueue() # 在异步任务中向队列推送用户文本或音频 queue.send_content( contenttypes.Content( roleuser, parts[types.Part.from_text(textHello! Can you hear me?)], ) ) async for event in runner.run_live( user_iduser_123, session_idsession_live, live_request_queuequeue, ): if event.content and event.content.parts: for part in event.content.parts: if part.text: print(Live model response:, part.text)队列允许调用方在run_live持续流式回传响应事件的同时异步地把 PCM 音频帧或文本消息推入活动会话。LiveRequestQueue还支持send_realtime()推送实时音频 Blob、send_activity_start()/send_activity_end()标记用户活动边界以及close()关闭队列。工作原理四阶段生命周期live 执行生命周期在Runner、LiveRequestQueue、BaseLlmFlow与 Gemini Live API 后端之间协调可概括为四个阶段连接建立Connection Setuprun_live使用types.LiveConnectConfig建立持久的双向连接可指定AUDIO或TEXT等模态以及语音设置。从 _runner_utils.py 的实现看当run_config.response_modalities为None时runner 会将其默认设为AUDIO部分原生音频模型要求显式设置模态该默认值写在一个浅拷贝的配置副本上避免污染调用方后续复用的RunConfig。异步输入摄入Asynchronous Input IngestionLiveRequestQueue内部包装了一个asyncio.Queue[LiveRequest]见 live_request_queue.py。调用方通过queue.send_realtime()流式发送音频 PCM 分片或用queue.send_content()发送文本 tokenrun_live会将这些请求转发到已打开的 WebSocket 上。事件分类与会话过滤Event Categorization Session Filtering内联音频事件inline_data直接流向调用方以实现低延迟音频播放但不会写入会话历史避免会话膨胀制品媒体事件save_live_blob当RunConfig.save_live_blobTrue时视频和音频数据保存到制品存储并持久化到会话历史转录与工具调用非分片non-partial转录、用量元数据与函数调用始终写入会话历史。非阻塞后台工具Non-Blocking Background Toolslive 流期间工具函数被调用时runner 将工具调用派发到后台任务音频输出不被阻塞工具完成后结果被推回LiveRequestQueue更新模型。LiveRequest 的优先级语义LiveRequestQueue传输的LiveRequest模型源码见 live_request_queue.py在多个字段同时设置时按如下优先级处理activity_startactivity_endaudio_stream_endblobcontent而state_delta无论其他字段如何设置都会被应用。队列方法映射如下方法底层请求用途send_content(content, partialFalse)LiveRequest(content...)回合制模式发送文本/多模态内容partialTrue表示部分轮次更新send_realtime(blob)LiveRequest(blob...)实时模式发送音频/视频 Blobsend_activity_start()LiveRequest(activity_start...)标记用户输入开始send_activity_end()LiveRequest(activity_end...)标记用户输入结束send_audio_stream_end()LiveRequest(audio_stream_endTrue)强制冲刷音频、结束音频流配合 VAD 使用close()LiveRequest(closeTrue)关闭队列终止处理send(req)任意LiveRequest直接入队自定义请求一个与 VAD语音活动检测相关的常见误区当 Gemini Live API 的 VAD 生效时模型会自动检测用户话语的起止不需要在每个话语末尾发送audio_stream_end该方法只用于通知后端整个音频输入流已结束例如用户关闭/静音麦克风时请勿在每个对话回合后调用否则需要新的音频消息才能重新打开流。更多细节可参考 LiveRequestQueue 指南。run_live 配置参数详解run_live接受以下配置参数与 runners.py 的签名一致参数类型默认值说明user_idstr \| NoneNone会话的用户 IDsession为None时必填session_idstr \| NoneNone会话 IDsession为None时必填live_request_queueLiveRequestQueue必填用于向会话推送实时用户输入、音频分片与工具结果的队列run_configRunConfig \| NoneNone执行配置含speech_config、response_modalities、save_live_blob等sessionSession \| NoneNone预取的会话实例已弃用推荐使用user_id与session_id注意源码中的校验逻辑session为None时必须同时提供user_id和session_id否则抛出ValueErrorlive_request_queue为None同样直接报错。传入session会触发DeprecationWarning。另外run_live的根节点既可以是BaseAgent也可以是非 Agent 的BaseNode如Workflow——后者会走run_node_live分支经由DynamicNodeScheduler驱动根节点并合并事件流。RunConfig Live 选项通过run_configRunConfig(...)配置字段定义位于 run_config.py选项类型默认值说明speech_configtypes.SpeechConfig \| NoneNonelive Agent 的语音选择与音频编码配置response_modalitieslist[types.Modality] \| NoneNone模型返回的输出模态AUDIO或TEXT未设置时默认AUDIOsave_live_blobboolFalse将 live 视频与音频数据保存到会话与制品服务session_resumptiontypes.SessionResumptionConfig \| NoneNone配置透明的会话恢复机制tool_thread_pool_configToolThreadPoolConfig \| NoneNone在后台线程池执行器中运行工具保持事件循环响应从源码注释可以补充几个值得注意的细节tool_thread_pool_config设置为非空时工具在独立线程池执行器中运行而非主事件循环默认None时在主事件循环运行同一事件循环上的所有调用共享一个线程池该池随事件循环销毁而关闭工作线程不会存活到事件循环之外。这能防止耗时工具阻塞音频事件的实时投递。音频转录配置input_audio_transcription/output_audio_transcription默认均为AudioTranscriptionConfig控制输入/输出音频的转录对于带sub_agents的 live 多 Agent 系统_runner_utils.py 会自动补齐转录配置因为子 Agent 需要模型文本转录作为上下文。其他 live 相关选项explicit_vad_signal显式 VAD 信号、translation_config实时语音到语音翻译仅受翻译类模型支持、enable_affective_dialog情感检测、proactivity模型主动响应以及streaming_modeNONE/SSE/BIDI等也可一并配置。高级应用非阻塞流式工具回调工具函数可以声明input_stream: LiveRequestQueue参数在后台运行期间把部分工具结果或状态更新流式回传给 live 会话from google.adk.live import LiveRequestQueue from google.genai import types async def fetch_stock_ticker( symbol: str, input_stream: LiveRequestQueue ) - dict[str, float]: Fetches live stock price while streaming progress. # 通知 live 模型会话查询已在进行中 input_stream.send_content( contenttypes.Content( roleuser, parts[ types.Part.from_text( textfFetching latest price for {symbol}... ) ], ) ) # 执行查询 return {symbol: symbol, price: 154.25}关键点这里的send_content作用对象是模型的输入流模拟用户发言使模型能够听到工具正在工作的状态从而在等待期间继续与用户自然对话。工具最终返回值仍然作为函数响应进入原有工具结果回路。仓库中的 live_non_blocking_tool_agent 示例 展示了完整模式用FunctionTool包装一个await asyncio.sleep(10)的慢任务并通过types.FunctionResponseScheduling.WHEN_IDLE声明非阻塞调度策略可选值还包括SILENT、INTERRUPT使 Agent 在任务执行期间持续与用户交谈而不中断音频流。该示例使用的模型为gemini-live-2.5-flash-native-audio可替换为你所用环境支持的 Live 模型。局限性与注意事项Gemini Live 模型要求run_live需要支持 Gemini Multimodal Live API 的模型端点例如gemini-2.0-flash-exp普通生成式模型无法走此路径。内联音频不持久化原始 PCM 音频 Blobinline_data被有意排除在会话存储之外。若需要保留会话的音频/视频历史请在RunConfig上设置save_live_blobTrue。实验性 API源码文档字符串明确标注 live 功能为experimentalrunners.py其 API 或行为可能在后续版本中变更session参数已弃用应改用user_idsession_id。相关指南与示例Runner 与 InMemoryRunner 指南 —— 标准回合制 runner 执行的主指南LiveRequestQueue 指南 —— 实时输入排队、音频分片与非阻塞流式工具App 容器指南 —— 将 Agent 与插件打包进AppLive 双向流式单 Agent 示例 —— 单 Agent 实时流式应用样例Live 非阻塞工具 Agent 示例 —— 在后台工具回调中使用LiveRequestQueue的样例以上示例均位于 contributing/samples/live 目录可结合 live 相关文档docs/guides/live一起阅读完整掌握 ADK 实时流式 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 Runner 实时直播流（run_live）实战指南：基于 Gemini Live API 的双向流式语音 Agent
ADK Runner 实时直播流run_live实战指南基于 Gemini Live API 的双向流式语音 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-pythonRunner.run_live是 ADKAgent Development Kit提供的实时执行模式用于与 Gemini Multimodal Live API 模型建立持久的双向流式会话。它通过LiveRequestQueue持续接收用户的 PCM 音频帧与文本分片将工具调用派发到后台任务以保持音频输出不被阻塞并向调用方实时产出模型Event流。阅读本篇后你将掌握如何用run_live构建可运行的实时语音/多模态 Agent、如何配置RunConfig控制音频持久化与响应模态以及如何编写不阻塞音频流的后台工具回调。为什么需要 run_live从回合制到流式标准的聊天模型遵循请求—响应回合制模式一次请求必须完整结束后模型才能给出整体回复。对于实时语音、会话式音频和流式多模态应用而言等待完整回合会带来不可接受的延迟。Gemini Live 模型需要持久、低延迟的 WebSocket 或 gRPC 连接要求能够一边持续接收 PCM 音频帧一边流式回传音频响应并执行工具。run_live子系统正是为此设计它将Runner连接到 Gemini Multimodal Live 端点调用方通过LiveRequestQueue注入实时的用户音频或文本分片Runner 维持活动的流式会话将非阻塞工具调用路由到后台执行任务不打断音频流并产出实时的模型内容Event。从 runners.py 的源码可以看到Runner.run_live是一个AsyncGenerator[Event, None]其核心实现委托给_live_runner_utils.run_live并通过aclosing保证调用方提前退出时后台任务被正确清理。需要特别注意的是并非所有产出的事件都会写入会话——这是 live 模式与普通run模式的关键差异之一详见下文事件分类与持久化策略。快速上手最小可运行的实时会话将LlmAgent挂载到App连接InMemoryRunner再用LiveRequestQueue驱动一次 live 会话from google.adk.agents import LlmAgent from google.adk.apps import App from google.adk.live import LiveRequestQueue from google.adk.runners import InMemoryRunner from google.genai import types root_agent LlmAgent( namevoice_agent, instructionYou are a voice assistant. Answer queries concisely in spoken English., ) app App(namevoice_app, root_agentroot_agent) runner InMemoryRunner(appapp) queue LiveRequestQueue() # 在异步任务中向队列推送用户文本或音频 queue.send_content( contenttypes.Content( roleuser, parts[types.Part.from_text(textHello! Can you hear me?)], ) ) async for event in runner.run_live( user_iduser_123, session_idsession_live, live_request_queuequeue, ): if event.content and event.content.parts: for part in event.content.parts: if part.text: print(Live model response:, part.text)队列允许调用方在run_live持续流式回传响应事件的同时异步地把 PCM 音频帧或文本消息推入活动会话。LiveRequestQueue还支持send_realtime()推送实时音频 Blob、send_activity_start()/send_activity_end()标记用户活动边界以及close()关闭队列。工作原理四阶段生命周期live 执行生命周期在Runner、LiveRequestQueue、BaseLlmFlow与 Gemini Live API 后端之间协调可概括为四个阶段连接建立Connection Setuprun_live使用types.LiveConnectConfig建立持久的双向连接可指定AUDIO或TEXT等模态以及语音设置。从 _runner_utils.py 的实现看当run_config.response_modalities为None时runner 会将其默认设为AUDIO部分原生音频模型要求显式设置模态该默认值写在一个浅拷贝的配置副本上避免污染调用方后续复用的RunConfig。异步输入摄入Asynchronous Input IngestionLiveRequestQueue内部包装了一个asyncio.Queue[LiveRequest]见 live_request_queue.py。调用方通过queue.send_realtime()流式发送音频 PCM 分片或用queue.send_content()发送文本 tokenrun_live会将这些请求转发到已打开的 WebSocket 上。事件分类与会话过滤Event Categorization Session Filtering内联音频事件inline_data直接流向调用方以实现低延迟音频播放但不会写入会话历史避免会话膨胀制品媒体事件save_live_blob当RunConfig.save_live_blobTrue时视频和音频数据保存到制品存储并持久化到会话历史转录与工具调用非分片non-partial转录、用量元数据与函数调用始终写入会话历史。非阻塞后台工具Non-Blocking Background Toolslive 流期间工具函数被调用时runner 将工具调用派发到后台任务音频输出不被阻塞工具完成后结果被推回LiveRequestQueue更新模型。LiveRequest 的优先级语义LiveRequestQueue传输的LiveRequest模型源码见 live_request_queue.py在多个字段同时设置时按如下优先级处理activity_startactivity_endaudio_stream_endblobcontent而state_delta无论其他字段如何设置都会被应用。队列方法映射如下方法底层请求用途send_content(content, partialFalse)LiveRequest(content...)回合制模式发送文本/多模态内容partialTrue表示部分轮次更新send_realtime(blob)LiveRequest(blob...)实时模式发送音频/视频 Blobsend_activity_start()LiveRequest(activity_start...)标记用户输入开始send_activity_end()LiveRequest(activity_end...)标记用户输入结束send_audio_stream_end()LiveRequest(audio_stream_endTrue)强制冲刷音频、结束音频流配合 VAD 使用close()LiveRequest(closeTrue)关闭队列终止处理send(req)任意LiveRequest直接入队自定义请求一个与 VAD语音活动检测相关的常见误区当 Gemini Live API 的 VAD 生效时模型会自动检测用户话语的起止不需要在每个话语末尾发送audio_stream_end该方法只用于通知后端整个音频输入流已结束例如用户关闭/静音麦克风时请勿在每个对话回合后调用否则需要新的音频消息才能重新打开流。更多细节可参考 LiveRequestQueue 指南。run_live 配置参数详解run_live接受以下配置参数与 runners.py 的签名一致参数类型默认值说明user_idstr \| NoneNone会话的用户 IDsession为None时必填session_idstr \| NoneNone会话 IDsession为None时必填live_request_queueLiveRequestQueue必填用于向会话推送实时用户输入、音频分片与工具结果的队列run_configRunConfig \| NoneNone执行配置含speech_config、response_modalities、save_live_blob等sessionSession \| NoneNone预取的会话实例已弃用推荐使用user_id与session_id注意源码中的校验逻辑session为None时必须同时提供user_id和session_id否则抛出ValueErrorlive_request_queue为None同样直接报错。传入session会触发DeprecationWarning。另外run_live的根节点既可以是BaseAgent也可以是非 Agent 的BaseNode如Workflow——后者会走run_node_live分支经由DynamicNodeScheduler驱动根节点并合并事件流。RunConfig Live 选项通过run_configRunConfig(...)配置字段定义位于 run_config.py选项类型默认值说明speech_configtypes.SpeechConfig \| NoneNonelive Agent 的语音选择与音频编码配置response_modalitieslist[types.Modality] \| NoneNone模型返回的输出模态AUDIO或TEXT未设置时默认AUDIOsave_live_blobboolFalse将 live 视频与音频数据保存到会话与制品服务session_resumptiontypes.SessionResumptionConfig \| NoneNone配置透明的会话恢复机制tool_thread_pool_configToolThreadPoolConfig \| NoneNone在后台线程池执行器中运行工具保持事件循环响应从源码注释可以补充几个值得注意的细节tool_thread_pool_config设置为非空时工具在独立线程池执行器中运行而非主事件循环默认None时在主事件循环运行同一事件循环上的所有调用共享一个线程池该池随事件循环销毁而关闭工作线程不会存活到事件循环之外。这能防止耗时工具阻塞音频事件的实时投递。音频转录配置input_audio_transcription/output_audio_transcription默认均为AudioTranscriptionConfig控制输入/输出音频的转录对于带sub_agents的 live 多 Agent 系统_runner_utils.py 会自动补齐转录配置因为子 Agent 需要模型文本转录作为上下文。其他 live 相关选项explicit_vad_signal显式 VAD 信号、translation_config实时语音到语音翻译仅受翻译类模型支持、enable_affective_dialog情感检测、proactivity模型主动响应以及streaming_modeNONE/SSE/BIDI等也可一并配置。高级应用非阻塞流式工具回调工具函数可以声明input_stream: LiveRequestQueue参数在后台运行期间把部分工具结果或状态更新流式回传给 live 会话from google.adk.live import LiveRequestQueue from google.genai import types async def fetch_stock_ticker( symbol: str, input_stream: LiveRequestQueue ) - dict[str, float]: Fetches live stock price while streaming progress. # 通知 live 模型会话查询已在进行中 input_stream.send_content( contenttypes.Content( roleuser, parts[ types.Part.from_text( textfFetching latest price for {symbol}... ) ], ) ) # 执行查询 return {symbol: symbol, price: 154.25}关键点这里的send_content作用对象是模型的输入流模拟用户发言使模型能够听到工具正在工作的状态从而在等待期间继续与用户自然对话。工具最终返回值仍然作为函数响应进入原有工具结果回路。仓库中的 live_non_blocking_tool_agent 示例 展示了完整模式用FunctionTool包装一个await asyncio.sleep(10)的慢任务并通过types.FunctionResponseScheduling.WHEN_IDLE声明非阻塞调度策略可选值还包括SILENT、INTERRUPT使 Agent 在任务执行期间持续与用户交谈而不中断音频流。该示例使用的模型为gemini-live-2.5-flash-native-audio可替换为你所用环境支持的 Live 模型。局限性与注意事项Gemini Live 模型要求run_live需要支持 Gemini Multimodal Live API 的模型端点例如gemini-2.0-flash-exp普通生成式模型无法走此路径。内联音频不持久化原始 PCM 音频 Blobinline_data被有意排除在会话存储之外。若需要保留会话的音频/视频历史请在RunConfig上设置save_live_blobTrue。实验性 API源码文档字符串明确标注 live 功能为experimentalrunners.py其 API 或行为可能在后续版本中变更session参数已弃用应改用user_idsession_id。相关指南与示例Runner 与 InMemoryRunner 指南 —— 标准回合制 runner 执行的主指南LiveRequestQueue 指南 —— 实时输入排队、音频分片与非阻塞流式工具App 容器指南 —— 将 Agent 与插件打包进AppLive 双向流式单 Agent 示例 —— 单 Agent 实时流式应用样例Live 非阻塞工具 Agent 示例 —— 在后台工具回调中使用LiveRequestQueue的样例以上示例均位于 contributing/samples/live 目录可结合 live 相关文档docs/guides/live一起阅读完整掌握 ADK 实时流式 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），仅供参考