麻瓜教程

Appearance

记忆管理 ​

记忆（Memory）是 LangChain 中的重要组件，它用于存储和管理会话历史，使 LLM 应用能够保持上下文信息，提高交互的连贯性。本节将详细介绍 LangChain 1.2 中的记忆管理方法。

安装依赖 ​

在开始使用记忆管理之前，需要安装以下依赖：

bash
pip install langchain langchain-openai python-dotenv langchain-community

什么是记忆？ ​

记忆是用于存储和管理会话历史的组件，它允许 LLM 应用在多次交互中保持上下文信息，使对话更加连贯自然。记忆可以存储用户的输入、模型的输出以及其他相关信息。

在 LangChain 1.2 中，推荐使用 RunnableWithMessageHistory 来管理对话历史，它提供了更灵活、更强大的记忆管理功能。

使用 RunnableWithMessageHistory ​

RunnableWithMessageHistory 是 LangChain 1.2 推荐使用的记忆管理方式，它可以将任何 Runnable 包装成支持对话历史的链。

基本用法 ​

python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 创建消息历史存储
store = {}

def get_session_history(session_id: str):
    """获取会话历史"""
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 创建提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# 初始化模型
llm = ChatOpenAI(api_key="your-api-key")

# 创建输出解析器
output_parser = StrOutputParser()

# 使用 LCEL 语法创建链
chain = prompt | llm | output_parser

# 创建带记忆的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 运行链
response1 = chain_with_history.invoke(
    {"input": "What is LangChain?"},
    config={"configurable": {"session_id": "user_123"}}
)
print("Response 1:", response1)

response2 = chain_with_history.invoke(
    {"input": "Can you give me an example?"},
    config={"configurable": {"session_id": "user_123"}}
)
print("Response 2:", response2)

# 查看会话历史
print("History:", get_session_history("user_123").messages)

使用不同的 ChatMessageHistory 后端 ​

LangChain 1.2 支持多种 ChatMessageHistory 后端，适用于不同的场景：

内存存储（默认） ​

python
from langchain_community.chat_message_histories import ChatMessageHistory

store = {}

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

Redis 存储 ​

python
from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_session_history(session_id: str):
    return RedisChatMessageHistory(
        session_id=session_id,
        url="redis://localhost:6379/0"
    )

数据库存储 ​

python
from langchain_community.chat_message_histories import SQLChatMessageHistory

def get_session_history(session_id: str):
    return SQLChatMessageHistory(
        session_id=session_id,
        connection_string="sqlite:///memory.db"
    )

MongoDB 存储 ​

python
from langchain_community.chat_message_histories import MongoDBChatMessageHistory

def get_session_history(session_id: str):
    return MongoDBChatMessageHistory(
        session_id=session_id,
        connection_string="mongodb://localhost:27017/",
        database_name="chat_db",
        collection_name="chat_history"
    )

传统记忆类型（向后兼容） ​

虽然 LangChain 1.2 推荐使用 RunnableWithMessageHistory，但以下传统记忆类型仍然可用：

记忆类型描述适用场景
ConversationBufferMemory简单地存储所有对话历史对话历史较短的场景
ConversationBufferWindowMemory只保留最近的 N 条对话对话历史较长的场景
ConversationSummaryMemory对对话历史进行总结对话历史较长的场景
ConversationSummaryBufferMemory结合了摘要和窗口的记忆对话历史较长的场景

使用 ConversationBufferMemory ​

python
from langchain.memory import ConversationBufferMemory

# 初始化记忆
memory = ConversationBufferMemory()

# 存储对话
memory.save_context(
    {"input": "What is LangChain?"},
    {"output": "LangChain is a framework for building LLM applications."}
)

# 加载记忆
print(memory.load_memory_variables({}))

使用 ConversationBufferWindowMemory ​

python
from langchain.memory import ConversationBufferWindowMemory

# 初始化记忆（只保留最近 2 条对话）
memory = ConversationBufferWindowMemory(k=2)

# 存储对话
memory.save_context(
    {"input": "What is LangChain?"},
    {"output": "LangChain is a framework for building LLM applications."}
)
memory.save_context(
    {"input": "Can you give me an example?"},
    {"output": "Sure, you can use LangChain to build a chatbot."}
)
memory.save_context(
    {"input": "How does it work?"},
    {"output": "LangChain provides components and chains to build LLM applications."}
)

# 加载记忆
print(memory.load_memory_variables({}))

使用 ConversationSummaryMemory ​

python
from langchain.memory import ConversationSummaryMemory
from langchain_openai import OpenAI

# 初始化语言模型
llm = OpenAI(api_key="your-api-key")

# 初始化记忆
memory = ConversationSummaryMemory(llm=llm)

# 存储对话
memory.save_context(
    {"input": "What is LangChain?"},
    {"output": "LangChain is a framework for building LLM applications."}
)
memory.save_context(
    {"input": "Can you give me an example?"},
    {"output": "Sure, you can use LangChain to build a chatbot."}
)

# 加载记忆
print(memory.load_memory_variables({}))

使用 ConversationSummaryBufferMemory ​

python
from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import OpenAI

# 初始化语言模型
llm = OpenAI(api_key="your-api-key")

# 初始化记忆（只保留最近 2 条对话，其余进行摘要）
memory = ConversationSummaryBufferMemory(llm=llm, max_token_limit=100)

# 存储对话
memory.save_context(
    {"input": "What is LangChain?"},
    {"output": "LangChain is a framework for building LLM applications."}
)
memory.save_context(
    {"input": "Can you give me an example?"},
    {"output": "Sure, you can use LangChain to build a chatbot."}
)
memory.save_context(
    {"input": "How does it work?"},
    {"output": "LangChain provides components and chains to build LLM applications."}
)

# 加载记忆
print(memory.load_memory_variables({}))

记忆的高级配置 ​

自定义记忆键 ​

python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

# 创建提示词模板（使用自定义键名）
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{user_input}")
])

# 创建链
chain = prompt | llm | output_parser

# 创建带记忆的链（指定自定义键名）
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="user_input",
    history_messages_key="chat_history"
)

# 运行链
response = chain_with_history.invoke(
    {"user_input": "What is LangChain?"},
    config={"configurable": {"session_id": "user_123"}}
)

限制历史消息数量 ​

python
from langchain_community.chat_message_histories import ChatMessageHistory

class LimitedChatMessageHistory(ChatMessageHistory):
    """限制历史消息数量的 ChatMessageHistory"""

    def __init__(self, max_messages=10):
        super().__init__()
        self.max_messages = max_messages

    def add_message(self, message):
        super().add_message(message)
        # 只保留最近的消息
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]

store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = LimitedChatMessageHistory(max_messages=10)
    return store[session_id]

记忆的最佳实践 ​

1. 选择合适的记忆后端 ​

根据应用场景选择合适的 ChatMessageHistory 后端：

  • 内存存储：适用于开发和测试，数据不会持久化
  • Redis 存储：适用于生产环境，需要快速访问
  • 数据库存储：适用于需要持久化的场景
  • MongoDB 存储：适用于大规模分布式应用

2. 合理设置历史消息限制 ​

根据实际需求，合理设置历史消息的数量限制：

python
# 限制历史消息数量
max_history = 10  # 只保留最近 10 条消息

3. 定期清理记忆 ​

对于长时间运行的应用，定期清理记忆，避免记忆过大影响性能：

python
# 定期清理过期会话
def cleanup_expired_sessions(store, max_age_hours=24):
    """清理过期的会话"""
    from datetime import datetime, timedelta

    current_time = datetime.now()
    expired_sessions = []

    for session_id, history in store.items():
        # 检查最后一条消息的时间
        if history.messages:
            last_message_time = history.messages[-1].additional_kwargs.get('timestamp')
            if last_message_time:
                last_time = datetime.fromisoformat(last_message_time)
                if current_time - last_time > timedelta(hours=max_age_hours):
                    expired_sessions.append(session_id)

    # 删除过期会话
    for session_id in expired_sessions:
        del store[session_id]

4. 使用会话隔离 ​

在多用户场景中，确保每个用户的会话是隔离的：

python
# 使用用户 ID 作为会话 ID
def get_user_session_history(user_id: str):
    if user_id not in store:
        store[user_id] = ChatMessageHistory()
    return store[user_id]

# 运行链时指定用户 ID
response = chain_with_history.invoke(
    {"input": "What is LangChain?"},
    config={"configurable": {"session_id": user_id}}
)

示例：构建一个带记忆的聊天机器人 ​

python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 创建消息历史存储
store = {}

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

# 创建提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# 初始化模型
llm = ChatOpenAI(api_key="your-api-key")

# 创建输出解析器
output_parser = StrOutputParser()

# 使用 LCEL 语法创建链
chain = prompt | llm | output_parser

# 创建带记忆的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 交互循环
print("Chatbot: Hello! How can I help you today?")
while True:
    user_input = input("You: ")
    if user_input.lower() in ["exit", "quit", "bye"]:
        print("Chatbot: Goodbye!")
        break

    response = chain_with_history.invoke(
        {"input": user_input},
        config={"configurable": {"session_id": "user_123"}}
    )
    print(f"Chatbot: {response}")

总结 ​

记忆是 LangChain 中的重要组件，它用于存储和管理会话历史，使 LLM 应用能够保持上下文信息，提高交互的连贯性。在 LangChain 1.2 中，推荐使用 RunnableWithMessageHistory 来管理对话历史，它提供了更灵活、更强大的记忆管理功能。

通过本文的介绍，您应该已经了解了：

  1. 如何使用 RunnableWithMessageHistory 管理对话历史
  2. 如何使用不同的 ChatMessageHistory 后端
  3. 传统记忆类型的使用方法（向后兼容）
  4. 记忆的最佳实践

在实际应用中，您可以根据具体需求选择合适的记忆管理方式，配置适当的参数，构建功能强大的 LLM 应用。

Copyright © 2025-2026 京公网安备 11010102000001号 京ICP备2024078128号-1