本页目录
Backends / 后端Note 27

04 StoreBackend

学习目标

学习目标

理解一句话:StoreBackend 把文件存进 LangGraph Store,使不同 thread 可以共享文件。

示例

代码见 ../04_store_backend.py

核心配置:

agent = create_deep_agent(
    model=gpt_model,
    backend=StoreBackend(namespace=lambda _rt: ("demo-user",)),
    store=InMemoryStore(),
    checkpointer=InMemorySaver(),
)

示例执行两步:

thread-a 写入 /memory.txt
thread-b 读取 /memory.txt

两个 thread 使用相同 namespace ("demo-user",),所以能访问同一份文件。

运行

在项目根目录执行:

uv run python deepagent_src/backend_teach/04_store_backend.py

运行调用模型两次,产生 API 费用。

三个容易混淆的概念

thread_id

标识一次对话线程。它决定 checkpointer 保存和恢复哪一份 Agent state。

{"configurable": {"thread_id": "thread-a"}}

namespace

标识 Store 中的数据空间。它决定哪些运行共享 Backend 文件。

StoreBackend(namespace=lambda _rt: ("demo-user",))

Store

真正保存数据的容器。

store=InMemoryStore()

关系如下:

thread_id  -> 隔离对话 state
namespace  -> 隔离 StoreBackend 文件
Store      -> 实际保存文件数据

为什么不同 thread 能读取

thread-athread-b 的对话 state 是分开的,但它们的 namespace 相同:

thread-a -> namespace demo-user -> /memory.txt
thread-b -> namespace demo-user -> /memory.txt

如果 namespace 不同,两个 thread 就会看到不同的文件空间。

多用户隔离

生产环境应按登录用户生成 namespace:

StoreBackend(
    namespace=lambda rt: (rt.server_info.user.identity,),
)

结果:

user-a -> namespace ("user-a",)
user-b -> namespace ("user-b",)

千万不要在多用户应用里给所有人使用同一个固定 namespace,否则用户数据会互相可见。

InMemoryStore 是否真正持久化

本课使用 InMemoryStore 是为了让示例最简单:

  • 可以跨 thread 共享。
  • 只能在当前 Python 进程中保存。
  • 程序退出后数据消失。

生产环境需要换成 Redis、Postgres、云 Store,或使用 LangSmith Deployment 自动提供的 Store。

StoreBackend 与 StateBackend

对比项 StateBackend StoreBackend
隔离依据 thread_id namespace
默认跨 thread 可以
典型数据 草稿、中间结果 用户偏好、长期指令
依赖组件 checkpointer BaseStore

本课结论

thread_id 管对话,namespace 管共享范围,Store 管实际数据。长期记忆能不能安全使用,关键看 namespace 是否正确隔离。

常见误解:StoreBackend 能嵌套 CompositeBackend 吗?

不能。StoreBackend 是一个存储 Backend,不是路由器;它不知道“某个路径应该转给本地磁盘”。

CompositeBackend 才是外层路由器:

Agent 文件工具
    -> CompositeBackend
        -> /memories/**  转给 StoreBackend
        -> /workspace/** 转给 FilesystemBackend
        -> 其他路径       转给 StateBackend

正确写法:

from pathlib import Path

from deepagents.backends import CompositeBackend, FilesystemBackend, StateBackend, StoreBackend
from langgraph.store.memory import InMemoryStore

backend = CompositeBackend(
    default=StateBackend(),
    routes={
        "/memories/": StoreBackend(namespace=lambda _rt: ("demo-user",)),
        "/workspace/": FilesystemBackend(
            root_dir=Path("workspace").resolve(),
            virtual_mode=True,
        ),
    },
)

agent = create_deep_agent(
    model=gpt_model,
    backend=backend,
    store=InMemoryStore(),
)

结果:

/draft.md             -> StateBackend:本次对话临时草稿
/memories/profile.md  -> StoreBackend:同一用户跨 thread 共享
/workspace/report.md  -> FilesystemBackend:本地磁盘持久化

技术上也可以把 StoreBackend 放在 default

CompositeBackend(
    default=StoreBackend(namespace=lambda _rt: ("demo-user",)),
    routes={"/workspace/": FilesystemBackend(root_dir="...", virtual_mode=True)},
)

但通常不推荐。Deep Agents 的内部结果和对话历史也会进入 default,导致它们被长期保存;多数项目选择 StateBackend 作为 default,只把明确的 /memories/ 交给 Store。

StateBackend 与 StoreBackend:到底差在哪里?

最短答案:StateBackend 的文件属于“这条对话”;StoreBackend 的文件属于“一个 namespace”。

问题 StateBackend StoreBackend
文件保存在哪里 LangGraph 的 Agent state LangGraph BaseStore
用什么键隔离 thread_id namespace
不同 thread 能否读取同一文件 默认不能 namespace 相同就能
需要什么组件 checkpointer store=
重启后是否存在 取决于 checkpointer 取决于 Store 实现
典型用途 临时计划、草稿、工具结果 用户偏好、长期记忆、共享规则

看同一个文件路径最容易懂:

StateBackend
thread-a 的 /note.txt  !=  thread-b 的 /note.txt

StoreBackend(同一 namespace)
thread-a 的 /note.txt  ==  thread-b 的 /note.txt

StoreBackend(不同 namespace)
user-a 的 /note.txt    !=  user-b 的 /note.txt

thread_id 是“一次对话的身份证”;namespace 是“一组共享记忆的房间号”。路径 /note.txt 只是房间里的文件名,不能单独决定文件是否共享。

还有一个容易踩坑的点:

store=InMemoryStore()

它只证明“跨 thread 共享”,不证明“程序重启后仍存在”。进程退出后内存清空;要真正长期保存,需要替换成持久化的 BaseStore 实现或使用平台提供的 Store。

相关资源

  • 查看示例代码:deepagent_src/backend_teach/04_store_backend.py
    import os
    
    os.environ["LANGSMITH_TRACING"] = "false"
    
    from deepagents import create_deep_agent
    from deepagents.backends import StoreBackend
    from langgraph.checkpoint.memory import InMemorySaver
    from langgraph.store.memory import InMemoryStore
    
    from deepagent_src.llms import gpt_model
    
    
    def main():
        agent = create_deep_agent(
            model=gpt_model,
            backend=StoreBackend(namespace=lambda _rt: ("demo-user",)),
            store=InMemoryStore(),
            checkpointer=InMemorySaver(),
        )
    
        agent.invoke(
            {"messages": [{"role": "user", "content": "把‘用户喜欢精简代码’写入 /memory.txt"}]},
            {"configurable": {"thread_id": "thread-a"}},
        )
        result = agent.invoke(
            {"messages": [{"role": "user", "content": "读取 /memory.txt,告诉我用户偏好"}]},
            {"configurable": {"thread_id": "thread-b"}},
        )
        print(result["messages"][-1].content)
    
    
    if __name__ == "__main__":
        main()