2026/9/10 10:47:24
FEATURED · 精选文章 
openai-agents-python 高级 SQLite 会话对话分支、用量分析与结构化查询完整实战指南【免费下载链接】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 用量分析usage analytics与结构化对话查询能力为多智能体应用提供可持久化、可回滚探索的记忆层。本文将以官方文档 docs/sessions/advanced_sqlite_session.md 为骨架结合 src/agents/extensions/memory/advanced_sqlite_session.py 的源码实现与 tests/extensions/memory/test_advanced_sqlite_session.py 的测试用例带你掌握分支创建/切换/删除、按轮统计 token 消耗、按轮次组织对话等核心能力并深入理解其底层数据库设计与并发安全机制。功能特性一览AdvancedSQLiteSession在基础会话能力之上提供五类增强对话分支Conversation branching可以从任意一条用户消息处派生新的对话路径用于如果当时换个问法会怎样式的探索而无需破坏原对话。用量跟踪Usage tracking以轮turn为粒度记录每次运行的 token 消耗包含输入/输出 token 的完整 JSON 明细。结构化查询Structured queries按轮获取对话、统计工具调用次数、按内容检索用户轮次等。分支管理Branch management独立地创建、列出、切换与删除分支分支间互不干扰。消息结构元数据Message structure metadata自动记录消息类型、工具名、轮次号、序列号、分支归属与时间戳。快速开始最小可运行示例创建一个 Agent用AdvancedSQLiteSession承载两轮对话并显式保存用量数据。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.add_items写入消息但用量数据不会自动落库必须手动调用await session.store_run_usage(result)保存。这正是源码中store_run_usage方法的设计定位——它专门设计为在Runner.run()完成后调用见 advanced_sqlite_session.py 的 docstring。初始化与参数三种典型初始化方式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自定义日志器默认使用模块级 logger源码中的额外参数与关键行为从源码构造签名看advanced_sqlite_session.py构造器还接受session_settings: SessionSettings | dict[str, Any] | None用于配置会话的默认检索条数上限等设置对应基类 sqlite_session.py 中的sessions_table/messages_table参数默认分别为agent_sessions与agent_messages。create_tables语义需要特别留意它不只是建表开关还决定了数据库文件的归属权校验方式。从_init_db_for_connection的实现advanced_sqlite_session.py可以看到两条路径create_tablesTrue在BEGIN IMMEDIATE事务中创建基础表 高级表并声明归属权create_tablesFalse先执行_claim_structure_tables校验表结构归属再创建基础表。_claim_structure_tablesadvanced_sqlite_session.py会通过PRAGMA foreign_key_list检查每个高级表的外键是否恰好指向配置的那对基础表高级表不按sessions_table/messages_table命名因此一个数据库文件只能容纳一对基础表的归属。如果多个会话配置了不同的基础表名却共用同一个db_path会抛出ValueError提示 Give each sessions_table/messages_table pair its own db_path。也就是说一个数据库文件应专属一组基础表多个会话共享文件时应保持基础表名一致。另外从构造函数可以看到分支指针的初始化逻辑实例化后self._current_branch_id main默认始终位于main分支。用量跟踪Usage Tracking用量跟踪完全依赖每次运行后调用store_run_usage——官方文档原话强调This is entirely dependent on thestore_run_usagemethod being called after each agent run. 源码通过_capture_current_turnadvanced_sqlite_session.py在加锁状态下读取当前分支当前轮次并用该轮第一条message_structure.id作为turn anchor锚点如果该轮在写入提交前被删除即便后来有同名轮次复用相同数字 id写入会被跳过避免用量错误地归属到已不存在的轮次上。保存用量数据# 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)保存的字段与turn_usage表一一对应requests、input_tokens、output_tokens、total_tokens以及两个 JSON 明细列input_tokens_details/output_tokens_details。明细序列化逻辑位于_update_turn_usage_internaladvanced_sqlite_session.py取Usage对象上input_tokens_details/output_tokens_details的__dict__序列化为 JSON 存储序列化失败时记录警告并置空不影响主数据写入随后以INSERT OR REPLACE写入UNIQUE(session_id, branch_id, user_turn_number)约束保证同一分支同一轮只保留一条记录。查询用量统计# 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)实现要点来自 advanced_sqlite_session.pyget_session_usage(branch_idNone)通过SUM(requests)、SUM(input_tokens)、SUM(output_tokens)、SUM(total_tokens)与COUNT(*)聚合turn_usage不传branch_id时汇总所有分支传入时只统计指定分支。无数据时返回None。get_turn_usage(user_turn_numberNone, branch_idNone)指定轮次时返回单个字典无数据返回空字典{}否则按user_turn_number升序返回字典列表JSON 明细列自动json.loads解析为 Python 对象。两者都通过asyncio.to_thread把同步 SQLite 操作放到工作线程执行避免阻塞事件循环。对话分支Conversation Branching对话分支是AdvancedSQLiteSession的核心能力任意一条用户消息都可以成为新分支的起点从而探索替代对话路径。创建分支# 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 的整个生命周期内保持唯一。删除分支或清空会话会移除其对话数据但不会释放已使用的分支 ID创建新分支时必须使用新名称。源码对这条规则的落实非常严谨。branch_reservations表见下文 Schema承担原子预留职责_reserve_branch_idadvanced_sqlite_session.py对自定义名称执行INSERT OR IGNORE若rowcount 0说明 ID 已被占用抛出ValueError自动命名则生成branch_from_turn_{turn}_{timestamp}基名并循环加后缀直到预留成功。_copy_messages_to_new_branchadvanced_sqlite_session.py在BEGIN IMMEDIATE下先检查分支点是否存在用户消息再把分支点之前branch_turn_number from_turn_number的消息以共享同一message_id的方式复制到新分支仅复制message_structure行消息数据本体不重复存储最后提交并切换当前指针。create_branch_from_contentadvanced_sqlite_session.py内部调用find_turns_by_content取最早匹配的用户轮次作为分支点无匹配时抛出ValueError。测试test_create_branch_from_content验证了这一行为见 test_advanced_sqlite_session.py。分支管理# 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_branchesadvanced_sqlite_session.py按message_structure聚合branch_id用COUNT(*)统计消息数、COUNT(CASE WHEN message_type user ...)统计用户轮数并标记当前分支。switch_to_branchadvanced_sqlite_session.py先校验分支存在message_structure中计数为 0 则抛ValueError再通过_commit_branch_pointer在锁内原子更新指针若期间发生了clear_sessiongeneration 不匹配清空操作的重置优先指针仍回落到main。delete_branchadvanced_sqlite_session.py拒绝删除main分支与空分支 ID删除当前分支时必须forceTrue会自动先切回main删除顺序为先turn_usage再message_structure随后用_cleanup_orphaned_messages_sync清理不再被任何分支引用的孤儿消息——这正是分支间共享消息、删除一个分支不影响其他分支的底层保证。测试test_delete_branch_keeps_messages_still_referenced_by_another_branchtest_advanced_sqlite_session.py专门验证了该行为。分支工作流完整示例# 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传入session后get_items会按当前分支过滤消息WHERE s.branch_id ?因此切换分支即切换对话上下文。官方完整示例 examples/memory/advanced_sqlite_session_example.py 演示了完整的创建分支 → 新分支内多轮对话 → 切回 main → 再切回新分支流程并验证分支完全独立、互不干扰。会话清空与多实例安全值得补充的是源码在文档描述之外还维护了第四张表session_clear_generations见下节。它记录每个会话的清空代数generationclear_sessionadvanced_sqlite_session.py在清空消息、会话、message_structure、turn_usage的同时将 generation 递增任何持有旧 generation 的会话实例在下次操作时都会通过_refresh_branch_after_external_clear检测到不一致并把分支指针重置回main。这防止了旧会话实例在另一个实例清空会话后把历史写入后来复用相同 ID 的分支这一竞态测试test_branch_ids_remain_reserved_after_delete_and_cleartest_advanced_sqlite_session.py与test_branch_allocation_is_serialized_across_processestest_advanced_sqlite_session.py分别覆盖了 ID 预留与跨进程串行化场景。结构化查询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_turnsadvanced_sqlite_session.py按sequence_number升序读取当前分支的message_structure返回{轮次号: [{type, tool_name}, ...]}字典。get_tool_usageadvanced_sqlite_session.py识别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_output去重若同轮已有对应tool_search_call则不重复计数返回(tool_name, count, turn)元组列表。find_turns_by_contentadvanced_sqlite_session.py用message_data LIKE %term%在用户消息中做子串匹配返回与get_conversation_turns相同结构的字典列表。get_conversation_turnsadvanced_sqlite_session.py返回turn、content截断预览默认 100 字符、full_content、timestamp、can_branch恒为True因为所有用户消息都可作为分支点。内容预览由_content_preview实现advanced_sqlite_session.py兼容纯字符串与多模态结构化 content 列表如input_text/input_image超长时截断并追加...。工具名称的提取规则message_structure.tool_name的取值由_extract_tool_nameadvanced_sqlite_session.py决定规则包括MCP 工具mcp_call/mcp_approval_request优先拼接为server_label.tool_name内置工具computer_call、file_search_call、web_search_call、code_interpreter_call、tool_search_call等无name字段直接以消息类型作为工具名其中tool_search_call/tool_search_output统一归为tool_search普通函数调用使用name字段有namespace时通过tool_qualified_name拼成限定名保留合成命名空间如custom_tool_call场景时直接使用裸名。测试test_tool_usage_tracking_preserves_namespaces_and_tool_searchtest_advanced_sqlite_session.py与test_advanced_tool_name_extractiontest_advanced_sqlite_session.py对上述规则做了系统验证。消息结构元数据会话自动跟踪以下消息结构信息消息类型值user、assistant、tool_call等工具调用的工具名称轮次号user_turn_number与序列号sequence_number分支归属关系branch_id时间戳created_at在写入路径上add_itemsadvanced_sqlite_session.py把消息插入与结构元数据插入放在同一个事务中_insert_items_insert_structure_metadata后统一conn.commit()保证元数据写入失败不会留下孤儿消息测试test_add_items_rolls_back_messages_when_structure_metadata_failstest_advanced_sqlite_session.py验证了该原子性。_insert_structure_metadataadvanced_sqlite_session.py为每条消息分配全局递增的sequence_number并按分支内用户消息计数推进user_turn_number与branch_turn_number正确处理单批次含多条用户消息的情况测试见test_conversation_structure_with_multiple_turnstest_advanced_sqlite_session.py。数据库 SchemaAdvancedSQLiteSession在基础 SQLite Schemaagent_sessions、agent_messages两张表定义见 sqlite_session.py之上新增三张业务表外加一张并发控制表。基础库连接默认启用 WAL 模式PRAGMA journal_modeWAL见 sqlite_session.py。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 );源码在创建该表后还会建立 4 个索引见_init_structure_tablesadvanced_sqlite_session.pyidx_structure_session_seq(session_id, sequence_number)idx_structure_branch(session_id, branch_id)idx_structure_turn(session_id, branch_id, user_turn_number)idx_structure_branch_seq(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 的分支中——这是分支 ID 终身唯一性的持久化保证。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)约束保证每个分支的每个轮次只有一条用量记录配套索引idx_turn_usage_session_turn覆盖(session_id, branch_id, user_turn_number)。session_clear_generations 表源码补充文档正文之外源码还维护一张并发控制表advanced_sqlite_session.pyCREATE TABLE session_clear_generations ( session_id TEXT PRIMARY KEY, generation INTEGER NOT NULL DEFAULT 0 );每次clear_session使generation递增用于跨实例失效陈旧的本地分支指针详见前文会话清空与多实例安全。完整示例官方提供了一个覆盖全部特性的综合示例 examples/memory/advanced_sqlite_session_example.py包含四个部分基础会话记忆三轮连续对话 每轮store_run_usage用量跟踪与分析get_session_usage()输出会话级汇总get_turn_usage()输出逐轮 token 数结构化查询get_conversation_by_turns()按轮展示消息与工具调用get_tool_usage()展示工具统计对话分支与分支管理从第 2 轮创建分支、在新分支继续对话、list_branches()列出全部分支、switch_to_branch在主分支与新分支之间往返最后验证分支完全独立、无相互干扰并在结束时session.close()释放连接。示例中使用了tool装饰器注册的get_weather工具运行后可在输出中直观看到tool_usage与get_conversation_by_turns中的tool_name字段。该文件位于examples/memory/目录可直接运行观察完整效果。API 参考AdvancedSQLiteSession—— 主类继承自SQLiteSession提供分支、用量、结构化查询等增强能力Session—— 基础会话协议Protocol定义get_items/add_items/pop_item/clear_session四个异步接口所有会话实现含第三方自定义会话都遵循该结构契约SQLiteSessionsqlite_session.py—— 基础 SQLite 实现AdvancedSQLiteSession的父类负责建表、连接管理与 WAL 配置。适用场景与注意事项多方案对比探索对同一对话在不同轮次分叉出多条路径如换一种提问方式、切换关注点事后随时切回主线继续适合需要保留完整探索轨迹的 Agent 应用。成本与用量审计store_run_usageturn_usage表为按会话、按分支、按轮次的 token 消耗核算提供了精确数据源。对话回放与质检message_structure的轮次/序列/工具元数据可用于还原每一轮的完整调用链。务必显式保存用量只要忘记调用store_run_usageget_session_usage/get_turn_usage将查不到数据返回None或空列表。分支 ID 不可复用删除分支不释放 ID请使用新名称创建分支。数据库文件归属一个db_path对应一组基础表对不同基础表名自定义sessions_table/messages_table的会话应使用各自独立的数据库文件。持久化前提默认db_path:memory:为内存库进程退出即丢失需要跨进程/跨重启保留分支与用量数据时务必传入文件路径。【免费下载链接】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 会话：对话分支、用量分析与结构化查询完整实战指南
openai-agents-python 高级 SQLite 会话对话分支、用量分析与结构化查询完整实战指南【免费下载链接】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 用量分析usage analytics与结构化对话查询能力为多智能体应用提供可持久化、可回滚探索的记忆层。本文将以官方文档 docs/sessions/advanced_sqlite_session.md 为骨架结合 src/agents/extensions/memory/advanced_sqlite_session.py 的源码实现与 tests/extensions/memory/test_advanced_sqlite_session.py 的测试用例带你掌握分支创建/切换/删除、按轮统计 token 消耗、按轮次组织对话等核心能力并深入理解其底层数据库设计与并发安全机制。功能特性一览AdvancedSQLiteSession在基础会话能力之上提供五类增强对话分支Conversation branching可以从任意一条用户消息处派生新的对话路径用于如果当时换个问法会怎样式的探索而无需破坏原对话。用量跟踪Usage tracking以轮turn为粒度记录每次运行的 token 消耗包含输入/输出 token 的完整 JSON 明细。结构化查询Structured queries按轮获取对话、统计工具调用次数、按内容检索用户轮次等。分支管理Branch management独立地创建、列出、切换与删除分支分支间互不干扰。消息结构元数据Message structure metadata自动记录消息类型、工具名、轮次号、序列号、分支归属与时间戳。快速开始最小可运行示例创建一个 Agent用AdvancedSQLiteSession承载两轮对话并显式保存用量数据。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.add_items写入消息但用量数据不会自动落库必须手动调用await session.store_run_usage(result)保存。这正是源码中store_run_usage方法的设计定位——它专门设计为在Runner.run()完成后调用见 advanced_sqlite_session.py 的 docstring。初始化与参数三种典型初始化方式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自定义日志器默认使用模块级 logger源码中的额外参数与关键行为从源码构造签名看advanced_sqlite_session.py构造器还接受session_settings: SessionSettings | dict[str, Any] | None用于配置会话的默认检索条数上限等设置对应基类 sqlite_session.py 中的sessions_table/messages_table参数默认分别为agent_sessions与agent_messages。create_tables语义需要特别留意它不只是建表开关还决定了数据库文件的归属权校验方式。从_init_db_for_connection的实现advanced_sqlite_session.py可以看到两条路径create_tablesTrue在BEGIN IMMEDIATE事务中创建基础表 高级表并声明归属权create_tablesFalse先执行_claim_structure_tables校验表结构归属再创建基础表。_claim_structure_tablesadvanced_sqlite_session.py会通过PRAGMA foreign_key_list检查每个高级表的外键是否恰好指向配置的那对基础表高级表不按sessions_table/messages_table命名因此一个数据库文件只能容纳一对基础表的归属。如果多个会话配置了不同的基础表名却共用同一个db_path会抛出ValueError提示 Give each sessions_table/messages_table pair its own db_path。也就是说一个数据库文件应专属一组基础表多个会话共享文件时应保持基础表名一致。另外从构造函数可以看到分支指针的初始化逻辑实例化后self._current_branch_id main默认始终位于main分支。用量跟踪Usage Tracking用量跟踪完全依赖每次运行后调用store_run_usage——官方文档原话强调This is entirely dependent on thestore_run_usagemethod being called after each agent run. 源码通过_capture_current_turnadvanced_sqlite_session.py在加锁状态下读取当前分支当前轮次并用该轮第一条message_structure.id作为turn anchor锚点如果该轮在写入提交前被删除即便后来有同名轮次复用相同数字 id写入会被跳过避免用量错误地归属到已不存在的轮次上。保存用量数据# 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)保存的字段与turn_usage表一一对应requests、input_tokens、output_tokens、total_tokens以及两个 JSON 明细列input_tokens_details/output_tokens_details。明细序列化逻辑位于_update_turn_usage_internaladvanced_sqlite_session.py取Usage对象上input_tokens_details/output_tokens_details的__dict__序列化为 JSON 存储序列化失败时记录警告并置空不影响主数据写入随后以INSERT OR REPLACE写入UNIQUE(session_id, branch_id, user_turn_number)约束保证同一分支同一轮只保留一条记录。查询用量统计# 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)实现要点来自 advanced_sqlite_session.pyget_session_usage(branch_idNone)通过SUM(requests)、SUM(input_tokens)、SUM(output_tokens)、SUM(total_tokens)与COUNT(*)聚合turn_usage不传branch_id时汇总所有分支传入时只统计指定分支。无数据时返回None。get_turn_usage(user_turn_numberNone, branch_idNone)指定轮次时返回单个字典无数据返回空字典{}否则按user_turn_number升序返回字典列表JSON 明细列自动json.loads解析为 Python 对象。两者都通过asyncio.to_thread把同步 SQLite 操作放到工作线程执行避免阻塞事件循环。对话分支Conversation Branching对话分支是AdvancedSQLiteSession的核心能力任意一条用户消息都可以成为新分支的起点从而探索替代对话路径。创建分支# 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 的整个生命周期内保持唯一。删除分支或清空会话会移除其对话数据但不会释放已使用的分支 ID创建新分支时必须使用新名称。源码对这条规则的落实非常严谨。branch_reservations表见下文 Schema承担原子预留职责_reserve_branch_idadvanced_sqlite_session.py对自定义名称执行INSERT OR IGNORE若rowcount 0说明 ID 已被占用抛出ValueError自动命名则生成branch_from_turn_{turn}_{timestamp}基名并循环加后缀直到预留成功。_copy_messages_to_new_branchadvanced_sqlite_session.py在BEGIN IMMEDIATE下先检查分支点是否存在用户消息再把分支点之前branch_turn_number from_turn_number的消息以共享同一message_id的方式复制到新分支仅复制message_structure行消息数据本体不重复存储最后提交并切换当前指针。create_branch_from_contentadvanced_sqlite_session.py内部调用find_turns_by_content取最早匹配的用户轮次作为分支点无匹配时抛出ValueError。测试test_create_branch_from_content验证了这一行为见 test_advanced_sqlite_session.py。分支管理# 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_branchesadvanced_sqlite_session.py按message_structure聚合branch_id用COUNT(*)统计消息数、COUNT(CASE WHEN message_type user ...)统计用户轮数并标记当前分支。switch_to_branchadvanced_sqlite_session.py先校验分支存在message_structure中计数为 0 则抛ValueError再通过_commit_branch_pointer在锁内原子更新指针若期间发生了clear_sessiongeneration 不匹配清空操作的重置优先指针仍回落到main。delete_branchadvanced_sqlite_session.py拒绝删除main分支与空分支 ID删除当前分支时必须forceTrue会自动先切回main删除顺序为先turn_usage再message_structure随后用_cleanup_orphaned_messages_sync清理不再被任何分支引用的孤儿消息——这正是分支间共享消息、删除一个分支不影响其他分支的底层保证。测试test_delete_branch_keeps_messages_still_referenced_by_another_branchtest_advanced_sqlite_session.py专门验证了该行为。分支工作流完整示例# 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传入session后get_items会按当前分支过滤消息WHERE s.branch_id ?因此切换分支即切换对话上下文。官方完整示例 examples/memory/advanced_sqlite_session_example.py 演示了完整的创建分支 → 新分支内多轮对话 → 切回 main → 再切回新分支流程并验证分支完全独立、互不干扰。会话清空与多实例安全值得补充的是源码在文档描述之外还维护了第四张表session_clear_generations见下节。它记录每个会话的清空代数generationclear_sessionadvanced_sqlite_session.py在清空消息、会话、message_structure、turn_usage的同时将 generation 递增任何持有旧 generation 的会话实例在下次操作时都会通过_refresh_branch_after_external_clear检测到不一致并把分支指针重置回main。这防止了旧会话实例在另一个实例清空会话后把历史写入后来复用相同 ID 的分支这一竞态测试test_branch_ids_remain_reserved_after_delete_and_cleartest_advanced_sqlite_session.py与test_branch_allocation_is_serialized_across_processestest_advanced_sqlite_session.py分别覆盖了 ID 预留与跨进程串行化场景。结构化查询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_turnsadvanced_sqlite_session.py按sequence_number升序读取当前分支的message_structure返回{轮次号: [{type, tool_name}, ...]}字典。get_tool_usageadvanced_sqlite_session.py识别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_output去重若同轮已有对应tool_search_call则不重复计数返回(tool_name, count, turn)元组列表。find_turns_by_contentadvanced_sqlite_session.py用message_data LIKE %term%在用户消息中做子串匹配返回与get_conversation_turns相同结构的字典列表。get_conversation_turnsadvanced_sqlite_session.py返回turn、content截断预览默认 100 字符、full_content、timestamp、can_branch恒为True因为所有用户消息都可作为分支点。内容预览由_content_preview实现advanced_sqlite_session.py兼容纯字符串与多模态结构化 content 列表如input_text/input_image超长时截断并追加...。工具名称的提取规则message_structure.tool_name的取值由_extract_tool_nameadvanced_sqlite_session.py决定规则包括MCP 工具mcp_call/mcp_approval_request优先拼接为server_label.tool_name内置工具computer_call、file_search_call、web_search_call、code_interpreter_call、tool_search_call等无name字段直接以消息类型作为工具名其中tool_search_call/tool_search_output统一归为tool_search普通函数调用使用name字段有namespace时通过tool_qualified_name拼成限定名保留合成命名空间如custom_tool_call场景时直接使用裸名。测试test_tool_usage_tracking_preserves_namespaces_and_tool_searchtest_advanced_sqlite_session.py与test_advanced_tool_name_extractiontest_advanced_sqlite_session.py对上述规则做了系统验证。消息结构元数据会话自动跟踪以下消息结构信息消息类型值user、assistant、tool_call等工具调用的工具名称轮次号user_turn_number与序列号sequence_number分支归属关系branch_id时间戳created_at在写入路径上add_itemsadvanced_sqlite_session.py把消息插入与结构元数据插入放在同一个事务中_insert_items_insert_structure_metadata后统一conn.commit()保证元数据写入失败不会留下孤儿消息测试test_add_items_rolls_back_messages_when_structure_metadata_failstest_advanced_sqlite_session.py验证了该原子性。_insert_structure_metadataadvanced_sqlite_session.py为每条消息分配全局递增的sequence_number并按分支内用户消息计数推进user_turn_number与branch_turn_number正确处理单批次含多条用户消息的情况测试见test_conversation_structure_with_multiple_turnstest_advanced_sqlite_session.py。数据库 SchemaAdvancedSQLiteSession在基础 SQLite Schemaagent_sessions、agent_messages两张表定义见 sqlite_session.py之上新增三张业务表外加一张并发控制表。基础库连接默认启用 WAL 模式PRAGMA journal_modeWAL见 sqlite_session.py。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 );源码在创建该表后还会建立 4 个索引见_init_structure_tablesadvanced_sqlite_session.pyidx_structure_session_seq(session_id, sequence_number)idx_structure_branch(session_id, branch_id)idx_structure_turn(session_id, branch_id, user_turn_number)idx_structure_branch_seq(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 的分支中——这是分支 ID 终身唯一性的持久化保证。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)约束保证每个分支的每个轮次只有一条用量记录配套索引idx_turn_usage_session_turn覆盖(session_id, branch_id, user_turn_number)。session_clear_generations 表源码补充文档正文之外源码还维护一张并发控制表advanced_sqlite_session.pyCREATE TABLE session_clear_generations ( session_id TEXT PRIMARY KEY, generation INTEGER NOT NULL DEFAULT 0 );每次clear_session使generation递增用于跨实例失效陈旧的本地分支指针详见前文会话清空与多实例安全。完整示例官方提供了一个覆盖全部特性的综合示例 examples/memory/advanced_sqlite_session_example.py包含四个部分基础会话记忆三轮连续对话 每轮store_run_usage用量跟踪与分析get_session_usage()输出会话级汇总get_turn_usage()输出逐轮 token 数结构化查询get_conversation_by_turns()按轮展示消息与工具调用get_tool_usage()展示工具统计对话分支与分支管理从第 2 轮创建分支、在新分支继续对话、list_branches()列出全部分支、switch_to_branch在主分支与新分支之间往返最后验证分支完全独立、无相互干扰并在结束时session.close()释放连接。示例中使用了tool装饰器注册的get_weather工具运行后可在输出中直观看到tool_usage与get_conversation_by_turns中的tool_name字段。该文件位于examples/memory/目录可直接运行观察完整效果。API 参考AdvancedSQLiteSession—— 主类继承自SQLiteSession提供分支、用量、结构化查询等增强能力Session—— 基础会话协议Protocol定义get_items/add_items/pop_item/clear_session四个异步接口所有会话实现含第三方自定义会话都遵循该结构契约SQLiteSessionsqlite_session.py—— 基础 SQLite 实现AdvancedSQLiteSession的父类负责建表、连接管理与 WAL 配置。适用场景与注意事项多方案对比探索对同一对话在不同轮次分叉出多条路径如换一种提问方式、切换关注点事后随时切回主线继续适合需要保留完整探索轨迹的 Agent 应用。成本与用量审计store_run_usageturn_usage表为按会话、按分支、按轮次的 token 消耗核算提供了精确数据源。对话回放与质检message_structure的轮次/序列/工具元数据可用于还原每一轮的完整调用链。务必显式保存用量只要忘记调用store_run_usageget_session_usage/get_turn_usage将查不到数据返回None或空列表。分支 ID 不可复用删除分支不释放 ID请使用新名称创建分支。数据库文件归属一个db_path对应一组基础表对不同基础表名自定义sessions_table/messages_table的会话应使用各自独立的数据库文件。持久化前提默认db_path:memory:为内存库进程退出即丢失需要跨进程/跨重启保留分支与用量数据时务必传入文件路径。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
