维护对话上下文
LLM本身是无状态的!每次调用都是独立的,不记得之前说过什么。
# 没有记忆的对话
llm.invoke("我叫张三") # AI: 你好张三!
llm.invoke("我叫什么名字?") # AI: 抱歉,我不知道你的名字
# 👉 AI完全忘记了刚才的对话!记忆机制让AI记住对话历史,实现真正的多轮对话。
# 有记忆的对话
conversation.predict(input="我叫张三") # AI: 你好张三!
conversation.predict(input="我叫什么名字?") # AI: 你叫张三
# 👉 AI记住了之前的对话!LangChain提供了多种记忆类型,适应不同的使用场景。
本页所有示例代码都可以复制到 memory_x.py 里运行。最常见的问题不是代码写错,而是:
IDE 选的 Python 解释器 ≠ 你安装依赖的 Python 环境。
# 在你的项目根目录执行
python3 -m venv .venv
# macOS / Linux
source .venv/bin/activate
# Windows PowerShell
# .venv\Scripts\Activate.ps1# 关键点:永远用 "python -m pip",避免 pip 装到另一个 Python 上
python -m pip install -U pip setuptools wheel
# LangChain + Tongyi 依赖(本页示例用到)
python -m pip install langchain langchain-community langchain-corepython -c "from langchain.chains import ConversationChain; from langchain.memory import ConversationBufferMemory; print('import OK')"# 例如:运行窗口记忆示例
python3 memory_window.py.venv。
如果你看到 from langchain... 下面有红线,通常就是解释器没选对。
Failed to build pcapy / distutils 相关错误,这是 pcapy 自身不兼容 Python 3.12。
不降 Python 版本的做法是换成维护分支 pcapy-ng:
python -m pip install pcapy-ngimport pcapy,安装 pcapy-ng 后大多数场景仍然可以继续用;
若遇到导入差异,再根据项目实际把 import 调整为对应模块名。
特点:把对话历史作为消息列表自动注入到 Prompt 中,适合 LCEL / Runnable 风格的链式调用。
通过 RunnableWithMessageHistory 按 session_id 取出历史消息列表,并填充到 MessagesPlaceholder("history") 这个占位符位置。
prompt | llm 这种 Runnable 风格写链把所有历史原封不动塞进 prompt。好处是“信息不丢”,代价是“越聊越贵、越聊越慢”。
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
os.environ["DASHSCOPE_API_KEY"] = "your-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)核心问题:LLM 每次调用都是独立的,不记得之前说过什么。我们需要把历史对话塞进 Prompt 里。
MessagesPlaceholder 的作用:专门用来插入消息列表,确保格式正确。
不是随便起的!"history" 必须在三个地方保持一致:
可以!但必须三处统一。比如换成 "chat_history":
总结:MessagesPlaceholder 就是预留位置,"history" 是数据键名,两者配合实现动态历史注入。
ConversationBufferMemory 是 LangChain 早期/传统(Chain API)里的记忆组件:它会把所有历史对话拼成一段文本(memory.buffer),
并在每次调用 ConversationChain 时自动把这段历史注入到 prompt 里。
什么时候用它?
如果你在写 ConversationChain(而不是 prompt | llm 这种 LCEL),用 BufferMemory 最省事。
但它也有代价:对话越长,token 越多(需要时可换 Window/Summary)。
import os
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_models import ChatTongyi
# ⚠️ 注意:请使用 python3 运行
# python3 memory_buffer.py
os.environ["DASHSCOPE_API_KEY"] = "your-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)=== 第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: 你叫张三。'}
👉 所有对话都被完整保存!特点:只保留最近K轮对话,节省Token。
只保留最近的K轮对话,旧的对话会被自动丢弃。
只记住最近 K 轮。像聊天窗口一样“只看最近消息”,能把成本控制住,但早期信息可能被忘。
import os
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import PromptTemplate
from langchain_community.chat_models import ChatTongyi
# ⚠️ 注意:请使用 python3 运行
# python3 memory_window.py
os.environ["DASHSCOPE_API_KEY"] = "your-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)第5轮 - 测试记忆:
AI: 我不知道。
=== 当前记忆 ===
Human: 我住在北京
AI: (确认收到)
Human: 我喜欢跑步
AI: (确认收到)
👉 只保留了最近2轮对话(第3、4轮),第1轮里的“张三/Python开发者”已经不在窗口里,所以第5轮必须回答“不知道”。特点:使用LLM总结历史对话,压缩信息。
随着对话进行,使用LLM将历史对话总结成简短的摘要,保留关键信息。
把“很长的历史”压缩成一段摘要。模型每次只读“摘要 + 当前输入”,从而把 Token 控制住。
import os
from langchain.chains import ConversationChain
from langchain.memory import ConversationSummaryMemory
from langchain_community.chat_models import ChatTongyi
# ⚠️ 注意:请使用 python3 运行
# python3 memory_summary.py
os.environ["DASHSCOPE_API_KEY"] = "your-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)第4轮 - 测试记忆:
AI: 根据我们的对话,你的基本信息如下:
- 姓名:张三
- 职业:Python开发者
- 年龄:30岁
- 居住地:北京
- 爱好:看科幻电影
- 当前学习:LangChain框架
=== 对话摘要 ===
{'history': '用户张三是一名30岁的Python开发者,居住在北京,喜欢科幻电影,目前正在学习LangChain框架。'}
👉 历史对话被总结成简短的摘要,保留了所有关键信息!特点:结合完整记忆和摘要记忆的优点。
保留最近的完整对话 + 较早对话的摘要,平衡信息完整性和Token消耗。
“最近消息保真 + 早期消息摘要”。既能保证最近几轮的细节不丢,又能把很早的历史压缩。
import os
from langchain.chains import ConversationChain
from langchain.memory import ConversationSummaryBufferMemory
from langchain_community.chat_models import ChatTongyi
# ⚠️ 注意:请使用 python3 运行
# python3 memory_summary_buffer.py
os.environ["DASHSCOPE_API_KEY"] = "your-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)=== 记忆内容 ===
{'history': 'System: 前3轮对话的摘要...\nHuman: 这是第4轮对话\nAI: ...\nHuman: 这是第5轮对话\nAI: ...'}
👉 较早的对话被总结,最近的对话保持完整!from langchain_community.chat_message_histories import ChatMessageHistory
history = ChatMessageHistory()
# 手动添加消息
history.add_user_message("你好")
history.add_ai_message("你好!有什么可以帮你的?")
# 查看消息
print(history.messages)
# 清空记忆
history.clear()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)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)、并发写入、合规(隐私/脱敏)。
掌握了记忆管理后,下一章我们将学习Prompt模板, 让提示词更加灵活和可复用!