LangGraph教程|麻瓜教程

Appearance

多轮对话状态管理 ​

多轮对话的"记忆"，本质上就是用一个 messages 列表保存所有历史消息。本篇讲清楚 LangGraph 是怎么维护这个列表的，以及怎么让 AI 跨调用记住上下文。

一、多轮对话的核心：messages 列表 ​

一轮对话会产生若干消息（用户消息、AI 回复、工具调用结果等）。把这些消息按顺序存进一个列表，下一轮调用时把这个列表完整发给 LLM，LLM 就"记得"前面说过什么。

mermaid
flowchart LR
    U1[用户：我叫阿宝] --> L1[LLM：你好阿宝]
    L1 --> H[历史 messages 列表]
    H --> U2[用户：我叫什么？]
    U2 --> L2[LLM：你叫阿宝]

关键问题：列表怎么追加而不是覆盖？ 答案是 add_messages reducer。

二、MessagesState：开箱即用的预置状态 ​

LangGraph 内置了一个带 messages 字段的状态，省得每次手写：

python
from langgraph.graph import StateGraph, MessagesState, START, END

def chat(state: MessagesState):
    # state["messages"] 就是历史消息列表
    last = state["messages"][-1]
    reply = f"收到：{last.content}"
    return {"messages": [{"role": "assistant", "content": reply}]}

graph = StateGraph(MessagesState)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)
app = graph.compile()

MessagesState 等价于：

python
class MessagesState(TypedDict):
    messages: Annotated[list, add_messages]

注意 Annotated[list, add_messages]——这就是记忆的魔法。

三、add_messages reducer 的合并规则 ​

节点返回 {"messages": [新消息]} 时，reducer 不是简单覆盖，而是按规则合并：

情况行为
新消息没有 id直接追加到列表末尾
新消息有 id 且列表里已有同 id用新消息替换旧的那条
新消息有 id 且列表里没有追加到末尾

这就支持了"追加新对话"和"编辑/撤回某条历史"两种操作。

追加示例 ​

python
from langchain_core.messages import HumanMessage, AIMessage

# 第一轮
app.invoke({"messages": [HumanMessage(content="我叫阿宝", id="u1")]})
# 第二轮：列表会变成 [u1, ai1, u2, ai2]
app.invoke({"messages": [HumanMessage(content="我叫什么", id="u2")]})

同 id 更新（编辑历史） ​

python
# 把 u1 那条改成新内容，列表长度不变
app.invoke({"messages": [HumanMessage(content="我改名了叫阿强", id="u1")]})

四、节点如何返回消息 ​

最常见：节点返回 {"messages": [新消息]}，reducer 自动追加。

python
def chat(state: MessagesState):
    history = state["messages"]
    # 把整个历史发给 LLM，它就能"记住"上下文
    # reply = llm.invoke(history)
    reply = "模拟回复"
    return {"messages": [AIMessage(content=reply, id="ai-1")]}

给消息加 id 是好习惯，方便后续编辑、撤回、去重。

五、用 MemorySaver + thread_id 实现跨调用记忆 ​

前面的例子每次 invoke 都是独立的一次图执行。要实现"用户问完第一句、过一会儿问第二句，AI 还记得第一句"，需要：

  1. Checkpointer：把每次执行后的状态存起来。
  2. thread_id：用同一个 thread_id 标识同一会话，下次进来自动加载历史。
python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END

def chat(state: MessagesState):
    history = state["messages"]
    last_user = history[-1].content
    # 这里把 history 整体喂给 LLM，它就有了上下文
    reply = f"你说「{last_user}」，我记得我们聊过 {len(history)-1} 句"
    return {"messages": [{"role": "assistant", "content": reply}]}

graph = StateGraph(MessagesState)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

# 关键：compile 时传入 checkpointer
app = graph.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "user-001"}}

# 第一轮
r1 = app.invoke({"messages": [{"role": "user", "content": "我叫阿宝"}]}, config=config)
print(r1["messages"][-1].content)  # 你说「我叫阿宝」，我记得我们聊过 0 句

# 第二轮（同一 thread_id，自动加载历史）
r2 = app.invoke({"messages": [{"role": "user", "content": "我叫什么"}]}, config=config)
print(r2["messages"][-1].content)  # 你说「我叫什么」，我记得我们聊过 2 句

重点：两次 invoke 传了相同的 thread_id，第二次进来时框架自动从检查点恢复历史，state["messages"] 已经包含了第一轮的全部消息。

六、手动修改历史（回溯/编辑） ​

有时你想"撤回"AI 上一句、或改写用户输入后重跑。借助 id 和 update_state：

python
# 假设上一轮 AI 回复的 id 是 "ai-1"，想改成新内容
app.update_state(
    config,
    {"messages": [AIMessage(content="改写后的回复", id="ai-1")]},
)
# 同 id 会被 reducer 替换，而不是追加

也可以回滚到某个历史检查点继续对话，详见 中断与恢复。

七、完整多轮对话示例 ​

python
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver

def chatbot(state: MessagesState):
    history = state["messages"]
    last = history[-1].content
    # 简单模拟：如果历史里有"叫阿宝"就记得，否则不记得
    memory = ""
    for m in history[:-1]:  # 排除当前这句
        if "阿宝" in getattr(m, "content", ""):
            memory = "（我记得你叫阿宝）"
    reply = f"{memory}收到：{last}"
    return {"messages": [AIMessage(content=reply, id=f"ai-{len(history)}")]}

graph = StateGraph(MessagesState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile(checkpointer=MemorySaver())

cfg = {"configurable": {"thread_id": "s1"}}

for turn in ["我叫阿宝", "今天天气怎样", "我叫什么"]:
    r = app.invoke({"messages": [HumanMessage(content=turn)]}, config=cfg)
    print("用户:", turn)
    print("AI  :", r["messages"][-1].content, "\n")

输出会显示第三轮 AI 还记得"阿宝"——这就是多轮记忆。

八、常见踩坑 ​

踩坑 1：忘了传 thread_id ​

python
# ❌ 这样每次都是新会话，AI 永远失忆
app.invoke({"messages": [...]})  # 没有 config，没有 thread_id

# ✅ 正确
app.invoke({"messages": [...]}, config={"configurable": {"thread_id": "u1"}})

踩坑 2：messages 没挂 reducer，每次覆盖 ​

自己定义 State 时漏写 Annotated[list, add_messages]，结果每轮只剩最新一条消息。直接用 MessagesState 可避免。

踩坑 3：消息顺序错误 ​

messages 列表必须按对话时间顺序排列（用户→AI→用户→AI）。如果手动构造历史时乱序，LLM 会困惑。add_messages 是追加语义，正常使用不会乱序，但如果你用 update_state 整段替换历史要注意顺序。

踩坑 4：历史无限增长爆 token ​

聊得久了，messages 列表越来越长，每次都全量发给 LLM 会超 token 上限。需要做窗口裁剪或摘要压缩：

python
def chatbot(state: MessagesState):
    history = state["messages"]
    # 只保留最近 10 条作为上下文
    recent = history[-10:]
    # reply = llm.invoke(recent)
    ...

生产场景推荐用摘要节点定期把旧消息压成一条摘要。详见 持久化与检查点。

踩坑 5：用 MemorySaver 部署到生产 ​

MemorySaver 存在内存里，进程重启就全丢。开发可以，生产必须换 SqliteSaver 或 PostgresSaver。

九、小结 ​

  • 多轮对话核心 = 用 messages 列表存历史，靠 add_messages reducer 追加。
  • MessagesState 是开箱即用的预置状态。
  • 跨调用记忆 = MemorySaver + 相同 thread_id，框架自动加载/保存历史。
  • 给消息加 id，方便后续编辑、撤回、去重。
  • 注意历史增长要裁剪，避免超 token。

下一篇 工具调用 Tool Calling 讲怎么让 AI 调用外部工具。

Copyright © 2025-2026 京公网安备 11010102000001号 京ICP备2024078128号-1