LangChain-Memory对话历史管理
LLM 没有记忆。每次调用都是独立的:接收一段文本,处理完,输出一段文本,然后这次调用就结束了。下次调用,它对上次发生了什么一无所知。没有状态,没有持久化,每次都是白板。
LangChain Memory:对话历史管理
LLM 没有记忆。每次调用都是独立的:接收一段文本,处理完,输出一段文本,然后这次调用就结束了。下次调用,它对上次发生了什么一无所知。没有状态,没有持久化,每次都是白板。
以下这个测试展示了这个特性:
用户:我叫小明。
Bot:你好,小明!
用户:我叫什么?
Bot:我不知道你叫什么。
这不是 bug,这是 LLM 的工作原理。在 ChatGPT 里看到的"连续对话",是因为 OpenAI 在后台把每轮对话的消息都存起来,每次发送时把历史记录一起带上。所谓"记住上下文",其实是每次都传了一大段历史。
Memory 的本质,不是让模型"记住"什么,而是每次调用时把历史记录塞进 messages 列表。
1. 最简单的实现:手动维护消息列表
最原始的实现方式如下:
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
llm = ChatOpenAI(
model="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com",
)
# 维护一个消息列表
messages = [
SystemMessage(content="你是一个友好的助手。")
]
def chat(user_input: str) -> str:
# 把用户消息加进列表
messages.append(HumanMessage(content=user_input))
# 把整个历史传给模型
response = llm.invoke(messages)
# 把模型回复也加进列表
messages.append(AIMessage(content=response.content))
return response.content
# 测试
print(chat("我叫小明。"))
print(chat("我叫什么?"))
运行这段代码,第二个问题就能正确回答了。逻辑很简单:每次都把 messages 全部传给模型,模型看到完整上下文,自然知道你叫小明。
这就是 Memory 的全部秘密。剩下的问题是:怎么管理好这个列表,以及当历史太长时怎么处理。
2. ChatMessageHistory:LangChain 的消息历史管理
手动维护列表容易出错,LangChain 提供了 ChatMessageHistory 来管理:
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage
history = ChatMessageHistory()
# 添加消息
history.add_user_message("我叫小明。")
history.add_ai_message("你好,小明!")
history.add_user_message("我叫什么?")
# 查看历史
for msg in history.messages:
role = "用户" if isinstance(msg, HumanMessage) else "助手"
print(f"{role}: {msg.content}")
ChatMessageHistory 本质是一个消息列表的包装,提供了 add_user_message、add_ai_message、clear 等方法,用起来更清晰。
3. ConversationBufferMemory vs ConversationSummaryMemory
LangChain Memory 类型对比 — 存储方式、Token 消耗与适用场景
历史越来越长,token 消耗就越来越大。一次对话聊了 50 轮,每次都要把 50 轮历史带上,费用会很夸张。LangChain 提供了两种策略来处理这个问题。
ConversationBufferMemory:保留全部历史
最简单,全部保留,原汁原味。适合短对话,或者 token 预算充裕的场景。问题是上下文窗口(LLM 单次能处理的最大文本量)有上限,超过了就会报错或者截断(各模型具体上限以官方文档为准,通常在 128K token 左右)。
ConversationSummaryMemory:用 LLM 压缩历史
超过一定长度后,用一个 LLM 调用把旧历史压缩成摘要,再把摘要和最近几轮历史一起带上。
比如 10 轮对话:
- 原始方式:带上全部 10 轮,假设 3000 tokens
- Summary 方式:把前 7 轮压缩成摘要(500 tokens),加上最近 3 轮(900 tokens),共 1400 tokens
代价是需要额外一次 LLM 调用来生成摘要,有延迟,也有摘要信息损失。适合长对话场景,比如一个客服 Bot 要处理几十轮的复杂咨询。
4. RunnableWithMessageHistory:新版推荐方式
LangChain 0.2 之后,推荐用 RunnableWithMessageHistory 而不是老版的 Memory 类。它把历史管理和 LCEL 链结合起来,更清晰。
截至 2026 年,对于新项目(特别是涉及 Agent 的场景),LangChain 官方更推荐使用 LangGraph 内置的 checkpointer 来管理对话状态(见 LangGraph Human-in-the-loop 文章)。
RunnableWithMessageHistory仍然完全可用,适合简单的多轮对话场景;如果项目已经在用 LangGraph,优先用 checkpointer 方案。
下面是一个完整可运行的有记忆多轮对话示例:
import os
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_community.chat_message_histories import ChatMessageHistory
llm = ChatOpenAI(
model="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com",
temperature=0.7,
)
# Prompt 模板,留一个 MessagesPlaceholder 给历史记录
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个友好的助手,记得用户说过的信息。"),
MessagesPlaceholder(variable_name="history"), # 历史记录插入这里
("human", "{input}"),
])
# 基础链
chain = prompt | llm | StrOutputParser()
# 用字典管理多个会话的历史
store = {}
def get_session_history(session_id: str) -> ChatMessageHistory:
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
# 包装成有记忆的链
chain_with_history = RunnableWithMessageHistory(
chain,
get_session_history,
input_messages_key="input",
history_messages_key="history",
)
# 会话 1:用户小明
config_1 = {"configurable": {"session_id": "user_xiaoming"}}
response = chain_with_history.invoke(
{"input": "我叫小明,是一名 Java 开发者。"},
config=config_1
)
print(f"Bot: {response}")
response = chain_with_history.invoke(
{"input": "我是做什么的?"},
config=config_1
)
print(f"Bot: {response}")
# 会话 2:用户小红(完全独立的历史)
config_2 = {"configurable": {"session_id": "user_xiaohong"}}
response = chain_with_history.invoke(
{"input": "我叫小红。"},
config=config_2
)
print(f"Bot: {response}")
response = chain_with_history.invoke(
{"input": "你刚才跟我说了什么?"},
config=config_2
)
print(f"Bot: {response}")
session_id 是关键。不同的 session_id 对应不同的历史,互不干扰。小明和小红的对话完全隔离,这在多用户应用里是基本要求。
5. 流式输出 + 历史记录
有记忆的链同样支持流式输出,只需要把 invoke 换成 stream:
print("Bot: ", end="", flush=True)
for chunk in chain_with_history.stream(
{"input": "你好,今天天气怎么样?"},
config=config_1
):
print(chunk, end="", flush=True)
print()
历史会在流式输出结束后自动保存。
6. 生产环境:用 Redis 存历史,别存内存
上面的例子里,历史存在 store 字典里,也就是内存里。这在生产环境是行不通的:
- 服务重启,历史全没了
- 多实例部署,不同实例的内存不共享
- 用户量大了,内存撑不住
生产环境应该用持久化存储。LangChain 提供了 RedisChatMessageHistory:
from langchain_community.chat_message_histories import RedisChatMessageHistory
def get_session_history(session_id: str) -> RedisChatMessageHistory:
return RedisChatMessageHistory(
session_id=session_id,
url=os.getenv("REDIS_URL", "redis://localhost:6379"),
ttl=86400, # 24 小时后自动过期
)
把 get_session_history 函数替换掉就行,其他代码不动。这就是 LangChain 接口设计的好处:存储层可以任意替换。
没有 Redis 也可以用数据库。LangChain 还支持 SQLChatMessageHistory,把历史存进 SQLite 或 PostgreSQL:
from langchain_community.chat_message_histories import SQLChatMessageHistory
def get_session_history(session_id: str) -> SQLChatMessageHistory:
return SQLChatMessageHistory(
session_id=session_id,
connection_string="sqlite:///chat_history.db"
)
7. token 消耗要心里有数
Memory 是 ChatBot 的基础,但用不好会让 token 消耗失控。
一轮对话平均 200 tokens,聊 20 轮就是 4000 tokens 的历史要带上。如果应用有 1000 个并发用户,每人每天聊 50 轮,光历史记录带来的 token 消耗就是个不小的数字。
几个控制手段:
- 限制历史长度,只保留最近 N 轮
- 定期用 Summary 压缩旧历史
- 给会话设置 TTL(Time To Live,存活时间:数据在一段时间无人访问后自动删除),自动清理不活跃的历史
- 监控每个会话的 token 消耗,设置上限
Memory 管理好了,ChatBot 才算真的可用,不然要么失忆,要么账单炸了。