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

工具缓存:让 Agent 不做重复的苦活

Agent 转圈时经常会用同一个工具、同样的参数重复调用——比如反复搜同一个词。每次真跑一遍既慢又费钱(有些工具还收费)。今天读缓存三件套:agents/tools_handler.py(记账中枢 ToolsHandler)、agents/cache/cache_handler.py(带读写锁的 CacheHandler)、以及 tools/tool_usage.py 里的缓存分支。看清楚:什么时候读缓存、什么时候写、cache_function 怎么细粒度控制、以及"连续调同一个工具同参数"怎么被拦下。

📍 你在 60 天里的位置(阶段5 工具系统 · 共 6 天)
D27 BaseTool D28 StructuredTool D29 工具调用 D30 缓存复用 D31 MCP 工具 D32 自定义与最佳实践
💡 先用一个类比兜住今天 缓存就像公司前台的"常见问题登记本"。有人问"三楼会议室 WiFi 密码",前台查一次、记本子上;下次再有人问一模一样的问题,直接翻本子答,不用再跑一趟机房。ToolsHandler 是那个前台,CacheHandler 是那本登记本(key = "工具名-参数")。而 cache_function 是一条规矩:"哪些问题值得记"(比如实时股价就别记,会过期)。还有一条铁律:同一个人连问两遍完全一样的问题,前台会说"你刚问过了"——这就是防重复调用。
L01

痛点:重复劳动又慢又烧钱

🤔 痛点Agent 的 ReAct 循环(Day 08)里,模型可能因为"没记住上一步"或推理绕圈,反复用同样的参数调同一个工具。如果这个工具是"调用付费 API""跑一个几秒的爬虫",每重复一次就是实打实的钱和时间。更糟的是模型可能陷入"调 A → 看结果 → 又调一模一样的 A"的死循环。怎么破?
💡 一句话本质 用 (工具名 + 参数) 当 key,把工具结果存起来,下次相同调用直接返回缓存。加上两道额外保险:① cache_function 让每个工具自定义"这次结果值不值得缓存"(实时数据就别缓存);② _check_tool_repeated_usage 拦住"紧接着又调完全相同的一次",直接回一句"你刚用过"逼模型换招。缓存省钱省时,防重复救死循环。

三个角色的分工:

角色文件职责
ToolsHandlertools_handler.py:15记账中枢:记"最后用的工具"、决定写不写缓存
CacheHandlercache_handler.py真正的存储:一个带读写锁的字典
缓存分支tool_usage.py:516执行流程里"先读缓存 / 后写缓存"的钩子
L02

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),只负责在每次工具用完后拍板"记不记、怎么记",并顺手记下"刚才用的是谁"。它是执行循环和缓存存储之间的中间人。
数据结构:缓存三角色的持有关系 ToolsHandler last_used_tool · cache CacheHandler _cache: dict + _lock: RWLock key = "工具名-参数串" last_used_tool ToolCalling(上一次调用) → 供防重复比对 持有(存储) 记录
图注:ToolsHandler 持有一个 CacheHandler(真存储)和一个 last_used_tool(防重复依据)。职责清晰分离。
L03

CacheHandler:一个带读写锁的字典

真正的存储 CacheHandlercache_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 序列化——缓存是运行时状态,不该被存档。
💡 设计取舍①:为什么用读写锁而不是普通锁 工具执行是高度并发的(Day 08 原生模式的线程池同时跑多个工具)。普通互斥锁:读和读之间也互斥,多个线程想同时查缓存得排队——而查缓存本是无冲突的只读操作,白白串行化。读写锁:允许"多读并发、写时独占"。缓存场景是典型的"读多写少",读写锁能让大量并发的缓存查询不互相阻塞。代价是读写锁实现更复杂、写饥饿风险。对读多写少的缓存,这个复杂度换来的吞吐提升很值。
L04

执行前:先敲敲缓存的门

_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 生成必须一致,否则永远命不中。
控制流:一次工具调用的缓存路径 收到 ToolCalling cache.read(工具名-参数) 命中 → 直接用缓存值(不执行) 未命中 → tool.invoke 真执行 on_tool_use → 按 cache_function 写回 from_cache=True,省一次执行
图注:读缓存命中就跳过执行;未命中才真跑,跑完按 cache_function 决定写不写回。
L05

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_useshould_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 缓存。
L06

防连续重复:你刚刚才用过

缓存解决"重复调用还得走一遍流程",而 _check_tool_repeated_usagetool_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)解决主要问题——简单且够用。
L07

用量上限与"结果即答案"的收尾

执行前还会查用量上限(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,不再继续推理。适合"生成报告""最终交付物"类工具。
💡 这些机制其实是同一个思想的不同侧面缓存、防重复、用量上限、结果直答——看似四件事,本质都是"控制工具被调用的次数和方式,把 Agent 的自由发挥约束在成本可控的范围内"。LLM 天生倾向"多试几次""绕圈子",这些机制就是给它套上的缰绳:能省则省(缓存)、别原地打转(防重复)、别无限用(上限)、该停就停(直答)。
L08

边界 + 今日小结

⚠️ 边界:缓存是"进程内内存字典",且不会自动过期 CacheHandler._cache 就是个普通 dict,存在当前进程内存里:进程一停,缓存全没;多进程/分布式部署时各进程各存各的,不共享。而且它没有 TTL(过期时间)——一旦缓存了某结果,同参数调用永远返回它,直到进程结束。坑:把一个"结果会随时间变化"的工具用默认缓存(比如"今天天气"缓存了,明天还返回昨天的)。解法:这类工具务必用 cache_function 返回 False,或在参数里带上日期让 key 天然区分。生产级需求(跨进程、带过期)得自己接外部缓存。

👶 小白:那个 CacheToolscache_tools.py)是干嘛的?和 CacheHandler 有啥区别?

👨‍🏫 老师:CacheHandler 是"存储"(后台的登记本)。CacheTools 是一个特殊工具——它把"读缓存"这个动作包装成一个 Agent 能调的工具(hit_cache),用于某些让模型"主动查缓存"的场景。L02 里 on_tool_use 特意排除 CacheTools().name,就是防止"缓存工具的调用结果又被缓存"的套娃。日常你几乎不用直接碰它。

🧠 今天你应该能回答

  • 缓存 key 是怎么构成的?读和写为什么必须用一致的 key?
  • ToolsHandlerCacheHandler 分别管什么?
  • 为什么用读写锁而不是普通互斥锁?
  • 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
"
明日预告 · Day 31:前四天讲的都是"本地工具"。但现在流行 MCP(Model Context Protocol)——把工具做成独立服务,Agent 远程连过去用。明天读 mcp/ 目录 + tools/mcp_tool_wrapper.py/mcp_native_tool.py:MCP 服务器怎么配置(stdio/http/sse)、远程工具怎么被发现并包装成 BaseTool、连接超时/重试/并行隔离怎么处理。
← Day 29 工具调用 Day 31 · MCP 工具 →