工具缓存:让 Agent 不做重复的苦活
Agent 转圈时经常会用同一个工具、同样的参数重复调用——比如反复搜同一个词。每次真跑一遍既慢又费钱(有些工具还收费)。今天读缓存三件套:agents/tools_handler.py(记账中枢 ToolsHandler)、agents/cache/cache_handler.py(带读写锁的 CacheHandler)、以及 tools/tool_usage.py 里的缓存分支。看清楚:什么时候读缓存、什么时候写、cache_function 怎么细粒度控制、以及"连续调同一个工具同参数"怎么被拦下。
ToolsHandler 是那个前台,CacheHandler 是那本登记本(key = "工具名-参数")。而 cache_function 是一条规矩:"哪些问题值得记"(比如实时股价就别记,会过期)。还有一条铁律:同一个人连问两遍完全一样的问题,前台会说"你刚问过了"——这就是防重复调用。痛点:重复劳动又慢又烧钱
cache_function 让每个工具自定义"这次结果值不值得缓存"(实时数据就别缓存);② _check_tool_repeated_usage 拦住"紧接着又调完全相同的一次",直接回一句"你刚用过"逼模型换招。缓存省钱省时,防重复救死循环。三个角色的分工:
| 角色 | 文件 | 职责 |
|---|---|---|
ToolsHandler | tools_handler.py:15 | 记账中枢:记"最后用的工具"、决定写不写缓存 |
CacheHandler | cache_handler.py | 真正的存储:一个带读写锁的字典 |
| 缓存分支 | tool_usage.py:516 | 执行流程里"先读缓存 / 后写缓存"的钩子 |
ToolsHandler:记账中枢
整个 tools_handler.py 很短(tools_handler.py:15):
# tools_handler.py:15
class ToolsHandler(BaseModel):
cache: CacheHandler | None = Field(default=None)
last_used_tool: ToolCalling | InstructorToolCalling | None = Field(default=None)
def on_tool_use(self, calling, output, should_cache=True) -> None:
self.last_used_tool = calling # ① 记下"最后用的工具"
if self.cache and should_cache and calling.tool_name != CacheTools().name:
input_str = ""
if calling.arguments:
if isinstance(calling.arguments, dict):
input_str = json.dumps(calling.arguments) # 参数序列化成字符串
else:
input_str = str(calling.arguments)
self.cache.add( # ② 写缓存
tool=calling.tool_name, input=input_str, output=output)
last_used_tool记住"上一次调的是哪个工具、什么参数"。这是 L06 防重复调用的依据。on_tool_use每次工具执行完都调它。做两件事:更新 last_used_tool、(满足条件时)写缓存。三个写缓存条件① 有 cache;② should_cache 为真(由 cache_function 决定,L05);③ 不是缓存工具自己(CacheTools().name,避免自己缓存自己造成套娃)。参数序列化把 arguments 转成字符串当 key 的一部分——因为字典不能直接当稳定 key,序列化后 {"a":1} 才有确定的字符串形式。ToolsHandler 就是那个"前台记录员",本身不存东西(存的活交给 CacheHandler),只负责在每次工具用完后拍板"记不记、怎么记",并顺手记下"刚才用的是谁"。它是执行循环和缓存存储之间的中间人。CacheHandler:一个带读写锁的字典
真正的存储 CacheHandler(cache_handler.py)——本质是个字典 + 读写锁:
# cache_handler.py
class CacheHandler(BaseModel):
_cache: dict[str, Any] = PrivateAttr(default_factory=dict)
_lock: RWLock = PrivateAttr(default_factory=RWLock) # 读写锁
def add(self, tool: str, input: str, output: Any) -> None:
with self._lock.w_locked(): # ★写锁:独占
self._cache[f"{tool}-{input}"] = output # key = "工具名-参数串"
def read(self, tool: str, input: str) -> Any | None:
with self._lock.r_locked(): # ★读锁:可并发
return self._cache.get(f"{tool}-{input}")
key = f"{tool}-{input}"缓存键就是"工具名 + 参数串"拼起来。相同工具 + 相同参数 = 相同 key = 命中缓存。w_locked()(写锁)写入时独占:同一时刻只能一个线程写,防止并发写坏字典。r_locked()(读锁)★读取时可并发:多个线程能同时读(读不改数据,无冲突),只在有写时才互斥。这就是读写锁比普通互斥锁高效的地方。PrivateAttr_cache/_lock 是私有属性,不参与 Pydantic 序列化——缓存是运行时状态,不该被存档。执行前:先敲敲缓存的门
_use 一开始就尝试读缓存(tool_usage.py:516):
# tool_usage.py:516
from_cache = False
result = None
try:
if self.tools_handler and self.tools_handler.cache:
input_str = ""
if calling.arguments:
if isinstance(calling.arguments, dict):
input_str = json.dumps(calling.arguments)
else:
input_str = str(calling.arguments)
result = self.tools_handler.cache.read( # ★先读缓存
tool=sanitize_tool_name(calling.tool_name), input=input_str)
from_cache = result is not None # 命中标记
...
elif result is None: # ★没命中才真执行
# ... 真正 tool.invoke(...) ...
cache.read(...)用和写入时一致的 key(sanitize_tool_name + 序列化参数)去查。查到了 result 就有值。from_cache = result is not None命中缓存的标记。后面事件(on_tool_use_finished)会带上这个标记,让观测端知道"这次是走缓存的、没真跑"。elif result is None → 真执行★只有缓存没命中(result 还是 None)才走真正的 tool.invoke。命中了就跳过执行,直接用缓存值。key 对齐是关键读用 sanitize_tool_name(calling.tool_name),写(on_tool_use)用 calling.tool_name——注意 D30 里读写 key 生成必须一致,否则永远命不中。cache_function:这次结果到底该不该缓存
执行完后要不要写缓存,由每个工具的 cache_function 决定(tool_usage.py:591):
# tool_usage.py:591
if self.tools_handler:
should_cache = True
# 从原工具(to_structured_tool 转来的)上找 cache_function
original_tool = getattr(available_tool, "_original_tool", None)
cache_func = None
if original_tool and hasattr(original_tool, "cache_function"):
cache_func = original_tool.cache_function
elif hasattr(available_tool, "cache_function"):
cache_func = available_tool.cache_function
if cache_func:
should_cache = cache_func(calling.arguments, result) # ★传入参数和结果,返回布尔
self.tools_handler.on_tool_use(
calling=calling, output=result, should_cache=should_cache) # 把决定传给中枢
默认的 cache_function(Day 27 见过,base_tool.py:81)永远返回 True:
# base_tool.py:81
def _default_cache_function(_args=None, _result=None) -> bool:
"""Default cache function that always allows caching."""
return True
should_cache 默认 True没自定义就默认缓存一切。多数工具(搜索、计算)缓存都安全。先找 _original_tool 的因为执行态是 CrewStructuredTool(D28),cache_function 挂在原 BaseTool 上,要通过 _original_tool 指针取回。取不到再看执行态自己有没有。cache_func(arguments, result)★把"这次的参数"和"这次的结果"都传给它,让它据此判断。比如结果是"服务暂时不可用"就别缓存,参数含时间戳就别缓存。结果传给 on_tool_use把 should_cache 交给 L02 的中枢,由它最终决定写不写。from crewai.tools import tool
@tool
def stock_price(symbol: str) -> str:
"""查询实时股价"""
return fetch_price(symbol)
# 实时数据缓存了就过期 → 让 cache_function 永远返回 False
stock_price.cache_function = lambda args, result: False
这样每次都真查,不会返回一个过时的旧价。而"计算 π 的前 100 位"这种确定性结果,就该用默认的 True 缓存。防连续重复:你刚刚才用过
缓存解决"重复调用还得走一遍流程",而 _check_tool_repeated_usage(tool_usage.py:728)直接拦住"紧接着又调完全一样的一次":
# tool_usage.py:728
def _check_tool_repeated_usage(self, calling) -> bool:
if not self.tools_handler:
return False
if last_tool_usage := self.tools_handler.last_used_tool: # 拿"上一次用的工具"
return (
sanitize_tool_name(calling.tool_name)
== sanitize_tool_name(last_tool_usage.tool_name) # 工具名相同
) and (calling.arguments == last_tool_usage.arguments) # 且参数完全相同
return False
命中时 _use 会直接返回一句提示(tool_usage.py:476):
# tool_usage.py:476
if self._check_tool_repeated_usage(calling=calling):
result = I18N_DEFAULT.errors("task_repeated_usage").format(tool_names=self.tools_names)
self.last_raw_result = result
self._telemetry.tool_repeated_usage(...)
return self._format_result(result=result) # 直接回"你重复了",不执行
比对 last_used_tool只看"紧挨着的上一次"。工具名 + 参数都完全一样,才算"重复"。命中 → 返回提示返回一句"你刚用过这个工具,别重复了,可用工具有……",喂回给 LLM,逼它换个动作(换工具或给答案),打破死循环。和缓存的区别缓存:相同调用返回上次结果(还是有个"结果");防重复:相同的连续调用直接拒绝执行并提醒。前者省成本,后者破死循环。最近一次。因为真正有害的是"卡在原地反复调同一个"(死循环),而隔了几步又调回同一个工具往往是合理的(比如查完 A、算了别的、又回来查 A 的新角度)。只看上一次,既能斩断紧邻死循环,又不误伤正常的"迂回复用"。用最小的状态(一个 last_used_tool)解决主要问题——简单且够用。用量上限与"结果即答案"的收尾
执行前还会查用量上限(tool_usage.py:546,配合 Day 27 的 max_usage_count):
# tool_usage.py:546
usage_limit_error = self._check_usage_limit(available_tool, sanitize_tool_name(tool.name))
if usage_limit_error:
result = usage_limit_error
self.last_raw_result = result
self._telemetry.tool_usage_error(llm=self.function_calling_llm)
result = self._format_result(result=result) # 超限 → 返回提示,不执行
elif result is None:
... # 未超限且未命中缓存 → 真执行
执行成功后,若工具设了 result_as_answer,会标记"这就是最终答案"(tool_usage.py:623):
# tool_usage.py:623
if (hasattr(available_tool, "result_as_answer")
and available_tool.result_as_answer):
result_as_answer = available_tool.result_as_answer
data["result_as_answer"] = result_as_answer # 传给执行循环 → 直接收尾
_check_usage_limitDay 27 的 max_usage_count 在这里生效:超限返回一段"这个工具用到上限了"的话(不 raise,喂回 LLM),让它别再用。顺序:防重复 → 缓存 → 上限 → 执行层层过滤,每一层能拦下就不进下一层。缓存命中和超限都直接返回,省掉真执行。result_as_answer 标记回到 Day 08 循环:带这个标记的结果会让执行器直接把它当 Final Answer,不再继续推理。适合"生成报告""最终交付物"类工具。边界 + 今日小结
CacheHandler._cache 就是个普通 dict,存在当前进程内存里:进程一停,缓存全没;多进程/分布式部署时各进程各存各的,不共享。而且它没有 TTL(过期时间)——一旦缓存了某结果,同参数调用永远返回它,直到进程结束。坑:把一个"结果会随时间变化"的工具用默认缓存(比如"今天天气"缓存了,明天还返回昨天的)。解法:这类工具务必用 cache_function 返回 False,或在参数里带上日期让 key 天然区分。生产级需求(跨进程、带过期)得自己接外部缓存。👶 小白:那个 CacheTools(cache_tools.py)是干嘛的?和 CacheHandler 有啥区别?
👨🏫 老师:CacheHandler 是"存储"(后台的登记本)。CacheTools 是一个特殊工具——它把"读缓存"这个动作包装成一个 Agent 能调的工具(hit_cache),用于某些让模型"主动查缓存"的场景。L02 里 on_tool_use 特意排除 CacheTools().name,就是防止"缓存工具的调用结果又被缓存"的套娃。日常你几乎不用直接碰它。
🧠 今天你应该能回答
- 缓存 key 是怎么构成的?读和写为什么必须用一致的 key?
ToolsHandler和CacheHandler分别管什么?- 为什么用读写锁而不是普通互斥锁?
cache_function收哪两个参数?什么工具该关掉缓存?- "缓存命中"和"防重复调用"有什么本质区别?
- 防重复为什么只看上一次而非全部历史?
- 缓存的三个边界(进程内/不共享/不过期)分别是什么坑?
✋ 10 分钟动手
P=lib/crewai/src/crewai
cat $P/agents/tools_handler.py # 记账中枢(很短)
cat $P/agents/cache/cache_handler.py # 读写锁字典
sed -n '516,606p' $P/tools/tool_usage.py # 读缓存 + cache_function
sed -n '728,738p' $P/tools/tool_usage.py # 防重复
python -c "
from crewai.agents.cache.cache_handler import CacheHandler
c=CacheHandler()
c.add(tool='search', input='{\"q\":\"a\"}', output='结果A')
print(c.read('search','{\"q\":\"a\"}')) # 命中 → 结果A
print(c.read('search','{\"q\":\"b\"}')) # 未命中 → None
"
mcp/ 目录 + tools/mcp_tool_wrapper.py/mcp_native_tool.py:MCP 服务器怎么配置(stdio/http/sse)、远程工具怎么被发现并包装成 BaseTool、连接超时/重试/并行隔离怎么处理。