记忆总览:CrewAI 的"统一记忆"到底长什么样
前 32 天我们把 Agent、Task、Crew、工具都拆透了,但有个根本问题一直没解决:大模型没有记性——每次调用都是"失忆"的。CrewAI 用一个 memory/ 模块给它装上"长期记忆"。今天先鸟瞰整个模块:memory/ 目录里有哪些文件、核心数据结构 MemoryRecord 有哪些字段、记忆是怎么"排序"的(语义 + 时间新鲜度 + 重要性三合一评分)、以及存储后端是怎么被抽象成一个"协议"的。这是深入 D34~D40 的地图。
remember),它会自动判断"这条该归到哪个文件夹(scope)、贴什么标签(categories)、有多重要(importance)",还会把跟已有笔记重复的自动合并。等你要用的时候(recall),它不是简单地按关键词翻找,而是综合考虑"内容有多贴题 + 记得有多近 + 当初标记有多重要",把最该看的翻到最前面。今天就是先认识这个笔记本的"零件清单"。痛点:大模型天生"失忆"
crewai/memory/ 一个模块,对外只暴露一个 Memory 类,内部用两条 Flow(EncodingFlow 编码、RecallFlow 检索)+ 可插拔的 StorageBackend(默认 LanceDB)实现。memory/ 目录地图:8 个核心文件
先看目录结构(真实 find 结果,lib/crewai/src/crewai/memory/):
memory/
├── __init__.py # 对外导出 + 懒加载 Memory / EncodingFlow
├── types.py # MemoryRecord / MemoryConfig / 评分函数(数据层)
├── unified_memory.py # Memory 类:对外总入口(1104 行) ← D34
├── encoding_flow.py # 存的流水线(5 步 Flow) ← D35
├── recall_flow.py # 取的流水线(自适应深度 Flow) ← D35
├── analyze.py # LLM 分析:分类/合并/查询蒸馏
├── memory_scope.py # MemoryScope / MemorySlice 作用域视图 ← D37
├── utils.py # scope 路径规整工具
└── storage/
├── backend.py # StorageBackend 协议(接口) ← D40
├── factory.py # 存储后端工厂(可替换)
├── lancedb_storage.py # 默认后端:LanceDB(669 行) ← D40
└── qdrant_edge_storage.py # 备选后端:Qdrant
入口文件用了懒加载(__init__.py:23)——这是个值得注意的设计:
# memory/__init__.py:23
_LAZY_IMPORTS: dict[str, tuple[str, str]] = {
"Memory": ("crewai.memory.unified_memory", "Memory"),
"EncodingFlow": ("crewai.memory.encoding_flow", "EncodingFlow"),
}
def __getattr__(name: str) -> Any:
"""Lazily import Memory / EncodingFlow to avoid pulling in lancedb at import time."""
if name in _LAZY_IMPORTS:
import importlib
module_path, attr = _LAZY_IMPORTS[name]
mod = importlib.import_module(module_path)
val = getattr(mod, attr)
globals()[name] = val
return val
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
模块级 __getattr__Python 3.7+ 特性:当访问模块里不存在的属性时触发。这里拦截 Memory/EncodingFlow,第一次访问才真正 import。importlib.import_module动态导入真正的实现模块。之所以要"懒",是因为 lancedb 是重依赖(几百 MB),不该在 import crewai 时就加载。globals()[name] = val导入后缓存到模块全局,第二次访问就直接命中、不再触发 __getattr__。import crewai does not initialise at runtime — critical for Celery pre-fork and similar deployment patterns." 朴素做法:在 __init__.py 直接 from .unified_memory import Memory——简单,但代价是每次 import crewai 都连带把 lancedb / 向量库全加载进来,拖慢启动、还在 Celery "预分叉"(fork 前加载、fork 后子进程继承)场景下容易出问题。源码做法:谁用到记忆谁才付出加载成本。这是"库要对不用某功能的用户零成本"的典型考量。MemoryRecord:一条记忆的"身份证"
最核心的数据结构是 MemoryRecord(types.py:20),一条记忆存的所有东西都在这:
# types.py:20
class MemoryRecord(BaseModel):
id: str = Field(default_factory=lambda: str(uuid4())) # 唯一 ID
content: str = Field(...) # 记忆正文(必填)
scope: str = Field(default="/") # 层级路径 /company/team/user
categories: list[str] = Field(default_factory=list) # 标签
metadata: dict[str, Any] = Field(default_factory=dict) # 任意元数据
importance: float = Field(default=0.5, ge=0.0, le=1.0) # 重要性 0~1
created_at: datetime = Field(default_factory=datetime.utcnow)
last_accessed: datetime = Field(default_factory=datetime.utcnow)
embedding: list[float] | None = Field(
default=None, exclude=True, repr=False, # ★不参与序列化
description="Vector embedding ... Excluded from serialization to save tokens.")
source: str | None = Field(default=None) # 来源(用户ID/会话ID)
private: bool = Field(default=False) # 私有:只对同 source 可见
content唯一必填字段——记忆的正文。其他字段要么有默认值、要么 LLM 帮你推断。scope="/"层级路径,像文件夹。默认根 /。crew 会自动设成 /crew/研究组 这类前缀(D37 详解)。importance 带 ge/lePydantic 约束死在 0.0~1.0,越界直接校验报错。这个分数直接进入检索排序。embedding exclude=True★关键:向量字段不参与序列化。3072 维浮点数若跟着 JSON 到处传会爆 token/带宽——只在存储层用,对外表示时剔除。source + private一对做"隐私隔离"的字段:private=True 的记忆只有 source 相同的检索请求才看得到(多租户/多用户共享一个库时防串味)。MemoryConfig:控制记忆行为的所有旋钮
MemoryConfig(types.py:135)把所有可调参数收进一个对象,最关键的是三个权重:
# types.py:135(节选核心字段)
class MemoryConfig(BaseModel):
# 复合分 = semantic_weight*相似度 + recency_weight*新鲜度 + importance_weight*重要性
recency_weight: float = Field(default=0.3) # 新鲜度权重
semantic_weight: float = Field(default=0.5) # 语义相似权重(占大头)
importance_weight: float = Field(default=0.2) # 重要性权重
recency_half_life_days: int = Field(default=30) # 新鲜度"半衰期":30 天减半
consolidation_threshold: float = Field(default=0.85) # 相似度超此值触发"合并"
consolidation_limit: int = Field(default=5) # 合并时最多比对几条
batch_dedup_threshold: float = Field(default=0.98) # 批量去重阈值(近乎重复才丢)
confidence_threshold_high: float = Field(default=0.8) # 检索置信度≥此值直接返回
confidence_threshold_low: float = Field(default=0.5) # 低于此值触发深挖
exploration_budget: int = Field(default=1) # 深挖轮数预算
query_analysis_threshold: int = Field(default=250) # 查询短于此字符数跳过 LLM 分析
三权重 0.5/0.3/0.2默认语义占一半、新鲜度三成、重要性两成,加起来约 1.0(注释要求"应求和约等于 1 以便直觉化 0-1 打分")。recency_half_life_days=30指数衰减的半衰期:一条记忆放 30 天,它的"新鲜分"减半、60 天减到 1/4。调小 = 老记忆掉得快。consolidation_threshold=0.85存新记忆时,若跟已有记忆相似度超 0.85,就叫 LLM 决定"合并/更新/删除"(D35 讲),设 1.0 则关闭合并。batch_dedup_threshold=0.98同一批里近乎完全重复(≥0.98)才丢弃——故意设得很高,注释说"避免丢掉只是相似而有用的记忆"。query_analysis_threshold=250检索时查询短于 250 字符就跳过 LLM 分析、直接向量搜——省 1~3 秒(D35 详解)。MemoryConfig 注释明说它不是公开 API(types.py:135-142):用户是通过 Memory(recency_weight=..., ...) 的关键字参数配置的,Memory 内部再把这些值组装成一个 MemoryConfig 传给两条 Flow 和评分函数——把散落的参数打包成一个对象传递,避免函数签名里挂十几个参数。复合评分:语义 + 新鲜 + 重要,三合一
整套记忆的"灵魂公式"就是 compute_composite_score(types.py:345):
# types.py:345
def compute_composite_score(
record: MemoryRecord, semantic_score: float, config: MemoryConfig,
) -> tuple[float, list[str]]:
age_seconds = (datetime.utcnow() - record.created_at).total_seconds()
age_days = max(age_seconds / 86400.0, 0.0)
decay = 0.5 ** (age_days / config.recency_half_life_days) # ★指数衰减
composite = (
config.semantic_weight * semantic_score # 语义:向量搜出来的相似度
+ config.recency_weight * decay # 新鲜:越新 decay 越接近 1
+ config.importance_weight * record.importance # 重要:存时标记的分
)
reasons: list[str] = ["semantic"]
if decay > 0.5:
reasons.append("recency") # 新鲜度还高就记一笔"因为新"
if record.importance > 0.5:
reasons.append("importance") # 重要就记一笔"因为重要"
return composite, reasons
age_days 计算用当前时间减 created_at 得秒数、换算成天。max(..., 0.0) 防止时钟回拨出现负数。decay = 0.5 ** (age_days/half_life)★指数衰减的标准写法:过了 1 个半衰期指数为 1、0.5^1=0.5;2 个 → 0.25。这就是"新鲜度分"。三项加权求和把语义相似度、新鲜度、重要性各乘权重相加——一个 0~1 的综合分,检索就按它排序。match_reasons顺带返回"为什么匹配":总有 semantic,新鲜/重要达标才追加。给用户看"这条为什么被翻出来",可解释性。A:3 天前记的普通笔记(importance=0.5)。
decay=0.5^(3/30)≈0.93,复合 = 0.5×0.8 + 0.3×0.93 + 0.2×0.5 = 0.779。B:90 天前记的重要决策(importance=0.9)。
decay=0.5^(90/30)=0.125,复合 = 0.5×0.8 + 0.3×0.125 + 0.2×0.9 = 0.618。→ A 排前面:虽然 B 更重要,但太旧了,新鲜度把它拉了下来。三个维度互相制衡,而不是单看相似度。
👶 小白:为什么不直接按向量相似度排就好,非要搞这么复杂?
👨🏫 老师:纯语义搜有两个毛病。一是过时信息:"公司地址在北京"和"公司已搬到上海"语义都很像你的问题,但你要的是最新的——新鲜度权重帮你压住旧的。二是抓不住重点:一句随口闲聊和一条关键决策,语义都可能贴题,但重要性权重让关键决策更容易冒头。记忆不只是"找像的",而是"找该看的"。
StorageBackend:把"存哪里"抽象成协议
记忆到底存进 LanceDB 还是 Qdrant?Memory 不关心——它只依赖一个 Protocol(storage/backend.py:44):
# storage/backend.py:44
@runtime_checkable
class StorageBackend(Protocol):
"""Protocol for pluggable memory storage backends."""
def save(self, records: list[MemoryRecord]) -> None: ...
def search(self, query_embedding: list[float],
scope_prefix: str | None = None,
categories: list[str] | None = None,
metadata_filter: dict[str, Any] | None = None,
limit: int = 10, min_score: float = 0.0,
) -> list[tuple[MemoryRecord, float]]: ...
def delete(self, ...) -> int: ...
def update(self, record: MemoryRecord) -> None: ...
def get_record(self, record_id: str) -> MemoryRecord | None: ...
def list_records(self, ...) -> list[MemoryRecord]: ...
def get_scope_info(self, scope: str) -> ScopeInfo: ...
def list_scopes(self, parent: str = "/") -> list[str]: ...
def list_categories(self, ...) -> dict[str, int]: ...
def count(self, ...) -> int: ...
def reset(self, ...) -> None: ...
# 还有 asave / asearch / adelete 三个 async 版本
Protocol(不是基类)Python 的"结构化子类型":任何类只要长得像(有这些方法)就算实现了协议,不用显式继承。LanceDBStorage 从没写 class LanceDBStorage(StorageBackend),但它满足协议。@runtime_checkable让 isinstance(x, StorageBackend) 在运行时能检查(只查方法名是否齐全)。search 返回 (记录, 分数) 列表约定:后端只负责"按向量找相似 + 返回相似分",复合评分/新鲜度这些在上层算。职责清晰。同步 + async 双份协议同时定义 save/asave 等——上层可按需选同步或异步路径。class LanceDBStorage(StorageBackend) 显式继承 + 用 @abstractmethod 强制实现。Protocol 做法:只描述"需要哪些方法",谁长得像谁就是。后者的好处是解耦——你可以拿一个第三方向量库的客户端,甚至一个测试用的假对象(in-memory fake),只要方法签名对得上就能直接塞给 Memory(storage=...),不需要去改它的继承链、不需要 import CrewAI 的基类。对"可插拔后端"这种场景,Protocol 是更轻、更开放的契约。backend.py 里的 EmbeddingDimensionMismatchError 也故意不继承 RuntimeError(D40 讲),足见接口设计的细致。Memory 类骨架:默认后端与 scope 前缀
把上面的零件拼起来,就是对外的 Memory 类(unified_memory.py:76):
# unified_memory.py:76(节选)
class Memory(BaseModel):
"""Unified memory: standalone, LLM-analyzed, with intelligent recall flow.
Works without agent/crew. Uses LLM to infer scope, categories, importance on save."""
llm: ... = Field(default="gpt-5.4-mini") # 分析用的 LLM
storage: ... = Field(default="lancedb") # ★默认存储:字符串 "lancedb"
embedder: Any = Field(default=None) # None = 默认 OpenAI 嵌入
recency_weight: float = Field(default=0.3)
semantic_weight: float = Field(default=0.5)
importance_weight: float = Field(default=0.2)
...
root_scope: str | None = Field(default=None,
description="Structural root scope prefix. ... a crew with "
"root_scope='/crew/research' will store memories at '/crew/research/<inferred_scope>'.")
storage 默认是字符串 "lancedb",不是实例——Memory 在初始化时(model_post_init,D34 讲)才根据这个字符串去 new 一个 LanceDBStorage。传路径字符串就用那个目录,传 "qdrant-edge" 就换 Qdrant。这叫"配置即字符串,延迟到用时才实例化"。root_scope 是理解 CrewAI 记忆组织的钥匙。一个开了记忆的 crew 会自动拿到 /crew/<crew名> 作为根前缀(crew.py:637 crew_root_scope = f"/crew/{crew_name}"),之后这个 crew 存的所有记忆都挂在这个前缀下、按 LLM 推断的子 scope 再分文件夹。这样多个 crew 共用一个物理库也不会互相看到对方的记忆——靠的就是 scope 路径前缀隔离(D37 深讲)。边界 + 今日小结
backend.py:11 专门定义了一个 EmbeddingDimensionMismatchError。场景:你先用 text-embedding-3-small(1536 维)建了本地记忆库,后来升级 CrewAI,默认嵌入换成 text-embedding-3-large(3072 维)——新旧向量维度对不上,存/搜都会算错。源码主动抛这个带修复指引的错(提示你 crewai reset-memories 或钉住旧模型),而不是让维度不同的向量悄悄混进去污染检索结果。记住:换嵌入模型 = 旧向量库作废,必须重建。👶 小白:这套"统一记忆"和老版 CrewAI 的短期/长期/实体记忆是什么关系?
👨🏫 老师:这是新一代架构,把老版三种记忆合并成了一个 Memory。老版"短期/长期"的区别,现在体现为新鲜度权重 + 半衰期(新的自然靠前);"实体记忆"体现为 LLM 自动抽取的 entities 元数据 + categories。D36 会专门讲这个"三合一"的对应关系。你现在只需知道:一个 Memory 通过 scope/importance/recency 就覆盖了过去要三个类才能做的事。
🧠 今天你应该能回答
- 大模型"失忆",CrewAI 用什么补?(编码存向量 + 智能检索)
- memory/ 目录有哪几块?入口为什么懒加载?
- MemoryRecord 有哪些字段?embedding 为什么 exclude?
- 复合评分公式是什么?三个权重默认多少?半衰期怎么算?
- StorageBackend 为什么用 Protocol 而不是 ABC?
- root_scope 起什么作用?crew 的记忆怎么隔离?
✋ 10 分钟动手
P=lib/crewai/src/crewai/memory
sed -n '20,73p' $P/types.py # MemoryRecord 全字段
sed -n '345,380p' $P/types.py # 复合评分公式
sed -n '44,80p' $P/storage/backend.py # StorageBackend 协议
# 亲手记一条、取一条
python -c "
from crewai.memory import Memory
m = Memory()
m.remember('公司 Q3 营收增长 20%')
for hit in m.recall('上季度营收如何'):
print(round(hit.score,3), hit.match_reasons, hit.record.content)
"
Memory 的字段。明天逐行读 unified_memory.py 的方法:remember 怎么走后台线程池、recall 的"读屏障"(drain_writes)怎么保证读到最新写入、shallow 与 deep 两种检索深度怎么分流。