Day 33 / 共 60 天 · 阶段6 记忆与知识

记忆总览:CrewAI 的"统一记忆"到底长什么样

前 32 天我们把 Agent、Task、Crew、工具都拆透了,但有个根本问题一直没解决:大模型没有记性——每次调用都是"失忆"的。CrewAI 用一个 memory/ 模块给它装上"长期记忆"。今天先鸟瞰整个模块:memory/ 目录里有哪些文件、核心数据结构 MemoryRecord 有哪些字段、记忆是怎么"排序"的(语义 + 时间新鲜度 + 重要性三合一评分)、以及存储后端是怎么被抽象成一个"协议"的。这是深入 D34~D40 的地图。

📍 你在 60 天里的位置(阶段6 记忆与知识 · 共 8 天)
S5 工具系统 D33 记忆总览 D34 unified_memory D35 recall/encoding D36 短/长/entity D37 scope 作用域 D38 RAG D39 knowledge D40 embedding/存储
💡 先用一个类比兜住今天 CrewAI 的记忆就像一个会自动整理的私人笔记本:你随手记一句话(remember),它会自动判断"这条该归到哪个文件夹(scope)、贴什么标签(categories)、有多重要(importance)",还会把跟已有笔记重复的自动合并。等你要用的时候(recall),它不是简单地按关键词翻找,而是综合考虑"内容有多贴题 + 记得有多近 + 当初标记有多重要",把最该看的翻到最前面。今天就是先认识这个笔记本的"零件清单"。
L01

痛点:大模型天生"失忆"

🤔 痛点你让一个 Agent 昨天调研了"公司 Q3 营收增长 20%",今天再问它"我们上季度营收怎样",它完全不记得——因为每次 LLM 调用都是独立的、无状态的。要让 Agent 跨任务、跨会话记住东西,就得在框架层面自己造一套"外挂记忆":存进去、还要能智能地取出来(不是全塞回 prompt,那样又贵又超窗口)。这套东西该怎么设计?
💡 一句话本质 记忆 = 把文本编码成向量存进数据库(remember),检索时按"综合相关度"排序取回最相关的几条(recall)。CrewAI 把这套逻辑收拢进 crewai/memory/ 一个模块,对外只暴露一个 Memory 类,内部用两条 Flow(EncodingFlow 编码、RecallFlow 检索)+ 可插拔的 StorageBackend(默认 LanceDB)实现。
大白话"编码"就是把一句话变成一串数字(向量),意思相近的句子数字也相近;"检索"就是把你的问题也变成数字,然后在数据库里找"数字最接近"的几条。这套叫向量检索,是所有 Agent 记忆/RAG 的地基。CrewAI 在这之上加了"新鲜度衰减""重要性加权""LLM 自动分类合并"等一堆智能。
L02

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__
💡 设计取舍①:为什么记忆入口要懒加载? 文件顶部注释写得很明确(__init__.py:1-6):"Heavy dependencies are lazily imported so that 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 后子进程继承)场景下容易出问题。源码做法:谁用到记忆谁才付出加载成本。这是"库要对不用某功能的用户零成本"的典型考量。
L03

MemoryRecord:一条记忆的"身份证"

最核心的数据结构是 MemoryRecordtypes.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 相同的检索请求才看得到(多租户/多用户共享一个库时防串味)。
数据结构:一条 MemoryRecord 的字段分组 身份 & 正文 id (uuid4) content ★必填 embedding (exclude) 组织 & 排序 scope (路径) categories (标签) importance 0~1 时间 & 隐私 created_at last_accessed source / private ↓ 检索时这三组共同决定"排在哪" embedding→语义分 created_at→新鲜分 importance→重要分 composite_score 复合分
图注:MemoryRecord 字段分三组,其中 embedding / created_at / importance 三者在检索时汇成一个复合分数(下一讲)。
L04

MemoryConfig:控制记忆行为的所有旋钮

MemoryConfigtypes.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 注释明说它不是公开 APItypes.py:135-142):用户是通过 Memory(recency_weight=..., ...) 的关键字参数配置的,Memory 内部再把这些值组装成一个 MemoryConfig 传给两条 Flow 和评分函数——把散落的参数打包成一个对象传递,避免函数签名里挂十几个参数。
L05

复合评分:语义 + 新鲜 + 重要,三合一

整套记忆的"灵魂公式"就是 compute_composite_scoretypes.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,新鲜/重要达标才追加。给用户看"这条为什么被翻出来",可解释性。
📝 例子:两条记忆谁排前面 查询"公司营收",向量搜出两条,语义分都 0.8。
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 更重要,但太旧了,新鲜度把它拉了下来。三个维度互相制衡,而不是单看相似度。

👶 小白:为什么不直接按向量相似度排就好,非要搞这么复杂?

👨‍🏫 老师:纯语义搜有两个毛病。一是过时信息:"公司地址在北京"和"公司已搬到上海"语义都很像你的问题,但你要的是最新的——新鲜度权重帮你压住旧的。二是抓不住重点:一句随口闲聊和一条关键决策,语义都可能贴题,但重要性权重让关键决策更容易冒头。记忆不只是"找像的",而是"找该看的"。

L06

StorageBackend:把"存哪里"抽象成协议

记忆到底存进 LanceDB 还是 Qdrant?Memory 不关心——它只依赖一个 Protocolstorage/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_checkableisinstance(x, StorageBackend) 在运行时能检查(只查方法名是否齐全)。
search 返回 (记录, 分数) 列表约定:后端只负责"按向量找相似 + 返回相似分",复合评分/新鲜度这些在上层算。职责清晰。
同步 + async 双份协议同时定义 save/asave 等——上层可按需选同步或异步路径。
💡 设计取舍②:为什么用 Protocol 而不是抽象基类(ABC)? ABC 做法:让每个后端 class LanceDBStorage(StorageBackend) 显式继承 + 用 @abstractmethod 强制实现。Protocol 做法:只描述"需要哪些方法",谁长得像谁就是。后者的好处是解耦——你可以拿一个第三方向量库的客户端,甚至一个测试用的假对象(in-memory fake),只要方法签名对得上就能直接塞给 Memory(storage=...),不需要去改它的继承链、不需要 import CrewAI 的基类。对"可插拔后端"这种场景,Protocol 是更轻、更开放的契约。backend.py 里的 EmbeddingDimensionMismatchError 也故意不继承 RuntimeError(D40 讲),足见接口设计的细致。
L07

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:一个 crew 的记忆有自己的"命名空间" root_scope 是理解 CrewAI 记忆组织的钥匙。一个开了记忆的 crew 会自动拿到 /crew/<crew名> 作为根前缀(crew.py:637 crew_root_scope = f"/crew/{crew_name}"),之后这个 crew 存的所有记忆都挂在这个前缀下、按 LLM 推断的子 scope 再分文件夹。这样多个 crew 共用一个物理库也不会互相看到对方的记忆——靠的就是 scope 路径前缀隔离(D37 深讲)。
L08

边界 + 今日小结

⚠️ 边界:embedding 维度不匹配会怎样? 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)
"
明日预告 · Day 34:今天只看了 Memory 的字段。明天逐行读 unified_memory.py方法remember 怎么走后台线程池、recall 的"读屏障"(drain_writes)怎么保证读到最新写入、shallow 与 deep 两种检索深度怎么分流。
← 总目录 Day 34 · unified_memory →