尧图网站建设 
openai-agents-python 高级 SQLite 会话实战基于 AdvancedSQLiteSession 实现对话分叉、用量分析与结构化查询【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonAdvancedSQLiteSession是 openai-agents-pythonAgents SDK中内置于agents.extensions.memory的会话后端它在基础SQLiteSession之上叠加了对话分叉conversation branching、逐轮 token 用量分析与结构化会话查询三大能力适用于需要多时间线探索、用量审计和会话结构回溯的多轮 Agent 应用。读完本文你将掌握该会话类的完整初始化方式、分叉/切换/删除分支的工作流、按轮与按分支聚合用量的查询方法以及其底层 SQLite 表结构设计与一致性保障机制。一、功能总览AdvancedSQLiteSession继承自基础SQLiteSession源码见 src/agents/extensions/memory/advanced_sqlite_session.py在保留自动读写会话历史能力的同时额外提供以下能力对话分叉Conversation branching可以从任意一条用户消息处创建替代对话路径探索不同的对话走向用量追踪Usage tracking按轮turn记录详细的 token 用量并以 JSON 形式保存完整的输入/输出 token 明细结构化查询Structured queries按轮获取会话、查询工具调用统计、按内容检索轮次等分支管理Branch management独立地列出、切换、删除分支消息结构元数据Message structure metadata自动记录消息类型、工具使用情况与对话流转顺序。从 SDK 内置会话选型表见 docs/sessions/index.md看它定位于SQLite 分支/分析场景比基础SQLiteSession功能更重适合对会话进行编辑、回溯与分析的应用。二、快速开始与基础会话一样AdvancedSQLiteSession直接作为Runner.run(..., sessionsession)的session参数传入即可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在每次运行前自动读取会话历史并拼接到输入运行后自动把本轮新增的输入、输出与工具调用写回会话无需手动调用.to_input_list()用量数据必须手动落库store_run_usage(result)是让用量分析生效的关键调用它读取result.context_wrapper.usage并写入turn_usage表对应源码 store_run_usage。跳过这一步get_session_usage()/get_turn_usage()将查不到数据。三、初始化与参数说明3.1 三种典型初始化方式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 )3.2 参数详解参数类型说明默认值session_idstr会话的唯一标识符必填db_pathstr \| PathSQLite 数据库文件路径:memory:表示纯内存存储进程结束即丢失:memory:create_tablesbool是否自动创建扩展表message_structure、branch_reservations、session_clear_generations、turn_usage及索引Falseloggerlogging.Logger \| None自定义日志器默认使用模块级 logger模块 logger此外构造函数还接受session_settingsSessionSettings或字典见 构造签名用于控制每次运行前检索的历史条数上限并可透传sessions_table/messages_table等底层表名参数继承自SQLiteSession。3.3 create_tables 的底层行为从源码 advanced_sqlite_session.py 可以看到连接初始化时有两种路径create_tablesTrue先执行BEGIN IMMEDIATE建表并初始化结构表create_tablesFalse不建表而是调用_claim_structure_tables校验已有结构表归属。结构表通过外键记录归属message_structure与turn_usage必须指向当前会话配置的sessions_table/messages_table组合。若一个数据库文件已被其他表名组合占用或基础表缺失会抛出带明确提示的ValueError见 源码注释。因此首次使用某个数据库文件时应以create_tablesTrue初始化后续再打开同一文件时可以不建表但要求表结构归属一致。四、用量追踪store_run_usage 与聚合查询4.1 存储用量数据store_run_usage的设计目标是在每次Runner.run()完成后调用它保存总 token 数输入/输出 token 细分请求次数详细的 JSON token 明细如果模型返回了input_tokens_details/output_tokens_details。# 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会先通过_capture_current_turn()锁定读取当前轮次、分支以及该轮首个message_structure行的 IDturn anchor再写入turn_usage。anchor 机制保证即使该轮在写入前被pop_item删除、或轮次编号被复用过期的用量写入也会被跳过不会记到幻影轮次上见 源码注释。4.2 查询会话级与轮次级用量# 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会话级用量并非独立存储而是对turn_usage表按session_id全部分支或session_id branch_id指定分支执行SUM(requests / input_tokens / output_tokens / total_tokens)与COUNT(*)动态聚合而来。get_turn_usage在指定user_turn_number时返回单轮字典否则返回按轮排序的列表并会把 JSON 字段反序列化为 Python 对象见 get_turn_usage。五、会话分叉从任意用户消息开启新对话路径5.1 创建分支# 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 )说明get_conversation_turns()返回当前分支的用户轮次列表每个元素含turn分支内轮号、content截断到 100 字符的预览、full_content、timestamp和恒为True的can_branch字段源码见 get_conversation_turnscreate_branch_from_turn(turn_number, branch_nameNone)以分支内轮号定位用户消息把该轮之前的全部消息复制到新分支并自动切换过去。不传branch_name时会生成branch_from_turn_{轮号}_{时间戳}形式的 ID若同秒冲突则追加数字后缀见 _reserve_branch_idcreate_branch_from_content(search_term, branch_nameNone)先在用户消息中做LIKE %term%匹配取最早命中轮次再走create_branch_from_turn无匹配时抛出ValueError见 create_branch_from_content。5.2 分支 ID 的唯一性与保留语义分支 ID 在 session_id 的整个生命周期内唯一。删除分支或清空会话会删除对应对话数据但不会让已用过的分支 ID 重新可用——这是通过branch_reservations表实现的详见下文 Schema 一节。因此再次创建分支时请使用新名称否则会抛出ValueError。5.3 分支管理# 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()返回各分支的branch_id、message_count、user_turnsmessage_typeuser的消息数、is_current与created_at见 list_branchesswitch_to_branch会先校验分支存在否则抛ValueErrordelete_branch禁止删除main分支删除当前分支时必须forceTrue内部会先切回main。删除时在同一事务中清理该分支的turn_usage与message_structure记录并顺手清理不再被任何分支引用的孤儿消息见 delete_branch。5.4 分支工作流完整示例# 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)注意新分支复制的是分叉点之前的共享消息记录message_structure中同一message_id可被多个分支引用而不是复制消息内容分支之间互不干扰pop_item、delete_branch只影响当前/目标分支仅在消息不再被任何分支引用时才删除底层数据见 pop_item 实现注释。六、结构化查询与会话分析6.1 会话分析# 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()返回{轮号: [{type: 消息类型, tool_name: 工具名或 None}, ...]}字典按sequence_number排序见 get_conversation_by_turnsget_tool_usage()返回(tool_name, count, turn_number)三元组列表统计范围涵盖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等消息类型见 get_tool_usagefind_turns_by_content(term)按LIKE模糊匹配用户消息内容返回与get_conversation_turns()相同格式的轮次列表见 find_turns_by_content。6.2 消息结构元数据会话在写入每条消息时会自动记录见 _insert_structure_metadata消息类型值user、assistant、tool_call、function_call等分类逻辑见 _classify_message_type工具调用的工具名对 MCP 调用会拼成server_label.tool_name形式对computer_call、file_search_call等无name字段的类型直接取类型名见 _extract_tool_name轮次号与序列号user_turn_number该分支内随用户消息递增的轮号与全局递增的sequence_number用于精确排序分支关联每条结构记录都带有branch_id时间戳created_at。七、数据库 Schema三张扩展表与一张保障表AdvancedSQLiteSession在基础 SQLite Schema 之上扩展了结构表。其中message_structure与turn_usage通过外键声明归属另有一张session_clear_generations表用于跨实例清空检测在创建时一并生成见 _init_structure_tables。7.1 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 );该表是分支机制的枢纽同一message_id可出现在多行多个分支共享底层消息通过branch_id sequence_number精确定位每个分支内消息的顺序与归属。创建时还会自动建立 4 个索引idx_structure_session_seq、idx_structure_branch、idx_structure_turn、idx_structure_branch_seq分别加速按会话/分支/轮次的查询。7.2 branch_reservations 表CREATE TABLE branch_reservations ( session_id TEXT NOT NULL, branch_id TEXT NOT NULL, PRIMARY KEY (session_id, branch_id) );这张表原子地保留分支 ID包括那些复制前缀为空的分支。保留行在分支被删除、甚至会话被清空后依然存在因此过期的会话实例无法把历史合并进后续复用了同一 ID 的新分支。从源码看创建分支时通过INSERT OR IGNORE插入保留行若rowcount 0说明 ID 已被占用并抛出ValueError自动生成 ID 时也会循环探测未占用的名称见 _reserve_branch_id。7.3 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 文本保存模型的 token 明细对象。7.4 session_clear_generations 表内部保障CREATE TABLE session_clear_generations ( session_id TEXT PRIMARY KEY, generation INTEGER NOT NULL DEFAULT 0 );该表记录每次clear_session()时递增的代数。所有写操作与分支指针切换都会先比对代数一旦检测到其他实例已清空会话本地分支指针会重置回main从而防止过期的写操作或分支切换污染新会话见 _commit_branch_pointer 与 clear_session。八、完整示例与测试验证仓库提供了覆盖全部特性的可运行示例examples/memory/advanced_sqlite_session_example.py。该示例分为四部分基础会话记忆连续三轮对话金门大桥所在城市 → 天气 → 人口每轮调用store_run_usage用量与分析打印get_session_usage()、get_turn_usage()、get_conversation_by_turns()与get_tool_usage()结果对话分叉从第 2 轮创建分支后改问纽约天气与景点验证新分支只包含分叉点之前的内容分支管理list_branches()对比主分支与分叉分支的消息数switch_to_branch往返切换证明分支完全独立、互不干扰。对应的单元测试位于 tests/extensions/memory/test_advanced_sqlite_session.py覆盖了基础功能、消息结构追踪、工具用量统计、分叉功能、分支保留语义、跨进程分支分配串行化、store_run_usage的陈旧写入跳过、清空会话重置分支等场景如test_branch_ids_remain_reserved_after_delete_and_clear、test_clear_session_removes_structure_and_usage_metadata、test_stale_store_run_usage_skipped_when_turn_removed_by_pop。这些测试同时验证了文中提到的分支隔离、保留语义与原子性保障。九、API 参考AdvancedSQLiteSession主类继承自SQLiteSession入口位于agents.extensions.memory见 src/agents/extensions/memory/init.pySession协议所有会话后端的公共接口定义session_id、session_settings以及get_items/add_items/pop_item/clear_session四个历史方法详见 docs/sessions/index.md。十、适用场景与注意事项小结何时选用需要多时间线探索例如假设从某轮换一种问法、需要逐轮成本/用量审计、需要结构化回溯会话工具调用统计、按内容检索轮次时AdvancedSQLiteSession是内置方案中直接可用的选择若仅需轻量自动记忆基础SQLiteSession更简洁。务必记得每次Runner.run()后调用store_run_usage(result)否则用量分析为空。分支 ID 一经使用即被保留删除/清空后不可复用请用新名称创建分支。首开数据库用create_tablesTrue后续以create_tablesFalse复用同一文件时结构表归属必须与既有配置一致否则抛出ValueError。会话与运行级续接互斥在同一轮运行中会话模式不能与conversation_id、previous_response_id、auto_previous_response_id同时使用见 docs/sessions/index.md若运行因审批暂停请用相同 session_id 与同一存储后端恢复以保证历史延续。【免费下载链接】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 实现对话分叉、用量分析与结构化查询
openai-agents-python 高级 SQLite 会话实战基于 AdvancedSQLiteSession 实现对话分叉、用量分析与结构化查询【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonAdvancedSQLiteSession是 openai-agents-pythonAgents SDK中内置于agents.extensions.memory的会话后端它在基础SQLiteSession之上叠加了对话分叉conversation branching、逐轮 token 用量分析与结构化会话查询三大能力适用于需要多时间线探索、用量审计和会话结构回溯的多轮 Agent 应用。读完本文你将掌握该会话类的完整初始化方式、分叉/切换/删除分支的工作流、按轮与按分支聚合用量的查询方法以及其底层 SQLite 表结构设计与一致性保障机制。一、功能总览AdvancedSQLiteSession继承自基础SQLiteSession源码见 src/agents/extensions/memory/advanced_sqlite_session.py在保留自动读写会话历史能力的同时额外提供以下能力对话分叉Conversation branching可以从任意一条用户消息处创建替代对话路径探索不同的对话走向用量追踪Usage tracking按轮turn记录详细的 token 用量并以 JSON 形式保存完整的输入/输出 token 明细结构化查询Structured queries按轮获取会话、查询工具调用统计、按内容检索轮次等分支管理Branch management独立地列出、切换、删除分支消息结构元数据Message structure metadata自动记录消息类型、工具使用情况与对话流转顺序。从 SDK 内置会话选型表见 docs/sessions/index.md看它定位于SQLite 分支/分析场景比基础SQLiteSession功能更重适合对会话进行编辑、回溯与分析的应用。二、快速开始与基础会话一样AdvancedSQLiteSession直接作为Runner.run(..., sessionsession)的session参数传入即可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在每次运行前自动读取会话历史并拼接到输入运行后自动把本轮新增的输入、输出与工具调用写回会话无需手动调用.to_input_list()用量数据必须手动落库store_run_usage(result)是让用量分析生效的关键调用它读取result.context_wrapper.usage并写入turn_usage表对应源码 store_run_usage。跳过这一步get_session_usage()/get_turn_usage()将查不到数据。三、初始化与参数说明3.1 三种典型初始化方式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 )3.2 参数详解参数类型说明默认值session_idstr会话的唯一标识符必填db_pathstr \| PathSQLite 数据库文件路径:memory:表示纯内存存储进程结束即丢失:memory:create_tablesbool是否自动创建扩展表message_structure、branch_reservations、session_clear_generations、turn_usage及索引Falseloggerlogging.Logger \| None自定义日志器默认使用模块级 logger模块 logger此外构造函数还接受session_settingsSessionSettings或字典见 构造签名用于控制每次运行前检索的历史条数上限并可透传sessions_table/messages_table等底层表名参数继承自SQLiteSession。3.3 create_tables 的底层行为从源码 advanced_sqlite_session.py 可以看到连接初始化时有两种路径create_tablesTrue先执行BEGIN IMMEDIATE建表并初始化结构表create_tablesFalse不建表而是调用_claim_structure_tables校验已有结构表归属。结构表通过外键记录归属message_structure与turn_usage必须指向当前会话配置的sessions_table/messages_table组合。若一个数据库文件已被其他表名组合占用或基础表缺失会抛出带明确提示的ValueError见 源码注释。因此首次使用某个数据库文件时应以create_tablesTrue初始化后续再打开同一文件时可以不建表但要求表结构归属一致。四、用量追踪store_run_usage 与聚合查询4.1 存储用量数据store_run_usage的设计目标是在每次Runner.run()完成后调用它保存总 token 数输入/输出 token 细分请求次数详细的 JSON token 明细如果模型返回了input_tokens_details/output_tokens_details。# 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会先通过_capture_current_turn()锁定读取当前轮次、分支以及该轮首个message_structure行的 IDturn anchor再写入turn_usage。anchor 机制保证即使该轮在写入前被pop_item删除、或轮次编号被复用过期的用量写入也会被跳过不会记到幻影轮次上见 源码注释。4.2 查询会话级与轮次级用量# 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会话级用量并非独立存储而是对turn_usage表按session_id全部分支或session_id branch_id指定分支执行SUM(requests / input_tokens / output_tokens / total_tokens)与COUNT(*)动态聚合而来。get_turn_usage在指定user_turn_number时返回单轮字典否则返回按轮排序的列表并会把 JSON 字段反序列化为 Python 对象见 get_turn_usage。五、会话分叉从任意用户消息开启新对话路径5.1 创建分支# 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 )说明get_conversation_turns()返回当前分支的用户轮次列表每个元素含turn分支内轮号、content截断到 100 字符的预览、full_content、timestamp和恒为True的can_branch字段源码见 get_conversation_turnscreate_branch_from_turn(turn_number, branch_nameNone)以分支内轮号定位用户消息把该轮之前的全部消息复制到新分支并自动切换过去。不传branch_name时会生成branch_from_turn_{轮号}_{时间戳}形式的 ID若同秒冲突则追加数字后缀见 _reserve_branch_idcreate_branch_from_content(search_term, branch_nameNone)先在用户消息中做LIKE %term%匹配取最早命中轮次再走create_branch_from_turn无匹配时抛出ValueError见 create_branch_from_content。5.2 分支 ID 的唯一性与保留语义分支 ID 在 session_id 的整个生命周期内唯一。删除分支或清空会话会删除对应对话数据但不会让已用过的分支 ID 重新可用——这是通过branch_reservations表实现的详见下文 Schema 一节。因此再次创建分支时请使用新名称否则会抛出ValueError。5.3 分支管理# 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()返回各分支的branch_id、message_count、user_turnsmessage_typeuser的消息数、is_current与created_at见 list_branchesswitch_to_branch会先校验分支存在否则抛ValueErrordelete_branch禁止删除main分支删除当前分支时必须forceTrue内部会先切回main。删除时在同一事务中清理该分支的turn_usage与message_structure记录并顺手清理不再被任何分支引用的孤儿消息见 delete_branch。5.4 分支工作流完整示例# 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)注意新分支复制的是分叉点之前的共享消息记录message_structure中同一message_id可被多个分支引用而不是复制消息内容分支之间互不干扰pop_item、delete_branch只影响当前/目标分支仅在消息不再被任何分支引用时才删除底层数据见 pop_item 实现注释。六、结构化查询与会话分析6.1 会话分析# 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()返回{轮号: [{type: 消息类型, tool_name: 工具名或 None}, ...]}字典按sequence_number排序见 get_conversation_by_turnsget_tool_usage()返回(tool_name, count, turn_number)三元组列表统计范围涵盖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等消息类型见 get_tool_usagefind_turns_by_content(term)按LIKE模糊匹配用户消息内容返回与get_conversation_turns()相同格式的轮次列表见 find_turns_by_content。6.2 消息结构元数据会话在写入每条消息时会自动记录见 _insert_structure_metadata消息类型值user、assistant、tool_call、function_call等分类逻辑见 _classify_message_type工具调用的工具名对 MCP 调用会拼成server_label.tool_name形式对computer_call、file_search_call等无name字段的类型直接取类型名见 _extract_tool_name轮次号与序列号user_turn_number该分支内随用户消息递增的轮号与全局递增的sequence_number用于精确排序分支关联每条结构记录都带有branch_id时间戳created_at。七、数据库 Schema三张扩展表与一张保障表AdvancedSQLiteSession在基础 SQLite Schema 之上扩展了结构表。其中message_structure与turn_usage通过外键声明归属另有一张session_clear_generations表用于跨实例清空检测在创建时一并生成见 _init_structure_tables。7.1 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 );该表是分支机制的枢纽同一message_id可出现在多行多个分支共享底层消息通过branch_id sequence_number精确定位每个分支内消息的顺序与归属。创建时还会自动建立 4 个索引idx_structure_session_seq、idx_structure_branch、idx_structure_turn、idx_structure_branch_seq分别加速按会话/分支/轮次的查询。7.2 branch_reservations 表CREATE TABLE branch_reservations ( session_id TEXT NOT NULL, branch_id TEXT NOT NULL, PRIMARY KEY (session_id, branch_id) );这张表原子地保留分支 ID包括那些复制前缀为空的分支。保留行在分支被删除、甚至会话被清空后依然存在因此过期的会话实例无法把历史合并进后续复用了同一 ID 的新分支。从源码看创建分支时通过INSERT OR IGNORE插入保留行若rowcount 0说明 ID 已被占用并抛出ValueError自动生成 ID 时也会循环探测未占用的名称见 _reserve_branch_id。7.3 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 文本保存模型的 token 明细对象。7.4 session_clear_generations 表内部保障CREATE TABLE session_clear_generations ( session_id TEXT PRIMARY KEY, generation INTEGER NOT NULL DEFAULT 0 );该表记录每次clear_session()时递增的代数。所有写操作与分支指针切换都会先比对代数一旦检测到其他实例已清空会话本地分支指针会重置回main从而防止过期的写操作或分支切换污染新会话见 _commit_branch_pointer 与 clear_session。八、完整示例与测试验证仓库提供了覆盖全部特性的可运行示例examples/memory/advanced_sqlite_session_example.py。该示例分为四部分基础会话记忆连续三轮对话金门大桥所在城市 → 天气 → 人口每轮调用store_run_usage用量与分析打印get_session_usage()、get_turn_usage()、get_conversation_by_turns()与get_tool_usage()结果对话分叉从第 2 轮创建分支后改问纽约天气与景点验证新分支只包含分叉点之前的内容分支管理list_branches()对比主分支与分叉分支的消息数switch_to_branch往返切换证明分支完全独立、互不干扰。对应的单元测试位于 tests/extensions/memory/test_advanced_sqlite_session.py覆盖了基础功能、消息结构追踪、工具用量统计、分叉功能、分支保留语义、跨进程分支分配串行化、store_run_usage的陈旧写入跳过、清空会话重置分支等场景如test_branch_ids_remain_reserved_after_delete_and_clear、test_clear_session_removes_structure_and_usage_metadata、test_stale_store_run_usage_skipped_when_turn_removed_by_pop。这些测试同时验证了文中提到的分支隔离、保留语义与原子性保障。九、API 参考AdvancedSQLiteSession主类继承自SQLiteSession入口位于agents.extensions.memory见 src/agents/extensions/memory/init.pySession协议所有会话后端的公共接口定义session_id、session_settings以及get_items/add_items/pop_item/clear_session四个历史方法详见 docs/sessions/index.md。十、适用场景与注意事项小结何时选用需要多时间线探索例如假设从某轮换一种问法、需要逐轮成本/用量审计、需要结构化回溯会话工具调用统计、按内容检索轮次时AdvancedSQLiteSession是内置方案中直接可用的选择若仅需轻量自动记忆基础SQLiteSession更简洁。务必记得每次Runner.run()后调用store_run_usage(result)否则用量分析为空。分支 ID 一经使用即被保留删除/清空后不可复用请用新名称创建分支。首开数据库用create_tablesTrue后续以create_tablesFalse复用同一文件时结构表归属必须与既有配置一致否则抛出ValueError。会话与运行级续接互斥在同一轮运行中会话模式不能与conversation_id、previous_response_id、auto_previous_response_id同时使用见 docs/sessions/index.md若运行因审批暂停请用相同 session_id 与同一存储后端恢复以保证历史延续。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考