用 ADK 与 Agent Platform Vector Search 2.0 构建可落地的 RAG 问答 Agentcore/rag-vector-search 实战指南【免费下载链接】adk-samplesA collection of sample agents built with Agent Development Kit (ADK)项目地址: https://gitcode.com/GitHub_Trending/ad/adk-samples本指南围绕开源仓库adk-samples中的 core/rag-vector-search 示例展开讲解如何基于 Agent Development KitADK构建一个文档索引 语义检索 大模型回答的检索增强生成RAG问答 Agent。读完本文你将掌握 Vector Search 2.0 Collection 的 Terraform 创建、Kubeflow PipelinesKFP数据摄取管道的编排与参数调优、ADK 工具化检索的实现方式以及 CI/CD 定时调度与测试策略的完整实战方案。示例概览一个单 Agent 语义检索工具的 RAG 参考实现rag-vector-search是一个起步级starterRAG Agent其核心思路是回答用户问题时先从索引在Agent Platform Vector SearchVector Search 2.0中的文档集合里检索相关片段再让大模型基于检索上下文作答。文档的加载、分块与导入由一套 KFP 摄取管道完成而向量嵌入由 Collection 上配置的嵌入模型在摄取时自动生成无需手动调用嵌入服务。原 README 中给出的 Agent 属性表如下可作为评估该示例定位的参考属性详情交互类型对话式Conversational复杂度中级IntermediateAgent 类型单 AgentSingle Agent组件工具Tools、RAG、Terraform、评估Evaluation、摄取管道Ingestion Pipeline工作原理四大模块如何协作从源码结构看整个示例由四个目录协同完成建索引—喂数据—检索—回答的闭环app/agent.py— ADK Agent 定义将retrieve_docs工具接线到对话模型app/retrievers.py—search_collection()通过vectorsearch_v1beta对 Vector Search 2.0 Collection 执行语义搜索设置INTEGRATION_TESTTRUE时返回 mock 数据data_ingestion/— KFP 摄取管道data_ingestion_pipeline/负责加载、分块并导入文档到 Collection通过make>root_agent Agent( nameroot_agent, modelGemini( modelLLM, retry_optionstypes.HttpRetryOptions(attempts3), ), instructioninstruction, tools[retrieve_docs], ) app App( root_agentroot_agent, nameapp, )系统提示词instruction定义了问答助手的边界优先使用提供的上下文作答可借助工具获取补充信息若已知道答案也可以直接回答而不调用工具。模型层配置了 3 次 HTTP 重试增强线上稳定性。语义检索器app/retrievers.pyapp/retrievers.py 中的search_collection(query, collection_path, top_k10)是检索核心集成测试模式下INTEGRATION_TEST TRUE直接返回 mock 的Document 0片段避免测试依赖真实云端资源正常模式下构建vectorsearch_v1beta.SearchDataObjectsRequest使用SemanticSearch语义搜索search_fieldtext_embedding指向 Collection 的自动嵌入字段task_typeRETRIEVAL_QUERY标记这是检索查询类型的嵌入请求top_k默认返回 10 条结果output_fields.data_fields[question_id, text_chunk, full_text_md]指定返回的文档字段结果被格式化为Document {i}...\n/Document {i}的结构化文本注入上下文无结果时返回No relevant documents found.。该检索结果格式与摄取管道写入的数据字段一一对应详见下文process_data组件保证写入什么、检索到什么、模型看到什么三者一致。环境准备.env 配置逐项说明项目通过 .env.example 模板管理运行配置cp .env.example .env后逐项填写。文件按用途分为三段Agent 运行时段app/agent.py 读取变量默认值/示例说明GOOGLE_CLOUD_PROJECT必填GCP 项目 ID不填则尝试从 ADC 解析GOOGLE_CLOUD_LOCATIONglobalVertex AI 模型调用地域GOOGLE_GENAI_USE_VERTEXAITrue走 Vertex AI 后端而非 Gemini APIVECTOR_SEARCH_COLLECTIONprojects/项目/locations/us-central1/collections/rag-vector-search-collectionCollection 完整资源路径由 infra 创建MODEL_NAMEgemini-flash-latest问答大模型名称MODEL_NAME_EMBEDDINGgemini-embedding-001Collection 的自动嵌入模型供建库脚本使用数据摄取段make>cp .env.example .env按上文表格替换TODO: update-this-value占位值尤其保证项目 ID 与 Collection 路径一致。Step 2创建 Collectionmake setup-infra先编辑 infra/terraform/vars/env.tfvars该文件本身是示例注释提示复制到自己的 tfvars 再填写project_id your-gcp-project-id project_name rag-vector-search region us-central1然后执行make setup-infra对应 Makefile 中的实现为cd infra/terraform terraform init terraform apply -var-filevars/env.tfvars即初始化并应用 infra/terraform 下的全部资源定义。Terraform 配置会调用 setup_vector_search_collection.py 创建 Collection并额外创建管道所需的 GCS 存储桶。这里值得展开的是 Collection 的创建细节。该脚本是幂等的——创建前先get_collection若已存在则直接跳过。创建请求同时声明了数据 Schema 与向量 Schemarequest vectorsearch_v1beta.CreateCollectionRequest( parentparent, collection_idcollection_id, collection{ data_schema: { type: object, properties: { question_id: {type: string}, text_chunk: {type: string}, full_text_md: {type: string}, }, }, vector_schema: { text_embedding: { dense_vector: { dimensions: 3072, vertex_embedding_config: { model_id: os.getenv(MODEL_NAME_EMBEDDING), text_template: {text_chunk}, task_type: RETRIEVAL_DOCUMENT, }, }, }, }, }, )要点有三其一data_schema定义每个数据对象携带question_id、text_chunk、full_text_md三个字符串字段与检索时的output_fields对齐其二vector_schema声明text_embedding为 3072 维稠密向量由MODEL_NAME_EMBEDDING默认gemini-embedding-001自动生成text_template{text_chunk}指明嵌入的文本来源是分块内容task_typeRETRIEVAL_DOCUMENT匹配文档侧的嵌入任务类型其三正因为嵌入由 Collection 托管摄取管道完全不需要自己生成向量。Step 3摄取文档make>make>set -a . ./.env set a \ cd data_ingestion uv run python data_ingestion_pipeline/submit_pipeline.py --local \ --project $PROJECT_ID --region $REGION --collection-id $VECTOR_SEARCH_COLLECTION_ID本地模式借助 KFPSubprocessRunnerlocal.init(runnerlocal.SubprocessRunner(use_venvFalse))把管道组件作为本地子进程执行无需部署 Vertex AI Pipelines 基础设施适合开发与验证阶段。深入数据摄取管道加载、清洗、分块、去重、导入摄取管道位于 data_ingestion/data_ingestion_pipeline由编排文件与两个组件构成。管道编排pipeline.pypipeline.py 定义了一个两步 KFP 管道先process_data处理数据再ingest_data导入数据两个步骤都设置了set_retry(num_retries2)。管道参数及其默认值如下参数默认值说明project_id/location必传GCP 项目与地域schedule_time1970-01-01T00:00:00Z调度时间用于增量窗口计算默认值会被识别为未设置并替换为当前时间is_incrementalTrue是否只处理最近look_back_days天的数据look_back_days1增量回看天数chunk_size1500文本分块大小字符数chunk_overlap20相邻分块重叠字符数max_rows100最大抓取行数0 表示不限destination_datasetrag_vector_search_qa_dataBigQuery 结果数据集destination_tableincremental_questions_embeddings增量结果表deduped_tablequestions_embeddings去重后的最终表collection_id目标 Vector Search 2.0 Collection IDingestion_batch_size250单次批量创建的数据对象数清洗与分块process_data 组件process_data.py 是一个自包含的 KFP 组件base_imagepython:3.11-slim依赖bigframes、langchain-text-splitters、markdownify等其文件头注释明确说明开箱即用的情况下它会在 BigQuery 中内联生成一个合成 QA 数据集12 个 Python 编程问答以便管道无需任何外部数据即可端到端跑通接入真实数据时只需把fetch_sample_data中的 CTE 换成源表查询并适配下游清洗逻辑。整个处理流程分五步取数按schedule_time与look_back_days计算日期窗口从 BigQuery 读取合成问答数据is_incremental为真时用TIMESTAMP_TRUNC(creation_date, DAY) BETWEEN ...过滤窗口max_rows 0时追加LIMIT清洗用markdownify把 HTML 问题正文与答案转成 Markdownconvert_html_to_markdown再拼出full_text_md 标题(H1) 正文 各答案(H2)同时按question_id排序去重分块使用RecursiveCharacterTextSplitter(chunk_sizechunk_size, chunk_overlapchunk_overlap)切分full_text_md生成稳定块 ID将每个question_id与块序号拼接为question_id__索引。注释指出这是幂等设计的关键——重跑管道会重建相同的数据对象摄取步骤会跳过已存在的对象避免重复积累代价是若某文档块数变少历史遗留的孤儿块不会被自动删除落库创建按creation_timestamp按天分区TimePartitioningType.DAY的 BigQuery 表增量表用append写入再按question_id取最新creation_timestamp生成去重表replace写入最后通过output_table工件元数据把bq://...URI 传给下游。注意第 3、4 步之间没有生成嵌入——代码用注释明确写出 No embedding generation needed — Vector Search 2.0 auto-generates embeddings这正是该示例与经典先嵌入再入库方案的最大区别。批量导入ingest_data 组件ingest_data.py 读取上游 BigQuery 表中的question_id、full_text_md、text_chunk、chunk_id四列然后对 Vector Search 2.0 Collection 分批执行batch_create_data_objects单批上限为 250 个数据对象代码batch_size min(ingestion_batch_size, 250)注释说明这是自动嵌入请求的硬上限默认参数 250 恰好就是该上限每个数据对象以chunk_id作为data_object_id幂等重试的依据data字段携带question_id、text_chunk、full_text_md而vectors传空字典{}交由 Collection 自动生成嵌入捕获AlreadyExists异常时跳过整批并计数最终日志输出Ingestion complete. {created} created, {skipped} skipped。启动与运行 Agent基础设施与数据就绪后按原 README 运行make install make playgroundmake install执行uv sync --dev --extra eval含评估依赖make playground执行uv run adk web . --port 8501 --reload_agents——即在 8501 端口拉起 ADK Web 调试台并开启 Agent 热重载启动后在界面中选择app文件夹即可与该 RAG Agent 对话。此外make lintuv run ruff check . 格式检查可用于代码质量校验。测试策略集成测试与凭据要求原 README 明确make test运行的是集成测试。其实现位于 tests/integration/test_agent.pyMakefile中的对应命令为uv sync --dev INTEGRATION_TESTTRUE uv run pytest tests/integration测试的关键设计是**检索器被 mock模型是真实调用**通过INTEGRATION_TESTTRUE让search_collection()返回固定 mock 数据但 Agent 仍会发起一次真实的 Gemini 调用因此要求环境具备 Google Cloud ADC 凭据与 Vertex AI 访问权限。测试用google.auth.default()探测凭据_has_gcp_credentials()返回 False 时整个用例被pytest.mark.skipif跳过导入app.agent也刻意放在函数内部避免无凭据环境下模块导入即失败。用例通过InMemorySessionServiceRunner以StreamingMode.SSE流式模式发送提问断言至少产生一条带文本内容的事件。CI/CD定时调度数据摄取生产环境下数据摄取应周期性执行而非手动触发。deployment/cloudbuild.yaml 负责在 Vertex AI Pipelines 上**重新调度**摄取管道它通过SCHEDULE_ONLYTRUE只创建/更新一个周期性的PipelineJobSchedule并不内联运行管道。可将其接入 Cloud Build 触发器例如 main 分支合并时触发或手动提交gcloud builds submit --config deployment/cloudbuild.yaml \ --substitutions\ _PROD_PROJECT_IDmy-project,\ _REGIONus-central1,\ _VECTOR_SEARCH_COLLECTION_IDrag-vector-search-collection,\ _PIPELINE_GCS_ROOTgs://my-project-rag-vector-search-rag,\ _PIPELINE_SA_EMAILrag-vector-search-ragmy-project.iam.gserviceaccount.com,\ _PIPELINE_NAMErag-vector-search-ingestion,\ _CRON_SCHEDULE0 2 * * *存储桶与服务账号由make setup-infra创建名称见 Terraform 的pipeline_gcs_bucket_name输出。构建步骤会现场安装 uv然后在data_ingestion目录执行submit_pipeline.py并通过环境变量注入替换值替换变量含义默认值_PROD_PROJECT_ID生产项目 IDYOUR_PROD_PROJECT_ID_REGIONVertex AI Pipelines 地域us-central1_VECTOR_SEARCH_LOCATIONVector Search 地域us-central1_VECTOR_SEARCH_COLLECTION_IDCollection IDrag-vector-search-collection_PIPELINE_GCS_ROOT管道根目录GCSgs://YOUR_PIPELINE_BUCKET_PIPELINE_SA_EMAIL管道服务账号占位值_PIPELINE_NAME管道显示名rag-vector-search-ingestion_CRON_SCHEDULE调度 cron 表达式0 2 * * *每天凌晨 2 点同时固定DISABLE_CACHINGTRUE、SCHEDULE_ONLYTRUE。submit_pipeline.py本身承担三种执行模式可通过命令行参数切换均支持环境变量兜底--local本地运行只要求 project/region/collection-id默认远程模式提交PipelineJob并等待完成要求补齐--service-account、--pipeline-root、--pipeline-name外层用backoff指数退避重试最多 3 次、最长 1 小时--schedule-only结合--cron-schedule时创建或更新PipelineJobSchedule按display_name精确过滤查找既有调度存在则update(cron...)否则create。调度模式还通过dsl.PIPELINE_JOB_SCHEDULE_TIME_UTC_PLACEHOLDER把调度时间占位符注入schedule_time供增量窗口计算使用。部署 Agent 本体数据摄取是调度化的Agent 本身则走 ADK 原生部署路径例如uv run adk deploy cloud_run .或adk deploy agent_engineAgent Engine 托管模式。原 README 特别提示不要使用已废弃的 agent-starter-pack 工具链。部署前请确保.env中的VECTOR_SEARCH_COLLECTION与MODEL_NAME指向真实资源。相关示例与选型对比原 README 末尾给出了两条参考路线core/rag-agent-search —— 同一 RAG 思路的Agent Platform SearchDiscovery Engine变体通过 GCS 数据连接器同步文档适合想用托管搜索服务而非自管 Collection 的场景contrib/python/multiformat-hybrid-rag —— 面向生产的**混合检索语义 关键词**系统适合对召回精度有更高要求的线上环境。对比可见本示例的定位用最少的组件单 Agent 一个检索工具 一条 KFP 管道打通 Vector Search 2.0 全链路是理解自动嵌入型 RAG与 ADK 工具化检索的最佳起点其上可继续演进为多 Agent 编排、混合检索或生产级增量管道相关文件AGENTS.md、manifest.yaml、Makefile均可作为深入阅读的入口。【免费下载链接】adk-samplesA collection of sample agents built with Agent Development Kit (ADK)项目地址: https://gitcode.com/GitHub_Trending/ad/adk-samples创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考