适用场景
内部知识库问答最容易翻车的两个点：一是"答得像但查无实据"，二是"该拒答时硬编"。这套方案适合已经有一批结构化程度一般的文档（制度、手册、工单、合同模板），希望智能体先规划检索、多轮补充、再带着原文出处回答的团队。它解决的核心问题不是"能不能答"，而是"每条结论能不能指回某个文档的具体片段"。
环境与前置条件
- 操作系统：Linux 或 macOS 本机即可；Windows 建议用 WSL2，避免路径与编码问题。
- Python：3.10 及以上（低于此版本部分 SDK 依赖会装不上）。
- 必备工具：
gcloudCLI、git、curl、jq。 - 硬件：本机 8GB 内存、10GB 可用磁盘足够。推理与检索都在托管侧完成，不需要本地 GPU。
- 云端：一个已绑定结算账号的 Google Cloud 项目；你的账号需要能创建服务账号、启用 API、部署 Cloud Run 的权限（Owner 或等价的组合角色）。
- 数据：语料原始文件建议先放到 Cloud Storage 桶里，作为唯一事实来源，后续重建索引都从这里取。
版本、配额、区域可用性这类信息变化较快，涉及具体数值时以官方文档当前版本为准。
分步骤部署
步骤 1：初始化项目与本地配置
把项目固定下来，后面所有命令都指向它，避免"改了半天发现改的是另一个项目"。
```bash
export PROJECT_ID="your-project-id"
export REGION="your-region" # 选一个支持检索与 Cloud Run 的区域
export BUCKET="gs://${PROJECT_ID}-rag-corpus"
gcloud auth login
gcloud config set project "$PROJECT_ID"
gcloud config set ai/region "$REGION"
```
成功标志：gcloud config list 输出的 project 与你设置的一致。
步骤 2：启用所需 API
```bash
gcloud services enable \
aiplatform.googleapis.com \
discoveryengine.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
logging.googleapis.com \
--project="$PROJECT_ID"
```
这条命令是幂等的，重复执行没副作用。完成后再跑一次 gcloud services list --enabled --project="$PROJECT_ID" | grep -E 'aiplatform|discoveryengine'，能看到对应条目即成功。API 名称以官方文档当前版本为准。
步骤 3：创建运行时服务账号
不要让服务用你的个人账号跑，后续换人、审计都会很麻烦。
```bash
gcloud iam service-accounts create rag-runtime \
--display-name="Agentic RAG runtime" \
--project="$PROJECT_ID"
export SA="rag-runtime@${PROJECT_ID}.iam.gserviceaccount.com"
```
给它绑定调用模型与检索的角色，角色名以官方 IAM 文档当前版本为准，常见的是模型调用侧的 roles/aiplatform.user 与检索侧的只读角色：
```bash
for ROLE in roles/aiplatform.user roles/discoveryengine.viewer roles/logging.logWriter; do
gcloud projects add-iam-policy-binding "$PROJECT_ID" \
--member="serviceAccount:${SA}" --role="$ROLE" --condition=None
done
```
步骤 4：本地开发认证
```bash
gcloud auth application-default login
```
这一步生成的是"应用默认凭据"（ADC），本地代码通过它访问云端服务。浏览器授权完成后，终端提示 Credentials saved to file 即成功。
步骤 5：语料入库与切分
先把原始文档上传到桶里，建立唯一来源：
```bash
gcloud storage buckets create "$BUCKET" --location="$REGION" --uniform-bucket-level-access
gcloud storage cp -r ./docs/* "$BUCKET"/raw/
```
切分策略直接决定检索质量，给几个可落地的默认值：按语义段落切，单块 500–800 字（中文按字符算），块间重叠 10%–15%，每个块必须保留三个字段——chunk_id、doc_uri（原始文件路径）、title_path（章节路径，例如"员工手册 > 第三章 报销 > 3.2 差旅"）。title_path 在生成引用时非常有用，模型写"依据《员工手册》3.2 节"比写"依据某文档"可信得多。
chunk_id 建议用 文档哈希 + 块序号 生成，保证同一份文档重复入库时 id 稳定，方便做增量更新和引用回溯。
步骤 6：建立检索索引
在控制台或通过 SDK 创建语料库/数据存储，把上一步的文件导入。导入完成后记录三个值到环境变量里，后面代码要用：
```bash
export CORPUS_ID="your-corpus-or-datastore-id"
export LOCATION="$REGION"
export MODEL_ID="a-tool-calling-capable-model" # 具体模型名以官方文档当前版本为准
```
导入是异步任务，页面或接口返回成功后不要立刻测，等导入任务状态变为完成。判断标准很简单：用一句你确定在文档里的话去搜，能返回对应片段就算导入生效。
步骤 7：封装检索工具
智能体要"会用"检索，检索就得是一个边界清晰、返回结构固定的函数。下面的封装只做一件事：把后端返回映射成统一的 Chunk，这样以后换检索后端不用改上层逻辑。
```python
retriever.py
import os
from dataclasses import dataclass, asdict
@dataclass
class Chunk:
chunk_id: str
doc_uri: str
title_path: str
text: str
TOP_K_DEFAULT = 8
def search(query: str, top_k: int = TOP_K_DEFAULT, filters: dict | None = None) -> list[Chunk]:
"""调用托管检索服务，返回带来源信息的片段列表。
具体的 SDK 方法与参数名以官方文档当前版本为准，
这里只负责把返回值映射成 Chunk。"""
raw = _call_backend(query=query, top_k=top_k, filters=filters)
out = []
for item in raw:
out.append(Chunk(
chunk_id=item["id"],
doc_uri=item["source_uri"],
title_path=item.get("title_path", ""),
text=item["content"],
))
return out
```
两个工程细节值得加：一是 top_k 不要一上来就设 20，块越多、上下文越脏、引用越容易串；二是给检索加超时和一次重试，网络抖动不该让整个问答报错。
步骤 8：实现智能体式检索循环
不要把所有片段一次性塞给模型，而是让它决定"先搜什么、够不够、要不要再搜"。用一个显式的循环，最多 3 轮，好处是每一轮的查询和命中都能被记录下来。
```python
agent.py
import json
MAX_ROUNDS = 3
def plan(question, evidence_summary):
"""让模型输出下一步动作：继续检索，还是可以作答。
强制 JSON 输出，字段固定，避免解析失败。"""
prompt = f"""用户问题：{question}
已获得的证据摘要：
{evidence_summary}
请只输出 JSON：
{{"action": "search" 或 "answer",
"query": "若 action=search，写下这一轮要检索的查询",
"filters": {{}},
"reason": "一句话说明理由"}}"""
return json.loads(call_model(prompt, json_mode=True))
def run(question, scope_filter):
evidence = {} # chunk_id -> Chunk
trace = []
for r in range(MAX_ROUNDS):
step = plan(question, summarize(evidence))
trace.append({"round": r, "step": step})
if step["action"] == "answer" or r == MAX_ROUNDS - 1:
break
filters = {scope_filter, (step.get("filters") or {})}
for c in search(step["query"], filters=filters):
evidence[c.chunk_id] = c
return generate_answer(question, list(evidence.values())), evidence, trace
```
这里有个容易被忽略的点：scope_filter 是权限过滤，必须由服务端根据当前登录用户注入，绝不能交给模型从问题里推断。否则用户问一句"帮我看看别人的合同"，智能体可能真的去检索了。
步骤 9：生成带引用的回答并做校验
提示词里把输出协议写死，并且要求模型"只引用给定片段，引用不到就说不知道"：
```python
ANSWER_PROMPT = """你是内部知识库助手。只能依据下面提供的片段回答。
片段列表：
{blocks}
要求：
1. 只输出 JSON，形如
{"answer": "...", "citations": [{"chunk_id": "...", "quote": "片段原文摘录"}], "answer_type": "answered 或 no_answer"}
2. citations 里的 chunk_id 必须来自上面列表，不许自造。
3. quote 必须是片段里连续出现的一段原文，不超过 60 字。
4. 片段无法支撑结论时，answer_type 填 no_answer，answer 说明缺什么信息。
"""
```
关键是拿到回答后不要直接返回给用户，先做机器校验：
```python
def verify(answer_obj, evidence):
problems = []
cites = answer_obj.get("citations") or []
if answer_obj.get("answer_type") == "answered" and not cites:
problems.append("answered 但没有引用")
for c in cites:
cid = c.get("chunk_id")
if cid not in evidence:
problems.append(f"引用了不存在的片段 {cid}")
continue
quote = normalize(c.get("quote", ""))
if quote and quote not in normalize(evidence[cid].text):
problems.append(f"引用原文与片段不符 {cid}")
return problems
```
校验不通过时的处理顺序：先用"你的引用有问题，请重新作答"回问模型一次；仍不通过，就降级返回保守结果——把检索到的片段原文和来源链接一并展示，标注"自动校验未通过，请人工确认"。这比返回一段看起来流畅但无出处的答案安全得多。
步骤 10：部署为可访问的服务
用 Cloud Run 从源码部署，不需要自己打镜像：
```bash
gcloud run deploy agentic-rag \
--source . \
--region="$REGION" \
--service-account="$SA" \
--no-allow-unauthenticated \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID},LOCATION=${LOCATION},CORPUS_ID=${CORPUS_ID},MODEL_ID=${MODEL_ID}" \
--memory=1Gi \
--timeout=180
```
--no-allow-unauthenticated 意味着服务不对外公开，调用方必须带身份令牌。部署成功会打印服务 URL，形如 https://agentic-rag-xxxx.a.run.app。
验证部署是否成功
第一步，本地跑一次冒烟测试，确认检索通了：
```bash
python -c "
from retriever import search
r = search('差旅住宿标准', top_k=5)
print(len(r), r[0].chunk_id, r[0].title_path)
"
```
预期：打印的不是 0，且 title_path 能看出章节位置。
第二步，测服务端点：
```bash
export URL="https://agentic-rag-xxxx.a.run.app"
TOKEN=$(gcloud auth print-identity-token)
curl -s -X POST "$URL/query" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"question":"出差住宿一晚最多报销多少？"}' | jq
```
预期返回里同时满足：answer 有具体数字；citations 数组非空；每个 chunk_id 都能在本次检索的 evidence 里找到；quote 字段是片段里的原话。
第三步，测"应该拒答"的问题。准备几个文档里确实没有的问题，例如"公司有没有在火星建分部"。
```bash
curl -s -X POST "$URL/query" -H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"question":"公司火星分部的补贴标准是多少？"}' | jq '.answer_type'
```
预期输出 "no_answer"。这一步比第一步重要——绝大多数 RAG 上线翻车，都是因为从来没测过"不知道"。
建议再准备 20 条左右的固定评测集（15 条有答案、5 条无答案），每次改提示词或换模型都跑一遍，记录三个数：有答案命中率、拒答正确率、引用校验失败率。
常见报错与解决
报错：DefaultCredentialsError: Could not automatically determine credentials
原因：本地没做 ADC 登录，或环境变量 GOOGLE_APPLICATION_CREDENTIALS 指向了一个不存在的文件。
解决：
```bash
gcloud auth application-default login
unset GOOGLE_APPLICATION_CREDENTIALS
```
报错：403 PERMISSION_DENIED: ... does not have ... permission on resource
原因：运行时服务账号没拿到检索或模型调用的角色，常见于手动改了服务账号却没重新部署。
解决：
```bash
gcloud projects add-iam-policy-binding "$PROJECT_ID" \
--member="serviceAccount:${SA}" --role="roles/aiplatform.user" --condition=None
gcloud run services update agentic-rag --region="$REGION" --service-account="$SA"
```
角色名以官方 IAM 文档当前版本为准。改完权限有几十秒生效延迟，别急着反复重试。
报错：429 RESOURCE_EXHAUSTED 或 Quota exceeded
原因：短时间内请求量超过配额，或某轮循环里一次性发了太多检索请求。
解决：先在代码里加指数退避重试（1s、2s、4s，最多 3 次），同时把 MAX_ROUNDS 从 3 降到 2、top_k 从 8 降到 5 观察；确属业务量增长，再到控制台提交配额调整申请。
```bash
gcloud run services update agentic-rag --region="$REGION" \
--set-env-vars="MAX_ROUNDS=2,TOP_K=5"
```
报错：504 Deadline Exceeded 或 Cloud Run 返回 503
原因：多轮检索叠加，单次请求超过服务超时；或检索后端慢。
解决：给检索单独设超时（例如 10 秒），超时就用已收集的片段直接作答；同时确认 Cloud Run 的 --timeout 大于最坏情况耗时。
现象：回答内容正确，但 citations 大量校验失败
原因：提示词没强制 JSON、片段太长导致模型难以精确摘录、或切分把一句话截断了。
解决：先加 json_mode 与固定字段协议；再把单块长度压到 500–800 字；最后检查切分是否在句子中间断开。
后续维护
备份：语料唯一来源放在 Cloud Storage，建议开启对象版本控制，误删可回滚：
```bash
gcloud storage buckets update "$BUCKET" --versioning
```
索引本身视为可重建产物，不要把它当数据源。
升级：换模型、改切分、调提示词都要走"影子索引 + 灰度"：新建一个语料库导入同一批文档，把评测集在上面跑一遍，指标不降再切流量；切的时候先放 10% 请求，观察一天引用校验失败率和拒答率。直接原地改索引，出问题时无法对比。
日志：每次请求落一条结构化日志，字段至少包括 trace_id、question、每轮的 query 与命中 chunk_id、answer_type、校验问题列表、耗时、token 用量。排查"为什么这次答错了"时，看 trace 比看最终答案有用十倍。
监控：盯四个指标——p95 延迟、拒答率、引用校验失败率、单请求平均检索轮次。拒答率突然下降要警惕（可能是模型开始瞎编），突然升高则可能是索引导入没完成或权限过滤写错了。
成本控制：常见问题做一层结果缓存（键用规范化后的问题 + 权限范围），top_k 与 MAX_ROUNDS 都设上限，避免某类问题把检索轮次打满。
最后提醒一句：这套架构里，检索质量决定上限，引用校验决定下限。上线初期宁可让它多说几次"没找到依据"，也别急着把校验关掉。