Function tools¶
當內建的 Agent Development Kit (ADK) 工具無法滿足你的需求時，你可以自行建立自訂的 function tools（函式工具）。建構 function tools 讓你能夠打造專屬的功能，例如連接專有資料庫或實作獨特的演算法。
舉例來說，一個 function tool myfinancetool，可能是一個計算特定財務指標的函式。Agent Development Kit (ADK) 也支援長時間執行的函式，因此如果該計算需要較長時間，agent 仍可繼續處理其他任務。
Agent Development Kit (ADK) 提供多種建立 function tools 的方式，適用於不同複雜度與控制需求：
Function Tools¶
將 Python 函式轉換為工具，是將自訂邏輯整合進 agent 的簡單方法。當你將函式指派給 agent 的 tools 清單時，框架會自動將其包裝為 FunctionTool。
運作方式¶
Agent Development Kit (ADK) 框架會自動檢查你的 Python 函式簽章，包括名稱、docstring、參數、型別提示以及預設值，並據此產生 schema。這個 schema 讓大型語言模型 (LLM) 能夠理解該工具的用途、使用時機，以及所需的參數。
定義函式簽章¶
良好定義的函式簽章對於大型語言模型 (LLM) 正確使用你的工具至關重要。
參數¶
你可以定義帶有必要參數、選用參數與可變參數的函式。以下說明各種參數的處理方式：
必要參數¶
如果參數有型別提示但沒有預設值，則視為必要參數。大型語言模型 (LLM) 在呼叫該工具時，必須為此參數提供值。
範例：必要參數
def get_weather(city: str, unit: str):
"""
Retrieves the weather for a city in the specified unit.
Args:
city (str): The city name.
unit (str): The temperature unit, either 'Celsius' or 'Fahrenheit'.
"""
# ... function logic ...
return {"status": "success", "report": f"Weather for {city} is sunny."}
在此範例中，city 和 unit 都是必填參數。如果大型語言模型 (LLM) 嘗試在缺少其中一個的情況下呼叫 get_weather，Agent Development Kit (ADK) 會回傳錯誤給 LLM，提示其修正呼叫內容。
具有預設值的選用參數¶
如果你為參數提供預設值，該參數就被視為選用。這是 Python 定義選用參數的標準方式。Agent Development Kit (ADK) 會正確解析這些參數，並且不會將它們列在傳送給 LLM 的工具 schema 的 required 欄位中。
範例：具有預設值的選用參數
def search_flights(destination: str, departure_date: str, flexible_days: int = 0):
"""
Searches for flights.
Args:
destination (str): The destination city.
departure_date (str): The desired departure date.
flexible_days (int, optional): Number of flexible days for the search. Defaults to 0.
"""
# ... function logic ...
if flexible_days > 0:
return {"status": "success", "report": f"Found flexible flights to {destination}."}
return {"status": "success", "report": f"Found flights to {destination} on {departure_date}."}
這裡的 flexible_days 是選填的。大型語言模型 (LLM) 可以選擇提供它，但不是必須的。
使用 typing.Optional 的選填參數¶
你也可以使用 typing.Optional[SomeType] 或 Python 3.10+ 的 | None 語法來標記參數為選填。這表示該參數可以是 None。當與預設值 None 結合時，會如同標準的選填參數運作。
範例：typing.Optional
from typing import Optional
def create_user_profile(username: str, bio: Optional[str] = None):
"""
Creates a new user profile.
Args:
username (str): The user's unique username.
bio (str, optional): A short biography for the user. Defaults to None.
"""
# ... function logic ...
if bio:
return {"status": "success", "message": f"Profile for {username} created with a bio."}
return {"status": "success", "message": f"Profile for {username} created."}
可變參數（*args 與 **kwargs）¶
雖然你可以在函式簽名中加入 *args（可變位置參數）與 **kwargs（可變關鍵字參數）以用於其他目的，但在 Agent Development Kit (ADK) 框架為大型語言模型 (LLM) 產生工具 schema 時，這些參數會被忽略。LLM 不會知道這些參數的存在，也無法傳遞參數給它們。因此，建議你對所有期望由 LLM 傳入的資料，都明確定義參數。
回傳型態¶
Function Tool（函式工具）建議的回傳型態是在 Python 中使用dictionary（字典），或在 Java 中使用Map。這樣可以用鍵值對的方式結構化回應，為 LLM 提供更多上下文與清晰度。如果你的函式回傳的型態不是字典，框架會自動將其包裝成一個僅有單一鍵 "result" 的字典。
請盡量讓你的回傳值具有描述性。例如，與其回傳數字型的錯誤代碼，不如回傳一個包含 "error_message" 鍵且帶有人類可讀說明的字典。請記住，理解結果的是 LLM，而不是一段程式碼。最佳實踐是，在回傳的字典中加入 "status" 鍵，用來標示整體執行結果（例如 "success"、"error"、"pending"），讓 LLM 能夠明確判斷操作狀態。
Docstring¶
你的函式 docstring 會作為工具的描述，並傳送給 LLM。因此，一份撰寫良好且詳盡的 docstring 對於 LLM 有效理解如何使用該工具至關重要。請清楚說明函式的用途、各參數的意義，以及預期的回傳值。
工具間資料傳遞¶
當 agent 依序呼叫多個工具時，你可能需要將資料從一個工具傳遞到另一個工具。建議的做法是使用 session state 中的 temp: 前綴。
一個工具可以將資料寫入 temp: 變數，後續的工具則可以讀取這個變數。這些資料僅在目前的呼叫過程中有效，之後會被清除。
Shared Invocation Context
在單一 agent 回合（agent turn）內的所有工具呼叫（tool calls）都會共用相同的 InvocationContext。這也表示它們會共用相同的暫存（temp:）狀態，這就是資料能夠在它們之間傳遞的方式。
範例¶
範例
This tool is a python function which obtains the Stock price of a given Stock ticker/ symbol.
Note: You need to pip install yfinance library before using this tool.
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import yfinance as yf
APP_NAME = "stock_app"
USER_ID = "1234"
SESSION_ID = "session1234"
def get_stock_price(symbol: str):
"""
Retrieves the current stock price for a given symbol.
Args:
symbol (str): The stock symbol (e.g., "AAPL", "GOOG").
Returns:
float: The current stock price, or None if an error occurs.
"""
try:
stock = yf.Ticker(symbol)
historical_data = stock.history(period="1d")
if not historical_data.empty:
current_price = historical_data['Close'].iloc[-1]
return current_price
else:
return None
except Exception as e:
print(f"Error retrieving stock price for {symbol}: {e}")
return None
stock_price_agent = Agent(
model='gemini-2.0-flash',
name='stock_agent',
instruction= 'You are an agent who retrieves stock prices. If a ticker symbol is provided, fetch the current price. If only a company name is given, first perform a Google search to find the correct ticker symbol before retrieving the stock price. If the provided ticker symbol is invalid or data cannot be retrieved, inform the user that the stock price could not be found.',
description='This agent specializes in retrieving real-time stock prices. Given a stock ticker symbol (e.g., AAPL, GOOG, MSFT) or the stock name, use the tools and reliable data sources to provide the most up-to-date price.',
tools=[get_stock_price], # You can add Python functions directly to the tools list; they will be automatically wrapped as FunctionTools.
)
# Session and Runner
async def setup_session_and_runner():
session_service = InMemorySessionService()
session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=stock_price_agent, app_name=APP_NAME, session_service=session_service)
return session, runner
# Agent Interaction
async def call_agent_async(query):
content = types.Content(role='user', parts=[types.Part(text=query)])
session, runner = await setup_session_and_runner()
events = runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
async for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("Agent Response: ", final_response)
# Note: In Colab, you can directly use 'await' at the top level.
# If running this code as a standalone Python script, you'll need to use asyncio.run() or manage the event loop.
await call_agent_async("stock price of GOOG")
The return value from this tool will be wrapped into a dictionary.
This tool retrieves the mocked value of a stock price.
import com.google.adk.agents.LlmAgent;
import com.google.adk.events.Event;
import com.google.adk.runner.InMemoryRunner;
import com.google.adk.sessions.Session;
import com.google.adk.tools.Annotations.Schema;
import com.google.adk.tools.FunctionTool;
import com.google.genai.types.Content;
import com.google.genai.types.Part;
import io.reactivex.rxjava3.core.Flowable;
import java.util.HashMap;
import java.util.Map;
public class StockPriceAgent {
private static final String APP_NAME = "stock_agent";
private static final String USER_ID = "user1234";
// Mock data for various stocks functionality
// NOTE: This is a MOCK implementation. In a real Java application,
// you would use a financial data API or library.
private static final Map<String, Double> mockStockPrices = new HashMap<>();
static {
mockStockPrices.put("GOOG", 1.0);
mockStockPrices.put("AAPL", 1.0);
mockStockPrices.put("MSFT", 1.0);
}
@Schema(description = "Retrieves the current stock price for a given symbol.")
public static Map<String, Object> getStockPrice(
@Schema(description = "The stock symbol (e.g., \"AAPL\", \"GOOG\")",
name = "symbol")
String symbol) {
try {
if (mockStockPrices.containsKey(symbol.toUpperCase())) {
double currentPrice = mockStockPrices.get(symbol.toUpperCase());
System.out.println("Tool: Found price for " + symbol + ": " + currentPrice);
return Map.of("symbol", symbol, "price", currentPrice);
} else {
return Map.of("symbol", symbol, "error", "No data found for symbol");
}
} catch (Exception e) {
return Map.of("symbol", symbol, "error", e.getMessage());
}
}
public static void callAgent(String prompt) {
// Create the FunctionTool from the Java method
FunctionTool getStockPriceTool = FunctionTool.create(StockPriceAgent.class, "getStockPrice");
LlmAgent stockPriceAgent =
LlmAgent.builder()
.model("gemini-2.0-flash")
.name("stock_agent")
.instruction(
"You are an agent who retrieves stock prices. If a ticker symbol is provided, fetch the current price. If only a company name is given, first perform a Google search to find the correct ticker symbol before retrieving the stock price. If the provided ticker symbol is invalid or data cannot be retrieved, inform the user that the stock price could not be found.")
.description(
"This agent specializes in retrieving real-time stock prices. Given a stock ticker symbol (e.g., AAPL, GOOG, MSFT) or the stock name, use the tools and reliable data sources to provide the most up-to-date price.")
.tools(getStockPriceTool) // Add the Java FunctionTool
.build();
// Create an InMemoryRunner
InMemoryRunner runner = new InMemoryRunner(stockPriceAgent, APP_NAME);
// InMemoryRunner automatically creates a session service. Create a session using the service
Session session = runner.sessionService().createSession(APP_NAME, USER_ID).blockingGet();
Content userMessage = Content.fromParts(Part.fromText(prompt));
// Run the agent
Flowable<Event> eventStream = runner.runAsync(USER_ID, session.id(), userMessage);
// Stream event response
eventStream.blockingForEach(
event -> {
if (event.finalResponse()) {
System.out.println(event.stringifyContent());
}
});
}
public static void main(String[] args) {
callAgent("stock price of GOOG");
callAgent("What's the price of MSFT?");
callAgent("Can you find the stock price for an unknown company XYZ?");
}
}
The return value from this tool will be wrapped into a Map
最佳實踐¶
雖然你在定義工具函式時擁有相當大的彈性，但請記住，簡單性有助於提升大型語言模型 (LLM) 的可用性。建議遵循以下指引：
- 參數越少越好： 盡量減少參數數量，以降低複雜度。
- 簡單資料型別： 優先使用像是
str和int這類的原始資料型別，盡量避免自訂類別。 - 具意義的命名： 函式名稱與參數名稱對大型語言模型 (LLM) 如何解讀與運用該工具有重大影響。請選擇能清楚反映函式用途及輸入意義的名稱，避免使用像
do_stuff()或beAgent()這類泛用名稱。 - 為平行執行而設計： 當多個工具同時運行時，透過設計為非同步操作來提升工具呼叫 (tool calls) 的效能。關於如何啟用工具的平行執行，請參閱 Increase tool performance with parallel execution。
長時間運行的 Function Tools¶
此工具設計用於協助你啟動並管理需要在 agent 工作流程之外執行、且需大量處理時間的任務，同時不會阻塞 agent 的執行。這個工具是 FunctionTool 的子類別。
當你使用 LongRunningFunctionTool 時，函式可以啟動長時間運行的操作，並可選擇性地回傳初始結果，例如長時間運行操作的 ID。一旦長時間運行的 function tool 被呼叫，agent runner 會暫停 agent 的運行，並讓 agent client 決定是否繼續或等待長時間運行的操作完成。agent client 可以查詢長時間運行操作的進度，並回傳中間或最終回應。agent 隨後可以繼續執行其他任務。舉例來說，在 human-in-the-loop（人類參與）情境下，agent 需要獲得人類批准後才能繼續執行任務。
Warning: Execution handling
長時間執行函式工具（Long Running Function Tools）旨在協助你在 agent 工作流程中啟動並管理長時間執行的任務，但不會實際執行這些長時間任務。對於需要花費大量時間才能完成的任務，建議你實作一個獨立的伺服器來處理該任務。
Tip: Parallel execution
根據你所建構的工具類型，設計為非同步（asynchronous）操作通常會比建立長時間運行的工具來得更好。欲了解更多資訊，請參閱利用平行執行提升工具效能。
運作方式說明¶
在 Python 中，你可以使用 LongRunningFunctionTool 將函式包裝起來。在 Java 中，則是將 Method 名稱傳遞給 LongRunningFunctionTool.create()。
-
啟動階段：當大型語言模型 (LLM) 呼叫該工具時，你的函式會啟動長時間運行的操作。
-
初始回報：你的函式可以選擇性地回傳初步結果（例如長時間運行操作的 id）。Agent Development Kit (ADK) 框架會將這個結果包裝在
FunctionResponse中回傳給 LLM。這讓 LLM 可以通知使用者（例如狀態、完成百分比、訊息等）。隨後，agent 執行會結束或暫停。 -
繼續或等待：每次 agent 執行結束後，agent client 可以查詢長時間運行操作的進度，並決定是要以中間回應（用於更新進度）繼續 agent 執行，還是等待直到取得最終回應。agent client 應將中間或最終回應回傳給 agent，以進行下一次執行。
-
框架處理：Agent Development Kit (ADK) 框架負責管理整個執行流程。它會將 agent client 傳回的中間或最終
FunctionResponse傳送給 LLM，以產生對使用者友善的訊息。
建立工具¶
定義你的工具函式（tool function），並使用 LongRunningFunctionTool 類別將其包裝：
# 1. Define the long running function
def ask_for_approval(
purpose: str, amount: float
) -> dict[str, Any]:
"""Ask for approval for the reimbursement."""
# create a ticket for the approval
# Send a notification to the approver with the link of the ticket
return {'status': 'pending', 'approver': 'Sean Zhou', 'purpose' : purpose, 'amount': amount, 'ticket-id': 'approval-ticket-1'}
def reimburse(purpose: str, amount: float) -> str:
"""Reimburse the amount of money to the employee."""
# send the reimbrusement request to payment vendor
return {'status': 'ok'}
# 2. Wrap the function with LongRunningFunctionTool
long_running_tool = LongRunningFunctionTool(func=ask_for_approval)
import com.google.adk.agents.LlmAgent;
import com.google.adk.tools.LongRunningFunctionTool;
import java.util.HashMap;
import java.util.Map;
public class ExampleLongRunningFunction {
// Define your Long Running function.
// Ask for approval for the reimbursement.
public static Map<String, Object> askForApproval(String purpose, double amount) {
// Simulate creating a ticket and sending a notification
System.out.println(
"Simulating ticket creation for purpose: " + purpose + ", amount: " + amount);
// Send a notification to the approver with the link of the ticket
Map<String, Object> result = new HashMap<>();
result.put("status", "pending");
result.put("approver", "Sean Zhou");
result.put("purpose", purpose);
result.put("amount", amount);
result.put("ticket-id", "approval-ticket-1");
return result;
}
public static void main(String[] args) throws NoSuchMethodException {
// Pass the method to LongRunningFunctionTool.create
LongRunningFunctionTool approveTool =
LongRunningFunctionTool.create(ExampleLongRunningFunction.class, "askForApproval");
// Include the tool in the agent
LlmAgent approverAgent =
LlmAgent.builder()
// ...
.tools(approveTool)
.build();
}
}
中間 / 最終結果更新¶
Agent client 會收到包含長時間執行工具呼叫 (function calls) 的事件，並檢查該 ticket 的狀態。接著，Agent client 可以將中間或最終回應傳回，以更新進度。框架會將這個值（即使是 None）包裝在FunctionResponse的內容中，並回傳給大型語言模型 (LLM)。
僅適用於 Java Agent Development Kit (ADK)
When passing ToolContext with Function Tools, ensure that one of the following is true:
-
The Schema is passed with the ToolContext parameter in the function signature, like:
OR -
The following
-parametersflag is set to the mvn compiler plugin
# Agent Interaction
async def call_agent_async(query):
def get_long_running_function_call(event: Event) -> types.FunctionCall:
# Get the long running function call from the event
if not event.long_running_tool_ids or not event.content or not event.content.parts:
return
for part in event.content.parts:
if (
part
and part.function_call
and event.long_running_tool_ids
and part.function_call.id in event.long_running_tool_ids
):
return part.function_call
def get_function_response(event: Event, function_call_id: str) -> types.FunctionResponse:
# Get the function response for the fuction call with specified id.
if not event.content or not event.content.parts:
return
for part in event.content.parts:
if (
part
and part.function_response
and part.function_response.id == function_call_id
):
return part.function_response
content = types.Content(role='user', parts=[types.Part(text=query)])
session, runner = await setup_session_and_runner()
print("\nRunning agent...")
events_async = runner.run_async(
session_id=session.id, user_id=USER_ID, new_message=content
)
long_running_function_call, long_running_function_response, ticket_id = None, None, None
async for event in events_async:
# Use helper to check for the specific auth request event
if not long_running_function_call:
long_running_function_call = get_long_running_function_call(event)
else:
_potential_response = get_function_response(event, long_running_function_call.id)
if _potential_response: # Only update if we get a non-None response
long_running_function_response = _potential_response
ticket_id = long_running_function_response.response['ticket-id']
if event.content and event.content.parts:
if text := ''.join(part.text or '' for part in event.content.parts):
print(f'[{event.author}]: {text}')
if long_running_function_response:
# query the status of the correpsonding ticket via tciket_id
# send back an intermediate / final response
updated_response = long_running_function_response.model_copy(deep=True)
updated_response.response = {'status': 'approved'}
async for event in runner.run_async(
session_id=session.id, user_id=USER_ID, new_message=types.Content(parts=[types.Part(function_response = updated_response)], role='user')
):
if event.content and event.content.parts:
if text := ''.join(part.text or '' for part in event.content.parts):
print(f'[{event.author}]: {text}')
（譯註：此區塊標示為 Java 相關內容，請依照後續內容進行翻譯。如原文僅有此標題，則無需額外翻譯內容。）
```java
import com.google.adk.agents.LlmAgent;
import com.google.adk.events.Event;
import com.google.adk.runner.InMemoryRunner;
import com.google.adk.runner.Runner;
import com.google.adk.sessions.Session;
import com.google.adk.tools.Annotations.Schema;
import com.google.adk.tools.LongRunningFunctionTool;
import com.google.adk.tools.ToolContext;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.genai.types.Content;
import com.google.genai.types.FunctionCall;
import com.google.genai.types.FunctionResponse;
import com.google.genai.types.Part;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
public class LongRunningFunctionExample {
private static String USER_ID = "user123";
@Schema(
name = "create_ticket_long_running",
description = """
Creates a new support ticket with a specified urgency level.
Examples of urgency are 'high', 'medium', or 'low'.
The ticket creation is a long-running process, and its ID will be provided when ready.
""")
public static void createTicketAsync(
@Schema(
name = "urgency",
description =
"The urgency level for the new ticket, such as 'high', 'medium', or 'low'.")
String urgency,
@Schema(name = "toolContext") // Ensures ADK injection
ToolContext toolContext) {
System.out.printf(
"TOOL_EXEC: 'create_ticket_long_running' called with urgency: %s (Call ID: %s)%n",
urgency, toolContext.functionCallId().orElse("N/A"));
}
public static void main(String[] args) {
LlmAgent agent =
LlmAgent.builder()
.name("ticket_agent")
.description("Agent for creating tickets via a long-running task.")
.model("gemini-2.0-flash")
.tools(
ImmutableList.of(
LongRunningFunctionTool.create(
LongRunningFunctionExample.class, "createTicketAsync")))
.build();
Runner runner = new InMemoryRunner(agent);
Session session =
runner.sessionService().createSession(agent.name(), USER_ID, null, null).blockingGet();
// --- Turn 1: User requests ticket ---
System.out.println("\n--- Turn 1: User Request ---");
Content initialUserMessage =
Content.fromParts(Part.fromText("Create a high urgency ticket for me."));
AtomicReference<String> funcCallIdRef = new AtomicReference<>();
runner
.runAsync(USER_ID, session.id(), initialUserMessage)
.blockingForEach(
event -> {
printEventSummary(event, "T1");
if (funcCallIdRef.get() == null) { // Capture the first relevant function call ID
event.content().flatMap(Content::parts).orElse(ImmutableList.of()).stream()
.map(Part::functionCall)
.flatMap(Optional::stream)
.filter(fc -> "create_ticket_long_running".equals(fc.name().orElse("")))
.findFirst()
.flatMap(FunctionCall::id)
.ifPresent(funcCallIdRef::set);
}
});
if (funcCallIdRef.get() == null) {
System.out.println("ERROR: Tool 'create_ticket_long_running' not called in Turn 1.");
return;
}
System.out.println("ACTION: Captured FunctionCall ID: " + funcCallIdRef.get());
// --- Turn 2: App provides initial ticket_id (simulating async tool completion) ---
System.out.println("\n--- Turn 2: App provides ticket_id ---");
String ticketId = "TICKET-" + UUID.randomUUID().toString().substring(0, 8).toUpperCase();
FunctionResponse ticketCreatedFuncResponse =
FunctionResponse.builder()
.name("create_ticket_long_running")
.id(funcCallIdRef.get())
.response(ImmutableMap.of("ticket_id", ticketId))
.build();
Content appResponseWithTicketId =
Content.builder()
.parts(
ImmutableList.of(
Part.builder().functionResponse(ticketCreatedFuncResponse).build()))
.role("user")
.build();
runner
.runAsync(USER_ID, session.id(), appResponseWithTicketId)
.blockingForEach(event -> printEventSummary(event, "T2"));
System.out.println("ACTION: Sent ticket_id " + ticketId + " to agent.");
// --- Turn 3: App provides ticket status update ---
System.out.println("\n--- Turn 3: App provides ticket status ---");
FunctionResponse ticketStatusFuncResponse =
FunctionResponse.builder()
.name("create_ticket_long_running")
.id(funcCallIdRef.get())
.response(ImmutableMap.of("status", "approved", "ticket_id", ticketId))
.build();
Content appResponseWithStatus =
Content.builder()
.parts(
ImmutableList.of(Part.builder().functionResponse(ticketStatusFuncResponse).build()))
.role("user")
.build();
runner
.runAsync(USER_ID, session.id(), appResponseWithStatus)
.blockingForEach(event -> printEventSummary(event, "T3_FINAL"));
System.out.println("Long running function completed successfully.");
}
private static void printEventSummary(Event event, String turnLabel) {
event
.content()
.ifPresent(
content -> {
String text =
content.parts().orElse(ImmutableList.of()).stream()
.map(part -> part.text().orElse(""))
.filter(s -> !s.isEmpty())
.collect(Collectors.joining(" "));
if (!text.isEmpty()) {
System.out.printf("[%s][%s_TEXT]: %s%n", turnLabel, event.author(), text);
}
content.parts().orElse(ImmutableList.of()).stream()
.map(Part::functionCall)
.flatMap(Optional::stream)
.findFirst() // Assuming one function call per relevant event for simplicity
.ifPresent(
fc ->
System.out.printf(
"[%s][%s_CALL]: %s(%s) ID: %s%n",
turnLabel,
event.author(),
fc.name().orElse("N/A"),
fc.args().orElse(ImmutableMap.of()),
fc.id().orElse("N/A")));
});
}
}
```
Python 完整範例：檔案處理模擬
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
from typing import Any
from google.adk.agents import Agent
from google.adk.events import Event
from google.adk.runners import Runner
from google.adk.tools import LongRunningFunctionTool
from google.adk.sessions import InMemorySessionService
from google.genai import types
# 1. Define the long running function
def ask_for_approval(
purpose: str, amount: float
) -> dict[str, Any]:
"""Ask for approval for the reimbursement."""
# create a ticket for the approval
# Send a notification to the approver with the link of the ticket
return {'status': 'pending', 'approver': 'Sean Zhou', 'purpose' : purpose, 'amount': amount, 'ticket-id': 'approval-ticket-1'}
def reimburse(purpose: str, amount: float) -> str:
"""Reimburse the amount of money to the employee."""
# send the reimbrusement request to payment vendor
return {'status': 'ok'}
# 2. Wrap the function with LongRunningFunctionTool
long_running_tool = LongRunningFunctionTool(func=ask_for_approval)
# 3. Use the tool in an Agent
file_processor_agent = Agent(
# Use a model compatible with function calling
model="gemini-2.0-flash",
name='reimbursement_agent',
instruction="""
You are an agent whose job is to handle the reimbursement process for
the employees. If the amount is less than $100, you will automatically
approve the reimbursement.
If the amount is greater than $100, you will
ask for approval from the manager. If the manager approves, you will
call reimburse() to reimburse the amount to the employee. If the manager
rejects, you will inform the employee of the rejection.
""",
tools=[reimburse, long_running_tool]
)
APP_NAME = "human_in_the_loop"
USER_ID = "1234"
SESSION_ID = "session1234"
# Session and Runner
async def setup_session_and_runner():
session_service = InMemorySessionService()
session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=file_processor_agent, app_name=APP_NAME, session_service=session_service)
return session, runner
# Agent Interaction
async def call_agent_async(query):
def get_long_running_function_call(event: Event) -> types.FunctionCall:
# Get the long running function call from the event
if not event.long_running_tool_ids or not event.content or not event.content.parts:
return
for part in event.content.parts:
if (
part
and part.function_call
and event.long_running_tool_ids
and part.function_call.id in event.long_running_tool_ids
):
return part.function_call
def get_function_response(event: Event, function_call_id: str) -> types.FunctionResponse:
# Get the function response for the fuction call with specified id.
if not event.content or not event.content.parts:
return
for part in event.content.parts:
if (
part
and part.function_response
and part.function_response.id == function_call_id
):
return part.function_response
content = types.Content(role='user', parts=[types.Part(text=query)])
session, runner = await setup_session_and_runner()
print("\nRunning agent...")
events_async = runner.run_async(
session_id=session.id, user_id=USER_ID, new_message=content
)
long_running_function_call, long_running_function_response, ticket_id = None, None, None
async for event in events_async:
# Use helper to check for the specific auth request event
if not long_running_function_call:
long_running_function_call = get_long_running_function_call(event)
else:
_potential_response = get_function_response(event, long_running_function_call.id)
if _potential_response: # Only update if we get a non-None response
long_running_function_response = _potential_response
ticket_id = long_running_function_response.response['ticket-id']
if event.content and event.content.parts:
if text := ''.join(part.text or '' for part in event.content.parts):
print(f'[{event.author}]: {text}')
if long_running_function_response:
# query the status of the correpsonding ticket via tciket_id
# send back an intermediate / final response
updated_response = long_running_function_response.model_copy(deep=True)
updated_response.response = {'status': 'approved'}
async for event in runner.run_async(
session_id=session.id, user_id=USER_ID, new_message=types.Content(parts=[types.Part(function_response = updated_response)], role='user')
):
if event.content and event.content.parts:
if text := ''.join(part.text or '' for part in event.content.parts):
print(f'[{event.author}]: {text}')
# Note: In Colab, you can directly use 'await' at the top level.
# If running this code as a standalone Python script, you'll need to use asyncio.run() or manage the event loop.
# reimbursement that doesn't require approval
# asyncio.run(call_agent_async("Please reimburse 50$ for meals"))
await call_agent_async("Please reimburse 50$ for meals") # For Notebooks, uncomment this line and comment the above line
# reimbursement that requires approval
# asyncio.run(call_agent_async("Please reimburse 200$ for meals"))
await call_agent_async("Please reimburse 200$ for meals") # For Notebooks, uncomment this line and comment the above line
此範例的重點¶
-
LongRunningFunctionTool：包裝所提供的方法／函式；框架會自動處理將讓渡（yielded）的更新與最終回傳值，依序作為 FunctionResponse 傳送。 -
Agent 指令：指示大型語言模型 (LLM) 使用該工具，並理解收到的 FunctionResponse 串流（進度 vs. 完成），以便向使用者提供更新。
-
最終回傳：該函式會回傳最終的結果字典，並於最後的 FunctionResponse 中傳送，以表示已完成。
Agent-as-a-Tool¶
這項強大的功能讓你可以在系統內，將其他 agent 當作工具來調用。Agent-as-a-Tool 讓你能夠呼叫另一個 agent 來執行特定任務，實現責任委派。這在概念上類似於建立一個 Python 函式，該函式會呼叫另一個 agent，並將該 agent 的回應作為函式的回傳值。
與子 agent 的主要差異¶
需特別區分 Agent-as-a-Tool 與 Sub-Agent（子 agent）。
-
Agent-as-a-Tool： 當 Agent A 以工具方式呼叫 Agent B（使用 Agent-as-a-Tool）時，Agent B 的答案會傳回給 Agent A，接著由 Agent A 彙整答案並產生回應給使用者。Agent A 保持主控權，並持續處理後續的使用者輸入。
-
Sub-agent： 當 Agent A 以子 agent 方式呼叫 Agent B 時，回答使用者的責任會完全轉移給 Agent B。Agent A 實際上就不再參與流程。所有後續的使用者輸入都會由 Agent B 回應。
使用方式¶
若要將 agent 當作工具使用，請使用 AgentTool 類別包裝該 agent。
```java
AgentTool.create(agent)
```
自訂化¶
AgentTool 類別提供以下屬性，以便自訂其行為：
- skip_summarization: bool： 若設為 True，框架將略過基於大型語言模型 (LLM) 的工具 agent 回應摘要處理。當工具的回應已經格式良好且不需要進一步處理時，這個選項會很有用。
範例
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.agent_tool import AgentTool
from google.genai import types
APP_NAME="summary_agent"
USER_ID="user1234"
SESSION_ID="1234"
summary_agent = Agent(
model="gemini-2.0-flash",
name="summary_agent",
instruction="""You are an expert summarizer. Please read the following text and provide a concise summary.""",
description="Agent to summarize text",
)
root_agent = Agent(
model='gemini-2.0-flash',
name='root_agent',
instruction="""You are a helpful assistant. When the user provides a text, use the 'summarize' tool to generate a summary. Always forward the user's message exactly as received to the 'summarize' tool, without modifying or summarizing it yourself. Present the response from the tool to the user.""",
tools=[AgentTool(agent=summary_agent, skip_summarization=True)]
)
# Session and Runner
async def setup_session_and_runner():
session_service = InMemorySessionService()
session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)
return session, runner
# Agent Interaction
async def call_agent_async(query):
content = types.Content(role='user', parts=[types.Part(text=query)])
session, runner = await setup_session_and_runner()
events = runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
async for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("Agent Response: ", final_response)
long_text = """Quantum computing represents a fundamentally different approach to computation,
leveraging the bizarre principles of quantum mechanics to process information. Unlike classical computers
that rely on bits representing either 0 or 1, quantum computers use qubits which can exist in a state of superposition - effectively
being 0, 1, or a combination of both simultaneously. Furthermore, qubits can become entangled,
meaning their fates are intertwined regardless of distance, allowing for complex correlations. This parallelism and
interconnectedness grant quantum computers the potential to solve specific types of incredibly complex problems - such
as drug discovery, materials science, complex system optimization, and breaking certain types of cryptography - far
faster than even the most powerful classical supercomputers could ever achieve, although the technology is still largely in its developmental stages."""
# Note: In Colab, you can directly use 'await' at the top level.
# If running this code as a standalone Python script, you'll need to use asyncio.run() or manage the event loop.
await call_agent_async(long_text)
import com.google.adk.agents.LlmAgent;
import com.google.adk.events.Event;
import com.google.adk.runner.InMemoryRunner;
import com.google.adk.sessions.Session;
import com.google.adk.tools.AgentTool;
import com.google.genai.types.Content;
import com.google.genai.types.Part;
import io.reactivex.rxjava3.core.Flowable;
public class AgentToolCustomization {
private static final String APP_NAME = "summary_agent";
private static final String USER_ID = "user1234";
public static void initAgentAndRun(String prompt) {
LlmAgent summaryAgent =
LlmAgent.builder()
.model("gemini-2.0-flash")
.name("summaryAgent")
.instruction(
"You are an expert summarizer. Please read the following text and provide a concise summary.")
.description("Agent to summarize text")
.build();
// Define root_agent
LlmAgent rootAgent =
LlmAgent.builder()
.model("gemini-2.0-flash")
.name("rootAgent")
.instruction(
"You are a helpful assistant. When the user provides a text, always use the 'summaryAgent' tool to generate a summary. Always forward the user's message exactly as received to the 'summaryAgent' tool, without modifying or summarizing it yourself. Present the response from the tool to the user.")
.description("Assistant agent")
.tools(AgentTool.create(summaryAgent, true)) // Set skipSummarization to true
.build();
// Create an InMemoryRunner
InMemoryRunner runner = new InMemoryRunner(rootAgent, APP_NAME);
// InMemoryRunner automatically creates a session service. Create a session using the service
Session session = runner.sessionService().createSession(APP_NAME, USER_ID).blockingGet();
Content userMessage = Content.fromParts(Part.fromText(prompt));
// Run the agent
Flowable<Event> eventStream = runner.runAsync(USER_ID, session.id(), userMessage);
// Stream event response
eventStream.blockingForEach(
event -> {
if (event.finalResponse()) {
System.out.println(event.stringifyContent());
}
});
}
public static void main(String[] args) {
String longText =
"""
Quantum computing represents a fundamentally different approach to computation,
leveraging the bizarre principles of quantum mechanics to process information. Unlike classical computers
that rely on bits representing either 0 or 1, quantum computers use qubits which can exist in a state of superposition - effectively
being 0, 1, or a combination of both simultaneously. Furthermore, qubits can become entangled,
meaning their fates are intertwined regardless of distance, allowing for complex correlations. This parallelism and
interconnectedness grant quantum computers the potential to solve specific types of incredibly complex problems - such
as drug discovery, materials science, complex system optimization, and breaking certain types of cryptography - far
faster than even the most powerful classical supercomputers could ever achieve, although the technology is still largely in its developmental stages.""";
initAgentAndRun(longText);
}
}
運作方式¶
- 當
main_agent收到長文本時，其指令會告訴它針對長文本使用summarize工具。 - 框架會將
summarize辨識為包裝summary_agent的AgentTool。 - 在背景流程中，
main_agent會以長文本作為輸入，呼叫summary_agent。 summary_agent會依據其指令處理文本並產生摘要。- 來自
summary_agent的回應會再傳回給main_agent。 main_agent便可利用該摘要，組成最終回應給使用者（例如：「這是該文本的摘要：...」）