尧图网页设计 
openai-agents-python 高级 SQLite 会话AdvancedSQLiteSession指南对话分支、Token 用量分析与结构化查询【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonAdvancedSQLiteSession是 openai-agents-python 中基于 SQLite 的增强版会话存储实现在基础SQLiteSession之上额外提供对话分支conversation branching、逐轮 Token 用量统计和结构化会话查询三大能力。本文将以 官方文档 为主体结合 源码实现 与 完整示例带你从初始化、用量追踪、分支管理到数据库表结构完整掌握这套适合构建带对话编辑 / 多时间线能力的多轮 Agent 应用的会话方案。特性总览与基础SQLiteSession相比AdvancedSQLiteSession提供的核心能力包括对话分支Conversation branching可从任意一条用户消息出发创建替代对话路径探索不同的后续走向各分支互不干扰用量追踪Usage tracking按轮turn记录详细的 Token 用量分析支持完整 JSON 明细input_tokens_details/output_tokens_details结构化查询Structured queries按轮获取对话、统计工具调用频次、按内容检索轮次等分支管理Branch management独立的分支切换、列举、删除分支 ID 在会话生命周期内唯一保留消息结构元数据Message structure metadata自动记录消息类型、工具名、轮次号、序号与分支归属。从仓库的会话选型表见 docs/sessions/index.md可以看到它的定位SQLite plus branching/analytics功能集更重见专门页面。也就是说当你只需要简单的历史记忆时用基础的SQLiteSession即可当你需要分支探索、用量分析和结构检索时才选择本方案。快速开始最基本的用法与基础会话几乎一致只需把会话类换成AdvancedSQLiteSession并在每次Runner.run之后调用store_run_usage持久化用量数据from agents import Agent, Runner from agents.extensions.memory import AdvancedSQLiteSession # Create agent agent Agent( nameAssistant, instructionsReply very concisely., ) # Create an advanced session session AdvancedSQLiteSession( session_idconversation_123, db_pathconversations.db, create_tablesTrue ) # First conversation turn result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession ) print(result.final_output) # San Francisco # IMPORTANT: Store usage data await session.store_run_usage(result) # Continue conversation result await Runner.run( agent, What state is it in?, sessionsession ) print(result.final_output) # California await session.store_run_usage(result)Runner.run传入session参数后SDK 会在每次运行前自动读取会话历史、运行后自动写回新增消息用户输入、助手回复、工具调用等无需手动调用.to_input_list()。需要提醒的是会话记忆与运行级续接选项conversation_id、previous_response_id、auto_previous_response_id在同一轮运行中不能混用参见 docs/sessions/index.md。初始化与参数AdvancedSQLiteSession的构造函数完全以关键字参数调用与基础类保持一致的接口风格from agents.extensions.memory import AdvancedSQLiteSession # Basic initialization session AdvancedSQLiteSession( session_idmy_conversation, create_tablesTrue # Auto-create advanced tables ) # With persistent storage session AdvancedSQLiteSession( session_iduser_123, db_pathconversations.db, create_tablesTrue ) # With custom logger import logging logger logging.getLogger(my_app) session AdvancedSQLiteSession( session_idsession_456, create_tablesTrue, loggerlogger )参数说明参数类型说明session_idstr会话的唯一标识用于区分不同对话db_pathstr \| PathSQLite 数据库文件路径默认:memory:内存存储进程结束即丢失create_tablesbool是否自动创建高级结构表默认Falseloggerlogging.Logger \| None会话使用的自定义日志器默认使用模块级日志器源码级的补充说明从 类定义 可以看到AdvancedSQLiteSession直接继承自SQLiteSession构造函数还透传session_settings与**kwargs给父类父类额外支持sessions_table/messages_table自定义基础表名默认分别为agent_sessions与agent_messages见 sqlite_session.py。create_tables的行为在 初始化逻辑 中有两类路径create_tablesTrue在BEGIN IMMEDIATE事务中先建基础表再创建全部高级结构表message_structure、turn_usage、branch_reservations、session_clear_generations及索引create_tablesFalse不建表而是先认领claim已有结构表——源码会通过PRAGMA foreign_key_list校验结构表的外键归属确认这些表确实由当前配置的基础表对创建否则抛出ValueError提示先用create_tablesTrue初始化并认领数据库。这种所有权校验见 认领逻辑是为了防止两个不同配置的会话实例共享同一个数据库文件时把message_structure的message_id外键错误地关联到另一对sessions_table/messages_table上。因此实操上请遵守源码的提示每个sessions_table/messages_table组合使用独立的db_path。另外db_path为:memory:时进程内共享连接以规避线程隔离问题为文件路径时采用线程本地连接并获得进程内的文件级RLock见 sqlite_session.py同一文件被多个会话共享时会复用同一把锁。用量追踪Usage trackingAdvancedSQLiteSession通过store_run_usage把每次运行的 Token 用量按轮写入turn_usage表并可在需要时聚合出会话级统计。注意这完全依赖每次 Agent 运行后调用store_run_usage方法如果不调用turn_usage表将没有任何数据get_session_usage会返回None。存储用量数据# After each agent run, store the usage data result await Runner.run(agent, Hello, sessionsession) await session.store_run_usage(result) # This stores: # - Total tokens used # - Input/output token breakdown # - Request count # - Detailed JSON token information (if available)从 store_run_usage 的实现 看写入前会以当前轮第一条message_structure行的自增 ID作为turn_anchor锚点写入事务提交前会再次校验该锚点行是否仍存在见 内部写入。这意味着如果该轮已被删除、或轮号被新轮复用这条用量写入会被安全跳过绝不会把用量错误地记到一个已不存在的轮次上也不会计入幻影轮 0。序列化 Token 明细失败时只记录警告而不中断主流程。检索用量统计# Get session-level usage (all branches) session_usage await session.get_session_usage() if session_usage: print(fTotal requests: {session_usage[requests]}) print(fTotal tokens: {session_usage[total_tokens]}) print(fInput tokens: {session_usage[input_tokens]}) print(fOutput tokens: {session_usage[output_tokens]}) print(fTotal turns: {session_usage[total_turns]}) # Get usage for specific branch branch_usage await session.get_session_usage(branch_idmain) # Get usage by turn turn_usage await session.get_turn_usage() for turn_data in turn_usage: print(fTurn {turn_data[user_turn_number]}: {turn_data[total_tokens]} tokens) if turn_data[input_tokens_details]: print(f Input details: {turn_data[input_tokens_details]}) if turn_data[output_tokens_details]: print(f Output details: {turn_data[output_tokens_details]}) # Get usage for specific turn turn_2_usage await session.get_turn_usage(user_turn_number2)两个检索方法的关键语义对应 get_session_usage 与 get_turn_usageget_session_usage(branch_idNone)不传branch_id时跨所有分支聚合传入时只统计指定分支返回字典含requests、input_tokens、output_tokens、total_tokens、total_turns五个字段无数据时返回Noneget_turn_usage(user_turn_numberNone, branch_idNone)不指定轮号时返回当前分支全部轮次的列表按user_turn_number升序指定轮号时返回单个轮次的字典JSON 明细列会在读取时自动json.loads反序列化解析失败则保留None。对话分支Conversation branching分支是本方案最核心的能力你可以从任意一条用户消息分叉出一条新的对话时间线历史消息复制到新分支之后的新对话在新分支中独立演进。创建分支# Get available turns for branching turns await session.get_conversation_turns() for turn in turns: print(fTurn {turn[turn]}: {turn[content]}) print(fCan branch: {turn[can_branch]}) # Create a branch from turn 2 branch_id await session.create_branch_from_turn(2) print(fCreated branch: {branch_id}) # Create a branch with custom name branch_id await session.create_branch_from_turn( 2, branch_namealternative_path ) # Create branch by searching for content branch_id await session.create_branch_from_content( weather, branch_nameweather_focus )分支 ID 在会话 ID 的生命周期内是唯一的删除分支或清空会话会移除其对话数据但不会让已使用过的分支 ID 重新可用创建新分支时应使用新的名称。从源码看分支创建流程create_branch_from_turn → 复制逻辑大致是以BEGIN IMMEDIATE获取 SQLite 写锁防止多进程并发通过同一校验校验指定轮号确实包含用户消息否则抛出ValueError通过branch_reservations表原子性地保留新分支 ID见 _reserve_branch_id自定义名称已被占用会立即报错自动命名格式为branch_from_turn_{轮号}_{时间戳}冲突时自动追加_1、_2后缀把源分支中branch_turn_number 分支点轮号的所有消息结构行复制到新分支消息数据本身message_data通过复用同一message_id共享不产生数据冗余提交后在锁内更新分支指针。create_branch_from_content(search_term, branch_name)则先通过find_turns_by_content找到第一条内容匹配的用户轮次再转调create_branch_from_turn没有任何匹配时抛出ValueError。分支管理# List all branches branches await session.list_branches() for branch in branches: current (current) if branch[is_current] else print(f{branch[branch_id]}: {branch[user_turns]} turns, {branch[message_count]} messages{current}) # Switch between branches await session.switch_to_branch(main) await session.switch_to_branch(branch_id) # Delete a branch await session.delete_branch(branch_id, forceTrue) # forceTrue allows deleting current branch源码中的行为细节对应 list_branches、switch_to_branch、delete_branchlist_branches()返回每条分支的branch_id、message_count消息总数、user_turns用户轮次数按message_type user计数、is_current是否为当前分支与created_atswitch_to_branch()会先校验分支存在message_structure中计数为 0 则抛ValueError再在锁内以清除代数generation守卫更新分支指针若期间发生了clear_session则指针复位为main的语义优先delete_branch()拒绝删除main分支也不允许无force时删除当前分支forceTrue会先自动切回main再删除删除时按顺序清理turn_usage、message_structure并通过_cleanup_orphaned_messages_sync删除不再被任何分支引用的孤儿消息行最后日志输出各表删除条数。分支工作流示例# Original conversation result await Runner.run(agent, Whats the capital of France?, sessionsession) await session.store_run_usage(result) result await Runner.run(agent, Whats the weather like there?, sessionsession) await session.store_run_usage(result) # Create branch from turn 2 (weather question) branch_id await session.create_branch_from_turn(2, weather_focus) # Continue in new branch with different question result await Runner.run( agent, What are the main tourist attractions in Paris?, sessionsession ) await session.store_run_usage(result) # Switch back to main branch await session.switch_to_branch(main) # Continue original conversation result await Runner.run( agent, How expensive is it to visit?, sessionsession ) await session.store_run_usage(result)可以看到创建分支后会话指针自动落在新分支上Runner.run会基于新分支的历史继续switch_to_branch(main)后又能回到原时间线继续推进。两条时间线各自独立演进、互不干扰很适合同一问题多方案推演用户编辑对话历史等场景。结构化查询Structured queries会话分析# Get conversation organized by turns conversation_by_turns await session.get_conversation_by_turns() for turn_num, items in conversation_by_turns.items(): print(fTurn {turn_num}: {len(items)} items) for item in items: if item[tool_name]: print(f - {item[type]} (tool: {item[tool_name]})) else: print(f - {item[type]}) # Get tool usage statistics tool_usage await session.get_tool_usage() for tool_name, count, turn in tool_usage: print(f{tool_name}: used {count} times in turn {turn}) # Find turns by content matching_turns await session.find_turns_by_content(weather) for turn in matching_turns: print(fTurn {turn[turn]}: {turn[content]})各方法语义对应 get_conversation_by_turns、get_tool_usage、find_turns_by_content、get_conversation_turnsget_conversation_by_turns(branch_idNone)返回{轮号: [{type: ..., tool_name: ...}, ...]}按sequence_number排序用于宏观把握每轮的消息构成get_tool_usage(branch_idNone)返回(tool_name, 次数, 轮号)三元组列表。源码对tool_call、function_call、computer_call、file_search_call、web_search_call、code_interpreter_call、tool_search_call、custom_tool_call、mcp_call、mcp_approval_request等类型统一计数并对没有配套tool_search_call的tool_search_output做去重避免重复统计find_turns_by_content(search_term, branch_idNone)基于message_data LIKE %term%在用户消息中做子串匹配返回与get_conversation_turns同结构的结果含turn、content、full_content、timestamp、can_branchget_conversation_turns(branch_idNone)按branch_turn_number升序列出所有用户轮次content为最多 100 字符的预览超出截断加...full_content为完整内容can_branch恒为True所有用户消息都可作为分支点。消息结构Message structure会话在每次add_items写入消息时会在同一个事务里同步写入message_structure元数据见 add_items 实现 与 元数据写入自动跟踪以下信息消息类型取值user、assistant、tool_call、function_call、mcp_call等见 类型分类工具调用的工具名含 MCP 工具的server_label.tool_name格式以及computer_call、web_search_call等无name字段类型的推导见 工具名提取轮次号user_turn_number按分支独立计数与全局顺序号sequence_number分支归属branch_id时间戳created_at。写入消息与写入元数据处于同一事务_insert_items_insert_structure_metadata后统一commit因此元数据失败不会留下孤儿消息行_cleanup_orphaned_messages还会兜底清理历史路径可能残留的孤儿数据。数据库 SchemaAdvancedSQLiteSession在基础 SQLite schemaagent_sessions、agent_messages两表之上额外增加了高级结构表。文档给出了其中三张表的结构源码见 建表逻辑中还包含第四张session_clear_generations表用于跨实例的清除协调。message_structure 表记录每条消息的结构元数据是分支与结构化查询的数据基础CREATE TABLE message_structure ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, message_id INTEGER NOT NULL, branch_id TEXT NOT NULL DEFAULT main, message_type TEXT NOT NULL, sequence_number INTEGER NOT NULL, user_turn_number INTEGER, branch_turn_number INTEGER, tool_name TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES agent_sessions(session_id) ON DELETE CASCADE, FOREIGN KEY (message_id) REFERENCES agent_messages(id) ON DELETE CASCADE );sequence_number为会话内全局单调递增的排序号保证跨分支的稳定顺序user_turn_number按分支独立计数源码中每个分支分配轮次号而非全局轮次号见 元数据写入因此同一条原始消息在不同分支里保持相同的轮次号branch_turn_number用于定位分支点create_branch_from_turn依据branch_turn_number 分支点轮号复制历史基础表被清空时此表数据通过ON DELETE CASCADE联动清理源码额外在clear_session中显式DELETE因为 SQLite 默认不启用外键约束见 clear_session。源码在建表后还会创建 4 个索引以加速查询(session_id, sequence_number)、(session_id, branch_id)、(session_id, branch_id, user_turn_number)、(session_id, branch_id, sequence_number)。branch_reservations 表CREATE TABLE branch_reservations ( session_id TEXT NOT NULL, branch_id TEXT NOT NULL, PRIMARY KEY (session_id, branch_id) );该表原子性地保留分支 ID包括复制前缀为空的分支。保留行在分支被删除、会话被清空后依然保留因此过期的会话实例无法把历史合并到后来复用同一 ID 的分支上。此外_ensure_branch_reservations_table还会在写入前把历史遗留的、已在message_structure中出现但尚未登记的分支 ID 自动回填backfill保证保留语义对旧数据同样生效。turn_usage 表CREATE TABLE turn_usage ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, branch_id TEXT NOT NULL DEFAULT main, user_turn_number INTEGER NOT NULL, requests INTEGER DEFAULT 0, input_tokens INTEGER DEFAULT 0, output_tokens INTEGER DEFAULT 0, total_tokens INTEGER DEFAULT 0, input_tokens_details JSON, output_tokens_details JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES agent_sessions(session_id) ON DELETE CASCADE, UNIQUE(session_id, branch_id, user_turn_number) );UNIQUE(session_id, branch_id, user_turn_number)保证每个分支的每一轮至多一条用量记录store_run_usage采用INSERT OR REPLACE写入同轮多次运行会覆盖更新input_tokens_details/output_tokens_details为 JSON 列保存Usage对象明细的序列化结果见 写入实现读取时自动反序列化表上建有(session_id, branch_id, user_turn_number)索引pop_item弹出某轮最后一条消息时会同步删除该轮已失效的turn_usage行保证用量统计不会报告一个已不存在的轮次见 pop_item 实现。第四张表session_clear_generationsCREATE TABLE session_clear_generations ( session_id TEXT PRIMARY KEY, generation INTEGER NOT NULL DEFAULT 0 );这是源码中为跨实例一致性引入的表文档正文未列出每次clear_session会将该会话的generation加 1其他会话实例在切换分支、创建分支或写入前会校验本地缓存的generation发现不匹配即把本地分支指针复位为main见 分支指针提交 与 外部清除刷新从而防止过期的会话实例在会话被清空后继续向已删除的分支写入。完整示例仓库提供了覆盖全部特性的完整可运行示例examples/memory/advanced_sqlite_session_example.py。它演示了四大部分基础会话记忆多轮对话自动记忆每轮后store_run_usage用量与结构分析get_session_usage、get_turn_usage、get_conversation_by_turns、get_tool_usage并展示get_items输出当前分支全部消息对话分支从第 2 轮创建分支后改问其他问题验证新分支只继承分支点之前的历史、之后完全独立分支管理list_branches、switch_to_branch(main)/switch_to_branch(branch_id)来回切换最终对比两个分支的消息数量验证分支完全隔离。示例中还注册了一个get_weather工具见 示例代码用于演示带工具调用的轮次在结构元数据与工具统计中的表现。运行方式python examples/memory/advanced_sqlite_session_example.pyAPI 参考AdvancedSQLiteSession— 主类继承SQLiteSession新增分支、用量与结构化查询能力SQLiteSession— 基础 SQLite 会话实现父类Session— 会话协议/抽象基类SessionABC自定义会话后端需遵循的接口含get_items、add_items、pop_item、clear_session四个历史方法见 docs/sessions/index.md 的自定义实现章节相关基础文档Sessions 总览、SQLAlchemy Sessions、Encrypted Sessions。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 openai-agents-python 高级 SQLite 会话（AdvancedSQLiteSession）指南：对话分支、Token 用量分析与结构化查询
openai-agents-python 高级 SQLite 会话AdvancedSQLiteSession指南对话分支、Token 用量分析与结构化查询【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonAdvancedSQLiteSession是 openai-agents-python 中基于 SQLite 的增强版会话存储实现在基础SQLiteSession之上额外提供对话分支conversation branching、逐轮 Token 用量统计和结构化会话查询三大能力。本文将以 官方文档 为主体结合 源码实现 与 完整示例带你从初始化、用量追踪、分支管理到数据库表结构完整掌握这套适合构建带对话编辑 / 多时间线能力的多轮 Agent 应用的会话方案。特性总览与基础SQLiteSession相比AdvancedSQLiteSession提供的核心能力包括对话分支Conversation branching可从任意一条用户消息出发创建替代对话路径探索不同的后续走向各分支互不干扰用量追踪Usage tracking按轮turn记录详细的 Token 用量分析支持完整 JSON 明细input_tokens_details/output_tokens_details结构化查询Structured queries按轮获取对话、统计工具调用频次、按内容检索轮次等分支管理Branch management独立的分支切换、列举、删除分支 ID 在会话生命周期内唯一保留消息结构元数据Message structure metadata自动记录消息类型、工具名、轮次号、序号与分支归属。从仓库的会话选型表见 docs/sessions/index.md可以看到它的定位SQLite plus branching/analytics功能集更重见专门页面。也就是说当你只需要简单的历史记忆时用基础的SQLiteSession即可当你需要分支探索、用量分析和结构检索时才选择本方案。快速开始最基本的用法与基础会话几乎一致只需把会话类换成AdvancedSQLiteSession并在每次Runner.run之后调用store_run_usage持久化用量数据from agents import Agent, Runner from agents.extensions.memory import AdvancedSQLiteSession # Create agent agent Agent( nameAssistant, instructionsReply very concisely., ) # Create an advanced session session AdvancedSQLiteSession( session_idconversation_123, db_pathconversations.db, create_tablesTrue ) # First conversation turn result await Runner.run( agent, What city is the Golden Gate Bridge in?, sessionsession ) print(result.final_output) # San Francisco # IMPORTANT: Store usage data await session.store_run_usage(result) # Continue conversation result await Runner.run( agent, What state is it in?, sessionsession ) print(result.final_output) # California await session.store_run_usage(result)Runner.run传入session参数后SDK 会在每次运行前自动读取会话历史、运行后自动写回新增消息用户输入、助手回复、工具调用等无需手动调用.to_input_list()。需要提醒的是会话记忆与运行级续接选项conversation_id、previous_response_id、auto_previous_response_id在同一轮运行中不能混用参见 docs/sessions/index.md。初始化与参数AdvancedSQLiteSession的构造函数完全以关键字参数调用与基础类保持一致的接口风格from agents.extensions.memory import AdvancedSQLiteSession # Basic initialization session AdvancedSQLiteSession( session_idmy_conversation, create_tablesTrue # Auto-create advanced tables ) # With persistent storage session AdvancedSQLiteSession( session_iduser_123, db_pathconversations.db, create_tablesTrue ) # With custom logger import logging logger logging.getLogger(my_app) session AdvancedSQLiteSession( session_idsession_456, create_tablesTrue, loggerlogger )参数说明参数类型说明session_idstr会话的唯一标识用于区分不同对话db_pathstr \| PathSQLite 数据库文件路径默认:memory:内存存储进程结束即丢失create_tablesbool是否自动创建高级结构表默认Falseloggerlogging.Logger \| None会话使用的自定义日志器默认使用模块级日志器源码级的补充说明从 类定义 可以看到AdvancedSQLiteSession直接继承自SQLiteSession构造函数还透传session_settings与**kwargs给父类父类额外支持sessions_table/messages_table自定义基础表名默认分别为agent_sessions与agent_messages见 sqlite_session.py。create_tables的行为在 初始化逻辑 中有两类路径create_tablesTrue在BEGIN IMMEDIATE事务中先建基础表再创建全部高级结构表message_structure、turn_usage、branch_reservations、session_clear_generations及索引create_tablesFalse不建表而是先认领claim已有结构表——源码会通过PRAGMA foreign_key_list校验结构表的外键归属确认这些表确实由当前配置的基础表对创建否则抛出ValueError提示先用create_tablesTrue初始化并认领数据库。这种所有权校验见 认领逻辑是为了防止两个不同配置的会话实例共享同一个数据库文件时把message_structure的message_id外键错误地关联到另一对sessions_table/messages_table上。因此实操上请遵守源码的提示每个sessions_table/messages_table组合使用独立的db_path。另外db_path为:memory:时进程内共享连接以规避线程隔离问题为文件路径时采用线程本地连接并获得进程内的文件级RLock见 sqlite_session.py同一文件被多个会话共享时会复用同一把锁。用量追踪Usage trackingAdvancedSQLiteSession通过store_run_usage把每次运行的 Token 用量按轮写入turn_usage表并可在需要时聚合出会话级统计。注意这完全依赖每次 Agent 运行后调用store_run_usage方法如果不调用turn_usage表将没有任何数据get_session_usage会返回None。存储用量数据# After each agent run, store the usage data result await Runner.run(agent, Hello, sessionsession) await session.store_run_usage(result) # This stores: # - Total tokens used # - Input/output token breakdown # - Request count # - Detailed JSON token information (if available)从 store_run_usage 的实现 看写入前会以当前轮第一条message_structure行的自增 ID作为turn_anchor锚点写入事务提交前会再次校验该锚点行是否仍存在见 内部写入。这意味着如果该轮已被删除、或轮号被新轮复用这条用量写入会被安全跳过绝不会把用量错误地记到一个已不存在的轮次上也不会计入幻影轮 0。序列化 Token 明细失败时只记录警告而不中断主流程。检索用量统计# Get session-level usage (all branches) session_usage await session.get_session_usage() if session_usage: print(fTotal requests: {session_usage[requests]}) print(fTotal tokens: {session_usage[total_tokens]}) print(fInput tokens: {session_usage[input_tokens]}) print(fOutput tokens: {session_usage[output_tokens]}) print(fTotal turns: {session_usage[total_turns]}) # Get usage for specific branch branch_usage await session.get_session_usage(branch_idmain) # Get usage by turn turn_usage await session.get_turn_usage() for turn_data in turn_usage: print(fTurn {turn_data[user_turn_number]}: {turn_data[total_tokens]} tokens) if turn_data[input_tokens_details]: print(f Input details: {turn_data[input_tokens_details]}) if turn_data[output_tokens_details]: print(f Output details: {turn_data[output_tokens_details]}) # Get usage for specific turn turn_2_usage await session.get_turn_usage(user_turn_number2)两个检索方法的关键语义对应 get_session_usage 与 get_turn_usageget_session_usage(branch_idNone)不传branch_id时跨所有分支聚合传入时只统计指定分支返回字典含requests、input_tokens、output_tokens、total_tokens、total_turns五个字段无数据时返回Noneget_turn_usage(user_turn_numberNone, branch_idNone)不指定轮号时返回当前分支全部轮次的列表按user_turn_number升序指定轮号时返回单个轮次的字典JSON 明细列会在读取时自动json.loads反序列化解析失败则保留None。对话分支Conversation branching分支是本方案最核心的能力你可以从任意一条用户消息分叉出一条新的对话时间线历史消息复制到新分支之后的新对话在新分支中独立演进。创建分支# Get available turns for branching turns await session.get_conversation_turns() for turn in turns: print(fTurn {turn[turn]}: {turn[content]}) print(fCan branch: {turn[can_branch]}) # Create a branch from turn 2 branch_id await session.create_branch_from_turn(2) print(fCreated branch: {branch_id}) # Create a branch with custom name branch_id await session.create_branch_from_turn( 2, branch_namealternative_path ) # Create branch by searching for content branch_id await session.create_branch_from_content( weather, branch_nameweather_focus )分支 ID 在会话 ID 的生命周期内是唯一的删除分支或清空会话会移除其对话数据但不会让已使用过的分支 ID 重新可用创建新分支时应使用新的名称。从源码看分支创建流程create_branch_from_turn → 复制逻辑大致是以BEGIN IMMEDIATE获取 SQLite 写锁防止多进程并发通过同一校验校验指定轮号确实包含用户消息否则抛出ValueError通过branch_reservations表原子性地保留新分支 ID见 _reserve_branch_id自定义名称已被占用会立即报错自动命名格式为branch_from_turn_{轮号}_{时间戳}冲突时自动追加_1、_2后缀把源分支中branch_turn_number 分支点轮号的所有消息结构行复制到新分支消息数据本身message_data通过复用同一message_id共享不产生数据冗余提交后在锁内更新分支指针。create_branch_from_content(search_term, branch_name)则先通过find_turns_by_content找到第一条内容匹配的用户轮次再转调create_branch_from_turn没有任何匹配时抛出ValueError。分支管理# List all branches branches await session.list_branches() for branch in branches: current (current) if branch[is_current] else print(f{branch[branch_id]}: {branch[user_turns]} turns, {branch[message_count]} messages{current}) # Switch between branches await session.switch_to_branch(main) await session.switch_to_branch(branch_id) # Delete a branch await session.delete_branch(branch_id, forceTrue) # forceTrue allows deleting current branch源码中的行为细节对应 list_branches、switch_to_branch、delete_branchlist_branches()返回每条分支的branch_id、message_count消息总数、user_turns用户轮次数按message_type user计数、is_current是否为当前分支与created_atswitch_to_branch()会先校验分支存在message_structure中计数为 0 则抛ValueError再在锁内以清除代数generation守卫更新分支指针若期间发生了clear_session则指针复位为main的语义优先delete_branch()拒绝删除main分支也不允许无force时删除当前分支forceTrue会先自动切回main再删除删除时按顺序清理turn_usage、message_structure并通过_cleanup_orphaned_messages_sync删除不再被任何分支引用的孤儿消息行最后日志输出各表删除条数。分支工作流示例# Original conversation result await Runner.run(agent, Whats the capital of France?, sessionsession) await session.store_run_usage(result) result await Runner.run(agent, Whats the weather like there?, sessionsession) await session.store_run_usage(result) # Create branch from turn 2 (weather question) branch_id await session.create_branch_from_turn(2, weather_focus) # Continue in new branch with different question result await Runner.run( agent, What are the main tourist attractions in Paris?, sessionsession ) await session.store_run_usage(result) # Switch back to main branch await session.switch_to_branch(main) # Continue original conversation result await Runner.run( agent, How expensive is it to visit?, sessionsession ) await session.store_run_usage(result)可以看到创建分支后会话指针自动落在新分支上Runner.run会基于新分支的历史继续switch_to_branch(main)后又能回到原时间线继续推进。两条时间线各自独立演进、互不干扰很适合同一问题多方案推演用户编辑对话历史等场景。结构化查询Structured queries会话分析# Get conversation organized by turns conversation_by_turns await session.get_conversation_by_turns() for turn_num, items in conversation_by_turns.items(): print(fTurn {turn_num}: {len(items)} items) for item in items: if item[tool_name]: print(f - {item[type]} (tool: {item[tool_name]})) else: print(f - {item[type]}) # Get tool usage statistics tool_usage await session.get_tool_usage() for tool_name, count, turn in tool_usage: print(f{tool_name}: used {count} times in turn {turn}) # Find turns by content matching_turns await session.find_turns_by_content(weather) for turn in matching_turns: print(fTurn {turn[turn]}: {turn[content]})各方法语义对应 get_conversation_by_turns、get_tool_usage、find_turns_by_content、get_conversation_turnsget_conversation_by_turns(branch_idNone)返回{轮号: [{type: ..., tool_name: ...}, ...]}按sequence_number排序用于宏观把握每轮的消息构成get_tool_usage(branch_idNone)返回(tool_name, 次数, 轮号)三元组列表。源码对tool_call、function_call、computer_call、file_search_call、web_search_call、code_interpreter_call、tool_search_call、custom_tool_call、mcp_call、mcp_approval_request等类型统一计数并对没有配套tool_search_call的tool_search_output做去重避免重复统计find_turns_by_content(search_term, branch_idNone)基于message_data LIKE %term%在用户消息中做子串匹配返回与get_conversation_turns同结构的结果含turn、content、full_content、timestamp、can_branchget_conversation_turns(branch_idNone)按branch_turn_number升序列出所有用户轮次content为最多 100 字符的预览超出截断加...full_content为完整内容can_branch恒为True所有用户消息都可作为分支点。消息结构Message structure会话在每次add_items写入消息时会在同一个事务里同步写入message_structure元数据见 add_items 实现 与 元数据写入自动跟踪以下信息消息类型取值user、assistant、tool_call、function_call、mcp_call等见 类型分类工具调用的工具名含 MCP 工具的server_label.tool_name格式以及computer_call、web_search_call等无name字段类型的推导见 工具名提取轮次号user_turn_number按分支独立计数与全局顺序号sequence_number分支归属branch_id时间戳created_at。写入消息与写入元数据处于同一事务_insert_items_insert_structure_metadata后统一commit因此元数据失败不会留下孤儿消息行_cleanup_orphaned_messages还会兜底清理历史路径可能残留的孤儿数据。数据库 SchemaAdvancedSQLiteSession在基础 SQLite schemaagent_sessions、agent_messages两表之上额外增加了高级结构表。文档给出了其中三张表的结构源码见 建表逻辑中还包含第四张session_clear_generations表用于跨实例的清除协调。message_structure 表记录每条消息的结构元数据是分支与结构化查询的数据基础CREATE TABLE message_structure ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, message_id INTEGER NOT NULL, branch_id TEXT NOT NULL DEFAULT main, message_type TEXT NOT NULL, sequence_number INTEGER NOT NULL, user_turn_number INTEGER, branch_turn_number INTEGER, tool_name TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES agent_sessions(session_id) ON DELETE CASCADE, FOREIGN KEY (message_id) REFERENCES agent_messages(id) ON DELETE CASCADE );sequence_number为会话内全局单调递增的排序号保证跨分支的稳定顺序user_turn_number按分支独立计数源码中每个分支分配轮次号而非全局轮次号见 元数据写入因此同一条原始消息在不同分支里保持相同的轮次号branch_turn_number用于定位分支点create_branch_from_turn依据branch_turn_number 分支点轮号复制历史基础表被清空时此表数据通过ON DELETE CASCADE联动清理源码额外在clear_session中显式DELETE因为 SQLite 默认不启用外键约束见 clear_session。源码在建表后还会创建 4 个索引以加速查询(session_id, sequence_number)、(session_id, branch_id)、(session_id, branch_id, user_turn_number)、(session_id, branch_id, sequence_number)。branch_reservations 表CREATE TABLE branch_reservations ( session_id TEXT NOT NULL, branch_id TEXT NOT NULL, PRIMARY KEY (session_id, branch_id) );该表原子性地保留分支 ID包括复制前缀为空的分支。保留行在分支被删除、会话被清空后依然保留因此过期的会话实例无法把历史合并到后来复用同一 ID 的分支上。此外_ensure_branch_reservations_table还会在写入前把历史遗留的、已在message_structure中出现但尚未登记的分支 ID 自动回填backfill保证保留语义对旧数据同样生效。turn_usage 表CREATE TABLE turn_usage ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, branch_id TEXT NOT NULL DEFAULT main, user_turn_number INTEGER NOT NULL, requests INTEGER DEFAULT 0, input_tokens INTEGER DEFAULT 0, output_tokens INTEGER DEFAULT 0, total_tokens INTEGER DEFAULT 0, input_tokens_details JSON, output_tokens_details JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES agent_sessions(session_id) ON DELETE CASCADE, UNIQUE(session_id, branch_id, user_turn_number) );UNIQUE(session_id, branch_id, user_turn_number)保证每个分支的每一轮至多一条用量记录store_run_usage采用INSERT OR REPLACE写入同轮多次运行会覆盖更新input_tokens_details/output_tokens_details为 JSON 列保存Usage对象明细的序列化结果见 写入实现读取时自动反序列化表上建有(session_id, branch_id, user_turn_number)索引pop_item弹出某轮最后一条消息时会同步删除该轮已失效的turn_usage行保证用量统计不会报告一个已不存在的轮次见 pop_item 实现。第四张表session_clear_generationsCREATE TABLE session_clear_generations ( session_id TEXT PRIMARY KEY, generation INTEGER NOT NULL DEFAULT 0 );这是源码中为跨实例一致性引入的表文档正文未列出每次clear_session会将该会话的generation加 1其他会话实例在切换分支、创建分支或写入前会校验本地缓存的generation发现不匹配即把本地分支指针复位为main见 分支指针提交 与 外部清除刷新从而防止过期的会话实例在会话被清空后继续向已删除的分支写入。完整示例仓库提供了覆盖全部特性的完整可运行示例examples/memory/advanced_sqlite_session_example.py。它演示了四大部分基础会话记忆多轮对话自动记忆每轮后store_run_usage用量与结构分析get_session_usage、get_turn_usage、get_conversation_by_turns、get_tool_usage并展示get_items输出当前分支全部消息对话分支从第 2 轮创建分支后改问其他问题验证新分支只继承分支点之前的历史、之后完全独立分支管理list_branches、switch_to_branch(main)/switch_to_branch(branch_id)来回切换最终对比两个分支的消息数量验证分支完全隔离。示例中还注册了一个get_weather工具见 示例代码用于演示带工具调用的轮次在结构元数据与工具统计中的表现。运行方式python examples/memory/advanced_sqlite_session_example.pyAPI 参考AdvancedSQLiteSession— 主类继承SQLiteSession新增分支、用量与结构化查询能力SQLiteSession— 基础 SQLite 会话实现父类Session— 会话协议/抽象基类SessionABC自定义会话后端需遵循的接口含get_items、add_items、pop_item、clear_session四个历史方法见 docs/sessions/index.md 的自定义实现章节相关基础文档Sessions 总览、SQLAlchemy Sessions、Encrypted Sessions。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考