基于 ADK 与 Gemini 3.8 Live 构建永不静默的语音 Agent

关于 流式工具（streaming tools）、非阻塞函数调用以及 Agent Development Kit (ADK) Gemini Live API Toolkit 中响应调度的实战指南。

在文本对话中，耗时 13 秒的数据库查询或 RAG 流程通常只会展示一个加载转轮（spinner）；但在实时语音通话中，长达 13 秒的阻塞式工具调用则意味着 13 秒的彻底死寂（dead air），这往往会让用户误以为通话掉线而直接挂断。

你不可能总能把后端查询优化到亚秒级别，因此解决之道必须依赖架构层面：在慢速工具于后台运行的同时，让 Agent 保持对话状态。开源的 ADK 与 gemini-3.8-live 无需开发者自行编写复杂的客户端 WebSocket 事件循环来管理异步工具状态，而是允许直接在 Python 中原生实现非阻塞语音工具。在 gemini-3.8-live 的销售报告基准工作流中，将阻塞式工具切换为流式工具后，Agent 的首字响应延迟从 14.4 秒直接缩短到了 1.2 秒。

本指南将详细介绍如何使用流式工具（Streaming Tools）、非阻塞式 FunctionTool 定义以及 response_scheduling 来构建响应迅速、自然的语音 Agent。

观看 1 分钟的 ADK Live AI Concierge 演示视频，直观感受调度响应机制的工作流程：

视频/嵌入内容：ADK Live AI Concierge 演示视频，解析调度响应（Scheduled Responses）机制

在 ADK 中实现非阻塞工具调用可带来立竿见影的用户体验提升：

• 零死寂（Zero Dead Air）： Agent 不会在对话中途卡死，而是能立即确认请求（如“正在为您调取这些数据……”），让对话持续处于活跃状态。
• 语音化进度播报（Audible Progress Updates）： 针对多阶段工作流（如连接、查询、核对），Agent 可以实时口述中间进展，使等待过程听起来像是在主动推进。
• 多任务与并行对话（Multitasking and Parallel Dialogue）： 后台执行长耗时查询期间，用户仍可提出后续问题或毫不相干的问题，并及时获得解答，会话不会因此停滞。
• 精细化结果交付控制（Granular Delivery Control）： 你可以精准决定最终结果切入对话的时机与方式：等待自然的停顿、针对紧急通知立即打断当前说话，或是静默更新对话上下文。

在传统的 Agent 开发中，工具通常被定义为一个完成任务后返回单一值的 async def 函数：

python 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)       # 例如耗时 13 秒的操作
return {"status": "success", "report": report}   # 最终单一返回

当将其直接传递给 ADK 的 Agent 时，该函数会引发阻塞式工具调用。在 Gemini Live API 的双向语音会话（run_live）期间，执行循环会一直等待工具返回结果，然后才会生成后续的模型轮次。

如果 query_sales_database 耗时 13 秒，Agent 就会陷入 13 秒的完全静音。即使在提示词中加上“告诉用户你正在处理”也无济于事，因为模型的轮次已经被阻塞在等待函数响应 payload 的过程中。

为了保持对话流畅，你可以根据所需的用户体验在两种主要的非阻塞模式之间进行选择。

模式 1：流式工具（当等待过程有故事可讲时）

如果一个操作包含有意义的阶段划分（例如连接数据库、汇总记录、格式化结果），**流式工具（streaming tool）**可以让 Agent 实时播报当前进度。

在 ADK 中，将标准工具转换为流式工具只需将 return 改为 yield。通过将函数定义为 AsyncGenerator，ADK 会自动将该工具声明标记为 NON_BLOCKING，并向模型返回即时的状态确认。

python 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." ), )

实际运行机制

1. 即时确认（Immediate Acknowledgment）： 当模型触发 query_sales_database 时，ADK 会立刻向模型派发一个异步状态事件。Agent 随即可直接开口回应（“正在为您查询 EMEA 区域的数据……”）。
2. 部分更新（Partial Updates）： 每次 yield 都会向上游发送一个部分函数响应。模型在接收到更新时会对其进行转述（“正在连接数据仓库……”、“正在聚合产品线数据……”），不过口述进度可能会落后工具实际进度数秒。
3. 无缝多线程交互（Seamless Multi-Threading）： 若用户在查询运行时提出无关问题，Agent 可以直接回答，而不会取消或打断后台任务。

• 单飞轮工具调用（Single-Flight Tool Calls）： 由于流式工具会跨多个对话轮次推送更新，如果用户反复提示，模型有时可能会尝试再次调用该工具。请务必在 docstring 中保留明确的“Call this ONCE”指令，并在生产环境中考虑在工具实现内加入内存锁（in-memory latch）或幂等键（idempotency key）以确保安全。
• 通过 stop_streaming 实现用户取消： ADK 为长时间运行的流式工具提供了内置的退出机制。只需在 Agent 的 tools 列表中包含一个名为 stop_streaming(function_name: str) 的函数，ADK 就会拦截该调用并取消生成器底层的 asyncio.Task，在你的函数内部抛出 asyncio.CancelledError，以便你妥善关闭数据库连接或文件句柄。
• 用户打断（Barge-In）不会取消工具： 当用户在 Agent 发声时插话，Live API 会停止模型的音频输出，ADK 会发出一个 interrupted=True 事件以便客户端停止音频播放。但正在运行的流式或后台工具不会被取消，其结果仍会正常交付。若要中止工具，模型必须显式调用 stop_streaming。

模式 2：后台执行（当等待只是单纯的等待时）

持续播报进度并不总是适宜的。如果用户请求了一份报告后立刻转入另一个话题，Agent 不断插话通报状态反而会造成打扰。

当等待过程没有值得讲述的情节时，应当让任务在后台静默运行。

方案 A：单次 Yield 生成器（最快速配置）

要想既能获得即时的口头确认，又无需中途的碎碎念进度播报，一个高效的方法是使用仅在完成时 yield 一次的异步生成器：

python 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) # 后台执行（约 13 秒） yield report # 仅在完成时 yield 一次

因为这是一个异步生成器，ADK 会立刻发出 pending 状态响应。Agent 能立刻确认请求（“我正在为您生成报告”），在计算期间保持静默，并在准备就绪或用户发言完毕后呈现数据。

如果你倾向于保留标准协程（async def ... -> dict），可以将函数包装进 FunctionTool 并配置 response_scheduling：

python 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

（注意：在 ADK 2.10+ 中，你还可以设置 sales_tool.behavior = types.Behavior.NON_BLOCKING，在不改变调度模式的情况下在后台运行工具。）

理解权衡取舍： 当在单次返回的协程上使用 response_scheduling 时，模型在整个函数完成前收不到任何 payload。因此，除非用户继续说话，否则模型通常不会给出最初的口头确认。当用户提问后紧接着提出其他问题、且你希望在下一个自然停顿将结果融入对话时，这种模式最为理想。

利用 response_scheduling 控制结果交付

response_scheduling 属性控制着模型对到达的工具响应在何时做出反应。WHEN_IDLE 绝不会打断进行中的轮次：如果模型正在说话或用户正在发音，结果会耐心等待该轮次结束，并在下一个自然停顿时由模型做出回应。ADK 支持三种调度行为：

在结合 gemini-3.8-live 的基准测试中：

• 在 WHEN_IDLE 模式下，当模型正在回答另一个耗时 20 秒的问题时，新收到的销售报告会礼貌等待，直到解答完毕后再无缝呈现报告数据。
• 在 INTERRUPT 模式下，新结果会在句子中途打断当前发言；随后模型会结束之前的回答并附上最新数据。
• 在 SILENT 模式下，结果会被静默添加到对话历史记录中，供后续轮次使用，而不会触发未经要求的公告播报。

流式与调度的结合使用

执行形态（流式 vs 单次返回）与交付时机（response_scheduling）是正交独立的，可以自由组合。在流式工具上设置 response_scheduling 会将该行为赋予每一个部分更新。

此外，在 ADK 2.8 及更高版本中，流式工具可以通过 yield types.FunctionResponse 来动态指定每次更新的调度行为：

python

确认并保持中间进度静默：

yield types.FunctionResponse( response={"stage": "aggregating"}, scheduling=types.FunctionResponseScheduling.SILENT, )

当说话者停顿时交付最终结果：

yield types.FunctionResponse( response={"report": final_data}, scheduling=types.FunctionResponseScheduling.WHEN_IDLE, )

架构总结：如何选择恰当的工具模式

经验法则：如果等待过程有故事可讲，就用流式（Streaming）；如果等待就只是单纯的等待，就置于后台运行，让结果在空闲时切入。

结语

非阻塞式工具并不会减少后端的实际执行时间：耗时 13 秒的查询仍然需要 13 秒。真正改变的是 Agent 的对话在场感（conversational presence）。

加载需要十秒的 Web 应用只会被认为速度慢；但若一个对话式语音 Agent 静音十秒，则会被用户直接判定为瘫痪故障。通过将 ADK 的流式工具与灵活的响应调度，与 gemini-3.8-live 的实时能力相结合，开发者便能打造出在复杂业务操作全程中始终保持专注、自然且始终在场的语音助理。

动手体验： 欢迎通过 交互式 Demo 体验，该 Demo 在 gemini-3.8-live 上针对同一条 13 秒查询分别运行所有工具模式与 response_scheduling 选项，方便你直观对比其差异。

依赖环境与后续步骤

• 框架要求： google-adk >= 2.10.0
• 适用模型： Gemini Enterprise 上的 gemini-3.8-live
• API 环境： 流式工具与实时调度依赖双向实时会话（run_live）。

参考链接

Building a Voice Agent That Never Goes Silent with ADK and Gemini 3.8 Live

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:

Sorry, your browser doesn't support playback for this video

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

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.

Pattern 1: Streaming Tools (When Progress Has a Story)

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."
    ),
)

Plain text

How It Works in Practice

1. 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...").
2. 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.
3. 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.

Pattern 2: Background Execution (When Waiting Is Just Waiting)

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.

Approach A: The Single-Yield Generator (Fastest Setup)

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

Plain text

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.)

Plain text

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.

Controlling Result Delivery with response_scheduling

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.

Combining Streaming and Scheduling

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,
)

Plain text

Architecture Summary: Choosing the Right Tool Pattern

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.

Conclusion

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.

Requirements & Next Steps

• 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).

References