openai-agents-python 实战：用 SQLAlchemySession 将 Agent 会话历史接入生产级数据库

openai-agents-python 实战：用 SQLAlchemySession 将 Agent 会话历史接入生产级数据库

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

SQLAlchemySession是 openai-agents-python（OpenAI Agents SDK）内置的、基于 SQLAlchemy 的生产级会话（Session）记忆实现，它允许你把 Agent 的多轮对话历史持久化到 SQLAlchemy 支持的任何数据库（PostgreSQL、MySQL、SQLite 等）。本文围绕 docs/ko/sessions/sqlalchemy_session.md 的完整内容展开，并结合仓库源码讲解安装步骤、两种初始化方式、构造参数、底层表结构与存储细节，帮助你为已有数据库的线上服务快速接入可用的会话记忆。

会话记忆与 SQLAlchemySession 的定位

Agents SDK 内置了会话记忆机制：你不再需要手动调用.to_input_list()在轮次之间拼接历史，Runner 会在每次运行前自动从 Session 拉取历史输入，运行结束后把本次产生的新条目（用户输入、助手回复、工具调用等）自动写回 Session。这套机制的抽象定义在 src/agents/memory/session.py 的Session协议中，任何实现只需提供四个方法：

• get_items(limit=None)：读取会话历史，limit指定时返回按时间正序的最新 N 条；
• add_items(items)：追加新的会话条目；
• pop_item()：移除并返回最近一条条目；
• clear_session()：清空该会话全部条目。

SQLAlchemySession就是该协议的 SQLAlchemy 落地实现（见 sqlalchemy_session.py 源码）。相比内置的轻量SQLiteSession，它的价值在于复用你已有的数据库与连接池：只要你的应用已经在用 PostgreSQL、MySQL 等关系型数据库，就可以把 Agent 的会话历史与业务数据存放在同一套基础设施中，而不需要额外引入 Redis 之类的独立存储。

安装：sqlalchemy 可选依赖与异步驱动

使用SQLAlchemySession需要安装openai-agents包的sqlalchemy可选依赖（extra）：

pip install openai-agents[sqlalchemy]

从 pyproject.toml 可以看到该 extra 的真实内容：

sqlalchemy = ["SQLAlchemy>=2.0", "asyncpg>=0.29.0"]

也就是说，extra 本身只携带 SQLAlchemy 2.x 以及PostgreSQL的异步驱动asyncpg。SQLAlchemySession内部使用的是异步引擎（sqlalchemy.ext.asyncio），因此还需要安装与你数据库 URL 匹配的异步驱动：

对应安装命令：

pip install 'openai-agents[sqlalchemy]' aiosqlite pip install 'openai-agents[sqlalchemy]' 'aiomysql[rsa]'

快速开始

方式一：通过数据库 URL 创建

最简方式是使用from_url类方法，传入会话 ID 与数据库 URL：

import asyncio from agents import Agent, Runner from agents.extensions.memory import SQLAlchemySession async def main(): agent = Agent("Assistant") # Create session using database URL session = SQLAlchemySession.from_url( "user-123", url="sqlite+aiosqlite:///:memory:", create_tables=True ) result = await Runner.run(agent, "Hello", session=session) print(result.final_output) if __name__ == "__main__": asyncio.run(main())

from_url的源码实现（src/agents/extensions/memory/sqlalchemy_session.py#L241-L268）会先用create_async_engine(url, **engine_kwargs)创建并自持一个异步引擎，再调用主构造函数。它额外接受一个engine_kwargs参数，可以透传给create_async_engine（例如配置连接池大小、超时等）：

session = SQLAlchemySession.from_url( "user-123", url="postgresql+asyncpg://app:secret@db.example.com/agents", create_tables=True, engine_kwargs={"pool_size": 10, "max_overflow": 20}, )

方式二：复用应用中已有的引擎

如果应用已经管理着 SQLAlchemy 异步引擎，直接构造即可，引擎生命周期仍由你掌控：

import asyncio from agents import Agent, Runner from agents.extensions.memory import SQLAlchemySession from sqlalchemy.ext.asyncio import create_async_engine async def main(): # Create your database engine engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") agent = Agent("Assistant") session = SQLAlchemySession( "user-456", engine=engine, create_tables=True ) result = await Runner.run(agent, "Hello", session=session) print(result.final_output) # Clean up await engine.dispose() if __name__ == "__main__": asyncio.run(main())

注意引擎必须是异步驱动创建的（如postgresql+asyncpg://、mysql+aiomysql://、sqlite+aiosqlite://）。由于引擎由外部传入，用完后的await engine.dispose()需要由应用自行负责；源码还通过 engine 属性 暴露底层AsyncEngine，方便你在高级场景下检查连接池状态或手动释放资源。

一个完整的多轮对话示例

仓库中的 examples/memory/sqlalchemy_session_example.py 演示了会话记忆的完整用法：同一个session实例连续传给多轮Runner.run，Agent 能自动记住前文（"What city is the Golden Gate Bridge in?" → "What state is it in?" → "What's the population of that state?"），最后还能用get_items(limit=2)只取最近两条历史：

import asyncio from agents import Agent, Runner from agents.extensions.memory.sqlalchemy_session import SQLAlchemySession async def main(): agent = Agent( name="Assistant", instructions="Reply very concisely.", ) # In-memory SQLite; create_tables=True is useful for development and testing. session = SQLAlchemySession.from_url( "conversation_123", url="sqlite+aiosqlite:///:memory:", create_tables=True, ) result = await Runner.run( agent, "What city is the Golden Gate Bridge in?", session=session, ) print(f"Assistant: {result.final_output}") # Second turn - the agent remembers the previous conversation result = await Runner.run(agent, "What state is it in?", session=session) print(f"Assistant: {result.final_output}") # Only fetch the latest 2 items latest_items = await session.get_items(limit=2) print(f"Fetched {len(latest_items)} latest items") if __name__ == "__main__": asyncio.run(main())

构造参数全解

SQLAlchemySession主构造函数的完整签名位于 src/agents/extensions/memory/sqlalchemy_session.py#L146-L156：

from_url相比主构造函数额外接受url（任意 SQLAlchemy 异步 URL）与engine_kwargs（透传给create_async_engine），其余关键字参数原样转发给主构造函数。

session_settings接受 SessionSettings 对象或等价字典；SessionSettings(limit=N)会在不显式传limit时，让get_items默认只取最新 N 条历史，适合超长对话场景下控制每轮的上下文规模。Runner.run(..., run_config=RunConfig(session_settings=SessionSettings(limit=50)))可以按轮次覆盖。

底层存储结构：两张表

从源码 src/agents/extensions/memory/sqlalchemy_session.py#L188-L231 可以看到，SQLAlchemySession使用两张表（表名均可通过构造参数覆盖）：

会话表agent_sessions：

• session_id（String，主键）；
• created_at（TIMESTAMP，服务端默认CURRENT_TIMESTAMP）；
• updated_at（TIMESTAMP，服务端默认CURRENT_TIMESTAMP，且配置了onupdate，每次写入消息时会自动刷新，见 add_items 实现）。

消息表agent_messages：

• id（Integer，自增主键，SQLite 下额外启用sqlite_autoincrement）；
• session_id（String，外键指向agent_sessions.session_id，ondelete="CASCADE"）；
• message_data（Text，存放序列化后的 JSON 条目）；
• created_at（TIMESTAMP，服务端默认CURRENT_TIMESTAMP）；
• 复合索引idx_{messages_table}_session_time(session_id, created_at)，用于加速按会话按时间取历史。

消息以 JSON 文本形式按行存储。在写入路径上，add_items先检查会话行是否存在，不存在时通过嵌套事务插入父行并捕获IntegrityError，从而在并发首写场景下避免 check-then-insert 竞态；随后批量insert消息并刷新updated_at（src/agents/extensions/memory/sqlalchemy_session.py#L388-L418）。

Session 协议方法的行为细节

• get_items(limit=None)（源码 L301-L368）：limit为None时按时间正序返回全部历史；limit=N时先用DESC + LIMIT取最新 N 条再反转成正序。读取时遇到 JSON 损坏的行会跳过而不是抛错；当最新几条中存在损坏行时，会动态扩大读取窗口，保证limit统计的是有效条目数。
• add_items(items)：空列表直接返回；先确保会话父行存在（处理并发竞态），再批量插入消息并刷新updated_at。
• pop_item()（源码 L420-L494）：删除并返回最近一条。实现上优先使用DELETE ... RETURNING作为“认领”手段以避免依赖 DBAPI rowcount；对不支持DELETE ... RETURNING的方言（以及部分 SQLite 场景）则退化为SELECT ... FOR UPDATE加锁删除，SQLite 下还会先执行BEGIN IMMEDIATE独占写锁。该方法是实现“撤回/修正最后一条消息”的关键操作：pop掉助手回复和用户提问后，重新以修正后的问题发起新一轮运行。
• `clear_session()``：删除该会话的消息行与会话行（外键级联删除）。

这些写操作都经由_await_mutation包装，即使调用方协程被取消，底层写事务也会等其落定，避免半途终止导致的状态不一致。

存储非 ASCII 文本：ensure_ascii

默认情况下，SQLAlchemySession在把会话条目序列化为 JSON 时会对非 ASCII 字符做转义（ensure_ascii=True），这保留了历史存储格式，同时在读取时仍能无损还原原始文本。如果你希望落库的 JSON 中多语言文本保持可读，设置ensure_ascii=False：

session = SQLAlchemySession.from_url( "user-123", url="sqlite+aiosqlite:///conversations.db", create_tables=True, ensure_ascii=False, )

使用已有引擎时，把同样的选项直接传给SQLAlchemySession(...)即可。需要明确的是：该设置只改变数据库中存储的 JSON 表示（对应源码 serialize/deserialize 钩子），不改变get_items等会话方法返回给调用方的值——无论ensure_ascii取值如何，还原出的都是原始文本。

生产环境落地要点

1. create_tables默认False：自动建表只适合开发与测试（源码中建表动作仅执行一次，见 L281-L299）。生产环境建议用 Alembic 等迁移工具管理agent_sessions/agent_messages两张表，再以create_tables=False接入。
2. SQLite 专属优化：源码对 SQLite 引擎做了特殊配置（L100-L144）：连接时执行PRAGMA busy_timeout = 5000与PRAGMA journal_mode = WAL减少瞬时锁失败；写操作遇到 "database is locked" 时按(0.05, 0.1, 0.2, 0.4, 0.8)秒的有界退避重试（L129-L144）。这些行为对用户透明，PostgreSQL / MySQL 不受影响。
3. 引擎生命周期：from_url创建的引擎由 Session 自持，记得在进程退出前await engine.dispose()；复用已有引擎时，释放责任在应用侧。可通过session.engine访问底层引擎做连接池检查等高级操作。
4. 会话续跑：如果一轮运行因审批（interruption）暂停，请用同一个 session 实例（或同 session ID + 同存储后端的另一实例）继续运行，确保续跑回合沿用同一份历史。
5. 会话隔离与共享：不同session_id各自维护独立历史（适合按用户、按线程或按工单维度组织）；同一个 session 也可以跨多个 Agent 共享，让不同 Agent 看到一致的对话上下文。更多会话模式可参考 docs/ko/sessions/index.md 与 docs/sessions/index.md。

与其他内置会话实现的取舍

Agents SDK 提供了多种会话后端（详见 docs/sessions/index.md 中的对比表）：本地开发可用轻量的SQLiteSession/AsyncSQLiteSession；多进程共享、低延迟场景可选RedisSession；已有 MongoDB 的应用可选MongoDBSession。而SQLAlchemySession的定位非常明确——面向已有关系型数据库的生产应用：复用现有 PostgreSQL / MySQL / SQLite 基础设施与运维体系，与业务数据同库管理，且天然支持跨进程共享（由数据库自身保证一致性）。

API 参考

• SQLAlchemySession—— 主类：SQLAlchemy 驱动的会话实现，从agents.extensions.memory导入；
• Session—— 基础会话协议：get_items/add_items/pop_item/clear_session；
• 配套文档：英文版 SQLAlchemy sessions 文档、会话总览（韩文）；
• 完整示例：examples/memory/sqlalchemy_session_example.py；
• 依赖声明：pyproject.toml 中的 sqlalchemy extra。

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python