跳至主要內容

Memory

JustQ2026/2/18大约 5 分钟

Memory

LangChain Memory 使用指南（新方案）

LangChain 的 Memory 模块经历了一次重要重构。本文聚焦新方案，旧方案仅作背景了解。核心思路：LLM 本身无状态，所谓"记忆"就是把历史对话以合适的形式拼入 prompt，再发送给模型。

一、两套方案背景

LangChain 目前存在新旧两套 Memory 方案并行，很多教程混用，容易踩坑。

1.1 旧方案（Legacy，已废弃）

基于 ConversationBufferMemory、ConversationSummaryMemory 等 Memory 类，配合 ConversationChain、LLMChain 使用。

# 旧方案写法（不推荐新项目使用）
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

memory = ConversationBufferMemory()
chain = ConversationChain(llm=llm, memory=memory)
chain.predict(input="你好")

已废弃的原因：与 LCEL 管道不兼容、不支持多用户 session、无法对接持久化存储、扩展困难。

1.2 新方案（推荐）

基于 LCEL + RunnableWithMessageHistory，手动或自动管理 chat_history，存储后端完全可插拔。核心变化是将记忆的管理权还给开发者，不再黑盒封装。

📌 选型原则：新项目一律用新方案。维护老代码时才需要了解旧方案。

二、核心概念：记忆的本质

在深入代码之前，先理解 LangChain Memory 的底层逻辑。

每次调用 LLM，实际发送的内容：

┌─────────────────────────────────────────┐
│  System: 你是一个helpful助手             │
│  Human:  你好                            │  ← 第1轮
│  AI:     你好！有什么可以帮你？           │
│  Human:  我叫小明                        │  ← 第2轮
│  AI:     好的，小明！                    │
│  Human:  【本轮新消息】我叫什么名字？     │  ← 第3轮（当前）
└─────────────────────────────────────────┘

所有 Memory 方案的本质差异只有两点：

  • 存哪里：内存 / Redis / 数据库 / 向量库
  • 存多少：全部 / 最近 N 条 / 摘要 / 语义检索

三、短期记忆：对话上下文管理

短期记忆指当前会话内的历史消息，随会话结束而失效（或持久化到数据库）。

3.1 基础：手动管理 chat_history

最简单直接的方式，完全透明可控。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

llm = ChatOpenAI(model="gpt-4o")

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个helpful助手。"),
    MessagesPlaceholder(variable_name="chat_history"),  # ← 历史消息插入点
    ("human", "{input}"),
])

chain = prompt | llm

chat_history = []  # 本地维护历史列表

def chat(user_input: str) -> str:
    response = chain.invoke({
        "chat_history": chat_history,
        "input": user_input,
    })
    # 每轮结束后手动追加
    chat_history.append(HumanMessage(content=user_input))
    chat_history.append(AIMessage(content=response.content))
    return response.content

chat("我叫小明")    # → 好的，小明！
chat("我叫什么？")  # → 你叫小明

💡 关键：MessagesPlaceholder 是插槽，它告诉 prompt"这里放一个消息对象列表"，接受 List[BaseMessage] 类型。

3.2 进阶：RunnableWithMessageHistory 自动管理

不想手动 append，用这个包装器自动完成历史的读取和写入，同时支持多用户 session 隔离。

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# session_id → 历史对象 的映射表
store = {}

def get_session_history(session_id: str) -> ChatMessageHistory:
    """根据 session_id 返回对应的历史对象，不存在则创建"""
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 用包装器包裹 chain
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",          # chain 输入中"本轮问题"的 key
    history_messages_key="chat_history", # prompt 中历史消息占位符的 key
)

# 通过 session_id 区分不同用户/会话
def chat(user_input: str, session_id: str) -> str:
    response = chain_with_history.invoke(
        {"input": user_input},
        config={"configurable": {"session_id": session_id}}
    )
    return response.content

# 两个用户的历史完全独立
chat("我叫小明", session_id="user_001")
chat("我叫小红", session_id="user_002")
chat("我叫什么？", session_id="user_001")  # → 你叫小明
chat("我叫什么？", session_id="user_002")  # → 你叫小红

⚠️ 注意：上面的 store 存在内存里，进程重启即丢失。生产环境需要替换为持久化后端（见第四节）。

3.3 历史消息裁剪策略

随着对话变长，历史消息会撑爆 context window，并造成 token 费用线性增长。

方式一：按条数裁剪

from langchain_core.runnables import RunnableLambda

def trim_to_last_n(input: dict, n: int = 10) -> dict:
    """只保留最近 n 条消息"""
    input["chat_history"] = input["chat_history"][-n:]
    return input

chain = RunnableLambda(trim_to_last_n) | prompt | llm

方式二：按 token 数裁剪（推荐）

from langchain_core.messages import trim_messages, SystemMessage

trimmer = trim_messages(
    max_tokens=4096,       # 历史消息最多保留 4096 token
    strategy="last",       # 保留最新的消息，丢弃最旧的
    token_counter=llm,     # 用 llm 计算 token 数量
    include_system=True,   # system message 不参与裁剪，始终保留
    allow_partial=False,   # 不截断单条消息（保证消息完整性）
)

# 插入 chain 的最前面
chain = (
    RunnableLambda(lambda x: {**x, "chat_history": trimmer.invoke(x["chat_history"])})
    | prompt
    | llm
)

方式三：摘要压缩（超长对话场景）

from langchain_core.messages import SystemMessage

summarize_prompt = ChatPromptTemplate.from_messages([
    ("system", "请将以下对话历史压缩为一段简洁的摘要，保留关键信息："),
    MessagesPlaceholder(variable_name="messages"),
])
summarize_chain = summarize_prompt | llm | StrOutputParser()

def summarize_history(messages: list, threshold: int = 20) -> list:
    """当消息超过阈值时，对早期消息做摘要压缩，保留最近消息"""
    if len(messages) <= threshold:
        return messages

    split = len(messages) - threshold // 2
    early, recent = messages[:split], messages[split:]

    summary_text = summarize_chain.invoke({"messages": early})
    summary_msg = SystemMessage(content=f"[早期对话摘要]\n{summary_text}")

    return [summary_msg] + recent

💡 三种裁剪策略的选择：

  • 简单聊天机器人 → 按条数裁剪，简单够用
  • 需要精确控制费用 → 按 token 裁剪
  • 客服/长会话场景 → 摘要压缩，平衡信息保留与成本

四、短期记忆持久化：存储后端

RunnableWithMessageHistory 的 get_session_history 函数是可插拔的，只需要替换返回的对象类型，即可切换存储后端。

4.1 内存存储（开发/测试）

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]

进程重启即清空，仅用于本地开发和测试。

4.2 Redis（生产首选）

适合高并发、低延迟场景，支持消息 TTL 自动过期。

from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_session_history(session_id: str):
    return RedisChatMessageHistory(
        session_id,
        url="redis://localhost:6379",
        ttl=3600,  # 可选：1小时后自动过期
    )

4.3 SQLite / PostgreSQL（需要持久化查询）

适合需要检索、统计历史对话的场景。

from langchain_community.chat_message_histories import SQLChatMessageHistory

def get_session_history(session_id: str):
    return SQLChatMessageHistory(
        session_id=session_id,
        connection="sqlite:///chat_history.db",
        # PostgreSQL: "postgresql://user:password@localhost/dbname"
    )

4.4 MongoDB（文档型存储）

from langchain_community.chat_message_histories import MongoDBChatMessageHistory

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

📌 session_id 设计建议：格式用 {user_id}:{conversation_id}，如 "user_42:conv_001"。同一用户可以拥有多个独立会话，便于管理和查询。

五、长期记忆：跨会话的持久知识

短期记忆随会话结束而"遗忘"。长期记忆是指把重要信息跨会话持久保存，在未来对话中按需调取，模拟人类的"记住你这个人"。

5.1 长期记忆的两种类型

类型说明举例
显式记忆（结构化）明确提取并存储的事实用户姓名、偏好、职业
语义记忆（向量化）嵌入为向量，按语义相似度召回历史对话片段、知识笔记
默认页脚
Copyright © 2026 JustQ