跳转至

情景与语义记忆

引言

Endel Tulving（1972）将长期记忆分为情景记忆（episodic）、语义记忆（semantic）和程序记忆（procedural）。这一分类为 Agent 记忆系统的设计提供了重要的理论指导。不同类型的记忆服务于不同的目的，需要不同的存储和检索机制。

Tulving 的记忆分类

记忆类型 内容 人类示例 Agent 示例
情景记忆 个人经历和事件 "昨天我去了公园" "用户昨天请求了数据分析"
语义记忆 一般知识和事实 "巴黎是法国首都" "Python 的 list 是可变类型"
程序记忆 技能和操作方式 骑自行车 调用 API 的标准流程

情景记忆（Episodic Memory）

定义与特征

情景记忆记录的是具体的经历和事件，具有以下特征：

  • 时间标记：发生在特定时间
  • 情境依赖：与特定场景相关
  • 自传性：与"我"（Agent 自身）相关
  • 可重放：可以回忆和重新体验

Agent 中的情景记忆

对话历史

最基本的情景记忆——记录 Agent 与用户的每次交互：

class EpisodicMemory:
    def __init__(self, vector_store):
        self.vector_store = vector_store

    def store_episode(self, episode):
        """存储一个交互事件"""
        record = {
            "id": generate_id(),
            "timestamp": datetime.now().isoformat(),
            "user_query": episode["query"],
            "agent_response": episode["response"],
            "tools_used": episode.get("tools", []),
            "outcome": episode.get("outcome", "unknown"),  # success/failure
            "context": episode.get("context", ""),
        }

        # 生成摘要用于嵌入
        summary = f"用户问: {record['user_query']}. Agent 回答后结果: {record['outcome']}"

        self.vector_store.add(
            documents=[summary],
            metadatas=[record],
            ids=[record["id"]]
        )

    def recall(self, query, n=5, time_range=None):
        """回忆相关经历"""
        filters = {}
        if time_range:
            filters["timestamp"] = {"$gte": time_range[0], "$lte": time_range[1]}

        return self.vector_store.query(
            query_texts=[query],
            n_results=n,
            where=filters if filters else None
        )

经验回放（Experience Replay）

借鉴强化学习的概念，Agent 可以从过去的经验中学习：

class ExperienceReplay:
    def __init__(self, memory, llm):
        self.memory = memory
        self.llm = llm

    def learn_from_failures(self, current_task):
        """从过去的失败经验中学习"""
        # 检索类似任务的失败案例
        failures = self.memory.recall(
            query=current_task,
            n=3,
            filters={"outcome": "failure"}
        )

        if failures:
            lessons = self.llm.generate(
                f"分析以下失败案例，提取对当前任务有用的教训：\n"
                f"当前任务: {current_task}\n"
                f"历史失败: {failures}\n"
                f"教训:"
            )
            return lessons
        return None

    def learn_from_successes(self, current_task):
        """从过去的成功经验中学习"""
        successes = self.memory.recall(
            query=current_task,
            n=3,
            filters={"outcome": "success"}
        )

        if successes:
            strategy = self.llm.generate(
                f"基于以下成功案例，为当前任务推荐策略：\n"
                f"当前任务: {current_task}\n"
                f"历史成功: {successes}\n"
                f"推荐策略:"
            )
            return strategy
        return None

Generative Agents 中的记忆流

Park et al.（2023）在 "Generative Agents" 中设计了记忆流（Memory Stream）机制：

class MemoryStream:
    """Generative Agents 风格的记忆流"""

    def __init__(self):
        self.memories = []

    def add_observation(self, description, importance=None):
        """添加观察"""
        if importance is None:
            importance = self.score_importance(description)

        self.memories.append({
            "description": description,
            "timestamp": datetime.now(),
            "importance": importance,  # 1-10
            "last_accessed": datetime.now(),
        })

    def retrieve(self, query, n=10):
        """基于相关性、时效性和重要性的综合检索"""
        scores = []
        for mem in self.memories:
            recency = self.recency_score(mem)
            importance = mem["importance"] / 10.0
            relevance = self.relevance_score(query, mem["description"])

            # 综合评分
            score = recency + importance + relevance
            scores.append((mem, score))

        scores.sort(key=lambda x: x[1], reverse=True)
        return [mem for mem, score in scores[:n]]

    def recency_score(self, memory):
        """时效性评分：指数衰减"""
        hours_ago = (datetime.now() - memory["last_accessed"]).total_seconds() / 3600
        return 0.99 ** hours_ago  # 衰减因子

语义记忆（Semantic Memory）

定义与特征

语义记忆存储的是一般性知识和事实，与具体经历脱钩：

  • 去情境化：不依赖于特定时间和场景
  • 结构化：可以组织为概念网络
  • 可推理：支持逻辑推断

知识图谱作为语义记忆

class SemanticMemory:
    """基于知识图谱的语义记忆"""

    def __init__(self):
        self.triples = []  # (subject, predicate, object)
        self.entity_index = {}  # entity -> related triples

    def add_knowledge(self, subject, predicate, obj, confidence=1.0):
        """添加知识三元组"""
        triple = {
            "subject": subject,
            "predicate": predicate,
            "object": obj,
            "confidence": confidence,
            "source": "observation",
            "updated_at": datetime.now(),
        }
        self.triples.append(triple)

        # 更新索引
        for entity in [subject, obj]:
            if entity not in self.entity_index:
                self.entity_index[entity] = []
            self.entity_index[entity].append(triple)

    def query(self, subject=None, predicate=None, obj=None):
        """查询知识"""
        results = self.triples
        if subject:
            results = [t for t in results if t["subject"] == subject]
        if predicate:
            results = [t for t in results if t["predicate"] == predicate]
        if obj:
            results = [t for t in results if t["object"] == obj]
        return results

    def get_entity_knowledge(self, entity, depth=1):
        """获取实体的相关知识（可多跳）"""
        if depth == 0 or entity not in self.entity_index:
            return []

        direct = self.entity_index[entity]
        if depth == 1:
            return direct

        # 多跳：获取邻居实体的知识
        result = list(direct)
        for triple in direct:
            neighbor = triple["object"] if triple["subject"] == entity else triple["subject"]
            result.extend(self.get_entity_knowledge(neighbor, depth - 1))
        return result

结构化事实存储

class StructuredFactStore:
    """用户相关的结构化事实"""

    def __init__(self):
        self.facts = {}  # category -> list of facts

    def add_fact(self, category, key, value, source="conversation"):
        if category not in self.facts:
            self.facts[category] = {}

        self.facts[category][key] = {
            "value": value,
            "source": source,
            "confidence": 1.0,
            "updated_at": datetime.now(),
        }

    def get_user_profile(self):
        """获取用户画像"""
        return self.facts

    # 使用示例：
    # store.add_fact("preferences", "response_style", "简洁")
    # store.add_fact("background", "programming_language", "Python")
    # store.add_fact("project", "current_task", "构建 RAG 系统")

程序记忆（Procedural Memory）

定义与特征

程序记忆存储的是技能、习惯和操作流程：

  • 隐式知识：通常难以用语言描述
  • 自动化：熟练后可以自动执行
  • 抗遗忘：一旦学会很难忘记

Agent 的程序记忆

学习到的工作流

class ProceduralMemory:
    """存储 Agent 学习到的操作流程"""

    def __init__(self, vector_store):
        self.vector_store = vector_store
        self.workflows = {}

    def store_workflow(self, task_type, steps, success_rate=None):
        """存储一个成功的工作流"""
        workflow = {
            "task_type": task_type,
            "steps": steps,
            "success_rate": success_rate,
            "usage_count": 1,
            "created_at": datetime.now().isoformat(),
        }

        self.workflows[task_type] = workflow

        # 同时存入向量数据库以支持语义检索
        description = f"任务类型: {task_type}. 步骤: {' -> '.join(steps)}"
        self.vector_store.add(
            documents=[description],
            metadatas=[workflow],
            ids=[f"workflow_{task_type}"]
        )

    def find_workflow(self, task_description, threshold=0.8):
        """查找适用的工作流"""
        results = self.vector_store.query(
            query_texts=[task_description],
            n_results=3,
        )

        if results and results["distances"][0][0] < threshold:
            return results["metadatas"][0][0]
        return None

Prompt 模板库

class PromptTemplateMemory:
    """存储和复用成功的 prompt 模板"""

    def __init__(self):
        self.templates = {}

    def store_template(self, name, template, performance_score):
        self.templates[name] = {
            "template": template,
            "score": performance_score,
            "usage_count": 0,
        }

    def get_best_template(self, task_type):
        """获取某类任务的最佳模板"""
        candidates = {k: v for k, v in self.templates.items() if task_type in k}
        if not candidates:
            return None
        return max(candidates.values(), key=lambda x: x["score"])

三种记忆的协同

graph TB
    Q[用户查询] --> C{查询分析}

    C -->|"做过类似的事吗？"| EP[情景记忆<br/>检索历史经验]
    C -->|"相关知识是什么？"| SM[语义记忆<br/>检索知识事实]
    C -->|"该怎么做？"| PM[程序记忆<br/>检索工作流]

    EP --> INT[记忆整合]
    SM --> INT
    PM --> INT

    INT --> LLM[LLM 推理]
    LLM --> ACT[行动/回答]

    ACT -->|记录经验| EP
    ACT -->|提取知识| SM
    ACT -->|更新流程| PM

实际应用示例

class IntegratedMemory:
    def __init__(self):
        self.episodic = EpisodicMemory(...)
        self.semantic = SemanticMemory()
        self.procedural = ProceduralMemory(...)

    def comprehensive_recall(self, query, task_type=None):
        """综合三种记忆进行回忆"""
        context = {}

        # 情景记忆：相关经验
        context["past_experiences"] = self.episodic.recall(query, n=3)

        # 语义记忆：相关知识
        entities = extract_entities(query)
        context["knowledge"] = []
        for entity in entities:
            context["knowledge"].extend(
                self.semantic.get_entity_knowledge(entity, depth=2)
            )

        # 程序记忆：适用的工作流
        if task_type:
            context["workflow"] = self.procedural.find_workflow(task_type)

        return context

记忆的更新与遗忘

知识冲突处理

当新信息与已有记忆冲突时：

def resolve_conflict(existing_fact, new_fact, llm):
    """处理知识冲突"""
    if new_fact["source"] == "user_explicit":
        # 用户明确陈述的优先级最高
        return new_fact

    if new_fact["timestamp"] > existing_fact["timestamp"]:
        # 更新的信息通常更可靠
        return new_fact

    # 让 LLM 判断
    resolution = llm.evaluate(
        f"已有知识: {existing_fact}\n新信息: {new_fact}\n哪个更可靠？为什么？"
    )
    return resolution

遗忘机制

并非所有记忆都需要永久保留：

  • 时间衰减：长时间未访问的记忆降低优先级
  • 重要性筛选：低重要性的记忆可以归档或删除
  • 容量管理：当存储接近上限时，清理最不重要的记忆

延伸阅读

  • 记忆流与反思机制 - Generative Agents 中的完整记忆架构
  • Tulving, E. (1972). "Episodic and semantic memory"
  • Park, J. S., et al. (2023). "Generative Agents: Interactive Simulacra of Human Behavior"
  • Shinn, N., et al. (2023). "Reflexion: Language Agents with Verbal Reinforcement Learning"

评论 #