大模型：长期记忆的智能判断逻辑 MD5组件

补齐短板：长期记忆的智能判断逻辑

在面试中，你需要展示你不仅知道“长期记忆和短期记忆的区别”，还要知道“如何设计一个系统，让 Agent 自动决定该记住什么，该遗忘什么”。下面是一个完整的设计方案和代码实现，涵盖了三层过滤机制：语义分类 → 重要性评分 → 摘要压缩。

🧠 设计目标

• 长期记忆：存储对用户有长期价值的信息（如姓名、喜好、职业、重要计划）。
• 短期记忆：存储当前会话的上下文，用于多轮对话。
• 自动决策：每次收到用户消息后，系统自动判断该消息是否值得存入长期记忆。

🏗️ 架构与流程

用户输入 ↓ ① 语义分类（意图识别） ├── 个人信息（姓名、年龄、职业）→ 高优先级 ├── 偏好设置（喜欢什么、讨厌什么）→ 高优先级 ├── 重要事件（计划、承诺）→ 中优先级 └── 闲聊、问候、无关问题 → 低优先级 ↓ ② 重要性评分（规则 + 小模型） - 计算实体数量（人名、地名、日期） - 情感强度 - 是否包含“记住”、“以后”等关键词 - 综合得分 0~1 ↓ ③ 决定是否存储 - 如果得分 > 阈值（如 0.6）→ 存入长期记忆（向量库） - 否则 → 只保留在短期记忆中（会话窗口） ↓ ④ 定期摘要压缩（每周/每月） - 对长期记忆中的相似内容合并 - 对旧的短期记忆做摘要，提炼关键信息存入长期记忆

💻 代码实现（LangChain + 简单规则）

以下是一个完整的MemoryManager类，它实现了上述三层逻辑。

importrefromtypingimportList,Dict,Anyfromlangchain.embeddingsimportOpenAIEmbeddingsfromlangchain.vectorstoresimportChromafromlangchain.llmsimportOpenAIfromlangchain.chainsimportLLMChainfromlangchain.promptsimportPromptTemplatefromlangchain.schemaimportDocumentimportnumpyasnpclassMemoryManager:def__init__(self,llm,embeddings,vectorstore):self.llm=llm self.embeddings=embeddings self.vectorstore=vectorstore self.conversation_history=[]# 短期记忆（会话内）self.long_term_memory=[]# 长期记忆（文档列表）self.importance_threshold=0.6# 定义规则库（用于快速过滤）self.high_priority_keywords=["我叫","我的名字是","我是","喜欢","讨厌","爱好","生日","职业","工作"]self.medium_priority_keywords=["计划","打算","约定","答应","承诺","以后"]def_classify_intent(self,text:str)->str:"""① 语义分类：基于规则快速判断意图"""text_lower=text.lower()ifany(kwintext_lowerforkwin["我叫","我的名字是","我是"]):return"personal_info"ifany(kwintext_lowerforkwin["喜欢","讨厌","爱好","偏好"]):return"preference"ifany(kwintext_lowerforkwin["计划","打算","约定","承诺","以后"]):return"important_event"ifany(kwintext_lowerforkwin["今天天气","你好","谢谢"]):return"chitchat"return"unknown"def_compute_importance_score(self,text:str,intent:str)->float:"""② 计算重要性得分（0~1）"""score=0.0# 基于意图的基础分intent_score_map={"personal_info":0.9,"preference":0.8,"important_event":0.7,"chitchat":0.2,"unknown":0.3}score+=intent_score_map.get(intent,0.3)# 实体数量加分（人名、地名、日期）entity_count=len(re.findall(r'[\u4e00-\u9fa5]{2,4}',text))# 简单中文名字/地点ifentity_count>=2:score+=0.1# 包含“记住”、“以后”等词加分ifany(kwintextforkwin["记住","以后","一定","务必"]):score+=0.1# 长度加分（较长的消息可能包含更多信息）iflen(text)>20:score+=0.05# 情感强度（简单规则：包含“很”、“非常”、“特别”）ifany(kwintextforkwin["很","非常","特别","超级"]):score+=0.05returnmin(score,1.0)def_needs_long_term_storage(self,text:str)->bool:"""③ 综合判断：是否存入长期记忆"""intent=self._classify_intent(text)score=self._compute_importance_score(text,intent)print(f"[评估] 消息: '{text[:30]}...' 意图:{intent}, 得分:{score:.2f}")returnscore>=self.importance_thresholddefadd_message(self,role:str,content:str):"""处理一条新消息（来自用户或AI）"""# 先存入短期记忆（不管是否重要）self.conversation_history.append({"role":role,"content":content})# 只对用户消息做长期存储判断ifrole=="user":ifself._needs_long_term_storage(content):# 存入长期记忆（向量库）doc=Document(page_content=content,metadata={"source":"user","timestamp":"now"})self.vectorstore.add_documents([doc])self.long_term_memory.append(content)print(f"[长期记忆] 已存储:{content}")defget_conversation_context(self,max_turns=5):"""获取短期记忆（最近N轮）"""returnself.conversation_history[-max_turns*2:]defget_long_term_memories(self,query:str,k=3):"""从长期记忆中检索相关内容（RAG）"""docs=self.vectorstore.similarity_search(query,k=k)return[doc.page_contentfordocindocs]defsummarize_long_term_memories(self):"""④ 摘要压缩：定期对长期记忆做合并和压缩"""# 如果长期记忆条目超过50条，触发摘要iflen(self.long_term_memory)>50:prompt=PromptTemplate.from_template("以下是用户的所有长期记忆条目，请将它们合并成一份简洁的摘要（保留关键个人信息和事件）：\n{memories}\n\n摘要：")chain=LLMChain(llm=self.llm,prompt=prompt)summary=chain.run(memories="\n".join(self.long_term_memory))# 清空旧记忆，存入摘要（实际生产可保留部分最新）self.vectorstore.delete_collection()new_doc=Document(page_content=summary,metadata={"type":"summary"})self.vectorstore.add_documents([new_doc])self.long_term_memory=[summary]print(f"[摘要] 长期记忆已压缩为：{summary[:100]}...")

🧪 测试示例

fromlangchain_openaiimportChatOpenAI,OpenAIEmbeddingsfromlangchain.vectorstoresimportChroma# 初始化组件llm=ChatOpenAI(model="gpt-3.5-turbo",temperature=0)embeddings=OpenAIEmbeddings()vectorstore=Chroma(collection_name="long_term_memory",embedding_function=embeddings)# 创建记忆管理器mem_manager=MemoryManager(llm,embeddings,vectorstore)# 模拟对话user_inputs=["你好，我叫张三，是一名软件工程师。","我今天心情不错。","我喜欢打篮球和看电影。","北京今天天气怎么样？","我计划下个月去日本旅游。","谢谢你的帮助。",]formsginuser_inputs:mem_manager.add_message("user",msg)print("---")# 查询长期记忆query="我的职业是什么？"results=mem_manager.get_long_term_memories(query,k=2)print(f"查询结果:{results}")# 获取短期上下文context=mem_manager.get_conversation_context(max_turns=3)print(f"最近对话:{context}")

输出示例（可能因模型和阈值不同而略有差异）：

[评估] 消息: '你好，我叫张三，是一名软件工程师。...' 意图: personal_info, 得分: 0.95 [长期记忆] 已存储: 你好，我叫张三，是一名软件工程师。 --- [评估] 消息: '我今天心情不错。...' 意图: chitchat, 得分: 0.25 --- [评估] 消息: '我喜欢打篮球和看电影。...' 意图: preference, 得分: 0.85 [长期记忆] 已存储: 我喜欢打篮球和看电影。 --- ... 查询结果: ['你好，我叫张三，是一名软件工程师。', '我喜欢打篮球和看电影。']

🗣️ 面试回答话术

“在设计长期记忆系统时，我采用了三层判断逻辑：

1. 语义分类：我预先定义了一组规则（或小模型），将用户输入分为个人信息、偏好、重要事件、闲聊等类别。个人信息（如姓名、职业）和偏好会被标记为高优先级。
2. 重要性评分：对于每一条用户消息，我会计算一个综合得分，考量因素包括：是否包含姓名/地点等实体、是否包含承诺性词汇（如‘以后’、‘一定’）、消息长度、情感强度等。得分超过阈值（如0.6）才会被存入长期记忆。
3. 摘要压缩：为了防止长期记忆无限膨胀，我会定期对存储的内容进行摘要合并。当条目过多时，使用LLM将多条记忆压缩成一段概述，既保留关键信息，又节省存储空间。

这套方案结合了规则引擎的快速性和LLM的语义理解能力，在实际项目中取得了不错的效果。当然，如果资源允许，我会用一个小型的分类器模型（如BERT）替换规则，进一步提升判断精度。”

MD5 组件：哈希世界的“老牌主力”

1. MD5 是什么？

MD5（Message Digest Algorithm 5，消息摘要算法第五版）是一种广泛使用的密码学哈希函数，由 Ronald Rivest 于 1991 年设计。它接收任意长度的输入，输出一个128 位（16 字节）的固定长度哈希值，通常表示为 32 位的十六进制字符串（例如d41d8cd98f00b204e9800998ecf8427e）。

在 Python 中，MD5 通过内置的hashlib模块提供：

importhashlib md5=hashlib.md5()md5.update(b"hello world")print(md5.hexdigest())# 输出: 5eb63bbbe01eeed093cb22bb8f5acdc3

2. MD5 的核心特性

3. MD5 在 LLM/RAG 项目中的典型应用（结合你的经历）

在你之前的 Excel 比较智能体代码中，hashlib.md5被用来生成文件的版本指纹，以判断两次上传的内容是否发生变化：

content_hash=hashlib.md5(json.dumps(all_sheets).encode()).hexdigest()

为什么要这样做？

• 避免重复处理相同内容的文件（去重）。
• 快速判断文件是否被修改，无需逐行比较整个文件。
• 作为缓存键（例如，如果哈希没变，直接从缓存返回结果）。

除此之外，MD5 在 RAG 系统中还有这些常见用途：

4. MD5 的安全隐患（面试重点！）

重要结论：MD5 已不再安全，不应在安全敏感场景中使用（如密码存储、数字证书）。

• 碰撞攻击：2004 年研究人员已发现 MD5 的碰撞方法，两个不同文件可生成相同哈希。2010 年进一步改进，普通 PC 也能快速制造碰撞。
• 破解速度：如今 GPU 可在数秒内破解短密码的 MD5 哈希。

替代方案：

5. 面试高频追问与回答

Q1：MD5 和 SHA-256 有什么区别？

Q2：既然 MD5 不安全，为什么还在用？
A：因为它的速度优势在非安全场景中依然显著，例如去重、缓存、版本检测，这些场景不需要防御恶意碰撞。但如果是涉及用户隐私或防篡改的场景，必须弃用。

Q3：在 LangChain/RAG 系统中，你如何用 MD5 优化性能？
A：我会对文档块（Chunk）生成 MD5 哈希作为文档 ID，存入向量库时检查 ID 是否存在，实现增量更新。另外，对高频 Query 做 MD5 缓存键，减少 LLM 调用成本（但需配合时效性设计，如 TTL）。

Q4：如何应对 MD5 的碰撞问题？
A：在安全场景，我使用 SHA-256 或 SHA-3。在非安全场景（如去重），碰撞概率极低（2^128），且即使发生也无大碍，因此可以接受。

6. 实战代码示例（结合 LangChain）

importhashlibfromlangchain.schemaimportDocumentdefget_document_id(doc:Document)->str:"""使用 MD5 生成文档唯一 ID（基于内容+元数据）"""content=doc.page_content meta_str=str(sorted(doc.metadata.items()))combined=content+meta_strreturnhashlib.md5(combined.encode('utf-8')).hexdigest()# 使用示例doc=Document(page_content="hello world",metadata={"source":"test.txt"})doc_id=get_document_id(doc)print(f"文档 ID:{doc_id}")

7. 总结话术（面试收尾）

“MD5 是一个经典的哈希算法，虽然因为碰撞问题不再适合安全场景，但它在去重、缓存和版本控制等性能敏感场景中依然有巨大价值。在实际项目中，我会区分用途：用 MD5 做快速指纹生成，用 SHA-256 做数据完整性校验，用 bcrypt 做密码存储。理解每个算法的强弱，是工程选型的基本功。”