LangChain 学习笔记 · Day 5-6 Memory（短期与长期记忆）

🎯 学习目标

1. 理解 Memory 在 LangChain 中的作用
2. 掌握三种记忆策略：Buffer、Window、Summary
3. 学会用 RunnableWithMessageHistory + LCEL 实现对话机器人
4. 通过实验验证：机器人能记住短期上下文 & 长期关键信息

🔹 1. 什么是 Memory？

• Memory 是 LangChain 的记忆模块，用来存储和回顾对话历史。
• 无记忆：模型每次只知道当前输入，完全无上下文。
• 有记忆：模型能回顾之前的对话，实现“连贯交流”。

🔹 2. ConversationBufferMemory（完整上下文）

• 保存完整的对话历史。
• 缺点：历史过长时，Prompt 会越来越大，导致 token 消耗过高。
• 适用场景：短对话。

代码示例

# day5_memory_buffer.py
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_core.chat_history import InMemoryChatMessageHistory

llm = ChatOpenAI(model="kimi-k2-0711-preview", temperature=0.7)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好而简洁的中文助手。"),
    MessagesPlaceholder("history"),
    ("human", "{input}")
])

chain = prompt | llm | StrOutputParser()

# 不同 session_id 有不同的会话历史
hist_store = {}
def get_history(session_id: str):
    if session_id not in hist_store:
        hist_store[session_id] = InMemoryChatMessageHistory()
    return hist_store[session_id]

chat = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
    output_messages_key="output",
)

if __name__ == "__main__":
    cfg = {"configurable": {"session_id": "user-001"}}
    print(chat.invoke({"input": "你好，我叫小明。"}, cfg))
    print(chat.invoke({"input": "你记得我叫什么吗？"}, cfg))
    print(chat.invoke({"input": "请用一句话介绍我。"}, cfg))

🔹 3. ConversationBufferWindowMemory（滑窗）

• 只保存最近 k 轮对话。
• 好处：避免历史无限膨胀，节省 token。
• 适用场景：关注“最近上下文”，不关心很久之前的内容。

代码示例

# day6_memory_window.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

llm = ChatOpenAI(model="kimi-k2-0711-preview", temperature=0.7)

K = 6  # 只保留最近6条消息
def last_k_messages(inputs: dict):
    hist = inputs.get("history", [])
    return {"history": hist[-K:], "input": inputs["input"]}

trim = RunnableLambda(last_k_messages)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你只会参考最近几轮对话。"),
    MessagesPlaceholder("history"),
    ("human", "{input}")
])

base = prompt | llm | StrOutputParser()
pipeline = trim | base

hist_store = {}
def get_history(session_id: str):
    if session_id not in hist_store:
        hist_store[session_id] = InMemoryChatMessageHistory()
    return hist_store[session_id]

chat = RunnableWithMessageHistory(
    pipeline,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
    output_messages_key="output",
)

if __name__ == "__main__":
    cfg = {"configurable": {"session_id": "user-002"}}
    print(chat.invoke({"input": "记住：我叫小明，住在上海。"}, cfg))
    for i in range(8):
        print(chat.invoke({"input": f"第{i+1}轮闲聊"}, cfg))
    print(chat.invoke({"input": "我住在哪？"}, cfg))  # 如果超出窗口范围，模型可能忘记

🔹 4. ConversationSummaryMemory（摘要）

• 自动将历史归纳成摘要，不断更新。
• 保留关键信息，但避免 token 爆炸。
• 适用场景：长对话，需要记住关键信息（职业、兴趣、目标…）。

代码示例

# day6_memory_summary_fixed.py
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_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import SystemMessage

llm = ChatOpenAI(model="kimi-k2-0711-preview", temperature=0.3)

# 小链：把历史对话总结成摘要
summarize_prompt = ChatPromptTemplate.from_messages([
    ("system", "请把以下对话历史总结为不超过120字的摘要，保留姓名、职业、兴趣、城市等长期信息。"),
    ("human", "{history_text}")
])
summarizer = summarize_prompt | llm | StrOutputParser()

hist_store = {}
def get_history(session_id: str):
    if session_id not in hist_store:
        hist_store[session_id] = InMemoryChatMessageHistory()
    return hist_store[session_id]

SUMMARIZE_THRESHOLD = 12
def maybe_rollup_summary(session_id: str):
    history = get_history(session_id)
    if len(history.messages) >= SUMMARIZE_THRESHOLD:
        text = "\n".join(f"{m.type.upper()}: {getattr(m,'content','')}" for m in history.messages)
        summary_text = summarizer.invoke({"history_text": text}).strip()
        history.clear()
        history.add_message(SystemMessage(content=f"【长期摘要】{summary_text}"))

prompt = ChatPromptTemplate.from_messages([
    ("system", "你会利用【长期摘要】中的信息来回答问题。"),
    MessagesPlaceholder("history"),
    ("human", "{input}")
])

chat_runnable = prompt | llm | StrOutputParser()

chat = RunnableWithMessageHistory(
    chat_runnable,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
    output_messages_key="output",
)

if __name__ == "__main__":
    cfg = {"configurable": {"session_id": "user-003"}}
    print(chat.invoke({"input": "你好，我叫小明，在上海做前端开发，爱好是篮球和摄影。"}, cfg))
    for i in range(15):
        print(chat.invoke({"input": f"随便聊聊第{i+1}轮……"}, cfg))
        maybe_rollup_summary(cfg["configurable"]["session_id"])
    print(chat.invoke({"input": "我之前说过我的职业和城市是什么？"}, cfg))

🔹 5. 三种记忆的对比

🔹 6. 今日练习

1. 用 Buffer 写一个短对话机器人，问它“我叫什么”，看是否记住。
2. 改成 Window，尝试连续 10 轮对话，再问早期信息，观察是否遗忘。
3. 改成 Summary，连续 20 轮对话，再问“我的职业/城市是什么”，验证是否从【长期摘要】中回忆。

🔹 7. 总结

• LangChain 的 Memory 是对话应用的核心能力。
• 推荐用 RunnableWithMessageHistory + InMemoryChatMessageHistory（或 Redis/DB）。
• 根据需求选择记忆策略：
  • 短对话 → Buffer
  • 关注最近 → Window
  • 长对话 → Summary
• 迁移要点：旧的 ConversationChain、ConversationBufferMemory 仍可用，但未来推荐全用 RunnableWithMessageHistory。