BytesRouter

主题

LangSmith 文档版本 跟随上游 源快照 254877da 更新于 2026-10-18

添加静态加密

Agent Server 支持对检查点数据和元数据进行静态加密。你可以在使用单个密钥的基础加密与用于高级用例的自定义加密之间选择。

选择加密方法

方法加密什么用例
Basic encryption检查点 blob，可选地包括 JSON 字段单个静态密钥、自动 AES 加密、选择性字段加密
Custom encryption检查点、线程、运行、助手、cron 和存储每租户密钥、KMS 集成

基础加密

对于用单个静态密钥的简单加密，设置 LANGGRAPH_AES_KEY 环境变量。LangGraph 会自动使用 AES 加密检查点 blob。

  1. 在 langgraph.json 中把 pycryptodome 添加到你的依赖：

    json
    {
  "dependencies": [".", "pycryptodome"],
  "graphs": {
    "agent": "./agent.py:graph"
  }
}

  2. 将 LANGGRAPH_AES_KEY 环境变量设为一个 16、24 或 32 字节的密钥（分别用于 AES-128、AES-192 或 AES-256）。

加密 JSON 字段

要同时加密特定的 JSON 字段，将 LANGGRAPH_AES_JSON_KEYS 设为一个逗号分隔的、要加密的键列表：

bash
export LANGGRAPH_AES_KEY="your-16-24-or-32-byte-key"
export LANGGRAPH_AES_JSON_KEYS="api_key,secret_token,user_credentials"

这些键在线程、助手、运行、cron 和存储数据中无论它们出现之处都会被加密。

WARNING

加密字段无法被搜索或过滤。

系统字段不能被加密：langgraph_version、langgraph_api_version、langgraph_plan、langgraph_host、langgraph_api_url、langgraph_request_id、langgraph_auth_user_id 和 langgraph_auth_permissions。

自定义加密

NOTE

需要 Agent Server 版本 0.6.22+ 和 Python SDK 版本 langgraph-sdk>=0.3.1。

WARNING

Agent Server 版本 0.5.34–0.6.21 包含了一个预发布版本的自定义加密。用这些版本加密的数据在升级到 0.6.22+ 时会损坏。不要在这些版本上使用自定义加密。

WARNING

仅当基础加密无法满足你的需求时才使用自定义加密。自定义加密需要你实现并维护加密处理器，并增加运维复杂度。如果你只需要单个静态密钥配合可选的选择性字段加密，请改用基础加密。

当你需要以下内容时使用自定义加密：

  • 每租户密钥隔离 — 为不同客户使用不同的加密密钥
  • KMS 集成 — 使用 AWS KMS、Google Cloud KMS 或 HashiCorp Vault 进行密钥管理、轮换和审计日志

工作原理

  1. 在 langgraph.json 中配置加密模块路径
  2. 定义你的加密模块，包含用于 blob 和 JSON 加密的处理器
  3. 通过 X-Encryption-Context 标头传入加密上下文（例如租户 ID）
  4. LangGraph 在存储数据之前和检索数据之后调用你的处理器

对于带密钥轮换和审计日志的生产部署，参阅使用 AWS Encryption SDK 的信封加密。

配置

将你的加密模块添加到 langgraph.json：

json
{
  "dependencies": ["."],
  "graphs": {
    "agent": "./agent.py:graph"
  },
  "encryption": {
    "path": "./encryption.py:encryption"
  }
}

NOTE

如果你从基础加密迁移，请保持 LANGGRAPH_AES_KEY 已配置。自定义加密处理新写入，而现有的 AES 加密数据仍可读。

定义你的加密模块

Blob 加密（检查点）

Blob 处理器加密检查点数据——即从图执行序列化的状态。这里是一个使用每租户密钥配合 Fernet（cryptography 库中的一种对称加密方案）的简化示例：

python
import os
from cryptography.fernet import Fernet
from langgraph_sdk import Encryption, EncryptionContext

encryption = Encryption()

# In production, fetch from a secrets manager
TENANT_KEYS = {
    "tenant-a": Fernet(os.environ["TENANT_A_KEY"]),
    "tenant-b": Fernet(os.environ["TENANT_B_KEY"]),
}

def _get_fernet(ctx: EncryptionContext) -> Fernet:
    tenant_id = ctx.metadata.get("tenant_id")
    if not tenant_id or tenant_id not in TENANT_KEYS:
        raise ValueError(f"Unknown tenant: {tenant_id}")
    return TENANT_KEYS[tenant_id]

@encryption.encrypt.blob
async def encrypt_blob(ctx: EncryptionContext, data: bytes) -> bytes:
    return _get_fernet(ctx).encrypt(data)

@encryption.decrypt.blob
async def decrypt_blob(ctx: EncryptionContext, data: bytes) -> bytes:
    return _get_fernet(ctx).decrypt(data)

ctx.metadata 字典来自 X-Encryption-Context 标头，并以明文与加密数据一并存储，因此在解密时使用正确的密钥。

JSON 加密（元数据）

JSON 处理器加密结构化数据，例如线程元数据、助手上下文和运行 kwargs。与 blob 加密不同，你可以选择加密哪些字段——让某些字段保持未加密以用于搜索和过滤。

python
import json
import os
from cryptography.fernet import Fernet
from langgraph_sdk import Encryption, EncryptionContext

encryption = Encryption()

TENANT_KEYS = {
    "tenant-a": Fernet(os.environ["TENANT_A_KEY"]),
    "tenant-b": Fernet(os.environ["TENANT_B_KEY"]),
}

SKIP_FIELDS = {
    "tenant_id", "owner",
    "run_id", "thread_id", "graph_id", "assistant_id", "user_id", "checkpoint_id",
    "source", "step", "parents", "run_attempt",
    "langgraph_version", "langgraph_api_version", "langgraph_plan", "langgraph_host",
    "langgraph_api_url", "langgraph_request_id", "langgraph_auth_user",
    "langgraph_auth_user_id", "langgraph_auth_permissions",
}
ENCRYPTED_PREFIX = "encrypted:"

def _get_fernet(ctx: EncryptionContext) -> Fernet:
    tenant_id = ctx.metadata.get("tenant_id")
    if not tenant_id or tenant_id not in TENANT_KEYS:
        raise ValueError(f"Unknown tenant: {tenant_id}")
    return TENANT_KEYS[tenant_id]

@encryption.encrypt.json
async def encrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    fernet = _get_fernet(ctx)
    result = {}
    for k, v in data.items():
        if k in SKIP_FIELDS or v is None:
            result[k] = v
        else:
            value_json = json.dumps(v)
            encrypted = fernet.encrypt(value_json.encode()).decode()
            result[k] = ENCRYPTED_PREFIX + encrypted
    return result

@encryption.decrypt.json
async def decrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    fernet = _get_fernet(ctx)
    result = {}
    for k, v in data.items():
        if isinstance(v, str) and v.startswith(ENCRYPTED_PREFIX):
            encrypted_value = v[len(ENCRYPTED_PREFIX):]
            decrypted = fernet.decrypt(encrypted_value.encode()).decode()
            result[k] = json.loads(decrypted)
        else:
            result[k] = v
    return result

JSON 加密的注意事项

WARNING

加密字段无法被搜索或过滤。 设计你的元数据 schema，让需要查询的字段保持未加密。

WARNING

JSON 加密器必须保留键结构。 SQL JSONB 合并操作在键级别工作。更改键的加密器——无论是通过合并字段（例如把敏感数据移入 __encrypted__）还是加密键名本身——都会在合并期间导致数据丢失。使用按密钥加密：就地将值转换，同时保留键。

NOTE

迁移注意事项： 在加密值中使用可识别的前缀或格式，以便你的解密器能够检测并跳过未加密的数据。这允许你将来加密更多字段而无需重新加密现有记录。上面的示例使用了这一模式。

NOTE

性能注意事项： 按密钥加密意味着每个字段一次加密调用。如果你的加密涉及对外部服务（例如 KMS）的往返，这会显著影响延迟。考虑在本地缓存数据密钥，或使用信封加密，即用 KMS 加密一个本地数据密钥并将其用于多个字段。

用于授权的、用户定义的字段（例如 tenant_id、owner）通常应保持未加密，用于搜索和过滤的字段也是如此。此外，某些系统管理的字段永远不会被加密：

  • 资源标识符（thread_id、run_id、assistant_id、graph_id、checkpoint_id、task_id）
  • 大多数以 langgraph_ 开头的字段（langgraph_auth_user 除外）
  • 必需的检查点元数据（source、step、parents、run_attempt）
  • 用于调度和编排的内部字段（__after_seconds__、__request_start_time_ms__，以及大多数以 __pregel 开头的字段）
  • 在运行的 config 中指定的运行级执行限制（max_concurrency、recursion_limit）
  • 在运行的 config.configurable 中指定的线程 TTL 更新（ttl）

加密了什么

JSON 处理器（@encryption.encrypt.json / @encryption.decrypt.json）被递归地应用于以下字段：

  • thread.metadata、thread.values
  • assistant.metadata、assistant.context
  • run.metadata、run.kwargs
  • cron.metadata、cron.payload
  • store.value

某些字段被排除在加密之外。 除非另有说明，这些排除适用于嵌套 JSON 对象的每一层，而不仅是根层。

Blob 处理器（@encryption.encrypt.blob / @encryption.decrypt.blob）被应用于检查点 blob（图执行状态）。

从身份验证派生上下文

不必显式传入 X-Encryption-Context，你可以从已认证的用户派生加密上下文：

python
from langgraph_sdk import Encryption, EncryptionContext
from starlette.authentication import BaseUser

encryption = Encryption()

@encryption.context
async def get_encryption_context(user: BaseUser, ctx: EncryptionContext) -> dict:
    return {
        **ctx.metadata,
        "tenant_id": user["tenant_id"],
    }

此处理器在每次请求身份验证后运行一次。返回的字典成为该请求中所有加密操作的 ctx.metadata。

传入加密上下文

通过 X-Encryption-Context 标头传入加密上下文。上下文是你定义的任意数据——你掌控 schema，并可包含你的加密逻辑所需的任何字段（例如 tenant_id、key_version）。上下文在你的处理器中作为 ctx.metadata 可用，并以明文存储以便在解密时使用。

python
import base64
import json
from langgraph_sdk import get_client

encryption_context = base64.b64encode(
    json.dumps({"tenant_id": "tenant-a"}).encode()
).decode()

client = get_client(url="http://localhost:2024")

result = await client.runs.wait(
    thread_id=None,
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "Hello"}]},
    headers={"X-Encryption-Context": encryption_context},
)

NOTE

加密上下文以明文存储。在解密时它会被自动恢复——调用方读取时无需传入该标头。

使用 AWS Encryption SDK 的信封加密

对于 AWS 上的生产部署，使用 AWS Encryption SDK 配合 AWS KMS，或在你云提供商中的等价方案。这一方式：

  • 自动处理信封加密（无需手动打包密钥）
  • 提供密钥轮换和审计日志
  • 将密文绑定到加密上下文（租户隔离）
  • 在本地缓存数据密钥以避免重复的 KMS 调用、延迟和速率限制

完整示例

python
import base64
import json
import os

import aws_encryption_sdk
from aws_encryption_sdk import (
    CachingCryptoMaterialsManager,
    CommitmentPolicy,
    LocalCryptoMaterialsCache,
    StrictAwsKmsMasterKeyProvider,
)
from langgraph_sdk import Encryption, EncryptionContext

encryption = Encryption()

# The SDK uses envelope encryption: one KMS API call generates a data key,
# then encrypts/decrypts locally. The cache reuses data keys across operations.
client = aws_encryption_sdk.EncryptionSDKClient(
    commitment_policy=CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT
)
key_provider = StrictAwsKmsMasterKeyProvider(key_ids=[os.environ["KMS_KEY_ARN"]])
cache = LocalCryptoMaterialsCache(capacity=100)
cmm = CachingCryptoMaterialsManager(
    master_key_provider=key_provider,
    cache=cache,
    max_age=300.0,
    max_messages_encrypted=100,
)

SKIP_FIELDS = {
    "tenant_id", "owner",
    "run_id", "thread_id", "graph_id", "assistant_id", "user_id", "checkpoint_id",
    "source", "step", "parents", "run_attempt",
    "langgraph_version", "langgraph_api_version", "langgraph_plan", "langgraph_host",
    "langgraph_api_url", "langgraph_request_id", "langgraph_auth_user",
    "langgraph_auth_user_id", "langgraph_auth_permissions",
}
ENCRYPTED_PREFIX = "encrypted:"

@encryption.encrypt.blob
async def encrypt_blob(ctx: EncryptionContext, data: bytes) -> bytes:
    ciphertext, _ = client.encrypt(
        source=data,
        materials_manager=cmm,
        encryption_context={"tenant_id": ctx.metadata["tenant_id"]},
    )
    return ciphertext

@encryption.decrypt.blob
async def decrypt_blob(ctx: EncryptionContext, data: bytes) -> bytes:
    plaintext, _ = client.decrypt(source=data, key_provider=key_provider)
    return plaintext

@encryption.encrypt.json
async def encrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    tenant_id = ctx.metadata["tenant_id"]
    result = {}
    for k, v in data.items():
        if k in SKIP_FIELDS or v is None:
            result[k] = v
        else:
            ciphertext, _ = client.encrypt(
                source=json.dumps(v).encode(),
                materials_manager=cmm,
                encryption_context={"tenant_id": tenant_id},
            )
            result[k] = ENCRYPTED_PREFIX + base64.b64encode(ciphertext).decode()
    return result

@encryption.decrypt.json
async def decrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    result = {}
    for k, v in data.items():
        if isinstance(v, str) and v.startswith(ENCRYPTED_PREFIX):
            ciphertext = base64.b64decode(v[len(ENCRYPTED_PREFIX):])
            plaintext, _ = client.decrypt(source=ciphertext, key_provider=key_provider)
            result[k] = json.loads(plaintext.decode())
        else:
            result[k] = v
    return result

encryption_context 通过 KMS 被密码学地绑定到密文——如果上下文不匹配，解密就会失败。上下文被嵌入到密文中，因此 decrypt 处理器不需要引用 ctx.metadata。

密钥轮换

KMS 自动处理主密钥轮换。当你在 KMS 密钥上启用自动轮换时，旧的已加密数据密钥仍可被解密，而新操作使用轮换后的密钥材料。无需重新加密现有数据。

相关

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可