告别 Token 焦虑：LangGraph 工业级记忆模块实现深度指南

在构建长对话 AI Agent 时，开发者通常会面临一个“不可能三角”：对话长度、推理成本（Token 消耗）以及记忆准确度。

传统的滑动窗口（Sliding Window）直接丢弃旧消息，会导致 AI 忘记半小时前聊过的核心需求；而全量存储则会让上下文迅速膨胀，最终触及模型限制并让你的账单爆炸。

今天，我们分享一种在 LangGraph 中实现的工业级方案：滚动摘要（Rolling Summary）+ 状态裁剪（State Trimming）+ 异构存储（Heterogeneous Storage）。

---

1. 核心架构设计

我们的目标是建立一套“像人类一样记忆”的系统：短期细节清晰，长期重点明确，完整历史可查。

三位一体的存储策略

1. Checkpointer (热存储)：存储在数据库（如 SQLite/Postgres）中的精简 State。只包含最新的 $N$ 条消息和一份不断更新的全局摘要。
2. State (运行时内存)：利用 LangGraph 的 add_messages 机制，通过 RemoveMessage 指令动态裁剪对话列表。
3. MySQL (冷存储)：异步记录每一条原始对话，仅供前端 UI 展示历史记录，不参与模型推理。

---

2. 关键技术实现

A. 定义智能状态 (State)

在 LangGraph 中，State 是唯一的真理。我们必须通过 Annotated 告诉系统如何处理消息的增量更新。

from typing import Annotated, List, Union
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage, RemoveMessage
from langgraph.graph.message import add_messages

class RAGState(TypedDict):
    # add_messages 会处理逻辑：如果是新消息就追加，如果是 RemoveMessage 就删除
    messages: Annotated[List[AnyMessage], add_messages]
    # 全局滚动摘要
    summary: str
    # 业务 ID
    conversation_id: str

B. 滚动压缩节点 (The Compactor Node)

这是记忆模块的心脏。当消息超过阈值时，它负责将“旧闻”提炼成“摘要”。

async def memory_compact_node(state: RAGState):
    messages = state["messages"]
    
    # 触发检查：如果消息没满，直接跳过
    if len(messages) <= MAX_MESSAGES:
        return {}

    # 1. 划分归档区和保留区
    to_archive = messages[:-KEEP_RECENT]
    
    # 2. 迭代摘要：旧摘要 + 旧消息片段 = 新摘要
    new_summary = await llm.ainvoke(f"现有摘要: {state['summary']}\n追加内容: {to_archive}")
    
    # 3. 构造删除指令
    # 关键：必须传入消息的 ID，LangGraph 才会执行逻辑删除
    delete_actions = [RemoveMessage(id=m.id) for m in to_archive]
    
    return {
        "messages": delete_actions,
        "summary": new_summary.content,
        "_archived_data": to_archive  # 传递给下游异步节点
    }

---

3. 为什么这是“最佳实践”？

1. 彻底解决 Token 膨胀

无论用户与 Agent 聊了 100 轮还是 1000 轮，由于 memory_compact_node 的存在，输入给 LLM 的 messages 永远被控制在 $N$ 条以内。

2. 完美的断点续传

依靠 LangGraph 内置的 Checkpointer，你只需要一个 thread_id 就能恢复现场：

# 自动加载该用户最后的 summary 和 裁剪后的 messages
config = {"configurable": {"thread_id": "user_888"}}
app.invoke(input_data, config)

3. 用户体验与模型性能的平衡

• 模型侧：只读精简后的 State，推理速度快，事实聚焦。
• 用户侧：前端通过接口读取 MySQL，看到的是完整的对话流（包括那些已经被模型“裁剪”掉的消息），体验连贯。

---

4. 实施建议与避坑指南

• ID 陷阱：在使用 RemoveMessage 时，务必确保消息对象拥有唯一的 id。如果是手动构造的消息，记得手动指定 ID。
• 摘要策略：在重写摘要的 Prompt 中，务必加入“保留关键事实（如姓名、地址、特定偏好）”的强约束，防止摘要在多次迭代后变得模糊。
• 异步落库：将 MySQL 的写入逻辑放在一个 archive_node 中，并使用 asyncio.create_task 处理，避免数据库 IO 阻塞 Agent 的响应速度。

---

结语

在 AI Agent 的长跑中，记忆力决定了它能跑多远。通过 LangGraph 的 State 机制配合滚动摘要方案，你不仅在技术上实现了真正的“长效记忆”，更在工程架构上实现了极高的性价比。

你的 Agent，现在准备好应对无限的对话了吗？

---

作者注：本方案已在生产环境下验证，能有效支撑 50 轮以上的深度对话场景。