Day 27 / 共 60 天 · 阶段5 工具系统

BaseTool:所有工具的"出厂标准"

前 26 天我们把 Agent / Task / Crew / 流程 / 事件都拆完了。从今天起进入阶段5 工具系统——Agent 之所以能"上网查、算数、读文件",全靠工具。今天读 tools/base_tool.py 里的 BaseTool:它规定了每个工具必须长什么样(名字、描述、参数 schema)、怎么被安全地调用(参数校验、用量上限、缓存开关),以及一个能把普通函数一键变工具的 @tool 装饰器。这是整个工具生态的地基。

📍 你在 60 天里的位置(阶段5 工具系统 · 共 6 天)
阶段4 Crew与流程 D19-26 D27 BaseTool D28 StructuredTool D29 工具调用 D30 缓存复用 D31 MCP 工具 D32 自定义与最佳实践 阶段6 记忆
💡 先用一个类比兜住今天 BaseTool 就像公司给每个新员工发的"岗位说明书 + 工牌"。说明书写清楚:你叫什么(name)、你是干嘛的(description)、干活需要哪些材料(args_schema)。工牌背后还有规矩:一次任务最多刷几次门禁(max_usage_count)、要不要把结果记小本本上省得重复干(cache_function)。有了这套统一标准,Agent 这个"项目经理"面对一百种工具,也能用同一套流程去调用——不用为每个工具单独写对接代码。
L01

痛点:工具五花八门,Agent 怎么统一调?

🤔 痛点一个搜索工具要传"关键词",一个计算器要传"两个数字",一个数据库工具要传"SQL"——它们的输入、输出、实现全不一样。可 Agent 的执行循环(Day 08)里只有一句 tool.invoke(...)。凭什么一句话能调所有工具?框架又怎么知道该给某个工具喂什么参数、参数写错了怎么办、同一个工具会不会被无限调用烧钱?
💡 一句话本质 用一个抽象基类 BaseTool 把"工具"这个概念标准化:它是 Pydantic 模型,强制每个工具都有 name/description/args_schema 三件套,并提供统一的 run() 入口,里面封装了"校验参数 → 检查用量 → 真正执行"三步。真正干活的逻辑塞进抽象方法 _run(),由子类实现。Agent 只认 BaseTool 这个接口,不关心背后是搜索还是算数——这就是"面向接口编程"。

类的骨架(base_tool.py:103),它同时继承 Pydantic 的 BaseModelABC(抽象基类):

# base_tool.py:103
class BaseTool(BaseModel, ABC):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    name: str          # 工具的唯一名字
    description: str   # 告诉模型"何时/为何用它"
    args_schema: type[PydanticBaseModel]      # 参数的 schema
    result_schema: type[PydanticBaseModel] | None
    cache_function: SerializableCallable      # 决定要不要缓存
    result_as_answer: bool = False            # 结果直接当最终答案?
    max_usage_count: int | None = None        # 最多用几次

    @abstractmethod
    def _run(self, *args, **kwargs) -> Any: ...   # ★子类必须实现的"真正干活"
    def run(self, *args, **kwargs) -> Any: ...     # :314 统一入口
    def to_structured_tool(self) -> CrewStructuredTool: ...  # :393 转执行态
大白话BaseModel 让工具的字段能被自动校验、能序列化存档;ABC + @abstractmethod_run 变成"占位坑"——你继承 BaseTool 却不写 _run,程序直接不让你创建实例。这就逼着每个工具作者都得填上"到底干什么活"。而 run(外壳)是框架写好的,你不用碰。
L02

字段全景:一个工具由哪些"标签"组成

核心字段的完整声明(base_tool.py:139 起):

# base_tool.py:139
name: str = Field(description="The unique name of the tool ...")
description: str = Field(description="Used to tell the model how/when/why to use the tool.")
env_vars: list[EnvVar] = Field(default_factory=list, ...)   # 依赖哪些环境变量
args_schema: type[PydanticBaseModel] = Field(
    default=_ArgsSchemaPlaceholder, validate_default=True, ...)  # 参数 schema
result_schema: type[PydanticBaseModel] | None = Field(default=None, ...)  # 输出 schema
# base_tool.py:176
cache_function: SerializableCallable = Field(default=_default_cache_function, ...)
result_as_answer: bool = Field(default=False, ...)   # 工具结果=最终答案
max_usage_count: int | None = Field(default=None, ...)  # 用量上限
current_usage_count: int = Field(default=0, ...)        # 已用次数
_usage_lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)  # 计数锁
name / description最关键的两个。description 是写给 LLM 看的——模型靠它判断"这个活该不该用这个工具"。写得含糊,模型就选错工具。
args_schema参数的 Pydantic 模型。既用来校验模型传来的参数,又用来生成给模型看的"这个工具要哪些参数"的说明。默认是占位符,会自动推断(L03)。
result_as_answer特殊开关:设 True 时,这个工具一返回,Agent 就直接把它当最终答案,不再继续想(Day 08 循环里见过它的身影)。
max_usage_count / current_usage_count一对搭档:上限 + 已用计数。防止 Agent 对同一个工具无限调用(比如死循环搜索)。
_usage_lock私有的线程锁。因为工具可能被并行调用(Day 08 原生模式的线程池),计数必须加锁才不会数错。
数据结构:BaseTool 的字段分三组 ① 身份/说明(给 LLM 看) name description args_schema result_schema ② 行为开关 result_as_answer cache_function max_usage_count env_vars ③ 运行时状态 current_usage_count _usage_lock(私有) 随执行不断变化 _run() ← 抽象方法:子类必须填的"真正干活" ①②③ 是声明式配置,_run 是命令式逻辑——配置与逻辑分离
图注:字段分"给模型看的说明 / 行为开关 / 运行时状态"三组,加一个抽象方法 _run。配置和逻辑彻底分开。
L03

args_schema:你不写,它替你从签名推断

🤔 痛点让工具作者手写一个 Pydantic 参数模型太啰嗦——明明 _run(self, query: str, top_k: int = 5) 已经把参数、类型、默认值都写在签名里了,为啥还要再抄一遍?

所以 BaseTool 用一个 字段校验器 在创建时自动推断(base_tool.py:200):

# base_tool.py:200
@field_validator("args_schema", mode="before")
@classmethod
def _default_args_schema(cls, v):
    if isinstance(v, dict):          # 从存档反序列化:dict → 模型类
        restored = _deserialize_schema(v)
        if restored is not None:
            return restored
    if v is None or v == cls._ArgsSchemaPlaceholder:
        pass                         # 没提供 → 往下走,从签名生成
    elif isinstance(v, type):
        return v                     # 已经给了一个模型类 → 直接用

    run_sig = signature(cls._run)    # ★读 _run 的函数签名
    fields: dict[str, Any] = {}
    for param_name, param in run_sig.parameters.items():
        if param_name in ("self", "return"):
            continue
        if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
            continue   # 跳过 *args / **kwargs
        annotation = param.annotation if param.annotation != param.empty else Any
        if param.default is param.empty:
            fields[param_name] = (annotation, ...)      # 无默认 → 必填
        else:
            fields[param_name] = (annotation, param.default)  # 有默认 → 可选
    if not fields:                   # _run 没参数?再看 _arun(异步版)
        arun_sig = signature(cls._arun)
        ...
    return create_model(f"{cls.__name__}Schema", **fields)   # 动态造一个模型类
mode="before"在 Pydantic 真正校验字段之前先跑——因为要把"没填/占位符"这种情况变成一个真的模型类。
三种输入分流dict(从存档恢复)→ 反序列化;已是类 → 原样用;没填 → 从签名生成。一个校验器兼容三种来源。
signature(cls._run)★核心:用 inspect_run 的参数名、类型标注、默认值。
param.empty 判断没默认值 → (类型, ...)... 是 Pydantic 的"必填"标记);有默认值 → 变可选。
create_model(...)Pydantic 的"动态造类"函数:运行时凭字段字典拼出一个全新的模型类。工具作者一行 schema 都没写。
📝 例子:一个函数签名会推断出什么_run(self, query: str, top_k: int = 5),自动生成的 schema 等价于:
class XxxSchema(BaseModel): query: str; top_k: int = 5
于是 LLM 会被告知:"这个工具需要一个必填的字符串 query 和一个可选的整数 top_k(默认 5)"。
💡 设计取舍①:为什么"约定优于配置" 朴素做法:要求每个工具都显式写 args_schema=MySchema。清晰,但啰嗦、易和签名不一致。源码做法:默认从 _run 签名自动推断,同时保留手写 schema 的口子(传了 type 就用你的)。这就是"约定优于配置"——常见情况零配置,特殊情况可覆盖。代价是"魔法感":新手可能不知道 schema 是从哪冒出来的,所以文档必须写清"我们读你的类型标注"。
L04

run():调用工具前的三道关卡

统一入口 runbase_tool.py:314)——注意它不是直接调 _run,而是先过三关:

# base_tool.py:314
def run(self, *args, **kwargs) -> Any:
    if not args:
        kwargs = self._validate_kwargs(kwargs)   # ① 关卡一:校验参数

    limit_error = self._claim_usage()             # ② 关卡二:占用量名额
    if limit_error:
        return limit_error                        # 超限 → 直接返回错误字符串

    result = self._run(*args, **kwargs)           # ③ 真正干活

    if asyncio.iscoroutine(result):
        result = asyncio.run(result)              # _run 若返回协程 → 跑完它
    return result

关卡一 _validate_kwargsbase_tool.py:272)用 args_schema 做校验,失败时给出可读的提示

# base_tool.py:272
def _validate_kwargs(self, kwargs):
    if self.args_schema is not None and self.args_schema.model_fields:
        try:
            validated = self.args_schema.model_validate(kwargs)
            return validated.model_dump()          # 校验通过 → 用清洗后的值
        except Exception as e:
            hint = build_schema_hint(self.args_schema)   # 拼出"期望哪些参数"
            raise ValueError(
                f"Tool '{self.name}' arguments validation failed: {e}{hint}")
    return kwargs
if not args只有当用关键字方式传参(kwargs)才校验。位置参数(args)跳过校验,交给 _run 自己处理。
model_validate + model_dump用 schema 校验一遍,返回的是清洗/类型转换后的值(比如字符串 "5" 转成 int 5)。
build_schema_hint★体贴设计:校验失败不只是报错,还附上"期望哪些参数、哪些必填"——这段提示会喂回给 LLM,帮它下次把参数写对。
iscoroutine 兜底如果 _run 返回的是协程(async 函数),这里用 asyncio.run 同步跑完,保证 run() 永远返回具体结果。
💡 为什么把校验/限流放外壳 run,把逻辑放 _run这叫模板方法模式:外壳 run 定义"调用工具的固定流程"(校验→限流→执行),可变的只有 _run 那一步。工具作者只需专注写业务逻辑,参数校验、用量控制这些"横切关注点"框架统一兜底。好处:所有工具的安全行为一致,改一处全生效。
L05

用量上限:一把锁保证并行下不数错

关卡二 _claim_usagebase_tool.py:295)——"检查上限 + 计数 +1"必须是一个原子动作

# base_tool.py:295
def _claim_usage(self) -> str | None:
    with self._usage_lock:            # ★加锁:同一时刻只有一个线程能进来
        if (self.max_usage_count is not None
                and self.current_usage_count >= self.max_usage_count):
            return (f"Tool '{self.name}' has reached its usage limit of "
                    f"{self.max_usage_count} times and cannot be used anymore.")
        self.current_usage_count += 1   # 没超限 → 占一个名额
        return None                     # None = 允许执行
大白话把"看还剩几个名额"和"抢一个名额"这两步用锁绑成一个动作。否则并行时会出现:线程 A 和 B 同时看到"还剩最后 1 次",两个都以为轮到自己,结果超额执行。加了锁,A 抢到后 B 只能等 A 出锁再看,这时计数已经满了,B 就被拦下。返回 None 表示"放行",返回一段错误文字表示"到顶了"。
⚠️ 边界:超限不是抛异常,而是返回一段"话" 注意 _claim_usage 超限时返回的是一个字符串,然后 run 里直接 return limit_error——不 raise。为什么?因为这个字符串会作为"工具的观察结果"喂回给 LLM(Day 08 的循环)。LLM 看到"这个工具用到上限了,别再用了",就会改用别的工具或直接给答案。如果这里 raise,整个 Agent 就崩了。把"限制"表达成模型能理解的自然语言反馈,而不是程序异常——这是 Agent 框架里很典型的"面向 LLM 的错误处理"。

👶 小白:current_usage_count 存在工具实例上,那两次 kickoff 之间会不会串?

👨‍🏫 老师:会,所以有 reset_usage_count()base_tool.py:371)把计数清零。框架在合适的时机调它,保证每次新任务从 0 开始数。这跟 Day 08 里 invoke 每次把 iterations=0 是一个道理——运行时状态要在新一轮开始时复位

L06

@tool 装饰器:三行把普通函数变工具

大多数人不会去继承 BaseTool,而是用 @tool 装饰器(base_tool.py:676)。它支持三种写法:

# base_tool.py:676(精简)
def tool(*args, result_schema=None, result_as_answer=False, max_usage_count=None):
    def _make_with_name(tool_name: str):
        def _make_tool(f):
            if f.__doc__ is None:
                raise ValueError("Function must have a docstring")   # ★强制写文档
            func_sig = signature(f)
            fields = {}
            for param_name, param in func_sig.parameters.items():
                ...   # 和 L03 一样:从签名抽字段
            class_name = "".join(tool_name.split()).title()
            args_schema = create_model(class_name, **fields)
            return Tool(name=tool_name, description=f.__doc__, func=f,   # ★描述=docstring
                        args_schema=args_schema, result_as_answer=result_as_answer,
                        max_usage_count=max_usage_count, ...)
        return _make_tool
    if len(args) == 1 and callable(args[0]):        # 用法1:@tool(裸用)
        return _make_with_name(args[0].__name__)(args[0])
    if len(args) == 1 and isinstance(args[0], str): # 用法2:@tool("自定义名")
        return _make_with_name(args[0])
    if len(args) == 0:                              # 用法3:@tool(result_as_answer=True)
        def decorator(f): return _make_with_name(f.__name__)(f)
        return decorator
    raise ValueError("Invalid arguments")
f.__doc__ is None → 报错★强制函数有 docstring——因为 docstring 会成为工具的 description,是给 LLM 的说明书。没说明书模型就不会用。
description=f.__doc__直接把函数的文档字符串当描述。所以写工具时,docstring 要写"这个工具做什么、什么时候用",写给模型看。
返回 Tool(不是 BaseTool)ToolBaseTool 的具体子类(base_tool.py:496),它的 _run 就是"调你包进来的那个函数 func"。
三种 if 分流靠判断"第一个位置参数是函数/字符串/没有"来区分裸用、带名字、带选项三种调用方式。这是 Python 装饰器兼容"带不带括号"的经典写法。
📝 例子:@tool 实战
from crewai.tools import tool

@tool("天气查询")
def get_weather(city: str) -> str:
    """查询某城市当前天气,输入城市名,返回温度描述。"""
    return f"{city} 当前 18°C 晴"

# get_weather 现在是一个 Tool 实例
print(get_weather.run(city="上海"))   # → 上海 当前 18°C 晴
这个 get_weather 就能直接塞进 Agent(tools=[get_weather])。docstring 变成描述,city: str 变成 args_schema。
L07

description 的秘密:它被偷偷塞进了 schema

工具创建后有个 model_post_init 钩子会调 _generate_descriptionbase_tool.py:482),把参数 schema拼进描述里:

# base_tool.py:482
def _generate_description(self) -> None:
    schema = generate_model_description(self.args_schema)
    args_json = json.dumps(schema["json_schema"]["schema"], indent=2)
    self.description = (
        f"Tool Name: {sanitize_tool_name(self.name)}\n"
        f"Tool Arguments: {args_json}\n"        # ★把参数 JSON schema 拼进来
        f"Tool Description: {self.description}")
model_post_init 触发Pydantic 在实例创建完成后自动调 model_post_initbase_tool.py:267),它调 _generate_description。所以你无需手动触发。
拼三段最终的 description = 名字 + 参数 JSON schema + 你写的描述。LLM 看到的是这一整段——它由此知道"这工具叫啥、要哪些参数、干嘛用"。
sanitize_tool_name把名字里的空格等清洗成规范格式(后面 D29 选择工具时会反复用到,保证名字可比对)。
控制流:从函数到"喂给模型的说明" 你的函数 + docstring + 类型标注 @tool / 签名推断 → args_schema Tool 实例 (BaseTool 子类) _generate_description Name + Arguments(JSON) + Description → 喂给 LLM
图注:函数 → 推断 schema → 造 Tool → 生成含 schema 的 description。这段文字最终决定 LLM 能不能正确调用工具。
💡 设计取舍②:为什么把 JSON schema 塞进 description 文字里 文本 ReAct 模式(Day 08)下,模型不吃"结构化的工具定义",只吃 prompt 里的一段文字。所以框架把参数 schema 序列化成 JSON 塞进 description,让不支持原生函数调用的模型也能"看懂参数"。代价是 description 变长、占 token。但换来的是全模型兼容——原生模式会另走结构化路径(D29),这段冗余文字主要服务于老模型的兜底。这是"最大兼容"压倒"最省 token"的取舍。
L08

边界 + 今日小结

⚠️ 边界/坑:忘了写 docstring 或类型标注@tool 时如果函数没有 docstring,直接 raise ValueError("Function must have a docstring")base_tool.py:710)——因为没描述,LLM 无从判断何时用它。同理,参数没写类型标注不会报错,但会被当成 Any,schema 就失去约束,模型可能乱传参。反模式:写工具时图省事省掉 docstring 和类型标注——这等于把工具变成"哑巴 + 不设防",Agent 要么不用它、要么用错它。

👶 小白:BaseTool(抽象基类)和 Tool(具体类)到底啥关系?我该继承哪个?

👨‍🏫 老师:想把"一个已有的函数"变工具 → 用 @tool 装饰器(背后造 Tool)。想写一个"有状态、逻辑复杂"的工具(比如带连接池的数据库工具)→ 继承 BaseTool 并实现 _runTool 只是 BaseTool 的一个"包函数"特化子类。D32 会专门讲怎么选。

🧠 今天你应该能回答

  • 为什么工具要有一个统一基类?Agent 靠什么"一句话调所有工具"?
  • BaseTool 的字段分哪三组?result_as_answer 是干嘛的?
  • args_schema 不写会怎样?它从哪里推断出来?
  • run() 在调 _run 前过哪三关?为什么这样分层(模板方法)?
  • 用量上限为什么要加锁?超限为什么"返回文字"而不是抛异常?
  • @tool 的三种写法?为什么强制 docstring?

✋ 10 分钟动手

P=lib/crewai/src/crewai/tools
sed -n '103,199p' $P/base_tool.py    # 字段全景
sed -n '200,258p' $P/base_tool.py    # args_schema 自动推断
sed -n '295,331p' $P/base_tool.py    # _claim_usage + run
sed -n '676,761p' $P/base_tool.py    # @tool 装饰器
python -c "
from crewai.tools import tool
@tool
def add(a:int,b:int)->int:
    '''把两个整数相加'''
    return a+b
print(type(add), add.name)
print(add.description)          # 看 schema 被拼进描述
print(add.run(a=2,b=3))
"
明日预告 · Day 28:今天 run() 是"配置态"的入口,但 Agent 执行循环真正调用的是 to_structured_tool() 转出的 CrewStructuredTool——那才是"执行态"的工具。明天读 structured_tool.py:它的 invoke/_parse_args 怎么把 LLM 传来的字符串参数解析成 dict、怎么处理同步/异步、result_schema 怎么格式化输出。
← Day 26 事件系统 Day 28 · StructuredTool →