跳转至

记忆流与反思机制

概述

记忆流（Memory Stream）是 Park et al. (2023) 在 Generative Agents 中提出的核心记忆架构。它为虚拟具身智能体提供了一个完整的经历记录，并通过三因子检索评分和反思机制实现高效的记忆利用。

相关内容

关于情景记忆和语义记忆的基础理论，参见 情景与语义记忆。

记忆流架构

记忆流是一个按时间排序的记忆对象列表，每个记忆对象包含：

class MemoryObject:
    """记忆流中的单个记忆条目"""
    def __init__(self):
        self.description: str      # 自然语言描述
        self.creation_time: float  # 创建时间戳
        self.last_access: float    # 最近访问时间
        self.importance: int       # 重要性评分 (1-10)
        self.embedding: list       # 语义嵌入向量
        self.type: str             # "observation" | "reflection" | "plan"
        self.related_ids: list     # 关联记忆ID列表

graph TD
    subgraph 记忆流 Memory Stream
        M1[观察: 看到John在图书馆<br/>t=8:00, imp=2]
        M2[观察: 与Maria讨论项目<br/>t=9:30, imp=5]
        M3[观察: 听到关于聚会的消息<br/>t=10:15, imp=6]
        M4[反思: John最近经常在图书馆<br/>t=10:30, imp=7]
        M5[观察: 收到会议邀请<br/>t=11:00, imp=4]
        M6[反思: 应该邀请Maria参加聚会<br/>t=11:30, imp=8]
    end

    M1 --> M4
    M3 --> M4
    M3 --> M6
    M2 --> M6

三因子检索评分

当智能体需要从记忆流中检索相关记忆时，使用三个因子的加权组合：

\[\text{score}(m, q) = \alpha \cdot \text{recency}(m) + \beta \cdot \text{importance}(m) + \gamma \cdot \text{relevance}(q, m)\]

其中： - \(m\) 是记忆对象 - \(q\) 是当前查询（当前情境） - \(\alpha, \beta, \gamma\) 是可调权重超参数

因子1: 时效性 (Recency)

时效性评分使用指数衰减函数，最近的记忆得分更高：

\[\text{recency}(m) = e^{-\lambda \Delta t}\]

其中： - \(\Delta t = t_{\text{now}} - t_{\text{last\_access}}(m)\) 是距上次访问的时间差 - \(\lambda\) 是衰减率参数

import math

def recency_score(memory, current_time, decay_rate=0.995):
    """计算时效性评分

    Args:
        memory: 记忆对象
        current_time: 当前时间（小时）
        decay_rate: 衰减率，越小衰减越快

    Returns:
        0-1之间的时效性评分
    """
    hours_passed = current_time - memory.last_access
    return math.exp(-decay_rate * hours_passed)

衰减率 \(\lambda\) 的选择影响记忆的"保鲜期"：

\(\lambda\) 半衰期 适用场景
0.01 ~69小时 长期社会关系
0.05 ~14小时 日常事件
0.1 ~7小时 即时对话
0.5 ~1.4小时 短期任务

因子2: 重要性 (Importance)

重要性评分由LLM在记忆创建时一次性评估：

\[\text{importance}(m) \in [1, 10]\]

评分Prompt: 

On the scale of 1 to 10, where 1 is purely mundane (e.g., brushing 
teeth, making bed) and 10 is extremely poignant (e.g., a break up, 
college acceptance), rate the likely poignancy of the following 
piece of memory.

Memory: {memory_description}
Rating: <fill in>

重要性评分归一化到 \([0, 1]\)：

\[\text{importance\_norm}(m) = \frac{\text{importance}(m) - 1}{9}\]

重要性评分示例

  • "刷牙" → 1 (完全日常)
  • "在咖啡店看到同事" → 3 (轻度社交)
  • "得知好友要结婚" → 8 (重要社交事件)
  • "被告知失去工作" → 10 (重大人生事件)

因子3: 相关性 (Relevance)

相关性通过语义嵌入的余弦相似度计算：

\[\text{relevance}(q, m) = \frac{\mathbf{e}_q \cdot \mathbf{e}_m}{\|\mathbf{e}_q\| \cdot \|\mathbf{e}_m\|}\]

其中 \(\mathbf{e}_q\) 和 \(\mathbf{e}_m\) 分别是查询和记忆的嵌入向量。

import numpy as np

def relevance_score(query_embedding, memory_embedding):
    """计算相关性评分（余弦相似度）"""
    dot_product = np.dot(query_embedding, memory_embedding)
    norm_product = np.linalg.norm(query_embedding) * np.linalg.norm(memory_embedding)
    if norm_product == 0:
        return 0.0
    return dot_product / norm_product

综合检索流程

def retrieve_memories(memory_stream, query, current_time, top_k=10,
                      alpha=1.0, beta=1.0, gamma=1.0, decay_rate=0.995):
    """从记忆流中检索最相关的记忆

    三因子加权检索：recency + importance + relevance
    """
    query_embedding = get_embedding(query)
    scored_memories = []

    for memory in memory_stream:
        rec = recency_score(memory, current_time, decay_rate)
        imp = (memory.importance - 1) / 9.0  # 归一化到[0,1]
        rel = relevance_score(query_embedding, memory.embedding)

        total = alpha * rec + beta * imp + gamma * rel
        scored_memories.append((memory, total))

    # 按总分排序，返回top_k
    scored_memories.sort(key=lambda x: x[1], reverse=True)

    # 更新访问时间
    for memory, _ in scored_memories[:top_k]:
        memory.last_access = current_time

    return [m for m, _ in scored_memories[:top_k]]

反思机制 (Reflection)

反思是记忆流架构中最关键的创新之一。智能体不仅存储原始观察，还能从多个记忆中合成更高层次的抽象洞察。

反思触发条件

当最近记忆的重要性评分之和超过阈值时触发反思：

\[\sum_{m \in \text{recent}} \text{importance}(m) \geq \theta_{\text{reflect}}\]

通常 \(\theta_{\text{reflect}} = 150\)（即重要性评分累计达到150时触发一次反思）。

反思生成过程

graph TD
    A[触发反思] --> B[确定反思主题]
    B --> C[检索相关记忆<br/>top 100]
    C --> D[LLM生成高层洞察]
    D --> E[创建反思记忆对象]
    E --> F[加入记忆流]
    F --> G[重置重要性累计]

    B -->|Prompt| B1["Given only the information above,<br/>what are 3 most salient high-level<br/>questions we can answer?"]
    D -->|Prompt| D1["What 5 high-level insights can you<br/>infer from the above statements?"]

反思层次

反思可以递归进行，形成多层抽象：

\[\text{Level 0: 观察} \rightarrow \text{Level 1: 反思} \rightarrow \text{Level 2: 元反思} \rightarrow \cdots\]
  • Level 0（观察）: "Klaus在画室画油画"
  • Level 1（反思）: "Klaus对艺术充满热情"
  • Level 2（元反思）: "Klaus是一个以创造力为核心身份的人"
def maybe_reflect(agent, threshold=150):
    """检查是否需要触发反思"""
    recent_importance_sum = sum(
        m.importance for m in agent.memory_stream
        if m.creation_time > agent.last_reflection_time
    )

    if recent_importance_sum >= threshold:
        # 生成反思主题
        recent_memories = get_recent_memories(agent, n=100)
        topics = generate_reflection_topics(recent_memories)

        for topic in topics:
            # 检索与主题相关的记忆
            relevant = retrieve_memories(agent.memory_stream, topic, 
                                        current_time=agent.current_time)
            # 生成反思洞察
            insights = generate_insights(relevant, topic)

            for insight in insights:
                # 创建反思记忆对象
                reflection = MemoryObject()
                reflection.description = insight
                reflection.type = "reflection"
                reflection.importance = rate_importance(insight)
                reflection.creation_time = agent.current_time
                reflection.related_ids = [m.id for m in relevant]

                agent.memory_stream.append(reflection)

        agent.last_reflection_time = agent.current_time

消融实验结果

Park et al. 对三因子检索和反思机制进行了系统的消融实验：

检索因子消融

配置 行为合理性评分 说明
完整模型（三因子 + 反思） 8.4 / 10 基线
移除时效性 7.2 / 10 智能体重复提及过时信息
移除重要性 7.5 / 10 无法区分琐事与重大事件
移除相关性 6.8 / 10 检索到不相关的记忆
仅时效性 5.9 / 10 只记得最近的事
仅相关性 6.3 / 10 无时间感知

反思机制消融

配置 行为合理性评分 关键观察
有反思 8.4 / 10 智能体展现深度理解
无反思 6.1 / 10 行为停留在表面反应
无反思 + 无规划 4.8 / 10 行为几乎随机

关键发现

反思机制的移除造成了最大的性能下降（-2.3分），表明高层抽象能力对于可信的智能体行为至关重要。相关性因子的移除是三个检索因子中影响最大的（-1.6分）。

记忆流的工程优化

向量索引

当记忆流增长到数千条记忆时，暴力检索变得低效：

# 使用FAISS进行近似最近邻搜索
import faiss

class OptimizedMemoryStream:
    def __init__(self, embedding_dim=1536):
        self.index = faiss.IndexFlatIP(embedding_dim)  # 内积索引
        self.memories = []

    def add_memory(self, memory):
        self.memories.append(memory)
        embedding = np.array([memory.embedding], dtype='float32')
        faiss.normalize_L2(embedding)  # 归一化后内积=余弦相似度
        self.index.add(embedding)

    def search_relevant(self, query_embedding, top_k=50):
        query = np.array([query_embedding], dtype='float32')
        faiss.normalize_L2(query)
        scores, indices = self.index.search(query, top_k)
        return [(self.memories[i], scores[0][j]) 
                for j, i in enumerate(indices[0])]

记忆压缩

长期运行的智能体需要记忆压缩策略：

\[\text{compress}(M) = \text{summarize}(\{m \in M : \text{importance}(m) < \theta_{\text{prune}}\})\]
  • 遗忘: 低重要性 + 低访问频率的记忆被删除
  • 合并: 相似的低层记忆被合并为摘要
  • 分层: 老记忆移入"长期存储"，降低检索频率

重要性评分缓存

# 批量评估重要性，减少LLM调用
def batch_rate_importance(descriptions, batch_size=10):
    """批量评估记忆重要性，减少API调用"""
    results = []
    for i in range(0, len(descriptions), batch_size):
        batch = descriptions[i:i+batch_size]
        prompt = format_batch_importance_prompt(batch)
        ratings = call_llm(prompt)
        results.extend(parse_ratings(ratings))
    return results

与其他记忆系统的比较

特性 记忆流 (Park) RAG MemGPT 传统数据库
时间感知 指数衰减 无 有限 可查询
重要性过滤 LLM评分 无 分层管理 手动标记
语义检索 嵌入相似度 嵌入相似度 嵌入相似度 SQL查询
抽象能力 反思机制 无 有限 无
可扩展性 中等 高 中等 高

总结

记忆流与反思机制的核心贡献：

  1. 三因子检索提供了比纯语义检索更加人性化的记忆访问模式
  2. 反思机制使智能体能够从经验中提取高层洞察，而非停留在表面
  3. 消融实验验证了每个组件的必要性，尤其是反思的关键作用
  4. 该架构为后续的虚拟具身智能体研究奠定了记忆系统的设计范式

评论 #