每个 LangSmith Deployment 包含一个基于 Postgres 的 存储 用于长期记忆和跨线程数据。默认情况下,存储命名空间在所有调用者之间共享。要为每个用户提供独立的隔离存储,请配置 自定义认证 并添加一个存储 授权处理程序 来重写命名空间以包含已认证用户的身份。
本指南展示了如何在 LangSmith Deployment 中配置该隔离。同样的模式适用于 self-hosted 启用了自定义认证的部署。
工作原理
存储项通过 **命名空间** 元组(例如, ("memories", "preferences"))和一个 **键**组织。与线程和助手不同,存储命名空间由你的应用程序定义,因此授权的工作方式不同:
- 你的
@auth.authenticate处理程序验证调用者并返回一个唯一的identity给每个用户。 - 你的
@auth.on.store处理程序要么 **验证** 命名空间以该身份开头,要么 **重写** 命名空间并在其前面添加该身份。 - 服务器使用经过验证或重写的命名空间执行实际读取或写入操作。
写入逻辑命名空间的用户 ("memories",) 通过自动前缀重写实际将数据存储在 ("user-123", "memories")。其他用户无法读取或覆盖该数据,因为他们的请求被限定在 ("user-456", ...).
前提条件
- - 一个带有 自定义认证 configured.
- - An
@auth.authenticate处理程序,且该处理程序为每个最终用户返回稳定且唯一的identity。
在 LangSmith Deployment 中配置认证
在你的部署中指向你的认证模块 langgraph.json:
{
"dependencies": ["."],
"graphs": {
"agent": "./src/agent/graph.py:graph"
},
"auth": {
"path": "./src/security/auth.py:auth"
}
}
该 auth 对象必须公开一个 langgraph_sdk.Auth (通常命名为 auth).
选择一种存储授权模式
存储隔离需要一个 @auth.on.store 处理程序。两种常见模式效果良好;根据你希望用户作用域所在的位置来选择。
| 模式 | 用户范围设置位置 | 最佳使用场景 |
|---|---|---|
| **显式命名空间 + 拒绝** | 应用程序代码将 user_id 放在每个命名空间的首位 | 您已经在将用户身份传递给图形代码,或者您希望命名空间反映完整存储路径 |
| **自动前缀重写** | 认证处理器预先添加 ctx.user.identity 到逻辑命名空间 | 您希望代码更简单且在 API 层实现透明隔离 |
两种模式都使用单个 @auth.on.store 处理器覆盖所有存储操作(put, get, search, delete和 list_namespaces)。仅当您需要对不同操作使用不同规则时才需要特定于操作的处理器,例如 @auth.on.store.put (如果您希望每个操作使用不同的规则)。
显式命名空间 + 拒绝
应用程序代码将用户身份作为每个命名空间的第一段(例如, ("user-123", "memories"))。认证处理器验证调用者是否与此前缀匹配,并拒绝不匹配的情况:
from langgraph_sdk import Auth
auth = Auth()
# ... your @auth.authenticate handler ...
@auth.on.store
async def authorize_store(
ctx: Auth.types.AuthContext,
value: Auth.types.on.store.value,
):
"""Require the user identity as the first namespace segment."""
namespace = tuple(value["namespace"])
if not namespace or namespace[0] != ctx.user.identity:
raise Auth.exceptions.HTTPException(
status_code=403,
detail="Not authorized to access this namespace.",
)
此模式使存储布局在图形代码中明确。对命名空间的请求 ("user-456", "memories") 如果经过身份验证的用户是 user-123,则会立即失败。权衡在于代理中的每个存储调用必须包含用户前缀,通常来自运行时上下文,例如 config["configurable"]["langgraph_auth_user_id"].
自动前缀重写
应用程序代码使用不包含用户前缀的逻辑命名空间(例如, ("memories", "preferences"))。认证处理器在每次存储操作时预先添加经过身份验证的用户身份:
from langgraph_sdk import Auth
auth = Auth()
# ... your @auth.authenticate handler ...
@auth.on.store
async def scope_store(
ctx: Auth.types.AuthContext,
value: Auth.types.on.store.value,
):
"""Isolate store data per user by rewriting namespaces."""
namespace = tuple(value["namespace"]) if value.get("namespace") else ()
if not namespace or namespace[0] != ctx.user.identity:
namespace = (ctx.user.identity, *namespace)
value["namespace"] = namespace
当客户端调用 list_namespaces 而没有前缀时, value["namespace"] 为空。上面的处理器将空命名空间视为 (ctx.user.identity,),因此用户只能看到自己的命名空间。
此模式使代理代码更简单,因为认证层透明地处理用户隔离。权衡在于您不能在应用程序代码中也为命名空间添加前缀,否则会导致数据双重作用域并破坏读取。选择一个作用域层:认证重写 **or** 应用程序级命名空间,而不是两者兼用。
在代理代码中使用命名空间
如果您使用自动前缀重写,图形代码使用不包含用户前缀的逻辑命名空间。认证层在请求时自动添加用户范围。将以下内容放在图形节点或工具中(例如, graph.py):
from langgraph.store.base import BaseStore
async def save_preference(state: State, *, store: BaseStore):
await store.aput(
("memories", "preferences"), # logical namespace
"settings",
{"theme": "dark"},
)
async def load_preference(state: State, *, store: BaseStore):
item = await store.aget(
("memories", "preferences"),
"settings",
)
return item.value if item else {}
当用户 A 调用存储 API 时,服务器将数据持久化到 ("user-a", "memories", "preferences")。用户 B 的请求永远不会到达该命名空间。
如果您使用 **显式命名空间 + 拒绝**,请改为从图形代码中在每个命名空间包含用户身份:
from langchain_core.runnables import RunnableConfig
from langgraph.store.base import BaseStore
async def save_preference(state: State, *, store: BaseStore, config: RunnableConfig):
user_id = config["configurable"]["langgraph_auth_user_id"]
await store.aput(
(user_id, "memories", "preferences"),
"settings",
{"theme": "dark"},
)
深度代理和 StoreBackend
如果您使用 深度代理 与 StoreBackend,请将您的命名空间工厂与认证作用域布局对齐。当认证预先添加 ctx.user.identity,在后端使用逻辑命名空间,让 auth 处理用户隔离:
from deepagents.backends import StoreBackend
StoreBackend(
namespace=lambda rt: ("memories",), # auth adds user identity
)
或者,如果你在图内部按用户进行作用域划分(例如,使用 rt.server_info.user.identity),请确保你的 auth 处理程序不会对命名空间进行双重前缀。选择一个作用域划分层:auth 重写 **or** 应用级命名空间,而不是两者。
有关命名空间设计的更多信息,请参阅 深度智能体后端 和 用户作用域内存.
测试存储隔离
使用 langgraph dev 运行且在 auth 处理程序中配置了两个测试用户,验证存储数据不会跨用户泄露:
from langgraph_sdk import get_client
async def main():
alice = get_client(
url="http://localhost:60058",
headers={"Authorization": "Bearer user1-token"},
)
bob = get_client(
url="http://localhost:60058",
headers={"Authorization": "Bearer user2-token"},
)
# Alice writes a store item
await alice.store.put_item(
["memories"],
key="note",
value={"text": "Alice private note"},
)
# Bob cannot read Alice's item
bob_item = await bob.store.get_item(["memories"], key="note")
if bob_item is None or bob_item["value"]["text"] != "Alice private note":
print("✅ Bob correctly denied access to Alice's item")
else:
print("❌ Bob should not see Alice's store item")
# Bob writes his own item at the same logical namespace
await bob.store.put_item(
["memories"],
key="note",
value={"text": "Bob private note"},
)
alice_item = await alice.store.get_item(["memories"], key="note")
bob_item = await bob.store.get_item(["memories"], key="note")
assert alice_item["value"]["text"] == "Alice private note"
assert bob_item["value"]["text"] == "Bob private note"
print("✅ Each user sees only their own store data")
if __name__ == "__main__":
asyncio.run(main())
与线程隔离结合使用
存储隔离和 线程隔离 解决不同的问题:
| 关注点 | 机制 | 作用域 |
|---|---|---|
| 对话历史 | 线程元数据过滤器(owner) | 每线程检查点和消息 |
| 长期记忆 | 存储命名空间重写 | 跨线程记忆、偏好设置和文档 |
对于多用户智能体,请同时配置两者。参阅 使对话私有化 了解线程和运行作用域划分。
另请参阅
- - 身份验证和访问控制
- - 设置自定义身份验证
- - 使对话私有化
- - 部署中的语义搜索
- - LangGraph 存储