Agno 接入 Google Cloud Firestore：Agent 会话持久化存储配置指南与源码解析

本篇文章聚焦于 Agno 框架与 Google Cloud Firestore 的集成实践，围绕 cookbook/06_storage/firestore/README.md 提供的安装、认证、最小配置与运行示例展开，并结合 libs/agno/agno/db/firestore/ 下 FirestoreDb 的真实实现，解析其集合（Collection）组织、索引自动创建、批量写入限制与会话读写链路。读完本文，你将掌握如何在自己的 GCP 项目中为 Agno Agent 启用 Firestore 持久化、如何复用与续接历史会话（add_history_to_context=True），并能理解底层存储结构的核心设计。

一、Firestore 在 Agno 存储体系中的定位

在 Agno 的示例库中，Agent 的会话持久化有多种后端可选，Firestore 是面向 Google Cloud 生态、主打免运维 Serverless 文档数据库的一支。它适合以下场景：

• 希望将 Agent 会话数据直接存放在已有 GCP 项目中，与 Cloud Run、Cloud Functions、Vertex AI 等云端组件共享同一套 IAM 与网络环境；
• 需要按 Collection / Document 粒度天然区分**用户维度（user_id）**数据，便于做多用户多会话隔离；
• 项目尚处于低成本起步阶段——示例代码中的注释明确指出：FirestoreDb 默认连接 Firestore 的 (default) 数据库，从而可以使用免费层级额度访问 Firestore（见 firestore_for_agent.py）。

在仓库的 cookbook 目录中，本模块对应的完整示例集合位于 cookbook/06_storage/firestore/，整个 06_storage 大目录下还并列提供 Postgres、SQLite、Mongo、Redis 等其他后端作为对比参考。

二、环境准备：安装与前置条件

1. 安装依赖

示例文档给出的安装命令基于 uv：

uv pip install google-cloud-firestore

而完整的可运行示例还需要 openai（作为默认模型后端）与 agno 本体：

uv pip install openai google-cloud-firestore agno

如果你使用原生 pip，等价命令为 pip install google-cloud-firestore openai agno。注意，google-cloud-firestore 是 FirestoreDb 硬性依赖：在源码 libs/agno/agno/db/firestore/firestore.py 中，导入失败时会主动抛出带有安装提示的 ImportError：

`google-cloud-firestore` not installed. Please install it using `pip install google-cloud-firestore`

2. 项目侧前置条件

运行示例前需要保证（详见 firestore_for_agent.py 开头的步骤说明）：

1. 你的 gcloud 项目已启用 Firestore（需先创建 Firestore 数据库）；
2. 当前环境具备访问 Firestore 的权限与凭据；
3. 需要访问云端真实环境执行，示例属于有真实外部依赖的集成用例。

三、Google Cloud 认证配置

Firestore 客户端使用 Google Cloud 的标准认证体系。官方 README 提供了两种方式：

# 方式一：通过 gcloud CLI 登录获取应用默认凭据（推荐本地开发）
gcloud auth application-default login

# 方式二：通过环境变量指定服务账号密钥文件
export GOOGLE_APPLICATION_CREDENTIALS="path/to/service-account.json"

从实现角度看，FirestoreDb 在未显式传入客户端时会执行 Client(project=project_id)，而该构造过程正是通过 GOOGLE_APPLICATION_CREDENTIALS 或 ADC（Application Default Credentials）链来解析凭据。在 CI、服务器或本地环境，只要其中一个认证通道可用，FirestoreDb 就能自动连接，无需额外代码。

四、最小集成配置

README 给出的核心用法非常简洁——把 FirestoreDb 作为 db 参数注入 Agent：

from agno.agent import Agent
from agno.db.firestore import FirestoreDb

db = FirestoreDb(project_id="your-project-id")

agent = Agent(
    db=db,
    add_history_to_context=True,
)

要点说明：

• project_id 指向你的 GCP 项目 ID；如果省略 project_id，则必须通过 db_client 显式传入一个 google.cloud.firestore.Client 实例，二者必填其一——否则构造函数会抛出 ValueError("One of project_id or db_client must be provided")（见 firestore.py）；
• add_history_to_context=True 使 Agent 在每次请求时将已持久化的历史轮次注入上下文，从而实现“重启进程后仍记得之前聊过什么”的长期会话能力；
• 公开导入路径为 from agno.db.firestore import FirestoreDb，该符号在 libs/agno/agno/db/firestore/init.py 中导出。

五、完整示例：带联网搜索的多轮持久化 Agent

仓库提供了可运行的示例 cookbook/06_storage/firestore/firestore_for_agent.py，完整逻辑如下：

"""
This recipe shows how to store agent sessions in a Firestore database.
"""
from agno.agent import Agent
from agno.db.firestore import FirestoreDb
from agno.tools.websearch import WebSearchTools

PROJECT_ID = "agno-os-test"  # 替换为你的 GCP 项目 ID

# FirestoreDb 唯一必填参数是集合名（使用默认值即可），
# 连接凭据会自动从 google cloud credentials 读取；
# 默认连接 (default) 数据库以兼容 Firestore 免费层级。
db = FirestoreDb(project_id=PROJECT_ID)

agent = Agent(
    db=db,
    tools=[WebSearchTools()],
    add_history_to_context=True,
)

if __name__ == "__main__":
    agent.print_response("How many people live in Canada?")
    agent.print_response("What is their national anthem called?")

运行方式：

python cookbook/06_storage/firestore/firestore_for_agent.py

该示例演示了最典型的实战链路：

1. 创建连接项目 agno-os-test 的 FirestoreDb（请替换为你自己的项目 ID）；
2. 为 Agent 挂载 WebSearchTools，使其在需要实时事实时执行联网搜索工具；
3. 连续抛出两个有上下文关联的问题——第二个问题（国歌）依赖第一个问题（人口）建立的对话语境；
4. 由于 add_history_to_context=True，第一轮会话被写入 Firestore 后，第二轮会自动读取并带上历史，确保 Agent 能正确回答“他们的国歌”。

六、FirestoreDb 参数与底层集合（Collection）模型

1. 构造参数总览

FirestoreDb.__init__（firestore.py）接受以下参数：

参数	默认值	作用
db_client	None	直接传入 google.cloud.firestore.Client，与 project_id 二选一
project_id	None	GCP 项目 ID；缺省时内部执行 Client(project=project_id)
session_collection	agno_sessions	存储 Agent/Team/Workflow 会话文档
runs_collection	agno_runs	存储单次运行记录（每个 run 一个文档）
memory_collection	agno_memories	存储用户记忆
metrics_collection	agno_metrics	存储按天聚合的用量指标
eval_collection	agno_eval_runs	存储评测运行记录
knowledge_collection	agno_knowledge	存储知识文档
traces_collection	agno_traces	存储追踪数据（span 的父级）
spans_collection	agno_spans	存储链路 span
id	由项目 ID 派生	数据库实例标识

默认集合名并非 Firestore 特有，而是继承自 BaseDb：在 libs/agno/agno/db/base.py 中统一落地为 agno_sessions、agno_runs、agno_memories、agno_metrics、agno_eval_runs、agno_knowledge、agno_traces、agno_spans，另有 agno_schema_versions 专门存放 schema 版本戳，供迁移管理器判断是否需要对某张表执行升级。当需要多 Agent/多用户数据分库隔离时，可通过这些参数自定义集合名。

2. 集合按实体类型拆分

_get_collection()（firestore.py）按 sessions / runs / memories / metrics / evals / knowledge / traces / spans 八种类型分发集合引用，并在首次访问时执行 create_collection_indexes() 自动创建单字段与复合索引。值得注意的两个细节：

• spans 集合依赖 traces 集合存在（“spans reference traces”），因此在创建 spans 集合前会先确保 traces 集合就绪；
• 每个集合只做一次初始化动作，通过 self._{collection_name}_initialized 标记避免重复建索引。

3. 复合索引的自动创建与降级策略

索引逻辑位于 libs/agno/agno/db/firestore/utils.py：它读取 schemas.py 中定义的集合 schema，通过 google.cloud.firestore_admin_v1.FirestoreAdminClient 调用 Admin API 在 projects/{project_id}/databases/(default)/collectionGroups/{collection_name} 下以编程方式创建复合索引。

例如会话集合的 schema（schemas.py）声明了多种复合索引，与 get_sessions() 的查询模式一一对应：

• (session_type ASCENDING, created_at DESCENDING)
• (session_type, agent_id, created_at DESC) / (session_type, team_id, created_at DESC) / (session_type, workflow_id, created_at DESC)
• (user_id, session_type, created_at DESC) 等多级组合

而 runs 集合 schema（schemas.py）则为“按会话顺序拉取历史”这一热路径声明了 (session_id ASCENDING, run_index ASCENDING) 索引，并为“按状态筛选 + 按时间倒序”的 HITL（人类介入）轮询场景提供组合索引。

如果 Admin API 不可用（例如缺少 Firestore Admin 权限），utils.py 会打印降级提示，并建议通过 Firebase Console 或 gcloud CLI 手工创建复合索引；缺失索引时相关查询会在运行时因 FAILED_PRECONDITION 失败。这是排查“查询忽然报错”时首先要检查的点。

七、会话读写链路与批量写入工程细节

1. 会话与运行记录的写入

Agent 一次交互会同时产生两类数据：

• 会话文档（session）：通过 upsert_session()（firestore.py）落库。代码按 AgentSession、TeamSession、WorkflowSession 分支组装记录，字段包括 session_id、session_type、各自的 agent_id / team_id / workflow_id、user_id、业务数据（agent_data / team_data / workflow_data）、session_data、summary、metadata、created_at、updated_at。写操作采用 doc_ref.set(record, merge=True) 的合并语义，避免覆盖历史遗留字段；
• 运行记录（run）：v3 起运行记录被拆入独立的 runs 集合（每个 run 一个文档），由 Agent 主循环通过 upsert_run() 单独写入，设计上是一个 O(1) 的单文档更新操作，特别适合 HITL / 后台模式下频繁的状态变更（PENDING → RUNNING → COMPLETED）。

读取时，get_session() 与 get_sessions() 会先把 runs 集合中属于该会话的运行记录按 (run_index, created_at) 排序取回，再与旧版 session 文档内嵌的 legacy runs 字段合并（merge_runs_table_with_legacy_blob），从而保证新旧数据的兼容可见性。

2. Firestore 500 条批量上限的处理

Firestore 对单次 commit 的批量写入有 500 条操作的硬性限制，源码中将其定义为常量 FIRESTORE_BATCH_LIMIT = 500（firestore.py）。在仓库实现中，凡是可能超量的删除与批量写入都会按 500 条分块：

• delete_runs()、delete_sessions()、delete_session() 的级联删除对每个 chunk 单独 batch.commit()；
• cleanup_legacy_runs_field()（清除 v3 迁移遗留的 runs 字段）同样分块处理，并在预检阶段拒绝删除仍持有非空 runs 数据的会话，除非显式传 force=True；
• 批量指标写入 bulk_upsert_metrics() 也按 500 条分块提交。

相关行为在仓库测试中可找到对应验证，例如 test_firestore_batch_chunking.py 与 test_firestore_run_indexes.py。对于会话/运行规模较大的生产使用，这一点直接决定了删除类接口是否报 INVALID_ARGUMENT。

3. 排序、分页与查询限制的实现细节

由于 Firestore 对 OR 查询与“任意字段 + 排序”组合的支持受限，实现中采取了两个务实策略（firestore.py）：

• 当未指定 session_type 却按 component_id（agent_id/team_id/workflow_id 之一）过滤时，先把流式查询结果全部拉回内存，再统一匹配三个 ID 字段；
• 分页在内存中切片完成（先 start_index = (page - 1) * limit 再截取），因为文档型数据库原生不支持 SQL 式 OFFSET 下推。

此外 get_runs() 的 in 过滤（批量查多个会话的运行记录）受 Firestore 单查询 30 个值上限约束，源码在 firestore.py 中对 session_ids 做了每 30 个一分组的自动 chunk。

4. 数据兼容与 Schema 版本管理

会话文档最初把 session_data 存成 JSON 字符串；新版改为原生 map，以便利用 Firestore 点号写法（如 "session_data.session_name"）做高效字段更新。rename_session()（firestore.py）中专门对两种形态做了区分处理：遇到旧式 JSON 字符串时走“读-改-写”并顺带迁移为原生 map，避免点号更新直接覆盖整个字段造成数据丢失。

为支撑跨版本升级，FirestoreDb 在 agno_schema_versions 集合中读写版本戳（get_latest_schema_version 缺省返回 "2.0.0"，firestore.py），使迁移管理器能够识别表是否需要升级。

八、能力边界与适用提醒

结合源码实现，以下几点值得在实际选型与排障时留意：

1. 真实云端依赖：示例需要可用的 GCP 项目、已启用的 Firestore 与有效凭据，无法在纯离线环境跑通；仓库内对应的 TEST_LOG.md 也标注该用例仍处于 PENDING 状态，读者在自己环境中验证时需以真实云端表现为准。
2. 默认数据库与免费层级：实现默认使用 (default) 数据库（Admin API 父路径即硬编码 databases/(default)），无需自建多数据库即可满足起步需求。
3. 批量写性能：FirestoreDb 不提供真正的批量 upsert 通道——upsert_sessions() 会明确打日志并退化为逐个 upsert_session()（firestore.py）。若追求大数据量批写吞吐，建议评估 SQL 系后端（同目录下其他 storage 示例可对比）。
4. 单实体类型覆盖：本目录仅提供一个 Agent 示例，但 FirestoreDb 在类型层面完整支持 Agent、Team、Workflow 三类会话（分别对应 AGENT / TEAM / WORKFLOW 的 session_type），可被 Team、Workflow 直接复用。

九、小结

从 cookbook/06_storage/firestore/README.md 出发，本文完整覆盖了依赖安装、ADC / 服务账号认证、FirestoreDb + Agent 的最小接线方式以及多轮联网搜索示例的运行流程；随后下钻到 libs/agno/agno/db/firestore/ 的源码，澄清了“集合（Collection）即表”的存储模型、八大实体集合、复合索引自动创建机制、runs 独立集合的新架构、500 条批量上限约束与旧版 JSON 数据兼容策略。对于希望在 GCP 上快速落地 Agno Agent 持久化会话的开发者，这套组合已经提供了开箱即用的接入路径与清晰的排障思路。