《Agentic Design Patterns：构建智能系统的实战指南》- 第五章工具使用

书籍名称：Agentic Design Patterns: A Hands-On Guide to Building Intelligent Systems
本书作者：Antonio Gulli
链接地址：https://docs.google.com/document/d/1bE4iMljhppqGY1p48gQWtZvk6MfRuJRCiba1yRykGNE
内容摘要：本文是对《智能体设计模式》第五章的翻译。此章节介绍了智能体工具（Tools）和结合工具的智能体设计模式。
翻译：煤矿工厂

概述

到目前为止，我们已经讨论了主要涉及协调语言模型之间的交互以及管理代理内部工作流中信息流的代理模式（链式、路由、并行化、反射）。然而，为了使代理真正有用并与现实世界或外部系统交互，它们需要使用工具的能力。

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

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

工具调用十分重要，因为它打破了大型语言模型训练数据的限制，使其能够访问最新信息、执行其内部无法完成的计算、与用户特定数据交互或触发真实世界的行动。函数调用是弥合大型语言模型推理能力与大量可用外部功能之间鸿沟的技术机制。

虽然“函数调用”恰如其分地描述了调用特定的、预定义的代码函数，但考虑更广阔的“工具调用”概念会很有用。这个更广泛的术语承认代理的能力可以远远超出简单的函数执行。“工具”可以是传统的函数，但它也可以是一个复杂的API端点、一个对数据库的请求，甚至是针对另一个专业代理的指令。这种视角使我们能够设想更复杂的系统，例如，一个主代理可能会将复杂的“数据分析任务”委托给一个专门的“分析师代理”，或者通过其API查询外部知识库。以“工具调用”来思考更能捕捉代理作为协调者在数字资源和其他智能实体多样化生态系统中发挥作用的全部潜力。

LangChain、LangGraph 和 Google Agent Developer Kit (ADK) 等框架为定义工具并将其集成到代理工作流中提供了强大的支持。通常，利用 Gemini 或 OpenAI 系列等现代大型语言模型的原生函数调用能力。在这些框架的“画布”上，您可以定义工具，然后配置代理（通常是大型语言模型代理）以了解并能够使用这些工具。工具调用是构建强大、交互式和外部感知代理的基石。

典型应用

工具使用模式几乎适用于所有代理需要超越生成文本来执行操作或检索特定动态信息的场景：

从外部源检索信息：访问大型语言模型训练数据中不存在的实时数据或信息。用例：天气代理。工具：一个天气API，它接受一个位置并返回当前天气状况。流程：用户询问“伦敦天气如何？”，大型语言模型识别出需要天气工具，以“伦敦”调用该工具，工具返回数据，大型语言模型将数据格式化为用户友好的响应
与数据库和API交互：对结构化数据执行查询、更新或其他操作。用例：电子商务代理。工具：API调用以检查产品库存、获取订单状态或处理支付。流程：用户询问“产品X有库存吗？”，大型语言模型调用库存API，工具返回库存数量，大型语言模型告诉用户库存状态。
执行计算和数据分析：使用外部计算器、数据分析库或统计工具。用例：金融代理。工具：计算器功能、股票市场数据API、电子表格工具。流程：用户询问“AAPL当前价格是多少？如果我以150美元购买100股，潜在利润是多少？”，大型语言模型调用股票API，获取当前价格，然后调用计算器工具，获得结果，格式化响应。
发送通信：发送电子邮件、消息或调用外部通信服务的API。用例：个人助理代理。工具：电子邮件发送API。流程：用户说“请给约翰发一封关于明天会议的电子邮件。”，大型语言模型使用从请求中提取的收件人、主题和正文调用电子邮件工具。
执行代码：在安全环境中运行代码片段以执行特定任务。用例：编码助理代理。工具：代码解释器。流程：用户提供一个Python代码片段并询问“这段代码是做什么的？”，大型语言模型使用解释器工具运行代码并分析其输出。
控制其他系统或设备：与智能家居设备、物联网平台或其他连接系统交互。用例：智能家居代理。工具：一个控制智能灯的API。流程：用户说“关掉客厅的灯。”大型语言模型使用命令和目标设备调用智能家居工具。

工具使用将语言模型从文本生成器转变为能够在数字或物理世界中感知、推理和行动的智能体（见图 1）。

《Agentic Design Patterns：构建智能系统的实战指南》- 第五章工具使用 — 图1 Agent使用工具的一些示例

代码实例上手

LangChain

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

以下实现将通过首先定义一个模拟信息检索工具的简单函数来演示。在这之后，我们将构建并配置一个代理，使其能够响应用户输入来利用此工具。此示例的执行需要安装核心LangChain库和模型特定的提供程序包。此外，与所选语言模型服务的正确身份验证（通常通过在本地环境中配置API密钥）是必要的先决条件。

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

# 取消注释
# 安全地提示用户并设置API密钥作为环境变量
os.environ["GOOGLE_API_KEY"] = getpass.getpass("请输入您的Google API密钥: ")
os.environ["OPENAI_API_KEY"] = getpass.getpass("请输入您的OpenAI API密钥: ")

try:
# 需要一个具有函数/工具调用功能的模型。
  llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
  print(f"✅ 语言模型已初始化: {llm.model}")
except Exception as e:
  print(f"🛑 初始化语言模型时出错: {e}")
  llm = None

# --- 定义一个工具 ---
@langchain_tool
def search_information(query: str) -> str:
"""
  提供给定主题的事实信息。使用此工具来查找诸如“法国的首都”或“伦敦的天气？”等短语的答案。
  """
  print(f"\n--- 🛠️ 工具已调用: search_information，查询: '{query}' ---")
# 使用预定义结果的字典模拟一个搜索工具。
  simulated_results = {
      "weather in london": "伦敦目前多云，气温15°C。",
      "capital of france": "法国的首都是巴黎。",
      "population of earth": "地球的估计人口约为80亿。",
      "tallest mountain": "珠穆朗玛峰是地球上最高的山峰（海拔）。",
      "default": f"“{query}”的模拟搜索结果: 未找到具体信息，但该主题似乎很有趣。"
  }
  result = simulated_results.get(query.lower(), simulated_results["default"])
  print(f"--- 工具结果: {result} ---")
return result

tools = [search_information]

# --- 创建一个工具调用代理 ---
if llm:
# 此提示模板需要一个`agent_scratchpad`占位符用于代理的内部步骤。
  agent_prompt = ChatPromptTemplate.from_messages([
      ("system", "你是一个乐于助人的助手。"),
      ("human", "{input}"),
      ("placeholder", "{agent_scratchpad}"),
  ])

# 创建代理，将LLM、工具和提示绑定在一起。
  agent = create_tool_calling_agent(llm, tools, agent_prompt)

# AgentExecutor 是一个运行时，用于调用代理并执行所选择的工具。
# 'tools'参数在这里不是必需的，因为它们已经绑定到代理。
  agent_executor = AgentExecutor(agent=agent, verbose=True, tools=tools)

asyncdef run_agent_with_tool(query: str):
"""使用查询调用代理执行器并打印最终响应。"""
  print(f"\n--- 🏃 正在运行代理，查询: '{query}' ---")
try:
      response = await agent_executor.ainvoke({"input": query})
      print("\n--- ✅ 代理最终响应 ---")
      print(response["output"])
except Exception as e:
      print(f"\n🛑 代理执行期间发生错误: {e}")

asyncdef main():
"""并发运行所有代理查询。"""
  tasks = [
      run_agent_with_tool("法国的首都是什么？"),
      run_agent_with_tool("伦敦天气怎么样？"),
      run_agent_with_tool("告诉我一些关于狗的事情。") # 应该触发默认工具响应
  ]
await asyncio.gather(*tasks)

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

该代码使用 LangChain 库和 Google Gemini 模型设置了一个工具调用代理。它定义了一个 search_information 工具，该工具模拟为特定查询提供事实性答案。该工具对“伦敦天气”、“法国首都”和“地球人口”有预定义响应，并对其他查询采取默认响应。

ChatGoogleGenerativeAI 模型已初始化，确保其具有工具调用功能。
ChatPromptTemplate 用于指导代理的交互。
create_tool_calling_agent 函数用于将语言模型、工具和提示组合成一个代理。然后设置 AgentExecutor 以管理代理的执行和工具调用。
定义 run_agent_with_tool 异步函数，用于使用给定查询调用代理并打印结果。main 异步函数准备多个查询以并发运行。

这些查询旨在测试 search_information 工具的特定响应和默认响应。最后，asyncio.run(main()) 调用执行所有代理任务。代码包括在继续代理设置和执行之前检查 LLM 初始化是否成功的检查。

CrewAI

此代码提供了一个实用示例，说明如何在 CrewAI 框架中实现函数调用（工具）。它设置了一个简单场景，其中代理配备了一个查找信息的工具。该示例专门演示了如何使用此代理和工具获取模拟股票价格。

# pip install crewai langchain-openai

import os
from crewai import Agent, Task, Crew
from crewai.tools import tool
import logging

# --- 最佳实践：配置日志 ---
# 一个基本的日志设置有助于调试和跟踪团队的执行。
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# --- 设置您的API密钥 ---
# 对于生产环境，建议使用更安全的密钥管理方法，
# 例如在运行时加载环境变量或使用秘密管理器。
#
# 为您选择的LLM提供商设置环境变量（例如，OPENAI_API_KEY）
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
# os.environ["OPENAI_MODEL_NAME"] = "gpt-4o"

# --- 1. 重构的工具：返回干净的数据 ---
# 该工具现在返回原始数据（一个浮点数）或引发标准的Python错误。
# 这使其更具可重用性，并强制代理正确处理结果。
@tool("股票价格查询工具")
def get_stock_price(ticker: str) -> float:
   """
   获取给定股票代码的最新模拟股票价格。
   以浮点数形式返回价格。如果未找到股票代码，则引发ValueError。
   """
   logging.info(f"工具调用: get_stock_price，股票代码 '{ticker}'")
   simulated_prices = {
       "AAPL": 178.15,
       "GOOGL": 1750.30,
       "MSFT": 425.50,
   }
   price = simulated_prices.get(ticker.upper())

   if price isnotNone:
       return price
   else:
       # 引发特定错误比返回字符串更好。
       # 代理能够处理异常并决定下一步操作。
       raise ValueError(f"未找到股票代码 '{ticker.upper()}' 的模拟价格。")

# --- 2. 定义代理 ---
# 代理的定义保持不变，但它现在将利用改进后的工具。
financial_analyst_agent = Agent(
 role='高级金融分析师',
 goal='使用提供的工具分析股票数据并报告关键价格。',
 backstory="你是一位经验丰富的金融分析师，擅长使用数据源查找股票信息。你提供清晰、直接的答案。",
 verbose=True,
 tools=[get_stock_price],
# 允许委托可能很有用，但对于这个简单的任务来说不是必需的。
 allow_delegation=False,
)

# --- 3. 改进的任务：更清晰的说明和错误处理 ---
# 任务描述更具体，并指导代理如何在成功检索数据和潜在错误时做出反应。
analyze_aapl_task = Task(
 description=(
     "苹果（股票代码：AAPL）当前的模拟股票价格是多少？"
     "使用“股票价格查询工具”查找。 "
     "如果未找到股票代码，您必须报告未能检索到价格。"
 ),
 expected_output=(
     "一个清晰的句子，说明AAPL的模拟股票价格。 "
     "例如：“AAPL的模拟股票价格为$178.15。” "
     "如果无法找到价格，请明确说明。"
 ),
 agent=financial_analyst_agent,
)

# --- 4. 组建团队 ---
# 团队协调代理和任务如何协同工作。
financial_crew = Crew(
 agents=[financial_analyst_agent],
 tasks=[analyze_aapl_task],
 verbose=True# 在生产环境中设置为False以减少日志详细程度
)

# --- 5. 在主执行块中运行团队 ---
# 使用__name__ == "__main__": 块是标准的Python最佳实践。
def main():
   """运行团队的主函数。"""
   # 在启动前检查API密钥，以避免运行时错误。
   ifnot os.environ.get("OPENAI_API_KEY"):
       print("错误：未设置OPENAI_API_KEY环境变量。")
       print("请在运行脚本前进行设置。")
       return

   print("\n## 正在启动金融团队...")
   print("---------------------------------")

   # kickoff 方法开始执行。
   result = financial_crew.kickoff()

   print("\n---------------------------------")
   print("## 团队执行完毕。")
   print("\n最终结果:\n", result)

if __name__ == "__main__":
   main()

此代码演示了一个使用 Crew.ai 库模拟财务分析任务的简单应用程序。它定义了一个自定义工具 get_stock_price，该工具模拟查找预定义股票代码的股价。该工具旨在为有效股票代码返回浮点数，或为无效股票代码引发 ValueError。创建了一个名为 financial_analyst_agent 的 Crew.ai 代理，其角色是高级财务分析师。该代理被赋予 get_stock_price工具进行交互。定义了一个任务 analyze_aapl_task，专门指示代理使用该工具查找 AAPL 的模拟股价。任务描述包含了在使用工具时如何处理成功和失败情况的清晰说明。组装了一个 Crew，包括 financial_analyst_agent 和 analyze_aapl_task。为代理和 Crew 都启用了详细设置，以便在执行期间提供详细日志。脚本的主要部分在标准的 if __name__ == "__main__": 块中，使用 kickoff() 方法运行 Crew 的任务。在启动 Crew 之前，它会检查是否设置了 OPENAI_API_KEY 环境变量，这是代理正常运行所必需的。然后将 Crew 执行的结果（即任务的输出）打印到控制台。该代码还包括基本的日志配置，以便更好地跟踪 Crew 的操作和工具调用。它使用环境变量进行 API 密钥管理，尽管它指出生产环境建议使用更安全的方法。简而言之，核心逻辑展示了如何定义工具、代理和任务，以在 Crew.ai 中创建协作工作流。

Google ADK

Google Agent Developer Kit (ADK) 包含一个原生集成工具库，可以直接整合到代理的功能中。

Google 搜索：此类组件的一个主要示例是 Google 搜索工具。该工具充当 Google 搜索引擎的直接接口，使代理能够执行网络搜索和检索外部信息。

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
# 定义会话设置和代理执行所需的变量
APP_NAME="Google Search_agent"
USER_ID="user1234"
SESSION_ID="1234"
# 定义具有搜索工具访问权限的代理
root_agent = Agent( # 修正了这里，ADKAgent 应该为 Agent
  name="basic_search_agent",
  model="gemini-2.0-flash-exp",
  description="使用 Google 搜索回答问题的代理。",
  instruction="我可以通过搜索互联网来回答你的问题。尽管问我任何问题！",
  tools=[google_search] # Google Search 是一个预构建的工具，用于执行 Google 搜索。
)
# 代理交互
asyncdef call_agent(query):
"""
  调用代理并进行查询的辅助函数。
  """
# 会话和运行器
  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("代理响应: ", final_response)
nest_asyncio.apply()
asyncio.run(call_agent("最新的AI新闻是什么？"))

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

代码执行：Google ADK 具有用于专业任务的集成组件，包括用于动态代码执行的环境。内置的 code_execution 工具为代理提供了一个沙盒 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

# 定义会话设置和代理执行所需的变量
APP_NAME="calculator"
USER_ID="user1234"
SESSION_ID="session_code_exec_async"

# 代理定义
code_agent = LlmAgent(
  name="calculator_agent",
  model="gemini-2.0-flash",
  code_executor=BuiltInCodeExecutor(),
  instruction="""你是一个计算器代理。
  当给定一个数学表达式时，编写并执行 Python 代码来计算结果。
  只以纯文本形式返回最终的数值结果，不要使用 Markdown 或代码块。
  """,
  description="执行 Python 代码来执行计算。",
)

# 代理交互 (异步)
asyncdef call_agent_async(query):

# 会话和运行器
  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--- 正在运行查询: {query} ---")
  final_response_text = "未捕获到最终文本响应。"
try:
      # 使用 run_async
      asyncfor event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
          print(f"事件 ID: {event.id}, 作者: {event.author}")

          # --- 首先检查特定部分 ---
          # has_specific_part = False
          if event.content and event.content.parts and event.is_final_response():
              for part in event.content.parts: # 遍历所有部分
                  if part.executable_code:
                      # 通过 .code 访问实际代码字符串
                      print(f"  调试: 代理生成的代码:\n```python\n{part.executable_code.code}\n```")
                      # has_specific_part = True # 此行被注释掉，保持原样
                  elif part.code_execution_result:
                      # 正确访问结果和输出
                      print(f"  调试: 代码执行结果: {part.code_execution_result.outcome} - 输出:\n{part.code_execution_result.output}")
                      # has_specific_part = True # 此行被注释掉，保持原样
                  # 也打印在任何事件中找到的任何文本部分以进行调试
                  elif part.text andnot part.text.isspace():
                      print(f"  文本: '{part.text.strip()}'")
                      # 不要在此处设置 has_specific_part=True，因为我们希望下面的最终响应逻辑生效

              # --- 在特定部分之后检查最终响应 ---
              text_parts = [part.text for part in event.content.parts if part.text]
              final_result = "".join(text_parts)
              print(f"==> 代理最终响应: {final_result}")

except Exception as e:
      print(f"代理运行期间发生错误: {e}")
  print("-" * 30)

# 运行示例的主异步函数
asyncdef main():
await call_agent_async("计算 (5 + 7) * 3 的值")
await call_agent_async("10的阶乘是多少？")

# 执行主异步函数
try:
  nest_asyncio.apply()
  asyncio.run(main())
except RuntimeError as e:
# 处理在已运行的循环中（如 Jupyter/Colab）运行 asyncio.run 时的特定错误
if"cannot be called from a running event loop"in str(e):
      print("\n正在现有事件循环中运行 (如 Colab/Jupyter)。")
      print("请改为在Colab/Jupyter单元格中运行 `await main()`。")
      # 如果在像Colab/Jupyter这样的交互式环境中，您可能需要运行：
      # await main()
else:
      raise e # 重新抛出其他运行时错误

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

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

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

# --- 配置 ---
# 确保您已设置 GOOGLE_API_KEY 和 DATASTORE_ID 环境变量
# 例如：
# os.environ["GOOGLE_API_KEY"] = "您的API密钥"
# os.environ["DATASTORE_ID"] = "您的数据存储ID"

DATASTORE_ID = os.environ.get("DATASTORE_ID")

# --- 应用程序常量 ---
APP_NAME = "vsearch_app"
USER_ID = "user_123"# 示例用户ID
SESSION_ID = "session_456"# 示例会话ID

# --- 代理定义 (已根据指南中的新模型更新) ---
vsearch_agent = agents.VSearchAgent(
   name="q2_strategy_vsearch_agent",
   description="使用 Vertex AI Search 回答有关第二季度战略文档的问题。",
   model="gemini-2.0-flash-exp", # 根据指南的示例更新的模型
   datastore_id=DATASTORE_ID,
   model_parameters={"temperature": 0.0}
)

# --- 运行器和会话初始化 ---
runner = Runner(
   agent=vsearch_agent,
   app_name=APP_NAME,
   session_service=InMemorySessionService(),
)

# --- 代理调用逻辑 ---
asyncdef call_vsearch_agent_async(query: str):
   """初始化一个会话并流式传输代理的响应。"""
   print(f"用户: {query}")
   print("代理: ", end="", flush=True)

   try:
       # 正确构造消息内容
       content = types.Content(role='user', parts=[types.Part(text=query)])


       # 处理从异步运行器收到的事件
       asyncfor event in runner.run_async(
           user_id=USER_ID,
           session_id=SESSION_ID,
           new_message=content
       ):
           # 对于响应文本的逐令牌流式传输
           if hasattr(event, 'content_part_delta') and event.content_part_delta:
               print(event.content_part_delta.text, end="", flush=True)

           # 处理最终响应及其关联的元数据
           if event.is_final_response():
               print() # 流式响应后的换行
               if event.grounding_metadata:
                   print(f"  (来源归因: 找到 {len(event.grounding_metadata.grounding_attributions)} 个来源)")
               else:
                   print("  (未找到来源归因元数据)")
               print("-" * 30)

   except Exception as e:
       print(f"\n发生错误: {e}")
       print("请确保您的数据存储ID正确，并且服务帐号具有必要的权限。")
       print("-" * 30)

# --- 运行示例 ---
asyncdef run_vsearch_example():
   # 替换为与您的数据存储内容相关的问题
   await call_vsearch_agent_async("总结一下关于第二季度战略文档的主要观点。")
   await call_vsearch_agent_async("X实验室提到了哪些安全程序？")

# --- 执行 ---
if __name__ == "__main__":
   ifnot DATASTORE_ID:
       print("错误: 未设置 DATASTORE_ID 环境变量。")
   else:
       try:
           asyncio.run(run_vsearch_example())
       except RuntimeError as e:
           # 这处理了在已经运行事件循环的环境中（例如 Jupyter notebook）
           # 调用 asyncio.run 的情况。
           if"cannot be called from a running event loop"in str(e):
               print("在运行中的事件循环中跳过执行。请直接运行此脚本。")
           else:
               raise e

本章摘要

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

为什么：工具使用模式，通常通过函数调用实现，为这个问题提供了一个标准化的解决方案。它的工作原理是，以大型语言模型能理解的方式，向其描述可用的外部函数或“工具”。根据用户的请求，具有代理能力的大型语言模型可以决定是否需要工具，并生成一个结构化数据对象（如JSON），指定要调用哪个函数以及使用哪些参数。编排层执行此函数调用，检索结果，并将其反馈给大型语言模型。这使得大型语言模型能够将最新的外部信息或操作结果整合到其最终响应中，从而有效地赋予它行动的能力。

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

要点总结：

工具使用（函数调用）允许代理与外部系统交互并访问动态信息。
它涉及定义具有清晰描述和参数的工具，以便LLM能够理解。
LLM决定何时使用工具并生成结构化的函数调用。
代理框架执行实际的工具调用并将结果返回给LLM。
工具使用对于构建能够执行实际操作和提供最新信息的代理至关重要。
LangChain使用@tool装饰器简化了工具定义，并提供了create_tool_calling_agent和AgentExecutor来构建使用工具的代理。
Google ADK拥有许多非常有用的预构建工具，例如Google搜索、代码执行和Vertex AI搜索工具。

结论

如果要将大型语言模型的功能范围扩展到其固有的文本生成能力之外，那么工具调用是一项非常关键且基础性的设计原则。通过赋予模型与外部软件和数据源交互的能力，这种范式允许代理执行操作、进行计算以及从其他系统检索信息。此过程涉及模型在确定需要调用外部工具以满足用户查询时生成结构化请求。LangChain、Google ADK 和 Crew AI 等框架提供了结构化的抽象和组件，以促进这些外部工具的集成。这些框架管理向模型公开工具规范并解析其后续工具使用请求的过程。这简化了复杂代理系统的开发，这些系统可以与外部数字环境交互并在其中采取行动。

参考文献

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