函数工具¶

什么是函数工具？¶

当现成的工具不能完全满足特定需求时，开发人员可以创建自定义函数工具。这允许定制功能，例如连接到专有数据库或实现独特算法。

例如，函数工具"myfinancetool"可能是计算特定财务指标的函数。ADK 还支持长时运行函数，因此如果该计算需要一段时间，智能体可以继续处理其他任务。

ADK 提供了几种创建函数工具的方法，每种方法适合不同的复杂性和控制级别：

函数工具
长时运行函数工具
智能体作为工具

1. 函数工具¶

将函数转换为工具是将自定义逻辑集成到智能体中的直接方法。这种方法提供灵活性和快速集成。

参数¶

使用标准的 JSON 可序列化类型（例如，字符串、整数、列表、字典）定义你的函数参数。重要的是避免为参数设置默认值，因为语言模型（LLM）目前不支持解释它们。

返回类型¶

Python 函数工具的首选返回类型是字典。这允许你用键值对构建响应，为 LLM 提供上下文和清晰度。如果你的函数返回的类型不是字典，框架会自动将其封装到一个名为 "result" 的单一键的字典中。

努力使你的返回值尽可能具有描述性。例如，不要返回数字错误代码，而是返回带有 "error_message" 键的字典，其中包含人类可读的解释。记住，是 LLM 而不是一段代码需要理解结果。作为最佳实践，在你的返回字典中包含一个 "status" 键来表示整体结果（例如，"success"、"error"、"pending"），为 LLM 提供关于操作状态的明确信号。

文档字符串¶

你的函数的文档字符串作为工具的描述，并被发送给 LLM。因此，一个写得好且全面的文档字符串对于 LLM 有效地理解如何使用工具至关重要。清楚地解释函数的目的、参数的含义和预期的返回值。

??? "示例" {: #example}

这个工具是一个 Python 函数，用于获取给定股票代码/符号的股票价格。

<u>注意</u>：在使用此工具之前，你需要 `pip install yfinance` 库。

```py
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],
)


# 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")

```

这个工具的返回值将被封装到一个字典中。

```json
{"result": "$123"}
```

最佳实践¶

虽然你在定义函数方面有相当大的灵活性，但请记住，简单性增强了 LLM 的可用性。考虑这些指导原则：

参数越少越好： 最小化参数数量以减少复杂性。
简单数据类型： 尽可能使用 str 和 int 等原始数据类型而不是自定义类。
有意义的名称： 函数的名称和参数名称显著影响 LLM 如何解释和利用工具。选择清晰反映函数目的和输入含义的名称。避免像 do_stuff() 这样的通用名称。

2. 长时运行函数工具¶

设计用于需要大量处理时间而不阻塞智能体执行的任务。这个工具是 FunctionTool 的子类。

当使用 LongRunningFunctionTool 时，你的 Python 函数可以启动长时间运行的操作，并可选择返回中间结果，以便让模型和用户了解进度。然后，智能体可以继续处理其他任务。一个例子是人在环路场景，智能体在继续任务前需要人类批准。

工作原理¶

你用 LongRunningFunctionTool 包装一个 Python 生成器函数（使用 yield 的函数）。

初始化： 当 LLM 调用工具时，你的生成器函数开始执行。
中间更新（yield）： 你的函数应定期产生中间 Python 对象（通常是字典）以报告进度。ADK 框架获取每个产生的值，并将其封装在 FunctionResponse 中发送回 LLM。这允许 LLM 通知用户（例如，状态、完成百分比、消息）。
完成（return）： 任务完成后，生成器函数使用 return 提供最终的 Python 对象结果。
框架处理： ADK 框架管理执行。它将每个产生的值作为中间 FunctionResponse 发送回去。当生成器完成时，框架将返回值作为最终 FunctionResponse 的内容发送，向 LLM 发出长时间运行操作结束的信号。

创建工具¶

定义你的生成器函数并使用 LongRunningFunctionTool 类包装它：

from google.adk.tools import LongRunningFunctionTool

# 定义你的生成器函数（见下面的例子）
def my_long_task_generator(*args, **kwargs):
    # ... 设置 ...
    yield {"status": "pending", "message": "Starting task..."} # 框架将此作为 FunctionResponse 发送
    # ... 增量执行工作 ...
    yield {"status": "pending", "progress": 50}               # 框架将此作为 FunctionResponse 发送
    # ... 完成工作 ...
    return {"status": "completed", "result": "Final outcome"} # 框架将此作为最终 FunctionResponse 发送

# 包装函数
my_tool = LongRunningFunctionTool(func=my_long_task_generator)

中间更新¶

产生结构化 Python 对象（如字典）对于提供有意义的更新至关重要。包括以下键：

status：例如，"pending"、"running"、"waiting_for_input"
progress：例如，百分比，已完成的步骤
message：用户/LLM 的描述性文本
estimated_completion_time：如果可计算

你产生的每个值都被框架打包到 FunctionResponse 中并发送给 LLM。

最终结果¶

你的生成器函数返回的 Python 对象被视为工具执行的最终结果。框架将这个值（即使是 None）打包到发送回 LLM 的最终 FunctionResponse 的内容中，表明工具执行完成。

??? "示例：文件处理模拟" {: #example-file-processing-simulation}

```py
import time
from google.adk.agents import Agent
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 generator function
def process_large_file(file_path: str) -> dict:
    """
    Simulates processing a large file, yielding progress updates.

    Args:
      file_path: Path to the file being processed.

    Returns: 
      A final status dictionary.
    """
    total_steps = 5

    # This dict will be sent in the first FunctionResponse
    yield {"status": "pending", "message": f"Starting processing for {file_path}..."}

    for i in range(total_steps):
        time.sleep(1)  # Simulate work for one step
        progress = (i + 1) / total_steps
        # Each yielded dict is sent in a subsequent FunctionResponse
        yield {
            "status": "pending",
            "progress": f"{int(progress * 100)}%",
            "estimated_completion_time": f"~{total_steps - (i + 1)} seconds remaining"
        }

    # This returned dict will be sent in the final FunctionResponse
    return {"status": "completed", "result": f"Successfully processed file: {file_path}"}

# 2. Wrap the function with LongRunningFunctionTool
long_running_tool = LongRunningFunctionTool(func=process_large_file)

# 3. Use the tool in an Agent
file_processor_agent = Agent(
    # Use a model compatible with function calling
    model="gemini-2.0-flash",
    name='file_processor_agent',
    instruction="""You are an agent that processes large files. When the user provides a file path, use the 'process_large_file' tool. Keep the user informed about the progress based on the tool's updates (which arrive as function responses). Only provide the final result when the tool indicates completion in its final function response.""",
    tools=[long_running_tool]
)


APP_NAME = "file_processor"
USER_ID = "1234"
SESSION_ID = "session1234"

# 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=file_processor_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("Replace with a path to your file...")

```

此示例的关键方面¶

process_large_file：这个生成器模拟了一个冗长的操作，产生中间状态/进度字典。
LongRunningFunctionTool：包装生成器；框架处理发送产生的更新和最终返回值作为顺序 FunctionResponses。
智能体指令：指导 LLM 使用工具并理解传入的 FunctionResponse 流（进度与完成）以更新用户。
最终返回：函数返回最终结果字典，它在结束的 FunctionResponse 中发送，表示完成。

3. 智能体作为工具¶

这个强大的功能允许你通过将系统中的其他智能体作为工具调用来利用它们的能力。智能体作为工具使你能够调用另一个智能体执行特定任务，有效地委托责任。这在概念上类似于创建一个调用另一个智能体并使用该智能体的响应作为函数返回值的 Python 函数。

与子智能体的关键区别¶

重要的是区分智能体作为工具和子智能体。

智能体作为工具： 当智能体 A 调用智能体 B 作为工具（使用智能体作为工具）时，智能体 B 的答案传回给智能体 A，然后智能体 A 对答案进行总结并生成对用户的响应。智能体 A 保持控制并继续处理未来的用户输入。
子智能体： 当智能体 A 调用智能体 B 作为子智能体时，回答用户的责任完全转移给智能体 B。智能体 A 实际上脱离了循环。所有后续用户输入都将由智能体 B 回答。

使用方法¶

要将智能体用作工具，请使用 AgentTool 类包装智能体。

tools=[AgentTool(agent=agent_b)]

自定义¶

AgentTool 类提供以下属性来自定义其行为：

skip_summarization: bool： 如果设置为 True，框架将绕过对工具智能体响应的基于 LLM 的总结。当工具的响应已经格式良好且不需要进一步处理时，这很有用。

??? "示例" {: #example-1}

```py
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)

```

工作原理¶

当 main_agent 收到长文本时，其指令告诉它对长文本使用 'summarize' 工具。
框架将 'summarize' 识别为包装 summary_agent 的 AgentTool。
在后台，main_agent 将以长文本作为输入调用 summary_agent。
summary_agent 将根据其指令处理文本并生成摘要。
summary_agent 的响应然后传回给 main_agent。
然后，main_agent 可以获取摘要并制定其对用户的最终响应（例如，"这是文本的摘要：..."）