LogiChat：从问答对到文档RAG的架构重构

LogiChat 是一个聊天机器人（chatbot）平台。客户上传文档，获取聊天组件（chat widget），无需接触模型。过去两年里，“训练机器人”意味着在仪表盘表单中手动整理问答列表——question 和 answer 对，一次一行，作为少样本（few-shot）对注入提示词（prompt）。我刚刚删除了这一切，并基于 Google 的 Agent 开发工具包 (ADK) 重建了整个流水线（pipeline），使用 Vertex AI 作为模型，Firestore 作为向量存储（vector storage），其余部分使用 Cloud Run。

这不是一个“hello world”的 agent 演示——这是我在过去几个月重建整个流水线后确定的架构，涉及三个 Cloud Run 服务（第四个，Stripe 充值令牌订阅器，未改动），以及一个最终被证明比其他任何设计选择都重要的决定：将检索（retrieval）作为模型前的步骤，而不是 ADK 的工具（tool）。

端到端的技术栈（stack）

下面的流水线涉及其中三个服务——apps/api、apps/subscribers/doc-processor 和 apps/agent——以及 Firestore 向量索引（vector index）。apps/api 在图中出现了两次，因为它既充当上传入口（ingress），又充当为 agent 生成 ID 令牌（ID token）的网关（gateway）。点击五个阶段，查看每个跳转点哪个服务处于活动状态，以及跨越每个边界的是什么负载（payload）：

交互式图表： 逐步查看五阶段流水线——上传（Upload）→ 处理（Process）→ 检索（Retrieve）→ 接地（Ground）→ 回答（Answer）——请访问 dalenguyen.me 上的原始文章。每个跳转点显示哪个服务处于活动状态（仪表盘 → apps/api → Cloud Storage → doc-processor → Firestore → apps/agent → Vertex AI）以及跨越边界的是什么负载。

与教科书式的 RAG 技术栈相比，有两个不同之处：

两者都是有意为之。两者也正是本文真正要讨论的内容。

为什么选择 ADK

我选择 Google 的 Agent 开发工具包 而不是 LangGraph、CrewAI 或手动编写的循环（loop），原因只有一个：agent 服务完全运行在 GCP 上，而 ADK 是这些工具中唯一一个从一开始就为 Vertex AI 构建的。 Python SDK 提供了 google.adk.agents.Agent 和 google.adk.runners.InMemoryRunner，它们通过 google-genai 进行路由，当 GOOGLE_GENAI_USE_VERTEXAI=TRUE 时，google-genai 已经能够与 Vertex AI 通信。没有 LangChain 适配器（adapter），没有提供商垫片（provider shim），在预发布（staging）和生产（prod）环境之间不会出现意外的模型名称漂移（model name drift）。

具体来说，我的 apps/agent 是一个 53 行的 pyproject.toml Python 3.12 服务：

[project]
name = "logichat-agent"
version = "0.1.0"
description = "LogiChat ADK agent service (Python) — Gemini 2.5 Flash via Vertex AI."
requires-python = ">=3.12"

dependencies = [
"fastapi>=0.115",
"pydantic>=2.7",
"uvicorn>=0.30",
# google-adk 是 Agent 开发工具包 Python SDK。宽松地锁定版本
# 因为 ADK 的 API 仍在变动；后续问题将涉及
# 真正的 Runner（会话 / 工具 / 流式传输）。
"google-adk>=0.1.0",
# google-genai 随 ADK 一起提供，但我在 runner 适配器中
# 直接导入 google.genai.types，因此显式声明它。
"google-genai>=1.0",
# google-cloud-firestore 是我用于每个应用块
# 向量搜索的客户端。检索层需要它。
"google-cloud-firestore>=2.20",
]

五个运行时依赖（runtime dep）。FastAPI 用于 HTTP 接口，Pydantic 用于传输契约（wire contract），uvicorn 用于 Cloud Run 入口点（entrypoint），google-adk 用于 agent，google-genai 用于类型化消息部分（typed message parts），google-cloud-firestore 用于 KNN 搜索。ADC 适用于所有这些——任何地方都不需要 API 密钥。

Agent 服务，自上而下

apps/agent 分为四个文件，每个文件负责一项工作：

| 文件 | 职责 |

| --- | --- |

| src/agent/main.py | uvicorn 入口点；绑定 $PORT（Cloud Run 注入） |

| src/agent/agent_definition.py | 构建 ADK Agent + InMemoryRunner，封装在 RunnerProtocol 中 |

| src/agent/retrieval.py | 通过 Vertex AI 进行 Firestore find_nearest + text-embedding-005 |

| src/agent/api/app.py | FastAPI 工厂：POST /run，GET /healthz |

main.py 是六行编排代码（orchestration）：

# apps/agent/src/agent/main.py
def main() -> None:
runner = build_agent()           # 构建 ADK 支持的适配器
app = create_app(runner=runner)  # 将 runner 接入 FastAPI
port = int(os.environ.get("PORT", "8080"))
uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")

build_agent() 是连接 ADK 的地方。以下是全部代码：

# apps/agent/src/agent/agent_definition.py
MODEL_ID = "gemini-2.5-flash"

SYSTEM_INSTRUCTION = (
"You are LogiChat's support assistant. Answer the user's question using "
"ONLY the context retrieved from the company's knowledge base. If the "
"retrieved context does not contain the answer, say you don't know and "
"suggest the user rephrase or contact a human. Never invent facts, "
"product names, or policies that are not present in the retrieved "
"context. Each retrieved block is prefixed with an internal "
"[doc:<id>] marker so you can tell sources apart while reasoning — "
"these markers are for your own use only and must never appear "
"anywhere in the text you show the user."
)

def build_agent() -> RunnerProtocol:
_require_env("GOOGLE_GENAI_USE_VERTEXAI")  # TRUE → 通过 Vertex AI 路由
_require_env("GOOGLE_CLOUD_PROJECT")       # logichat-dev / logichat-prod
_require_env("GOOGLE_CLOUD_LOCATION")      # us-central1

# 内联导入：允许单元测试在未安装 google-adk 的情况下运行。
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner

root_agent = Agent(
name="logichat_faq_agent",
model=MODEL_ID,
instruction=SYSTEM_INSTRUCTION,
# tools=[firestore_retrieval_tool]  ← 有意不作为 ADK 工具
)
runner = InMemoryRunner(agent=root_agent)
return _AdkRunnerAdapter(runner)

环境变量（env var）在启动时进行验证，因此配置错误的部署会在第一个请求时失败——而不是在第一次部署时。模型 ID 在代码中固定，因此意外的环境变量覆盖不会静默地切换到不同的模型系列。区域（region）和项目（project）仍然来自环境变量，因此预发布和生产环境可以在不重新构建的情况下进行区分。

`RunnerProtocol` 接缝（seam）

google.adk.runners.InMemoryRunner 暴露的接口比我的 HTTP 层需要的更多——会话（session）、事件流（event stream）、函数调用（function call）。我将其封装在一个协议（protocol）中，以便在上游 SDK 变动时，路由处理器（route handler）保持稳定：

# apps/agent/src/agent/api/dependencies.py
class RunnerProtocol(Protocol):
async def run(
self,
*,
app_id: str,
session_id: str,
question: str,
config: dict[str, Any],
) -> RunnerResult: ...

_AdkRunnerAdapter 实现了该协议，并将请求转发到真正的 ADK runner。HTTP 层只知道 RunnerProtocol——它从不导入 google.adk。这之所以重要，是因为：ADK Python SDK 的 API 仍在变动中。 封装它使我以后可以交换实现（会话、评估框架、流式传输），而无需触及 app.py。

适配器在调用 ADK Runner 之前 执行检索。这是值得理解的设计选择。

检索作为模型前的步骤，而不是 ADK 的工具（tool）

这是整个系统中最重要的架构决策，也是大多数 ADK 教程搞错的地方。

在 ADK 中，“显而易见”的模式是将检索暴露为一个工具（tool）：

# 我没有做的事情。
firestore_retrieval_tool = FunctionTool(firestore_retrieval)

root_agent = Agent(
name="logichat_faq_agent",
model="gemini-2.5-flash",
instruction=...,
tools=[firestore_retrieval_tool],  # 模型自行决定是否调用
)

我评估并拒绝了它。有两个原因：

因此，检索位于适配器中，在 ADK runner 被调用之前：

# apps/agent/src/agent/agent_definition.py
async def run(self, *, app_id, session_id, question, config):
# 1. 检索 — 模型前，不是 ADK 工具
chunks = await self._retrieve(app_id=app_id, question=question)

if not chunks:
# 没有有用上下文 → 短路到 defaultAnswer。
# 也涵盖了检索引发异常的情况：_retrieve 在失败时
# 返回空列表和一个标志。
return RunnerResult(answer=_default_answer(config), grounded=False, retrieval=[])

# 2. 组合接地后的用户消息
chunk_block = "

".join(_format_chunk(c) for c in chunks)
user_text = f"{chunk_block}

User question: {question.strip()}"

# 3. 针对接地后的消息运行 ADK 流程
runner = self._adk_runner
user_id = app_id  # 每个 LogiChat 应用对应一个 ADK 用户
session = await runner.session_service.get_session(
app_name=runner.app_name, user_id=user_id, session_id=session_id
)
if session is None:
await runner.session_service.create_session(
app_name=runner.app_name, user_id=user_id, session_id=session_id
)

message = types.Content(role="user", parts=[types.Part(text=user_text)])
final_text = ""
async for event in runner.run_async(
user_id=user_id, session_id=session_id, new_message=message
):
if event.is_final_response() and event.content and event.content.parts:
final_text = "".join(part.text or "" for part in event.content.parts)

return RunnerResult(
answer=sanitize_answer(final_text),
grounded=True,
retrieval=[
RetrievalHit(
chunk_id=c.chunk_id, document_id=c.document_id, distance=c.distance
)
for c in chunks
],
)

有三点值得指出：

检索层

apps/agent/src/agent/retrieval.py 故意保持小巧——它拥有适配器需要的两个操作：嵌入问题，拉取最相关的块。agent 服务中的其他任何部分都不会直接接触 Firestore 或嵌入 API。

EMBEDDING_MODEL = "text-embedding-005"
EMBEDDING_DIMENSION = 768
DEFAULT_DISTANCE_THRESHOLD = 0.5  # COSINE: 0 = 相同，2 = 相反
DEFAULT_TOP_K = 5

KNN 搜索使用 COSINE 距离：

# 注意：这两个符号都没有从包根目录重新导出。两者
# 都会针对真正的 google-cloud-firestore 2.x 安装引发 ImportError。
# 不要“修复”为包根目录——它会在每个请求上静默地
# 回退到非接地后备方案。
from google.cloud.firestore_v1.base_vector_query import DistanceMeasure
from google.cloud.firestore_v1.vector import Vector

query = db.collection(f"logichat-apps/{app_id}/chunks").find_nearest(
vector_field="embedding",
query_vector=Vector(query_vector),
distance_measure=DistanceMeasure.COSINE,
limit=limit,
distance_result_field="vector_distance",
)

阈值在 KNN 返回后在 Python 中应用。原因：find_nearest 按距离返回最近的 limit 个块；阈值是一个独立的问题（召回率 vs 精确度）。将它们混在查询中意味着每个应用所有者都必须知道正确的 top-k 来平衡两者。最好让参数保持独立。

写入路径（write path）和读取路径（read path）必须在维度上保持一致。 firestore.indexes.json 中的 KNN 索引是为 768 维构建的，平面索引（flat index）：

{
"collectionGroup": "chunks",
"queryScope": "COLLECTION",
"fields": [
{
"fieldPath": "embedding",
"vectorConfig": {
"dimension": 768,
"flat": {}
}
}
]
}

这里的漂移是静默且灾难性的——一个使用 1536 维模型嵌入的块，存储到 768 维索引中，要么写入失败，要么更糟，产生看起来合理但实际无意义的结果。我将 EMBEDDING_MODEL 和 EMBEDDING_DIMENSION 固定为模块常量，并在 API 返回错误形状时大声失败：

if len(values) != EMBEDDING_DIMENSION:
raise RuntimeError(
f"embedding model {EMBEDDING_MODEL} returned {len(values)} dims, "
f"expected {EMBEDDING_DIMENSION}"
)

文档处理器（doc-processor）——块的来源

agent 的工作是回答问题。文档处理器的工作是确保有块可以从中回答。

apps/subscribers/doc-processor 是一个 Cloud Run 服务，订阅了项目存储桶上的 google.cloud.storage.object.v1.finalized 事件。触发器过滤器是路径前缀 apps/{appId}/docs/{documentId}/{filename}——该前缀之外的任何内容都会被静默忽略。为什么基于 Storage 事件而不是 Firestore 写入来驱动？因为 API 在返回签名 URL（signed URL）之前 写入 pending 文档，因此元数据可能比实际上传延迟几秒钟。基于 Storage 事件驱动意味着我等待真正的字节。

当事件触发时：

// apps/subscribers/doc-processor/src/app/doc-processor.handler.ts
const [buffer] = await storage.bucket(event.bucket).file(event.name).download()
const parsedDoc = await parseDocument(buffer, mimeType)
const chunks = chunkText(parsedDoc.text)
const vectors = await embedBatch(chunks)
await persistChunks(appId, documentId, uid, chunks, vectors, embeddingModel)
await markReady(appId, documentId, chunks.length, embeddingModel)

其中三行隐藏了真正的选择。

分块器（Chunker）（libs/server/ai/src/lib/chunker.utils.ts）。 800 令牌的块，100 令牌重叠。首先在段落边界处分割，当段落溢出时回退到句子边界，并通过将前一个块的尾部附加到下一个块来实现重叠。*.utils.ts 中的纯函数——没有注入的依赖项，没有生命周期，这是仓库的约定。

解析器（Parser）。 纯文本、markdown、HTML 和 PDF。HTML 解析器使用 cheerio，并在提取文本之前去除