创建自定义工具

在 CrewAI 中创建和使用工具

本指南详细说明了如何为 CrewAI 框架创建自定义工具，以及如何高效管理和使用这些工具，涵盖工具委派、错误处理和动态工具调用等最新功能。它还强调了协作工具的重要性，使智能体能够执行更广泛的操作。

继承 `BaseTool`

要创建一个个性化工具，请继承 BaseTool 并定义必要属性，包括用于输入校验的 args_schema 和 _run 方法。

from typing import Type
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"

使用 `tool` 装饰器

你也可以使用 @tool 装饰器。这种方式允许你直接在函数中定义工具的属性和功能，从而以简洁高效的方式创建符合需求的专用工具。

from crewai.tools import tool

@tool("Tool Name")
def my_simple_tool(question: str) -> str:
    """Tool description for clarity."""
    # Tool logic here
    return "Tool output"

最佳实践：定义带类型的输出

当工具返回结构化数据时，请定义一个 Pydantic 输出模型。这能帮助智能体把结果读成清晰的字段，而不是从纯文本里猜测。

对于具有稳定字段的结果，例如 ID、状态值、分数、价格或列表，类型化输出特别有用。简短的纯文本结果仍然完全可以接受。

直接的 Python 调用仍然会收到工具返回的值。当智能体使用类型化工具时，CrewAI 会根据输出模型向智能体发送 JSON。

返回 Pydantic 模型

当你的 BaseTool 带有 Pydantic 返回类型注解时，CrewAI 会推断输出模式。

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

class InventoryResult(BaseModel):
    sku: str = Field(description="The product SKU.")
    quantity: int = Field(description="Units available.")
    needs_reorder: bool = Field(description="Whether the item should be reordered.")

class InventoryTool(BaseTool):
    name: str = "Inventory Check"
    description: str = "Check 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()
result = tool.run(sku="SKU-123")

# Direct Python calls receive the raw Pydantic object.
print(result.quantity)

当智能体调用 InventoryTool 时，它会收到如下 JSON：

{"sku":"SKU-123","quantity":14,"needs_reorder":false}

将 `result_schema` 与字典结果一起使用

如果你的工具返回字典，请显式设置 result_schema。你可以在 BaseTool 子类上这样做，也可以使用 @tool 装饰器：

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

class ProductResult(BaseModel):
    sku: str = Field(description="The product SKU.")
    name: str = Field(description="The product name.")
    in_stock: bool = Field(description="Whether the product is available.")

@tool("Product Lookup", result_schema=ProductResult)
def product_lookup(sku: str) -> dict[str, object]:
    """Look up product availability by SKU."""
    catalog = {
        "SKU-123": ("Noise-canceling headset", True),
        "SKU-456": ("USB-C dock", False),
    }
    name, in_stock = catalog.get(sku, ("Unknown product", False))
    return {
        "sku": sku,
        "name": name,
        "in_stock": in_stock,
    }

自定义发送给智能体的文本

默认情况下，类型化工具输出会以 JSON 形式发送给智能体。如果你希望智能体接收到简短摘要，请继承 BaseTool 并重写 format_output_for_agent。

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

class InventoryResult(BaseModel):
    sku: str = Field(description="The product SKU.")
    quantity: int = Field(description="Units available.")
    needs_reorder: bool = Field(description="Whether the item should be reordered.")

class InventoryTool(BaseTool):
    name: str = "Inventory Check"
    description: str = "Check 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}."

tool = InventoryTool()
result = tool.run(sku="SKU-123")

# Direct Python calls receive the raw Pydantic object.
print(result.quantity)

这个重写只会改变智能体看到的内容。对 tool.run(...) 的直接调用仍然返回正常的 Python 值。

为工具定义缓存函数

要通过缓存优化工具性能，可以定义自定义的缓存策略并赋给 cache_function 属性。

@tool("Tool with Caching")
def cached_tool(argument: str) -> str:
    """Tool functionality description."""
    return "Cacheable result"

def my_cache_strategy(arguments: dict, result: str) -> bool:
    # Define custom caching logic
    return True if some_condition else False

cached_tool.cache_function = my_cache_strategy

创建异步工具

CrewAI 支持用于非阻塞 I/O 操作的异步工具。当你的工具需要进行 HTTP 请求、数据库查询或其他 I/O 密集型操作时，这尤其有用。

将 `@tool` 装饰器与异步函数一起使用

创建异步工具最简单的方式，就是把 @tool 装饰器与异步函数结合使用：

import aiohttp
from crewai.tools import tool

@tool("Async Web Fetcher")
async def fetch_webpage(url: str) -> str:
    """Fetch content from a webpage asynchronously."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

继承 `BaseTool` 并支持异步

如果你需要更多控制，可以继承 BaseTool，同时实现 _run（同步）和 _arun（异步）方法：

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

class WebFetcherInput(BaseModel):
    """Input schema for WebFetcher."""
    url: str = Field(..., description="The URL to fetch")

class WebFetcherTool(BaseTool):
    name: str = "Web Fetcher"
    description: str = "Fetches content from a URL"
    args_schema: type[BaseModel] = WebFetcherInput

    def _run(self, url: str) -> str:
        """Synchronous implementation."""
        return requests.get(url).text

    async def _arun(self, url: str) -> str:
        """Asynchronous implementation for non-blocking I/O."""
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.text()

遵循这些指南，并把新的功能和协作工具纳入你的工具创建与管理流程，你就能充分发挥 CrewAI 框架的能力，同时提升开发体验和智能体效率。