构建一个LangChain代理需要为其配备自定义工具,这些工具可以与外部数据源交互或执行特定的计算。这种方法模拟了常见的生产场景,在这些场景中,代理需要超出LLM自身知识的专业能力。我们的目标是创建一个代理,它能够回答有关当前天气的问题并估算地点间的驾驶时间。这需要通过自定义工具,让代理能够使用两种不同的功能。场景定义设想您需要一个助手,能够回答以下问题:“东京目前的温度是多少?”“从旧金山开车到洛杉矶,实际需要多长时间?”“告诉我巴黎的天气预报,并估算从巴黎到里昂的驾驶时间。”为处理这些请求,代理需要:一个工具,用于获取指定城市的当前天气信息。一个工具,用于估算两个城市之间的驾驶时间。我们将把这些功能作为自定义LangChain工具实现,并将其整合到一个工具调用代理中。步骤1:实现天气工具首先,我们创建一个获取天气数据的工具。对于实际应用,您可能会使用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:异步执行方法。步骤2:实现驾驶时间工具接下来,我们需要一个工具来估算驾驶时间。同样,实现中可能会使用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)。步骤3:创建和配置代理既然我们有了自定义工具,现在将它们整合到一个代理中。我们将使用工具调用代理。这是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("代理执行器创建成功。")让我们分析一下代理的创建过程:LLM 初始化: 我们实例化ChatOpenAI。温度设置为0,以便为工具使用提供更确定的响应。工具列表: 我们将自定义的weather_tool和driving_tool实例收集到一个列表中。提示模板: 我们从LangChain Hub获取hwchase17/openai-tools-agent。此提示专为处理工具调用模型所需的系统指令而设计。create_tool_calling_agent: 此函数构建代理逻辑。与需要复杂文本解析指令的传统代理不同,此代理在内部使用bind_tools方法将我们的工具定义直接附加到API调用中。AgentExecutor: 这是代理的运行时环境。它管理循环:将输入发送到LLM,执行LLM请求的工具,并将输出反馈给LLM。步骤4:运行代理并观察行为代理执行器准备就绪后,我们用不同的查询来测试它。# --- 运行代理 --- 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的输出并进行记录。(必要时重复): 对于多工具查询,代理可能会在收到天气数据后立即调用驾驶工具。最终答案: LLM将工具输出综合成自然语言响应。请密切注意代理如何使用BaseTool子类中定义的精确名称和输入模式。工具描述的质量对代理选择正确工具的能力非常必要。总结与后续步骤本实践练习演示了创建具有自定义能力的LangChain代理的基本工作流程:明确需求: 确定代理必须执行的特定任务。实现工具: 使用BaseTool或@tool装饰器为每种能力创建函数或类。请特别注意name、description和args_schema。配置代理: 为现代LLM使用create_tool_calling_agent,以确保可靠的工具使用,避免解析错误。实例化执行器: 创建AgentExecutor来管理运行时循环。测试和观察: 运行查询并使用verbose=True来验证代理是否选择了正确的工具。从这里,您可以进一步查看:更复杂的工具: 整合与数据库或专有API交互的工具。LangGraph: 对于需要复杂状态管理、循环或人机协作工作流程的生产应用,可以考虑从AgentExecutor迁移到LangGraph,它提供了一个更易于控制的基于图的执行环境。异步执行: 通过确保您的工具和代理执行器为I/O密集型任务有效使用异步操作(_arun方法)来优化性能。