openai-agents-python 中的 SQLAlchemySession用任意 SQLAlchemy 数据库构建生产级 Agent 会话记忆【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonSQLAlchemySession是 openai-agents-pythonAgents SDK提供的一个基于 SQLAlchemy 的会话Session存储后端允许你将 Agent 的多轮对话历史持久化到 PostgreSQL、MySQL、SQLite 等任何 SQLAlchemy 支持的数据库中。读完本文你将掌握它的安装与驱动选型、两种接入方式数据库 URL 与既有 AsyncEngine、底层表结构与序列化机制、四个核心会话操作方法的实现原理以及如何将其无缝接入Runner.run(...)构建生产级的多轮对话应用。会话记忆是什么SQLAlchemySession 处在什么位置在 Agents SDK 中Session 用于为「特定会话」保存对话历史使 Agent 无需手动维护.to_input_list()即可在多次运行之间保持上下文。其行为可以概括为三步参见 会话总览文档运行前Runner 自动读取该会话的历史记录并把它拼接到本次输入之前运行后本次运行产生的新条目用户输入、助手回复、工具调用等自动写入会话上下文保持后续使用同一会话的每次运行都包含完整历史Agent 因此能记住之前的交互。SDK 内置了多种会话实现各有侧重详见 docs/sessions/index.md 的内置实现对照表会话类型适用场景SQLiteSession/AsyncSQLiteSession本地开发、简单应用RedisSession跨 worker/服务的低延迟共享记忆SQLAlchemySession已有数据库的生产应用支持任意 SQLAlchemy 数据库MongoDBSession/DaprSessionMongoDB 生态 / 云原生 Dapr sidecar 部署OpenAIConversationsSessionOpenAI 服务端托管存储SQLAlchemySession的价值在于「复用你现有的数据库基础设施」如果你的业务已经在使用 PostgreSQL 或 MySQL无需再引入新的存储组件就能获得与SQLiteSession完全一致的会话协议只是底层数据落到了关系型数据库里。从源码结构看SQLAlchemySession继承自SessionABC抽象基类见 src/agents/memory/session.py而Session本身是一个带session_id、session_settings与四个历史操作方法的 Protocolsrc/agents/memory/session.py。这也意味着它可以直接传给Runner.run(agent, input, sessionsession)SDK 会自动完成「取历史 → 拼输入 → 存新条目」的完整闭环。安装与数据库驱动选型SQLAlchemySession属于可选依赖optional extra需要通过sqlalchemy这个 extra 安装同时还要根据数据库 URL 搭配对应的异步驱动异步驱动是硬性要求因为会话方法全部是async的。SQLitesqliteaiosqlite://extra 之外需额外安装aiosqlitepip install openai-agents[sqlalchemy] aiosqlitePostgreSQLpostgresqlasyncpg://sqlalchemyextra 已经自带asyncpg无需额外安装。这一点可以在 pyproject.toml 中得到确认sqlalchemy [SQLAlchemy2.0, asyncpg0.29.0]。MySQLmysqlaiomysql://需要额外安装aiomysql其rsa附加依赖用于支持 MySQL 的 SHA-256 认证方式pip install openai-agents[sqlalchemy] aiomysql[rsa]安装完成后SQLAlchemySession会通过agents.extensions.memory命名空间导出可以直接导入见 src/agents/extensions/memory/init.py该模块采用惰性导入仅在使用到时才加载 sqlalchemy 相关依赖。快速上手两种接入方式方式一通过数据库 URLfrom_urlfrom_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, urlsqliteaiosqlite:///:memory:, create_tablesTrue, ) result await Runner.run(agent, Hello, sessionsession) print(result.final_output) if __name__ __main__: asyncio.run(main())对于 PostgreSQL 只需替换 URLpostgresqlasyncpg://user:passhost/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(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) # 清理关闭连接池 await engine.dispose() if __name__ __main__: asyncio.run(main())两种方式完全等价区别仅在于引擎的所有权from_url内部创建引擎并由会话持有直接构造则要求你自行管理引擎生命周期见下文「生产实践」。构造参数与底层表结构SQLAlchemySession的构造函数定义在 src/agents/extensions/memory/sqlalchemy_session.py参数如下参数类型默认值说明session_idstr必填会话唯一标识engineAsyncEngine必填预配置的异步引擎必须使用异步驱动postgresqlasyncpg://、mysqlaiomysql://或sqliteaiosqlite://create_tablesboolFalse是否自动建表建索引。生产环境默认False配合迁移工具开发与测试时设为Truesessions_tablestragent_sessions会话表的表名可按需覆盖messages_tablestragent_messages消息表的表名可按需覆盖session_settingsSessionSettings \| dictNone会话配置如默认的条目获取上限ensure_asciiboolTrue序列化为 JSON 时是否转义非 ASCII 字符默认True以保持历史存储格式from_url的完整签名为from_url(session_id, *, url, engine_kwargsNone, session_settingsNone, **kwargs)源码位置其中engine_kwargs会透传给create_async_engine例如池大小、超时等其余关键字参数透传给构造函数。自动建表两张表 一个复合索引启用create_tablesTrue时_ensure_tables()会通过self._metadata.create_all创建以下结构定义见 源码 L188-L231agent_sessions会话主表列类型约束session_idString主键created_atTIMESTAMP无时区NOT NULL服务端默认CURRENT_TIMESTAMPupdated_atTIMESTAMP无时区NOT NULL服务端默认CURRENT_TIMESTAMP更新时自动刷新agent_messages消息明细表列类型约束idInteger主键自增session_idStringNOT NULL外键引用agent_sessions.session_idON DELETE CASCADEmessage_dataTextNOT NULL存放 JSON 序列化后的对话条目created_atTIMESTAMP无时区NOT NULL服务端默认CURRENT_TIMESTAMP此外还包含复合索引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_asciiTrue默认非 ASCII 字符中文、日文、Emoji 等会被转义为\uXXXX。这保持了历史上的存储格式读取时仍能还原出原始文本ensure_asciiFalse多语言文本在数据库中以可读的原始字符保存方便直接查库调试。session SQLAlchemySession.from_url( user-123, urlsqliteaiosqlite:///conversations.db, create_tablesTrue, ensure_asciiFalse, )使用现有引擎时也可以把同样的参数直接传给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.limitSessionSettings(limitNone)表示返回全部定义见 src/agents/memory/session_settings.py按created_at ASC, id ASC升序返回全部历史传入正数limitN时返回时间上最新的 N 条且保持时间正序。实现上先按created_at DESC, id DESC取尾部 N 条再反转效率更高当最新若干条里混有损坏记录时实现会以「窗口翻倍」的方式扩大拉取范围确保limit统计的是有效条目数与 SQLite 后端行为保持一致非正数limit保留各数据库方言定义的原始语义直接透传给 SQL。add_items写入新条目签名async def add_items(self, items: list[TResponseInputItem]) - None源码 L370-L418。写入采用「先确保会话行存在再批量插入消息」的事务流程查询agent_sessions中是否已有该session_id若无则在嵌套事务中插入会话行——若此时另一个并发写入者已抢先创建了该行捕获IntegrityError后静默忽略避免 check-then-insert 竞态将待写入条目整体批量INSERT进agent_messages最后刷新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通过引擎connect事件为每个连接执行PRAGMA busy_timeout 5000与PRAGMA journal_mode WAL从连接层面降低瞬时锁失败对写入型操作采用有界退避重试延迟序列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( nameAssistant, instructionsReply very concisely., ) # 会话 ID 建议使用有业务含义的命名如 user_12345、thread_abc123 session SQLAlchemySession.from_url( conversation_123, urlsqliteaiosqlite:///:memory:, create_tablesTrue, ) # 第一轮提问并得到回答 result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession, ) print(fAssistant: {result.final_output}) # 第二轮不重复提供历史Agent 依然记得上下文 result await Runner.run(agent, What state is it in?, sessionsession) print(fAssistant: {result.final_output}) # 读取历史只取最新的 2 条 latest_items await session.get_items(limit2) for i, msg in enumerate(latest_items, 1): print(f {i}. {msg.get(role)}: {msg.get(content)}) # 读取全部历史 all_items await session.get_items() print(fTotal items in session: {len(all_items)}) if __name__ __main__: asyncio.run(main())对超长对话可以用SessionSettings限制每次运行拉取的历史条数通过RunConfig.session_settings按运行覆盖会话的默认设置在构造时传入session_settingsfrom agents import Agent, RunConfig, Runner, SessionSettings result await Runner.run( agent, Summarize our recent discussion., sessionsession, run_configRunConfig(session_settingsSessionSettings(limit50)), )另外有两点使用约束值得注意详见 会话总览文档同一运行中Session 与运行级续接选项互斥不能同时使用conversation_id、previous_response_id或auto_previous_response_id二者只能选其一中断恢复如果运行因审批approval暂停请使用同一 session 实例或相同 session ID 相同存储后端的另一个实例恢复运行以便续接同一份存储历史。生产实践建议建表交给迁移工具生产环境保持create_tablesFalse把两张表及索引纳入 Alembic 等迁移流程管理create_tablesTrue仅用于开发与测试快速起跑引擎生命周期用from_url时引擎由会话内部持有直接构造时引擎归你所有应用退出前记得await engine.dispose()。类还暴露了engine只读属性源码 L512-L523方便你在高级场景下检查连接池状态或手动释放资源多进程/多 worker 共享相比文件型 SQLitePostgreSQL/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会话协议与SessionABCsrc/agents/memory/session.pySessionSettings定义src/agents/memory/session_settings.py可运行示例examples/memory/sqlalchemy_session_example.py依赖声明sqlalchemyextrapyproject.toml【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
