LangChain核心组件Tools(工具)

简介

大模型不仅可以生成文本，还能根据需要调用外部的工具来完成一些特定的任务。例如，模型可能需要从数据库中获取数据、在互联网上搜索信息，或者运行一段代码来实现某个功能。通过调用这些工具，模型能够更有效地处理复杂的任务，扩展自身的功能边界，从而更好地满足用户的需求。

• 组件：代理调用以执行动作的组件
• 桥梁作用：连接模型与外部系统（API、数据库、文件系统）
• 结构化接口：通过明确定义的输入输出与世界互动
• 能力扩展：扩展模型能力，使其能够执行具体任务

工具调用流程：

graph LR
    A["用户输入"] --> B["模型决策"]
    B --> C["工具调用"]
    C --> D["执行动作"]
    D --> E["返回结果"]

工具的组成：

• 模式（Schema）：模式是工具的一个重要组成部分，它定义了工具的基本信息和使用规范。具体包括：
  • 工具名称（Name of the tool）：用于唯一标识这个工具，方便模型在调用时能够明确指定要使用的是哪一个工具。
  • 描述（Description）：对工具的功能和用途进行简要说明，帮助模型理解这个工具是做什么的，以便在合适的情况下调用它。
  • 参数定义（Argument definitions）：详细说明了调用该工具时需要提供哪些参数，以及这些参数的类型、格式和含义等。通常会采用JSON模式（JSON schema）来定义参数，这样可以清晰地描述参数的结构和约束条件，确保模型在调用工具时能够提供正确和完整的参数信息。
• 执行函数或协程（Function or coroutine to execute）：这是工具的核心部分，它是一个具体的函数或者协程，用于实际执行工具所定义的任务。当模型调用工具时，就会触发这个函数或协程的执行，从而完成相应的操作，比如查询数据库、搜索网络或者运行代码等。函数或协程的实现会根据工具的具体功能而有所不同，它们是工具能够完成任务的关键所在。

以官网的示例进行说明

from langchain.tools import tool

@tool
def get_weather(locations: list[str]) -> str:
    """Get the weather at multiple locations."""
    results = []
    for location in locations:
        if location == 'SF':
            results.append('SF: 72°F sunny')
        elif location == 'NYC':
            results.append('NYC: 68°F cloudy')
        else:
            results.append(f'{location}: Weather not available')
    return ", ".join(results)

model_with_tools = model.bind_tools([get_weather])  

response = model_with_tools.invoke("What's the weather in SF and NYC?")
for tool_call in response.tool_calls:
    # View tool calls made by the model
    print(f"Tool: {tool_call['name']}")
    print(f"Args: {tool_call['args']}")

用户和模型之间的基本工具调用流程

sequenceDiagram
    participant User as 用户
    participant Model as AI模型
    participant Tools as 外部工具(天气查询)

    %% 用户发起天气查询请求
    User->>Model: What's the weather in SF and NYC?
    note over Model: 分析请求，确定需调用天气查询工具

    %% 模型并行调用工具（SF和NYC天气查询）
    Model->>Tools: 并行调用 get_weather("San Francisco")
    Model->>Tools: 并行调用 get_weather("New York")
    note over Tools: 分别获取两地天气数据

    %% 工具返回结果给模型
    Tools-->>Model: 返回旧金山天气数据（72°F sunny）
    Tools-->>Model: 返回纽约天气数据（68°F cloudy）
    note over Model: 整理工具返回结果，生成自然语言回复

    %% 模型向用户反馈最终结果
    Model-->>User: SF: 72°F sunny, NYC: 68°F cloudy

核心价值

1. 能力扩展：让模型可以访问外部 API、数据库、文件系统等，获取实时信息或执行操作，例如查询天气、查询数据库、读写文件等。
2. 结构化交互：通过明确的输入输出模式（如函数定义、参数规范），确保模型与外部系统交互的准确性，避免因理解偏差导致的调用失败。
3. 自主决策：让模型能够根据任务需求，自主决定何时调用工具、调用哪个工具，以及如何解析工具返回的结果，实现更智能的任务处理。
4. 任务自动化：使模型能够完成复杂的端到端任务，而不仅仅是生成文本。例如，自动规划行程、处理订单、分析数据等。

创建基本工具

使用 @tool 装饰器

# @Time：2025/12/15 09:20
# @Author：jinglv
import pymysql
from langchain.agents import create_agent
from langchain.tools import tool

from src.core.llms import model_client


# 第一步定义工具函数
@tool('数据库操作', description='连接数据库,执行sql语句，并返回执行的结果')
def mysql_executor(sql: str):
    """
    连接数据库，执行sql语句
    需要先安装pymysql
    :param sql:
    :return:
    """
    try:
        connect = pymysql.connect(host='127.0.0.1',
                                  port=3306,
                                  user='root',
                                  password='mysql',
                                  cursorclass=pymysql.cursors.DictCursor,
                                  db='agent',
                                  autocommit=True,
                                  )
        cursor = connect.cursor()
        cursor.execute(sql)
        connect.commit()
    except pymysql.err.OperationalError as e:
        return 0
    result = cursor.fetchall()
    cursor.close()
    connect.cursor()
    return result


# 创建模型并绑定工具
agent = create_agent(
    model=model_client,
    tools=[mysql_executor],
    system_prompt="""
    您是一个数据库操作助手
    您可以：
    1. 执行SQL创建、查询、更新和删除操作
    2. 提供详细的执行过程反馈
    请根据用户需求执行相应的数据库操作。
    """
)

重要说明:

• 类型提示必需：定义工具的输入模式
• 文档字符串重要：帮助模型理解工具用途
• 简洁明了：文档字符串应具备信息性且简洁

自定义工具属性

自定义工具名称

@tool("web_search")  # 自定义名称
def search(query: str) -> str:
    """在网络上搜索信息。"""
    return f"搜索结果: {query}"

print(search.name)  # 输出: web_search

自定义工具描述

@tool("web_search")  # 自定义名称
def search(query: str，description="用于网络搜索") -> str:
    """在网络上搜索信息。"""
    return f"搜索结果: {query}"

print(search.name)  # 输出: web_search

工具参数的声明

使用 Pydantic 模型定义复杂输入

# @Time：2025/12/15 09:20
# @Author：jinglv
import pymysql
from langchain.agents import create_agent
from langchain.tools import tool
from pydantic import BaseModel, Field

from src.core.llms import model_client


# 定义工具参数
class DataBaseConfig(BaseModel):
    host: str = Field(..., description="数据库地址")
    port: int = Field(..., description="数据库端口")
    user: str = Field(..., description="数据库用户名")
    password: str = Field(..., description="数据库密码")
    db: str = Field(..., description="数据库名称")


# 第一步定义工具函数
@tool('数据库操作', description='连接数据库,执行sql语句，并返回执行的结果')
def mysql_executor(sql: str, data_config: DataBaseConfig):
    """
    连接数据库，执行sql语句
    需要先安装pymysql
    :param sql:
    :return:
    """
    try:
        connect = pymysql.connect(host=data_config.host,
                                  port=data_config.port,
                                  user=data_config.user,
                                  password=data_config.password,
                                  cursorclass=pymysql.cursors.DictCursor,
                                  db=data_config.db,
                                  # 自动提交事物
                                  autocommit=True,
                                  )
        cursor = connect.cursor()
        cursor.execute(sql)
        connect.commit()
    except pymysql.err.OperationalError as e:
        return 0
    result = cursor.fetchall()
    cursor.close()
    connect.cursor()
    return result


@tool('获取数据库连接配置', description='获取数据库的连接参数')
def get_database_config() -> DataBaseConfig:
    """
    获取数据库连接参数
    :return:
    """
    return DataBaseConfig(
        host="127.0.0.1",
        port=3306,
        user="root",
        password="12345678",
        db="test"
    )


# 创建模型并绑定工具
agent = create_agent(
    model=model_client,
    tools=[get_database_config, mysql_executor],
    system_prompt="""
    您是一个数据库操作助手
    您可以：
    1. 执行SQL创建、查询、更新和删除操作
    2. 提供详细的执行过程反馈
    请根据用户需求执行相应的数据库操作。
    """
)

注意：

自定义工具参数的名称，以下参数名称是保留的，不能用作工具参数：

参数名称	用途
config	保留给内部工具 RunnableConfig
runtime	保留给参数 ToolRuntime

要访问运行时信息，请使用 ToolRuntime 参数。

访问运行时上下文

当工具能够访问代理状态、运行时上下文和长期记忆时，它们最为强大。这使得工具能够做出基于上下文的决策、个性化响应，并在对话之间保持信息。

运行时上下文提供了一种方式，可以在运行时将依赖项（如数据库连接、用户ID或配置）注入到工具中，这使得工具更具可测试性和可重用性。

工具可以通过ToolRuntime参数访问运行时信息，它提供以下内容：

• State(状态)：在执行过程中流动的可变数据（例如消息、计数器、自定义字段）。
• Context(上下文)：像用户ID、会话详情或应用程序特定配置这样的不可变配置。
• Store(存储)：跨对话的持久长期记忆。
• Stream Writer(流式写入器)：在工具执行时流式传输自定义更新。
• Config(配置)：执行的RunnableConfig。
• Tool Call ID(工具调用ID)：当前工具调用的ID。这段内容主要介绍了LangChain中工具（Tools）如何通过ToolRuntime参数访问运行时上下文信息，以增强工具的功能性和灵活性。

from langchain.tools import tool, ToolRuntime

@tool
def summarize_conversation(runtime: ToolRuntime) -> str:
    """总结当前对话历史。"""
    # 从 runtime.state 中获取对话历史消息
    messages = runtime.state["messages"]
    
    # 统计不同类型消息的数量
    human_msgs = sum(1 for m in messages if m.__class__.__name__ == "HumanMessage")
    ai_msgs = sum(1 for m in messages if m.__class__.__name__ == "AIMessage")
    tool_msgs = sum(1 for m in messages if m.__class__.__name__ == "ToolMessage")
    
    # 返回统计结果
    return f"对话包含{human_msgs}条用户消息，{ai_msgs}条AI回复，{tool_msgs}条工具结果"

工具访问运行时上下文的重要性

• 增强工具能力：当工具能够访问代理状态、运行时上下文和长期记忆时，它们可以基于上下文做出更智能的决策，提供个性化的响应，并在多次对话中保持信息的连贯性。
• 提高可测试性和可重用性：运行时上下文允许在运行时将依赖项（如数据库连接、用户ID或配置）注入到工具中，这使得工具在不同环境下更容易被测试和重用。

flowchart LR
    %% 定义容器（子图）
    subgraph ToolRuntimeContext [工具运行时上下文]
        direction LR
        ToolCall[工具调用] --> ToolRuntime[工具运行时]

        ToolRuntime --> ContextAccess[上下文访问]
        ToolRuntime --> StateAccess[状态访问]
        ToolRuntime --> StoreAccess[存储访问]
        ToolRuntime --> StreamWriter[流写入器]
    end

    subgraph AvailableResources [可用资源]
        SessionInfo[会话信息]
        UserID[用户ID]
        Messages[消息记录]
        CustomState[自定义状态]
        UserPreferences[用户偏好]
        LongtermMemory[长期记忆]
    end

    subgraph EnhancedToolCapabilities [增强的工具能力]
        ContextAwareTools[上下文感知工具]
        StatefulTools[有状态工具]
        MemoryEnabledTools[记忆赋能工具]
        StreamingTools[流式工具]
    end

    %% 连接各模块
    ContextAccess --> SessionInfo
    ContextAccess --> UserID
    StateAccess --> Messages
    StateAccess --> CustomState
    StoreAccess --> UserPreferences
    StoreAccess --> LongtermMemory

    SessionInfo --> ContextAwareTools
    UserID --> ContextAwareTools
    Messages --> ContextAwareTools
    CustomState --> StatefulTools
    UserPreferences --> MemoryEnabledTools
    LongtermMemory --> MemoryEnabledTools
    StreamWriter --> StreamingTools

    %% 样式优化（提升视觉辨识度）
    style ToolRuntimeContext fill:#fff8e1,stroke:#333,stroke-width:1px
    style AvailableResources fill:#fff8e1,stroke:#333,stroke-width:1px
    style EnhancedToolCapabilities fill:#fff8e1,stroke:#333,stroke-width:1px
    classDef boxStyle fill:#e3eafd,stroke:#333,stroke-width:1px
    class ToolCall,ToolRuntime,ContextAccess,StateAccess,StoreAccess,StreamWriter,SessionInfo,UserID,Messages,CustomState,UserPreferences,LongtermMemory,ContextAwareTools,StatefulTools,MemoryEnabledTools,StreamingTools boxStyle

上下文数据

ToolRuntime：这是一个工具运行时的上下文环境或参数，它为工具提供了一个统一的方式来访问所有运行时的信息。

# @Time：2025/12/15 09:20
# @Author：jinglv
from dataclasses import dataclass

import pymysql
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from pydantic import BaseModel, Field

from src.core.llms import model_client


# 定义工具参数
class DataBaseConfig(BaseModel):
    host: str = Field(..., description="数据库地址")
    port: int = Field(..., description="数据库端口")
    user: str = Field(..., description="数据库用户名")
    password: str = Field(..., description="数据库密码")
    db: str = Field(..., description="数据库名称")


# 第一步定义工具函数
@tool('数据库操作', description='连接数据库,执行sql语句，并返回执行的结果')
def mysql_executor(sql: str, data_config: DataBaseConfig):
    """
    连接数据库，执行sql语句
    需要先安装pymysql
    :param sql:
    :return:
    """
    try:
        connect = pymysql.connect(host=data_config.host,
                                  port=data_config.port,
                                  user=data_config.user,
                                  password=data_config.password,
                                  cursorclass=pymysql.cursors.DictCursor,
                                  db=data_config.db,
                                  # 自动提交事物
                                  autocommit=True,
                                  )
        cursor = connect.cursor()
        cursor.execute(sql)
        connect.commit()
    except pymysql.err.OperationalError as e:
        return 0
    result = cursor.fetchall()
    cursor.close()
    connect.cursor()
    return result


# 定义上下文类型
@dataclass
class AgentRunContext:
    user_id: str


@tool('获取数据库连接配置', description='获取数据库的连接参数')
def get_database_config(runtime: ToolRuntime[AgentRunContext]) -> DataBaseConfig:
    """
    获取数据库连接参数
    :return:
    """
    print("user_id", runtime.context.user_id)
    return DataBaseConfig(
        host="127.0.0.1",
        port=3306,
        user="root",
        password="12345678",
        db="test"
    )


# 创建模型并绑定工具
agent = create_agent(
    model=model_client,
    tools=[get_database_config, mysql_executor],
    context_schema=AgentRunContext,
    system_prompt="""
    您是一个数据库操作助手
    您可以：
    1. 执行SQL创建、查询、更新和删除操作
    2. 提供详细的执行过程反馈
    请根据用户需求执行相应的数据库操作。
    """
)

result = agent.stream(
    {"messages": [{"role": "user", "content": "用户获取数据的连接时什么？"}]},
    # 通过上下文传递参数
    context=AgentRunContext(user_id="user123")
)

ToolRuntime的核心工作原理：

• 类型化上下文：通过 @dataclass 定义结构化数据模型，避免无类型的字典传递。
• 泛型声明：工具函数使用 ToolRuntime[ContextType] 声明所需上下文类型，实现编译期类型检查。
• 自动注入：运行时自动注入正确类型的上下文对象，通过 runtime.context 安全访问，无需手动传递。
• Agent 集成：创建 Agent 时指定 context_schema，调用时传入具体实例，实现上下文与工具的无缝衔接。

存储访问

通过记忆可以访问跨对话的持久数据。存储可通过以下方式访问，并允许您保存和检索用户特定或应用特定数据。runtime.store

# @Time：2025/12/15 09:20
# @Author：jinglv
from dataclasses import dataclass

import pymysql
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langgraph.store.memory import InMemoryStore
from pydantic import BaseModel, Field

from src.core.llms import model_client


# 定义工具参数
class DataBaseConfig(BaseModel):
    host: str = Field(..., description="数据库地址")
    port: int = Field(..., description="数据库端口")
    user: str = Field(..., description="数据库用户名")
    password: str = Field(..., description="数据库密码")
    db: str = Field(..., description="数据库名称")


# 第一步定义工具函数
@tool('数据库操作', description='连接数据库,执行sql语句，并返回执行的结果')
def mysql_executor(sql: str, data_config: DataBaseConfig):
    """
    连接数据库，执行sql语句
    需要先安装pymysql
    :param sql:
    :return:
    """
    try:
        connect = pymysql.connect(host=data_config.host,
                                  port=data_config.port,
                                  user=data_config.user,
                                  password=data_config.password,
                                  cursorclass=pymysql.cursors.DictCursor,
                                  db=data_config.db,
                                  # 自动提交事物
                                  autocommit=True,
                                  )
        cursor = connect.cursor()
        cursor.execute(sql)
        connect.commit()
    except pymysql.err.OperationalError as e:
        return 0
    result = cursor.fetchall()
    cursor.close()
    connect.cursor()
    return result


@tool('获取数据库连接配置', description='获取数据库的连接参数')
def get_database_config() -> DataBaseConfig:
    """
    获取数据库连接参数
    :return:
    """
    print("user_id", runtime.context.user_id)
    return DataBaseConfig(
        host="127.0.0.1",
        port=3306,
        user="root",
        password="12345678",
        db="test"
    )


# 定义上下文类型
@dataclass
class AgentRunContext:
    user_id: str


@tool('获取当前用户详情', description='获取当前的用户id')
def get_user_id(runtime: ToolRuntime[AgentRunContext]) -> str:
    """
    获取当前用户id
    :return:
    """
    return runtime.context.user_id


@tool("存储对话的记录", description="保存一个agent对话的记录")
def save_conversation(runtime: ToolRuntime[AgentRunContext]):
    # 获取存储对象
    store = runtime.store
    user_id = runtime.context.user_id
    message = runtime.state['message']
    store.put(('user_id', user_id), user_id, message)
    return "保存成功"


@tool("读取历史对话记录", description="读取历史对话记录")
def read_history_conversation(runtime: ToolRuntime[AgentRunContext]):
    # 获取存储对象
    store = runtime.store
    user_id = runtime.context.user_id
    info_list = store.get(('user_id', user_id), user_id)
    return info_list


# 初始化store对象
store = InMemoryStore()

# 创建模型并绑定工具
agent = create_agent(
    model=model_client,
    tools=[get_database_config, mysql_executor, get_user_id, save_conversation, read_history_conversation],
    context_schema=AgentRunContext,
    store=store,  # 绑定store
    system_prompt="""
    您是一个数据库操作助手
    您可以：
    1. 执行SQL创建、查询、更新和删除操作
    2. 提供详细的执行过程反馈
    3. 保存和读取对话记录
    请根据用户需求执行相应的数据库操作。
    """
)

result = agent.stream(
    {"messages": [{"role": "user", "content": "查询users表中的详情"}]},
    # 通过上下文传递参数
    context=AgentRunContext(user_id="user123")
)

流式写入器

当工具运行时，使用“runtime.stream_writer”来流式传输来自工具的自定义更新。这有助于向用户实时反馈工具正在执行的操作。

from langchain.tools import tool, ToolRuntime

@tool
def get_weather(city: str, runtime: ToolRuntime) -> str:
    """获取指定城市的天气。"""
    writer = runtime.stream_writer
    
    # 在工具执行时流式传输自定义更新
    writer(f"正在查找城市数据: {city}")
    writer(f"已获取城市数据: {city}")
    
    return f"{city} 的天气总是晴朗的！"

更新会话状态

• Command 对象：来自 langgraph.types，是一种特殊的返回类型，用于向执行引擎（如 LangGraph）发出指令，以更新状态或控制流程。
• 工具返回 Command：当工具函数返回 Command 时，它不再只是返回文本，而是告诉系统 “执行以下操作”。

from langgraph.types import Command
from langchain.tools import tool, ToolRuntime

@tool
def update_user_name(new_name: str, runtime: ToolRuntime) -> Command:
    """更新用户的名称。"""
    # 返回一个 Command，指示更新状态中的 "user_name" 字段
    return Command(update={"user_name": new_name})

• 工具接收 new_name 参数。
• 返回 Command(update={"user_name": new_name})，让系统直接更新会话状态中的 user_name 字段。

@tool
def clear_conversation() -> Command:
    """清除对话历史。"""
    from langchain.messages import RemoveMessage
    from langchain.graph.message import REMOVE_ALL_MESSAGES
    
    # 返回一个 Command，指示移除所有消息
    return Command(
        update={
            "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)],
        }
    )

• 使用 RemoveMessage 和 REMOVE_ALL_MESSAGES 常量。
• 返回 Command，让系统清空对话历史（messages 列表）。

工具的最佳实战

1. 工具描述最佳实践

核心目标是让模型和开发者都能清晰理解工具的用途和用法：

• 编写详细清晰的函数文档字符串：用自然语言描述工具的功能，让模型能准确判断何时调用。
• 为每个参数添加准确的描述：明确参数的含义、类型和约束，避免模型生成无效参数。
• 说明工具的预期用途和返回值：让模型知道调用后能得到什么，从而更好地规划下一步。
• 包含示例输入和输出：提供具体示例，帮助模型理解正确的调用格式和结果格式。

2. 安全性考虑

核心目标是防止工具被滥用或恶意利用：

• 对工具输入进行验证和清理：校验参数格式，过滤危险字符，防止注入攻击。
• 设置合理的权限和限制：限制工具可执行的操作范围，例如只读操作或特定用户数据访问。
• 避免在工具中执行未经过滤的用户输入：不直接执行用户提供的代码或命令，防止远程代码执行。
• 实施访问控制和审计日志：记录工具调用的用户、时间和参数，便于事后审计和问题追溯。

3. 错误处理

核心目标是让工具在失败时也能稳定、可调试：

• 捕获并优雅地处理异常：避免工具崩溃导致整个 Agent 流程中断。
• 提供有用的错误消息：返回清晰的错误信息，帮助模型理解失败原因并尝试修复。
• 实现超时机制：防止长时间运行的操作阻塞对话，提升用户体验。
• 记录详细的错误日志：记录完整的错误栈和上下文，便于开发者排查问题。

4. 性能优化

核心目标是让工具高效、可扩展：

• 缓存频繁使用的工具结果：对重复查询（如天气、股票）进行缓存，减少重复调用。
• 异步执行长时间运行的操作：将耗时任务（如文件处理、数据库查询）转为异步，避免阻塞对话。
• 优化工具的执行路径：减少不必要的步骤，提升单次调用的响应速度。
• 实现批量处理机制：支持批量操作，减少网络或 IO 开销。