Charles Shao
Go back

LangGraph 长期记忆：跨 Thread 持久化与全局 Store 注入

创建于:
v1.0.0
–views

在时光旅行：回溯状态、断点重跑与分支分叉中，我们彻底挖掘了 Checkpointer 的能力：它可以将单个 thread_id 内的每一次对话状态保存为不可变的版本树，让我们能够断点续传甚至篡改分叉。

然而，无论 Checkpointer 记录得多么精细，它的生命周期都被死死锁在了当前这个 thread_id 内部。

如果用户今天开启了一个新线程，说：“我们账户以后追投都只用 aggressive 档”，明天开启了另一个全新的线程说：“帮「双11-美妆拉新」追投两天，就按我的偏好来”。因为 thread_id 不同，新线程的 Checkpointer 从零开始，Agent 会像失忆一样傻傻反问：“请问这次追投用哪个档位？”

要想跨越 thread_id 的次元壁，让 Agent 沉淀出跨会话的长期认知（例如用户画像、全局知识库、全局配置），我们必须引入 LangGraph 持久化体系的另一块核心拼图——Store（存储）。

Table of contents

Open Table of contents

一、双库架构：Checkpointer 与 Store 的职责分工

在构建复杂的 Agent 系统时，记忆系统往往会被切分为两层结构：工作记忆（Working Memory） 与 长期记忆（Long-Term Memory）。在 LangGraph 中，它们分别由 Checkpointer 与 Store 来承载。

必须严格区分它们的定位与数据组织形式：

  1. Checkpointer（工作记忆 / 状态栈）：

    • 作用域：局限于单一的 thread_id，本质是一次任务、一张追投单或一次连贯对话的现场。
    • 数据结构：追加写入的不可变版本树（StateSnapshot）。
    • 核心职责：管理图节点状态，支持打断点（Human-in-the-loop）、断点续传、时光旅行和并发回滚。

  2. Store（长期记忆 / 事实库）：

    • 作用域：全局的，或者与特定 user_id / org_id 绑定，能在无数个 thread_id 间共享。
    • 数据结构：Key-Value 文档，按 Namespace（命名空间元组）分层组织，就地覆写更新（Mutable）。
    • 核心职责：管理跨会话的用户偏好、静态规则、企业知识文档或向量化数据。

在一次对话中，Agent 的大脑不仅要看自己在这个 thread_id 中走到哪一步了（Checkpointer），还要时刻能查阅并更新关于该用户的长效事实（Store）。

二、引入 Store 记忆库

为了让 Agent 在追投过程中自动记住用户的档位偏好，我们需要把 BaseStore 实例挂载到图中。

与 Checkpointer 类似，LangGraph 为 Store 提供了多种实现：

在调用 create_agent（或 compile()）时，将实例传入 store 参数即可：

from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.sqlite import SqliteSaver

# ... 之前的 SQLite 连接建立
store = InMemoryStore()

graph = create_agent(
    model=chat_model(),
    tools=TOOLS,
    system_prompt=SYSTEM,
    # 通过 Middleware 拦截器注入记忆（下文详解）
    middleware=[PreferenceMiddleware()],
    state_schema=State,
    checkpointer=SqliteSaver(conn),
    store=store,  # 挂载长期记忆库
)

三、操作 Store 的三个维度

Agent 如果要利用 Store，就必须掌握如何写和如何读。LangGraph 提供了非常优雅的原生接口，其中最核心的概念就是 Namespace。

1. 命名空间（Namespace）设计

Store 并非扁平的字典，它强制通过层级化的命名空间（元组类型）来隔离数据，防范数据污染。例如，针对用户的偏好，我们可以设计这样的结构： ("users", "u_001", "preferences")

在这个 Namespace 之下，我们可以存放不同的键，比如 tier 档位偏好、blocked_placements 屏蔽版位等。

2. 在 Tool 中写入长期记忆：InjectedStore

我们希望 Agent 能够主动在对话中提取并记忆用户的偏好。最好的方式是提供一个工具（Tool），并在这个工具中把 Store 实例以依赖注入（Dependency Injection）的方式拿出来使用。

使用 InjectedStore 类型标注，LangGraph 会在调用工具时自动将全局配置的 Store 实例塞进去：

from langgraph.prebuilt import InjectedStore
from langgraph.store.base import BaseStore
from typing import Annotated
from langchain.tools import tool

USER_ID = "u_001"

@tool
def save_user_preference(
    preference: str,
    # 框架检测到 InjectedStore，会自动注入当前挂载的 store 实例
    store: Annotated[BaseStore, InjectedStore()]
) -> str:
    """保存用户的偏好以供未来对话使用。当用户提到自己的习惯、偏好时调用。"""

    # 存入 Store：命名空间、键名、字典值
    store.put(
        ("users", USER_ID, "preferences"),
        "tier",
        {"value": preference}
    )
    return "Preference saved."

通过这一层封装，Agent 只需要像调用普通工具一样思考：“我应该调用 save_user_preference 把这个喜好记下来”。剩下的脏活累活全由底层图机制搞定。

3. 在 Middleware 中动态读取并注入提示词

记忆存下去了，怎么让 Agent 在下一次完全不同的对话（另一个 thread_id）里回忆起来呢？

LangGraph/LangChain 的现代架构中，通过 AgentMiddleware（中间件拦截器）是极其优雅的解法。我们在每次大模型调用前，在它的 SystemMessage 开头动态拼上属于当前用户的记忆文本。

from langchain.agents.middleware import AgentMiddleware, ModelRequest
from langchain_core.messages import SystemMessage

class PreferenceMiddleware(AgentMiddleware):
    def wrap_model_call(self, request: ModelRequest, handler):
        # 1. 从 Request 运行时中拿到 store 实例
        if request.runtime.store:
            # 2. 搜索属于该用户的长效记忆
            prefs = request.runtime.store.search(("users", USER_ID, "preferences"))
            if prefs:
                pref_text = "\n\n[用户偏好]\n"
                for p in prefs:
                    pref_text += f"- {p.key}: {p.value.get('value')}\n"

                # 3. 追加到原来的 System Prompt 末尾，构成新的 SystemMessage
                existing = request.system_message.content if request.system_message else ""
                new_msg = SystemMessage(content=existing + pref_text)

                # 4. 覆盖请求中的 system_message，原样不可变（Immutable）传递给 handler
                request = request.override(system_message=new_msg)

        # 放行请求给后续链路（大模型调用）
        return handler(request)

为什么这种方案好？ 如果强迫 Agent 每次都调用 read_preferences 工具去读取，会凭空消耗一轮 Agent -> Tool -> Agent 的大模型往返，大幅增加延迟和成本。在 before_model 层面前置拦截读取，等于把环境知识在物理层无缝塞入了模型大脑。

四、跨会话实测：让历史指引未来

写个主函数来跑两个完全独立的 Thread，验证 Store 是否真正跨越了次元壁。

第一幕：在旧线程中记忆偏好

# 线程 1 开启
config1 = {"configurable": {"thread_id": "thread-1-save-pref"}}
goal1 = "我们账户以后追投都只用 aggressive 档。"

first = graph.invoke(
    {"messages": [HumanMessage(content=goal1)], "seen_adgroup_ids": [], "orders": {}},
    config1
)

底层执行轨迹：

  1. 拦截器 PreferenceMiddleware 读取 Store，空无一物。
  2. 模型看到用户的指令，决定调用 save_user_preference(preference="aggressive")。
  3. InjectedStore 捕获到操作，在 ("users", "u_001", "preferences") 命名空间下落盘 {"tier": {"value": "aggressive"}}。

第二幕：在新线程中自动应用

隔天，用户开了一个完全独立的新会话：

# 线程 2 开启，全新的 thread_id
config2 = {"configurable": {"thread_id": "thread-2-use-pref"}}
goal2 = "计划「双11-美妆拉新」11 月 1 日起追投两天，帮我看看有哪些广告组，按我的偏好挑档位。"

second = graph.invoke(
    {"messages": [HumanMessage(content=goal2)], "seen_adgroup_ids": [], "orders": {}},
    config2
)

底层执行轨迹：

  1. 拦截器发威！PreferenceMiddleware 去 Store 捞取，命中昨天存下的 aggressive。
  2. 拦截器默默篡改大模型的 SystemMessage，塞入 [用户偏好] - tier: aggressive。
  3. 模型读取到上下文，先调用 search_campaign("双11-美妆拉新") 拿到 CP-1024，再调用 search_adgroups("CP-1024", "2026-11-01")。
  4. 搜索工具回灌计划下的三个广告组。
  5. 模型精准挑出唯一支持 aggressive 档（已过学习期）的 AG-002，调用 boost_adgroup 下追投单！

没有任何多余的提问或确认，Agent 的智力获得了跨越时间维度的升华。

接下来

至此，我们的**存储三板斧（State 业务状态通道、Checkpointer 线程版本树、Store 跨线程记忆库）**已全部集齐。借助它们，哪怕节点挂机、代码重跑或跨月聊天，Agent 系统都拥有了健壮且连贯的状态韧性。

然而在这个流程中，依然有一个薄弱点：当我们的 get_campaign_report 等工具内部网络请求阻塞或缓慢时，整个图依然会被同步卡死；当我们的系统规模变大，我们希望能以流式（Streaming）将节点的思考过程推送到前端。 在下一篇流式输出与事件溯源中，我们将从底层的 invoke 切换到现代框架标配的 stream，剖析 LangGraph 事件订阅的设计。

–views
Share this post on:
Share this post via WhatsApp Share this post on Facebook Share this post on X Share this post via Telegram Share this post on Pinterest Share this post via email
Previous Post
Kafka 生产踩坑与排障实战：AdTech 企业级线上事故复盘与反模式
Next Post
LangGraph 时光旅行：回溯状态、断点重跑与分支分叉