【LangChain 多轮对话】记忆篇
·
LangChain 多轮对话完全指南
目录
什么是多轮对话
单轮对话 vs 多轮对话
单轮对话：每次独立问答，不记住之前的内容
# 单轮：每次都是全新的开始
response = model.invoke("北京天气如何？")
response = model.invoke("那上海呢？") # AI不知道"那上海"指的是什么
多轮对话：记住对话历史，上下文关联
# 多轮：记住之前的对话
messages = [
HumanMessage(content="北京天气如何？"),
AIMessage(content="北京今天晴天，25度"),
HumanMessage(content="那上海呢？") # AI知道"那上海"指的是上海的天气
]
response = model.invoke(messages)
为什么需要多轮对话？
| 场景 | 单轮问题 | 多轮优势 |
|---|---|---|
| 追问 | "那上海呢？" | 自动理解指代上文 |
| 澄清 | "不对，是冬天" | 记住之前的错误修正 |
| 任务延续 | "继续写代码" | 记得之前写了什么 |
| 个性化 | "我的名字是小明" | 记住用户信息 |
对话历史管理
最简单的多轮对话
from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage, HumanMessage, AIMessage
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
model = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 构建对话历史
messages = [
SystemMessage(content="你是一个旅游助手"),
HumanMessage(content="推荐几个国内旅游城市"),
AIMessage(content="我推荐：成都、杭州、厦门、丽江"),
HumanMessage(content="成都有什么好吃的？"),
]
response = model.invoke(messages)
print(response.content)
对话历史累积
# 初始化对话历史
history = []
# 第1轮对话
messages = history + [
HumanMessage(content="推荐几个国内旅游城市")
]
response = model.invoke(messages)
print(f"AI: {response.content}")
# 更新历史
history.append(HumanMessage(content="推荐几个国内旅游城市"))
history.append(AIMessage(content=response.content))
# 第2轮对话（带上历史）
messages = history + [HumanMessage(content="成都有什么好吃的？")]
response = model.invoke(messages)
print(f"AI: {response.content}")
# 更新历史
history.append(HumanMessage(content="成都有什么好吃的？"))
history.append(AIMessage(content=response.content))
# 第3轮对话
messages = history + [HumanMessage(content="那杭州呢？")]
response = model.invoke(messages)
print(f"AI: {response.content}")
MessagesPlaceholder 占位符
为什么需要占位符？
在构建 Prompt 模板时，对话历史的长度是动态的，不可能写死：
-
第1轮：0条历史
-
第10轮：10条历史
这时候需要 MessagesPlaceholder 占位符，在运行时动态插入历史消息。
显式使用 MessagesPlaceholder
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 构建模板，占位符用于插入对话历史
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个资深的Python开发工程师，请认真回答我提出的Python相关的问题"),
MessagesPlaceholder("memory"), # 动态插入对话历史
("human", "{question}")
])
# 调用时传入历史消息
prompt_value = prompt.invoke({
"memory": [
HumanMessage(content="我的名字叫亮仔，是一名程序员"),
AIMessage(content="好的，亮仔你好，很高兴认识你"),
],
"question": "Python的装饰器是什么？"
})
隐式使用占位符
# 隐式写法：("placeholder", "{memory}") 等价于 MessagesPlaceholder("memory")
prompt = ChatPromptTemplate.from_messages([
("placeholder", "{memory}"),
("system", "你是一个资深的Python开发工程师"),
("human", "{question}")
])
prompt_value = prompt.invoke({
"memory": [
HumanMessage(content="我的名字叫亮仔"),
AIMessage(content="好的，亮仔你好"),
],
"question": "装饰器是什么？"
})
两种写法对比
# 显式（更清晰，推荐）
MessagesPlaceholder("memory")
# 隐式（更简洁）
("placeholder", "{memory}")
对话记忆机制
什么是记忆机制？
记忆机制是 AI 应用记住之前对话内容的能力。常见实现方式：
| 类型 | 说明 | 适用场景 |
|---|---|---|
| ConversationBufferMemory | 完整保存所有历史 | 对话短、上下文重要 |
| ConversationTokenBufferMemory | 按Token数限制 | 对话长、需要控制成本 |
| ConversationSummaryMemory | 自动总结历史 | 超长对话 |
| VectorStoreRetrieverMemory | 向量检索记忆 | 超长对话、语义检索 |
ConversationBufferMemory
最简单的记忆方式，保存完整对话历史：
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
# 创建记忆对象
memory = ConversationBufferMemory()
# 添加对话历史
memory.save_context({"input": "我叫小明"}, {"output": "你好小明！"})
memory.save_context({"input": "我最喜欢的颜色是蓝色"}, {"output": "蓝色是很棒的选择！"})
# 加载历史
history = memory.load_memory_variables({})
print(history)
# {'history': 'Human: 我叫小明\nAI: 你好小明！\nHuman: 我最喜欢的颜色是蓝色\nAI: 蓝色是很棒的选择！'}
结合 LangChain 链使用
from langchain.chat_models import init_chat_model
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.prompts import PromptTemplate
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
model = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 创建记忆链
memory = ConversationBufferMemory()
# 自定义Prompt
template = """你是一个友好的AI助手。
历史对话：
{history}
用户新消息：{input}
你的回复："""
prompt = PromptTemplate(
template=template,
input_variables=["history", "input"]
)
# 创建对话链
chain = ConversationChain(
llm=model,
memory=memory,
prompt=prompt,
verbose=True
)
# 对话
response = chain.invoke({"input": "我叫小明"})
print(response)
response = chain.invoke({"input": "我叫什么名字？"})
print(response) # 会记住叫小明
多轮对话实战
基础版本：手动管理历史
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
model = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 构建带记忆的模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个旅游助手，为用户提供旅行建议"),
MessagesPlaceholder("history"),
("human", "{question}")
])
def chat(question, history):
"""对话函数"""
messages = prompt.format_messages(history=history, question=question)
response = model.invoke(messages)
# 更新历史
history.append(HumanMessage(content=question))
history.append(AIMessage(content=response.content))
return response.content, history
# 开始对话
history = []
print("=== 第1轮 ===")
answer1, history = chat("推荐几个国内旅游城市", history)
print(f"AI: {answer1}")
print("\n=== 第2轮 ===")
answer2, history = chat("北京有什么好吃的？", history)
print(f"AI: {answer2}")
print("\n=== 第3轮 ===")
answer3, history = chat("那上海呢？", history) # "那上海呢"需要结合上文理解
print(f"AI: {answer3}")
进阶版本：支持上下文理解
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
model = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 改进的系统提示词，让AI更好地理解上下文
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的旅游顾问。
请根据对话历史，理解用户的上下文。
如果用户用"那...呢"、"还有呢"等指代词，要结合之前的对话来理解。
例如：
- 问"北京之后上海呢？" 要理解这是在问上海的旅游推荐"""),
MessagesPlaceholder("history"),
("human", "{question}")
])
def chat(question, history):
messages = prompt.format_messages(history=history, question=question)
response = model.invoke(messages)
history.append(HumanMessage(content=question))
history.append(AIMessage(content=response.content))
return response.content, history
# 对话测试
history = []
answer1, history = chat("北京有什么景点？", history)
print(f"AI: {answer1}")
answer2, history = chat("那上海呢？", history)
print(f"AI: {answer2}") # 应该理解是在问上海的景点
生产版本：带用户信息记忆
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
model = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 构建带用户信息的模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个私人旅行助手。用户信息：{user_info}"),
MessagesPlaceholder("history"),
("human", "{question}")
])
class TravelChat:
def __init__(self):
self.history = []
self.user_info = {
"name": "小明",
"preferences": "喜欢美食和自然风光",
"budget": "中等预算",
"travel_style": "自由行"
}
def chat(self, question):
messages = prompt.format_messages(
user_info=self.user_info,
history=self.history,
question=question
)
response = model.invoke(messages)
self.history.append(HumanMessage(content=question))
self.history.append(AIMessage(content=response.content))
return response.content
def reset(self):
"""重置对话历史"""
self.history = []
# 使用
chat = TravelChat()
print("=== 对话1 ===")
print(chat.chat("我叫小明，喜欢吃辣，帮我推荐一个旅游城市"))
print("\n=== 对话2 ===")
print(chat.chat("那里有什么好吃的辣味美食？"))
进阶用法
Token数限制（控制成本）
对话历史太长会导致 Token 消耗过大，需要限制：
from langchain.memory import ConversationTokenBufferMemory
from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
model = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 限制最大Token数为2000
memory = ConversationTokenBufferMemory(
llm=model,
max_token_limit=2000
)
# 添加大量对话
for i in range(20):
memory.save_context({"input": f"这是第{i}轮对话"}, {"output": f"这是第{i}轮回复"})
# 自动截断旧的历史
history = memory.load_memory_variables({})
print(f"历史长度: {len(history['history'])}")
# 会自动保留最新的对话，删除旧的
自动总结历史（超长对话）
from langchain.memory import ConversationSummaryMemory
from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
model = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 自动总结历史
memory = ConversationSummaryMemory(llm=model)
# 添加对话
memory.save_context({"input": "我叫小明，是一名程序员"}, {"output": "你好小明！"})
memory.save_context({"input": "我工作5年了"}, {"output": "经验丰富啊！"})
memory.save_context({"input": "我会Python、Java、Go"}, {"output": "技术栈很广！"})
# 获取总结后的历史
history = memory.load_memory_variables({})
print(history['history'])
# 会自动总结为一段话，而不是保留所有对话
向量记忆（语义检索）
from langchain.memory import VectorStoreRetrieverMemory
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
# 创建向量存储
vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
memory = VectorStoreRetrieverMemory(
vectorstore=vectorstore,
search_kwargs={"k": 3} # 返回最相关的3条记忆
)
# 保存记忆（带描述）
memory.save_context(
{"input": "用户喜欢日料"},
{"output": "已记录用户的日料偏好"}
)
memory.save_context(
{"input": "用户下周要去东京"},
{"output": "已记录用户的东京行程"}
)
# 检索相关记忆
memory.load_memory_variables({"prompt": "用户有什么美食偏好？"})
# 会语义检索返回"喜欢日料"相关记忆
常见问题
1. 对话历史越来越长怎么办？
# 方法1：限制Token数
from langchain.memory import ConversationTokenBufferMemory
memory = ConversationTokenBufferMemory(llm=model, max_token_limit=2000)
# 方法2：自动总结
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(llm=model)
# 方法3：只保留最近N轮
def limit_history(history, max_turns=5):
return history[-max_turns * 2:] # 每轮2条消息（问+答）
2. 如何让AI记住用户信息？
# 方法1：放在系统提示词里
prompt = ChatPromptTemplate.from_messages([
("system", "用户叫{name}，喜欢{preference}"),
MessagesPlaceholder("history"),
("human", "{question}")
])
# 方法2：每次传入用户信息
messages = prompt.format_messages(
user_info=user_info,
history=history,
question=question
)
3. 如何让AI理解指代词？
# 改进系统提示词 system_prompt = """你是一个专业的AI助手。 请注意理解对话中的指代词： - "那...呢" 通常指代之前提到的事物 - "他/她/它" 指代之前提到的人或物 - "然后呢" 通常继续之前的话题 结合上下文理解用户真实意图。"""
4. 如何持久化对话历史？
import json
# 保存到文件
def save_history(history, filename="history.json"):
with open(filename, "w", encoding="utf-8") as f:
json.dump([msg.to_json() for msg in history], f, ensure_ascii=False)
# 从文件加载
def load_history(filename="history.json"):
with open(filename, "r", encoding="utf-8") as f:
data = json.load(f)
return [HumanMessage(**m) if m["type"]=="human" else AIMessage(**m) for m in data]
总结
多轮对话核心概念
| 概念 | 作用 | 代码 |
|---|---|---|
| 对话历史 | 记住之前的问答 | history.append(HumanMessage(...)) |
| MessagesPlaceholder | 动态插入历史到模板 | MessagesPlaceholder("memory") |
| ConversationBufferMemory | 完整保存历史 | memory.load_memory_variables({}) |
| ConversationTokenBufferMemory | 限制Token数 | max_token_limit=2000 |
| ConversationSummaryMemory | 自动总结历史 | memory.load_memory_variables({}) |
选择记忆类型
对话长度短 → ConversationBufferMemory（最简单） 对话长度中等 → ConversationTokenBufferMemory（控制成本） 对话长度长 → ConversationSummaryMemory（自动总结） 对话长度超长 → VectorStoreRetrieverMemory（向量检索）
多轮对话模板
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个AI助手"),
MessagesPlaceholder("history"), # 对话历史
("human", "{question}")
])
def chat(question, history):
messages = prompt.format_messages(history=history, question=question)
response = model.invoke(messages)
history.append(HumanMessage(content=question))
history.append(AIMessage(content=response.content))
return response.content, history
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
7
0- 0
已为社区贡献25条内容
所有评论(0)