Day 25 / 共 60 天 · 阶段4 Crew 与流程

记忆开关:一个 memory=True 背后发生了什么

Day 19 说 memory 字段是个"bool 或实例"的联合类型;Day 22 说收尾要 _drain_memory_writes。今天把 crew 级的记忆配置这一层串起来:memory=Truecreate_crew_memorycrew.py:627)如何造出一个统一记忆、root_scope 命名空间怎么定、embedder 怎么接、_memory_llmcrew.py:664)怎么在你没指定时兜底选模型、私有 _memory 和公开 memory 两个属性怎么分工、后台写入又是何时 drain 落盘。记忆的存储细节留到阶段6(D33-40),今天只讲"crew 这层的开关与配置"。

📍 你在 60 天里的位置(阶段4 Crew 与流程 · 共 8 天)
D19 Crew 全字段 D20 顺序流程 D21 层级+manager D22 kickoff 家族 D23 规划 D24 训练+replay D25 记忆开关 D26 事件系统
💡 先用一个类比兜住今天 crew 的记忆配置就像给团队开一个"共享档案柜"memory=True = "给我配个标准档案柜"(框架自动买一个装好);传 Memory(...) 实例 = "我自己搬来一个特制柜子";False = "不要柜子"。档案柜有个门牌号root_scope,如 /crew/研究组),保证这个团队的档案不会和别的团队混在一起。柜子还需要一个"会整理归档的管理员"(记忆用的 LLM),你没指定的话框架会顺手借用团队里现成的模型。
L01

痛点:团队跑完就"失忆"

🤔 痛点你的客服 crew 昨天帮用户张三解决了退货问题。今天张三又来了,crew 却完全不记得——又从头问一遍"您遇到什么问题"。默认情况下每次 kickoff 都是一张白纸。你想让团队"记住"跨会话的信息(用户偏好、历史结论),但又不想手动管理一个数据库。这就是 memory 开关要解决的:一个开关,背后自动配好一整套记忆基础设施。
💡 一句话本质 crew 级记忆配置 = 一个 @model_validatorcreate_crew_memory)在建对象时,根据 memory 字段造出统一记忆并挂到私有 _memory它做三件事:① 按 crew 名生成一个命名空间 root_scope(隔离不同团队的记忆);② 若配了 embedder 就建好向量化器(记忆要靠语义检索);③ 给记忆配一个分析用的 LLM(没指定就兜底借用)。今天只看"配置层",具体怎么存/怎么召回是阶段6 的事。
L02

memory 字段的三态与 _ensure_memory_kind

回顾 Day 19 的字段声明(crew.py:225)与它的前置处理器(memory/memory_scope.py:20):

# crew.py:225
memory: Annotated[
    bool | Annotated[Memory | MemoryScope | MemorySlice, Field(discriminator="memory_kind")] | None,
    BeforeValidator(_ensure_memory_kind),
] = Field(default=False, description="Enable crew memory. Pass True for default Memory(), "
          "or a Memory/MemoryScope/MemorySlice instance for custom configuration.")

# memory/memory_scope.py:20  兼容老配置:给旧 dict 补上 memory_kind
def _ensure_memory_kind(value: Any) -> Any:
    """Backfill ``memory_kind`` on legacy dicts that predate the discriminator.
    Inference: ``scopes`` key → ``slice``; ``root_path`` → ``scope``; else ``memory``."""
    if isinstance(value, dict) and "memory_kind" not in value:
        if "scopes" in value:      value["memory_kind"] = "slice"
        elif "root_path" in value: value["memory_kind"] = "scope"
        else:                       value["memory_kind"] = "memory"
    return value                    # 非 dict(实例/bool/None)原样放行
三态:bool / 实例 / NoneTrue→自动造;Memory/Scope/Slice 实例→用你的;False/None→关闭。一个字段三种含义(Day 19 的联合类型取舍)。
discriminator="memory_kind"传实例时,Pydantic 靠 memory_kind 字段秒判是 Memory 还是 Scope 还是 Slice,不用逐个 try。
_ensure_memory_kind向后兼容垫片:老版本存的 dict 没有 memory_kind 字段,这里按有没有 scopes/root_path 推断并补上,让老 checkpoint/配置能加载不崩。
非 dict 原样放行实例、bool、None 不用处理,直接 return——垫片只管"老 dict"这一种情况。
_ensure_memory_kind 的 docstring 明说是为"pre-1.14.6 configs/checkpoints"服务的。这种"补字段"的兼容代码是框架演进的常见产物:加了新的判别字段后,得让存量数据平滑过渡,不能让老用户一升级就全崩。读源码碰到"backfill/legacy"字样,基本都是这类兼容垫片。
L03

create_crew_memory:开关落地的地方

核心校验器(crew.py:627):

# crew.py:627
@model_validator(mode="after")
def create_crew_memory(self) -> Crew:
    """Initialize unified memory, respecting crew embedder config."""
    from crewai.memory.utils import sanitize_scope_name
    crew_name = sanitize_scope_name(self.name or "crew")
    crew_root_scope = f"/crew/{crew_name}"          # ① 命名空间(L04)

    if self.memory is True:                          # —— 传 True:自动造 ——
        from crewai.memory.unified_memory import Memory
        embedder = None
        if self.embedder is not None:                # ② 配了 embedder 就建向量化器
            from crewai.rag.embeddings.factory import build_embedder
            embedder = build_embedder(cast(dict[str, Any], self.embedder))
        memory_kwargs = {"embedder": embedder, "root_scope": crew_root_scope}
        memory_llm = self._memory_llm()              # ③ 选分析用 LLM(L05)
        if memory_llm is not None:
            memory_kwargs["llm"] = memory_llm
        self._memory = Memory(**memory_kwargs)       # ★造好挂到私有 _memory
    elif self.memory:                                # —— 传了实例:尊重用户配置 ——
        # 不自动设 root_scope,用户自己的配置说了算
        self._memory = self.memory
    else:                                            # —— False/None:关闭 ——
        self._memory = None
    return self
@model_validator(mode="after")和 Day 19 的校验器一样,建对象时就把记忆配好——不拖到 kickoff。
if self.memory is True★注意是 is True(严格身份判断),只有传布尔 True 才走"自动造"分支——不会把实例误判进来。
build_embedder记忆靠语义检索召回(把文本转向量比相似度)。你配了 embedder 就建对应的向量化器,否则用默认。
elif self.memory:★传了实例走这里:直接用你的,且覆盖你的 root_scope——尊重高级用户的完整配置。
self._memory = ...★结果统一挂到私有 _memory(不是公开 memory)。为什么分两个属性,L06 讲。
💡 设计取舍①:为什么 True 分支设 root_scope,实例分支不设?True 意味着"我不想操心,你看着办"——框架就贴心地按 crew 名自动划一个命名空间,避免多个 crew 记忆互相串。但如果你亲自传了 Memory(...) 实例,说明你想完全掌控(可能故意让多个 crew 共享一个 scope 做跨团队记忆)——这时框架绝不越俎代庖去改你的 root_scope"约定优于配置,但用户显式配置永远优先"——这条原则贯穿整个记忆设计。
L04

root_scope:给团队记忆一个门牌号

命名空间由 crew 名清洗而来(memory/utils.py:8):

# memory/utils.py:8
def sanitize_scope_name(name: str) -> str:
    """Sanitize a name for use in hierarchical scope paths.
    Converts to lowercase, replaces non-alphanumeric chars (except underscore
    and hyphen) with hyphens, collapses multiple hyphens, strips leading/trailing
    hyphens. ... Returns 'unknown' if the result would be empty."""
    ...

# 回到 crew.py:634 —— 拼成层级路径
crew_name = sanitize_scope_name(self.name or "crew")   # "研究 Crew!" → "研究-crew"
crew_root_scope = f"/crew/{crew_name}"                   # → "/crew/研究-crew"
sanitize_scope_name把 crew 名清洗成安全的路径片段:转小写、非法字符换成连字符、去掉首尾连字符。空的话返回 "unknown"。
f"/crew/{crew_name}"★拼成层级命名空间,形如 /crew/research-crew。这个 crew 及其 agent 存的所有记忆都归到这个前缀下。
self.name or "crew"没给 crew 名就用默认 "crew"。Day 19 说过 name 默认就是 "crew"。
📝 例子:命名空间隔离 你有两个 crew:name="客服组"name="研发组",都 memory=True。它们的记忆分别落在 /crew/客服组/crew/研发组 两个命名空间下。客服组召回记忆时只会翻自己命名空间的档案,绝不会翻出研发组的内部讨论。就像两个部门用同一个档案系统、但各有各的文件夹前缀。
💡 为什么用"层级路径"而不是随便一个 id 当命名空间?层级路径(/crew/xxx)天然支持作用域嵌套与查询:将来可以按 /crew/研发组/* 查这个团队所有记忆,也可以给某个 agent 再细分 /crew/研发组/agent/李四(这就是 MemoryScope/MemorySlice 干的事,阶段6 详解)。用扁平 id 就没法表达这种"团队→成员"的层级归属。路径式命名空间 = 可组合的记忆作用域。
L05

_memory_llm:你没指定时,替记忆借个模型

记忆也要用 LLM(做摘要/分析),没配就兜底(crew.py:664):

# crew.py:664
def _memory_llm(self) -> str | BaseLLM | None:
    """Return the LLM auto-created memory should use for analysis."""
    if self.chat_llm is not None:
        return self.chat_llm                     # ① 优先用 chat_llm
    for agent in self.agents:
        agent_llm = getattr(agent, "llm", None)
        if agent_llm is not None:
            return agent_llm                     # ② 否则借第一个 agent 的 llm
    return None                                  # ③ 都没有 → None(记忆用它自己的默认)
优先 chat_llm★如果你设了 Day 19 的 chat_llm,记忆分析就用它。因为 chat_llm 本就是"和 crew 交互/统筹"用的,语义上最贴。
否则借 agent 的 llm没有 chat_llm,就借用团队里第一个 agent 的模型——反正它们都在同一个团队,模型选择大概率一致。
都没有 → None返回 None,让 Memory 用它自己的默认模型。绝不硬崩。
💡 设计取舍②:为什么记忆的 LLM 要"层层兜底"而不是必填? 记忆是个可选增强——用户开 memory=True 时想的是"帮我记东西",通常不会额外再想"记忆用哪个模型分析"。如果框架强制要求单独配一个记忆 LLM,就是给用户添负担、劝退新手。所以源码用"chat_llm → agent.llm → 默认"三级兜底:你想管就管(设 chat_llm),不想管框架自动挑一个合理的。这是"合理默认 + 可覆盖"设计哲学在又一处的体现——和 Day 23 规划 LLM 的兜底、Day 21 经理人设的内置一脉相承。
L06

私有 _memory vs 公开 memory:为什么两个属性

恢复检查点时的重绑逻辑,暴露了两者的分工(crew.py:514):

# crew.py:514
def _rebind_memory_views(self) -> None:
    """Reattach a live ``Memory`` to restored ``MemoryScope``/``MemorySlice`` views.
    ... Prefer the crew's restored ``Memory`` (from ``create_crew_memory``
    or a ``Crew.memory=Memory(...)`` instance) so all views share one
    backing store; fall back to a fresh ``Memory()`` only if nothing available."""
    backing: Memory | None = None
    if isinstance(self._memory, Memory):        # 优先私有 _memory(自动造的)
        backing = self._memory
    elif isinstance(self.memory, Memory):        # 其次公开 memory(用户传的实例)
        backing = self.memory
    def _ensure(view):
        ...
        if backing is None:
            backing = Memory()                   # 兜底:新建一个
        view.bind(backing)                       # ★让 scope/slice 视图绑到同一个底层 Memory
    _ensure(self.memory)
    for agent in self.agents:
        _ensure(agent.memory)                    # 每个 agent 的记忆视图也绑到同一底层
公开 memory用户设的字段:可能是 True、实例、或 None。它是"用户的意图声明"。
私有 _memory框架算出来的真身create_crew_memory 把意图落实成的真正 Memory 对象(或 None)。运行时用它。
view.bind(backing)★关键:让 crew 和各 agent 的记忆视图都绑到同一个底层 Memory——这样团队才是"共享一个档案柜",而不是各记各的。
用于 checkpoint 恢复从 JSON 恢复时,scope/slice 视图丢了 live Memory 依赖,这个方法把它们重新接上同一个底层,避免首次使用就 RuntimeError。
💡 为什么"意图"和"真身"要分成两个属性?因为它们的生命周期/职责不同:公开 memory可序列化的配置(要能存进 checkpoint/JSON),所以它保存"用户填了啥";私有 _memory运行时的活对象(含数据库连接、embedder 等不可序列化的东西),不进 JSON。分开后:存档时只存意图,恢复时再用 create_crew_memory/_rebind_memory_views 把活对象重建出来。"配置与运行时实例分离"是可持久化系统的常见架构。
L07

后台写入与 drain:记忆什么时候真落盘

回顾 Day 22 的 _drain_memory_writescrew.py:1848),从"配置层"再看一遍它收集了谁:

# crew.py:1848
def _drain_memory_writes(self) -> None:
    """Block until all pending background memory saves have completed."""
    seen = set()
    candidates = [self._memory, self.memory,                     # crew 自己的(真身+意图)
                  getattr(self.manager_agent, "memory", None),   # 经理的
                  *(getattr(agent, "memory", None) for agent in self.agents)]  # 每个 agent 的
    for mem in candidates:
        if mem is None or isinstance(mem, bool):
            continue                          # None / bool 跳过(bool 是"意图"不是真身)
        backing = getattr(mem, "_memory", None) or mem   # 视图→取其底层 Memory
        if id(backing) in seen:
            continue                          # 同一底层只 drain 一次
        seen.add(id(backing))
        drain = getattr(backing, "drain_writes", None)
        if callable(drain):
            drain()                           # ★阻塞等后台写入落盘
记忆是"后台异步写"的★为什么需要 drain?因为存记忆不阻塞主流程——agent 干完活继续跑,记忆在后台线程慢慢存。所以结束前必须等它们写完。
isinstance(mem, bool) 跳过★这里能看出 memory=True(bool)是意图不是真身——真身在 _memory。bool 没有 drain_writes,跳过。
backing = _memory or mem如果 mem 是 scope/slice 视图,取它的底层 _memory;否则它自己就是底层。呼应 L06 的"视图共享底层"。
seen 去重crew 和多个 agent 可能共享同一底层 Memory(L06 绑的),用 id 去重只 drain 一次。
⚠️ 边界:不 drain 就可能"记了但没存进去" 因为记忆是后台异步写,如果进程在 drain 之前就退出(或完成事件抢先触发监听器拆除,Day 22 讲过),后台那几条还没落盘的记忆就丢了——你以为存了,下次却召回不到。所以 kickoff 的收尾(成功路径 + finally)都要 drain。异步写性能好,但必须配一个"结束前强制等待"的 drain 点,否则数据静默丢失——这是所有"异步写 + 需要持久保证"系统的通用铁律。
L08

今日小结

数据结构:memory 字段 → create_crew_memory → 共享底层 memory 字段(意图:True/实例/False) create_crew_memory(建对象时落地) _memory (真身) root_scope + embedder + llm 各 agent 记忆视图 经理记忆视图 bind 到同一底层 → 全队共享一个"档案柜";结束前 drain 落盘
图注:意图字段经 create_crew_memory 落成真身 _memory,各视图 bind 到同一底层共享,结束前 drain。

👶 小白:memory=True 之后,记忆存哪了?我没配数据库啊。

👨‍🏫 老师:框架给了默认存储后端(本地文件/向量库),你不配也能用——这就是 memory=True 的"开箱即用"。想换后端、换 embedder、换存储路径,就别传 True,改传一个配置好的 Memory(...) 实例(走 elif self.memory 分支)。具体后端、短期/长期/实体记忆的区别,是阶段6(D33-40)的内容。今天你只要理解 crew 这层"开关怎么把配置落地"。

🧠 今天你应该能回答

  • memory 字段的三态?_ensure_memory_kind 为什么存在?
  • create_crew_memory 做的三件事是什么?
  • 为什么 True 分支自动设 root_scope,实例分支不设?
  • root_scope 的作用是什么?为什么用层级路径?
  • _memory_llm 的三级兜底顺序?为什么不做成必填?
  • 公开 memory 和私有 _memory 分别代表什么?为什么要分?
  • 记忆为什么是后台异步写?不 drain 会怎样?

✋ 10 分钟动手

P=lib/crewai/src/crewai
sed -n '627,675p'   $P/crew.py            # create_crew_memory + _memory_llm
sed -n '514,546p'   $P/crew.py            # _rebind_memory_views(视图共享底层)
sed -n '8,30p'      $P/memory/utils.py    # sanitize_scope_name
sed -n '20,37p'     $P/memory/memory_scope.py   # _ensure_memory_kind 兼容垫片
# 开记忆跑一次,观察 root_scope 命名
python -c "
from crewai import Agent, Task, Crew
a=Agent(role='客服', goal='解答', backstory='耐心')
t=Task(description='记住用户叫张三,然后打个招呼', expected_output='招呼语', agent=a)
c=Crew(name='客服组', agents=[a], tasks=[t], memory=True, verbose=True)
print('私有真身 _memory =', c._memory)   # 不是 None,说明记忆已配好
print(c.kickoff())
"
明日预告 · Day 26:本阶段最后一天。今天多次出现 crewai_event_bus.emit(...) —— kickoff 开始/结束/失败、训练开始/完成、记忆保存完成……全靠事件。明天拆事件系统:单例事件总线 CrewAIEventsBus@on(EventType) 注册监听器、emit 如何把事件分发给同步/异步 handler、BaseEventListener 怎么自动注册。这是日志、遥测、流式输出、tracing 的统一底座。
← Day 24 训练+replay Day 26 · 事件系统 →