Antigravity で構築するサーバーレスAIパイプライン — Cloud Run × Pub/Sub × BigQuery 本番実装ガイド

AI 処理を本番環境で動かしていると、どこかで必ず壁にぶつかります。「リクエストが急増したとき、API コールが連鎖してタイムアウトした」「処理が途中で失敗しても、リトライの仕組みがなくてデータが消えた」「ユーザーの行動ログを溜めたいのに、どこに保存すればいいか分からない」——そういった課題が積み重なって、開発者は同期処理の限界を実感します。

Antigravity でコードを書いていると、非同期パイプラインの実装もかなりスムーズになりました。以前は Cloud Run + Pub/Sub の連携を一から設計するのが面倒で後回しにしていたのですが、Antigravity に「イベント駆動の AI 処理パイプラインを作りたい」と伝えるだけで、Pub/Sub のトピック定義から Cloud Run のサービスコード、BigQuery のスキーマ設計まで、一貫した構成を提案してくれます。

ここでは私が実際のプロジェクトで試行錯誤しながら確立したパイプライン構成を、動作するコードとともに紹介します。「とりあえず動くもの」ではなく、デッドレターキュー・べき等性・監視まで揃えた、本番に持っていける設計です。

なぜサーバーレスAIパイプラインが必要なのか

同期 API だけで AI 処理を組んでいると、いくつかの問題が顕在化します。

まず レイテンシの問題です。Gemini API などの大規模モデルへの呼び出しは、数百ミリ秒から数秒かかることがあります。これをユーザーリクエストの処理フロー内に直接置くと、レスポンスが遅くなるだけでなく、上流のサービスとの接続がタイムアウトするリスクも生まれます。

次に 信頼性の問題です。同期処理で AI API がエラーを返した場合、そのリクエストは失敗として扱われます。リトライをアプリ側で実装するにしても、べき等性の管理が難しく、二重処理が発生しやすくなります。

そして スケーラビリティの問題です。トラフィックが急増したとき、処理キューを持たない同期アーキテクチャは即座に詰まります。

これらをまとめて解決するのが、Pub/Sub を中心としたイベント駆動パイプラインです。

ユーザーリクエスト
      ↓
  API (Cloud Run) ── publish ──→  Pub/Sub Topic
                                       ↓
                               処理ワーカー (Cloud Run)
                                       ↓
                           Gemini API / AI 処理
                                       ↓
                              BigQuery (結果格納)

この構成のメリットは「受付」と「処理」が完全に分離されることです。受付側の API は Pub/Sub にメッセージを投げたら即座にレスポンスを返せます。処理側のワーカーは自分のペースで処理を進め、失敗してもPub/Sub が自動でリトライしてくれます。

アーキテクチャ全体像と各コンポーネントの役割

実際のプロジェクトで使っている構成を示します。

受付サービス（Ingestion Service）: Cloud Run で動く HTTP エンドポイント。ユーザーからのリクエストをバリデーションして Pub/Sub に投げ、202 Accepted を返します。

AI 処理ワーカー（Worker Service）: Pub/Sub の push サブスクリプションを受け取る Cloud Run サービス。Gemini API 呼び出し、ベクトル検索、分類処理など、重い処理を担います。

BigQuery（結果ストレージ）: AI 処理の結果を構造化データとして格納します。ユーザーの利用ログ、AI の出力、レイテンシなどを記録し、後から Looker Studio で可視化します。

デッドレターキュー（DLQ）: 一定回数リトライしても失敗し続けたメッセージを別トピックに転送します。手動調査・再投入ができる安全網です。

Antigravity でこのアーキテクチャ図を伝えると、各サービスの骨格コードを一気に生成してくれます。私が好むのは最初に アーキテクチャの意図を言語化してから実装させることです。「なぜ Pub/Sub を使うか」「DLQ が必要な理由」まで伝えると、コードのコメントや変数名にも意図が反映されて、後からメンテしやすくなります。

受付サービスの実装 — Pub/Sub へのメッセージ投入

まずは受付 API から作ります。Cloud Run 上で動く Node.js/TypeScript サービスです。

// ingestion-service/src/index.ts
import { PubSub } from "@google-cloud/pubsub";
import express from "express";
import { z } from "zod";
import crypto from "crypto";
 
const app = express();
app.use(express.json());
 
const pubsub = new PubSub({ projectId: process.env.GCP_PROJECT_ID });
const TOPIC_NAME = process.env.PUBSUB_TOPIC_NAME ?? "ai-processing-jobs";
 
// リクエストスキーマ（Zod でバリデーション）
const ProcessRequestSchema = z.object({
  userId: z.string().min(1),
  inputText: z.string().min(1).max(10000),
  jobType: z.enum(["summarize", "classify", "extract"]),
  callbackUrl: z.string().url().optional(),
});
 
app.post("/api/process", async (req, res) => {
  // バリデーション
  const result = ProcessRequestSchema.safeParse(req.body);
  if (!result.success) {
    return res.status(400).json({ error: result.error.issues });
  }
 
  const { userId, inputText, jobType, callbackUrl } = result.data;
 
  // べき等性キーの生成（同一リクエストの二重処理を防ぐ）
  const idempotencyKey = crypto
    .createHash("sha256")
    .update(`${userId}-${jobType}-${inputText.slice(0, 100)}-${Date.now()}`)
    .digest("hex")
    .slice(0, 32);
 
  const jobId = `job-${idempotencyKey}`;
 
  const message = {
    jobId,
    userId,
    inputText,
    jobType,
    callbackUrl,
    createdAt: new Date().toISOString(),
  };
 
  try {
    const topic = pubsub.topic(TOPIC_NAME);
    const messageId = await topic.publishMessage({
      data: Buffer.from(JSON.stringify(message)),
      attributes: {
        jobType,
        userId,
        idempotencyKey,
      },
    });
 
    console.log(`Published message ${messageId} for job ${jobId}`);
 
    // 202 Accepted で即座に返す（処理完了を待たない）
    return res.status(202).json({
      jobId,
      messageId,
      status: "accepted",
      message: "処理キューに追加しました。結果はコールバックURLに通知されます。",
    });
  } catch (error) {
    console.error("Pub/Sub publish failed:", error);
    return res.status(503).json({
      error: "キューへの投入に失敗しました。しばらく待ってから再試行してください。",
    });
  }
});
 
app.get("/health", (_req, res) => res.json({ status: "ok" }));
 
const PORT = parseInt(process.env.PORT ?? "8080");
app.listen(PORT, () => console.log(`Ingestion service listening on :${PORT}`));

ポイント解説: 202 Accepted を返すのが非同期処理の基本です。200 OK では「処理が完了した」という意味になってしまいます。また idempotencyKey を Pub/Sub の attributes に含めることで、ワーカー側で二重処理を検出できます。

AI 処理ワーカーの実装 — Pub/Sub push サブスクリプション受信

次がパイプラインの核心部分です。Pub/Sub は push 型と pull 型の2通りの受信方法があります。Cloud Run との組み合わせでは push 型が推奨されます。Pub/Sub がワーカーの Cloud Run エンドポイントに HTTP POST してくれるので、ポーリングが不要です。

// worker-service/src/index.ts
import { VertexAI } from "@google-cloud/vertexai";
import { BigQuery } from "@google-cloud/bigquery";
import express from "express";
 
const app = express();
app.use(express.json());
 
const vertexAI = new VertexAI({
  project: process.env.GCP_PROJECT_ID!,
  location: process.env.GCP_REGION ?? "us-central1",
});
const bigquery = new BigQuery({ projectId: process.env.GCP_PROJECT_ID });
 
const DATASET_ID = process.env.BQ_DATASET_ID ?? "ai_pipeline_results";
const TABLE_ID = process.env.BQ_TABLE_ID ?? "processing_jobs";
 
// 処理済みジョブIDのインメモリキャッシュ（べき等性対策）
// 本番では Redis や Firestore を使うことを推奨
const processedJobs = new Set<string>();
 
app.post("/pubsub/push", async (req, res) => {
  const startTime = Date.now();
 
  // Pub/Sub メッセージのデコード
  const pubsubMessage = req.body?.message;
  if (!pubsubMessage?.data) {
    // ACK を返して無効メッセージを破棄（無限リトライを防ぐ）
    return res.status(200).json({ status: "invalid_message_acked" });
  }
 
  let job: {
    jobId: string;
    userId: string;
    inputText: string;
    jobType: "summarize" | "classify" | "extract";
    callbackUrl?: string;
    createdAt: string;
  };
 
  try {
    const decoded = Buffer.from(pubsubMessage.data, "base64").toString("utf-8");
    job = JSON.parse(decoded);
  } catch {
    // JSON パースエラー → ACK して破棄
    return res.status(200).json({ status: "parse_error_acked" });
  }
 
  // べき等性チェック（同一ジョブの二重処理防止）
  if (processedJobs.has(job.jobId)) {
    console.log(`Duplicate job detected: ${job.jobId}, skipping`);
    return res.status(200).json({ status: "duplicate_acked" });
  }
 
  try {
    // AI 処理の実行
    const aiResult = await processWithGemini(job.inputText, job.jobType);
 
    const processingTimeMs = Date.now() - startTime;
 
    // BigQuery への結果格納
    await insertToBigQuery({
      job_id: job.jobId,
      user_id: job.userId,
      job_type: job.jobType,
      input_length: job.inputText.length,
      output_text: aiResult.text,
      output_tokens: aiResult.tokens,
      processing_time_ms: processingTimeMs,
      created_at: job.createdAt,
      completed_at: new Date().toISOString(),
      status: "success",
    });
 
    // 処理済みとしてキャッシュ
    processedJobs.add(job.jobId);
    // キャッシュが膨張しないよう古いエントリを削除
    if (processedJobs.size > 10000) {
      const oldest = processedJobs.values().next().value;
      processedJobs.delete(oldest);
    }
 
    // コールバックURL がある場合は通知
    if (job.callbackUrl) {
      await notifyCallback(job.callbackUrl, {
        jobId: job.jobId,
        status: "completed",
        result: aiResult.text,
      }).catch((e) => console.warn("Callback notification failed:", e));
    }
 
    // 200 を返すと Pub/Sub は ACK（=処理完了・再配信しない）とみなす
    return res.status(200).json({ status: "success", jobId: job.jobId });
  } catch (error) {
    console.error(`Job ${job.jobId} failed:`, error);
 
    // BigQuery にエラーログを記録
    await insertToBigQuery({
      job_id: job.jobId,
      user_id: job.userId,
      job_type: job.jobType,
      input_length: job.inputText.length,
      output_text: null,
      output_tokens: 0,
      processing_time_ms: Date.now() - startTime,
      created_at: job.createdAt,
      completed_at: new Date().toISOString(),
      status: "error",
      error_message: error instanceof Error ? error.message : String(error),
    }).catch(console.error);
 
    // 500 を返すと Pub/Sub は NACK（=再配信）とみなす
    // 設定したリトライ上限に達すると DLQ に転送される
    return res.status(500).json({ status: "error", jobId: job.jobId });
  }
});
 
async function processWithGemini(
  text: string,
  jobType: "summarize" | "classify" | "extract"
): Promise<{ text: string; tokens: number }> {
  const model = vertexAI.getGenerativeModel({
    model: "gemini-2.0-flash",
  });
 
  const prompts: Record<string, string> = {
    summarize: `以下のテキストを3〜5文で要約してください:\n\n${text}`,
    classify: `以下のテキストのカテゴリを ["positive", "negative", "neutral", "technical", "other"] から1つ選んでJSONで返してください:\n\n${text}`,
    extract: `以下のテキストから重要なキーワードを最大10個抽出してJSON配列で返してください:\n\n${text}`,
  };
 
  const response = await model.generateContent(prompts[jobType]);
  const candidate = response.response.candidates?.[0];
 
  if (!candidate?.content?.parts?.[0]?.text) {
    throw new Error("Gemini API returned empty response");
  }
 
  return {
    text: candidate.content.parts[0].text,
    tokens: response.response.usageMetadata?.totalTokenCount ?? 0,
  };
}
 
async function insertToBigQuery(row: Record<string, unknown>): Promise<void> {
  const table = bigquery.dataset(DATASET_ID).table(TABLE_ID);
  await table.insert([row], { skipInvalidRows: false, ignoreUnknownValues: false });
}
 
async function notifyCallback(url: string, payload: unknown): Promise<void> {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), 5000);
  try {
    await fetch(url, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify(payload),
      signal: controller.signal,
    });
  } finally {
    clearTimeout(timeout);
  }
}
 
const PORT = parseInt(process.env.PORT ?? "8080");
app.listen(PORT, () => console.log(`Worker service listening on :${PORT}`));

ここで重要なのは HTTP ステータスコードが Pub/Sub の再配信判断に直結する点です。200 を返すと「処理完了」として ACK され、メッセージは削除されます。4xx や 5xx を返すと「処理失敗」として NACK となり、指定回数まで再配信されます。再配信上限を超えると DLQ トピックに転送されます。

Pub/Sub トピックとサブスクリプションの設定

Antigravity に gcloud コマンドも書かせると、インフラ設定の漏れが減ります。

# メイントピックの作成
gcloud pubsub topics create ai-processing-jobs \
  --project=$GCP_PROJECT_ID
 
# デッドレターキュートピックの作成
gcloud pubsub topics create ai-processing-jobs-dlq \
  --project=$GCP_PROJECT_ID
 
# push サブスクリプションの作成（DLQ 付き）
gcloud pubsub subscriptions create ai-processing-worker \
  --topic=ai-processing-jobs \
  --push-endpoint="https://worker-service-xxxx-uc.a.run.app/pubsub/push" \
  --push-auth-service-account="pubsub-invoker@$GCP_PROJECT_ID.iam.gserviceaccount.com" \
  --ack-deadline=300 \
  --max-delivery-attempts=5 \
  --dead-letter-topic=ai-processing-jobs-dlq \
  --min-retry-delay=10s \
  --max-retry-delay=300s \
  --project=$GCP_PROJECT_ID
 
# Cloud Run サービスアカウントに Pub/Sub の起動権限を付与
gcloud run services add-iam-policy-binding worker-service \
  --member="serviceAccount:pubsub-invoker@$GCP_PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/run.invoker" \
  --region=$GCP_REGION \
  --project=$GCP_PROJECT_ID

--ack-deadline=300 は処理の最大許容時間（秒）です。AI 処理が長くなる場合はここを伸ばします。--max-delivery-attempts=5 は DLQ に移す前の再試行回数です。

BigQuery スキーマとデータ活用

処理結果の格納スキーマも、Antigravity に設計を任せると整合性の取れたものを出してくれます。

[
  { "name": "job_id",           "type": "STRING",    "mode": "REQUIRED" },
  { "name": "user_id",          "type": "STRING",    "mode": "REQUIRED" },
  { "name": "job_type",         "type": "STRING",    "mode": "REQUIRED" },
  { "name": "input_length",     "type": "INTEGER",   "mode": "NULLABLE" },
  { "name": "output_text",      "type": "STRING",    "mode": "NULLABLE" },
  { "name": "output_tokens",    "type": "INTEGER",   "mode": "NULLABLE" },
  { "name": "processing_time_ms","type": "INTEGER",  "mode": "NULLABLE" },
  { "name": "status",           "type": "STRING",    "mode": "REQUIRED" },
  { "name": "error_message",    "type": "STRING",    "mode": "NULLABLE" },
  { "name": "created_at",       "type": "TIMESTAMP", "mode": "REQUIRED" },
  { "name": "completed_at",     "type": "TIMESTAMP", "mode": "NULLABLE" }
]

このスキーマで溜めたデータを使うと、以下のような分析クエリが動きます。

-- ジョブタイプ別の平均処理時間と成功率
SELECT
  job_type,
  COUNT(*) AS total_jobs,
  COUNTIF(status = 'success') AS success_count,
  ROUND(COUNTIF(status = 'success') * 100.0 / COUNT(*), 1) AS success_rate_pct,
  ROUND(AVG(IF(status = 'success', processing_time_ms, NULL)) / 1000.0, 2) AS avg_processing_sec,
  SUM(output_tokens) AS total_tokens_used
FROM `your-project.ai_pipeline_results.processing_jobs`
WHERE created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY job_type
ORDER BY total_jobs DESC;

BigQuery + Looker Studio の連携については Antigravity × Firebase・BigQuery・Looker Studio 収益ダッシュボード構築ガイドも参考にしてください。

本番環境での監視・アラート設計

パイプラインが本番で動き始めると、監視が欠かせません。私がいつも設定するのは3つのアラートです。

DLQ メッセージ数アラート: DLQ にメッセージが溜まり始めたら、処理が詰まっているサインです。pubsub.googleapis.com/subscription/num_undelivered_messages で ai-processing-jobs-dlq-sub が閾値（例：10件）を超えたら通知するよう Cloud Monitoring に設定します。

処理レイテンシアラート: processing_time_ms の P95 が急上昇したら、AI API の応答が遅くなっている可能性があります。BigQuery のスケジュールクエリで定期チェックするのが手軽です。

エラー率アラート: Cloud Run のログベースメトリクスで status=error が5%を超えたらアラートを飛ばします。

# Cloud Monitoring アラートポリシー（Terraform 形式）
resource "google_monitoring_alert_policy" "dlq_alert" {
  display_name = "AI Pipeline DLQ Backlog Alert"
  conditions {
    display_name = "DLQ message count exceeds threshold"
    condition_threshold {
      filter         = "resource.type=\"pubsub_subscription\" AND metric.type=\"pubsub.googleapis.com/subscription/num_undelivered_messages\" AND resource.label.subscription_id=\"ai-processing-jobs-dlq-sub\""
      comparison     = "COMPARISON_GT"
      threshold_value = 10
      duration        = "60s"
      aggregations {
        alignment_period   = "60s"
        per_series_aligner = "ALIGN_MAX"
      }
    }
  }
  notification_channels = [var.notification_channel_id]
  alert_strategy {
    auto_close = "1800s"
  }
}

Cloud Run のデプロイ手順は Antigravity × Cloud Run サーバーレスデプロイ実践ガイドを参照してください。

よくある落とし穴と解決策

実際に運用してよく踏んだ問題を3つ挙げます。

① Pub/Sub の push 認証エラー

Cloud Run に push するとき、サービスアカウントの権限が足りないと 403 Forbidden が発生します。Pub/Sub のサービスアカウント（service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com）に iam.serviceAccountTokenCreator ロールが付いていないことが原因です。Antigravity に「Cloud Run への Pub/Sub push の認証設定」を聞くと正確なコマンドを出してくれますが、自分でも確認しておくと安心です。

② ack_deadline を超えた処理の NACK ループ

Gemini API の処理が重くなると ack_deadline（デフォルト10秒）を超えて自動 NACK され、同じメッセージが何度も届きます。対策は2つです。（a）ack_deadline を処理の最大想定時間より長く設定する（先ほどの設定例では300秒）。（b）ワーカー内で処理開始時に modifyAckDeadline を呼んで延長します。どちらか一方だけでも効きますが、両方組み合わせると確実です。

③ BigQuery の insertAll エラー後の無限リトライ

table.insert() がスキーマ不一致でエラーを返した場合、ワーカーが 500 を返し続けて再配信ループに入ります。この場合は BigQuery への書き込み失敗をキャッチして、ジョブ自体の成功・失敗とは切り離すか（ログだけ記録してACK）、もしくはスキーマを厳格に守るバリデーションを入れて不正データを事前に弾くかのどちらかが解決策です。私は後者を採用しています。BigQuery への書き込みはパイプラインの必須ステップですが、「書き込み失敗ならジョブ全体を再試行」よりも「書き込みはベストエフォートでエラーログを残す」設計のほうが安定しました。

④ インメモリべき等性キャッシュの限界

サンプルコードの processedJobs = new Set() はメモリ内キャッシュです。Cloud Run がスケールアウトすると複数インスタンス間でキャッシュが共有されず、異なるインスタンスが同一ジョブを処理する可能性があります。本番では Redis（Cloud Memorystore）か Firestore を使い、SET NX EX でべき等性を担保することをお勧めします。

// Redis を使ったべき等性チェック（Upstash または Cloud Memorystore）
import { Redis } from "@upstash/redis";
 
const redis = new Redis({
  url: process.env.UPSTASH_REDIS_REST_URL!,
  token: process.env.UPSTASH_REDIS_REST_TOKEN!,
});
 
async function isAlreadyProcessed(jobId: string): Promise<boolean> {
  // SET NX EX: キーが存在しない場合のみセット（1時間TTL）
  const result = await redis.set(`processed:${jobId}`, "1", {
    nx: true,
    ex: 3600,
  });
  // null が返ったらキーが既存 = 処理済み
  return result === null;
}

Antigravity でこのパイプラインを実装する際のコツ

Antigravity に「Cloud Run × Pub/Sub のパイプラインを作りたい」と伝えるだけでも骨格は出てきます。ただ、私の経験では プロジェクト固有の制約を先に伝えると質が上がります。

たとえば「処理時間は最大3分かかる可能性がある」「BigQuery のスキーマはすでに決まっている（スキーマを貼り付ける）」「Upstash Redis を使いたい（Cloud Memorystore ではなく）」といった情報です。制約が多いほど、Antigravity は余計な選択肢を提案せずに核心のコードに集中してくれます。

また、コードを生成させた後に「このコードで DLQ に入るケースを列挙して」と聞くのが個人的なお気に入りです。自分では気づかなかったエッジケースが出てきて、テストコードに反映できます。DevContainer 環境での開発については Antigravity × DevContainer 再現性ある AI 開発環境の構築も参考になります。

全体を振り返って — パイプライン化で変わること

同期処理から非同期パイプラインに移行すると、一番変わるのは「障害に対する心理的負担」です。何かが失敗しても DLQ が受け止めてくれる。リトライは自動でかかる。ログは BigQuery に溜まっている——この状態になると、夜中にアラートが来たとしても「何が起きたか」がすぐ分かり、対応が落ち着いてできます。

まず今日できる一歩は、一番重い AI API 呼び出しを1つ選んで、Pub/Sub に切り出してみることです。受付 API とワーカーの分割は30分もあればできます。最初から完璧な構成を目指さず、まず1つのジョブタイプだけパイプライン化してみてください。