如何将RedisChatMessageHistory接入ConversationSummaryBufferMemory优化Token用量？

问题原因 #

你遇到的AttributeError是因为类型接口不匹配：

• RunnableWithMessageHistory要求传入的是BaseChatMessageHistory类型的实例（比如RedisChatMessageHistory），这类对象拥有.messages属性来直接获取对话消息列表。
• ConversationSummaryBufferMemory属于BaseMemory类型，它的核心接口是load_memory_variables()，并不暴露.messages属性，所以直接将其传给RunnableWithMessageHistory会触发属性不存在的错误。

你的核心思路（通过汇总历史降低Token消耗）是正确的，只是选错了组件组合方式。

可行解决方案 #

方案1：自定义带汇总功能的Redis聊天历史类 #

继承BaseChatMessageHistory，内部整合RedisChatMessageHistory的持久化能力和ConversationSummaryBufferMemory的摘要逻辑，让对象同时满足RunnableWithMessageHistory的接口要求和历史汇总需求。

示例代码：

from langchain.memory import BaseChatMessageHistory, ConversationSummaryBufferMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain.schema import BaseMessage, HumanMessage, AIMessage
from langchain.llms.base import BaseLLM

class SummarizedRedisChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, redis_url: str, session_id: str, llm: BaseLLM, max_token_limit: int = 1000):
        # 初始化Redis持久化历史
        self.redis_history = RedisChatMessageHistory(redis_url=redis_url, session_id=session_id)
        # 初始化摘要内存，绑定Redis历史作为数据源
        self.summary_memory = ConversationSummaryBufferMemory(
            llm=llm,
            max_token_limit=max_token_limit,
            chat_memory=self.redis_history,
            return_messages=True
        )
    
    @property
    def messages(self):
        # 返回汇总后的历史：摘要消息 + 最新未被汇总的对话
        summary_content = self.summary_memory.load_memory_variables({})["history"]
        summary_msg = AIMessage(content=f"对话历史摘要：{summary_content}")
        # 保留最新的几条原始对话（可根据token阈值调整数量）
        recent_messages = self.redis_history.messages[-3:]
        return [summary_msg] + recent_messages
    
    def add_message(self, message: BaseMessage) -> None:
        # 同步消息到Redis
        self.redis_history.add_message(message)
        # 更新摘要内存
        if isinstance(message, HumanMessage):
            self.summary_memory.save_context({"input": message.content}, {"output": ""})
        elif isinstance(message, AIMessage):
            self.summary_memory.save_context({"input": ""}, {"output": message.content})
    
    def clear(self) -> None:
        # 清空Redis历史和摘要内存
        self.redis_history.clear()
        self.summary_memory.clear()

使用方式：
将这个自定义类传入RunnableWithMessageHistory的get_session_history参数即可：

chain = RunnableWithMessageHistory(
    retrieval_chain,
    get_session_history=lambda session_id: SummarizedRedisChatMessageHistory(
        redis_url="redis://localhost:6379",
        session_id=session_id,
        llm=your_llm_instance
    ),
    input_messages_key="question",
    history_messages_key="chat_history"
)

方案2：手动在历史获取环节做汇总 #

不依赖ConversationSummaryBufferMemory，直接在每次获取历史时，判断历史长度，超过阈值则调用LLM生成摘要，再将摘要+最新对话拼接成精简上下文。

示例代码：

from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain.schema import AIMessage, Document
from langchain.chains.summarize import load_summarize_chain

def get_summarized_history(session_id: str, llm, max_history_rounds: int = 5):
    history = RedisChatMessageHistory(redis_url="redis://localhost:6379", session_id=session_id)
    total_messages = len(history.messages)
    
    if total_messages <= max_history_rounds:
        # 历史较短，直接返回原始消息
        return history.messages
    else:
        # 生成历史摘要
        history_text = "\n".join([f"{msg.type}: {msg.content}" for msg in history.messages[:-max_history_rounds]])
        summary_chain = load_summarize_chain(llm, chain_type="stuff")
        summary = summary_chain.run([Document(page_content=history_text)])
        # 返回摘要+最新N轮对话
        summary_msg = AIMessage(content=f"对话历史摘要：{summary}")
        return [summary_msg] + history.messages[-max_history_rounds:]

# 构建带汇总历史的检索链
chain = RunnableWithMessageHistory(
    retrieval_chain,
    get_session_history=lambda session_id: get_summarized_history(session_id, your_llm_instance),
    input_messages_key="question",
    history_messages_key="chat_history"
)

总结 #

你的核心思路（减少历史Token消耗）是正确的，错误在于混淆了BaseChatMessageHistory和BaseMemory的接口差异。通过自定义兼容接口的历史类，或者手动在历史获取环节做摘要处理，就能实现Redis持久化+历史汇总的需求。

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