流式工具¶
Supported in ADKPython v0.5.0Java v0.2.0Experimental
流式工具允许工具（函数）将中间结果流式传回智能体，智能体可以对这些中间结果做出响应。例如，我们可以使用流式工具监控股票价格变化并让智能体对此做出反应。
Info
此功能仅在 ADK Gemini 实时 API 中支持。
要定义流式工具，你需要遵守以下规则：
- 异步函数： 该工具必须是一个
asyncPython 函数。 - AsyncGenerator 返回类型： 函数的返回类型必须标注为
AsyncGenerator。AsyncGenerator的第一个类型参数是yield数据的类型（例如，文本消息使用str，结构化数据使用自定义对象）。第二个类型参数通常是None，表示生成器不会通过send()接收值。
我们支持两种类型的流式工具： - 简单类型。这是一种仅接受非视频/非音频流（即你通过 adk web 或 adk runner 提供的流）作为输入的流式工具。 - 视频流式工具。此类型仅在视频流中有效，视频流（即你通过 adk web 或 adk runner 提供的流）会被传入该函数。
现在让我们定义一个能够监控股价变化和视频流变化的智能体。
import asyncio
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
async def monitor_stock_price(stock_symbol: str) -> AsyncGenerator[str, None]:
"""此函数将以持续、流式和异步的方式监控给定 stock_symbol 的价格。"""
print(f"开始监控股票 {stock_symbol} 的价格！")
# 模拟股价变化。
await asyncio.sleep(4)
price_alert1 = f"the price for {stock_symbol} is 300"
yield price_alert1
print(price_alert1)
await asyncio.sleep(4)
price_alert1 = f"the price for {stock_symbol} is 400"
yield price_alert1
print(price_alert1)
await asyncio.sleep(20)
price_alert1 = f"the price for {stock_symbol} is 900"
yield price_alert1
print(price_alert1)
await asyncio.sleep(20)
price_alert1 = f"the price for {stock_symbol} is 500"
yield price_alert1
print(price_alert1)
# 对于视频流，`input_stream: LiveRequestQueue` 是必需的保留关键字参数，供 ADK 传入视频流。
async def monitor_video_stream(
input_stream: LiveRequestQueue,
) -> AsyncGenerator[str, None]:
"""监控视频流中有多少人。"""
print("开始监控视频流！")
client = Client(enterprise=False)
prompt_text = (
"Count the number of people in this image. Just respond with a numeric"
" number."
)
last_count = None
while True:
last_valid_req = None
print("Start monitoring loop")
# 使用此循环拉取最新图像并丢弃旧图像
while input_stream._queue.qsize() != 0:
live_req = await input_stream.get()
if live_req.blob is not None and live_req.blob.mime_type == "image/jpeg":
last_valid_req = live_req
# 如果找到有效图像，则进行处理
if last_valid_req is not None:
print("Processing the most recent frame from the queue")
# 使用 blob 的数据和 mime 类型创建图像部分
image_part = genai_types.Part.from_bytes(
data=last_valid_req.blob.data, mime_type=last_valid_req.blob.mime_type
)
contents = genai_types.Content(
role="user",
parts=[image_part, genai_types.Part.from_text(prompt_text)],
)
# 调用模型根据提供的图像和提示生成内容
response = client.models.generate_content(
model="gemini-flash-latest",
contents=contents,
config=genai_types.GenerateContentConfig(
system_instruction=(
"You are a helpful video analysis assistant. You can count"
" the number of people in this image or video. Just respond"
" with a numeric number."
)
),
)
if not last_count:
last_count = response.candidates[0].content.parts[0].text
elif last_count != response.candidates[0].content.parts[0].text:
last_count = response.candidates[0].content.parts[0].text
yield response
print("response:", response)
# 等待后再检查新图像
await asyncio.sleep(0.5)
# 请使用此精确函数来帮助 ADK 在需要时停止你的流式工具。
# 例如，如果我们想停止 `monitor_stock_price`，则智能体将
# 调用此函数：stop_streaming(function_name=monitor_stock_price)。
def stop_streaming(function_name: str):
"""停止流式传输
Args:
function_name: 要停止的流式函数的名称。
"""
pass
root_agent = Agent(
model="gemini-flash-latest",
name="video_streaming_agent",
instruction="""
你是一个监控智能体。你可以使用提供的工具/函数进行视频监控和股价监控。
当用户想要监控视频流时，
你可以使用 monitor_video_stream 函数来实现。当 monitor_video_stream
返回警报时，你应该告知用户。
当用户想要监控股票价格时，你可以使用 monitor_stock_price。
不要问太多问题。不要过于健谈。
""",
tools=[
monitor_video_stream,
monitor_stock_price,
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.GenerateContentResponse;
import com.google.genai.types.Part;
import io.reactivex.rxjava3.core.Flowable;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;
public class StreamingTools {
@Schema(description = "此函数将以持续、流式和异步的方式监控给定 stock_symbol 的价格。")
public static Flowable<Map<String, Object>> monitorStockPrice(@Schema(name = "stockSymbol") String stockSymbol) {
System.out.println("开始监控股票 " + stockSymbol + " 的价格！");
return Flowable.concat(
Flowable.<Map<String, Object>>just(Collections.singletonMap("result", "the price for " + stockSymbol + " is 300")).delay(4, TimeUnit.SECONDS),
Flowable.<Map<String, Object>>just(Collections.singletonMap("result", "the price for " + stockSymbol + " is 400")).delay(4, TimeUnit.SECONDS),
Flowable.<Map<String, Object>>just(Collections.singletonMap("result", "the price for " + stockSymbol + " is 900")).delay(20, TimeUnit.SECONDS),
Flowable.<Map<String, Object>>just(Collections.singletonMap("result", "the price for " + stockSymbol + " is 500")).delay(20, TimeUnit.SECONDS)
);
}
// 对于视频流，`inputStream` 是必需的保留参数，供 ADK 传入视频流。
@Schema(description = "监控视频流中有多少人。")
public static Flowable<Map<String, Object>> monitorVideoStream(@Schema(name = "inputStream") LiveRequestQueue inputStream) {
System.out.println("开始监控视频流！");
Client client = Client.builder().build();
String promptText = "Count the number of people in this image. Just respond with a numeric number.";
// 使用 RxJava 处理流
return inputStream.get()
.filter(req -> req.blob().isPresent() && "image/jpeg".equals(req.blob().get().mimeType()))
.sample(500, TimeUnit.MILLISECONDS) // 每 0.5 秒处理一帧
.map(req -> {
System.out.println("正在处理队列中的最新帧");
Part imagePart = Part.builder().inlineData(req.blob().get()).build();
Content contents = Content.builder()
.role("user")
.parts(Arrays.asList(imagePart, Part.fromText(promptText)))
.build();
GenerateContentResponse response = client.models().generateContent(
"gemini-flash-latest",
contents,
GenerateContentConfig.builder()
.systemInstruction(Content.builder().parts(Arrays.asList(
Part.fromText("You are a helpful video analysis assistant. You can count the number of people in this image or video. Just respond with a numeric number.")
)).build())
.build()
);
return (Map<String, Object>) Collections.<String, Object>singletonMap("result", response.text());
})
.distinctUntilChanged()
.doOnNext(res -> System.out.println("response: " + res));
}
// 请使用此精确函数来帮助 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-flash-latest")
.name("video_streaming_agent")
.instruction(
"你是一个监控智能体。你可以使用提供的工具/函数进行视频监控和股价监控。\n" +
"当用户想要监控视频流时，\n" +
"你可以使用 monitorVideoStream 函数来实现。当 monitorVideoStream\n" +
"返回警报时，你应该告知用户。\n" +
"当用户想要监控股票价格时，你可以使用 monitorStockPrice。\n" +
"不要问太多问题。不要过于健谈。"
)
.tools(Arrays.asList(
FunctionTool.create(StreamingTools.class, "monitorVideoStream"),
FunctionTool.create(StreamingTools.class, "monitorStockPrice"),
FunctionTool.create(StreamingTools.class, "stopStreaming")
))
.build();
}
}
以下是一些用于测试的示例查询： - 帮我监控 $XYZ 股票的价格。 - 帮我监控视频流中有多少人。