实时智能体的工具¶
工具在实时智能体中的工作方式与 ADK 其他地方基本一致：你将函数传递给智能体，模型会调用它们。在实时连接下编写工具的方式不会改变，因此工具定义、工具上下文、回调和认证都遵循自定义工具的规则。
实时连接在此基础上新增了两项能力。ADK 会在 run_live() 循环中为你执行工具调用，因此你无需编写原始 Live API 所需的函数调用管道代码。实时智能体还可以使用流式工具：持续运行并将中间结果推送回智能体的函数，这样智能体就能对股价变动或视频画面中出现的人做出反应，而无需用户再次提问。
自动工具执行¶
在智能体上定义工具，ADK 会在 run_live() 循环中为你调用它们：它会检测模型的函数调用、运行工具（并行执行，带前后回调）、格式化响应，并将调用和响应作为事件产出。你只需编写函数，无需关心管道代码。
import os
from google.adk.agents import Agent
from google.adk.tools import google_search
agent = Agent(
name="google_search_agent",
model=os.getenv("DEMO_AGENT_MODEL", "gemini-live-2.5-flash-native-audio"),
tools=[google_search],
instruction="你是一个可以搜索网络的有用助手。",
)
你通过事件流观察工具活动，无需手动驱动：
async for event in runner.run_live(...):
if event.get_function_calls():
print(f"Model calling: {event.get_function_calls()[0].name}")
if event.get_function_responses():
print(f"Tool result: {event.get_function_responses()[0].response}")
保持智能体的响应性¶
在聊天窗口中，慢速工具尚可接受，用户可以看着加载动画等待。但在实时语音对话中就不行了：如果智能体调用了一个耗时十秒的 API 然后沉默，用户会认为通话断了。你需要一个在运行时不阻塞对话的工具。ADK 提供了两种方式来实现这一点，外加针对快速场景的普通阻塞方式：
| 你的场景 | 使用方式 | 实现方法 |
|---|---|---|
| 工具在一秒内返回 | 阻塞（默认方式） | 普通的 return 工具 |
| 长时间等待且无需播报 | 非阻塞工具 | 在工具上设置 response_scheduling |
| 长时间等待且值得播报 | 流式工具 | 从异步生成器中 yield 进度 |
非阻塞工具¶
有些等待不值得播报：一个耗时的分析查询、一次批量导出、一个媒体生成任务。用户未请求的进度更新会无益地打断对话。保持你普通的单次 return 工具不变，设置 response_scheduling 将其移到后台：
from google.adk.tools import FunctionTool
from google.genai import types
async def export_report(region: str) -> dict:
"""生成并存储季度报告。导出完成后返回。"""
await run_export(region) # 一个耗时较长、仅返回一次的普通操作
return {"status": "done", "region": region}
report_tool = FunctionTool(export_report)
report_tool.response_scheduling = types.FunctionResponseScheduling.WHEN_IDLE
在工具运行期间智能体仍然自由，可以回应用户提出的其他话题，并在结果就绪时将其纳入对话。可运行的示例见 non_blocking_tool_agent 示例。
需要 ADK Python 2.4 或更高版本
response_scheduling 功能在 adk-python 2.4 中加入，且支持情况因模型而异。参见
实时模型。
response_scheduling 还控制已完成结果何时送达用户：
| 值 | 行为 | 适用场景 |
|---|---|---|
WHEN_IDLE |
等待自然停顿 | 报告和查询，通常选择此项 |
INTERRUPT |
立即送达 | 告警、失败、"转账失败" |
SILENT |
进入上下文，仅在相关时播报 | 模型稍后可能使用的背景信息 |
流式工具¶
流式工具持续运行并将中间结果推送回智能体，这样智能体就能播报进度或对变化的输入（股价、有人进入视频画面）做出反应，而无需用户再次提问。将工具改为流式只需一行改动：用 yield 替代 return 的 async 函数。ADK 会自动将任何异步生成器工具视为非阻塞的。
import asyncio
from typing import AsyncGenerator
async def query_sales_database(region: str) -> AsyncGenerator[str, None]:
"""运行季度销售报告。只需调用一次；它会自行流式输出更新。"""
yield "正在连接数据仓库……"
await asyncio.sleep(4)
yield "正在按产品线汇总……"
await asyncio.sleep(4)
yield f"完成。{summarise(region)}"
像其他工具一样将其传入 tools=[...]。模型将每个 yield 作为实时更新接收，因此用户不会听到沉默，而是听到"让我查一下... 还在汇总中... 查到了：EMEA 收入 481 万美元，增长 12.4%。"这适用于 RAG 管道、多阶段聚合和构建测试运行等任何值得播报进度的场景。
添加 ADK 保留的 stop_streaming 工具（一个 ADK 按名称拦截的空函数），这样用户可以取消操作："算了，取消吧。"
视频流式工具¶
添加一个 input_stream: LiveRequestQueue 参数，ADK 会将用户的实时输入馈送到该工具的专用队列中，以便它拉取视频帧并做出反应。
任何流式工具的要求：
- 它必须是一个
async函数，类型标注为返回AsyncGenerator[T, None]，其中T是你yield的类型。 - 对于视频，添加
input_stream: LiveRequestQueue；ADK 会自动填充。
下面的模式清空队列到最新帧，丢弃过时的帧，仅在答案变化时才 yield，因此智能体在其他时候保持安静。
import asyncio
import os
from typing import AsyncGenerator
from google.adk.agents import LiveRequestQueue
from google.adk.agents.llm_agent import Agent
from google.adk.tools.function_tool import FunctionTool
from google.genai import Client
from google.genai import types as genai_types
PROMPT = "这张图片里有多少人？只回复一个数字。"
async def monitor_video_stream(
input_stream: LiveRequestQueue,
) -> AsyncGenerator[str, None]:
"""每当可见人数发生变化时报告该人数。"""
client = Client()
last_count = None
while True:
# 清空队列并只保留最新一帧；较早的帧已经过时。
latest = None
while input_stream._queue.qsize() != 0:
req = await input_stream.get()
if req.blob and req.blob.mime_type == "image/jpeg":
latest = req
if latest is not None:
response = client.models.generate_content(
model="gemini-flash-latest",
contents=genai_types.Content(
role="user",
parts=[
genai_types.Part.from_bytes(
data=latest.blob.data, mime_type=latest.blob.mime_type
),
genai_types.Part.from_text(text=PROMPT),
],
),
)
count = response.candidates[0].content.parts[0].text.strip()
if count != last_count:
last_count = count
yield count
await asyncio.sleep(0.5)
# ADK 按名称拦截此函数；函数体保持为空。
def stop_streaming(function_name: str):
"""停止正在运行的流式工具。
Args:
function_name: 要停止的流式函数的名称。
"""
root_agent = Agent(
# 流式工具在 run_live() 下运行，因此根智能体需要一个 Live
# 模型。上面的 gemini-flash-latest 仅用于工具内部的一次性调用。
model=os.getenv("DEMO_AGENT_MODEL", "gemini-live-2.5-flash-native-audio"),
name="video_monitoring_agent",
instruction=(
"你负责监控用户的视频流。当被要求时调用 monitor_video_stream 一次，"
"然后报告它发送的每一次更新。绝不要为了轮询而再次调用它。"
),
tools=[monitor_video_stream, FunctionTool(stop_streaming)],
)
import com.google.adk.agents.LiveRequestQueue;
import com.google.adk.agents.LlmAgent;
import com.google.adk.tools.Annotations.Schema;
import com.google.adk.tools.FunctionTool;
import com.google.genai.Client;
import com.google.genai.types.Content;
import com.google.genai.types.GenerateContentConfig;
import com.google.genai.types.Part;
import io.reactivex.rxjava3.core.Flowable;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.TimeUnit;
public class StreamingTools {
private static final String PROMPT =
"这张图片里有多少人？只回复一个数字。";
// `inputStream` 是保留参数名；ADK 会把视频流传入其中。
@Schema(description = "每当可见人数发生变化时报告该人数。")
public static Flowable<Map<String, Object>> monitorVideoStream(
@Schema(name = "inputStream") LiveRequestQueue inputStream) {
Client client = Client.builder().build();
return inputStream
.get()
.filter(req -> req.blob().isPresent()
&& "image/jpeg".equals(req.blob().get().mimeType()))
.sample(500, TimeUnit.MILLISECONDS) // 每 0.5 秒取最新一帧
.map(req -> client.models().generateContent(
"gemini-flash-latest",
Content.builder()
.role("user")
.parts(Arrays.asList(
Part.builder().inlineData(req.blob().get()).build(),
Part.fromText(PROMPT)))
.build(),
GenerateContentConfig.builder().build())
.text())
.distinctUntilChanged() // 仅在人数变化时才 yield
.map(count -> Map.of("result", count));
}
// ADK 按名称拦截此函数；函数体保持为空。
@Schema(description = "停止正在运行的流式工具。")
public static void stopStreaming(
@Schema(name = "functionName", description = "要停止的流式函数。")
String functionName) {}
public static void main(String[] args) {
LlmAgent rootAgent =
LlmAgent.builder()
.model("gemini-live-2.5-flash-native-audio")
.name("video_monitoring_agent")
.instruction(
"你负责监控用户的视频流。当被要求时调用 monitorVideoStream 一次，"
+ "然后报告它发送的每一次更新。绝不要为了轮询而再次调用它。")
.tools(Arrays.asList(
FunctionTool.create(StreamingTools.class, "monitorVideoStream"),
FunctionTool.create(StreamingTools.class, "stopStreaming")))
.build();
}
}
试试让智能体监控视频流中有多少人，然后走进走出画面。
工具执行上下文¶
工具或回调接收一个 InvocationContext，用于获取状态、历史和制品。它的工作方式与任何 ADK 智能体中相同——参见智能体上下文——但有一个在实时场景中很重要的区别：一个 InvocationContext 贯穿整个 run_live() 循环，在你调用 run_live() 时创建，在每个智能体和每轮对话中持续存在，直到会话结束。在请求/响应模式的智能体中，一次调用就是一轮对话；在实时会话中，一次调用就是整个对话。
实时工具中最常用的两个字段：
| 字段 | 提供的信息 |
|---|---|
context.run_config |
会话的配置——响应模态、转录、限制 |
context.end_invocation |
设为 True 可立即终止整个流式会话 |
