関数ツール¶
関数ツールとは？¶
標準のツールが特定の要件を完全に満たさない場合、開発者はカスタムの関数ツールを作成できます。これにより、独自のデータベースへの接続やユニークなアルゴリズムの実装など、カスタマイズされた機能が可能になります。
例えば、関数ツール「myfinancetool」は、特定の財務指標を計算する関数かもしれません。ADKは長時間実行される関数もサポートしているため、その計算に時間がかかる場合でも、エージェントは他のタスクの作業を続けることができます。
ADKは、関数ツールを作成するためのいくつかの方法を提供しており、それぞれが異なるレベルの複雑さと制御に適しています：
- 関数ツール
- 長時間実行関数ツール
- ツールとしてエージェント
1. 関数ツール¶
関数をツールに変換することは、カスタムロジックをエージェントに統合する簡単な方法です。実際、エージェントのツールリストに関数を割り当てると、フレームワークは自動的にそれを関数ツールとしてラップします。このアプローチは、柔軟性と迅速な統合を提供します。
パラメータ¶
標準のJSONシリアライズ可能な型（例：文字列、整数、リスト、辞書）を使用して関数パラメータを定義します。言語モデル（LLM）は現在、パラメータのデフォルト値を解釈することをサポートしていないため、デフォルト値を設定しないことが重要です。
戻り値の型¶
関数ツールの推奨される戻り値の型は、Pythonでは辞書（dictionary）、JavaではMapです。これにより、応答をキーと値のペアで構造化し、LLMにコンテキストと明確さを提供できます。関数が辞書以外の型を返す場合、フレームワークは自動的にそれを「result」という単一のキーを持つ辞書にラップします。
戻り値はできるだけ説明的にするように努めてください。例えば、数値のエラーコードを返す代わりに、「error_message」キーに人間が読める説明を含む辞書を返します。コードの一部ではなく、LLMが結果を理解する必要があることを忘れないでください。ベストプラクティスとして、戻り値の辞書に「status」キーを含め、操作全体の成果（例：「success」、「error」、「pending」）を示すことで、LLMに操作の状態に関する明確なシグナルを提供します。
Docstring / ソースコードのコメント¶
関数のdocstring（またはその上のコメント）は、ツールの説明として機能し、LLMに送信されます。したがって、LLMがツールを効果的に使用する方法を理解するためには、よく書かれた包括的なdocstringが不可欠です。関数の目的、そのパラメータの意味、および期待される戻り値を明確に説明してください。
例
このツールは、指定された株式のティッカー/シンボルの株価を取得するPython関数です。
注：このツールを使用する前に、pip install yfinanceライブラリをインストールする必要があります。
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
session_service = InMemorySessionService()
session = 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)
# Agent Interaction
def call_agent(query):
content = types.Content(role='user', parts=[types.Part(text=query)])
events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("Agent Response: ", final_response)
call_agent("stock price of GOOG")
このツールからの戻り値は辞書にラップされます。
このツールは、株価のモック値を取得します。
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?");
}
}
このツールからの戻り値はMap
ベストプラクティス¶
関数の定義にはかなりの柔軟性がありますが、シンプルさがLLMの使いやすさを向上させることを忘れないでください。以下のガイドラインを考慮してください：
- パラメータは少ない方が良い： 複雑さを減らすためにパラメータの数を最小限に抑えます。
- 単純なデータ型： 可能な限り、カスタムクラスよりも
strやintのようなプリミティブなデータ型を優先します。 - 意味のある名前： 関数の名前とパラメータ名は、LLMがツールをどのように解釈し、利用するかに大きく影響します。関数の目的とその入力の意味を明確に反映する名前を選択してください。
do_stuff()やbeAgent()のような一般的な名前は避けてください。
2. 長時間実行関数ツール¶
エージェントの実行をブロックすることなく、かなりの処理時間を必要とするタスクのために設計されています。このツールはFunctionToolのサブクラスです。
LongRunningFunctionToolを使用する場合、関数は長時間実行される操作を開始し、オプションで初期結果（例：長時間実行操作ID）を返すことができます。長時間実行関数ツールが呼び出されると、エージェントランナーはエージェントの実行を一時停止し、エージェントクライアントに続行するか、長時間実行操作が終了するまで待つかを決定させます。エージェントクライアントは長時間実行操作の進捗を照会し、中間または最終的な応答を送り返すことができます。その後、エージェントは他のタスクを続行できます。例としては、エージェントがタスクを進める前に人間の承認が必要なヒューマンインザループのシナリオがあります。
仕組み¶
Pythonでは、関数をLongRunningFunctionToolでラップします。Javaでは、メソッド名をLongRunningFunctionTool.create()に渡します。
-
開始： LLMがツールを呼び出すと、関数が長時間実行操作を開始します。
-
初期更新： 関数はオプションで初期結果（例：長時間実行操作ID）を返す必要があります。ADKフレームワークはその結果を受け取り、
FunctionResponseにパッケージ化してLLMに送り返します。これにより、LLMはユーザーに（例：ステータス、完了率、メッセージ）を通知できます。そして、エージェントの実行は終了/一時停止されます。 -
続行または待機： 各エージェントの実行が完了した後。エージェントクライアントは長時間実行操作の進捗を照会し、中間応答でエージェントの実行を続行するか（進捗を更新するため）、最終的な応答が取得されるまで待つかを決定できます。エージェントクライアントは、次回の実行のために中間または最終的な応答をエージェントに送り返す必要があります。
-
フレームワークの処理： ADKフレームワークは実行を管理します。エージェントクライアントから送信された中間または最終的な
FunctionResponseをLLMに送信し、ユーザーフレンドリーなメッセージを生成します。
ツールの作成¶
ツール関数を定義し、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 {
// 長時間実行関数を定義します。
// 払い戻しの承認を求めます。
public static Map<String, Object> askForApproval(String purpose, double amount) {
// チケット作成と通知送信をシミュレート
System.out.println(
"目的：" + purpose + "、金額：" + amount + "のチケット作成をシミュレート中");
// 承認者にチケットのリンク付きで通知を送信
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 {
// メソッドをLongRunningFunctionTool.createに渡す
LongRunningFunctionTool approveTool =
LongRunningFunctionTool.create(ExampleLongRunningFunction.class, "askForApproval");
// ツールをエージェントに含める
LlmAgent approverAgent =
LlmAgent.builder()
// ...
.tools(approveTool)
.build();
}
}
中間/最終結果の更新¶
エージェントクライアントは、長時間実行関数呼び出しを含むイベントを受け取り、チケットのステータスを確認します。その後、エージェントクライアントは進捗を更新するために中間または最終的な応答を送り返すことができます。フレームワークはこの値（Noneであっても）をFunctionResponseのコンテンツにパッケージ化してLLMに送り返します。
Java ADKにのみ適用
Function ToolsでToolContextを渡す場合、以下のいずれかが真であることを確認してください：
-
スキーマが関数シグネチャのToolContextパラメータと一緒に渡されている。例：
または -
以下の
-parametersフラグがmvnコンパイラプラグインに設定されている。
# Agent Interaction
async def call_agent(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)])
events = runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
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:
long_running_function_response = get_function_response(event, long_running_function_call.id)
if long_running_function_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}')
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
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)
# Agent Interaction
async def call_agent(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)])
events = runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
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:
long_running_function_response = get_function_response(event, long_running_function_call.id)
if long_running_function_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}')
# reimbursement that doesn't require approval
asyncio.run(call_agent("Please reimburse 50$ for meals"))
# await call_agent("Please reimburse 50$ for meals") # For Notebooks, uncomment this line and comment the above line
# reimbursement that requires approval
asyncio.run(call_agent("Please reimburse 200$ for meals"))
# await call_agent("Please reimburse 200$ for meals") # For Notebooks, uncomment this line and comment the above line
この例の重要な側面¶
-
LongRunningFunctionTool: 提供されたメソッド/関数をラップします。フレームワークは、yieldされた更新と最終的な戻り値をシーケンシャルなFunctionResponseとして送信する処理を担当します。 -
エージェントの指示: LLMにツールを使用させ、ユーザーへの更新のために受信するFunctionResponseストリーム（進捗 vs 完了）を理解させます。
-
最終的な戻り値: 関数は最終的な結果の辞書を返し、それが完了を示すための最後のFunctionResponseで送信されます。
3. ツールとしてのエージェント¶
この強力な機能により、システム内の他のエージェントをツールとして呼び出すことで、その能力を活用できます。ツールとしてのエージェントは、別のエージェントを呼び出して特定のタスクを実行させることができ、効果的に責任を委任します。これは概念的に、別のエージェントを呼び出し、そのエージェントの応答を関数の戻り値として使用するPython関数を作成するのと似ています。
サブエージェントとの主な違い¶
ツールとしてのエージェントとサブエージェントを区別することが重要です。
-
ツールとしてのエージェント: エージェントAがエージェントBをツールとして呼び出すと（ツールとしてのエージェントを使用）、エージェントBの回答はエージェントAに返され、エージェントAはその回答を要約してユーザーへの応答を生成します。エージェントAは制御を保持し、将来のユーザー入力を処理し続けます。
-
サブエージェント: エージェントAがエージェントBをサブエージェントとして呼び出すと、ユーザーに答える責任は完全にエージェントBに移譲されます。エージェントAは事実上ループの外に出ます。以降のすべてのユーザー入力はエージェントBによって回答されます。
使用法¶
エージェントをツールとして使用するには、エージェントをAgentToolクラスでラップします。
カスタマイズ¶
AgentToolクラスは、その振る舞いをカスタマイズするための以下の属性を提供します：
- skip_summarization: bool: Trueに設定すると、フレームワークはツールエージェントの応答のLLMベースの要約をバイパスします。これは、ツールの応答が既によくフォーマットされており、さらなる処理が不要な場合に役立ちます。
例
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)]
)
# Session and Runner
session_service = InMemorySessionService()
session = 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)
# Agent Interaction
def call_agent(query):
content = types.Content(role='user', parts=[types.Part(text=query)])
events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
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."""
call_agent(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はその要約を受け取り、ユーザーへの最終的な応答を（例：「テキストの要約はこちらです：...」）を組み立てることができます。