一、背景与核心问题
多轮对话应用(如智能客服、AI 助手)需要记住用户历史,才能实现连续对话。
如何让模型“记住”对话历史?怎么自动保存、切换不同用户的对话?这就是“消息历史管理”的核心意义。
LangGraph(LangChain 的新一代工作流引擎)为我们自动管理消息历史和多用户隔离提供了非常好的机制。
本文将用浅显的语言,带你快速掌握这一过程。
二、核心概念
1. 消息历史(message history)
- 就是用户和 AI 互相说的每一句话的顺序列表。
- 例如:
[HumanMessage("你好"), AiMessage("你好,有什么可以帮您?"), HumanMessage("今天天气怎么样?")]
2. 状态(state)
- 不光能存消息历史,还能存别的上下文信息(如当前语言、用户 ID 等)。
- 所有流程节点都共享这份“状态”,就像对话机器人随身带的记事本。
3. 多线程/多用户支持
- 每个对话都有唯一 thread_id,就像每个人有自己的历史,不会串台。
三、消息历史的自动管理与持久化
LangGraph 的一大亮点是:
你只要定义好状态和节点,历史自动保存!
还可以选用内存、数据库等多种持久化方式。
最简单的例子:自动保存消息历史
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import START, MessagesState, StateGraph
workflow = StateGraph(state_schema=MessagesState)
def call_model(state: MessagesState):
response = llm.invoke(state["messages"])
return {"messages": response} # 自动加入历史!
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)
memory = MemorySaver() # 用内存存历史
app = workflow.compile(checkpointer=memory)
- 每次调用 app.invoke,历史都会自动更新,无需手动管理。
- 支持多用户,只要 thread_id 不一样,历史互不干扰。
对话示例(支持多轮 & 多用户)
config = {"configurable": {"thread_id": "abc123"}}
# 第一次对话
input_messages = [HumanMessage("Hi! I'm Bob.")]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
# 第二次对话(同一个 thread_id,历史自动串起来)
input_messages = [HumanMessage("What's my name?")]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
# 换个 thread_id(新用户/新会话,历史是空的)
config = {"configurable": {"thread_id": "abc234"}}
input_messages = [HumanMessage("What's my name?")]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
效果说明:
- 同一个 thread_id,历史递增,模型记得你的名字
- 换 thread_id,历史清空,模型不认识你
四、复杂场景:自定义状态(带多参数)
除了消息历史,你还可以在状态中加入其它参数,比如当前对话语言。
from typing import Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from typing_extensions import Annotated, TypedDict
class State(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
language: str
用法举例:
- 每次对话可以指定 language(如 “Chinese”, “English”, “Spanish”)
- 消息历史会自动添加,语言会自动覆盖
五、历史的访问与手动更新
1. 查看当前历史
state = app.get_state(config).values
print(state["language"])
for message in state["messages"]:
message.pretty_print()
这样可以随时取出当前 thread_id 的全部历史和上下文。
2. 手动追加消息
from langchain_core.messages import HumanMessage
_ = app.update_state(config, {"messages": [HumanMessage("Test")]})
这样可以在自动对话流程外,自己加一条历史(方便测试或人工干预)。
六、测试工程师关注点与实用建议
1. 多线程/多用户测试
- 用不同 thread_id 验证历史是否真正隔离。
2. 断点续传 & 持久化
- 用 MemorySaver 测试本地内存;
- 生产可切换 SQLite、Redis 等持久化方案,验证断电/重启后历史是否能恢复。
3. 历史正确性
- 用 get_state 检查所有历史内容;
- 用 update_state 人为插入边界场景,验证模型上下文是否准确。
4. 参数多样性
- 状态不仅能存历史,还能存语言/偏好等多参数,便于多场景测试。
七、总结
- LangGraph自动管理消息历史,极大简化了多轮对话和多用户的开发、测试和维护。
- thread_id 隔离机制天然适合并发场景。
- 灵活的状态管理为复杂对话和个性化配置打下了基础。
- 测试工程师可以用 get_state、update_state 等接口,轻松覆盖各种历史与边界场景。
一句话总结:
用 LangGraph 构建多轮对话,历史和上下文“全自动保存”,多用户/多场景一把梭,测试/生产都省心!
如需具体代码场景或集成建议,可继续留言。