
openai-agents-python 实战用 SQLAlchemySession 将 Agent 会话历史接入生产级数据库【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonSQLAlchemySession是 openai-agents-pythonOpenAI 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(limitNone)读取会话历史limit指定时返回按时间正序的最新 N 条add_items(items)追加新的会话条目pop_item()移除并返回最近一条条目clear_session()清空该会话全部条目。SQLAlchemySession就是该协议的 SQLAlchemy 落地实现见 sqlalchemy_session.py 源码。相比内置的轻量SQLiteSession它的价值在于复用你已有的数据库与连接池只要你的应用已经在用 PostgreSQL、MySQL 等关系型数据库就可以把 Agent 的会话历史与业务数据存放在同一套基础设施中而不需要额外引入 Redis 之类的独立存储。安装sqlalchemy 可选依赖与异步驱动使用SQLAlchemySession需要安装openai-agents包的sqlalchemy可选依赖extrapip install openai-agents[sqlalchemy]从 pyproject.toml 可以看到该 extra 的真实内容sqlalchemy [SQLAlchemy2.0, asyncpg0.29.0]也就是说extra 本身只携带 SQLAlchemy 2.x 以及PostgreSQL的异步驱动asyncpg。SQLAlchemySession内部使用的是异步引擎sqlalchemy.ext.asyncio因此还需要安装与你数据库 URL 匹配的异步驱动数据库URL 前缀需要的额外包PostgreSQLpostgresqlasyncpg://已随 extra 内置asyncpgSQLitesqliteaiosqlite://aiosqliteMySQLmysqlaiomysql://aiomysql[rsa]rsaextra 为 MySQL 的 SHA-256 认证方式提供依赖对应安装命令pip install openai-agents[sqlalchemy] aiosqlite pip install openai-agents[sqlalchemy] aiomysql[rsa]快速开始方式一通过数据库 URL 创建最简方式是使用from_url类方法传入会话 ID 与数据库 URLimport 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的源码实现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, urlpostgresqlasyncpg://app:secretdb.example.com/agents, create_tablesTrue, 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(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())注意引擎必须是异步驱动创建的如postgresqlasyncpg://、mysqlaiomysql://、sqliteaiosqlite://。由于引擎由外部传入用完后的await engine.dispose()需要由应用自行负责源码还通过 engine 属性 暴露底层AsyncEngine方便你在高级场景下检查连接池状态或手动释放资源。一个完整的多轮对话示例仓库中的 examples/memory/sqlalchemy_session_example.py 演示了会话记忆的完整用法同一个session实例连续传给多轮Runner.runAgent 能自动记住前文What city is the Golden Gate Bridge in? → What state is it in? → Whats the population of that state?最后还能用get_items(limit2)只取最近两条历史import asyncio from agents import Agent, Runner from agents.extensions.memory.sqlalchemy_session import SQLAlchemySession async def main(): agent Agent( nameAssistant, instructionsReply very concisely., ) # In-memory SQLite; create_tablesTrue is useful for development and testing. 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}) # Second turn - the agent remembers the previous conversation result await Runner.run(agent, What state is it in?, sessionsession) print(fAssistant: {result.final_output}) # Only fetch the latest 2 items latest_items await session.get_items(limit2) print(fFetched {len(latest_items)} latest items) if __name__ __main__: asyncio.run(main())构造参数全解SQLAlchemySession主构造函数的完整签名位于 src/agents/extensions/memory/sqlalchemy_session.py#L146-L156参数类型默认值说明session_idstr必填会话唯一标识例如user-123engineAsyncEngine必填已配置的 SQLAlchemy 异步引擎必须使用异步驱动create_tablesboolFalse是否自动建表建索引。生产环境建议False并配合迁移工具开发与测试可设为Truesessions_tablestragent_sessions覆盖会话表的默认表名messages_tablestragent_messages覆盖消息表的默认表名session_settingsSessionSettings \| dictNone会话配置目前核心是limit默认拉取条数ensure_asciiboolTrue序列化会话条目到 JSON 时是否转义非 ASCII 字符from_url相比主构造函数额外接受url任意 SQLAlchemy 异步 URL与engine_kwargs透传给create_async_engine其余关键字参数原样转发给主构造函数。session_settings接受 SessionSettings 对象或等价字典SessionSettings(limitN)会在不显式传limit时让get_items默认只取最新 N 条历史适合超长对话场景下控制每轮的上下文规模。Runner.run(..., run_configRunConfig(session_settingsSessionSettings(limit50)))可以按轮次覆盖。底层存储结构两张表从源码 src/agents/extensions/memory/sqlalchemy_session.py#L188-L231 可以看到SQLAlchemySession使用两张表表名均可通过构造参数覆盖会话表agent_sessionssession_idString主键created_atTIMESTAMP服务端默认CURRENT_TIMESTAMPupdated_atTIMESTAMP服务端默认CURRENT_TIMESTAMP且配置了onupdate每次写入消息时会自动刷新见 add_items 实现。消息表agent_messagesidInteger自增主键SQLite 下额外启用sqlite_autoincrementsession_idString外键指向agent_sessions.session_idondeleteCASCADEmessage_dataText存放序列化后的 JSON 条目created_atTIMESTAMP服务端默认CURRENT_TIMESTAMP复合索引idx_{messages_table}_session_time(session_id, created_at)用于加速按会话按时间取历史。消息以 JSON 文本形式按行存储。在写入路径上add_items先检查会话行是否存在不存在时通过嵌套事务插入父行并捕获IntegrityError从而在并发首写场景下避免 check-then-insert 竞态随后批量insert消息并刷新updated_atsrc/agents/extensions/memory/sqlalchemy_session.py#L388-L418。Session 协议方法的行为细节get_items(limitNone)源码 L301-L368limit为None时按时间正序返回全部历史limitN时先用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_asciiTrue这保留了历史存储格式同时在读取时仍能无损还原原始文本。如果你希望落库的 JSON 中多语言文本保持可读设置ensure_asciiFalsesession SQLAlchemySession.from_url( user-123, urlsqliteaiosqlite:///conversations.db, create_tablesTrue, ensure_asciiFalse, )使用已有引擎时把同样的选项直接传给SQLAlchemySession(...)即可。需要明确的是该设置只改变数据库中存储的 JSON 表示对应源码 serialize/deserialize 钩子不改变get_items等会话方法返回给调用方的值——无论ensure_ascii取值如何还原出的都是原始文本。生产环境落地要点create_tables默认False自动建表只适合开发与测试源码中建表动作仅执行一次见 L281-L299。生产环境建议用 Alembic 等迁移工具管理agent_sessions/agent_messages两张表再以create_tablesFalse接入。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 不受影响。引擎生命周期from_url创建的引擎由 Session 自持记得在进程退出前await engine.dispose()复用已有引擎时释放责任在应用侧。可通过session.engine访问底层引擎做连接池检查等高级操作。会话续跑如果一轮运行因审批interruption暂停请用同一个 session 实例或同 session ID 同存储后端的另一实例继续运行确保续跑回合沿用同一份历史。会话隔离与共享不同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创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考