工具¶

什么是工具？¶

在 ADK 的上下文中，工具代表提供给 AI 智能体的特定能力，使其能够执行行动并与其核心文本生成和推理能力之外的世界进行交互。通常，有能力的智能体与基础语言模型的区别在于它们有效地使用工具。

从技术上讲，工具通常是一个模块化的代码组件—比如 Python 函数、类方法，甚至是另一个专门的智能体—设计用于执行独特的、预定义的任务。这些任务通常涉及与外部系统或数据交互。

Agent tool call

主要特点¶

面向行动： 工具执行特定行动，例如：

查询数据库
发出 API 请求（例如，获取天气数据、预订系统）
搜索网络
执行代码片段
从文档中检索信息（RAG）
与其他软件或服务交互

扩展智能体能力： 它们使智能体能够访问实时信息，影响外部系统，并克服其训练数据固有的知识限制。

执行预定义逻辑： 至关重要的是，工具执行特定的、由开发者定义的逻辑。它们不像智能体的核心大型语言模型（LLM）那样具有自己独立的推理能力。LLM 会推理使用哪个工具、何时使用以及使用什么输入，但工具本身只是执行其指定的功能。

智能体如何使用工具¶

智能体通过通常涉及函数调用的机制动态地利用工具。该过程通常遵循以下步骤：

推理： 智能体的 LLM 分析其系统指令、对话历史和用户请求。
选择： 基于分析，LLM 根据智能体可用的工具和描述每个工具的文档字符串决定执行哪个工具（如果有的话）。
调用： LLM 生成所选工具所需的参数（输入）并触发其执行。
观察： 智能体接收工具返回的输出（结果）。
完成： 智能体将工具的输出纳入其持续的推理过程中，以制定下一个响应、决定后续步骤或确定是否已达成目标。

可以将工具视为智能体的智能核心（LLM）可以根据需要访问和利用的专用工具包，以完成复杂任务。

ADK 中的工具类型¶

ADK 通过支持几种类型的工具提供灵活性：

函数工具： 由你创建的工具，根据你的特定应用需求量身定制。
- 函数/方法： 在你的代码中定义标准同步函数或方法（例如，Python def）。
- 智能体作为工具： 使用另一个可能专门的智能体作为父智能体的工具。
- 长时运行函数工具： 支持执行异步操作或需要大量时间完成的工具。
内置工具： 框架提供的可直接使用的常见任务工具。示例：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?")

工具上下文¶

对于更高级的场景，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.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.

控制智能体流程¶

tool_context.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")

解释¶

我们定义了两个智能体：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()：返回当前通过 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 的结构取决于特定的记忆服务实现，但通常包含相关的文本片段或对话摘录。

示例¶

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 = 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=...)

通过利用 ToolContext，开发人员可以创建更复杂和上下文感知的自定义工具，这些工具与 ADK 的架构无缝集成并增强其智能体的整体能力。

定义有效的工具函数¶

当使用标准 Python 函数作为 ADK 工具时，你如何定义它会显著影响智能体正确使用它的能力。智能体的大型语言模型（LLM）严重依赖函数的名称、参数（参数）、类型提示和文档字符串来理解其目的并生成正确的调用。

以下是定义有效工具函数的关键指南：

函数名称：
- 使用描述性的、基于动词 - 名词的名称，清楚地表明行动（例如，get_weather、search_documents、schedule_meeting）。
- 避免像 run、process、handle_data 这样的通用名称，或者像 do_stuff 这样过于模糊的名称。即使有很好的描述，像 do_stuff 这样的名称可能会使模型对何时使用该工具与例如 cancel_flight 感到困惑。
- LLM 在工具选择过程中使用函数名称作为主要标识符。
参数（参数）：
- 你的函数可以有任意数量的参数。
- 使用清晰且描述性的名称（例如，city 而不是 c，search_query 而不是 q）。
- 为所有参数提供类型提示（例如，city: str、user_id: int、items: list[str]）。这对于 ADK 生成 LLM 的正确模式至关重要。
- 确保所有参数类型都是 JSON 可序列化的。标准 Python 类型如 str、int、float、bool、list、dict 及其组合通常是安全的。避免使用复杂的自定义类实例作为直接参数，除非它们有明确的 JSON 表示。
- 不要为参数设置默认值。例如，def my_func(param1: str = "default")。底层模型在函数调用生成期间不可靠地支持或使用默认值。所有必要的信息应由 LLM 从上下文中派生或在缺失时显式请求。
返回类型：
- 函数的返回值必须是字典（dict）。
- 如果你的函数返回非字典类型（例如，字符串、数字、列表），ADK 框架会在将结果传回模型之前自动将其包装成像 {'result': your_original_return_value} 这样的字典。
- 设计字典键和值使其易于被 LLM 理解。记住，模型会阅读这个输出来决定下一步。
- 包含有意义的键。例如，不要只返回错误代码如 500，而是返回 {'status': 'error', 'error_message': 'Database connection failed'}。
- 强烈建议包含一个 status 键（例如，'success'、'error'、'pending'、'ambiguous'）以清楚地表明工具执行的结果供模型使用。

文档字符串：

这一点至关重要。 文档字符串是 LLM 的主要描述性信息来源。
清楚地说明工具做什么。 明确其目的和限制。
解释何时应该使用工具。 提供上下文或示例场景来指导 LLM 的决策。
清楚地描述每个参数。 解释 LLM 需要为该参数提供什么信息。
描述预期的 dict 返回值的结构和含义，特别是不同的 status 值和相关的数据键。

良好定义的例子：

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."}

简单性和专注性：
- 保持工具专注： 每个工具理想情况下应执行一个明确定义的任务。
- 参数越少越好： 相比于有许多可选或复杂参数的工具，模型通常更可靠地处理具有较少、明确定义参数的工具。
- 使用简单数据类型： 在可能的情况下，更倾向于基本类型（str、int、bool、float、List[str] 等）而不是复杂的自定义类或深度嵌套结构作为参数。
- 分解复杂任务： 将执行多个不同逻辑步骤的函数分解为更小、更专注的工具。例如，不要使用单个 update_user_profile(profile: ProfileObject) 工具，而是考虑单独的工具，如 update_user_name(name: str)、update_user_address(address: str)、update_user_preferences(preferences: list[str]) 等。这使 LLM 更容易选择和使用正确的功能。

通过遵循这些准则，你为 LLM 提供了它需要的清晰度和结构，以有效地利用你的自定义函数工具，从而导致更有能力和可靠的智能体行为。