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 ADK（Agent Development Kit）Python 版中，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）本身是无状态的——模型只能看到你喂给它的东西。真正把对话跨轮次"背"起来的是Session：

• Session是一个纯 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 来创建、列举和删除会话。

快速上手：零配置的 InMemorySessionService

InMemorySessionService不需要任何配置即可使用。下面的示例（源自官方文档，可直接运行）创建会话、追加两个事件、再读回验证：

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_name=APP_NAME, user_id=USER_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(author="user", message="What is the weather?") ) await session_service.append_event( session, Event( author="weather_agent", message="It 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_name=APP_NAME, user_id=USER_ID, session_id=session.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.py），InMemorySessionService内部用三层嵌套字典组织数据：

• 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 Scoping）

state中的键通过前缀划分作用域，前缀常量定义在State类上（state.py）：

带前缀的键可以像普通键一样写在create_session(state=...)或事件的 state delta 中。service 会将其路由到正确的存储作用域，并在读取时（前缀保留）合并回session.state。temp:键是唯一例外：它先被应用到内存会话（保证同一 invocation 内后续 agent 可读，例如SequentialAgent中的output_key='temp: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_events=0 for metadata and state # only, or after_timestamp=<unix seconds> to cut the history by time instead. recent = await session_service.get_session( app_name=APP_NAME, user_id=USER_ID, session_id=session_id, config=GetSessionConfig(num_recent_events=20), )

GetSessionConfig的两个字段（base_session_service.py）：

• num_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实现。

选择会话存储后端

补充：google.adk.sessions包根的__init__.py中（init.py），InMemorySessionService与VertexAiSessionService采用惰性导入，而DatabaseSessionService在缺少 SQLAlchemy 依赖时会抛出带dbextra 提示的ImportError，这正是文档中"requires the db extra"的源码依据。

DatabaseSessionService：URL 或自持引擎，二选一

DatabaseSessionService接受一个 URL 或一个你已拥有的 engine，且恰好只能提供其一：

from google.adk.sessions import DatabaseSessionService async with DatabaseSessionService("sqlite+aiosqlite:///./sessions.db") as svc: await svc.prepare_tables() # optional; otherwise done on first use session = await svc.create_session(app_name=APP_NAME, user_id=USER_ID)

注意事项（均有源码支撑，见 database_session_service.py）：

• URL 必须使用异步驱动：如sqlite+aiosqlite、postgresql+asyncpg等。若 URL 解析到同步驱动，构造器会抛出带修复建议的ValueError（如提示改用{backend}+{async_driver}://）。
• 传db_engine=<AsyncEngine>则复用应用自己的引擎，service不会关闭它不拥有的引擎。
• 作为异步上下文管理器退出时，只关闭自己创建的引擎；否则需要手动调用close()。
• SQLite 内存库会自动配置StaticPool与check_same_thread=False；非 SQLite 后端默认开启pool_pre_ping。
• prepare_tables()可选，首次使用时也会自动建表（表创建有asyncio.Lock保证线程安全）。
• 内部会为每个会话建立 per-session 锁来串行化进程内的append_event调用（self._session_locks），并使用行级锁（with_for_update）保护并发写入。

VertexAiSessionService：app_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。

SqliteSessionService：ADK CLI 的本地落盘选择

SqliteSessionService（sqlite_session_service.py）基于aiosqlite将事件以 JSON 形式存入本地 SQLite 文件，构造参数db_path接受文件系统路径或sqlite:/sqlite+aiosqlite:URL（_parse_db_path会做归一化），不需要任何服务器。若数据库文件使用的是旧 schema，初始化时会给出迁移提示——仓库 migration 目录 中提供了从旧 SQLAlchemy 存储迁移到新格式的工具（migrate_from_sqlalchemy_pickle/migrate_from_sqlalchemy_sqlite）。

进阶应用

把 service 接入 Runner

• 要解决的问题：让"每一段对话存到哪里"由一个地方统一决定。
• 实现方式：把 service 传给Runner(session_service=...)，并在第一次运行前创建好会话。Runner默认auto_create_session=False（runners.py），因此遇到未知的session_id会抛出SessionNotFoundError，而不是静默开启一段新对话（对应逻辑见 runners.py）。

from google.adk.runners import Runner runner = Runner( agent=my_agent, session_service=session_service, # 统一决定会话存储位置 auto_create_session=False, # 显式创建，避免误开新会话 )

编写自己的后端

• 要解决的问题：会话需要存到 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_sessions+get_session枚举的方式读取用户状态。

总结

ADK 的会话体系用"纯模型 + 存储接口"的极简抽象，把多轮对话的持久化问题压缩为一个可插拔的session_service参数：开发期用InMemorySessionService零配置起步，本地发布用SqliteSessionService落盘，生产期切到DatabaseSessionService或VertexAiSessionService而 agent 代码零改动。理解会话生命周期、app:/user:/temp:状态作用域、GetSessionConfig历史裁剪以及 stale 检测机制，是写出健壮多轮应用的基础——这些知识同样适用于编写自定义后端，将 ADK 的会话体系无缝接入你自己的存储设施。

延伸阅读：本指南对应的状态详解参见 docs/guides/sessions/state/index.md；会话状态相关工具与测试分别位于 sessions 源码目录 与 tests/unittests/sessions；Runner 与会话服务的协作细节见 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