LangChain v0.2中如何结合RunnableWithMessageHistory使用ConversationSummaryMemory？

在LangChain v0.2中，ConversationSummaryMemory不再继承自BaseChatMessageHistory，无法直接用于RunnableWithMessageHistory的会话历史回调。推荐通过自定义BaseChatMessageHistory子类的方式，将对话摘要逻辑封装到会话历史管理中，无缝对接原有RunnableWithMessageHistory的使用流程。

方案实现：自定义摘要式会话历史 #

我们创建一个继承自BaseChatMessageHistory的类，内部维护对话摘要和少量最新消息（用于更新摘要），自动处理摘要生成与持久化。

1. 自定义摘要会话历史类 #

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain.chat_models import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import json

class SummaryChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, file_path: str, llm: ChatOpenAI):
        self.file_path = file_path
        self.llm = llm
        self.summary = ""
        # 保留最近4条消息（2轮对话）用于更新摘要，平衡准确性与效率
        self.recent_messages = []
        self._load_from_file()

    def _load_from_file(self):
        """从文件加载已保存的摘要和最近消息"""
        try:
            with open(self.file_path, "r") as f:
                data = json.load(f)
                self.summary = data.get("summary", "")
                self.recent_messages = [self._dict_to_message(msg) for msg in data.get("recent_messages", [])]
        except FileNotFoundError:
            self.summary = ""
            self.recent_messages = []

    def _save_to_file(self):
        """将摘要和最近消息保存到文件"""
        data = {
            "summary": self.summary,
            "recent_messages": [self._message_to_dict(msg) for msg in self.recent_messages]
        }
        with open(self.file_path, "w") as f:
            json.dump(data, f)

    def _message_to_dict(self, msg: BaseMessage) -> dict:
        """将消息对象转为可序列化的字典"""
        return {"type": msg.type, "content": msg.content}

    def _dict_to_message(self, msg_dict: dict) -> BaseMessage:
        """将字典转为消息对象"""
        msg_type = msg_dict["type"]
        content = msg_dict["content"]
        if msg_type == "human":
            return HumanMessage(content=content)
        elif msg_type == "ai":
            return AIMessage(content=content)
        elif msg_type == "system":
            return SystemMessage(content=content)
        raise ValueError(f"未知消息类型：{msg_type}")

    def add_messages(self, messages: list[BaseMessage]) -> None:
        """添加新消息并更新对话摘要"""
        self.recent_messages.extend(messages)
        # 限制最近消息数量，避免冗余
        if len(self.recent_messages) > 4:
            self.recent_messages = self.recent_messages[-4:]
        
        # 生成更新后的摘要
        if self.summary:
            summary_prompt = ChatPromptTemplate.from_messages([
                SystemMessage(content="结合之前的对话摘要和最新对话内容，生成简洁的更新后摘要。"),
                SystemMessage(content=f"原有摘要：{self.summary}"),
                SystemMessage(content="最新对话："),
                *self.recent_messages,
                SystemMessage(content="更新后的摘要：")
            ])
        else:
            summary_prompt = ChatPromptTemplate.from_messages([
                SystemMessage(content="将以下对话内容生成简洁的摘要。"),
                *self.recent_messages,
                SystemMessage(content="对话摘要：")
            ])
        
        summary_chain = summary_prompt | self.llm
        self.summary = summary_chain.invoke({}).content
        self._save_to_file()

    @property
    def messages(self) -> list[BaseMessage]:
        """返回用于prompt的历史摘要（以SystemMessage形式传入）"""
        return [SystemMessage(content=f"对话历史摘要：{self.summary}")] if self.summary else []

    def clear(self) -> None:
        """清空会话历史"""
        self.summary = ""
        self.recent_messages = []
        self._save_to_file()

2. 对接原有 RunnableWithMessageHistory 流程 #

修改你的原有代码，替换FileChatMessageHistory为自定义的SummaryChatMessageHistory：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder, HumanMessagePromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain_core.runnables.history import RunnableWithMessageHistory

# 初始化LLM
chat = ChatOpenAI(model="gpt-3.5-turbo")

prompt = ChatPromptTemplate.from_messages([
    MessagesPlaceholder(variable_name="messages"),
    HumanMessagePromptTemplate.from_template("{content}"),
])

chain = prompt | chat

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    return SummaryChatMessageHistory(f"summary_{session_id}.json", llm=chat)

with_message_history = RunnableWithMessageHistory(
    chain, 
    get_session_history=get_session_history,
    input_messages_key="content",
    history_messages_key="messages",
)

# 测试对话流程
while True:
    content = input(">>> ")
    if content.lower() == "exit":
        break
    result = with_message_history.invoke(
        input={"content": content},
        config={"configurable": {"session_id": "abc123"}}
    )
    print(result.content)

核心优势 #

• 无缝兼容RunnableWithMessageHistory的原有逻辑，无需大幅修改代码结构
• 自动处理摘要生成与持久化，无需手动管理历史消息的添加和摘要更新
• 通过保留少量最新消息，在保证摘要准确性的同时降低LLM调用成本

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