LangChain核心组件Tools深度解析

还记得你的第一个 Agent吗？

它能回答问题、能聊天、能推理，但它只能"说"，不能"做"。

想让它查询数据库?？行。

想让它调用API？不行。

想让它移动文件？还是不行。

这时候，Tools就是 Agent走向世界的关键。

今天,我们深入剖析LangChain v1.0中 Tools的核心机制。

一、Tools的本质

Tools是 Agent与外部世界交互的接口。

现实场景：

你(用户): "帮我查一下北京今天的天气"

Agent(大脑): "我需要调用天气查询工具"

Tool(手): 实际调用天气API,获取数据

Agent(大脑)： "北京今天晴,25℃"

没有Tool，Agent就只能回答"抱歉,，无法查询实时天气"。

Tools是模块化设计，每个Tool专注一个功能，可以像搭积木一样组合使用。

二、Tool的构成

一个完整的Tool包含以下要素：

Tool(
    name="search_database",           # ① 工具名称
    description="搜索客户数据库",       # ② 功能描述
    args_schema=SearchInput,          # ③ 输入参数的JSON Schema
    func=search_function,             # ④ 实际执行的函数
    return_direct=False               # ⑤ 是否直接返回结果给用户
)

① name（工具名称）

● 必选，且在工具集中唯一

● Agent通过名称识别和选择工具

● 命名要直观、准确，比如 get_weather比 tool1好得多

② description（功能描述）

● 描述要清晰、具体，告诉Agent什么时候用这个工具

对比示例:

❌ description="查询数据"

✅ description="从客户数据库中搜索匹配的记录,支持按姓名、邮箱、手机号查询"

③ args_schema（参数定义）

● 定义工具需要哪些输入参数

● 使用 Pydantic模型或 JSON Schema

● 类型提示是必需的，Agent根据这个生成调用参数

④ func（执行函数）

● 工具的实际业务逻辑

● 可以是同步函数或异步函数

⑤ return_direct（直接返回）

● 仅对Agent相关

● False（默认）：工具结果返回给 Agent，由 Agent决定下一步

● True：直接返回结果给用户，中断Agent循环

实战场景：

● 数据查询工具:return_direct=False，Agent可能需要基于查询结果继续推理

● 文件下载工具:return_direct=True，下载完成就可以直接告诉用户了

2.2 工作流程详解

步骤1: 定义Tools并绑定到Model
↓
步骤2: LLM根据name、description和schema理解工具能力
↓
步骤3: 用户输入 → LLM推理 → 决定是否需要调用工具
↓
步骤4: 如果需要,LLM生成工具调用请求(包含参数)
↓
步骤5: Agent框架执行工具,获取结果
↓
步骤6: 结果返回给LLM继续推理
↓
步骤7: 生成最终答案返回用户

关键理解：LLM本身不执行工具，它只是"请求"执行。真正的执行由Agent框架完成。

三、  创建自定义工具

1.  方式1

@tool装饰器(最简单)

这是创建工具最快捷的方式：

from langchain.tools import tool

@tool
def add_numbers(a: int, b: int) -> int:
    """两个整数相加。
    
    Args:
        a: 第一个整数
        b: 第二个整数
    
    Returns:
        两数之和
    """
    return a + b

# 查看工具属性
print(f"name: {add_numbers.name}")           # add_numbers
print(f"description: {add_numbers.description}")  # 两个整数相加
print(f"args: {add_numbers.args}")           # {'a': {'type': 'integer'}, 'b': {'type': 'integer'}}

# 调用工具
result = add_numbers.invoke({"a": 10, "b": 20})
print(result)  # 30

关键点：

● 装饰器自动从函数名生成工具名

● 自动从docstring生成描述

● 自动从类型提示生成参数schema

● 类型提示是必需的，否则LLM不知道参数类型

2.  自定义工具属性

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

class AddInput(BaseModel):
    """加法运算的输入"""
    a: int = Field(description="被加数")
    b: int = Field(description="加数")

@tool(
    name="calculator_add",                    # 自定义名称
    description="执行两个整数的加法运算",    # 自定义描述
    args_schema=AddInput,                     # 详细的参数定义
    return_direct=True                        # 直接返回结果
)
def add_numbers(a: int, b: int) -> int:
    """两个整数相加"""
    return a + b

print(add_numbers.name)         # calculator_add
print(add_numbers.return_direct) # True

为什么要用 Pydantic定义 args_schema?

● 更详细的参数说明：Field的description会传递给LLM

● 参数验证:自动验证类型、范围等

● 复杂结构支持:嵌套对象、可选参数、默认值

3.  方式2：:StructuredTool.from_function

这种方式提供更多配置灵活性:：

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    """搜索输入参数"""
    query: str = Field(description="搜索关键词")
    limit: int = Field(default=10, description="返回结果数量上限")

def search_database(query: str, limit: int = 10) -> str:
    """搜索客户数据库"""
    # 这里应该是真实的数据库查询逻辑
    return f"找到{limit}条关于'{query}'的记录"

search_tool = StructuredTool.from_function(
    func=search_database,
    name="search_db",
    description="从客户数据库中搜索记录,支持模糊匹配",
    args_schema=SearchInput,
    return_direct=False
)

# 使用工具
result = search_tool.invoke({"query": "张三", "limit": 5})
print(result)  # 找到5条关于'张三'的记录

@tool vs StructuredTool.from_function:

● @tool：代码简洁，适合简单工具

● StructuredTool：配置灵活，适合需要精确控制的场景

4 实战示例：天气查询工具

from langchain.tools import tool
from pydantic import BaseModel, Field
from typing import Literal
import requests

class WeatherInput(BaseModel):
    """天气查询输入"""
    location: str = Field(description="城市名称,如'北京'、'上海'")
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="温度单位"
    )

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius") -> str:
    """获取指定城市的实时天气信息。
    
    支持全球主要城市查询,返回当前温度、天气状况等。
    """
    # 实际应用中应调用真实的天气API
    # 这里用模拟数据演示
    temp = 25 if units == "celsius" else 77
    return f"{location}当前天气:晴,温度{temp}°{units[0].upper()}"

# 测试工具
result = get_weather.invoke({
    "location": "北京",
    "units": "celsius"
})
print(result)  # 北京当前天气:晴,温度25°C

四、将 Tools绑定到 Model

1.  基本绑定流程

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_core.messages import HumanMessage

load_dotenv()

# 1. 初始化模型
model = init_chat_model("openai:gpt-4o-mini")

# 2. 定义工具
@tool
def get_weather(location: str) -> str:
    """获取指定地点的天气"""
    return f"{location}今天晴,25℃"

@tool
def search_database(query: str) -> str:
    """搜索数据库"""
    return f"找到3条关于'{query}'的记录"

# 3. 绑定工具到模型
model_with_tools = model.bind_tools([get_weather, search_database])

# 4. 调用模型
response = model_with_tools.invoke([
    HumanMessage(content="北京今天天气怎么样?")
])

# 5. 检查是否请求调用工具
if response.tool_calls:
    for tool_call in response.tool_calls:
        print(f"工具: {tool_call['name']}")
        print(f"参数: {tool_call['args']}")

输出：

工具: get_weather
参数: {'location': '北京'}

2.  理解tool_calls的结构

response.tool_calls = [
    {
        'name': 'get_weather',          # 工具名称
        'args': {'location': '北京'},   # 调用参数
        'id': 'call_abc123'             # 唯一标识
    }
]

关键点：

● tool_calls是一个列表，支持并行调用多个工具

● 每个调用都有唯一的id，用于追踪

● args是LLM根据schema生成的参数

3.  两种调用场景对比

场景1：需要调用工具

response = model_with_tools.invoke([
    HumanMessage(content="北京今天天气怎么样?")
])

print(response.content)      # 通常为空
print(response.tool_calls)   # [{'name': 'get_weather', ...}]

特点：

● content为空（因为模型选择调用工具而非直接回答）

● tool_calls包含工具调用信息

场景2：不需要调用工具

response = model_with_tools.invoke([
    HumanMessage(content="你好")
])

print(response.content)      # "你好!我是AI助手,很高兴为你服务"
print(response.tool_calls)   # []

特点：

● content包含自然语言回复

● tool_calls为空列表

4.  完整的工具执行流程

from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

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

@tool
def calculator(expression: str) -> str:
    """执行数学计算"""
    try:
        result = eval(expression)
        return str(result)
    except:
        return "计算错误"

model_with_tools = model.bind_tools([calculator])

# 第一轮:用户提问
messages = [HumanMessage(content="计算123 * 456")]
response = model_with_tools.invoke(messages)

print("=== 第一轮响应 ===")
print(f"tool_calls: {response.tool_calls}")

# 第二轮:执行工具并返回结果
if response.tool_calls:
    # 模拟Agent执行工具
    tool_call = response.tool_calls[0]
    tool_result = calculator.invoke(tool_call['args'])
    
    # 将工具结果添加到对话历史
    messages.append(response)  # AI的工具调用请求
    messages.append(ToolMessage(
        content=tool_result,
        tool_call_id=tool_call['id']
    ))
    
    # 让模型基于工具结果生成最终答案
    final_response = model_with_tools.invoke(messages)
    print("\n=== 最终响应 ===")
    print(final_response.content)  # "123 * 456 = 56088"

关键点：

1.  第一次调用：模型决定需要用工具,返回tool_calls

2.  执行工具：获取实际结果

3.  第二次调用：把工具结果传回模型,生成最终答案

这就是 Agent循环的基础！

五、高级特性

1.  强制工具调用

有时你希望模型必须调用某个工具：

response = model_with_tools.invoke(
    messages,
    tool_choice="get_weather"  # 强制调用这个工具
)

适用场景：

● 表单填充场景：必须调用数据验证工具

● 数据提取场景：必须调用解析工具

● 审批流程：必须调用权限检查工具

2.  并行工具调用

高级模型（如GPT-4o）支持同时调用多个工具:

response = model_with_tools.invoke([
    HumanMessage(content="查询北京和上海的天气")
])

# 模型可能同时请求两次get_weather
print(len(response.tool_calls))  # 2
print(response.tool_calls[0]['args'])  # {'location': '北京'}
print(response.tool_calls[1]['args'])  # {'location': '上海'}

注意：不是所有模型都支持并行调用，使用前查看文档。

3.  访问运行时上下文（ToolRuntime）

LangChain v1.0引入了 ToolRuntime，让工具可以访问：

● State：Agent的状态

● Context：用户ID、会话信息等不可变配置

● Store：持久化存储(跨会话记忆)

● Stream Writer：实时流式输出

from langchain.tools import tool, ToolRuntime

@tool
def get_user_info(
    user_id: str,
    runtime: ToolRuntime  # 运行时上下文,不暴露给LLM
) -> str:
    """查询用户信息"""
    # 访问持久化存储
    store = runtime.store
    user_info = store.get(("users",), user_id)
    
    if user_info:
        return f"用户{user_id}的信息:{user_info.value}"
    return "用户不存在"

@tool
def save_user_preference(
    pref_name: str,
    pref_value: str,
    runtime: ToolRuntime
) -> str:
    """保存用户偏好设置"""
    # 访问当前状态
    preferences = runtime.state.get("user_preferences", {})
    preferences[pref_name] = pref_value
    
    # 可以使用Command更新状态
    return f"已保存偏好:{pref_name}={pref_value}"

关键点：runtime参数不会暴露给LLM，模型只看到其他业务参数。

4.  流式输出工具执行进度

from langchain.tools import tool, ToolRuntime

@tool
def complex_analysis(data: str, runtime: ToolRuntime) -> str:
    """执行复杂的数据分析"""
    writer = runtime.stream_writer
    
    # 实时反馈进度
    writer("开始加载数据...")
    # ... 加载逻辑
    
    writer("数据预处理中...")
    # ... 预处理逻辑
    
    writer("正在执行分析...")
    # ... 分析逻辑
    
    writer("生成分析报告...")
    return "分析完成!"

用户体验：用户可以实时看到工具执行的进度，而不是傻等结果。

六、实战案例：文件管理Agent

场景设计

我们要创建一个文件管理Agent，支持：

● 移动文件

● 查看文件列表

● 读取文件内容

1.  定义工具集

from langchain.tools import tool
from pathlib import Path
import shutil

@tool
def move_file(source_path: str, destination_path: str) -> str:
    """移动文件到指定位置。
    
    Args:
        source_path: 源文件路径
        destination_path: 目标路径
    """
    try:
        shutil.move(source_path, destination_path)
        return f"文件已从 {source_path} 移动到 {destination_path}"
    except Exception as e:
        return f"移动失败: {str(e)}"

@tool
def list_files(directory: str) -> str:
    """列出目录中的所有文件。
    
    Args:
        directory: 目录路径
    """
    try:
        path = Path(directory)
        files = [f.name for f in path.iterdir() if f.is_file()]
        return f"目录 {directory} 中的文件:\n" + "\n".join(files)
    except Exception as e:
        return f"列出文件失败: {str(e)}"

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

2.  创建 Agent并测试

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, ToolMessage

# 初始化模型
model = init_chat_model("openai:gpt-4o-mini", temperature=0)

# 绑定工具
tools = [move_file, list_files, read_file]
model_with_tools = model.bind_tools(tools)

# 测试1:移动文件
print("=== 测试1:移动文件 ===")
messages = [HumanMessage(content="把当前目录下的test.txt移动到桌面")]
response = model_with_tools.invoke(messages)

if response.tool_calls:
    tool_call = response.tool_calls[0]
    print(f"调用工具: {tool_call['name']}")
    print(f"参数: {tool_call['args']}")
    
    # 执行工具
    tool_result = move_file.invoke(tool_call['args'])
    print(f"执行结果: {tool_result}")

# 测试2:列出文件
print("\n=== 测试2:列出文件 ===")
messages = [HumanMessage(content="列出当前目录的所有文件")]
response = model_with_tools.invoke(messages)

if response.tool_calls:
    tool_call = response.tool_calls[0]
    tool_result = list_files.invoke(tool_call['args'])
    print(f"执行结果:\n{tool_result}")

3.  优化：添加错误处理和日志

import logging
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@tool
def move_file_safe(source_path: str, destination_path: str) -> str:
    """移动文件(带错误处理和日志)"""
    logger.info(f"准备移动文件: {source_path} -> {destination_path}")
    
    try:
        # 检查源文件是否存在
        if not Path(source_path).exists():
            return f"错误:源文件 {source_path} 不存在"
        
        # 检查目标目录是否存在
        dest_dir = Path(destination_path).parent
        if not dest_dir.exists():
            return f"错误:目标目录 {dest_dir} 不存在"
        
        # 执行移动
        shutil.move(source_path, destination_path)
        logger.info(f"文件移动成功")
        return f"✅ 文件已成功移动到 {destination_path}"
        
    except PermissionError:
        logger.error(f"权限不足")
        return "❌ 错误:没有权限操作该文件"
    except Exception as e:
        logger.error(f"移动失败: {str(e)}")
        return f"❌ 移动失败: {str(e)}"

七、设计原则与常见陷阱

工具设计原则

1.  单一职责

# ❌ 不好:一个工具做太多事
@tool
def file_operations(operation: str, path: str, content: str = "") -> str:
    """执行各种文件操作"""
    if operation == "read":
        return read_file(path)
    elif operation == "write":
        return write_file(path, content)
    elif operation == "delete":
        return delete_file(path)

# ✅ 好:每个工具专注一个功能
@tool
def read_file(path: str) -> str:
    """读取文件内容"""
    ...

@tool
def write_file(path: str, content: str) -> str:
    """写入文件内容"""
    ...

2.  描述具体

# ❌ 描述太泛
@tool
def search(query: str) -> str:
    """搜索"""
    ...

# ✅ 描述清晰
@tool
def search_customer_by_name(name: str) -> str:
    """在客户数据库中按姓名搜索客户记录。
    
    支持模糊匹配,返回包含该姓名的所有客户信息。
    适用场景:客服查询客户资料、销售跟进客户等。
    """
    ...

3.  参数验证

from pydantic import BaseModel, Field, validator

class SearchInput(BaseModel):
    """搜索参数"""
    keyword: str = Field(min_length=1, max_length=100)
    page: int = Field(default=1, ge=1, le=100)
    
    @validator('keyword')
    def keyword_not_empty(cls, v):
        if not v.strip():
            raise ValueError("关键词不能为空")
        return v.strip()

@tool(args_schema=SearchInput)
def search_products(keyword: str, page: int = 1) -> str:
    """搜索商品"""
    ...

常见陷阱

陷阱1：忘记处理异常

# ❌ 危险:没有异常处理
@tool
def divide(a: float, b: float) -> float:
    """除法运算"""
    return a / b  # b=0会崩溃

# ✅ 安全:完善的异常处理
@tool
def divide_safe(a: float, b: float) -> str:
    """安全的除法运算"""
    try:
        if b == 0:
            return "错误:除数不能为0"
        result = a / b
        return f"{a} ÷ {b} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

陷阱2：工具执行时间过长

# ❌ 问题:可能执行很久
@tool
def analyze_big_data(file_path: str) -> str:
    """分析大数据文件"""
    # 可能需要几分钟
    df = pd.read_csv(file_path)  # 100GB的文件
    return df.describe()

# ✅ 优化:添加超时和进度反馈
@tool
def analyze_big_data_safe(
    file_path: str,
    runtime: ToolRuntime
) -> str:
    """分析大数据文件(带进度反馈)"""
    writer = runtime.stream_writer
    
    writer("开始加载数据...")
    # 设置超时或分块处理
    df = pd.read_csv(file_path, nrows=10000)  # 只读前1万行
    
    writer("数据加载完成,正在分析...")
    result = df.describe()
    
    return str(result)

陷阱3：返回结果太长

# ❌ 问题:返回几万字
@tool
def get_log_file() -> str:
    """获取系统日志"""
    with open('system.log', 'r') as f:
        return f.read()  # 可能有100MB

# ✅ 优化:返回摘要或分页
@tool
def get_log_summary(lines: int = 100) -> str:
    """获取最近的系统日志摘要"""
    with open('system.log', 'r') as f:
        all_lines = f.readlines()
        recent_lines = all_lines[-lines:]  # 只返回最后100行
        return "".join(recent_lines)

性能优化建议

1.  工具缓存：对于相同参数的重复调用，可以缓存结果

2.  异步工具：IO密集型工具使用异步实现

3.  批量处理：支持批量参数,减少调用次数

4.  超时设置：避免工具执行时间过长阻塞Agent