langgraph检查点信息加密持久化-开发者社区

一.背景

LangGraph 作为 LangChain 生态中聚焦大模型应用流程编排与状态管理的核心框架，其 ** 检查点（Checkpoint）** 机制是实现流程中断恢复、时间旅行、流程重放的核心基础 —— 通过持久化存储流程执行的全量状态（节点执行记录、上下文数据、决策路径、大模型交互记录等），支撑流程的柔性管控。但在企业级生产环境中，原生的检查点持久化机制（多为明文存储，如本地文件、Redis、PostgreSQL 明文存储）存在严重的安全隐患，因此 “检查点信息加密持久化” 能力应运而生。这一能力的需求源于原生检查点存储在数据安全、合规性上的核心痛点，也是 LangGraph 适配企业级 “敏感数据保护、合规审计、隐私管控” 诉求的关键升级。

1.LangGraph 原生检查点持久化的核心痛点

LangGraph 原生支持将检查点数据持久化到本地内存、文件系统、Redis、PostgreSQL 等存储介质，但未内置加密机制，检查点信息以明文形式存储，在企业级场景中暴露以下致命问题：

1. 敏感数据泄露风险，威胁业务安全

检查点中包含大量敏感信息，这些数据明文存储时，一旦存储介质被攻破（如 Redis 未授权访问、数据库脱库、服务器被入侵），敏感数据会直接泄露：

业务敏感数据：流程执行中的用户输入（如客户的手机号、订单信息、企业核心业务数据）、业务决策结果（如合同审查的风险评估结论、财务数据分析结果）；
隐私数据：用户的个人身份信息（PII）、对话记录、偏好数据（如智能助手的交互历史）；
系统敏感信息：大模型的调用参数（如 API Key、Prompt 模板）、流程的权限配置、智能体的协作密钥。例如，金融行业的风控决策流程检查点中包含用户的征信数据、交易记录，明文存储可能导致用户隐私泄露，引发业务风险与法律纠纷。

2. 合规性不达标，无法满足监管要求

金融、政企、医疗等行业受《网络安全法》《数据安全法》《个人信息保护法》（PIPL）、GDPR 等法规约束，要求对敏感数据进行加密存储、访问审计、数据脱敏：

原生检查点明文存储无法满足 “个人信息加密保存” 的合规要求，企业可能面临监管处罚；
检查点数据缺乏加密溯源能力，无法证明数据在存储全生命周期中未被篡改或泄露，合规审计时无法提供有效证据；
部分行业（如医疗）要求数据存储需达到等保三级以上标准，明文存储直接违反该要求。

3. 数据篡改风险，破坏流程完整性

检查点数据是流程恢复、重放的唯一依据，明文存储时，恶意攻击者或内部人员可轻易篡改检查点内容（如修改流程执行的决策结果、伪造节点执行记录）：

例如，在合同审查流程中，攻击者篡改检查点中的 “风险评估结论”（将高风险改为低风险），流程恢复后会生成错误的审查报告，导致企业遭受经济损失；
篡改后的检查点数据还会导致流程重放、时间旅行功能失效，无法还原真实的流程执行轨迹，破坏流程的完整性与可信度。

4. 多租户场景下数据隔离不足

在 Saas 模式的 LangGraph 应用中，多个租户的检查点数据通常存储在同一存储介质（如共享的 PostgreSQL 数据库）中，原生明文存储仅能通过逻辑隔离（如租户 ID 分表）实现数据分离，无法做到物理级隔离：

内部运维人员可轻易访问其他租户的检查点数据，违反租户数据隔离的商业协议；
一旦出现配置错误（如租户 ID 过滤逻辑失效），会导致不同租户的检查点数据混淆，引发数据泄露与业务混乱。

5. 持久化介质适配性差，加密逻辑需重复开发

LangGraph 支持多种检查点持久化介质（本地文件、Redis、PostgreSQL、S3 等），企业若要实现加密存储，需针对不同介质手动开发加密逻辑（如 Redis 数据加密、PostgreSQL 字段加密、文件加密）：

开发成本高，且不同介质的加密逻辑不一致，易出现加密漏洞；
加密逻辑与业务代码耦合，后续升级存储介质时，需重新修改加密逻辑，维护成本剧增。

2.LangGraph 检查点信息加密持久化的核心价值

LangGraph 检查点信息加密持久化，本质是通过 “加密层 + 持久化层” 的分层架构，在检查点数据写入存储前进行加密，读取时解密，实现检查点数据全生命周期的加密保护。这一能力解决了原生存储的痛点，核心价值体现在：

1. 全维度加密保护，杜绝敏感数据泄露

通过对称加密（如 AES-256）、非对称加密（如 RSA）或混合加密方案，对检查点中的敏感数据进行端到端加密：

整体加密：将整个检查点对象序列化为字节流后，使用 AES-256 加密后再存储，确保所有数据均被保护；
字段级加密：对检查点中的核心敏感字段（如用户手机号、API Key）单独使用非对称加密，提升加密粒度；
密钥管理：对接企业级密钥管理服务（KMS，如 AWS KMS、阿里云 KMS、HashiCorp Vault），避免密钥硬编码，实现密钥的安全存储与轮换。即使存储介质被攻破，攻击者获取的也只是加密后的密文，无法解析出有效信息，从根本上杜绝敏感数据泄露。

2. 满足合规要求，通过监管审计

加密持久化机制可满足各类法规与行业标准的要求，为企业合规审计提供有力支撑：

符合《个人信息保护法》中 “个人信息存储需采取加密等安全措施” 的规定，避免监管处罚；
生成加密审计日志，记录检查点数据的加密时间、加密算法、密钥版本、访问人员等信息，满足合规审计的可追溯要求；
支持数据脱敏与加密存储结合，对检查点中的个人信息进行脱敏后再加密，进一步满足隐私保护要求。

3. 防止数据篡改，保障流程完整性

在加密的基础上，为检查点数据添加数字签名（如 SHA-256 哈希 + RSA 签名）：

写入检查点时，生成数据哈希值并使用私钥签名，与加密数据一同存储；
读取检查点时，先验证签名的有效性，确认数据未被篡改后再解密。即使攻击者篡改了加密后的密文，签名验证也会失败，流程会拒绝加载被篡改的检查点，保障流程执行的完整性与可信度。

4. 强化多租户数据隔离，适配 Saas 场景

针对多租户场景，采用租户级密钥隔离的加密策略：

为每个租户分配独立的加密密钥，检查点数据使用对应租户的密钥加密；
即使多个租户的检查点数据存储在同一介质中，不同租户的密钥无法互相解密，实现物理级的数据隔离。这一策略解决了原生逻辑隔离的不足，满足 Saas 平台的租户数据安全要求，同时避免运维人员越权访问租户数据。

5. 统一加密层，适配多持久化介质

构建独立的加密持久化层，与 LangGraph 的检查点存储接口解耦：

无论检查点存储在 Redis、PostgreSQL、S3 还是本地文件，均通过统一的加密层处理，无需为不同介质开发单独的加密逻辑；
支持加密算法的动态切换（如从 AES-256 切换为 ChaCha20），无需修改业务代码，提升适配性与可维护性。

3.典型应用场景

金融行业风控决策流程：风控流程的检查点包含用户征信数据、交易记录、风控评分等敏感信息，加密持久化可防止数据泄露，满足金融行业的合规要求。
政企智能客服 / 助手流程：检查点中包含政企用户的身份信息、业务咨询记录、涉密决策数据，加密持久化可保障数据安全，符合等保三级标准。
医疗行业病历分析流程：检查点存储患者的病历数据、诊断结果等隐私信息，加密持久化可满足《医疗保障基金使用监督管理条例》的隐私保护要求。
Saas 模式的大模型流程编排平台：多租户的检查点数据通过租户级密钥加密存储，实现数据物理隔离，避免租户数据泄露。
企业级合同审查 / 项目审批流程：检查点包含合同条款、审批意见、企业核心业务数据，加密 + 数字签名可防止数据篡改，保障流程决策的可信度。

4.关键优势总结

LangGraph 检查点信息加密持久化的核心价值，是将检查点数据从 “明文裸存” 转化为 “加密存储 + 防篡改 + 可追溯” 的安全状态：既解决了原生存储的敏感数据泄露、篡改、合规性不足的痛点，又通过统一加密层适配多存储介质，降低企业开发与维护成本。这一能力让 LangGraph 能够适配金融、政企、医疗等对数据安全要求严苛的行业场景，是大模型应用从 “演示级” 走向 “生产级、合规级” 的关键安全支撑。

综上，LangGraph 检查点信息加密持久化的需求，源于企业对大模型流程数据 “安全性、合规性、可信度” 的核心诉求：解决了原生检查点存储的安全隐患，支撑金融、医疗、政企等敏感行业的落地，为复杂大模型应用的稳定运行提供了安全保障。

二.具体实现

1.安装依赖包

pip install langgraph-checkpoint-postgres pip install psycopg[binary] pip install pycryptodome

2.引入依赖

import sys import io from langgraph.store.memory import InMemoryStore from langgraph.constants import START, END from langgraph.graph import StateGraph from typing_extensions import TypedDict from langgraph.store.base import BaseStore from langgraph.checkpoint.memory import MemorySaver from langgraph.checkpoint.serde.encrypted import EncryptedSerializer from langgraph.checkpoint.postgres import PostgresSaver

3.设置加密秘钥

if sys.platform == 'win32': # Windows系统设置控制台编码为UTF-8 sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8') # 设置环境变量 import os os.environ['PYTHONIOENCODING'] = 'utf-8' # 设置加密密钥（如果未设置）- 需要32字符的字符串（UTF-8编码后为32字节） if 'LANGGRAPH_AES_KEY' not in os.environ: # 生成32个ASCII字符，确保UTF-8编码后正好32字节 from Crypto.Random import get_random_bytes key_bytes = get_random_bytes(32) # 将字节转换为可打印的ASCII字符（每个字节对应一个字符） os.environ['LANGGRAPH_AES_KEY'] = ''.join(chr(b) if 32 <= b < 127 else chr(65 + (b % 26)) for b in key_bytes)

4.定义状态类

# 1. 定义状态 class EventState(TypedDict): count_status: int #累计计数状态

5.定义节点函数

def first_node(state: EventState, *, store: BaseStore) -> EventState: state["count_status"] += 1 print(f"first_node: {state['count_status']}") return state def second_node(state: EventState, *, store: BaseStore) -> EventState: state["count_status"] += 3 print(f"second_node: {state['count_status']}") return state

6.构建图

# 初始化状态 state = {"count_status": 0} # 创建事件流 graph = StateGraph(EventState) graph.add_node("first_node", first_node) graph.add_node("second_node", second_node) graph.add_edge(START, "first_node") graph.add_edge("first_node", "second_node") graph.add_edge("second_node", END)

7.定义pg持久层，设置加密，作为图运行配置

serde = EncryptedSerializer.from_pycryptodome_aes() with PostgresSaver.from_conn_string("postgresql://postgres:xxxxxx@localhost:5432/test") as checkpointer: checkpointer.serde = serde # 设置序列化器 checkpointer.setup() graph = graph.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "thread-155"}} graph.invoke(state, config=config)

8.运行后可以在postgres看到生成四个表

分别明细如下，存储了加密的检查点信息

langgraph检查点信息加密持久化