developersIO
Google DriveのKB記事をVertex AI RAGコーパスへ自動同期する仕組みを作った話 — イベント駆動を断念してスマートポーリングに至るまで

Google DriveのKB記事をVertex AI RAGコーパスへ自動同期する仕組みを作った話 — イベント駆動を断念してスマートポーリングに至るまで

Google DriveのKB記事をVertex AI RAGコーパスへ自動同期する仕組みを構築。Workspace Events APIのOAuth制約を乗り越え、GCSメタデータを活用したスマートポーリングに至るまでの試行錯誤を共有します。
Google Cloud (GCP)Vertex AIRAGGoogle Drive
2026.07.15

はじめに

社内向けのGoogle ChatボットにRAG（Retrieval-Augmented Generation）を組み込み、KB（ナレッジベース）記事を基に回答を生成する仕組みを運用しています。Vertex AI RAG Engineにドキュメントを取り込み、ボットがユーザーの質問に対してコーパスを検索して回答するという構成です。

当初、KB記事はGitリポジトリ内のMarkdownファイルとして管理し、CLIスクリプトでGCS経由でRAGコーパスにアップロードしていました。しかし、記事の追加・編集にはgit操作とCLI実行が必要で、非技術者の担当者にとってハードルが高いという課題がありました。

「Google Driveで記事を編集したら、自動的にRAGコーパスに反映される」— これを実現するために調査・実装した記録です。最初はイベント駆動で設計しましたが、OAuth制約という壁にぶつかり、最終的にスマートポーリングという現実的な解に落ち着きました。

全体アーキテクチャ

最終的に採用した構成は以下の通りです。

google-drive-vertex-ai-rag-auto-sync-smart-polling

ポイントは、Google DocsをKB記事のフォーマットとして採用したことです。DriveフォルダにMarkdownファイルを置くのではなく、Google Docsを使うことで、担当者はWYSIWYGエディタで記事を編集できます。Markdown記法を知らなくても、見出し・リスト・太字などの書式を直感的に操作できます。

Vertex AI RAG EngineはGoogle Docsを含むGoogle Workspaceファイルをネイティブサポートしており、DriveフォルダURLを rag.import_files() に渡すだけでインポートできます。ただし、import_files() にフォルダURLを渡すと毎回全ファイルを再処理するため、15分間隔のポーリングでは非効率です。そこで、GCSを中間層に挟んで変更検知・削除検知を行うスマートポーリングを実装しました。

当初の設計: Workspace Events APIによるイベント駆動

最初に設計したのは、リアルタイム同期のアーキテクチャでした。

google-drive-vertex-ai-rag-auto-sync-event-driven

Google Workspace Events API を使うと、Driveのファイル作成・更新・削除イベントをPub/Subに配信できます。サポートされるイベントタイプは以下の通りです。

  • google.workspace.drive.file.v3.created — ファイル作成
  • google.workspace.drive.file.v3.updated — ファイル更新
  • google.workspace.drive.file.v3.trashed — ファイル削除（ゴミ箱へ移動）

この設計であれば、編集が発生した瞬間にRAGコーパスが更新され、ボットは常に最新の情報で回答できます。

サブスクリプションの作成

Workspace Events APIのサブスクリプションは以下のように作成します。

scripts/create_subscription.py
EVENT_TYPES = [
    "google.workspace.drive.file.v3.created",
    "google.workspace.drive.file.v3.updated",
    "google.workspace.drive.file.v3.trashed",
]

credentials, _ = google.auth.default(
    scopes=["https://www.googleapis.com/auth/drive"]
)
service = build("workspaceevents", "v1", credentials=credentials)

body = {
    "targetResource": f"//drive.googleapis.com/files/{folder_id}",
    "eventTypes": EVENT_TYPES,
    "notificationEndpoint": {
        "pubsubTopic": "projects/YOUR_PROJECT_ID/topics/drive-kb-events",
    },
    "payloadOptions": {
        "includeResource": True,
        "fieldMask": "mimeType,name,parents",
    },
}
result = service.subscriptions().create(body=body).execute()

重要な注意点: Workspace Events APIのDriveイベントに必要なOAuthスコープは https://www.googleapis.com/auth/drive です。apps.events.subscriptions というスコープは存在しません。公式ドキュメントでも、ターゲットリソースに応じた既存のスコープ（Driveならdriveスコープ）を使うよう記載されています。

サブスクリプションの有効期限

Workspace Events APIのサブスクリプションには最大7日間のTTLがあり、定期的な更新が必要です。Cloud Schedulerで6日ごとに更新するか、サブスクリプション自体が配信するexpirationReminderイベントをハンドリングすることで対応できます。

OAuth・組織セキュリティの壁

イベント駆動の設計を進める中で、いくつかの壁にぶつかりました。

壁1: サービスアカウントではサブスクリプションを作成できない

Workspace Events APIのサブスクリプション作成は、ユーザーのOAuth認証が必須です。サービスアカウントでは作成できません。また、ドメイン全体の委任（Domain-Wide Delegation / DWD）も Workspace Events API ではサポートされていません。

つまり、初回のサブスクリプション作成には必ず人間がOAuthフローを通る必要があります。

壁2: gcloud CLIのOAuthクライアントがブロックされる

gcloud auth application-default login を使ってOAuth認証を試みたところ、Google Workspaceの管理コンソールで「このアプリはブロックされます」というエラーが発生しました。

これはGoogle Workspace管理者が設定するアプリアクセス制御（Admin Console → セキュリティ → APIの制御 → アプリのアクセス制御）によるもので、2つの独立した設定があります：

設定 内容
OAuthスコープの許可リスト 特定のスコープ（例: drive）を許可するか
アプリの信頼レベル 個々のOAuthクライアントを「信頼済み」「制限あり」「ブロック」に分類

スコープが許可されていても、アプリ自体がブロックされていると認証できません。gcloud CLIのOAuthクライアントIDは 764086051850-6qr4p6gpi6hn506pt8ejuq83di341hur.apps.googleusercontent.com で、これを管理者が「信頼済み」に設定する必要があります。

壁3: 組織のポリシー変更が必要

結局、Workspace Events APIを使うには以下の条件をすべて満たす必要がありました：

  1. drive スコープが組織で許可されていること
  2. gcloud CLIのOAuthクライアントが信頼済みに設定されていること
  3. 管理者がApp access controlでアプリを承認すること

組織のセキュリティポリシーによっては、これらの変更を申請・承認するプロセスが必要になります。

判断: ポーリングへの切り替え

イベント駆動アプローチのために組織のセキュリティ設定を変更してもらう調整コストと、ポーリングで十分な更新頻度が得られることを天秤にかけ、スマートポーリングを採用しました。

観点 イベント駆動 スマートポーリング
反映速度 数秒〜数十秒 最大15分
組織設定変更 必要（OAuth許可、アプリ信頼） 不要
インフラ Pub/Sub + Workspace Events subscription Cloud Scheduler のみ
保守性 サブスクリプション更新（7日TTL） 不要
信頼性 サブスクリプション失効リスク Schedulerは安定稼働

KB記事の更新頻度を考えると、15分間隔の同期で実用上の問題はありません。イベント駆動は将来的にOAuth設定が整えば追加できるよう、Pub/Subトピックとイベントハンドラのコードは残してあります。

GCS経由のスマート差分同期の実装

「ポーリング」と言っても、毎回全ファイルを再インポートするのは非効率です。変更があったファイルだけを同期するスマートポーリングを実装しました。

差分検知: GCSメタデータにmodifiedTimeを保存

DriveからエクスポートしたファイルをGCSにアップロードする際、GCSのblob metadataにDriveのmodifiedTimeを記録します。

sync/rag_sync.py
def _export_doc_to_gcs(file_id: str, file_name: str, modified_time: str) -> str:
    drive = _get_drive_service()
    content = drive.files().export(
        fileId=file_id,
        mimeType="text/plain",
    ).execute()

    client = _get_gcs_client()
    bucket = client.bucket(GCS_BUCKET)
    blob = bucket.blob(_blob_name(file_name))
    blob.metadata = {"drive_modified_time": modified_time, "drive_file_id": file_id}
    blob.upload_from_string(content, content_type="text/plain")

    return f"gs://{GCS_BUCKET}/{blob.name}"

同期ロジック: 変更・削除の両方を検知

full_sync() は以下の流れで動作します：

  1. Drive APIで対象フォルダ内のGoogle Docs一覧を取得
  2. GCS上の既存blobのメタデータ（drive_modified_time）を取得
  3. 両者を比較し、変更があったファイルだけを再エクスポート
  4. GCSにあるがDriveにないファイル（＝削除されたファイル）を検出し、GCSとRAGコーパスから削除
  5. 変更があった場合のみ rag.import_files() を呼び出し
sync/rag_sync.py
def full_sync() -> dict:
    drive_docs = _list_drive_docs()
    gcs_blobs = _get_gcs_blobs()

    # GCSメタデータからmodifiedTimeを取得
    gcs_modified = {}
    for blob in gcs_blobs.values():
        blob.reload()
        meta = blob.metadata or {}
        gcs_modified[blob.name] = meta.get("drive_modified_time", "")

    exported = 0
    unchanged = 0
    drive_blob_names = set()

    for doc in drive_docs:
        bn = _blob_name(doc["name"])
        drive_blob_names.add(bn)
        gcs_mtime = gcs_modified.get(bn, "")

        if gcs_mtime == doc["modifiedTime"]:
            unchanged += 1
            continue

        _export_doc_to_gcs(doc["id"], doc["name"], doc["modifiedTime"])
        exported += 1

    # 削除検知: GCSにあるがDriveにないファイル
    deleted = 0
    for bn, blob in gcs_blobs.items():
        if bn not in drive_blob_names:
            file_name = bn.removeprefix(GCS_PREFIX).removesuffix(".txt")
            blob.delete()
            _delete_rag_file(file_name)
            deleted += 1

    # 変更があった場合のみインポート
    imported = 0
    if exported > 0:
        response = rag.import_files(
            corpus_name=_corpus_name(),
            paths=[f"gs://{GCS_BUCKET}/{GCS_PREFIX}"],
        )
        imported = response.imported_rag_files_count

    return {"exported": exported, "unchanged": unchanged, "deleted": deleted, "imported": imported}

rag.import_files() 自体も内部でファイルの変更を検知してスキップする仕組みがありますが、GCS側の差分チェックにより不要なエクスポート処理を省略できるため、同期全体が高速になります。

なぜGCSを中間層に挟むのか

rag.import_files() にDriveフォルダURLを直接渡せばGoogle Docsをインポートできますが、この方法では毎回フォルダ内の全ファイルを再処理します。15分間隔のポーリングで170件以上のファイルを毎回再インポートするのは非効率です。

GCSを中間層にすることで、アプリケーション側で差分検知を行い、変更があったファイルだけを処理できます：

  • 差分検知: GCSのblob metadataにDriveのmodifiedTimeを保存し、変更の有無を正確に判定できる。変更がなければエクスポートもインポートもスキップ
  • 削除検知: GCSのblob一覧とDriveのファイル一覧をdiffすることで、削除されたファイルを特定できる。rag.import_files() は削除を検知しないため、この仕組みが必須
  • デバッグ容易性: GCS上のファイルを直接確認でき、同期状態のトラブルシューティングが容易

Cloud Functionのエントリポイント

Cloud Function（2nd gen / Cloud Run-based）は、Cloud Schedulerからの定期呼び出しとPub/Subイベント（将来のイベント駆動用）の両方に対応します。

sync/main.py
@functions_framework.http
def handle_sync(request):
    body = request.get_json(silent=True) or {}

    try:
        # Pub/Sub push message（将来のイベント駆動用）
        if "message" in body:
            return _handle_pubsub_event(body["message"])

        action = body.get("action", "")
        if action == "full_sync":
            summary = full_sync()
            return (summary, 200)

        return ("Unknown request", 400)
    except Exception:
        log("ERROR", "sync_error", error=traceback.format_exc())
        # 200を返してPub/Subの無限リトライを防止
        return ("error logged", 200)

デプロイとスケジューラの設定

# Cloud Function（sync）のデプロイ
gcloud functions deploy kb-drive-sync \
  --gen2 --runtime=python314 --region=asia-northeast1 \
  --source=sync/ --entry-point=handle_sync \
  --trigger-http --no-allow-unauthenticated \
  --memory=512Mi --cpu=0.5 \
  --set-env-vars="GCP_PROJECT_ID=YOUR_PROJECT_ID,GCP_LOCATION=asia-northeast1,RAG_CORPUS_ID=YOUR_CORPUS_ID,DRIVE_FOLDER_ID=YOUR_FOLDER_ID,GCS_BUCKET=YOUR_BUCKET"

# Cloud Schedulerで15分間隔の定期実行
gcloud scheduler jobs create http kb-drive-sync-scheduler \
  --location=asia-northeast1 \
  --schedule="*/15 * * * *" \
  --uri="https://kb-drive-sync-HASH-an.a.run.app" \
  --http-method=POST \
  --headers="Content-Type=application/json" \
  --message-body='{"action":"full_sync"}' \
  --oidc-service-account-email="YOUR_SA@developer.gserviceaccount.com"

KB記事のマイグレーション

数百件のMarkdownファイルをGoogle Docsに移行する必要がありました。

課題: ローカルからDrive APIを叩けない

前述のOAuth制約により、ローカル環境からDrive APIを使ってファイルをアップロードすることができません。そこで、Cloud Function経由でマイグレーションを行う方法を取りました。

  1. ローカルからMarkdownファイルをGCSの一時プレフィックス（kb_migrate/）にアップロード
  2. Cloud Functionのmigrateアクションを呼び出し
  3. Cloud Function内でGCSからファイルを読み取り、Drive APIでGoogle Docsとして作成
sync/rag_sync.py
def migrate_from_gcs() -> dict:
    client = _get_gcs_client()
    bucket = client.bucket(GCS_BUCKET)
    drive = _get_drive_service()

    blobs = list(bucket.list_blobs(prefix=GCS_MIGRATE_PREFIX))
    md_blobs = [b for b in blobs if b.name.endswith(".md")]

    # 重複スキップ（再実行時の安全性）
    existing_docs = {doc["name"] for doc in _list_drive_docs()}

    for blob in md_blobs:
        file_name = blob.name.removeprefix(GCS_MIGRATE_PREFIX).removesuffix(".md")
        if file_name in existing_docs:
            continue

        content = blob.download_as_bytes()
        media = MediaInMemoryUpload(content, mimetype="text/markdown")
        metadata = {
            "name": file_name,
            "parents": [DRIVE_FOLDER_ID],
            "mimeType": "application/vnd.google-apps.document",
        }
        drive.files().create(
            body=metadata,
            media_body=media,
            fields="id",
            supportsAllDrives=True,
        ).execute()

Drive APIの files().create() で mimeType に application/vnd.google-apps.document を指定すると、アップロードしたファイルが自動的にGoogle Docsに変換されます。text/markdown として送信したMarkdownファイルは、見出し・リスト・コードブロックなどが適切にGoogle Docsのフォーマットに変換されます。

タイムアウトへの対応

Cloud Functions（2nd gen）のデフォルトタイムアウトは540秒（9分）です。数百件のファイルをすべてアップロードするには時間が足りず、初回実行では約半数でタイムアウトしました。

しかし、existing_docs のチェックにより重複スキップが効くため、再実行するだけで残りのファイルが処理されます。冪等性のある設計にしておくことで、タイムアウトは致命的な問題にはなりません。

まとめ

学んだこと：

  • Workspace Events APIはOAuth必須: サービスアカウントやDWDでは使えない。組織のセキュリティポリシーとの調整が必要
  • OAuthスコープの罠: Workspace Events API用の独自スコープ（apps.events.subscriptionsのような）は存在しない。ターゲットリソースに応じた既存のスコープ（driveなど）を使う
  • GCS中間層の役割: RAG EngineはGoogle Docsを直接インポートできるが、毎回全ファイルを再処理する。GCSのblobメタデータで差分検知・削除検知を行うことで、ポーリングの効率を大幅に改善できる
  • 冪等性は大事: マイグレーションスクリプトを再実行可能に設計しておくことで、タイムアウトや一時的な障害に対処できる

今後の展望：

イベント駆動のコード（Pub/Subハンドラ）は残してあるので、将来的にOAuth設定が整えば、スマートポーリングとイベント駆動を併用できます。ポーリングをベースラインの信頼性として維持しつつ、イベント駆動でリアルタイム性を追加する — この「ポーリング + イベント駆動」のハイブリッド構成が、最も堅牢な選択だと考えています。

この記事をシェアする

関連記事

Vertex AI RAG Engineのナレッジベース更新で踏んだ落とし穴 ― GCSアップロードだけでは同期されない
lin-yuchen
2026.07.10
Google Chat Bot に Vertex AI RAG Engine を繋いでナレッジベース検索を実装した話
lin-yuchen
2026.06.19
Storage Transfer Service のイベントドリブン転送を試してみた
hanzawa.yuya
2025.12.21
Vertex AI SDKからgoogle-genaiへ移行した — Pydanticネイティブな構造化出力が便利
lin-yuchen
2026.07.10