趋近智
构建一个LangChain代理需要为其配备自定义工具,这些工具可以与外部数据源交互或执行特定的计算。这种方法模拟了常见的生产场景,在这些场景中,代理需要超出LLM自身知识的专业能力。
我们的目标是创建一个代理,它能够回答有关当前天气的问题并估算地点间的驾驶时间。这需要通过自定义工具,让代理能够使用两种不同的功能。
设想您需要一个助手,能够回答以下问题:
为处理这些请求,代理需要:
我们将把这些功能作为自定义LangChain工具实现,并将其整合到一个工具调用代理中。
首先,我们创建一个获取天气数据的工具。对于实际应用,您可能会使用OpenWeatherMap、WeatherAPI或类似的服务提供商。这通常需要获取一个API密钥。为简化本实践部分,我们将定义一个函数来返回模拟天气数据。不过,我们将以调用真实API的方式来构建它。
# 前提条件:请确保已安装langchain、langchain-openai和langchain-core
# pip install langchain langchain-openai langchain-core python-dotenv
import os
import random
from dotenv import load_dotenv
from langchain_core.tools import BaseTool, Tool
from typing import Type, Optional
from pydantic import BaseModel, Field
# 加载环境变量(可选,用于真实API的API密钥)
load_dotenv()
# --- 天气工具实现 ---
class WeatherInput(BaseModel):
"""天气工具的输入模式。"""
location: str = Field(description="需要获取天气的城市名称。")
def get_current_weather(location: str) -> str:
"""
模拟获取某个地点的当前天气。
在实际应用中,这将调用外部天气API。
"""
print(f"---> Calling Weather Tool for: {location}")
# 模拟API调用
try:
# 模拟数据生成
temp_celsius = random.uniform(5.0, 35.0)
conditions = random.choice(["晴朗", "多云", "有雨", "有风", "下雪(不太可能!)"])
humidity = random.randint(30, 90)
return f"The current weather in {location} is {temp_celsius:.1f}°C, {conditions}, with {humidity}% humidity."
except Exception as e:
return f"Error fetching weather for {location}: {e}"
# 选项1:使用Tool装饰器(适用于基本功能,更简单)
# from langchain_core.tools import tool
# @tool("weather_checker", args_schema=WeatherInput)
# def weather_tool(location: str) -> str:
# """用于查找特定城市当前天气状况的工具。"""
# return get_current_weather(location)
# 选项2:继承BaseTool类(控制力更强,更适合复杂逻辑/状态)
class WeatherTool(BaseTool):
name: str = "天气查询器"
description: str = "用于查找特定城市当前天气状况的工具。输入应为城市名称。"
args_schema: Type[BaseModel] = WeatherInput
def _run(self, location: str) -> str:
"""使用该工具。"""
return get_current_weather(location)
async def _arun(self, location: str) -> str:
"""异步使用该工具。"""
# 对于这个简单的模拟函数,异步并非严格必要,
# 但它演示了真实异步API调用的模式。
# 在实际场景中,您会使用异步HTTP客户端(例如aiohttp)。
return self._run(location) # 模拟异步调用
weather_tool = WeatherTool()
# 直接测试工具(可选)
# print(weather_tool.invoke({"location": "London"}))
# print(weather_tool.invoke("Paris")) # 如果args_schema允许,也接受直接字符串输入
关于此工具的要点:
WeatherInput 模式: 我们定义了一个Pydantic模型WeatherInput来指定预期输入(location)。这有助于LangChain验证输入,并为LLM工具调用API提供结构。get_current_weather 函数: 这是核心逻辑。它目前使用随机数据,但模仿了API调用处理程序的结构,包括基本的错误处理。print语句有助于跟踪工具的执行。WeatherTool 类: 我们从langchain_core继承BaseTool以进行明确控制。
name:工具的简洁标识符。description:对代理来说非常必要。LLM使用此描述来决定何时使用该工具以及提供什么输入。务必使其清晰且信息丰富。args_schema:链接到我们的Pydantic输入模型。_run:同步执行方法。_arun:异步执行方法。接下来,我们需要一个工具来估算驾驶时间。同样,实现中可能会使用Google Maps Distance Matrix或OSRM等API。我们将通过一个简单的计算来模拟它。
# --- 驾驶时间工具实现 ---
class DrivingTimeInput(BaseModel):
"""驾驶时间工具的输入模式。"""
origin: str = Field(description="起始城市或地点。")
destination: str = Field(description="目的城市或地点。")
def estimate_driving_time(origin: str, destination: str) -> str:
"""
模拟估算两个地点之间的驾驶时间。
为简化起见,假设一个固定的平均速度。
"""
print(f"---> Calling Driving Time Tool for: {origin} to {destination}")
# 基于城市名称长度的极简距离模拟
# (在生产环境中请替换为真实的距离计算或API调用)
simulated_distance_km = abs(len(origin) - len(destination)) * 50 + random.randint(50, 500)
average_speed_kph = 80
if simulated_distance_km == 0: # 避免起点/终点相同时除以零
return f"Origin and destination ({origin}) are the same."
time_hours = simulated_distance_km / average_speed_kph
hours = int(time_hours)
minutes = int((time_hours - hours) * 60)
return f"The estimated driving time from {origin} to {destination} is approximately {hours} hours and {minutes} minutes ({simulated_distance_km} km)."
class DrivingTimeTool(BaseTool):
name: str = "驾驶时间估算器"
description: str = ("用于估算两个城市之间驾驶时间的工具。"
"输入应为起始城市和目的城市。")
args_schema: Type[BaseModel] = DrivingTimeInput
def _run(self, origin: str, destination: str) -> str:
"""使用该工具。"""
return estimate_driving_time(origin, destination)
async def _arun(self, origin: str, destination: str) -> str:
"""异步使用该工具。"""
# 模拟异步调用以作演示
return self._run(origin, destination)
driving_tool = DrivingTimeTool()
# 直接测试工具(可选)
# print(driving_tool.invoke({"origin": "Paris", "destination": "Berlin"}))
此工具遵循与天气工具相同的模式:一个输入模式(DrivingTimeInput)、一个核心逻辑函数(estimate_driving_time)和一个BaseTool子类(DrivingTimeTool)。
既然我们有了自定义工具,现在将它们整合到一个代理中。我们将使用工具调用代理。这是GPT-3.5和GPT-4等模型的现代标准,它采用模型原生的API进行函数调用,而不是依赖脆弱的文本解析(如旧的ReAct模式)。
# --- 代理设置 ---
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import create_tool_calling_agent, AgentExecutor
# 确保在您的环境变量或.env文件中设置了OPENAI_API_KEY
# os.environ["OPENAI_API_KEY"] = "your_api_key"
if not os.getenv("OPENAI_API_KEY"):
print("警告:OPENAI_API_KEY未设置。代理执行很可能会失败。")
# 1. 初始化LLM
# 工具调用需要支持此功能的聊天模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 2. 定义工具列表
tools = [weather_tool, driving_tool]
# 3. 获取提示模板
# 获取一个为工具调用代理优化的预定义提示
# 您可以在LangChain Hub上查看其他提示
prompt = hub.pull("hwchase17/openai-tools-agent")
# 4. 创建工具调用代理
# 这将LLM、工具和提示绑定在一起,借助OpenAI工具API。
agent = create_tool_calling_agent(llm, tools, prompt)
# 5. 创建代理执行器
# 这将运行代理循环
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 设置为True以查看代理的工具使用情况
max_iterations=5 # 防止潜在的无限循环
)
print("代理执行器创建成功。")
让我们分析一下代理的创建过程:
ChatOpenAI。温度设置为0,以便为工具使用提供更确定的响应。weather_tool和driving_tool实例收集到一个列表中。hwchase17/openai-tools-agent。此提示专为处理工具调用模型所需的系统指令而设计。create_tool_calling_agent: 此函数构建代理逻辑。与需要复杂文本解析指令的传统代理不同,此代理在内部使用bind_tools方法将我们的工具定义直接附加到API调用中。AgentExecutor: 这是代理的运行时环境。它管理循环:将输入发送到LLM,执行LLM请求的工具,并将输出反馈给LLM。代理执行器准备就绪后,我们用不同的查询来测试它。
# --- 运行代理 ---
print("\n--- 运行简单天气查询 ---")
response1 = agent_executor.invoke({
"input": "What's the weather like right now in Toronto?"
})
print("\n最终答案:", response1['output'])
print("\n--- 运行简单驾驶时间查询 ---")
response2 = agent_executor.invoke({
"input": "How long does it take to drive from Berlin to Munich?"
})
print("\n最终答案:", response2['output'])
print("\n--- 运行多工具查询 ---")
response3 = agent_executor.invoke({
"input": "Can you tell me the current weather in Rome and also how long it might take to drive there from Naples?"
})
print("\n最终答案:", response3['output'])
# 代理可以直接回答的查询示例(如果可能)
# print("\n--- 运行非工具查询 ---")
# response4 = agent_executor.invoke({
# "input": "What is the capital of France?"
# })
# print("\n最终答案:", response4['output'])
当verbose=True时观察输出。您将看到与旧ReAct代理不同的模式:
Invoking: Weather Checker with {'location': 'Toronto'}。weather_tool的输出并进行记录。请密切注意代理如何使用BaseTool子类中定义的精确名称和输入模式。工具描述的质量对代理选择正确工具的能力非常必要。
本实践练习演示了创建具有自定义能力的LangChain代理的基本工作流程:
BaseTool或@tool装饰器为每种能力创建函数或类。请特别注意name、description和args_schema。create_tool_calling_agent,以确保可靠的工具使用,避免解析错误。AgentExecutor来管理运行时循环。verbose=True来验证代理是否选择了正确的工具。从这里,您可以进一步查看:
AgentExecutor迁移到LangGraph,它提供了一个更易于控制的基于图的执行环境。_arun方法)来优化性能。简洁的语法。内置调试功能。从第一天起就可投入生产。
为 ApX 背后的 AI 系统而构建
这部分内容有帮助吗?
© 2026 ApX Machine LearningAI伦理与透明度•