先看 ToolsHandler 的字段（tools_handler.py:15），麻雀虽小：

# tools_handler.py:15
class ToolsHandler(BaseModel):
    """Callback handler for tool usage."""
    cache: CacheHandler | None = Field(default=None)             # 那本"速查本"
    last_used_tool: ToolCalling | InstructorToolCalling | None = Field(default=None)  # 最近一次调用

缓存的 key 从"一次调用"里提取，所以先看"一次调用"的数据结构（tools/tool_calling.py:11）：

# tools/tool_calling.py:11
class ToolCalling(BaseModel):
    tool_name: str = Field(..., description="The name of the tool to be called.")
    arguments: dict[str, Any] | None = Field(
        ..., description="A dictionary of arguments to be passed to the tool.")

class InstructorToolCalling(PydanticBaseModel):
    tool_name: str = PydanticField(..., description="The name of the tool to be called.")
    arguments: dict[str, Any] | None = PydanticField(
        ..., description="A dictionary of arguments to be passed to the tool.")

ToolsHandler 的核心方法 on_tool_use（tools_handler.py:26），在工具执行完被调用：

# tools_handler.py:26
def on_tool_use(self, calling, output, should_cache: bool = True) -> None:
    """Run when tool ends running."""
    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)   # ③ 参数字典 → JSON 字符串当 key
            else:
                input_str = str(calling.arguments)
        self.cache.add(tool=calling.tool_name, input=input_str, output=output)  # ④ 写进缓存

真正存数据的是 CacheHandler（cache/cache_handler.py:10）：

# cache/cache_handler.py:10
class CacheHandler(BaseModel):
    """Handles caching of tool execution results.
    Provides thread-safe in-memory caching ... Uses a read-write lock to allow
    concurrent reads while ensuring exclusive write access."""
    _cache: dict[str, Any] = PrivateAttr(default_factory=dict)   # 真正的字典
    _lock: RWLock = PrivateAttr(default_factory=RWLock)          # 读写锁

    def add(self, tool: str, input: str, output: Any) -> None:   # :23
        with self._lock.w_locked():                              # 写：独占锁
            self._cache[f"{tool}-{input}"] = output              # ★key = "工具名-输入串"

    def read(self, tool: str, input: str) -> Any | None:        # :37
        with self._lock.r_locked():                              # 读：共享锁
            return self._cache.get(f"{tool}-{input}")

写缓存在 on_tool_use，那"读缓存"在哪触发？在工具真正执行前（tools/tool_usage.py，tool_usage.py:276）：

# tools/tool_usage.py:276
from_cache = False
result = None
try:
    if self.tools_handler and self.tools_handler.cache:      # 有缓存本子
        input_str = ""
        if calling.arguments:
            input_str = json.dumps(calling.arguments) if isinstance(calling.arguments, dict) \
                        else 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:                                     # ★只有没命中缓存才真执行
        ...
        result = await tool.ainvoke(input=arguments, config=fingerprint_config)

不是所有结果都该缓存。执行完，源码给工具一个"投票权"（tool_usage.py:351）：

# tools/tool_usage.py:351
if self.tools_handler:
    should_cache = True                                   # 默认缓
    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)   # 把决定传给写缓存

还有个有趣的角色 CacheTools（tools/cache_tools/cache_tools.py）——把"读缓存"本身也包装成一个工具：

# tools/cache_tools/cache_tools.py
class CacheTools(BaseModel):
    name: str = "Hit Cache"
    cache_handler: CacheHandler = Field(default_factory=CacheHandler)
    def hit_cache(self, key: str) -> str | None:
        split = key.split("tool:")
        tool = split[1].split("|input:")[0].strip()
        tool_input = split[1].split("|input:")[1].strip()
        return self.cache_handler.read(tool, tool_input)

P=lib/crewai/src/crewai
cat -n            $P/agents/tools_handler.py          # ToolsHandler（50 行全看）
cat -n            $P/agents/cache/cache_handler.py    # CacheHandler + 读写锁
sed -n '276,366p' $P/tools/tool_usage.py              # 执行前查缓存 + cache_function
sed -n '1,60p'    $P/tools/cache_tools/cache_tools.py # CacheTools
# 亲手验证缓存命中
python -c "
from crewai.agents.cache.cache_handler import CacheHandler
c = CacheHandler()
c.add(tool='search', input='{\"q\":\"CrewAI\"}', output='一个多智能体框架')
print('命中:', c.read('search', '{\"q\":\"CrewAI\"}'))
print('未命中:', c.read('search', '{\"q\":\"别的\"}'))
"