LangChain开发-全量记忆方案：完整保存与检索对话历史

一、ConversationBufferMemory详解

1.1 工作原理

全量记忆保存所有对话历史：

第1轮: 用户:你好 → AI:你好！ 第2轮: 用户:我是小明 → AI:你好小明！ 第3轮: 用户:我叫什么？ → AI:你叫小明 完整历史被传递给LLM

1.2 基本使用

fromlangchain.memoryimportConversationBufferMemoryfromlangchain_openaiimportChatOpenAIfromlangchain.chainsimportConversationChain# 创建记忆组件memory=ConversationBufferMemory()# 创建对话链model=ChatOpenAI(model="gpt-4o-mini")chain=ConversationChain(llm=model,memory=memory,verbose=True)# 多轮对话print(chain.invoke({"input":"我叫小明，今年25岁"})['response'])print(chain.invoke({"input":"我喜欢打篮球"})['response'])print(chain.invoke({"input":"我叫什么？今年多大？喜欢什么？"})['response'])

二、返回格式控制

2.1 文本格式

memory=ConversationBufferMemory(return_messages=False)memory.save_context({"input":"你好"},{"output":"你好！"})result=memory.load_memory_variables({})print(result['history'])# Human: 你好# AI: 你好！

2.2 消息对象格式

memory=ConversationBufferMemory(return_messages=True)memory.save_context({"input":"你好"},{"output":"你好！"})result=memory.load_memory_variables({})print(result['history'])# [HumanMessage(content='你好'), AIMessage(content='你好！')]

2.3 应用场景

三、自定义记忆键名

3.1 修改memory_key

fromlangchain.memoryimportConversationBufferMemoryfromlangchain_core.promptsimportPromptTemplate# 自定义键名memory=ConversationBufferMemory(memory_key="chat_history")# 提示模板中使用对应键名template=""" 当前对话历史： {chat_history} 用户：{question} AI： """prompt=PromptTemplate.from_template(template)

3.2 完整示例

fromlangchain_openaiimportChatOpenAIfromlangchain.memoryimportConversationBufferMemoryfromlangchain.chainsimportConversationChainfromlangchain_core.promptsimportChatPromptTemplate,MessagesPlaceholder# 自定义提示词prompt=ChatPromptTemplate.from_messages([("system","你是一个友好的AI助手"),MessagesPlaceholder(variable_name="chat_history"),("human","{input}")])# 创建记忆组件（键名与提示词对应）memory=ConversationBufferMemory(memory_key="chat_history",return_messages=True)# 创建链model=ChatOpenAI(model="gpt-4o-mini")chain=ConversationChain(llm=model,memory=memory,prompt=prompt,verbose=True)# 对话chain.invoke({"input":"你好"})

四、消息历史操作

4.1 手动添加消息

fromlangchain.memoryimportConversationBufferMemoryfromlangchain_core.messagesimportHumanMessage,AIMessage memory=ConversationBufferMemory()# 方式一：使用save_contextmemory.save_context({"input":"用户消息"},{"output":"AI回复"})# 方式二：直接添加消息对象memory.chat_memory.add_message(HumanMessage(content="另一条用户消息"))memory.chat_memory.add_message(AIMessage(content="另一条AI回复"))print(memory.load_memory_variables({}))

4.2 查看历史

# 获取消息列表messages=memory.chat_memory.messagesformsginmessages:print(f"{msg.type}:{msg.content}")

4.3 清空历史

memory.clear()print(memory.load_memory_variables({}))# {'history': ''}

五、持久化存储

5.1 保存到文件

importjsondefsave_memory(memory,filepath:str):"""保存记忆到文件"""messages=memory.chat_memory.messages data=[{"type":msg.type,"content":msg.content}formsginmessages]withopen(filepath,'w',encoding='utf-8')asf:json.dump(data,f,ensure_ascii=False,indent=2)defload_memory(filepath:str)->ConversationBufferMemory:"""从文件加载记忆"""memory=ConversationBufferMemory(return_messages=True)withopen(filepath,'r',encoding='utf-8')asf:data=json.load(f)foritemindata:ifitem['type']=='human':memory.chat_memory.add_user_message(item['content'])else:memory.chat_memory.add_ai_message(item['content'])returnmemory# 使用memory=ConversationBufferMemory()memory.save_context({"input":"你好"},{"output":"你好！"})# 保存save_memory(memory,"chat_history.json")# 加载loaded_memory=load_memory("chat_history.json")

5.2 使用Redis存储

fromlangchain_community.chat_message_historiesimportRedisChatMessageHistory# 连接Redishistory=RedisChatMessageHistory(session_id="user_123",url="redis://localhost:6379")# 添加消息history.add_user_message("你好")history.add_ai_message("你好！")# 获取历史print(history.messages)

5.3 使用SQLite存储

fromlangchain_community.chat_message_historiesimportSQLChatMessageHistory# 连接SQLitehistory=SQLChatMessageHistory(session_id="user_123",connection="sqlite:///chat_history.db")# 使用方式相同history.add_user_message("你好")history.add_ai_message("你好！")

六、限制与注意事项

6.1 Token限制

全量记忆会累积所有历史，可能超出上下文窗口限制：

# 检查历史消息数量defcheck_history_length(memory,max_messages=20):messages=memory.chat_memory.messagesiflen(messages)>max_messages:print(f"警告：历史消息数量{len(messages)}超过限制{max_messages}")

6.2 成本问题

历史越长，每次调用消耗的Token越多：

每次调用Token = 历史Token + 新输入Token + 输出Token

6.3 适用场景

✅ 适合： - 对话轮次少（< 10轮） - 需要完整上下文 - 短期交互场景 ❌ 不适合： - 长期对话 - 高频交互 - 成本敏感场景

七、实战：带记忆的聊天机器人

fromlangchain_openaiimportChatOpenAIfromlangchain.memoryimportConversationBufferMemoryfromlangchain.chainsimportConversationChainclassChatBot:"""带记忆的聊天机器人"""def__init__(self,name:str="小智"):self.name=name self.memory=ConversationBufferMemory()self.model=ChatOpenAI(model="gpt-4o-mini")self.chain=ConversationChain(llm=self.model,memory=self.memory,verbose=False)defchat(self,user_input:str)->str:response=self.chain.invoke({"input":user_input})returnresponse['response']defshow_history(self):print("=== 对话历史 ===")formsginself.memory.chat_memory.messages:role="用户"ifmsg.type=="human"else"AI"print(f"{role}:{msg.content}")defclear_history(self):self.memory.clear()print("历史已清空")# 使用bot=ChatBot()print(bot.chat("你好，我叫小明"))print(bot.chat("我喜欢Python"))print(bot.chat("我叫什么？喜欢什么？"))bot.show_history()

八、小结

适用场景：对话轮次少、需要完整上下文的短期交互场景。

下一篇文章：18 记忆优化策略：滑动窗口与摘要压缩的应用场景