第5章：工具使用（函数调用）Tool Use (Function Calling)
工具使用模式概述
到目前为止，我们已经讨论了主要涉及语言模型之间交互编排以及管理智能体内部工作流程信息流的智能体模式（链式调用、路由、并行化、反思）。然而，为了使智能体真正有用并能够与现实世界或外部系统交互，它们需要具备使用工具的能力。
工具使用模式通常通过一种称为函数调用的机制来实现，使智能体能够与外部 API、数据库、服务甚至执行代码进行交互。它允许作为智能体核心的 LLM（大型语言模型）根据用户请求或任务当前状态决定何时以及如何使用特定的外部函数。
该过程通常包括以下步骤：
- 工具定义： 向 LLM 定义并描述外部函数或能力。这些描述包括函数的用途、名称、接受的参数及其类型和描述。
- LLM 决策： LLM 接收用户请求和可用的工具定义。基于对请求和工具的理解，LLM 决定是否需要调用一个或多个工具来满足请求。
- 函数调用生成： 如果 LLM 决定使用工具，它会生成一个结构化输出（通常是一个 JSON 对象），指定要调用的工具名称以及从用户请求中提取的参数（参数）传递给工具。
- 工具执行： 智能体框架或编排层拦截该结构化输出，识别请求的工具，并使用提供的参数执行实际的外部函数。
- 观察/结果： 工具执行的输出或结果返回给智能体。
- LLM 处理（可选但常见）： LLM 将工具的输出作为上下文接收，并利用它来制定最终的用户响应或决定工作流程的下一步（可能涉及调用另一个工具、反思或提供最终答案）。
这一模式至关重要，因为它突破了 LLM 的训练数据限制，使其能够访问最新信息、执行内部无法完成的计算、与用户特定数据交互或触发现实世界中的动作。函数调用是连接 LLM 推理能力与外部功能广泛生态系统之间的技术桥梁。
虽然“函数调用”准确描述了调用特定的预定义代码函数，但考虑更广泛的“工具调用”概念是有益的。这一更广泛的术语承认智能体的能力可以远远超出简单的函数执行。“工具”可以是传统的函数，也可以是复杂的 API 端点、数据库请求，甚至是针对其他专用智能体的指令。这种视角使我们能够设想更复杂的系统，例如，一个主智能体可能将复杂的数据分析任务委托给专用的“分析智能体”或通过其 API 查询外部知识库。从“工具调用”的角度思考，更好地捕捉了智能体作为跨多样化数字资源和其他智能实体的编排者的全部潜力。
像 LangChain、LangGraph 和 Google Agent Developer Kit (ADK) 这样的框架提供了强大的支持，用于定义工具并将其集成到智能体工作流程中，通常利用现代 LLM（如 Gemini 或 OpenAI 系列）的原生函数调用能力。在这些框架的“画布”上，您可以定义工具，然后配置智能体（通常是 LLM 智能体）使其能够识别并使用这些工具。
工具使用是构建强大、交互性强且具有外部感知能力智能体的基石。
实际应用与使用场景
工具使用模式适用于几乎任何需要智能体超越文本生成以执行操作或检索特定动态信息的场景：
1. 从外部来源检索信息：
访问实时数据或语言模型（LLM）训练数据中不存在的信息。
- 使用场景： 天气智能体。
- 工具： 一个天气 API，接收位置并返回当前天气状况。
- 智能体流程： 用户询问“伦敦的天气怎么样？”，LLM识别出需要使用天气工具，调用工具并传入“伦敦”，工具返回数据，LLM将数据格式化为用户友好的响应。
2. 与数据库和 API 交互：
对结构化数据执行查询、更新或其他操作。
- 使用场景： 电商智能体。
- 工具： API 调用以检查产品库存、获取订单状态或处理支付。
- 智能体流程： 用户询问“产品X有库存吗？”，LLM调用库存API，工具返回库存数量，LLM告知用户库存状态。
3. 执行计算和数据分析：
使用外部计算器、数据分析库或统计工具。
- 使用场景： 财务智能体。
- 工具： 计算器功能、股票市场数据 API、电子表格工具。
- 智能体流程： 用户询问“当前AAPL的价格是多少？如果我以150美元购买100股，潜在利润是多少？”，LLM调用股票API获取当前价格，然后调用计算器工具计算结果，最后格式化响应。
4. 发送通信：
发送电子邮件、消息或调用外部通信服务的 API。
- 使用场景： 个人助理智能体。
- 工具： 电子邮件发送 API。
- 智能体流程： 用户说“给John发送一封关于明天会议的邮件。”，LLM调用电子邮件工具，并从请求中提取收件人、主题和正文。
5. 执行代码：
在安全环境中运行代码片段以完成特定任务。
- 使用场景： 编码助手智能体。
- 工具： 代码解释器。
- 智能体流程： 用户提供一个Python代码片段并询问“这段代码的功能是什么？”，LLM使用解释器工具运行代码并分析其输出。
6. 控制其他系统或设备：
与智能家居设备、物联网平台或其他连接系统交互。
- 使用场景： 智能家居智能体。
- 工具： 控制智能灯的 API。
- 智能体流程： 用户说“关闭客厅的灯。”，LLM调用智能家居工具并传递命令和目标设备。
工具使用是将语言模型从文本生成器转变为能够在数字或物理世界中感知、推理和行动的智能体的关键（见图1）。
图1：智能体使用工具的一些示例
实操代码示例（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)
async def 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}")
async def 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，用于通过给定查询调用智能体并打印结果。主异步函数准备了多个查询以并发运行。这些查询旨在测试 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"工具调用：获取股票价格，股票代码 '{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:
# 引发特定错误比返回字符串更好。
# 智能体能够处理异常并决定下一步行动。
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密钥，避免运行时错误。
if not 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的模拟股价。任务描述包括了如何处理工具使用中的成功和失败情况的清晰指令。组建了一个团队financial_crew，由financial_analyst_agent和analyze_aapl_task组成。团队和智能体均启用了详细日志记录（verbose设置为True），以便在执行过程中提供详细的日志。
脚本的主要部分通过标准的if __name__ == "__main__":块使用kickoff()方法运行团队任务。在启动团队之前，脚本会检查是否设置了OPENAI_API_KEY环境变量，这是智能体运行所必需的。团队执行的结果，即任务的输出，将打印到控制台。
代码还包含基本的日志配置，以便更好地跟踪团队的动作和工具调用。它使用环境变量进行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 = ADKAgent(
name="basic_search_agent",
model="gemini-2.0-flash-exp",
description="通过使用 Google 搜索回答问题的智能体。",
instruction="我可以通过搜索互联网来回答您的问题。尽管问我任何问题！",
tools=[google_search] # Google Search 是一个用于执行 Google 搜索的预构建工具。
)
## 智能体交互
async def 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("最新的人工智能新闻是什么？"))
此代码演示了如何使用 Google ADK for Python 创建和使用一个基本智能体。该智能体旨在通过利用 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 对象。使用 runner.run 方法调用智能体，传入用户 ID、会话 ID 和新的消息内容。runner.run 方法返回一个事件列表，表示智能体的动作和响应。代码遍历这些事件以找到最终的响应。如果某个事件被识别为最终响应，则提取该响应的文本内容。然后将提取的智能体响应打印到控制台。最后，调用 call_agent 函数并传入查询 "最新的人工智能新闻是什么？"，以演示智能体的运行。
代码执行： Google ADK 提供了用于专门任务的集成组件，包括动态代码执行环境。内置的 built_in_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代码以进行计算。",
)
## 智能体交互（异步）
async def 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
async for 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 and not 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)
## 主异步函数以运行示例
async def main():
await call_agent_async("计算 (5 + 7) * 3 的值")
await call_agent_async("10 的阶乘是多少？")
## 执行主异步函数
try:
nest_asyncio.apply()
asyncio.run(main())
except RuntimeError as e:
# 处理在已运行的事件循环中调用 asyncio.run 时的特定错误（例如 Jupyter/Colab）
if "cannot be called from a running event loop" in str(e):
print("\n正在现有事件循环中运行（例如 Colab/Jupyter）。")
print("请在笔记本单元中运行 `await main()`。")
# 如果在交互式环境（如笔记本）中，可能需要运行：
# await main()
else:
raise e # 重新引发其他运行时错误
此脚本使用 Google 的 Agent Development Kit (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 预计会被设置为环境变量。然后，它为智能体设置了一个 Runner，使用 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"] = "YOUR_API_KEY"
## os.environ["DATASTORE_ID"] = "YOUR_DATASTORE_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 回答关于 Q2 策略文档的问题。",
model="gemini-2.0-flash-exp", # 根据指南示例更新的模型
datastore_id=DATASTORE_ID,
model_parameters={"temperature": 0.0}
)
## --- Runner 和会话初始化 ---
runner = Runner(
agent=vsearch_agent,
app_name=APP_NAME,
session_service=InMemorySessionService(),
)
## --- 智能体调用逻辑 ---
async def call_vsearch_agent_async(query: str):
"""初始化会话并流式传输智能体的响应。"""
print(f"用户: {query}")
print("智能体: ", end="", flush=True)
try:
# 正确构建消息内容
content = types.Content(role='user', parts=[types.Part(text=query)])
# 异步处理来自运行器的事件
async for 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)
## --- 示例运行 ---
async def run_vsearch_example():
# 替换为与您的数据存储内容相关的问题
await call_vsearch_agent_async("总结关于 Q2 战略文件的主要内容。")
await call_vsearch_agent_async("实验室 X 提到的安全程序是什么？")
## --- 执行 ---
if __name__ == "__main__":
if not DATASTORE_ID:
print("错误: DATASTORE_ID 环境变量未设置。")
else:
try:
asyncio.run(run_vsearch_example())
except RuntimeError as e:
# 处理在已经运行事件循环的环境中调用 asyncio.run 的情况
# （例如 Jupyter notebook）。
if "cannot be called from a running event loop" in str(e):
print("在运行事件循环中跳过执行。请直接运行此脚本。")
else:
raise e
总体而言，此代码提供了一个基本框架，用于构建一个对话式 AI 应用程序，该应用程序利用 Vertex AI Search 根据存储在数据存储中的信息回答问题。它展示了如何定义智能体、设置运行器以及异步与智能体交互，同时流式传输响应。重点是从特定的数据存储中检索和综合信息以回答用户查询。
Vertex 扩展: Vertex AI 扩展是一种结构化的 API 包装器，使模型能够连接外部 API 以进行实时数据处理和执行操作。扩展提供企业级安全性、数据隐私和性能保障。它们可用于生成和运行代码、查询网站以及分析来自私有数据存储的信息等任务。Google 为常见用例（如代码解释器和 Vertex AI Search）提供了预构建扩展，同时支持创建自定义扩展。扩展的主要优势包括强大的企业控制和与其他 Google 产品的无缝集成。扩展与函数调用的主要区别在于其执行方式：Vertex AI 自动执行扩展，而函数调用需要用户或客户端手动执行。
概览
定义（What）
大型语言模型（LLMs）是强大的文本生成工具，但它们与外部世界基本上是脱节的。其知识是静态的，仅限于训练数据，并且缺乏执行操作或检索实时信息的能力。这种固有的局限性使得它们无法完成需要与外部 API、数据库或服务交互的任务。如果没有与这些外部系统的桥梁，它们在解决现实世界问题方面的实用性将受到严重限制。
设计意图（Why）
工具使用模式（Tool Use pattern），通常通过函数调用实现，为解决这一问题提供了一种标准化的解决方案。该模式通过向 LLM 描述可用的外部函数或“工具”，以一种它可以理解的方式进行操作。基于用户的请求，具备智能体能力的 LLM可以决定是否需要使用工具，并生成一个结构化的数据对象（如 JSON），指定需要调用的函数及其参数。一个编排层负责执行该函数调用，检索结果，并将其反馈给 LLM。这使得 LLM能够将最新的外部信息或操作结果整合到最终的响应中，从而有效地赋予其执行操作的能力。
使用原则（Rule of Thumb）
每当智能体需要突破 LLM的内部知识并与外部世界交互时，应使用工具使用模式。这对于需要实时数据（例如，天气、股票价格查询）、访问私有或专有信息（例如，查询公司数据库）、执行精确计算、运行代码或触发其他系统中的操作（例如，发送电子邮件、控制智能设备）等任务至关重要。
图解 （Visual Summary）
图2：工具使用设计模式
关键要点
- 工具使用（函数调用）使智能体能够与外部系统交互并访问动态信息。
- 它涉及定义具有清晰描述和参数的工具，使 LLM 可以理解。
- LLM 决定何时使用工具并生成结构化的函数调用。
- 智能体框架执行实际的工具调用并将结果返回给 LLM。
- 工具使用对于构建能够执行现实世界操作并提供最新信息的智能体至关重要。
- LangChain 通过 @tool 装饰器简化工具定义，并提供
create_tool_calling_agent和AgentExecutor用于构建使用工具的智能体。 - Google ADK 提供了一些非常有用的预构建工具，例如 Google 搜索、代码执行和 Vertex AI 搜索工具。
结论
工具使用模式是扩展大型语言模型功能范围的重要架构原则，超越其固有的文本生成能力。通过赋予模型与外部软件和数据源交互的能力，这一范式使智能体能够执行操作、进行计算以及从其他系统中检索信息。该过程涉及模型生成一个结构化请求以调用外部工具，当它判断这样做对于满足用户查询是必要的时。诸如 LangChain、Google ADK 和 Crew AI 等框架提供了结构化的抽象和组件，简化了这些外部工具的集成。这些框架管理向模型公开工具规范的过程，并解析其后续的工具使用请求，从而简化了能够与外部数字环境交互并采取行动的复杂智能体系统的开发。
参考文献
- LangChain 文档 (工具): https://python.langchain.com/docs/integrations/tools/
- Google Agent Developer Kit (ADK) 文档 (工具): https://google.github.io/adk-docs/tools/
- OpenAI 函数调用文档: https://platform.openai.com/docs/guides/function-calling
- CrewAI 文档 (工具): https://docs.crewai.com/concepts/tools
```