Gemini混合RAG教程：结合文件搜索与Google搜索
技术背景
在实际 AI 应用中，模型常需同时访问私有文档（如企业内部手册）和实时公共信息（如新闻、统计数据）。单一知识源无法满足复杂查询需求，而传统 RAG 仅处理私有数据，无法获取最新网络信息。Gemini API 提供文件搜索（托管 RAG）和 Google 搜索接地两种工具，分别解决私有与公共知识检索问题。混合 RAG 将两者结合，使模型能根据问题动态选择或融合知识源，提升回答的准确性和时效性。此方案适用于客服系统、研究助手、企业知识库等场景，但需注意 API 密钥权限管理、存储成本及路由策略的合理性。
核心原理
混合 RAG 的核心在于两种接地工具的协同：文件搜索将文档分块并嵌入向量，查询时检索相关块；Google 搜索接地则实时查询网络。两种工具均通过 grounding_metadata 返回引用来源，其中文件搜索填充 retrieved_context（文本、标题、页码），Google 搜索填充 web（URI、标题）。确定性回退路由由代码控制：先查询文件搜索，通过启发式规则（如无检索结果或模型明确表示文档未覆盖）判断是否回退到 Google 搜索。原生多工具路由则在同一请求中传递两个工具，由 Gemini 自主选择或组合调用，减少请求次数，但路由决策不透明。grounding_supports 提供文本片段与引用块的映射，用于行内引用。
架构分析
系统架构分为三层：客户端层（Python SDK）、Gemini API 层（模型与工具）和知识源层（文件搜索存储与 Google 搜索）。文件搜索存储通过异步摄取操作处理文档分块与嵌入，上传后需轮询状态直至完成。两种路由模式共享相同的工具定义，但调用方式不同：确定性路由分步调用，原生路由一次调用。响应处理层通过 extract_citations 函数统一规范化引用格式，便于展示。设计上强调可审计性：确定性路由可记录路由决策，原生路由通过 grounding_metadata 事后分析。生产环境可扩展为微服务，将文件搜索存储独立部署，并通过 API 网关管理密钥与限流。
完整实战教程
##### 版权所有 2026 Google LLC.
# @title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
Gemini API：混合 RAG，结合文件搜索与 Google 搜索
<a target="_blank" href="https://colab.research.google.com/github/google-gemini/cookbook/blob/main/examples/hybrid_file_search_and_google_search.ipynb">
</a>
大多数真实的助手需要同时具备两种知识：私有知识，存在于你自己的文档中；以及公共知识，存在于开放网络上且每日变化。Gemini 为每种知识都提供了一个接地工具：
* 文件搜索 是一个托管的 RAG 工具。你将文档上传到存储库，Gemini 在生成时检索相关的块。
* Google 搜索接地 将模型连接到网络上最新的公开信息。
本笔记本向你展示两种将它们组合成单一*混合*流水线的方法：
- 确定性回退路由 — 你先查询文件搜索，检查返回的内容，仅当你的文档无法回答问题时才回退到 Google 搜索。你完全控制路由决策，这使得它可审计且易于推理。
- 原生多工具路由 — 你在同一个请求中传递两个工具，让 Gemini 选择正确的工具，或在单个答案中组合两者。
在此过程中，你将提取每个响应中的 grounding_metadata，以便向用户准确展示答案来自哪个文档块或哪个网页。
注意： 本笔记本使用 generateContent API 而非更新的
Interactions API，因为
grounding_metadata（混合路由所依赖的引用）是在generateContent响应中返回的。
工具定义本身在两个 API 中是相同的。
安装依赖
安装 google-genai Python SDK。
%pip install -U -q "google-genai>=2.9.0"
设置你的 API 密钥
要运行以下单元格，你的 API 密钥必须存储在名为 GEMINI_API_KEY 的 Colab Secret 中。
如果你还没有 API 密钥，或者不确定如何创建 Colab Secret，请参阅
获取详细指南。
重要提示： 文件搜索会将上传的文档与 API 密钥的云项目关联，因此你的
API 密钥也会授予对你上传的所有内容的访问权限。请妥善保管，并遵循 Google 的
from google import genai
from google.colab import userdata
from google.genai import types
GEMINI_API_KEY = userdata.get("GEMINI_API_KEY")
client = genai.Client(api_key=GEMINI_API_KEY)
选择模型
当前 Gemini 模型系列均支持 Grounding 工具。在本指南中，请选择您要使用的模型。
MODEL_ID = "gemini-3.7-flash" # @param ["gemini-3.1-pro-preview", "gemini-3.7-flash", "gemini-3.5-flash-lite", "gemini-2.5-pro"] {"allow-input":true, isTemplate: true}
创建示例文档
为了让笔记本自包含，你为一家虚构公司生成一份小型内部手册，而不是下载任何内容。本文档中的任何内容都不存在于公共网络上，因此它可以很好地代表你自己的私有语料库：Gemini 能够回答关于它的问题的唯一方式是通过文件搜索检索它。
HANDBOOK = """
Zephyr Robotics — Employee Handbook (Internal)
About the company
Company Name: Zephyr Robotics
Founded: 2019
Headquarters: San Francisco, CA
Time off
PTO Policy: Every full-time employee receives 20 days of paid time off per year.
Unused PTO does not roll over into the next calendar year.
Benefits
Health Insurance: Zephyr Robotics covers 90% of medical, dental, and vision premiums
for employees and their dependents.
Working arrangements
Remote Work: Zephyr Robotics runs a hybrid schedule of 3 days remote and 2 days in the
San Francisco office. Teams pick their own in-office days.
"""
with open("sample_handbook.txt", "w") as f:
f.write(HANDBOOK)
print(HANDBOOK.strip())
创建文件搜索存储并上传文档
文件搜索存储用于保存您的文档及其嵌入向量。上传会启动一个长时间运行的摄取操作（分块和嵌入），因此请轮询该操作，直到其状态报告为 done。
file_search_store = client.file_search_stores.create(
config=types.CreateFileSearchStoreConfig(
display_name="Zephyr Robotics Handbook Store"
)
)
# Keep the store name around; every File Search call and the cleanup step needs it.
FILE_SEARCH_STORE_NAME = file_search_store.name
print(f"Created store: {FILE_SEARCH_STORE_NAME}")
import time
upload_op = client.file_search_stores.upload_to_file_search_store(
file_search_store_name=FILE_SEARCH_STORE_NAME,
file="sample_handbook.txt",
config=types.UploadToFileSearchStoreConfig(
display_name="Zephyr Robotics Employee Handbook",
),
)
print(f"Ingestion started: {upload_op.name}")
# Wait for chunking and embedding to finish before querying the store.
while not (upload_op := client.operations.get(upload_op)).done:
time.sleep(2)
print(".", end="")
print()
print("Ingestion complete.")
从响应中读取引用
两种 grounding 工具都以相同的方式报告其来源，即通过 response.candidates[0].grounding_metadata。区别在于 grounding_chunks 中每个条目的哪个字段被填充：
* File Search 填充 chunk.retrieved_context —— 检索到的 text、文档 title，通常还有 page_number。
* Google Search 填充 chunk.web —— 页面的 uri 和 title。
编写一个小辅助函数，将两种形状规范化为普通字典，再编写另一个辅助函数来打印它们。你将在两种路由模式中复用这些函数。
def extract_citations(response):
"""Normalizes File Search and Google Search grounding chunks into dicts."""
citations = []
if response is None:
return citations
candidates = response.candidates or []
if not candidates:
return citations
metadata = candidates[0].grounding_metadata
if metadata is None:
return citations
for chunk in metadata.grounding_chunks or []:
if chunk.retrieved_context is not None:
snippet = chunk.retrieved_context.text or ""
page_num = chunk.retrieved_context.page_number
location = (
f"page {page_num}" if page_num is not None else "retrieved chunk"
)
citations.append({
"source_type": "document",
"title": chunk.retrieved_context.title,
"location": location,
"snippet": snippet[:200],
})
elif chunk.web is not None:
citations.append({
"source_type": "web",
"title": chunk.web.title,
"location": chunk.web.uri,
"snippet": "",
})
return citations
def show(label, response):
"""Prints a response together with its normalized citations."""
if response is None:
print("\n[no response returned]")
return
print(f"=== {label} ===")
print((response.text or "").strip())
citations = extract_citations(response)
if not citations:
print("\n[no grounding citations returned]")
return
print("\nSources:")
for i, citation in enumerate(citations, start=1):
print(f" [{i}] ({citation['source_type']}) {citation['title']}")
print(f" {citation['location']}")
if citation["snippet"]:
print(f" \"{citation['snippet']}...\"")
模式 1：确定性回退路由
在此模式中，*你的代码*决定哪个来源回答问题，分两步：
- 仅启用文件搜索工具询问 Gemini。
- 用简单的启发式规则判断结果。如果你的文档明显未覆盖问题，则仅启用 Google 搜索工具重新提问。
下面的启发式规则刻意保持基础——没有评分模型，只有两个检查，既便宜又易于解释：
* 无检索结果。 如果 grounding_chunks 为空，说明文件搜索未找到相关内容。
* 明确的不回答。 如果模型检索到了内容，但其回复称文档中不包含答案，也视为不足。
当你需要保证私有文档始终被优先查阅、希望记录路由决策以供审计，或当网络访问必须是明确、可观察的升级而非模型选择时，确定性路由值得额外调用。
NO_ANSWER_PHRASES = (
"does not contain",
"doesn't contain",
"no information",
"not mentioned",
"not specified",
"unable to answer",
"cannot answer",
"i don't know",
)
def is_sufficient(response):
"""Simple heuristic: did File Search actually ground this answer?"""
if response is None:
return False, "No response returned."
citations = extract_citations(response)
if not citations:
return False, "File Search returned no chunks."
text = (response.text or "").lower()
if any(phrase in text for phrase in NO_ANSWER_PHRASES):
return False, "The model reported that the documents do not cover the question."
return True, f"Grounded in {len(citations)} document chunk(s)."
def ask_with_fallback(question, store_name=None, verbose=True):
"""Queries File Search first, then falls back to Google Search if needed."""
store_name = store_name or FILE_SEARCH_STORE_NAME
try:
doc_response = client.models.generate_content(
model=MODEL_ID,
contents=question,
config=types.GenerateContentConfig(
tools=[types.Tool(
file_search=types.FileSearch(
file_search_store_names=[store_name]
)
)],
),
)
except Exception as e: # Surface API errors without stopping the notebook.
print(f"File Search call failed: {type(e).__name__}: {e}")
return None, "error"
sufficient, reason = is_sufficient(doc_response)
if verbose:
print(f"Router: File Search -> {'sufficient' if sufficient else 'insufficient'}")
print(f" {reason}")
if sufficient:
return doc_response, "file_search"
if verbose:
print("Router: falling back to Google Search.")
try:
web_response = client.models.generate_content(
model=MODEL_ID,
contents=question,
config=types.GenerateContentConfig(
tools=[types.Tool(google_search=types.GoogleSearch())],
),
)
except Exception as e:
print(f"Google Search call failed: {type(e).__name__}: {e}")
return None, "error"
return web_response, "google_search"
问题1：由你的文档回答
PTO政策只存在于员工手册中，因此文件搜索应返回相关片段，路由器应在步骤1后停止。
question_1 = "What is Zephyr Robotics' PTO policy?"
response_1, route_1 = ask_with_fallback(question_1)
print()
if response_1 is not None:
show(f"Route: {route_1}", response_1)
问题2：你的文档未回答的问题
手册中没有任何内容提到宏观经济学，因此启发式方法应将文件搜索尝试标记为不足，并升级到谷歌搜索。请注意，返回的引用是网页而不是文档片段。
question_2 = "What is the current US inflation rate?"
response_2, route_2 = ask_with_fallback(question_2)
print()
if response_2 is not None:
show(f"Route: {route_2}", response_2)
模式 2：原生多工具路由
与其在自己的代码中进行路由，不如在同一请求中将两个工具都交给 Gemini，让它自行决定。
模型会读取问题，调用合适的工具，并且当问题同时涉及私有和公共知识时，可以同时调用两个工具。
这样只需一次请求，而不是最多两次，并且能够处理确定性路由需要手动拆分的混合问题。
代价是路由决策现在由模型内部做出，因此你需要事后通过阅读 grounding_metadata 来了解发生了什么，而不是事先决定。
HYBRID_TOOLS = [
types.Tool(
file_search=types.FileSearch(
file_search_store_names=[FILE_SEARCH_STORE_NAME]
)
),
types.Tool(google_search=types.GoogleSearch()),
]
def ask_hybrid(question):
"""Passes both grounding tools and lets Gemini choose."""
try:
return client.models.generate_content(
model=MODEL_ID,
contents=question,
config=types.GenerateContentConfig(tools=HYBRID_TOOLS),
)
except Exception as e:
print(f"Hybrid call failed: {type(e).__name__}: {e}")
return None
再次提出同样两个问题。答案应与模式1产生的结果一致，但现在每个问题只需一次请求即可完成。
for question in (question_1, question_2):
response = ask_hybrid(question)
if response is not None:
show(question, response)
print()
一个需要两种来源的问题
由于两个工具同时在线，模型可以将答案的一部分基于你的手册，另一部分基于网络。提出一个刻意跨越这两者的问题。
blended_question = (
"Zephyr Robotics covers 90% of health insurance premiums. "
"How does that compare to the average employer contribution in the US?"
)
blended_response = ask_hybrid(blended_question)
if blended_response is not None:
show("Blended question", blended_response)
检查原始 grounding 元数据
show 辅助函数为了可读性简化了引用，但值得查看一次原始对象，以便了解还有哪些其他信息可用。除了 grounding_chunks 之外，元数据还包含：
* grounding_supports — 生成文本中映射回支持它们的块索引的片段，这是渲染行内脚注标记所需的。
* web_search_queries — Gemini 实际向 Google 搜索发出的查询。
* retrieval_queries — 针对你的文件搜索存储发出的查询。
这些信息共同告诉你模型使用了*什么*以及*它是如何查找的*。
import json
def describe_grounding(response):
"""Dumps the interesting parts of grounding_metadata."""
if response is None:
print("No response returned.")
return
candidates = response.candidates or []
if not candidates or candidates[0].grounding_metadata is None:
print("No grounding metadata on this response.")
return
metadata = candidates[0].grounding_metadata
print(f"web_search_queries: {metadata.web_search_queries}")
print(f"retrieval_queries: {metadata.retrieval_queries}")
print(f"grounding_chunks: {len(metadata.grounding_chunks or [])}")
print(f"grounding_supports: {len(metadata.grounding_supports or [])}")
print("\nFirst chunk, as JSON:")
if metadata.grounding_chunks:
print(json.dumps(
metadata.grounding_chunks[0].model_dump(exclude_none=True),
indent=2,
default=str,
)[:800])
print("\nText spans and the chunks that support them:")
for support in (metadata.grounding_supports or [])[:5]:
segment_text = (support.segment.text or "")[:80]
print(f" {support.grounding_chunk_indices} <- \"{segment_text}\"")
if blended_response is not None:
describe_grounding(blended_response)
清理
文件搜索存储会持久保存并计入存储空间，因此完成后请删除该存储。本地示例文件也会随之消失。
import os
try:
client.file_search_stores.delete(
name=FILE_SEARCH_STORE_NAME,
config=types.DeleteFileSearchStoreConfig(force=True),
)
print(f"Deleted store: {FILE_SEARCH_STORE_NAME}")
except Exception as e:
print(f"Could not delete store: {type(e).__name__}: {e}")
if os.path.exists("sample_handbook.txt"):
os.remove("sample_handbook.txt")
print("Deleted local file: sample_handbook.txt")
print("Cleanup complete.")
下一步
将本教程适配到您自己的文档
将生成的示例手册替换为您的真实语料库，笔记本的其余部分无需更改即可正常工作：
使用 upload_to_file_search_store 上传每个文件，并附加自定义元数据（如团队或文档类型），以便您可以通过 metadata_filter 缩小检索范围。如果您的回退触发过于频繁，请收紧 is_sufficient —— 要求最小块数，或检查检索到的片段是否与问题共享关键词，这些都是在使用评分模型之前合理的下一步。
了解更多
* Google 搜索接地指南，包括内联引用归属
* 文件搜索快速入门，了解元数据过滤和多模态存储
* 接地快速入门，了解地图、YouTube 和 URL 上下文
* 搜索接地示例，了解更长的研究报告工作流程
* 在 Gemini API Cookbook 中查看更多想法
常见问题
常见错误：1) API 密钥未设置或权限不足，导致认证失败；2) 文件搜索存储未创建或名称错误，引发 NotFound；3) 上传文档格式不支持，需检查文件类型；4) 摄取未完成即查询，返回空结果；5) 启发式规则误判，导致不必要的回退或遗漏。排查方法：检查 API 密钥环境变量，确认存储名称正确，轮询摄取状态，打印 grounding_metadata 调试。生产优化：使用元数据过滤缩小检索范围，调整 is_sufficient 阈值（如要求最小块数），采用评分模型替代简单启发式，缓存高频查询结果，监控 API 调用成本。
企业应用扩展
可扩展方向：1) 将示例手册替换为真实语料库，利用自定义元数据（如团队、文档类型）进行过滤；2) 结合 Interactions API 实现多轮对话中的动态路由；3) 集成其他接地工具（如地图、YouTube）丰富知识源；4) 构建评估集，量化路由准确率与回答质量；5) 实现流式输出，提升用户体验；6) 部署为 REST 服务，支持高并发。参考官方文档：文件搜索指南、Google 搜索接地指南、文件搜索快速入门、接地快速入门。
总结
本教程面向有编程基础的开发者，深入讲解如何结合 Gemini API 的文件搜索（私有知识）与 Google 搜索接地（公共知识）构建混合 RAG 流水线。通过两种路由模式——确定性回退路由和原生多工具路由，实现知识源的灵活选择与融合。教程涵盖环境配置、文档上传、引用提取、元数据解析及清理流程，并强调 grounding_metadata 在溯源中的关键作用。同时提供常见错误排查与生产环境优化建议，帮助开发者构建可审计、可扩展的 AI 应用。
- --
内容来源与验证
- 内容等级：PREMIUM_A
- 质量等级：S_CANDIDATE
- 难度：中级
- 验证等级：V1
- 原文作者：
- 原文日期：
- 来源许可：Apache-2.0
- 来源版本：
4723788811a38911cbc502f0e9a28e68a1ac315d - 查看原文
站内还有更多完整实战教程
加入会员，解锁全部教程、完整代码与持续更新内容。
加入会员