BaseTool:所有工具的"出厂标准"
前 26 天我们把 Agent / Task / Crew / 流程 / 事件都拆完了。从今天起进入阶段5 工具系统——Agent 之所以能"上网查、算数、读文件",全靠工具。今天读 tools/base_tool.py 里的 BaseTool:它规定了每个工具必须长什么样(名字、描述、参数 schema)、怎么被安全地调用(参数校验、用量上限、缓存开关),以及一个能把普通函数一键变工具的 @tool 装饰器。这是整个工具生态的地基。
BaseTool 就像公司给每个新员工发的"岗位说明书 + 工牌"。说明书写清楚:你叫什么(name)、你是干嘛的(description)、干活需要哪些材料(args_schema)。工牌背后还有规矩:一次任务最多刷几次门禁(max_usage_count)、要不要把结果记小本本上省得重复干(cache_function)。有了这套统一标准,Agent 这个"项目经理"面对一百种工具,也能用同一套流程去调用——不用为每个工具单独写对接代码。痛点:工具五花八门,Agent 怎么统一调?
tool.invoke(...)。凭什么一句话能调所有工具?框架又怎么知道该给某个工具喂什么参数、参数写错了怎么办、同一个工具会不会被无限调用烧钱?BaseTool 把"工具"这个概念标准化:它是 Pydantic 模型,强制每个工具都有 name/description/args_schema 三件套,并提供统一的 run() 入口,里面封装了"校验参数 → 检查用量 → 真正执行"三步。真正干活的逻辑塞进抽象方法 _run(),由子类实现。Agent 只认 BaseTool 这个接口,不关心背后是搜索还是算数——这就是"面向接口编程"。类的骨架(base_tool.py:103),它同时继承 Pydantic 的 BaseModel 和 ABC(抽象基类):
# 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(外壳)是框架写好的,你不用碰。字段全景:一个工具由哪些"标签"组成
核心字段的完整声明(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 原生模式的线程池),计数必须加锁才不会数错。args_schema:你不写,它替你从签名推断
_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 是从哪冒出来的,所以文档必须写清"我们读你的类型标注"。run():调用工具前的三道关卡
统一入口 run(base_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_kwargs(base_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 那一步。工具作者只需专注写业务逻辑,参数校验、用量控制这些"横切关注点"框架统一兜底。好处:所有工具的安全行为一致,改一处全生效。用量上限:一把锁保证并行下不数错
关卡二 _claim_usage(base_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 = 允许执行
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 是一个道理——运行时状态要在新一轮开始时复位。
@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)Tool 是 BaseTool 的具体子类(base_tool.py:496),它的 _run 就是"调你包进来的那个函数 func"。三种 if 分流靠判断"第一个位置参数是函数/字符串/没有"来区分裸用、带名字、带选项三种调用方式。这是 Python 装饰器兼容"带不带括号"的经典写法。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。description 的秘密:它被偷偷塞进了 schema
工具创建后有个 model_post_init 钩子会调 _generate_description(base_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_init(base_tool.py:267),它调 _generate_description。所以你无需手动触发。拼三段最终的 description = 名字 + 参数 JSON schema + 你写的描述。LLM 看到的是这一整段——它由此知道"这工具叫啥、要哪些参数、干嘛用"。sanitize_tool_name把名字里的空格等清洗成规范格式(后面 D29 选择工具时会反复用到,保证名字可比对)。边界 + 今日小结
@tool 时如果函数没有 docstring,直接 raise ValueError("Function must have a docstring")(base_tool.py:710)——因为没描述,LLM 无从判断何时用它。同理,参数没写类型标注不会报错,但会被当成 Any,schema 就失去约束,模型可能乱传参。反模式:写工具时图省事省掉 docstring 和类型标注——这等于把工具变成"哑巴 + 不设防",Agent 要么不用它、要么用错它。👶 小白:BaseTool(抽象基类)和 Tool(具体类)到底啥关系?我该继承哪个?
👨🏫 老师:想把"一个已有的函数"变工具 → 用 @tool 装饰器(背后造 Tool)。想写一个"有状态、逻辑复杂"的工具(比如带连接池的数据库工具)→ 继承 BaseTool 并实现 _run。Tool 只是 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))
"
run() 是"配置态"的入口,但 Agent 执行循环真正调用的是 to_structured_tool() 转出的 CrewStructuredTool——那才是"执行态"的工具。明天读 structured_tool.py:它的 invoke/_parse_args 怎么把 LLM 传来的字符串参数解析成 dict、怎么处理同步/异步、result_schema 怎么格式化输出。