短期 / 长期 / 实体记忆:三种记忆如何被"一个 Memory"统一
很多 Agent 框架(包括老版 CrewAI)把记忆分成三类:短期记忆(本次会话的上下文)、长期记忆(跨会话沉淀的经验)、实体记忆(人/组织/地点等实体信息)。新版 CrewAI 没有三个类,而是用一个 Memory 通过字段和评分把三者揉进一套机制:短期 = 新鲜度衰减、长期 = 重要性 + 合并持久化、实体 = LLM 抽取的 entities/categories 元数据。今天用真代码把这三条对应关系讲透。
痛点:三种记忆各造一个类,重复又割裂
MemoryRecord 带上 created_at(时间)、importance(价值)、metadata.entities+categories(实体),再用复合评分把这些维度融进排序,三种记忆就自然浮现,不用三个类。| 经典分类 | 在新版里对应 | 真实字段/机制 |
|---|---|---|
| 短期记忆(近期上下文) | 新鲜度权重 + 半衰期衰减 | recency_weight · recency_half_life_days · created_at |
| 长期记忆(沉淀经验) | 重要性权重 + 合并持久化 | importance · consolidation_* · LanceDB 落盘 |
| 实体记忆(人/组织/地点) | LLM 抽取的结构化元数据 | ExtractedMetadata.entities · categories |
一份记录,同时承载三个维度
回看 MemoryRecord(types.py:20),三种记忆的"料"全在一条里:
# types.py:20(挑出与三种记忆相关的字段)
class MemoryRecord(BaseModel):
content: str # 记忆正文
importance: float = Field(default=0.5, ...) # ★长期性:越高越该长期留存/优先召回
created_at: datetime = Field(default_factory=datetime.utcnow) # ★短期性:越新越"短期"
last_accessed: datetime = ... # 最近被翻到的时间
categories: list[str] = Field(default_factory=list) # ★实体性:主题/类目标签
metadata: dict[str, Any] = Field(default_factory=dict) # ★实体性:entities/dates/topics 落在这
importance存时由 LLM 或用户给的 0~1 分。它是"这条值不值得长期记住"的度量——长期记忆的核心信号。created_at创建时间。复合评分里换算成新鲜度衰减——它让"最近的"表现得像短期记忆(自动靠前、慢慢淡出)。categories + metadataLLM 抽出的实体(entities)、日期(dates)、主题(topics)分别落进 categories 和 metadata——这就是实体记忆的载体。短期记忆 = 新鲜度衰减
"短期"的本质是"越近的越该先看、越久越淡"。这正是复合评分里的 decay(types.py:364):
# types.py:364(复合评分的新鲜度部分)
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) # 半衰期默认 30 天
composite = (config.semantic_weight * semantic_score
+ config.recency_weight * decay # ★新鲜度贡献
+ config.importance_weight * record.importance)
decay 就是"短期性"刚存 decay≈1(很"短期"),过一个半衰期减到 0.5,过两个减到 0.25——记忆随时间自动"变旧"、排名自然下滑,模拟"淡忘"。recency_weight 调强弱想要"更看重最近"(更像短期记忆主导),把 recency_weight 调高、recency_half_life_days 调小即可。反之更看重沉淀。不需要单独的"短期库"老版要专门存一份短期上下文;新版靠时间字段 + 衰减,让"最近的"天然浮上来,不用第二个存储。Memory(recency_weight=0.6, recency_half_life_days=3):新鲜度权重拉到 0.6、半衰期缩到 3 天。效果——3 天前的记忆新鲜分就砍半,一周前的几乎沉底。这就得到一个"偏重近期上下文"的短期型记忆,全靠调参数,不用换类。长期记忆 = 重要性 + 合并持久化
"长期"要解决两件事:重要的东西不因时间沉底、以及持久落盘不丢。前者靠 importance 权重顶住 decay,后者靠合并(consolidation)防止越攒越乱。合并计划的数据结构(analyze.py:124):
# analyze.py:103
class ConsolidationAction(BaseModel):
action: str = Field(description="One of 'keep', 'update', or 'delete'.")
record_id: str = Field(...)
new_content: str | None = Field(default=None) # update 时的新内容
reason: str = Field(default="")
# analyze.py:124
class ConsolidationPlan(BaseModel):
actions: list[ConsolidationAction] = Field(default_factory=list) # 对旧记录的动作
insert_new: bool = Field(default=True) # 要不要同时存新的
insert_reason: str = Field(default="")
importance 顶住时间复合评分里 importance_weight * importance 这一项不随时间衰减。一条 importance=0.9 的记忆哪怕很旧,重要分仍稳定贡献——这就是"长期记忆不轻易被冲掉"。keep/update/delete存新记忆时若发现跟旧的重叠,LLM 给出对每条旧记录的动作:保留、用新内容更新、或删掉——让长期记忆不断被修订而非无脑堆积。insert_new是否把新内容也作为独立一条存。若旧记录已被 update 成新内容,可能就不必再插入。LanceDBStorage 把记录写进本地 $CREWAI_STORAGE_DIR/memory 目录的 LanceDB 表(D40 讲)。所以你今天存的、明天甚至下个月重启程序还在——这就是长期记忆的物理基础。短期与长期在存储上没有区别,区别只在评分权重和合并策略。实体记忆 = LLM 抽取的结构化元数据
实体记忆的载体是 ExtractedMetadata(analyze.py:18),存记忆时由 LLM 填充:
# analyze.py:18
class ExtractedMetadata(BaseModel):
model_config = ConfigDict(extra="forbid") # OpenAI 要求 additionalProperties:false
entities: list[str] = Field(default_factory=list,
description="Entities (people, orgs, places) mentioned in the content.")
dates: list[str] = Field(default_factory=list,
description="Dates or time references in the content.")
topics: list[str] = Field(default_factory=list,
description="Topics or themes in the content.")
# analyze.py:37
class MemoryAnalysis(BaseModel):
suggested_scope: str # 建议归到哪个 scope
categories: list[str] = Field(default_factory=list) # 类目标签
importance: float = Field(default=0.5, ge=0.0, le=1.0)
extracted_metadata: ExtractedMetadata = Field(default_factory=ExtractedMetadata) # ★实体
entities/dates/topicsLLM 从内容里抽出的"人/组织/地点""日期""主题"。存"张三上周从阿里跳到字节",entities≈[张三,阿里,字节]、dates≈[上周]——这就是实体记忆。extra="forbid"★固定 schema(禁止额外字段):因为 OpenAI 的结构化输出要求 additionalProperties:false。用固定三字段而非任意 dict,才能走 LLM 的原生结构化输出。抽取即分析,无独立实体库老版实体记忆是单独的库;新版把实体作为每条记忆的元数据顺手抽出来,检索时可按 categories 过滤、按 entities 追踪,不用第二套存储。parallel_analyze(D35 L07)被并进记录:item.resolved_metadata = dict(item.metadata or {}, **analysis.extracted_metadata.model_dump())(encoding_flow.py:328)——用户给的 metadata 和 LLM 抽的实体合并成最终元数据。合并:让长期记忆"越记越精"而非"越堆越乱"
analyze_for_consolidation(analyze.py:321)是长期记忆保持整洁的关键:
# analyze.py:321(节选)
def analyze_for_consolidation(new_content, existing_records, llm) -> ConsolidationPlan:
if not existing_records:
return ConsolidationPlan(actions=[], insert_new=True) # 没相似的→直接存
records_lines = []
for r in existing_records:
created = r.created_at.isoformat() if r.created_at else ""
records_lines.append(
f"- id={r.id} | scope={r.scope} | importance={r.importance:.2f} | created={created}\n"
f" content: {r.content[:200]}...")
user = _get_prompt("consolidation_user").format(
new_content=new_content, records_summary="\n\n".join(records_lines))
messages = [{"role": "system", "content": _get_prompt("consolidation_system")},
{"role": "user", "content": user}]
... # 让 LLM 输出 ConsolidationPlan(结构化)
except Exception:
return _CONSOLIDATION_DEFAULT # 失败→安全默认:insert_new=True,不丢
喂旧记录摘要给 LLM把相似的旧记录(含 id/scope/importance/时间/正文前 200 字)列给 LLM,问它"新内容跟这些什么关系"。LLM 决定动作可能回"这条旧的过时了→delete 它、insert 新的",或"新旧是同一件事的更新→update 旧的、不 insert",或"无关→都保留"。触发条件只有相似度 ≥ consolidation_threshold(0.85)才触发这个分析(D35 的 B/D 组)——不相似的直接存,省 LLM。失败兜底 insert_new合并分析挂了就默认"照存不误",宁可暂时有点重复,也不丢记忆(D35 L08 的容错哲学)。crew 一键开记忆:自动命名空间
用户其实很少直接 new Memory——通常是 Crew(memory=True),crew 自动装配(crew.py:627):
# crew.py:627(节选)
def create_crew_memory(self) -> Crew:
from crewai.memory.utils import sanitize_scope_name
crew_name = sanitize_scope_name(self.name or "crew")
crew_root_scope = f"/crew/{crew_name}" # ★每个 crew 一个命名空间
if self.memory is True:
from crewai.memory.unified_memory import Memory
embedder = None
if self.embedder is not None:
embedder = build_embedder(cast(dict, self.embedder))
memory_kwargs = {"embedder": embedder, "root_scope": crew_root_scope}
memory_llm = self._memory_llm() # 复用 chat_llm 或某个 agent 的 llm
if memory_llm is not None:
memory_kwargs["llm"] = memory_llm
self._memory = Memory(**memory_kwargs)
elif self.memory: # 用户传了 Memory/Scope/Slice 实例
self._memory = self.memory # 尊重用户配置,不强塞 root_scope
else:
self._memory = None
return self
memory=True 简写只写 Crew(memory=True),crew 就 new 一个默认 Memory,并自动设 root_scope=/crew/<名字>——所有该 crew 的记忆挂这个前缀下。_memory_llm 复用记忆分析用的 LLM 不用单配:优先用 crew 的 chat_llm,否则借某个 agent 的 llm(:664)。省心。传实例则尊重用户如果你传的是自己配好的 Memory/MemoryScope/MemorySlice,crew 不强行改你的 root_scope——你的配置优先。root_scope=/crew/研究组 就像给这个 crew 开了一个专属文件夹,它存的所有记忆(不管短期长期实体)都放进这个文件夹的子目录里。这样十个 crew 共用一个物理记忆库也不会互相看到对方的东西——下一天(D37)专门讲这套 scope 作用域怎么隔离、怎么嵌套。边界 + 今日小结
delete 过时记录;② 你手动调 forget(older_than=...)(unified_memory.py:818)按时间/scope/类目批量删;③ reset() 清空某 scope。所以长期跑的服务要自己定期 forget 老数据,别指望它自己瘦身——这是"设计上把清理权交给使用者"的取舍,别踩坑。👶 小白:那我还需要区分"这条是短期还是长期"吗?
👨🏫 老师:基本不用。你只管 remember(content),剩下的交给系统:LLM 判重要性(决定长期性)、系统记时间(决定短期性)、LLM 抽实体(决定实体性)。你顶多在建 Memory 时调权重来偏向近期或偏向沉淀。真要强制某条重要就传 importance=0.95,要它归到某类就传 categories=[...]。三种记忆从"你要选类"变成了"系统按维度自动融合"。
🧠 今天你应该能回答
- 老版三种记忆在新版分别对应哪些字段/机制?
- 短期记忆为什么不用单独的库?decay 怎么模拟"淡忘"?
- 长期记忆靠什么"不被时间冲掉"?靠什么"持久"?
- 实体记忆存在哪?ExtractedMetadata 为什么 extra=forbid?
- 合并为什么要用 LLM 而不是"相似就删"?
- Crew(memory=True) 自动做了什么?记忆会自动瘦身吗?
✋ 10 分钟动手
P=lib/crewai/src/crewai/memory
sed -n '18,56p' $P/analyze.py # ExtractedMetadata / MemoryAnalysis(实体)
sed -n '103,140p' $P/analyze.py # ConsolidationPlan(长期合并)
sed -n '360,375p' ../crew.py 2>/dev/null; sed -n '627,662p' lib/crewai/src/crewai/crew.py # crew 开记忆
python -c "
from crewai.memory import Memory
m = Memory(recency_weight=0.6, recency_half_life_days=3) # 偏短期
r = m.remember('张三从阿里跳槽到字节,负责推荐算法')
print('scope=',r.scope,'importance=',r.importance,'cats=',r.categories,'meta=',r.metadata)
"
scope、root_scope 到底怎么隔离和嵌套?明天读 memory_scope.py 的 MemoryScope(单作用域视图)和 MemorySlice(多作用域合并视图),讲清 CrewAI 记忆的"文件夹式"权限与共享。