让对话变为私密（第 2 部分/共 3 部分）¶
在本教程中，我们将扩展我们的聊天机器人，为每个用户提供他们自己的私密对话。我们将添加资源级访问控制，以便用户只能看到他们自己的对话线程。
占位符令牌
正如我们在第 1 部分中所做的那样，在本节中，我们将使用硬编码的令牌进行说明。 在掌握基础知识后，我们将在第 3 部分中实现一个“适用于生产环境”的认证方案。
理解资源授权¶
在上一个教程中，我们控制了哪些人可以访问我们的机器人。但目前，任何经过身份验证的用户都可以查看其他所有人的对话！让我们通过添加资源授权来解决这个问题。
首先，请确保你已经完成了基本身份验证教程，并且你的安全机器人可以无错误运行：
添加资源授权¶
回顾一下，在上一个教程中，Auth 对象允许我们注册一个认证函数，LangGraph 平台使用该函数来验证传入请求中的承载令牌。现在，我们将使用它来注册一个**授权**处理程序。
授权处理程序是在认证成功**之后**运行的函数。这些处理程序可以为资源添加元数据（例如资源的所有者是谁），并过滤每个用户可以查看的内容。
让我们更新 src/security/auth.py 文件，并添加一个在每个请求上都会运行的授权处理程序：
src/security/auth.py
from langgraph_sdk import Auth
# 保留上一个教程中的测试用户
VALID_TOKENS = {
"user1-token": {"id": "user1", "name": "Alice"},
"user2-token": {"id": "user2", "name": "Bob"},
}
auth = Auth()
@auth.authenticate
async def get_current_user(authorization: str | None) -> Auth.types.MinimalUserDict:
"""我们上一个教程中的认证处理程序。"""
assert authorization
scheme, token = authorization.split()
assert scheme.lower() == "bearer"
if token not in VALID_TOKENS:
raise Auth.exceptions.HTTPException(status_code=401, detail="无效的令牌")
user_data = VALID_TOKENS[token]
return {
"identity": user_data["id"],
}
@auth.on
async def add_owner(
ctx: Auth.types.AuthContext, # 包含当前用户的信息
value: dict, # 正在创建/访问的资源
):
"""使资源对其创建者私有。"""
# 示例：
# ctx: AuthContext(
# permissions=[],
# user=ProxyUser(
# identity='user1',
# is_authenticated=True,
# display_name='user1'
# ),
# resource='threads',
# action='create_run'
# )
# value:
# {
# 'thread_id': UUID('1e1b2733-303f-4dcd-9620-02d370287d72'),
# 'assistant_id': UUID('fe096781-5601-53d2-b2f6-0d3403f7e9ca'),
# 'run_id': UUID('1efbe268-1627-66d4-aa8d-b956b0f02a41'),
# 'status': 'pending',
# 'metadata': {},
# 'prevent_insert_if_inflight': True,
# 'multitask_strategy': 'reject',
# 'if_not_exists': 'reject',
# 'after_seconds': 0,
# 'kwargs': {
# 'input': {'messages': [{'role': 'user', 'content': 'Hello!'}]},
# 'command': None,
# 'config': {
# 'configurable': {
# 'langgraph_auth_user': ... 你的用户对象...
# 'langgraph_auth_user_id': 'user1'
# }
# },
# 'stream_mode': ['values'],
# 'interrupt_before': None,
# 'interrupt_after': None,
# 'webhook': None,
# 'feedback_keys': None,
# 'temporary': False,
# 'subgraphs': False
# }
# }
# 做两件事：
# 1. 将用户的 ID 添加到资源的元数据中。每个 LangGraph 资源都有一个 `metadata` 字典，该字典会与资源一起持久化。
# 此元数据在读取和更新操作的过滤中很有用
# 2. 返回一个过滤器，使用户只能看到自己的资源
filters = {"owner": ctx.user.identity}
metadata = value.setdefault("metadata", {})
metadata.update(filters)
# 只允许用户查看自己的资源
return filters
该处理程序接收两个参数：
ctx(AuthContext)：包含当前用户的信息、用户的权限、资源（“threads”、“crons”、“assistants”）以及正在执行的操作（“create”、“read”、“update”、“delete”、“search”、“create_run”）value(dict)：正在创建或访问的数据。此字典的内容取决于正在访问的资源和操作。有关如何获得更精细范围的访问控制的信息，请参阅下面的添加范围授权处理程序。
请注意，我们这个简单的处理程序做了两件事：
- 将用户的 ID 添加到资源的元数据中。
- 返回一个元数据过滤器，使用户只能看到自己拥有的资源。
测试私密对话¶
让我们来测试一下我们的授权功能。如果我们正确地设置了相关内容，应该会看到所有的 ✅ 消息。确保你的开发服务器正在运行（运行 langgraph dev）：
from langgraph_sdk import get_client
# 为两个用户分别创建客户端
alice = get_client(
url="http://localhost:2024",
headers={"Authorization": "Bearer user1-token"}
)
bob = get_client(
url="http://localhost:2024",
headers={"Authorization": "Bearer user2-token"}
)
# 爱丽丝创建一个助手
alice_assistant = await alice.assistants.create()
print(f"✅ 爱丽丝创建了助手: {alice_assistant['assistant_id']}")
# 爱丽丝创建一个线程并进行聊天
alice_thread = await alice.threads.create()
print(f"✅ 爱丽丝创建了线程: {alice_thread['thread_id']}")
await alice.runs.create(
thread_id=alice_thread["thread_id"],
assistant_id="agent",
input={"messages": [{"role": "user", "content": "嗨，这是爱丽丝的私密聊天"}]}
)
# 鲍勃尝试访问爱丽丝的线程
try:
await bob.threads.get(alice_thread["thread_id"])
print("❌ 鲍勃不应该看到爱丽丝的线程！")
except Exception as e:
print("✅ 鲍勃被正确拒绝访问:", e)
# 鲍勃创建他自己的线程
bob_thread = await bob.threads.create()
await bob.runs.create(
thread_id=bob_thread["thread_id"],
assistant_id="agent",
input={"messages": [{"role": "user", "content": "嗨，这是鲍勃的私密聊天"}]}
)
print(f"✅ 鲍勃创建了他自己的线程: {bob_thread['thread_id']}")
# 列出线程 - 每个用户只能看到自己的线程
alice_threads = await alice.threads.search()
bob_threads = await bob.threads.search()
print(f"✅ 爱丽丝看到 {len(alice_threads)} 个线程")
print(f"✅ 鲍勃看到 {len(bob_threads)} 个线程")
运行测试代码，你应该会看到如下输出：
✅ 爱丽丝创建了助手: fc50fb08-78da-45a9-93cc-1d3928a3fc37
✅ 爱丽丝创建了线程: 533179b7-05bc-4d48-b47a-a83cbdb5781d
✅ 鲍勃被正确拒绝访问: Client error '404 Not Found' for url 'http://localhost:2024/threads/533179b7-05bc-4d48-b47a-a83cbdb5781d'
For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404
✅ 鲍勃创建了他自己的线程: 437c36ed-dd45-4a1e-b484-28ba6eca8819
✅ 爱丽丝看到 1 个线程
✅ 鲍勃看到 1 个线程
这意味着：
- 每个用户可以创建并在自己的线程中聊天
- 用户无法看到彼此的线程
- 列出线程时只会显示自己的线程
添加作用域授权处理程序¶
宽泛的 @auth.on 处理程序会匹配所有授权事件。这很简洁，但这意味着 value 字典的内容作用域不明确，并且我们对每个资源应用相同的用户级访问控制。如果我们想更精细地控制，也可以控制对资源的特定操作。
更新 src/security/auth.py 以添加针对特定资源类型的处理程序：
# 保留我们之前的处理程序...
from langgraph_sdk import Auth
@auth.on.threads.create
async def on_thread_create(
ctx: Auth.types.AuthContext,
value: Auth.types.on.threads.create.value,
):
"""创建线程时添加所有者。
此处理程序在创建新线程时运行，执行两件事：
1. 在正在创建的线程上设置元数据以跟踪所有权
2. 返回一个过滤器，确保只有创建者可以访问它
"""
# 示例值：
# {'thread_id': UUID('99b045bc-b90b-41a8-b882-dabc541cf740'), 'metadata': {}, 'if_exists': 'raise'}
# 向正在创建的线程添加所有者元数据
# 此元数据与线程一起存储并持久化
metadata = value.setdefault("metadata", {})
metadata["owner"] = ctx.user.identity
# 返回过滤器以限制只有创建者可以访问
return {"owner": ctx.user.identity}
@auth.on.threads.read
async def on_thread_read(
ctx: Auth.types.AuthContext,
value: Auth.types.on.threads.read.value,
):
"""只允许用户读取自己的线程。
此处理程序在读取操作时运行。由于线程已经存在，我们不需要设置元数据 - 我们只需要返回一个过滤器，以确保用户只能看到自己的线程。
"""
return {"owner": ctx.user.identity}
@auth.on.assistants
async def on_assistants(
ctx: Auth.types.AuthContext,
value: Auth.types.on.assistants.value,
):
# 为了说明目的，我们将拒绝所有涉及助手资源的请求
# 示例值：
# {
# 'assistant_id': UUID('63ba56c3-b074-4212-96e2-cc333bbc4eb4'),
# 'graph_id': 'agent',
# 'config': {},
# 'metadata': {},
# 'name': 'Untitled'
# }
raise Auth.exceptions.HTTPException(
status_code=403,
detail="User lacks the required permissions.",
)
# 假设你在存储中按照 (user_id, resource_type, resource_id) 组织信息
@auth.on.store()
async def authorize_store(ctx: Auth.types.AuthContext, value: dict):
# 每个存储项的 "namespace" 字段是一个元组，你可以将其视为项的目录。
namespace: tuple = value["namespace"]
assert namespace[0] == ctx.user.identity, "Not authorized"
注意，现在我们不是只有一个全局处理程序，而是有针对以下操作的特定处理程序：
- 创建线程
- 读取线程
- 访问助手
前三个处理程序匹配每个资源上的特定**操作**（请参阅资源操作），而最后一个（@auth.on.assistants）匹配 assistants 资源上的_任何_操作。对于每个请求，LangGraph 将运行与正在访问的资源和操作最匹配的特定处理程序。这意味着上述四个处理程序将运行，而不是宽泛作用域的 “@auth.on” 处理程序。
尝试将以下测试代码添加到你的测试文件中：
# ... 和之前一样
# 尝试创建一个助手。这应该失败
try:
await alice.assistants.create("agent")
print("❌ Alice 不应该能够创建助手！")
except Exception as e:
print("✅ Alice 被正确拒绝访问:", e)
# 尝试搜索助手。这也应该失败
try:
await alice.assistants.search()
print("❌ Alice 不应该能够搜索助手！")
except Exception as e:
print("✅ Alice 被正确拒绝访问搜索助手:", e)
# Alice 仍然可以创建线程
alice_thread = await alice.threads.create()
print(f"✅ Alice 创建了线程: {alice_thread['thread_id']}")
然后再次运行测试代码：
✅ Alice 创建了线程: dcea5cd8-eb70-4a01-a4b6-643b14e8f754
✅ Bob 被正确拒绝访问: Client error '404 Not Found' for url 'http://localhost:2024/threads/dcea5cd8-eb70-4a01-a4b6-643b14e8f754'
更多信息请查看: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404
✅ Bob 创建了自己的线程: 400f8d41-e946-429f-8f93-4fe395bc3eed
✅ Alice 看到 1 个线程
✅ Bob 看到 1 个线程
✅ Alice 被正确拒绝访问:
更多信息请查看: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500
✅ Alice 被正确拒绝访问搜索助手:
恭喜！你已经构建了一个聊天机器人，每个用户都有自己的私人对话。虽然此系统使用简单的基于令牌的身份验证，但我们学到的授权模式适用于实现任何实际的身份验证系统。在下一个教程中，我们将使用 OAuth2 用真实用户账户替换我们的测试用户。
下一步是什么？¶
既然你已经能够控制对资源的访问，你可能想：