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

🤔 为什么需要记忆？

❌ 没有记忆的问题

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

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

✅ 有记忆的效果

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

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

💡 记忆的核心作用

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

📚 记忆类型详解

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

1. ConversationBufferMemory - 完整记忆 💾

特点：保存所有对话历史，完整但可能很长。

📖 工作原理

把所有对话消息都存储在内存中，每次调用时都把完整历史传给模型。

优点：信息完整，不会丢失任何上下文
缺点：对话越长，Token消耗越大
适用：短对话、需要完整上下文的场景

🧩 一句话理解（BufferMemory）

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

✅ 适用场景

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

⚠️ 不适用场景

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

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

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 中预留 history 占位符（由 RunnableWithMessageHistory 自动注入）
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个有帮助的助手。"),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)

chain = prompt | llm

# 简单的内存存储：一个 session_id 对应一个 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("=== 记忆内容（messages 列表）===")
print(store[session_id].messages)

✅ 运行结果

=== 第1轮 ===
> Prompt after formatting:
Current conversation:

Human: 我叫张三，是一名Python开发者
AI: 你好张三！很高兴认识你。作为Python开发者，你一定对编程很有热情...

=== 第2轮 ===
> Prompt after formatting:
Current conversation:
Human: 我叫张三，是一名Python开发者
AI: 你好张三！很高兴认识你...
Human: 我的职业是什么？
AI: 你是一名Python开发者。

=== 第3轮 ===
AI: 你叫张三。

=== 记忆内容 ===
{'history': 'Human: 我叫张三，是一名Python开发者\nAI: 你好张三！...\nHuman: 我的职业是什么？\nAI: 你是一名Python开发者。\nHuman: 我叫什么名字？\nAI: 你叫张三。'}
👉 所有对话都被完整保存！

2. ConversationBufferWindowMemory - 滑动窗口记忆 🪟

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

📖 工作原理

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

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

🧩 一句话理解（WindowMemory）

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

✅ 适用场景

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

⚠️ 不适用场景

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

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

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)

✅ 运行结果

第5轮 - 测试记忆:
AI: 抱歉，我不记得你的名字和职业了。我只记得你今年30岁，住在北京。

=== 当前记忆 ===
{'history': 'Human: 我今年30岁\nAI: ...\nHuman: 我住在北京\nAI: ...'}
👉 只保留了最近2轮对话（第3、4轮），前面的信息（名字、职业）被丢弃了！

3. ConversationSummaryMemory - 摘要记忆 📝

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

📖 工作原理

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

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

🧩 一句话理解（SummaryMemory）

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

✅ 适用场景

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

⚠️ 风险点

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

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

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}"),
    ]
)

# 摘要器：把当前 history 总结成短摘要（用于压缩早期信息）
summary_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "请把下面对话总结成不超过80字的摘要，只保留用户关键信息："),
        MessagesPlaceholder("history"),
    ]
)

summary = ""

def chat(user_input: str) -> str:
    global summary
    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)

    # 每轮更新摘要（真实业务可降低频率，比如每 N 轮更新一次）
    s_messages = summary_prompt.format_messages(history=history.messages)
    summary = llm.invoke(s_messages).content
    return ai.content

print("第1轮:")
chat("我叫张三，是一名Python开发者，今年30岁")

print("\n第2轮:")
chat("我住在北京，喜欢看科幻电影")

print("\n第3轮:")
chat("我最近在学习LangChain框架")

print("\n第4轮 - 测试记忆:")
response = chat("总结一下我的基本信息")
print("AI:", response)

print("\n=== 对话摘要（summary）===")
print(summary)

✅ 运行结果

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

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

4. ConversationSummaryBufferMemory - 混合记忆 🔄

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

📖 工作原理

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

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

🧩 一句话理解（SummaryBufferMemory）

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

✅ 适用场景

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

⚙️ 实战建议

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

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

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"),
    ]
)

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)

✅ 运行结果

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

📊 记忆类型对比

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

💡 实战技巧

1. 手动管理记忆

from langchain_community.chat_message_histories import ChatMessageHistory

history = ChatMessageHistory()

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

# 查看消息
print(history.messages)

# 清空记忆
history.clear()

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

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. 获取记忆变量

import os

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_community.chat_models import ChatTongyi
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()

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}")

✅ 课后习题与常见面试题

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

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

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模板，让提示词更加灵活和可复用！