langgraph教程系列-08-让agent记住过去-长期记忆

2401_88756258

902人浏览 · 2026-08-03 22:49:49

2401_88756258 · 2026-08-03 22:49:49 发布

本文是「LangGraph 教程系列」第 8 篇。写作时基于 langgraph 1.2.10、langchain 1.3.14、langchain-openai 1.4.1、Python 3.12+。配套代码仓库 https://github.com/wxj006007/deep-research-assistant ，本篇对应 tag v2.2。

上一版的研究助手已经会停下来等人审批。审批后恢复得很好，只要还是同一个 thread_id，它就记得这轮检索查过什么、现在该去哪一个节点。

但用户换一条新会话呢？比如昨天研究过 checkpoint，今天继续问 Store；又比如他每次都希望中文回答、先给结论再列步骤。checkpoint 不应承担这些跨会话信息——它记录的是某一次执行，不是用户档案。

这一篇把“记住过去”拆清楚：短期记忆仍归 checkpointer，长期记忆交给 Store。

一、三种东西，不要都叫 memory

“记忆”在 agent 系统里很容易被说成一个模糊概念。这里至少有三层：

层次	存放内容	生命周期	本例实现
当前 state	当前节点需要的 question、docs、round	一次图运行	`ResearchState`
短期记忆	可暂停、恢复、回放的执行快照	一个 `thread_id`	`MemorySaver`
长期记忆	用户偏好、历史研究摘要	跨多个 thread	`InMemoryStore`

前两层跟着图的执行走；第三层以用户为边界，独立于某次执行。把它们混在一起，常见后果是：新会话读不到信息，或者所有旧资料不断塞进 state，checkpoint 越来越大。

二、v2.2 的记忆分层

一个用户可以有多条 thread；每条 thread 的 checkpoint 相互隔离，但它们都可以读取同一个用户 namespace 下的长期 Store。反过来，另一个用户即便问了同样的问题，也不应读取到前者的资料。

三、Store 的最小模型：namespace、key 和 value

这一篇故意不引入 embedding 或向量库。我们先用可复现的 namespace/key 模型讲清数据归属：

("users", user_id, "profile")   -> "preferences"
("users", user_id, "research")  -> "summaries"

第一条记录保存稳定的用户偏好，第二条记录保存一个摘要列表。它们的 value 都是普通字典：

store.put(
    ("users", "alice", "profile"),
    "preferences",
    {
        "topics": ["LangGraph"],
        "style": "先给结论，再给简洁步骤",
        "language": "中文",
    },
    index=False,
)

index=False 明确表示本例不建向量索引、不做语义检索。以后需要从大量历史资料中按自然语言召回时，再讨论 search()、embedding 和索引策略；现在的重点是把跨会话存取边界设计正确。

四、把用户身份放进运行时上下文

用户身份不是问题本身，也不该作为每个节点都写入的 state 字段。LangGraph 的 Runtime 适合承载这类运行时信息：

from dataclasses import dataclass

@dataclass(frozen=True)
class ResearchContext:
    user_id: str
    research_id: str

编译图时同时传入 checkpointer 与 store：

graph = builder.compile(
    checkpointer=MemorySaver(),
    store=InMemoryStore(),
)

节点通过 runtime.context 知道当前用户，通过 runtime.store 读写长期记忆：

def load_memory_node(
    _: MemoryResearchState, runtime: Runtime[ResearchContext]
) -> dict:
    profile = runtime.store.get(profile_namespace(runtime.context.user_id), "preferences")
    summaries = runtime.store.get(research_namespace(runtime.context.user_id), "summaries")
    return {
        "memory_context": {
            "preferences": profile.value if profile else {},
            "summaries": summaries.value.get("items", []) if summaries else [],
        }
    }

memory_context 只是这次执行读取到的记忆副本，放入 state 供 planner 和 writer 使用；真正长期保存的数据仍在 Store 中。这样 checkpoint 保存的是“本轮曾读取了什么上下文”，Store 保存的是“下次还能取到什么档案”。

五、图结构：先读，后写

v2.2 保留第7篇的人在回路。新增两个节点：开始时 load_memory，成稿后 persist_memory。

读取节点在 plan 前，因此查询规划可以根据用户的语言、风格和已有研究避免重复。保存节点只在成功成稿后执行；用户在审批阶段拒绝任务时，不应凭空新增一条“研究完成”记忆。

六、保存摘要时为什么要有 `research_id`

图可能因重试、进程恢复或调用方重复提交而再次执行。若每次都 append，一项研究会在长期记忆里留下多份重复摘要。

因此 v2.2 的上下文里还有 research_id。persist_memory_node 写入前会删除同 id 的旧项，再保存新项，并把历史限制在最近 5 条：

items = [item for item in items if item.get("research_id") != research_id]
items.append(
    {
        "research_id": research_id,
        "question": state["question"],
        "summary": state.get("answer", "")[:500],
    }
)

store.put(namespace, "summaries", {"items": items[-5:]}, index=False)

这不是分布式事务的完整方案，但已经给出一个重要的工程习惯：外部写入要有稳定业务 id，并且能安全重试。

七、跨会话演示

先为 Alice 保存偏好，再在 thread A 中完成研究：

context_a = ResearchContext(user_id="alice", research_id="checkpoint-basics")
thread_a = {"configurable": {"thread_id": "alice-thread-a"}}

first = graph.invoke(
    {"question": "checkpoint 和长期记忆有什么区别？"},
    config=thread_a,
    context=context_a,
)

之后 Alice 在一个全新的 thread B 提问：

context_b = ResearchContext(user_id="alice", research_id="memory-followup")
thread_b = {"configurable": {"thread_id": "alice-thread-b"}}

second = graph.invoke(
    {"question": "那 Store 应该保存哪些数据？"},
    config=thread_b,
    context=context_b,
)

assert second["memory_context"]["preferences"]["language"] == "中文"
assert any(
    item["research_id"] == "checkpoint-basics"
    for item in second["memory_context"]["summaries"]
)