openai-agents-python Sessions 会话记忆完全指南：从多轮对话持久化到自定义存储后端

openai-agents-python Sessions 会话记忆完全指南从多轮对话持久化到自定义存储后端【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本指南围绕 openai-agents-pythonAgents SDK内置的 Sessions 会话记忆机制展开系统讲解如何利用 SDK 自动维护跨多次 Agent 运行run的对话历史涵盖快速上手、历史合并与裁剪、内置存储后端选型SQLite/Redis/MongoDB/SQLAlchemy/Dapr 等、记忆操作、对话压缩compaction以及自定义会话实现。读完本文你将掌握在多轮对话、聊天应用与生产级多进程场景下如何用最少的代码让 Agent 记住上下文并在需要时接管存储层的全部控制权。Sessions 是什么让 Agent 记住上下文的官方方案在多轮对话场景中Agent 需要记住用户上一次问了什么、自己答了什么才能理解它指的是什么这类指代。在没有 Sessions 之前开发者必须自己调用.to_input_list()之类的逻辑把历史消息手工拼接到每一次新的模型输入中。Agents SDK 提供的 Sessions 机制正是为了消除这种手工管理负担它为一个特定会话session存储对话历史让 Agent 在多轮运行之间自动保持上下文无需显式的内存管理。从源码角度看Sessions 的核心是 Session 协议它规定了四个异步历史操作详见后文自定义 Session 实现章节SDK 的所有内置实现SQLiteSession、RedisSession、MongoDBSession等都遵循这一结构。其设计文档与使用说明即本仓库的 docs/sessions/index.md。何时使用 Sessions何时不用当你希望 SDK 替你管理客户端侧的内存时就使用 Sessions。需要特别注意的是在同一个 run 中Session 不能与运行级续接选项conversation_id、previous_response_id、auto_previous_response_id同时使用。这三个选项走的是 OpenAI 服务端托管续接server-managed continuation路线如果你想要服务端管理的方式应选择其中一种机制而不是在 Session 之上叠加使用。快速开始三次对话零手工拼接最简单的会话用法只需三步创建 Agent、创建 Session 实例、每次运行都传入同一个 session。from agents import Agent, Runner, SQLiteSession # Create agent agent Agent( nameAssistant, instructionsReply very concisely., ) # Create a session instance with a session ID session SQLiteSession(conversation_123) # First turn result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession ) print(result.final_output) # San Francisco # Second turn - agent automatically remembers previous context result await Runner.run( agent, What state is it in?, sessionsession ) print(result.final_output) # California # Also works with synchronous runner result Runner.run_sync( agent, Whats the population?, sessionsession ) print(result.final_output) # Approximately 39 million可以看到第二次、第三次运行完全不需要把第一次的输出手工传给模型Agent 自动记得桥在旧金山、旧金山在加利福尼亚。Runner.run_sync同样支持session参数同步场景下享受相同的行为。用同一 Session 恢复被中断的运行如果一次运行因审批approval等原因暂停需要用同一个 session 实例或另一个配置了相同 session ID 与相同底层存储后端、指向同一份数据的实例恢复运行这样被恢复的轮次会继续使用同一份已存储的对话历史result await Runner.run(agent, Delete temporary files that are no longer needed., sessionsession) if result.interruptions: state result.to_state() for interruption in result.interruptions: state.approve(interruption) result await Runner.run(agent, state, sessionsession)关于审批human-in-the-loop的更多细节可参考 docs/human_in_the_loop.md。Session 的核心行为运行前取历史、运行后存新项启用会话记忆后Runner 在每次运行前后做三件事运行前Runner 自动取回该会话的历史记录并将其前置prepend到输入 items 中运行后本次运行产生的所有新 items用户输入、助手回复、工具调用等自动存入会话上下文保持每次用同一 session 的后续运行都包含完整对话历史Agent 因此持续保持上下文。这彻底省去了手工调用.to_input_list()以及在两次运行之间管理对话状态的麻烦。在源码层面这一先取历史、再拼新输入、最后只持久化新轮次内容的合并逻辑实现在 src/agents/run_internal/session_persistence.py 的prepare_input_with_session中先按SessionSettings.limit若设置取历史把历史规范化为 input-item 格式再把当前轮的输入转为新 items 列表最后拼接成模型输入而需要持久化的内容严格限定为属于当前轮的新 items即使合并回调对历史做了重排或过滤也不会把旧历史当作新输入重复写入存储。控制历史与新输入如何合并session_input_callback传入 session 时Runner 默认按以下顺序组装模型输入会话历史通过session.get_items(...)取回本轮新输入如果默认的历史在前、新输入在后不满足需求可以用RunConfig.session_input_callback自定义合并步骤发生在调用模型之前。回调接收两个列表history取回的会话历史已规范化为 input-item 格式new_input当前轮的新输入 items回调需要返回最终应发送给模型的 input items 列表。from agents import Agent, RunConfig, Runner, SQLiteSession def keep_recent_history(history, new_input): # Keep only the last 10 history items, then append the new turn. return history[-10:] new_input agent Agent(nameAssistant) session SQLiteSession(conversation_123) result await Runner.run( agent, Continue from the latest updates only., sessionsession, run_configRunConfig(session_input_callbackkeep_recent_history), )两个要点回调收到的是两份列表的拷贝可以安全地就地修改排序、过滤、去重而不影响存储返回的列表决定该轮模型输入但 SDK仍然只持久化属于本轮的新 items。因此对旧历史做重排或过滤不会导致旧的 session items 被当作新输入再次保存。这一语义在 session_persistence.py 中有精细实现合并后SDK 通过对象身份identity与内容频次frequency双重比对从回调输出中精确还原出哪些属于新轮次再入库。需要自定义裁剪、重排或选择性包含历史、但又不想改变 session 的存储方式时就用这个回调。如果你还需要在调用模型前做一次最终兜底处理例如追加系统提示、控制 token 长度可以使用 running_agents 指南 中介绍的RunConfig.call_model_input_filter它发生在session_input_callback之后、紧邻模型调用之前。限制取回的历史数量SessionSettings(limit)长对话中全部历史可能超出模型上下文窗口。用SessionSettings控制每次运行前取回多少历史SessionSettings(limitNone)默认取回全部会话 itemsSessionSettings(limitN)只取回最近 N 个 items每次运行可通过RunConfig.session_settings应用该限制from agents import Agent, RunConfig, Runner, SessionSettings, SQLiteSession agent Agent(nameAssistant) session SQLiteSession(conversation_123) result await Runner.run( agent, Summarize our recent discussion., sessionsession, run_configRunConfig(session_settingsSessionSettings(limit50)), )如果 session 实现本身暴露了默认的session_settings例如SQLiteSession(session_settings...)那么RunConfig.session_settings中的每个非 None 值都会覆盖对应字段的默认值对本次运行生效。这一覆盖而非替换的语义在 SessionSettings.resolve 中实现它把 override 中非 None 的字段叠加到实例自身的设置之上。对于想在不改变 session 默认行为的前提下临时封顶取回规模的长对话这个参数非常实用。记忆操作增、查、弹、清Session 提供四个基本的历史管理操作from agents import SQLiteSession session SQLiteSession(user_123, conversations.db) # Get all items in a session items await session.get_items() # Add new items to a session new_items [ {role: user, content: Hello}, {role: assistant, content: Hi there!} ] await session.add_items(new_items) # Remove and return the most recent item last_item await session.pop_item() print(last_item) # {role: assistant, content: Hi there!} # Clear all items from a session await session.clear_session()以SQLiteSession为例其实现位于 src/agents/memory/sqlite_session.py数据以 JSON 序列化后的字符串存入agent_messages表会话元数据在agent_sessions表get_items按id ASC返回时间正序历史pop_item使用DELETE ... RETURNING原子地删除并返回最近一条记录同时会跳过损坏的 JSON 行继续向前寻找有效条目clear_session则级联清空消息与会话两表。文件型数据库自动启用 WAL 模式并对共享同一数据库文件的多个 session 做进程内文件锁协调保证并发访问安全。用 pop_item 做纠错撤销上一轮对话pop_item最典型的用途是撤销或修改对话的最后一条消息。比如用户问错了问题、希望 Agent 忘掉刚才那轮问答from agents import Agent, Runner, SQLiteSession agent Agent(nameAssistant) session SQLiteSession(correction_example) # Initial conversation result await Runner.run( agent, Whats 2 2?, sessionsession ) print(fAgent: {result.final_output}) # User wants to correct their question assistant_item await session.pop_item() # Remove agents response user_item await session.pop_item() # Remove users question # Ask a corrected question result await Runner.run( agent, Whats 2 3?, sessionsession ) print(fAgent: {result.final_output})由于pop_item是弹出最近一条连续调用两次即可先后移除助手回复与用户提问之后重新提问时历史中不再有旧的错误问答。内置 Session 实现选型与用法SDK 提供了多种内置 session 实现覆盖从本地开发到生产分布式部署的各类场景。先看选型总表再按需阅读各小节Session 类型适用场景说明SQLiteSession本地开发与简单应用内置、轻量支持文件存储或内存存储AsyncSQLiteSession使用aiosqlite的异步 SQLite扩展后端异步驱动支持RedisSession多 worker/服务间共享内存适合低延迟分布式部署SQLAlchemySession已有数据库的生产应用支持所有 SQLAlchemy 支持的数据库MongoDBSession已用 MongoDB 或需要多进程存储的应用异步 pymongo原子序列计数器保证排序DaprSession带 Dapr sidecar 的云原生部署支持多种状态存储外加 TTL 与一致性控制OpenAIConversationsSessionOpenAI 服务端托管存储基于 OpenAI Conversations API 的历史OpenAIResponsesCompactionSession长对话自动压缩包装另一个 session 后端AdvancedSQLiteSessionSQLite 分支/分析功能更重见独立文档页EncryptedSession在另一 session 之上加加密 TTL包装器先选好底层后端部分实现有独立文档页下方各小节内联链接。另外需要特别说明如果你在实现 ChatKit 的 Python 服务端应使用chatkit.store.Store实现来完成 ChatKit 的线程与消息持久化Agents SDK 的 session如SQLAlchemySession管理的是 SDK 侧的对话历史不能作为 ChatKit store 的替代品。OpenAI Conversations API 会话服务端托管通过OpenAIConversationsSession使用 OpenAI 的 Conversations API 托管历史from agents import Agent, Runner, OpenAIConversationsSession # Create agent agent Agent( nameAssistant, instructionsReply very concisely., ) # Create a new conversation session OpenAIConversationsSession() # Optionally resume a previous conversation by passing a conversation ID # session OpenAIConversationsSession(conversation_idconv_123) # Start conversation result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession ) print(result.final_output) # San Francisco # Continue the conversation result await Runner.run( agent, What state is it in?, sessionsession ) print(result.final_output) # California不传conversation_id时自动新建会话传入已有 ID 即可恢复之前会话。OpenAI Responses 压缩会话compactionOpenAIResponsesCompactionSession使用 Responses API 的responses.compact对已存储的历史做压缩从而控制上下文增长。它包装一个底层 session默认在每轮之后根据should_trigger_compaction决定是否自动压缩。不要用它包装OpenAIConversationsSession——两者管理历史的方式不同前者在 SDK 侧维护、后者在服务端维护源码中也会直接抛出ValueError拒绝这种组合见 src/agents/memory/openai_responses_compaction_session.py。典型用法自动压缩from agents import Agent, Runner, SQLiteSession from agents.memory import OpenAIResponsesCompactionSession underlying SQLiteSession(conversation_123) session OpenAIResponsesCompactionSession( session_idconversation_123, underlying_sessionunderlying, ) agent Agent(nameAssistant) result await Runner.run(agent, Hello, sessionsession) print(result.final_output)默认行为每轮结束后 SDK 检查压缩候选是否达到阈值达到才压缩。默认阈值是候选 items 数量 ≥ 10候选即排除了用户消息与既有 compaction 项之后的历史条目见源码中的default_should_trigger_compaction。关于压缩的时序与计费语义官方文档明确了以下几点自动压缩发生时SDK 会等待压缩完成后才让Runner.run(...)返回或流式事件迭代器关闭压缩请求产生的用量会计入该次运行的Usage汇总而之后手动调用run_compaction()时默认没有外层运行上下文不会更新已完成运行的 usage 对象compaction_modeprevious_response_id依赖压缩 session 保留的 Responses API response ID在该响应链仍然可用时效果最好compaction_modeinput改为从当前 session items 重建压缩请求适用于响应链不可用、或希望以 session 内容为准的场景默认auto会自动选择当前最安全可用的选项。特别地如果 Agent 以ModelSettings(storeFalse)运行Responses API 不会保留最后一条响应供后续查询。在这种无状态配置下默认的auto模式会回退到基于输入的压缩而不是依赖previous_response_id。完整示例参见 examples/memory/compaction_session_stateless_example.py。自动压缩可能阻塞流式输出压缩会清空并重写会话历史因此 SDK 必须等压缩完成才认为运行结束。在流式模式下这意味着如果压缩开销较大run.stream_events()可能在最后一个输出 token 之后仍保持打开数秒。压缩包装器在边界处把清空并重写视为可恢复的替换操作相关实现见 src/agents/memory/openai_responses_compaction_session.py如果底层历史已变更后替换失败或被取消包装器会尝试恢复先前历史并在原始异常/取消到达调用方之前等待恢复完成若底层后端在恢复期间也失败则历史可能无法恢复SDK 会记录恢复失败日志。包装器会用一把锁把add_items()、pop_item()、clear_session()与加锁的替换和恢复阶段串行化但一次变更可能在远端压缩请求仍在进行时就完成、随后被成功的替换覆盖。因此手动压缩应在轮次之间进行避免与包装器并发变更压缩进行中不要直接修改底层 session。如果你追求低延迟流式或快速回合建议关闭自动压缩改在轮次之间或空闲期自己调用run_compaction()用自己的标准决定何时强制压缩from agents import Agent, Runner, SQLiteSession from agents.memory import OpenAIResponsesCompactionSession underlying SQLiteSession(conversation_123) session OpenAIResponsesCompactionSession( session_idconversation_123, underlying_sessionunderlying, # Disable triggering the auto compaction should_trigger_compactionlambda _: False, ) agent Agent(nameAssistant) result await Runner.run(agent, Hello, sessionsession) # Decide when to compact (e.g., on idle, every N turns, or size thresholds). await session.run_compaction({force: True})run_compaction({force: True})会无视阈值强制压缩一次适合在空闲期主动收敛历史。SQLite 会话默认实现内置的轻量级默认实现from agents import SQLiteSession # In-memory database (lost when process ends) session SQLiteSession(user_123) # Persistent file-based database session SQLiteSession(user_123, conversations.db) # Use the session result await Runner.run( agent, Hello, sessionsession )SQLiteSession(session_id)使用内存数据库进程结束即丢失SQLiteSession(session_id, conversations.db)使用文件持久化。还可选传sessions_table/messages_table自定义表名、session_settings设置默认取回限制签名见 src/agents/memory/sqlite_session.py。异步 SQLite 会话需要aiosqlite驱动的 SQLite 持久化时使用AsyncSQLiteSessionpip install aiosqlitefrom agents import Agent, Runner from agents.extensions.memory import AsyncSQLiteSession agent Agent(nameAssistant) session AsyncSQLiteSession(user_123, db_pathconversations.db) result await Runner.run(agent, Hello, sessionsession)Redis 会话多服务共享需要跨多个 worker 或服务共享会话内存时使用RedisSessionpip install openai-agents[redis]from agents import Agent, Runner from agents.extensions.memory import RedisSession agent Agent(nameAssistant) session RedisSession.from_url( user_123, urlredis://localhost:6379/0, ) result await Runner.run(agent, Hello, sessionsession) await session.close()生命周期语义值得注意from_url(...)创建并拥有Redis 客户端close()之后该 session 进入终态后续操作抛出RuntimeError重复或并发调用close()是安全的。如果你的应用已经自己管理 Redis 客户端应直接用RedisSession(..., redis_client...)构造此时close()是 no-op客户端所有权与 session 可用性都由调用方保留。SQLAlchemy 会话生产级面向生产、基于任意 SQLAlchemy 支持数据库的会话持久化from agents.extensions.memory import SQLAlchemySession # Using database URL session SQLAlchemySession.from_url( user_123, urlpostgresqlasyncpg://user:passlocalhost/db, create_tablesTrue ) # Using existing engine from sqlalchemy.ext.asyncio import create_async_engine engine create_async_engine(postgresqlasyncpg://user:passlocalhost/db) session SQLAlchemySession(user_123, engineengine, create_tablesTrue)详细文档参见 docs/sessions/sqlalchemy_session.md。Dapr 会话云原生状态存储已经在跑 Dapr sidecar、或想在不改 Agent 代码的前提下切换状态存储后端时使用DaprSessionpip install openai-agents[dapr]from agents import Agent, Runner from agents.extensions.memory import DaprSession agent Agent(nameAssistant) async with DaprSession.from_address( user_123, state_store_namestatestore, dapr_addresslocalhost:50001, ) as session: result await Runner.run(agent, Hello, sessionsession) print(result.final_output)注意事项from_address(...)创建并拥有 Dapr 客户端若应用已自行管理直接用DaprSession(..., dapr_client...)构造退出上下文或调用close()会让拥有客户端的 session 进入终态后续操作抛出RuntimeError重复或并发close()安全。注入客户端时close()为 no-opsession 仍可用若底层状态存储支持 TTL传ttl...可自动对 session 数据施加 TTL 过期传consistencyDAPR_CONSISTENCY_STRONG可对状态写入与删除请求强一致性取决于存储注意线上读取的一致性还需要上游 Dapr Python 客户端在get_state上支持设置一致性Dapr Python SDK 也会检查 HTTP sidecar 端点本地开发时除了dapr_address指定的 gRPC 端口还要用--dapr-http-port 3500启动 Dapr完整的环境搭建与故障排查流程参见 examples/memory/dapr_session_example.py。MongoDB 会话多进程水平扩展已使用 MongoDB、或需要水平可扩展的多进程会话存储时使用MongoDBSessionpip install openai-agents[mongodb]from agents import Agent, Runner from agents.extensions.memory import MongoDBSession agent Agent(nameAssistant) # Create from URI — owns the client and closes it when session.close() is called session MongoDBSession.from_uri( user-123, urimongodb://localhost:27017, databaseagents, ) result await Runner.run(agent, Hello, sessionsession) print(result.final_output) await session.close()注意事项from_uri(...)创建并拥有AsyncMongoClient在session.close()时关闭它。拥有客户端的 session 在close()后进入终态后续操作抛出RuntimeError若应用自行管理客户端直接用MongoDBSession(..., client...)构造此时close()为 no-op调用方保留客户端生命周期责任session 仍可用通过向from_uri(...)传mongodbsrv://user:passwordcluster.example.mongodb.net形式的 URI 即可连接 MongoDB Atlas无需其他改动使用两个集合名字都可通过sessions_collection默认agent_sessions与messages_collection默认agent_messages配置首次使用自动建索引。每次非空add_items()调用会写入一个逻辑批次文档其单调递增的seq字段按批次最后一条消息排序旧的逐消息文档仍然可读。一个逻辑批次必须小于 MongoDB 单文档大小上限超限的批次会原子失败不会存储半截批次首次运行前可用await session.ping()验证连通性。高级 SQLite 会话分支与分析AdvancedSQLiteSession提供对话分支branching、用量分析usage analytics与结构化查询等增强能力from agents.extensions.memory import AdvancedSQLiteSession # Create with advanced features session AdvancedSQLiteSession( session_iduser_123, db_pathconversations.db, create_tablesTrue ) # Automatic usage tracking result await Runner.run(agent, Hello, sessionsession) await session.store_run_usage(result) # Track token usage # Conversation branching await session.create_branch_from_turn(2) # Branch from turn 2详细文档参见 docs/sessions/advanced_sqlite_session.md。加密会话透明加密 TTLEncryptedSession是任意 session 实现之上的透明加密包装器from agents.extensions.memory import EncryptedSession, SQLAlchemySession # Create underlying session underlying_session SQLAlchemySession.from_url( user_123, urlsqliteaiosqlite:///conversations.db, create_tablesTrue ) # Wrap with encryption and TTL session EncryptedSession( session_iduser_123, underlying_sessionunderlying_session, encryption_keyyour-secret-key, ttl600 # 10 minutes ) result await Runner.run(agent, Hello, sessionsession)ttl600表示会话数据 10 分钟后过期。详细文档参见 docs/sessions/encrypted_session.md。其他内置 Session 类型SDK 还提供少量其他内置选项可参考 examples/memory/ 目录下的示例以及src/agents/extensions/memory/下的扩展源码见 src/agents/extensions/memory。操作模式与最佳实践Session ID 命名使用有意义的 session ID 便于组织会话按用户user_12345按线程thread_abc123按上下文support_ticket_456持久化选型速查临时会话内存 SQLiteSQLiteSession(session_id)持久会话文件 SQLiteSQLiteSession(session_id, path/to/db.sqlite)需要aiosqliteAsyncSQLiteSession(session_id, db_path...)共享低延迟会话内存RedisSession.from_url(session_id, urlredis://...)已有 SQLAlchemy 支持数据库的生产系统SQLAlchemySession(session_id, engineengine, create_tablesTrue)已用 MongoDB 或需多进程水平扩展MongoDBSession.from_uri(session_id, urimongodb://localhost:27017)云原生生产部署自带遥测、追踪、数据隔离支持 30 数据库后端DaprSession.from_address(session_id, state_store_namestatestore, dapr_addresslocalhost:50001)希望历史存于 OpenAI Conversations APIOpenAIConversationsSession()需要透明加密与 TTL 过期EncryptedSession(session_id, underlying_session, encryption_key)其他生产系统例如 Django的进阶需求可考虑实现自定义 session 后端多个 Session互相隔离的对话历史from agents import Agent, Runner, SQLiteSession agent Agent(nameAssistant) # Different sessions maintain separate conversation histories session_1 SQLiteSession(user_123, conversations.db) session_2 SQLiteSession(user_456, conversations.db) result1 await Runner.run( agent, Help me with my account, sessionsession_1 ) result2 await Runner.run( agent, What are my charges?, sessionsession_2 )不同 session 维护各自独立的对话历史互不干扰。Session 共享多个 Agent 看同一份历史# Different agents can share the same session support_agent Agent(nameSupport) billing_agent Agent(nameBilling) session SQLiteSession(user_123) # Both agents will see the same conversation history result1 await Runner.run( support_agent, Help me with my account, sessionsession ) result2 await Runner.run( billing_agent, What are my charges?, sessionsession )不同 Agent 共享同一 session 时后者能看到前者写入的历史适合客服转接/多角色接力的场景。完整示例贯穿三轮对话的会话记忆import asyncio from agents import Agent, Runner, SQLiteSession async def main(): # Create an agent agent Agent( nameAssistant, instructionsReply very concisely., ) # Create a session instance that will persist across runs session SQLiteSession(conversation_123, conversation_history.db) print( Sessions Example ) print(The agent will remember previous messages automatically.\n) # First turn print(First turn:) print(User: What city is the Golden Gate Bridge in?) result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession ) print(fAssistant: {result.final_output}) print() # Second turn - the agent will remember the previous conversation print(Second turn:) print(User: What state is it in?) result await Runner.run( agent, What state is it in?, sessionsession ) print(fAssistant: {result.final_output}) print() # Third turn - continuing the conversation print(Third turn:) print(User: Whats the population of that state?) result await Runner.run( agent, Whats the population of that state?, sessionsession ) print(fAssistant: {result.final_output}) print() print( Conversation Complete ) print(Notice how the agent remembered the context from previous turns!) print(Sessions automatically handles conversation history.) if __name__ __main__: asyncio.run(main())运行时注意观察Agent 在第二轮、第三轮都自动记得前文语境无需任何手工拼接。自定义 Session 实现遵循 Session 协议你可以实现自己的会话存储。只需创建一个在结构上遵循Session协议的类即可——不需要继承SessionABC。需要提供session_id与session_settings属性并直接实现四个历史方法from agents import Agent, Runner, SessionSettings from agents.items import TResponseInputItem class MyCustomSession: Custom session implementation following the Session protocol. session_settings: SessionSettings | None None def __init__(self, session_id: str) - None: self.session_id session_id self.items: list[TResponseInputItem] [] async def get_items(self, limit: int | None None) - list[TResponseInputItem]: if limit is None: return list(self.items) if limit 0: return [] return list(self.items[-limit:]) async def add_items(self, items: list[TResponseInputItem]) - None: self.items.extend(items) async def pop_item(self) - TResponseInputItem | None: return self.items.pop() if self.items else None async def clear_session(self) - None: self.items.clear() # Use your custom session agent Agent(nameAssistant) result await Runner.run( agent, Hello, sessionMyCustomSession(my_session) )协议要求的方法语义取自 session.pyget_items(limit)取回会话历史limitNone取全部指定时按时间正序返回最近 N 条add_items(items)向历史追加 itemspop_item()移除并返回最近一条空会话返回Noneclear_session()清空该会话所有 items。从自定义 Session 中访问运行上下文Agents SDK 可以把当前的RunContextWrapper传给自定义 session用于租户路由、鉴权或其他应用级存储决策。要让 SDK 传入 wrapper需要在全部四个历史方法上显式声明一个可按关键字传入的wrapper参数from typing import Any from agents import RunContextWrapper from agents.items import TResponseInputItem class ContextAwareSession: async def get_items( self, limit: int | None None, *, wrapper: RunContextWrapper[Any] | None None, ) - list[TResponseInputItem]: ... async def add_items( self, items: list[TResponseInputItem], *, wrapper: RunContextWrapper[Any] | None None, ) - None: ... async def pop_item( self, *, wrapper: RunContextWrapper[Any] | None None, ) - TResponseInputItem | None: ... async def clear_session( self, *, wrapper: RunContextWrapper[Any] | None None, ) - None: ...启用条件很严格只有当get_items、add_items、pop_item、clear_session全部声明了wrapper时SDK 才会启用这一集成。通用的**kwargs参数不满足签名检查。已有未声明wrapper的 session 实现保持原有调用形态无需改动即可继续工作。这一检测逻辑位于 session.py 的_session_accepts_wrapper它会逐个检查四个方法签名中是否存在名为wrapper的位置或关键字参数。社区 Session 实现社区也贡献了额外的 session 实现包描述openai-django-sessions基于 Django ORM 的 session支持任意 Django 支持的数据库PostgreSQL、MySQL、SQLite 等如果你构建了自己的 session 实现欢迎提交文档 PR 将其补充到官方列表中。API 参考更详细的 API 文档可对照以下源码文件Session协议协议接口定义OpenAIConversationsSessionOpenAI Conversations API 实现OpenAIResponsesCompactionSessionResponses API 压缩包装器SQLiteSession基础 SQLite 实现AsyncSQLiteSession基于aiosqlite的异步实现RedisSessionRedis 后端实现SQLAlchemySessionSQLAlchemy 驱动实现MongoDBSessionMongoDB 后端实现DaprSessionDapr 状态存储实现AdvancedSQLiteSession带分支与分析功能的增强 SQLiteEncryptedSession任意 session 的加密包装器相关配置类还涉及RunConfigsession_input_callback、session_settings字段与SessionSettings运行时合并逻辑见 src/agents/run_internal/session_persistence.py。结合 docs/running_agents.md、docs/sessions/advanced_sqlite_session.md、docs/sessions/sqlalchemy_session.md 与 docs/sessions/encrypted_session.md即可把会话记忆从本地原型平滑演进到生产级架构。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考