工具

概览

CrewAI 工具让智能体具备从网页搜索、数据分析到协作以及在同事之间委派任务等能力。本文档概述了如何在 CrewAI 框架中创建、集成并利用这些工具，并特别强调协作工具这一新重点。

什么是工具？

CrewAI 中的工具是一种技能或函数，智能体可以利用它来执行各种操作。这包括来自 CrewAI Toolkit 和 LangChain Tools 的工具，使智能体能够完成从简单搜索到复杂交互，以及高效的团队协作。

工具的关键特性

实用性：适用于网页搜索、数据分析、内容生成和智能体协作等任务。
集成性：通过将工具无缝集成到工作流中，增强智能体能力。
可定制性：可灵活开发自定义工具或使用现有工具，以满足智能体的具体需求。
错误处理：内置稳健的错误处理机制，确保运行顺畅。
缓存机制：具备智能缓存，可优化性能并减少重复操作。
异步支持：同时支持同步和异步工具，避免阻塞式操作。
类型化输出：可使用可选的 Pydantic 模型为智能体提供清晰的 JSON 字段，而直接的 Python 调用仍会接收工具的常规返回值。

使用 CrewAI 工具

要通过 crewAI 工具增强智能体能力，请先安装我们的额外工具包：

pip install 'crewai[tools]'

下面是一个演示其用法的示例：

import os
from crewai import Agent, Task, Crew
# 导入 crewAI 工具
from crewai_tools import (
    DirectoryReadTool,
    FileReadTool,
    SerperDevTool,
    WebsiteSearchTool
)

# 设置 API 密钥
os.environ["SERPER_API_KEY"] = "Your Key" # serper.dev API key
os.environ["OPENAI_API_KEY"] = "Your Key"

# 实例化工具
docs_tool = DirectoryReadTool(directory='./blog-posts')
file_tool = FileReadTool()
search_tool = SerperDevTool()
web_rag_tool = WebsiteSearchTool()

# 创建智能体
researcher = Agent(
    role='Market Research Analyst',
    goal='Provide up-to-date market analysis of the AI industry',
    backstory='An expert analyst with a keen eye for market trends.',
    tools=[search_tool, web_rag_tool],
    verbose=True
)

writer = Agent(
    role='Content Writer',
    goal='Craft engaging blog posts about the AI industry',
    backstory='A skilled writer with a passion for technology.',
    tools=[docs_tool, file_tool],
    verbose=True
)

# 定义任务
research = Task(
    description='Research the latest trends in the AI industry and provide a summary.',
    expected_output='A summary of the top 3 trending developments in the AI industry with a unique perspective on their significance.',
    agent=researcher
)

write = Task(
    description='Write an engaging blog post about the AI industry, based on the research analyst\'s summary. Draw inspiration from the latest blog posts in the directory.',
    expected_output='A 4-paragraph blog post formatted in markdown with engaging, informative, and accessible content, avoiding complex jargon.',
    agent=writer,
    output_file='blog-posts/new_post.md'  # 最终博客文章将保存在这里
)

# 组装一个启用规划的 crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research, write],
    verbose=True,
    planning=True,  # 启用规划功能
)

# 执行任务
crew.kickoff()

可用的 CrewAI 工具

错误处理：所有工具都具备错误处理能力，使智能体能够优雅地处理异常并继续其任务。
缓存机制：所有工具都支持缓存，使智能体能够高效复用先前获得的结果，减少对外部资源的负载并加快执行时间。你也可以使用工具上的 cache_function 属性来更精细地控制缓存机制。

下面是可用工具及其说明列表：

工具	说明
ApifyActorsTool	将 Apify Actors 集成到你的工作流中，用于网页抓取和自动化任务。
BrowserbaseLoadTool	用于与浏览器交互并从浏览器中提取数据的工具。
CodeDocsSearchTool	一个经过优化的 RAG 工具，用于搜索代码文档和相关技术文档。
CodeInterpreterTool	一个用于解释 Python 代码的工具。
ComposioTool	支持使用 Composio 工具。
CSVSearchTool	一个专为搜索 CSV 文件而设计的 RAG 工具，适合处理结构化数据。
DALL-E Tool	一个用于通过 DALL-E API 生成图像的工具。
DirectorySearchTool	一个用于在目录中搜索的 RAG 工具，非常适合浏览文件系统。
DOCXSearchTool	一个面向 DOCX 文档搜索的 RAG 工具，非常适合处理 Word 文件。
DirectoryReadTool	便于读取和处理目录结构及其内容。
ExaSearchTool	使用 Exa 进行网页搜索，这是最快且最准确的网页搜索 API。支持节省 token 的高亮和完整页面内容。
FileReadTool	支持读取文件并提取数据，兼容多种文件格式。
FirecrawlSearchTool	使用 Firecrawl 搜索网页并返回结果的工具。
FirecrawlCrawlWebsiteTool	使用 Firecrawl 抓取网页的工具。
FirecrawlScrapeWebsiteTool	使用 Firecrawl 抓取网页 URL 并返回其内容的工具。
GithubSearchTool	一个用于搜索 GitHub 仓库的 RAG 工具，适用于代码和文档搜索。
SerperDevTool	一个面向开发用途的专用工具，具体功能仍在开发中。
TXTSearchTool	一个专注于搜索文本（.txt）文件的 RAG 工具，适合非结构化数据。
JSONSearchTool	一个面向 JSON 文件搜索的 RAG 工具，适合处理结构化数据。
LlamaIndexTool	支持使用 LlamaIndex 工具。
MDXSearchTool	一个专为搜索 Markdown（MDX）文件而设计的 RAG 工具，适合文档检索。
PDFSearchTool	一个面向 PDF 文档搜索的 RAG 工具，非常适合处理扫描文档。
PGSearchTool	一个优化用于搜索 PostgreSQL 数据库的 RAG 工具，适合数据库查询。
Vision Tool	一个用于通过 DALL-E API 生成图像的工具。
RagTool	一个通用的 RAG 工具，能够处理多种数据源和数据类型。
ScrapeElementFromWebsiteTool	支持从网站抓取特定元素，适合定向数据提取。
ScrapeWebsiteTool	便于抓取整个网站，适合全面的数据收集。
WebsiteSearchTool	一个用于搜索网站内容的 RAG 工具，针对网页数据提取进行了优化。
XMLSearchTool	一个专为搜索 XML 文件而设计的 RAG 工具，适合结构化数据格式。
YoutubeChannelSearchTool	一个用于搜索 YouTube 频道的 RAG 工具，适合视频内容分析。
YoutubeVideoSearchTool	一个面向 YouTube 视频搜索的 RAG 工具，非常适合提取视频数据。

创建你自己的工具

创建 CrewAI 工具主要有两种方式：

继承 `BaseTool`

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class MyToolInput(BaseModel):
    """MyCustomTool 的输入模式。"""
    argument: str = Field(..., description="参数说明。")

class MyCustomTool(BaseTool):
    name: str = "Name of my tool"
    description: str = "What this tool does. It's vital for effective utilization."
    args_schema: Type[BaseModel] = MyToolInput

    def _run(self, argument: str) -> str:
        # 你的工具逻辑写在这里
        return "Tool's result"

类型化工具输出

当工具返回结构化数据时，请定义一个 Pydantic 输出模型。这样智能体就能信任诸如 sku、quantity 或 needs_reorder 之类的字段名。

直接的 Python 调用仍会接收工具返回的值。智能体使用该工具时，CrewAI 会基于输出模型向智能体发送 JSON 字符串。

from crewai.tools import BaseTool
from pydantic import BaseModel

class InventoryResult(BaseModel):
    sku: str
    quantity: int
    needs_reorder: bool

class InventoryTool(BaseTool):
    name: str = "Inventory Check"
    description: str = "Checks current stock for a product SKU."

    def _run(self, sku: str) -> InventoryResult:
        quantity = {"SKU-123": 14, "SKU-456": 0}.get(sku, 0)
        return InventoryResult(sku=sku, quantity=quantity, needs_reorder=quantity < 5)

tool = InventoryTool()

# 直接调用会收到原始的 Pydantic 对象。
result = tool.run(sku="SKU-123")
print(result.quantity)

如果你不覆盖 format_output_for_agent，类型化输出会以 JSON 形式发送给智能体。普通字符串结果则保持原样。

class InventoryTool(BaseTool):
    name: str = "Inventory Check"
    description: str = "Checks current stock for a product SKU."

    def _run(self, sku: str) -> InventoryResult:
        quantity = {"SKU-123": 14, "SKU-456": 0}.get(sku, 0)
        return InventoryResult(sku=sku, quantity=quantity, needs_reorder=quantity < 5)

    def format_output_for_agent(self, raw_result: object) -> str:
        result = InventoryResult.model_validate(raw_result)
        status = "reorder needed" if result.needs_reorder else "stock is healthy"
        return f"{result.sku}: {result.quantity} units. {status}."

如果你没有覆盖 format_output_for_agent，类型化输出会以 JSON 发送给智能体。普通字符串结果仍然按以前的方式处理。

异步工具支持

CrewAI 支持异步工具，让你可以实现执行网络请求、文件 I/O 或其他异步操作而不会阻塞主执行线程的工具。

创建异步工具

你可以通过两种方式创建异步工具：

1. 使用带有异步函数的 `tool` 装饰器

from crewai.tools import tool

@tool("fetch_data_async")
async def fetch_data_async(query: str) -> str:
    """根据查询异步获取数据。"""
    # 模拟异步操作
    await asyncio.sleep(1)
    return f"Data retrieved for {query}"

2. 在自定义工具类中实现异步方法

from crewai.tools import BaseTool

class AsyncCustomTool(BaseTool):
    name: str = "async_custom_tool"
    description: str = "An asynchronous custom tool"

    async def _run(self, query: str = "") -> str:
        """异步运行该工具"""
        # 你的异步实现写在这里
        await asyncio.sleep(1)
        return f"Processed {query} asynchronously"

使用异步工具

异步工具可以无缝用于标准 Crew 工作流和基于 Flow 的工作流：

# 在标准 Crew 中
agent = Agent(role="researcher", tools=[async_custom_tool])

# 在 Flow 中
class MyFlow(Flow):
    @start()
    async def begin(self):
        crew = Crew(agents=[agent])
        result = await crew.kickoff_async()
        return result

CrewAI 框架会自动处理同步与异步工具的执行，因此你无需担心它们的调用方式有何不同。

使用 `tool` 装饰器

from crewai.tools import tool
@tool("Name of my tool")
def my_tool(question: str) -> str:
    """清晰说明这个工具的用途，智能体需要这些信息才能使用它。"""
    # 函数逻辑写在这里
    return "Result from your custom tool"

自定义缓存机制

from crewai.tools import tool

@tool
def multiplication_tool(first_number: int, second_number: int) -> str:
    """当你需要将两个数字相乘时非常有用。"""
    return first_number * second_number

def cache_func(args, result):
    # 在这里，我们只在结果是 2 的倍数时缓存
    cache = result % 2 == 0
    return cache

multiplication_tool.cache_function = cache_func

writer1 = Agent(
        role="Writer",
        goal="You write lessons of math for kids.",
        backstory="You're an expert in writing and you love to teach kids but you know nothing of math.",
        tools=[multiplication_tool],
        allow_delegation=False,
    )
    #...

结论

工具对于扩展 CrewAI 智能体的能力至关重要，使它们能够承担广泛的任务并高效协作。在构建 CrewAI 解决方案时，请同时利用自定义工具和现有工具，赋予你的智能体更强能力，并提升 AI 生态系统。也请考虑使用错误处理、缓存机制以及工具参数的灵活性来优化智能体的性能和能力。