第 8 章 · 自定义工具开发

本章目标：掌握 BaseTool 子类与 @tool 装饰器两种自定义工具写法，会用 Pydantic 做参数校验、用 cache_function 控制缓存，并能把企业内部 REST API 安全地包装成 Agent 可用工具。

8.1 两种自定义工具写法

CrewAI 提供两条路径，官方文档都位于 Tools 概念页：

继承 BaseTool：适合复杂工具——需要状态、多个方法、类型化输出；
@tool 装饰器：适合"一个函数就是一件事"的简单工具，代码最少。

python

from crewai.tools import BaseTool, tool


# 写法一：@tool 装饰器（推荐起步用）
# 第一个参数是工具名；函数 docstring 会成为工具描述——模型靠它决定何时调用
@tool("汇率换算")
def exchange_rate(amount: float, currency: str) -> str:
    """把金额按固定演示汇率换算成人民币。当需要货币换算时使用。"""
    rates = {"USD": 7.2, "EUR": 7.8, "JPY": 0.048}
    if currency not in rates:
        return f"错误：暂不支持币种 {currency}，支持的币种：{list(rates)}"
    return f"{amount} {currency} ≈ {amount * rates[currency]:.2f} CNY"


# 写法二：继承 BaseTool（需要状态或更细控制时使用）
from typing import Type

from pydantic import BaseModel, Field


class WeatherInput(BaseModel):
    """天气查询工具的输入模式。"""
    city: str = Field(..., description="城市名，例如：北京")
    days: int = Field(3, ge=1, le=7, description="预报天数，1-7")


class WeatherTool(BaseTool):
    name: str = "weather_forecast"
    description: str = "查询城市未来几天的天气预报。回答天气相关问题前必须调用。"
    args_schema: Type[BaseModel] = WeatherInput   # 用 Pydantic 模型声明并校验参数

    def _run(self, city: str, days: int = 3) -> str:
        # 真实场景这里调用天气 API；演示返回固定数据
        return f"{city} 未来 {days} 天：晴转多云，18-26℃。"


weather_tool = WeatherTool()

description 决定调用质量

模型完全依据 description 和参数的 Field(description=...) 来判断"什么时候用这个工具、传什么参数"。写得含糊的工具 = 不会被正确调用的工具。描述里最好包含：功能 + 适用时机 + 关键约束。

8.2 参数校验与失败设计

工具参数经 Pydantic 校验后才会进入 _run。此外还有两个关键约定：

返回值必须是字符串（或定义类型化输出模型）：这是喂给模型的文本；
可恢复的错误返回错误说明，而不是抛异常：让 agent 有机会自我纠正。

python

from typing import Type

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


class OrderQueryInput(BaseModel):
    order_id: str = Field(..., description="订单号，格式如 ORD-2024-0001")

    @field_validator("order_id")
    @classmethod
    def check_format(cls, v: str) -> str:
        if not v.startswith("ORD-"):
            raise ValueError("订单号必须以 ORD- 开头")
        return v


class OrderQueryTool(BaseTool):
    name: str = "query_order"
    description: str = "按订单号查询订单状态。仅在用户提供合法订单号时使用。"
    args_schema: Type[BaseModel] = OrderQueryInput

    def _run(self, order_id: str) -> str:
        db = {"ORD-2024-0001": "已发货"}
        status = db.get(order_id)
        if status is None:
            # 返回可操作的提示而非抛异常，agent 可据此向用户追问
            return f"未找到订单 {order_id}，请与用户核对订单号后重试。"
        return f"订单 {order_id} 当前状态：{status}"

新版 CrewAI 还提供了 ToolFailure（从 crewai.tools.tool_failure 导入）：当工具"没抛异常但确实失败了"时，返回 ToolFailure(message=..., retryable=...) 可以让框架明确记录失败，配合 Crew 的 tool_failure_policy（warn/raise）做统一治理——比往字符串里塞 "error" 让模型猜要可靠得多。

8.3 缓存：cache_function

所有工具默认开启结果缓存（相同参数直接复用上次结果），这对费钱费时的外部调用非常重要。用 cache_function 可以精细控制"什么结果值得缓存"：

python

from crewai.tools import tool


@tool("实时金价")
def gold_price(city: str = "shanghai") -> str:
    """查询当前金价。价格敏感场景请勿依赖缓存。"""
    return "768.50 元/克"   # 演示值


def cache_func(args, result) -> bool:
    # 返回 True 表示"这次结果允许被缓存"
    # 例如：查询失败的结果不缓存，避免错误答案被反复复用
    return "错误" not in result


gold_price.cache_function = cache_func

不需要缓存的工具直接设 gold_price.cache_function = lambda args, result: False 即可每次强制执行。

8.4 实战：把内部 REST API 包装成工具

企业落地最常见的诉求：让 agent 能安全地查内部系统。核心要点是白名单操作 + 参数收敛 + 不泄露实现细节：

python

import os
from typing import Type

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


class CrmCustomerInput(BaseModel):
    """CRM 客户查询输入。"""
    customer_name: str = Field(..., min_length=2, description="客户全名或唯一编号")


class CrmLookupTool(BaseTool):
    name: str = "crm_customer_lookup"
    description: str = (
        "查询公司 CRM 中客户的等级、归属销售和最近订单时间。"
        "仅支持精确名称/编号查询，不支持模糊搜索，一次只查一个客户。"
    )
    args_schema: Type[BaseModel] = CrmCustomerInput

    def _run(self, customer_name: str) -> str:
        try:
            resp = requests.get(
                "https://crm.internal.example.com/api/v1/customers",
                params={"q": customer_name},
                headers={
                    # 服务账号 token 从环境变量读取，绝不硬编码
                    "Authorization": f"Bearer {os.environ['CRM_TOKEN']}",
                },
                timeout=10,
            )
            resp.raise_for_status()
            data = resp.json()
        except requests.Timeout:
            return "错误：CRM 服务超时，请稍后重试。"
        except requests.HTTPError as e:
            return f"错误：CRM 返回 {e.response.status_code}，请检查查询条件。"

        items = data.get("results", [])
        if not items:
            return f"CRM 中没有找到客户「{customer_name}」。"
        c = items[0]
        return (
            f"客户：{c['name']}｜等级：{c['tier']}｜"
            f"归属销售：{c['owner']}｜最近订单：{c['last_order_at']}"
        )


crm_tool = CrmLookupTool()

设计要点复盘：

超时必须设置（timeout=10），否则一次卡死会拖住整个 crew；
异常翻译成人话返回给模型，而不是让它看到 Python traceback 自行猜测；
只暴露必要字段：内部 API 的几十个字段不要原样透传，挑任务相关的输出，省 token 也防信息泄露。

本章小结

简单工具用 @tool 装饰器，复杂工具继承 BaseTool 并实现 _run；
args_schema 用 Pydantic 模型声明参数，校验和文档一步到位；
工具描述（含每个参数的 description）是模型选择与传参的唯一依据；
可恢复错误应返回说明文字或 ToolFailure，让 agent 能自我纠正；
cache_function 控制哪些结果可缓存，外部 API 工具务必设置超时。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 在 BaseTool 子类中，工具的实际执行逻辑应该写在哪个方法里？

2. 关于 args_schema，下列说法正确的是？

3. cache_function 返回 False 意味着什么？

4. 包装内部 REST API 为工具时，下列做法不推荐的是？

🛠️ 动手实践

用 @tool 写一个"密码强度检查"工具，输入密码返回强度评级与改进建议，并用 Pydantic 思路思考：如果改写成 BaseTool 版本，args_schema 应该怎么定义。
把任意一个公开 REST API（如 open-meteo 天气接口）包装成带超时、异常翻译和字段精简的工具。
给第 2 题的工具加上 cache_function：只有成功响应才缓存，然后用同一组参数连续调用两次验证缓存命中。

工具是单个 agent 的手脚。下一章看多个 agent 如何在管理者带领下协作：第 9 章 · 分层流程 hierarchical 与管理者。

第 8 章 · 自定义工具开发 ​

8.1 两种自定义工具写法 ​

8.2 参数校验与失败设计 ​

8.3 缓存：cache_function ​

8.4 实战：把内部 REST API 包装成工具 ​

本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 8 章 · 自定义工具开发

8.1 两种自定义工具写法

8.2 参数校验与失败设计

8.3 缓存：cache_function

8.4 实战：把内部 REST API 包装成工具

本章小结

🛠️ 动手实践