主题
按用户隔离存储数据
使用身份验证处理程序,在 LangSmith Deployment 中将长期存储数据的作用域限定为每个经过身份验证的用户。
每个 LangSmith Deployment 都包含一个由 Postgres 支持的存储,用于长期记忆和跨线程数据。默认情况下,存储命名空间在所有调用者之间共享。要为每个用户提供自己隔离的存储,请配置自定义身份验证并添加一个存储授权处理程序,将命名空间重写为包含已身份验证用户的标识。
本指南展示如何在 LangSmith Deployment 中配置这种隔离。同一模式也适用于启用了自定义身份验证的自托管部署。
工作原理
存储项由命名空间元组(例如 ("memories", "preferences"))和键组织。与线程和助手不同,存储命名空间由你的应用程序定义,因此授权的工作方式不同:
- 你的
@auth.authenticate处理程序验证调用者,并为每个用户返回唯一的identity。 - 你的
@auth.on.store处理程序要么验证命名空间以该身份开头,要么重写命名空间以在其前面添加该身份。 - 服务器使用经过验证或重写的命名空间进行实际读取或写入。
使用自动前缀重写写入逻辑命名空间 ("memories",) 的用户,实际将数据存储在 ("user-123", "memories")。另一个用户无法读取或覆盖该数据,因为他们的请求作用域为 ("user-456", ...)。
NOTE
存储授权处理程序接收一个可变的 value 字典。对 value["namespace"] 的更改会对操作生效。你不需要像对线程那样返回元数据筛选器。
前置条件
- 一个已配置自定义身份验证的 LangSmith Deployment。
- 一个为每个最终用户返回稳定且唯一
identity的@auth.authenticate处理程序。
在 LangSmith Deployment 中配置身份验证
在 langgraph.json 中将部署指向你的身份验证模块:
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 添加到逻辑命名空间前面 | 你希望 agent 代码更简单,并在 API 层实现透明隔离 |
两种模式都使用单个 @auth.on.store 处理程序,覆盖所有存储操作(put、get、search、delete 和 list_namespaces)。只有当你希望为每个操作使用不同规则时,才需要 @auth.on.store.put 等特定操作的处理程序。
TIP
如果你使用拒绝未处理请求的全局 @auth.on 处理程序,请显式注册 @auth.on.store,以便允许存储操作。有关特定操作的处理程序名称,请参阅 Auth 参考。
显式命名空间 + 拒绝
你的应用程序代码将用户身份作为每个命名空间的第一个段(例如 ("user-123", "memories"))。授权处理程序验证调用者与该前缀匹配,并拒绝不匹配的请求:
python
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-123,则对命名空间 ("user-456", "memories") 的请求会立即失败。代价是 agent 中的每次存储调用都必须包含用户前缀,通常来自运行时上下文,例如 config["configurable"]["langgraph_auth_user_id"]。
自动前缀重写
你的应用程序代码使用不带用户前缀的逻辑命名空间(例如 ("memories", "preferences"))。授权处理程序会在每次存储操作前添加已身份验证用户的身份:
python
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,),因此用户只能看到自己的命名空间。
此模式使 agent 代码更简单,因为身份验证层透明地处理用户隔离。代价是你不能在应用程序代码中同时为命名空间添加前缀,否则会造成数据双重作用域并破坏读取。请只选择一个作用域层:身份验证重写或应用程序级命名空间,不要同时使用两者。
在 agent 代码中使用命名空间
如果你使用自动前缀重写,图代码应使用不带用户前缀的逻辑命名空间。身份验证层会在请求时自动添加用户作用域。将以下内容放入图节点或工具中(例如 graph.py):
python
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 的请求永远不会到达该命名空间。
如果你使用显式命名空间 + 拒绝,请在图代码中的每个命名空间里包含用户身份:
python
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"},
)Deep Agents 和 StoreBackend
如果你使用 Deep Agents 和 StoreBackend,请将命名空间工厂与身份验证作用域布局对齐。当身份验证添加 ctx.user.identity 前缀时,在后端使用逻辑命名空间,并让身份验证处理用户隔离:
python
from deepagents.backends import StoreBackend
StoreBackend(
namespace=lambda rt: ("memories",), # auth adds user identity
)或者,如果你在图内按用户限定作用域(例如使用 rt.server_info.user.identity),请确保身份验证处理程序不会重复为命名空间添加前缀。请只选择一个作用域层:身份验证重写或应用程序级命名空间,不要同时使用两者。
有关命名空间设计的更多信息,请参阅 Deep Agents 后端和用户范围记忆。
测试存储隔离
在 langgraph dev 运行且身份验证处理程序中配置了两个测试用户的情况下,验证存储数据不会在用户之间泄漏:
python
import asyncio
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) | 每线程检查点和消息 |
| 长期记忆 | 存储命名空间重写 | 跨线程记忆、偏好和文档 |
对于多用户 agent,请同时配置两者。有关线程和运行作用域,请参阅将对话设为私有。