工具详解 #

工具概述 #

在 CrewAI 框架中，工具是 Agent 可以使用的技能或功能。工具能够：

  • 执行各种操作（搜索、文件操作、API 调用等）
  • 扩展 Agent 的能力范围
  • 连接外部系统和服务
  • 实现复杂的数据处理

工具特性 #

核心特性 #

text 
┌─────────────────────────────────────────────────────────────┐
│                    CrewAI 工具特性                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  实用性：                                                     │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ - 网络搜索                                          │   │
│  │ - 数据分析                                          │   │
│  │ - 内容生成                                          │   │
│  │ - Agent 协作                                        │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  集成性：                                                     │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ - 无缝集成到 Agent 工作流                            │   │
│  │ - 支持 CrewAI 工具包                                │   │
│  │ - 支持 LangChain 工具                               │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  可定制性：                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ - 开发自定义工具                                    │   │
│  │ - 使用现有工具                                      │   │
│  │ - 满足特定需求                                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  高级功能：                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ - 错误处理                                          │   │
│  │ - 缓存机制                                          │   │
│  │ - 异步支持                                          │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

安装工具包 #

bash 
# 安装 CrewAI 工具包
pip install 'crewai[tools]'

内置工具 #

搜索工具 #

SerperDevTool #

python 
from crewai_tools import SerperDevTool
import os

os.environ["SERPER_API_KEY"] = "your-serper-api-key"

search_tool = SerperDevTool()

agent = Agent(
    role="研究员",
    goal="搜索最新信息",
    tools=[search_tool]
)

WebsiteSearchTool #

python 
from crewai_tools import WebsiteSearchTool

# 搜索特定网站
web_tool = WebsiteSearchTool(
    website="https://docs.python.org"
)

# 或搜索任意网站
web_tool = WebsiteSearchTool()

EXASearchTool #

python 
from crewai_tools import EXASearchTool

search_tool = EXASearchTool()

文件工具 #

FileReadTool #

python 
from crewai_tools import FileReadTool

# 读取特定文件
file_tool = FileReadTool(filepath="data/report.txt")

# 或通用文件读取
file_tool = FileReadTool()

DirectoryReadTool #

python 
from crewai_tools import DirectoryReadTool

# 读取目录内容
dir_tool = DirectoryReadTool(directory="./documents")

PDFSearchTool #

python 
from crewai_tools import PDFSearchTool

# 搜索 PDF 文件
pdf_tool = PDFSearchTool(pdf="docs/report.pdf")

# 或通用 PDF 搜索
pdf_tool = PDFSearchTool()

CSVSearchTool #

python 
from crewai_tools import CSVSearchTool

csv_tool = CSVSearchTool(csv="data/sales.csv")

代码工具 #

CodeInterpreterTool #

python 
from crewai_tools import CodeInterpreterTool

code_tool = CodeInterpreterTool()

agent = Agent(
    role="开发者",
    goal="执行代码",
    tools=[code_tool],
    allow_code_execution=True
)

CodeDocsSearchTool #

python 
from crewai_tools import CodeDocsSearchTool

docs_tool = CodeDocsSearchTool(docs_url="https://docs.python.org")

网页抓取工具 #

ScrapeWebsiteTool #

python 
from crewai_tools import ScrapeWebsiteTool

scrape_tool = ScrapeWebsiteTool(
    website_url="https://example.com"
)

FirecrawlScrapeWebsiteTool #

python 
from crewai_tools import FirecrawlScrapeWebsiteTool
import os

os.environ["FIRECRAWL_API_KEY"] = "your-api-key"

scrape_tool = FirecrawlScrapeWebsiteTool(
    website_url="https://example.com"
)

RAG 工具 #

RagTool #

python 
from crewai_tools import RagTool

rag_tool = RagTool()

agent = Agent(
    role="知识库助手",
    goal="从知识库检索信息",
    tools=[rag_tool]
)

其他工具 #

JSONSearchTool #

python 
from crewai_tools import JSONSearchTool

json_tool = JSONSearchTool(json_path="data/config.json")

TXTSearchTool #

python 
from crewai_tools import TXTSearchTool

txt_tool = TXTSearchTool(txt_file="notes.txt")

GithubSearchTool #

python 
from crewai_tools import GithubSearchTool
import os

os.environ["GITHUB_TOKEN"] = "your-github-token"

github_tool = GithubSearchTool(
    gh_query="repo:owner/repo language:python"
)

创建自定义工具 #

方式一：使用 @tool 装饰器 #

python 
from crewai.tools import tool

@tool("股票价格查询")
def get_stock_price(symbol: str) -> str:
    """
    查询指定股票的当前价格。
    
    Args:
        symbol: 股票代码，如 'AAPL'
    
    Returns:
        股票的当前价格信息
    """
    import requests
    response = requests.get(f"https://api.example.com/stock/{symbol}")
    data = response.json()
    return f"股票 {symbol} 的当前价格是 ${data['price']}"

# 使用工具
agent = Agent(
    role="金融分析师",
    goal="分析股票",
    tools=[get_stock_price]
)

方式二：继承 BaseTool 类 #

python 
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type

class WeatherInput(BaseModel):
    """天气查询工具的输入参数"""
    city: str = Field(..., description="城市名称")
    unit: str = Field(default="celsius", description="温度单位")

class WeatherTool(BaseTool):
    name: str = "天气查询"
    description: str = "查询指定城市的当前天气情况"
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, city: str, unit: str = "celsius") -> str:
        import requests
        response = requests.get(
            f"https://api.example.com/weather",
            params={"city": city, "unit": unit}
        )
        data = response.json()
        return f"{city} 当前天气：{data['weather']}，温度：{data['temp']}°"

# 使用工具
weather_tool = WeatherTool()
agent = Agent(
    role="天气助手",
    goal="提供天气信息",
    tools=[weather_tool]
)

方式三：异步工具 #

python 
from crewai.tools import tool
import asyncio

@tool("异步数据获取")
async def fetch_data_async(url: str) -> str:
    """
    异步获取 URL 数据。
    
    Args:
        url: 要获取的 URL
    
    Returns:
        获取的数据内容
    """
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

# 或使用类的方式
from crewai.tools import BaseTool

class AsyncCustomTool(BaseTool):
    name: str = "异步自定义工具"
    description: str = "执行异步操作的工具"

    async def _run(self, query: str = "") -> str:
        await asyncio.sleep(1)
        return f"处理完成: {query}"

工具缓存 #

启用缓存 #

python 
from crewai import Agent

agent = Agent(
    role="分析师",
    goal="分析数据",
    tools=[search_tool],
    cache=True
)

自定义缓存逻辑 #

python 
from crewai.tools import tool

@tool
def multiplication_tool(first_number: int, second_number: int) -> str:
    """用于两个数字相乘"""
    return str(first_number * second_number)

def cache_func(args, result):
    return int(result) % 2 == 0

multiplication_tool.cache_function = cache_func

agent = Agent(
    role="数学助手",
    tools=[multiplication_tool],
    cache=True
)

工具使用示例 #

研究团队 #

python 
import os
from crewai import Agent, Task, Crew
from crewai_tools import (
    SerperDevTool,
    WebsiteSearchTool,
    PDFSearchTool,
    FileReadTool
)

os.environ["SERPER_API_KEY"] = "your-key"
os.environ["OPENAI_API_KEY"] = "your-key"

search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()
pdf_tool = PDFSearchTool()
file_tool = FileReadTool()

researcher = Agent(
    role="研究分析师",
    goal="收集和分析信息",
    backstory="专业的研究分析师",
    tools=[search_tool, web_tool, pdf_tool],
    verbose=True
)

analyst = Agent(
    role="数据分析师",
    goal="分析数据并生成报告",
    backstory="资深数据分析师",
    tools=[file_tool],
    verbose=True
)

research_task = Task(
    description="研究 AI 行业趋势",
    expected_output="研究报告",
    agent=researcher
)

analysis_task = Task(
    description="分析研究结果",
    expected_output="分析报告",
    agent=analyst,
    context=[research_task]
)

crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    verbose=True
)

result = crew.kickoff()

开发团队 #

python 
from crewai import Agent, Task, Crew
from crewai_tools import (
    GithubSearchTool,
    CodeInterpreterTool,
    FileReadTool
)

github_tool = GithubSearchTool()
code_tool = CodeInterpreterTool()
file_tool = FileReadTool()

developer = Agent(
    role="开发者",
    goal="编写和调试代码",
    backstory="全栈开发者",
    tools=[github_tool, code_tool, file_tool],
    allow_code_execution=True,
    verbose=True
)

dev_task = Task(
    description="开发一个数据处理脚本",
    expected_output="Python 脚本",
    agent=developer
)

crew = Crew(
    agents=[developer],
    tasks=[dev_task],
    verbose=True
)

LangChain 工具集成 #

python 
from crewai import Agent
from langchain.tools import DuckDuckGoSearchRun

search = DuckDuckGoSearchRun()

agent = Agent(
    role="研究员",
    goal="搜索信息",
    tools=[search],
    verbose=True
)

工具最佳实践 #

1. 清晰的描述 #

python 
@tool("数据库查询")
def query_database(query: str) -> str:
    """
    执行 SQL 查询并返回结果。
    
    Args:
        query: SQL 查询语句
    
    Returns:
        查询结果的 JSON 格式字符串
    
    注意：
        - 仅支持 SELECT 语句
        - 结果限制在 1000 行
    """
    pass

2. 错误处理 #

python 
@tool("API 调用")
def call_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.RequestException as e:
        return f"错误：{str(e)}"

3. 参数验证 #

python 
from pydantic import BaseModel, Field, validator

class QueryInput(BaseModel):
    query: str = Field(..., description="搜索查询")
    max_results: int = Field(default=10, ge=1, le=100)
    
    @validator('query')
    def query_not_empty(cls, v):
        if not v.strip():
            raise ValueError('查询不能为空')
        return v

class QueryTool(BaseTool):
    name: str = "查询工具"
    args_schema: Type[BaseModel] = QueryInput

4. 性能优化 #

python 
@tool("批量处理")
def batch_process(items: list) -> str:
    """批量处理数据"""
    results = []
    for item in items:
        result = process_item(item)
        results.append(result)
    return str(results)

常见问题 #

工具未找到 #

python 
# 确保正确导入和实例化工具
from crewai_tools import SerperDevTool

search_tool = SerperDevTool()

agent = Agent(
    role="研究员",
    tools=[search_tool]
)

API 密钥问题 #

python 
import os

# 设置必要的环境变量
os.environ["SERPER_API_KEY"] = "your-key"
os.environ["OPENAI_API_KEY"] = "your-key"

工具执行超时 #

python 
# 在工具中设置超时
@tool("网络请求")
def fetch_url(url: str) -> str:
    import requests
    response = requests.get(url, timeout=30)
    return response.text

下一步 #

现在你已经深入了解了工具系统，接下来学习 高级特性，掌握 CrewAI 的高级功能！

最后更新：2026-04-04