基于 SQLAlchemy 的生产级会话存储openai-agents-python SQLAlchemySession 实战指南【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonSQLAlchemySession是 openai-agents-python 提供的生产级会话Session存储实现它把 Agent 的对话历史持久化到 SQLAlchemy 支持的任何数据库PostgreSQL、MySQL、SQLite 等让多轮对话上下文不再依赖内存。本文以 docs/zh/sessions/sqlalchemy_session.md 为主线结合仓库源码与测试完整讲解安装、两种初始化方式、表结构与并发细节帮你把 Agent 会话接入现有数据库体系。一、SQLAlchemySession 是什么在 openai-agents-python 中Session负责存储某个会话以session_id标识的对话历史使 Agent 无需手动管理记忆即可保持上下文。仓库通过 src/agents/memory/session.py 定义了Session协议要求实现四个异步方法get_items(limitNone)取回会话历史limit指定时返回最新 N 条按时间正序add_items(items)向历史追加条目pop_item()弹出并返回最近一条空会话返回Noneclear_session()清空该会话全部条目。SQLAlchemySession正是该协议的 SQLAlchemy 实现见 src/agents/extensions/memory/sqlalchemy_session.py它属于agents.extensions.memory扩展命名空间下生产级、引入第三方数据库驱动依赖的会话后端之一可像替换SQLiteSession一样直接使用。官方文档对会话后端的选型建议是生产系统且已有 SQLAlchemy 支持的数据库时优先使用SQLAlchemySession见 docs/sessions/index.md。二、安装与数据库驱动SQLAlchemySession需要安装openai-agents的sqlalchemy可选依赖 extrapip install openai-agents[sqlalchemy]从 pyproject.toml 可以看到该 extra 的实际内容sqlalchemy [SQLAlchemy2.0, asyncpg0.29.0]即安装时已经自带asyncpg驱动可直接用于postgresqlasyncpg://开头的 PostgreSQL 连接串。对于文档示例中使用的 SQLitesqliteaiosqlite://还需要额外安装aiosqlitepip install openai-agents[sqlalchemy] aiosqlite对于mysqlaiomysql://开头的 MySQL 连接串则需安装aiomysql其rsa子 extra 提供 MySQL SHA-256 认证方式所需的依赖pip install openai-agents[sqlalchemy] aiomysql[rsa]值得注意的是SQLAlchemySession在 src/agents/extensions/memory/init.py 中通过懒加载导出只有真正访问SQLAlchemySession时才会导入sqlalchemy模块未安装依赖时会抛出带有 extra 名称的清晰错误提示因此不必担心它拖慢agents包的导入速度。三、快速入门两种初始化方式3.1 通过数据库 URL 创建from_url最简方式是用SQLAlchemySession.from_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, urlsqliteaiosqlite:///:memory:, create_tablesTrue ) result await Runner.run(agent, Hello, sessionsession) print(result.final_output) if __name__ __main__: asyncio.run(main())from_url是类方法实现见 sqlalchemy_session.py其签名参数如下参数说明session_id会话唯一标识如用户 ID对应一张会话主键url任意 SQLAlchemy 异步连接串如postgresqlasyncpg://user:passhost/dbengine_kwargs额外关键字参数透传给sqlalchemy.ext.asyncio.create_async_engine如连接池、超时等配置session_settings会话配置如默认拉取条数上限见下文**kwargs透传给主构造函数例如create_tables、自定义表名等在源码层面from_url等价于先执行create_async_engine(url, **engine_kwargs)再调用主构造函数因此它返回的会话内部持有一个由该 URL 独立创建的引擎。3.2 复用现有引擎SQLAlchemySession(...)对于已经管理了 SQLAlchemy 引擎的应用程序直接传入AsyncEngineimport 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(postgresqlasyncpg://user:passlocalhost/db) agent Agent(Assistant) session SQLAlchemySession( user-456, engineengine, create_tablesTrue ) result await Runner.run(agent, Hello, sessionsession) print(result.final_output) # Clean up await engine.dispose() if __name__ __main__: asyncio.run(main())这里有两个关键约束引擎必须是异步驱动创建的AsyncEngine文档明确要求postgresqlasyncpg://、mysqlaiomysql://或sqliteaiosqlite://之一引擎生命周期由调用方管理SQLAlchemySession不会替你关闭引擎示例在结束后显式调用await engine.dispose()释放连接池。若需在运行期访问底层引擎例如查看连接池状态、手动 dispose可使用session.engine属性见 sqlalchemy_session.py。四、会话的底层数据模型与表结构SQLAlchemySession在初始化时sqlalchemy_session.py会构建两张表agent_sessions会话表可用sessions_table覆盖表名列类型说明session_idString主键created_atTIMESTAMP服务端默认CURRENT_TIMESTAMP非空updated_atTIMESTAMP服务端默认CURRENT_TIMESTAMP写入时自动更新agent_messages消息表可用messages_table覆盖表名列类型说明idInteger自增主键sqlite_autoincrementTruesession_idString外键指向会话表并ondeleteCASCADE删会话即级联删消息message_dataText条目序列化后的 JSON 文本created_atTIMESTAMP服务端默认CURRENT_TIMESTAMP非空此外还建立了(session_id, created_at)复合索引idx_agent_messages_session_time用于加速按会话、按时间的查询与排序。主构造函数完整签名sqlalchemy_session.pySQLAlchemySession( session_id: str, *, engine: AsyncEngine, create_tables: bool False, sessions_table: str agent_sessions, messages_table: str agent_messages, session_settings: SessionSettings | dict[str, Any] | None None, ensure_ascii: bool True, )关于create_tables源码文档特别说明默认False用于生产环境配合数据库迁移工具管理表结构开发测试时设为True让会话自动建表。表创建由_ensure_tables()通过metadata.create_all完成并保证只执行一次见 sqlalchemy_session.py。会话如何接入多轮对话把session传给Runner.run(agent, input, sessionsession)后Runner 会在每轮运行中自动调用add_items追加对话、get_items读取历史从而让 Agent 在多轮对话中保持上下文pop_item与clear_session则用于回滚最近一步或重置会话。仓库测试 tests/extensions/memory/test_sqlalchemy_session.py 用内存 SQLite 完整验证了追加 → 读取 → 弹出 → 清空这条链路。拉取条数限制limit 与 session_settingsget_items(limit)的语义是返回最新 N 条、且按时间正序底层先按created_at DESC, id DESC取最近 N 条再反转见 sqlalchemy_session.py。未传limit时使用SessionSettings.limit定义于 src/agents/memory/session_settings.py两者都未设置则返回全部历史。测试中还覆盖了最新几条中存在损坏 JSON 行时自动扩大拉取窗口、保证返回的有效条目数满足limit的边界行为。五、多语言文本存储ensure_ascii详解默认情况下SQLAlchemySession在把会话条目序列化为 JSON 时会转义非 ASCII 字符ensure_asciiTrue。这样做的意义是保留历史存储格式同时在加载条目时仍能无损还原原始文本即 round-trip 无损。如果希望中文等多语言文本在数据库存储的 JSON 中保持可读设置ensure_asciiFalsesession SQLAlchemySession.from_url( user-123, urlsqliteaiosqlite:///conversations.db, create_tablesTrue, ensure_asciiFalse, )使用现有引擎时同样的选项直接传给SQLAlchemySession(...)即可session SQLAlchemySession( user-456, engineengine, create_tablesTrue, ensure_asciiFalse, )文档强调此设置只改变数据库中存储的 JSON 表示形式不会改变会话方法返回的值——get_items()读取后仍还原为原始的 Python 对象/文本。这一行为的实现位于序列化钩子 sqlalchemy_session.py其中_serialize_item用json.dumps(item, ensure_asciiself._ensure_ascii, separators(,, :))紧凑序列化且两个方法均可被子类覆写。仓库测试 test_sqlalchemy_session.py 专门验证了默认转义存储格式与读取还原的往返一致性。六、生产级细节并发安全与 SQLite 特殊处理SQLAlchemySession之所以被称为生产可用在于它对并发和方言差异做了大量处理这些细节在源码中清晰可见1. 并发建表不冲突_get_table_init_lock以引擎 URL隐藏密码 两张表名为键维护进程级锁sqlalchemy_session.py多个协程并发初始化同一数据库时只有一方执行建表其余轮询等待。2. SQLite 专属优化当检测到引擎方言为 sqlite 时会在连接事件中自动设置PRAGMA busy_timeout 5000与PRAGMA journal_mode WALsqlalchemy_session.py显著减少并发写导致的瞬时锁失败同时对database is locked类OperationalError按0.05/0.1/0.2/0.4/0.8秒的有界退避重试sqlalchemy_session.py。3. 首次写入防竞态add_items先查会话行是否存在不存在则以begin_nested()插入父会话行若并发写者已先创建则捕获IntegrityError静默跳过sqlalchemy_session.py避免先检查后插入的经典竞态。4.pop_item的原子取号支持DELETE ... RETURNING的方言PostgreSQL、MySQL 等直接用删除返回语句认领最新一条不支持该特性的旧版 SQLite 则先执行BEGIN IMMEDIATE独占写锁再用SELECT ... FOR UPDATE锁定行后删除sqlalchemy_session.py确保并发弹出不会取到同一条。5. 取消安全的写入add_items、pop_item、clear_session均经由_await_mutation定义于 src/agents/memory/session.py包装即使调用方协程被取消也要等底层事务真正落盘后再抛出取消异常避免数据丢失。七、测试验证仓库为SQLAlchemySession提供了较完整的异步测试套件 tests/extensions/memory/test_sqlalchemy_session.py以sqliteaiosqlite:///:memory:为测试库覆盖直连数据库的增/查/弹/清全流程含与AgentScriptedModel的集成运行非 ASCII 文本默认转义存储与读取还原ensure_asciiFalse时存储格式的可读性get_items的 limit 语义与损坏行跳过逻辑并发写入与pop_item的原子性、SQLite 锁重试等边界场景。这些测试既是对实现行为的约束也可以作为你接入SQLAlchemySession时的行为参考。八、API 参考与进一步阅读SQLAlchemySession—— 主要类位于agents.extensions.memorySession—— 基础会话协议Protocol自定义会话后端需实现其四个异步方法SessionSettings—— 会话配置如默认limit会话后端横向对比与选型建议见 docs/sessions/index.md其中也说明SQLAlchemySession可配合EncryptedSession做加密包装见 docs/sessions/index.md内存版会话的官方示例可参考 examples/memory/sqlite_session_example.py理解 Session 在 Runner 中的实际用法。综上SQLAlchemySession的核心价值在于用一套符合Session协议的实现把 Agent 对话历史无缝接入你现有的 SQLAlchemy 数据库栈——从开发期的内存 SQLite到生产环境的 PostgreSQL/MySQL只需更换 URL 或复用引擎即可平滑迁移。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考