Langgraph logo Langgraph
登录
13. 认证与安全防护

13.认证与安全防护

在构建生产级 AI 应用时，认证与安全防护是不可或缺的环节。LangGraph 提供了一套灵活且强大的安全体系，既能满足云平台的快速部署需求，也支持自托管场景下的深度定制。本章将深入探讨如何在 LangGraph 应用中实现完整的认证授权机制，从中间件实现到资源级访问控制，再到与外部认证提供方的集成。

认证与授权的核心概念

在深入技术细节前，需要明确两个常被混淆但本质不同的概念：认证（Authentication）和授权（Authorization）。认证解决"你是谁"的问题，而授权解决"你能做什么"的问题。LangGraph 平台将这两层安全机制清晰地分离，分别通过不同的处理器来实现。

认证与授权的区别

认证（AuthN）是验证用户身份的过程。每次请求到达 LangGraph 后端时，认证中间件会首先介入，验证请求中携带的凭证是否有效。这个过程类似于机场安检时查验护照——系统确认请求者的身份是否真实可信。

授权（AuthZ）则发生在认证之后，决定已认证用户能够访问哪些资源、执行哪些操作。这相当于登机牌上的舱位等级决定了你能进入哪些区域。在 LangGraph 中，授权处理器会检查用户身份，并根据预设规则允许或拒绝特定操作。

默认安全模型

LangGraph 平台根据部署方式提供不同的默认安全策略：

对于 LangGraph Cloud SaaS 部署，默认使用 LangSmith API 密钥进行认证。每个请求必须在 x-api-key 请求头中携带有效的 API 密钥。这种方式开箱即用，适合快速验证和开发测试。

自托管部署则默认不启用任何认证机制，完全由开发者自行实现安全模型。这种设计给予了企业级用户最大的灵活性，可以根据内部安全规范定制认证流程。

值得注意的是，无论是哪种部署方式，LangGraph 都支持自定义认证处理器。这意味着即使在使用云平台时，也可以替换掉默认的 API 密钥机制，接入企业现有的身份认证系统。

系统架构概览

典型的 LangGraph 认证架构涉及三个核心组件：

首先是认证提供方（Identity Provider），如 Auth0、Supabase Auth 或企业自建的认证服务。它负责管理用户身份凭证，处理注册、登录、密码重置等流程，并在用户成功登录后颁发令牌（JWT 或会话令牌）。

其次是 LangGraph 后端（资源服务器），包含业务逻辑和受保护的资源。它接收客户端请求，验证令牌的有效性，并基于用户身份执行访问控制。重要的是，LangGraph 后端不直接存储用户密码，而是委托给专业的认证服务。

最后是客户端应用，负责收集用户凭证并发送给认证提供方，获取令牌后将其包含在对 LangGraph 后端的每个请求中。

整个交互流程形成一个清晰的安全闭环：用户登录认证提供方获取令牌，客户端携带令牌访问 LangGraph 后端，后端验证令牌并执行授权检查，最终返回受保护的资源。

认证中间件实现

LangGraph 的认证机制通过 Auth 对象实现，它作为容器注册认证和授权处理器。认证处理器在每个请求上运行，负责验证凭证并提取用户身份。

创建认证处理器

认证处理器是一个异步函数，接收请求信息并返回用户数据。如果凭证无效，应抛出 HTTPException 或 AssertionError。

from langgraph_sdk import Auth

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    # 从请求头中提取 API 密钥
    api_key = headers.get("x-api-key")
    if not api_key or not is_valid_key(api_key):
        raise Auth.exceptions.HTTPException(
            status_code=401,
            detail="Invalid API key"
        )
    
    # 返回用户信息，identity 是必需字段
    return {
        "identity": "user-123",        # 必需：唯一用户标识
        "is_authenticated": True,      # 可选：默认为 True
        "permissions": ["read", "write"] # 可选：用于基于权限的授权
    }

这个处理器展示了认证的核心逻辑：提取凭证、验证有效性、返回用户身份。identity 字段是后续授权决策的基础，而 permissions 等额外字段则为更细粒度的访问控制提供数据支持。

认证处理器可以接收多种参数，包括原始请求对象、请求体、路径、HTTP 方法、路径参数、查询参数、请求头以及 Authorization 头值。这种灵活性使得可以实现各种复杂的认证方案，从简单的 API 密钥到完整的 OAuth2 流程。

在图中访问认证信息

认证成功后，LangGraph 会将用户信息注入到运行时的配置对象中，使其在整个图执行过程中可用。在节点函数中，可以通过 config 参数访问当前用户：

def my_node(state, config):
    # 从配置中获取用户身份信息
    user_config = config["configurable"].get("langgraph_auth_user")
    if user_config:
        user_id = user_config.get("identity")
        # 基于用户身份执行特定逻辑
        # ...

对于工具函数等嵌套调用场景，可以使用 get_config 函数获取当前配置：

from langgraph.config import get_config

def search_everything(query: str):
    # 在工具内部获取用户身份信息
    config = get_config()
    organization_id = config["configurable"].get("x-organization-id")
    # 使用组织 ID 过滤搜索结果
    # ...

这种设计使得用户身份信息能够穿透整个调用栈，无论是主流程节点还是深层工具调用，都能获取到当前请求的用户上下文。

代理认证与委托访问

一个强大的特性是，LangGraph 支持代理认证，允许 AI 代理以用户身份访问外部资源。这在构建需要调用外部 API 的智能体时尤为重要。

认证流程扩展后包括从安全密钥存储中获取用户特定令牌，并将其传递给代理。代理在调用外部服务时携带这些令牌，外部服务验证令牌后执行操作，最终返回结果。

例如，当代理需要访问用户的 GitHub 或 Jira 时，可以在认证阶段从密钥存储中获取这些服务的访问令牌，并将其包含在返回的用户信息中。代理节点随后可以提取这些令牌，在调用相应服务时进行身份验证。

@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    api_key = headers.get("x-api-key")
    if not api_key or not is_valid_key(api_key):
        raise Auth.exceptions.HTTPException(status_code=401, detail="Invalid API key")
    
    # 从密钥存储获取用户特定令牌
    user_tokens = await fetch_user_tokens(api_key)
    
    return {
        "identity": api_key,
        "github_token": user_tokens.github_token,
        "jira_token": user_tokens.jira_token,
        # ... 其他自定义字段
    }

在图中使用这些令牌时，务必从安全的密钥存储中获取，避免将敏感信息直接存储在图状态中。

资源授权过滤规则

认证确认用户身份后，授权机制决定用户能访问哪些资源。LangGraph 通过 @auth.on 装饰器注册授权处理器，这些处理器可以修改资源元数据并返回过滤规则。

基础授权模式

最简单的授权模式是为所有资源添加所有者信息，确保用户只能访问自己的数据：

@auth.on
async def add_owner(ctx: Auth.types.AuthContext, value: dict):
    """为资源添加所有者信息"""
    # 创建过滤器，限制只能访问当前用户的资源
    filters = {"owner": ctx.user.identity}
    
    # 获取或创建元数据字典
    metadata = value.setdefault("metadata", {})
    # 将所有者信息添加到元数据
    metadata.update(filters)
    
    # 返回过滤器，应用于所有操作
    return filters

这个处理器实现了两个关键功能：在资源创建时添加所有者元数据，并返回过滤器限制后续访问。当用户尝试读取或更新资源时，系统会自动应用这些过滤器，确保只能看到自己的数据。

资源特定处理器

对于更精细的控制，可以为特定资源和操作注册专用处理器。LangGraph 支持三级处理器匹配机制：

  1. 全局处理器（@auth.on）：匹配所有资源和操作
  2. 资源级处理器（如 @auth.on.threads）：匹配特定资源的所有操作
  3. 操作级处理器（如 @auth.on.threads.create）：匹配特定资源的特定操作

系统总是选择最具体的匹配处理器。例如，@auth.on.threads.create 会优先于 @auth.on.threads 和 @auth.on 执行。

# 拒绝所有未处理的请求
@auth.on
async def reject_unhandled_requests(ctx: Auth.types.AuthContext, value: Any):
    raise Auth.exceptions.HTTPException(
        status_code=403,
        detail="Forbidden"
    )

# 线程创建处理器
@auth.on.threads.create
async def on_thread_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create.value
):
    if "write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="User lacks the required permissions."
        )
    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.threads.read.value
):
    # 读取操作只需返回过滤器
    return {"owner": ctx.user.identity}

# 助手创建处理器
@auth.on.assistants.create
async def on_assistant_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.assistants.create.value
):
    if "assistants:create" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="User lacks the required permissions."
        )

这种分层设计使得可以为不同资源设置不同的访问策略。例如，普通用户可能只能创建和读取自己的线程，但只有管理员才能创建新的助手配置。

基于权限的访问控制

当需要更复杂的角色体系时，可以基于权限进行访问控制。首先在认证阶段定义用户的权限列表，然后在授权处理器中检查这些权限：

# 认证阶段定义权限
@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    # ... 验证逻辑
    return {
        "identity": "user-123",
        "permissions": ["threads:write", "threads:read", "assistants:create"]
    }

# 授权阶段检查权限
@auth.on.threads.create
async def create_thread(ctx: Auth.types.AuthContext, value: dict):
    if "threads:write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="Unauthorized"
        )
    # ... 添加元数据和过滤器

@auth.on.threads.read
async def read_thread(ctx: Auth.types.AuthContext, value: dict):
    if "threads:read" not in ctx.permissions and "threads:write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="Unauthorized"
        )
    # ... 返回过滤器

这种模式支持 RBAC（基于角色的访问控制）和更细粒度的权限管理。权限可以在认证时从外部系统（如 LDAP、OAuth2 范围）动态获取，使得安全策略能够与企业现有的身份管理体系集成。

存储命名空间隔离

对于长期记忆存储（Store），授权机制同样适用。可以通过检查存储项的命名空间来实现用户级隔离：

@auth.on.store
async def on_store_access(ctx: Auth.types.AuthContext, value: dict):
    # 命名空间是类似文件路径的元组
    namespace = value.get("namespace", [])
    # 确保用户只能访问自己的命名空间
    if namespace and namespace[0] != ctx.user.identity:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="Not authorized to access this namespace"
        )

这种机制确保了用户数据的完全隔离，防止跨用户数据泄露。

自定义认证提供方集成

生产环境通常需要集成企业现有的认证系统。LangGraph 支持接入任何 OAuth2 兼容的认证提供方，如 Supabase Auth、Auth0、Okta 等。

OAuth2 集成架构

OAuth2 集成涉及三个主要角色：授权服务器（身份提供方）、应用后端（LangGraph）和客户端应用。标准流程中，用户在客户端发起登录，重定向到授权服务器进行身份验证，获取访问令牌后，客户端携带令牌访问 LangGraph 后端，后端验证令牌并服务于受保护资源。

sequenceDiagram
    participant User
    participant Client
    participant AuthServer
    participant LangGraph
    
    User->>Client: 发起登录
    Client->>AuthServer: 重定向到登录页
    User->>AuthServer: 输入凭证
    AuthServer-->>Client: 返回访问令牌
    Client->>LangGraph: 请求携带令牌
    LangGraph->>AuthServer: 验证令牌
    AuthServer-->>LangGraph: 确认有效性
    LangGraph-->>Client: 返回受保护资源

实现令牌验证

在 LangGraph 中实现 OAuth2 令牌验证，需要在认证处理器中调用认证提供方的验证端点：

import os
import httpx
from langgraph_sdk import Auth

auth = Auth()

SUPABASE_URL = os.environ["SUPABASE_URL"]
SUPABASE_SERVICE_KEY = os.environ["SUPABASE_SERVICE_KEY"]

@auth.authenticate
async def authenticate(authorization: str | None):
    """验证 JWT 令牌并提取用户信息"""
    assert authorization
    scheme, token = authorization.split()
    assert scheme.lower() == "bearer"
    
    try:
        # 调用 Supabase 验证端点
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{SUPABASE_URL}/auth/v1/user",
                headers={
                    "Authorization": authorization,
                    "apiKey": SUPABASE_SERVICE_KEY,
                },
            )
            assert response.status_code == 200
            user = response.json()
            return {
                "identity": user["id"],
                "email": user["email"],
                "is_authenticated": True,
            }
    except Exception as e:
        raise Auth.exceptions.HTTPException(status_code=401, detail=str(e))

这种方式将令牌验证委托给专业的认证服务，确保安全性。验证成功后，可以从响应中提取用户 ID、邮箱等信息，用于后续授权决策。

客户端实现

客户端需要实现 OAuth2 登录流程，获取令牌后通过请求头传递给 LangGraph：

from langgraph_sdk import get_client

# 用户登录后获取的访问令牌
access_token = "user-access-token"

# 创建客户端时添加认证头
client = get_client(
    url="https://your-deployment.langgraph.app",
    headers={"Authorization": f"Bearer {access_token}"}
)

# 后续所有请求都会自动携带认证信息
thread = await client.threads.create()

对于 Web 应用，通常会在登录成功后将令牌存储在 localStorage 或 sessionStorage 中，并在每次 API 调用时从存储中读取并添加到请求头。

处理 Studio 用户

在开发阶段，可能需要允许 LangGraph Studio 的访问，即使启用了自定义认证。可以通过检查用户类型来实现特殊处理：

from langgraph_sdk.auth import is_studio_user

@auth.on
async def add_owner(ctx: Auth.types.AuthContext, value: dict):
    # 对 Studio 用户放宽限制
    if is_studio_user(ctx.user):
        return {}
    
    # 普通用户执行常规授权
    filters = {"owner": ctx.user.identity}
    metadata = value.setdefault("metadata", {})
    metadata.update(filters)
    return filters

is_studio_user 函数用于识别来自 LangGraph Studio 的请求，这在开发和调试阶段非常有用。生产环境中可以通过配置禁用此行为。

OpenAPI 安全架构

良好的 API 文档应当清晰说明认证要求。LangGraph 允许自定义 OpenAPI 规范中的安全方案，帮助 API 使用者理解如何认证，甚至支持自动生成客户端代码。

默认安全方案

LangGraph Cloud 默认使用 API 密钥认证，在 OpenAPI 文档中体现为：

components:
  securitySchemes:
    apiKeyAuth:
      type: apiKey
      in: header
      name: x-api-key
security:
  - apiKeyAuth: []

自托管部署默认没有安全方案，需要手动配置。

自定义安全方案

在 langgraph.json 中通过 auth.openapi 字段自定义安全方案：

{
  "auth": {
    "path": "./auth.py:my_auth",
    "openapi": {
      "securitySchemes": {
        "OAuth2": {
          "type": "oauth2",
          "flows": {
            "implicit": {
              "authorizationUrl": "https://your-auth-server.com/oauth/authorize",
              "scopes": {
                "me": "读取当前用户信息",
                "threads": "访问和管理线程"
              }
            }
          }
        }
      },
      "security": [
        {"OAuth2": ["me", "threads"]}
      ]
    }
  }
}

对于 API 密钥方案，配置更为简单：

{
  "auth": {
    "path": "./auth.py:my_auth",
    "openapi": {
      "securitySchemes": {
        "apiKeyAuth": {
          "type": "apiKey",
          "in": "header",
          "name": "X-API-Key"
        }
      },
      "security": [
        {"apiKeyAuth": []}
      ]
    }
  }
}

配置完成后，部署应用并访问 /docs 端点即可看到更新后的 API 文档。API 使用者可以直观地了解认证方式，部分工具还能基于 OpenAPI 规范自动生成认证代码。

需要注意的是，OpenAPI 配置仅影响文档，实际的认证逻辑必须在 auth.authenticate 处理器中实现。两者需要保持一致，避免文档与实际行为不符。

总结

本章全面介绍了 LangGraph 的认证与安全防护体系。从认证与授权的基础概念出发，深入讲解了认证中间件的实现方式、资源授权过滤规则的配置方法、自定义认证提供方的集成步骤，以及 OpenAPI 安全架构的文档化。

认证机制通过 Auth 对象实现，支持灵活的凭证验证和用户身份提取。授权系统提供三级处理器匹配机制，从全局到特定操作，满足不同粒度的访问控制需求。集成 OAuth2 等外部认证提供方时，LangGraph 的架构能够无缝衔接企业现有身份体系。

安全是生产应用的基石。通过合理配置认证授权规则，可以确保用户数据隔离、防止未授权访问，同时为 AI 代理提供安全的委托访问能力。这些机制共同构成了 LangGraph 应用的安全防线。

下一章将探讨 LangGraph 的高级开发模式，包括状态流控制、子图嵌套调用和可恢复任务设计，进一步提升应用的架构能力和可维护性。

上一章

生产部署策略

 下一章

高级开发模式