openai-agents-python 会话内存Sessions完全指南从 SQLite 到加密会话的多轮对话持久化【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python会话Session是 openai-agents-pythonAgents SDK内置的会话记忆机制它跨多次 Agent 运行自动维护对话历史让 Agent 无需手动调用.to_input_list()就能记住上下文。本文以官方文档 docs/ja/sessions.md 为主体结合仓库源码src/agents/memory/与src/agents/extensions/memory/与示例examples/memory/系统讲解会话内存的工作原理、SQLite / OpenAI Conversations / SQLAlchemy / 加密会话四种后端选型、pop_item纠错技巧以及如何基于SessionABC编写自定义会话实现。读完本文你将能直接为聊天应用、多轮工具调用与多 Agent 协作场景落地可靠的会话持久化方案。会话内存是什么为什么需要它在 Agents SDK 中一次Runner.run()默认是无状态的Agent 只看到本轮输入。要让 Agent 记住“上一轮说了什么”开发者需要手动拼接历史例如反复调用to_input_list()维护input列表。会话内存Session Memory解决了这一问题会话为特定会话 ID 保存对话历史使 Agent 无需显式手动内存管理即可保持上下文。这对希望 Agent 记住历史交互的聊天应用和多轮对话场景尤其有用。会话的抽象定义位于 session.pySession协议与SessionABC抽象基类均要求实现四个核心方法——get_items()、add_items()、pop_item()、clear_session()。所有后端SQLite、OpenAI Conversations、SQLAlchemy、加密会话都是这套接口的具体实现Runner 只与协议交互因此后端可以按需替换。快速开始三行代码让 Agent 记住上下文from agents import Agent, Runner, SQLiteSession # Create agent agent Agent( nameAssistant, instructionsReply very concisely., ) # Create a session instance with a session ID session SQLiteSession(conversation_123) # First turn result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession ) print(result.final_output) # San Francisco # Second turn - agent automatically remembers previous context result await Runner.run( agent, What state is it in?, sessionsession ) print(result.final_output) # California # Also works with synchronous runner result Runner.run_sync( agent, Whats the population?, sessionsession ) print(result.final_output) # Approximately 39 millionSQLiteSession(conversation_123)不传数据库路径时使用内存数据库进程结束即丢失Runner.run(..., sessionsession)支持异步运行器Runner.run_sync(..., sessionsession)同样支持同步运行器。第二、三轮运行时 Agent 不再需要重复提及“金门大桥”因为完整历史已自动注入。工作原理运行前后的自动挂载与回写启用会话内存后Runner 内部见 session_persistence.py按以下三步工作每次运行前Runner 自动调用session.get_items()取出该会话的对话历史并将其拼接到输入项input items之前作为本轮模型调用的上下文每次运行后本轮产生的全部新条目——用户输入、Assistant 回复、工具调用tool calls与工具结果——通过session.add_items()自动写入会话上下文保持同一会话的后续运行都包含完整历史Agent 由此维持跨轮上下文。这消除了开发者手动调用.to_input_list()维护对话状态的工作。值得一提的是源码中_call_session_methodsession.py允许自定义会话通过可选wrapper参数获取RunContextWrapper从而在读取/写入时感知当前运行上下文——这是扩展会话能力如按用户维度做权限过滤的切入点。会话基本操作读取、追加、弹出、清空会话提供四类基本操作对应SessionABC的四个抽象方法from agents import SQLiteSession session SQLiteSession(user_123, conversations.db) # Get all items in a session items await session.get_items() # Add new items to a session new_items [ {role: user, content: Hello}, {role: assistant, content: Hi there!} ] await session.add_items(new_items) # Remove and return the most recent item last_item await session.pop_item() print(last_item) # {role: assistant, content: Hi there!} # Clear all items from a session await session.clear_session()各方法的语义来自 session.py 协议定义方法签名行为get_items(limit: int \| None None) - list[TResponseInputItem]按时间正序返回历史limit指定时返回最近 N 条并按时间正序排列add_items(items: list[TResponseInputItem]) - None追加一批条目到历史末尾pop_item() - TResponseInputItem \| None移除并返回最近一条会话为空返回Noneclear_session() - None清空该会话全部条目其中get_items的limit也可以不传——此时生效的是session_settings.limit见 session_settings.pySessionSettings.limit默认为None表示取全部。如果你在Session构造时传入session_settings{limit: 20}运行前注入的历史将自动截取最近 20 条避免长会话上下文爆炸。用 pop_item 修正对话撤销上一条提问对话中用户想撤销或更正最后一条消息时pop_item非常实用——它按“后进先出”顺序逐条弹出可以精确移除 Agent 回复与用户提问from agents import Agent, Runner, SQLiteSession agent Agent(nameAssistant) session SQLiteSession(correction_example) # Initial conversation result await Runner.run( agent, Whats 2 2?, sessionsession ) print(fAgent: {result.final_output}) # User wants to correct their question assistant_item await session.pop_item() # Remove agents response user_item await session.pop_item() # Remove users question # Ask a corrected question result await Runner.run( agent, Whats 2 3?, sessionsession ) print(fAgent: {result.final_output})注意pop_item弹出顺序与入队顺序相反先弹掉 Agent 回复再弹掉用户问题历史就回到了提问之前的状态。从实现看SQLite 后端用DELETE ... RETURNING原子地删除并返回最新一条sqlite_session.py并会跳过损坏的 JSON 条目继续向下寻找有效条目。内存选项从默认无记忆到多后端持久化无记忆默认不传session参数即为默认行为每次运行互不感知# Default behavior - no session memory result await Runner.run(agent, Hello)OpenAI Conversations API 记忆云端托管如果不想自建数据库可以让 OpenAI 托管会话状态Conversations API。当你的应用已经依赖 OpenAI 托管的存储时这是最省事的选择from agents import OpenAIConversationsSession session OpenAIConversationsSession() # Optionally resume a previous conversation by passing a conversation ID # session OpenAIConversationsSession(conversation_idconv_123) result await Runner.run( agent, Hello, sessionsession, )从源码openai_conversations_session.py看OpenAIConversationsSession是惰性初始化的不传conversation_id时首次调用get_items()/add_items()会通过conversations.create(items[])在服务端创建一个新会话并缓存其 IDclear_session()会调用conversations.delete()删除远端会话并重置 IDsession_id属性在未初始化前访问会抛出ValueError。它同样可以传入自定义AsyncOpenAI客户端openai_client参数与session_settings。SQLite 内存本地文件SQLite 是零依赖的本地方案支持内存库与文件库两种形态from agents import SQLiteSession # In-memory database (lost when process ends) session SQLiteSession(user_123) # Persistent file-based database session SQLiteSession(user_123, conversations.db) # Use the session result await Runner.run( agent, Hello, sessionsession )构造参数sqlite_session.py还包括sessions_table默认agent_sessions、messages_table默认agent_messages与session_settings。实现上它使用两条表agent_sessions存会话元数据session_id主键、created_at、updated_atagent_messages存消息message_data为 JSON 文本按session_id外键级联删除并建有(session_id, id)索引。底层连接启用了WAL 模式PRAGMA journal_modeWAL以提升并发读写能力文件库场景下同一进程共享同一 SQLite 文件的多个会话实例会复用一把进程级文件锁内存库则使用共享连接避免线程隔离问题。多会话隔离不同会话 ID 维护相互独立的对话历史适合多用户/多线程场景from agents import Agent, Runner, SQLiteSession agent Agent(nameAssistant) # Different sessions maintain separate conversation histories session_1 SQLiteSession(user_123, conversations.db) session_2 SQLiteSession(user_456, conversations.db) result1 await Runner.run( agent, Hello, sessionsession_1 ) result2 await Runner.run( agent, Hello, sessionsession_2 )SQLAlchemy 会话接入 PostgreSQL / MySQL / SQLite高级场景下可使用 SQLAlchemy 会话后端sqlalchemy_session.py从而接入 SQLAlchemy 支持的任何数据库PostgreSQL、MySQL、SQLite 等。例 1from_url创建内存 SQLite开发/测试最简方式import asyncio from agents import Agent, Runner from agents.extensions.memory.sqlalchemy_session import SQLAlchemySession async def main(): agent Agent(Assistant) session SQLAlchemySession.from_url( user-123, urlsqliteaiosqlite:///:memory:, create_tablesTrue, # Auto-create tables for the demo ) result await Runner.run(agent, Hello, sessionsession) if __name__ __main__: asyncio.run(main())例 2复用现有 SQLAlchemy 引擎生产推荐import asyncio from agents import Agent, Runner from agents.extensions.memory.sqlalchemy_session import SQLAlchemySession from sqlalchemy.ext.asyncio import create_async_engine async def main(): # In your application, you would use your existing engine engine create_async_engine(sqliteaiosqlite:///conversations.db) agent Agent(Assistant) session SQLAlchemySession( user-456, engineengine, create_tablesTrue, # Auto-create tables for the demo ) result await Runner.run(agent, Hello, sessionsession) print(result.final_output) await engine.dispose() if __name__ __main__: asyncio.run(main())关键参数说明来自 sqlalchemy_session.py 的 docstringsession_id会话唯一标识engine必须是异步驱动的AsyncEngine例如postgresqlasyncpg://、mysqlaiomysql://、sqliteaiosqlite://create_tables是否自动建表默认False生产环境建议用迁移工具管理表结构开发/测试可设Truesessions_table/messages_table自定义表名默认agent_sessions/agent_messagessession_settings会话配置如默认limitensure_ascii序列化时是否转义非 ASCII 字符默认True以保持历史存储格式一致。从源码看from_url本质是内部调用create_async_engine(url, **engine_kwargs)再走主构造器表结构包含sessionssession_id主键 created_at/updated_at与messages自增id、session_id外键ON DELETE CASCADE、message_data文本、(session_id, created_at)索引。针对 SQLite 后端它还自动设置busy_timeout5000与 WAL 模式并对写操作遭遇database is locked时做带退避的指数重试0.05s → 0.1s → 0.2s → 0.4s → 0.8s降低多写者竞争下的瞬态锁失败率。加密会话透明加密 TTL 自动过期需要对落盘会话数据加密的应用可用EncryptedSession包装任意会话后端提供透明加密与基于 TTL 的自动过期。它需要encrypt可选依赖pip install openai-agents[encrypt]EncryptedSession使用带会话级密钥派生HKDF的 Fernet 加密并支持旧消息自动过期——条目超过 TTL 后读取时被静默跳过。例加密 SQLAlchemy 会话数据import asyncio from agents import Agent, Runner from agents.extensions.memory import EncryptedSession, SQLAlchemySession async def main(): # Create underlying session (works with any SessionABC implementation) underlying_session SQLAlchemySession.from_url( session_iduser-123, urlpostgresqlasyncpg://app:secretdb.example.com/agents, create_tablesTrue, ) # Wrap with encryption and TTL-based expiration session EncryptedSession( session_iduser-123, underlying_sessionunderlying_session, encryption_keyyour-encryption-key, # Use a secure key from your secrets management ttl600, # 10 minutes - items older than this are silently skipped ) agent Agent(Assistant) result await Runner.run(agent, Hello, sessionsession) print(result.final_output) if __name__ __main__: asyncio.run(main())主要特性对应 encrypt_session.py 的实现透明加密写入前自动加密所有会话条目读取时自动解密会话级密钥派生以会话 ID 为盐通过 HKDF-SHA256 从主密钥派生每会话唯一密钥infobagents.session-store.hkdf.v1TTL 过期按可配置的存活时长自动过期旧消息默认 10 分钟灵活的密钥输入加密密钥既可以是 Fernet 密钥urlsafe-base64 解码后 32 字节也可以是任意原始字符串——_ensure_fernet_key_bytes会自动识别可包装任意会话适用于 SQLite、SQLAlchemy 或任何自定义会话实现。⚠️ 重要的安全注意事项加密密钥必须妥善保管例如放在环境变量或密钥管理服务中源码中空密钥会直接抛出ValueError(encryption_key not set; required for EncryptedSession.)过期令牌的拒绝基于应用服务器的系统时钟——请确保所有服务器通过 NTP 同步时间避免合法令牌因时钟偏移被误拒底层会话存储的仍是加密后的数据因此数据库基础设施的管理权限仍保留在你的手中。从实现细节看EncryptedSession通过__getattr__透传底层会话属性并覆写四个协议方法完成“加密信封__enc__、v、kid、payload”的写入与解析解密失败InvalidToken或已过 TTL 的条目在读取时返回None被静默跳过因此过期条目不会污染历史。进阶用法可参考 docs/sessions/encrypted_session.md 与示例 examples/memory/encrypted_session_example.py。自定义会话实现实现 SessionABC 接入自己的存储想接入 Redis、Django 或其他自研存储时只需实现SessionABC或结构上满足Session协议的四个方法from agents.memory.session import SessionABC from agents.items import TResponseInputItem from typing import List class MyCustomSession(SessionABC): Custom session implementation following the Session protocol. def __init__(self, session_id: str): self.session_id session_id # Your initialization here async def get_items(self, limit: int | None None) - List[TResponseInputItem]: Retrieve conversation history for this session. # Your implementation here pass async def add_items(self, items: List[TResponseInputItem]) - None: Store new items for this session. # Your implementation here pass async def pop_item(self) - TResponseInputItem | None: Remove and return the most recent item from this session. # Your implementation here pass async def clear_session(self) - None: Clear all items for this session. # Your implementation here pass # Use your custom session agent Agent(nameAssistant) result await Runner.run( agent, Hello, sessionMyCustomSession(my_session) )从 session.py 可以看到两条接口路径Session是runtime_checkable的Protocol结构类型任何实现了四个方法、带session_id: str与可选session_settings的类都能被接受适合第三方库SessionABC是抽象基类供 SDK 内部与具体实现继承使用docstring 明确建议第三方实现Session协议而非继承 ABC。可选的扩展点实现OpenAIResponsesCompactionAwareSession协议含run_compaction()方法支持previous_response_id/input/auto三种压缩模式可以让会话参与 OpenAI Responses 的上下文压缩四个方法若接受名为wrapper的关键字参数则可在调用时收到RunContextWrapper见_session_method_accepts_wrapper的检测逻辑。仓库中还提供了RedisSession、MongoDBSession、DaprSession等参考实现src/agents/extensions/memory/可作为自定义后端的范本。会话管理最佳实践会话 ID 命名用可读、有业务含义的会话 ID 组织对话按用户user_12345按线程thread_abc123按上下文support_ticket_456记忆持久化选型建议场景推荐方案临时会话进程内内存 SQLiteSQLiteSession(session_id)需持久化的会话文件型 SQLiteSQLiteSession(session_id, path/to/db.sqlite)已有数据库的生产系统SQLAlchemy 会话SQLAlchemySession(session_id, engineengine, create_tablesTrue)想让 OpenAI 托管历史OpenAIConversationsSession()需要加密 TTL 过期EncryptedSession(session_id, underlying_session, encryption_key)更进阶的需求为 Redis、Django 等生产系统实现自定义会话后端清空与跨 Agent 共享# Clear a session when conversation should start fresh await session.clear_session() # Different agents can share the same session support_agent Agent(nameSupport) billing_agent Agent(nameBilling) session SQLiteSession(user_123) # Both agents will see the same conversation history result1 await Runner.run( support_agent, Help me with my account, sessionsession ) result2 await Runner.run( billing_agent, What are my charges?, sessionsession )同一会话可被多个 Agent 共享——客服与账单 Agent 看到相同的完整历史这是实现多 Agent 交接handoff记忆延续的基础模式。完整示例三轮对话展示自动记忆import asyncio from agents import Agent, Runner, SQLiteSession async def main(): # Create an agent agent Agent( nameAssistant, instructionsReply very concisely., ) # Create a session instance that will persist across runs session SQLiteSession(conversation_123, conversation_history.db) print( Sessions Example ) print(The agent will remember previous messages automatically.\n) # First turn print(First turn:) print(User: What city is the Golden Gate Bridge in?) result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession ) print(fAssistant: {result.final_output}) print() # Second turn - the agent will remember the previous conversation print(Second turn:) print(User: What state is it in?) result await Runner.run( agent, What state is it in?, sessionsession ) print(fAssistant: {result.final_output}) print() # Third turn - continuing the conversation print(Third turn:) print(User: Whats the population of that state?) result await Runner.run( agent, Whats the population of that state?, sessionsession ) print(fAssistant: {result.final_output}) print() print( Conversation Complete ) print(Notice how the agent remembered the context from previous turns!) print(Sessions automatically handles conversation history.) if __name__ __main__: asyncio.run(main())注意本示例使用文件型 SQLiteconversation_history.db因此进程重启后再次运行历史依然存在——这正是会话持久化的价值。仓库中的可运行对照版见 examples/memory/sqlite_session_example.py。更深入的会话主题本文覆盖了会话记忆的完整主干若需要进阶主题仓库中还有专门文档docs/sessions/index.md会话指南总览docs/sessions/sqlalchemy_session.mdSQLAlchemy 后端进阶连接池、表管理、迁移docs/sessions/advanced_sqlite_session.md高级 SQLite 会话压缩、TTL 等docs/sessions/encrypted_session.md加密会话深入密钥轮换、TTL 语义docs/ja/sessions.md本文对应的日文原文参考实现与测试src/agents/extensions/memory/、tests/memory/test_session.py、tests/memory/test_openai_conversations_session.py。API 参考Session协议 /SessionABC会话接口定义含四个核心方法、session_settings、可选的run_compaction扩展协议SQLiteSessionSQLite 实现内存库 / 文件库、WAL、表结构、进程内文件锁OpenAIConversationsSessionOpenAI Conversations API 实现惰性初始化、远端创建/删除会话SQLAlchemySessionSQLAlchemy 后端from_url便捷构造、异步引擎、自动建表、SQLite 锁重试EncryptedSession带 TTL 的加密会话包装器HKDF 派生、Fernet 加密、过期静默跳过SessionSettings会话配置limit默认读取条数。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考