AI助手跨会话记忆：三层记忆架构设计与实践

“在上一篇文章中我们完成了 AI 助手从零到一的对话闭环用户输入消息大模型返回回答整个流程能跑通了。但如果你真的把这个助手放到实际业务里很快会遇到一个很尴尬的场景——用户昨天刚说过‘我是 Java 后端项目里用的 Spring Boot 3’今天再问它‘帮我看看这段配置为什么加载失败’它却一脸茫然完全不知道你在说什么。”这个问题的根源就是 AI 助手没有跨会话记忆。单个会话内它还能靠 Prompt 里的上下文维持“短期记忆”一旦会话结束所有信息清空它就变成了一个每次都重新认识用户的“陌生人”。本文是 AI 助手记忆系统的下篇重点解决“跨会话也记得你”的问题。我会带你实现一套三层记忆架构工作记忆、情景记忆、语义记忆分别对应人类记忆的不同层次。文章会从概念讲起给出完整的 Python 代码实现最后跑通一个带长期记忆的 AI 助手 demo。如果你正在开发 AI 助理、客服机器人、Agent 应用这篇文章可以直接作为你的记忆模块设计参考。1. 为什么 AI 助手需要跨会话记忆1.1 没有记忆的助手会遇到什么问题没有跨会话记忆的 AI 助手在实际使用中会暴露出几个非常明显的问题第一信息重复采集。用户每次开启新会话都要重新交代自己的身份、偏好、项目背景。比如用户上一周已经告诉助手自己用的是 MySQL 8.0这周再问“我的数据库连接池该怎么配”助手完全不知道这个前提条件回答只能是泛泛而谈。第二无法形成连续性服务。很多真实任务不是一次对话能完成的。比如用户让助手帮忙分析一段线上日志第二天接着问“昨天那个错误后来查到原因了吗”没有记忆的助手根本不知道“那个错误”指什么只能让用户重新描述一遍。第三个性化体验差。同一个助手面对不同用户应该有不同的回答风格和知识背景。没有跨会话记忆助手对所有用户一视同仁自然谈不上“懂你”。所以跨会话记忆不是“锦上添花”的功能而是 AI 助手能否承担真实工作的基础设施。1.2 跨会话记忆不是简单的上下文保留有人可能会说跨会话记忆不就是把上次对话的 Prompt 再拼回来吗这是一个常见的误区。如果只是机械地把历史消息全部塞回 Prompt会带来三个新的问题Token 成本爆炸。长对话累积起来非常快全部塞进 Prompt 很快就能打爆大模型的上下文窗口。噪声干扰判断。历史消息里大量无关内容会稀释关键信息导致大模型的回答质量反而下降。没有层次。人类回忆过去时不会把每一句话都重放一遍而是会按“当时发生了什么“”这个人是什么性格“”这类问题以前怎么解决“去组织记忆。所以跨会话记忆的核心不是“把更多东西检索出来”而是让 Agent 学会像人一样“回忆”——该想起来的时候想起来不该想起来的时候不打扰。这也是本文要讲的三层记忆架构的核心思想。1.3 三层记忆架构的设计思路三层记忆架构借鉴了认知心理学对人类记忆的分类方式工作记忆Working Memory对应当前对话的短期上下文容量小、时效性强用来维持当前这段对话的连贯性。情景记忆Episodic Memory对应“发生过什么事”记录用户与助手之间的历史互动比如用户某天问过什么问题、某个问题最后是怎么解决的。语义记忆Semantic Memory对应“用户的稳定事实”从历史对话中提炼出用户画像比如用户的职业、技术栈、偏好、长期目标。这三层各有分工合在一起就形成了一个完整的记忆系统。下面我们先搭建开发环境再逐层实现。2. 环境准备与项目结构2.1 开发环境与依赖本文的示例代码基于 Python 3建议使用 3.10 或更高版本代码中用到了较新的类型注解语法。项目本身对第三方依赖的要求很低核心存储我们用 SQLite 实现标准库即可完成大部分工作只有在调用大模型 API 做“记忆抽取”时才会用到 openai 库。如果你使用的是其他模型服务只要提供兼容 OpenAI 格式的 HTTP 接口代码结构基本相同。版本需要根据你的项目实际情况调整下面是一个最小依赖文件# requirements.txt openai1.0.0安装命令pip install -r requirements.txt为了方便调试建议在项目根目录创建.env文件用环境变量保存大模型服务的配置。本文示例中我们需要配置三个变量LLM_API_KEYsk-xxxxxxxxxxxxxxxx LLM_BASE_URLhttps://api.example.com/v1 LLM_MODELgpt-4o-mini请根据你实际使用的模型服务商填写注意不要把密钥提交到 Git 仓库。2.2 项目目录设计为了让代码职责清晰我们把记忆系统拆成独立模块。目录结构如下ai_memory_demo/ ├── main.py # 主程序交互式对话入口 ├── requirements.txt ├── memory/ │ ├── __init__.py │ ├── models.py # 记忆数据模型 │ ├── working.py # 工作记忆 │ ├── episodic.py # 情景记忆 │ ├── semantic.py # 语义记忆 │ └── service.py # 记忆服务统一封装三层记忆下面每个文件会逐个实现。你可能注意到这个项目里没有用任何 Web 框架原因是我们先把核心逻辑做对之后再接到 FastAPI 或 Flask 上都很容易。3. 记忆数据模型设计3.1 记忆项的通用结构在设计数据库表之前先想清楚一条“记忆”应该包含哪些字段。不管是什么类型的记忆我认为下面这些字段是通用的。# 文件路径memory/models.py from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Optional def _now() - str: 返回当前 UTC 时间的 ISO 格式字符串保证时间可比较、可序列化。 return datetime.now(timezone.utc).isoformat() dataclass class MemoryItem: memory_id: str # 唯一 ID memory_type: str # working / episodic / semantic content: str # 记忆内容文本 metadata: dict[str, Any] field(default_factorydict) # 扩展字段 importance: float 0.5 # 重要度0~1用于排序和遗忘 access_count: int 0 # 被访问次数用于热度排序 created_at: str field(default_factory_now) last_access_at: str field(default_factory_now)解释几个关键字段memory_type标识这条记忆属于哪一层。虽然三层记忆物理上可以分表存储但在同一套数据结构下定义便于后续统一做序列化和迁移。importance重要度评分。不是所有记忆都值得长期保存这个字段会在“记忆遗忘”阶段起到关键作用。access_count和last_access_at记录记忆的访问频率和最后访问时间。这两项指标用于实现 LRU最近最少使用思想帮助系统自动淘汰不再有价值的记忆。3.2 三类记忆的存储方案选择三类记忆对存储的要求不同。工作记忆体量小、变化快最简单的实现是用内存中的队列或列表例如 Python 的collections.deque。它的优势是读写 O(1)天然支持按长度淘汰。情景记忆的特点是“数量多、增长快、需要检索”。SQLite 是一个很好的默认选择单文件、零运维、支持 SQL能够满足万条级别的记忆检索。如果你希望在生产环境中做大规模向量检索后面可以替换为 FAISS、ChromaDB 或云数据库但核心检索逻辑是类似的。语义记忆是“用户画像”特点是条目少、更新频率低、但对准确性要求高。同样存储在 SQLite 中以“主体—谓词—客体”三元组的形式记录相当于一个精简版的知识图谱。4. 逐层实现三层记忆4.1 工作记忆会话内的短期缓冲工作记忆的本质就是当前会话的上下文缓冲。它的职责是维持本轮对话的流畅性让大模型能理解“刚才聊到哪了”。我用deque实现了一个固定长度的环形队列超过容量上限时最旧的消息会自动被挤出。# 文件路径memory/working.py from collections import deque from typing import Optional class WorkingMemory: 工作记忆保存当前会话最近的若干条消息。 def __init__(self, max_items: int 20): self.max_items max_items self._items deque(maxlenmax_items) def add(self, role: str, content: str) - None: 新增一条消息。role 取值为 user 或 assistant。 if not content.strip(): return self._items.append({role: role, content: content.strip()}) def recent(self, n: Optional[int] None) - list[dict]: 返回最近 n 条消息默认返回全部。 items list(self._items) if n is None: return items return items[-n:] def clear(self) - None: 清空工作记忆。新会话开始时通常需要调用。 self._items.clear() def estimate_tokens(self) - int: 粗略估算当前工作记忆占用的 token 数用于控制 Prompt 长度。 # 中文字符大约 1 个字符 ≈ 0.6~1 token这里做粗略估算即可 total_chars sum(len(str(m.get(content, ))) for m in self._items) return int(total_chars * 0.7)estimate_tokens方法虽然在示例中不是必须的但在生产环境里非常有用——当工作记忆过长时你可以根据估算值决定是否丢弃更早的消息或者先从稍远的历史里做摘要。4.2 情景记忆记录发生过的事情景记忆要回答的问题是“以前发生过什么”。我以“用户提问内容 关键词”为一条 episode 保存。检索时用当前用户问题的关键词去匹配历史 episode 的关键词按命中数量排序。# 文件路径memory/episodic.py import json import sqlite3 import uuid from datetime import datetime, timezone from typing import Optional class EpisodicMemory: 情景记忆记录用户与助手的历史互动事件。 def __init__(self, db_path: str episodic.db): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS episodes ( id TEXT PRIMARY KEY, user_id TEXT, content TEXT, keywords TEXT, created_at TEXT, importance REAL, access_count INTEGER, last_access_at TEXT ) ) self.conn.commit() def save( self, user_id: str, content: str, keywords: list[str], importance: float 0.5, ) - None: 保存一条情景记忆。 episode_id uuid.uuid4().hex now datetime.now(timezone.utc).isoformat() self.conn.execute( INSERT INTO episodes (id, user_id, content, keywords, created_at, importance, access_count, last_access_at) VALUES (?, ?, ?, ?, ?, ?, 0, ?) , ( episode_id, user_id, content, json.dumps(keywords, ensure_asciiFalse), now, importance, now, ), ) self.conn.commit() def search( self, user_id: str, query: str, limit: int 5, min_score: int 1, ) - list[dict]: 根据关键词匹配检索历史记录。返回按相关度降序排列的结果。 rows self.conn.execute( SELECT * FROM episodes WHERE user_id ? ORDER BY last_access_at DESC LIMIT 50 , (user_id,), ).fetchall() scored [] for row in rows: keywords json.loads(row[3]) score sum(1 for kw in keywords if kw in query) if score min_score: # row 结构 # 0:id, 1:user_id, 2:content, 3:keywords, # 4:created_at, 5:importance, 6:access_count, 7:last_access_at scored.append((score, row)) # 先按相关度降序再按创建时间降序 scored.sort(keylambda x: (x[0], x[1][4]), reverseTrue) return [self._row_to_dict(row) for score, row in scored[:limit]] staticmethod def _row_to_dict(row: tuple) - dict: return { id: row[0], user_id: row[1], content: row[2], keywords: json.loads(row[3]), created_at: row[4], importance: row[5], access_count: row[6], last_access_at: row[7], }这里需要说明一个设计细节为什么不用全文索引或者向量检索因为本文示例的目标是讲清楚链路。关键词评分在演示阶段足够用而且不依赖外部服务。如果你希望检索质量更高可以把search方法内部的匹配逻辑替换为 embedding 向量相似度计算——这不会影响上层调用方式。4.3 语义记忆提炼用户的稳定画像语义记忆存储的是“不被时间冲淡”的用户事实。比如用户是 Java 后端开发用户的项目用的是 Spring Boot 3用户偏好简洁的回答风格用户的长期目标是搭建一个自动化运维平台这些事实适合用三元组(subject, predicate, object)表示。SQLite 表设计如下# 文件路径memory/semantic.py import sqlite3 import uuid from datetime import datetime, timezone class SemanticMemory: 语义记忆保存从对话中提炼出的用户稳定事实。 def __init__(self, db_path: str semantic.db): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS facts ( id TEXT PRIMARY KEY, user_id TEXT, subject TEXT, predicate TEXT, object TEXT, confidence REAL, source TEXT, created_at TEXT, updated_at TEXT ) ) self.conn.commit() def upsert_fact( self, user_id: str, subject: str, predicate: str, object: str, confidence: float 0.5, source: str manual, ) - None: 写入或更新一条事实。 同一个 (user_id, subject, predicate) 只保留一条避免重复。 now datetime.now(timezone.utc).isoformat() existing self.conn.execute( SELECT id FROM facts WHERE user_id ? AND subject ? AND predicate ? , (user_id, subject, predicate), ).fetchone() if existing: self.conn.execute( UPDATE facts SET object ?, confidence ?, source ?, updated_at ? WHERE id ? , (object, confidence, source, now, existing[0]), ) else: fact_id uuid.uuid4().hex self.conn.execute( INSERT INTO facts (id, user_id, subject, predicate, object, confidence, source, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) , (fact_id, user_id, subject, predicate, object, confidence, source, now, now), ) self.conn.commit() def query_facts( self, user_id: str, subject: Optional[str] None, predicate: Optional[str] None, ) - list[dict]: 按条件查询用户事实默认返回全部并按置信度降序排列。 sql SELECT * FROM facts WHERE user_id ? params: list [user_id] if subject: sql AND subject ? params.append(subject) if predicate: sql AND predicate ? params.append(predicate) sql ORDER BY confidence DESC rows self.conn.execute(sql, params).fetchall() return [self._row_to_dict(row) for row in rows] staticmethod def _row_to_dict(row: tuple) - dict: return { id: row[0], user_id: row[1], subject: row[2], predicate: row[3], object: row[4], confidence: row[5], source: row[6], created_at: row[7], updated_at: row[8], }upsert_fact是语义记忆最重要的操作。它保证同一事实不会被重复插入——用户今天说“我喜欢 Python”明天说“我用 Python 写脚本”这两条事实会合并成同一条记录后面的信息覆盖前面而不是产生两条互相冲突的记忆。4.4 三个记忆层的协作关系三个记忆层不是孤立的而是按一条清晰的数据流协作用户输入 │ ▼ 工作记忆临时保存当前对话────┐ │ │ ▼ │ LLM 生成回答 ←──── 检索情景记忆 ←─┘ │ ▲ ▼ │ LLM 抽取语义事实 ────────┘ │ ▼ 更新语义记忆用户画像简单来说工作记忆承载当前会话的短期上下文每轮对话都会写入。检索器根据用户当前问题从情景记忆中找到相关的历史事件。语义记忆提供用户长期画像这部分与当前问题无关也会带上因为个性化信息对回答风格和准确性有帮助。每当一轮对话结束记忆抽取模块会判断有没有值得沉淀的长期事实有则写入语义记忆。这个链路是后面MemoryService的核心逻辑。5. 记忆的写入、检索与遗忘5.1 记忆写入链路记忆写入分为两层第一层是“记录”也就是把每轮对话都保存到情景记忆。这一步是无条件执行的因为谁也无法预判哪句话将来会被重新想起。我把用户的问题作为一条 episode 保存关键词由简单的提取函数生成。import re def extract_keywords(text: str, top_n: int 5) - list[str]: 极简关键词提取按汉字单字和英文单词切分去掉停用词按频率排序。 生产环境建议替换为 jieba 分词或通用分词库。 tokens re.findall(r[\u4e00-\u9fa5]|[a-zA-Z0-9], text) stopwords {我, 你, 您, 的, 了, 吗, 呢, 是, 在, 和, 或, 对, 就, 都} counter: dict[str, int] {} for token in tokens: if token not in stopwords and len(token.strip()) 0: counter[token] counter.get(token, 0) 1 sorted_tokens sorted(counter.items(), keylambda x: x[1], reverseTrue) return [token for token, _ in sorted_tokens[:top_n]]第二层是“提炼”也就是从自然对话中抽取稳定事实写入语义记忆。这一步建议交给大模型来做。下面给出一段兼容 OpenAI 格式的调用代码# 文件路径memory/service.py 内部使用的工具函数 import json from openai import OpenAI def extract_facts_with_llm( dialogue: list[dict], api_key: str, base_url: str, model: str, ) - list[dict]: 调用大模型抽取稳定事实。 dialogue 示例 [{role: user, content: 我是Java后端开发}, {role: assistant, content: 好的我记住了。}] 返回示例 [{subject: 用户, predicate: 职业, object: Java后端开发, confidence: 0.9}] client OpenAI(api_keyapi_key, base_urlbase_url) system_prompt ( 你是一个记忆抽取器。请从用户与助手的对话中抽取关于用户的稳定事实。\n 要求\n 1. 只抽取不容易改变的事实比如职业、技术栈、项目背景、长期目标、明确偏好。\n 2. 不抽取临时的、一次性的请求。\n 3. 输出 JSON 数组每个元素包含 subject、predicate、object、confidence 四个字段。\n 4. 如果没有稳定事实输出空数组 []。\n ) dialogue_text \n.join( [f{item[role]}: {item[content]} for item in dialogue] ) response client.chat.completions.create( modelmodel, temperature0, messages[ {role: system, content: system_prompt}, {role: user, content: dialogue_text}, ], ) content response.choices[0].message.content.strip() # 有些模型会输出 json 代码块做一次清洗 if content.startswith(): content content.strip() if content.startswith(json): content content[4:] try: return json.loads(content) except json.JSONDecodeError: return []为什么抽取稳定事实需要大模型而不是用规则因为自然语言里的“事实”往往藏在语境里。用户说“我在用 Spring Boot 做微服务”这句话里可以抽取的事实有“用户 - 技术栈 - Spring Boot”“用户 - 开发领域 - 微服务”。用规则很难覆盖所有说法而大模型在理解自然语言上有先天优势。5.2 记忆检索链路检索是记忆系统最容易出错的地方。我把检索拆成三个步骤。第一步先访问工作记忆拿到最近几轮消息保证对话的连贯性。第二步用当前用户问题去情景记忆里搜索相关历史。这里我们用的是关键词匹配。如果你希望更精准可以将关键词匹配替换成向量检索将问题转为 embedding与历史记录 embedding 计算余弦相似度取 Top-K。第三步直接读取语义记忆中该用户的全部事实按照置信度排序。因为用户画像通常不会太多几十条以内全部放入 Prompt 的成本可接受。这三个结果会经过一个“记忆上下文组装器”合并成一段结构文本供大模型参考。这个合并逻辑在下一节完整实现。5.3 记忆融合与评分合并后的记忆上下文需要按“对当前问题是否有帮助”做排序。重要程度排序规则如下语义记忆用户画像优先级最高因为它是长期稳定信息。情景记忆中与当前问题关键词重叠越多的排得越靠前。工作记忆最近几条始终在 Prompt 的最前面因为当前的对话流最相关。下面是一段组装代码也是MemoryService的核心方法def build_context(self, query: str, recent_n: int 6) - str: sections [] # 1. 工作记忆最近的对话 recent self.working_memory.recent(recent_n) if recent: lines [【最近对话】] for msg in recent: lines.append(f{msg[role]}: {msg[content]}) sections.append(\n.join(lines)) # 2. 情景记忆相关的历史事件 episodes self.episodic_memory.search(self.user_id, query, limit3) if episodes: lines [【相关历史对话】] for ep in episodes: lines.append(f- {ep[content]}) sections.append(\n.join(lines)) # 3. 语义记忆用户长期画像 facts self.semantic_memory.query_facts(self.user_id) if facts: lines [【用户画像】] for f in facts: lines.append(f- {f[subject]} {f[predicate]} {f[object]}) sections.append(\n.join(lines)) return \n\n.join(sections)这段代码的目的是把“记忆”变成大模型能直接理解的自然语言段落而不是直接把数据库记录丢给它。你会发现最终 Prompt 结构非常简单你是一个带长期记忆的 AI 助手。请结合下面的记忆上下文回答用户问题。 【最近对话】 user: ... assistant: ... 【相关历史对话】 - ... 【用户画像】 - 用户 职业 Java后端开发大模型看到这样的结构后回答时自然会把“用户画像”当作默认背景知识来使用。5.4 记忆遗忘与存储控制记忆不可能无限增长所以必须设计遗忘机制。常见策略有三类。基于容量的遗忘比如情景记忆默认只保留每个用户最近 2000 条记录超出后删除最旧的。实现上很直接SQL 删除即可DELETE FROM episodes WHERE user_id ? AND id NOT IN ( SELECT id FROM episodes WHERE user_id ? ORDER BY created_at DESC LIMIT 2000 );基于时间的遗忘比如只保留最近 90 天的情景记忆。适合隐私要求较高的业务场景。基于重要度的遗忘每条记忆都有importance字段系统定期清理低重要度、长期未被访问的记录。公式可以设计为保留分数 importance * 0.6 最近访问衰减分数 * 0.4不过要注意遗忘机制不能设计得太激进。用户说过的话即使当时看起来不重要未来也可能有价值。实际业务中建议同时提供“手动固定”功能把重点记忆标记为重要防止被自动清理掉。6. 完整实战给 AI 助手装上三层记忆6.1 创建项目文件与依赖现在把上面的模块组合成一个可运行的 demo。首先创建项目目录mkdir ai_memory_demo cd ai_memory_demo mkdir memory按照第 2 节的内容把models.py、working.py、episodic.py、semantic.py分别放到memory目录下。然后编写内存服务入口memory/service.py# 文件路径memory/service.py from .working import WorkingMemory from .episodic import EpisodicMemory from .semantic import SemanticMemory class MemoryService: 统一封装三层记忆对外提供记录、检索、提炼、组装上下文的接口。 def __init__(self, user_id: str): self.user_id user_id self.working_memory WorkingMemory(max_items20) self.episodic_memory EpisodicMemory(episodic.db) self.semantic_memory SemanticMemory(semantic.db) def record_exchange(self, user_message: str, assistant_reply: str) - None: 记录一轮对话到工作记忆和情景记忆。 self.working_memory.add(user, user_message) self.working_memory.add(assistant, assistant_reply) # 情景记忆以用户问题为主体保存 keywords extract_keywords(user_message) self.episodic_memory.save( user_idself.user_id, contentuser_message, keywordskeywords, importance0.5, ) def save_extracted_facts(self, facts: list[dict]) - int: 把大模型抽取的事实写入语义记忆返回新增/更新条数。 count 0 for fact in facts: subject fact.get(subject, 用户) predicate fact.get(predicate, 属性) obj fact.get(object, ) confidence float(fact.get(confidence, 0.5)) if not obj: continue self.semantic_memory.upsert_fact( user_idself.user_id, subjectsubject, predicatepredicate, objectobj, confidenceconfidence, sourcellm_extract, ) count 1 return count def build_context(self, query: str) - str: 组装记忆上下文。 这个字符串会拼进 System Prompt让大模型基于记忆回答问题。 sections [] # 1. 工作记忆最近的对话 recent self.working_memory.recent(6) if recent: lines [【最近对话】] for msg in recent: lines.append(f{msg[role]}: {msg[content]}) sections.append(\n.join(lines)) # 2. 情景记忆相关的历史事件 episodes self.episodic_memory.search(self.user_id, query, limit3) if episodes: lines [【相关历史对话】] for ep in episodes: lines.append(f- {ep[content]}) sections.append(\n.join(lines)) # 3. 语义记忆用户长期画像 facts self.semantic_memory.query_facts(self.user_id) if facts: lines [【用户画像】] for f in facts: lines.append(f- {f[subject]} {f[predicate]} {f[object]}) sections.append(\n.join(lines)) return \n\n.join(sections) def clear_working_memory(self) - None: 新会话开始时可以只清空工作记忆保留情景记忆和语义记忆。 self.working_memory.clear()6.2 实现对话主程序接下来写main.py这是交互式运行入口# 文件路径main.py import os import re from openai import OpenAI from memory.service import MemoryService def load_api_config(): 从环境变量读取 LLM 配置。 return { api_key: os.getenv(LLM_API_KEY, ), base_url: os.getenv(LLM_BASE_URL, https://api.example.com/v1), model: os.getenv(LLM_MODEL, gpt-4o-mini), } def extract_keywords(text: str, top_n: int 5) - list[str]: 简单关键词提取用于情景记忆检索。 tokens re.findall(r[\u4e00-\u9fa5]|[a-zA-Z0-9], text) stopwords {我, 你, 您, 的, 了, 吗, 呢, 是, 在, 和, 或, 对, 就, 都} counter: dict[str, int] {} for token in tokens: if token not in stopwords and len(token.strip()) 0: counter[token] counter.get(token, 0) 1 sorted_tokens sorted(counter.items(), keylambda x: x[1], reverseTrue) return [token for token, _ in sorted_tokens[:top_n]] def extract_facts_with_llm(dialogue, api_key, base_url, model): 调用 LLM 抽取稳定事实。 client OpenAI(api_keyapi_key, base_urlbase_url) system_prompt ( 你是一个记忆抽取器。请从用户与助手的对话中抽取关于用户的稳定事实。\n 要求\n 1. 只抽取不容易改变的事实。\n 2. 输出 JSON 数组每个元素包含 subject、predicate、object、confidence 四个字段。\n 3. 如果没有稳定事实输出空数组 []。\n ) dialogue_text \n.join([f{item[role]}: {item[content]} for item in dialogue]) response client.chat.completions.create( modelmodel, temperature0, messages[ {role: system, content: system_prompt}, {role: user, content: dialogue_text}, ], ) content response.choices[0].message.content.strip() if content.startswith(): content content.strip() if content.startswith(json): content content[4:] try: import json return json.loads(content) except json.JSONDecodeError: return [] def call_llm(messages, api_key, base_url, model) - str: 调用 LLM 生成回答。 client OpenAI(api_keyapi_key, base_urlbase_url) response client.chat.completions.create( modelmodel, messagesmessages, temperature0.7, ) return response.choices[0].message.content.strip() def main(): config load_api_config() if not config[api_key]: print(请先设置环境变量 LLM_API_KEY、LLM_BASE_URL、LLM_MODEL) return # 每个用户对应一个 MemoryService。真实项目中 user_id 来自登录态。 service MemoryService(user_iduser_demo_001) print( AI 助手带三层记忆) print(输入 exit 退出) print(输入 /clear 清空工作记忆情景记忆和语义记忆保留) # 记录最近一轮对话用于事实抽取 last_dialogue [] while True: user_input input(\n你: ).strip() if user_input.lower() exit: break if user_input /clear: service.clear_working_memory() print(系统: 工作记忆已清空) continue # 1. 组装记忆上下文 context service.build_context(user_input) # 2. 构造带记忆的 Prompt system_prompt ( 你是一个带长期记忆的 AI 助手。 请结合下面的记忆上下文回答用户问题。 如果记忆与问题无关请忽略记忆正常回答。\n\n context ) messages [ {role: system, content: system_prompt}, {role: user, content: user_input}, ] # 3. 调用 LLM reply call_llm(messages, config[api_key], config[base_url], config[model]) print(f助手: {reply}) # 4. 记录到工作记忆 情景记忆 service.record_exchange(user_input, reply) # 5. 抽取语义事实并沉淀 last_dialogue [ {role: user, content: user_input}, {role: assistant, content: reply}, ] facts extract_facts_with_llm( last_dialogue, config[api_key], config[base_url], config[model], ) if facts: count service.save_extracted_facts(facts) print(f[记忆系统] 已沉淀 {count} 条长期事实) if __name__ __main__: main()需要注意extract_keywords在main.py和memory/service.py里各有一份。正常项目中你应该把它抽到一个公共utils.py文件里这里是示例代码便于读者直接复制。6.3 运行与验证确保环境变量已经配置好然后运行python main.py第一轮对话输入你: 我是Java后端开发主要用Spring Boot做微服务助手回答后终端会打印类似信息[记忆系统] 已沉淀 2 条长期事实此时强制退出程序再重新运行python main.py。注意数据库文件semantic.db和episodic.db已经保存在磁盘上情景记忆和语义记忆并没有丢失。新会话里直接问你: 我平时用什么技术栈正常情况下助手的回答会提到“你之前说过自己是 Java 后端开发主要使用 Spring Boot”而不是反问“你用什么技术栈呀之前好像没聊过”。这说明语义记忆成功跨会话生效了。再试一个自然衔接的问题你: 帮我写一个Spring Boot的HelloWorld接口这次工作记忆和情景记忆会提供上一轮的相关上下文助手能直接结合“用户是 Java 开发者、用 Spring Boot”这个背景生成代码而不需要用户重复解释。6.4 结果说明从上面的演示可以看出三层记忆各司其职工作记忆负责跨“消息”的连贯性保证同一会话内不重复提问。情景记忆负责跨“会话”的联想当用户再次提到类似话题时能召回历史讨论。语义记忆负责跨“时间”的稳定画像无论多久之后再对话关键用户事实都不会丢。这套设计还有一个很重要的优点即使大模型上下文窗口很小也能通过“先检索、再组装”的方式把最相关的记忆放进去而不是把所有历史全量塞入。7. 常见问题与排查思路问题现象常见原因解决思路跨会话重启后助手完全“失忆”情景记忆、语义记忆没有持久化数据存在内存里确认episodic.db和semantic.db是否正常创建检查MemoryService是否正确加载同一个数据库文件情景记忆检索不到相关内容关键词提取效果差把extract_keywords替换为 jieba 分词或向量检索增加检索返回条数limit语义记忆里存了太多冲突事实没有对重复事实做合并使用upsert_fact方法按(user_id, subject, predicate)去重LLM 抽取事实返回空数组模型输出格式不符合预期在 system prompt 中明确要求输出 JSON部分模型需要把response_format设为json_objectToken 占用过高记忆上下文组装过多限制recent_n、episodic search limit、语义事实数量用estimate_tokens做预算同一个用户不同会话之间数据串了没有按user_id隔离所有查询都必须带user_id条件生产环境建议为每个用户分表或增加 tenant_id数据库文件损坏异常退出或并发写入SQLite 适合单进程写入多线程场景建议改用 PostgreSQL或者加 WAL 模式记忆更新后效果没生效缓存问题或没有刷新上下文检查组装上下文时机确保每次请求都重新调用build_context()如果你遇到跨会话记忆不生效建议按以下顺序排查先确认对话结束后有没有调用record_exchange。确认数据库文件里有没有数据sqlite3 episodic.db SELECT * FROM episodes LIMIT 3; sqlite3 semantic.db SELECT * FROM facts LIMIT 3;在build_context方法中打印组装好的上下文确认拼接结果。最终把组装好的 System Prompt 复制到模型调试台里人工判断记忆是否被正确引用。8. 最佳实践与工程建议8.1 记忆写入准确率优化记忆系统最怕“什么都记、什么都不敢记”。无论是情景记忆还是语义记忆在写入前都要做质量控制。情景记忆建议做去重。用户可能在一次会话里多次提问同一个问题连续记录会占用大量存储空间。可以在写入前判断内容相似度如果与最近一条记录字数相近、关键词重合度高就跳过写入。语义记忆建议做“置信度门槛”。大模型抽取事实时会返回 confidence 字段比如 0.95 表示非常确定0.3 表示猜测。业务上可以设定阈值低于 0.6 的事实不写入或者写入后标记为“待确认”状态等用户下一次对话时再验证。另外重要度importance字段不要长期用一个固定值。可以根据用户确认来调整用户明确说“记住了”“对对对”时给这条记忆提高重要度如果用户对助手的记忆内容表示否定则应立即删除或降低重要度。8.2 存储与性能优化当用户量和记忆条数增长后SQLite 单文件存储会逐渐遇到瓶颈。建议做下面几件事第一加索引。episodes表的检索条件集中在user_id和keywords至少要为user_id建索引。facts表同理。CREATE INDEX idx_episodes_user_id ON episodes(user_id); CREATE INDEX idx_facts_user_id ON facts(user_id);第二引入向量检索。当情景记忆超过几万条后关键词匹配的召回率会明显下降。建议把所有历史记录离线生成 embedding在线检索时使用 FAISS 或向量数据库做 ANN 近似最近邻搜索。你可以把search方法抽象成一个接口内部实现替换为向量检索上层逻辑不用改动。第三控制记忆上下文长度。即使检索做得很好也不能把所有命中结果都塞进 Prompt。建议给各部分设置预算比如工作记忆 1000 token情景记忆 1000 token语义记忆 800 token超出部分按重要度截断。8.3 安全与隐私边界记忆系统存储的数据高度敏感因为它记录的是用户说过的话、用户画像和偏好。这部分要特别注意不存储明文敏感信息。密码、身份证号、银行卡号等数据不仅不应该存进记忆在对话记录里也应该做脱敏处理。支持用户主动删除。产品必须提供“清除我的记忆”功能。实现上就是按user_id删除对应数据库记录。最小权限原则。不是所有系统组件都应该能读全部记忆。建议把记忆服务单独部署只开放必要的读接口。合规审计。每次记忆写入和读取都应有日志包括时间、操作方、记忆 ID。这样如果出现数据泄漏可以快速定位。这里需要特别提醒在生产环境执行任何删除操作前必须先备份数据库并且在测试环境充分验证 SQL避免误删全量数据。8.4 可观测性与调试记忆系统是个“黑盒”如果不可观测排查问题会非常痛苦。建议为每一条记忆添加source字段标注记忆来源例如user_input、llm_extract、manual_confirm。这样将来发现某条记忆不准确时能追溯到是哪个环节产生了它。同时建议在开发环境打印每一步的记忆上下文方便核对# 示例开发环境的调试图 context service.build_context(user_input) print( 本次请求的记忆上下文 ) print(context) print( 记忆上下文结束 )生产环境不要打印完整上下文可能包含用户敏感信息建议只记录记忆命中的 ID 列表。9. 下一步可以继续做什么从“无记忆”到“三层记忆”我们已经走完了最关键的一步。下一步我建议你按自己产品的形态从下面三个方向继续深入。第一把记忆服务改造成独立中间件。目前MemoryService和主程序耦合在同一个进程里真实项目中你可能希望它作为独立服务通过 HTTP 接口对外提供“写入记忆”“检索记忆”“删除记忆”能力。这样前端、后端、Agent 都能复用同一套记忆也不会互相污染。第二完善记忆可视化。给用户提供一个“助手记得什么”的管理页面展示三层记忆内容允许用户编辑、确认和删除。这一步对提升 AI 产品的可信度非常重要。第三引入多轮对话摘要。工作记忆的容量有限当一段会话特别长时与其丢弃较早信息不如让大模型把之前的对话压缩成摘要再存回情景记忆。这样既节省 token又不丢关键上下文。记忆系统是 AI 助手从“能用”走向“好用”的关键模块。希望这篇文章能帮你少走弯路动手把属于自己的记忆系统搭起来。如果你在实现过程中遇到问题欢迎在评论区留言交流。