LangMem / LangGraph：记忆应该在 hot path，还是交给后台 manager

同样是“用户偏好要长期保存”，放在不同执行路径里，系统性格会完全不同。让 agent 当场调用工具保存，记忆写入可解释、可控， 但会占用模型注意力和工具调用预算；让后台 manager 稍后处理，主对话更干净，也更适合合并和去重，但用户可能马上问起刚说过的偏好， 这时系统要面对最终一致性。

本篇的代表性单位是一条偏好：“以后默认用英文回答我”。它可以在当前回答路径里立刻写入，也可以等后台 manager 看完对话快照和相似旧记忆后再写。两条路径都可能正确，差别在于谁能马上看见它、谁负责合并旧状态，以及失败时会影响哪一层。

一、LangMem 先把记忆拆成两条执行路径

1.1 同一条记忆，放在前台和后台会变成不同系统

LangMem README 的关键功能列表直接给出两条路线：一条是 memory management tools， 让 agents 在 active conversations 里记录和搜索信息，也就是 hot path；另一条是 background memory manager， 自动抽取、合并、更新 agent knowledge；两条路线都能接到 LangGraph 的 Long-term Memory Store。 这个定位可以看 README 功能列表。

这不是实现细节，而是产品性格。用户明确说“以后都用英文回答我”，如果当前回答路径里立刻写入，下一轮就能生效， 但模型要多承担一次工具选择和写入动作；如果交给后台 manager，当前回答更干净，也更容易合并多条信号， 但短时间内用户可能追问“你刚刚不是记住了吗”。LangMem 的价值，是把这两种性格都做成显式路径。

同一句话：“以后默认用英文回答我”

Hot path：
  当前 agent 判断这是长期偏好
  -> 调用 manage_memory 写入 BaseStore
  -> 下一轮 prompt / search_memory 可以立刻取到

Background：
  当前 agent 先正常回答
  -> 对话片段稍后交给 manager
  -> manager 检索旧偏好、判断新增还是更新
  -> 再把结果写回 BaseStore

后面所有机制都可以放回这个例子里理解：hot path 追求“现在就记住”，background 追求“整理得更稳”。 LangGraph 的 store 和 checkpointer 分工，则决定这条偏好究竟是跨会话长期可用， 还是只属于当前 thread 的可恢复状态。

1.2 先按可见性、延迟和治理压力选路径

README 的 quickstart 也体现了这点。示例先创建带 embedding index 的 InMemoryStore， 再把 create_manage_memory_tool 和 create_search_memory_tool 放进 create_react_agent 的 tools 里，同时把同一个 store 传给 agent。 README 解释说，这些 tools 让 agent 控制保存什么；当用户问起过去交互，LLM 可以调用 search tool 找相似内容。 代码和解释分别在 README 创建 agent 示例 与 后续说明。

二、Hot path：agent 自己决定要不要记住

2.1 manage tool 是模型主动触发的副作用

Hot path quickstart 一开头就把这条路线说清楚：memory 可以由 agent 在当前路径里“consciously saves notes using tools”， 也可以由后台自动抽取。示例里 prompt 函数通过 get_store() 取到当前图运行时配置的 store， 先用 store.search(("memories",), query=...) 找相关记忆，再把结果放进 system message 的 <memories> 区块。随后 agent 带着 create_manage_memory_tool、 store 和 checkpointer 被创建出来。 这条完整示例在 hot path 定义 和 agent 示例。

def prompt(state):
    store = get_store()
    memories = store.search(
        ("memories",),
        query=state["messages"][-1].content,
    )
    return [
        {"role": "system", "content": f"<memories>\n{memories}\n</memories>"},
        *state["messages"],
    ]

agent = create_react_agent(
    model,
    prompt=prompt,
    tools=[create_manage_memory_tool(namespace=("memories",))],
    store=store,
    checkpointer=checkpointer,
)

从这段代码可以看出两件事。读取并不神秘：prompt 函数先用最后一条 user message 做 store.search()， 把命中的记忆放回 system message。写入也不自动：只有当 agent 判断“以后默认用英文回答”值得长期保存时， 才会调用 manage_memory。这带来两个好处：用户可以在工具轨迹里看到为什么被记住； 模型也能结合当前任务上下文，区分长期偏好、一次性约束和普通补充说明。

2.2 search tool 让长期记忆重新进入当前模型视图

工具源码更直接。create_manage_memory_tool 的默认 instructions 会在发现新用户偏好、收到明确记忆请求、想记录重要上下文、或发现旧 memory 错误过时时主动调用工具。 真正执行时，amanage_memory 会解析 namespace；delete 走 store.adelete()，create/update 走 store.aput()， value 里保存 JSON 化后的 content。

async def amanage_memory(content=None, action="create", *, id=None):
    namespace = namespacer()
    if action == "delete":
        await store.adelete(namespace, key=str(id))
        return f"Deleted memory {id}"

    id = id or uuid.uuid4()
    await store.aput(
        namespace,
        key=str(id),
        value={"content": _ensure_json_serializable(content)},
    )
    return f"{action}d memory {id}"

这段代码把 LangMem 的写入责任边界讲得很清楚：tool 不替你重新发明一套 memory store， 它把模型决定出的 content 存到 LangGraph BaseStore 的某个 namespace 下。 id 决定是新增还是覆盖旧条目；delete 则必须指定已有 memory id。

读取路径对应 create_search_memory_tool。 它暴露 query、limit、offset 和 filter， 再调用 store.asearch(namespace, query=..., filter=..., limit=..., offset=...)。 所以 hot path 的本质是把“写记忆”和“搜记忆”变成模型可调用工具，让 agent 在当前执行路径里自己承担判断。

async def asearch_memory(query, *, limit=10, offset=0, filter=None):
    namespace = namespacer()
    memories = await store.asearch(
        namespace,
        query=query,
        filter=filter,
        limit=limit,
        offset=offset,
    )
    return utils.dumps([m.dict() for m in memories])

所以，如果你问“这里有没有 BM25、RRF 或别的混合检索”，答案不应该靠猜。 从这个工具路径看，LangMem 把检索委托给配置的 BaseStore 实现：tool 只传 namespace、query、filter 和分页参数。具体是 embedding 语义检索、 精确过滤，还是某个数据库适配器的能力，取决于 store 是否配置了 index 以及所用实现支持什么。

这条路线和 Mem0 的差别也在这里。Mem0 更像应用层先检索，再把结果交给模型；LangMem hot path 可以让模型在需要时自己搜索， 再决定是否继续回答或更新记忆。它把长期记忆变成 workflow 中的可调用能力，而不只是 prompt 组装前的固定步骤。

三、Store 是长期记忆，checkpointer 是线程状态

3.1 BaseStore 解决跨线程、跨会话的长期状态

Hot path quickstart 特别提醒：Store 和 MemorySaver / checkpointer 不是同一个东西。 store 可以按 user、agent、organization 等 namespace 保存和检索任意信息，更适合 long-term、cross-thread memory； checkpointer 维护每个 thread 内的 graph state 和 conversation history，用于 durable execution。 这段边界说明在 hot path quickstart。

LangGraph 源码也这样分。BaseStore 被描述为 persistent key-value store，可跨 threads 和 conversations 共享，按 user IDs、assistant IDs 或任意 namespace 分区， 部分实现支持 optional semantic search。它的 search() 支持 namespace prefix、query、filter、limit、offset；put() 则以 namespace + key 形成完整路径，并可控制索引字段。

class BaseStore(ABC):
    """Persistent key-value store shared across threads."""

    def search(
        self,
        namespace_prefix: tuple[str, ...],
        *,
        query: str | None = None,
        filter: dict | None = None,
        limit: int = 10,
        offset: int = 0,
    ) -> list[SearchItem]: ...

    def put(
        self,
        namespace: tuple[str, ...],
        key: str,
        value: dict,
        index: Literal[False] | list[str] | None = None,
    ) -> None: ...

这也是 namespace 为什么重要。("memories", "user_123") 可以保存某个用户的长期偏好； ("memories", "org_7", "user_123") 可以在组织和用户之间做层级隔离。 同一句“默认用英文回答”，如果写进共享 namespace，就可能影响所有用户；写进用户 namespace，才是个人偏好。

3.2 checkpointer 负责可恢复执行，不负责长期画像

StateGraph.compile() 里，checkpointer 参数被描述为 fully versioned short-term memory，可暂停、恢复和 replay； 同一个 compile 方法还有单独的 store 参数。 运行时 Runtime.store 的注释是 enabling persistence and memory；节点内部也能通过 get_store() 取到当前图或 entrypoint 配置的 store。把这两层拆开，才能避免把“当前线程可恢复”和“跨线程长期偏好”混在一起。

graph = builder.compile(
    checkpointer=checkpointer,  # 当前 thread 的短期、可恢复执行状态
    store=store,                # 跨 thread 的长期记忆
)

@dataclass
class Runtime:
    store: BaseStore | None = None

一个实际判断是：如果没有它，当前任务会不能恢复，还是用户长期偏好会丢失？前者属于 checkpointer，后者属于 store。 混用以后会出现两类问题：把用户画像塞进 thread checkpoint，导致另一个线程读不到；或者把每轮中间状态写入长期 store， 让长期记忆被执行噪声污染。

四、Background manager：把整理记忆移出当前回答

4.1 manager 看到的是对话快照和相似旧记忆

Background quickstart 从另一边切入：agent 正常回答，memory manager 在后台从 conversation history 中抽取和合并记忆。 示例创建 create_memory_store_manager(..., namespace=("memories",))， 在 @entrypoint(store=store) 的 chat 函数里先调用 LLM 得到 response， 再把 user message 和 response 组成 to_process，交给 memory_manager.ainvoke()。 文档也提醒，实际活跃对话里可以用 delayed processing 做 debounce，避免每条消息都处理。 这些内容在 background 定义、 基础示例 和 处理说明。

这样做的关键，不是把当前对话摘要塞进 store，而是让 manager 带着旧记忆一起判断。比如用户前面说“我喜欢 Python”，后来又说 “最近主要写 Rust”，后台 manager 可以同时看 conversation history 和相似旧 memory，决定是新增、更新、保留并列事实， 还是删除错误记忆。它承担的是“整理长期状态”的角色，而不是“记录所有新句子”的角色。

4.2 后台路径把写入变成可延迟、可合并的事务

源码里的 manager 做了三步。第一层 create_memory_manager 会分析 conversation messages 和 existing memories，生成或更新结构化 memory entries，并通过 enable_inserts、enable_updates、enable_deletes 控制允许的操作。 第二层 create_memory_store_manager 把这个过程接到 BaseStore：自动搜索相关 memories、抽取新信息、更新已有 memories，并维护变更历史。 docstring 里的 data flow 写成一条链：Client 给 Manager conversation history，Manager 先找 similar memories， 再让 LLM analyze & extract，最后把 changes 应用回 Store。

async def ainvoke(self, input, config=None):
    namespace = self.namespace(config)
    search_results_lists = await asyncio.gather(
        *[store.asearch(namespace, query=query) for query in queries]
    )
    store_map = self._sort_results(search_results_lists, self.query_limit)

    enriched = await self.memory_manager.ainvoke({
        "messages": input["messages"],
        "existing": store_based,
        "max_steps": input.get("max_steps"),
    })

    await asyncio.gather(
        *(store.aput(**put) for put in final_puts),
        *(store.adelete(ns, key) for (ns, key) in final_deletes),
    )

形状示例：后台整理后写回 BaseStore

before:
  namespace = ("memories", user_id)
  key = "pref-1"
  value = { "content": "用户默认希望用中文回答" }

conversation snapshot:
  "以后默认用英文回答我"

after manager:
  final_puts:
    namespace = ("memories", user_id)
    key = "pref-1"
    value = { "content": "用户希望以后默认用英文回答" }
  final_deletes: []

执行细节也能印证这条链。MemoryStoreManager.ainvoke() 会先算 namespace，再根据 query generator 或对话窗口并发调用 store.asearch() 找旧记忆； 后面会整理 final puts 和 deletes，并通过 store.aput() / store.adelete() 应用到 store。换句话说，background manager 会带着旧记忆一起做更新判断，而不是把对话摘要直接塞进存储。

代价是这条路径天然存在一致性窗口：短期 thread state 已经知道刚才发生了什么，长期 store 可能还没处理完。 所以适合后台的记忆，最好是“稍后更准确”比“立刻可见”更重要的内容，例如长对话总结、重复偏好合并、知识 profile 清洗。 如果用户明确要求“现在记住”，让 hot path 工具处理会更符合预期。

五、什么时候选哪条路

5.1 用延迟、可观测性和一致性做最后判断

Hot path 适合那些用户能感知、需要即时生效、最好可解释的记忆动作。用户说“以后都用英文回答我”，agent 当场调用 manage_memory，下一轮就能被 prompt 或 search tool 取到；如果写错了，也能在工具轨迹里定位。 代价是它会消耗一次工具调用，并且让模型在回答任务和记忆管理之间分配注意力。

Background manager 适合长对话后的整理、重复事实合并、结构化 profile 更新、低优先级偏好沉淀。 它把记忆维护移出主对话，可以减少干扰，也更容易批处理和延迟处理；代价是刚说完的信息未必立刻进入长期 store， 系统需要清楚地处理“短期 thread state 已有，但 long-term store 尚未整理完成”的窗口。

把它放回整个 Agent Memory 系列，LangMem / LangGraph 提供的是执行路径维度的答案。 Mem0 更像外置 memory layer，Letta 更像 agent state，Graphiti 更像 temporal graph； LangMem / LangGraph 则提醒我们：长期记忆还要决定由哪条 workflow path 改写。下一篇的 TencentDB-Agent-Memory 会把视角推进到当前任务本身：当工具日志已经挤满上下文，系统怎样卸载它们，又怎样保留可恢复的证据链。

参考资料

• LangMem README：核心功能
• LangMem README：创建带 memory tools 的 agent
• LangMem hot path quickstart：两种记忆路径
• LangMem hot path quickstart：store、tools、checkpointer
• LangMem create_manage_memory_tool
• LangMem manage_memory write path
• LangMem create_search_memory_tool
• LangGraph BaseStore
• LangGraph StateGraph.compile()
• LangMem background quickstart
• LangMem create_memory_store_manager data flow
• LangMem MemoryStoreManager.ainvoke()