记忆开关:一个 memory=True 背后发生了什么
Day 19 说 memory 字段是个"bool 或实例"的联合类型;Day 22 说收尾要 _drain_memory_writes。今天把 crew 级的记忆配置这一层串起来:memory=True 时 create_crew_memory(crew.py:627)如何造出一个统一记忆、root_scope 命名空间怎么定、embedder 怎么接、_memory_llm(crew.py:664)怎么在你没指定时兜底选模型、私有 _memory 和公开 memory 两个属性怎么分工、后台写入又是何时 drain 落盘。记忆的存储细节留到阶段6(D33-40),今天只讲"crew 这层的开关与配置"。
memory=True = "给我配个标准档案柜"(框架自动买一个装好);传 Memory(...) 实例 = "我自己搬来一个特制柜子";False = "不要柜子"。档案柜有个门牌号(root_scope,如 /crew/研究组),保证这个团队的档案不会和别的团队混在一起。柜子还需要一个"会整理归档的管理员"(记忆用的 LLM),你没指定的话框架会顺手借用团队里现成的模型。痛点:团队跑完就"失忆"
memory 开关要解决的:一个开关,背后自动配好一整套记忆基础设施。@model_validator(create_crew_memory)在建对象时,根据 memory 字段造出统一记忆并挂到私有 _memory。它做三件事:① 按 crew 名生成一个命名空间 root_scope(隔离不同团队的记忆);② 若配了 embedder 就建好向量化器(记忆要靠语义检索);③ 给记忆配一个分析用的 LLM(没指定就兜底借用)。今天只看"配置层",具体怎么存/怎么召回是阶段6 的事。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"字样,基本都是这类兼容垫片。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 意味着"我不想操心,你看着办"——框架就贴心地按 crew 名自动划一个命名空间,避免多个 crew 记忆互相串。但如果你亲自传了 Memory(...) 实例,说明你想完全掌控(可能故意让多个 crew 共享一个 scope 做跨团队记忆)——这时框架绝不越俎代庖去改你的 root_scope。"约定优于配置,但用户显式配置永远优先"——这条原则贯穿整个记忆设计。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"。name="客服组" 和 name="研发组",都 memory=True。它们的记忆分别落在 /crew/客服组 和 /crew/研发组 两个命名空间下。客服组召回记忆时只会翻自己命名空间的档案,绝不会翻出研发组的内部讨论。就像两个部门用同一个档案系统、但各有各的文件夹前缀。/crew/xxx)天然支持作用域嵌套与查询:将来可以按 /crew/研发组/* 查这个团队所有记忆,也可以给某个 agent 再细分 /crew/研发组/agent/李四(这就是 MemoryScope/MemorySlice 干的事,阶段6 详解)。用扁平 id 就没法表达这种"团队→成员"的层级归属。路径式命名空间 = 可组合的记忆作用域。_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 用它自己的默认模型。绝不硬崩。memory=True 时想的是"帮我记东西",通常不会额外再想"记忆用哪个模型分析"。如果框架强制要求单独配一个记忆 LLM,就是给用户添负担、劝退新手。所以源码用"chat_llm → agent.llm → 默认"三级兜底:你想管就管(设 chat_llm),不想管框架自动挑一个合理的。这是"合理默认 + 可覆盖"设计哲学在又一处的体现——和 Day 23 规划 LLM 的兜底、Day 21 经理人设的内置一脉相承。私有 _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 把活对象重建出来。"配置与运行时实例分离"是可持久化系统的常见架构。后台写入与 drain:记忆什么时候真落盘
回顾 Day 22 的 _drain_memory_writes(crew.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 一次。kickoff 的收尾(成功路径 + finally)都要 drain。异步写性能好,但必须配一个"结束前强制等待"的 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())
"
crewai_event_bus.emit(...) —— kickoff 开始/结束/失败、训练开始/完成、记忆保存完成……全靠事件。明天拆事件系统:单例事件总线 CrewAIEventsBus、@on(EventType) 注册监听器、emit 如何把事件分发给同步/异步 handler、BaseEventListener 怎么自动注册。这是日志、遥测、流式输出、tracing 的统一底座。