智能体通过与大型语言模型(LLM)内部知识之外的环境互动来获得能力。这种互动通过工具实现。LangChain提供了一系列预设工具,用于常见的任务,比如网络搜索或使用计算器,但应用程序通常需要为特定API、数据库或私有逻辑定制功能。因此,创建自定义工具是构建精良、可用于生产的智能体的一项重要技能。LangChain工具本质上是一个封装特定功能的组件,智能体可以调用它。它将执行逻辑与元数据捆绑在一起,而最主要的就是一个name和一个description,智能体的LLM使用这些信息来决定何时以及如何使用该工具。自定义工具的构成:BaseToolLangChain中的自定义工具,其核心是继承自BaseTool类。我们来检查一下您需要定义的主要组成部分:name (str): 工具的唯一标识符。此名称在提供给智能体的所有工具中必须是不同的。它应具有描述性但简洁,通常使用snake_case命名法(例如,weather_reporter,database_query_executor)。智能体在决定调用工具时会在内部使用此名称。description (str): 这可以说是自定义工具中最重要的部分。描述会告诉智能体的LLM该工具的作用、它期望哪些输入以及它产生什么输出。编写清晰、准确且有内容的描述对于智能体有效使用工具非常必要。可以将其视为专门为LLM编写的文档。糟糕的描述会导致工具使用不当,或智能体未能在适当时候使用工具。_run(self, *args, **kwargs) (方法): 此方法包含您工具的同步执行逻辑。它接收由智能体确定的输入参数,并执行预期的操作,将结果以字符串形式返回。_arun(self, *args, **kwargs) (可选方法): 如果您的工具涉及I/O密集型操作(如网络请求或数据库查询),强烈建议实现异步_arun方法以获得更好的性能,尤其是在并发应用程序中。此方法使用Python的async/await语法。如果未实现_arun,LangChain通常会将同步的_run方法包装起来用于异步调用,这可能会阻塞事件循环。下面是一个使用BaseTool类的自定义工具的基本示例:import os import requests from langchain_core.tools import BaseTool from typing import Type, Optional from pydantic import BaseModel, Field # 可选,用于结构化参数 # 定义输入架构(可选但建议) class WeatherInput(BaseModel): location: str = Field(description="城市和州,例如:San Francisco, CA") class GetCurrentWeatherTool(BaseTool): name: str = "get_current_weather" description: str = ( "当你需要查询特定地点的当前天气状况时很有用。" "输入应为地点字符串。" ) # 如果使用结构化输入,请取消以下行的注释: # args_schema: Type[BaseModel] = WeatherInput # 示例:安全存储API密钥(例如,环境变量) api_key: str = os.environ.get("OPENWEATHERMAP_API_KEY") def _run(self, location: str) -> str: """同步使用此工具。""" if not self.api_key: return "错误:天气API密钥未设置。" if not location: return "错误:必须提供地点。" try: base_url = "http://api.openweathermap.org/data/2.5/weather" params = {"q": location, "appid": self.api_key, "units": "metric"} response = requests.get(base_url, params=params) response.raise_for_status() # 对于错误响应(4xx或5xx)引发HTTPError data = response.json() # 提取相关信息 main_weather = data['weather'][0]['main'] description = data['weather'][0]['description'] temp = data['main']['temp'] feels_like = data['main']['feels_like'] humidity = data['main']['humidity'] return ( f"Current weather in {location}: {main_weather} ({description}). " f"Temperature: {temp}°C (Feels like: {feels_like}°C). " f"Humidity: {humidity}%." ) except requests.exceptions.RequestException as e: return f"Error fetching weather data: {e}" except KeyError: return f"Error: Unexpected response format from weather API for {location}." except Exception as e: # 在处理过程中捕获意外错误 return f"An unexpected error occurred: {e}" async def _arun(self, location: str) -> str: """异步使用此工具。""" # 在生产环境中,请使用aiohttp等异步HTTP客户端 # 此示例为简化起见,在异步函数中使用了同步请求 # 但这并非实现真正异步性能的最佳实践。 import asyncio # 通过在线程池中运行同步_run来模拟异步调用 # 在真实的异步实现中,您将使用异步库(例如aiohttp) loop = asyncio.get_running_loop() result = await loop.run_in_executor(None, self._run, location) return result # 示例用法(假设已设置OPENWEATHERMAP_API_KEY) # weather_tool = GetCurrentWeatherTool() # sync_result = weather_tool.run("London, UK") # print(sync_result) # async_result = await weather_tool.arun("Tokyo, JP") # print(async_result)在此示例中:我们定义了一个GetCurrentWeatherTool。name是get_current_weather。description清楚地说明了其用途和预期输入。_run使用requests库实现逻辑,以调用外部天气API。它包含了基本的错误处理。_arun提供了一个异步接口。请注意,此示例为简化起见使用了run_in_executor,但生产实现应使用异步HTTP客户端(如aiohttp)来实现真正的非阻塞I/O。使用@tool装饰器简化工具创建手动定义继承自BaseTool的类可能会变得冗长,特别是对于较简单的工具。LangChain提供了一个方便的@tool装饰器,可以直接将任何Python函数或协程转换为工具对象。该装饰器从函数名称推断name,并使用函数的文档字符串作为description。from langchain_core.tools import tool import math @tool def simple_calculator(expression: str) -> str: """ 用于计算涉及加法、减法、乘法、除法和幂运算的简单数学表达式。 输入必须是有效的Python数值表达式字符串。 示例输入:'2 * (3 + 4) / 2**2' """ try: # 如果可能,请使用安全的求值方法,或限制操作。 # eval()通常对任意用户输入不安全。 # 对于生产环境,请考虑使用ast.literal_eval或专用的数学解析器。 allowed_chars = "0123456789+-*/(). " if not all(c in allowed_chars for c in expression): # 一个非常基本的净化检查 return "错误:表达式中包含无效字符。" # 此处为简化起见使用了eval,但在生产环境中请警惕安全风险。 result = eval(expression, {"__builtins__": None}, {'math': math}) return f"The result of '{expression}' is {result}" except Exception as e: return f"Error evaluating expression '{expression}': {e}" # 'simple_calculator'对象现在是一个LangChain工具 print(f"工具名称: {simple_calculator.name}") print(f"工具描述: {simple_calculator.description}") # print(simple_calculator.run("5 * (10 - 2)")) @tool装饰器会自动为您创建BaseTool子类结构。它检查函数的类型提示以确定输入参数。对于异步函数(async def),它会自动填充_arun方法。使用Pydantic的结构化工具输入尽管传递简单字符串作为输入是可行的,但复杂工具通常会因结构化输入而更有效用,这种输入包含多个参数、类型验证以及对每个参数的清晰说明。您可以通过定义Pydantic BaseModel并将其分配给工具的args_schema属性来实现此目的。当您提供args_schema时,LLM会收到指令,将其为工具准备的输入格式化为符合该架构的JSON对象。LangChain会处理解析此JSON并将参数正确传递给您的_run或_arun方法。from langchain_core.tools import BaseTool from pydantic import BaseModel, Field from typing import Type, Optional import datetime # 使用 Pydantic 定义输入架构 class FlightSearchInput(BaseModel): departure_city: str = Field(description="航班出发城市。") arrival_city: str = Field(description="航班抵达城市。") departure_date: str = Field(description="期望的出发日期,格式为 YYYY-MM-DD。") max_stops: Optional[int] = Field(None, description="可选的最多允许经停次数。") class FlightSearchTool(BaseTool): name: str = "flight_search_engine" description: str = ( "根据出发城市、抵达城市、出发日期以及可选的最多经停次数,搜索航班选项。" "返回可用的航班详情。" ) args_schema: Type[BaseModel] = FlightSearchInput def _run( self, departure_city: str, arrival_city: str, departure_date: str, max_stops: Optional[int] = None ) -> str: """同步执行,使用结构化参数。""" # 输入验证部分由 Pydantic 处理 print(f"Searching flights from {departure_city} to {arrival_city} on {departure_date}...") if max_stops is not None: print(f"Constraint: Maximum {max_stops} stops.") # --- 实际航班搜索API调用的占位符 --- # 在真实工具中,您将在此处使用参数调用航班API。 # 示例模拟响应: if departure_city.lower() == "london" and arrival_city.lower() == "new york": return (f"Found flights for {departure_date}: " f"Flight BA001 (Direct, $850), Flight UA934 (1 Stop, $720)") else: return f"No flights found for the specified route on {departure_date}." # --- 占位符结束 --- async def _arun( self, departure_city: str, arrival_city: str, departure_date: str, max_stops: Optional[int] = None ) -> str: """异步执行,使用结构化参数。""" # 在生产环境中,请在此处使用异步HTTP客户端(例如 aiohttp) print(f"(Async) Searching flights from {departure_city} to {arrival_city} on {departure_date}...") if max_stops is not None: print(f"(Async) Constraint: Maximum {max_stops} stops.") # 模拟异步工作 import asyncio await asyncio.sleep(0.5) # --- 实际异步航班搜索API调用的占位符 --- if departure_city.lower() == "london" and arrival_city.lower() == "new york": return (f"(Async) Found flights for {departure_date}: " f"Flight BA001 (Direct, $850), Flight UA934 (1 Stop, $720)") else: return f"(Async) No flights found for the specified route on {departure_date}." # --- 占位符结束 --- # 实例化工具 # flight_tool = FlightSearchTool() # LangChain 可能如何在内部调用它(简化版): # tool_input = '{"departure_city": "London", "arrival_city": "New York", "departure_date": "2024-12-25"}' # result = flight_tool.run(tool_input) # print(result) # 或者如果直接使用 run 方法,则直接使用结构化输入: # result_structured = flight_tool.run( # departure_city="London", # arrival_city="New York", # departure_date="2024-12-25" # ) # print(result_structured)使用args_schema使工具互动更加精确和明白,尤其是在处理多个参数或可选参数时。它还有助于LLM确切地了解需要向工具提供哪些信息。@tool装饰器也可以从类型提示的函数参数中自动推断args_schema,特别是当它们使用Pydantic模型时。工具设计最佳实践描述清晰: 描述很重要。明确说明工具的功能、确切的输入(包括格式)以及输出的含义。提及限制或先决条件(例如,“需要城市和州”或“返回摄氏温度”)。原子性: 设计工具以执行一个特定、定义明确的任务。避免创建试图完成过多任务的单一工具。更小、更集中的工具更易于智能体理解和组合。输入/输出格式: 确保工具始终返回一个字符串,这符合大多数标准智能体执行器的预期。如果返回复杂数据,请将其序列化为清晰的字符串格式(例如JSON、格式化文本),以便LLM在后续步骤中需要时进行解析。错误处理: 在_run / _arun方法中实现基本的错误处理,以捕获常见问题(例如API错误、无效输入)。返回有用的错误消息作为字符串,以便智能体知道工具执行失败。更复杂的错误处理策略将在后面讨论。安全性: 如果您的工具执行代码、与文件系统互动或根据LLM生成的输入调用外部API,请务必非常谨慎。严格净化输入并限制工具的权限,以防止提示注入等安全漏洞导致任意代码执行。这将在安全章节中进行更详细的介绍。异步实现: 如果您的工具执行I/O操作,请使用适当的异步库(aiohttp、asyncpg等)来实现_arun,以在并发智能体设置中获得更好的性能。通过熟练掌握自定义工具开发,您将能够赋予LangChain智能体专门的能力,使它们能够与您的应用程序所需的几乎任何系统或数据源进行互动。请记住,您的智能体的效用在很大程度上取决于您提供的工具的质量和清晰度。