知识库记忆，把对话中的重要信息存入长期记忆

短期记忆管的是当前对话，长期记忆管的是跨对话的持久信息。

上一篇讲的对话记忆组件，都是短期记忆。对话结束就没了。下次用户来，Agent又是"失忆"状态。

这一篇我们讲怎么构建长期记忆。把对话中的重要信息提取出来，存到知识库里。下次对话的时候，Agent能回忆起之前交互的关键内容。

什么信息值得记住

不是所有对话内容都值得存。把所有对话都存进长期记忆，信息会越来越多，检索质量会下降。

值得记住的信息有这几类。

用户身份信息。姓名、职业、技能、所在地。这些基本不变化的信息，记住一次就能长期用。

用户偏好。喜欢什么编程语言、偏好什么回答风格、用什么工具。这些偏好能帮助Agent给出更贴合用户需求的回答。

重要决策和结论。用户在对话中做出的决策、达成的结论。比如"用户决定用LangChain做Agent开发"、“用户的项目用PostgreSQL做数据库”。

纠正过的错误。用户纠正过Agent的错误，Agent应该记住，下次不犯同样的错。

任务进度。跨对话的长任务，记录进行到哪一步了，下次继续。

不值得记住的信息包括闲聊内容、临时性的问题、已经过时的信息。

信息提取

从对话中提取重要信息，通常用大模型来做。

对话结束后，把整个对话发给大模型，让它提取出值得记住的信息。

fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplate llm=ChatOpenAI(model="gpt-3.5-turbo",temperature=0)extract_prompt=ChatPromptTemplate.from_template("""请从以下对话中提取值得长期记住的重要信息。 提取规则： 1. 只提取重要的事实、偏好、决策，不要提取闲聊内容 2. 每条信息用一句完整的话描述 3. 如果没有值得记住的信息，返回"无" 对话内容： {conversation} 请输出提取的信息，每条一行。如果有多条，用编号列出。""")extract_chain=extract_prompt|llm# 示例对话conversation=""" 用户：我叫张三，在阿里巴巴做算法工程师。 Agent：你好张三！请问有什么可以帮你的？ 用户：我最近在用LangChain做Agent开发，遇到了记忆管理的问题。 Agent：你可以试试ConversationSummaryBufferMemory... 用户：谢谢，我用GPT-4做模型，Token限制不是大问题。 Agent：那你可以考虑用BufferMemory配合向量数据库做长期记忆。 用户：好的，我试试。对了，我偏好Python，不太喜欢Java。 """result=extract_chain.invoke({"conversation":conversation})print(result.content)# 输出类似：# 1. 用户叫张三，在阿里巴巴做算法工程师# 2. 用户在用LangChain做Agent开发# 3. 用户使用GPT-4作为模型# 4. 用户偏好Python，不太喜欢Java# 5. 用户在探索Agent的记忆管理方案

提取出来的信息，就可以存入长期记忆了。

存入向量数据库

提取出来的信息，存到向量数据库里。下次对话时用语义检索来回忆。

fromlangchain_openaiimportOpenAIEmbeddingsfromlangchain_community.vectorstoresimportChromafromdatetimeimportdatetimeclassKnowledgeBaseMemory:"""基于向量数据库的长期记忆"""def__init__(self,db_path="./agent_long_memory"):self.embeddings=OpenAIEmbeddings()self.store=Chroma(persist_directory=db_path,embedding_function=self.embeddings,)defsave(self,information,source="conversation"):"""保存一条记忆"""metadata={"source":source,"timestamp":datetime.now().isoformat(),}self.store.add_texts(texts=[information],metadatas=[metadata],)self.store.persist()defsave_batch(self,informations,source="conversation"):"""批量保存记忆"""texts=[]metadatas=[]forinfoininformations:texts.append(info)metadatas.append({"source":source,"timestamp":datetime.now().isoformat(),})self.store.add_texts(texts=texts,metadatas=metadatas)self.store.persist()defrecall(self,query,k=3):"""根据查询回忆相关信息"""results=self.store.similarity_search(query,k=k)return[doc.page_contentfordocinresults]defrecall_with_details(self,query,k=3):"""回忆并返回详细信息"""results=self.store.similarity_search(query,k=k)return[{"content":doc.page_content,"metadata":doc.metadata,}fordocinresults]# 使用memory=KnowledgeBaseMemory()# 保存提取的信息informations=["用户叫张三，在阿里巴巴做算法工程师","用户在用LangChain做Agent开发","用户使用GPT-4作为模型","用户偏好Python，不太喜欢Java",]memory.save_batch(informations)# 下次对话时回忆results=memory.recall("用户的职业是什么")print(results)# 返回："用户叫张三，在阿里巴巴做算法工程师"

把记忆注入对话

回忆出来的信息，要注入到当前对话的Prompt里，Agent才能用上。

fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.messagesimportSystemMessageclassMemoryAwareConversation:"""带长期记忆的对话"""def__init__(self,llm,long_term_memory):self.llm=llm self.memory=long_term_memory self.short_term=[]# 短期对话历史self.system_prompt="""你是一个智能助手。 你可以参考以下长期记忆信息来更好地服务用户。如果长期记忆中没有相关信息，就正常回答。 长期记忆： {long_term_memory} 请根据长期记忆和当前对话来回答用户的问题。"""defchat(self,user_input):# 1. 从长期记忆中检索相关信息memories=self.memory.recall(user_input,k=3)memory_text="\n".join(f"-{m}"forminmemories)ifmemorieselse"暂无相关记忆"# 2. 构建Promptsystem_msg=self.system_prompt.format(long_term_memory=memory_text)messages=[SystemMessage(content=system_msg)]messages.extend(self.short_term)messages.append(HumanMessage(content=user_input))# 3. 调用大模型response=self.llm.invoke(messages)# 4. 更新短期记忆self.short_term.append(HumanMessage(content=user_input))self.short_term.append(response)# 控制短期记忆长度iflen(self.short_term)>10:self.short_term=self.short_term[-10:]returnresponse.contentdefend_conversation(self):"""对话结束时，提取重要信息存入长期记忆"""# 把短期记忆转成文本conv_text="\n".join(f"{'用户'ifisinstance(m,HumanMessage)else'助手'}:{m.content}"forminself.short_term)# 提取重要信息extract_prompt=f"""请从以下对话中提取值得长期记住的重要信息。 只提取重要的事实、偏好、决策。每条一行。如果没有，返回"无"。 对话：{conv_text}"""result=self.llm.invoke(extract_prompt)ifresult.content.strip()!="无":# 解析并保存lines=[l.strip()forlinresult.content.split("\n")ifl.strip()]# 去掉编号前缀infos=[]forlineinlines:# 去掉 "1. " 这样的编号clean=line.split(". ",1)[-1]ifline[0].isdigit()elseline infos.append(clean)ifinfos:self.memory.save_batch(infos)print(f"已保存{len(infos)}条记忆")# 清空短期记忆self.short_term=[]

记忆去重和更新

长期记忆会遇到一个问题。同一条信息可能被重复保存。用户每次对话都提到自己的名字，长期记忆里就存了十条"用户叫张三"。

需要做去重和更新。

在保存新记忆之前，先检索一下，看看有没有相似的记忆。有的话，更新而不是新增。

defsave_with_dedup(self,information,similarity_threshold=0.85):"""保存记忆，自动去重"""# 检查是否已有相似记忆existing=self.store.similarity_search_with_score(information,k=1)ifexisting:doc,score=existing[0]# Chroma返回的是距离，越小越相似# 转换成相似度（近似）similarity=1-score/2ifsimilarity>similarity_threshold:# 已有相似记忆，跳过print(f"跳过重复记忆：{information[:30]}...")return# 没有相似的，保存新记忆self.save(information)

更新策略也可以更智能。比如用户之前说"用Python"，后来改成"改用Go了"。新记忆应该覆盖旧记忆。可以让大模型判断新旧记忆是否冲突，冲突的话以新的为准。

记忆的时效性

有些信息会过时。用户半年前说"在学Python"，现在已经学完了。过时的记忆如果还检索出来，可能会误导Agent。

解决方法是给每条记忆加时间戳，检索的时候考虑时间因素。近期的记忆权重高，远期的权重低。

defrecall_with_time_weight(self,query,k=5,days_decay=90):"""带时间衰减的回忆"""results=self.store.similarity_search(query,k=k)weighted=[]fordocinresults:# 计算时间衰减timestamp=doc.metadata.get("timestamp","")iftimestamp:fromdatetimeimportdatetime created=datetime.fromisoformat(timestamp)age_days=(datetime.now()-created).days decay=max(0.5,1-age_days/(days_decay*2))else:decay=1.0weighted.append((doc,decay))# 按相似度乘以时间衰减排序weighted.sort(key=lambdax:x[1],reverse=True)return[doc.page_contentfordoc,_inweighted[:k]]

下一篇讲用户画像记忆。怎么从交互中构建用户画像，让Agent越来越懂用户。