CrewAI 工具集成与自定义工具

概念速查

CrewAI 工具是 Agent 可调用的函数式能力单元。每个工具封装一个原子操作——搜索、计算、读写文件——Agent 在执的 Task 中自主决定何时调用。框架内置两类创建接口：

@tool 装饰器（快捷方式，crewai>=0.30.0）：

from crewai.tools import tool

@tool("网页抓取器")
def fetch_page(url: str) -> str:
    """抓取指定 URL 的文本内容"""
    # 工具逻辑
    return "页面内容"

BaseTool 子类化（完整控制，crewai>=0.30.0）：

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

class SearchInput(BaseModel):
    query: str = Field(..., description="搜索关键词")

class SearchTool(BaseTool):
    name: str = "搜索工具"
    description: str = "执行搜索引擎查询，返回结构化结果"
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str) -> str:
        # 调用搜索 API 并返回
        return "搜索结果"

装饰器适用于无状态、单函数工具；BaseTool 适用于需输入校验（Pydantic schema）、状态管理或混合同步/异步逻辑的场景。

工具分类

CrewAI 预置工具按职能大致分四类：

类别	代表工具	用途
搜索检索	SerperDevTool, WebsiteSearchTool, FirecrawlSearchTool	网页搜索与内容抓取
代码执行	CodeInterpreterTool	运行 Python 代码
文件操作	FileReadTool, DirectoryReadTool, CSVSearchTool, PDFSearchTool	读/搜本地文件
RAG 检索	RagTool, PGSearchTool, GithubSearchTool	对文档/数据库做语义搜索

底层原理

工具协定与执行流程

CrewAI 在底层将每个工具包装为一个 CrewStructuredTool，内部维护 name、description、args_schema、_run 与可选的 _arun。当 LLM 在 Agent 的 System Prompt 中看到工具描述后，自主决策调用并输出 JSON 格式参数。框架反序列化该 JSON 为 Pydantic 模型，传入 _run 执行。

关键约束：description 是 LLM 选工具的唯一依据。描述越精确，Agent 选错工具的概率越低。字段名与 LLM 能理解的自然语言一致。

缓存机制

缓存发生在工具层，key 由工具名 + 参数 JSON 组成：

# crewai>=0.30.0
from crewai.tools import tool

@tool("乘法器")
def multiply(a: int, b: int) -> int:
    """将两个数相乘"""
    return a * b

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

multiply.cache_function = cache_only_even

cache_function 签名是 (args: dict, result: Any) -> bool，返回 True 则写入缓存。默认所有工具开启缓存，若某工具结果不宜复用（如余额查询），通过 cache_function 返回 False 关闭。

缓存存储在 Crew 级别的 Cache 实例中，跨 Task 共享——同一 Crew 内多次相同参数调用只产生一次真实执行。

错误处理

_run 内的任何异常会被框架捕获并包装为工具调用失败消息返回给 LLM。Agent 可据此尝试修正参数后重试，或放弃该工具改用其他手段。

# crewai>=0.30.0
from crewai.tools import BaseTool

class SafeSearchTool(BaseTool):
    name: str = "安全搜索"
    description: str = "执行搜索，出错时返回友好提示"

    def _run(self, query: str) -> str:
        try:
            # 可能抛异常的 API 调用
            return self.api.search(query)
        except ConnectionError:
            return "[搜索服务不可用，使用缓存摘要]"  # LLM 看到后继续工作
        except RateLimitError:
            return "[请求过快，等待后重试]"

工具应返回字符串而非抛出异常。框架虽会兜底捕获，但自定义错误消息让 LLM 有上下文做补救决策，而裸异常则只让 Agent 知道"失败了"。

LangChain 工具互操作

所有 langchain 工具均可通过包装接入 CrewAI。crewai>=0.30.0 不直接提供 from_langchain 工厂，而是通过 BaseTool 子类化封装 LangChain 原生工具实例：

# crewai>=0.30.0, langchain-community>=0.3.0
from crewai.tools import BaseTool
from pydantic import Field
from langchain_community.utilities import SerpAPIWrapper

class LangChainSearchTool(BaseTool):
    name: str = "Web Search"
    description: str = "通过 Google Search 获取实时信息"
    search: SerpAPIWrapper = Field(default_factory=SerpAPIWrapper)

    def _run(self, query: str) -> str:
        return self.search.run(query)

这是推荐的集成方式——不引入多余依赖，在 CrewAI 的 Pydantic 序列化体系内工作。若需批量包装已有 LangChain 工具列表，可自行实现一个循环工厂方法。

架构设计原则

单一职责 + 描述即契约

一个工具做一件事，且用 description 精准告诉 LLM 这件事是什么。不要写"全能工具"（一个工具既能搜网页又能写文件），那会迫使 LLM 靠参数猜测行为，增加决策成本。

反例：

description: "通用数据处理工具"

正例：

description: "当需要根据关键词搜索最新技术新闻时使用该工具。返回标题、来源和发布日期。"

缓存策略与结果幂等

缓存提升效率但引入因果风险。设计原则：

• 读操作（搜索、查 DB）—— 默认缓存，cache_function 细化条件
• 写操作（发邮件、写文件）—— cache_function 返回 False
• 实时数据（余额、汇率）—— cache_function 返回 False 或使用短 TTL

错误信息的语义价值

工具返回的错误字符串要带有状态提示而非只传递原因。LLM 不理解 HTTP 状态码，但它能理解 “[API 扣费余额不足]”。错误消息的原则：让 Agent 知道发生了什么 + 下一步能做什么。

工具粒度与 Task 层级匹配

Task 是"一篇文章"时，对应工具是"搜索"+“写作”。不要提供"搜索+总结+翻译+格式化"的"大统一工具"——那让 LLM 不得不一次性做完所有事，失去中间推理步骤。粗粒度工具适合简单子 Task，细粒度工具交给负责推理的 Agent。

---

以上原则贯穿 CrewAI 工具系统全链路：从 @tool 装饰器的快速定义，到 BaseTool 的完整控制，再到缓存、错误、LangChain 互操作的工程落地。按接口选方式、按场景调缓存、按职责拆工具，即可获得干净高效的 Agent 工具层。