开发代理专用定制工具

代理通过与外部系统和数据源交互，从而扩展大型语言模型（LLM）的功能，实现更强的性能。这种互动是通过工具实现的。尽管LangChain提供了一套预置工具，可用于网络搜索或访问计算器等常见任务，但实际应用常需要针对特定API、数据库或专属逻辑的定制功能。因此，创建定制工具是构建精巧且可用于生产环境的代理的一项基础技能。

LangChain工具本质上是一个组件，它封装了代理可调用的特定功能。它将执行逻辑与元数据（最主要的是name和description）捆绑在一起，代理的LLM会用这些信息来决定何时以及如何使用该工具。

定制工具的结构：BaseTool

核心而言，LangChain中的定制工具都继承自BaseTool类。让我们来看看你需要定义的必要组件：

1. name (字符串): 工具的唯一标识符。此名称在提供给代理的所有工具中必须独有。它应具有描述性但保持简洁，通常使用snake_case命名法（例如，weather_reporter，database_query_executor）。代理在决定调用工具时会内部使用此名称。
2. description (字符串): 这可以说是定制工具中最重要的部分。描述会告诉代理的LLM该工具做什么、它预期什么输入以及它产生什么输出。编写清晰、准确且信息丰富的描述对于代理有效使用工具极其重要。可以将其视为专门为LLM编写的说明文档。描述不佳会导致工具使用错误，或者代理在适当时候未能使用该工具。
3. _run(self, *args, **kwargs) (方法): 此方法包含工具的同步执行逻辑。它接收代理确定的输入参数 (parameter)，并执行预期操作，将结果以字符串形式返回。
4. _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="城市和州，例如：旧金山, CA")

class GetCurrentWeatherTool(BaseTool):
    name: str = "get_current_weather"
    description: str = (
        "在需要查询特定地点当前天气情况时很有用。输入应为地点字符串。"
    )
    # 如果使用结构化输入，请取消注释以下行：
    # args_schema: Type[BaseModel] = WeatherInput 

    # 示例：安全存储API密钥（例如，环境变量）
    api_key: Optional[str] = os.environ.get("OPENWEATHERMAP_API_KEY") 

    def _run(self, location: str) -> str:
        """同步使用工具。"""
        if not self.api_key:
            return "Error: Weather API key not set."
        if not location:
            return "Error: Location must be provided."

        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.invoke({"location": "伦敦, UK"})
# print(sync_result) 
# async_result = await weather_tool.ainvoke({"location": "东京, 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函数或协程直接转换为Tool对象。

该装饰器从函数名推断出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 "Error: Invalid characters in expression."

        # 此处为简化起见使用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"Tool Name: {simple_calculator.name}")
print(f"Tool Description: {simple_calculator.description}")
# print(simple_calculator.invoke({"expression": "5 * (10 - 2)"})) 

@tool装饰器会自动为你处理BaseTool子类结构的创建。它会检查函数的类型提示以确定输入参数 (parameter)。对于异步函数（async def），它会自动填充_arun方法。

结合Pydantic实现结构化工具输入

尽管将简单字符串作为输入是可行的，但复杂的工具通常受益于包含多个参数 (parameter)、类型验证和每个参数清晰描述的结构化输入。你可通过定义一个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": "伦敦", "arrival_city": "纽约", "departure_date": "2024-12-25"}
# result = flight_tool.invoke(tool_input) 
# print(result)

# 或者直接使用结构化输入：
# result_structured = flight_tool.invoke({
#     "departure_city": "伦敦", 
#     "arrival_city": "纽约", 
#     "departure_date": "2024-12-25"
# })
# print(result_structured)

使用args_schema可以使工具互动更精确和清晰，尤其是在处理多个参数或可选参数时。它还能帮助LLM明确需要向工具提供哪些信息。@tool装饰器也可以从带有类型提示的函数参数中自动推断args_schema，特别是当它们使用Pydantic模型时。

工具设计的最佳实践

• 描述清晰： 描述十分重要。清楚说明工具的功能、确切的输入（包括格式）以及输出的含义。提及局限性或前提条件（例如，“需要城市和州”或“返回摄氏温度”）。
• 原子性： 设计工具以执行一个特定的、定义明确的任务。避免创建试图完成过多任务的单一工具。更小、专注的工具更容易让代理进行推理 (inference)和组合。
• 输入/输出格式： 确保工具始终返回字符串，这是大多数标准代理执行器所期望的。如果返回复杂数据，请将其序列化为清晰的字符串格式（例如JSON、格式化文本），以便LLM在后续步骤中需要时进行解析。
• 错误处理： 在_run / _arun方法中实现基本的错误处理，以捕获常见问题（例如API错误、无效输入）。以字符串形式返回信息丰富的错误消息，以便代理知道工具执行失败。更高级的错误处理策略将在之后讨论。
• 安全性： 如果你的工具根据LLM生成的输入执行代码、与文件系统交互或调用外部API，请务必极其小心。严密净化输入，并限制工具的权限，以防止像提示注入这类安全漏洞导致任意代码执行。安全章节将更详细地说明这一点。
• 异步实现： 如果你的工具执行I/O操作，请使用适当的异步库（aiohttp、asyncpg等）实现_arun，以在并发代理设置中获得更佳性能。

通过熟练掌握定制工具的开发，你将获得赋予LangChain代理专业技能的能力，使它们能够与你的应用所需的几乎任何系统或数据源进行互动。请记住，你的代理的有效性在很大程度上依赖于你为其提供的工具的质量和清晰度。