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

memory_scope.py:记忆的"文件夹"与"多柜合看"

D34~D36 反复出现 scoperoot_scope。今天读 memory_scope.py,把作用域彻底讲清。MemoryScope 是"锁定在某个路径下"的记忆视图——所有存/取都自动加这个路径前缀,像一个 cd 进去的子文件夹。MemorySlice 是"同时看多个路径"的合并视图——跨几个 scope 检索后合并去重。配合 utils.py 的三个路径工具,CrewAI 用纯粹的"路径前缀"就实现了多用户/多 crew/多 agent 的记忆隔离与共享,不需要多个物理库。

📍 你在 60 天里的位置(阶段6 记忆与知识 · 共 8 天)
D33 记忆总览 D34 unified_memory D35 recall/encoding D36 短/长/entity D37 scope 作用域 D38 RAG D39 knowledge D40 embedding/存储
💡 先用一个类比兜住今天 scope 就是文件系统的路径。一个物理记忆库 = 一块大硬盘;/crew/研究组/市场 = 硬盘上的一个文件夹路径。MemoryScope("/crew/研究组") 相当于你 cd 进这个文件夹——之后你说"存到 /市场"其实是存到 /crew/研究组/市场,你看不到也动不了别的 crew 的文件夹。MemorySlice(["/A","/B"]) 则像同时打开两个文件夹一起搜。隔离与共享,全靠路径前缀,不用给每个人一块新硬盘。
L01

痛点:一个库,多个主体,怎么不串味

🤔 痛点真实系统里,同一个记忆库常常被多个主体共用:多个 crew、一个 crew 里多个 agent、一个应用里多个用户/会话。如果全堆在一起,A crew 会检索到 B crew 的记忆、用户甲会看到用户乙的隐私。但要是给每个主体开一个独立的物理库,又会有一堆库要管理、还没法做"跨主体共享某些公共知识"。怎么在一个库里既隔离又能选择性共享?
💡 一句话本质层级路径 scope:每条记忆有一个像 /crew/研究组/agent/分析师 的路径;检索时给一个前缀,只匹配"以此前缀开头"的记录(存储层 scope LIKE '前缀%')。MemoryScope 帮你把前缀自动拼好(视图),MemorySlice 帮你跨多个前缀合并搜。隔离=各用各的前缀,共享=让多个视图指向同一段前缀。
L02

scope 路径三工具:清洗、规整、拼接

路径处理全靠 utils.py 三个纯函数(memory/utils.py:8):

# memory/utils.py:8
def sanitize_scope_name(name: str) -> str:
    """把 crew名/agent角色 洗成安全的路径片段。"""
    if not name: return "unknown"
    name = name.lower().strip()
    name = re.sub(r"[^a-z0-9_-]", "-", name)   # 非字母数字下划线连字符 → 连字符
    name = re.sub(r"-+", "-", name)            # 连续连字符压成一个
    name = name.strip("-")
    return name or "unknown"

# memory/utils.py:39
def normalize_scope_path(path: str) -> str:
    if not path or path == "/": return "/"
    path = re.sub(r"/+", "/", path)            # 多个斜杠压成一个 //→/
    if not path.startswith("/"): path = "/" + path   # 保证前导斜杠
    if len(path) > 1: path = path.rstrip("/")  # 去尾部斜杠
    return path

# memory/utils.py:67
def join_scope_paths(root: str | None, inner: str | None) -> str:
    root = root.rstrip("/") if root else ""
    inner = inner.strip("/") if inner else ""
    if root and inner:  result = f"{root}/{inner}"
    elif root:          result = root
    elif inner:         result = f"/{inner}"
    else:               result = "/"
    return normalize_scope_path(result)
sanitize_scope_name把"Research Crew"这种人类名字洗成 research-crew——路径里不能有空格/特殊字符/大小写混乱。空的兜底成 unknown
normalize_scope_path规整路径:/crew/test//agent///crew/test/agent。统一"前导有斜杠、尾部无斜杠、中间无双斜杠",后面所有前缀匹配才不出错。
join_scope_paths★把 root 和 inner 安全拼接,处理各种边界(root 有尾斜杠、inner 有前斜杠、任一为 None/"/")。join("/crew/test","/市场")→"/crew/test/市场"
大白话为什么要三个函数专门伺候路径?因为路径拼接是"魔鬼在细节"的事:多一个斜杠、少一个前导斜杠、大小写不一致,前缀匹配就会漏或错。把这些清洗/规整/拼接的逻辑收进三个纯函数(无副作用、好测试),到处复用,就不会各处各写一套、各有各的 bug。
L03

MemoryScope:绑定一个根路径的视图

MemoryScopememory_scope.py:38)是一个"薄视图",持有底层 Memory + 一个根路径:

# memory_scope.py:38(节选)
class MemoryScope(BaseModel):
    memory_kind: Literal["scope"] = "scope"      # 判别器(D06 用于反序列化)
    root_path: str = Field(default="/")
    _memory: Memory | None = PrivateAttr(default=None)   # 底层真 Memory(不序列化)
    _root: str = PrivateAttr(default="")

    @model_validator(mode="wrap")
    @classmethod
    def _accept_memory(cls, data, handler) -> MemoryScope:
        if isinstance(data, MemoryScope): return data
        memory = data.pop("memory", None)        # ★从入参里抽出 memory 依赖
        instance = handler(data)                 # 正常校验剩下的字段
        if memory is not None: instance._memory = memory
        root = instance.root_path.rstrip("/") or ""
        if root and not root.startswith("/"): root = "/" + root
        instance._root = root                    # 规整后的根
        return instance

    def _scope_path(self, scope: str | None) -> str:   # ★核心:把相对 scope 拼成绝对
        if not scope or scope == "/": return self._root or "/"
        s = scope.rstrip("/")
        if not s.startswith("/"): s = "/" + s
        if not self._root: return s
        return f"{self._root.rstrip('/')}{s}"
_memory 是 PrivateAttr底层 Memory 不参与序列化(它含线程池等,且不该被存进检查点)——这埋下 L06 "反序列化后要 bind 重连"的伏笔。
model_validator(mode="wrap")包裹式校验:能在校验前后动手。这里在校验前把 memory 依赖从入参 pop 出来(它不是普通字段),校验后再塞回私有属性。
_scope_path 拼前缀★视图的灵魂:你传相对 /市场,它拼成绝对 /crew/研究组/市场。你永远在根路径之下操作,出不去。
数据结构:一个物理库里的 scope 树 / (根) /crew/研究组 /crew/写作组 …/市场 …/agent/分析师 MemoryScope("/crew/研究组") 只见左子树,写"/市场"→"/crew/研究组/市场"
图注:所有记忆共处一棵 scope 树;MemoryScope 把你锁在某个子树里,前缀匹配天然隔离兄弟子树。
L04

相对路径操作与 subscope 下钻

视图的每个方法都先过 _scope_path 拼前缀再转发给底层(memory_scope.py:102):

# memory_scope.py:102 / 148 / 217(节选)
def remember(self, content, scope="/", ...) -> MemoryRecord | None:
    path = self._scope_path(scope)                 # 相对 → 绝对
    return self._require_memory().remember(content, scope=path, ...)

def recall(self, query, scope=None, ..., depth="deep", ...) -> list[MemoryMatch]:
    search_scope = self._scope_path(scope) if scope else (self._root or "/")
    return self._require_memory().recall(query, scope=search_scope, ...)

def subscope(self, path: str) -> MemoryScope:      # ★往下再钻一层
    child = path.strip("/")
    if not child:
        return MemoryScope(memory=self._memory, root_path=self._root or "/")
    base = self._root.rstrip("/") or ""
    new_root = f"{base}/{child}" if base else f"/{child}"
    return MemoryScope(memory=self._memory, root_path=new_root)
remember/recall 都拼前缀你在 scope 视图里写的每个路径都是"相对这个 scope 的",视图自动补全成绝对路径——用起来就像操作独立的小库。
_require_memory()转发前先确认底层 Memory 存在(:77),没有就抛清晰错误"call .bind(memory) after restore"——防止反序列化后忘了重连就用(L06)。
subscope 复用同一底层下钻出的子作用域共享同一个 _memory,只是 root_path 更长。scope("/crew/研究组").subscope("agent/分析师") → 根变 /crew/研究组/agent/分析师
📝 例子:给每个 agent 一个专属子作用域 crew 的记忆根是 /crew/研究组。想让"分析师"这个 agent 有自己的记忆区、又能读到 crew 公共区:
crew_mem = m.scope("/crew/研究组")(crew 级)
analyst_mem = crew_mem.subscope("agent/分析师")(agent 级,根 = /crew/研究组/agent/分析师
分析师存的在自己子区;而 crew_mem.recall(...) 因为搜的是 /crew/研究组 前缀,能搜到所有 agent 的(含子区)——层级前缀天然实现"下级私有、上级可见全部"。
L05

MemorySlice:跨多个作用域合并检索

有时你要"同时看几个不相邻的 scope"——这是 MemorySlicememory_scope.py:227):

# memory_scope.py:227 / 292(节选)
class MemorySlice(BaseModel):
    memory_kind: Literal["slice"] = "slice"
    scopes: list[str] = Field(default_factory=list)   # 要合看的多个路径
    categories: list[str] | None = Field(default=None)
    read_only: bool = Field(default=True)             # ★默认只读!

    def recall(self, query, ..., limit=10, ...) -> list[MemoryMatch]:
        cats = categories or self.categories
        all_matches = []
        for sc in self.scopes:                        # ★逐个 scope 检索
            matches = self._require_memory().recall(
                query, scope=sc, categories=cats,
                limit=limit * _RECALL_OVERSAMPLE_FACTOR, ...)   # 每个多取一些
            all_matches.extend(matches)
        seen_ids, unique = set(), []
        for m in sorted(all_matches, key=lambda x: x.score, reverse=True):  # 全局按分排
            if m.record.id not in seen_ids:
                seen_ids.add(m.record.id)
                unique.append(m)
                if len(unique) >= limit: break        # 截断到 limit
        return unique
scopes 多路径比如 ["/crew/研究组","/company/公共知识"]——既看自己的又看公司公共的。
逐个 recall 再合并对每个 scope 单独走一遍检索(各自 oversample 多取),把所有结果汇总,全局按复合分重排、去重、截断到 limit。
read_only 默认 True★slice 默认只读:remember 直接返回 None 不写(:280)。因为"合看多个域"通常是消费场景,往哪个域写会有歧义,默认禁写更安全。
info/list_categories 也聚合slice 的统计类方法会把各 scope 的结果合并(记录数相加、类目并集、取最早/最晚时间,:338)。
控制流:MemorySlice 跨域检索合并 recall /crew/A recall /company/公共 各自 oversample 多取 汇总 all_matches 按分排序 + 去重 seen_ids 截断 limit ✅ 每个 scope 内仍走完整智能检索;应用层负责全局重排与去重
图注:slice 对每个 scope 各搜一遍(多取),汇总后全局按复合分排序去重、截断到 limit。
💡 设计取舍①:为什么 slice 检索是"逐个搜再合并"而不是"一次搜多前缀"? 存储层的 search 只接受单个 scope_prefix(一次 LIKE '前缀%')。slice 要的多个前缀往往不连续、不同层/crew/A/company/公共),没法用一个 LIKE 表达。所以 slice 选择对每个 scope 各搜一次、在应用层合并去重重排。代价是 N 个 scope 就 N 次搜,但换来灵活组合任意路径的能力,且每个 scope 内部仍享受完整的智能检索。为弥补合并可能漏掉边缘结果,每个 scope 都 oversample 多取一些再统一筛。
L06

序列化与 bind:检查点恢复后重连底层

视图能被存进检查点(Flow 持久化),但底层 Memory 存不了——所以要 bind 重连(memory_scope.py:68):

# memory_scope.py:68
def bind(self, memory: Memory) -> Self:
    """Rebind the runtime Memory dependency after restore.
    Required after deserializing from a checkpoint, since the live Memory
    cannot be serialized."""
    self._memory = memory
    return self

# memory_scope.py:20 —— 老配置兼容:给缺判别器的旧字典补 memory_kind
def _ensure_memory_kind(value: Any) -> Any:
    if isinstance(value, dict) and "memory_kind" not in value:
        if "scopes" in value:      value["memory_kind"] = "slice"   # 有 scopes → slice
        elif "root_path" in value: value["memory_kind"] = "scope"   # 有 root_path → scope
        else:                      value["memory_kind"] = "memory"
    return value
bind(memory)反序列化出来的 scope/slice 只有 root_path/scopes,没有底层 Memory。crew 恢复时调 bind 把活的 Memory 接回去(crew.py 的 _rebind_memory_views),之后才能用。
memory_kind 判别器Memory/MemoryScope/MemorySlice 三者用 memory_kind 字段区分(Pydantic 判别联合)——反序列化时靠它决定还原成哪个类。
_ensure_memory_kind 兼容旧版★向后兼容:1.14.6 之前的检查点没有 memory_kind,这个 BeforeValidator 按"有没有 scopes/root_path"猜出类型补上,让老检查点也能加载不崩。
crew 里的字段声明是 Annotated[Memory | MemoryScope | MemorySlice, Field(discriminator="memory_kind")] 再套 BeforeValidator(_ensure_memory_kind)crew.py:227)——判别联合 + 老数据补丁,双保险。
L07

root_scope 在 Memory 内部怎么落地

视图是"外挂"的前缀;而 Memory 自己也有个 root_scope,在每个方法里统一处理(unified_memory.py:715):

# unified_memory.py:715(recall 里的 scope 计算,forget/list_* 同款模式)
effective_scope = scope
if effective_scope is None and self.root_scope:
    effective_scope = self.root_scope                          # 没传 → 用根
elif effective_scope is not None and self.root_scope:
    effective_scope = join_scope_paths(self.root_scope, effective_scope)  # 传了 → 拼在根下
# ... 之后所有搜索/删除都用 effective_scope 当前缀
root_scope 是"内建前缀"MemoryScope 是包一层的视图;Memory.root_scope 是 Memory 自己带的前缀。crew 用后者(Memory(root_scope="/crew/x"),D36 L07)。二者最终都归到 join_scope_paths
三分支统一处理没传 scope → 落到 root;传了 → 拼在 root 下;无 root → 原样。recall/forget/reset/list_scopes/list_records/info/tree/list_categories 全用这套模式,保证"永远在 root 之内"。
encoding 侧也遵守存的时候 EncodingFlow_apply_defaults/字段解析都 join_scope_paths(item.root_scope, inner_scope)encoding_flow.py:313/355)——存和取用同一套拼接,前缀才对得上。

👶 小白:MemoryScope 视图和 Memory.root_scope 有啥区别,用哪个?

👨‍🏫 老师:功能上很像,都是"加前缀"。区别在层次root_scope 是把某个 Memory 实例整个钉在一个根下(crew 自动用它);MemoryScope 是在一个 Memory 之上临时开一个受限视图,可以随时 subscope 再钻、可以给不同 agent 各开一个视图共享同一个底层库。要"一个 crew 一个命名空间"用 root_scope;要"同一个库里给多个主体各切一块、灵活组合"用 scope/slice 视图。

L08

取舍 + 今日小结

💡 设计取舍②:为什么用"路径前缀"而不是"每主体一个库/一张表"? 多库/多表做法:每个 crew/用户开一个独立库或表,隔离最彻底。但代价大:库/表数量随主体线性膨胀、跨主体检索("搜我的 + 公司公共")得连多个库合并、连接数/文件句柄爆炸。前缀做法:一个库、一张表、一列 scope,隔离靠 LIKE '前缀%'、共享靠"多个视图指向重叠前缀"、跨域靠 slice。好处是主体再多也只有一个库、组合无限灵活、还能建 scope 列的索引(D40 的 BTREE 索引)加速前缀过滤。代价是隔离是"逻辑的"而非"物理的"——但配合 private/source 字段做隐私过滤,对绝大多数场景足够。用一列路径 + 索引,换掉"一堆库"的运维复杂度。
⚠️ 边界:scope 前缀匹配的"越界"陷阱 前缀匹配是字符串 LIKE '/crew/研究%',它会同时匹配 /crew/研究组/crew/研究员——如果两个 crew 名字有前缀包含关系就会串。源码靠两点缓解:① sanitize_scope_name + normalize_scope_path 规整路径;② 存储层 list_scopes/get_scope_info"/" 切分组件、只认完整路径段(D40 会看到 child_prefix = prefix + "/")。但直接用 search(scope_prefix=...) 时仍是纯前缀 LIKE——给 scope 取名时避免一个名字是另一个的前缀,或始终带上完整层级(/crew/研究组/),是稳妥习惯。

🧠 今天你应该能回答

  • 为什么用一个库 + scope 路径就能隔离多主体?
  • 三个路径工具各做什么?为什么要专门规整路径?
  • MemoryScope 的 _scope_path 起什么作用?subscope 怎么下钻?
  • MemorySlice 为什么"逐个搜再合并"?为什么默认只读?
  • 反序列化后为什么要 bind?memory_kind 判别器干嘛的?
  • root_scope 和 MemoryScope 视图怎么选?前缀匹配有什么坑?

✋ 10 分钟动手

P=lib/crewai/src/crewai/memory
sed -n '8,103p'   $P/utils.py           # 三个路径工具
sed -n '38,101p'  $P/memory_scope.py    # MemoryScope + _scope_path
sed -n '292,324p' $P/memory_scope.py    # MemorySlice.recall 合并
python -c "
from crewai.memory import Memory
m = Memory()
research = m.scope('/crew/研究组')
research.remember('市场调研:Q3 SaaS 增长放缓', scope='/市场')
analyst = research.subscope('agent/分析师')
analyst.remember('我的私人笔记:关注留存率', scope='/')
print('研究组能看到全部(含子区):', [h.record.scope for h in research.recall('调研', depth='shallow')])
"
明日预告 · Day 38:记忆用的是 memory/ 自带的向量存储;而 CrewAI 还有一套独立的 RAG 子系统rag/)给"知识库"用。明天读 rag/BaseClient 向量库客户端协议、ChromaDB 实现、以及 distance→score 的换算。
← Day 36 短/长/entity Day 38 · RAG →