工具系统
1. 工具概述
工具（Tools）是 Agent 与外部世界交互的能力。没有工具的 Agent 只能基于自身知识回答问题；有了工具，Agent 可以搜索网络、读写文件、调用 API、操作数据库等。
工具的作用
┌──────────┐ 使用工具 ┌──────────────┐
│ Agent │ ──────────────→ │ 外部能力 │
│ (思考者) │ │ - 网络搜索 │
│ │ ←────────────── │ - 文件读写 │
└──────────┘ 返回结果 │ - API 调用 │
│ - 数据库查询 │
└──────────────┘2. 内置工具
CrewAI 提供了丰富的内置工具，通过 crewai_tools 包获取：
bash
pip install 'crewai[tools]'2.1 常用内置工具
| 工具 | 作用 | 依赖 |
|---|---|---|
SerperDevTool | Google 搜索（Serper API） | SERPER_API_KEY |
WebsiteSearchTool | 基于 RAG 的网站内容搜索 | 嵌入模型 |
ScrapeWebsiteTool | 抓取网页内容 | 无 |
FileReadTool | 读取文件 | 无 |
FileWriteTool | 写入文件 | 无 |
DirectoryReadTool | 读取目录结构 | 无 |
DirectorySearchTool | 在目录中搜索内容 | 无 |
CSVSearchTool | CSV 文件搜索 | 嵌入模型 |
JSONSearchTool | JSON 文件搜索 | 嵌入模型 |
PDFSearchTool | PDF 文件搜索 | 嵌入模型 |
XMLSearchTool | XML 文件搜索 | 嵌入模型 |
TXTSearchTool | 文本文件搜索 | 嵌入模型 |
DOCXSearchTool | Word 文档搜索 | 嵌入模型 |
GithubSearchTool | GitHub 仓库搜索 | GITHUB_TOKEN |
SeleniumScrapingTool | 浏览器级网页抓取 | Selenium |
CodeDocsSearchTool | 代码文档搜索 | 嵌入模型 |
2.2 使用内置工具
python
from crewai import Agent
from crewai_tools import SerperDevTool, WebsiteSearchTool, FileReadTool
# 搜索工具
search = SerperDevTool()
# 网站搜索（RAG）
website_search = WebsiteSearchTool(
website_url="https://docs.example.com"
)
# 文件读取
file_reader = FileReadTool()
# 分配给 Agent
researcher = Agent(
role="研究员",
goal="调研技术信息",
backstory="资深研究员",
tools=[search, website_search, file_reader]
)2.3 工具配置示例
python
from crewai_tools import SerperDevTool, WebsiteSearchTool
# SerperDevTool 配置
search = SerperDevTool(
n_results=10, # 返回结果数量
country="cn", # 搜索区域
locale="zh-cn" # 语言
)
# WebsiteSearchTool 配置
web_search = WebsiteSearchTool(
website_url="https://example.com",
config=dict(
embedder=dict(
provider="openai",
config=dict(model="text-embedding-3-small")
)
)
)3. 自定义工具 — @tool 装饰器
对于简单的工具，使用 @tool 装饰器是最快的方式：
3.1 基础用法
python
from crewai import tool
@tool("搜索数据库")
def search_database(query: str) -> str:
"""根据关键词搜索产品数据库，返回匹配的产品信息。"""
# 模拟数据库查询
products = {
"python": "Python 编程入门教程 - 销量 10000+",
"java": "Java 高级编程 - 销量 8000+",
"ai": "AI 实战指南 - 销量 15000+"
}
return products.get(query.lower(), "未找到匹配的产品")
# 使用
agent = Agent(
role="产品顾问",
goal="帮助用户找到合适的产品",
backstory="资深产品顾问",
tools=[search_database]
)3.2 带类型注解的工具
python
from crewai import tool
@tool("计算两个数的和")
def add_numbers(a: float, b: float) -> str:
"""计算两个数字的和并返回结果。"""
return f"{a} + {b} = {a + b}"
@tool("获取当前天气")
def get_weather(city: str, unit: str = "celsius") -> str:
"""获取指定城市的当前天气信息。
Args:
city: 城市名称（如 "北京"、"上海"）
unit: 温度单位，celsius 或 fahrenheit
"""
# 实际应用中调用天气 API
return f"{city} 当前温度 25°{'C' if unit == 'celsius' else 'F'}，晴天"3.3 返回结构化数据
python
import json
from crewai import tool
@tool("查询股票信息")
def query_stock(symbol: str) -> str:
"""查询股票的实时价格和基本信息。
Args:
symbol: 股票代码（如 AAPL, TSLA）
"""
# 模拟 API 调用
data = {
"symbol": symbol,
"price": 150.25,
"change": "+2.3%",
"volume": "1.2M"
}
return json.dumps(data, ensure_ascii=False)4. 自定义工具 — BaseTool 子类
对于复杂的工具，推荐继承 BaseTool，获得完整的 Pydantic 验证和更好的可维护性：
4.1 基础用法
python
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type
# 定义输入 Schema
class StockQueryInput(BaseModel):
symbol: str = Field(description="股票代码，如 AAPL、TSLA")
period: str = Field(default="1d", description="时间周期：1d/5d/1m/3m/1y")
# 定义工具
class StockAnalysisTool(BaseTool):
name: str = "股票分析工具"
description: str = "查询股票的历史数据和技术指标，返回分析结果"
args_schema: Type[BaseModel] = StockQueryInput
def _run(self, symbol: str, period: str = "1d") -> str:
"""执行股票分析"""
# 实际实现：调用金融 API
return f"股票 {symbol} 在 {period} 周期内的分析结果：上涨趋势，RSI=65"
# 使用
agent = Agent(
role="金融分析师",
goal="分析股票走势",
backstory="资深金融分析师",
tools=[StockAnalysisTool()]
)4.2 完整示例：API 调用工具
python
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type
import requests
class WeatherInput(BaseModel):
city: str = Field(description="城市名称，如 北京、上海、广州")
days: int = Field(default=3, description="预报天数，1-7")
class WeatherTool(BaseTool):
name: str = "天气查询工具"
description: str = "查询指定城市的天气预报，包括温度、湿度、风力等信息"
args_schema: Type[BaseModel] = WeatherInput
def _run(self, city: str, days: int = 3) -> str:
"""调用天气 API"""
try:
# 示例：使用和风天气 API
api_key = "your-api-key"
url = f"https://api.weatherapi.com/v1/forecast.json"
params = {
"key": api_key,
"q": city,
"days": days,
"lang": "zh"
}
response = requests.get(url, params=params, timeout=10)
data = response.json()
# 格式化结果
location = data["location"]["name"]
current = data["current"]
result = f"📍 {location}\n"
result += f"🌡️ 温度：{current['temp_c']}°C\n"
result += f"💧 湿度：{current['humidity']}%\n"
result += f"💨 风力：{current['wind_kph']} km/h\n"
result += f"☁️ 天气：{current['condition']['text']}\n"
# 预报
result += f"\n📅 未来 {days} 天预报：\n"
for day in data["forecast"]["forecastday"]:
result += f" {day['date']}: {day['day']['condition']['text']}"
result += f" ({day['day']['mintemp_c']}~{day['day']['maxtemp_c']}°C)\n"
return result
except Exception as e:
return f"天气查询失败：{str(e)}"
# 使用
agent = Agent(
role="旅行规划师",
goal="根据天气情况规划旅行",
backstory="资深旅行规划师",
tools=[WeatherTool()]
)4.3 使用第三方工具包
python
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel
class CalculatorInput(BaseModel):
expression: str = Field(description="数学表达式，如 '2 + 3 * 4'")
class CalculatorTool(BaseTool):
name: str = "计算器"
description: str = "执行数学计算，支持基本运算和常用数学函数"
args_schema: Type[BaseModel] = CalculatorInput
def _run(self, expression: str) -> str:
try:
# 安全的数学表达式求值
import ast
import operator as op
operators = {
ast.Add: op.add, ast.Sub: op.sub,
ast.Mult: op.mul, ast.Div: op.truediv,
ast.Pow: op.pow, ast.USub: op.neg
}
def eval_expr(node):
if isinstance(node, ast.Num):
return node.n
elif isinstance(node, ast.BinOp):
return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))
elif isinstance(node, ast.UnaryOp):
return operators[type(node.op)](eval_expr(node.operand))
else:
raise TypeError(f"不支持的操作: {type(node)}")
result = eval_expr(ast.parse(expression, mode='eval').body)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误：{str(e)}"5. 工具分配方式
5.1 Agent 级别分配
python
agent = Agent(
role="研究员",
goal="调研",
backstory="研究员",
tools=[search_tool, file_tool] # Agent 默认可用的工具
)5.2 Task 级别覆盖
python
task = Task(
description="执行特定任务",
expected_output="结果",
agent=agent,
tools=[specialized_tool] # 覆盖 Agent 的默认工具
)TIP
Task 级别的 tools 会替换（而非追加）Agent 的默认工具。如果 Task 需要 Agent 的工具 + 额外工具，需要手动合并：
python
task = Task(
description="...",
expected_output="...",
agent=agent,
tools=agent.tools + [extra_tool] # 合并
)6. 工具对比总结
| 方式 | 适用场景 | 输入验证 | 代码量 |
|---|---|---|---|
@tool 装饰器 | 简单、快速、一次性工具 | 基于类型注解 | 少 |
BaseTool 子类 | 复杂、可复用、需要完整验证 | Pydantic Schema | 中 |
内置 crewai_tools | 常见场景（搜索、文件、网页） | 已内置 | 无 |
总结
| 知识点 | 要点 |
|---|---|
| 内置工具 | crewai_tools 包提供搜索、文件、网页等常用工具 |
@tool 装饰器 | 快速定义简单工具 |
BaseTool 子类 | 定义复杂工具，支持完整 Pydantic 验证 |
| 工具分配 | Agent 级别（默认）和 Task 级别（覆盖） |
| Task 覆盖 | Task 的 tools 替换而非追加 Agent 的 tools |