从零构建聊天机器人记忆系统：基于LLM与向量检索的工程实践

1. 项目概述：从零构建聊天机器人记忆系统

最近几年，大语言模型驱动的聊天机器人遍地开花，但很多开发者都踩过同一个坑：聊着聊着，机器人就“失忆”了。你刚告诉它你养了一只叫“奥利奥”的猫，三句话之后它可能就问你“你养宠物吗？”。这种上下文遗忘问题，本质上是模型本身有限的“工作记忆”窗口导致的。一个真正智能、能持续对话的助手，必须拥有自己的“记忆系统”。市面上有很多现成的库和框架，但如果你真想理解记忆机制的核心，并能在不同场景下灵活定制，亲手从零实现一套算法，是性价比最高的学习路径。这个项目，就是带你一步步拆解“聊天机器人记忆”这个黑盒，用代码构建属于你自己的记忆引擎。

简单来说，我们要做的，是给聊天机器人装上一个外置的“大脑皮层”。这个系统需要能存储历史对话中的关键信息（比如用户偏好、事实陈述、任务状态），能检索当前对话所需的相关记忆，并能管理记忆的存储、更新和遗忘。它不依赖于模型自身那有限的上下文长度，而是作为一个独立的、可持久化的模块存在。无论你是想做一个能记住用户口味的点餐助手，一个能跟进项目进度的协作机器人，还是一个能陪你聊天的虚拟伙伴，这套底层能力都是刚需。

2. 记忆系统的核心架构与设计思路

2.1 记忆的本质：从对话流中提取结构化信息

首先我们要破除一个迷思：记忆不是简单地把所有聊天记录存进一个文本文件。那样做，检索效率会低得可怕，而且会塞满大量无关噪音。记忆的本质，是从非结构化的对话流中，提取出结构化的、高信息密度的“知识单元”。

举个例子，用户说：“我住在北京朝阳区，最喜欢吃四川火锅，对花生过敏。” 原始对话是一句话。但我们的记忆系统应该从中提取出至少三个独立的知识单元（或称为记忆片段）：

1. 属性：居住地 -> 北京朝阳区
2. 偏好：食物 -> 四川火锅
3. 禁忌：过敏源 -> 花生

每个知识单元都应该包含几个核心字段：内容、类型（是事实、偏好、任务还是其他）、实体（涉及的人、物、地点）、时间戳、访问频率/强度。这种结构化的存储，是后续高效检索和推理的基础。

2.2 核心组件拆解：一个记忆系统的四大支柱

一个完整的记忆系统，通常由四个核心组件构成，理解它们是你设计时的蓝图：

记忆存储（Memory Storage）：这是记忆的“仓库”。最简单的可以用一个Python列表或字典在内存中维护。但为了持久化和更复杂的查询，我们通常会引入数据库。对于入门，SQLite是绝佳选择，它轻量、无需服务器，可以直接用sqlite3库操作。我们会设计一张memories表，字段就对应上面提到的知识单元。

记忆编码（Memory Encoding）：这是把原始对话文本变成结构化记忆的“翻译官”。这里是大语言模型（LLM）大显身手的地方。我们需要设计一个“提示词（Prompt）”，引导LLM从一段对话中提取关键信息，并以我们约定的格式（比如JSON）输出。这一步的质量直接决定了记忆的可用性。

记忆检索（Memory Retrieval）：这是整个系统的“搜索引擎”。当新对话发生时，我们需要从海量记忆中快速找到最相关的几条。这里的关键是“相关性计算”。最经典的方法是使用向量检索。我们将记忆的文本内容（或经过处理的摘要）通过一个嵌入模型（Embedding Model）转换成高维向量，同样把当前用户的问题也转换成向量。然后计算它们之间的余弦相似度，相似度最高的记忆就是最相关的。我们会在本地使用像sentence-transformers这样的库来生成向量。

记忆管理（Memory Management）：这是系统的“管家”，负责记忆的更新、合并、衰减和遗忘。比如，用户之前说“我喜欢蓝色”，后来又说“我现在最喜欢绿色了”，系统需要能更新这条偏好记忆，而不是存储两条矛盾的。另外，长期不用的记忆应该逐渐“淡忘”（降低检索优先级或归档），避免仓库无限膨胀。

2.3 技术选型与工具链

为了专注于算法逻辑，我们选择最轻量、最易上手的工具链：

• 语言：Python。生态丰富，适合快速原型开发。
• 向量数据库：初期为了简化，我们不用专门的向量数据库（如Pinecone, Weaviate），而是自己实现一个简单的基于FAISS（Facebook AI Similarity Search）的索引。FAISS是一个高效的向量相似度搜索库，能轻松处理成千上万的向量。
• 嵌入模型：选用sentence-transformers库中的all-MiniLM-L6-v2模型。这个模型在速度和效果上取得了很好的平衡，且完全可以在本地CPU上运行，无需GPU。
• 大语言模型（用于编码）：为了可复现性和零成本，我们使用一个开源的、可在本地运行的轻量级模型，例如通过Ollama运行的Llama 3.2或Qwen2.5系列。这能保证我们的记忆编码环节完全离线、可控。如果你没有本地条件，也可以暂时用OpenAI等云端API替代，但会引入网络依赖和成本。
• 传统数据库：使用Python内置的sqlite3来存储记忆的元数据和原始文本。

注意：选择本地化工具链的核心目的是教学和可控。在实际生产环境中，根据数据量、并发量和延迟要求，你可能需要升级到更强大的向量数据库（如Qdrant）和更大型的LLM API服务。

3. 逐步实现：搭建你的记忆引擎

3.1 第一步：定义记忆的数据结构

一切从定义开始。我们先创建一个memory.py文件，定义一个记忆条目的类。

# memory.py import json from datetime import datetime from typing import Optional, Dict, Any import uuid class MemoryItem: def __init__(self, content: str, memory_type: str = "fact", entities: list = None, metadata: Optional[Dict[str, Any]] = None, strength: float = 1.0): """ 初始化一个记忆单元。 :param content: 记忆的文本内容，例如“用户喜欢咖啡”。 :param memory_type: 记忆类型，如 'fact'(事实), 'preference'(偏好), 'task'(任务), 'emotion'(情绪)。 :param entities: 该记忆涉及到的实体列表，如 ['用户', '咖啡']。 :param metadata: 其他元数据，如来源对话ID、创建时间等。 :param strength: 记忆强度，初始为1.0。随着时间推移和反复提及，强度会变化。 """ self.id = str(uuid.uuid4()) # 唯一标识符 self.content = content self.memory_type = memory_type self.entities = entities if entities is not None else [] self.metadata = metadata if metadata is not None else {} self.strength = strength self.created_at = datetime.now().isoformat() self.last_accessed_at = self.created_at # 向量表示，初始为None，在编码后填充 self.embedding = None # 确保metadata中有时间戳 if 'created_at' not in self.metadata: self.metadata['created_at'] = self.created_at def to_dict(self) -> Dict[str, Any]: """将记忆对象转换为字典，便于序列化存储。""" return { 'id': self.id, 'content': self.content, 'memory_type': self.memory_type, 'entities': self.entities, 'metadata': self.metadata, 'strength': self.strength, 'created_at': self.created_at, 'last_accessed_at': self.last_accessed_at, 'embedding': self.embedding.tolist() if self.embedding is not None else None } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'MemoryItem': """从字典还原记忆对象。""" memory = cls( content=data['content'], memory_type=data['memory_type'], entities=data['entities'], metadata=data['metadata'], strength=data['strength'] ) memory.id = data['id'] memory.created_at = data['created_at'] memory.last_accessed_at = data['last_accessed_at'] if data['embedding']: import numpy as np memory.embedding = np.array(data['embedding']) return memory

这个类封装了一条记忆的所有信息。embedding字段稍后我们会用向量模型来填充。strength字段是实现记忆衰减与巩固的关键。

3.2 第二步：实现记忆存储层（SQLite + FAISS）

接下来，我们创建memory_store.py，实现一个同时管理结构化数据（SQLite）和向量数据（FAISS）的存储层。

# memory_store.py import sqlite3 import json import numpy as np import faiss from typing import List, Optional, Tuple from memory import MemoryItem class MemoryStore: def __init__(self, db_path: str = "chatbot_memory.db", embedding_dim: int = 384): """ 初始化记忆存储。 :param db_path: SQLite数据库文件路径。 :param embedding_dim: 向量嵌入的维度，需与使用的嵌入模型匹配。 """ self.db_path = db_path self.embedding_dim = embedding_dim self._init_db() self._init_faiss_index() def _init_db(self): """初始化SQLite数据库表。""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, content TEXT NOT NULL, memory_type TEXT, entities TEXT, -- 存储为JSON字符串 metadata TEXT, -- 存储为JSON字符串 strength REAL DEFAULT 1.0, created_at TEXT, last_accessed_at TEXT, embedding_blob BLOB -- 存储FAISS索引所需的向量（可选，我们主要用FAISS存） ) ''') conn.commit() conn.close() def _init_faiss_index(self): """初始化FAISS向量索引。这里使用最简单的L2距离索引。""" self.index = faiss.IndexFlatL2(self.embedding_dim) # 使用L2（欧氏距离）度量 self.id_to_memory = {} # 用于通过FAISS返回的索引ID映射回MemoryItem对象 def add_memory(self, memory: MemoryItem): """添加一条记忆到存储。""" # 1. 存入SQLite conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO memories (id, content, memory_type, entities, metadata, strength, created_at, last_accessed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ''', ( memory.id, memory.content, memory.memory_type, json.dumps(memory.entities), json.dumps(memory.metadata), memory.strength, memory.created_at, memory.last_accessed_at )) conn.commit() conn.close() # 2. 如果记忆有向量，则加入FAISS索引 if memory.embedding is not None: # 确保向量是二维数组 [1, embedding_dim] vector = np.array([memory.embedding]).astype('float32') faiss_id = len(self.id_to_memory) # 使用一个简单的自增ID作为FAISS的内部ID self.index.add(vector) self.id_to_memory[faiss_id] = memory.id # 映射：FAISS ID -> 内存ID def search_by_vector(self, query_vector: np.ndarray, k: int = 5) -> List[Tuple[MemoryItem, float]]: """ 通过向量相似度搜索记忆。 :param query_vector: 查询向量，形状应为 [1, embedding_dim]。 :param k: 返回最相关的k条记忆。 :return: 包含（记忆对象，相似度距离）的列表。距离越小越相似。 """ if self.index.ntotal == 0: # 索引为空 return [] query_vector = query_vector.astype('float32') distances, indices = self.index.search(query_vector, k) results = [] for i, (idx, dist) in enumerate(zip(indices[0], distances[0])): if idx != -1: # FAISS可能返回-1 memory_id = self.id_to_memory.get(idx) if memory_id: memory = self.get_memory_by_id(memory_id) if memory: # 更新该记忆的最后访问时间（模拟记忆的“激活”） memory.last_accessed_at = datetime.now().isoformat() self.update_memory(memory) results.append((memory, float(dist))) return results def get_memory_by_id(self, memory_id: str) -> Optional[MemoryItem]: """根据ID从数据库获取记忆。""" conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute('SELECT * FROM memories WHERE id = ?', (memory_id,)) row = cursor.fetchone() conn.close() if row: data = dict(row) data['entities'] = json.loads(data['entities']) data['metadata'] = json.loads(data['metadata']) return MemoryItem.from_dict(data) return None def update_memory(self, memory: MemoryItem): """更新一条已有的记忆。""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' UPDATE memories SET content=?, memory_type=?, entities=?, metadata=?, strength=?, last_accessed_at=? WHERE id=? ''', ( memory.content, memory.memory_type, json.dumps(memory.entities), json.dumps(memory.metadata), memory.strength, memory.last_accessed_at, memory.id )) conn.commit() conn.close() # 注意：更新内容后，其向量可能已变，但FAISS索引更新复杂。简化处理：不更新FAISS中的旧向量。 # 生产环境需要考虑更复杂的向量索引更新策略，如标记删除后重新添加。 def get_all_memories(self, memory_type: Optional[str] = None) -> List[MemoryItem]: """获取所有记忆，可按类型过滤。""" conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() if memory_type: cursor.execute('SELECT * FROM memories WHERE memory_type = ? ORDER BY last_accessed_at DESC', (memory_type,)) else: cursor.execute('SELECT * FROM memories ORDER BY last_accessed_at DESC') rows = cursor.fetchall() conn.close() memories = [] for row in rows: data = dict(row) data['entities'] = json.loads(data['entities']) data['metadata'] = json.loads(data['metadata']) memories.append(MemoryItem.from_dict(data)) return memories

这个MemoryStore类是我们的核心数据管理层。它巧妙地结合了SQLite的关系型存储（适合精确查询和元数据管理）和FAISS的向量检索（适合相似度搜索）。id_to_memory字典是连接FAISS索引ID和我们业务逻辑记忆ID的桥梁。

实操心得：这里有一个简化处理。当记忆内容更新时，我们没有同步更新FAISS索引中的向量，因为更新向量需要先删除再添加，操作相对复杂。在原型阶段，我们可以接受这种轻微的不一致，或者采用“惰性更新”策略：在下次检索时，如果发现记忆内容与向量不匹配（通过一个版本号或哈希校验），则重新计算向量并更新索引。对于本项目，我们暂不实现此逻辑，但你需要知道这是生产环境必须考虑的问题。

3.3 第三步：实现记忆编码器（利用LLM提取信息）

这是最具“智能”的一步。我们需要一个MemoryEncoder类，它调用LLM，将一段对话文本解析成结构化的MemoryItem列表。我们使用Ollama本地运行Llama 3.2模型为例。

# memory_encoder.py import requests import json from typing import List from memory import MemoryItem class MemoryEncoder: def __init__(self, ollama_base_url: str = "http://localhost:11434"): self.ollama_url = ollama_base_url self.model_name = "llama3.2" # 替换成你本地安装的模型 def _create_extraction_prompt(self, conversation_text: str) -> str: """构建引导LLM提取记忆的提示词。提示词工程是关键！""" prompt = f""" 你是一个精准的信息提取助手。请从以下对话中，提取出需要被长期记忆的关键信息。 这些信息通常是关于用户或世界的**事实**、**偏好**、**目标**或**情感状态**。 提取要求： 1. 每条记忆必须是独立的、原子性的陈述。 2. 为每条记忆判断一个类型：`fact`（客观事实）、`preference`（个人偏好）、`task`（任务目标）、`emotion`（情绪感受）。 3. 识别每条记忆涉及的主要实体（如人、物、地点、抽象概念）。 4. 用简洁、完整的句子描述记忆内容。 请以严格的JSON数组格式输出，每个元素是一个记忆对象，包含 `content`, `memory_type`, `entities` 三个字段。 对话内容：

{conversation_text}

示例输出格式： ```json [ {{ "content": "用户居住在北京市海淀区。", "memory_type": "fact", "entities": ["用户", "北京", "海淀区"] }}, {{ "content": "用户喜欢在周末看电影。", "memory_type": "preference", "entities": ["用户", "电影", "周末"] }} ]

现在，请分析上述对话并输出JSON数组： """ return prompt

def encode_conversation(self, conversation_text: str) -> List[MemoryItem]: """ 将对话文本编码为记忆列表。 :param conversation_text: 一段或多轮对话的文本。 :return: 提取出的MemoryItem列表。 """ prompt = self._create_extraction_prompt(conversation_text) try: # 调用Ollama API response = requests.post( f"{self.ollama_url}/api/generate", json={ "model": self.model_name, "prompt": prompt, "stream": False, "options": { "temperature": 0.1, # 低温度，保证输出稳定性 "num_predict": 500 # 限制生成长度 } } ) response.raise_for_status() result = response.json() response_text = result.get('response', '').strip() # 尝试从响应中解析JSON。LLM的输出可能包含markdown代码块或额外文本。 # 简单的处理：找到第一个'['和最后一个']'之间的内容 start_idx = response_text.find('[') end_idx = response_text.rfind(']') + 1 if start_idx != -1 and end_idx != 0: json_str = response_text[start_idx:end_idx] memories_data = json.loads(json_str) else: # 如果解析失败，回退到将整个响应作为一条记忆（不理想，但可容错） print(f"警告：无法从LLM响应中解析JSON。响应内容：{response_text[:200]}...") memories_data = [{"content": response_text, "memory_type": "fact", "entities": []}] except (requests.RequestException, json.JSONDecodeError) as e: print(f"编码过程中发生错误: {e}") # 返回一个空的记忆列表，避免因单次失败导致系统崩溃 memories_data = [] # 将解析出的数据转换为MemoryItem对象 memories = [] for mem_data in memories_data: # 确保数据有必要的字段 content = mem_data.get('content', '').strip() if not content: continue memory_type = mem_data.get('memory_type', 'fact') entities = mem_data.get('entities', []) if isinstance(entities, str): # 处理LLM可能将列表输出为字符串的情况 try: entities = json.loads(entities) except: entities = [e.strip() for e in entities.split(',')] memory = MemoryItem( content=content, memory_type=memory_type, entities=entities, metadata={'source_text': conversation_text[:100]} # 记录来源摘要 ) memories.append(memory) return memories

这个类的核心是`_create_extraction_prompt`方法。提示词的质量决定了提取的准确性和结构化程度。我们通过示例（Few-shot Learning）和严格的格式要求，来引导LLM输出我们想要的JSON。温度（temperature）设置为较低值（0.1），是为了减少输出的随机性，让提取结果更稳定。 > 注意事项：LLM的输出并不总是完美的JSON。在实际操作中，你需要编写更健壮的解析逻辑来处理LLM可能输出的额外文本、markdown代码块标记（```json ... ```）或格式错误。上述代码提供了一个简单的容错方案。在生产环境中，可以考虑使用`json5`库来解析更宽松的JSON，或者加入重试机制。 ### 3.4 第四步：实现记忆检索器（向量化与搜索） 检索器负责将用户的当前查询与记忆库中的内容进行匹配。它需要做两件事：1. 将文本转换为向量；2. 使用向量进行相似度搜索。 ```python # memory_retriever.py import numpy as np from sentence_transformers import SentenceTransformer from typing import List, Tuple from memory import MemoryItem from memory_store import MemoryStore class MemoryRetriever: def __init__(self, model_name: str = 'all-MiniLM-L6-v2'): """ 初始化检索器，加载嵌入模型。 :param model_name: sentence-transformers 模型名称。 """ # 首次加载模型可能需要一点时间 self.embedding_model = SentenceTransformer(model_name) self.embedding_dim = self.embedding_model.get_sentence_embedding_dimension() def get_embedding(self, text: str) -> np.ndarray: """将单条文本转换为向量。""" # 模型返回的是numpy数组 embedding = self.embedding_model.encode(text, convert_to_numpy=True) # 确保是二维数组 [1, dim] return np.expand_dims(embedding, axis=0) def retrieve(self, query: str, memory_store: MemoryStore, top_k: int = 3, memory_type_filter: str = None) -> List[Tuple[MemoryItem, float]]: """ 检索与查询最相关的记忆。 :param query: 用户当前的查询或对话内容。 :param memory_store: 记忆存储对象。 :param top_k: 返回最相关的K条记忆。 :param memory_type_filter: 可选，按记忆类型过滤。 :return: 包含（记忆对象，相似度距离）的列表。 """ # 1. 将查询文本向量化 query_embedding = self.get_embedding(query) # 2. 从存储中搜索 raw_results = memory_store.search_by_vector(query_embedding, k=top_k*2) # 多取一些，用于后续过滤 # 3. 应用类型过滤（如果指定） filtered_results = [] for memory, distance in raw_results: if memory_type_filter is None or memory.memory_type == memory_type_filter: filtered_results.append((memory, distance)) if len(filtered_results) >= top_k: break return filtered_results def update_memory_embedding(self, memory: MemoryItem): """为一条记忆计算并更新其向量表示。""" # 我们通常使用记忆的`content`字段来生成向量。 # 你也可以结合`content`和`entities`来生成更丰富的表示。 text_to_embed = memory.content memory.embedding = self.get_embedding(text_to_embed).squeeze() # 从[1, dim]变为[dim]

这里我们使用了sentence-transformers库，它封装了预训练的Transformer模型，专门用于生成句子级别的向量。all-MiniLM-L6-v2模型在速度和效果上取得了很好的平衡，生成的向量维度是384。get_embedding方法返回的是二维数组，这是因为FAISS的add和search方法通常期望批量输入。

在retrieve方法中，我们先进行向量搜索，拿到初步结果后再进行业务逻辑的过滤（比如按记忆类型）。这是一种“先粗筛，后精滤”的策略，效率较高。

3.5 第五步：实现记忆管理器与整合应用

现在，我们把存储、编码、检索三个模块组合起来，形成一个完整的、可用的记忆系统。我们创建一个MemoryManager类作为对外的统一接口。

# memory_manager.py from memory_store import MemoryStore from memory_encoder import MemoryEncoder from memory_retriever import MemoryRetriever from memory import MemoryItem from typing import List, Tuple, Optional class MemoryManager: def __init__(self, db_path: str = "chatbot_memory.db"): self.store = MemoryStore(db_path) self.encoder = MemoryEncoder() self.retriever = MemoryRetriever() # 初始化时，需要为所有已有的记忆计算向量（如果之前没有） self._initialize_embeddings() def _initialize_embeddings(self): """启动时，为所有尚未有向量的记忆计算向量并更新存储和FAISS索引。""" all_memories = self.store.get_all_memories() for memory in all_memories: if memory.embedding is None: self.retriever.update_memory_embedding(memory) # 注意：直接更新memory对象后，需要重新添加到store，以更新FAISS索引。 # 由于我们之前add_memory时未处理向量，这里需要先“模拟”添加。 # 更干净的做法是修改store.add_memory，使其能处理已存在向量的更新。 # 简化处理：删除旧的（无向量）记录，添加新的（有向量）记录。 # 这里我们调用一个内部方法（需要在MemoryStore中实现delete功能） # 为了简化，我们假设这是一个新系统，或者我们在首次运行时清空旧数据。 # 本示例跳过此复杂逻辑，假设启动时记忆都是新的。 self.store.add_memory(memory) def process_conversation(self, conversation_text: str): """ 处理一段对话，提取记忆并存储。 :param conversation_text: 完整的对话文本。 """ print(f"正在处理对话，提取记忆...") extracted_memories = self.encoder.encode_conversation(conversation_text) print(f"提取到 {len(extracted_memories)} 条潜在记忆。") for memory in extracted_memories: # 在存储前，为记忆生成向量 self.retriever.update_memory_embedding(memory) # 检查是否已存在类似记忆（基于内容简单去重） if not self._is_duplicate(memory): self.store.add_memory(memory) print(f" 已存储记忆: {memory.content[:50]}...") else: print(f" 跳过重复记忆: {memory.content[:50]}...") def _is_duplicate(self, new_memory: MemoryItem, threshold: float = 0.9) -> bool: """ 简单基于内容相似度的去重检查。 :param threshold: 余弦相似度阈值，超过则认为重复。 :return: 是否重复。 """ # 计算新记忆的向量 new_embedding = self.retriever.get_embedding(new_memory.content) # 在现有记忆中搜索最相似的 similar_memories = self.store.search_by_vector(new_embedding, k=1) if similar_memories: _, distance = similar_memories[0] # FAISS L2距离，转换为相似度（近似）。距离越小越相似。 # 这是一个简化处理，更准确的做法是使用余弦相似度索引。 similarity = 1 / (1 + distance) # 将距离映射到(0,1]的相似度 if similarity > threshold: return True return False def query_memories(self, user_query: str, top_k: int = 5, memory_type: Optional[str] = None) -> List[Tuple[MemoryItem, float]]: """ 查询与用户当前问题相关的记忆。 :param user_query: 用户当前输入。 :param top_k: 返回几条最相关的记忆。 :param memory_type: 可选的记忆类型过滤。 :return: 记忆列表和相似度分数。 """ return self.retriever.retrieve(user_query, self.store, top_k=top_k, memory_type_filter=memory_type) def get_conversation_context(self, user_query: str, max_tokens: int = 500) -> str: """ 获取用于构建LLM上下文的记忆文本。 这是记忆系统的最终输出：将检索到的记忆格式化成一段自然语言文本，供LLM参考。 :param user_query: 用户当前输入。 :param max_tokens: 上下文的近似最大token数（粗略估算）。 :return: 格式化后的上下文字符串。 """ relevant_memories = self.query_memories(user_query, top_k=5) if not relevant_memories: return "没有找到相关的历史记忆。" context_parts = ["以下是与当前对话相关的历史信息："] current_length = len(context_parts[0]) for memory, score in relevant_memories: # 简单估算：一个中文字符约等于1个token（粗略） memory_text = f"- {memory.content} (类型：{memory.memory_type}, 相关度：{1/(1+score):.2f})" if current_length + len(memory_text) < max_tokens: context_parts.append(memory_text) current_length += len(memory_text) else: break return "\n".join(context_parts) def list_memories(self, memory_type: Optional[str] = None): """列出所有记忆，用于调试和管理。""" memories = self.store.get_all_memories(memory_type=memory_type) for i, mem in enumerate(memories): print(f"{i+1}. [{mem.memory_type}] {mem.content} (强度: {mem.strength:.2f})")

MemoryManager是我们整个系统的门面。它提供了两个最核心的方法：

1. process_conversation: 消化一段对话，提取并存储记忆。
2. get_conversation_context: 根据当前用户查询，检索相关记忆并格式化成LLM可用的上下文。

_is_duplicate方法实现了一个简单的基于向量相似度的去重逻辑，防止存储大量重复或高度相似的记忆。get_conversation_context方法展示了如何将检索到的记忆“注入”到给LLM的提示词中——通常，我们会把这些记忆作为“系统提示词”或“用户历史”的一部分。

4. 实战演练：让记忆系统跑起来

让我们写一个简单的main.py来演示整个工作流程。

# main.py from memory_manager import MemoryManager import time def main(): # 初始化记忆管理器 print("初始化记忆系统...") manager = MemoryManager() # 模拟几轮对话 conversations = [ "用户：你好，我叫小明。我是一名软件工程师，住在上海。\n助手：你好小明，很高兴认识你！上海是个很棒的城市。", "用户：我喜欢打篮球和阅读科幻小说。对了，我对芒果过敏。\n助手：收到，已记录你的爱好和过敏信息。", "用户：我最近在做一个基于Python的聊天机器人项目。\n助手：听起来很有趣，Python很适合做这个。", "用户：还记得我对什么过敏吗？我晚上想去吃水果沙拉。\n助手：", ] # 处理前三轮对话，提取和存储记忆 for i, conv in enumerate(conversations[:3]): print(f"\n--- 处理第 {i+1} 轮对话 ---") manager.process_conversation(conv) time.sleep(1) # 避免请求过快（如果使用本地LLM） # 列出所有存储的记忆 print("\n=== 当前所有记忆 ===") manager.list_memories() # 模拟第四轮对话：用户查询，系统检索相关记忆 user_query = "我晚上想去吃水果沙拉，需要注意什么？" print(f"\n--- 用户查询：'{user_query}' ---") relevant_memories = manager.query_memories(user_query, top_k=2) print("检索到的相关记忆：") for mem, dist in relevant_memories: print(f" - {mem.content} (距离: {dist:.4f})") # 获取格式化后的上下文，这可以直接拼接到给LLM的提示词中 context = manager.get_conversation_context(user_query) print(f"\n生成的对话上下文：\n{context}") # 模拟LLM的响应生成（这里只是打印） print(f"\n--- 模拟LLM基于上下文的回答 ---") print(f"系统提示词（包含记忆）：\n[系统]: {context}\n[用户]: {user_query}") print(f"[助手]: 根据记录，您对芒果过敏。在制作或选择水果沙拉时，请务必避开芒果或任何含有芒果成分的食材。") if __name__ == "__main__": main()

运行这个脚本，你会看到系统如何从对话中提取出“姓名”、“职业”、“地点”、“爱好”、“过敏史”、“项目”等记忆，并在用户询问过敏相关问题时，精准地检索出“对芒果过敏”这条记忆，并生成包含此记忆的上下文供LLM使用。

5. 进阶优化与深度思考

一个可用的基础系统已经搭建完成，但要使其健壮、高效，还需要考虑很多进阶问题。

5.1 记忆的衰减、巩固与合并

我们引入了strength（强度）字段，但还没使用它。一个真实的记忆系统需要模拟记忆的遗忘曲线。

• 衰减（Forgetting）：可以定期（例如每天）运行一个后台任务，对所有记忆的强度进行衰减，例如new_strength = old_strength * decay_rate（decay_rate略小于1，如0.95）。强度低于某个阈值（如0.1）的记忆可以被自动归档或删除。
• 巩固（Consolidation）：当一条记忆被频繁检索（last_accessed_at更新），其强度应该增加，模拟“温故知新”。在search_by_vector方法中更新访问时间的同时，可以轻微提升其强度。
• 合并（Merging）：当提取到两条高度相似但表述不同的记忆时（例如“用户喜欢咖啡”和“用户最爱喝美式咖啡”），系统应该能合并它们，而不是存储两条。这需要更复杂的语义理解和冲突解决策略，可以在process_conversation的_is_duplicate环节之后，增加一个_try_merge函数。

5.2 检索策略的优化

我们目前使用的是简单的向量相似度检索。但在复杂场景下，可以结合多种策略：

1. 混合检索（Hybrid Search）：结合向量检索（语义相似）和关键词检索（精确匹配）。例如，使用BM25算法进行关键词检索，然后将两者的结果按分数融合（Reciprocal Rank Fusion）。这对于需要精确匹配名称、日期等信息的场景特别有效。
2. 分层检索：先按记忆类型（memory_type）过滤，再进行向量检索。例如，当用户问“我今天心情如何？”，系统可以优先检索emotion类型的记忆。
3. 时间加权：在计算最终相关性分数时，引入时间衰减因子，让近期记忆的权重更高。公式可以是：final_score = semantic_similarity * (recency_factor)。

5.3 处理冲突与错误记忆

LLM提取的记忆可能有误。系统需要机制来处理冲突和修正错误。

• 置信度评分：可以让LLM在提取记忆时，同时输出一个置信度分数。低置信度的记忆可以标记为待确认，或者不立即存入主记忆库，而是放入一个“缓冲区”。
• 用户确认：对于关键信息（如地址、电话号码、重要偏好），系统可以主动询问用户进行确认。“您刚才说您住在上海，对吗？”
• 记忆溯源与修正：每条记忆都应记录其来源（哪段对话）。当发现冲突时（用户后来纠正了信息），系统需要能定位到旧记忆并将其删除或降权，同时存储新记忆。

5.4 性能与扩展性考量

• 向量索引的规模：当记忆数量超过数十万时，简单的IndexFlatL2会变慢。需要升级到更高效的索引，如IndexIVFFlat（基于聚类的倒排索引）或IndexHNSW（基于图的近似搜索），这些FAISS都支持。
• 记忆的分区与分片：如果服务于多用户，每个用户的记忆必须严格隔离。可以在MemoryStore中增加一个user_id字段，并在构建FAISS索引时，为每个用户维护独立的索引或在一个大索引中通过user_id过滤。
• 异步处理：记忆编码（调用LLM）和向量化是比较耗时的操作。在生产环境中，应该将这些操作放入任务队列（如Celery、RQ）异步执行，避免阻塞主对话流程。

5.5 与聊天机器人框架的集成

我们的MemoryManager最终需要与一个聊天机器人框架（如LangChain、LlamaIndex，或自定义的FastAPI服务）集成。

集成模式通常是这样的：

1. 用户输入到来。
2. 调用manager.get_conversation_context(user_input)获取相关记忆。
3. 将记忆上下文、系统指令、最近的对话历史一起，构造成最终的提示词（Prompt）。
4. 将提示词发送给LLM生成回复。
5. 将本轮完整的对话（用户输入+助手回复）传递给manager.process_conversation()，以提取新的记忆。

这个过程可以封装成一个中间件或一个Memory插件，对上层对话逻辑透明。

从零实现一个聊天机器人记忆系统，远不止是调用几个API那么简单。它涉及自然语言理解、信息检索、数据结构设计和认知心理学等多个领域的交叉。通过这个项目，你不仅获得了一套可运行的代码，更重要的是，你深入理解了记忆系统每一个环节的设计权衡和潜在挑战。当你下次使用LangChain的ConversationBufferMemory或VectorStoreRetrieverMemory时，你会清楚地知道它们底层在做什么，以及如何在它们的基础上进行定制和优化。这才是“从零实现”带来的最大价值——知其然，更知其所以然。你可以基于这个基础，继续探索更复杂的记忆结构（如知识图谱）、更智能的遗忘算法，甚至赋予你的机器人“记忆梦境”（离线记忆重组与强化）的能力。