通俗解析:如何在 LangChain/LangGraph 中管理消息历史

一、背景与核心问题

多轮对话应用(如智能客服、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 构建多轮对话,历史和上下文“全自动保存”,多用户/多场景一把梭,测试/生产都省心!


如需具体代码场景或集成建议,可继续留言。

评论
添加红包

请填写红包祝福语或标题

红包个数最小为10个

红包金额最低5元

当前余额3.43前往充值 >
需支付:10.00
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包

打赏作者

Python测试之道

你的鼓励将是我创作的最大动力

¥1 ¥2 ¥4 ¥6 ¥10 ¥20
扫码支付:¥1
获取中
扫码支付

您的余额不足,请更换扫码支付或充值

打赏作者

实付
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明:

1.余额是钱包充值的虚拟货币,按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载,可以购买VIP、付费专栏及课程。

余额充值