跳转到主要内容

文档索引

获取完整文档索引： https://docs.crewai.com.cn/llms.txt

在深入了解之前，请使用此文件来浏览所有可用页面。

​
在 CrewAI 中创建和使用工具

本指南提供了在 CrewAI 框架内创建自定义工具的详细说明，以及如何高效管理和使用这些工具，包括工具委托、错误处理和动态工具调用等最新功能。指南还强调了协作工具的重要性，使智能体能够执行广泛的操作。
想要向社区发布您的工具吗？如果您构建的工具能为他人带来帮助，请查看发布自定义工具指南，了解如何打包并在 PyPI 上分发您的工具。

​
继承 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"

​
为工具定义缓存函数

若要通过缓存优化工具性能，可使用 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 框架的强大功能，从而提升开发体验以及 AI 智能体的工作效率。