如何将LangChain中ChatMessageHistory替换为ConversationBufferWindowMemory？

问题背景 #

基于LangChain实现RAG聊天历史功能，当前可完整保留聊天记录，但希望仅保留最近x条消息。尝试替换为ConversationBufferWindowMemory时出现报错：

TypeError: __init__() takes 1 positional argument but 3 were given

原代码：

store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

conversational_rag_chain = RunnableWithMessageHistory(
    rag_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="answer",
)

错误尝试代码：

store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ConversationBufferWindowMemory(memory_key="chat_history", k=2)
    return store[session_id]

conversational_rag_chain = ConversationChain(
    rag_chain,
    get_session_history,
    verbose=True, 
    memory="chat_history"
)

错误原因 #

1. 类型不匹配：RunnableWithMessageHistory要求get_session_history返回BaseChatMessageHistory子类实例（如ChatMessageHistory），但ConversationBufferWindowMemory属于BaseMemory子类，两者并非同一类型，无法直接替换。
2. ConversationChain初始化错误：错误替换为ConversationChain后，参数传递完全不符合要求——ConversationChain第一个参数需传入语言模型（LLM）实例，而非RAG链；memory参数需传入Memory实例，而非字符串。

正确实现方案 #

方案一：使用WindowChatMessageHistory适配原有架构（推荐） #

WindowChatMessageHistory是BaseChatMessageHistory的子类，专门用于保留最近k条消息，完美适配你原有的RunnableWithMessageHistory逻辑：

from langchain.memory.chat_message_histories import WindowChatMessageHistory

store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        # k值设置为需要保留的最近消息条数，这里示例为2
        store[session_id] = WindowChatMessageHistory(k=2)
    return store[session_id]

# 原有的RunnableWithMessageHistory初始化逻辑无需改动
conversational_rag_chain = RunnableWithMessageHistory(
    rag_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="answer",
)

方案二：改用ConversationChain搭配ConversationBufferWindowMemory #

如果需要使用ConversationBufferWindowMemory，需正确初始化ConversationChain，注意传入LLM实例：

from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferWindowMemory

# 初始化你的语言模型实例，例如OpenAI()
llm = ... 

# 单会话版本
conversational_rag_chain = ConversationChain(
    llm=llm,
    memory=ConversationBufferWindowMemory(k=2, memory_key="chat_history"),
    verbose=True
)

# 多会话管理版本
store = {}
def get_conversation_chain(session_id: str):
    if session_id not in store:
        store[session_id] = ConversationChain(
            llm=llm,
            memory=ConversationBufferWindowMemory(k=2, memory_key="chat_history"),
            verbose=True
        )
    return store[session_id]

内容的提问来源于stack exchange，提问作者dogma897