跳到主要内容

Tools 工具

Tools（工具）是 LangChain Agent 执行具体操作的函数。本章将详细介绍如何定义和使用工具。

什么是 Tools？​

Tools 是 Agent 可以调用的外部函数，用于执行特定任务，如搜索、计算、文件操作等。

工具的作用​

定义工具​

使用 @tool 装饰器​

from langchain.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积"""
    return a * b

# 调用
result = multiply.invoke({"a": 3, "b": 4})
print(result)  # 12

使用 Tool 类​

from langchain.tools import Tool

def add(a: int, b: int) -> str:
    return str(a + b)

add_tool = Tool(
    name="add",
    func=add,
    description="用于计算两个数的和，输入为两个整数"
)

# 调用
result = add_tool.invoke({"a": 5, "b": 3})

多参数工具​

@tool
def search_hotels(location: str, price_level: str = "all") -> str:
    """搜索酒店
    
    参数:
        location: 城市或地区名称
        price_level: 价格等级 (budget/mid-range/luxury)，默认为 all
    """
    if price_level == "budget":
        return f"{location}的经济型酒店：xxx酒店、yyy青年旅舍"
    elif price_level == "luxury":
        return f"{location}的豪华酒店：zzz酒店、aaa度假村"
    else:
        return f"{location}的中档酒店：..."

无参数工具​

@tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def get_random_joke() -> str:
    """随机获取一个笑话"""
    jokes = [
        "为什么程序员总是弄混圣诞节和万圣节？因为 Oct 31 = Dec 25",
        "为什么程序员不喜欢户外活动？因为他们的生活已经充满了bugs",
    ]
    import random
    return random.choice(jokes)

返回结构化数据​

使用 Pydantic​

from langchain.tools import tool
from pydantic import BaseModel, Field

class WeatherOutput(BaseModel):
    city: str = Field(description="城市名称")
    temperature: str = Field(description="温度")
    condition: str = Field(description="天气状况")

@tool
def get_weather(city: str) -> WeatherOutput:
    """获取城市天气信息"""
    # 模拟API调用
    return WeatherOutput(
        city=city,
        temperature="25°C",
        condition="晴"
    )

工具返回格式​

简单返回​

@tool
def simple_tool(input: str) -> str:
    """简单工具，返回字符串"""
    return f"处理结果: {input}"

带错误处理​

@tool
def safe_divide(a: float, b: float) -> str:
    """安全除法，参数为两个数字"""
    if b == 0:
        return "错误：除数不能为零"
    return str(a / b)

在 Agent 中使用工具​

基础示例​

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 定义工具
@tool
def get_weather(city: str) -> str:
    """获取城市天气"""
    return f"{city}今天晴，25°C"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

# 创建 Agent
agent = create_agent(
    model=model,
    tools=[get_weather, calculate],
    system_prompt="你是一个有帮助的助手，可以使用工具"
)

# 调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "北京天气怎么样?"}]
})
print(result["messages"][-1].content)

多轮对话​

# 第一次调用
result1 = agent.invoke({
    "messages": [{"role": "user", "content": "上海天气如何?"}]
})

# Agent 会自动调用 get_weather 工具
# 返回天气信息

工具选择机制​

工具描述的重要性​

LLM 根据工具的 name 和 description 选择要使用的工具：

# ✅ 好的描述
@tool
def get_stock_price(symbol: str) -> str:
    """获取股票价格
    
    参数:
        symbol: 股票代码，如 AAPL, GOOGL
        
    返回:
        股票当前价格
    """
    return "150.00 美元"

# ❌ 差的描述
@tool
def get_price(s: str) -> str:
    """获取价格"""
    return "..."

避免工具冲突​

# 使用不同的名称
@tool
def search_google(query: str) -> str:
    """使用Google搜索信息"""
    ...

@tool
def search_baidu(query: str) -> str:
    """使用百度搜索信息"""
    ...

# Agent 会根据用户意图选择合适的搜索引擎

高级用法​

异步工具​

import asyncio

@tool
async def async_search(query: str) -> str:
    """异步搜索"""
    await asyncio.sleep(0.1)  # 模拟异步操作
    return f"搜索结果: {query}"

工具响应格式化​

@tool(response_format="content")
def get_formatted_weather(city: str) -> str:
    """获取格式化天气"""
    return f"""城市：{city}
天气：晴
温度：25°C
湿度：60%"""

自定义工具类​

from langchain.tools import BaseTool
from typing import Optional

class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "自定义搜索工具，用于搜索网络信息"
    
    def _run(self, query: str, source: Optional[str] = None) -> str:
        if source == "wiki":
            return f"维基百科结果: {query}"
        return f"搜索结果: {query}"

# 使用
search_tool = CustomSearchTool()
result = search_tool.invoke("Python")

错误处理工具​

from langchain.tools import tool
from langchain_core.runnables import Runnable

class RetryTool(Runnable):
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
    
    def invoke(self, input, config=None):
        for attempt in range(self.max_retries):
            try:
                return self._execute(input)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise
                continue

实际应用示例​

1. 天气查询助手​

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

@tool
def get_weather(city: str) -> str:
    """查询城市天气"""
    weather_map = {
        "北京": "晴，25°C",
        "上海": "多云，28°C",
        "广州": "雨，30°C"
    }
    return weather_map.get(city, "天气数据未找到")

@tool
def get_forecast(city: str, days: int = 3) -> str:
    """查询城市天气预报
    
    参数:
        city: 城市名称
        days: 预报天数，默认为3天
    """
    return f"{city}未来{days}天：晴、多云、晴"

agent = create_agent(
    model=model,
    tools=[get_weather, get_forecast],
    system_prompt="你是一个天气助手"
)

2. 数据处理工具​

@tool
def process_csv_data(csv_content: str, operation: str) -> str:
    """处理CSV数据
    
    参数:
        csv_content: CSV格式的字符串内容
        operation: 操作类型 (sum/avg/count)
    
    操作说明:
        - sum: 计算指定列的总和
        - avg: 计算指定列的平均值
        - count: 统计行数
    """
    import csv
    from io import StringIO
    
    reader = csv.DictReader(StringIO(csv_content))
    rows = list(reader)
    
    if operation == "count":
        return f"总行数: {len(rows)}"
    elif operation in ["sum", "avg"]:
        return "请指定要计算的列名"
    
    return "未知操作"

@tool
def generate_chart(data: str, chart_type: str = "bar") -> str:
    """生成图表
    
    参数:
        data: 图表数据
        chart_type: 图表类型 (bar/line/pie)
    """
    return f"生成了{chart_type}类型图表"

3. 文件操作工具​

import os

@tool
def read_file(file_path: str) -> str:
    """读取文件内容
    
    参数:
        file_path: 文件路径
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        return f"读取失败: {e}"

@tool
def write_file(file_path: str, content: str) -> str:
    """写入文件内容
    
    参数:
        file_path: 文件路径
        content: 要写入的内容
    """
    try:
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(content)
        return "写入成功"
    except Exception as e:
        return f"写入失败: {e}"

工具集合​

预定义工具​

LangChain 提供了许多预定义工具：

from langchain.agents import load_tools

# 加载数学工具
tools = load_tools(["calculator"])

# 加载搜索工具
tools = load_tools(["serpapi"], serpapi_api_key="...")

# 加载数学和搜索
tools = load_tools(["calculator", "serpapi"])

工具列表​

工具说明需要的依赖
calculator数学计算-
serpapiGoogle 搜索google-search-results
wikipedia维基百科wikipedia
arxivArXiv 搜索arxiv
ddg-searchDuckDuckGo 搜索duckduckgo-search

最佳实践​

1. 清晰的工具描述​

# ✅ 好的示例
@tool
def book_flight(origin: str, destination: str, date: str) -> str:
    """预订机票
    
    用于帮用户预订机票，需要提供以下信息：
    - origin: 出发城市
    - destination: 目的城市  
    - date: 出发日期 (格式: YYYY-MM-DD)
    
    返回预订结果或错误信息
    """
    pass

2. 错误处理​

@tool
def safe_api_call(endpoint: str) -> str:
    """调用API端点"""
    try:
        # 调用逻辑
        return "成功"
    except Exception as e:
        return f"错误: {str(e)}"

3. 适当的返回值​

# ✅ 返回有意义的信息
@tool
def get_user_info(user_id: str) -> str:
    """获取用户信息"""
    return f"用户ID: {user_id}, 姓名: 张三, 邮箱: [email protected]"

下一步​

现在你已经掌握了 Tools 的使用方法，接下来学习：