工具使用

大语言模型只能生成文本，无法直接执行操作。工具（Tools）功能让 AI 能够调用外部函数、API 和服务，大大扩展了应用的能力。

什么是工具

工具是 AI 可以调用的函数，用于：

  • 🔍 搜索网络
  • 🧮 执行计算
  • 📊 查询数据库
  • 📧 发送邮件
  • 🌐 调用 API
  • 📁 操作文件

工具的工作原理

用户问题 → LLM 判断是否需要工具 → 选择工具 → 执行工具 → 将结果返回 LLM → 生成最终回答

创建自定义工具

使用 @tool 装饰器

最简单的创建方式：

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """将两个数字相乘。
    
    Args:
        a: 第一个数字
        b: 第二个数字
    """
    return a * b

# 工具信息
print(f"工具名称: {multiply.name}")
print(f"工具描述: {multiply.description}")
print(f"参数: {multiply.args}")

# 直接调用
result = multiply.invoke({"a": 4, "b": 5})
print(f"结果: {result}")  # 20

使用 StructuredTool

更灵活的工具定义：

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

class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, description="最大结果数")

def search_func(query: str, max_results: int = 5) -> str:
    """模拟搜索功能"""
    return f"搜索 '{query}' 的前 {max_results} 条结果..."

search_tool = StructuredTool.from_function(
    func=search_func,
    name="web_search",
    description="搜索网络获取信息",
    args_schema=SearchInput
)

异步工具

import asyncio
from langchain_core.tools import tool

@tool
async def async_fetch(url: str) -> str:
    """异步获取网页内容"""
    # 模拟异步请求
    await asyncio.sleep(1)
    return f"从 {url} 获取的内容..."

# 异步调用
result = await async_fetch.ainvoke({"url": "https://example.com"})

内置工具

搜索工具

# DuckDuckGo 搜索
from langchain_community.tools import DuckDuckGoSearchRun

search = DuckDuckGoSearchRun()
result = search.invoke("LangChain 是什么")
print(result)

# Tavily 搜索（推荐，需要 API Key）
from langchain_community.tools.tavily_search import TavilySearchResults

search = TavilySearchResults(max_results=3)
results = search.invoke("最新的 AI 新闻")

数学计算

from langchain_community.tools import LLMMathChain
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo")
math_tool = LLMMathChain.from_llm(llm)

result = math_tool.invoke("计算 123 * 456 + 789")
print(result)

Python 代码执行

from langchain_experimental.tools import PythonREPLTool

python_tool = PythonREPLTool()

code = """
import math
result = sum([math.sqrt(i) for i in range(1, 11)])
print(f"1到10的平方根之和: {result:.2f}")
"""

result = python_tool.invoke(code)
print(result)

安全警告

PythonREPLTool 会执行任意 Python 代码，在生产环境中使用需要特别注意安全性。

维基百科查询

from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
result = wikipedia.invoke("人工智能")
print(result)

工具与模型绑定

使用 bind_tools

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    # 模拟天气数据
    weather_data = {
        "北京": "晴天，25°C",
        "上海": "多云，22°C",
        "广州": "小雨，28°C"
    }
    return weather_data.get(city, "未找到该城市的天气信息")

@tool
def get_time(timezone: str = "Asia/Shanghai") -> str:
    """获取指定时区的当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 绑定工具
llm = ChatOpenAI(model="gpt-3.5-turbo")
llm_with_tools = llm.bind_tools([get_weather, get_time])

# 调用
response = llm_with_tools.invoke("北京现在的天气怎么样？")
print(response)

解析工具调用

from langchain_core.messages import HumanMessage

response = llm_with_tools.invoke([
    HumanMessage(content="北京的天气和当前时间是多少？")
])

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

自动工具执行

完整的工具调用流程

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def add(a: int, b: int) -> int:
    """加法运算"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """乘法运算"""
    return a * b

tools = [add, multiply]
tool_map = {t.name: t for t in tools}

llm = ChatOpenAI(model="gpt-3.5-turbo")
llm_with_tools = llm.bind_tools(tools)

def run_with_tools(query: str) -> str:
    """运行带工具的查询"""
    messages = [HumanMessage(content=query)]
    
    while True:
        response = llm_with_tools.invoke(messages)
        messages.append(response)
        
        # 如果没有工具调用，返回结果
        if not response.tool_calls:
            return response.content
        
        # 执行工具调用
        for tool_call in response.tool_calls:
            tool = tool_map[tool_call["name"]]
            result = tool.invoke(tool_call["args"])
            
            # 添加工具结果
            messages.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"]
            ))

# 测试
result = run_with_tools("计算 (3 + 5) * 7")
print(result)

工具链

创建工具执行链

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_core.output_parsers import StrOutputParser

@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"关于'{query}'的搜索结果：这是一些相关信息..."

@tool
def summarize(text: str) -> str:
    """总结文本"""
    return f"摘要：{text[:100]}..."

# 创建链
llm = ChatOpenAI(model="gpt-3.5-turbo")

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手，可以使用工具来帮助回答问题。"),
    ("human", "{input}")
])

# 绑定工具
llm_with_tools = llm.bind_tools([search, summarize])

chain = prompt | llm_with_tools

实用工具示例

天气查询工具

import requests
from langchain_core.tools import tool

@tool
def get_weather_real(city: str) -> str:
    """获取真实的天气信息（需要 API Key）"""
    api_key = "your_api_key"
    url = f"http://api.weatherapi.com/v1/current.json?key={api_key}&q={city}&lang=zh"
    
    try:
        response = requests.get(url)
        data = response.json()
        
        return f"""
城市: {data['location']['name']}
天气: {data['current']['condition']['text']}
温度: {data['current']['temp_c']}°C
体感温度: {data['current']['feelslike_c']}°C
湿度: {data['current']['humidity']}%
"""
    except Exception as e:
        return f"获取天气失败: {str(e)}"

数据库查询工具

import sqlite3
from langchain_core.tools import tool

@tool
def query_database(sql: str) -> str:
    """执行 SQL 查询（仅限 SELECT）"""
    if not sql.strip().upper().startswith("SELECT"):
        return "安全限制：只允许执行 SELECT 查询"
    
    try:
        conn = sqlite3.connect("database.db")
        cursor = conn.execute(sql)
        results = cursor.fetchall()
        conn.close()
        
        if not results:
            return "查询无结果"
        
        return str(results)
    except Exception as e:
        return f"查询错误: {str(e)}"

文件操作工具

from langchain_core.tools import tool
import os

@tool
def read_file(filepath: str) -> str:
    """读取文件内容"""
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            content = f.read()
        return content[:1000] + "..." if len(content) > 1000 else content
    except FileNotFoundError:
        return f"文件不存在: {filepath}"
    except Exception as e:
        return f"读取错误: {str(e)}"

@tool
def list_files(directory: str = ".") -> str:
    """列出目录中的文件"""
    try:
        files = os.listdir(directory)
        return "\n".join(files)
    except Exception as e:
        return f"错误: {str(e)}"

完整示例：智能助手

"""
智能助手 - 工具使用示例
"""

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from datetime import datetime
import math
from dotenv import load_dotenv

load_dotenv()

# 定义工具
@tool
def calculator(expression: str) -> str:
    """计算数学表达式。支持基本运算和数学函数（如 sin, cos, sqrt）。
    
    Args:
        expression: 要计算的数学表达式，如 "2 + 3 * 4" 或 "sqrt(16)"
    """
    try:
        # 安全的数学计算
        allowed = {
            'sqrt': math.sqrt,
            'sin': math.sin,
            'cos': math.cos,
            'tan': math.tan,
            'log': math.log,
            'pow': pow,
            'abs': abs,
            'pi': math.pi,
            'e': math.e
        }
        result = eval(expression, {"__builtins__": {}}, allowed)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def get_current_time() -> str:
    """获取当前日期和时间"""
    now = datetime.now()
    return now.strftime("当前时间：%Y年%m月%d日 %H:%M:%S，星期%w").replace("星期0", "星期日")

@tool
def unit_converter(value: float, from_unit: str, to_unit: str) -> str:
    """单位转换。支持长度、重量、温度等单位转换。
    
    Args:
        value: 要转换的数值
        from_unit: 原单位（如 km, m, kg, g, celsius, fahrenheit）
        to_unit: 目标单位
    """
    conversions = {
        ("km", "m"): lambda x: x * 1000,
        ("m", "km"): lambda x: x / 1000,
        ("kg", "g"): lambda x: x * 1000,
        ("g", "kg"): lambda x: x / 1000,
        ("celsius", "fahrenheit"): lambda x: x * 9/5 + 32,
        ("fahrenheit", "celsius"): lambda x: (x - 32) * 5/9,
        ("m", "feet"): lambda x: x * 3.28084,
        ("feet", "m"): lambda x: x / 3.28084,
    }
    
    key = (from_unit.lower(), to_unit.lower())
    if key in conversions:
        result = conversions[key](value)
        return f"{value} {from_unit} = {result:.2f} {to_unit}"
    else:
        return f"不支持从 {from_unit} 到 {to_unit} 的转换"

@tool
def word_counter(text: str) -> str:
    """统计文本的字数、词数和行数。
    
    Args:
        text: 要统计的文本
    """
    chars = len(text)
    words = len(text.split())
    lines = len(text.split('\n'))
    
    return f"字符数: {chars}\n词数: {words}\n行数: {lines}"


class SmartAssistant:
    """智能助手类"""
    
    def __init__(self):
        self.tools = [calculator, get_current_time, unit_converter, word_counter]
        self.tool_map = {t.name: t for t in self.tools}
        
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
        self.llm_with_tools = self.llm.bind_tools(self.tools)
        
        self.system_prompt = """你是一个智能助手，可以使用以下工具帮助用户：

1. calculator - 数学计算
2. get_current_time - 获取当前时间
3. unit_converter - 单位转换
4. word_counter - 文本统计

当用户的问题需要使用工具时，请调用相应的工具。
回答要简洁明了。"""
        
        self.history = []
    
    def chat(self, message: str) -> str:
        """处理用户消息"""
        # 构建消息列表
        messages = [
            {"role": "system", "content": self.system_prompt}
        ] + self.history + [
            {"role": "user", "content": message}
        ]
        
        # 循环直到获得最终回答
        while True:
            response = self.llm_with_tools.invoke(messages)
            
            # 如果没有工具调用，返回结果
            if not response.tool_calls:
                # 更新历史
                self.history.append({"role": "user", "content": message})
                self.history.append({"role": "assistant", "content": response.content})
                return response.content
            
            # 执行工具调用
            messages.append(response)
            
            for tool_call in response.tool_calls:
                tool = self.tool_map[tool_call["name"]]
                result = tool.invoke(tool_call["args"])
                
                messages.append(ToolMessage(
                    content=str(result),
                    tool_call_id=tool_call["id"]
                ))
    
    def clear_history(self):
        """清除历史"""
        self.history = []


def main():
    assistant = SmartAssistant()
    
    print("=" * 50)
    print("  智能助手（支持工具调用）")
    print("  可用工具：计算器、时间查询、单位转换、文本统计")
    print("  输入 '退出' 结束对话")
    print("=" * 50)
    
    while True:
        user_input = input("\n你：").strip()
        
        if not user_input:
            continue
        
        if user_input.lower() in ['退出', 'quit', 'exit']:
            print("\n再见！")
            break
        
        response = assistant.chat(user_input)
        print(f"\n助手：{response}")


if __name__ == "__main__":
    main()

小结

本章介绍了：

✅ 工具的概念和工作原理
✅ 创建自定义工具的多种方式
✅ LangChain 内置工具的使用
✅ 工具与模型的绑定
✅ 自动工具执行流程
✅ 实用工具示例
✅ 完整的智能助手实现

下一步

掌握工具使用后，让我们学习检索增强生成（RAG），让 AI 能够基于你的文档回答问题。

练习

  1. 创建一个网络搜索工具
  2. 实现一个数据库查询工具
  3. 构建一个文件管理助手
  4. 尝试组合多个工具完成复杂任务