知识库记忆把对话中的重要信息存入长期记忆短期记忆管的是当前对话长期记忆管的是跨对话的持久信息。上一篇讲的对话记忆组件都是短期记忆。对话结束就没了。下次用户来Agent又是失忆状态。这一篇我们讲怎么构建长期记忆。把对话中的重要信息提取出来存到知识库里。下次对话的时候Agent能回忆起之前交互的关键内容。什么信息值得记住不是所有对话内容都值得存。把所有对话都存进长期记忆信息会越来越多检索质量会下降。值得记住的信息有这几类。用户身份信息。姓名、职业、技能、所在地。这些基本不变化的信息记住一次就能长期用。用户偏好。喜欢什么编程语言、偏好什么回答风格、用什么工具。这些偏好能帮助Agent给出更贴合用户需求的回答。重要决策和结论。用户在对话中做出的决策、达成的结论。比如用户决定用LangChain做Agent开发、“用户的项目用PostgreSQL做数据库”。纠正过的错误。用户纠正过Agent的错误Agent应该记住下次不犯同样的错。任务进度。跨对话的长任务记录进行到哪一步了下次继续。不值得记住的信息包括闲聊内容、临时性的问题、已经过时的信息。信息提取从对话中提取重要信息通常用大模型来做。对话结束后把整个对话发给大模型让它提取出值得记住的信息。fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplate llmChatOpenAI(modelgpt-3.5-turbo,temperature0)extract_promptChatPromptTemplate.from_template(请从以下对话中提取值得长期记住的重要信息。 提取规则 1. 只提取重要的事实、偏好、决策不要提取闲聊内容 2. 每条信息用一句完整的话描述 3. 如果没有值得记住的信息返回无 对话内容 {conversation} 请输出提取的信息每条一行。如果有多条用编号列出。)extract_chainextract_prompt|llm# 示例对话conversation 用户我叫张三在阿里巴巴做算法工程师。 Agent你好张三请问有什么可以帮你的 用户我最近在用LangChain做Agent开发遇到了记忆管理的问题。 Agent你可以试试ConversationSummaryBufferMemory... 用户谢谢我用GPT-4做模型Token限制不是大问题。 Agent那你可以考虑用BufferMemory配合向量数据库做长期记忆。 用户好的我试试。对了我偏好Python不太喜欢Java。 resultextract_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.embeddingsOpenAIEmbeddings()self.storeChroma(persist_directorydb_path,embedding_functionself.embeddings,)defsave(self,information,sourceconversation):保存一条记忆metadata{source:source,timestamp:datetime.now().isoformat(),}self.store.add_texts(texts[information],metadatas[metadata],)self.store.persist()defsave_batch(self,informations,sourceconversation):批量保存记忆texts[]metadatas[]forinfoininformations:texts.append(info)metadatas.append({source:source,timestamp:datetime.now().isoformat(),})self.store.add_texts(textstexts,metadatasmetadatas)self.store.persist()defrecall(self,query,k3):根据查询回忆相关信息resultsself.store.similarity_search(query,kk)return[doc.page_contentfordocinresults]defrecall_with_details(self,query,k3):回忆并返回详细信息resultsself.store.similarity_search(query,kk)return[{content:doc.page_content,metadata:doc.metadata,}fordocinresults]# 使用memoryKnowledgeBaseMemory()# 保存提取的信息informations[用户叫张三在阿里巴巴做算法工程师,用户在用LangChain做Agent开发,用户使用GPT-4作为模型,用户偏好Python不太喜欢Java,]memory.save_batch(informations)# 下次对话时回忆resultsmemory.recall(用户的职业是什么)print(results)# 返回用户叫张三在阿里巴巴做算法工程师把记忆注入对话回忆出来的信息要注入到当前对话的Prompt里Agent才能用上。fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.messagesimportSystemMessageclassMemoryAwareConversation:带长期记忆的对话def__init__(self,llm,long_term_memory):self.llmllm self.memorylong_term_memory self.short_term[]# 短期对话历史self.system_prompt你是一个智能助手。 你可以参考以下长期记忆信息来更好地服务用户。如果长期记忆中没有相关信息就正常回答。 长期记忆 {long_term_memory} 请根据长期记忆和当前对话来回答用户的问题。defchat(self,user_input):# 1. 从长期记忆中检索相关信息memoriesself.memory.recall(user_input,k3)memory_text\n.join(f-{m}forminmemories)ifmemorieselse暂无相关记忆# 2. 构建Promptsystem_msgself.system_prompt.format(long_term_memorymemory_text)messages[SystemMessage(contentsystem_msg)]messages.extend(self.short_term)messages.append(HumanMessage(contentuser_input))# 3. 调用大模型responseself.llm.invoke(messages)# 4. 更新短期记忆self.short_term.append(HumanMessage(contentuser_input))self.short_term.append(response)# 控制短期记忆长度iflen(self.short_term)10:self.short_termself.short_term[-10:]returnresponse.contentdefend_conversation(self):对话结束时提取重要信息存入长期记忆# 把短期记忆转成文本conv_text\n.join(f{用户ifisinstance(m,HumanMessage)else助手}:{m.content}forminself.short_term)# 提取重要信息extract_promptf请从以下对话中提取值得长期记住的重要信息。 只提取重要的事实、偏好、决策。每条一行。如果没有返回无。 对话{conv_text}resultself.llm.invoke(extract_prompt)ifresult.content.strip()!无:# 解析并保存lines[l.strip()forlinresult.content.split(\n)ifl.strip()]# 去掉编号前缀infos[]forlineinlines:# 去掉 1. 这样的编号cleanline.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_threshold0.85):保存记忆自动去重# 检查是否已有相似记忆existingself.store.similarity_search_with_score(information,k1)ifexisting:doc,scoreexisting[0]# Chroma返回的是距离越小越相似# 转换成相似度近似similarity1-score/2ifsimilaritysimilarity_threshold:# 已有相似记忆跳过print(f跳过重复记忆{information[:30]}...)return# 没有相似的保存新记忆self.save(information)更新策略也可以更智能。比如用户之前说用Python后来改成改用Go了。新记忆应该覆盖旧记忆。可以让大模型判断新旧记忆是否冲突冲突的话以新的为准。记忆的时效性有些信息会过时。用户半年前说在学Python现在已经学完了。过时的记忆如果还检索出来可能会误导Agent。解决方法是给每条记忆加时间戳检索的时候考虑时间因素。近期的记忆权重高远期的权重低。defrecall_with_time_weight(self,query,k5,days_decay90):带时间衰减的回忆resultsself.store.similarity_search(query,kk)weighted[]fordocinresults:# 计算时间衰减timestampdoc.metadata.get(timestamp,)iftimestamp:fromdatetimeimportdatetime createddatetime.fromisoformat(timestamp)age_days(datetime.now()-created).days decaymax(0.5,1-age_days/(days_decay*2))else:decay1.0weighted.append((doc,decay))# 按相似度乘以时间衰减排序weighted.sort(keylambdax:x[1],reverseTrue)return[doc.page_contentfordoc,_inweighted[:k]]下一篇讲用户画像记忆。怎么从交互中构建用户画像让Agent越来越懂用户。 知识库记忆把对话中的重要信息存入长期记忆短期记忆管的是当前对话长期记忆管的是跨对话的持久信息。上一篇讲的对话记忆组件都是短期记忆。对话结束就没了。下次用户来Agent又是失忆状态。这一篇我们讲怎么构建长期记忆。把对话中的重要信息提取出来存到知识库里。下次对话的时候Agent能回忆起之前交互的关键内容。什么信息值得记住不是所有对话内容都值得存。把所有对话都存进长期记忆信息会越来越多检索质量会下降。值得记住的信息有这几类。用户身份信息。姓名、职业、技能、所在地。这些基本不变化的信息记住一次就能长期用。用户偏好。喜欢什么编程语言、偏好什么回答风格、用什么工具。这些偏好能帮助Agent给出更贴合用户需求的回答。重要决策和结论。用户在对话中做出的决策、达成的结论。比如用户决定用LangChain做Agent开发、“用户的项目用PostgreSQL做数据库”。纠正过的错误。用户纠正过Agent的错误Agent应该记住下次不犯同样的错。任务进度。跨对话的长任务记录进行到哪一步了下次继续。不值得记住的信息包括闲聊内容、临时性的问题、已经过时的信息。信息提取从对话中提取重要信息通常用大模型来做。对话结束后把整个对话发给大模型让它提取出值得记住的信息。fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplate llmChatOpenAI(modelgpt-3.5-turbo,temperature0)extract_promptChatPromptTemplate.from_template(请从以下对话中提取值得长期记住的重要信息。 提取规则 1. 只提取重要的事实、偏好、决策不要提取闲聊内容 2. 每条信息用一句完整的话描述 3. 如果没有值得记住的信息返回无 对话内容 {conversation} 请输出提取的信息每条一行。如果有多条用编号列出。)extract_chainextract_prompt|llm# 示例对话conversation 用户我叫张三在阿里巴巴做算法工程师。 Agent你好张三请问有什么可以帮你的 用户我最近在用LangChain做Agent开发遇到了记忆管理的问题。 Agent你可以试试ConversationSummaryBufferMemory... 用户谢谢我用GPT-4做模型Token限制不是大问题。 Agent那你可以考虑用BufferMemory配合向量数据库做长期记忆。 用户好的我试试。对了我偏好Python不太喜欢Java。 resultextract_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.embeddingsOpenAIEmbeddings()self.storeChroma(persist_directorydb_path,embedding_functionself.embeddings,)defsave(self,information,sourceconversation):保存一条记忆metadata{source:source,timestamp:datetime.now().isoformat(),}self.store.add_texts(texts[information],metadatas[metadata],)self.store.persist()defsave_batch(self,informations,sourceconversation):批量保存记忆texts[]metadatas[]forinfoininformations:texts.append(info)metadatas.append({source:source,timestamp:datetime.now().isoformat(),})self.store.add_texts(textstexts,metadatasmetadatas)self.store.persist()defrecall(self,query,k3):根据查询回忆相关信息resultsself.store.similarity_search(query,kk)return[doc.page_contentfordocinresults]defrecall_with_details(self,query,k3):回忆并返回详细信息resultsself.store.similarity_search(query,kk)return[{content:doc.page_content,metadata:doc.metadata,}fordocinresults]# 使用memoryKnowledgeBaseMemory()# 保存提取的信息informations[用户叫张三在阿里巴巴做算法工程师,用户在用LangChain做Agent开发,用户使用GPT-4作为模型,用户偏好Python不太喜欢Java,]memory.save_batch(informations)# 下次对话时回忆resultsmemory.recall(用户的职业是什么)print(results)# 返回用户叫张三在阿里巴巴做算法工程师把记忆注入对话回忆出来的信息要注入到当前对话的Prompt里Agent才能用上。fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.messagesimportSystemMessageclassMemoryAwareConversation:带长期记忆的对话def__init__(self,llm,long_term_memory):self.llmllm self.memorylong_term_memory self.short_term[]# 短期对话历史self.system_prompt你是一个智能助手。 你可以参考以下长期记忆信息来更好地服务用户。如果长期记忆中没有相关信息就正常回答。 长期记忆 {long_term_memory} 请根据长期记忆和当前对话来回答用户的问题。defchat(self,user_input):# 1. 从长期记忆中检索相关信息memoriesself.memory.recall(user_input,k3)memory_text\n.join(f-{m}forminmemories)ifmemorieselse暂无相关记忆# 2. 构建Promptsystem_msgself.system_prompt.format(long_term_memorymemory_text)messages[SystemMessage(contentsystem_msg)]messages.extend(self.short_term)messages.append(HumanMessage(contentuser_input))# 3. 调用大模型responseself.llm.invoke(messages)# 4. 更新短期记忆self.short_term.append(HumanMessage(contentuser_input))self.short_term.append(response)# 控制短期记忆长度iflen(self.short_term)10:self.short_termself.short_term[-10:]returnresponse.contentdefend_conversation(self):对话结束时提取重要信息存入长期记忆# 把短期记忆转成文本conv_text\n.join(f{用户ifisinstance(m,HumanMessage)else助手}:{m.content}forminself.short_term)# 提取重要信息extract_promptf请从以下对话中提取值得长期记住的重要信息。 只提取重要的事实、偏好、决策。每条一行。如果没有返回无。 对话{conv_text}resultself.llm.invoke(extract_prompt)ifresult.content.strip()!无:# 解析并保存lines[l.strip()forlinresult.content.split(\n)ifl.strip()]# 去掉编号前缀infos[]forlineinlines:# 去掉 1. 这样的编号cleanline.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_threshold0.85):保存记忆自动去重# 检查是否已有相似记忆existingself.store.similarity_search_with_score(information,k1)ifexisting:doc,scoreexisting[0]# Chroma返回的是距离越小越相似# 转换成相似度近似similarity1-score/2ifsimilaritysimilarity_threshold:# 已有相似记忆跳过print(f跳过重复记忆{information[:30]}...)return# 没有相似的保存新记忆self.save(information)更新策略也可以更智能。比如用户之前说用Python后来改成改用Go了。新记忆应该覆盖旧记忆。可以让大模型判断新旧记忆是否冲突冲突的话以新的为准。记忆的时效性有些信息会过时。用户半年前说在学Python现在已经学完了。过时的记忆如果还检索出来可能会误导Agent。解决方法是给每条记忆加时间戳检索的时候考虑时间因素。近期的记忆权重高远期的权重低。defrecall_with_time_weight(self,query,k5,days_decay90):带时间衰减的回忆resultsself.store.similarity_search(query,kk)weighted[]fordocinresults:# 计算时间衰减timestampdoc.metadata.get(timestamp,)iftimestamp:fromdatetimeimportdatetime createddatetime.fromisoformat(timestamp)age_days(datetime.now()-created).days decaymax(0.5,1-age_days/(days_decay*2))else:decay1.0weighted.append((doc,decay))# 按相似度乘以时间衰减排序weighted.sort(keylambdax:x[1],reverseTrue)return[doc.page_contentfordoc,_inweighted[:k]]下一篇讲用户画像记忆。怎么从交互中构建用户画像让Agent越来越懂用户。 
