ADK-Python 会话机制完全指南Session 与 BaseSessionService 的原理、存储与实战【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python导读在 Google ADKAgent Development KitPython 版中Session是贯穿多轮对话的会话记录——它承载会话 id、归属app/user、state状态字典与有序的事件历史event history而BaseSessionService则是负责创建、读取、列举、删除这些记录并向其追加事件的存储接口。本文以官方指南 docs/guides/sessions/session/index.md 为核心骨架结合仓库源码sessions 模块 与 runners.py深入讲解会话的生命周期、状态作用域app:/user:/temp:前缀、历史裁剪配置、四种内置存储后端的选择与切换、以及如何将自定义后端接入 Runner。读完本文你将能独立完成内存开发 → SQLite/数据库持久化 → Vertex AI 生产部署的存储层平滑迁移并掌握自定义会话存储的完整实现路径。Session 与 BaseSessionService一文一存职责分离ADK 对会话的设计有一个核心理念单个 agent 的一次运行run本身是无状态的——模型只能看到你喂给它的东西。真正把对话跨轮次背起来的是SessionSession是一个纯 Pydantic 模型定义见 session.py包含id、app_name、user_id、state会话状态字典和events有序事件历史涵盖用户输入、模型回复、工具调用/结果等以及last_update_time最近一次更新的 Unix 时间戳。它自身从不接触存储。所有需要持久化的工作都经由BaseSessionService定义见 base_session_service.py完成。该抽象基类声明了四个抽象方法——create_session、get_session、list_sessions、delete_session——并附带一个所有后端都继承的具体实现append_event。正是这种Session 只管数据结构、Service 只管读写的拆分让同一套 agent 代码在开发期使用进程内字典InMemorySessionService、在生产期使用共享数据库时完全无需改动——你只换 service不换 agent。Runner见 runners.py把session_service作为必填参数并在运行过程中替你驱动get_session与append_event因此绝大多数应用只需要直接调用 service 来创建、列举和删除会话。快速上手零配置的 InMemorySessionServiceInMemorySessionService不需要任何配置即可使用。下面的示例源自官方文档可直接运行创建会话、追加两个事件、再读回验证import asyncio from google.adk.events import Event from google.adk.sessions import InMemorySessionService APP_NAME hello_world USER_ID user-123 async def main() - None: session_service InMemorySessionService() # 1. Create. Omit session_id to have one generated for you. session await session_service.create_session( app_nameAPP_NAME, user_idUSER_ID, state{locale: en-US}, ) # 2. Append events. Each one lands in session.events, and any state the # event carries is merged into session.state. await session_service.append_event( session, Event(authoruser, messageWhat is the weather?) ) await session_service.append_event( session, Event( authorweather_agent, messageIt is sunny., state{last_city: Zurich}, ), ) # 3. Read it back. get_session returns None when nothing is stored. loaded await session_service.get_session( app_nameAPP_NAME, user_idUSER_ID, session_idsession.id ) assert loaded is not None print(len(loaded.events), loaded.state) if __name__ __main__: asyncio.run(main())输出为2 {locale: en-US, last_city: Zurich}。几个必须记住的约定参数风格除append_event以位置参数接收 session 与 event 外其余所有方法均为关键字专用参数keyword-only。三元组标识一个会话由(app_name, user_id, session_id)三元组唯一标识而不是仅靠session_id因此每次读取都必须提供全部三个参数。事件即状态载体事件中的state字段最终会被映射为EventActions.state_delta见 event_actions.py并在append_event时合并进session.state。底层实现内存存储的数据结构从源码看in_memory_session_service.pyInMemorySessionService内部用三层嵌套字典组织数据sessions[app_name][user_id][session_id] - Session会话本体user_state[app_name][user_id][key] - value用户级状态app_state[app_name][key] - value应用级状态。这解释了文档中state lives in process dicts的说法。此外append_event在追加前会做事件去重if any(e event for e in storage_session.events if e.id event.id)防止编排器广播共享状态 delta 时对同一事件的重复应用in_memory_session_service.py。工作原理生命周期、状态作用域与裁剪会话生命周期create_session不传session_id时自动生成 UUID传入的 id 已被占用时抛出AlreadyExistsError定义见 already_exists_error.py。初始state中的前缀键会被拆分到对应的 app/user/session 作用域存储in_memory_session_service.py。get_session会话不存在时返回None而非抛异常。list_sessions返回ListSessionsResponse见 base_session_service.py按last_update_time从旧到新排序且省略事件历史。append_event这是两份会话交汇的地方——基类实现先把事件的state_delta应用到内存中的Session并追加到session.events各后端再覆写该方法把事件写入存储。event.partial为True的部分事件会被原样返回且绝不落库这正是流式输出分片不进历史的原因。失败要响亮向存储不认识的会话追加事件时所有后端都会抛出SessionNotFoundError继承自ValueError见 session_not_found_error.py而不是静默丢弃事件。基类append_event的具体流程base_session_service.py依次是partial 事件短路返回 →_apply_temp_state把temp:状态应用到内存会话 →_trim_temp_delta_state从事件 delta 中剥离temp:键 →_update_session_state合并状态 → 追加事件。状态作用域State Scopingstate中的键通过前缀划分作用域前缀常量定义在State类上state.py前缀常量作用域无—仅当前会话。app:State.APP_PREFIX该应用的所有会话。user:State.USER_PREFIX该用户在该应用内的所有会话。temp:State.TEMP_PREFIX仅当前 invocation绝不持久化。带前缀的键可以像普通键一样写在create_session(state...)或事件的 state delta 中。service 会将其路由到正确的存储作用域并在读取时前缀保留合并回session.state。temp:键是唯一例外它先被应用到内存会话保证同一 invocation 内后续 agent 可读例如SequentialAgent中的output_keytemp:my_key随后在事件写入前被剥离对应源码_apply_temp_state与_trim_temp_delta_state。特别地get_user_state(app_name..., user_id...)可以在没有 session id的情况下读取用户级状态返回的是去掉user:前缀的原始键——非常适合在create_session之前引导上下文避免为读取用户数据而做昂贵的list_sessions。注意它不是抽象方法基类默认实现直接抛出NotImplementedError所以自定义后端若不覆写它调用时会失败base_session_service.py。关于状态校验值得一提State类支持可选的 Pydanticschema无前缀键的写入会按 schema 校验前缀键含:的键跳过校验state.py。裁剪加载的历史GetSessionConfig通过向get_session传入GetSessionConfig可以限制读回的历史量。注意它位于google.adk.sessions.base_session_service子模块而非包根目录from google.adk.sessions.base_session_service import GetSessionConfig # The 20 most recent events. Use num_recent_events0 for metadata and state # only, or after_timestampunix seconds to cut the history by time instead. recent await session_service.get_session( app_nameAPP_NAME, user_idUSER_ID, session_idsession_id, configGetSessionConfig(num_recent_events20), )GetSessionConfig的两个字段base_session_service.pynum_recent_events返回最近 N 个事件0表示只取元数据与状态不含事件None表示不过滤负数会触发ValueError字段校验器强制 0对应测试见 test_session_service.py。after_timestamp只返回时间戳给定 Unix 秒的事件。关键点在于过滤发生在 service 内部因此在数据库后端上它会真正减少读取量而不仅仅是读回来再裁给你看。内存后端的裁剪逻辑参考 in_memory_session_service.py数据库后端的读取过滤见 database_session_service.py 中的get_session实现。选择会话存储后端服务导入路径适用场景InMemorySessionServicegoogle.adk.sessions开发与测试。状态存于进程内字典类注释明确声明不适用于多线程生产环境。DatabaseSessionServicegoogle.adk.sessions需要持久化或多个进程共享同一个会话。底层是 SQLAlchemy 异步引擎需要安装dbextra。VertexAiSessionServicegoogle.adk.sessions部署在 Vertex AI Agent Engine 上使用其托管的会话存储需要gcpextra。SqliteSessionServicegoogle.adk.sessions.sqlite_session_service想要一个本地 SQLite 文件、无需服务器。ADK CLI 使用的正是它注意它不会从包根目录再导出。补充google.adk.sessions包根的__init__.py中init.pyInMemorySessionService与VertexAiSessionService采用惰性导入而DatabaseSessionService在缺少 SQLAlchemy 依赖时会抛出带dbextra 提示的ImportError这正是文档中requires the db extra的源码依据。DatabaseSessionServiceURL 或自持引擎二选一DatabaseSessionService接受一个 URL 或一个你已拥有的 engine且恰好只能提供其一from google.adk.sessions import DatabaseSessionService async with DatabaseSessionService(sqliteaiosqlite:///./sessions.db) as svc: await svc.prepare_tables() # optional; otherwise done on first use session await svc.create_session(app_nameAPP_NAME, user_idUSER_ID)注意事项均有源码支撑见 database_session_service.pyURL 必须使用异步驱动如sqliteaiosqlite、postgresqlasyncpg等。若 URL 解析到同步驱动构造器会抛出带修复建议的ValueError如提示改用{backend}{async_driver}://。传db_engineAsyncEngine则复用应用自己的引擎service不会关闭它不拥有的引擎。作为异步上下文管理器退出时只关闭自己创建的引擎否则需要手动调用close()。SQLite 内存库会自动配置StaticPool与check_same_threadFalse非 SQLite 后端默认开启pool_pre_ping。prepare_tables()可选首次使用时也会自动建表表创建有asyncio.Lock保证线程安全。内部会为每个会话建立 per-session 锁来串行化进程内的append_event调用self._session_locks并使用行级锁with_for_update保护并发写入。VertexAiSessionServiceapp_name 的特别约束切换到VertexAiSessionService前有一个必须知道的差异在这里app_name不是自由字符串。它必须是 reasoning engine id纯数字或完整的projects/.../locations/.../reasoningEngines/N资源名除非你向构造函数传了agent_engine_id。源码中的解析逻辑vertex_ai_session_service.py用正则^projects/([a-zA-Z0-9-_])/locations/([a-zA-Z0-9-_])/reasoningEngines/(\d)$校验完整资源名app_name.isdigit()时直接视为 engine id否则抛出ValueError提示使用完整资源名或 engine id。SqliteSessionServiceADK CLI 的本地落盘选择SqliteSessionServicesqlite_session_service.py基于aiosqlite将事件以 JSON 形式存入本地 SQLite 文件构造参数db_path接受文件系统路径或sqlite:/sqliteaiosqlite:URL_parse_db_path会做归一化不需要任何服务器。若数据库文件使用的是旧 schema初始化时会给出迁移提示——仓库 migration 目录 中提供了从旧 SQLAlchemy 存储迁移到新格式的工具migrate_from_sqlalchemy_pickle/migrate_from_sqlalchemy_sqlite。进阶应用把 service 接入 Runner要解决的问题让每一段对话存到哪里由一个地方统一决定。实现方式把 service 传给Runner(session_service...)并在第一次运行前创建好会话。Runner默认auto_create_sessionFalserunners.py因此遇到未知的session_id会抛出SessionNotFoundError而不是静默开启一段新对话对应逻辑见 runners.py。from google.adk.runners import Runner runner Runner( agentmy_agent, session_servicesession_service, # 统一决定会话存储位置 auto_create_sessionFalse, # 显式创建避免误开新会话 )编写自己的后端要解决的问题会话需要存到 ADK 未内置的存储如 Redis、MongoDB 等。实现方式继承BaseSessionService实现四个抽象方法create_session、get_session、list_sessions、delete_session。随后覆写append_event持久化事件并务必调用await super().append_event(session, event)让内存中的会话状态保持同步基类会处理 temp 状态应用、delta 裁剪与状态合并。如果存储能回答用户级状态查询覆写get_user_state基类默认实现抛NotImplementedError。如果采用缓冲写入覆写flush——基类的flush是空操作Runner关闭时会调用它见 runners.py。检测过期会话Stale Session要解决的问题两个 worker 持有同一个Session对象并同时追加事件其中一个会静默覆盖另一个的历史。实现方式无需自己写代码。DatabaseSessionService为每个会话跟踪一个存储修订标记storage revision。从数据库读回的Session会携带精确的内部标记Session._storage_update_marker私有属性见 session.py当append_event发现内存副本的标记落后于存储时抛出StaleSessionError即文档所述的ValueError语义定义见 _stale_session_error.py。恢复方法重新调用get_session拿到最新会话再对新会话重放追加。相关实现位于 database_session_service.py并被测试覆盖见 test_session_service.py 中test_append_event_to_stale_session与并发场景测试。限制与边界InMemorySessionService不适合生产重启后一切消失、worker 之间不共享、且不加锁。它只适合开发与测试类 docstring 亦明确声明见 in_memory_session_service.py。list_sessions返回的是残缺会话事件历史被丢弃state填充多少取决于后端实现。需要完整数据时请用get_session按需加载。DatabaseSessionService需要dbextra、VertexAiSessionService需要gcpextra缺失对应依赖时导入会失败并得到明确的 extra 提示。get_user_state依赖后端支持自定义后端若不覆写调用即抛NotImplementedError可按文档提示改用list_sessionsget_session枚举的方式读取用户状态。总结ADK 的会话体系用纯模型 存储接口的极简抽象把多轮对话的持久化问题压缩为一个可插拔的session_service参数开发期用InMemorySessionService零配置起步本地发布用SqliteSessionService落盘生产期切到DatabaseSessionService或VertexAiSessionService而 agent 代码零改动。理解会话生命周期、app:/user:/temp:状态作用域、GetSessionConfig历史裁剪以及 stale 检测机制是写出健壮多轮应用的基础——这些知识同样适用于编写自定义后端将 ADK 的会话体系无缝接入你自己的存储设施。延伸阅读本指南对应的状态详解参见 docs/guides/sessions/state/index.md会话状态相关工具与测试分别位于 sessions 源码目录 与 tests/unittests/sessionsRunner 与会话服务的协作细节见 runners.py。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ADK-Python 会话机制完全指南：Session 与 BaseSessionService 的原理、存储与实战
ADK-Python 会话机制完全指南Session 与 BaseSessionService 的原理、存储与实战【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python导读在 Google ADKAgent Development KitPython 版中Session是贯穿多轮对话的会话记录——它承载会话 id、归属app/user、state状态字典与有序的事件历史event history而BaseSessionService则是负责创建、读取、列举、删除这些记录并向其追加事件的存储接口。本文以官方指南 docs/guides/sessions/session/index.md 为核心骨架结合仓库源码sessions 模块 与 runners.py深入讲解会话的生命周期、状态作用域app:/user:/temp:前缀、历史裁剪配置、四种内置存储后端的选择与切换、以及如何将自定义后端接入 Runner。读完本文你将能独立完成内存开发 → SQLite/数据库持久化 → Vertex AI 生产部署的存储层平滑迁移并掌握自定义会话存储的完整实现路径。Session 与 BaseSessionService一文一存职责分离ADK 对会话的设计有一个核心理念单个 agent 的一次运行run本身是无状态的——模型只能看到你喂给它的东西。真正把对话跨轮次背起来的是SessionSession是一个纯 Pydantic 模型定义见 session.py包含id、app_name、user_id、state会话状态字典和events有序事件历史涵盖用户输入、模型回复、工具调用/结果等以及last_update_time最近一次更新的 Unix 时间戳。它自身从不接触存储。所有需要持久化的工作都经由BaseSessionService定义见 base_session_service.py完成。该抽象基类声明了四个抽象方法——create_session、get_session、list_sessions、delete_session——并附带一个所有后端都继承的具体实现append_event。正是这种Session 只管数据结构、Service 只管读写的拆分让同一套 agent 代码在开发期使用进程内字典InMemorySessionService、在生产期使用共享数据库时完全无需改动——你只换 service不换 agent。Runner见 runners.py把session_service作为必填参数并在运行过程中替你驱动get_session与append_event因此绝大多数应用只需要直接调用 service 来创建、列举和删除会话。快速上手零配置的 InMemorySessionServiceInMemorySessionService不需要任何配置即可使用。下面的示例源自官方文档可直接运行创建会话、追加两个事件、再读回验证import asyncio from google.adk.events import Event from google.adk.sessions import InMemorySessionService APP_NAME hello_world USER_ID user-123 async def main() - None: session_service InMemorySessionService() # 1. Create. Omit session_id to have one generated for you. session await session_service.create_session( app_nameAPP_NAME, user_idUSER_ID, state{locale: en-US}, ) # 2. Append events. Each one lands in session.events, and any state the # event carries is merged into session.state. await session_service.append_event( session, Event(authoruser, messageWhat is the weather?) ) await session_service.append_event( session, Event( authorweather_agent, messageIt is sunny., state{last_city: Zurich}, ), ) # 3. Read it back. get_session returns None when nothing is stored. loaded await session_service.get_session( app_nameAPP_NAME, user_idUSER_ID, session_idsession.id ) assert loaded is not None print(len(loaded.events), loaded.state) if __name__ __main__: asyncio.run(main())输出为2 {locale: en-US, last_city: Zurich}。几个必须记住的约定参数风格除append_event以位置参数接收 session 与 event 外其余所有方法均为关键字专用参数keyword-only。三元组标识一个会话由(app_name, user_id, session_id)三元组唯一标识而不是仅靠session_id因此每次读取都必须提供全部三个参数。事件即状态载体事件中的state字段最终会被映射为EventActions.state_delta见 event_actions.py并在append_event时合并进session.state。底层实现内存存储的数据结构从源码看in_memory_session_service.pyInMemorySessionService内部用三层嵌套字典组织数据sessions[app_name][user_id][session_id] - Session会话本体user_state[app_name][user_id][key] - value用户级状态app_state[app_name][key] - value应用级状态。这解释了文档中state lives in process dicts的说法。此外append_event在追加前会做事件去重if any(e event for e in storage_session.events if e.id event.id)防止编排器广播共享状态 delta 时对同一事件的重复应用in_memory_session_service.py。工作原理生命周期、状态作用域与裁剪会话生命周期create_session不传session_id时自动生成 UUID传入的 id 已被占用时抛出AlreadyExistsError定义见 already_exists_error.py。初始state中的前缀键会被拆分到对应的 app/user/session 作用域存储in_memory_session_service.py。get_session会话不存在时返回None而非抛异常。list_sessions返回ListSessionsResponse见 base_session_service.py按last_update_time从旧到新排序且省略事件历史。append_event这是两份会话交汇的地方——基类实现先把事件的state_delta应用到内存中的Session并追加到session.events各后端再覆写该方法把事件写入存储。event.partial为True的部分事件会被原样返回且绝不落库这正是流式输出分片不进历史的原因。失败要响亮向存储不认识的会话追加事件时所有后端都会抛出SessionNotFoundError继承自ValueError见 session_not_found_error.py而不是静默丢弃事件。基类append_event的具体流程base_session_service.py依次是partial 事件短路返回 →_apply_temp_state把temp:状态应用到内存会话 →_trim_temp_delta_state从事件 delta 中剥离temp:键 →_update_session_state合并状态 → 追加事件。状态作用域State Scopingstate中的键通过前缀划分作用域前缀常量定义在State类上state.py前缀常量作用域无—仅当前会话。app:State.APP_PREFIX该应用的所有会话。user:State.USER_PREFIX该用户在该应用内的所有会话。temp:State.TEMP_PREFIX仅当前 invocation绝不持久化。带前缀的键可以像普通键一样写在create_session(state...)或事件的 state delta 中。service 会将其路由到正确的存储作用域并在读取时前缀保留合并回session.state。temp:键是唯一例外它先被应用到内存会话保证同一 invocation 内后续 agent 可读例如SequentialAgent中的output_keytemp:my_key随后在事件写入前被剥离对应源码_apply_temp_state与_trim_temp_delta_state。特别地get_user_state(app_name..., user_id...)可以在没有 session id的情况下读取用户级状态返回的是去掉user:前缀的原始键——非常适合在create_session之前引导上下文避免为读取用户数据而做昂贵的list_sessions。注意它不是抽象方法基类默认实现直接抛出NotImplementedError所以自定义后端若不覆写它调用时会失败base_session_service.py。关于状态校验值得一提State类支持可选的 Pydanticschema无前缀键的写入会按 schema 校验前缀键含:的键跳过校验state.py。裁剪加载的历史GetSessionConfig通过向get_session传入GetSessionConfig可以限制读回的历史量。注意它位于google.adk.sessions.base_session_service子模块而非包根目录from google.adk.sessions.base_session_service import GetSessionConfig # The 20 most recent events. Use num_recent_events0 for metadata and state # only, or after_timestampunix seconds to cut the history by time instead. recent await session_service.get_session( app_nameAPP_NAME, user_idUSER_ID, session_idsession_id, configGetSessionConfig(num_recent_events20), )GetSessionConfig的两个字段base_session_service.pynum_recent_events返回最近 N 个事件0表示只取元数据与状态不含事件None表示不过滤负数会触发ValueError字段校验器强制 0对应测试见 test_session_service.py。after_timestamp只返回时间戳给定 Unix 秒的事件。关键点在于过滤发生在 service 内部因此在数据库后端上它会真正减少读取量而不仅仅是读回来再裁给你看。内存后端的裁剪逻辑参考 in_memory_session_service.py数据库后端的读取过滤见 database_session_service.py 中的get_session实现。选择会话存储后端服务导入路径适用场景InMemorySessionServicegoogle.adk.sessions开发与测试。状态存于进程内字典类注释明确声明不适用于多线程生产环境。DatabaseSessionServicegoogle.adk.sessions需要持久化或多个进程共享同一个会话。底层是 SQLAlchemy 异步引擎需要安装dbextra。VertexAiSessionServicegoogle.adk.sessions部署在 Vertex AI Agent Engine 上使用其托管的会话存储需要gcpextra。SqliteSessionServicegoogle.adk.sessions.sqlite_session_service想要一个本地 SQLite 文件、无需服务器。ADK CLI 使用的正是它注意它不会从包根目录再导出。补充google.adk.sessions包根的__init__.py中init.pyInMemorySessionService与VertexAiSessionService采用惰性导入而DatabaseSessionService在缺少 SQLAlchemy 依赖时会抛出带dbextra 提示的ImportError这正是文档中requires the db extra的源码依据。DatabaseSessionServiceURL 或自持引擎二选一DatabaseSessionService接受一个 URL 或一个你已拥有的 engine且恰好只能提供其一from google.adk.sessions import DatabaseSessionService async with DatabaseSessionService(sqliteaiosqlite:///./sessions.db) as svc: await svc.prepare_tables() # optional; otherwise done on first use session await svc.create_session(app_nameAPP_NAME, user_idUSER_ID)注意事项均有源码支撑见 database_session_service.pyURL 必须使用异步驱动如sqliteaiosqlite、postgresqlasyncpg等。若 URL 解析到同步驱动构造器会抛出带修复建议的ValueError如提示改用{backend}{async_driver}://。传db_engineAsyncEngine则复用应用自己的引擎service不会关闭它不拥有的引擎。作为异步上下文管理器退出时只关闭自己创建的引擎否则需要手动调用close()。SQLite 内存库会自动配置StaticPool与check_same_threadFalse非 SQLite 后端默认开启pool_pre_ping。prepare_tables()可选首次使用时也会自动建表表创建有asyncio.Lock保证线程安全。内部会为每个会话建立 per-session 锁来串行化进程内的append_event调用self._session_locks并使用行级锁with_for_update保护并发写入。VertexAiSessionServiceapp_name 的特别约束切换到VertexAiSessionService前有一个必须知道的差异在这里app_name不是自由字符串。它必须是 reasoning engine id纯数字或完整的projects/.../locations/.../reasoningEngines/N资源名除非你向构造函数传了agent_engine_id。源码中的解析逻辑vertex_ai_session_service.py用正则^projects/([a-zA-Z0-9-_])/locations/([a-zA-Z0-9-_])/reasoningEngines/(\d)$校验完整资源名app_name.isdigit()时直接视为 engine id否则抛出ValueError提示使用完整资源名或 engine id。SqliteSessionServiceADK CLI 的本地落盘选择SqliteSessionServicesqlite_session_service.py基于aiosqlite将事件以 JSON 形式存入本地 SQLite 文件构造参数db_path接受文件系统路径或sqlite:/sqliteaiosqlite:URL_parse_db_path会做归一化不需要任何服务器。若数据库文件使用的是旧 schema初始化时会给出迁移提示——仓库 migration 目录 中提供了从旧 SQLAlchemy 存储迁移到新格式的工具migrate_from_sqlalchemy_pickle/migrate_from_sqlalchemy_sqlite。进阶应用把 service 接入 Runner要解决的问题让每一段对话存到哪里由一个地方统一决定。实现方式把 service 传给Runner(session_service...)并在第一次运行前创建好会话。Runner默认auto_create_sessionFalserunners.py因此遇到未知的session_id会抛出SessionNotFoundError而不是静默开启一段新对话对应逻辑见 runners.py。from google.adk.runners import Runner runner Runner( agentmy_agent, session_servicesession_service, # 统一决定会话存储位置 auto_create_sessionFalse, # 显式创建避免误开新会话 )编写自己的后端要解决的问题会话需要存到 ADK 未内置的存储如 Redis、MongoDB 等。实现方式继承BaseSessionService实现四个抽象方法create_session、get_session、list_sessions、delete_session。随后覆写append_event持久化事件并务必调用await super().append_event(session, event)让内存中的会话状态保持同步基类会处理 temp 状态应用、delta 裁剪与状态合并。如果存储能回答用户级状态查询覆写get_user_state基类默认实现抛NotImplementedError。如果采用缓冲写入覆写flush——基类的flush是空操作Runner关闭时会调用它见 runners.py。检测过期会话Stale Session要解决的问题两个 worker 持有同一个Session对象并同时追加事件其中一个会静默覆盖另一个的历史。实现方式无需自己写代码。DatabaseSessionService为每个会话跟踪一个存储修订标记storage revision。从数据库读回的Session会携带精确的内部标记Session._storage_update_marker私有属性见 session.py当append_event发现内存副本的标记落后于存储时抛出StaleSessionError即文档所述的ValueError语义定义见 _stale_session_error.py。恢复方法重新调用get_session拿到最新会话再对新会话重放追加。相关实现位于 database_session_service.py并被测试覆盖见 test_session_service.py 中test_append_event_to_stale_session与并发场景测试。限制与边界InMemorySessionService不适合生产重启后一切消失、worker 之间不共享、且不加锁。它只适合开发与测试类 docstring 亦明确声明见 in_memory_session_service.py。list_sessions返回的是残缺会话事件历史被丢弃state填充多少取决于后端实现。需要完整数据时请用get_session按需加载。DatabaseSessionService需要dbextra、VertexAiSessionService需要gcpextra缺失对应依赖时导入会失败并得到明确的 extra 提示。get_user_state依赖后端支持自定义后端若不覆写调用即抛NotImplementedError可按文档提示改用list_sessionsget_session枚举的方式读取用户状态。总结ADK 的会话体系用纯模型 存储接口的极简抽象把多轮对话的持久化问题压缩为一个可插拔的session_service参数开发期用InMemorySessionService零配置起步本地发布用SqliteSessionService落盘生产期切到DatabaseSessionService或VertexAiSessionService而 agent 代码零改动。理解会话生命周期、app:/user:/temp:状态作用域、GetSessionConfig历史裁剪以及 stale 检测机制是写出健壮多轮应用的基础——这些知识同样适用于编写自定义后端将 ADK 的会话体系无缝接入你自己的存储设施。延伸阅读本指南对应的状态详解参见 docs/guides/sessions/state/index.md会话状态相关工具与测试分别位于 sessions 源码目录 与 tests/unittests/sessionsRunner 与会话服务的协作细节见 runners.py。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考