工具

理解并运用 CrewAI 框架中的工具，实现智能体协作与任务执行。

概述

CrewAI 工具为智能体赋予了从网页搜索、数据分析到协作与在同事之间委派任务等多种能力。 本文档介绍如何在 CrewAI 框架中创建、集成和运用这些工具，并新增对协作工具的关注。

NOTE

工具为智能体提供可执行动作的 可调用函数。它们与 MCP（远程工具服务器）、Apps（平台集成）、技能（领域专业知识）和知识（检索到的事实）配合使用。请参阅智能体能力概览，了解何时使用哪一种。

什么是工具？

CrewAI 中的工具是一种技能或函数，智能体可以利用它来执行各种动作。 这包括来自 CrewAI Toolkit 和 LangChain Tools 的工具， 从简单搜索到复杂交互以及智能体之间高效协作，一切皆可实现。

NOTE

CrewAI AMP 提供全面的工具仓库，内含针对常见业务系统和 API 的预构建集成。几分钟（而非数天）即可部署具备企业级工具的智能体。

企业版工具仓库包括：

• 针对主流企业系统的预构建连接器
• 自定义工具创建界面
• 版本控制与共享能力
• 安全与合规特性

工具的关键特性

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

使用 CrewAI 工具

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

pip install 'crewai[tools]'

以下示例演示了它们的使用方式：

import os
from crewai import Agent, Task, Crew
# Importing crewAI tools
from crewai_tools import (
    DirectoryReadTool,
    FileReadTool,
    SerperDevTool,
    WebsiteSearchTool
)

# Set up API keys
os.environ["SERPER_API_KEY"] = "Your Key" # serper.dev API key
os.environ["OPENAI_API_KEY"] = "Your Key"

# Instantiate tools
docs_tool = DirectoryReadTool(directory='./blog-posts')
file_tool = FileReadTool()
search_tool = SerperDevTool()
web_rag_tool = WebsiteSearchTool()

# Create agents
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
)

# Define tasks
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'  # The final blog post will be saved here
)

# Assemble a crew with planning enabled
crew = Crew(
    agents=[researcher, writer],
    tasks=[research, write],
    verbose=True,
    planning=True,  # Enable planning feature
)

# Execute tasks
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 高效的 highlights 和整页内容。
FileReadTool	支持从文件中读取和提取数据，兼容多种文件格式。
FirecrawlSearchTool	使用 Firecrawl 搜索网页并返回结果的工具。
FirecrawlCrawlWebsiteTool	使用 Firecrawl 抓取网页的工具。
FirecrawlScrapeWebsiteTool	使用 Firecrawl 抓取网页 URL 并返回其内容的工具。
GithubSearchTool	用于在 GitHub 仓库内搜索的 RAG 工具，适合代码和文档搜索。
SerperDevTool	面向开发用途的专用工具，具备正在开发中的特定功能。
TXTSearchTool	专注于在文本（.txt）文件内搜索的 RAG 工具，适用于非结构化数据。
JSONSearchTool	专为在 JSON 文件内搜索而设计的 RAG 工具，适合处理结构化数据。
LlamaIndexTool	支持使用 LlamaIndex 工具。
MDXSearchTool	为在 Markdown（MDX）文件内搜索而定制的 RAG 工具，适合文档场景。
MongoDBVectorSearchTool	用于在 MongoDB Atlas 集合上执行向量相似度搜索的 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 工具，非常适合视频数据提取。

创建你自己的工具

TIP

开发者可以打造契合智能体需求的 custom tools，也可以 使用预构建的选项。

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

子类化 BaseTool

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

class MyToolInput(BaseModel):
    """Input schema for MyCustomTool."""
    argument: str = Field(..., description="Description of the argument.")

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:
        # Your tool's logic here
        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()

# Direct calls receive the raw Pydantic object.
result = tool.run(sku="SKU-123")
print(result.quantity)

若要向智能体发送 Markdown 或其他简短文本格式，请重写 format_output_for_agent。直接调用 tool.run(...) 仍会返回正常的 Python 值。

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:
    """Asynchronously fetch data based on the query."""
    # Simulate async operation
    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:
        """Asynchronously run the tool"""
        # Your async implementation here
        await asyncio.sleep(1)
        return f"Processed {query} asynchronously"

使用异步工具

异步工具在标准 Crew 工作流和基于 Flow 的工作流中都能无缝运行：

# In standard Crew
agent = Agent(role="researcher", tools=[async_custom_tool])

# In 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:
    """Clear description for what this tool is useful for, your agent will need this information to use it."""
    # Function logic here
    return "Result from your custom tool"

自定义缓存机制

TIP

工具可以选择实现 cache_function，以微调缓存 行为。该函数根据特定条件决定何时缓存结果， 让你可以细粒度地控制缓存逻辑。

from crewai.tools import tool

@tool
def multiplication_tool(first_number: int, second_number: int) -> str:
    """Useful for when you need to multiply two numbers together."""
    return first_number * second_number

def cache_func(args, result):
    # In this case, we only cache the result if it's a multiple of 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,
    )
    #...

报告工具失败

工具可以在不抛出异常的情况下结束，却仍然没有完成它被要求做的事。Slack 会返回 HTTP 200 和 {"ok": false, "error": "channel_not_found"}；MCP 服务器会设置 isError；平台操作会返回错误负载。工具调用 “成功了”，于是错误文本作为普通结果传给智能体——智能体 在最终答案中叙述该问题，而这次运行被记录为成功。

返回 ToolFailure 而不是错误字符串，框架就能分辨出 两者的差别：

from typing import Any

from crewai.tools import BaseTool
from crewai.tools.tool_failure import ToolFailure

class SendSlackMessage(BaseTool):
    name: str = "send_slack_message"
    description: str = "Post a message to a Slack channel."

    def _run(self, channel: str, text: str) -> Any:
        payload = slack.post(channel=channel, text=text)
        if not payload["ok"]:
            return ToolFailure(
                message=f"Slack rejected the message: {payload['error']}",
                code=payload["error"],
                retryable=payload["error"] == "rate_limited",
            )
        return payload

智能体读到的仍然是纯文本——ToolFailure.as_agent_message() 会渲染该 消息——因此模型行为没有变化。改变的是，这个失败现在 对下游的所有环节都可见。

检测严格是声明式的。CrewAI 从不猜测某个字符串是否“看起来 像”错误，因此一个合法返回错误相关文本的工具永远不会 被误读为失败。以下情况会记录失败：工具返回 ToolFailure、工具抛出异常、MCP 服务器设置 isError、工具的 max_usage_count 用完，或智能体调用了不存在的工具。

选择失败策略

tool_failure_policy 控制接下来发生什么：

策略	行为
ignore	不记录、不发出事件，也不采取任何动作。
warn （默认）	记录失败，发出 ToolFailureDetectedEvent，然后继续执行。
raise	记录并发出事件，然后以 ToolExecutionFailedError 中止。

from crewai import Agent, Crew, Task
from crewai.tools.tool_failure import ToolFailurePolicy

agent = Agent(
    role="Slack Messenger",
    goal="Post the report to Slack",
    backstory="...",
    tools=[SendSlackMessage()],
    tool_failure_policy=ToolFailurePolicy.WARN,
)

# Tighten a single high-stakes task without changing the agent.
task = Task(
    description="Post the final report to #engineering",
    expected_output="Confirmation the message was posted",
    agent=agent,
    tool_failure_policy=ToolFailurePolicy.RAISE,
)

# Or set a baseline once for every agent in the crew.
crew = Crew(
    agents=[agent],
    tasks=[task],
    tool_failure_policy=ToolFailurePolicy.WARN,
)

最具体的设置优先：tool → task → agent → crew → warn。每一 层默认都是 None，表示“从外一层继承”，因此在任何地方都 未配置时，实际生效的默认值是 warn。

检查失败

记录下来的失败是结构化的，因此下游无需解析字符串：

result = crew.kickoff()

if result.has_tool_failures:
    for record in result.tool_failures:
        print(record.tool_name)          # "send_slack_message"
        print(record.failure.code)       # "channel_not_found"
        print(record.failure.reason)     # ToolFailureReason.TOOL_REPORTED
        print(record.summary())

tool_failures 可在 TaskOutput、CrewOutput 和 LiteAgentOutput 上获取。团队可能带着非空列表成功结束——在 把 raw 当作完整结果之前请先检查它。

要在失败发生时做出反应，请订阅该事件：

from crewai.events import ToolFailureDetectedEvent
from crewai.events.event_bus import crewai_event_bus

@crewai_event_bus.on(ToolFailureDetectedEvent)
def on_tool_failure(source, event):
    print(f"{event.tool_name} failed: {event.failure.message} ({event.policy})")

事件会在 raise 策略中止之前发出，因此订阅者总能 观察到该失败。ToolUsageFinishedEvent 还带有一个 failure 字段，让 追踪 UI 无需关联两个事件就能把该调用标记为失败。

结论

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