工具定义与集成

工具的核心概念

在LangChain中，工具是代理可以调用的任何函数或服务。工具通过让模型访问外部信息和能力，大大扩展其功能范围。

工具的三个关键要素

1. 函数实现

def get_weather(city: str) -> str:
    """实现具体功能"""
    # 调用天气API
    api_url = f"https://api.weather.com/forecast?city={city}"
    response = requests.get(api_url)
    return response.json()["weather"]

2. 类型注解

def calculate(operation: str, a: int, b: int) -> int:
    """类型注解让模型理解参数"""
    if operation == "add":
        return a + b
    elif operation == "multiply":
        return a * b

3. 文档字符串（Docstring）

def search_knowledge_base(query: str, limit: int = 10) -> list:
    """在知识库中搜索相关信息
    
    这个文档字符串会被发送给模型，帮助其理解工具的用途。
    
    Args:
        query: 搜索关键词
        limit: 返回结果数上限，默认为10
        
    Returns:
        匹配的文档列表，每个文档包含标题和内容
    """
    pass

创建工具的方法

方法1：使用@tool装饰器（推荐）

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息
    
    Args:
        city: 城市名称
    """
    return f"{city}目前是晴天，温度25°C"

# 创建代理时使用
agent = create_agent(
    model="anthropic:claude-sonnet-4",
    tools=[get_weather]  # 直接传入工具
)

方法2：使用Tool类

from langchain_core.tools import Tool

def weather_func(city: str) -> str:
    return f"{city}目前是晴天"

tool = Tool(
    name="weather",
    description="获取天气信息",
    func=weather_func
)

方法3：使用Pydantic模型（精确控制）

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

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    units: str = Field(
        default="celsius",
        description="温度单位：celsius或fahrenheit"
    )

@tool
def get_weather_advanced(params: WeatherInput) -> str:
    """获取天气信息，支持自定义温度单位"""
    # 实现逻辑
    pass

常见工具模式

1. API查询工具

@tool
def search_web(query: str, max_results: int = 10) -> list:
    """在网络上搜索信息
    
    Args:
        query: 搜索关键词
        max_results: 最大结果数
    """
    import requests
    
    try:
        response = requests.get(
            "https://api.search.com/search",
            params={"q": query, "limit": max_results},
            timeout=10
        )
        return response.json()["results"]
    except Exception as e:
        return [{"error": str(e)}]

2. 数据库查询工具

@tool
def query_database(sql: str) -> list:
    """执行数据库查询
    
    Args:
        sql: SQL查询语句
    """
    import sqlite3
    
    # 安全考虑：验证SQL语句
    if "DROP" in sql.upper() or "DELETE" in sql.upper():
        return [{"error": "Dangerous SQL operations not allowed"}]
    
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()
    
    try:
        cursor.execute(sql)
        results = cursor.fetchall()
        return results
    except Exception as e:
        return [{"error": str(e)}]
    finally:
        conn.close()

3. 文件操作工具

@tool
def read_file(file_path: str, encoding: str = "utf-8") -> str:
    """读取文件内容
    
    Args:
        file_path: 文件路径
        encoding: 文件编码方式
    """
    try:
        with open(file_path, "r", encoding=encoding) as f:
            return f.read()
    except FileNotFoundError:
        return "文件不存在"
    except Exception as e:
        return f"错误: {str(e)}"

4. 计算工具

@tool
def evaluate_expression(expression: str) -> str:
    """计算数学表达式
    
    Args:
        expression: 数学表达式，例如 "2+3*4"
    """
    try:
        # 注意：eval有安全风险，实际应用中应使用更安全的方式
        result = eval(expression)
        return f"结果: {result}"
    except SyntaxError:
        return "表达式语法错误"
    except Exception as e:
        return f"计算失败: {e}"

5. 外部服务集成工具

from anthropic import Anthropic

@tool
def get_user_preferences(user_id: str) -> dict:
    """从用户服务获取用户偏好
    
    Args:
        user_id: 用户ID
    """
    # 调用用户服务API
    import requests
    
    try:
        response = requests.get(
            f"https://user-service.com/api/users/{user_id}/preferences",
            timeout=5
        )
        return response.json()
    except Exception as e:
        return {"error": str(e)}

工具的高级特性

1. 工具结果转换

@tool
def format_data(raw_data: str) -> str:
    """处理和格式化数据"""
    import json
    
    # 解析JSON
    data = json.loads(raw_data)
    
    # 格式化为易读形式
    formatted = json.dumps(data, indent=2, ensure_ascii=False)
    return formatted

2. 工具链式调用

@tool
def chain_tools(source: str, transformation: str) -> str:
    """链式调用多个工具"""
    
    # 第一步：获取数据
    raw_data = get_data(source)
    
    # 第二步：转换数据
    transformed = transform_data(raw_data, transformation)
    
    # 第三步：验证结果
    validated = validate_result(transformed)
    
    return validated

3. 异步工具

import asyncio

@tool
async def fetch_multiple_urls(urls: list) -> dict:
    """并发获取多个URL的内容
    
    Args:
        urls: URL列表
    """
    import aiohttp
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return {url: content for url, content in zip(urls, results)}

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

工具的安全性

1. 参数验证

from pydantic import BaseModel, validator

class FileOperationInput(BaseModel):
    file_path: str
    
    @validator("file_path")
    def validate_path(cls, v):
        # 确保路径在允许的目录中
        allowed_dirs = ["/data", "/uploads"]
        for allowed_dir in allowed_dirs:
            if v.startswith(allowed_dir):
                return v
        raise ValueError(f"不允许访问 {v}")

@tool
def safe_read_file(params: FileOperationInput) -> str:
    """安全地读取文件"""
    with open(params.file_path, "r") as f:
        return f.read()

2. 权限检查

def require_admin(func):
    """装饰器：检查管理员权限"""
    def wrapper(user_id: str, *args, **kwargs):
        # 检查用户权限
        if not is_admin(user_id):
            raise PermissionError(f"用户 {user_id} 没有权限执行此操作")
        return func(*args, **kwargs)
    return wrapper

@tool
@require_admin
def delete_user(user_id: str) -> str:
    """删除用户（需要管理员权限）"""
    # 实现删除逻辑
    pass

3. 速率限制

from functools import wraps
import time

def rate_limit(max_calls: int, time_window: int):
    """装饰器：限制调用频率"""
    def decorator(func):
        calls = []
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            # 删除时间窗口外的调用
            calls[:] = [call for call in calls if call > now - time_window]
            
            if len(calls) >= max_calls:
                raise RuntimeError(f"超过速率限制：{max_calls}/{time_window}秒")
            
            calls.append(now)
            return func(*args, **kwargs)
        
        return wrapper
    return decorator

@tool
@rate_limit(max_calls=10, time_window=60)
def api_call() -> str:
    """每分钟最多10次调用"""
    pass

工具的错误处理

1. 优雅的失败

@tool
def robust_tool(query: str) -> str:
    """处理各种异常情况的工具"""
    try:
        # 主要逻辑
        result = perform_operation(query)
        return result
    
    except TimeoutError:
        return "操作超时，请稍后重试"
    
    except ValueError:
        return "输入参数无效，请检查格式"
    
    except Exception as e:
        # 记录错误
        logger.error(f"未预期的错误: {e}")
        return f"处理失败: {e}"

2. 重试机制

from tenacity import retry, stop_after_attempt, wait_exponential

@tool
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def unreliable_tool(query: str) -> str:
    """会重试失败的工具"""
    # 可能失败的操作
    response = requests.get("https://unstable-api.com", timeout=5)
    return response.json()

工具的组织

1. 工具分组

# tools.py
from langchain.tools import tool

# 天气相关工具
@tool
def get_weather(city: str) -> str:
    """获取天气"""
    pass

@tool
def get_forecast(city: str, days: int) -> str:
    """获取天气预报"""
    pass

# 数据库相关工具
@tool
def query_users(name: str) -> list:
    """查询用户"""
    pass

@tool
def update_user(user_id: str, **fields) -> bool:
    """更新用户信息"""
    pass

2. 动态工具加载

def get_tools_for_agent(agent_type: str) -> list:
    """根据代理类型加载不同的工具"""
    
    basic_tools = [get_weather, search_web]
    
    if agent_type == "admin":
        return basic_tools + [delete_user, update_system]
    
    elif agent_type == "user":
        return basic_tools + [get_preferences]
    
    else:
        return basic_tools

# 使用
agent_type = request.headers.get("X-Agent-Type", "user")
tools = get_tools_for_agent(agent_type)
agent = create_agent(model=model, tools=tools)

测试工具

def test_weather_tool():
    """测试天气工具"""
    result = get_weather("北京")
    assert "天气" in result
    assert "温度" in result

def test_tool_with_agent():
    """测试工具与代理的集成"""
    agent = create_agent(
        model="anthropic:claude-sonnet-4",
        tools=[get_weather]
    )
    
    result = agent.invoke({
        "messages": [{"role": "user", "content": "北京天气如何?"}]
    })
    
    # 验证代理确实使用了工具
    assert any(msg.get("name") == "get_weather" for msg in result["messages"])

常见问题

Q: 为什么模型不调用我的工具？ A: 检查工具的描述是否清晰，参数文档是否完整。改进docstring通常能解决问题。

Q: 如何处理工具执行超时？ A: 在工具中设置超时参数，使用try-except捕获timeout异常。

Q: 可以有工具依赖其他工具吗？ A: 可以，但要小心递归问题和循环依赖。最好使用LangGraph来管理复杂的工作流。