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

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

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

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

一、临时会话记忆

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

1. RunnableWithMessageHistory类

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

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

简单示例如下：

from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain.schema.runnable import RunnableWithMessageHistory
from langchain.memory import InMemoryChatMessageHistory

# 初始化基础链
llm = ChatOpenAI(temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手，基于历史会话回答问题。"),
    ("human", "{question}")
])
base_chain = prompt | llm

# 为基础链添加会话历史功能
chain_with_history = RunnableWithMessageHistory(
    runnable=base_chain,
    history_factory=lambda session_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作为文件名，每个会话对应一个独立文件，实现不同会话历史的隔离存储。文件存储方式确保会话历史在应用重启后不丢失，适用于对持久化要求较低、无需高并发访问的场景。

使用示例：

from langchain.memory import FileChatMessageHistory

# 初始化文件存储的会话历史，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属性）。

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

from langchain.schema import BaseMessage
from langchain.memory.chat_message_histories.base import BaseChatMessageHistory

class DatabaseChatMessageHistory(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": []}

    def add_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属性：

@property
def messages(self) -> List[BaseMessage]:
    # 从数据库读取消息并转换为BaseMessage对象
    from langchain.schema import HumanMessage, AIMessage
    message_list = []
    for msg in self.db_conn.get(self.session_id, []):
        if msg["role"] == "human":
            message_list.append(HumanMessage(content=msg["content"]))
        elif msg["role"] == "ai":
            message_list.append(AIMessage(content=msg["content"]))
    return message_list

（3）clear：同步清除消息

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

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

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

总结

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