agent-platform-rag-engine-management

Agent Platform RAG Engine の管理

このスキルでは、Agent Platform Python SDK を使用して Agent Platform RAG Engine を操作する方法を説明します。このコードは外部クライアントによる実行を 想定しているため、RAG Engine の操作には生の REST 呼び出しや MCP ツールではなく、 必ず vertexai Python SDK を使用してください。

安全性と確認のティア（重要）

ユーザーに代わってコマンドやスクリプトを実行する前に、要求された操作に基づき、 以下の安全性ティアを遵守する必要があります。

1. ティア R：読み取り専用（list_corpora、list_files、get_corpus、 retrieval_query）

   • 確認は不要です。情報の収集や根拠のあるコンテキストの取得のため、 すぐに実行してください。

2. ティア RC：読み取り専用だがコンピューティングリソースを消費する （client.models.generate_content）

   • 根拠のあるコンテンツ生成を実行する前に、「Yes」／「No」の選択肢を含む 対話形式の確認が必要です。確認プロンプトでは、実行する生成処理と その主要なパラメーター（例：対象コーパス ID、クエリテキスト、対象モデル）を 明確に説明する必要があります。正確なパラメーターを指定しない自然言語での 言い換えでは不十分です。ユーザーが特定のリソースと設定を明確に承認できるよう、 パラメーターを明示する必要があります。

   • 同一ターンでの実行制限：確認プロンプトを提示した同じターン内に、 生成コードを実行してはいけません。ユーザーの返答を待ち、明示的な「Yes」／ 承認を得てから実行してください。

   • 模範例：

     以下のパラメーターで、根拠に基づくコンテンツ生成を実行します。 続行する前に、内容をご確認ください。

     • 対象コーパス ID：projects/123/locations/us/ragCorpora/abc
     • 対象モデル：gemini-2.5-pro
     • クエリテキスト："What are the company policies on remote work?"

     承認しますか？［Yes/No］

フェーズ 0：環境のセットアップ

重要：以下の Python スニペットを実行する前に、次の手順に従って 環境が正しく初期化されていることを確認してください。

1. Google Cloud 認証：Google Cloud の認証情報で認証し、Agent Platform に アクセスするための有効な Application Default Credentials（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. 実行：各スニペットは python3 -c '...' ではなく、引用符付きヒアドキュメント python3 - <<'PY' ... PY として実行してください。後者は入れ子の引用符で 簡単に問題が起きるためです。事前に有効化する環境はありません。以下のスニペットを 複数使う手順（たとえば、コーパスを一覧表示してから、それぞれのファイルを一覧表示 する場合）は、1 つのスクリプトと 1 つのコマンドにまとめてください。 標準エラー出力に表示される vertexai.preview.rag の非推奨警告は想定どおりです。 この警告を理由にクライアントを切り替えず、引き続きこれらのスニペットを使用して ください。

ワークフローの判断ツリー

1. 情報の収集：ユーザーはプロジェクト ID、リージョン、コーパス ID を指定していますか？

   • いいえ -> [1. コーパスとファイルの一覧表示] に進み、必要なリソース名と ID を見つけます。見つからない場合にのみ、ユーザーに確認してください。
   • はい -> 次に進みます。

2. タスクの種類：ユーザーは何をしたいと考えていますか？

   • コーパスとファイルの一覧表示 -> [1. コーパスとファイルの一覧表示] に進みます。
   • コーパスの確認 -> [2. RAG Engine コーパスの取得／確認] に進みます。
   • コンテキストの検索 -> [3. コンテキストの取得] に進みます。
   • RAG Engine を使って質問に回答する -> [4. 取得したコンテキストを使ったユーザーへの回答] に進みます。

ヒント

プレースホルダーのパラメーター置換：以下の Python スクリプトでは、角括弧で囲まれた文字列のプレースホルダー（"{project_id}"、"{region}"、"{corpus_id}" など）を使用しています。スクリプトを生成、提示、または実行する前に、これらのプレースホルダーをユーザーのプロンプト（またはアクティブなコンテキスト）で指定された実際のプロジェクト ID、リージョン、コーパス 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)