【LangChain】—— History模块使用-洪萨配资

【LangChain】—— 使用History实现上下文记忆功能

在对话式AI应用开发中，会话记忆是实现上下文连贯交互的核心能力。LangChain提供了完善的History模块，用于管理会话历史记录，支持临时存储与长期持久化存储两种场景，满足不同业务对会话记忆的需求。本文将围绕LangChain的会话记忆机制，拆解临时与长期会话记忆的实现方式及核心类用法。

History：管理会话历史记录（记忆）

LangChain的History模块通过封装各类会话存储逻辑，为链（Chain）或智能体（Agent）提供会话历史的增、查、清操作接口，使开发者无需关注底层存储细节，即可快速为应用添加上下文记忆能力。根据存储时效，可分为临时会话记忆与长期会话记忆两大类。

一、临时会话记忆

临时会话记忆适用于单次会话场景，会话结束后历史记录自动销毁，不进行持久化存储。LangChain通过RunnableWithMessageHistory类与InMemoryChatMessageHistory类配合实现该功能，前者负责为链注入记忆能力，后者提供内存级存储支持。

1. RunnableWithMessageHistory类

该类是LangChain中为原有链添加会话历史功能的核心封装类，本质是对基础链的装饰器，能够在链的执行过程中自动管理会话历史的传入与更新。其核心作用是将历史会话消息与当前用户输入结合，作为链的输入参数，同时将链的输出结果补充到会话历史中，形成闭环。

使用时需指定三个关键参数：基础链（runnable）、会话历史存储实例（history_factory）、会话标识（session_id）。其中session_id用于区分不同会话，确保多用户并发场景下历史记录不混淆。

简单示例如下：

fromlangchain.chainsimportLLMChainfromlangchain.promptsimportChatPromptTemplatefromlangchain.chat_modelsimportChatOpenAIfromlangchain.schema.runnableimportRunnableWithMessageHistoryfromlangchain.memoryimportInMemoryChatMessageHistory# 初始化基础链llm=ChatOpenAI(temperature=0)prompt=ChatPromptTemplate.from_messages([("system","你是一个助手，基于历史会话回答问题。"),("human","{question}")])base_chain=prompt|llm# 为基础链添加会话历史功能chain_with_history=RunnableWithMessageHistory(runnable=base_chain,history_factory=lambdasession_id:InMemoryChatMessageHistory(session_id=session_id),input_messages_key="question",# 指定输入中用户问题的键output_messages_key="output"# 指定输出中助手回复的键)# 执行带历史记忆的链response1=chain_with_history.invoke({"question":"我叫小明"},config={"configurable":{"session_id":"session-001"}})print(response1)response2=chain_with_history.invoke({"question":"我叫什么名字？"},config={"configurable":{"session_id":"session-001"}})print(response2)# 助手会基于历史会话回答“你叫小明”

2. InMemoryChatMessageHistory类

该类为会话历史提供内存级存储支持，属于临时存储方案，会话结束后内存释放，历史记录丢失，适用于无需持久化的场景（如单次临时对话）。其内部维护一个列表存储消息对象，自动处理消息的顺序管理，无需开发者手动维护。

核心特性：轻量高效，无外部存储依赖，仅适用于单进程、单次会话场景，无法跨会话、跨服务共享历史记录。

二、长期会话记忆

当需要持久化存储会话历史（如多轮对话断点续聊、会话回溯）时，需使用长期会话记忆方案。LangChain提供FileChatMessageHistory类实现基于文件的持久化存储，同时支持通过继承BaseChatMessageHistory类自定义存储方案（如数据库存储）。

1. FileChatMessageHistory类

该类基于文件系统存储会话记录，以session_id作为文件名，每个会话对应一个独立文件，实现不同会话历史的隔离存储。文件存储方式确保会话历史在应用重启后不丢失，适用于对持久化要求较低、无需高并发访问的场景。

使用示例：

fromlangchain.memoryimportFileChatMessageHistory# 初始化文件存储的会话历史，session_id为文件名history=FileChatMessageHistory(session_id="session-001")# 添加消息history.add_message({"role":"human","content":"我叫小明"})history.add_message({"role":"ai","content":"你好小明！有什么可以帮你？"})# 获取消息print(history.messages)# 输出所有历史消息# 清除消息history.clear()print(history.messages)# 输出空列表

该类默认将文件存储在当前工作目录下，文件名格式为{session_id}.json，消息以JSON格式序列化存储，便于读取与解析。

2. 继承BaseChatMessageHistory类自定义存储

FileChatMessageHistory类本质是继承BaseChatMessageHistory类的实现类。若需自定义存储介质（如MySQL、Redis），可直接继承BaseChatMessageHistory类，并实现其要求的三个同步方法，即可接入LangChain的会话记忆体系。

BaseChatMessageHistory类是会话历史的抽象基类，定义了会话记忆的核心接口，强制子类实现以下三个同步方法：

（1）add_message：同步添加消息

功能：将用户消息或AI回复消息同步添加到存储介质中，需保证消息的顺序性（按时间先后存储）。方法参数为message对象，类型为BaseMessage（LangChain的基础消息类，包含role和content属性）。

实现示例（以自定义数据库存储为例）：

fromlangchain.schemaimportBaseMessagefromlangchain.memory.chat_message_histories.baseimportBaseChatMessageHistoryclassDatabaseChatMessageHistory(BaseChatMessageHistory):def__init__(self,session_id:str):self.session_id=session_id# 初始化数据库连接self.db_conn=self._init_db()def_init_db(self):# 模拟数据库连接初始化return{"session-001":[]}defadd_message(self,message:BaseMessage)->None:# 同步添加消息到数据库message_dict={"role":message.role,"content":message.content,"timestamp":datetime.now().isoformat()}self.db_conn[self.session_id].append(message_dict)

（2）messages：同步获取消息

功能：从存储介质中同步读取当前会话（对应session_id）的所有历史消息，返回值为List[BaseMessage]类型，确保消息顺序与发送顺序一致。

补充实现上述类的messages属性：

@propertydefmessages(self)->List[BaseMessage]:# 从数据库读取消息并转换为BaseMessage对象fromlangchain.schemaimportHumanMessage,AIMessage message_list=[]formsginself.db_conn.get(self.session_id,[]):ifmsg["role"]=="human":message_list.append(HumanMessage(content=msg["content"]))elifmsg["role"]=="ai":message_list.append(AIMessage(content=msg["content"]))returnmessage_list

（3）clear：同步清除消息

功能：同步清空当前会话（对应session_id）的所有历史消息，删除存储介质中的对应数据，确保数据清理的原子性。

补充实现上述类的clear方法：

defclear(self)->None:# 清空数据库中当前会话的消息self.db_conn[self.session_id]=[]

总结

LangChain的History模块通过分层设计，兼顾了临时会话与长期会话的记忆需求。RunnableWithMessageHistory+InMemoryChatMessageHistory适用于轻量临时对话，FileChatMessageHistory满足基础持久化需求，而继承BaseChatMessageHistory自定义实现则支持复杂存储场景。开发者可根据业务对会话时效、存储介质的需求，灵活选择对应的实现方案，快速为AI应用赋予上下文交互能力。