第5章：Tool Use (Function Calling)

Tool Use 模式概述

到目前为止，我们已经讨论了主要涉及编排语言模型交互和管理 agent 内部工作流中信息流的 agentic 模式（Chaining、Routing、Parallelization、Reflection）。然而，要使 agent 真正有用并与真实世界或外部系统进行交互，它们需要使用 Tool 的能力。

Tool Use 模式（通常通过一种称为 Function Calling 的机制实现）使 agent 能够与外部 API、数据库、服务进行交互，甚至执行代码。它允许 agent 核心的 LLM 根据用户的请求或任务的当前状态来决定何时以及如何使用特定的外部函数。

该过程通常包括：

Tool 定义：外部函数或能力被定义并描述给 LLM。此描述包括函数的目的、名称、以及它接受的参数及其类型和描述。
LLM 决策：LLM 接收用户的请求和可用的 tool 定义。基于其对请求和 tool 的理解，LLM 决定是否需要调用一个或多个 tool 来满足请求。
函数调用生成：如果 LLM 决定使用 tool，它会生成一个结构化输出（通常是 JSON 对象），指定要调用的 tool 名称和要传递给它的参数（从用户请求中提取）。
Tool 执行：agentic 框架或编排层拦截此结构化输出。它识别请求的 tool 并使用提供的参数执行实际的外部函数。
观察/结果：tool 执行的输出或结果返回给 agent。
LLM 处理（可选但常见）：LLM 将 tool 的输出作为上下文接收，并使用它来形成对用户的最终响应或决定工作流中的下一步（这可能涉及调用另一个 tool、进行 Reflection 或提供最终答案）。

该模式是基础性的，因为它打破了 LLM 训练数据的限制，使其能够访问最新信息、执行其内部无法完成的计算、与用户特定的数据进行交互，或触发真实世界的行动。Function calling 是弥合 LLM 推理能力与广泛的外部功能之间差距的技术机制。

虽然"function calling"恰当地描述了调用特定的、预定义的代码函数，但考虑更广泛的概念"tool calling"是有益的。这个更广泛的术语承认 agent 的能力可以远超简单的函数执行。一个"tool"可以是一个传统函数，但它也可以是一个复杂的 API 端点、对数据库的请求，甚至是指向另一个专门的 agent 的指令。这种视角使我们能够构想更复杂的系统，例如，一个主 agent 可能将复杂的数据分析任务委托给一个专门的"分析师 agent"，或通过其 API 查询外部知识库。

以"tool calling"的思维方式更好地捕捉了 agent 作为编排者跨越多样化数字资源生态系统和其他智能实体发挥全部潜力的能力。

LangChain、LangGraph 和 Google Agent Developer Kit (ADK) 等框架为定义 tool 和将它们集成到 agent 工作流中提供了强大的支持，通常利用现代 LLM（如 Gemini 或 OpenAI 系列）的原生 Function Calling 能力。在这些框架的"画布"上，你定义 tool，然后配置 agent（通常是 LLM Agent）使其能够感知并使用这些 tool。

Tool Use 是构建强大、交互式且具有外部感知能力的 agent 的基石模式。

实际应用与用例

Tool Use 模式几乎适用于 agent 需要超越生成文本来执行操作或检索特定动态信息的任何场景：

1. 从外部来源检索信息：

访问 LLM 训练数据中不存在的实时数据或信息。

用例：一个天气 agent。
- Tool：一个接受地点并返回当前天气状况的天气 API。
- Agent 流程：用户问"伦敦天气如何？"，LLM 识别出需要天气 tool，使用"London"调用该 tool，tool 返回数据，LLM 将数据格式化为用户友好的响应。

2. 与数据库和 API 交互：

对结构化数据执行查询、更新或其他操作。

用例：一个电商 agent。
- Tool：用于检查产品库存、获取订单状态或处理支付的 API 调用。
- Agent 流程：用户问"产品 X 有库存吗？"，LLM 调用库存 API，tool 返回库存数量，LLM 告知用户库存状态。

3. 执行计算和数据分析：

使用外部计算器、数据分析库或统计工具。

用例：一个金融 agent。
- Tool：计算器函数、股票市场数据 API、电子表格工具。
- Agent 流程：用户问"AAPL 的当前价格是多少？如果我在150美元时买入100股，潜在的利润是多少？"，LLM 调用股票 API，获取当前价格，然后调用计算器 tool，获取结果，格式化响应。

4. 发送通信：

发送电子邮件、消息或对外部通信服务进行 API 调用。

用例：一个个人助理 agent。
- Tool：一个电子邮件发送 API。
- Agent 流程：用户说"给 John 发一封关于明天会议的电子邮件。"，LLM 使用从请求中提取的收件人、主题和正文调用电子邮件 tool。

5. 执行代码：

在安全的环境中运行代码片段以执行特定任务。

用例：一个编码助手 agent。
- Tool：一个代码解释器。
- Agent 流程：用户提供一个 Python 代码片段并问"这段代码做什么？"，LLM 使用解释器 tool 运行代码并分析其输出。

6. 控制其他系统或设备：

与智能家居设备、IoT 平台或其他连接系统进行交互。

用例：一个智能家居 agent。
- Tool：一个控制智能灯的 API。
- Agent 流程：用户说"关掉客厅的灯。"，LLM 使用命令和目标设备调用智能家居 tool。

Tool Use 是将语言模型从文本生成器转变为能够在数字或物理世界中感知、推理和行动的 agent 的关键（见图1）。

Agent 使用 Tool 的示例

实战代码示例（LangChain）

在 LangChain 框架中实现 Tool Use 是一个两阶段的过程。首先，定义一个或多个 tool，通常通过封装现有的 Python 函数或其他可运行组件。随后，这些 tool 被绑定到语言模型，从而使模型在确定需要外部函数调用来满足用户查询时能够生成结构化的 tool 使用请求。

以下实现将通过首先定义一个简单的函数来模拟信息检索 tool 来演示此原理。随后，将构建一个 agent 并配置其利用此 tool 来响应用户输入。执行此示例需要安装核心 LangChain 库和特定模型的提供程序包。此外，必须通过在本地环境中配置的 API key 与所选语言模型服务进行正确的身份验证。

python

import os, getpass
import asyncio
import nest_asyncio
from typing import List
from dotenv import load_dotenv
import logging
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool as langchain_tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

# UNCOMMENT
# Prompt the user securely and set API keys as an environment variables
os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter your Google API key: ")
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

try:
    # A model with function/tool calling capabilities is required.
    llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
    print(f" Language model initialized: {llm.model}")
except Exception as e:
    print(f" Error initializing language model: {e}")
    llm = None

# --- Define a Tool ---
@langchain_tool
def search_information(query: str) -> str:
    """
    Provides factual information on a given topic. Use this tool to find answers to phrases
    like 'capital of France' or 'weather in London?'.
    """
    print(f"\n--- Tool Called: search_information with query: '{query}' ---")
    # Simulate a search tool with a dictionary of predefined results.
    simulated_results = {
        "weather in london": "The weather in London is currently cloudy with a temperature of 15°C.",
        "capital of france": "The capital of France is Paris.",
        "population of earth": "The estimated population of Earth is around 8 billion people.",
        "tallest mountain": "Mount Everest is the tallest mountain above sea level.",
        "default": f"Simulated search result for '{query}': No specific information found, but the topic seems interesting."
    }
    result = simulated_results.get(query.lower(), simulated_results["default"])
    print(f"--- TOOL RESULT: {result} ---")
    return result

tools = [search_information]

# --- Create a Tool-Calling Agent ---
if llm:
    # This prompt template requires an `agent_scratchpad` placeholder for the agent's internal steps.
    agent_prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])

    # Create the agent, binding the LLM, tools, and prompt together.
    agent = create_tool_calling_agent(llm, tools, agent_prompt)

    # AgentExecutor is the runtime that invokes the agent and executes the chosen tools.
    # The 'tools' argument is not needed here as they are already bound to the agent.
    agent_executor = AgentExecutor(agent=agent, verbose=True, tools=tools)

async def run_agent_with_tool(query: str):
    """Invokes the agent executor with a query and prints the final response."""
    print(f"\n--- Running Agent with Query: '{query}' ---")
    try:
        response = await agent_executor.ainvoke({"input": query})
        print("\n--- Final Agent Response ---")
        print(response["output"])
    except Exception as e:
        print(f"\n An error occurred during agent execution: {e}")

async def main():
    """Runs all agent queries concurrently."""
    tasks = [
        run_agent_with_tool("What is the capital of France?"),
        run_agent_with_tool("What's the weather like in London?"),
        run_agent_with_tool("Tell me something about dogs.") # Should trigger the default tool response
    ]
    await asyncio.gather(*tasks)

nest_asyncio.apply()
asyncio.run(main())

该代码使用 LangChain 库和 Google Gemini 模型设置了一个 Tool Calling agent。它定义了一个 search_information tool，模拟为特定查询提供事实性答案。该 tool 对"weather in london"、"capital of france"和"population of earth"有预定义的响应，对其他查询有默认响应。初始化了一个 ChatGoogleGenerativeAI 模型，确保其具有 Tool Calling 能力。创建了一个 ChatPromptTemplate 来指导 agent 的交互。create_tool_calling_agent 函数用于将语言模型、tool 和提示组合成一个 agent。然后设置了 AgentExecutor 来管理 agent 的执行和 tool 调用。定义了 run_agent_with_tool 异步函数来使用给定查询调用 agent 并打印结果。main 异步函数准备了多个并发运行的查询。这些查询旨在测试 search_information tool 的特定和默认响应。最后，asyncio.run(main()) 调用执行所有 agent 任务。代码包括在继续进行 agent 设置和执行之前检查 LLM 初始化是否成功的逻辑。

实战代码示例（CrewAI）

这段代码提供了一个在 CrewAI 框架中实现 Function Calling (Tool) 的实用示例。它设置了一个简单的场景，一个 agent 配备了一个查找信息的 tool。该示例专门演示了使用此 agent 和 tool 获取模拟股票价格。

python

# pip install crewai langchain-openai
import os
from crewai import Agent, Task, Crew
from crewai.tools import tool
import logging

# --- Best Practice: Configure Logging ---
# A basic logging setup helps in debugging and tracking the crew's execution.
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# --- Set up your API Key ---
# For production, it's recommended to use a more secure method for key management
# like environment variables loaded at runtime or a secret manager.
#
# Set the environment variable for your chosen LLM provider (e.g., OPENAI_API_KEY)
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
# os.environ["OPENAI_MODEL_NAME"] = "gpt-4o"

# --- 1. Refactored Tool: Returns Clean Data ---
# The tool now returns raw data (a float) or raises a standard Python error.
# This makes it more reusable and forces the agent to handle outcomes properly.
@tool("Stock Price Lookup Tool")
def get_stock_price(ticker: str) -> float:
    """
    Fetches the latest simulated stock price for a given stock ticker symbol.
    Returns the price as a float. Raises a ValueError if the ticker is not found.
    """
    logging.info(f"Tool Call: get_stock_price for ticker '{ticker}'")

    simulated_prices = {
        "AAPL": 178.15,
        "GOOGL": 1750.30,
        "MSFT": 425.50,
    }

    price = simulated_prices.get(ticker.upper())
    if price is not None:
        return price
    else:
        # Raising a specific error is better than returning a string.
        # The agent is equipped to handle exceptions and can decide on the next action.
        raise ValueError(f"Simulated price for ticker '{ticker.upper()}' not found.")

# --- 2. Define the Agent ---
# The agent definition remains the same, but it will now leverage the improved tool.
financial_analyst_agent = Agent(
    role='Senior Financial Analyst',
    goal='Analyze stock data using provided tools and report key prices.',
    backstory="You are an experienced financial analyst adept at using data sources to find stock information. You provide clear, direct answers.",
    verbose=True,
    tools=[get_stock_price],
    # Allowing delegation can be useful, but is not necessary for this simple task.
    allow_delegation=False,
)

# --- 3. Refined Task: Clearer Instructions and Error Handling ---
# The task description is more specific and guides the agent on how to react
# to both successful data retrieval and potential errors.
analyze_aapl_task = Task(
    description=(
        "What is the current simulated stock price for Apple (ticker: AAPL)? "
        "Use the 'Stock Price Lookup Tool' to find it. "
        "If the ticker is not found, you must report that you were unable to retrieve the price."
    ),
    expected_output=(
        "A single, clear sentence stating the simulated stock price for AAPL. "
        "For example: 'The simulated stock price for AAPL is $178.15.' "
        "If the price cannot be found, state that clearly."
    ),
    agent=financial_analyst_agent,
)

# --- 4. Formulate the Crew ---
# The crew orchestrates how the agent and task work together.
financial_crew = Crew(
    agents=[financial_analyst_agent],
    tasks=[analyze_aapl_task],
    verbose=True # Set to False for less detailed logs in production
)

# --- 5. Run the Crew within a Main Execution Block ---
# Using a __name__ == "__main__": block is a standard Python best practice.
def main():
    """Main function to run the crew."""
    # Check for API key before starting to avoid runtime errors.
    if not os.environ.get("OPENAI_API_KEY"):
        print("ERROR: The OPENAI_API_KEY environment variable is not set.")
        print("Please set it before running the script.")
        return

    print("\n## Starting the Financial Crew...")
    print("---------------------------------")

    # The kickoff method starts the execution.
    result = financial_crew.kickoff()

    print("\n---------------------------------")
    print("## Crew execution finished.")
    print("\nFinal Result:\n", result)

if __name__ == "__main__":
    main()

这段代码演示了一个使用 CrewAI 库模拟金融分析任务的简单应用。它定义了一个自定义 tool get_stock_price，模拟查找预定义股票代码的股票价格。该 tool 被设计为对有效的股票代码返回浮点数，对无效的股票代码抛出 ValueError。创建了一个名为 financial_analyst_agent 的 CrewAI Agent，其角色为高级金融分析师。该 agent 被赋予了 get_stock_price tool 进行交互。定义了一个 Task analyze_aapl_task，专门指示 agent 使用该 tool 查找 AAPL 的模拟股票价格。任务描述包括关于如何处理使用 tool 时的成功和失败情况的清晰指令。组装了一个 Crew，包含 financial_analyst_agent 和 analyze_aapl_task。agent 和 crew 都启用了详细设置，以便在执行期间提供详细的日志记录。脚本的主要部分使用标准 if __name__ == "__main__": 块中的 kickoff() 方法运行 crew 的任务。在启动 crew 之前，它检查是否设置了 OPENAI_API_KEY 环境变量，这是 agent 运行所需的。crew 执行的结果（即任务的输出）随后被打印到控制台。代码还包括基本的日志配置，以便更好地跟踪 crew 的行动和 tool 调用。它使用环境变量进行 API key 管理，但指出在生产环境中推荐使用更安全的方法。简而言之，核心逻辑展示了如何定义 tool、agent 和任务以在 CrewAI 中创建协作工作流。

实战代码（Google ADK）

Google Agent Developer Kit (ADK) 包含一个原生集成 tool 库，可以直接整合到 agent 的能力中。

Google 搜索：这类组件的一个主要示例是 Google Search tool。该 tool 作为 Google 搜索引擎的直接接口，使 agent 具备执行网络搜索和检索外部信息的功能。

python

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.genai import types
import nest_asyncio
import asyncio

# Define variables required for Session setup and Agent execution
APP_NAME="Google Search_agent"
USER_ID="user1234"
SESSION_ID="1234"

# Define Agent with access to search tool
root_agent = ADKAgent(
    name="basic_search_agent",
    model="gemini-2.0-flash-exp",
    description="Agent to answer questions using Google Search.",
    instruction="I can answer your questions by searching the internet. Just ask me anything!",
    tools=[google_search] # Google Search is a pre-built tool to perform Google searches.
)

# Agent Interaction
async def call_agent(query):
    """
    Helper function to call the agent with a query.
    """
    # 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)

    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)

nest_asyncio.apply()
asyncio.run(call_agent("what's the latest ai news?"))

这段代码演示了如何创建和使用一个由 Google ADK for Python 驱动的基本 agent。该 agent 被设计为通过使用 Google Search 作为 tool 来回答问题。首先，导入来自 IPython、google.adk 和 google.genai 的必要库。定义了应用名称、用户 ID 和会话 ID 的常量。创建了一个名为"basic_search_agent"的 Agent 实例，并附带描述和指示其目的的指令。它被配置为使用 Google Search tool，这是 ADK 提供的一个预构建 tool。初始化了一个 InMemorySessionService（见第8章）来管理 agent 的会话。为指定的应用、用户和会话 ID 创建了一个新会话。实例化了一个 Runner，将创建的 agent 与会话服务链接起来。此 Runner 负责在会话中执行 agent 的交互。定义了一个辅助函数 call_agent 来简化向 agent 发送查询和处理响应的过程。在 call_agent 内部，用户的查询被格式化为一个角色为 'user' 的 types.Content 对象。使用用户 ID、会话 ID 和新消息内容调用 runner.run 方法。runner.run 方法返回一个表示 agent 行动和响应的事件列表。代码遍历这些事件以查找最终响应。如果某个事件被标识为最终响应，则提取该响应的文本内容。提取的 agent 响应随后被打印到控制台。最后，使用查询"what's the latest ai news?"调用 call_agent 函数来演示 agent 的实际运行。

代码执行：Google ADK 包含用于专门任务的集成组件，包括一个用于动态代码执行的环境。built_in_code_execution tool 为 agent 提供了一个沙盒化的 Python 解释器。这允许模型编写和运行代码来执行计算任务、操作数据结构和执行程序脚本。此类功能对于解决需要确定性逻辑和精确计算的问题至关重要，这些问题超出了概率性语言生成的范围。

python

import os, getpass
import asyncio
import nest_asyncio
from typing import List
from dotenv import load_dotenv
import logging
from google.adk.agents import Agent as ADKAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.adk.code_executors import BuiltInCodeExecutor
from google.genai import types

# Define variables required for Session setup and Agent execution
APP_NAME="calculator"
USER_ID="user1234"
SESSION_ID="session_code_exec_async"

# Agent Definition
code_agent = LlmAgent(
    name="calculator_agent",
    model="gemini-2.0-flash",
    code_executor=BuiltInCodeExecutor(),
    instruction="""You are a calculator agent.
    When given a mathematical expression, write and execute Python code to calculate the result.
    Return only the final numerical result as plain text, without markdown or code blocks.
    """,
    description="Executes Python code to perform calculations.",
)

# Agent Interaction (Async)
async def call_agent_async(query):
    # 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=code_agent, app_name=APP_NAME, session_service=session_service)

    content = types.Content(role='user', parts=[types.Part(text=query)])
    print(f"\n--- Running Query: {query} ---")
    final_response_text = "No final text response captured."

    try:
        # Use run_async
        async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
            print(f"Event ID: {event.id}, Author: {event.author}")
            # --- Check for specific parts FIRST ---
            # has_specific_part = False
            if event.content and event.content.parts and event.is_final_response():
                for part in event.content.parts: # Iterate through all parts
                    if part.executable_code:
                        # Access the actual code string via .code
                        print(f" Debug: Agent generated code:\n```python\n{part.executable_code.code}\n```")
                        has_specific_part = True
                    elif part.code_execution_result:
                        # Access outcome and output correctly
                        print(f" Debug: Code Execution Result: {part.code_execution_result.outcome} - Output:\n{part.code_execution_result.output}")
                        has_specific_part = True
                    # Also print any text parts found in any event for debugging
                    elif part.text and not part.text.isspace():
                        print(f" Text: '{part.text.strip()}'")
                        # Do not set has_specific_part=True here, as we want the final response logic below

                # --- Check for final response AFTER specific parts ---
                text_parts = [part.text for part in event.content.parts if part.text]
                final_result = "".join(text_parts)
                print(f"==> Final Agent Response: {final_result}")
    except Exception as e:
        print(f"ERROR during agent run: {e}")

    print("-" * 30)

# Main async function to run the examples
async def main():
    await call_agent_async("Calculate the value of (5 + 7) * 3")
    await call_agent_async("What is 10 factorial?")

# Execute the main async function
try:
    nest_asyncio.apply()
    asyncio.run(main())
except RuntimeError as e:
    # Handle specific error when running asyncio.run in an already running loop (like Jupyter/Colab)
    if "cannot be called from a running event loop" in str(e):
        print("\nRunning in an existing event loop (like Colab/Jupyter).")
        print("Please run `await main()` in a notebook cell instead.")
        # If in an interactive environment like a notebook, you might need to run:
        # await main()
    else:
        raise e # Re-raise other runtime errors

该脚本使用 Google 的 Agent Development Kit (ADK) 创建一个 agent，通过编写和执行 Python 代码来解决数学问题。它定义了一个 LlmAgent，专门被指示充当计算器，并配备了 built_in_code_execution tool。主要逻辑位于 call_agent_async 函数中，该函数将用户的查询发送到 agent 的 runner 并处理生成的事件。在此函数内部，一个异步循环遍历事件，打印生成的 Python 代码及其执行结果以供调试。代码仔细区分了这些中间步骤和包含数字答案的最终事件。最后，一个 main 函数使用两个不同的数学表达式运行 agent，以演示其执行计算的能力。

企业搜索：这段代码使用 Python 中的 google.adk 库定义了一个 Google ADK 应用程序。它专门使用了一个 VSearchAgent，该 agent 被设计为通过搜索指定的 Vertex AI Search 数据存储来回答问题。代码初始化了一个名为"q2_strategy_vsearch_agent"的 VSearchAgent，提供了描述、要使用的模型（"gemini-2.0-flash-exp"）以及 Vertex AI Search 数据存储的 ID。DATASTORE_ID 预期设置为环境变量。然后为 agent 设置了一个 Runner，使用 InMemorySessionService 来管理对话历史。定义了一个异步函数 call_vsearch_agent_async 来与 agent 交互。该函数接受查询，构造消息内容对象，并调用 runner 的 run_async 方法将查询发送给 agent。该函数随后将 agent 的响应以流式方式回传到控制台。它还打印关于最终响应的信息，包括来自数据存储的任何来源归属。包含了错误处理以捕获 agent 执行期间的异常，提供关于潜在问题（如不正确的数据存储 ID 或缺少权限）的信息性消息。提供了另一个异步函数 run_vsearch_example 来演示如何使用示例查询调用 agent。主执行块检查 DATASTORE_ID 是否已设置，然后使用 asyncio.run 运行示例。它包含一个检查来处理代码在已有运行事件循环的环境（如 Jupyter notebook）中运行的情况。

python

import asyncio
from google.genai import types
from google.adk import agents
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
import os

# --- Configuration ---
# Ensure you have set your GOOGLE_API_KEY and DATASTORE_ID environment variables
# For example:
# os.environ["GOOGLE_API_KEY"] = "YOUR_API_KEY"
# os.environ["DATASTORE_ID"] = "YOUR_DATASTORE_ID"
DATASTORE_ID = os.environ.get("DATASTORE_ID")

# --- Application Constants ---
APP_NAME = "vsearch_app"
USER_ID = "user_123" # Example User ID
SESSION_ID = "session_456" # Example Session ID

# --- Agent Definition (Updated with the newer model from the guide) ---
vsearch_agent = agents.VSearchAgent(
    name="q2_strategy_vsearch_agent",
    description="Answers questions about Q2 strategy documents using Vertex AI Search.",
    model="gemini-2.0-flash-exp", # Updated model based on the guide's examples
    datastore_id=DATASTORE_ID,
    model_parameters={"temperature": 0.0}
)

# --- Runner and Session Initialization ---
runner = Runner(
    agent=vsearch_agent,
    app_name=APP_NAME,
    session_service=InMemorySessionService(),
)

# --- Agent Invocation Logic ---
async def call_vsearch_agent_async(query: str):
    """Initializes a session and streams the agent's response."""
    print(f"User: {query}")
    print("Agent: ", end="", flush=True)
    try:
        # Construct the message content correctly
        content = types.Content(role='user', parts=[types.Part(text=query)])

        # Process events as they arrive from the asynchronous runner
        async for event in runner.run_async(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
        ):
            # For token-by-token streaming of the response text
            if hasattr(event, 'content_part_delta') and event.content_part_delta:
                print(event.content_part_delta.text, end="", flush=True)

            # Process the final response and its associated metadata
            if event.is_final_response():
                print() # Newline after the streaming response
                if event.grounding_metadata:
                    print(f" (Source Attributions: {len(event.grounding_metadata.grounding_attributions)} sources found)")
                else:
                    print(" (No grounding metadata found)")
                print("-" * 30)
    except Exception as e:
        print(f"\nAn error occurred: {e}")
        print("Please ensure your datastore ID is correct and that the service account has the necessary permissions.")
        print("-" * 30)

# --- Run Example ---
async def run_vsearch_example():
    # Replace with a question relevant to YOUR datastore content
    await call_vsearch_agent_async("Summarize the main points about the Q2 strategy document.")
    await call_vsearch_agent_async("What safety procedures are mentioned for lab X?")

# --- Execution ---
if __name__ == "__main__":
    if not DATASTORE_ID:
        print("Error: DATASTORE_ID environment variable is not set.")
    else:
        try:
            asyncio.run(run_vsearch_example())
        except RuntimeError as e:
            # This handles cases where asyncio.run is called in an environment
            # that already has a running event loop (like a Jupyter notebook).
            if "cannot be called from a running event loop" in str(e):
                print("Skipping execution in a running event loop. Please run this script directly.")
            else:
                raise e

总体而言，这段代码提供了一个基本框架，用于构建利用 Vertex AI Search 根据数据存储中存储的信息回答问题的对话式 AI 应用程序。它演示了如何定义 agent、设置 runner，以及如何以异步方式与 agent 交互同时流式传输响应。重点在于从特定数据存储中检索和综合信息以回答用户查询。

Vertex Extensions：Vertex AI 扩展是一个结构化的 API 包装器，使模型能够连接外部 API 进行实时数据处理和行动执行。Extensions 提供企业级的安全性、数据隐私和性能保证。它们可用于生成和运行代码、查询网站以及分析来自私有数据存储的信息等任务。Google 为常见用例（如 Code Interpreter 和 Vertex AI Search）提供了预构建的扩展，并可选择创建自定义扩展。Extensions 的主要优势包括强大的企业控制以及与其他 Google 产品的无缝集成。Extensions 和 Function Calling 之间的关键区别在于它们的执行方式：Vertex AI 自动执行 Extensions，而 Function Call 需要用户或客户端手动执行。

一览

是什么：LLM 是强大的文本生成器，但它们从根本上与外部世界是断开的。它们的知识是静态的，仅限于训练数据，并且缺乏执行操作或检索实时信息的能力。这种固有的限制使它们无法完成需要与外部 API、数据库或服务交互的任务。如果没有与这些外部系统的桥梁，它们解决现实世界问题的实用性就会受到严重制约。

为什么：Tool Use 模式（通常通过 Function Calling 实现）为这个问题提供了标准化的解决方案。它的工作方式是将可用的外部函数或"tool"以 LLM 可以理解的方式描述给它。基于用户的请求，agentic LLM 可以判断是否需要 tool，并生成一个结构化数据对象（如 JSON），指定要调用哪个函数以及使用什么参数。编排层执行此函数调用，检索结果，并将其反馈给 LLM。这使 LLM 能够将最新的外部信息或操作的结果整合到其最终响应中，有效地赋予其行动能力。

经验法则：当 agent 需要突破 LLM 的内部知识并与外部世界交互时，使用 Tool Use 模式。这对于需要实时数据（如检查天气、股票价格）、访问私有或专有信息（如查询公司数据库）、执行精确计算、执行代码或在其他系统中触发操作（如发送电子邮件、控制智能设备）的任务至关重要。

视觉总结

Tool Use 设计模式

关键要点

Tool Use (Function Calling) 允许 agent 与外部系统交互并访问动态信息。
它涉及定义具有 LLM 可以理解的清晰描述和参数的 tool。
LLM 决定何时使用 tool 并生成结构化的函数调用。
Agentic 框架执行实际的 tool 调用并将结果返回给 LLM。
Tool Use 对于构建能够执行真实世界操作和提供最新信息的 agent 至关重要。
LangChain 使用 @tool 装饰器简化了 tool 定义，并提供了 create_tool_calling_agent 和 AgentExecutor 来构建使用 tool 的 agent。
Google ADK 有许多非常有用的预构建 tool，如 Google Search、Code Execution 和 Vertex AI Search Tool。

结论

Tool Use 模式是扩展大语言模型功能范围超越其固有的文本生成能力的关键架构原则。通过赋予模型与外部软件和数据源交互的能力，这种范式使 agent 能够执行操作、执行计算以及从其他系统检索信息。此过程涉及模型在确定有必要满足用户查询时生成结构化请求以调用外部 tool。LangChain、Google ADK 和 Crew AI 等框架提供了结构化的抽象和组件，以促进这些外部 tool 的集成。这些框架管理着将 tool 规范暴露给模型以及解析其后续 tool 使用请求的过程。这简化了能够在外部数字环境中进行交互和采取行动的复杂 agentic 系统的开发。

参考资料

LangChain Documentation (Tools): https://python.langchain.com/docs/integrations/tools/
Google Agent Developer Kit (ADK) Documentation (Tools): https://google.github.io/adk-docs/tools/
OpenAI Function Calling Documentation: https://platform.openai.com/docs/guides/function-calling
CrewAI Documentation (Tools): https://docs.crewai.com/concepts/tools