构建LangChain智能体涉及为其配备自定义工具,这些工具能够与外部数据源交互或执行特定计算。这种方法模拟了常见的生产应用场景,其中智能体需要LLM固有知识之外的专门能力。我们的目标是创建一个智能体,它能回答有关当前天气的问题并估算地点间的驾驶时长。这要求通过自定义工具给予智能体访问两种不同功能的能力。场景设定设想您需要一个助手,能够回答以下问题:“东京目前的温度是多少?”“从旧金山开车到洛杉矶实际需要多久?”“告诉我巴黎的天气预报,并估算从巴黎到里昂的驾驶时长。”为处理这些请求,智能体需要:获取给定城市当前天气信息的工具。估算两座城市间驾驶时长的工具。我们将把这些功能实现为自定义LangChain工具,并将其集成到ReAct智能体中。步骤1:实现天气工具首先,我们创建一个工具,用于获取天气数据。在真实场景中,您可能会使用像OpenWeatherMap、WeatherAPI或类似的服务提供商。这通常需要获取API密钥。为简化本练习节,我们将定义一个返回模拟天气数据的函数。但我们将按照调用真实API的模式来构建它。# 前提条件:请确保您已安装langchain和langchain_openai # pip install langchain langchain_openai python-dotenv import os import random from dotenv import load_dotenv from langchain.tools import BaseTool, Tool from typing import Type, Optional from pydantic.v1 import BaseModel, Field # Use pydantic v1 for BaseTool compatibility # 加载环境变量(可选,用于使用真实API时设置API密钥) load_dotenv() # --- 天气工具实现 --- class WeatherInput(BaseModel): """天气工具的输入模式。""" location: str = Field(description="需要获取天气的城市名称。") def get_current_weather(location: str) -> str: """ 模拟获取某个地点的当前天气。 在实际应用中,这将调用外部天气API。 """ print(f"---> 调用天气工具获取:{location}") # 模拟API调用 try: # 模拟数据生成 temp_celsius = random.uniform(5.0, 35.0) conditions = random.choice(["晴朗", "多云", "有雨", "刮风", "下雪(不太可能!)"]) humidity = random.randint(30, 90) return f"{location}目前的温度是{temp_celsius:.1f}°C, {conditions}, 湿度为{humidity}%。" except Exception as e: return f"获取{location}天气时出错:{e}" # 选项1:使用Tool装饰器(适用于基本函数,更简单) # weather_tool = Tool.from_function( # func=get_current_weather, # name="Weather Checker", # description="用于查询指定城市的当前天气情况。输入应为城市名称。", # args_schema=WeatherInput # ) # 选项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.run({"location": "伦敦"})) # print(weather_tool.run("巴黎")) # 如果args_schema允许,也接受直接字符串输入关于此工具的要点:WeatherInput Schema: 我们定义了一个Pydantic模型WeatherInput,用于指定预期输入(location)。这有助于LangChain验证输入并提供结构。使用Pydantic v1(pydantic.v1)通常是为了与旧版BaseTool结构兼容。get_current_weather Function: 这是核心逻辑。它目前使用随机数据,但模拟了API调用处理程序的结构,包括基本的错误处理。print语句有助于追踪工具的执行过程。WeatherTool 类: 我们继承BaseTool以获得更明确的控制。name:工具的简洁标识符。description:对智能体很重要。LLM使用此描述来决定何时使用该工具以及提供什么输入。使其清晰明了,信息充分。args_schema:链接到我们的Pydantic输入模型。_run:同步执行方法。_arun:异步执行方法。即使底层函数是同步的,实现_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"---> 调用驾驶时长工具计算:从{origin}到{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})相同。" time_hours = simulated_distance_km / average_speed_kph hours = int(time_hours) minutes = int((time_hours - hours) * 60) return f"从{origin}到{destination}的估算驾驶时长约为{hours}小时{minutes}分钟({simulated_distance_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.run({"origin": "巴黎", "destination": "柏林"}))该工具遵循与天气工具相同的模式:一个输入模式(DrivingTimeInput)、一个核心逻辑函数(estimate_driving_time),以及一个BaseTool子类(DrivingTimeTool),提供名称、描述和执行方法。其描述清晰地说明了其用途和预期输入。步骤3:创建与配置智能体现在我们有了自定义工具,让我们将它们集成到智能体中。我们将使用一个ReAct(推理与行动)智能体,这是一种常见模式,LLM会根据输入和之前的步骤来推断下一步要采取的行动(工具)。# --- 智能体设置 --- from langchain_openai import ChatOpenAI from langchain import hub from langchain.agents import create_react_agent, AgentExecutor # 确保您的环境变量或.env文件中已设置OPENAI_API_KEY # os.environ["OPENAI_API_KEY"] = "您的API密钥" 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. 获取提示模板 # 拉取一个为聊天模型优化的预定义ReAct提示 # 您可以在LangChain Hub上查看其他提示:https://smith.langchain.com/hub prompt = hub.pull("hwchase17/react-chat") # 如果好奇,可以检查提示模板:print(prompt.template) # 4. 创建ReAct智能体 # 这将LLM、工具和提示绑定在一起 agent = create_react_agent(llm, tools, prompt) # 5. 创建智能体执行器 # 这运行智能体循环(思考 -> 行动 -> 观察 -> ...) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 设置为True可查看智能体的推理步骤 handle_parsing_errors=True, # 增加对潜在LLM输出解析问题的鲁棒性 max_iterations=5 # 防止潜在的无限循环 ) print("智能体执行器创建成功。")让我们细分智能体的创建过程:LLM初始化: 我们实例化ChatOpenAI。温度设置为0,以获得更确定性的响应,适用于工具使用。请记住安全处理您的API密钥(环境变量是常见做法)。工具列表: 我们将自定义的weather_tool和driving_tool实例收集到一个列表中。提示模板: 我们从LangChain Hub(hwchase17/react-chat)获取一个经过验证的ReAct提示模板。此模板构建了交互方式,引导LLM逐步思考并决定使用哪个工具(或者是否可以直接回答)。create_react_agent: 此函数构建核心智能体逻辑,将LLM的推理能力与可用工具和提示结构连接起来。AgentExecutor: 这是智能体的运行时环境。它接收智能体逻辑和工具,然后管理执行循环:将输入和对话历史传递给智能体。智能体(LLM + 提示)决定一个行动(例如,使用“天气查询器”,输入“伦敦”)。执行器调用相应的工具(weather_tool.run(“伦敦”))。工具的输出(观察结果)被传回给智能体。此循环持续进行,直到智能体确定已得到最终答案。verbose=True在开发过程中强烈推荐,以便观察智能体的思考过程。handle_parsing_errors=True使智能体更具弹性,即使LLM生成的输出与预期格式不完全匹配也能应对。max_iterations是一种安全机制。步骤4:运行智能体并观察行为智能体执行器就绪后,让我们用不同的查询来测试它。# --- 运行智能体 --- print("\n--- 运行简单天气查询 ---") response1 = agent_executor.invoke({ "input": "多伦多现在天气怎么样?", "chat_history": [] # 单轮查询时使用空历史记录 }) print("\n最终答案:", response1['output']) print("\n--- 运行简单驾驶时长查询 ---") response2 = agent_executor.invoke({ "input": "从柏林开车到慕尼黑需要多久?", "chat_history": [] }) print("\n最终答案:", response2['output']) print("\n--- 运行多工具查询 ---") response3 = agent_executor.invoke({ "input": "能告诉我罗马现在的天气吗,以及从那不勒斯开车到罗马可能需要多久?", "chat_history": [] }) print("\n最终答案:", response3['output']) # 智能体应直接回答的查询示例(如果可能) # print("\n--- 运行非工具查询 ---") # response4 = agent_executor.invoke({ # "input": "法国的首都是哪里?", # "chat_history": [] # }) # print("\n最终答案:", response4['output'])当verbose=True时观察输出。对于每个涉及工具的查询,您应该看到类似以下模式的内容:思考: LLM分析输入并决定需要特定信息(例如,多伦多的天气)。它识别出“天气查询器”工具的描述符合这一需求。行动: LLM格式化一个行动,指定工具名称(Weather Checker)以及所需的输入({"location": "Toronto"})。观察: AgentExecutor运行weather_tool并传入“Toronto”,捕获其输出(模拟的天气字符串),并将其反馈给LLM。(如有必要,重复此过程): 对于多工具查询,在获取天气后,智能体的思考过程可能会识别出对驾驶时长的需求,选择“驾驶时长估算器”工具,执行它,获得观察结果,然后综合两部分信息。思考: LLM现在已拥有所需信息。最终答案: LLM向用户形成最终回复。请密切关注智能体如何使用BaseTool子类中定义的精确名称和输入模式。工具描述的质量对智能体选择正确工具的能力不可或缺。总结与后续步骤这项实践练习展示了创建具备自定义能力的LangChain智能体的基本工作流程:确定需求: 确定智能体必须执行的特定任务。实现工具: 为每种能力创建函数或类,使用Tool.from_function或通过继承BaseTool来封装它们。仔细关注name、description和args_schema。配置智能体: 选择合适的智能体类型(例如ReAct),选择LLM,收集工具,并使用合适的提示模板。实例化执行器: 创建AgentExecutor以管理运行时循环。测试与观察: 运行查询并使用verbose=True来理解智能体的推理和工具使用情况。在此基础上,您可以进一步考虑:更复杂的工具: 集成与数据库、专有API交互或执行复杂计算的工具。错误处理: 在您的工具内部以及可能在智能体执行器本身中实现更精密的错误处理(如“处理工具错误和智能体恢复”部分所述)。结构化工具调用: 对于支持此功能的LLM(例如较新的OpenAI模型),尝试基于函数/工具调用的智能体(create_openai_tools_agent、create_tool_calling_agent),与ReAct基于文本的方法相比,这能提供更可靠的输入/输出解析。异步执行: 通过确保您的工具和智能体执行器有效使用异步操作(_arun方法)来优化I/O密集型任务(如API调用)的性能。构建带自定义工具的智能体是一项核心技术,用于创建功能强大、专业化的LLM应用,这些应用能够交互并解决复杂问题。