openai-agents-python 加密会话EncryptedSession实战指南Fernet 透明加密、HKDF 按会话派生密钥与 TTL 自动过期【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本文是 openai-agents-pythonAgents SDK会话Session记忆体系中EncryptedSession的完整技术指南。EncryptedSession是一个透明的加密包装器可对任意已有的会话实现如SQLiteSession、SQLAlchemySession进行 Fernet 加密并在 TTL 到期后自动跳过过期历史消息。读完本文你将掌握如何安装encrypt扩展、如何用主密钥或 Fernet 密钥包装底层会话、HKDF 按会话派生密钥的底层原理、TTL 自动过期的实现机制以及如何在Runner.run中无缝接入加密会话本指南对应文档docs/sessions/encrypted_session.md。一、EncryptedSession 是什么在多轮对话场景中Agents SDK 通过会话Session自动维护对话历史每次运行前把历史输入前置给模型运行后把新产生的 items 写回存储从而省去手动调用.to_input_list()的麻烦见 会话总览。但历史数据通常以明文落库一旦数据库文件、Redis 或云数据库泄露用户对话内容将直接暴露。EncryptedSession正是为这一问题而生它是一个包装器wrapper本身不负责存储而是把任何符合 [Session][agents.memory.session.Session] 协议的底层会话包一层加密对外暴露完全一致的操作接口。核心特性透明加密Transparent encryption基于 Fernet 对称加密包装任意会话读写路径自动加解密业务代码几乎无感知按会话派生密钥Per-session keys使用 HKDF 密钥派生函数结合主密钥与会话 ID 为每个会话派生唯一密钥自动过期Automatic expirationTTL 过期后旧消息在读取时被静默跳过不会影响会话行为即插即用Drop-in replacement与任意既有会话实现兼容可直接替换Runner.run(..., session...)中的 session 参数。需要说明的是会话机制本身存在一个使用前提在同一 run 中session 不能与 run 级续接选项conversation_id、previous_response_id、auto_previous_response_id混用详见 docs/sessions/index.md这一点对EncryptedSession同样适用。二、安装与依赖加密会话依赖cryptography库属于 SDK 的可选依赖extrapip install openai-agents[encrypt]在 pyproject.toml 中可以看到对应声明[project.optional-dependencies] encrypt [cryptography45.0, 46]即encryptextra 对应cryptography45.0, 46。如果未安装该依赖就导入EncryptedSession会触发agents.extensions.memory._optional_imports中定义的raise_optional_dependency_error提示src/agents/extensions/memory/_optional_imports.py错误信息会明确建议执行上述安装命令。三、快速开始以SQLAlchemySession作为底层存储、包一层EncryptedSession为例完整示例见 examples/memory/encrypted_session_example.pyimport asyncio from agents import Agent, Runner from agents.extensions.memory import EncryptedSession, SQLAlchemySession async def main(): agent Agent(Assistant) # 1. 创建底层会话真实的存储后端 underlying_session SQLAlchemySession.from_url( user-123, urlsqliteaiosqlite:///:memory:, create_tablesTrue ) # 2. 用加密包装 session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keyyour-secret-key-here, ttl600 # 10 minutes ) # 3. 像普通会话一样直接使用 result await Runner.run(agent, Hello, sessionsession) print(result.final_output) if __name__ __main__: asyncio.run(main())整个使用方式与普通会话完全一致传入Runner.run后SDK 在每次运行前自动取出解密后的历史并前置给模型运行后自动把新 items加密后写回底层存储。测试用例 tests/extensions/memory/test_encrypt_session.py 验证了这一点同一加密会话连续两轮运行模型第二轮能记住第一轮的上下文。四、配置详解EncryptedSession构造函数签名为src/agents/extensions/memory/encrypt_session.pyEncryptedSession( session_id: str, underlying_session: SessionABC, encryption_key: str, ttl: int 600, )session_id会话 ID同时充当 HKDF 的 salt见下文密钥派生underlying_session真实存储后端如SQLiteSession、SQLAlchemySession、RedisSession、MongoDBSession等任何符合SessionABC的实现encryption_key主密钥可为 Fernet 密钥或任意原始字符串ttlToken 有效期秒默认 600 秒10 分钟。4.1 加密密钥的两种形式encryption_key支持两种输入源码_ensure_fernet_key_bytes负责判别src/agents/extensions/memory/encrypt_session.pyFernet 密钥base64 编码、解码后为 32 字节的密钥。构造时先尝试base64.urlsafe_b64decode若解码成功且长度为 32 字节则按 Fernet 密钥使用否则回退为原始字符串处理。空字符串会直接抛出ValueError(encryption_key not set; required for EncryptedSession.)。from agents.extensions.memory import EncryptedSession # 形式一Fernet 密钥base64 编码 session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keyyour-fernet-key-here, ttl600 )原始字符串任意密码串内部以 UTF-8 编码后作为 HKDF 的输入材料输入密钥同样可以正常工作# 形式二原始字符串内部会派生为密钥 session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keymy-secret-password, ttl600 )测试用例 tests/extensions/memory/test_encrypt_session.py 专门验证了原始字符串密钥可正常完成加解密往返。生产环境建议用 Fernet 密钥形式可通过cryptography.fernet.Fernet.generate_key()生成并将密钥存放在环境变量或密钥管理服务中切勿硬编码在代码里。4.2 TTL自动过期时长ttl控制每个加密 Token 的有效期过期后读取时被静默跳过# Items expire after 1 hour session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keysecret, ttl3600 # 1 hour in seconds ) # Items expire after 1 day session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keysecret, ttl86400 # 24 hours in seconds )注意 TTL 判定基于应用服务器的系统时钟Fernet 的decrypt(token, ttl...)使用time.time()判断。源码类文档明确提示为避免时钟漂移导致有效 Token 被误判过期请确保环境中所有服务器通过 NTP 同步时钟src/agents/extensions/memory/encrypt_session.py。五、与不同类型会话的组合使用EncryptedSession的底层会话可以是任何会话实现选择取决于你的存储场景5.1 与 SQLite 会话组合本地开发from agents import SQLiteSession from agents.extensions.memory import EncryptedSession # 创建加密的 SQLite 会话文件型持久化 underlying SQLiteSession(user-123, conversations.db) session EncryptedSession( session_iduser-123, underlying_sessionunderlying, encryption_keysecret-key )SQLiteSession支持内存库SQLiteSession(user_123)与文件库两种形态前者进程结束即丢失后者可持久化。5.2 与 SQLAlchemy 会话组合生产级from agents.extensions.memory import EncryptedSession, SQLAlchemySession # 创建加密的 SQLAlchemy 会话任何 SQLAlchemy 支持的数据库 underlying SQLAlchemySession.from_url( user-123, urlpostgresqlasyncpg://user:passlocalhost/db, create_tablesTrue ) session EncryptedSession( session_iduser-123, underlying_sessionunderlying, encryption_keysecret-key )SQLAlchemySession的完整说明见 SQLAlchemy 会话指南若底层会话设置了session_settingsEncryptedSession会将其原样透传给底层见下文第七节。5.3 与高级会话组合时的注意事项!!! warning 高级会话特性当 EncryptedSession 与 AdvancedSQLiteSession 这类带高级查询能力的会话实现组合时需要注意 - find_turns_by_content() 这类**基于内容的方法无法有效工作**因为消息内容已被加密 - 基于内容的搜索本质上是针对密文操作效果极其有限。 也就是说加密会与内容检索/分析类功能互斥。如果业务需要按内容回溯如会话分析、审计应在加密前另行设计检索通道。AdvancedSQLiteSession 的完整能力见 [高级 SQLite 会话指南](https://link.gitcode.com/i/e7c7812ee378684214cccf31568f1e6c)。5.4 透明委托EncryptedSession通过__getattr__将未定义的属性访问委托给底层会话src/agents/extensions/memory/encrypt_session.py因此底层会话的自定义方法如自定义统计、扩展 API依然可以直接调用。测试 tests/extensions/memory/test_encrypt_session.py 验证了自定义同步/异步方法均可经由加密会话正常访问。六、密钥派生机制HKDF 保证每个会话密钥唯一EncryptedSession使用HKDFHMAC-based Key Derivation Function从主密钥为每个会话派生独立加密密钥核心逻辑在_derive_session_fernet_keysrc/agents/extensions/memory/encrypt_session.py参数取值作用算法SHA-256哈希算法输入密钥材料IKM你提供的主密钥Fernet 密钥或原始字符串全局机密不落库Salt会话 IDUTF-8 编码每个会话不同产生不同输出Info 字符串bagents.session-store.hkdf.v1域隔离标识防止跨用途派生输出长度32 字节恰好构成 Fernet 密钥派生出的 32 字节密钥经base64.urlsafe_b64encode后构造 Fernet 实例用于加密该会话的所有 items。这一设计带来的安全性保证每个会话拥有唯一加密密钥即使两个会话使用相同主密钥由于 salt会话 ID不同派生密钥也完全不同没有主密钥无法派生任何会话密钥HKDF 的单向性决定了仅凭密文和会话 ID 无法恢复密钥会话之间无法互相解密即使某个会话密钥泄露也无法解密其他会话的数据。七、存储格式与读写流程7.1 加密信封Envelope写入时_wrap把每条 item 序列化为 JSONPydantic 对象优先model_dump()普通对象取__dict__再用 Fernet 加密封装成固定结构的信封src/agents/extensions/memory/encrypt_session.py{__enc__: 1, v: 1, kid: hkdf-v1, payload: fernet-token}__enc__加密标记恒为 1读取时据此识别信封v版本号当前为 1kid密钥标识当前为hkdf-v1便于未来密钥轮换payloadFernet 密文 Token。_unwrap反向解密非信封 item 原样透传解密失败或密钥错误InvalidToken、KeyError时返回None即静默丢弃src/agents/extensions/memory/encrypt_session.py。测试 tests/extensions/memory/test_encrypt_session.py 验证了落库数据确实为密文底层存储中看到的是__enc__ 1的信封明文内容不可见。示例脚本 examples/memory/encrypted_session_example.py 也演示了如何直接检查底层存储确认加密生效。7.2 自动过期TTL 到期静默跳过过期判定发生在解密阶段_unwrap调用cipher.decrypt(token, ttlself.ttl)Fernet 内部根据 Token 时间戳与ttl判断是否过期过期抛InvalidToken进而被捕获并返回None实现读取时静默跳过# Items older than TTL are silently ignored items await session.get_items() # 只返回未过期 items # Expired items dont affect session behavior result await Runner.run(agent, Continue conversation, sessionsession)三个值得注意的边界行为均有测试佐证过期 item 只跳过、不删除底层存储中数据仍在只是读取时被过滤tests/extensions/memory/test_encrypt_session.pypop_item自动重试从栈顶弹出时若遇到过期 item会继续向下弹出直到找到有效 item 或空src/agents/extensions/memory/encrypt_session.py测试见 tests/extensions/memory/test_encrypt_session.pyget_items(limitN)以解密后的有效数量为准因为过期 item 被跳过EncryptedSession会以翻倍窗口window 初始为 limit不足则 ×2向底层多次取数直到凑够 N 个有效 item 或取尽为止src/agents/extensions/memory/encrypt_session.py。这意味着最新若干条恰好过期时会回退返回更早的有效历史不会因尾部过期导致 limit 空洞测试见 tests/extensions/memory/test_encrypt_session.py。八、SessionSettings 与检索数量限制EncryptedSession的session_settings属性会透传给底层会话src/agents/extensions/memory/encrypt_session.py因此可以像普通会话一样通过SessionSettings(limitN)控制每次运行前取回的历史条数from agents import Agent, RunConfig, Runner, SessionSettings result await Runner.run( agent, Summarize our recent discussion., sessionsession, run_configRunConfig(session_settingsSessionSettings(limit50)), )取数规则的优先级为显式传入get_items(limit...) 会话默认session_settings.limit 不限制。测试覆盖了默认使用底层 limittests/extensions/memory/test_encrypt_session.py与显式 limit 覆盖默认tests/extensions/memory/test_encrypt_session.py两种场景。另外EncryptedSession的四个历史操作方法get_items/add_items/pop_item/clear_session都支持可选的wrapper: RunContextWrapper参数并按底层会话是否声明wrapper参数来决定是否透传详见 src/agents/memory/session.py可用于租户路由、鉴权等定制存储逻辑。九、完整可运行示例以下是一个可直接运行的完整脚本来源examples/memory/encrypted_session_example.py演示加密会话的全流程import asyncio from typing import cast from agents import Agent, Runner, SQLiteSession from agents.extensions.memory import EncryptedSession from agents.extensions.memory.encrypt_session import EncryptedEnvelope async def main(): # 创建 Agent agent Agent( nameAssistant, instructionsReply very concisely., ) # 创建底层会话这里用 SQLiteSession session_id conversation_123 underlying_session SQLiteSession(session_id) # 用加密会话包装开启自动加密与 TTL session EncryptedSession( session_idsession_id, underlying_sessionunderlying_session, encryption_keymy-secret-encryption-key, ttl3600, # 1 hour TTL for messages ) # 第一轮 result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession, ) print(fAssistant: {result.final_output}) # 第二轮 —— 自动记住上一轮上下文 result await Runner.run(agent, What state is it in?, sessionsession) print(fAssistant: {result.final_output}) # 演示 limit 参数 —— 只取最近 2 条自动解密 latest_items await session.get_items(limit2) for i, msg in enumerate(latest_items, 1): print(f {i}. {msg.get(role)}: {msg.get(content)}) # 验证底层存储确实是密文 raw_items await underlying_session.get_items() for i, item in enumerate(raw_items, 1): if isinstance(item, dict) and item.get(__enc__) 1: enc_item cast(EncryptedEnvelope, item) print(f {i}. Encrypted envelope: __enc__{enc_item[__enc__]}, fpayload length{len(enc_item[payload])}) underlying_session.close() if __name__ __main__: asyncio.run(main())十、API 参考与延伸阅读[EncryptedSession][agents.extensions.memory.encrypt_session.EncryptedSession]加密包装器主类源码位于 src/agents/extensions/memory/encrypt_session.py[Session][agents.memory.session.Session]会话基础协议定义get_items/add_items/pop_item/clear_session四个方法源码位于 src/agents/memory/session.py会话体系总览与选型表含EncryptedSession在全部会话实现中的定位会话指南底层 SQLAlchemy 会话细节SQLAlchemy 会话底层高级 SQLite 会话细节注意加密后内容检索受限高级 SQLite 会话加密会话测试套件tests/extensions/memory/test_encrypt_session.py涵盖基本加解密、Runner 集成、TTL 过期、limit 语义、委托与 SessionSettings 等场景可运行示例examples/memory/encrypted_session_example.py安装依赖声明pyproject.toml。小结EncryptedSession以包装器姿态解决了会话记忆的落库安全问题上层业务零改动底层任意会话实现即可获得 Fernet 加密、HKDF 按会话派生密钥与 TTL 自动过期三大能力。选择建议本地开发可直接用SQLiteSession打底生产环境推荐SQLAlchemySession或 Redis/MongoDB 会话打底同时牢记两点限制——加密与内容检索类高级功能互斥以及过期判断依赖服务器时钟务必配置 NTP 同步。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 openai-agents-python 加密会话（EncryptedSession）实战指南：Fernet 透明加密、HKDF 按会话派生密钥与 TTL 自动过期
openai-agents-python 加密会话EncryptedSession实战指南Fernet 透明加密、HKDF 按会话派生密钥与 TTL 自动过期【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本文是 openai-agents-pythonAgents SDK会话Session记忆体系中EncryptedSession的完整技术指南。EncryptedSession是一个透明的加密包装器可对任意已有的会话实现如SQLiteSession、SQLAlchemySession进行 Fernet 加密并在 TTL 到期后自动跳过过期历史消息。读完本文你将掌握如何安装encrypt扩展、如何用主密钥或 Fernet 密钥包装底层会话、HKDF 按会话派生密钥的底层原理、TTL 自动过期的实现机制以及如何在Runner.run中无缝接入加密会话本指南对应文档docs/sessions/encrypted_session.md。一、EncryptedSession 是什么在多轮对话场景中Agents SDK 通过会话Session自动维护对话历史每次运行前把历史输入前置给模型运行后把新产生的 items 写回存储从而省去手动调用.to_input_list()的麻烦见 会话总览。但历史数据通常以明文落库一旦数据库文件、Redis 或云数据库泄露用户对话内容将直接暴露。EncryptedSession正是为这一问题而生它是一个包装器wrapper本身不负责存储而是把任何符合 [Session][agents.memory.session.Session] 协议的底层会话包一层加密对外暴露完全一致的操作接口。核心特性透明加密Transparent encryption基于 Fernet 对称加密包装任意会话读写路径自动加解密业务代码几乎无感知按会话派生密钥Per-session keys使用 HKDF 密钥派生函数结合主密钥与会话 ID 为每个会话派生唯一密钥自动过期Automatic expirationTTL 过期后旧消息在读取时被静默跳过不会影响会话行为即插即用Drop-in replacement与任意既有会话实现兼容可直接替换Runner.run(..., session...)中的 session 参数。需要说明的是会话机制本身存在一个使用前提在同一 run 中session 不能与 run 级续接选项conversation_id、previous_response_id、auto_previous_response_id混用详见 docs/sessions/index.md这一点对EncryptedSession同样适用。二、安装与依赖加密会话依赖cryptography库属于 SDK 的可选依赖extrapip install openai-agents[encrypt]在 pyproject.toml 中可以看到对应声明[project.optional-dependencies] encrypt [cryptography45.0, 46]即encryptextra 对应cryptography45.0, 46。如果未安装该依赖就导入EncryptedSession会触发agents.extensions.memory._optional_imports中定义的raise_optional_dependency_error提示src/agents/extensions/memory/_optional_imports.py错误信息会明确建议执行上述安装命令。三、快速开始以SQLAlchemySession作为底层存储、包一层EncryptedSession为例完整示例见 examples/memory/encrypted_session_example.pyimport asyncio from agents import Agent, Runner from agents.extensions.memory import EncryptedSession, SQLAlchemySession async def main(): agent Agent(Assistant) # 1. 创建底层会话真实的存储后端 underlying_session SQLAlchemySession.from_url( user-123, urlsqliteaiosqlite:///:memory:, create_tablesTrue ) # 2. 用加密包装 session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keyyour-secret-key-here, ttl600 # 10 minutes ) # 3. 像普通会话一样直接使用 result await Runner.run(agent, Hello, sessionsession) print(result.final_output) if __name__ __main__: asyncio.run(main())整个使用方式与普通会话完全一致传入Runner.run后SDK 在每次运行前自动取出解密后的历史并前置给模型运行后自动把新 items加密后写回底层存储。测试用例 tests/extensions/memory/test_encrypt_session.py 验证了这一点同一加密会话连续两轮运行模型第二轮能记住第一轮的上下文。四、配置详解EncryptedSession构造函数签名为src/agents/extensions/memory/encrypt_session.pyEncryptedSession( session_id: str, underlying_session: SessionABC, encryption_key: str, ttl: int 600, )session_id会话 ID同时充当 HKDF 的 salt见下文密钥派生underlying_session真实存储后端如SQLiteSession、SQLAlchemySession、RedisSession、MongoDBSession等任何符合SessionABC的实现encryption_key主密钥可为 Fernet 密钥或任意原始字符串ttlToken 有效期秒默认 600 秒10 分钟。4.1 加密密钥的两种形式encryption_key支持两种输入源码_ensure_fernet_key_bytes负责判别src/agents/extensions/memory/encrypt_session.pyFernet 密钥base64 编码、解码后为 32 字节的密钥。构造时先尝试base64.urlsafe_b64decode若解码成功且长度为 32 字节则按 Fernet 密钥使用否则回退为原始字符串处理。空字符串会直接抛出ValueError(encryption_key not set; required for EncryptedSession.)。from agents.extensions.memory import EncryptedSession # 形式一Fernet 密钥base64 编码 session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keyyour-fernet-key-here, ttl600 )原始字符串任意密码串内部以 UTF-8 编码后作为 HKDF 的输入材料输入密钥同样可以正常工作# 形式二原始字符串内部会派生为密钥 session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keymy-secret-password, ttl600 )测试用例 tests/extensions/memory/test_encrypt_session.py 专门验证了原始字符串密钥可正常完成加解密往返。生产环境建议用 Fernet 密钥形式可通过cryptography.fernet.Fernet.generate_key()生成并将密钥存放在环境变量或密钥管理服务中切勿硬编码在代码里。4.2 TTL自动过期时长ttl控制每个加密 Token 的有效期过期后读取时被静默跳过# Items expire after 1 hour session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keysecret, ttl3600 # 1 hour in seconds ) # Items expire after 1 day session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keysecret, ttl86400 # 24 hours in seconds )注意 TTL 判定基于应用服务器的系统时钟Fernet 的decrypt(token, ttl...)使用time.time()判断。源码类文档明确提示为避免时钟漂移导致有效 Token 被误判过期请确保环境中所有服务器通过 NTP 同步时钟src/agents/extensions/memory/encrypt_session.py。五、与不同类型会话的组合使用EncryptedSession的底层会话可以是任何会话实现选择取决于你的存储场景5.1 与 SQLite 会话组合本地开发from agents import SQLiteSession from agents.extensions.memory import EncryptedSession # 创建加密的 SQLite 会话文件型持久化 underlying SQLiteSession(user-123, conversations.db) session EncryptedSession( session_iduser-123, underlying_sessionunderlying, encryption_keysecret-key )SQLiteSession支持内存库SQLiteSession(user_123)与文件库两种形态前者进程结束即丢失后者可持久化。5.2 与 SQLAlchemy 会话组合生产级from agents.extensions.memory import EncryptedSession, SQLAlchemySession # 创建加密的 SQLAlchemy 会话任何 SQLAlchemy 支持的数据库 underlying SQLAlchemySession.from_url( user-123, urlpostgresqlasyncpg://user:passlocalhost/db, create_tablesTrue ) session EncryptedSession( session_iduser-123, underlying_sessionunderlying, encryption_keysecret-key )SQLAlchemySession的完整说明见 SQLAlchemy 会话指南若底层会话设置了session_settingsEncryptedSession会将其原样透传给底层见下文第七节。5.3 与高级会话组合时的注意事项!!! warning 高级会话特性当 EncryptedSession 与 AdvancedSQLiteSession 这类带高级查询能力的会话实现组合时需要注意 - find_turns_by_content() 这类**基于内容的方法无法有效工作**因为消息内容已被加密 - 基于内容的搜索本质上是针对密文操作效果极其有限。 也就是说加密会与内容检索/分析类功能互斥。如果业务需要按内容回溯如会话分析、审计应在加密前另行设计检索通道。AdvancedSQLiteSession 的完整能力见 [高级 SQLite 会话指南](https://link.gitcode.com/i/e7c7812ee378684214cccf31568f1e6c)。5.4 透明委托EncryptedSession通过__getattr__将未定义的属性访问委托给底层会话src/agents/extensions/memory/encrypt_session.py因此底层会话的自定义方法如自定义统计、扩展 API依然可以直接调用。测试 tests/extensions/memory/test_encrypt_session.py 验证了自定义同步/异步方法均可经由加密会话正常访问。六、密钥派生机制HKDF 保证每个会话密钥唯一EncryptedSession使用HKDFHMAC-based Key Derivation Function从主密钥为每个会话派生独立加密密钥核心逻辑在_derive_session_fernet_keysrc/agents/extensions/memory/encrypt_session.py参数取值作用算法SHA-256哈希算法输入密钥材料IKM你提供的主密钥Fernet 密钥或原始字符串全局机密不落库Salt会话 IDUTF-8 编码每个会话不同产生不同输出Info 字符串bagents.session-store.hkdf.v1域隔离标识防止跨用途派生输出长度32 字节恰好构成 Fernet 密钥派生出的 32 字节密钥经base64.urlsafe_b64encode后构造 Fernet 实例用于加密该会话的所有 items。这一设计带来的安全性保证每个会话拥有唯一加密密钥即使两个会话使用相同主密钥由于 salt会话 ID不同派生密钥也完全不同没有主密钥无法派生任何会话密钥HKDF 的单向性决定了仅凭密文和会话 ID 无法恢复密钥会话之间无法互相解密即使某个会话密钥泄露也无法解密其他会话的数据。七、存储格式与读写流程7.1 加密信封Envelope写入时_wrap把每条 item 序列化为 JSONPydantic 对象优先model_dump()普通对象取__dict__再用 Fernet 加密封装成固定结构的信封src/agents/extensions/memory/encrypt_session.py{__enc__: 1, v: 1, kid: hkdf-v1, payload: fernet-token}__enc__加密标记恒为 1读取时据此识别信封v版本号当前为 1kid密钥标识当前为hkdf-v1便于未来密钥轮换payloadFernet 密文 Token。_unwrap反向解密非信封 item 原样透传解密失败或密钥错误InvalidToken、KeyError时返回None即静默丢弃src/agents/extensions/memory/encrypt_session.py。测试 tests/extensions/memory/test_encrypt_session.py 验证了落库数据确实为密文底层存储中看到的是__enc__ 1的信封明文内容不可见。示例脚本 examples/memory/encrypted_session_example.py 也演示了如何直接检查底层存储确认加密生效。7.2 自动过期TTL 到期静默跳过过期判定发生在解密阶段_unwrap调用cipher.decrypt(token, ttlself.ttl)Fernet 内部根据 Token 时间戳与ttl判断是否过期过期抛InvalidToken进而被捕获并返回None实现读取时静默跳过# Items older than TTL are silently ignored items await session.get_items() # 只返回未过期 items # Expired items dont affect session behavior result await Runner.run(agent, Continue conversation, sessionsession)三个值得注意的边界行为均有测试佐证过期 item 只跳过、不删除底层存储中数据仍在只是读取时被过滤tests/extensions/memory/test_encrypt_session.pypop_item自动重试从栈顶弹出时若遇到过期 item会继续向下弹出直到找到有效 item 或空src/agents/extensions/memory/encrypt_session.py测试见 tests/extensions/memory/test_encrypt_session.pyget_items(limitN)以解密后的有效数量为准因为过期 item 被跳过EncryptedSession会以翻倍窗口window 初始为 limit不足则 ×2向底层多次取数直到凑够 N 个有效 item 或取尽为止src/agents/extensions/memory/encrypt_session.py。这意味着最新若干条恰好过期时会回退返回更早的有效历史不会因尾部过期导致 limit 空洞测试见 tests/extensions/memory/test_encrypt_session.py。八、SessionSettings 与检索数量限制EncryptedSession的session_settings属性会透传给底层会话src/agents/extensions/memory/encrypt_session.py因此可以像普通会话一样通过SessionSettings(limitN)控制每次运行前取回的历史条数from agents import Agent, RunConfig, Runner, SessionSettings result await Runner.run( agent, Summarize our recent discussion., sessionsession, run_configRunConfig(session_settingsSessionSettings(limit50)), )取数规则的优先级为显式传入get_items(limit...) 会话默认session_settings.limit 不限制。测试覆盖了默认使用底层 limittests/extensions/memory/test_encrypt_session.py与显式 limit 覆盖默认tests/extensions/memory/test_encrypt_session.py两种场景。另外EncryptedSession的四个历史操作方法get_items/add_items/pop_item/clear_session都支持可选的wrapper: RunContextWrapper参数并按底层会话是否声明wrapper参数来决定是否透传详见 src/agents/memory/session.py可用于租户路由、鉴权等定制存储逻辑。九、完整可运行示例以下是一个可直接运行的完整脚本来源examples/memory/encrypted_session_example.py演示加密会话的全流程import asyncio from typing import cast from agents import Agent, Runner, SQLiteSession from agents.extensions.memory import EncryptedSession from agents.extensions.memory.encrypt_session import EncryptedEnvelope async def main(): # 创建 Agent agent Agent( nameAssistant, instructionsReply very concisely., ) # 创建底层会话这里用 SQLiteSession session_id conversation_123 underlying_session SQLiteSession(session_id) # 用加密会话包装开启自动加密与 TTL session EncryptedSession( session_idsession_id, underlying_sessionunderlying_session, encryption_keymy-secret-encryption-key, ttl3600, # 1 hour TTL for messages ) # 第一轮 result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession, ) print(fAssistant: {result.final_output}) # 第二轮 —— 自动记住上一轮上下文 result await Runner.run(agent, What state is it in?, sessionsession) print(fAssistant: {result.final_output}) # 演示 limit 参数 —— 只取最近 2 条自动解密 latest_items await session.get_items(limit2) for i, msg in enumerate(latest_items, 1): print(f {i}. {msg.get(role)}: {msg.get(content)}) # 验证底层存储确实是密文 raw_items await underlying_session.get_items() for i, item in enumerate(raw_items, 1): if isinstance(item, dict) and item.get(__enc__) 1: enc_item cast(EncryptedEnvelope, item) print(f {i}. Encrypted envelope: __enc__{enc_item[__enc__]}, fpayload length{len(enc_item[payload])}) underlying_session.close() if __name__ __main__: asyncio.run(main())十、API 参考与延伸阅读[EncryptedSession][agents.extensions.memory.encrypt_session.EncryptedSession]加密包装器主类源码位于 src/agents/extensions/memory/encrypt_session.py[Session][agents.memory.session.Session]会话基础协议定义get_items/add_items/pop_item/clear_session四个方法源码位于 src/agents/memory/session.py会话体系总览与选型表含EncryptedSession在全部会话实现中的定位会话指南底层 SQLAlchemy 会话细节SQLAlchemy 会话底层高级 SQLite 会话细节注意加密后内容检索受限高级 SQLite 会话加密会话测试套件tests/extensions/memory/test_encrypt_session.py涵盖基本加解密、Runner 集成、TTL 过期、limit 语义、委托与 SessionSettings 等场景可运行示例examples/memory/encrypted_session_example.py安装依赖声明pyproject.toml。小结EncryptedSession以包装器姿态解决了会话记忆的落库安全问题上层业务零改动底层任意会话实现即可获得 Fernet 加密、HKDF 按会话派生密钥与 TTL 自动过期三大能力。选择建议本地开发可直接用SQLiteSession打底生产环境推荐SQLAlchemySession或 Redis/MongoDB 会话打底同时牢记两点限制——加密与内容检索类高级功能互斥以及过期判断依赖服务器时钟务必配置 NTP 同步。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考