在构建一个真正可用的 AI Agent 时，“记忆”不是一个可有可无的附加功能，而是系统能否持续理解用户、复用历史经验、稳定完成复杂任务的关键能力。

很多 Agent 在 Demo 阶段看起来表现不错，但一旦进入真实业务场景，就会暴露出几个典型问题：

  • 只能记住当前上下文，跨会话后就“失忆”。

  • 能检索知识库，但不知道用户过去问过什么。

  • 能生成答案，但无法从失败回答中总结经验。

  • 能调用工具，但不会根据历史效果调整工具使用策略。

因此，一个完整的 Agent 记忆系统通常不只是“把聊天记录存起来”，而是要区分不同层次的记忆：当前对话中的短期上下文、过去发生过的具体事件、长期沉淀的事实知识，以及从失败中反思和改进的机制。

本文围绕 工作记忆、情景记忆、语义记忆 三层记忆展开，并进一步对比 Hindsight / Hindsight Experience Replay 的作用，最后给出一个医疗 AI Agent 中的工程实现思路。

一、为什么 Agent 需要分层记忆？

人类的记忆并不是单一结构。

当我们正在解一道题时，脑中临时保存的题干、条件、当前推理步骤，类似于 工作记忆。当我们回忆上一次和某个人的对话、某次看病经历、某个项目事故时，调用的是 情景记忆。当我们知道“高血压可能与心血管风险相关”“Redis 可以做缓存和限流”这类事实时，依赖的是 语义记忆。

AI Agent 也类似。

如果所有信息都塞进一个数据库，系统很快会遇到三个问题：

  1. 检索粒度混乱：当前任务上下文、历史对话、领域知识混在一起，难以判断哪些信息应该优先使用。

  2. 生命周期不同：有些信息只在本轮对话有效，有些信息需要长期保留，有些信息需要被抽象成知识。

  3. 更新策略不同：工作记忆应该快速覆盖，情景记忆应该追加记录，语义记忆应该谨慎更新。

所以，分层记忆的核心价值是：让不同类型的信息进入不同的存储、检索和更新流程。

二、三层记忆的基本定义

1. 工作记忆：当前任务的临时操作区

工作记忆是 Agent 当前正在处理的信息缓冲区。

它保存的是和当前任务直接相关的信息，例如：

  • 用户本轮输入。

  • 最近几轮对话。

  • 当前任务目标。

  • 已识别出的关键实体。

  • 正在等待执行的工具调用。

  • 中间推理结果。

工作记忆的特点是生命周期短、容量小、更新频繁。它通常不适合作为长期知识库，而是更像当前任务的“草稿纸”。

在 LLM Agent 中，工作记忆常见实现包括：

  • Prompt 上下文窗口。

  • Redis 会话缓存。

  • 短期消息队列。

  • 当前请求内的内存对象。

例如医疗问答场景中，用户问：“高血压患者可以吃柚子吗？”如果前几轮用户已经说过自己正在服用某种降压药，那么这个药名就应该进入工作记忆，并直接影响本轮回答。

2. 情景记忆：过去发生过的具体事件

情景记忆记录的是具体发生过的事件，也就是 episode。

它保存的信息通常包括：

  • 用户过去问过什么。

  • Agent 当时如何回答。

  • 使用了哪些工具。

  • 检索到了哪些资料。

  • 用户是否追问、纠正或反馈。

  • 这次交互发生的时间、会话、上下文。

情景记忆的关键价值是让 Agent 能够“想起类似经历”。

比如用户曾经问过“我吃了降压药还能不能吃柚子”，后来又问“西柚会不会影响药效”，系统就可以从历史情景中召回相似对话，避免重复解释，也能保持用户体验的一致性。

工程上，情景记忆通常采用：

  • MySQL / PostgreSQL 存储结构化聊天记录。

  • Milvus / pgvector 存储向量索引。

  • 会话摘要表保存长对话压缩结果。

  • 时间索引支持按最近会话或特定阶段检索。

3. 语义记忆：长期沉淀的事实、概念和规则

语义记忆存储的是相对稳定的事实、知识、概念和关系。

它不依赖某一次具体对话，而是系统长期可复用的知识基础，例如：

  • 高血压的定义。

  • 药物相互作用。

  • 某疾病的典型症状。

  • 医学论文中的研究结论。

  • 题库系统中的考点关系。

  • 业务规则和领域概念。

语义记忆通常通过知识库、向量库或知识图谱实现。

常见技术组合包括：

  • Milvus / Elasticsearch / pgvector 做语义检索。

  • Neo4j 做实体关系查询。

  • BM25 + Embedding + Rerank 做混合检索。

  • 文档切块、摘要、关键词和元数据共同辅助召回。

在医疗 AI Agent 中，语义记忆可以回答“客观上有哪些已知事实”，例如“柚子中的呋喃香豆素可能影响 CYP3A4 酶，进而影响部分药物代谢”。

三、三层记忆的对比

维度工作记忆情景记忆语义记忆
本质当前任务缓冲区历史事件记录长期知识库
保存内容当前输入、最近上下文、中间状态对话、动作、反馈、结果事实、概念、规则、关系
生命周期单轮任务或短会话跨会话持久保存长期保存并持续更新
访问方式直接读取相似度检索、时间检索语义检索、图查询、关键词检索
更新方式实时覆盖或追加追加写入、定期摘要增量更新、重新嵌入、图谱维护
典型存储Prompt、Redis、内存对象MySQL + MilvusMilvus + Neo4j
主要作用支撑当前推理召回类似经历提供领域知识

可以把三层记忆理解成一个从短期到长期、从具体到抽象的结构：

当前请求
  ↓
工作记忆：当前任务上下文
  ↓
情景记忆：过去发生过的相似事件
  ↓
语义记忆：长期沉淀的领域知识

四、Hindsight 到底是什么？

在讨论记忆系统时，很多人会把 Hindsight 也放进“记忆”范畴。但严格来说，Hindsight 并不是一种普通记忆，而是一种 事后学习机制。

在认知心理学中，Hindsight 通常指“后见之明”偏差，也就是事情发生后，人会觉得自己早就知道结果。

但在 AI 和强化学习语境中，Hindsight 更常指 Hindsight Experience Replay，简称 HER。

HER 的核心思想是：

如果智能体没有达成原始目标，不要简单把这次轨迹当成失败丢掉，而是回头看它实际达成了什么目标，并把这条失败轨迹重标为另一个目标下的成功经验。

举一个简单例子。

假设机器人的原始目标是把物体推到 A 点，但最后只推到了 B 点。传统强化学习会认为这次失败，没有奖励。HER 会换一个角度：如果目标本来就是 B 点，那这条轨迹其实是成功的。于是系统把目标从 A 重标为 B，重新计算奖励，把失败经验转成可训练样本。

这就是 Hindsight 的价值：它不是“记住了什么”，而是“如何从失败中创造新的训练信号”。

五、Hindsight 与三层记忆的区别

维度工作记忆情景记忆语义记忆Hindsight / HER
本质在线处理缓冲区事件存储器知识存储器学习算法 / 数据增广机制
是否存储信息是是是本身不负责存储
处理对象当前上下文历史事件长期知识失败轨迹
关键动作保持上下文召回相似经历提供事实依据重标目标、重算奖励
时间方向使用当前和过去使用过去使用长期知识改造过去经验以改进未来
典型实现Prompt + RedisMySQL + 向量库向量库 + 知识图谱Replay Buffer + 目标重标

三层记忆是 存储结构，回答的是“系统记住了什么”。

Hindsight 是 学习机制，回答的是“系统如何从失败中学习”。

二者可以互补：情景记忆保存失败轨迹，Hindsight 对失败轨迹进行反思、重标和规则提炼，再把结果沉淀回语义记忆或策略模型中。

六、在医疗 Agent 中如何协同？

以一个医疗问答 Agent 为例，用户提问：

高血压患者可以吃柚子吗？

系统可以这样调用不同记忆层：

1. 工作记忆处理当前问题

工作记忆保存当前输入和最近对话。

如果用户前面提到“我正在吃硝苯地平”，系统需要把这个药名带入本轮推理，不能只孤立回答“高血压患者能否吃柚子”。

2. 语义记忆检索医学知识

系统从知识库或知识图谱中检索事实链：

柚子 / 西柚
  → 呋喃香豆素
  → 抑制 CYP3A4 酶
  → 可能影响部分降压药代谢
  → 增加药物浓度或不良反应风险

这部分提供的是客观知识依据。

3. 情景记忆召回历史类似对话

如果该用户之前问过“降压药和水果的相互作用”，系统可以召回当时的回答和用户反馈。

如果其他用户类似问题中出现过追问，比如“是不是所有降压药都不能吃柚子”，系统也可以提前补充解释，减少误解。

4. Hindsight 用于事后反思

如果 Agent 曾经给出过过度绝对的回答，例如“高血压患者不能吃柚子”，但后来被人工纠正为“关键取决于具体药物，尤其注意受 CYP3A4 影响的药物”，那么 Hindsight 机制可以事后分析这条失败轨迹。

它可以提炼出新的规则：

当用户询问食物与疾病关系时，
不要直接给出绝对禁止结论，
应进一步判断是否涉及具体药物、剂量、基础疾病和医生建议。

这条规则可以进入反思记忆、语义规则库或提示词策略中，用于改进未来回答。

七、工程实现：从概念到组件

一个可落地的三层记忆系统，可以按下面的方式映射到工程组件：

概念层             组件层                     数据层
------------------------------------------------------------
工作记忆     →     Redis / 会话上下文缓冲区     临时数据
情景记忆     →     MySQL + Milvus              历史消息 + 向量索引
语义记忆     →     Milvus + Neo4j              文档知识 + 实体关系
Hindsight    →     反思模块 / HER Buffer       训练样本或规则更新

1. 工作记忆实现

工作记忆适合放在 Redis 中，并设置 TTL。

class WorkingMemory:
    def __init__(self, redis_client, ttl_seconds=1800):
        self.redis = redis_client
        self.ttl = ttl_seconds
​
    def store(self, session_id: str, data: dict):
        key = f"session:{session_id}:working_memory"
        self.redis.setex(key, self.ttl, json.dumps(data, ensure_ascii=False))
​
    def retrieve(self, session_id: str) -> dict:
        key = f"session:{session_id}:working_memory"
        value = self.redis.get(key)
        return json.loads(value) if value else {}

典型字段包括：

{
  "last_input": "高血压患者可以吃柚子吗？",
  "recent_history": ["用户：我在吃降压药", "AI：请说明具体药名"],
  "active_entities": ["高血压", "柚子", "降压药"],
  "current_goal": "判断食物与药物相互作用风险",
  "pending_actions": []
}

2. 情景记忆实现

情景记忆建议同时存结构化记录和向量索引。

MySQL 负责保存完整、可追溯的对话数据：

messages
------------------------------------------------
id
user_id
session_id
role
content
tool_calls
sources
feedback
created_at

Milvus 负责相似历史召回：

class EpisodicMemory:
    def __init__(self, db, milvus_client):
        self.db = db
        self.milvus = milvus_client
​
    async def store(self, message):
        record_id = await self.db.insert_message(message)
        vector = await embed(message["content"])
        await self.milvus.insert(
            collection="episodic_memories",
            data=[{
                "message_id": record_id,
                "user_id": message["user_id"],
                "session_id": message["session_id"],
                "content": message["content"],
                "embedding": vector
            }]
        )
        return record_id
​
    async def retrieve(self, user_id: int, query: str, top_k: int = 5):
        vector = await embed(query)
        return await self.milvus.search(
            collection="episodic_memories",
            vector=vector,
            filter=f"user_id == {user_id}",
            top_k=top_k
        )

当会话过长时，可以异步生成摘要，避免未来检索时召回大量碎片化消息。

3. 语义记忆实现

语义记忆通常由 RAG 知识库和知识图谱共同承担。

RAG 负责从文档中召回语义相关片段：

用户问题
  → 查询改写
  → Embedding
  → Milvus 向量检索
  → BM25 关键词检索
  → RRF 融合
  → Rerank 精排
  → TopK 上下文注入 Prompt

知识图谱负责结构化关系查询：

MATCH (d:Disease)-[:HAS_SYMPTOM]->(s:Symptom)
WHERE d.name CONTAINS $disease
RETURN d.name, s.name
LIMIT 10

对医疗场景来说，语义记忆的优势是可追溯、可解释。向量检索能找到相关段落，图谱查询能返回实体关系，两者结合可以减少纯文本生成的幻觉风险。

八、完整数据流：一次对话如何经过记忆系统？

一次医疗问答可以拆成六步。

用户输入
  ↓
STEP 1：加载工作记忆
  从 Redis 读取当前会话上下文、最近历史、当前目标
  ↓
STEP 2：检索情景记忆
  从 MySQL / Milvus 召回该用户相似历史对话
  ↓
STEP 3：检索语义记忆
  并行调用 RAG 检索和 Neo4j 图谱查询
  ↓
STEP 4：融合与决策
  去重、排序、压缩，必要时触发联网搜索
  ↓
STEP 5：LLM 生成回答
  注入工作记忆、情景记忆、语义知识和来源信息
  ↓
STEP 6：写入记忆与反思
  保存消息、向量化索引、更新工作记忆、触发 Hindsight 反思

核心服务可以抽象成下面的结构：

class MedicalAgentService:
    def __init__(self):
        self.working_memory = WorkingMemory(redis_client)
        self.episodic_memory = EpisodicMemory(db, milvus_client)
        self.semantic_memory = SemanticMemory(milvus_client, neo4j_driver)
        self.reflector = HindsightReflector()
        self.llm = QwenLLM()
​
    async def chat(self, user_id: int, session_id: str, user_input: str):
        working_context = self.working_memory.retrieve(session_id)
​
        episodic_context = await self.episodic_memory.retrieve(
            user_id=user_id,
            query=user_input,
            top_k=5
        )
​
        rag_results = await self.semantic_memory.rag_search(user_input)
        graph_results = await self.semantic_memory.graph_search(user_input)
​
        fused_context = self.fuse_context(
            working_context=working_context,
            episodic_context=episodic_context,
            semantic_context=rag_results + graph_results
        )
​
        response = await self.llm.generate(
            query=user_input,
            context=fused_context
        )
​
        await self.episodic_memory.store({
            "user_id": user_id,
            "session_id": session_id,
            "role": "user",
            "content": user_input
        })
​
        await self.episodic_memory.store({
            "user_id": user_id,
            "session_id": session_id,
            "role": "assistant",
            "content": response.answer,
            "sources": response.sources
        })
​
        self.working_memory.store(session_id, {
            "recent_history": working_context.get("recent_history", []) + [
                user_input,
                response.answer
            ],
            "active_entities": response.entities,
            "current_goal": response.inferred_goal
        })
​
        await self.reflector.reflect_async(
            user_id=user_id,
            session_id=session_id,
            query=user_input,
            response=response,
            context=fused_context
        )
​
        return response

九、Hindsight 反思机制如何落地？

在普通问答系统中，回答结束就结束了。

但在具备 Hindsight 思想的 Agent 中，回答结束后还会进入一个“事后反思”阶段。

反思模块可以做几件事：

  1. 判断本次回答是否低置信度。

  2. 判断用户是否追问、否定或纠正。

  3. 判断引用来源是否不足。

  4. 判断工具调用是否失败。

  5. 从失败轨迹中提炼新的规则或偏好。

示例：

class HindsightReflector:
    async def reflect_async(self, user_id, session_id, query, response, context):
        if response.confidence >= 0.8 and not response.user_correction:
            return
​
        trajectory = await self.load_trajectory(session_id)
​
        reflection_prompt = f"""
        请分析这次 Agent 回答是否存在问题，并提炼可复用规则。
​
        用户问题：{query}
        Agent回答：{response.answer}
        置信度：{response.confidence}
        用户反馈：{response.user_feedback}
        检索上下文：{context}
​
        输出：
        1. 失败原因
        2. 应该如何改写目标
        3. 可复用的回答规则
        4. 是否需要更新用户画像或工具可靠性
        """
​
        reflection = await llm.invoke(reflection_prompt)
        await self.save_reflection(user_id, session_id, reflection)

在医疗场景中，Hindsight 反思尤其适合处理：

  • 回答过于绝对。

  • 没有补充免责声明。

  • 没有追问关键条件。

  • 检索来源不充分。

  • 工具调用失败但没有降级。

  • 用户纠正了 Agent 的错误理解。

这些反思结果不一定直接进入模型训练，也可以先沉淀为：

  • Prompt 规则。

  • 用户画像。

  • 工具可靠性评分。

  • 问答质量标签。

  • 高风险问题处理策略。

十、推荐的落地优先级

如果从零实现，不建议一开始就做复杂的长期反思系统。更稳妥的方式是分阶段落地。

第一阶段：先完成工作记忆和情景记忆

目标是让系统能正常保存对话、恢复会话、读取最近上下文。

建议实现：

  • Redis 保存当前会话上下文。

  • MySQL 保存完整消息记录。

  • 会话级 message_count、token_count、created_at、updated_at。

  • 支持按 session_id 查询历史消息。

第二阶段：加入语义记忆

目标是让系统不只依赖模型自身知识，而能基于外部知识库回答。

建议实现：

  • 文档切块。

  • Embedding 入库。

  • Milvus 向量检索。

  • BM25 关键词检索。

  • RRF 融合。

  • Rerank 精排。

  • 来源引用返回。

如果业务强依赖关系推理，再加入 Neo4j 知识图谱。

第三阶段：做记忆巩固

目标是解决长会话和跨会话检索质量问题。

建议实现：

  • 会话结束后生成摘要。

  • 长会话超过阈值后异步压缩。

  • 将摘要向量化后写入情景记忆库。

  • 从多次情景记忆中抽取稳定偏好，更新用户画像。

第四阶段：加入 Hindsight 反思机制

目标是让系统能从失败和反馈中改进。

建议实现：

  • 对低置信度回答触发反思。

  • 对用户纠错触发反思。

  • 对工具失败触发工具可靠性更新。

  • 将反思结果保存为规则或观点。

  • 定期人工审核高价值反思，避免错误规则污染系统。

十一、总结

三层记忆系统解决的是 Agent 的“记住什么”和“如何调用记忆”的问题。

工作记忆关注当前任务，负责保存短期上下文。情景记忆关注过去经历，负责召回相似事件。语义记忆关注长期知识，负责提供事实、概念和规则。

Hindsight 则不是普通记忆，而是一种从失败中学习的机制。它通过事后重标目标、分析失败轨迹、提炼规则和更新信念，让 Agent 不只是保存历史，还能从历史中改进。

一个更完整的 AI Agent 记忆架构应该是：

工作记忆：让我知道当前正在做什么
情景记忆：让我想起过去发生过什么
语义记忆：让我知道世界中有哪些事实
Hindsight：让我从失败中学到下次怎么做得更好

当这四部分协同起来，Agent 才能从一次性问答工具，逐步演进为能够持续理解用户、复用经验、沉淀知识并自我改进的智能系统。

# 数据库 # RAG # milvus # python
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

从想法到真机｜启元机器人开发者社区正式上线，面向所有具身智能创作者开放

avatar AtomGit开源社区
cover

最高赢取价值万元大奖！仓颉生态创新开发挑战赛等你来挑战

avatar AtomGit开源社区
cover

从创意画布到可编辑成片，开源 AI 正在重做视频工作流

avatar AtomGit开源社区