情景记忆这个术语最早来自认知心理学,用来描述人对特定时间、特定地点发生事件的记忆。它关注的不是抽象的规则,而是带着时间戳和情境线索的具体片段。比如你记得上周三下午在某台服务器上排查过数据库连接超时,这就是情景记忆;而你知道数据库连接超时通常与网络抖动或连接池耗尽有关,这属于语义记忆。当前大模型在参数里保存了大量语义知识,但无法记住某个用户在某次会话中发生过的具体事件。为了让智能体具备类似人类的回忆能力,需要在模型外部增加一层情景记忆模块,把交互记录以可检索的方式保存下来,并在后续推理时按相关性召回。

一、情景记忆与摘要记忆的边界
常见做法是把历史对话压缩成一段摘要,例如把五十轮对话总结成用户偏好和待办事项。这种做法能控制上下文长度,但会丢失大量可用于精确回答的细节。情景记忆不做过度压缩,它保留原始输入、工具返回、环境状态以及当时的时间信息。摘要回答不了像“上次那个超时错误后来是怎么解决的”这种问题,因为它只记得有超时问题,却不记得排查步骤和最终结论。而情景记忆可以通过时间和内容相似度定位到那一段记录。
具体来说,一条情景记忆通常包含会话标识、事件类型、时间戳、原始内容、附加元数据和向量表示。会话标识用于区分不同用户或不同任务,事件类型标记这是一条用户消息、一次工具调用还是一个执行结果,时间戳提供排序和衰减依据,原始内容保存未经改写的信息,向量表示则负责相似度检索。这样的结构比摘要更重,但它换来了可回溯性和精确性。
二、设计一个可落地的情景记忆存储结构
落地时可从最简单的数据类开始。下面用 Python 定义一条情景记录,字段尽量精简但保留扩展空间。
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional
@dataclass
class EpisodicRecord:
id: str
session_id: str
timestamp: datetime
event_type: str
content: str
metadata: dict[str, Any] = field(default_factory=dict)
embedding: Optional[list[float]] = None
这里 id 是记录的唯一标识,session_id 可以把同一次会话的经历聚合起来,timestamp 用于时间排序和后续衰减计算,event_type 区分用户输入、工具返回、系统提示等类型,content 直接保存原始文本,metadata 可以放入消息来源、模型版本、用户反馈标记等补充信息。embedding 字段在写入时填充,也可以由存储层延迟生成。
存储层不一定要上专门的向量数据库,早期可以用 SQLite 保存结构化字段,再单独用 FAISS 或 Chroma 管理向量索引。写入时先生成嵌入向量,再把向量和元数据分别落到对应存储,既保持事务一致性,又便于后续切换后端。下面给出一个简单写入封装:
class EpisodicStore:
def __init__(self, embedder):
self.embedder = embedder
self.records = []
def add(self, record: EpisodicRecord):
if record.embedding is None:
record.embedding = self.embedder.encode(record.content)
self.records.append(record)
def iter_records(self):
return iter(self.records)
这个示例只展示了内存列表作为存储,实际工程中需要把 records 替换为数据库操作。核心逻辑是每次写入前检查 embedding 是否已经存在,避免重复调用编码器。对于低频交互场景,这种延迟编码可以减少启动阶段的开销;对于高频写入场景,可以在消息产生时就同步生成向量,防止写入路径过长。
三、检索与时间衰减:让回忆既相似又新鲜
检索情景记忆不能只依赖向量相似度。如果用户现在问数据库连接超时,系统找到一条一年前相似的超时记录,虽然语义上接近,但当时的网络环境、数据库版本和解决方案可能已经失效。因此需要在相似度基础上加入时间衰减,让最近发生的经历拥有更高的权重。常见的做法是使用指数衰减因子 exp(-lam * delta_t),其中 delta_t 是当前时间与记录时间的间隔,lam 控制衰减速度。
下面的代码结合余弦相似度和时间衰减,对记录进行综合打分并返回前 k 条。
import math
from datetime import datetime, timezone
def cosine_similarity(a, b):
dot = sum(x * y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x * x for x in a))
norm_b = math.sqrt(sum(x * x for x in b))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
def score_record(query_vec, record, now, lam=0.01):
sim = cosine_similarity(query_vec, record.embedding)
delta_t = (now - record.timestamp).total_seconds() / 3600.0
decay = math.exp(-lam * delta_t)
return sim * decay
def retrieve(query, store, k=5):
now = datetime.now(timezone.utc)
query_vec = store.embedder.encode(query)
scored = []
for rec in store.iter_records():
s = score_record(query_vec, rec, now)
scored.append((s, rec))
scored.sort(key=lambda x: x[0], reverse=True)
return [rec for s, rec in scored[:k]]
注意时间比较需要统一使用 UTC 时间戳,避免服务器本地时区造成偏差。score_record 中的 lam 参数可以根据业务调整,如果任务变化很快,可以把 lam 调大,让较早记录迅速失去影响;如果任务具有长期稳定性,则把 lam 调小,保留更长时间窗内的经历。检索完成后,把返回记录按时间倒序或按相关度插入提示词,让模型在回答时显式看到这些历史片段。
上下文构造可以很简单:把每条记录格式化成一行,包含时间、事件类型和内容,再拼接到当前问题之前。这样模型不需要额外训练就能利用外部情景记忆,因为它看到的是自然语言形式的历史事件描述。
四、容量控制与遗忘策略
情景记忆如果只增不减,存储成本和检索延迟都会持续上升,而且过旧或低价值的记录会稀释有效信息。遗忘不是简单删除,而是根据时间、重要性和冗余度做分层清理。最直接的策略是设置最大记录数,超过阈值后删除最旧的记录。下面是一个基本修剪函数:
def prune(store, max_records=10000):
records = list(store.iter_records())
if len(records) <= max_records:
return
records.sort(key=lambda r: r.timestamp)
overflow = len(records) - max_records
for old_record in records[:overflow]:
store.delete(old_record.id)
仅有时间维度还不够,有些很久以前的重要事件仍应保留,比如一次罕见的故障恢复过程。可以在 metadata 中增加 importance 字段,用户反馈有效、人工确认的关键结论可以打高分,删除时优先清理低重要性记录。另一种做法是合并冗余:如果过去一周内出现了多条高度相似的用户咨询,可以保留其中信息量最大的一条,将其余转化为引用计数或频次标记,既减少存储量,又保留重复出现的信号。
在实际系统中,容量控制可以与检索联合优化。比如记录写入时先经过一层粗筛,如果与已有记录相似度超过 0.95,就只更新已有记录的频次和最近出现时间,避免写入重复数据。这样能显著降低向量索引体积,同时让重复事件在检索时获得更高的重要性。遗忘策略没有统一答案,需要根据业务对记忆长度和精确度的要求调整。
情景记忆给大模型智能体提供了一种保存具体经历的方式,与摘要记忆形成互补。通过结构化记录、向量检索、时间衰减和容量控制四个环节,系统可以在不无限膨胀上下文的前提下,按需回忆起与当前任务相关的历史片段。落地时建议从轻量级存储和简单检索开始,先验证记忆质量对任务完成率的提升,再逐步引入更复杂的重要性打分和遗忘策略。毕竟记忆系统的目的不是存得越多越好,而是在正确的时间把正确的经历放回模型眼前。

