06 生产化：动态用户 Memory 设计

学习目标

理解一句话：多用户场景不要给每个用户写一套代码，而是用同一个 Agent 和同一个 memory 路径，通过 namespace 动态隔离。

本地最小例子

代码见 ../06_production_user_memory_design.py。

这个例子演示动态新增用户：

ensure_user_memory(store, user_id)：用户第一次出现时创建默认 memory。
create_backend_for_user(store, user_id)：用 user_id 生成隔离 backend。
Alice、Bob、新用户共用同一个路径 /memories/AGENTS.md，但读到不同内容。

生产版写法

在 LangSmith / 托管运行时里，namespace 可以直接来自当前请求身份：

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

如果你自己用 FastAPI 包一层，通常在请求入口拿到 user_id，然后创建或选择这个用户对应的 graph/backend：

def create_user_graph(user_id: str):
    backend = CompositeBackend(
        default=StateBackend(),
        routes={
            "/memories/": StoreBackend(
                namespace=lambda _rt: ("users", user_id),
            )
        },
    )
    return create_deep_agent(
        model=model,
        backend=backend,
        memory=["/memories/AGENTS.md"],
        store=store,
    )

关键原则

不要为每个用户写不同的 memory 路径，路径固定，namespace 动态。
不要把用户 ID 交给模型决定，必须来自认证系统。
用户 memory 默认可写，组织 policy 默认只读。
不要把 API key、token、密码放进 memory。
并发写同一个 memory 文件会 last-write-wins，复杂场景用后台 consolidation 合并。

常见误区

最常见的坑是把用户隔离做在文件名上，比如 /memories/alice.md、/memories/bob.md。这能跑，但权限、检索、迁移都会越来越臭。namespace 才是 StoreBackend 的隔离边界，别把虚拟路径当数据库分区。

相关资源

查看示例代码：deepagent_src/memory_teach/06_production_user_memory_design.py

from __future__ import annotations

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


MEMORY_SOURCE = "/memories/AGENTS.md"
ROUTED_STORE_KEY = "/AGENTS.md"


def create_backend_for_user(store: InMemoryStore, user_id: str) -> CompositeBackend:
    return CompositeBackend(
        default=StateBackend(),
        routes={
            "/memories/": StoreBackend(
                store=store,
                namespace=lambda _rt: ("users", user_id),
            )
        },
    )


def ensure_user_memory(store: InMemoryStore, user_id: str) -> None:
    namespace = ("users", user_id)
    if store.get(namespace, ROUTED_STORE_KEY) is None:
        store.put(
            namespace,
            ROUTED_STORE_KEY,
            create_file_data("# User Memory\n\n## Preferences\n\n"),
        )


def read_memory(backend: CompositeBackend) -> str:
    response = backend.download_files([MEMORY_SOURCE])[0]
    assert response.error is None
    assert response.content is not None
    return response.content.decode("utf-8")


def main() -> None:
    store = InMemoryStore()

    for user_id in ["alice", "bob", "new-user-42"]:
        ensure_user_memory(store, user_id)

    store.put(
        ("users", "alice"),
        ROUTED_STORE_KEY,
        create_file_data("# User Memory\n\n## Preferences\n\n- Prefers Python\n"),
    )
    store.put(
        ("users", "bob"),
        ROUTED_STORE_KEY,
        create_file_data("# User Memory\n\n## Preferences\n\n- Prefers TypeScript\n"),
    )

    alice_backend = create_backend_for_user(store, "alice")
    bob_backend = create_backend_for_user(store, "bob")
    new_user_backend = create_backend_for_user(store, "new-user-42")

    assert "Prefers Python" in read_memory(alice_backend)
    assert "Prefers TypeScript" in read_memory(bob_backend)
    assert read_memory(new_user_backend).startswith("# User Memory")
    print("production user memory design ok")


if __name__ == "__main__":
    main()