LangGraph 会话切换与短期记忆管理

发表于 分类于
在 LangGraph 构建 Agent 应用时，会话记忆是核心能力，借助 Checkpointer 搭配 thread_id，可以自动保存每一轮对话轨迹与消息历史，实现多会话隔离。本文基于 InMemorySaver 内存存储，演示交互式会话切换、查看、清空等基础操作。讲解消息条数修剪、Token 阈值修剪、消息删除等处理方案；同时为避免直接丢弃历史造成信息丢失，实现对话历史摘要压缩方案，在控制上下文长度的同时保留关键对话信息，适合本地 GGUF 类大模型 Agent 开发学习参考。 Agent 开发等功能实现。

在 LangGraph 构建 Agent 应用时，会话记忆是核心能力，借助 Checkpointer 搭配 thread_id，可以自动保存每一轮对话轨迹与消息历史，实现多会话隔离。本文基于 InMemorySaver 内存存储，演示交互式会话切换、查看、清空等基础操作。讲解消息条数修剪、Token 阈值修剪、消息删除等处理方案；同时为避免直接丢弃历史造成信息丢失，实现对话历史摘要压缩方案，在控制上下文长度的同时保留关键对话信息，适合本地 GGUF 类大模型 Agent 开发学习参考。

LangGraph 记忆系统分为 Checkpointer 与 Store 两套能力。Checkpointer 依靠 thread_id 自动记录 Agent 会话轨迹和消息历史，无需额外工具；Store 用于存储用户档案、知识库等业务实体，基于 namespace+key，需要手动调用 put 写入。存储载体上，InMemorySaver 把会话数据存放内存，InMemoryStore 用于将重要数据手动持久化到磁盘。

消息交互

开发 Agent 应用时，多轮连贯对话与多会话隔离属于必备基础能力，LangGraph 借助 Checkpointer 检查点机制搭配thread_id，可以自动保存 Agent 会话轨迹与消息历史，无需手动维护消息列表；使用InMemorySaver内存检查点可快速实现短期会话记忆，不同线程之间会话相互隔离，但数据存放于内存，程序重启会话便会全部丢失，仅适用于开发调试，生产环境建议替换为 SqliteSaver、PostgresSaver 等持久化检查点完成数据落盘。

首先实现一个简单的临时会话功能，通过 Agent 绑定checkpointer=InMemorySaver()，调用invoke时传入configurable配置，指定唯一thread_id。相同thread_id会复用历史对话；不同thread_id生成完全独立会话，互相不会串话。

下面示例构建 Agent，创建session_001、session_002两个会话。session_001 记住用户基本信息，session_002 是全新会话，获取不到之前的用户信息。

from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

basic_model = ChatOpenAI(
    model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
    base_url="http://127.0.0.1:11433/v1",
    api_key="dummy",
    temperature=0.1,
    max_tokens=512,
)

if __name__ == "__main__":
    agent = create_agent(
        basic_model,
        tools=[],
        checkpointer=InMemorySaver(),
    )

    # 会话A
    res1 = agent.invoke(
        {"messages": [{"role": "user", "content": "我叫张三，今年25岁，男生"}]},
        {"configurable": {"thread_id": "session_001"}}
    )
    print(res1["messages"][-1].content)

    res2 = agent.invoke(
        {"messages": [{"role": "user", "content": "我叫什么名字？多大了？"}]},
        {"configurable": {"thread_id": "session_001"}}
    )
    print(f"[session_001] {res2['messages'][-1].content}")

    # 会话B
    res3 = agent.invoke(
        {"messages": [{"role": "user", "content": "我叫什么名字？多大了？"}]},
        {"configurable": {"thread_id": "session_002"}}
    )
    print(f"[session_002] {res3['messages'][-1].content}")

运行上述代码，此时会话1传入用户基本信息并询问得到输出结果，而会话2则无法记忆用户的输入内容，如下输出所示；

CMD > main.py
你好，张三！很高兴认识你。作为一个AI助手，我无法直接了解你的个人信息。
[session_001] 你好，张三！你叫张三，今年25岁。
[session_002] 抱歉，我无法回答这个问题。作为一个AI助手，我无法获取你的个人信息，包括你的名字和年龄。

上面是代码层面调用，实际调试希望有交互式终端，可以切换会话、查看全部历史、清空当前会话消息。借助agent.get_state()读取会话状态，agent.update_state()更新状态，配合RemoveMessage对象实现消息删除，实现一套简易终端命令行工具。

from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import RemoveMessage

basic_model = ChatOpenAI(
    model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
    base_url="http://127.0.0.1:11433/v1",
    api_key="dummy",
    temperature=0.1,
    max_tokens=512,
)

if __name__ == "__main__":
    agent = create_agent(
        basic_model,
        tools=[],
        checkpointer=InMemorySaver(),
    )

    print("------- 交互式会话测试 -------")
    print("  /switch <thread_id>  切换会话ID")
    print("  /show                打印会话全部历史消息")
    print("  /clear               清空当前会话消息")
    print("  /exit                退出程序")
    print("-----------------------------")

    # 默认会话
    current_thread_id = "session_01"

    while True:
        user_input = input("\nHuman: ").strip()
        if not user_input:
            continue
        if user_input.lower() == "/exit":
            break
        if user_input.startswith("/switch"):
            parts = user_input.split(maxsplit=1)
            if len(parts) == 2:
                current_thread_id = parts[1]
                print(f">>> 已切换会话 thread_id = {current_thread_id}")
            else:
                print(">>> 用法: /switch session_xx")
            continue
        if user_input == "/show":
            state = agent.get_state({"configurable": {"thread_id": current_thread_id}})
            print("\n---- 当前会话历史 ----")
            for msg in state.values["messages"]:
                print(f"{msg.type}: {msg.content}")
            continue
        if user_input == "/clear":
            state = agent.get_state({"configurable": {"thread_id": current_thread_id}})
            msgs = state.values.get("messages", [])
            if msgs:
                removes = [RemoveMessage(id=m.id) for m in msgs]
                agent.update_state(
                    {"configurable": {"thread_id": current_thread_id}},
                    {"messages": removes}
                    )
            print(">>> 当前会话消息已清空")
            continue

        # 正常对话调用
        res = agent.invoke(
            {"messages": [{"role": "user", "content": user_input}]},
            {"configurable": {"thread_id": current_thread_id}}
        )
        ai_reply = res["messages"][-1].content
        print(f"AI：{ai_reply}")

运行后，用户可以交互式与大模型进行提问，并且大模型也具备了基本的记忆力，如下效果；

CMD> main.py
------- 交互式会话测试 -------
命令说明：
  /switch <thread_id>  切换会话ID
  /show                打印会话全部历史消息
  /clear               清空当前会话消息
  /exit                退出程序
-----------------------------

Human: 你好，我是王瑞，今年28岁，男性，汉族人。
AI：你好，王瑞。很高兴认识你。我是阿里云的AI助手，我叫Qwen。有什么我可以帮助你的吗？
Human: 概述一下我的基本信息
AI：你好，王瑞，28岁，男性，汉族。

Human: /show
---- 当前会话历史 ----
human: 你好，我是王瑞，今年28岁，男性，汉族人。
ai: 你好，王瑞。很高兴认识你。我是阿里云的AI助手，我叫Qwen。有什么我可以帮助你的吗？
human: 概述一下我的基本信息
ai: 你好，王瑞，28岁，男性，汉族。

Human: /switch session_02
>>> 已切换会话 thread_id = session_02
Human: 概述一下我的基本信息
AI：对不起，我无法提供关于您的个人信息。如果您有其他问题或需要帮助，请随时告诉我，我会尽力提供帮助。
Human: /clear
>>> 当前会话消息已清空
Human: /exit

LangGraph 的 Checkpointer+thread_id，把会话状态管理交给框架，不用自己维护消息数组。通过简单配置就实现多会话隔离；结合get_state、update_state、RemoveMessage，可以实现会话查看、切换、清空等常见会话管理功能。这套短期记忆方案，是构建多轮对话 Agent 的基础。

消息修剪

大模型存在上下文窗口 token 上限，对话轮次不断增加后会出现超限问题，消息修剪是解决该问题的常用手段，修剪分为物理修剪直接删除消息、逻辑修剪仅推理时忽略消息两类，本节将依次实现基于 LangGraph 的两种物理修剪方案。

第一种按消息条数修剪，保留首条消息与最近 N 条对话，删除老旧消息；

from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import AnyMessage, RemoveMessage

basic_model = ChatOpenAI(
    model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
    base_url="http://127.0.0.1:11433/v1",
    api_key="dummy",
    temperature=0.1,
    max_tokens=512,
)

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def dump_msg(graph, config) -> list[dict]:
    snap = graph.get_state(config)
    msgs = snap.values.get("messages", [])
    return [
        {"type": m.type, "content": m.content, "id": m.id[:10]}
        for m in msgs
    ]

def trim_node(state: State):
    """按消息条数物理修剪，保留第一条 + 最近4条消息"""
    messages = state["messages"]
    keep_recent = 4
    if len(messages) <= keep_recent:
        return {}

    first = messages[0]
    recent = messages[-keep_recent:]
    keep_ids = {m.id for m in [first, *recent]}

    remove_list = [
        RemoveMessage(id=m.id)
        for m in messages
        if m.id not in keep_ids
    ]
    return {"messages": remove_list}

def llm_node(state: State):
    resp = basic_model.invoke(state["messages"])
    return {"messages": [resp]}

if __name__ == "__main__":
    builder = StateGraph(State)
    builder.add_node("trim", trim_node)
    builder.add_node("llm", llm_node)

    builder.add_edge("trim", "llm")
    builder.set_entry_point("trim")

    checkpointer = InMemorySaver()
    graph = builder.compile(checkpointer=checkpointer)

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

    for i in range(6):
        out = graph.invoke({"messages": [("user", f"第{i}轮对话随便聊")]}, cfg)
        print(f"\n===== 轮{i} =====")
        print(f"reply: {out['messages'][-1].content[:60]}")
        msg_list = dump_msg(graph, cfg)
        print(f"当前消息总条数：{len(msg_list)}")
        for idx, item in enumerate(msg_list):
            print(f"  [{idx}] {item['type']:6s} | {item['content']} | id={item['id']}")

    snap = graph.get_state(cfg)
    print("\n===== 最终 =====")
    print("最终消息条数：", len(snap.values["messages"]))

上述代码运行后大模型将持续问答，在第3轮开始消息的条数始终保持6条，除去首轮未变动以外，中间部分将自动被忽略，只保留底部的4条记录，如下所示；

CMD> main.py

===== 轮0 =====
reply: 好的，我们可以聊聊你最近的生活。
当前消息总条数：2
  [0] human  | 第0轮对话随便聊 | id=46440f0a-7
  [1] ai     | 好的，我们可以聊聊你最近的生活。 | id=lc_run--01

===== 轮1 =====
reply: 当然可以！我们可以聊聊你最近的生活。
当前消息总条数：4
  [0] human  | 第0轮对话随便聊 | id=46440f0a-7
  [1] ai     | 好的，我们可以聊聊你最近的生活。 | id=lc_run--01
  [2] human  | 第1轮对话随便聊 | id=8571ca1f-3
  [3] ai     | 当然可以！我们可以聊聊你最近的生活。 | id=lc_run--01

===== 轮2 =====
reply: 当然，我们可以聊聊你最近的生活。
当前消息总条数：6
  [0] human  | 第0轮对话随便聊 | id=46440f0a-7
  [1] ai     | 好的，我们可以聊聊你最近的生活。 | id=lc_run--01
  [2] human  | 第1轮对话随便聊 | id=8571ca1f-3
  [3] ai     | 当然可以！我们可以聊聊你最近的生活。 | id=lc_run--01
  [4] human  | 第2轮对话随便聊 | id=4c517732-b
  [5] ai     | 当然，我们可以聊聊你最近的生活。 | id=lc_run--01

===== 轮3 =====
reply: 好的，我们可以聊聊你最近的生活。
当前消息总条数：6
  [0] human  | 第0轮对话随便聊 | id=46440f0a-7
  [1] ai     | 当然可以！我们可以聊聊你最近的生活。 | id=lc_run--01
  [2] human  | 第2轮对话随便聊 | id=4c517732-b
  [3] ai     | 当然，我们可以聊聊你最近的生活。 | id=lc_run--01
  [4] human  | 第3轮对话随便聊 | id=42868fa7-7
  [5] ai     | 好的，我们可以聊聊你最近的生活。 | id=lc_run--01

===== 轮4 =====
reply: 当然，我们可以聊聊你最近的生活。
当前消息总条数：6
  [0] human  | 第0轮对话随便聊 | id=46440f0a-7
  [1] ai     | 当然，我们可以聊聊你最近的生活。 | id=lc_run--01
  [2] human  | 第3轮对话随便聊 | id=42868fa7-7
  [3] ai     | 好的，我们可以聊聊你最近的生活。 | id=lc_run--01
  [4] human  | 第4轮对话随便聊 | id=c86f64f3-0
  [5] ai     | 当然，我们可以聊聊你最近的生活。 | id=lc_run--01

===== 轮5 =====
reply: 好的，我们可以聊聊你最近的生活。
当前消息总条数：6
  [0] human  | 第0轮对话随便聊 | id=46440f0a-7
  [1] ai     | 好的，我们可以聊聊你最近的生活。 | id=lc_run--01
  [2] human  | 第4轮对话随便聊 | id=c86f64f3-0
  [3] ai     | 当然，我们可以聊聊你最近的生活。 | id=lc_run--01
  [4] human  | 第5轮对话随便聊 | id=71bbd1ed-d
  [5] ai     | 好的，我们可以聊聊你最近的生活。 | id=lc_run--01

===== 最终 =====
最终消息条数： 6

第二种按照 token 阈值修剪，采用字符简易估算 token，当整体 token 超出上限时逐步删除旧消息，同时保留首条消息，trim_node 里统计全部消息总 token，如果超过阈值，从最旧消息开始删除，直到总 token 低于限制，保留第一条消息不删。

  • 说明：这里用字符近似估算 token（中文≈2 字符 = 1token，英文≈1 字符 = 1token）
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import AnyMessage, RemoveMessage

basic_model = ChatOpenAI(
    model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
    base_url="http://127.0.0.1:11433/v1",
    api_key="dummy",
    temperature=0.1,
    max_tokens=512,
)

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def estimate_message_tokens(msg: AnyMessage) -> int:
    """简易估算单条消息token"""
    content = str(msg.content)
    return len(content) // 2

def calc_total_tokens(messages: list[AnyMessage]) -> int:
    """计算消息列表总估算token"""
    total = 0
    for m in messages:
        total += estimate_message_tokens(m)
    return total

def dump_msg(graph, config) -> list[dict]:
    snap = graph.get_state(config)
    msgs = snap.values.get("messages", [])
    return [
        {"type": m.type, "content": m.content, "id": m.id[:10]}
        for m in msgs
    ]

def trim_node(state: State):
    """按token阈值自动物理修剪，强制保留第一条消息，旧消息逐步删除"""
    # 设置上下文最大允许token
    MAX_ALLOW_TOKENS = 100
    messages = state["messages"]
    if not messages:
        return {}

    total_tok = calc_total_tokens(messages)
    # token没超限，直接返回，不做处理
    if total_tok <= MAX_ALLOW_TOKENS:
        return {}

    first_msg = messages[0]
    # 从第二条开始往后，候选待删除消息（第一条强制保留）
    candidate_messages = messages[1:]

    keep = [first_msg]
    remove_ids = []

    # 倒序遍历，优先保留最新消息
    for m in reversed(candidate_messages):
        temp_list = keep + [m]
        if calc_total_tokens(temp_list) <= MAX_ALLOW_TOKENS:
            keep.append(m)
        else:
            remove_ids.append(m.id)

    # 注意：上面是倒序收集remove_ids，直接生成删除对象
    remove_list = [RemoveMessage(id=mid) for mid in remove_ids]
    return {"messages": remove_list}

def llm_node(state: State):
    resp = basic_model.invoke(state["messages"])
    return {"messages": [resp]}

if __name__ == "__main__":
    builder = StateGraph(State)
    builder.add_node("trim", trim_node)
    builder.add_node("llm", llm_node)

    builder.add_edge("trim", "llm")
    builder.set_entry_point("trim")

    checkpointer = InMemorySaver()
    graph = builder.compile(checkpointer=checkpointer)

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

    for i in range(8):
        user_text = f"第{i}轮对话随便聊，多一点内容增加token占用测试测试测试"
        out = graph.invoke({"messages": [("user", user_text)]}, cfg)
        print(f"\n===== 轮{i} =====")
        print(f"reply: {out['messages'][-1].content[:60]}")
        msg_list = dump_msg(graph, cfg)
        current_tokens = calc_total_tokens([m for m in graph.get_state(cfg).values["messages"]])
        print(f"当前消息总条数：{len(msg_list)}，估算总token={current_tokens}")
        for idx, item in enumerate(msg_list):
            print(f"  [{idx}] {item['type']:6s} | {item['content'][:30]}... | id={item['id']}")

    snap = graph.get_state(cfg)
    print("\n===== 最终 =====")
    print("最终消息条数：", len(snap.values["messages"]))
    print(f"最终估算总token：{calc_total_tokens(snap.values['messages'])}")

上述代码中，我们通过改写trim_node内的MAX_ALLOW_TOKENS变量为100，规定了Token最大消耗为100，如果超过了100则自动的删除前面的消息内容，并始终保持Token消耗在100-150范围内，输出效果如下所示；

CMD> main.py

===== 轮0 =====
reply: 好的，那我们就开始吧！有什么话题你想聊的吗？
当前消息总条数：2，估算总token=25
  [0] human  | 第0轮对话随便聊，多一点内容增加token占用测试测试测试... | id=e2e87dd3-0
  [1] ai     | 好的，那我们就开始吧！有什么话题你想聊的吗？... | id=lc_run--01

===== 轮1 =====
reply: 当然可以！我们可以聊聊你最近的生活、工作、兴趣爱好或者是你对某个话题的看法。有什么特别想聊的吗？
当前消息总条数：4，估算总token=63
  [0] human  | 第0轮对话随便聊，多一点内容增加token占用测试测试测试... | id=e2e87dd3-0
  [1] ai     | 好的，那我们就开始吧！有什么话题你想聊的吗？... | id=lc_run--01
  [2] human  | 第1轮对话随便聊，多一点内容增加token占用测试测试测试... | id=a476bbca-f
  [3] ai     | 当然可以！我们可以聊聊你最近的生活、工作、兴趣爱好或者是你对... | id=lc_run--01

===== 轮2 =====
reply: 好的，那我们继续聊天。你最近有什么新鲜事想分享吗？或者你有什么特别感兴趣的话题想要讨论？
当前消息总条数：6，估算总token=99
  [0] human  | 第0轮对话随便聊，多一点内容增加token占用测试测试测试... | id=e2e87dd3-0
  [1] ai     | 好的，那我们就开始吧！有什么话题你想聊的吗？... | id=lc_run--01
  [2] human  | 第1轮对话随便聊，多一点内容增加token占用测试测试测试... | id=a476bbca-f
  [3] ai     | 当然可以！我们可以聊聊你最近的生活、工作、兴趣爱好或者是你对... | id=lc_run--01
  [4] human  | 第2轮对话随便聊，多一点内容增加token占用测试测试测试... | id=f6e1cf0e-a
  [5] ai     | 好的，那我们继续聊天。你最近有什么新鲜事想分享吗？或者你有什... | id=lc_run--01

===== 轮3 =====
reply: 当然可以！我们可以聊聊你最近的生活、工作、兴趣爱好或者是你对某个话题的看法。有什么特别想聊的吗？
当前消息总条数：7，估算总token=123
  [0] human  | 第0轮对话随便聊，多一点内容增加token占用测试测试测试... | id=e2e87dd3-0
  [1] ai     | 好的，那我们就开始吧！有什么话题你想聊的吗？... | id=lc_run--01
  [2] ai     | 当然可以！我们可以聊聊你最近的生活、工作、兴趣爱好或者是你对... | id=lc_run--01
  [3] human  | 第2轮对话随便聊，多一点内容增加token占用测试测试测试... | id=f6e1cf0e-a
  [4] ai     | 好的，那我们继续聊天。你最近有什么新鲜事想分享吗？或者你有什... | id=lc_run--01
  [5] human  | 第3轮对话随便聊，多一点内容增加token占用测试测试测试... | id=a0d43249-6
  [6] ai     | 当然可以！我们可以聊聊你最近的生活、工作、兴趣爱好或者是你对... | id=lc_run--01

===== 轮4 =====
reply: 好的，那我们继续聊天。你最近有什么新鲜事想分享吗？或者你有什么特别感兴趣的话题想要讨论？
当前消息总条数：7，估算总token=121
  [0] human  | 第0轮对话随便聊，多一点内容增加token占用测试测试测试... | id=e2e87dd3-0
  [1] ai     | 好的，那我们就开始吧！有什么话题你想聊的吗？... | id=lc_run--01
  [2] ai     | 好的，那我们继续聊天。你最近有什么新鲜事想分享吗？或者你有什... | id=lc_run--01
  [3] human  | 第3轮对话随便聊，多一点内容增加token占用测试测试测试... | id=a0d43249-6
  [4] ai     | 当然可以！我们可以聊聊你最近的生活、工作、兴趣爱好或者是你对... | id=lc_run--01
  [5] human  | 第4轮对话随便聊，多一点内容增加token占用测试测试测试... | id=c50a190c-8
  [6] ai     | 好的，那我们继续聊天。你最近有什么新鲜事想分享吗？或者你有什... | id=lc_run--01

===== 轮5 =====
reply: 当然，我们可以聊聊你最近的旅行经历、最喜欢的书籍、或者是你对某个领域的兴趣。有什么特别想聊的吗？
当前消息总条数：7，估算总token=123
  [0] human  | 第0轮对话随便聊，多一点内容增加token占用测试测试测试... | id=e2e87dd3-0
  [1] ai     | 好的，那我们就开始吧！有什么话题你想聊的吗？... | id=lc_run--01
  [2] ai     | 当然可以！我们可以聊聊你最近的生活、工作、兴趣爱好或者是你对... | id=lc_run--01
  [3] human  | 第4轮对话随便聊，多一点内容增加token占用测试测试测试... | id=c50a190c-8
  [4] ai     | 好的，那我们继续聊天。你最近有什么新鲜事想分享吗？或者你有什... | id=lc_run--01
  [5] human  | 第5轮对话随便聊，多一点内容增加token占用测试测试测试... | id=4720d737-8
  [6] ai     | 当然，我们可以聊聊你最近的旅行经历、最喜欢的书籍、或者是你对... | id=lc_run--01

===== 轮6 =====
reply: 好的，我们可以聊聊你最近的旅行经历、最喜欢的书籍、或者是你对某个领域的兴趣。有什么特别想聊的吗？
当前消息总条数：7，估算总token=123
  [0] human  | 第0轮对话随便聊，多一点内容增加token占用测试测试测试... | id=e2e87dd3-0
  [1] ai     | 好的，那我们就开始吧！有什么话题你想聊的吗？... | id=lc_run--01
  [2] ai     | 好的，那我们继续聊天。你最近有什么新鲜事想分享吗？或者你有什... | id=lc_run--01
  [3] human  | 第5轮对话随便聊，多一点内容增加token占用测试测试测试... | id=4720d737-8
  [4] ai     | 当然，我们可以聊聊你最近的旅行经历、最喜欢的书籍、或者是你对... | id=lc_run--01
  [5] human  | 第6轮对话随便聊，多一点内容增加token占用测试测试测试... | id=7a653ef4-4
  [6] ai     | 好的，我们可以聊聊你最近的旅行经历、最喜欢的书籍、或者是你对... | id=lc_run--01

===== 轮7 =====
reply: 当然，我们可以聊聊你最近的旅行经历、最喜欢的书籍、或者是你对某个领域的兴趣。有什么特别想聊的吗？
当前消息总条数：6，估算总token=114
  [0] human  | 第0轮对话随便聊，多一点内容增加token占用测试测试测试... | id=e2e87dd3-0
  [1] ai     | 当然，我们可以聊聊你最近的旅行经历、最喜欢的书籍、或者是你对... | id=lc_run--01
  [2] human  | 第6轮对话随便聊，多一点内容增加token占用测试测试测试... | id=7a653ef4-4
  [3] ai     | 好的，我们可以聊聊你最近的旅行经历、最喜欢的书籍、或者是你对... | id=lc_run--01
  [4] human  | 第7轮对话随便聊，多一点内容增加token占用测试测试测试... | id=30b7e043-1
  [5] ai     | 当然，我们可以聊聊你最近的旅行经历、最喜欢的书籍、或者是你对... | id=lc_run--01

===== 最终 =====
最终消息条数： 6
最终估算总token：114

消息删除

在 LangGraph 中可通过RemoveMessage对会话历史消息进行删除管理，该组件需要配合add_messages状态归约器使用，支持删除单条指定消息、批量清空会话全部消息以及图节点内部运行时自动裁剪消息等多种场景，借助update_state接口能够在图外部手动操作消息状态，是实现会话清理、消息修剪的底层核心能力。

from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import AnyMessage, RemoveMessage, HumanMessage

llm = ChatOpenAI(
    model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
    base_url="http://127.0.0.1:11433/v1",
    api_key="dummy",
    temperature=0.1,
    max_tokens=512,
)

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def chat_node(state: State):
    resp = llm.invoke(state["messages"])
    return {"messages": [resp]}

def trim_node(state: State):
    """裁剪：保留最近4条消息，删除更早消息"""
    messages = state["messages"]
    keep_last_n = 4
    if len(messages) <= keep_last_n:
        return {}
    # 生成要删除的消息对象
    to_remove = [RemoveMessage(id=m.id) for m in messages[:-keep_last_n]]
    return {"messages": to_remove}

def dump_message_list(graph, config) -> list[dict]:
    """导出消息为字典列表，方便打印对比"""
    snapshot = graph.get_state(config)
    msgs = snapshot.values.get("messages", [])
    return [
        {"type": m.type, "content": m.content, "id": m.id}
        for m in msgs
    ]

if __name__ == "__main__":
    builder = StateGraph(State)
    builder.add_node("trim", trim_node)
    builder.add_node("chat", chat_node)
    builder.add_edge("trim", "chat")
    builder.set_entry_point("trim")

    checkpointer = InMemorySaver()
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "demo_001"}}

    print("========= 填充多轮对话，制造待删除的消息队列 =========\n")
    for turn in range(1, 7):
        graph.invoke(
            {"messages": [HumanMessage(content=f"消息{turn}")]},
            config
        )

    before_list = dump_message_list(graph, config)
    print(f"[填充完成｜删除前]消息总数 = {len(before_list)}")
    for idx, item in enumerate(before_list):
        print(f"  [{idx}] {item['type']:5s} | {item['content']} | id={item['id'][:8]}")
    print("-" * 90)

    # ========== 案例A：外部 update_state 删除【指定单条消息】 ==========
    print("\n========= 案例A：外部API删除第0号消息 =========")
    before_a = dump_message_list(graph, config)
    del_target_id = before_a[0]["id"]
    print(f"准备删除: index=0, content={before_a[0]['content']}, id={del_target_id[:8]}")

    graph.update_state(
        config,
        {"messages": [RemoveMessage(id=del_target_id)]}
    )

    after_a = dump_message_list(graph, config)
    print(f"[删除单条后]消息总数 = {len(after_a)}")
    for idx, item in enumerate(after_a):
        print(f"  [{idx}] {item['type']:5s} | {item['content']} | id={item['id'][:8]}")
    print("-" * 90)

    # ========== 案例B：清空会话全部消息（批量RemoveMessage） ==========
    print("\n========= 案例B：批量清空当前会话全部消息 =========")
    before_b = dump_message_list(graph, config)
    print(f"清空前消息总数 = {len(before_b)}")

    remove_all = [RemoveMessage(id=m["id"]) for m in before_b]
    graph.update_state(config, {"messages": remove_all})

    after_b = dump_message_list(graph, config)
    print(f"[全部清空后]消息总数 = {len(after_b)}")
    for idx, item in enumerate(after_b):
        print(f"  [{idx}] {item['type']:5s} | {item['content']}")
    print("-" * 90)

    # ========== 案例C：节点内部自动裁剪删除（运行时删除） ==========
    print("\n========= 案例C：Graph节点内部运行时自动裁剪删除 =========")
    # 重新填充多轮消息，触发trim_node删除旧消息
    for turn in range(1, 9):
        graph.invoke({"messages": [HumanMessage(content=f"运行时消息{turn}")]}, config)

    before_c = dump_message_list(graph, config)
    print(f"[节点裁剪执行完成后]消息总数 = {len(before_c)}")
    for idx, item in enumerate(before_c):
        print(f"  [{idx}] {item['type']:5s} | {item['content']} | id={item['id'][:8]}")

运行上述代码，程序预先填充多轮对话得到 5 条消息；

  • 案例 A 通过update_state外部调用RemoveMessage成功删除指定索引 0 的 AI 消息，消息总数变为 4 条；
  • 案例 B 批量构造RemoveMessage对象，完成会话全部消息清空，消息数归零；
  • 案例 C 重新写入多轮对话，图内trim_node执行运行时裁剪，保留最近 4 条消息，最终得到 5 条消息（1 条 AI + 人机交互 4 条）。
CMD> main.py
========= 填充多轮对话，制造待删除的消息队列 =========

[填充完成｜删除前]消息总数 = 5
  [0] ai    | 对不起，我无法查看或处理消息4。如果您有其他问题或需要帮助。 | id=lc_run--
  [1] human | 消息5 | id=ecde0029
  [2] ai    | 对不起，我无法查看或处理消息5。如果您有其他问题或需要帮助。 | id=lc_run--
  [3] human | 消息6 | id=d372f076
  [4] ai    | 对不起，我无法查看或处理消息6。如果您有其他问题或需要帮助。 | id=lc_run--
------------------------------------------------------------------------------------------

========= 案例A：外部API删除第0号消息 =========
准备删除: index=0, content=对不起，我无法查看或处理消息4。如果您有其他问题或需要帮助。, id=lc_run--
[删除单条后]消息总数 = 4
  [0] human | 消息5 | id=ecde0029
  [1] ai    | 对不起，我无法查看或处理消息5。如果您有其他问题或需要帮助。 | id=lc_run--
  [2] human | 消息6 | id=d372f076
  [3] ai    | 对不起，我无法查看或处理消息6。如果您有其他问题或需要帮助。 | id=lc_run--
------------------------------------------------------------------------------------------

========= 案例B：批量清空当前会话全部消息 =========
清空前消息总数 = 4
[全部清空后]消息总数 = 0
------------------------------------------------------------------------------------------

========= 案例C：Graph节点内部运行时自动裁剪删除 =========
[节点裁剪执行完成后]消息总数 = 5
  [0] ai    | 对不起，我无法提供关于"运行时消息6"的信息。 | id=lc_run--
  [1] human | 运行时消息7 | id=0f39c38d
  [2] ai    | 对不起，我无法提供关于"运行时消息7"的信息。 | id=lc_run--
  [3] human | 运行时消息8 | id=0c293edf
  [4] ai    | 对不起，我无法提供关于"运行时消息8"的信息。 | id=lc_run--

消息总结

直接对历史消息进行修剪删除会丢失对话关键信息，消息摘要则是更优的处理方案。它不会直接丢弃旧对话，当消息总 token 超过设定阈值时，提取老旧历史消息交由大模型生成简短摘要，删除原始旧消息，将摘要消息加入消息列表，同时保留最近若干条原始对话，以此控制上下文 token 占用，最大程度留存对话关键信息。

示例代码实现完整摘要逻辑，采用字符简易估算 token，将旧对话压缩为 SystemMessage 摘要消息，生产环境建议替换模型原生 tokenizer 完成 token 统计。

from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import AnyMessage, RemoveMessage, HumanMessage, AIMessage, SystemMessage

basic_model = ChatOpenAI(
    model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
    base_url="http://127.0.0.1:11433/v1",
    api_key="dummy",
    temperature=0.1,
    max_tokens=512,
)

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def estimate_message_tokens(msg: AnyMessage) -> int:
    """简易估算token"""
    content = str(msg.content)
    return len(content) // 2

def calc_total_tokens(messages: list[AnyMessage]) -> int:
    total = 0
    for m in messages:
        total += estimate_message_tokens(m)
    return total

def dump_msg(graph, config) -> list[dict]:
    snap = graph.get_state(config)
    msgs = snap.values.get("messages", [])
    return [
        {"type": m.type, "content": m.content, "id": m.id[:10]}
        for m in msgs
    ]

def summarize_old_messages(messages: list[AnyMessage]) -> str:
    """调用LLM，把一批历史消息压缩成摘要文本"""
    prompt = """请把下面的用户与助手对话，精简为简短的对话摘要，保留关键信息，不要冗余：
对话记录：
"""
    for m in messages:
        role = "用户" if isinstance(m, HumanMessage) else "助手"
        prompt += f"\n{role}: {m.content}"
    resp = basic_model.invoke([HumanMessage(content=prompt)])
    return resp.content

def trim_or_summarize_node(state: State):
    """
    消息摘要压缩节点
    规则：
    1. 总token < 阈值：什么都不做
    2. 超过阈值：把前面老旧一批消息做摘要，删除原始旧消息，插入摘要消息，保留最近N条原始对话
    """
    # 总Token限制
    MAX_TOKENS = 100

    # 保留最近4条原始对话不压缩
    RECENT_KEEP_COUNT = 4
    messages = state["messages"]
    total_tok = calc_total_tokens(messages)

    if total_tok <= MAX_TOKENS:
        return {}

    # 需要压缩：拆分老旧待压缩消息 + 保留最近原始消息
    if len(messages) <= RECENT_KEEP_COUNT:
        return {}

    to_summarize = messages[:-RECENT_KEEP_COUNT]   # 前面旧消息，拿去做摘要
    keep_recent = messages[-RECENT_KEEP_COUNT:]    # 最近几条原样保留

    # 调用模型生成摘要
    summary_text = summarize_old_messages(to_summarize)
    summary_msg = SystemMessage(content=f"[历史对话摘要]: {summary_text}")

    # 删除全部待压缩的原始旧消息
    remove_ids = [m.id for m in to_summarize]
    remove_list = [RemoveMessage(id=mid) for mid in remove_ids]

    # remove_list：删掉旧消息；同时追加摘要消息
    return {
        "messages": [*remove_list, summary_msg]
    }

def llm_node(state: State):
    resp = basic_model.invoke(state["messages"])
    return {"messages": [resp]}

if __name__ == "__main__":
    builder = StateGraph(State)
    builder.add_node("summarize_trim", trim_or_summarize_node)
    builder.add_node("llm", llm_node)

    builder.add_edge("summarize_trim", "llm")
    builder.set_entry_point("summarize_trim")

    checkpointer = InMemorySaver()
    graph = builder.compile(checkpointer=checkpointer)

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

    for i in range(10):
        user_text = f"第{i}轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试"
        out = graph.invoke({"messages": [("user", user_text)]}, cfg)
        print(f"\n===== 轮{i} =====")
        print(f"reply: {out['messages'][-1].content[:60]}")
        msg_list = dump_msg(graph, cfg)
        current_tok = calc_total_tokens(graph.get_state(cfg).values["messages"])
        print(f"消息总条数={len(msg_list)} 估算token={current_tok}")
        for idx, item in enumerate(msg_list):
            print(f"  [{idx}] {item['type']:8s} | {item['content'][:55]}")

    snap = graph.get_state(cfg)
    print("\n=====最终消息队列=====")
    final_msgs = dump_msg(graph, cfg)
    for idx, item in enumerate(final_msgs):
        print(f"[{idx}] {item['type']:8s} | {item['content']}")

上述代码运行后，程序将依次循环执行 10 轮对话，前期消息数量与估算 token 持续上涨；当总 token 超过设置阈值 100 后，摘要压缩逻辑触发，会删除老旧原始消息，生成 SystemMessage 类型的历史摘要，同时保留最近 4 条原始消息，消息条数稳定维持在 6 条左右。

CMD> main.py

===== 轮0 =====
reply: 很抱歉，我无法理解您的问题。请提供更具体的信息或问题描述，以便我能够更好地帮助您。
消息总条数=2 估算token=36
  [0] human    | 第0轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [1] ai       | 很抱歉，我无法理解您的问题。请提供更具体的信息或问题描述，以便我能够更好地帮助您。

===== 轮1 =====
reply: 好的，我明白了。请提供第1轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。
消息总条数=4 估算token=74
  [0] human    | 第0轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [1] ai       | 很抱歉，我无法理解您的问题。请提供更具体的信息或问题描述，以便我能够更好地帮助您。
  [2] human    | 第1轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [3] ai       | 好的，我明白了。请提供第1轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。

===== 轮2 =====
reply: 好的，我明白了。请提供第2轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。
消息总条数=6 估算token=112
  [0] human    | 第0轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [1] ai       | 很抱歉，我无法理解您的问题。请提供更具体的信息或问题描述，以便我能够更好地帮助您。
  [2] human    | 第1轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [3] ai       | 好的，我明白了。请提供第1轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。
  [4] human    | 第2轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [5] ai       | 好的，我明白了。请提供第2轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。

===== 轮3 =====
reply: 好的，我明白了。请提供第3轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。
消息总条数=6 估算token=126
  [0] ai       | 好的，我明白了。请提供第1轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。
  [1] human    | 第2轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [2] ai       | 好的，我明白了。请提供第2轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。
  [3] human    | 第3轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [4] system   | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手表示无法理解问题并建议提供更具体的信息
  [5] ai       | 好的，我明白了。请提供第3轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。

===== 轮4 =====
reply: 对不起，我无法提供第2轮对话的内容。如果您有其他问题或需要帮助，请随时告诉我。
消息总条数=6 估算token=127
  [0] human    | 第3轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [1] system   | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手表示无法理解问题并建议提供更具体的信息
  [2] ai       | 好的，我明白了。请提供第3轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。
  [3] human    | 第4轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [4] system   | [历史对话摘要]: 助手：明白了，请提供第2轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。
  [5] ai       | 对不起，我无法提供第2轮对话的内容。如果您有其他问题或需要帮助，请随时告诉我。

===== 轮5 =====
reply: 明白了，用户需要提供第3轮对话的内容，以便进行文本长度增加和摘要压缩效果的测试。
消息总条数=6 估算token=147
  [0] human    | 第4轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [1] system   | [历史对话摘要]: 助手：明白了，请提供第2轮对话的内容，我会帮助您进行文本长度增加和摘要压缩效果的测试。
  [2] ai       | 对不起，我无法提供第2轮对话的内容。如果您有其他问题或需要帮助，请随时告诉我。
  [3] human    | 第5轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [4] system   | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手表示无法理解问题并建议提供更具体的信息
  [5] ai       | 明白了，用户需要提供第3轮对话的内容，以便进行文本长度增加和摘要压缩效果的测试。

===== 轮6 =====
reply: 对不起，我无法提供第2轮对话的内容。
消息总条数=6 估算token=126
  [0] human    | 第5轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [1] system   | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手表示无法理解问题并建议提供更具体的信息
  [2] ai       | 明白了，用户需要提供第3轮对话的内容，以便进行文本长度增加和摘要压缩效果的测试。
  [3] human    | 第6轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [4] system   | [历史对话摘要]: 助手：对不起，我无法提供第2轮对话的内容。
  [5] ai       | 对不起，我无法提供第2轮对话的内容。

===== 轮7 =====
reply: [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议用户提供第3轮对话的内容，以便进行测试。
消息总条数=6 估算token=114
  [0] human    | 第6轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [1] system   | [历史对话摘要]: 助手：对不起，我无法提供第2轮对话的内容。
  [2] ai       | 对不起，我无法提供第2轮对话的内容。
  [3] human    | 第7轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [4] system   | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议用户提供第3轮对话的内容，以便进行
  [5] ai       | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议用户提供第3轮对话的内容，以便进行

===== 轮8 =====
reply: [历史对话摘要]: 用户询问了关于第6轮对话的内容，但助手表示无法提供第2轮对话的内容。
消息总条数=6 估算token=134
  [0] human    | 第7轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [1] system   | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议用户提供第3轮对话的内容，以便进行
  [2] ai       | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议用户提供第3轮对话的内容，以便进行
  [3] human    | 第8轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [4] system   | [历史对话摘要]: 用户询问了关于第6轮对话的内容，但助手表示无法提供第2轮对话的内容。
  [5] ai       | [历史对话摘要]: 用户询问了关于第6轮对话的内容，但助手表示无法提供第2轮对话的内容。

===== 轮9 =====
reply: [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议提供第3轮对话的内容。
消息总条数=6 估算token=124
  [0] human    | 第8轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [1] system   | [历史对话摘要]: 用户询问了关于第6轮对话的内容，但助手表示无法提供第2轮对话的内容。
  [2] ai       | [历史对话摘要]: 用户询问了关于第6轮对话的内容，但助手表示无法提供第2轮对话的内容。
  [3] human    | 第9轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
  [4] system   | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议提供第3轮对话的内容。
  [5] ai       | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议提供第3轮对话的内容。

=====最终消息队列=====
[0] human    | 第8轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
[1] system   | [历史对话摘要]: 用户询问了关于第6轮对话的内容，但助手表示无法提供第2轮对话的内容。
[2] ai       | [历史对话摘要]: 用户询问了关于第6轮对话的内容，但助手表示无法提供第2轮对话的内容。
[3] human    | 第9轮对话，测试内容，增加文本长度，测试摘要压缩效果测试测试测试
[4] system   | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议提供第3轮对话的内容。
[5] ai       | [历史对话摘要]: 用户询问了关于增加文本长度的摘要压缩效果的测试。助手建议提供第3轮对话的内容。

随着对话不断推进，会反复执行摘要生成，新的摘要消息持续加入消息队列。受本地模型应答能力影响，模型有时会直接把摘要内容当作回复输出，造成消息列表中同时存在 system 摘要与 ai 重复摘要内容。整体来看压缩控制 token 条数的逻辑生效，但摘要质量依赖大模型本身能力。