Google 讲解用 ADK 和 gemini-3.8-live 构建不冷场的语音 Agent

2026年10月19日

Stephen Allen 解决方案架构师 AI 应用与平台

Kaz Sato Staff 开发者关系倡导者

使用 ADK 和 Gemini 3.8 Live 构建永不沉默的语音代理

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

在文本聊天中，一次 13 秒的数据库查询或 RAG 流水线只会显示一个加载旋转图标。而在实时语音对话中，13 秒的阻塞式工具执行会产生 13 秒的死寂，导致用户以为通话已断开而挂断。

你无法总是让后端查询达到亚秒级响应，因此解决方案必须是架构层面的：在慢速工具于后台运行时，让代理保持对话进行。开源 ADK 和 gemini-3.8-live 无需你自行编写客户端 WebSocket 事件循环来管理异步工具状态，而是让你可以用 Python 原生地实现非阻塞语音工具。在我们基于 gemini-3.8-live 的基准销售报告工作流中，从阻塞式工具切换到流式工具，将代理首次开口的时间从 14.4 秒缩短到了 1.2 秒。

本指南介绍如何使用流式工具、非阻塞 FunctionTool 定义以及 response_scheduling 来构建响应迅速、自然的语音代理。

观看这段 1 分钟的 ADK Live AI Concierge 演示视频，亲耳听听调度响应是如何工作的：

抱歉，你的浏览器不支持播放此视频

ADK Live AI Concierge 演示视频解释了调度响应是如何工作的

非阻塞工具能带来什么

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

零死寂：代理不会在对话中途卡住，而是可以立即回应请求（“正在查看那些数字……”）并让对话保持进行。
可听见的进度更新：对于多阶段工作流（例如连接、查询和核对），代理可以播报中间里程碑，让等待听起来像是在积极进展。
多任务与并行对话：用户可以在长时间查询于后台执行时提出后续或无关问题，并及时获得答复，而不会让会话停滞。
细粒度交付控制：你可以决定最终结果何时以及如何进入对话：等待一个自然的说话停顿、针对关键警报立即打断，或者静默更新对话上下文。

基线：为什么标准工具会阻塞对话

在标准代理开发中，工具被定义为一个 async def 函数，它完成工作并返回单个值：

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

纯文本

已复制

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

如果 query_sales_database 需要 13 秒，代理就会产生 13 秒的完全沉默。诸如“告诉用户你正在处理”这样的提示指令无法解决这个问题，因为模型的轮次正被阻塞，等待函数响应负载。

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

模式 1：流式工具（当进度有故事可讲时）

当一个操作由有意义的阶段组成（例如连接数据库、聚合记录、格式化结果）时，流式工具让智能体能够在操作进行时实时叙述进度。

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

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

纯文本

已复制

实际工作方式

即时确认：当模型触发 query_sales_database 时，ADK 会立即向模型发送一个异步状态事件。智能体会立刻开口（“正在为您查看 EMEA 的数据……”）。
部分更新：每个 yield 都会向上游发送一个部分函数响应。模型会在每个更新到达时进行转述（“正在连接数据仓库……”、“正在聚合产品线数据……”），不过叙述可能会比工具滞后几秒钟。
无缝多线程：如果用户在查询运行期间提出一个不相关的问题，智能体可以回答该问题，而不会取消或中断后台任务。

流式工具的生产环境注意事项

单次飞行工具调用：由于流式工具会跨多个轮次推送更新，模型在反复提示时偶尔可能会尝试再次调用该工具。请在 docstring 中保持明确的“仅调用此工具一次”指令，并考虑在工具实现中实现内存闩锁或幂等键，以确保生产环境的安全性。
使用 `stop_streaming` 进行用户取消：ADK 为长时间运行的流式工具提供了内置的应急出口。通过在智能体的工具列表中包含一个名为 stop_streaming(function_name: str) 的函数，ADK 会拦截该调用并取消生成器底层的 asyncio.Task，在你的函数中引发 asyncio.CancelledError，以便你可以干净地关闭数据库连接或文件句柄。
用户插话不会取消工具：当用户在智能体说话时插话，Live API 会停止模型的输出，ADK 会发出一个带有 interrupted=True 的事件，以便你的客户端可以停止音频播放。正在运行的流式或后台工具不会被取消，其结果仍会被传递。要停止某个工具，模型必须调用 stop_streaming。

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

连续不断的进度解说并不总是合适的。如果用户要求生成一份报告，然后立即转向另一个话题，一个不断用状态更新打断用户的智能体可能会让人觉得烦扰。

当等待没有什么可叙述的故事时，就让任务在后台安静地运行。

方法 A：单次产出生成器（最快的设置方式）

要获得即时的口头确认而不产生持续的进度絮叨，一个有效的方法是使用一个仅在完成时产出一次的异步生成器：

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

纯文本

已复制

由于这是一个异步生成器，ADK 会发出一个即时的待处理响应。智能体会立即确认该请求（“我现在正在为您汇总那份报告”），在计算期间保持沉默，并在准备好时或用户说完话后呈现数据。

方法 B：使用 `response_scheduling` 的 `FunctionTool`

如果你更愿意保留标准的协程（async def ... -> dict），请将该函数包装在 FunctionTool 中并配置 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.)

纯文本

已复制

理解权衡： 当将 response_scheduling 与返回一次结果的协程一起使用时，模型在整个函数完成之前不会收到任何负载。因此，除非用户继续说话，模型通常不会发出初始的口头确认。当用户立即继续提出其他问题，而你希望结果在下一个自然停顿时融入对话时，这种模式是理想之选。

使用 `response_scheduling` 控制结果传递

response_scheduling 属性控制模型对传入工具响应做出反应的时机。WHEN_IDLE 从不打断正在进行的轮次：如果模型仍在说话，或用户仍在讲话，结果会等到该轮次结束后，模型在下一个自然停顿时才做出响应。ADK 支持三种调度行为：

table1_response_scheduling_16x9

在 gemini-3.8-live 的基准会话中：

在 WHEN_IDLE 下，传入的销售报告礼貌地等待，直到模型完成回答另一个 20 秒的问题，然后无缝地传递数据。
在 INTERRUPT 下，传入的结果在句子中途打断了当前讲话；随后模型完成了自己的回答，并追加了数据。
在 SILENT 下，结果被添加到对话历史记录中，可供后续轮次使用，而不会触发未经请求的公告。

结合流式传输与调度

执行形态（流式传输 vs. 单次返回）和传递时机（response_scheduling）是相互独立且可以组合的。在流式工具上设置 response_scheduling 会为每个部分更新打上该行为的印记。

此外，在 ADK 2.8 及更新版本中，流式工具可以通过 yield 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,
)

纯文本

已复制

架构总结：选择正确的工具模式

table2_architecture_summary_4x3

经验法则：如果等待有故事可讲，就流式传输。如果等待只是等待，就在后台运行，让结果在空闲时落地。

结论

非阻塞工具不会减少后端执行时间：一个 13 秒的查询仍然需要 13 秒。改变的是你的 agent 在对话中的存在感。

一个需要十秒才能加载的网页应用会被认为是慢的；一个沉默十秒的对话式语音 agent 会被认为是故障的。通过将 ADK 的流式工具和灵活的响应调度与 gemini-3.8-live 的实时能力相结合，你可以构建出在复杂操作过程中始终保持专注、自然和存在的语音助手。

亲自试试： 交互式演示在 gemini-3.8-live 上通过每种工具模式和 response_scheduling 模式运行相同的 13 秒查询，让你听出其中的差异。

要求与后续步骤

框架： google-adk >= 2.10.0
模型： gemini-3.8-live on Gemini Enterprise
API 环境： 流式工具和实时调度需要双向实时会话（run_live）。