第42集：LangChain中工具的两种定义方式：@tool装饰器与BaseTool类

章节标题

LangChain工具定义的两种方法详解

核心知识点讲解

工具定义的重要性

在LangChain中，工具是智能体与外部世界交互的桥梁。正确定义工具对于智能体的能力至关重要：

1. 扩展智能体能力：通过工具，智能体可以执行各种任务，如搜索、计算、API调用等
2. 标准化接口：工具定义提供了标准化的接口，使智能体能够统一调用不同的功能
3. 参数验证：工具定义可以包含参数验证，确保智能体提供正确的参数
4. 错误处理：工具定义可以包含错误处理逻辑，提高智能体的鲁棒性
5. 文档生成：工具定义会自动生成文档，帮助智能体理解工具的功能

@tool装饰器

什么是@tool装饰器

@tool装饰器是LangChain中一种简单、直观的工具定义方式，它允许你通过装饰普通函数来创建工具。

@tool装饰器的优势

• 简洁易用：只需在函数上添加装饰器，无需继承类
• 自动文档：自动从函数文档字符串生成工具描述
• 参数自动提取：自动从函数签名提取参数信息
• 快速原型：适合快速创建简单工具

@tool装饰器的局限性

• 功能有限：不支持复杂的参数验证和错误处理
• 灵活性较低：难以实现高级功能，如异步执行
• 状态管理：不适合需要维护状态的工具

BaseTool类

什么是BaseTool类

BaseTool是LangChain中工具的基类，它提供了更强大、更灵活的工具定义方式。通过继承BaseTool类，你可以创建功能更丰富的工具。

BaseTool类的优势

• 功能强大：支持复杂的参数验证和错误处理
• 灵活性高：可以实现高级功能，如异步执行、状态管理等
• 可扩展性好：可以通过继承和组合创建复杂工具
• 标准化：提供了统一的接口和生命周期方法

BaseTool类的局限性

• 代码量较大：需要编写更多代码来定义工具
• 学习曲线较陡：需要了解类的继承和方法重写
• 初始化复杂：可能需要复杂的初始化逻辑

实用案例分析

案例1：使用@tool装饰器创建简单工具

场景：创建一个简单的天气查询工具，通过城市名称获取天气信息。

解决方案：使用@tool装饰器定义一个函数，接收城市名称参数，返回天气信息。

优势：代码简洁，易于理解和维护。

案例2：使用BaseTool类创建复杂工具

场景：创建一个股票查询工具，需要验证股票代码格式，处理API错误，并支持缓存功能。

解决方案：继承BaseTool类，实现参数验证、错误处理和缓存逻辑。

优势：功能强大，能够处理复杂场景。

代码示例

示例1：使用@tool装饰器定义工具

from langchain.tools import tool
from langchain.agents import AgentType, initialize_agent
from langchain.llms import OpenAI

# 初始化模型
llm = OpenAI(temperature=0.7)

# 使用@tool装饰器定义工具
@tool
async def get_weather(city: str) -> str:
    """获取指定城市的天气信息
    
    参数:
        city: 城市名称，例如：北京、上海、广州
    
    返回:
        天气信息字符串
    """
    # 模拟天气查询
    weather_data = {
        "北京": "晴天，温度10-18度",
        "上海": "多云，温度15-22度",
        "广州": "阴天，温度20-28度"
    }
    
    if city in weather_data:
        return f"{city}的天气：{weather_data[city]}"
    else:
        return f"抱歉，暂无法获取{city}的天气信息"

# 使用@tool装饰器定义另一个工具
@tool
async def calculate(expression: str) -> str:
    """执行数学计算
    
    参数:
        expression: 数学表达式，例如：1+1、2*3/4
    
    返回:
        计算结果字符串
    """
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# 获取工具列表
tools = [get_weather, calculate]

# 初始化智能体
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 测试智能体
print(agent.run("北京的天气怎么样？"))
print(agent.run("12345乘以6789等于多少？"))

示例2：使用BaseTool类定义工具

from langchain.tools import BaseTool
from langchain.pydantic_v1 import Field, BaseModel
from langchain.agents import AgentType, initialize_agent
from langchain.llms import OpenAI
from typing import Optional
import requests
import time

# 初始化模型
llm = OpenAI(temperature=0.7)

# 定义股票查询工具的输入参数
class StockQueryInput(BaseModel):
    symbol: str = Field(..., description="股票代码，例如：AAPL、MSFT、GOOG")
    days: Optional[int] = Field(5, description="查询最近几天的股票数据，默认5天")

# 继承BaseTool类定义股票查询工具
class StockQueryTool(BaseTool):
    name = "StockQuery"
    description = "查询股票价格信息"
    args_schema = StockQueryInput
    
    def __init__(self):
        super().__init__()
        self.cache = {}  # 缓存
        self.cache_expiry = 3600  # 缓存过期时间（秒）
    
    def _run(self, symbol: str, days: int = 5) -> str:
        """执行股票查询
        
        参数:
            symbol: 股票代码
            days: 查询天数
        
        返回:
            股票价格信息字符串
        """
        # 验证股票代码格式
        if not symbol.isalpha() or len(symbol) < 1 or len(symbol) > 5:
            return f"错误：无效的股票代码格式"
        
        # 检查缓存
        cache_key = f"{symbol}_{days}"
        current_time = time.time()
        
        if cache_key in self.cache:
            cached_data, timestamp = self.cache[cache_key]
            if current_time - timestamp < self.cache_expiry:
                return f"缓存数据：{cached_data}"
        
        try:
            # 模拟股票查询API调用
            # 实际应用中，这里应该调用真实的股票API
            stock_data = {
                "AAPL": {"price": 180.25, "change": 1.25},
                "MSFT": {"price": 420.50, "change": -0.75},
                "GOOG": {"price": 145.30, "change": 0.50}
            }
            
            if symbol in stock_data:
                data = stock_data[symbol]
                result = f"{symbol}的当前价格：${data['price']}，涨跌：${data['change']}"
                
                # 更新缓存
                self.cache[cache_key] = (result, current_time)
                
                return result
            else:
                return f"错误：未找到股票代码 {symbol}"
                
        except Exception as e:
            return f"错误：查询股票时发生错误 - {str(e)}"
    
    async def _arun(self, symbol: str, days: int = 5) -> str:
        """异步执行股票查询
        
        参数:
            symbol: 股票代码
            days: 查询天数
        
        返回:
            股票价格信息字符串
        """
        # 简单实现，实际应用中应该使用异步API调用
        return self._run(symbol, days)

# 定义天气查询工具
class WeatherTool(BaseTool):
    name = "Weather"
    description = "查询指定城市的天气信息"
    
    def _run(self, city: str) -> str:
        """执行天气查询
        
        参数:
            city: 城市名称
        
        返回:
            天气信息字符串
        """
        try:
            # 模拟天气查询API调用
            weather_data = {
                "北京": "晴天，温度10-18度",
                "上海": "多云，温度15-22度",
                "广州": "阴天，温度20-28度"
            }
            
            if city in weather_data:
                return f"{city}的天气：{weather_data[city]}"
            else:
                return f"抱歉，暂无法获取{city}的天气信息"
                
        except Exception as e:
            return f"错误：查询天气时发生错误 - {str(e)}"
    
    async def _arun(self, city: str) -> str:
        """异步执行天气查询
        
        参数:
            city: 城市名称
        
        返回:
            天气信息字符串
        """
        return self._run(city)

# 创建工具实例
tools = [StockQueryTool(), WeatherTool()]

# 初始化智能体
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 测试智能体
print(agent.run("苹果股票(AAPL)的价格是多少？"))
print(agent.run("北京的天气怎么样？"))

示例3：混合使用两种工具定义方式

from langchain.tools import tool, BaseTool
from langchain.agents import AgentType, initialize_agent
from langchain.llms import OpenAI
from langchain.pydantic_v1 import Field, BaseModel

# 初始化模型
llm = OpenAI(temperature=0.7)

# 使用@tool装饰器定义简单工具
@tool
def greet(name: str) -> str:
    """打招呼
    
    参数:
        name: 人名
    
    返回:
        打招呼的消息
    """
    return f"你好，{name}！很高兴认识你！"

# 使用BaseTool类定义复杂工具
class CalculatorTool(BaseTool):
    name = "Calculator"
    description = "执行数学计算"
    
    def _run(self, expression: str) -> str:
        """执行数学计算
        
        参数:
            expression: 数学表达式
        
        返回:
            计算结果
        """
        try:
            # 安全计算，避免执行危险代码
            # 只允许基本的算术运算
            safe_chars = set("0123456789+-*/.() ")
            if all(c in safe_chars for c in expression):
                result = eval(expression)
                return f"计算结果：{result}"
            else:
                return "错误：表达式包含不安全的字符"
        except Exception as e:
            return f"错误：{str(e)}"
    
    async def _arun(self, expression: str) -> str:
        """异步执行数学计算
        
        参数:
            expression: 数学表达式
        
        返回:
            计算结果
        """
        return self._run(expression)

# 创建工具列表
tools = [greet, CalculatorTool()]

# 初始化智能体
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 测试智能体
print(agent.run("你好，我是张三"))
print(agent.run("123加456等于多少？"))
print(agent.run("(100-25)*4/2等于多少？"))

高级技巧

1. 工具参数的高级定义

使用Pydantic模型定义参数

对于复杂工具，建议使用Pydantic模型定义参数，这样可以：

• 提供更详细的参数描述：通过Field的description参数
• 添加参数验证：通过Pydantic的验证功能
• 支持默认值：为参数设置默认值
• 提高可读性：使参数定义更加清晰

示例

from langchain.pydantic_v1 import Field, BaseModel
from langchain.tools import BaseTool

class ComplexToolInput(BaseModel):
    query: str = Field(..., description="查询内容")
    limit: int = Field(10, description="结果数量限制")
    offset: int = Field(0, description="结果偏移量")
    sort_by: str = Field("relevance", description="排序方式")

class ComplexTool(BaseTool):
    name = "ComplexTool"
    description = "执行复杂查询"
    args_schema = ComplexToolInput
    
    def _run(self, query: str, limit: int = 10, offset: int = 0, sort_by: str = "relevance") -> str:
        # 工具实现
        pass
    
    async def _arun(self, query: str, limit: int = 10, offset: int = 0, sort_by: str = "relevance") -> str:
        # 异步工具实现
        pass

2. 工具的异步实现

为什么需要异步工具

• 提高性能：异步执行可以避免阻塞主线程
• 更好的并发处理：支持同时执行多个工具调用
• 更适合I/O密集型操作：如网络请求、文件操作等

如何实现异步工具

• 重写_arun方法：在BaseTool子类中实现_arun方法
• 使用async/await：使用Python的异步语法
• 避免阻塞操作：在异步方法中避免使用阻塞操作

示例

import asyncio
import aiohttp
from langchain.tools import BaseTool

class AsyncAPITool(BaseTool):
    name = "AsyncAPI"
    description = "异步调用API"
    
    async def _arun(self, url: str) -> str:
        """异步调用API
        
        参数:
            url: API URL
        
        返回:
            API响应
        """
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                if response.status == 200:
                    return await response.text()
                else:
                    return f"错误：API调用失败，状态码：{response.status}"
    
    def _run(self, url: str) -> str:
        """同步调用API（备用）
        
        参数:
            url: API URL
        
        返回:
            API响应
        """
        return asyncio.run(self._arun(url))

3. 工具的组合与复用

工具组合

• 顺序组合：将多个工具组合成一个管道
• 并行组合：同时执行多个工具，然后整合结果
• 条件组合：根据条件选择执行不同的工具

工具复用

• 创建工具基类：定义通用功能的基类
• **使用混入(Mixin)**：通过混入添加特定功能
• 工具工厂：创建工具的工厂函数

示例

from langchain.tools import BaseTool

# 工具基类
class BaseAPITool(BaseTool):
    def __init__(self, api_key: str):
        super().__init__()
        self.api_key = api_key
        self.base_url = "https://api.example.com"
    
    def _get_headers(self):
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

# 具体工具
class UserTool(BaseAPITool):
    name = "UserTool"
    description = "用户相关操作"
    
    def _run(self, user_id: str) -> str:
        # 使用基类的方法获取headers
        headers = self._get_headers()
        # 实现具体逻辑
        pass

class ProductTool(BaseAPITool):
    name = "ProductTool"
    description = "产品相关操作"
    
    def _run(self, product_id: str) -> str:
        # 使用基类的方法获取headers
        headers = self._get_headers()
        # 实现具体逻辑
        pass

最佳实践

1. 工具定义的选择

何时使用@tool装饰器

• 简单工具：功能简单，参数较少的工具
• 快速原型：需要快速创建和测试的工具
• 一次性工具：只在特定场景使用的工具

何时使用BaseTool类

• 复杂工具：功能复杂，需要参数验证和错误处理的工具
• 需要状态管理：需要维护内部状态的工具
• 需要异步执行：需要异步执行的工具
• 可复用工具：将在多个项目中使用的工具

2. 工具命名和描述的最佳实践

工具命名

• 简洁明了：使用简洁、明确的名称
• 使用动词：工具名称应该描述其执行的操作
• 一致性：保持命名风格的一致性

工具描述

• 清晰准确：准确描述工具的功能和用途
• 包含参数：说明工具需要什么参数
• 包含返回值：说明工具返回什么
• 使用中文：对于中文智能体，使用中文描述

3. 工具实现的最佳实践

参数验证

• 严格验证：验证所有输入参数的类型和格式
• 提供默认值：为可选参数提供合理的默认值
• 错误提示：提供清晰、有用的错误提示

错误处理

• 全面捕获：捕获所有可能的错误
• 优雅降级：在错误发生时提供合理的降级方案
• 日志记录：记录错误信息，便于调试

性能优化

• 缓存：对于频繁调用的工具，实现缓存机制
• 异步执行：对于I/O密集型操作，使用异步执行
• 批量处理：支持批量操作，减少API调用次数

4. 工具测试的最佳实践

单元测试

• 测试正常情况：测试工具在正常输入下的行为
• 测试异常情况：测试工具在异常输入下的行为
• 测试边界情况：测试工具在边界输入下的行为

集成测试

• 测试工具与智能体的集成：确保智能体能够正确调用工具
• 测试工具链：测试多个工具的组合使用
• 测试真实场景：在真实场景中测试工具的性能和可靠性

故障排除

1. 工具不被智能体调用

症状：智能体知道有工具，但不调用它

原因：

• 工具描述不清晰，智能体不理解工具的功能
• 工具参数描述不准确，智能体不知道如何提供参数
• 智能体的提示词模板不鼓励使用工具

解决方案：

• 优化工具描述，使其更清晰、更具体
• 改进参数描述，明确说明参数的格式和示例
• 调整智能体的提示词模板，鼓励使用工具

2. 工具执行失败

症状：智能体调用工具，但执行失败

原因：

• 参数错误：智能体提供的参数格式不正确
• API错误：工具调用的API失败
• 网络问题：网络连接失败
• 代码错误：工具实现中有bug

解决方案：

• 加强参数验证，提供更清晰的错误提示
• 添加错误处理和重试机制
• 实现网络错误的处理逻辑
• 修复工具实现中的bug

3. 工具执行结果不被智能体理解

症状：工具执行成功，但智能体不理解执行结果

原因：

• 工具返回格式不清晰
• 工具返回内容过于复杂
• 智能体的提示词模板不包含如何处理工具结果的指导

解决方案：

• 优化工具返回格式，使其更清晰、更结构化
• 简化工具返回内容，突出关键信息
• 调整智能体的提示词模板，添加处理工具结果的指导

4. 工具性能问题

症状：工具执行速度慢，影响智能体的响应速度

原因：

• API调用慢：工具调用的API响应速度慢
• 计算密集型操作：工具执行计算密集型操作
• 网络延迟：网络延迟导致工具执行慢
• 缓存未实现：频繁调用相同的工具但没有缓存

解决方案：

• 优化API调用，减少不必要的API请求
• 对于计算密集型操作，考虑使用异步执行
• 实现缓存机制，避免重复计算
• 优化网络请求，减少网络延迟

总结与展望

LangChain提供了两种强大的工具定义方式：@tool装饰器和BaseTool类。通过本集的学习，你已经掌握了：

1. @tool装饰器：简洁易用，适合创建简单工具
2. BaseTool类：功能强大，适合创建复杂工具
3. 两种方式的混合使用：根据工具的复杂性选择合适的定义方式
4. 高级技巧：如参数验证、异步执行、工具组合等
5. 最佳实践：工具命名、描述、实现和测试的最佳实践
6. 故障排除：常见问题的解决方法

未来，LangChain的工具系统可能会进一步发展，包括：

• 更丰富的工具类型：支持更多类型的工具，如多模态工具
• 更智能的工具选择：智能体能够更智能地选择和使用工具
• 更标准化的工具接口：工具接口更加标准化，便于跨平台使用
• 更强大的工具生态：更多预定义的工具可供使用

通过掌握LangChain的工具定义方式，你可以为智能体赋予更强大的能力，使其能够执行各种复杂任务，与外部世界进行更丰富的交互。