ADK Python 使用 PostgreSQL 持久化会话DatabaseSessionService 配置与源码级原理解析【免费下载链接】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本文以 ADK Pythongoogle-adk官方示例 postgres_session_service 为主线系统讲解如何将DatabaseSessionService接入 PostgreSQL实现 Agent 会话Session、事件Event与状态State的跨进程、跨重启持久化。读完本文你将掌握连接串配置、连接池调优、自动建表 Schema、会话读写 API以及底层基于乐观锁与行级锁的并发一致性机制可直接照抄示例代码搭建生产可用的持久化会话服务。为什么需要数据库会话服务ADK 默认提供内存会话服务InMemorySessionService它在单进程内可用但 Agent 进程重启或水平扩容后会话上下文即丢失。DatabaseSessionService通过 SQLAlchemy 将会话数据落到关系型数据库中从而支持应用重启后恢复历史对话跨重启持久化多实例共享同一份会话数据水平扩展将会话、事件、状态沉淀为可查询、可备份的数据库记录。官方示例 main.py 演示的正是这一能力首次运行创建会话并写入事件再次运行读取同一会话并看到上一次的事件历史。环境准备与依赖安装前置条件一个可用的 PostgreSQL 实例本地或云上均可asyncpgPython 异步 PostgreSQL 驱动。安装所需 Python 包pip install google-adk asyncpg greenlet需要说明的是asyncpg是 PostgreSQL 后端的必需驱动。从源码看DatabaseSessionService内部通过sqlalchemy.ext.asyncio.create_async_engine创建异步引擎并维护了后端 → 异步驱动的映射表database_session_service.py 中_ASYNC_DRIVER_BY_BACKEND明确将postgresql映射到asyncpg。如果连接串误用了同步驱动如postgresql://初始化会抛出明确错误提示改用postgresqlasyncpg://形式的 URL。自动生成的数据库 SchemaDatabaseSessionService会在首次使用时自动建表源码中的prepare_tables()方法见 database_session_service.py无需手工执行 DDL。它还会把当前 schema 版本写入adk_internal_metadata表为后续迁移提供依据。sessions会话表列类型说明app_nameVARCHAR(128)应用标识联合主键user_idVARCHAR(128)用户标识联合主键idVARCHAR(128)会话 UUID联合主键stateJSONB会话级状态JSONcreate_timeTIMESTAMP创建时间update_timeTIMESTAMP最后更新时间events事件表列类型说明idVARCHAR(128)事件 UUID联合主键app_nameVARCHAR(128)应用标识联合主键user_idVARCHAR(128)用户标识联合主键session_idVARCHAR(128)所属会话引用联合主键、外键invocation_idVARCHAR(256)调用invocation标识timestampTIMESTAMP事件时间戳event_dataJSONB事件内容JSONapp_states应用级状态表列类型说明app_nameVARCHAR(128)应用标识主键stateJSONB应用级状态update_timeTIMESTAMP最后更新时间user_states用户级状态表列类型说明app_nameVARCHAR(128)应用标识联合主键user_idVARCHAR(128)用户标识联合主键stateJSONB用户级状态update_timeTIMESTAMP最后更新时间adk_internal_metadata内部元数据表列类型说明keyVARCHAR(128)元数据键valueVARCHAR(256)元数据值Schema 的源码级细节这些表对应 SQLAlchemy ORM 模型定义在 schemas/v1.pyv1 为当前最新版本JSONB 是 PostgreSQL 专属优化状态列的类型是自定义的DynamicJSON类型装饰器见 schemas/shared.py。当方言为 PostgreSQL 时它实际编译为postgresql.JSONB在其他数据库如 SQLite、MySQL则退化为 TEXT JSON 序列化。因此 README 中描述PostgreSQL 的 JSONB 为状态数据提供高效存储在源码层面是成立的。级联删除events表通过ForeignKeyConstraint引用sessions(app_name, user_id, id)并设置ondeleteCASCADE删除会话时其事件会被级联清理。查询索引events表自带组合索引idx_events_app_user_session_ts(app_name, user_id, session_id, timestamp DESC)支撑按会话按时间倒序拉取事件的场景。prepare_tables()在create_all之后还会调用_ensure_schema_indexes_exist为已有表补建缺失索引。微秒级时间戳时间列使用PreciseTimestamp类型装饰器保证时间精度到微秒这是后续陈旧会话检测依赖update_time作为版本标记的基础。连接配置连接 URL 格式postgresqlasyncpg://username:passwordhost:port/database例如postgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions。基础用法from google.adk.sessions.database_session_service import DatabaseSessionService from google.adk.runners import Runner # Initialize with PostgreSQL URL session_service DatabaseSessionService( postgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions ) # Use with Runner runner Runner( app_namemy_app, agentmy_agent, session_servicesession_service, )从源码看DatabaseSessionService.__init__还支持另一种初始化方式直接传入一个已构造好的sqlalchemy.ext.asyncio.AsyncEngine实例db_engine参数。两个参数db_url与db_engine互斥必须且只能提供一个否则抛出ValueError。若通过 URL 创建所有额外关键字参数都会透传给create_async_engine且对非 SQLite 后端源码会自动设置pool_pre_pingTrue在每次取连接前先探测连接是否可用避免使用失效连接。高级配置连接池参数session_service DatabaseSessionService( postgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions, pool_size10, max_overflow20, pool_timeout30, pool_recycle1800, )这些参数全部由 SQLAlchemy 引擎接受含义如下参数默认值说明pool_size5连接池保持的连接数max_overflow10连接池耗尽后最多额外创建的连接数pool_timeout30等待空闲连接的超时秒数pool_recycle-1连接被回收复用前的最大存活秒数建议配合数据库wait_timeout设置如 1800注意DatabaseSessionService内部维护了_rollback_on_exception_session上下文管理器任何异常都会显式回滚事务避免无效事务长期占用连接导致连接池耗尽。运行官方示例1. 启动 PostgreSQL示例目录自带 compose.yml一键启动 PostgreSQLdocker compose up -d该编排文件使用postgres:16-alpine镜像预置了用户postgres、密码postgres、数据库adk_sessions将宿主机5432端口映射到容器并通过命名卷postgres_data持久化数据容器销毁后数据不丢。也可以直接使用已有的 PostgreSQL 实例只需确保目标数据库已创建。2. 配置环境变量创建.env文件POSTGRES_URLpostgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions GOOGLE_CLOUD_PROJECTyour-gcp-project-id GOOGLE_CLOUD_LOCATIONus-central1 GOOGLE_GENAI_USE_ENTERPRISEtrue或者直接导出环境变量export POSTGRES_URLpostgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions export GOOGLE_CLOUD_PROJECT$(gcloud config get-value project) export GOOGLE_CLOUD_LOCATIONus-central1 export GOOGLE_GENAI_USE_ENTERPRISEtrue示例 main.py 使用python-dotenv的load_dotenv(overrideTrue)加载.env若未设置POSTGRES_URL会抛出带提示信息的ValueError。其余三个变量用于配置 Vertex AI Gemini 模型的调用模型走企业版入口。3. 运行 Agent直接运行示例脚本python main.py或者使用 ADK CLI 以当前目录作为 Agent 应用运行adk run .示例中的 Agent 定义在 agent.py名为postgres_session_agent挂载了get_current_time工具返回 UTC 当前时间指令中要求它记住同一会话内的历史对话。main.py依次向 Agent 发送两条消息——现在几点请记住这条信息和我刚才问了你什么——第二次提问时 Agent 需要依赖已持久化的会话事件才能正确回答。会话持久化核心读写 API创建与恢复会话# First run - creates a new session session await session_service.create_session( app_namemy_app, user_iduser1, session_idpersistent-session-123, ) # Later run - retrieves the existing session session await session_service.get_session( app_namemy_app, user_iduser1, session_idpersistent-session-123, )源码层面的行为create_session在写入前会检查(app_name, user_id, session_id)主键是否已存在存在则抛出AlreadyExistsError并发场景下即便两个请求同时通过检查底层flush触发的主键冲突也会被捕获并转换为同样的AlreadyExistsError。get_session返回None表示会话不存在可用于存在即恢复否则新建的分支逻辑同时支持GetSessionConfig通过num_recent_events只读取最近 N 条事件0表示只读元数据不查事件通过after_timestamp只读取某时间点之后的事件。读取时会合并三层状态会话状态 用户状态 应用状态其中用户/应用状态会以app.、user.前缀合入统一的 state 字典见源码_merge_state。状态管理PostgreSQL 的 JSONB 为状态数据提供高效存储三层状态分别落库会话级状态sessions.state用户级状态user_states.state应用级状态app_states.state写入时append_event会把事件携带的state_delta拆分extract_json_safe_state_delta后分别合并到对应状态行临时状态temp state只存在于内存、落库前会被修剪避免污染持久化数据。并发与一致性源码中的工程细节DatabaseSessionService面向生产设计源码中包含几处值得注意的并发控制陈旧会话检测乐观并发控制每次append_event都会比对存储中的update_time微秒级标记get_update_marker与内存会话携带的标记。若存储已被其他进程更新过则抛出StaleSessionError提示重新加载会话后再追加事件从根源上防止多实例并发写同一个会话导致事件交错。行级锁悲观锁在 PostgreSQL / MySQL / MariaDB 后端读取会话与状态行时使用SELECT ... FOR UPDATEwith_for_update进一步串行化同会话的关键路径写入。进程内会话锁append_event在进程内按(app_name, user_id, session_id)加asyncio.Lock带引用计数自动清理配合数据库锁形成进程内、跨进程双重保护。事务安全读操作走独立的只读引擎与只读会话工厂写操作通过async_sessionmaker(expire_on_commitFalse)避免提交后 ORM 属性过期导致对 asyncpg 的惰性加载。时区处理PostgreSQL 的 TIMESTAMP 不保留时区信息源码会先剥离tzinfo再写入、读出时再按 UTC 还原保证create_session写入值与append_event读回值一致避免陈旧检测误报。这些机制从 database_session_service.py 可直接读到也解释了为何文档建议生产环境使用连接池而非默认值。生产环境实践建议连接池调优高并发应用务必设置pool_size与max_overflow并配合pool_timeout、pool_recycle防止连接耗尽或使用被数据库回收的陈旧连接。SSL/TLS生产环境始终使用加密连接云厂商如 Cloud SQL、RDS一般要求开启 SSL。备份策略会话与事件是业务数据应纳入常规备份体系events表对会话级联删除需在备份与恢复方案中同步考虑。索引规划默认 Schema 已包含主键索引与events表的会话-时间组合索引若出现新的高频查询模式如按时间范围查所有用户会话可补充额外索引。监控关注连接池使用率、活跃连接数与慢查询。示例中还可用prepare_tables()在应用启动阶段提前完成建表而非首次请求触发避免首请求延迟。显式关闭资源服务提供close()方法释放 SQLAlchemy 引擎及连接池也支持async with上下文管理便于优雅停机。总结通过DatabaseSessionService接入 PostgreSQL只需一条postgresqlasyncpg://连接串即可获得自动建表、JSONB 高效状态存储、跨重启会话恢复等能力而源码层补充的乐观锁、行级锁、进程内锁与自动索引则让这套方案在并发多实例场景下依然可靠。官方示例 postgres_session_service 是可直接复用的最小工程骨架结合 main.py、compose.yml 与核心实现 database_session_service.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 使用 PostgreSQL 持久化会话：DatabaseSessionService 配置与源码级原理解析
ADK Python 使用 PostgreSQL 持久化会话DatabaseSessionService 配置与源码级原理解析【免费下载链接】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本文以 ADK Pythongoogle-adk官方示例 postgres_session_service 为主线系统讲解如何将DatabaseSessionService接入 PostgreSQL实现 Agent 会话Session、事件Event与状态State的跨进程、跨重启持久化。读完本文你将掌握连接串配置、连接池调优、自动建表 Schema、会话读写 API以及底层基于乐观锁与行级锁的并发一致性机制可直接照抄示例代码搭建生产可用的持久化会话服务。为什么需要数据库会话服务ADK 默认提供内存会话服务InMemorySessionService它在单进程内可用但 Agent 进程重启或水平扩容后会话上下文即丢失。DatabaseSessionService通过 SQLAlchemy 将会话数据落到关系型数据库中从而支持应用重启后恢复历史对话跨重启持久化多实例共享同一份会话数据水平扩展将会话、事件、状态沉淀为可查询、可备份的数据库记录。官方示例 main.py 演示的正是这一能力首次运行创建会话并写入事件再次运行读取同一会话并看到上一次的事件历史。环境准备与依赖安装前置条件一个可用的 PostgreSQL 实例本地或云上均可asyncpgPython 异步 PostgreSQL 驱动。安装所需 Python 包pip install google-adk asyncpg greenlet需要说明的是asyncpg是 PostgreSQL 后端的必需驱动。从源码看DatabaseSessionService内部通过sqlalchemy.ext.asyncio.create_async_engine创建异步引擎并维护了后端 → 异步驱动的映射表database_session_service.py 中_ASYNC_DRIVER_BY_BACKEND明确将postgresql映射到asyncpg。如果连接串误用了同步驱动如postgresql://初始化会抛出明确错误提示改用postgresqlasyncpg://形式的 URL。自动生成的数据库 SchemaDatabaseSessionService会在首次使用时自动建表源码中的prepare_tables()方法见 database_session_service.py无需手工执行 DDL。它还会把当前 schema 版本写入adk_internal_metadata表为后续迁移提供依据。sessions会话表列类型说明app_nameVARCHAR(128)应用标识联合主键user_idVARCHAR(128)用户标识联合主键idVARCHAR(128)会话 UUID联合主键stateJSONB会话级状态JSONcreate_timeTIMESTAMP创建时间update_timeTIMESTAMP最后更新时间events事件表列类型说明idVARCHAR(128)事件 UUID联合主键app_nameVARCHAR(128)应用标识联合主键user_idVARCHAR(128)用户标识联合主键session_idVARCHAR(128)所属会话引用联合主键、外键invocation_idVARCHAR(256)调用invocation标识timestampTIMESTAMP事件时间戳event_dataJSONB事件内容JSONapp_states应用级状态表列类型说明app_nameVARCHAR(128)应用标识主键stateJSONB应用级状态update_timeTIMESTAMP最后更新时间user_states用户级状态表列类型说明app_nameVARCHAR(128)应用标识联合主键user_idVARCHAR(128)用户标识联合主键stateJSONB用户级状态update_timeTIMESTAMP最后更新时间adk_internal_metadata内部元数据表列类型说明keyVARCHAR(128)元数据键valueVARCHAR(256)元数据值Schema 的源码级细节这些表对应 SQLAlchemy ORM 模型定义在 schemas/v1.pyv1 为当前最新版本JSONB 是 PostgreSQL 专属优化状态列的类型是自定义的DynamicJSON类型装饰器见 schemas/shared.py。当方言为 PostgreSQL 时它实际编译为postgresql.JSONB在其他数据库如 SQLite、MySQL则退化为 TEXT JSON 序列化。因此 README 中描述PostgreSQL 的 JSONB 为状态数据提供高效存储在源码层面是成立的。级联删除events表通过ForeignKeyConstraint引用sessions(app_name, user_id, id)并设置ondeleteCASCADE删除会话时其事件会被级联清理。查询索引events表自带组合索引idx_events_app_user_session_ts(app_name, user_id, session_id, timestamp DESC)支撑按会话按时间倒序拉取事件的场景。prepare_tables()在create_all之后还会调用_ensure_schema_indexes_exist为已有表补建缺失索引。微秒级时间戳时间列使用PreciseTimestamp类型装饰器保证时间精度到微秒这是后续陈旧会话检测依赖update_time作为版本标记的基础。连接配置连接 URL 格式postgresqlasyncpg://username:passwordhost:port/database例如postgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions。基础用法from google.adk.sessions.database_session_service import DatabaseSessionService from google.adk.runners import Runner # Initialize with PostgreSQL URL session_service DatabaseSessionService( postgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions ) # Use with Runner runner Runner( app_namemy_app, agentmy_agent, session_servicesession_service, )从源码看DatabaseSessionService.__init__还支持另一种初始化方式直接传入一个已构造好的sqlalchemy.ext.asyncio.AsyncEngine实例db_engine参数。两个参数db_url与db_engine互斥必须且只能提供一个否则抛出ValueError。若通过 URL 创建所有额外关键字参数都会透传给create_async_engine且对非 SQLite 后端源码会自动设置pool_pre_pingTrue在每次取连接前先探测连接是否可用避免使用失效连接。高级配置连接池参数session_service DatabaseSessionService( postgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions, pool_size10, max_overflow20, pool_timeout30, pool_recycle1800, )这些参数全部由 SQLAlchemy 引擎接受含义如下参数默认值说明pool_size5连接池保持的连接数max_overflow10连接池耗尽后最多额外创建的连接数pool_timeout30等待空闲连接的超时秒数pool_recycle-1连接被回收复用前的最大存活秒数建议配合数据库wait_timeout设置如 1800注意DatabaseSessionService内部维护了_rollback_on_exception_session上下文管理器任何异常都会显式回滚事务避免无效事务长期占用连接导致连接池耗尽。运行官方示例1. 启动 PostgreSQL示例目录自带 compose.yml一键启动 PostgreSQLdocker compose up -d该编排文件使用postgres:16-alpine镜像预置了用户postgres、密码postgres、数据库adk_sessions将宿主机5432端口映射到容器并通过命名卷postgres_data持久化数据容器销毁后数据不丢。也可以直接使用已有的 PostgreSQL 实例只需确保目标数据库已创建。2. 配置环境变量创建.env文件POSTGRES_URLpostgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions GOOGLE_CLOUD_PROJECTyour-gcp-project-id GOOGLE_CLOUD_LOCATIONus-central1 GOOGLE_GENAI_USE_ENTERPRISEtrue或者直接导出环境变量export POSTGRES_URLpostgresqlasyncpg://postgres:postgreslocalhost:5432/adk_sessions export GOOGLE_CLOUD_PROJECT$(gcloud config get-value project) export GOOGLE_CLOUD_LOCATIONus-central1 export GOOGLE_GENAI_USE_ENTERPRISEtrue示例 main.py 使用python-dotenv的load_dotenv(overrideTrue)加载.env若未设置POSTGRES_URL会抛出带提示信息的ValueError。其余三个变量用于配置 Vertex AI Gemini 模型的调用模型走企业版入口。3. 运行 Agent直接运行示例脚本python main.py或者使用 ADK CLI 以当前目录作为 Agent 应用运行adk run .示例中的 Agent 定义在 agent.py名为postgres_session_agent挂载了get_current_time工具返回 UTC 当前时间指令中要求它记住同一会话内的历史对话。main.py依次向 Agent 发送两条消息——现在几点请记住这条信息和我刚才问了你什么——第二次提问时 Agent 需要依赖已持久化的会话事件才能正确回答。会话持久化核心读写 API创建与恢复会话# First run - creates a new session session await session_service.create_session( app_namemy_app, user_iduser1, session_idpersistent-session-123, ) # Later run - retrieves the existing session session await session_service.get_session( app_namemy_app, user_iduser1, session_idpersistent-session-123, )源码层面的行为create_session在写入前会检查(app_name, user_id, session_id)主键是否已存在存在则抛出AlreadyExistsError并发场景下即便两个请求同时通过检查底层flush触发的主键冲突也会被捕获并转换为同样的AlreadyExistsError。get_session返回None表示会话不存在可用于存在即恢复否则新建的分支逻辑同时支持GetSessionConfig通过num_recent_events只读取最近 N 条事件0表示只读元数据不查事件通过after_timestamp只读取某时间点之后的事件。读取时会合并三层状态会话状态 用户状态 应用状态其中用户/应用状态会以app.、user.前缀合入统一的 state 字典见源码_merge_state。状态管理PostgreSQL 的 JSONB 为状态数据提供高效存储三层状态分别落库会话级状态sessions.state用户级状态user_states.state应用级状态app_states.state写入时append_event会把事件携带的state_delta拆分extract_json_safe_state_delta后分别合并到对应状态行临时状态temp state只存在于内存、落库前会被修剪避免污染持久化数据。并发与一致性源码中的工程细节DatabaseSessionService面向生产设计源码中包含几处值得注意的并发控制陈旧会话检测乐观并发控制每次append_event都会比对存储中的update_time微秒级标记get_update_marker与内存会话携带的标记。若存储已被其他进程更新过则抛出StaleSessionError提示重新加载会话后再追加事件从根源上防止多实例并发写同一个会话导致事件交错。行级锁悲观锁在 PostgreSQL / MySQL / MariaDB 后端读取会话与状态行时使用SELECT ... FOR UPDATEwith_for_update进一步串行化同会话的关键路径写入。进程内会话锁append_event在进程内按(app_name, user_id, session_id)加asyncio.Lock带引用计数自动清理配合数据库锁形成进程内、跨进程双重保护。事务安全读操作走独立的只读引擎与只读会话工厂写操作通过async_sessionmaker(expire_on_commitFalse)避免提交后 ORM 属性过期导致对 asyncpg 的惰性加载。时区处理PostgreSQL 的 TIMESTAMP 不保留时区信息源码会先剥离tzinfo再写入、读出时再按 UTC 还原保证create_session写入值与append_event读回值一致避免陈旧检测误报。这些机制从 database_session_service.py 可直接读到也解释了为何文档建议生产环境使用连接池而非默认值。生产环境实践建议连接池调优高并发应用务必设置pool_size与max_overflow并配合pool_timeout、pool_recycle防止连接耗尽或使用被数据库回收的陈旧连接。SSL/TLS生产环境始终使用加密连接云厂商如 Cloud SQL、RDS一般要求开启 SSL。备份策略会话与事件是业务数据应纳入常规备份体系events表对会话级联删除需在备份与恢复方案中同步考虑。索引规划默认 Schema 已包含主键索引与events表的会话-时间组合索引若出现新的高频查询模式如按时间范围查所有用户会话可补充额外索引。监控关注连接池使用率、活跃连接数与慢查询。示例中还可用prepare_tables()在应用启动阶段提前完成建表而非首次请求触发避免首请求延迟。显式关闭资源服务提供close()方法释放 SQLAlchemy 引擎及连接池也支持async with上下文管理便于优雅停机。总结通过DatabaseSessionService接入 PostgreSQL只需一条postgresqlasyncpg://连接串即可获得自动建表、JSONB 高效状态存储、跨重启会话恢复等能力而源码层补充的乐观锁、行级锁、进程内锁与自动索引则让这套方案在并发多实例场景下依然可靠。官方示例 postgres_session_service 是可直接复用的最小工程骨架结合 main.py、compose.yml 与核心实现 database_session_service.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），仅供参考