openai-agents-python 中的 SQLAlchemySession：用任意 SQLAlchemy 数据库构建生产级 Agent 会话记忆

openai-agents-python 中的 SQLAlchemySession：用任意 SQLAlchemy 数据库构建生产级 Agent 会话记忆

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

SQLAlchemySession是 openai-agents-python（Agents SDK）提供的一个基于 SQLAlchemy 的会话（Session）存储后端，允许你将 Agent 的多轮对话历史持久化到 PostgreSQL、MySQL、SQLite 等任何 SQLAlchemy 支持的数据库中。读完本文，你将掌握它的安装与驱动选型、两种接入方式（数据库 URL 与既有 AsyncEngine）、底层表结构与序列化机制、四个核心会话操作方法的实现原理，以及如何将其无缝接入Runner.run(...)构建生产级的多轮对话应用。

会话记忆是什么，SQLAlchemySession 处在什么位置

在 Agents SDK 中，Session 用于为「特定会话」保存对话历史，使 Agent 无需手动维护.to_input_list()即可在多次运行之间保持上下文。其行为可以概括为三步（参见 会话总览文档）：

1. 运行前：Runner 自动读取该会话的历史记录，并把它拼接到本次输入之前；
2. 运行后：本次运行产生的新条目（用户输入、助手回复、工具调用等）自动写入会话；
3. 上下文保持：后续使用同一会话的每次运行都包含完整历史，Agent 因此能记住之前的交互。

SDK 内置了多种会话实现，各有侧重（详见 docs/sessions/index.md 的内置实现对照表）：

SQLAlchemySession的价值在于「复用你现有的数据库基础设施」：如果你的业务已经在使用 PostgreSQL 或 MySQL，无需再引入新的存储组件，就能获得与SQLiteSession完全一致的会话协议，只是底层数据落到了关系型数据库里。

从源码结构看，SQLAlchemySession继承自SessionABC（抽象基类，见 src/agents/memory/session.py），而Session本身是一个带session_id、session_settings与四个历史操作方法的 Protocol（src/agents/memory/session.py）。这也意味着它可以直接传给Runner.run(agent, input, session=session)，SDK 会自动完成「取历史 → 拼输入 → 存新条目」的完整闭环。

安装与数据库驱动选型

SQLAlchemySession属于可选依赖（optional extra），需要通过sqlalchemy这个 extra 安装，同时还要根据数据库 URL 搭配对应的异步驱动（异步驱动是硬性要求，因为会话方法全部是async的）。

SQLite（sqlite+aiosqlite://）：extra 之外需额外安装aiosqlite：

pip install 'openai-agents[sqlalchemy]' aiosqlite

PostgreSQL（postgresql+asyncpg://）：sqlalchemyextra 已经自带asyncpg，无需额外安装。这一点可以在 pyproject.toml 中得到确认：sqlalchemy = ["SQLAlchemy>=2.0", "asyncpg>=0.29.0"]。

MySQL（mysql+aiomysql://）：需要额外安装aiomysql，其rsa附加依赖用于支持 MySQL 的 SHA-256 认证方式：

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

安装完成后，SQLAlchemySession会通过agents.extensions.memory命名空间导出，可以直接导入（见 src/agents/extensions/memory/init.py，该模块采用惰性导入，仅在使用到时才加载 sqlalchemy 相关依赖）。

快速上手：两种接入方式

方式一：通过数据库 URL（from_url）

from_url是上手最快的入口，你只需给出会话 ID 和数据库连接串，类内部会调用sqlalchemy.ext.asyncio.create_async_engine创建引擎：

import asyncio from agents import Agent, Runner from agents.extensions.memory import SQLAlchemySession async def main(): agent = Agent("Assistant") # 通过数据库 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())

对于 PostgreSQL 只需替换 URL：postgresql+asyncpg://user:pass@host/db。

方式二：复用应用已有的 AsyncEngine

如果你的应用已经管理着一个 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(): # 创建（或复用）数据库引擎 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) # 清理：关闭连接池 await engine.dispose() if __name__ == "__main__": asyncio.run(main())

两种方式完全等价，区别仅在于引擎的所有权：from_url内部创建引擎并由会话持有；直接构造则要求你自行管理引擎生命周期（见下文「生产实践」）。

构造参数与底层表结构

SQLAlchemySession的构造函数定义在 src/agents/extensions/memory/sqlalchemy_session.py，参数如下：

from_url的完整签名为from_url(session_id, *, url, engine_kwargs=None, session_settings=None, **kwargs)（源码位置），其中engine_kwargs会透传给create_async_engine（例如池大小、超时等），其余关键字参数透传给构造函数。

自动建表：两张表 + 一个复合索引

启用create_tables=True时，_ensure_tables()会通过self._metadata.create_all创建以下结构（定义见 源码 L188-L231）：

agent_sessions（会话主表）

agent_messages（消息明细表）

此外还包含复合索引idx_{messages_table}_session_time (session_id, created_at)，用于加速「按会话取历史并按时间排序」的查询。建表过程使用了类级别的线程锁（以引擎 URL + 表名作为键）确保并发创建时只执行一次，且建表完成后会立即置回_create_tables = False防止重复执行。

存储原理：JSON 序列化与 ensure_ascii

每条对话条目（TResponseInputItem）在写入前会被序列化为 JSON 字符串存入message_data列。序列化与反序列化逻辑集中在 src/agents/extensions/memory/sqlalchemy_session.py 的 _serialize_item/_deserialize_item，且这两个方法被设计为可被子类覆写——如果你想换一种存储格式（比如压缩或加密），覆写这两个方法即可。

序列化使用json.dumps(item, ensure_ascii=..., separators=(",", ":"))，其中separators=(",", ":")用于生成紧凑的 JSON（去掉多余空格）。ensure_ascii的语义需要特别说明：

• ensure_ascii=True（默认）：非 ASCII 字符（中文、日文、Emoji 等）会被转义为\uXXXX。这保持了历史上的存储格式，读取时仍能还原出原始文本；
• ensure_ascii=False：多语言文本在数据库中以可读的原始字符保存，方便直接查库调试。

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

使用现有引擎时也可以把同样的参数直接传给SQLAlchemySession(...)。注意：该参数只影响数据库中的 JSON 表示，不会改变会话方法返回的值——无论哪种设置，get_items()读出来的都是还原后的原始文本（官方文档在 docs/sessions/sqlalchemy_session.md 中有明确说明）。

另外，get_items在反序列化时会跳过损坏的行（捕获json.JSONDecodeError后continue），因此个别脏数据不会导致整个会话读取失败。

会话协议四个核心方法深度解析

作为Session协议的实现，SQLAlchemySession提供四个历史操作方法。下面结合 源码实现 逐一说明其行为与并发安全设计。

get_items：读取会话历史

签名：async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]（源码 L301-L368）。

• 不传limit时，使用session_settings.limit（SessionSettings(limit=None)表示返回全部，定义见 src/agents/memory/session_settings.py），按created_at ASC, id ASC升序返回全部历史；
• 传入正数limit=N时，返回时间上最新的 N 条，且保持时间正序。实现上先按created_at DESC, id DESC取尾部 N 条再反转，效率更高；
• 当最新若干条里混有损坏记录时，实现会以「窗口翻倍」的方式扩大拉取范围，确保limit统计的是有效条目数（与 SQLite 后端行为保持一致）；
• 非正数limit保留各数据库方言定义的原始语义，直接透传给 SQL。

add_items：写入新条目

签名：async def add_items(self, items: list[TResponseInputItem]) -> None（源码 L370-L418）。

写入采用「先确保会话行存在，再批量插入消息」的事务流程：

1. 查询agent_sessions中是否已有该session_id；若无，则在嵌套事务中插入会话行——若此时另一个并发写入者已抢先创建了该行，捕获IntegrityError后静默忽略（避免 check-then-insert 竞态）；
2. 将待写入条目整体批量INSERT进agent_messages；
3. 最后刷新agent_sessions.updated_at。

整个写入包在一个事务里，保证原子性。此外，写入/弹出等「变更型」操作会通过_await_mutation（定义于 src/agents/memory/session.py）等待事务真正落定，即使调用方协程被取消也不会让数据库停留在中间状态。

pop_item：弹出最新条目（用于纠错）

签名：async def pop_item(self) -> TResponseInputItem | None（源码 L420-L494）。

该方法返回并删除最新一条条目，常用于「用户想改口」的场景：先弹出助手回复、再弹出用户提问，然后重新发起一问。其并发安全设计是源码中最精妙的部分：

• 支持DELETE ... RETURNING的方言（如 PostgreSQL）：直接以删除语句的 RETURNING 结果作为「认领」依据，只有真正删掉当前尾部的那笔事务才能拿到它的载荷；若认领失败且表中仍有数据，则在新事务中重试；
• 不支持该特性的方言：先用SELECT ... FOR UPDATE锁定尾部行再删除，用事务级行锁而非 DBAPI rowcount 来确立所有权；
• SQLite 特例：SQLite 会忽略SELECT ... FOR UPDATE，因此在查询前先执行BEGIN IMMEDIATE抢占单写者锁，保证回退认领的唯一性。

clear_session：清空会话

签名：async def clear_session(self) -> None（源码 L496-L510）。

在同一事务中先删除该会话的所有消息，再删除会话主行，实现彻底清空。

SQLite 专属的锁容错

由于 SQLite 在并发写入时容易出现database is locked，当底层方言是 SQLite 时，该类会自动做两件事（源码 L100-L144）：

1. 通过引擎connect事件为每个连接执行PRAGMA busy_timeout = 5000与PRAGMA journal_mode = WAL，从连接层面降低瞬时锁失败；
2. 对写入型操作采用有界退避重试（延迟序列0.05, 0.1, 0.2, 0.4, 0.8秒），仅对「database is locked」类错误重试，其他OperationalError立即抛出。

这些配置只针对 SQLite 生效，PostgreSQL/MySQL 不受影响。从源码注释还可以看到，引擎级配置缓存以id(engine.sync_engine)为键并配合weakref.finalize清理，避免引擎被垃圾回收后地址复用导致配置遗漏。

接入 Runner 的完整实战：多轮对话与历史上限控制

把SQLAlchemySession与Runner组合，即可获得开箱即用的多轮记忆。仓库提供了完整可运行示例 examples/memory/sqlalchemy_session_example.py，核心流程如下：

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.", ) # 会话 ID 建议使用有业务含义的命名，如 "user_12345"、"thread_abc123" 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}") # 第二轮：不重复提供历史，Agent 依然记得上下文 result = await Runner.run(agent, "What state is it in?", session=session) print(f"Assistant: {result.final_output}") # 读取历史：只取最新的 2 条 latest_items = await session.get_items(limit=2) for i, msg in enumerate(latest_items, 1): print(f" {i}. {msg.get('role')}: {msg.get('content')}") # 读取全部历史 all_items = await session.get_items() print(f"Total items in session: {len(all_items)}") if __name__ == "__main__": asyncio.run(main())

对超长对话，可以用SessionSettings限制每次运行拉取的历史条数，通过RunConfig.session_settings按运行覆盖（会话的默认设置在构造时传入session_settings=）：

from agents import Agent, RunConfig, Runner, SessionSettings result = await Runner.run( agent, "Summarize our recent discussion.", session=session, run_config=RunConfig(session_settings=SessionSettings(limit=50)), )

另外有两点使用约束值得注意（详见 会话总览文档）：

• 同一运行中，Session 与运行级续接选项互斥：不能同时使用conversation_id、previous_response_id或auto_previous_response_id，二者只能选其一；
• 中断恢复：如果运行因审批（approval）暂停，请使用同一 session 实例（或相同 session ID + 相同存储后端的另一个实例）恢复运行，以便续接同一份存储历史。

生产实践建议

• 建表交给迁移工具：生产环境保持create_tables=False，把两张表（及索引）纳入 Alembic 等迁移流程管理；create_tables=True仅用于开发与测试快速起跑；
• 引擎生命周期：用from_url时引擎由会话内部持有；直接构造时引擎归你所有，应用退出前记得await engine.dispose()。类还暴露了engine只读属性（源码 L512-L523），方便你在高级场景下检查连接池状态或手动释放资源；
• 多进程/多 worker 共享：相比文件型 SQLite，PostgreSQL/MySQL 天然支持多 worker 并发访问同一张会话表，这正是SQLAlchemySession被推荐用于生产系统的原因；
• 加密增强：如果对存储敏感度要求更高，可以用EncryptedSession包装SQLAlchemySession实现透明加密与 TTL 过期（参见 docs/sessions/encrypted_session.md 中的组合示例）；
• 扩展定制：继承SQLAlchemySession并覆写_serialize_item/_deserialize_item，即可在不改变会话协议的前提下自定义存储格式。

参考资料

• 官方专项文档：docs/sessions/sqlalchemy_session.md
• 会话机制总览与内置实现对照：docs/sessions/index.md
• 核心实现：src/agents/extensions/memory/sqlalchemy_session.py
• 会话协议与SessionABC：src/agents/memory/session.py
• SessionSettings定义：src/agents/memory/session_settings.py
• 可运行示例：examples/memory/sqlalchemy_session_example.py
• 依赖声明（sqlalchemyextra）：pyproject.toml

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