agent-platform-rag-engine-management

Agent Platform RAG 引擎管理

本技能介绍如何使用 Agent Platform Python SDK 与 Agent Platform RAG 引擎交互。你必须使用 vertexai Python SDK 执行 RAG 引擎操作，而非直接调用 REST 或使用 MCP 工具， 因为此代码旨在由外部客户端运行。

安全与确认等级（重要）

在代表用户执行任何命令或脚本之前，必须根据所请求的操作遵守以下安全等级：

1. 等级 R：只读（list_corpora、list_files、get_corpus、 retrieval_query）

   • 无需确认。立即执行以收集信息或检索有依据的上下文。

2. 等级 RC：只读，但会消耗计算资源 （client.models.generate_content）

   • 执行有依据的内容生成前，必须通过“是”/“否”选项获得交互式确认。确认提示必须清楚说明拟执行的生成操作及其关键参数 （例如目标语料库 ID、查询文本、目标模型）。未明确列出确切参数的自然语言转述并不足够，因为必须明确列出参数，才能确保用户对具体资源和配置作出无歧义的批准。

   • 同一轮限制：展示确认提示的同一轮中，不得执行生成代码。暂停并等待用户回复；只有在用户明确回复“是”/批准后才能执行。

   • 标准示例：

     我将根据以下参数执行有依据的内容生成。请先确认这些信息，然后我再继续：

     • 目标语料库 ID：projects/123/locations/us/ragCorpora/abc
     • 目标模型：gemini-2.5-pro
     • 查询文本：“公司的远程办公政策是什么？”

     是否确认？[是/否]

阶段 0：环境设置

重要：运行以下任何 Python 代码片段之前，必须按照以下步骤确保环境已正确初始化：

1. Google Cloud 身份验证：使用你的 Google Cloud 凭据进行身份验证，并配置用于访问 Agent Platform 的有效应用默认凭据（ADC）：

   gcloud auth login
   gcloud auth application-default login
   

2. Python 依赖项：此技能需要 google-cloud-aiplatform 和 google-genai，沙盒已提供这两个依赖项。不要创建虚拟环境——虚拟环境初始为空，会隐藏环境已提供的包，导致不必要的重复安装。不要另行执行命令检查这些依赖项：直接运行代码片段，只有在出现 ModuleNotFoundError 时，才在重试的同一条命令中安装：

   pip install -q google-cloud-aiplatform google-genai && python3 - <<'PY'
   ...
   PY
   

3. 执行：将每个代码片段作为带引号的 heredoc 运行，即 python3 - <<'PY' ... PY，而不是使用 python3 -c '...'，因为后者的嵌套引号很容易出错。无需先激活任何环境。如果某个步骤需要运行以下多个代码片段（例如，先列出语料库，再列出每个语料库中的文件），请将它们合并为一个脚本并通过一条命令运行。vertexai.preview.rag 弃用警告会打印到 stderr，这是预期行为；请继续使用这些代码片段，不要因此改用其他客户端。

工作流决策树

1. 信息收集：用户是否提供了 Project ID、Region 和 Corpus ID？

   • 否 -> 前往 [1. 列出语料库和文件]，查找所需的资源名称和 ID。只有在查找失败时才询问用户。
   • 是 -> 继续。

2. 任务类型：用户想要做什么？

   • 列出语料库和文件 -> 前往 [1. 列出语料库和文件]。
   • 检查语料库 -> 前往 [2. 获取/检查 RAG Engine 语料库]。
   • 搜索上下文 -> 前往 [3. 检索上下文]。
   • 使用 RAG Engine 回答问题 -> 前往 [4. 使用检索到的上下文回答用户]。

提示

**占位参数替换：**以下 Python 脚本使用带方括号的字符串占位符（例如 "{project_id}"、"{region}" 和 "{corpus_id}"）。在生成、提供或执行脚本之前，必须根据用户提示词（或当前上下文）中提供的实际 Project ID、Region 和 Corpus ID 值，动态替换这些占位符。

1. 列出语料库和文件（发现）

如果不知道语料库或文件的资源名称，必须先列出它们以查找所需信息。转换为列表时，SDK 会自动处理分页；对于大型集合，也可以手动分页。

1.1 列出并查找语料库

import vertexai
from vertexai.preview import rag

vertexai.init(project="{project_id}", location="{region}")

# Approach A: List ALL (Automatic Pagination)
# The SDK's Pager iterates through all pages for you.
all_corpora = list(rag.list_corpora())
print(f"Found {len(all_corpora)} corpora in total.")
for c in all_corpora:
    print(f"Corpus Name: {c.name} | Display Name: {c.display_name}")

# Approach B: Manual Pagination (for very large projects)
pager = rag.list_corpora(page_size=10)
# Process first page
for c in pager:
    print(f"Corpus: {c.display_name}")

# Get next page if needed
if pager.next_page_token:
    second_page = rag.list_corpora(
        page_size=10, page_token=pager.next_page_token
    )

1.2 列出并查找文件

要了解语料库中的文件及其类型，请列出文件并检查 display_name（通常包含扩展名）。

import vertexai
from vertexai.preview import rag

vertexai.init(project="{project_id}", location="{region}")
corpus_name = (
    "projects/{project_id}/locations/{region}/ragCorpora/{corpus_id}"
)

# List files with automatic pagination
files = list(rag.list_files(corpus_name=corpus_name))
print(f"Found {len(files)} files.")

for f in files:
    # High-level SDK RagFile objects usually have name, display_name,
    # description
    print(f"File: {f.display_name} | Resource: {f.name}")
    # Tip: Check extension to understand file type (PDF, TXT, etc.)
    if f.display_name.lower().endswith(".pdf"):
        print("  Type: PDF")
    elif f.display_name.lower().endswith(".txt"):
        print("  Type: Plain Text")

2. 获取/检查 Agent Platform RAG Engine 语料库

要获取现有 Agent Platform RAG Engine 语料库的详细信息：

import vertexai
from vertexai.preview import rag

vertexai.init(project="{project_id}", location="{region}")

# To get details of a specific corpus
corpus_name = (
    "projects/{project_id}/locations/{region}/ragCorpora/{corpus_id}"
)
corpus = rag.get_corpus(name=corpus_name)
print(f"Corpus Name: {corpus.name}")
print(f"Display Name: {corpus.display_name}")

3. 检索上下文

根据查询，从 RAG Engine 语料库中检索相关上下文。如果没有返回任何上下文，这是有效的结果：请报告所用的语料库和查询。如果适合进行范围更广的搜索，请在同一个脚本中尝试（例如增大 similarity_top_k），而不是作为单独的命令执行。

import vertexai
from vertexai.preview import rag

vertexai.init(project="{project_id}", location="{region}")

corpus_name = (
    "projects/{project_id}/locations/{region}/ragCorpora/{corpus_id}"
)
query = "What is the speed of light?"

# Retrieve contexts
response = rag.retrieval_query(
    rag_corpora=[corpus_name],
    text=query,
    similarity_top_k=3
)

for context in response.contexts.contexts:
    print(f"Context text: {context.text}")
    print(f"Source: {context.source_uri}")

4. 使用检索到的上下文回答用户

要结合检索到的上下文和 Agent Platform 模型，生成有依据的回答：

from google import genai
from google.genai import types

client = genai.Client(enterprise=True, project="{project_id}", location="{region}")
corpus_name = (
    "projects/{project_id}/locations/{region}/ragCorpora/{corpus_id}"
)

# Define the Agent Platform RAG Engine tool pointing to the corpus
rag_tool = types.Tool(
    retrieval=types.Retrieval(
        vertex_rag_store=types.VertexRagStore(
            rag_resources=[types.VertexRagStoreRagResource(rag_corpus=corpus_name)],
            rag_retrieval_config=types.RagRetrievalConfig(
                top_k=3,
                filter=types.RagRetrievalConfigFilter(
                    vector_similarity_threshold=0.5,
                ),
            ),
        )
    )
)

# Generate content using the RAG Engine tool
response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents="What is the speed of light?",
    config=types.GenerateContentConfig(
        tools=[rag_tool]
    )
)
print(response.text)