实时智能体的工具¶

Supported in ADKPython v0.1.0Java v0.2.0

工具在实时智能体中的工作方式与 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="You are a helpful assistant that can search the web.",
)

你通过事件流观察工具活动，无需手动驱动：

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:
    """Generate and store the quarterly report. Returns when the export finishes."""
    await run_export(region)  # a long, plain return-once operation
    return {"status": "done", "region": region}

report_tool = FunctionTool(export_report)
report_tool.response_scheduling = types.FunctionResponseScheduling.WHEN_IDLE

工具运行期间智能体保持空闲，可以回答用户提出的其他问题，并在结果就绪时将其整合进来。一个可运行的示例在 live_non_blocking_tool_agent 示例中提供。

需要 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]:
    """Run the quarterly sales report. Call this once; it streams its own updates."""
    yield "Connecting to the warehouse..."
    await asyncio.sleep(4)
    yield "Aggregating by product line..."
    await asyncio.sleep(4)
    yield f"Done. {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，因此智能体在其他时候保持安静。

PythonJava

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 = "How many people are in this image? Reply with a number only."


async def monitor_video_stream(
    input_stream: LiveRequestQueue,
) -> AsyncGenerator[str, None]:
  """Report how many people are visible, whenever that number changes."""
  client = Client()
  last_count = None

  while True:
    # Drain the queue and keep only the newest frame; older ones are stale.
    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 intercepts this by name; the body stays empty.
def stop_streaming(function_name: str):
  """Stop a running streaming tool.

  Args:
    function_name: The name of the streaming function to stop.
  """


root_agent = Agent(
    # Streaming tools run under run_live(), so the root agent needs a Live
    # model. gemini-flash-latest above is only for the one-shot call in the tool.
    model=os.getenv("DEMO_AGENT_MODEL", "gemini-live-2.5-flash-native-audio"),
    name="video_monitoring_agent",
    instruction=(
        "You monitor the user's video stream. Call monitor_video_stream once when"
        " asked, then report each update it sends. Never call it again to poll."
    ),
    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 =
      "How many people are in this image? Reply with a number only.";

  // `inputStream` is a reserved parameter name; ADK passes the video stream in.
  @Schema(description = "Report how many people are visible, whenever that number changes.")
  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)  // newest frame every 0.5s
        .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 only when the count changes
        .map(count -> Map.of("result", count));
  }

  // ADK intercepts this by name; the body stays empty.
  @Schema(description = "Stop a running streaming tool.")
  public static void stopStreaming(
      @Schema(name = "functionName", description = "The streaming function to stop.")
      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(
                "You monitor the user's video stream. Call monitorVideoStream once when"
                    + " asked, then report each update it sends. Never call it again to poll.")
            .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` 可立即终止整个流式会话