企业官网建设流程全解析

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-python（Agents 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.py：Session协议与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( name="Assistant", instructions="Reply 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?", session=session ) print(result.final_output) # "San Francisco" # Second turn - agent automatically remembers previous context result = await Runner.run( agent, "What state is it in?", session=session ) print(result.final_output) # "California" # Also works with synchronous runner result = Runner.run_sync( agent, "What's the population?", session=session ) print(result.final_output) # "Approximately 39 million"

SQLiteSession("conversation_123")不传数据库路径时使用内存数据库（进程结束即丢失）；Runner.run(..., session=session)支持异步运行器，Runner.run_sync(..., session=session)同样支持同步运行器。第二、三轮运行时 Agent 不再需要重复提及“金门大桥”，因为完整历史已自动注入。

工作原理：运行前后的自动挂载与回写

启用会话内存后，Runner 内部（见 session_persistence.py）按以下三步工作：

1. 每次运行前：Runner 自动调用session.get_items()取出该会话的对话历史，并将其拼接到输入项（input items）之前，作为本轮模型调用的上下文；
2. 每次运行后：本轮产生的全部新条目——用户输入、Assistant 回复、工具调用（tool calls）与工具结果——通过session.add_items()自动写入会话；
3. 上下文保持：同一会话的后续运行都包含完整历史，Agent 由此维持跨轮上下文。

这消除了开发者手动调用.to_input_list()维护对话状态的工作。值得一提的是，源码中_call_session_method（session.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也可以不传——此时生效的是session_settings.limit（见 session_settings.py，SessionSettings.limit默认为None表示取全部）。如果你在Session构造时传入session_settings={"limit": 20}，运行前注入的历史将自动截取最近 20 条，避免长会话上下文爆炸。

用 pop_item 修正对话：撤销上一条提问

对话中用户想撤销或更正最后一条消息时，pop_item非常实用——它按“后进先出”顺序逐条弹出，可以精确移除 Agent 回复与用户提问：

from agents import Agent, Runner, SQLiteSession agent = Agent(name="Assistant") session = SQLiteSession("correction_example") # Initial conversation result = await Runner.run( agent, "What's 2 + 2?", session=session ) print(f"Agent: {result.final_output}") # User wants to correct their question assistant_item = await session.pop_item() # Remove agent's response user_item = await session.pop_item() # Remove user's question # Ask a corrected question result = await Runner.run( agent, "What's 2 + 3?", session=session ) print(f"Agent: {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_id="conv_123") result = await Runner.run( agent, "Hello", session=session, )

从源码（openai_conversations_session.py）看，OpenAIConversationsSession是惰性初始化的：不传conversation_id时，首次调用get_items()/add_items()会通过conversations.create(items=[])在服务端创建一个新会话并缓存其 ID；clear_session()会调用conversations.delete()删除远端会话并重置 ID；session_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", session=session )

构造参数（sqlite_session.py）还包括sessions_table（默认agent_sessions）、messages_table（默认agent_messages）与session_settings。实现上它使用两条表：agent_sessions存会话元数据（session_id主键、created_at、updated_at），agent_messages存消息（message_data为 JSON 文本，按session_id外键级联删除并建有(session_id, id)索引）。底层连接启用了WAL 模式（PRAGMA journal_mode=WAL）以提升并发读写能力；文件库场景下同一进程共享同一 SQLite 文件的多个会话实例会复用一把进程级文件锁，内存库则使用共享连接避免线程隔离问题。

多会话隔离

不同会话 ID 维护相互独立的对话历史，适合多用户/多线程场景：

from agents import Agent, Runner, SQLiteSession agent = Agent(name="Assistant") # 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", session=session_1 ) result2 = await Runner.run( agent, "Hello", session=session_2 )

SQLAlchemy 会话：接入 PostgreSQL / MySQL / SQLite

高级场景下可使用 SQLAlchemy 会话后端（sqlalchemy_session.py），从而接入 SQLAlchemy 支持的任何数据库（PostgreSQL、MySQL、SQLite 等）。

例 1：from_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", url="sqlite+aiosqlite:///:memory:", create_tables=True, # Auto-create tables for the demo ) result = await Runner.run(agent, "Hello", session=session) 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("sqlite+aiosqlite:///conversations.db") agent = Agent("Assistant") session = SQLAlchemySession( "user-456", engine=engine, create_tables=True, # Auto-create tables for the demo ) result = await Runner.run(agent, "Hello", session=session) print(result.final_output) await engine.dispose() if __name__ == "__main__": asyncio.run(main())

关键参数说明（来自 sqlalchemy_session.py 的 docstring）：

• session_id：会话唯一标识；
• engine：必须是异步驱动的AsyncEngine，例如postgresql+asyncpg://、mysql+aiomysql://、sqlite+aiosqlite://；
• create_tables：是否自动建表，默认False（生产环境建议用迁移工具管理表结构，开发/测试可设True）；
• sessions_table/messages_table：自定义表名，默认agent_sessions/agent_messages；
• session_settings：会话配置（如默认limit）；
• ensure_ascii：序列化时是否转义非 ASCII 字符，默认True以保持历史存储格式一致。

从源码看，from_url本质是内部调用create_async_engine(url, **engine_kwargs)再走主构造器；表结构包含sessions（session_id主键 +created_at/updated_at）与messages（自增id、session_id外键ON DELETE CASCADE、message_data文本、(session_id, created_at)索引）。针对 SQLite 后端，它还自动设置busy_timeout=5000与 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_id="user-123", url="postgresql+asyncpg://app:secret@db.example.com/agents", create_tables=True, ) # Wrap with encryption and TTL-based expiration session = EncryptedSession( session_id="user-123", underlying_session=underlying_session, encryption_key="your-encryption-key", # Use a secure key from your secrets management ttl=600, # 10 minutes - items older than this are silently skipped ) agent = Agent("Assistant") result = await Runner.run(agent, "Hello", session=session) print(result.final_output) if __name__ == "__main__": asyncio.run(main())

主要特性（对应 encrypt_session.py 的实现）：

• 透明加密：写入前自动加密所有会话条目，读取时自动解密；
• 会话级密钥派生：以会话 ID 为盐，通过 HKDF-SHA256 从主密钥派生每会话唯一密钥（info=b"agents.session-store.hkdf.v1"）；
• TTL 过期：按可配置的存活时长自动过期旧消息（默认 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(name="Assistant") result = await Runner.run( agent, "Hello", session=MyCustomSession("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"

记忆持久化选型建议

清空与跨 Agent 共享

# Clear a session when conversation should start fresh await session.clear_session() # Different agents can share the same session support_agent = Agent(name="Support") billing_agent = Agent(name="Billing") session = SQLiteSession("user_123") # Both agents will see the same conversation history result1 = await Runner.run( support_agent, "Help me with my account", session=session ) result2 = await Runner.run( billing_agent, "What are my charges?", session=session )

同一会话可被多个 Agent 共享——客服与账单 Agent 看到相同的完整历史，这是实现多 Agent 交接（handoff）记忆延续的基础模式。

完整示例：三轮对话展示自动记忆

import asyncio from agents import Agent, Runner, SQLiteSession async def main(): # Create an agent agent = Agent( name="Assistant", instructions="Reply 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?", session=session ) print(f"Assistant: {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?", session=session ) print(f"Assistant: {result.final_output}") print() # Third turn - continuing the conversation print("Third turn:") print("User: What's the population of that state?") result = await Runner.run( agent, "What's the population of that state?", session=session ) print(f"Assistant: {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())

注意本示例使用文件型 SQLite（conversation_history.db），因此进程重启后再次运行，历史依然存在——这正是会话持久化的价值。仓库中的可运行对照版见 examples/memory/sqlite_session_example.py。

更深入的会话主题

本文覆盖了会话记忆的完整主干，若需要进阶主题，仓库中还有专门文档：

• docs/sessions/index.md：会话指南总览；
• docs/sessions/sqlalchemy_session.md：SQLAlchemy 后端进阶（连接池、表管理、迁移）；
• 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扩展协议）；
• SQLiteSession：SQLite 实现（内存库 / 文件库、WAL、表结构、进程内文件锁）；
• OpenAIConversationsSession：OpenAI Conversations API 实现（惰性初始化、远端创建/删除会话）；
• SQLAlchemySession：SQLAlchemy 后端（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