本文介绍了如何使用LangGraph构建具有记忆功能的聊天机器人，重点讲解了如何在LangGraph中实现对话状态的持久化，以便于多轮对话应用开发。通过将LangChain runnables 包装在最小的LangGraph应用程序中，可以持久化消息历史记录和其他链状态元素。文章提供了两个示例，分别展示了如何处理消息输入和字典输入，以及如何管理消息历史记录。此外，还提到了LangGraph对多线程的支持，使单个应用程序能够与多个用户分别交互。通过使用LangGraph，开发者可以更轻松地构建更智能、更具交互性的聊天机器人。

💬 LangGraph 允许将对话状态传入和传出链，并提供内置的持久层，支持将链状态持久化到内存或外部后端，简化多轮应用程序的开发。

🤖 文章通过示例演示了如何将LangChain runnables 包装在 LangGraph 应用程序中，添加信息历史功能，从而实现消息历史记录和链状态的持久化。

📤 提供了两种输入方式的示例：消息输入和字典输入。对于消息输入，使用MessagesState来存储消息列表；对于字典输入，图状态定义为包括提示模板所需的参数，如language。

💾 消息历史记录可以通过 .get_state 访问，也可以通过 .update_state 更新，方便开发者管理和操作对话状态。

一、食用说明

在构建聊天机器人时，将对话状态传入和传出链至关重要。 LangGraph 实现了内置的持久层，允许链状态自动持久化在内存或外部后端（如 SQLite、Postgres 或 Redis）中。在本文我们将演示如何通过将任意 LangChain runnables 包装在最小的 LangGraph 应用程序中来添加信息历史功能，这使我们能够持久化消息历史记录和链状态的其他元素，从而简化多轮应用程序的开发。它还支持多线程，使单个应用程序能够与多个用户分别交互。

温馨提示：本文搭配 Jupyter notebooks 食用更佳，在交互式环境中学习是更好地理解它们的好方法。

二、使用 Qwen3 聊天模型

pip install -qU "langchain[openai]"

from pydantic import SecretStrimport osos.environ["OPENAI_BASE_URL"] = "https://api.siliconflow.cn/v1/"os.environ["OPENAI_API_KEY"] = "sk-xxx"from langchain.chat_models import init_chat_modelllm = init_chat_model("Qwen/Qwen3-8B", model_provider="openai")

三、示例：消息输入

聊天模型接受消息列表作为输入并输出消息。 LangGraph 包括一个内置的 MessagesState，我们可以将其用于此目的。下面，我们将图状态定义为消息列表，向图中添加一个调用聊天模型的节点，使用内存检查点编译器编译图，以在运行之间存储消息。

from langchain_core.messages import HumanMessagefrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import START, MessagesState, StateGraph# Define a new graphworkflow = StateGraph(state_schema=MessagesState)# Define the function that calls the modeldef call_model(state: MessagesState):    response = llm.invoke(state["messages"])    # Update message history with response:    return {"messages": response}# Define the (single) node in the graphworkflow.add_edge(START, "model")workflow.add_node("model", call_model)# Add memorymemory = MemorySaver()app = workflow.compile(checkpointer=memory)

可以查看一下图结构：

from IPython.display import Image, displaydisplay(Image(app.get_graph().draw_mermaid_png()))

当我们运行应用程序时，我们传入一个配置 dict，其中指定了 thread_id。此 ID 用于区分对话线程（例如，不同用户之间）。

config = {"configurable": {"thread_id": "abc123"}}

然后我们可以调用应用程序

query = "hi 我是小智"input_messages = [HumanMessage(query)]output = app.invoke({"messages": input_messages}, config)output["messages"][-1].pretty_print()  # output contains all messages in state

query = "我是谁？"input_messages = [HumanMessage(query)]output = app.invoke({"messages": input_messages}, config)output["messages"][-1].pretty_print()

请注意，状态对于不同的线程是分开的。如果我们向具有新 thread_id 的线程发出相同的查询，模型会指示它不知道答案

query = "我是谁？"config = {"configurable": {"thread_id": "abc234"}}input_messages = [HumanMessage(query)]output = app.invoke({"messages": input_messages}, config)output["messages"][-1].pretty_print()

四、示例：字典输入

LangChain runnables 通常通过单个 dict 参数中的单独键接受多个输入。一个常见的例子是具有多个参数的提示模板。

之前的 runnable 是聊天模型，这里我们将提示模板和聊天模型链接在一起。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderprompt = ChatPromptTemplate.from_messages(    [        ("system", "请用 {language} 回答问题"),        MessagesPlaceholder(variable_name="messages"),    ])runnable = prompt | llm

对于这种情况，我们将图状态定义为包括这些参数（除了消息历史记录之外）。然后，我们以与之前相同的方式定义单节点图。

请注意，在下面的状态中：

对 messages 列表的更新将附加消息；对 language 字符串的更新将覆盖该字符串。

from typing import Sequencefrom langchain_core.messages import BaseMessagefrom langgraph.graph.message import add_messagesfrom typing_extensions import Annotated, TypedDict"""‌作为状态机StateGraph的状态字段，用于跟踪对话历史，序列化时会保留消息的时间顺序和完整内容"""class State(TypedDict):    # Annotated包装器允许附加元数据add_messages，这是LangGraph提供的消息合并函数，当状态更新时，add_messages会自动处理新旧消息序列的合并逻辑    messages: Annotated[Sequence[BaseMessage], add_messages]    language: strworkflow = StateGraph(state_schema=State)def call_model(state: State):    response = runnable.invoke(state)    # Update message history with response:    return {"messages": [response]}workflow.add_edge(START, "model")workflow.add_node("model", call_model)memory = MemorySaver()app = workflow.compile(checkpointer=memory)

from IPython.display import Image, displaydisplay(Image(app.get_graph().draw_mermaid_png()))

config = {"configurable": {"thread_id": "abc345"}}input_dict = {    "messages": [HumanMessage("hi 我是小智")],    "language": "粤语",}output = app.invoke(input_dict, config)output["messages"][-1].pretty_print()

五、管理消息历史记录

消息历史记录（和应用程序状态的其他元素）可以通过 .get_state 访问

state = app.get_state(config).valuesprint(f'Language: {state["language"]}')for message in state["messages"]:    message.pretty_print()

我们还可以通过 .update_state 更新状态。例如，我们可以手动附加新消息

from langchain_core.messages import HumanMessage_ = app.update_state(config, {"messages": [HumanMessage("今天天气很好")]})

state = app.get_state(config).valuesprint(f'Language: {state["language"]}')for message in state["messages"]:    message.pretty_print()

参考文档

python.langchain.ac.cn/docs/how_to…