3.4 Tools

本节介绍 LangChain 中的工具定义和使用。

---

什么是 Tools？

Tools（工具） 是 Agent 可以调用的组件，用于执行特定操作。每个工具封装了一个可调用函数及其输入 schema，让模型能够决定何时调用以及传入什么参数。

---

基本工具定义

使用 @tool 装饰器是最简单的方式：

from langchain_core.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """搜索客户数据库中匹配的记录。

    Args:
        query: 搜索关键词
        limit: 返回结果数量限制
    """
    return f"找到 {limit} 条关于 '{query}' 的结果"

重要：类型提示是必需的，它们定义了工具的输入 schema。

---

工具命名

默认命名

默认使用函数名：

@tool
def get_weather(city: str) -> str:
    """获取天气"""
    return f"{city}：晴"

print(get_weather.name)  # "get_weather"

自定义命名

@tool("weather_lookup")
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    return f"{city}：晴"

print(get_weather.name)  # "weather_lookup"

---

工具描述

模型通过描述理解何时使用工具：

@tool
def calculate(expression: str) -> str:
    """计算数学表达式。

    适用于需要精确计算的场景，如：
    - 基本运算：1 + 2 * 3
    - 复杂表达式：(100 - 20) / 4

    不适用于：
    - 近似估算
    - 统计分析
    """
    return str(eval(expression))

---

复杂输入 Schema

使用 Pydantic 定义复杂输入：

from pydantic import BaseModel, Field
from typing import Literal
from langchain_core.tools import tool

class WeatherInput(BaseModel):
    """天气查询输入参数"""
    location: str = Field(description="城市名称或坐标")
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="温度单位"
    )
    include_forecast: bool = Field(
        default=False,
        description="是否包含未来预报"
    )

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
    """获取当前天气和可选的预报信息"""
    result = f"{location}：25度"
    if include_forecast:
        result += "，明天多云"
    return result

---

访问运行时上下文

使用 ToolRuntime 访问状态、配置和流式写入器：

from langchain_core.tools import tool
from langchain.agents import ToolRuntime

@tool
def summarize_conversation(runtime: ToolRuntime) -> str:
    """总结目前的对话内容"""
    # 访问对话消息
    messages = runtime.state["messages"]

    # 访问配置
    config = runtime.config

    # 处理消息
    summary = f"对话共 {len(messages)} 条消息"
    return summary

runtime 参数对模型不可见，不会出现在工具 schema 中。

---

保留参数名

以下参数名是保留的：

参数名	用途
config	运行时配置
runtime	ToolRuntime 对象

不要将它们用作工具的普通参数。

---

异步工具

@tool
async def fetch_data(url: str) -> str:
    """从 URL 获取数据"""
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

---

工具返回类型

返回字符串

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

返回字典

@tool
def structured_tool(query: str) -> dict:
    """返回结构化数据"""
    return {
        "query": query,
        "results": ["结果1", "结果2"],
        "count": 2
    }

返回列表

@tool
def list_tool(category: str) -> list[str]:
    """返回列表"""
    return ["项目1", "项目2", "项目3"]

---

错误处理

在工具内处理

@tool
def safe_divide(a: float, b: float) -> str:
    """安全除法"""
    try:
        result = a / b
        return f"结果: {result}"
    except ZeroDivisionError:
        return "错误: 不能除以零"
    except Exception as e:
        return f"错误: {str(e)}"

返回错误让模型重试

@tool
def validate_input(data: str) -> str:
    """验证输入数据"""
    if not data.strip():
        raise ValueError("输入不能为空，请提供有效数据")
    return f"验证通过: {data}"

---

流式更新

在工具执行中发送实时更新：

from langchain.agents import ToolRuntime, get_stream_writer

@tool
def long_running_task(task: str, runtime: ToolRuntime) -> str:
    """执行长时间运行的任务"""
    writer = get_stream_writer()

    writer(f"开始处理: {task}")
    # 第一步
    writer("步骤 1/3: 收集数据...")
    # 第二步
    writer("步骤 2/3: 分析数据...")
    # 第三步
    writer("步骤 3/3: 生成报告...")

    return "任务完成"

---

工具组合使用

from langchain.agents import create_agent

@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"关于 {query} 的搜索结果..."

@tool
def calculate(expression: str) -> str:
    """数学计算"""
    return str(eval(expression))

@tool
def translate(text: str, target_lang: str) -> str:
    """翻译文本"""
    return f"[翻译成{target_lang}]: {text}"

# 创建带有多个工具的 Agent
agent = create_agent(
    "gpt-4o",
    tools=[search, calculate, translate],
    system_prompt="你是一个多功能助手"
)

---

绑定工具到模型

直接将工具绑定到模型（不使用 Agent）：

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

# 绑定工具
model_with_tools = model.bind_tools([search, calculate])

# 调用
response = model_with_tools.invoke("100 * 25 等于多少？")

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

---

工具执行

手动执行工具调用：

# 获取模型的工具调用请求
response = model_with_tools.invoke("搜索最新的 AI 新闻")

# 执行工具
for call in response.tool_calls:
    tool_name = call["name"]
    tool_args = call["args"]

    if tool_name == "search":
        result = search.invoke(tool_args)
        print(f"结果: {result}")

---

最佳实践

实践	说明
清晰的描述	让模型理解何时使用工具
完整的类型提示	定义准确的输入 schema
适当的错误处理	返回有意义的错误信息
单一职责	每个工具只做一件事
有意义的返回值	返回模型能理解的结果

---

常见工具模式

# 1. 数据获取工具
@tool
def get_user_info(user_id: str) -> dict:
    """获取用户信息"""
    return {"id": user_id, "name": "张三", "email": "zhang@example.com"}

# 2. 操作执行工具
@tool
def send_notification(user_id: str, message: str) -> str:
    """发送通知给用户"""
    return f"已发送通知给用户 {user_id}"

# 3. 计算工具
@tool
def analyze_data(data: list[float]) -> dict:
    """分析数据"""
    return {
        "mean": sum(data) / len(data),
        "max": max(data),
        "min": min(data)
    }

# 4. 验证工具
@tool
def validate_email(email: str) -> bool:
    """验证邮箱格式"""
    import re
    pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
    return bool(re.match(pattern, email))

---

上一节：3.3 Messages

下一节：3.5 Short-term Memory