Tools 工具
Tools（工具）是 Agent 执行具体操作的函数。本章将详细介绍如何定义和使用工具，以及工具在 Agent 系统中的核心作用。
什么是 Tools？
Tools 是 Agent 与外部世界交互的桥梁。它们封装了具体的业务逻辑或操作能力，让大模型能够根据用户请求自主判断：要不要调用工具？调用哪个工具？
工具的本质
工具的本质是将具体的业务逻辑封装成标准化接口。例如：
- 查询天气 →
get_weather(city: str) - 搜索信息 →
search_web(query: str) - 计算数学 →
calculate(expression: str)
工具在 Agent 中的角色
用户提问 → Agent 理解意图 → 决定调用工具 → 执行工具 →
获取结果 → 生成最终回答
Agent 通过工具的名称和描述来决定何时以及如何使用工具。因此，清晰的工具定义是 Agent 表现良好的关键。
定义工具
使用 @tool 装饰器（推荐）
@tool 装饰器是定义工具的最简单方式，它自动从函数签名和文档字符串中提取信息：
from langchain_core.tools import tool
@tool
def multiply(a: int, b: int) -> int:
"""计算两个整数的乘积。
参数:
a: 第一个整数
b: 第二个整数
返回:
两个数的乘积
"""
return a * b
# 查看工具信息
print(multiply.name) # "multiply"
print(multiply.description) # "计算两个整数的乘积..."
print(multiply.args) # {'a': {'title': 'A', 'type': 'integer'}, ...}
# 调用工具
result = multiply.invoke({"a": 3, "b": 4})
print(result) # 12
关键点：
- 函数名会成为工具的
name - 文档字符串（docstring）会成为工具的
description - 类型注解会被用来生成参数 schema
使用 Pydantic 精确定义参数
对于复杂工具，使用 Pydantic 可以更精确地定义参数：
from langchain_core.tools import tool
from pydantic import BaseModel, Field
class WeatherInput(BaseModel):
city: str = Field(description="城市名称，如'北京'、'Shanghai'")
date: str = Field(default="today", description="日期，格式'YYYY-MM-DD'")
@tool(args_schema=WeatherInput)
def get_weather(city: str, date: str = "today") -> str:
"""获取指定城市和日期的天气预报。
当用户询问天气情况时使用此工具。
"""
return f"{city} {date} 的天气：晴天 25°C"
# 调用
result = get_weather.invoke({"city": "北京", "date": "2025-01-01"})
使用 Tool 类（底层方式）
如果你需要更多控制，可以直接使用 Tool 类：
from langchain_core.tools import Tool
def add(a: int, b: int) -> str:
"""实际的函数实现"""
return str(a + b)
add_tool = Tool(
name="add",
func=add,
description="用于计算两个数的和，输入为两个整数",
args_schema=AddInput # 可选：Pydantic schema
)
# 调用
result = add_tool.invoke({"a": 5, "b": 3})
多参数工具示例
@tool
def search_hotels(
location: str,
check_in: str,
check_out: str,
price_level: str = "all"
) -> str:
"""搜索酒店。
当用户需要预订酒店或查询酒店信息时使用此工具。
参数:
location: 城市或地区名称，如"北京"、"上海"
check_in: 入住日期，格式"YYYY-MM-DD"
check_out: 退房日期，格式"YYYY-MM-DD"
price_level: 价格等级 (budget/mid-range/luxury)，默认为 all
"""
if price_level == "budget":
return f"{location}的经济型酒店：如家、汉庭"
elif price_level == "luxury":
return f"{location}的豪华酒店：四季、丽思卡尔顿"
else:
return f"{location}的中档酒店：全季、亚朵"
无参数工具
@tool
def get_current_time() -> str:
"""获取当前系统时间。
当用户询问当前时间、现在几点了时使用此工具。
"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@tool
def get_random_joke() -> str:
"""随机获取一个程序员笑话。
当用户想要听笑话或放松心情时使用此工具。
"""
jokes = [
"为什么程序员总是弄混圣诞节和万圣节？因为 Oct 31 = Dec 25",
"为什么程序员不喜欢户外活动？因为他们的生活已经充满了 bugs",
"一个程序员走进酒吧，举起双手说：'我要一杯啤酒。' 酒保问：'一杯还是两杯？' 程序员回答：'一杯。' 然后举起两根手指。"
]
import random
return random.choice(jokes)
工具返回内容
简单字符串返回
@tool
def simple_tool(input: str) -> str:
"""简单工具，返回字符串"""
return f"处理结果: {input}"
结构化数据返回
from pydantic import BaseModel, Field
class WeatherOutput(BaseModel):
city: str = Field(description="城市名称")
temperature: str = Field(description="温度")
condition: str = Field(description="天气状况")
humidity: str = Field(description="湿度")
@tool
def get_weather(city: str) -> WeatherOutput:
"""获取城市天气信息"""
# 模拟API调用
return WeatherOutput(
city=city,
temperature="25°C",
condition="晴",
humidity="60%"
)
错误处理
工具应该优雅地处理错误，返回字符串而非抛出异常：
@tool
def safe_divide(a: float, b: float) -> str:
"""安全除法，参数为两个数字。
当需要进行除法运算时使用此工具。
"""
if b == 0:
return "错误：除数不能为零"
try:
result = a / b
return f"计算结果：{result}"
except Exception as e:
return f"计算错误：{str(e)}"
@tool
def read_file(file_path: str) -> str:
"""读取文件内容。
参数:
file_path: 文件路径
"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
return f.read()
except FileNotFoundError:
return f"错误：文件 '{file_path}' 不存在"
except Exception as e:
return f"读取失败：{str(e)}"
在 Agent 中使用工具
基础示例
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
# 定义工具
@tool
def get_weather(city: str) -> str:
"""获取城市天气"""
return f"{city}今天晴，25°C"
@tool
def calculate(expression: str) -> str:
"""计算数学表达式，如 '2 + 3 * 4'"""
try:
# 安全计算：只允许基本运算符
allowed_chars = set('0123456789+-*/.() ')
if not all(c in allowed_chars for c in expression):
return "错误：表达式包含非法字符"
result = eval(expression)
return str(result)
except Exception as e:
return f"计算错误: {e}"
# 初始化模型
model = init_chat_model(model="gpt-4o-mini", model_provider="openai")
# 创建 Agent
agent = create_agent(
model=model,
tools=[get_weather, calculate],
system_prompt="你是一个智能助手"
)
# 调用 Agent
messages = [{"role": "user", "content": "北京天气怎么样? 25度乘以3等于多少?"}]
result = agent.invoke({"messages": messages})
print(result["messages"][-1].content)
工具选择机制
LLM 根据工具的 name 和 description 选择要使用的工具：
# ✅ 好的描述：清晰说明使用场景
@tool
def get_stock_price(symbol: str) -> str:
"""获取股票价格。
当用户询问特定股票的当前价格时使用此工具。
参数:
symbol: 股票代码，如 AAPL(苹果)、GOOGL(谷歌)、MSFT(微软)
返回:
股票当前价格和涨跌信息
"""
return f"{symbol}: 150.00 美元 (+2.5%)"
# ❌ 差的描述：信息不足
@tool
def get_price(s: str) -> str:
"""获取价格"""
return "..."
避免工具冲突
当多个工具功能相似时，通过描述区分它们：
@tool
def search_google(query: str) -> str:
"""使用 Google 搜索全球信息。
当需要搜索国际新闻、英文资料或全球信息时使用此工具。
"""
return f"Google 搜索结果: {query}..."
@tool
def search_baidu(query: str) -> str:
"""使用百度搜索中文信息。
当需要搜索中文网页、国内新闻或中文资料时使用此工具。
"""
return f"百度搜索结果: {query}..."
@tool
def search_wikipedia(query: str) -> str:
"""搜索维基百科获取百科知识。
当需要查询人物、地点、事件等百科知识时使用此工具。
"""
return f"维基百科: {query}..."
高级用法
异步工具
import asyncio
import aiohttp
@tool
async def async_search(query: str) -> str:
"""异步搜索网络信息。
适合需要网络请求的工具，不会阻塞主线程。
"""
await asyncio.sleep(0.1) # 模拟异步操作
return f"异步搜索结果: {query}"
# 在异步环境中使用
async def main():
result = await async_search.ainvoke({"query": "Python"})
print(result)
asyncio.run(main())
带上下文依赖的工具
from typing import Optional
class Database:
"""模拟数据库连接"""
def query(self, sql: str) -> list:
return [{"id": 1, "name": "张三"}]
# 需要在创建时传入依赖
def create_db_tool(db: Database):
@tool
def query_database(sql: str) -> str:
"""执行 SQL 查询。
当需要从数据库获取数据时使用此工具。
参数:
sql: SQL 查询语句
"""
try:
results = db.query(sql)
return str(results)
except Exception as e:
return f"查询错误: {e}"
return query_database
# 使用
db = Database()
db_tool = create_db_tool(db)
工具组合
from langchain_core.tools import Tool
from typing import List
def create_calculator_tools() -> List[Tool]:
"""创建一组计算器工具"""
@tool
def add(a: float, b: float) -> str:
"""加法运算"""
return str(a + b)
@tool
def subtract(a: float, b: float) -> str:
"""减法运算"""
return str(a - b)
@tool
def multiply(a: float, b: float) -> str:
"""乘法运算"""
return str(a * b)
@tool
def divide(a: float, b: float) -> str:
"""除法运算"""
if b == 0:
return "错误：除数不能为零"
return str(a / b)
return [add, subtract, multiply, divide]
# 使用
calc_tools = create_calculator_tools()
app = create_agent(model=model, tools=calc_tools)
实际应用示例
示例 1：天气查询助手
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
model = init_chat_model(model="gpt-4o-mini", model_provider="openai")
@tool
def get_current_weather(city: str) -> str:
"""查询城市当前天气。
当用户询问今天天气时使用此工具。
"""
weather_map = {
"北京": "晴，25°C，空气质量良",
"上海": "多云，28°C，湿度65%",
"广州": "小雨，30°C，带伞出行",
"深圳": "晴，29°C，紫外线强"
}
return weather_map.get(city, f"抱歉，暂无{city}的天气数据")
@tool
def get_weather_forecast(city: str, days: int = 3) -> str:
"""查询城市天气预报。
当用户询问未来几天天气时使用此工具。
参数:
city: 城市名称
days: 预报天数，1-7天，默认为3天
"""
forecasts = {
"北京": ["晴", "多云", "晴"],
"上海": ["多云", "小雨", "多云"]
}
city_forecast = forecasts.get(city, ["晴"] * days)
return f"{city}未来{days}天：" + " → ".join(city_forecast[:days])
@tool
def get_clothing_advice(temperature: int) -> str:
"""根据温度给出穿衣建议。
当用户询问穿什么衣服合适时使用此工具。
参数:
temperature: 温度（摄氏度）
"""
if temperature < 10:
return "建议穿羽绒服、毛衣，注意保暖"
elif temperature < 20:
return "建议穿外套、长袖衬衫"
elif temperature < 28:
return "建议穿短袖、薄外套"
else:
return "建议穿短袖、短裤，注意防晒"
weather_agent = create_agent(
model=model,
tools=[get_current_weather, get_weather_forecast, get_clothing_advice],
system_prompt="你是一个天气助手，可以查询天气并给出穿衣建议。"
)
# 测试
# "北京今天天气怎么样？适合穿什么？"
# Agent 会先查天气（25°C），然后给出穿衣建议
示例 2：数据处理工具集
import csv
from io import StringIO
from typing import List
@tool
def parse_csv(csv_content: str) -> str:
"""解析 CSV 内容。
当用户提供了 CSV 格式数据时使用此工具。
参数:
csv_content: CSV 格式的字符串内容
"""
try:
reader = csv.DictReader(StringIO(csv_content))
rows = list(reader)
return f"成功解析 {len(rows)} 行数据，列：{list(rows[0].keys()) if rows else '无'}"
except Exception as e:
return f"解析错误: {e}"
@tool
def calculate_column(csv_content: str, column: str, operation: str) -> str:
"""对 CSV 列进行计算。
参数:
csv_content: CSV 内容
column: 要计算的列名
operation: 操作类型 (sum/avg/max/min/count)
"""
try:
reader = csv.DictReader(StringIO(csv_content))
values = [float(row[column]) for row in reader if column in row]
if operation == "sum":
return f"{column} 总和: {sum(values)}"
elif operation == "avg":
return f"{column} 平均值: {sum(values)/len(values) if values else 0}"
elif operation == "max":
return f"{column} 最大值: {max(values)}"
elif operation == "min":
return f"{column} 最小值: {min(values)}"
elif operation == "count":
return f"{column} 计数: {len(values)}"
else:
return f"不支持的操作: {operation}"
except Exception as e:
return f"计算错误: {e}"
@tool
def generate_chart(data_description: str, chart_type: str = "bar") -> str:
"""生成图表描述。
参数:
data_description: 数据描述
chart_type: 图表类型 (bar/line/pie/scatter)
"""
chart_names = {
"bar": "柱状图",
"line": "折线图",
"pie": "饼图",
"scatter": "散点图"
}
return f"已生成{chart_names.get(chart_type, '图表')}，展示：{data_description}"
示例 3：文件操作工具
import os
@tool
def list_files(directory: str = ".") -> str:
"""列出目录中的文件。
当用户想查看某个文件夹内容时使用此工具。
参数:
directory: 目录路径，默认为当前目录
"""
try:
files = os.listdir(directory)
return f"目录 '{directory}' 包含 {len(files)} 个项目:\n" + "\n".join(files[:20])
except Exception as e:
return f"列出文件失败: {e}"
@tool
def read_file(file_path: str) -> str:
"""读取文本文件内容。
当用户需要查看文件内容时使用此工具。
支持 .txt, .md, .py, .json 等文本文件。
参数:
file_path: 文件路径
"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# 限制返回长度
if len(content) > 2000:
return content[:2000] + "\n... (内容已截断)"
return content
except FileNotFoundError:
return f"错误：文件 '{file_path}' 不存在"
except UnicodeDecodeError:
return f"错误：文件 '{file_path}' 不是文本文件或编码不支持"
except Exception as e:
return f"读取失败: {e}"
@tool
def write_file(file_path: str, content: str) -> str:
"""写入文本文件。
当用户需要保存内容到文件时使用此工具。
参数:
file_path: 文件路径
content: 要写入的内容
"""
try:
# 确保目录存在
directory = os.path.dirname(file_path)
if directory and not os.path.exists(directory):
os.makedirs(directory)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return f"成功写入文件: {file_path} ({len(content)} 字符)"
except Exception as e:
return f"写入失败: {e}"
@tool
def get_file_info(file_path: str) -> str:
"""获取文件信息。
参数:
file_path: 文件路径
"""
try:
stat = os.stat(file_path)
size_kb = stat.st_size / 1024
return f"文件: {file_path}\n大小: {size_kb:.2f} KB\n修改时间: {stat.st_mtime}"
except Exception as e:
return f"获取信息失败: {e}"
预定义工具
LangChain 社区提供了许多现成的工具：
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
# DuckDuckGo 搜索
search_tool = DuckDuckGoSearchRun()
result = search_tool.invoke("Python programming")
# 维基百科
wikipedia_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
result = wikipedia_tool.invoke("Artificial Intelligence")
常用预定义工具列表
| 工具 | 说明 | 安装 |
|---|---|---|
DuckDuckGoSearchRun | DuckDuckGo 搜索 | pip install duckduckgo-search |
WikipediaQueryRun | 维基百科查询 | pip install wikipedia |
ShellTool | 执行 Shell 命令 | 内置 |
PythonREPLTool | 执行 Python 代码 | 内置 |
最佳实践
1. 清晰的工具描述
# ✅ 好的示例：说明使用场景、参数、返回值
@tool
def book_flight(origin: str, destination: str, date: str) -> str:
"""预订机票。
当用户明确表达想要预订机票时使用此工具。
需要提供以下信息：
- origin: 出发城市（必需）
- destination: 目的城市（必需）
- date: 出发日期，格式 YYYY-MM-DD（必需）
如果用户没有提供完整信息，不要调用此工具，
而是询问用户缺少的信息。
返回预订结果或错误信息。
"""
pass
2. 参数命名规范
# ✅ 好的命名：清晰易懂
@tool
def search_products(
keyword: str, # 不是 "q" 或 "query"
max_price: float, # 不是 "max_p"
category: str = "all" # 有默认值的可选参数
) -> str:
...
3. 错误处理
@tool
def call_external_api(endpoint: str) -> str:
"""调用外部 API"""
try:
# 调用逻辑
response = requests.get(endpoint, timeout=10)
response.raise_for_status()
return response.text
except requests.Timeout:
return "错误：请求超时，请稍后重试"
except requests.HTTPError as e:
return f"错误：HTTP {e.response.status_code}"
except Exception as e:
return f"错误：{str(e)}"
4. 安全性考虑
# ✅ 安全的文件操作：限制访问范围
ALLOWED_DIRECTORIES = ["/home/user/data", "/tmp"]
@tool
def safe_read_file(file_path: str) -> str:
"""安全地读取文件"""
# 检查路径是否在允许范围内
real_path = os.path.realpath(file_path)
if not any(real_path.startswith(d) for d in ALLOWED_DIRECTORIES):
return "错误：无权访问该路径"
# 继续读取...