OCT. 19, 2026 (英语).
Agent Development Kit(ADK) Gemini Live API工具包中的流媒体工具指南，非屏蔽功能调用，以及响应调度。
在文本聊天中，一个13秒的数据库查询或RAG管道显示是一个加载旋转器。 在现场语音对话中，13秒的屏蔽工具执行产生13秒的死气，导致用户假设呼叫被放弃并挂断。
你不能总是做后端查询子秒，所以修补必须是建筑：保持agent对话，同时在背景中慢化工具运行。 开源 ADK和gemini-3.8-live没有强迫您连接自定义客户端-WebSocket事件环路来管理同步工具状态，而是允许您在Python本土执行非屏蔽语音工具。 在我们的基准测试销售-报告工作流程中，在gemini-3.8-live上，从阻塞工具切换到流线工具将时间从14.4秒缩短到agent的第一个单词1.2秒。
本指南涵盖如何使用流线工具，非屏蔽函数工具定义，以及响应 scheduling来构建响应性，自然的声音agents.
观看1分钟的ADK Live AI Concierge演示视频
ADK Live AI Concierge 演示视频解释计划响应如何运作
在ADK中实施非屏蔽工具调用可提供几个即时用户体验的好处：
- Zero Dead Air:agent不是冻结中间对话，而是可以立即确认请求("现在查看这些数字。.....")，并保持对话的活力。
- Audible Progress Updates：对于多阶段工作流程(例如连接，查询，和调和),agent可以描述临时里程碑，因此等待听起来像积极的进展。
- 多任务和平行对话：用户可以在背景中执行一个长期运行的查询时，询问后续或不相关的问题，并及时得到答案而不会耽误会话。
- Granular 交付控制：您确定最终结果何时以及如何进入对话：等待自然的暂停发言，立即中断关键提示，或默默更新对话上下文。
在标准agent开发中，一个工具被定义为一个async def函数，可以完成它的工作并返回一个单一值：
从输入中导入
aync def 查询 sales 数据库(区域：str) - > dict[str, Any]:
""运行一个区域的季度销售报告查询。
参数：
区域：报告销售区域，例如“EMEA”或“APAC”。
回返情况：
完成后包含报告指标的命令。
"""
报告 = 等待 run excensive query(区域) #，例如工作13秒
返回 {“状态”: “成功”、“ 报告”：报告} {\fn方正粗倩简体\fs12\an8\1cHFFFF00\b0}最后的一次回归
纯文本
复制
当直接传递到ADK中的Agent时，此函数会创建一个阻塞工具调用。 在Gemini Live API双向语音会话(run live)期间，执行循环等待工具返回后再生成进一步的模型转折。
如果查询 sales 数据库需要13秒，agent会产生13秒的完全沉默。 诸如"告诉您正在研究的用户"等提示指令无法解决这一点，因为模型的转弯被阻塞等待函数响应有效载荷。
为了保持对话的流畅性，您可以根据期望的用户体验，在两种主要的非阻断模式之间做出选择。
当一个操作包含有意义的阶段(如连接到数据库，汇总记录，以及格式化结果)时，一个流线工具让agent在其发生时描述进度。
在ADK中，将标准工具变成流化工具需要只改变返回到收益。 通过将函数定义为 AsyncGenerator,ADK 自动将工具声明标记为 NON BLOCKING，并将即时状态确认返回到模型中。
导入 Ayncio
从输入输入 AsyncGenerator
从 google.adk.agents 导入 Agent
async def 查询 sales 数据库( 区域：str) - > AsyncGenerator[str, no]:
""运行季度销售报告。 调用此 ONCE；它自动推动更新。
参数：
区域：报告销售区域，例如“EMEA”或“APAC”。
仪表：
在查询运行时更新进度，然后是最后摘要。
"""
阶段 = [
("连接到销售仓库。..", 1.5)
("扫描订单记录。..",3.5)
("按产品线分列收入。..",3.5).
("校对报告。..", 1.5)
]
对于信件，分阶段延迟：
输出 f: {消息}'
等待 Ayncio.sleep(延迟)
产量f"DONE:{地区}Q2收入达到481M，年增长12.4%.
agent = Agent(
名称="销售分析"
型号="gemini-3.8-live"
工具=[query sales database]
指令=( )
"你是一个有用的销售助理。 当用户询问销售数据时，"
"呼叫查询 sales 数据库一次。 立即确认这一请求，"
"Briefly解释每一个进展更新，并提出最后的数字"
"当完成。 保持答案的对话和简洁".
),
)
- 立即确认：当模型触发查询 sales 数据库时，ADK立即向模型发送同步状态事件。 agent立刻说话("现在为您查看EMEA的数字。.....").
- 部分更新：每个输出向上游发送部分函数响应。 模型在每次更新到达时都会解释("连接到仓库。..","聚合产品线数据。.....")，虽然叙述可以跟踪工具几秒钟。
– 无缝多轨：如果用户在查询运行时询问一个无关的问题，agent可以在不取消或中断背景任务的情况下回答。
- 单一闪光工具呼叫：由于流线工具将更新推向多个转弯，模型有时可能会试图再次调用工具，如果反复被催动。 在您的文档中保留清晰的“ 调用此 ONCE” 指令，并考虑在您用于安全生产的工具执行中执行一个内链键或同位素键。
- 用户取消 " stop streaming ":ADK为长期流线工具提供内置逃生舱。 通过在 agent 的工具列表中包含一个名为 stop streaming(函数 name: str) 的函数，ADK 截取呼叫并取消生成器的内在 Ayncio. 任务，抚养Ayncio。 功能内已取消错误，这样您就可以清理关闭数据库连接或文件控件。
- 用户 Barge- in 不取消工具：当用户在 agent 上交谈时，Live API 停止了模型的输出，ADK 发出一个中断的事件 = 让你的客户停止播放音频运行的流或背景工具不会被取消，其结果仍然被交付。 要阻止一个工具，模型必须叫停止流体。
持续的进展评注并非总是适当的。 如果用户要求报告，并立即向另一个话题吐槽，持续中断状态更新的agent会感到侵入性。
当等待没有故事可以叙述时，在背景中悄悄地执行任务。
一个无需不断进展的立即口头承认的有效方式是，一个同步生成器，完成后只产生一次：
aync def 查询 sales 数据库(区域：str) - > AsyncGenerator[dit, no]:
""运行季度销售报告。 这叫做ONCE；结果一准备好就到了。"
报表 = 等待 run expensive query(区域)# 背景执行(~13s)
产量报告# 完成时的单一产量
由于这是一个Async生成器，ADK发布一个即时响应。 agent迅速确认请求("我现在为您把报告拉在一起")，在计算时保持缄默，并在准备好或用户完成演讲时给出数字。
如果您更愿意保留标准皮层( async def... - > dict)，将函数包裹在一个函数工具中，并配置响应排程：
从 google.adk.tools 导入函数工具
从 google.genai 导入类型
销售工具=函数工具(query sales database)
销售工具。 Reswer scheduling = 类型。 FunctionreseScheduling.WHEN IDLE
(注：在ADK 2.10+中，也可以设置‘sales tool.beavior = 类型。 Behavior.NON BLOCKING"在背景中运行一个工具而不修改调度模式。 )
了解Trade-Off：在使用回转式皮层反应时，模型在完成整个功能之前不会收到有效载荷。 因此，除非用户继续发言，否则模型通常不会表示初步口头承认。 当用户立即继续处理其他问题时，这个模式是理想的，您想要在下一次自然暂停时将结果编织到对话框中。
当模型 reacts 到一个工具响应时，响应 scheduling 属性控制。 当 IDLE从不将削减变成进动的转弯：如果模型还在说话，或者用户还在说话，结果会等到那个转弯结束，模型会在下一个自然暂停时响应。 ADK支持三种调度行为：
在基准测试会话中，使用gemini-3.8-live:
- 在When IDLE下，收到一份销售报告礼貌地等待，而模型完成了一个单独的20秒的问题，然后无缝地发送数字。
- 在INTRUPT下，即将到来的结果切断了目前中刑的言论；模型随后完成了答复，并附上了数字。
- 在SILENT下，结果被添加到对话历史中，可供以后的回合使用，而无需催促宣布。
执行形状(流对单返回)和交付时间(Resignation scheduling)是独立的，可以组成。 设定响应排程在流媒体工具印章上每一个部分更新。
此外，在ADK 2.8和较新的版本中，流线工具可以通过生成类型来动态指定每次更新的时间安排。 函数响应：
承认并保持中间进展安静：
产量类型。 函数响应( R)
反应 QQ"阶段":"汇总"]
排程=类型。 FunctionResponse Scheduling.SILENT (英语).
)
{\fn方正粗倩简体\fs12\an8\1cHFFFF00\b0}在演讲完毕后提供最后的结果
产量类型。 函数响应( R)
回复 QQ“ 报告 ”: final data}
排程=类型。 功能响应排程。 WHEN IDLE
)
拇指规则：如果等待有一个故事，流它。 如果等待只是一个等待，在背景中运行，让结果在闲置时着陆。
非屏蔽工具不会减少后端执行时间：13秒查询仍需要13秒。 变化是您agent的谈话。
一个需要10秒才能加载的网络应用程序被认为是慢的；一个会说话的声音agent会默默10秒就被认为是破碎的。 通过将ADK的流动工具和灵活的响应调度与gemini-3.8-live的实时能力配对，可以建立声音助手，这些助手保持注意力，自然，并贯穿整个复杂的操作。
自己试一下：交互演示通过gemini-3.8-live上的每个工具模式和响应排程模式运行相同的13秒查询，这样就可以听到差异。
- 框架：google-adk 2.1.0
- 型号：gemini-3.8-在Gemini企业上有效
- API环境：流线工具和直播调度需要双向直播(运行 live).
本文由 AI图鉴自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。
OCT. 19, 2026
A guide to streaming tools, non-blocking function calls, and response scheduling in the Agent Development Kit (ADK) Gemini Live API Toolkit.
In text chat, a 13-second database query or RAG pipeline shows a loading spinner. In a live voice conversation, 13 seconds of blocking tool execution produces 13 seconds of dead air, causing users to assume the call dropped and hang up.
You cannot always make backend queries sub-second, so the fix must be architectural: keep the agent conversational while slow tools run in the background. Rather than forcing you to wire up custom client-side WebSocket event loops to manage asynchronous tool state, the open-source ADK and gemini-3.8-live let you implement non-blocking voice tools natively in Python. In our benchmark sales-report workflow on gemini-3.8-live, switching from a blocking tool to a streaming tool cut the time to the agent's first words from 14.4 seconds to 1.2 seconds.
This guide covers how to use streaming tools, non-blocking FunctionTool definitions, and response_scheduling to build responsive, natural voice agents.
Watch the 1-minute ADK Live AI Concierge demo video to hear scheduled responses in action:
ADK Live AI Concierge demo video explains how scheduled responses work
Implementing non-blocking tool calls in ADK provides several immediate user experience benefits:
- Zero Dead Air: Instead of freezing mid-conversation, the agent can acknowledge requests immediately ("Looking into those numbers now...") and keep the dialogue alive.
- Audible Progress Updates: For multi-stage workflows (such as connecting, querying, and reconciling), the agent can narrate interim milestones so the wait sounds like active progress.
- Multitasking and Parallel Dialogue: Users can ask follow-up or unrelated questions while a long-running query executes in the background, and receive timely answers without stalling the session.
- Granular Delivery Control: You determine when and how the final result enters the conversation: waiting for a natural speaking pause, interrupting immediately for critical alerts, or silently updating conversation context.
In standard agent development, a tool is defined as an async def function that completes its work and returns a single value:
from typing import Any
async def query_sales_database(region: str) -> dict[str, Any]:
"""Runs the quarterly sales report query for a region.
Args:
region: Sales region to report on, e.g. "EMEA" or "APAC".
Returns:
A dict containing report metrics upon completion.
"""
report = await run_expensive_query(region) # e.g., 13 seconds of work
return {"status": "success", "report": report} # Single return at the end
Plain text
Copied
When passed directly to an Agent in ADK, this function creates a blocking tool call. During a Gemini Live API bidirectional voice session (run_live), the execution loop awaits the tool's return before generating further model turns.
If query_sales_database takes 13 seconds, the agent produces 13 seconds of complete silence. Prompting instructions such as "tell the user you are working on it" cannot resolve this, because the model's turn is blocked waiting for the function response payload.
To keep the conversation fluid, you can choose between two main non-blocking patterns depending on the desired user experience.
When an operation consists of meaningful stages (such as connecting to a database, aggregating records, and formatting results), a streaming tool lets the agent narrate progress as it happens.
In ADK, turning a standard tool into a streaming tool requires changing only return to yield. By defining the function as an AsyncGenerator, ADK automatically marks the tool declaration as NON_BLOCKING and returns an immediate status acknowledgement to the model.
import asyncio
from typing import AsyncGenerator
from google.adk.agents import Agent
async def query_sales_database(region: str) -> AsyncGenerator[str, None]:
"""Runs the quarterly sales report. Call this ONCE; it pushes updates automatically.
Args:
region: Sales region to report on, e.g. "EMEA" or "APAC".
Yields:
Progress updates while the query runs, followed by the final summary.
"""
stages = [
("Connecting to the sales warehouse...", 1.5),
("Scanning order records...", 3.5),
("Aggregating revenue by product line...", 3.5),
("Formatting the report...", 1.5),
]
for message, delay in stages:
yield f"PROGRESS: {message}"
await asyncio.sleep(delay)
yield f"DONE: {region} Q2 revenue reached $4.81M, up 12.4% year over year."
agent = Agent(
name="sales_analyst",
model="gemini-3.8-live",
tools=[query_sales_database],
instruction=(
"You are a helpful sales assistant. When the user asks about sales data, "
"call query_sales_database once. Acknowledge the request immediately, "
"briefly paraphrase each progress update, and present the final figures "
"when completed. Keep answers conversational and concise."
),
)
- Immediate Acknowledgment: When the model triggers query_sales_database, ADK immediately dispatches an asynchronous status event to the model. The agent speaks right away ("Looking into the EMEA figures for you now...").
- Partial Updates: Each yield sends a partial function response upstream. The model paraphrases each update as it arrives ("Connecting to the warehouse...", "Aggregating product line data..."), though narration can trail the tool by several seconds.
- Seamless Multi-Threading: If the user asks an unrelated question while the query is running, the agent can answer it without canceling or interrupting the background task.
- Single-Flight Tool Calls: Because streaming tools push updates across multiple turns, models may occasionally attempt to call the tool again if prompted repeatedly. Keep a clear "Call this ONCE" directive in your docstring, and consider implementing an in-memory latch or idempotency key in your tool implementation for production safety.
- User Cancellation with `stop_streaming`: ADK provides a built-in escape hatch for long-running streaming tools. By including a function named stop_streaming(function_name: str) in the agent's tools list, ADK intercepts the call and cancels the generator's underlying asyncio.Task, raising asyncio.CancelledError within your function so you can cleanly close database connections or file handles.
- User Barge-In Does Not Cancel Tools: When the user talks over the agent, the Live API stops the model's output and ADK emits an event with interrupted=True so your client can stop audio playback. Running streaming or background tools are not cancelled, and their results are still delivered. To stop a tool, the model must call stop_streaming.
Continuous progress commentary is not always appropriate. If a user asks for a report and immediately pivots to another topic, an agent that continually interrupts with status updates can feel intrusive.
When the wait has no story to narrate, run the task quietly in the background.
An effective way to get an immediate verbal acknowledgment without ongoing progress chatter is an async generator that yields only once, upon completion:
async def query_sales_database(region: str) -> AsyncGenerator[dict, None]:
"""Runs the quarterly sales report. Call this ONCE; results arrive when ready."""
report = await run_expensive_query(region) # Background execution (~13s)
yield report # Single yield upon completion
Because this is an async generator, ADK issues an immediate pending response. The agent promptly acknowledges the request ("I'm pulling that report together for you now"), remains silent during computation, and presents the figures once ready or when the user finishes speaking.
If you prefer to keep standard coroutines (async def ... -> dict), wrap the function in a FunctionTool and configure response_scheduling:
from google.adk.tools import FunctionTool
from google.genai import types
sales_tool = FunctionTool(query_sales_database)
sales_tool.response_scheduling = types.FunctionResponseScheduling.WHEN_IDLE
(Note: In ADK 2.10+, you can also set `sales_tool.behavior = types.Behavior.NON_BLOCKING` to run a tool in the background without modifying scheduling mode.)
Understanding the Trade-Off: When using response_scheduling with a return-once coroutine, the model receives no payload until the entire function completes. As a result, the model typically does not utter an initial verbal acknowledgment unless the user continues speaking. This pattern is ideal when the user immediately continues with other questions and you want the result woven into the dialogue at the next natural pause.
The response_scheduling attribute controls when the model reacts to an incoming tool response. WHEN_IDLE never cuts into a turn in progress: if the model is still speaking or the user is still talking, the result waits until that turn ends, and the model responds at the next natural pause. ADK supports three scheduling behaviors:
In benchmark sessions with gemini-3.8-live:
- Under WHEN_IDLE, an incoming sales report waited politely while the model finished answering a separate 20-second question, then delivered the figures seamlessly.
- Under INTERRUPT, the incoming result cut off current speech mid-sentence; the model then finished its answer and appended the figures.
- Under SILENT, the result was added to the conversation history, available for subsequent turns without prompting an unrequested announcement.
Execution shape (streaming vs. single-return) and delivery timing (response_scheduling) are independent and can be composed. Setting response_scheduling on a streaming tool stamps every partial update with that behavior.
Furthermore, in ADK 2.8 and newer, streaming tools can dynamically specify scheduling per update by yielding types.FunctionResponse:
# Acknowledge and keep intermediate progress quiet:
yield types.FunctionResponse(
response={"stage": "aggregating"},
scheduling=types.FunctionResponseScheduling.SILENT,
)
# Deliver the final result when the speaker finishes:
yield types.FunctionResponse(
response={"report": final_data},
scheduling=types.FunctionResponseScheduling.WHEN_IDLE,
)
The rule of thumb: If the wait has a story, stream it. If the wait is just a wait, run it in the background and let the result land when idle.
Non-blocking tools do not reduce backend execution time: a 13-second query still takes 13 seconds. What changes is the conversational presence of your agent.
A web application that takes ten seconds to load is perceived as slow; a conversational voice agent that goes silent for ten seconds is perceived as broken. By pairing ADK's streaming tools and flexible response scheduling with the real-time capabilities of gemini-3.8-live, you can build voice assistants that remain attentive, natural, and present throughout complex operations.
Try it yourself: the interactive demo runs the same 13-second query through every tool pattern and response_scheduling mode on gemini-3.8-live, so you can hear the difference.
- Framework: google-adk >= 2.10.0
- Model: gemini-3.8-live on Gemini Enterprise
- API Environment: Streaming tools and live scheduling require bidirectional live sessions (run_live).
Auto-fetched and translated by AI Atlas. All rights belong to the original author.