用ADK消除语音代理调用工具时的冷场

使用 ADK 和 Gemini 3.8 Live，构建一个不会陷入沉默的语音 Agent。

本文介绍如何在 Agent Development Kit（ADK）Gemini Live API Toolkit 中使用流式工具、非阻塞函数调用和响应调度。

在文字聊天中，数据库查询或 RAG 流水线耗时 13 秒，用户看到的是一个加载动画。但在实时语音对话中，工具执行阻塞 13 秒，就意味着整整 13 秒的冷场。用户往往会以为通话断了，直接挂断。

你不可能总把后端查询压缩到一秒以内，因此必须从架构上解决：让慢速工具在后台运行，同时保持 Agent 的对话能力。借助开源 ADK 和 gemini-3.8-live，你可以直接用 Python 实现非阻塞语音工具，无需自行编写客户端 WebSocket 事件循环来管理异步工具状态。在我们基于 gemini-3.8-live 的销售报表工作流基准测试中，将阻塞工具改为流式工具后，Agent 开口说出第一句话的时间从 14.4 秒缩短到了 1.2 秒。

本文将介绍如何使用流式工具、非阻塞 FunctionTool 定义和 response_scheduling，构建响应及时、交流自然的语音 Agent。

观看这段 1 分钟的 ADK Live AI Concierge 演示视频，听听响应调度的实际效果：

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

非阻塞工具能带来什么

在 ADK 中实现非阻塞工具调用，可以立即改善多个方面的用户体验：

• 消除冷场：Agent 不会在对话中途卡住，而是可以立即回应请求，比如“我这就查一下这些数据……”，让对话继续下去。
• 播报进度：对于包含多个阶段的工作流，例如连接、查询和核对，Agent 可以播报中间节点，让用户听到的等待过程体现出工作正在推进。
• 多任务与并行对话：耗时查询在后台执行时，用户可以继续追问，也可以聊其他话题，并及时获得回答，整个会话不会因此停滞。
• 精细控制结果交付：由你决定最终结果何时、以何种方式进入对话：等到自然停顿时再说、遇到关键提醒立即打断，或者只静默更新对话上下文。

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

在常规 Agent 开发中，工具通常被定义为一个 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 秒，Agent 就会完全沉默 13 秒。即使在 prompt 中要求“告诉用户你正在处理”，也无法解决这个问题，因为模型的轮次已经被阻塞，正在等待函数响应 payload。

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

模式一：流式工具，让等待过程有进展可讲

当一个操作包含多个有意义的阶段，例如连接数据库、聚合记录和格式化结果时，流式工具可以让 Agent 随着任务推进，实时讲述进度。

在 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 会立即向模型发送一个异步状态事件。Agent 随即开口：“我这就帮你查一下 EMEA 的数据……”
• 逐步更新：每次 yield 都会向上游发送一条部分函数响应。模型收到更新后，会用自己的话转述，比如“正在连接数据仓库……”“正在聚合产品线数据……”。不过，语音播报可能比工具的实际进度慢几秒。
• 无缝处理并行话题：如果用户在查询期间提出无关问题，Agent 可以直接回答，不必取消或中断后台任务。

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

• 避免重复发起工具调用：流式工具会跨越多个轮次推送更新，因此在用户反复催问时，模型偶尔可能再次调用同一个工具。在 docstring 中明确保留“Call this ONCE”指令，并考虑在生产实现中加入内存锁存标记或幂等键，防止重复执行。

• 通过 stop_streaming 响应用户取消：ADK 为长时间运行的流式工具提供了内置终止机制。在 Agent 的工具列表中加入一个名为 stop_streaming(function_name: str) 的函数后，ADK 会拦截该调用，取消生成器底层的 asyncio.Task。这会在你的函数内部抛出 asyncio.CancelledError，让你可以妥善关闭数据库连接或文件句柄。

• 用户插话不会取消工具：用户打断 Agent 说话时，Live API 会停止模型输出，ADK 则会发出一个带有 interrupted=True 的事件，让客户端停止音频播放。正在运行的流式工具或后台工具不会被取消，结果仍会继续交付。要停止工具，模型必须调用 stop_streaming。

模式二：后台执行，让单纯的等待安静进行

持续播报进度并不总是合适。如果用户请求一份报表后，马上转向另一个话题，Agent 还不断插话汇报状态，就会显得打扰。

如果等待过程中没有什么值得讲述的进展，就让任务在后台安静运行。

方案 A：只 yield 一次的生成器，最快上手

如果希望 Agent 立即口头回应，又不想持续播报进度，一个有效办法是使用异步生成器，只在任务完成时 yield 一次：

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 会立即发出一条表示任务待完成的响应。Agent 会迅速确认请求，比如“我正在帮你整理这份报表”，计算期间不再播报，等数据就绪或用户说完话后，再给出结果。

方案 B：为 FunctionTool 配置 response_scheduling

如果你希望保留标准协程（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 时，模型在整个函数执行完毕之前，都收不到任何 payload。因此，除非用户继续说话，否则模型通常不会先口头确认请求。如果用户会立即继续提出其他问题，而你希望结果在下一次自然停顿时融入对话，这种模式就很合适。

使用 response_scheduling 控制结果交付

response_scheduling 属性控制模型何时对收到的工具响应作出反应。WHEN_IDLE 绝不会打断正在进行的轮次：如果模型仍在说话，或用户还没说完，结果就会等待当前轮次结束，模型再在下一个自然停顿处回应。ADK 支持三种调度行为。

在使用 gemini-3.8-live 的基准测试会话中：

• 使用 WHEN_IDLE 时，销售报表结果到达后，会耐心等待模型回答完另一个耗时 20 秒的问题，再自然地给出报表数据。
• 使用 INTERRUPT 时，到达的结果会在一句话说到一半时打断当前语音；随后模型会完成原来的回答，并补充报表数据。
• 使用 SILENT 时，结果会加入对话历史，供后续轮次使用，不会触发用户未要求的主动播报。

组合使用流式输出与响应调度

执行形式，即流式输出还是单次返回，与交付时机 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,
)

架构总结：如何选择合适的工具模式

经验法则是：等待过程中有进展可讲，就使用流式工具；如果只是单纯等待，就放到后台运行，等对话空闲时再交付结果。

非阻塞工具不会缩短后端执行时间：一个 13 秒的查询，仍然需要 13 秒。改变的是 Agent 在对话中的参与感。

一个网页应用加载十秒，会让人觉得慢；一个语音 Agent 沉默十秒，会让人觉得它坏了。将 ADK 的流式工具、灵活的响应调度与 gemini-3.8-live 的实时能力结合起来，就能构建出这样的语音助手：即使执行复杂操作，也始终保持专注、自然，并持续参与对话。

亲自试试：交互式演示会在 gemini-3.8-live 上，使用每一种工具模式和 response_scheduling 模式执行同一个 13 秒查询，让你直接听出差别。

框架：google-adk >= 2.10.0

模型：Gemini Enterprise 上的 gemini-3.8-live

API 环境：流式工具和实时调度需要使用双向实时会话（run_live）。

ADK Gemini Live API Toolkit