はじめに

みなさんは Retrieval-Augmented Generation（RAG）についてご存知でしょうか？ RAG とは簡単に言うと「ファイル（PDFなど）を AI 専用のデータベースに保存しておいて、 AI はこの保存されたファイルを元に回答を生成できる」と言うものです。

これを聞いたあなたはこう思うかもしれません：
「でもそれって ChatGPT とかの LLM のプロンプト（ユーザーが入力する値）にファイルを投げるのと同じでは？ AI はそれを元に回答を生成してくれますよね？それがあるなら RAG なんていらないのでは？」

実を言うと直接プロンプトをファイルに入れるのには明確な欠点があります。

プロンプトにファイルを入れるデメリット

ChatGPT とかの LLM のプロンプトに PDF や Word ファイルを投げても AI はそれを元に回答を生成してくれます。しかし、プロンプトに入れたファイルは「プロンプト」扱いなので LLM に一度に入力できる情報量（トークン数）によって制限されてしまいます。そのため、数ページのファイルなら可能ですが、数十ページ、数百ページのドキュメント全体を一度に渡すことはできません。たとえ制限内に収まっても、長すぎる文書は LLM が重要なポイントを見落とす原因ともなってしまいます。

RAG の仕組み & メリット

RAG はアップロードされたファイル（今後はドキュメントと呼びます）を細かく分割（チャンク化）し、データベース（ベクトルデータベースという）に格納します。そして、AIに「RAGについて教えて」などのプロンプトを打つとそれにマッチする情報をデータベースから取得し、それを元に回答してくれます。なので数百ページの重くて長いファイルや数百を超えるファイルでもうまくAIが読み取り回答してくれます。

図1：RAG の仕組み

Google の Vertex AI Search を使って RAG アプリを作ろう！

ゴール

今回の記事では Google Cloud の Vertex AI Search を使って簡単な RAG アプリを作成する手順を解説します。ベクトルデータベースに以下のようなサンプルドキュメントを保存し、質問に対して適切に回答できることを目指します。さて、猫についての完全な出鱈目な解説ですが、果たして Vertex AI はこれを元に回答を生成してくれるのでしょうか？

cat-whiskers.txt

ネコのヒゲは「Wi-Fi」を受信している説
実は、ネコのヒゲは単なる触覚センサーではありません。

最新の非科学的な研究（※妄想）によると、ネコのヒゲは微弱な電磁波をキャッチする
**「生体アンテナ」**として機能しています。

なぜネコはルーターの近くに集まるのか？
多くの飼い主が「暖かいから」だと思い込んでいますが、実際は違います。
彼らはルーターから放出されるWi-Fi信号をヒゲで受信し、
脳内で高画質な「またたび動画」を再生しているのです。

ヒゲが抜けるとどうなる？
ネコのヒゲが抜けると、一時的に受信感度が落ちます。
稀にネコが何もない壁をじっと見つめていることがありますが、
あれは**「バッファリング中（読み込み待ち）」**の状態です。

設計

今回は Google Cloud の Vertex AI Search を使って簡単な RAG アプリを作成してみます。全体の設計は図2の通りです。今回のハンズオンでは、直接ファイルをベクトルデータベース（Vertex AI Datastore）へアップロードせず、一度 GCS に保存してからインポートする構成をとります。

ベクトルデータベースに直接保存せず、 GCS を挟む理由としては主に2つあります。

数百、数千のファイルを一括で Datastore へ入れる際、API経由の直接送信よりも GCS 経由の方が制限が緩く、安定している
元のファイルを GCS に保存しておくことで、後から Datastore へ再インポートや他の用途に使いやすい

図2：今回の RAG アプリ設計図

料金について： Google Cloud では、新規利用であれば無料クレジットが付与されるキャンペーンがあります。また、無料枠なども存在しています！使いすぎることなく、検証が終わった後にリソースを削除すれば、多額の請求が発生する心配はありません。

事前準備

1. Google Cloud Console へのログイン

Google Cloud コンソールにアクセスし、Google アカウントでログインします
初めて利用する場合は、無料トライアルに登録します（クレジットカード情報の登録が必要です）

画像1：Google Cloud Console へのログイン

注意：
セットアップにはクレジットカード情報の入力が必要です。無料クレジットや無料枠を超えて使いすぎると、実際に請求が発生する可能性があるため、検証が終わったら必ずリソースを削除するようにしましょう。

2. プロジェクトの作成

Google Cloud コンソールのトップバーにある「プロジェクトを選択」から「新しいプロジェクト」をクリックし、新しいプロジェクトを作成します（名前は任意）

画像2：プロジェクト作成

3. 請求先アカウント（Billing）の設定：

4. API の有効化

Google Cloud コンソール：API とサービスにアクセスし、以下の API を有効化します。

表1：有効化する API

対象サービス	有効化する API	画像
Vertex AI	Vertex AI API
Cloud Storage	Cloud Storage API
Vertex AI Search	Discovery Engine API

5. gcloud CLI のインストール

gcloud CLI のインストール手順に従って、gcloud CLI をインストールします。

7. uv のインストール

uv CLI のインストール手順に従って、uv をインストールします。これにより Python 仮想環境の管理とパッケージのインストールが簡単に行えます。

環境構築

1. gcloud CLI の初期設定

ターミナルを開き、以下のコマンドを実行して gcloud CLI によるプロジェクト認証と Quota Project の設定を行います。これにより、 Google Cloud 関連の API をプログラムから叩けるようになります。PROJECT_ID は先ほど作成した Google Cloud プロジェクトの ID に置き換えてください。

gcloud auth application-default login
gcloud auth application-default set-quota-project PROJECT_ID

2. uv プロジェクトの作成とパッケージのインストール

今回のアプリを作成したいディレクトリでターミナルを開いて以下のコマンドを実行し、 uv プロジェクトを作成します。そして、必要なパッケージをインストールします。

# uv プロジェクトの初期化
uv init                      
# Google Cloud Storage API 用のライブラリ         
uv add google-cloud-storage       
# Vertex AI Search API 用のライブラリ    
uv add google-cloud-discoveryengine   
# .env から環境変数を読み込む用
uv add python-dotenv
# Vertex AI と会話するための画面を作る用
uv add streamlit

3. 環境変数ファイルの作成

プロジェクトディレクトリに .env ファイルを作成し、以下の内容を記載します。

PROJECT_ID は先ほど作成した Google Cloud プロジェクトの ID を書いてください
GCS_BUCKET_NAMEは任意の名前を指定してください（ただし、その名前が既に使用されている場合は使えないので違う名前を指定してください）
GCS_DIRECTORY_PATH、DATASTORE_ID、ENGINE_ID はデフォルトのままでも大丈夫です（変えたい場合は変えても問題ありません）
LOCATION は触らなくて大丈夫です

PROJECT_ID=                         # Google Cloud プロジェクト ID を指定
GCS_BUCKET_NAME=                    # Google Cloud Storage バケット名を指定
GCS_DIRECTORY_PATH=test             # GCS 内のディレクトリパスを指定
DATASTORE_ID=my_datastore_1         # Vertex AI Search Datastore ID を指定
ENGINE_ID=my-rag-engine             # Vertex AI 用のエンジン ID を指定

# 固定
GCS_BUCKET_LOCATION=ASIA-NORTHEAST1 # GCS バケットのロケーションを指定
LOCATION=global

Vertex AI RAG アプリの構築

0. ディレクトリ構成

プロジェクトルート/
├── .env                          # 環境変数設定ファイル（実際の値を記載）
├── 1_create_gcs_bucket.py        # GCS バケット作成スクリプト
├── 2_upload_to_gcs_bucket.py     # GCS へファイルアップロードスクリプト
├── 3_create_datastore.py         # Vertex AI Search データストア作成スクリプト
├── 4_gcs_to_datastore.py         # GCS からデータストアへのインポートスクリプト
├── 5_create_vertex_ai_app.py     # Vertex AI エンジン作成スクリプト
├── 6_chat_with_vertex_ai.py      # Streamlit チャットアプリ
├── 10_cleanup.py                 # 作成したリソース削除用のスクリプト
└── data/
    └── cat-whiskers.txt          # アップロード対象のサンプルデータファイル

1. Google Cloud Storage バケットの作成

Google Cloud Storage バケットを作成するスクリプトを作成します。これにより、Google Cloud Storage にファイルを保存できるようになります。

1_create_gcs_bucket.py

"""
Google Cloud Storage バケット作成スクリプト

このスクリプトは、Vertex AI Search のデータソースとして使用する
Google Cloud Storage (GCS) バケットを作成します。

環境変数:
    PROJECT_ID: Google Cloud プロジェクトID
    GCS_BUCKET_NAME: 作成するバケットの名前
    GCS_BUCKET_LOCATION: バケットのロケーション（例: asia-northeast1）
"""
import os
from dotenv import load_dotenv
from google.cloud import storage

# .env ファイルから環境変数を読み込み
load_dotenv()
PROJECT_ID = os.getenv("PROJECT_ID")
BUCKET_NAME = os.getenv("GCS_BUCKET_NAME")
BUCKET_LOCATION = os.getenv("GCS_BUCKET_LOCATION")

# Google Cloud Storage クライアントの初期化
storage_client = storage.Client(project=PROJECT_ID)

# 指定された名前とロケーションでバケットを作成
new_bucket = storage_client.create_bucket(BUCKET_NAME, location=BUCKET_LOCATION)

print(f"バケット 'gs://{new_bucket.name}' を作成しました（場所: {BUCKET_LOCATION}）。")

2. ファイルを Google Cloud Storage バケットにアップロード

data/cat-whiskers.txt ファイルを先ほど作成した GCS バケットにアップロードします。

2_upload_to_gcs_bucket.py

"""
Google Cloud Storage へのファイルアップロードスクリプト

このスクリプトは、ローカルファイルを Google Cloud Storage (GCS) バケットに
アップロードします。アップロードされたファイルは後続のステップで
Vertex AI Search のデータストアにインポートされます。

環境変数:
    PROJECT_ID: Google Cloud プロジェクトID
    GCS_BUCKET_NAME: アップロード先のバケット名
    GCS_DIRECTORY_PATH: バケット内のアップロード先ディレクトリパス
"""
import os
from google.cloud import storage
from dotenv import load_dotenv

# .env ファイルから環境変数を読み込み
load_dotenv()

PROJECT_ID = os.getenv("PROJECT_ID")
GCS_BUCKET_NAME = os.getenv('GCS_BUCKET_NAME')
GCS_DIRECTORY_PATH = os.getenv('GCS_DIRECTORY_PATH')

# アップロード対象ファイルの設定
file_name = "cat-whiskers.txt"
upload_file_path = f"{GCS_DIRECTORY_PATH}/{file_name}"  # GCS上のパス
local_file_path = "data/" + file_name  # ローカルファイルのパス

# Cloud Storage クライアントの作成
client = storage.Client(PROJECT_ID)

# バケットへの参照を取得
bucket = client.bucket(bucket_name=GCS_BUCKET_NAME)

# ファイルをアップロード
blob = bucket.blob(upload_file_path)
blob.upload_from_filename(local_file_path)

print(f"ファイル '{local_file_path}' を 'gs://{GCS_BUCKET_NAME}/{upload_file_path}' にアップロードしました。")

画像3：ファイルがアップロードされた GCS バケット

3. Vertex AI Search Datastore の作成

Google Cloud Storage にファイルをアップロードしたら、次に Vertex AI Search Datastore を作成します。Datastore は検索対象のドキュメントを格納するためのベクトルデータベースです。

3_create_datastore.py

"""
Vertex AI Search データストア作成スクリプト

このスクリプトは、Vertex AI Search のデータストアを作成します。
データストアは検索対象のドキュメントを格納するためのコンテナです。

環境変数:
    PROJECT_ID: Google Cloud プロジェクトID
    DATASTORE_ID: 作成するデータストアのID
    LOCATION: データストアのロケーション（例: global）
"""
import os
from dotenv import load_dotenv
from google.cloud import discoveryengine_v1 as discoveryengine

# .env ファイルから環境変数を読み込み
load_dotenv()

PROJECT_ID = os.getenv("PROJECT_ID")
DATASTORE_ID = os.getenv("DATASTORE_ID")
LOCATION = os.getenv("LOCATION")

# Discovery Engine クライアントの初期化
client = discoveryengine.DataStoreServiceClient()

# データストアの親リソースパスを構築
parent = f"projects/{PROJECT_ID}/locations/{LOCATION}/collections/default_collection"

# データストアの設定を定義
datastore = discoveryengine.DataStore(
    display_name=DATASTORE_ID,
    industry_vertical=discoveryengine.IndustryVertical.GENERIC,  # 汎用的な業界垂直
    content_config=discoveryengine.DataStore.ContentConfig.CONTENT_REQUIRED,  # コンテンツ必須
    solution_types=[discoveryengine.SolutionType.SOLUTION_TYPE_SEARCH],  # 検索ソリューション
)

# データストア作成リクエストを構築
request = discoveryengine.CreateDataStoreRequest(
    parent=parent,
    data_store=datastore,
    data_store_id=DATASTORE_ID,
)

# データストア作成を実行（長時間実行オペレーション）
operation = client.create_data_store(request)

print(f"Waiting for operation to complete: {operation.operation.name}")

# オペレーションの完了を待機
response = operation.result()
print(f"Data Store Created: {response.name}")

Console 上では Datastore が作成されると以下のように表示されます

画像3：作成された Datastore

4. Google Cloud Storage から Datastore へのドキュメントインポート

Google Cloud Storage にアップロードしたファイルを Vertex AI Search Datastore にインポートするスクリプトを作成します。

4_gcs_to_datastore.py

"""
GCS から Vertex AI Search データストアへのドキュメントインポートスクリプト

このスクリプトは、Google Cloud Storage (GCS) に保存されているドキュメントを
Vertex AI Search のデータストアにインポートします。

環境変数:
    PROJECT_ID: Google Cloud プロジェクトID
    GCS_BUCKET_NAME: ドキュメントが保存されているGCSバケット名
    GCS_DIRECTORY_PATH: バケット内のドキュメントディレクトリパス
    DATASTORE_ID: インポート先のデータストアID
    LOCATION: データストアのロケーション
"""
import os
from dotenv import load_dotenv
from google.cloud import discoveryengine_v1 as discoveryengine

# .env ファイルから環境変数を読み込み
load_dotenv()

PROJECT_ID = os.getenv("PROJECT_ID")
GCS_BUCKET_NAME = os.getenv('GCS_BUCKET_NAME')
GCS_DIRECTORY_PATH = os.getenv('GCS_DIRECTORY_PATH')
DATASTORE_ID = os.getenv("DATASTORE_ID")
LOCATION = os.getenv("LOCATION")

# 1. Document Service クライアントの初期化
client = discoveryengine.DocumentServiceClient()

# 2. インポート先のブランチパスを構築
parent = client.branch_path(
    project=PROJECT_ID,
    location=LOCATION,
    data_store=DATASTORE_ID,
    branch="default_branch",  # デフォルトブランチにインポート
)

# 3. GCS ソースの設定
gcs_uri = f"gs://{GCS_BUCKET_NAME}/{GCS_DIRECTORY_PATH}/*"
gcsSource = discoveryengine.GcsSource(
    input_uris=[gcs_uri],  # ワイルドカードで複数ファイルを指定可能
    data_schema="content"  # コンテンツスキーマを指定（非構造化データ（PDF等）をインポートする場合の基本設定）
)

# 4. リクエストの構成
request = discoveryengine.ImportDocumentsRequest(
    parent=parent,
    gcs_source=gcsSource,
    # インポートモードの指定（ここでは増分インポートを使用）
    # INCREMENTAL: 既存ドキュメントを保持しつつ新規ドキュメントを追加
    # FULL: 既存ドキュメントを全て削除してから新規ドキュメントをインポート
    reconciliation_mode=discoveryengine.ImportDocumentsRequest.ReconciliationMode.INCREMENTAL,
)

# 5. インポートの実行（非同期処理の開始）
print(f"Importing documents from {gcs_uri}...")
operation = client.import_documents(request=request)

# 6. 完了まで待機
print(f"Waiting for operation to complete: {operation.operation.name}")
response = operation.result()

print(f"Import completed successfully.")

※ 実行完了には5分程度かかります。Google Cloud Console の Vertex AI Search Datastore のデータストアのページからステータスを確認できます。

実行中

画像4：実行中の importDocuments

終了時

画像5：実行終了時の importDocuments

5. Vertex AI エンジンの作成

Vertex AI が Datastore からファイルを読み取って検索と回答生成を行うにはエンジンというものが必要となります。以下のスクリプトでエンジンを作成しましょう！

5_create_vertex_ai_app.py

"""
Vertex AI Search エンジン（検索アプリ）作成スクリプト

このスクリプトは、Vertex AI Search のエンジン（検索アプリケーション）を作成します。
エンジンはデータストアを検索し、LLMベースのサマリー生成機能を提供します。

環境変数:
    PROJECT_ID: Google Cloud プロジェクトID
    LOCATION: エンジンのロケーション（例: global）
    DATASTORE_ID: 接続するデータストアのID
    ENGINE_ID: 作成するエンジンのID
"""
import os
from dotenv import load_dotenv
from google.cloud import discoveryengine_v1 as discoveryengine

# .env ファイルから環境変数を読み込み
load_dotenv()
PROJECT_ID = os.getenv("PROJECT_ID")
LOCATION = os.getenv("LOCATION")
DATASTORE_ID = os.getenv("DATASTORE_ID")
ENGINE_ID = os.getenv("ENGINE_ID")

# Engine Service クライアントの初期化（クォータプロジェクトを指定）
client_options = {"quota_project_id": PROJECT_ID}
client = discoveryengine.EngineServiceClient(client_options=client_options)

# エンジンの親リソースパスを構築
parent = f"projects/{PROJECT_ID}/locations/{LOCATION}/collections/default_collection"

# エンジンの設定を定義
engine = discoveryengine.Engine(
    display_name="My RAG App",  # アプリの表示名
    industry_vertical=discoveryengine.IndustryVertical.GENERIC,  # 汎用的な業界垂直
    solution_type=discoveryengine.SolutionType.SOLUTION_TYPE_SEARCH,  # 検索ソリューション
    search_engine_config=discoveryengine.Engine.SearchEngineConfig(
        search_tier=discoveryengine.SearchTier.SEARCH_TIER_ENTERPRISE,  # エンタープライズティア
        search_add_ons=[discoveryengine.SearchAddOn.SEARCH_ADD_ON_LLM]  # LLMベースのサマリー生成を有効化
    ),
    data_store_ids=[DATASTORE_ID]  # 関連付けるデータストアのID
)

# エンジン作成リクエストを構築
request = discoveryengine.CreateEngineRequest(
    parent=parent,
    engine=engine,
    engine_id=ENGINE_ID,
)

# エンジン作成を実行
print(f"Creating Engine '{ENGINE_ID}'...")
operation = client.create_engine(request=request)

# エンジン作成の完了を待機
response = operation.result()
print(f"Engine Created Successfully: {response.name}")

6. Streamlit を使ったチャットアプリの作成

Vertex AI Datastore へのファイルのアップロード（importDocuments）が終了した後は、Vertex AI がそのドキュメントを読み取ることができます。Streamlit を使って簡単なチャットアプリを作成しましょう！これにより、ユーザーはブラウザ上で質問を入力し、Vertex AI からの回答を受け取ることができます。

Vertex AI がアップロードされたドキュメントを読めるようになるまで数分ラグが生じることもあります。その場合、一旦数分置いてからリトライしてみてください。

6_chat_with_vertex_ai.py

"""
Vertex AI Search チャットインターフェース（Streamlit アプリ）

このアプリケーションは、Vertex AI Search を使用したチャットインターフェースを提供します。
ユーザーの質問に対して、データストアを検索し、LLMベースのサマリーを生成して回答します。

環境変数:
    PROJECT_ID: Google Cloud プロジェクトID
    LOCATION: データストアのロケーション
    DATASTORE_ID: 検索対象のデータストアID
"""
import os
import streamlit as st
from dotenv import load_dotenv
from google.cloud import discoveryengine_v1 as discoveryengine

# .env ファイルから環境変数を読み込み
load_dotenv()

PROJECT_ID = os.getenv("PROJECT_ID")
LOCATION = os.getenv("LOCATION")
DATASTORE_ID = os.getenv("DATASTORE_ID")

# UI設定
st.set_page_config(page_title="Vertex AI Search Chat", page_icon="💬")
st.title("Vertex AI Search Chat")
st.caption(f"Connected to DataStore: {DATASTORE_ID}")

# クライアント初期化（認証プロジェクトを指定）
client_options = {"quota_project_id": PROJECT_ID}
search_client = discoveryengine.SearchServiceClient(client_options=client_options)

# セッション状態の初期化（チャット履歴の保存用）
if "messages" not in st.session_state:
    st.session_state.messages = []

# チャット履歴の表示
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])


# 回答生成関数
def generate_answer(query_text):
    """データストアから検索してAI生成サマリーを返す"""
    serving_config = f"projects/{PROJECT_ID}/locations/{LOCATION}/collections/default_collection/dataStores/{DATASTORE_ID}/servingConfigs/default_config"

    request = discoveryengine.SearchRequest(
        serving_config=serving_config,
        query=query_text,
        page_size=10,
        content_search_spec=discoveryengine.SearchRequest.ContentSearchSpec(
            # スニペットを返す
            snippet_spec=discoveryengine.SearchRequest.ContentSearchSpec.SnippetSpec(
                return_snippet=True
            ),
            # AI生成サマリー設定
            summary_spec=discoveryengine.SearchRequest.ContentSearchSpec.SummarySpec(
                summary_result_count=5,
                include_citations=True,
            ),
        ),
    )

    response = search_client.search(request=request)
    return response


# ユーザー入力と処理
if prompt := st.chat_input("質問を入力してください..."):
    # ユーザーのメッセージを表示
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    # アシスタントの回答生成
    with st.chat_message("assistant"):
        with st.spinner("データストアを検索中..."):
            try:
                response = generate_answer(prompt)

                # AI生成サマリーがある場合（推奨）
                if hasattr(response, 'summary') and response.summary and hasattr(response.summary, 'summary_text') and response.summary.summary_text:
                    full_answer = response.summary.summary_text
                    st.markdown(full_answer)

                    # 引用元の表示
                    if hasattr(response.summary, 'summary_with_metadata') and response.summary.summary_with_metadata:
                        citations = response.summary.summary_with_metadata.references
                        if citations:
                            with st.expander("参照元を確認"):
                                for i, citation in enumerate(citations, 1):
                                    if hasattr(citation, 'document'):
                                        st.write(f"**[{i}]** {citation.document}")
                                    elif hasattr(citation, 'uri'):
                                        st.write(f"**[{i}]** {citation.uri}")
                else:
                    # サマリーがない場合は検索結果を直接表示
                    results = list(response.results)
                    if results:
                        full_answer = f"「{prompt}」に関する検索結果:\n\n"

                        for i, result in enumerate(results[:5], 1):
                            doc = result.document
                            if hasattr(doc, 'derived_struct_data'):
                                data = doc.derived_struct_data
                                title = data.get('title', data.get('link', 'ドキュメント'))

                                full_answer += f"**{i}. {title}**\n"

                                # スニペットを表示
                                snippets = data.get('snippets', [])
                                if snippets and len(snippets) > 0:
                                    snippet_text = snippets[0].get('snippet', '')
                                    if snippet_text:
                                        full_answer += f"{snippet_text}\n"

                                # リンクがあれば表示
                                link = data.get('link', '')
                                if link:
                                    full_answer += f"ソース: {link}\n"

                                full_answer += "\n"

                        st.markdown(full_answer)

                        # 詳細な検索結果
                        with st.expander("すべての検索結果を表示"):
                            for i, result in enumerate(results, 1):
                                doc = result.document
                                if hasattr(doc, 'derived_struct_data'):
                                    data = doc.derived_struct_data
                                    st.write(f"**[{i}] {data.get('title', 'タイトルなし')}**")
                                    if 'link' in data:
                                        st.write(f"リンク: {data['link']}")
                                    if 'snippets' in data and data['snippets']:
                                        for snippet in data['snippets']:
                                            st.write(f"> {snippet.get('snippet', '')}")
                                    st.write("---")
                    else:
                        full_answer = "申し訳ございません。データストアに関連する情報が見つかりませんでした。"
                        st.warning(full_answer)

                st.session_state.messages.append({"role": "assistant", "content": full_answer})

            except Exception as e:
                error_message = f"エラーが発生しました: {e}"
                st.error(error_message)
                import traceback
                st.error(traceback.format_exc())

実際にプロンプトを実行した例

画像6：AI とのチャット画面

GIF

画像7：AI とのチャット画面（GIF）

クリーニング

検証が終わったら、最後にリソースを削除しましょう。10_cleanup.py を実行することで、作成した Engine、Datastore、GCS バケットをすべて自動で削除し、余計な課金を防ぐことができます。

実行後、正しく削除されたか Google Cloud Console 上で確認してください

10_cleanup.py

"""
Vertex AI Search リソースクリーンアップスクリプト

このスクリプトは、Vertex AI Search で作成したリソースを削除します。
削除対象:
    1. Engine (検索アプリ)
    2. DataStore (データストア)
    3. GCS Bucket (Cloud Storage バケット)

環境変数:
    PROJECT_ID: Google Cloud プロジェクトID
    ENGINE_ID: 削除するエンジンのID
    DATASTORE_ID: 削除するデータストアのID
    GCS_BUCKET_NAME: 削除するGCSバケット名

注意: このスクリプトは全てのリソースを削除するため、実行前に確認を求めます。
"""
import os
import time
from dotenv import load_dotenv
from google.cloud import discoveryengine_v1 as discoveryengine
from google.cloud import storage

# .env ファイルから環境変数を読み込み
load_dotenv()

PROJECT_ID = os.getenv("PROJECT_ID")
LOCATION = os.getenv("LOCATION")
ENGINE_ID = os.getenv("ENGINE_ID")
DATASTORE_ID = os.getenv("DATASTORE_ID")
BUCKET_NAME = os.getenv("GCS_BUCKET_NAME")


def delete_engine():
    """
    Vertex AI Search Engine (検索アプリケーション) の削除

    Engine を削除します。既に削除されている場合はエラーメッセージを表示します。
    """
    print(f"--- 1. Engine '{ENGINE_ID}' を削除中 ---")

    # Engine Service クライアントの初期化
    client = discoveryengine.EngineServiceClient()

    # 削除対象の Engine のリソースパスを構築
    name = f"projects/{PROJECT_ID}/locations/{LOCATION}/collections/default_collection/engines/{ENGINE_ID}"

    try:
        # Engine 削除リクエストを送信（非同期処理）
        operation = client.delete_engine(name=name)
        print("削除リクエストを送信しました。完了を待機中...")

        # 削除完了まで待機
        operation.result()
        print("Engine を削除しました。")
    except Exception as e:
        print(f"Engine の削除に失敗しました（既に削除されている可能性があります）: {e}")


def delete_data_store():
    """
    Vertex AI Search DataStore の削除

    DataStore を削除します。Engine を先に削除してから実行する必要があります。
    """
    print(f"--- 2. DataStore '{DATASTORE_ID}' を削除中 ---")

    # DataStore Service クライアントの初期化
    client = discoveryengine.DataStoreServiceClient()

    # 削除対象の DataStore のリソースパスを構築
    name = f"projects/{PROJECT_ID}/locations/{LOCATION}/collections/default_collection/dataStores/{DATASTORE_ID}"

    try:
        # DataStore 削除リクエストを送信（非同期処理）
        operation = client.delete_data_store(name=name)
        print("削除リクエストを送信しました。完了を待機中...")

        # 削除完了まで待機
        operation.result()
        print("DataStore を削除しました。")
    except Exception as e:
        print(f"DataStore の削除に失敗しました: {e}")


def delete_gcs_bucket():
    """
    Google Cloud Storage バケットの削除

    GCS バケットとその中の全てのファイルを削除します。
    force=True により、バケット内にファイルが残っていても削除されます。
    """
    print(f"--- 3. GCSバケット '{BUCKET_NAME}' を削除中 ---")

    # Cloud Storage クライアントの初期化
    storage_client = storage.Client(project=PROJECT_ID)

    try:
        # バケットを取得
        bucket = storage_client.get_bucket(BUCKET_NAME)

        # バケット内が空でなくても強制削除
        # force=True: バケット内の全ファイルを含めて削除
        bucket.delete(force=True)
        print(f"バケット 'gs://{BUCKET_NAME}' を削除しました。")
    except Exception as e:
        print(f"バケットの削除に失敗しました: {e}")


if __name__ == "__main__":
    # ユーザーに削除の確認を求める
    confirm = input("全てのリソースを削除しますか？ (y/n): ")

    if confirm.lower() == 'y':
        # 1. Engine の削除（最初に削除する必要がある）
        delete_engine()

        # クラウド側の反映待ち（Engine 削除が完全に反映されるまで待機）
        time.sleep(5)

        # 2. DataStore の削除
        delete_data_store()

        # 3. GCS Bucket の削除
        delete_gcs_bucket()

        print("\n全てのクリーンアップ作業が完了しました。")
    else:
        print("キャンセルしました。")

まとめ

今回は、Google Cloud の Vertex AI Search を活用して RAG アプリケーションを構築する方法を解説しました。Google Cloud Storage を利用したドキュメントの保存から、Datastore へのアップロード、そして Streamlit を使ったチャットインターフェースの作成まで、一連の流れをカバーしました。

このように Vertex AI Search を使用すると簡単に RAG アプリケーションを作ることができちゃいます！ぜひ自分だけのドキュメントを読み込ませて試してみてください！

ちなみに Github リポジトリも公開していますので、コードを丸ごと確認したい方はこちらのリンクからどうぞ！

免責事項

本記事に掲載されている情報は執筆時点（2025年12月）のものです。Google Cloudの仕様変更や料金改定により、内容が異なる場合がありますのでご注意ください
料金や無料枠の詳細については、必ずGoogle Cloud 公式ドキュメントをご確認ください
検証が終わりましたら、不要な課金を防ぐため、必ずリソースの削除（10_cleanup.py の実行）を行ってください