工具¶
什么是工具？¶
在 ADK 的上下文中，工具代表提供给 AI 智能体的特定能力，使其能够执行行动并与其核心文本生成和推理能力之外的世界进行交互。通常，有能力的智能体与基础语言模型的区别在于它们有效地使用工具。
从技术上讲，工具通常是一个模块化的代码组件——如 Python/Java 函数、类方法，甚至是另一个专用智能体——用于执行一个独立且预定义的任务。这些任务通常涉及与外部系统或数据的交互。
主要特点¶
面向行动： 工具执行特定行动，例如：
- 查询数据库
- 发出 API 请求（例如，获取天气数据、预订系统）
- 搜索网络
- 执行代码片段
- 从文档中检索信息（RAG）
- 与其他软件或服务交互
扩展智能体能力： 它们使智能体能够访问实时信息，影响外部系统，并克服其训练数据固有的知识限制。
执行预定义逻辑： 至关重要的是，工具执行特定的、由开发者定义的逻辑。它们不像智能体的核心大型语言模型（LLM）那样具有自己独立的推理能力。LLM 会推理使用哪个工具、何时使用以及使用什么输入，但工具本身只是执行其指定的功能。
智能体如何使用工具¶
智能体通过通常涉及函数调用的机制动态地利用工具。该过程通常遵循以下步骤：
- 推理： 智能体的 LLM 分析其系统指令、对话历史和用户请求。
- 选择： 基于分析，LLM 根据智能体可用的工具和描述每个工具的文档字符串决定执行哪个工具（如果有的话）。
- 调用： LLM 生成所选工具所需的参数（输入）并触发其执行。
- 观察： 智能体接收工具返回的输出（结果）。
- 完成： 智能体将工具的输出纳入其持续的推理过程中，以制定下一个响应、决定后续步骤或确定是否已达成目标。
可以将工具视为智能体的智能核心（LLM）可以根据需要访问和利用的专用工具包，以完成复杂任务。
ADK 中的工具类型¶
ADK 通过支持几种类型的工具提供灵活性：
- 函数工具： 由你创建的工具，根据你的特定应用需求量身定制。
- 内置工具： 框架提供的可直接使用的常见任务工具。 示例：Google 搜索、代码执行、检索增强生成（RAG）。
- 第三方工具： 无缝集成来自流行外部库的工具。 示例：LangChain 工具、CrewAI 工具。
有关每种工具类型的详细信息和示例，请导航到上面链接的相应文档页面。
在智能体指令中引用工具¶
在智能体的指令中，你可以通过使用工具的函数名称直接引用它。如果工具的函数名称和文档字符串足够描述性，你的指令可以主要集中在大型语言模型（LLM）应该何时使用该工具。这促进了清晰度并帮助模型理解每个工具的预期用途。
清楚地指导智能体如何处理工具可能产生的不同返回值是至关重要的。例如，如果工具返回错误消息，你的指令应指定智能体是否应重试操作、放弃任务或向用户请求额外信息。
此外，ADK 支持工具的顺序使用，其中一个工具的输出可以作为另一个工具的输入。在实现此类工作流程时，在智能体的指令中描述工具使用的预期顺序很重要，以引导模型完成必要的步骤。
示例¶
以下示例展示了智能体如何通过在其指令中引用工具的函数名称来使用工具。它还演示了如何指导智能体处理工具的不同返回值，如成功或错误消息，以及如何编排多个工具的顺序使用以完成任务。
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
APP_NAME="weather_sentiment_agent"
USER_ID="user1234"
SESSION_ID="1234"
MODEL_ID="gemini-2.0-flash"
# Tool 1
def get_weather_report(city: str) -> dict:
"""Retrieves the current weather report for a specified city.
Returns:
dict: A dictionary containing the weather information with a 'status' key ('success' or 'error') and a 'report' key with the weather details if successful, or an 'error_message' if an error occurred.
"""
if city.lower() == "london":
return {"status": "success", "report": "The current weather in London is cloudy with a temperature of 18 degrees Celsius and a chance of rain."}
elif city.lower() == "paris":
return {"status": "success", "report": "The weather in Paris is sunny with a temperature of 25 degrees Celsius."}
else:
return {"status": "error", "error_message": f"Weather information for '{city}' is not available."}
weather_tool = FunctionTool(func=get_weather_report)
# Tool 2
def analyze_sentiment(text: str) -> dict:
"""Analyzes the sentiment of the given text.
Returns:
dict: A dictionary with 'sentiment' ('positive', 'negative', or 'neutral') and a 'confidence' score.
"""
if "good" in text.lower() or "sunny" in text.lower():
return {"sentiment": "positive", "confidence": 0.8}
elif "rain" in text.lower() or "bad" in text.lower():
return {"sentiment": "negative", "confidence": 0.7}
else:
return {"sentiment": "neutral", "confidence": 0.6}
sentiment_tool = FunctionTool(func=analyze_sentiment)
# Agent
weather_sentiment_agent = Agent(
model=MODEL_ID,
name='weather_sentiment_agent',
instruction="""You are a helpful assistant that provides weather information and analyzes the sentiment of user feedback.
**If the user asks about the weather in a specific city, use the 'get_weather_report' tool to retrieve the weather details.**
**If the 'get_weather_report' tool returns a 'success' status, provide the weather report to the user.**
**If the 'get_weather_report' tool returns an 'error' status, inform the user that the weather information for the specified city is not available and ask if they have another city in mind.**
**After providing a weather report, if the user gives feedback on the weather (e.g., 'That's good' or 'I don't like rain'), use the 'analyze_sentiment' tool to understand their sentiment.** Then, briefly acknowledge their sentiment.
You can handle these tasks sequentially if needed.""",
tools=[weather_tool, sentiment_tool]
)
# 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=weather_sentiment_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("weather in london?")
import com.google.adk.agents.BaseAgent;
import com.google.adk.agents.LlmAgent;
import com.google.adk.runner.Runner;
import com.google.adk.sessions.InMemorySessionService;
import com.google.adk.sessions.Session;
import com.google.adk.tools.Annotations.Schema;
import com.google.adk.tools.FunctionTool;
import com.google.adk.tools.ToolContext; // Ensure this import is correct
import com.google.common.collect.ImmutableList;
import com.google.genai.types.Content;
import com.google.genai.types.Part;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
public class WeatherSentimentAgentApp {
private static final String APP_NAME = "weather_sentiment_agent";
private static final String USER_ID = "user1234";
private static final String SESSION_ID = "1234";
private static final String MODEL_ID = "gemini-2.0-flash";
/**
* Retrieves the current weather report for a specified city.
*
* @param city The city for which to retrieve the weather report.
* @param toolContext The context for the tool.
* @return A dictionary containing the weather information.
*/
public static Map<String, Object> getWeatherReport(
@Schema(name = "city")
String city,
@Schema(name = "toolContext")
ToolContext toolContext) {
Map<String, Object> response = new HashMap<>();
if (city.toLowerCase(Locale.ROOT).equals("london")) {
response.put("status", "success");
response.put(
"report",
"The current weather in London is cloudy with a temperature of 18 degrees Celsius and a"
+ " chance of rain.");
} else if (city.toLowerCase(Locale.ROOT).equals("paris")) {
response.put("status", "success");
response.put(
"report", "The weather in Paris is sunny with a temperature of 25 degrees Celsius.");
} else {
response.put("status", "error");
response.put(
"error_message", String.format("Weather information for '%s' is not available.", city));
}
return response;
}
/**
* Analyzes the sentiment of the given text.
*
* @param text The text to analyze.
* @param toolContext The context for the tool.
* @return A dictionary with sentiment and confidence score.
*/
public static Map<String, Object> analyzeSentiment(
@Schema(name = "text")
String text,
@Schema(name = "toolContext")
ToolContext toolContext) {
Map<String, Object> response = new HashMap<>();
String lowerText = text.toLowerCase(Locale.ROOT);
if (lowerText.contains("good") || lowerText.contains("sunny")) {
response.put("sentiment", "positive");
response.put("confidence", 0.8);
} else if (lowerText.contains("rain") || lowerText.contains("bad")) {
response.put("sentiment", "negative");
response.put("confidence", 0.7);
} else {
response.put("sentiment", "neutral");
response.put("confidence", 0.6);
}
return response;
}
/**
* Calls the agent with the given query and prints the final response.
*
* @param runner The runner to use.
* @param query The query to send to the agent.
*/
public static void callAgent(Runner runner, String query) {
Content content = Content.fromParts(Part.fromText(query));
InMemorySessionService sessionService = (InMemorySessionService) runner.sessionService();
Session session =
sessionService
.createSession(APP_NAME, USER_ID, /* state= */ null, SESSION_ID)
.blockingGet();
runner
.runAsync(session.userId(), session.id(), content)
.forEach(
event -> {
if (event.finalResponse()
&& event.content().isPresent()
&& event.content().get().parts().isPresent()
&& !event.content().get().parts().get().isEmpty()
&& event.content().get().parts().get().get(0).text().isPresent()) {
String finalResponse = event.content().get().parts().get().get(0).text().get();
System.out.println("Agent Response: " + finalResponse);
}
});
}
public static void main(String[] args) throws NoSuchMethodException {
FunctionTool weatherTool =
FunctionTool.create(
WeatherSentimentAgentApp.class.getMethod(
"getWeatherReport", String.class, ToolContext.class));
FunctionTool sentimentTool =
FunctionTool.create(
WeatherSentimentAgentApp.class.getMethod(
"analyzeSentiment", String.class, ToolContext.class));
BaseAgent weatherSentimentAgent =
LlmAgent.builder()
.model(MODEL_ID)
.name("weather_sentiment_agent")
.description("Weather Sentiment Agent")
.instruction("""
You are a helpful assistant that provides weather information and analyzes the
sentiment of user feedback
**If the user asks about the weather in a specific city, use the
'get_weather_report' tool to retrieve the weather details.**
**If the 'get_weather_report' tool returns a 'success' status, provide the
weather report to the user.**
**If the 'get_weather_report' tool returns an 'error' status, inform the
user that the weather information for the specified city is not available
and ask if they have another city in mind.**
**After providing a weather report, if the user gives feedback on the
weather (e.g., 'That's good' or 'I don't like rain'), use the
'analyze_sentiment' tool to understand their sentiment.** Then, briefly
acknowledge their sentiment.
You can handle these tasks sequentially if needed.
""")
.tools(ImmutableList.of(weatherTool, sentimentTool))
.build();
InMemorySessionService sessionService = new InMemorySessionService();
Runner runner = new Runner(weatherSentimentAgent, APP_NAME, null, sessionService);
// Change the query to ensure the tool is called with a valid city that triggers a "success"
// response from the tool, like "london" (without the question mark).
callAgent(runner, "weather in paris");
}
}
工具上下文¶
对于更高级的场景，ADK 允许你通过在函数签名中包含特殊参数 tool_context: ToolContext 来访问工具函数内的额外上下文信息。通过在函数签名中包含此参数，ADK 将在智能体执行期间调用你的工具时自动提供 ToolContext 类的实例。
ToolContext 提供对几个关键信息和控制杠杆的访问：
-
state: State：读取和修改当前会话的状态。此处所做的更改会被跟踪和持久化。 -
actions: EventActions：影响工具运行后智能体的后续行动（例如，跳过总结，转移到另一个智能体）。 -
function_call_id: str：框架分配给此特定工具调用的唯一标识符。对于跟踪和与认证响应相关联很有用。当在单个模型响应中调用多个工具时，这也很有帮助。 -
function_call_event_id: str：此属性提供触发当前工具调用的 事件 的唯一标识符。这对于跟踪和日志记录目的很有用。 -
auth_response: Any：如果在此工具调用之前完成了认证流程，则包含认证响应/凭证。 -
访问服务：与配置的服务（如制品（Artifacts）和内存）进行交互的方法。
请注意，你不应在工具函数的文档字符串中包含 tool_context 参数。由于 ToolContext 是由 ADK 框架在 LLM 决定调用工具函数 之后 自动注入的，因此它与 LLM 的决策无关，包含它可能会使 LLM 产生混淆。
状态管理¶
tool_context.state 属性提供对与当前会话关联的状态的直接读写访问。它的行为类似于字典，但确保任何修改都被跟踪为增量并由会话服务持久化。这使工具能够在不同的交互和智能体步骤之间维护和共享信息。
-
读取状态：使用标准字典访问（
tool_context.state['my_key']）或.get()方法（tool_context.state.get('my_key', default_value)）。 -
写入状态：直接分配值（
tool_context.state['new_key'] = 'new_value'）。这些更改记录在结果事件的 state_delta 中。 -
状态前缀：记住标准状态前缀：
-
app:*：在应用程序的所有用户之间共享。 -
user:*：特定于当前用户的所有会话。 -
（无前缀）：特定于当前会话。
-
temp:*：临时的，不在调用之间持久化（在单个运行调用中传递数据很有用，但在工具上下文中通常不太有用，工具上下文在 LLM 调用之间运行）。
-
from google.adk.tools import ToolContext, FunctionTool
def update_user_preference(preference: str, value: str, tool_context: ToolContext):
"""Updates a user-specific preference."""
user_prefs_key = "user:preferences"
# Get current preferences or initialize if none exist
preferences = tool_context.state.get(user_prefs_key, {})
preferences[preference] = value
# Write the updated dictionary back to the state
tool_context.state[user_prefs_key] = preferences
print(f"Tool: Updated user preference '{preference}' to '{value}'")
return {"status": "success", "updated_preference": preference}
pref_tool = FunctionTool(func=update_user_preference)
# In an Agent:
# my_agent = Agent(..., tools=[pref_tool])
# When the LLM calls update_user_preference(preference='theme', value='dark', ...):
# The tool_context.state will be updated, and the change will be part of the
# resulting tool response event's actions.state_delta.
import com.google.adk.tools.FunctionTool;
import com.google.adk.tools.ToolContext;
// 更新用户偏好设置。
public Map<String, String> updateUserThemePreference(String value, ToolContext toolContext) {
String userPrefsKey = "user:preferences:theme";
// 获取当前偏好或初始化
String preference = toolContext.state().getOrDefault(userPrefsKey, "").toString();
if (preference.isEmpty()) {
preference = value;
}
// 将更新后的字典写回 state
toolContext.state().put("user:preferences", preference);
System.out.printf("工具：将用户偏好 %s 更新为 %s", userPrefsKey, preference);
return Map.of("status", "success", "updated_preference", toolContext.state().get(userPrefsKey).toString());
// 当 LLM 调用 updateUserThemePreference("dark") 时：
// toolContext.state 会被更新，且更改会作为结果事件的 actions.stateDelta 的一部分。
}
控制智能体流程¶
tool_context.actions 属性（Java 中为 ToolContext.actions()）包含一个 EventActions 对象。修改该对象上的属性可以让你的工具影响工具执行后智能体或框架的行为。
-
skip_summarization: bool：（默认：False）如果设置为 True，指示 ADK 绕过通常总结工具输出的 LLM 调用。如果你的工具的返回值已经是用户可用的消息，这很有用。 -
transfer_to_agent: str：将此设置为另一个智能体的名称。框架将停止当前智能体的执行并将对话控制权转移到指定的智能体。这允许工具动态地将任务交给更专业的智能体。 -
escalate: bool：（默认：False）将此设置为 True 表示当前智能体无法处理请求并应将控制权传递给其父智能体（如果在层次结构中）。在 LoopAgent 中，在子智能体的工具中设置 escalate=True 将终止循环。
示例¶
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import ToolContext
from google.genai import types
APP_NAME="customer_support_agent"
USER_ID="user1234"
SESSION_ID="1234"
def check_and_transfer(query: str, tool_context: ToolContext) -> str:
"""Checks if the query requires escalation and transfers to another agent if needed."""
if "urgent" in query.lower():
print("Tool: Detected urgency, transferring to the support agent.")
tool_context.actions.transfer_to_agent = "support_agent"
return "Transferring to the support agent..."
else:
return f"Processed query: '{query}'. No further action needed."
escalation_tool = FunctionTool(func=check_and_transfer)
main_agent = Agent(
model='gemini-2.0-flash',
name='main_agent',
instruction="""You are the first point of contact for customer support of an analytics tool. Answer general queries. If the user indicates urgency, use the 'check_and_transfer' tool.""",
tools=[check_and_transfer]
)
support_agent = Agent(
model='gemini-2.0-flash',
name='support_agent',
instruction="""You are the dedicated support agent. Mentioned you are a support handler and please help the user with their urgent issue."""
)
main_agent.sub_agents = [support_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=main_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("this is urgent, i cant login")
import com.google.adk.agents.LlmAgent;
import com.google.adk.runner.Runner;
import com.google.adk.sessions.InMemorySessionService;
import com.google.adk.sessions.Session;
import com.google.adk.tools.Annotations.Schema;
import com.google.adk.tools.FunctionTool;
import com.google.adk.tools.ToolContext;
import com.google.common.collect.ImmutableList;
import com.google.genai.types.Content;
import com.google.genai.types.Part;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
public class CustomerSupportAgentApp {
private static final String APP_NAME = "customer_support_agent";
private static final String USER_ID = "user1234";
private static final String SESSION_ID = "1234";
private static final String MODEL_ID = "gemini-2.0-flash";
/**
* Checks if the query requires escalation and transfers to another agent if needed.
*
* @param query The user's query.
* @param toolContext The context for the tool.
* @return A map indicating the result of the check and transfer.
*/
public static Map<String, Object> checkAndTransfer(
@Schema(name = "query", description = "the user query")
String query,
@Schema(name = "toolContext", description = "the tool context")
ToolContext toolContext) {
Map<String, Object> response = new HashMap<>();
if (query.toLowerCase(Locale.ROOT).contains("urgent")) {
System.out.println("Tool: Detected urgency, transferring to the support agent.");
toolContext.actions().setTransferToAgent("support_agent");
response.put("status", "transferring");
response.put("message", "Transferring to the support agent...");
} else {
response.put("status", "processed");
response.put(
"message", String.format("Processed query: '%s'. No further action needed.", query));
}
return response;
}
/**
* Calls the agent with the given query and prints the final response.
*
* @param runner The runner to use.
* @param query The query to send to the agent.
*/
public static void callAgent(Runner runner, String query) {
Content content =
Content.fromParts(Part.fromText(query));
InMemorySessionService sessionService = (InMemorySessionService) runner.sessionService();
// Fixed: session ID does not need to be an optional.
Session session =
sessionService
.createSession(APP_NAME, USER_ID, /* state= */ null, SESSION_ID)
.blockingGet();
runner
.runAsync(session.userId(), session.id(), content)
.forEach(
event -> {
if (event.finalResponse()
&& event.content().isPresent()
&& event.content().get().parts().isPresent()
&& !event.content().get().parts().get().isEmpty()
&& event.content().get().parts().get().get(0).text().isPresent()) {
String finalResponse = event.content().get().parts().get().get(0).text().get();
System.out.println("Agent Response: " + finalResponse);
}
});
}
public static void main(String[] args) throws NoSuchMethodException {
FunctionTool escalationTool =
FunctionTool.create(
CustomerSupportAgentApp.class.getMethod(
"checkAndTransfer", String.class, ToolContext.class));
LlmAgent supportAgent =
LlmAgent.builder()
.model(MODEL_ID)
.name("support_agent")
.description("""
The dedicated support agent.
Mentions it is a support handler and helps the user with their urgent issue.
""")
.instruction("""
You are the dedicated support agent.
Mentioned you are a support handler and please help the user with their urgent issue.
""")
.build();
LlmAgent mainAgent =
LlmAgent.builder()
.model(MODEL_ID)
.name("main_agent")
.description("""
The first point of contact for customer support of an analytics tool.
Answers general queries.
If the user indicates urgency, uses the 'check_and_transfer' tool.
""")
.instruction("""
You are the first point of contact for customer support of an analytics tool.
Answer general queries.
If the user indicates urgency, use the 'check_and_transfer' tool.
""")
.tools(ImmutableList.of(escalationTool))
.subAgents(supportAgent)
.build();
// Fixed: LlmAgent.subAgents() expects 0 arguments.
// Sub-agents are now added to the main agent via its builder,
// as `subAgents` is a property that should be set during agent construction
// if it's not dynamically managed.
InMemorySessionService sessionService = new InMemorySessionService();
Runner runner = new Runner(mainAgent, APP_NAME, null, sessionService);
// Agent Interaction
callAgent(runner, "this is urgent, i cant login");
}
}
解释¶
- 我们定义了两个智能体：
main_agent和support_agent。main_agent被设计为初始联系点。 - 当
main_agent调用check_and_transfer工具时，它会检查用户的查询。 - 如果查询包含单词 "urgent"，工具会访问
tool_context，特别是tool_context.actions，并将 transfer_to_agent 属性设置为support_agent。 - 此操作向框架发出信号，将对话控制权转移到名为
support_agent的智能体。 - 当
main_agent处理紧急查询时，check_and_transfer工具触发转移。后续响应理想情况下将来自support_agent。 - 对于没有紧急性的普通查询，工具只是处理它而不触发转移。
这个示例说明了工具如何通过其 ToolContext 中的 EventActions 动态地影响对话流程，将控制权转移到另一个专门的智能体。
认证¶
ToolContext 提供了工具与认证 API 交互的机制。如果你的工具需要处理认证，你可以使用以下方法：
-
auth_response：如果框架在调用你的工具之前已经处理了认证，则包含凭证（例如令牌）（常见于 RestApiTool 和 OpenAPI 安全方案）。 -
request_credential(auth_config: dict)：如果你的工具确定需要认证但凭证不可用，则调用此方法。这向框架发出信号，基于提供的 auth_config 开始认证流程。 -
get_auth_response()：在后续调用中调用此方法（在成功处理 request_credential 之后）以检索用户提供的凭证。
有关认证流程、配置和示例的详细说明，请参阅专门的工具认证文档页面。
上下文感知数据访问方法¶
这些方法为你的工具提供了方便的方式来与通过配置的服务管理的与会话或用户相关的持久数据进行交互。
-
list_artifacts()（Java 中为listArtifacts()）：返回通过 artifact_service 当前为会话存储的所有制品（Artifacts）的文件名（或键）列表。制品通常是用户上传或工具/智能体生成的文件（图片、文档等）。 -
load_artifact(filename: str)：通过文件名从 artifact_service 检索特定制品（Artifacts）。你可以选择指定版本；如果省略，则返回最新版本。返回包含制品（Artifacts）数据和 mime 类型的google.genai.types.Part对象，如果未找到则返回 None。 -
save_artifact(filename: str, artifact: types.Part)：将制品（Artifacts）的新版本保存到 artifact_service。返回新的版本号（从 0 开始）。 -
search_memory(query: str)使用配置的
memory_service查询用户的长期记忆。这对于从过去的交互或存储的知识中检索相关信息很有用。SearchMemoryResponse 的结构取决于具体的 memory service 实现，但通常包含相关的文本片段或对话摘录。
示例¶
# 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.tools import ToolContext, FunctionTool
from google.genai import types
def process_document(
document_name: str, analysis_query: str, tool_context: ToolContext
) -> dict:
"""Analyzes a document using context from memory."""
# 1. Load the artifact
print(f"Tool: Attempting to load artifact: {document_name}")
document_part = tool_context.load_artifact(document_name)
if not document_part:
return {"status": "error", "message": f"Document '{document_name}' not found."}
document_text = document_part.text # Assuming it's text for simplicity
print(f"Tool: Loaded document '{document_name}' ({len(document_text)} chars).")
# 2. Search memory for related context
print(f"Tool: Searching memory for context related to: '{analysis_query}'")
memory_response = tool_context.search_memory(
f"Context for analyzing document about {analysis_query}"
)
memory_context = "\n".join(
[
m.events[0].content.parts[0].text
for m in memory_response.memories
if m.events and m.events[0].content
]
) # Simplified extraction
print(f"Tool: Found memory context: {memory_context[:100]}...")
# 3. Perform analysis (placeholder)
analysis_result = f"Analysis of '{document_name}' regarding '{analysis_query}' using memory context: [Placeholder Analysis Result]"
print("Tool: Performed analysis.")
# 4. Save the analysis result as a new artifact
analysis_part = types.Part.from_text(text=analysis_result)
new_artifact_name = f"analysis_{document_name}"
version = await tool_context.save_artifact(new_artifact_name, analysis_part)
print(f"Tool: Saved analysis result as '{new_artifact_name}' version {version}.")
return {
"status": "success",
"analysis_artifact": new_artifact_name,
"version": version,
}
doc_analysis_tool = FunctionTool(func=process_document)
# In an Agent:
# Assume artifact 'report.txt' was previously saved.
# Assume memory service is configured and has relevant past data.
# my_agent = Agent(..., tools=[doc_analysis_tool], artifact_service=..., memory_service=...)
// 使用内存上下文分析文档。
// 你也可以使用 Callback Context 或 LoadArtifacts 工具列出、加载和保存制品。
public static @NonNull Maybe<ImmutableMap<String, Object>> processDocument(
@Annotations.Schema(description = "要分析的文档名称。") String documentName,
@Annotations.Schema(description = "分析的查询。") String analysisQuery,
ToolContext toolContext) {
// 1. 列出所有可用制品
System.out.printf(
"列出所有可用制品 %s:", toolContext.listArtifacts().blockingGet());
// 2. 加载制品到内存
System.out.println("工具：尝试加载制品：" + documentName);
Part documentPart = toolContext.loadArtifact(documentName, Optional.empty()).blockingGet();
if (documentPart == null) {
System.out.println("工具：文档 '" + documentName + "' 未找到。");
return Maybe.just(
ImmutableMap.<String, Object>of(
"status", "error", "message", "文档 '" + documentName + "' 未找到。"));
}
String documentText = documentPart.text().orElse("");
System.out.println(
"工具：已加载文档 '" + documentName + "' (" + documentText.length() + " 字符)。");
// 3. 执行分析（占位符）
String analysisResult =
"对文档 '"
+ documentName
+ "' 关于 '"
+ analysisQuery
+ " [占位符分析结果]" 的分析；
System.out.println("工具：已执行分析。");
// 4. 将分析结果保存为新制品
Part analysisPart = Part.fromText(analysisResult);
String newArtifactName = "analysis_" + documentName;
toolContext.saveArtifact(newArtifactName, analysisPart);
return Maybe.just(
ImmutableMap.<String, Object>builder()
.put("status", "success")
.put("analysis_artifact", newArtifactName)
.build());
}
// FunctionTool processDocumentTool =
// FunctionTool.create(ToolContextArtifactExample.class, "processDocument");
// 在智能体中包含此函数工具。
// LlmAgent agent = LlmAgent().builder().tools(processDocumentTool).build();
通过利用 ToolContext，开发人员可以创建更复杂和上下文感知的自定义工具，这些工具与 ADK 的架构无缝集成并增强其智能体的整体能力。
定义有效的工具函数¶
当你将方法或函数作为 ADK 工具使用时，你的定义方式会极大影响智能体能否正确使用它。智能体的大型语言模型（LLM）高度依赖函数的名称、参数（参数）、类型提示和docstring/源码注释来理解其用途并生成正确的调用。
以下是定义有效工具函数的关键指南：
-
函数名：
- 使用描述性、动词 - 名词式的名称，清楚地指示动作（如
get_weather、searchDocuments、schedule_meeting）。 - 避免使用如
run、process、handle_data这类通用名，或doStuff这类过于模糊的名称。即使有良好的描述，像do_stuff这样的名字也可能让模型混淆何时使用该工具与如cancelFlight等其他工具。 - LLM 在工具选择时主要依赖函数名。
- 使用描述性、动词 - 名词式的名称，清楚地指示动作（如
-
参数（Arguments）：
- 你的函数可以有任意数量的参数。
- 使用清晰和描述性的名称（例如，使用
city而不是c，使用search_query而不是q）。 - 在 Python 中为所有参数提供类型提示（例如，
city: str、user_id: int、items: list[str]）。这对于 ADK 为 LLM 生成正确的 schema 是必需的。 - 确保所有参数类型都是 JSON 可序列化的。所有 Java 基本类型以及标准 Python 类型，如
str、int、float、bool、list、dict及其组合通常是安全的。避免将复杂的自定义类实例作为直接参数，除非它们有明确的 JSON 表示。 - 不要为参数设置默认值。例如，
def my_func(param1: str = "default")。在函数调用生成期间，底层模型不可靠地支持或使用默认值。所有必要信息应由 LLM 从上下文中推导或在缺失时明确请求。 self/cls自动处理： 像self（用于实例方法）或cls（用于类方法）这样的隐式参数由 ADK 自动处理，并从显示给 LLM 的 schema 中排除。你只需要为工具要求 LLM 提供的逻辑参数定义类型提示和描述。
-
返回类型：
- 函数的返回值在 Python 中必须是字典（dict），在 Java 中必须是 Map。
- 如果你的函数返回非字典类型（如字符串、数字、列表），ADK 框架会自动将其包装为字典/Map，如
{'result': your_original_return_value}，再传递给模型。 - 设计字典/Map 的键和值时要对 LLM 友好且易于理解。记住，模型会读取这个输出来决定下一步。
- 包含有意义的键。例如，不要只返回错误码如
500，而要返回{'status': 'error', 'error_message': 'Database connection failed'}。 - 强烈建议包含
status键（如'success'、'error'、'pending'、'ambiguous'），以便模型清楚地了解工具执行的结果。
-
Docstring / 源码注释：
- 非常关键。 docstring 是 LLM 获取描述信息的主要来源。
- 清楚说明工具做什么。 明确其用途和局限。
- 说明何时应使用该工具。 提供上下文或示例场景，帮助 LLM 做决策。
- 清楚描述每个参数。 说明 LLM 需要为该参数提供什么信息。
- 说明预期返回字典的结构和含义，尤其是不同
status值及相关数据键。 - 不要描述注入的 ToolContext 参数。避免在 docstring 描述中提及可选的
tool_context: ToolContext参数，因为 LLM 不需要知道它。ToolContext 由 ADK 注入，在 LLM 决定调用后。
良好定义的例子：
def lookup_order_status(order_id: str) -> dict:
"""获取使用其 ID 的客户订单的当前状态。
仅当用户明确要求特定订单的状态并提供订单 ID 时才使用此工具。
不要将其用于一般查询。
参数：
order_id: 要查找的订单的唯一标识符。
返回：
包含订单状态的字典。
可能的状态：'shipped'、'processing'、'pending'、'error'。
成功示例：{'status': 'shipped', 'tracking_number': '1Z9...'}
错误示例：{'status': 'error', 'error_message': 'Order ID not found.'}
"""
# ... 获取状态的函数实现 ...
if status := fetch_status_from_backend(order_id):
return {"status": status.state, "tracking_number": status.tracking} # 示例结构
else:
return {"status": "error", "error_message": f"Order ID {order_id} not found."}
/**
* 获取指定城市的当前天气报告。
*
* @param city 要获取天气报告的城市。
* @param toolContext 工具上下文。
* @return 包含天气信息的字典。
*/
public static Map<String, Object> getWeatherReport(String city, ToolContext toolContext) {
Map<String, Object> response = new HashMap<>();
if (city.toLowerCase(Locale.ROOT).equals("london")) {
response.put("status", "success");
response.put(
"report",
"伦敦当前天气多云，气温 18 摄氏度，有降雨可能。");
} else if (city.toLowerCase(Locale.ROOT).equals("paris")) {
response.put("status", "success");
response.put("report", "巴黎天气晴朗，气温 25 摄氏度。");
} else {
response.put("status", "error");
response.put("error_message", String.format("无法获取 '%s' 的天气信息。", city));
}
return response;
}
- 简洁与聚焦：
- 保持工具聚焦： 每个工具应理想地只执行一个明确定义的任务。
- 参数越少越好： 模型通常更可靠地处理参数少且定义清晰的工具。
- 使用简单数据类型： 尽量使用基础类型（如 Python 的
str、int、bool、float、List[str]，或 Java 的int、byte、short、long、float、double、boolean和char），避免复杂自定义类或深层嵌套结构。 - 分解复杂任务： 将执行多个独立逻辑步骤的函数拆分为更小、更聚焦的工具。例如，不要用一个
update_user_profile(profile: ProfileObject)工具，而是用update_user_name(name: str)、update_user_address(address: str)、update_user_preferences(preferences: list[str])等分开实现。这样更便于 LLM 选择和使用正确的能力。
遵循这些指南，可以为 LLM 提供清晰的结构和语义，从而让你的自定义函数工具被更有效、更可靠地利用，提升智能体的能力。
工具集：分组与动态提供工具 
¶
除了单个工具，ADK 还通过 BaseToolset 接口（定义于 google.adk.tools.base_toolset）引入了工具集（Toolset）的概念。工具集允许你管理并为智能体动态提供一组 BaseTool 实例。
这种方式有如下好处：
- 组织相关工具： 将服务于同一目的的工具分组（如所有数学运算工具，或所有与某一 API 交互的工具）。
- 动态工具可用性： 允许智能体根据当前上下文（如用户权限、会话状态或其他运行时条件）拥有不同的工具。工具集的
get_tools方法可以决定暴露哪些工具。 - 集成外部工具提供方： 工具集可以作为适配器，将外部系统（如 OpenAPI 规范或 MCP 服务器）中的工具转换为 ADK 兼容的
BaseTool对象。
BaseToolset 接口¶
任何作为 ADK 工具集的类都应实现 BaseToolset 抽象基类。该接口主要定义了两个方法：
-
async def get_tools(...) -> list[BaseTool]:这是工具集的核心方法。当 ADK 智能体需要知道其可用工具时，会对其tools列表中的每个BaseToolset实例调用get_tools()。- 它接收一个可选的
readonly_context（ReadonlyContext实例）。该上下文提供只读访问，如当前会话状态（readonly_context.state）、智能体名称和调用 ID。工具集可以利用该上下文动态决定返回哪些工具。 - 必须返回
BaseTool实例的list（如FunctionTool、RestApiTool）。
- 它接收一个可选的
-
async def close(self) -> None:这是一个异步方法，当工具集不再需要时（如智能体服务器关闭或 Runner 被关闭）由 ADK 框架调用。实现此方法以执行必要的清理，如关闭网络连接、释放文件句柄或清理工具集管理的其他资源。
在智能体中使用工具集¶
你可以直接在 LlmAgent 的 tools 列表中包含你的 BaseToolset 实现实例，与单个 BaseTool 实例并列。
当智能体初始化或需要确定其可用能力时，ADK 框架会遍历 tools 列表：
- 如果项是
BaseTool实例，则直接使用。 - 如果项是
BaseToolset实例，则调用其get_tools()方法（传入当前ReadonlyContext），并将返回的BaseTool列表加入智能体可用工具。
示例：简单数学工具集¶
下面是一个为简单算术运算提供工具的工具集示例。
# 1. Define the individual tool functions
def add_numbers(a: int, b: int, tool_context: ToolContext) -> Dict[str, Any]:
"""Adds two integer numbers.
Args:
a: The first number.
b: The second number.
Returns:
A dictionary with the sum, e.g., {'status': 'success', 'result': 5}
"""
print(f"Tool: add_numbers called with a={a}, b={b}")
result = a + b
# Example: Storing something in tool_context state
tool_context.state["last_math_operation"] = "addition"
return {"status": "success", "result": result}
def subtract_numbers(a: int, b: int) -> Dict[str, Any]:
"""Subtracts the second number from the first.
Args:
a: The first number.
b: The second number.
Returns:
A dictionary with the difference, e.g., {'status': 'success', 'result': 1}
"""
print(f"Tool: subtract_numbers called with a={a}, b={b}")
return {"status": "success", "result": a - b}
# 2. Create the Toolset by implementing BaseToolset
class SimpleMathToolset(BaseToolset):
def __init__(self, prefix: str = "math_"):
self.prefix = prefix
# Create FunctionTool instances once
self._add_tool = FunctionTool(
func=add_numbers,
name=f"{self.prefix}add_numbers", # Toolset can customize names
)
self._subtract_tool = FunctionTool(
func=subtract_numbers, name=f"{self.prefix}subtract_numbers"
)
print(f"SimpleMathToolset initialized with prefix '{self.prefix}'")
async def get_tools(
self, readonly_context: Optional[ReadonlyContext] = None
) -> List[BaseTool]:
print(f"SimpleMathToolset.get_tools() called.")
# Example of dynamic behavior:
# Could use readonly_context.state to decide which tools to return
# For instance, if readonly_context.state.get("enable_advanced_math"):
# return [self._add_tool, self._subtract_tool, self._multiply_tool]
# For this simple example, always return both tools
tools_to_return = [self._add_tool, self._subtract_tool]
print(f"SimpleMathToolset providing tools: {[t.name for t in tools_to_return]}")
return tools_to_return
async def close(self) -> None:
# No resources to clean up in this simple example
print(f"SimpleMathToolset.close() called for prefix '{self.prefix}'.")
await asyncio.sleep(0) # Placeholder for async cleanup if needed
# 3. Define an individual tool (not part of the toolset)
def greet_user(name: str = "User") -> Dict[str, str]:
"""Greets the user."""
print(f"Tool: greet_user called with name={name}")
return {"greeting": f"Hello, {name}!"}
greet_tool = FunctionTool(func=greet_user)
# 4. Instantiate the toolset
math_toolset_instance = SimpleMathToolset(prefix="calculator_")
# 5. Define an agent that uses both the individual tool and the toolset
calculator_agent = LlmAgent(
name="CalculatorAgent",
model="gemini-2.0-flash", # Replace with your desired model
instruction="You are a helpful calculator and greeter. "
"Use 'greet_user' for greetings. "
"Use 'calculator_add_numbers' to add and 'calculator_subtract_numbers' to subtract. "
"Announce the state of 'last_math_operation' if it's set.",
tools=[greet_tool, math_toolset_instance], # Individual tool # Toolset instance
)
在本例中：
SimpleMathToolset实现了BaseToolset，其get_tools()方法返回add_numbers和subtract_numbers的FunctionTool实例，并用前缀自定义了它们的名称。calculator_agent配置了单独的greet_tool和一个SimpleMathToolset实例。- 当运行
calculator_agent时，ADK 会调用math_toolset_instance.get_tools()。智能体的 LLM 将能访问greet_user、calculator_add_numbers和calculator_subtract_numbers来处理用户请求。 add_numbers工具演示了如何写入tool_context.state，且智能体指令中提及了读取该 state。close()方法被调用以确保释放工具集持有的资源。
工具集为你的 ADK 智能体组织、管理和动态提供工具集合，带来了更模块化、可维护和适应性强的智能体应用。