第2章: 记忆管理：让聊天助手多轮不失忆

🤔 为什么需要记忆？

❌ 没有记忆的问题

LLM本身是无状态的！每次调用都是独立的，不记得之前说过什么。

text

# 没有记忆的对话
llm.invoke("我叫张三")  # AI: 你好张三！
llm.invoke("我叫什么名字？")  # AI: 抱歉，我不知道你的名字
# 👉 AI完全忘记了刚才的对话！

✅ 有记忆的效果

记忆机制让AI记住对话历史，实现真正的多轮对话。

text

# 有记忆的对话
conversation.predict(input="我叫张三")  # AI: 你好张三！
conversation.predict(input="我叫什么名字？")  # AI: 你叫张三
# 👉 AI记住了之前的对话！

💡 记忆的核心作用

维护上下文：让AI理解对话的来龙去脉
个性化体验：记住用户信息和偏好
连贯对话：实现自然的多轮交互
提升体验：避免重复询问相同问题

📚 记忆类型详解

LangChain提供了多种记忆类型，适应不同的使用场景。

✅ 环境准备 & 如何运行本页代码（强烈建议先看）

本页所有示例代码都可以复制到 memory_x.py 里运行。最常见的问题不是代码写错，而是： IDE 选的 Python 解释器 ≠ 你安装依赖的 Python 环境。

1) 创建并激活虚拟环境（推荐）

text

# 在你的项目根目录执行
python3 -m venv .venv

# macOS / Linux
source .venv/bin/activate

# Windows PowerShell
# .venv\Scripts\Activate.ps1

2) 安装依赖（用同一个解释器的 pip）

text

# 关键点：永远用 "python -m pip"，避免 pip 装到另一个 Python 上
python -m pip install -U pip setuptools wheel

# 经典 Chain API（ConversationChain / ConversationBufferMemory）建议使用 langchain-classic
# 注意：langchain-classic==1.x 依赖 langchain-core>=1.0（必须一起装对版本）
python -m pip install -U \
  "langchain-classic==1.0.0" \
  "langchain-core>=1,<2" \
  "langchain-community>=0.3,<0.4" \
  "langchain-text-splitters>=1,<2" \
  "langsmith>=0.1.17,<1" \
  dashscope \
  redis \
  pymysql

3) 验证导入是否正常（先验证再跑示例）

python

python -c "from langchain_classic.chains import ConversationChain; from langchain_classic.memory import ConversationBufferMemory; print('import OK')"

4) 运行示例（建议统一用 python3）

text

# 例如：运行窗口记忆示例
python3 memory_window.py

5) IDE 里怎么选解释器（非常关键）

在 VS Code / PyCharm 里把 Python Interpreter 切到你刚创建的 .venv。如果你看到 from langchain... 下面有红线，通常就是解释器没选对。

6) Python 3.12 + pcapy 安装报错怎么办？（不降版本）

如果你遇到 Failed to build pcapy / distutils 相关错误，这是 pcapy 自身不兼容 Python 3.12。 不降 Python 版本的做法是换成维护分支 pcapy-ng：

text

python -m pip install pcapy-ng

如果你的代码里原来写的是 import pcapy，安装 pcapy-ng 后大多数场景仍然可以继续用；若遇到导入差异，再根据项目实际把 import 调整为对应模块名。

7) 常见报错：ConversationSummaryBufferMemory 运行时报 Token/Tokenizer 问题

如果你在跑 ConversationSummaryBufferMemory（或 SummaryBuffer 类记忆）时遇到类似报错： Could not import transformers / GPT2TokenizerFast.from_pretrained("gpt2") / httpx.ConnectTimeout，本质原因通常是：它需要计算 token 数，而默认 token 计数会依赖 transformers，并且可能需要从 HuggingFace 下载 gpt2 的 tokenizer 文件。

✅ 解决方案 A：安装 transformers

text

# 必须用你正在运行脚本的那个 python 来安装
python -m pip install -U transformers

✅ 解决方案 B：国内网络配置 HF 镜像（避免 ConnectTimeout）

text

# 方式1：运行脚本时临时设置（推荐）
HF_ENDPOINT=https://hf-mirror.com python3 memory_11.py

✅ 解决方案 C：预缓存 gpt2 tokenizer（下载一次，后面复用）

python
# 先把 tokenizer 下载到本地缓存目录
HF_ENDPOINT=https://hf-mirror.com python -c "from transformers import GPT2TokenizerFast; GPT2TokenizerFast.from_pretrained('gpt2'); print('gpt2 tokenizer cached')"

提示：如果你已经缓存成功，但运行脚本仍然超时，通常是因为运行时没带 HF_ENDPOINT。记得在运行脚本时也一起设置。

1. MessagesPlaceholder + RunnableWithMessageHistory（LCEL 正统做法）💾

特点：把对话历史作为消息列表自动注入到 Prompt 中，适合 LCEL / Runnable 风格的链式调用。

📖 工作原理

通过 RunnableWithMessageHistory 按 session_id 取出历史消息列表，并填充到 MessagesPlaceholder("history") 这个占位符位置。

优点：“历史注入”逻辑清晰，和 LCEL 链天然契合
缺点：历史越长 token 越多（需要配合 window/summary 等策略）
适用：你在用 prompt | llm 这种 Runnable 风格写链

python
import os

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

# ⚠️ 注意：请使用 python3 运行（macOS 上 /usr/bin/python 可能不是你的 pip 环境）
# python3 memory_placeholder.py

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# 关键点1：Prompt 里用 MessagesPlaceholder 预留“消息列表”位置
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个有帮助的助手。"),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)

chain = prompt | llm

# 关键点2：用一个 store 存不同 session 的 ChatMessageHistory
store = {}


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


conversation = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
)

session_id = "user_1"

print("=== 第1轮 ===")
r1 = conversation.invoke(
    {"input": "我叫张三，是一名Python开发者"},
    config={"configurable": {"session_id": session_id}},
)
print("AI:", r1.content, "\n")

print("=== 第2轮 ===")
r2 = conversation.invoke(
    {"input": "我的职业是什么？"},
    config={"configurable": {"session_id": session_id}},
)
print("AI:", r2.content, "\n")

print("=== 第3轮 ===")
r3 = conversation.invoke(
    {"input": "我叫什么名字？"},
    config={"configurable": {"session_id": session_id}},
)
print("AI:", r3.content, "\n")

print("=== 当前 history messages ===")
print(store[session_id].messages)

🔍 MessagesPlaceholder 和 "history" 深度解析

🤔 为什么要用 MessagesPlaceholder？

核心问题：LLM 每次调用都是独立的，不记得之前说过什么。我们需要把历史对话塞进 Prompt 里。

                        # 没有 MessagesPlaceholder 的写法（错误❌）

                        prompt = "你是有帮助的助手。历史对话：[这里要手动塞历史] 用户说：{input}"

                        # 有 MessagesPlaceholder 的写法（正确✅）

                        MessagesPlaceholder("history")  ← 自动处理格式和位置

MessagesPlaceholder 的作用：专门用来插入消息列表，确保格式正确。

🔑 "history" 这个名字是随便起的吗？

不是随便起的！"history" 必须在三个地方保持一致：

❌ 错误示例（名字不一致）

                                # Prompt 中用 "history"

                                MessagesPlaceholder("history")

                                # 配置中用 "chat_history"

                                history_messages_key="chat_history"

                                # 🚨 结果：找不到数据！

✅ 正确示例（名字一致）

                                # Prompt 中用 "history"

                                MessagesPlaceholder("history")

                                # 配置中用 "history"

                                history_messages_key="history"

                                # ✅ 结果：数据正确注入！

� 三处统一的 "history"

第1处：Prompt 模板声明

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是有帮助的助手"),
    MessagesPlaceholder("history"),  ← 第1处：占位符名称
    ("human", "{input}"),
])
                        

第2处：RunnableWithMessageHistory 配置

conversation = RunnableWithMessageHistory(
    chain, get_history,
    history_messages_key="history",  ← 第2处：配置键名
)
                        

第3处：运行时数据传递

# RunnableWithMessageHistory 自动生成：
{
    "history": [HumanMessage(...), AIMessage(...)],  ← 第3处：实际数据
    "input": "我叫什么名字？"
}
                        

🔄 实际运行过程演示

用户调用：

conversation.invoke(
    {"input": "我叫什么名字？"},
    config={"configurable": {"session_id": "user_1"}}
)
                        

🔍 config 参数详解（关键！）

❓ 为什么必须用 config={"configurable": {"session_id": ...}} ？

1) config 不是给模型的输入，是给运行时的配置
• {"input": "..."} → 业务输入，会进入 Prompt
• config={...} → 运行参数，由 Runnable 框架消费，不会进 Prompt

2) configurable 是 LangChain 约定的可注入字段容器
• RunnableWithMessageHistory 声明了需要 session_id
• 框架约定通过 config["configurable"]["session_id"] 传递

3) session_id 的作用：会话隔离
• 告诉 RunnableWithMessageHistory "这次属于哪个会话"
• 框架调用你提供的 get_session_history(session_id)
• 返回该会话的 ChatMessageHistory 对象
• 历史被注入到 MessagesPlaceholder("history")

🔄 config 在链路中的完整流转

1️⃣ 用户调用 → config 传入整个 pipeline

2️⃣ RunnableWithMessageHistory 层 → 从 config["configurable"]["session_id"] 取值

3️⃣ 调用 get_session_history(session_id) → 得到该会话的历史对象

4️⃣ 把历史注入 MessagesPlaceholder("history") → 拼装完整 prompt

5️⃣ 执行底层 chain (prompt → model → parser)

6️⃣ 把本轮 Human/AI 消息追加回该 session 的历史

📦 config 还能放什么？（常用字段）

configurable

session_id、user_id、tenant_id 等运行时可注入参数

💡 可以换成其他名字吗？

可以！但必须三处统一。比如换成 "chat_history"：

                        # 三处都改成 "chat_history"

                        MessagesPlaceholder("chat_history")

                        history_messages_key="chat_history"

                        # 运行时数据变成：{"chat_history": [消息列表]}

总结：MessagesPlaceholder 就是预留位置，"history" 是数据键名，两者配合实现动态历史注入。

类型	典型类名	特点	适用
内存	ChatMessageHistory	零依赖、最快，但进程重启即丢	本地开发 / Demo
SQLite / MySQL	SQLChatMessageHistory	可持久化、可查询；需要 DB 连接	生产落盘 / 审计
Redis	RedisChatMessageHistory	性能高、支持 TTL；适合多实例共享	在线对话服务
文件	FileChatMessageHistory	简单直观，但并发/检索能力弱	个人项目 / 小规模

📦 ConversationBufferMemory（传统 Chain API）

这一节把 ConversationBufferMemory 单独拿出来讲：它属于 LangChain 早期的 ConversationChain（传统 Chain API） 生态，思路是“把历史拼成文本塞回 Prompt”。

📦 ConversationBufferMemory 是什么？

ConversationBufferMemory 会把所有历史对话拼成一段文本（memory.buffer），并在每次调用 ConversationChain 时自动注入到 prompt。

一句话：信息不丢，但对话越长 Token 越多。

🧩 一句话理解（BufferMemory）

把所有历史原封不动塞进 prompt。好处是"信息不丢"，代价是"越聊越贵、越聊越慢"。

✅ 适用场景

用户咨询流程短（<10轮），但需要完整回看所有细节：例如客服工单、一次性需求澄清、面试模拟。

⚠️ 不适用场景

长对话、长文本输入（例如复盘会议记录、长期陪聊）。这类场景容易出现： Token 爆炸 / 延迟上升 / 成本失控。

🖼️ 图示：所有历史都进入 prompt

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationBufferMemory
from langchain_community.chat_models import ChatTongyi

# ⚠️ 注意：请使用 python3 运行
# python3 memory_buffer.py

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# BufferMemory：完整保存所有对话历史（对话越长，token 越多）
memory = ConversationBufferMemory()

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

print(conversation.predict(input="我叫张三，是一名Python开发者"))
print(conversation.predict(input="我的职业是什么？"))
print(conversation.predict(input="我叫什么名字？"))

print("\n=== memory.buffer ===")
print(memory.buffer)

💾 BufferMemory 持久化：保存到磁盘/数据库

问题：默认的 ChatMessageHistory 只保存在内存里，程序重启后历史就丢了。
解决：把 chat_memory 换成持久化的 SQLChatMessageHistory / RedisChatMessageHistory / FileChatMessageHistory。

SQLite（单机/测试）

本地文件，零配置，适合开发和小规模部署。

Redis（分布式/高性能）

内存级速度，支持过期策略，适合多实例部署。

File（简单场景）

每个 session 一个 JSON 文件，直观可控。

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_community.chat_models import ChatTongyi

# BufferMemory + SQLite 持久化
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# 使用 SQLChatMessageHistory 替代内存存储
memory = ConversationBufferMemory(
    chat_memory=SQLChatMessageHistory(
        session_id="user_123",  # 每个用户的唯一标识
        connection_string="sqlite:///chat_history.db"  # 本地 SQLite 文件
    ),
    return_messages=True
)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 第一次运行：写入对话
print(conversation.predict(input="我叫张三，是一名Python开发者"))
print(conversation.predict(input="我今年30岁，住在北京"))

print("\n=== 重启程序后再次运行 ===")
print(conversation.predict(input="我叫什么名字？我的职业是什么？"))

# 数据已保存到 chat_history.db，程序重启后仍然存在

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_community.chat_models import ChatTongyi

# BufferMemory + MySQL 持久化（生产环境推荐）
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# 使用本地 MySQL 数据库
# 前置准备：
# 1) 安装 MySQL 驱动：python -m pip install pymysql
# 2) 创建数据库：CREATE DATABASE langchain_chat;
# 3) LangChain 会自动创建表结构
memory = ConversationBufferMemory(
    chat_memory=SQLChatMessageHistory(
        session_id="user_123",
        connection_string="mysql+pymysql://root:password@localhost:3306/langchain_chat"
    ),
    return_messages=True
)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 第一次运行：写入对话
print(conversation.predict(input="我叫张三，是一名Python开发者"))
print(conversation.predict(input="我今年30岁，住在北京"))

print("\n=== 重启程序后再次运行 ===")
print(conversation.predict(input="我叫什么名字？我的职业是什么？"))

# 数据已保存到 MySQL 数据库，多实例部署时共享

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_community.chat_models import ChatTongyi

# BufferMemory + Redis 持久化（适合分布式部署）
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

memory = ConversationBufferMemory(
    chat_memory=RedisChatMessageHistory(
        session_id="user_123",
        url="redis://localhost:6379/0",  # Redis 连接地址
        ttl=3600  # 1小时后自动过期（秒）
    ),
    return_messages=True
)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

print(conversation.predict(input="我叫张三，请多关照"))
print(conversation.predict(input="我叫什么名字？"))

# 数据保存在 Redis，可被多个服务实例共享

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_community.chat_models import ChatTongyi

# BufferMemory + 文件持久化（最简单直观）
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

memory = ConversationBufferMemory(
    chat_memory=FileChatMessageHistory(
        file_path="session_user_123.json"  # 每个 session 一个文件
    ),
    return_messages=True
)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

print(conversation.predict(input="我叫张三，是一名开发者"))
print(conversation.predict(input="我叫什么名字？"))

# 数据保存到 session_user_123.json，可随时查看和备份

2. ConversationBufferWindowMemory - 滑动窗口记忆 🪟

特点：只保留最近K轮对话，节省Token。

📖 工作原理

只保留最近的K轮对话，旧的对话会被自动丢弃。

优点：Token消耗可控，不会无限增长
缺点：会丢失较早的对话信息
适用：长对话、只需要近期上下文的场景

🧩 一句话理解（WindowMemory）

只记住最近 K 轮。像聊天窗口一样“只看最近消息”，能把成本控制住，但早期信息可能被忘。

✅ 适用场景

用户和你聊很久，但你只需要关心最近几轮：例如问答助手、代码修改协作、实时咨询。

⚠️ 不适用场景

用户在前面说过的身份/约束/偏好必须一直有效：例如长期陪伴、个性化助手、医疗/法律等强约束场景。

🖼️ 图示：窗口滑动，旧消息被丢弃

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationBufferWindowMemory
from langchain_classic.prompts import PromptTemplate
from langchain_community.chat_models import ChatTongyi

# ⚠️ 注意：请使用 python3 运行
# python3 memory_window.py

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# WindowMemory：只保留最近 k 轮（约 2k 条 message：user+ai）
memory = ConversationBufferWindowMemory(k=2)

# 关键点：用一个“更严格”的 prompt，避免模型凭常识乱猜
# 只允许基于 {history} 回答；如果历史里没有，就说不知道。
prompt = PromptTemplate(
    input_variables=["history", "input"],
    template=(
        "你是一个严谨的助手。\n"
        "规则：只能根据【对话历史】回答；不要猜测、不要脑补。\n"
        "如果【对话历史】里没有答案，必须回答：\"我不知道\"。\n"
        "并且：除非用户明确提问，否则不要在回复里重复用户的姓名/职业等身份信息。\n\n"
        "【对话历史】\n{history}\n\n"
        "【用户问题】\n{input}"
    ),
)

conversation = ConversationChain(llm=llm, memory=memory, prompt=prompt, verbose=True)

print("第1轮:")
print(conversation.predict(input="我叫张三，是一名Python开发者"))

print("\n第2轮:")
print(conversation.predict(input="我今年30岁"))

print("\n第3轮:")
print(conversation.predict(input="我住在北京"))

print("\n第4轮:")
print(conversation.predict(input="我喜欢跑步"))

print("\n第5轮 - 测试记忆:")
print(conversation.predict(input="我叫什么名字？我的职业是什么？"))

print("\n=== memory.buffer（窗口内当前历史）===")
print(memory.buffer)

✅ 运行结果

text

第5轮 - 测试记忆:
AI: 我不知道。

=== 当前记忆 ===
Human: 我住在北京
AI: （确认收到）
Human: 我喜欢跑步
AI: （确认收到）
👉 只保留了最近2轮对话（第3、4轮），第1轮里的“张三/Python开发者”已经不在窗口里，所以第5轮必须回答“不知道”。

💡 使用官方类的优势

自动管理：不需要手动实现 trim_window 逻辑
内置优化：官方类已经优化了性能和内存使用
标准化：使用 LangChain 标准接口，易于维护
功能完整：支持 save_context、load_memory、clear 等方法

💾 WindowMemory 持久化：滑动窗口 + 磁盘存储

问题：WindowMemory 只保留最近 K 轮，但默认也在内存中，重启后丢失。
解决：同样通过 chat_memory 注入持久化的 History 实现。

关键理解：WindowMemory 的 "窗口裁剪" 是内存层面的策略，但底层历史数据可以来自磁盘/数据库。每次加载时从持久化存储读取完整历史，然后在内存中只保留最近 K 轮。

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationBufferWindowMemory
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_community.chat_models import ChatTongyi

# WindowMemory + SQLite 持久化
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# WindowMemory 使用 SQL 持久化：只保留最近 k=3 轮，但数据存数据库
memory = ConversationBufferWindowMemory(
    k=3,  # 窗口大小：只保留最近 3 轮
    chat_memory=SQLChatMessageHistory(
        session_id="user_window_123",
        connection_string="sqlite:///chat_history.db"
    ),
    return_messages=True
)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 模拟长对话（超过窗口大小）
print("=== 前5轮对话 ===")
for i, topic in enumerate(["Python", "Java", "Go", "Rust", "TypeScript"], 1):
    print(f"\n第{i}轮:")
    print(conversation.predict(input=f"我喜欢用{topic}编程"))

print("\n=== 第6轮 - 测试窗口记忆 ===")
print(conversation.predict(input="我最喜欢的前3种编程语言是什么？"))

print("\n=== 重启程序后 ===")
# 重新创建 memory（从数据库加载），仍然只保留最近 3 轮
memory2 = ConversationBufferWindowMemory(
    k=3,
    chat_memory=SQLChatMessageHistory(
        session_id="user_window_123",
        connection_string="sqlite:///chat_history.db"
    ),
    return_messages=True
)
conversation2 = ConversationChain(llm=llm, memory=memory2, verbose=True)
print(conversation2.predict(input="我最喜欢的第一种编程语言是什么？"))

# 说明：虽然数据库保存了所有历史，但 WindowMemory 只加载最近 3 轮到内存

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationBufferWindowMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_classic.prompts import PromptTemplate
from langchain_community.chat_models import ChatTongyi

# WindowMemory + Redis 持久化（适合高并发分布式场景）
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# 使用 Redis 存储，设置 1 小时过期
memory = ConversationBufferWindowMemory(
    k=2,  # ✅ 为了演示效果，这里故意设小一点：只保留最近 2 轮
    chat_memory=RedisChatMessageHistory(
        session_id="session_live_123",
        url="redis://localhost:6379/0",
        ttl=3600  # 1小时后自动清理过期数据
    ),
    return_messages=True
)

prompt = PromptTemplate(
    input_variables=["history", "input"],
    template=(
        "你是一个严谨的助手。\n"
        "规则：只能根据【对话历史】回答；不要猜测、不要脑补。\n"
        "如果【对话历史】里没有答案，必须回答：\"我不知道\"。\n\n"
        "【对话历史】\n{history}\n\n"
        "【用户问题】\n{input}"
    ),
)

conversation = ConversationChain(llm=llm, memory=memory, prompt=prompt, verbose=True)

print("=== 写入 4 轮信息（超过 k=2）===")
print(conversation.predict(input="我的名字是张三"))
print(conversation.predict(input="我的职业是Python开发者"))
print(conversation.predict(input="我住在北京"))
print(conversation.predict(input="我喜欢跑步"))

print("\n=== 提问：验证窗口裁剪（应遗忘早期信息）===")
print(conversation.predict(input="我叫什么名字？我的职业是什么？"))

print("\n=== memory.buffer（窗口内当前历史：应该只剩最近 2 轮）===")
print(memory.buffer)

print("\n=== 模拟重启脚本：同 session_id 从 Redis 继续 ===")
memory2 = ConversationBufferWindowMemory(
    k=2,
    chat_memory=RedisChatMessageHistory(
        session_id="session_live_123",
        url="redis://localhost:6379/0",
        ttl=3600,
    ),
    return_messages=True,
)
conversation2 = ConversationChain(llm=llm, memory=memory2, prompt=prompt, verbose=True)

print(conversation2.predict(input="我住在哪里？我喜欢什么？"))

# 说明：
# - Redis 保存了该 session_id 的历史，多个服务实例可共享
# - WindowMemory 每次只取最近 k 轮注入 prompt，因此早期信息会被“窗口裁剪”

3. ConversationSummaryMemory - 摘要记忆 📝

特点：使用LLM总结历史对话，压缩信息。

📖 工作原理

随着对话进行，使用LLM将历史对话总结成简短的摘要，保留关键信息。

优点：保留关键信息，Token消耗可控
缺点：需要额外调用LLM，可能丢失细节
适用：长对话、需要保留关键信息的场景

🧩 一句话理解（SummaryMemory）

把“很长的历史”压缩成一段摘要。模型每次只读“摘要 + 当前输入”，从而把 Token 控制住。

✅ 适用场景

长期对话里只需要记住“事实”：姓名、职业、偏好、长期目标。例如长期助手、学习教练、用户画像。

⚠️ 风险点

摘要是“二次创作”，可能把细节总结错。生产中要配合：摘要可追溯（保留原文）、摘要版本、以及抽样回放。

🖼️ 图示：历史被压缩成 summary

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationSummaryMemory
from langchain_community.chat_models import ChatTongyi

# ⚠️ 注意：请使用 python3 运行
# python3 memory_summary.py

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# SummaryMemory：用 LLM 把历史压缩成一段摘要文本
memory = ConversationSummaryMemory(llm=llm)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

print(conversation.predict(input="我叫张三，是一名Python开发者，今年30岁"))
print(conversation.predict(input="我住在北京，喜欢看科幻电影"))
print(conversation.predict(input="我最近在学习LangChain框架"))

print("\n第4轮 - 测试记忆:")
print(conversation.predict(input="总结一下我的基本信息"))

print("\n=== memory.buffer ===")
print(memory.buffer)

✅ 运行结果

text

第4轮 - 测试记忆:
AI: 根据我们的对话，你的基本信息如下：
- 姓名：张三
- 职业：Python开发者
- 年龄：30岁
- 居住地：北京
- 爱好：看科幻电影
- 当前学习：LangChain框架

=== 对话摘要 ===
{'history': '用户张三是一名30岁的Python开发者，居住在北京，喜欢科幻电影，目前正在学习LangChain框架。'}
👉 历史对话被总结成简短的摘要，保留了所有关键信息！

💾 SummaryMemory 持久化：摘要 + 磁盘存储

问题：SummaryMemory 生成的摘要只保存在内存中，重启后丢失，需要重新总结。
解决：通过持久化存储保存摘要，避免每次重启后重新调用 LLM 生成摘要。

关键理解：SummaryMemory 的核心是“用 LLM 把长历史压缩成短摘要”。持久化后，重启程序可以直接从磁盘加载摘要，不需要重新调用 LLM 总结，节省 Token 和延迟。

python
import os
import pathlib

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationSummaryMemory
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_community.chat_models import ChatTongyi

# SummaryMemory + SQLite 持久化
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# ✅ 关键：SQLite 路径一定要稳定（避免“重启后找不到记录”）
# sqlite:///chat_history.db 是相对路径，取决于你运行脚本时的工作目录（cwd）。
# 用绝对路径可以保证每次都读写同一个 DB 文件。
DB_PATH = pathlib.Path("./chat_history.db").resolve()
CONN_STR = f"sqlite:///{DB_PATH}"  # 三个 / 表示绝对路径

session_id = "user_summary_123"  # ✅ 重启后必须保持一致，否则会被当成新会话

history = SQLChatMessageHistory(session_id=session_id, connection_string=CONN_STR)

# SummaryMemory 使用 SQL 持久化
memory = ConversationSummaryMemory(
    llm=llm,
    chat_memory=history,
    return_messages=True
)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 第一轮：写入对话并生成摘要
print("=== 第一次运行：写入对话 ===")
print(conversation.predict(input="我叫张三，是一名Python开发者，今年30岁"))
print(conversation.predict(input="我住在北京，喜欢看科幻电影"))

print("\n=== 当前摘要 ===")
print(memory.buffer)

print("\n=== 重启程序后 ===")
# 重新创建 memory（从数据库加载已保存的 messages）
# 注意：SQLChatMessageHistory 持久化的是“对话消息”（Human/AI），不是 summary 文本本身。
print(f"DB 文件：{DB_PATH}")

# ✅ 使用相同的 session_id 才能读到之前的记录
history2 = SQLChatMessageHistory(session_id=session_id, connection_string=CONN_STR)
print(f"DB 中已存在 messages 数量：{len(history2.messages)}")

# ⚠️ 重要：ConversationSummaryMemory 不会自动加载之前的摘要！
# 它只会基于 chat_memory 中的原始消息重新生成摘要
memory2 = ConversationSummaryMemory(
    llm=llm,
    chat_memory=history2,
    return_messages=True
)

# 🔧 关键修复：如果有历史消息，先预加载摘要
if len(history2.messages) > 0:
    print("📝 基于 DB 中的消息预生成摘要...")
    # 强制让 Memory 基于已有消息生成摘要
    # 通过调用 save_context 来触发摘要生成
    for i, msg in enumerate(history2.messages):
        if i % 2 == 0:  # Human message
            memory2.save_context(
                {"input": msg.content}, 
                {"output": ""}
            )
        else:  # AI message
            memory2.save_context(
                {"input": ""}, 
                {"output": msg.content}
            )
    print("✅ 摘要预生成完成")

# 🔧 简化方案：直接让 ConversationChain 在第一次对话时重新生成摘要
conversation2 = ConversationChain(llm=llm, memory=memory2, verbose=True)

print("📝 开始对话，Memory 会自动基于 DB 中的消息生成摘要...")
print(conversation2.predict(input="总结一下我的基本信息"))

print("\n=== 自动生成的摘要 ===")
print(memory2.buffer)

# 结论：
# - ✅ messages 可以通过 SQLChatMessageHistory 跨进程持久化
# - ⚠️ SummaryMemory 的摘要需要重新生成（不会自动保存到数据库）
# - 🔑 session_id 是会话的唯一标识符，换了就读不到之前的数据
# - 💡 实际应用：要么接受重新生成摘要，要么额外存储摘要本身

✅ 运行结果

text

=== 记忆内容 ===
{'history': '用户张三是一名30岁的Python开发者，居住在北京，喜欢科幻电影，目前正在学习LangChain框架。'}
👉 历史对话被总结成简短的摘要，保留了所有关键信息！

💾 SummaryMemory 持久化：摘要 + 磁盘存储

问题：SummaryMemory 生成的摘要只保存在内存中，重启后丢失，需要重新总结。
解决：通过持久化存储保存摘要，避免每次重启后重新调用 LLM 生成摘要。

关键理解：SummaryMemory 的核心是“用 LLM 把长历史压缩成短摘要”。持久化后，重启程序可以直接从磁盘加载摘要，不需要重新调用 LLM 总结，节省 Token 和延迟。

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationSummaryMemory
from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_community.chat_models import ChatTongyi

# SummaryMemory + 文件持久化（适合需要查看/备份摘要的场景）
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# 每个用户的摘要保存到独立的 JSON 文件
memory = ConversationSummaryMemory(
    llm=llm,
    chat_memory=FileChatMessageHistory(
        file_path="summary_user_123.json"
    ),
    return_messages=True
)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 长期对话场景
print(conversation.predict(input="我是一名产品经理，负责电商APP"))
print(conversation.predict(input="我在团队里带5个人，主要做用户增长"))
print(conversation.predict(input="今年OKR是提升DAU 30%"))
print(conversation.predict(input="我们最近在做一个新功能叫直播带货"))

print("\n=== 生成的摘要 ===")
print(memory.buffer)

# 摘要已保存到 summary_user_123.json，可以随时查看、备份、审计
# 文件内容示例：
# [
#   {"type": "human", "content": "我是一名产品经理..."},
#   {"type": "ai", "content": "..."},
#   ...
# ]

4. ConversationSummaryBufferMemory - 混合记忆 🔄

特点：结合完整记忆和摘要记忆的优点。

📖 工作原理

保留最近的完整对话 + 较早对话的摘要，平衡信息完整性和Token消耗。

优点：近期信息完整，历史信息有摘要
缺点：实现较复杂，需要调用LLM
适用：需要平衡完整性和效率的场景

🧩 一句话理解（SummaryBufferMemory）

“最近消息保真 + 早期消息摘要”。既能保证最近几轮的细节不丢，又能把很早的历史压缩。

✅ 适用场景

你需要“长期记住用户关键信息”，同时又要“对最近几轮细节敏感”。例如：个人助理、项目协作、长期学习规划。

⚙️ 实战建议

设定阈值（消息条数/Token），超过阈值再摘要；同时保留： 摘要 + 最近 N 条消息，并给摘要加入“可回放”的追踪 ID。

🖼️ 图示：summary + 最近消息一起进 prompt

⚠️ 常见报错：transformers 下载 gpt2 tokenizer 超时

现象：运行 SummaryBufferMemory 时报 httpx.ConnectTimeout: timed out，原因是 LangChain 需要下载 gpt2 tokenizer 来计算 token 数，但国内访问 HuggingFace 不稳定。

🔧 解决方案（推荐顺序）

1) 临时解决（运行时设置镜像）

text

# 设置国内镜像并运行
HF_ENDPOINT=https://hf-mirror.com python3 memory_summary_buffer.py

2) 预缓存 tokenizer（只需一次）

python
# 先缓存一次，以后就不需要网络了
HF_ENDPOINT=https://hf-mirror.com python -c "from transformers import GPT2TokenizerFast; GPT2TokenizerFast.from_pretrained('gpt2'); print('gpt2 tokenizer cached')"

3) 永久解决（环境变量）

text

# 添加到 ~/.zshrc 或 ~/.bashrc
echo 'export HF_ENDPOINT=https://hf-mirror.com' >> ~/.zshrc
source ~/.zshrc

# 之后直接运行即可
python3 memory_summary_buffer.py

✅ 验证是否成功：第一次运行后会显示 ✅ gpt2 tokenizer cached successfully，之后即使不设置镜像也能正常运行。

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationSummaryBufferMemory
from langchain_community.chat_models import ChatTongyi

# ⚠️ 注意：请使用 python3 运行
# python3 memory_summary_buffer.py

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# SummaryBufferMemory：超过 max_token_limit 后，把早期历史压成 summary，并保留最近消息原文
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=100,
)

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True,
)

print("第1轮:")
print(conversation.predict(input="我叫张三，是一名Python开发者"))

print("\n第2轮:")
print(conversation.predict(input="我今年30岁，住在北京"))

print("\n第3轮:")
print(conversation.predict(input="我喜欢看科幻电影，最近在学习LangChain"))

print("\n第4轮:")
print(conversation.predict(input="我的职业是什么？我住哪里？"))

print("\n=== memory.buffer ===")
print(memory.buffer)

✅ 运行结果

text

=== 记忆内容 ===
{'history': 'System: 前3轮对话的摘要...\nHuman: 这是第4轮对话\nAI: ...\nHuman: 这是第5轮对话\nAI: ...'}
👉 较早的对话被总结，最近的对话保持完整！

💾 SummaryBufferMemory 持久化：混合策略 + 磁盘存储

问题：SummaryBufferMemory 同时维护“摘要 + 最近消息”，数据量更大，更需要持久化。
解决：通过持久化存储保存混合记忆，确保重启后既能恢复摘要也能恢复最近完整消息。

关键理解：SummaryBufferMemory 是最复杂的记忆类型，它同时包含： 1）早期历史的摘要；2）最近 N 轮的完整原文。
持久化后重启程序，这两部分都能恢复，无需重新生成摘要或丢失最近消息。

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationSummaryBufferMemory
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_community.chat_models import ChatTongyi

# SummaryBufferMemory + SQLite 持久化（最完整的生产方案）
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# SummaryBufferMemory 使用 SQL 持久化
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=100,  # 超过100 tokens时，早期历史会被总结
    chat_memory=SQLChatMessageHistory(
        session_id="user_summary_buffer_123",
        connection_string="sqlite:///chat_history.db"
    ),
    return_messages=True
)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 第一阶段：写入足够多的对话触发摘要机制
print("=== 第一阶段：写入长对话（触发摘要） ===")
print(conversation.predict(input="我叫张三，是一名Python开发者"))
print(conversation.predict(input="我今年30岁，住在北京，工作5年了"))
print(conversation.predict(input="我喜欢看科幻电影，跑步，和编程"))
print(conversation.predict(input="最近在做一个AI项目，用LangChain"))

print("\n=== 当前记忆状态（摘要+最近消息） ===")
print(memory.buffer)

print("\n=== 重启程序后 ===")
# 重新创建 memory（从数据库加载已保存的摘要+消息）
memory2 = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=100,
    chat_memory=SQLChatMessageHistory(
        session_id="user_summary_buffer_123",
        connection_string="sqlite:///chat_history.db"
    ),
    return_messages=True
)
conversation2 = ConversationChain(llm=llm, memory=memory2, verbose=True)

# 验证：既能回答早期事实（来自摘要），也能回答近期细节（来自完整消息）
print(conversation2.predict(input="我叫什么名字？我的职业是什么？"))
print(conversation2.predict(input="我最近在做什麼项目？"))

print("\n=== 加载后的记忆状态 ===")
print(memory2.buffer)

# 说明：数据库同时保存了摘要和最近消息，重启后完整恢复

python
import os

from langchain_classic.chains import ConversationChain
from langchain_classic.memory import ConversationSummaryBufferMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_community.chat_models import ChatTongyi

# SummaryBufferMemory + Redis 持久化（高性能分布式方案）
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

# 使用 Redis 存储，设置合适的过期时间
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=150,
    chat_memory=RedisChatMessageHistory(
        session_id="enterprise_user_001",
        url="redis://localhost:6379/0",
        ttl=86400  # 24小时过期，适合企业级会话
    ),
    return_messages=True
)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 企业级场景：客户咨询历史
print(conversation.predict(input="我是某科技公司的CTO，正在评估AI方案"))
print(conversation.predict(input="我们有50人的技术团队"))
print(conversation.predict(input="主要关注数据安全和成本控制"))
print(conversation.predict(input="希望能快速落地，3个月内见效果"))

# 多实例共享：同一个 session_id 可以在多个服务实例间共享记忆
# 适合：微服务架构、负载均衡、客服系统

📊 记忆类型对比

记忆类型	保存方式	Token消耗	适用场景
BufferMemory	完整保存所有对话	高（随对话增长）	短对话
WindowMemory	只保留最近K轮	低（固定大小）	长对话，只需近期上下文
SummaryMemory	LLM总结历史	中（需要总结）	长对话，需保留关键信息
SummaryBufferMemory	近期完整+历史摘要	中（平衡）	需要平衡的场景

💡 实战技巧

1. 手动管理记忆

python
from langchain_community.chat_message_histories import ChatMessageHistory

history = ChatMessageHistory()

# 手动添加消息
history.add_user_message("你好")
history.add_ai_message("你好！有什么可以帮你的？")

# 查看消息
print(history.messages)

# 清空记忆
history.clear()

2. 持久化记忆（保存到文件）

python
import os

from dotenv import load_dotenv
from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

# 加载环境变量（从 .env 中读取 DASHSCOPE_API_KEY）
load_dotenv()

api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("请先在 .env 中配置 DASHSCOPE_API_KEY")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个有帮助的助手。"),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)

chain = prompt | llm

def get_file_history(session_id: str) -> FileChatMessageHistory:
    # 每个 session_id 绑定一个历史文件
    return FileChatMessageHistory(f"{session_id}_history.json")

conversation = RunnableWithMessageHistory(
    chain,
    get_file_history,
    input_messages_key="input",
    history_messages_key="history",
)

session_id = "demo_user"

print("=== 第1次运行：写入对话并持久化到文件 ===")
conversation.invoke(
    {"input": "我叫李雷，今年28岁，在杭州做前端工程师"},
    config={"configurable": {"session_id": session_id}},
)
conversation.invoke(
    {"input": "我喜欢跑步和看电影"},
    config={"configurable": {"session_id": session_id}},
)

print("\n=== 模拟下次启动：重新加载同一个文件 ===")
r = conversation.invoke(
    {"input": "请总结一下我的基本信息"},
    config={"configurable": {"session_id": session_id}},
)
print("AI:", r.content)

3. 获取记忆变量

python
import os

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)
history = ChatMessageHistory()

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个有帮助的助手。"),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)

def invoke(user_input: str) -> str:
    messages = prompt.format_messages(history=history.messages, input=user_input)
    ai = llm.invoke(messages)
    history.add_user_message(user_input)
    history.add_ai_message(ai.content)
    return ai.content

invoke("我叫韩梅梅，是一名数据分析师")
invoke("我擅长用Python做数据可视化")

print("=== 字符串形式的 history（自己拼接）===")
history_str = "\n".join([f"[{m.type}] {m.content}" for m in history.messages])
print(type(history_str))
print(history_str)

            print("\n=== 消息列表形式的 history ===")
            print(type(history.messages))
            for msg in history.messages:
                print(f"[{msg.type}] {msg.content}")

🧪 RunnableWithMessageHistory + Window 裁剪

演示如何用 RunnableWithMessageHistory 维护会话历史，并用 K 实现只保留最近几轮（Window）效果。

python
import os

from langchain_community.chat_models import ChatTongyi
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

# os.environ["DASHSCOPE_API_KEY"] = "your-api-key"

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个有帮助的助手。"),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)

chain = prompt | llm

store = {}

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

conversation = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
)

session_id = "user_2"
K = 2  # 只保留最近 K 轮（约 2K 条消息：user+ai）

def trim_window(history: ChatMessageHistory, k: int) -> None:
    keep = k * 2
    if len(history.messages) > keep:
        history.messages = history.messages[-keep:]

def ask(text: str) -> str:
    r = conversation.invoke(
        {"input": text},
        config={"configurable": {"session_id": session_id}},
    )
    trim_window(store[session_id], K)
    return r.content

print("第1轮:")
ask("我叫张三")

print("\n第2轮:")
ask("我是Python开发者")

print("\n第3轮:")
ask("我今年30岁")

print("\n第4轮:")
ask("我住在北京")

print("\n第5轮 - 测试记忆:")
response = ask("我叫什么名字？我的职业是什么？")
print("AI:", response)

print("\n=== 当前记忆（只保留最近 K 轮）===")
print(store[session_id].messages)

🧪 手写 SummaryBuffer（summary + 最近消息）

用一个最小可运行示例手写“摘要 + 最近消息”的混合记忆：超过阈值就把更早的消息压缩进 summary，并只保留最近 N 条原文消息。

python
import os

from langchain_community.chat_models import ChatTongyi
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

#os.environ["DASHSCOPE_API_KEY"] = "your-api-key"

llm = ChatTongyi(model="qwen-turbo", temperature=0.7)
history = ChatMessageHistory()
summary = ""

chat_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个有帮助的助手。历史摘要：{summary}"),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)

summary_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "把对话总结成不超过80字摘要，只保留关键事实："),
        MessagesPlaceholder("history"),
    ]
)

# 关键事实的作用：通过只保留最重要的信息，既节省了 token 成本，又确保了关键信息不会丢失

def maybe_summarize(max_messages: int = 8) -> None:
    global summary
    # 超过阈值：把较早消息压缩进 summary，只保留最近 max_messages 条原文
    if len(history.messages) <= max_messages:
        return
    s_messages = summary_prompt.format_messages(history=history.messages)
    summary = llm.invoke(s_messages).content
    history.messages = history.messages[-max_messages:]

for i in range(5):
    user_input = f"这是第{i+1}轮对话"
    messages = chat_prompt.format_messages(summary=summary, history=history.messages, input=user_input)
    ai = llm.invoke(messages)
    history.add_user_message(user_input)
    history.add_ai_message(ai.content)
    maybe_summarize(max_messages=6)
    print(f"第{i+1}轮: {ai.content}\n")

print("=== 记忆内容（summary + 最近消息）===")
print("summary:", summary)
print("history messages:", history.messages)

✅ 课后习题与常见面试题

这章的目标不是“记住名字”，而是让你能在真实业务里选对记忆方案，并能把记忆做成可控、可观测、可扩展的工程能力。

📌 基础练习题（概念 + 选型）

1. Buffer / Window / Summary / SummaryBuffer 的核心差异是什么？

请从“信息完整性、成本、风险（遗忘/误总结）”三方面对比，并举 1 个你项目里的典型场景选择其一。

2. 为什么“多用户/多会话”必须引入 session_id？

如果不区分 session，会出现什么线上事故？你如何在后端接口里设计它（例如 header/cookie/body）？

💻 代码题（复制即可运行）

建议：在一个不包含本项目 static/langchain 的路径下运行示例，避免目录名遮蔽 pip 安装的包。

C1. 【编程题】实现“多会话隔离”：同一个服务进程里同时维护 2 个用户的记忆

要求：用 RunnableWithMessageHistory；user_a 记住“张三/开发者”，user_b 记住“李四/产品经理”；互相提问时不能串台。

C2. 【编程题】实现“窗口记忆”：只保留最近 K 轮（并证明它会遗忘）

要求：写一个 trim_window()，每轮对话后裁剪 history.messages；让模型在第 5 轮问“我叫什么名字？”时答不出来。

C3. 【编程题】实现“文件持久化记忆”：重启程序后还能记得之前信息

要求：用 FileChatMessageHistory；第一次运行写入两轮消息；第二次运行用同一个 session_id 继续提问并能回答。

💼 常见面试题（偏工程）

1.【工程题】你如何评估“记忆”带来的成本，并让它可观测？

请覆盖：token 使用、延迟、失败率、session 维度统计、采样回放。

2.【架构题】如果要把记忆存到 Redis / 数据库，你会怎么设计？

请覆盖：数据结构、过期策略（TTL）、并发写入、合规（隐私/脱敏）。

📝 本章小结

🎓 你学到了什么？

记忆的重要性：LLM无状态，需要记忆维护上下文
BufferMemory：保存完整历史，适合短对话
WindowMemory：滑动窗口，只保留最近K轮
SummaryMemory：LLM总结历史，节省Token
SummaryBufferMemory：混合方案，平衡效率和完整性
实战技巧：手动管理、持久化、获取记忆变量

🚀 下一步学习

掌握了记忆管理后，下一章我们将学习Prompt模板，让提示词更加灵活和可复用！

第2章: 记忆管理：让聊天助手多轮不失忆

🤔 为什么需要记忆？

❌ 没有记忆的问题

✅ 有记忆的效果

💡 记忆的核心作用

📚 记忆类型详解

✅ 环境准备 & 如何运行本页代码（强烈建议先看）

1) 创建并激活虚拟环境（推荐）

2) 安装依赖（用同一个解释器的 pip）

3) 验证导入是否正常（先验证再跑示例）

4) 运行示例（建议统一用 python3）

5) IDE 里怎么选解释器（非常关键）

6) Python 3.12 + pcapy 安装报错怎么办？（不降版本）

7) 常见报错：ConversationSummaryBufferMemory 运行时报 Token/Tokenizer 问题

1. MessagesPlaceholder + RunnableWithMessageHistory（LCEL 正统做法）💾

📖 工作原理

🔍 MessagesPlaceholder 和 "history" 深度解析

🤔 为什么要用 MessagesPlaceholder？

🔑 "history" 这个名字是随便起的吗？

� 三处统一的 "history"

🔄 实际运行过程演示

💡 可以换成其他名字吗？

🧩 ChatMessageHistory：所有类型与用法总结

2) 常见实现类型（选型速查）

📦 ConversationBufferMemory（传统 Chain API）

📦 ConversationBufferMemory 是什么？

🧩 一句话理解（BufferMemory）

💾 BufferMemory 持久化：保存到磁盘/数据库

2. ConversationBufferWindowMemory - 滑动窗口记忆 🪟

📖 工作原理

🧩 一句话理解（WindowMemory）

✅ 运行结果

💡 使用官方类的优势

💾 WindowMemory 持久化：滑动窗口 + 磁盘存储

3. ConversationSummaryMemory - 摘要记忆 📝

📖 工作原理

🧩 一句话理解（SummaryMemory）

✅ 运行结果

💾 SummaryMemory 持久化：摘要 + 磁盘存储

✅ 运行结果

💾 SummaryMemory 持久化：摘要 + 磁盘存储

4. ConversationSummaryBufferMemory - 混合记忆 🔄

📖 工作原理

🧩 一句话理解（SummaryBufferMemory）

⚠️ 常见报错：transformers 下载 gpt2 tokenizer 超时

✅ 运行结果

💾 SummaryBufferMemory 持久化：混合策略 + 磁盘存储

📊 记忆类型对比

💡 实战技巧

1. 手动管理记忆

2. 持久化记忆（保存到文件）

3. 获取记忆变量

🧪 RunnableWithMessageHistory + Window 裁剪

🧪 手写 SummaryBuffer（summary + 最近消息）

✅ 课后习题与常见面试题

📌 基础练习题（概念 + 选型）

💻 代码题（复制即可运行）

💼 常见面试题（偏工程）

📝 本章小结

🎓 你学到了什么？

🚀 下一步学习