こんにちは！
KDDIアイレットの取り組みとして6月22日〜7月3日の期間で開催中の「Google Cloud Next '26 / Google I/O やってみた系ブログリレー」、最終日の投稿です。
今回はAI Inference SMTでメッセージをリアルタイム加工を試してみました！
前回の記事はこちらです。
推論するだけのFunctionが不要に？
「モデルを呼んで、結果をくっつけるだけ」のCloud Run / Cloud Run functionsを書いたことはありませんか。本記事では、それを1行も書かずに、レビューをpublishするとGeminiが感情を分類し、結果付きのメッセージが下流に届く構成を作ります。自分で用意するのは、サブスクリプションに登録するYAML1枚だけです。
Pub/Subに流れてくるメッセージにAIで一手間加えたい。たとえばレビューの感情分析をしてからBigQueryに入れたい、みたいなやつです。
これまでこういう要件があると、Pub/Subからメッセージを受けて、Gemini Enterprise Agent Platform（旧Vertex AI）を叩いて、結果を別のトピックにpublishし直す、という中継役のCloud Run functionsなりCloud Runなりを一つ挟むのが定番でした。やることは「モデルを呼んで結果をくっつける」だけなのに、レート制限を考慮したリトライ処理を書き、デプロイパイプラインを整え、監視も仕込む。本質的な処理は数行で、残りの数百行は全部その周辺コード。地味に面倒な割に、どのプロジェクトにも一つはある構成だと思います。
Google Cloud Next '26のタイミングでGAになったPub/Subの「AI Inference SMT」は、この中継役をPub/Sub自体に吸収してしまう機能です。メッセージがPub/Subを通過する途中でGeminiなどのモデルが呼ばれ、推論結果が付与された状態で下流に届きます。
そもそもSMTとは
SMT（Single Message Transforms）は、Pub/Sub内でメッセージを1件ずつ変換できる機能群です。第1弾としてJavaScriptのUDF（ユーザー定義関数）が先にGAしていて、フィールドのマスキングやフォーマット変換くらいの軽い処理なら、外部サービスなしでPub/Subの中で完結できるようになっていました。
ポイントは、変換をトピック側とサブスクリプション側のどちらにも仕込めることです。トピック側のSMTはメッセージが永続化される前に適用されるので全サブスクライバーに効き、サブスクリプション側のSMTは配信直前に適用されるので特定の購読者向けだけ変換できます。「全体では生データを保持しつつ、外部連携するサブスクリプションだけPIIをマスクする」といった使い分けができる設計です。
AI Inference SMTはこのSMTファミリーの新顔で、変換処理として「モデル推論」を選べるようになった、と理解するのが早いです。
AI Inference SMTの仕組み
設定はトピックまたはサブスクリプションに対して行います。使えるモデルは大きく2系統あります。
一つはMaaS（Model-as-a-Service）系で、GeminiやClaudeなど、Model Garden経由で提供されるモデルをデプロイ管理なしでそのまま指定できます。もう一つは自前デプロイ系で、Gemini Enterprise Agent Platform（旧Vertex AI。長い）のエンドポイントにデプロイした独自モデルやオープンモデルを指定します。
メッセージがSMTを通過すると、Pub/Subが裏でモデルへのリクエストを投げ、返ってきた推論結果を元のメッセージにマージして下流へ流します。呼び出しのオーケストレーションやフロー制御（モデルエンドポイントを過負荷にしないためのレート調整）はPub/Sub側がやってくれるので、アプリ側にリトライロジックを書く必要がありません。
こんな場合に使える？
概要だけだとピンとこないと思うので、自分なりに考えたユースケースをいくつか挙げてみます。
1. 問い合わせ・レビューの分類とセンチメント付与
一番わかりやすい定番です。カスタマーサポートの問い合わせやアプリのレビューがPub/Subに流れてくる構成で、SMTで「カテゴリ」「感情スコア」「緊急度」あたりを付与してからBigQueryサブスクリプションで直接テーブルに落とす。これまで分類のためだけに存在していた中間のFunctionとリトライ処理が丸ごと消えます。Export Subscriptionと組み合わせると、publish以降ノーコードで「分類済みデータがBigQueryに溜まる」パイプラインになるのが気持ちいいところです。
2. 非構造テキストからの情報抽出（構造化）
フォームの自由記述、メールの本文、チャットログのような非構造テキストから、日付・金額・製品名などを抜き出してJSONにするパターン。正規表現で頑張ると地獄になるやつです（そして半年後、誰も読めない正規表現だけが残る）。プロンプトで「このスキーマで抽出して」と指示しておけば、下流には構造化済みのメッセージが届きます。下流の各コンシューマーがそれぞれ抽出処理を持つ必要がなくなるので、トピック側SMTに置く価値が特に出ます。
3. RAG用の埋め込み生成をインジェスト時に
ドキュメントや商品情報の更新イベントを受けて、ベクトルDBに反映する構成。従来は「更新イベント → embedding生成Function → ベクトルDB書き込み」でしたが、embedding生成をSMTに寄せれば、サブスクライバーは受け取ったベクトルを書き込むだけになります。検索インデックスの鮮度が問われるサービスで、パイプラインを一段浅くできます。
4. アラート・通知文の要約と整形
監視システムが吐く長大なアラートJSONを、人間が読める2〜3行のサマリに変換してからSlack通知用サブスクリプションに流すパターン。深夜2時に叩き起こされて、寝ぼけた頭で数百行のJSONとにらめっこした経験がある人なら、この価値はすぐわかると思います。元のメッセージは別のサブスクリプションで生のまま保存しておけるので、「人間向けは要約、機械向けは原文」という出し分けがSMTの配置だけで表現できます。
5. 多言語コンテンツの正規化
グローバル展開しているサービスで、各言語のユーザー投稿をいったん英語（あるいは日本語）に翻訳してから分析パイプラインに流すケース。翻訳という「全メッセージに一律にかけたい軽い変換」はSMTの性質と相性が良いです。
逆に向かないのは、複雑な条件分岐を伴う処理、複数メッセージをまとめて扱いたい処理、推論結果によって後続の重い処理を呼び分けたいケースです。そういうのは素直にCloud RunなりDataflowなりでやったほうがいい。SMTはあくまで「1メッセージに1変換」の道具です。
やってみた：レビューの感情分析をSMTだけで
というわけで、冒頭に挙げた「レビューの感情分析をしてからBigQueryに入れたい」を実際にやってみます。今回はまず仕組みの確認として、SMT付きサブスクリプションからpullして結果を見るところまで。BigQueryサブスクリプションへの接続は同じ要領で置き換えられます。
事前準備：サービスアカウントの権限
AI Inference SMTは、モデル呼び出しにデフォルトでPub/Subのサービスエージェント（service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com）を使います。このアカウントにVertex AI Service Agentロールを付けておきます。
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com" \
--role="roles/aiplatform.serviceAgent" \
--condition=None
SMT定義を書く
変換の定義はYAMLで書いて、トピックまたはサブスクリプション作成時に渡します。まずは素直に書いたバージョン。
# ai-smt.yaml
- aiInference:
endpoint: projects/PROJECT_ID/locations/global/publishers/google/models/gemini-3.5-flash
unstructuredInference: {
parameters: {
"max_tokens": 1024
}
}
エンドポイントにはModel Gardenのモデルをフルパスで指定します。自前デプロイのモデルなら projects/.../endpoints/ENDPOINT 形式です。
使えるモデルは、公式ドキュメントの動作確認済みMaaSモデル一覧に載っているものです。今回はその中から最新の gemini-3.5-flash を使います。
gcloud pubsub topics create reviews
gcloud pubsub subscriptions create reviews-sub \
--ack-deadline=600 \
--topic=reviews \
--message-transforms-file=ai-smt.yaml
メッセージは「プロンプトそのもの」でないといけない
サブスクリプションができたので、次はレビューのpublishです。ここが最初のつまずきポイントでした。AI Inference SMTは、メッセージデータをそのままモデルへのリクエストJSONとして送ります。Pub/Subは中身を解釈してくれません。つまりGemini基盤モデルの場合、メッセージはChat Completions API形式になっている必要があります。
gcloud pubsub topics publish reviews --message=$'{
"model": "google/gemini-3.5-flash",
"messages": [{
"role": "user",
"content": "次のレビューの感情をpositive/negative/neutralで分類し、JSONで {\\"sentiment\\": ..., \\"score\\": ...} の形式で返して。レビュー: 配送は早かったが、箱に穴が空いていた。中身も傷がついていた。"
}]
}'
「え、じゃあpublisher側でプロンプトに整形しないといけないの？」と思いますよね。既存システムが素のレビューJSON（{"review_id": ..., "text": ...}）を流している場合、publisher側を改修するのは本末転倒です。
解決策：UDFとチェーンする
ここで効いてくるのが、SMTは複数チェーンできるという仕様です。AI Inference SMTの前段にJavaScript UDFを置いて、素のレビューJSONをプロンプト形式に包んでやります。
# review-smt.yaml
- javascriptUdf:
code: |
function wrapPrompt(message, metadata) {
const review = JSON.parse(message.data);
const request = {
model: "google/gemini-3.5-flash",
messages: [{
role: "user",
content: "次のレビューの感情をpositive/negative/neutralで分類し、" +
'JSONで {"sentiment": ..., "score": ...} の形式のみで返して。' +
"レビュー: " + review.text
}]
};
message.data = JSON.stringify(request);
return message;
}
functionName: wrapPrompt
- aiInference:
endpoint: projects/PROJECT_ID/locations/global/publishers/google/models/gemini-3.5-flash
unstructuredInference: {
parameters: {
"max_tokens": 1024
}
}
先ほど作ったサブスクリプションに、この定義を反映します。
gcloud pubsub subscriptions update reviews-sub \
--message-transforms-file=review-smt.yaml
なお、このJavaScriptのために別途デプロイ先を用意する必要はありません。updateコマンドを実行した時点で、YAMLの中身がまるごとサブスクリプションの設定としてCloudに反映され、Pub/Subのマネージド環境内で実行されます。Cloud Functionsのようなランタイム管理も不要です。
gcloud pubsub topics publish reviews --message='{"review_id": "r-001", "text": "配送は早かったが、箱に穴が空いていた。中身も傷がついていた。"}'
結果を確認してみる
# pullは空振りすることがあるので数回リトライ
for i in {1..5}; do gcloud pubsub subscriptions pull reviews-sub --auto-ack --limit=10; sleep 5; done
出ました。整形するとこういう形です。
{
"original_message": {
"model": "google/gemini-3.5-flash",
"messages": [{
"role": "user",
"content": "次のレビューの感情をpositive/negative/neutralで分類し、JSONで {\"sentiment\": ..., \"score\": ...} の形式のみで返して。レビュー: 配送は早かったが、箱に穴が空いていた。中身も傷がついていた。"
}]
},
"model_output": {
"choices": [{
"finish_reason": "stop",
"index": 0,
"message": {
"role": "assistant",
"content": "{\n \"sentiment\": \"negative\",\n \"score\": -0.8\n}"
}
}],
"model": "google/gemini-3.5-flash",
"usage": {
"prompt_tokens": 56,
"completion_tokens": 20,
"completion_tokens_details": { "reasoning_tokens": 926 },
"total_tokens": 1002
}
}
}
「配送は早いが箱に穴、中身にも傷」に対して negative。妥当な判定です。中継のFunctionは一行も書いていません。リトライも429対応も、全部Pub/Subの中の人がやってくれます。あとはBigQueryサブスクリプションに繋げば「publishするだけで分類済みレビューがテーブルに溜まる」パイプラインの完成です。
実際に流してみてわかったこと
複数メッセージを流してみると、机上ではわからなかったことがいくつも見えました。
LLMの出力は素直じゃない
「JSON形式のみで返して」と指示しても、出力がコードフェンス（```json）で包まれて返ってくる回があったり、モデルやリクエストによって変わったりと、出力形式は完全には揃いません。このままBigQueryに入れると下流のパースが割れます。AI Inference SMTの後段にもう一つUDFを置いて、model_output からJSONを抜き出して検証・整形する後処理チェーンは、「あると便利」ではなく実質必須だと感じました。
同じ入力でも結果は揺れる
同じレビュー文を2回流したら、スコアが 1.0 と 0.99 で返ってきました。当たり前といえば当たり前ですが、「リトライで再推論されたら結果が変わり得る」という冪等性の注意が、実際に目の前で起きると重みが違います。
review_idが消えた
届いたメッセージの original_message は、UDF適用後のリクエストJSONです。つまり素のレビューにあった review_id はUDFが message.data を上書きした時点で消えています。下流で元レコードと突合できないのは実務では致命傷なので、UDF内で message.attributes にIDを退避する（message.attributes.review_id = review.review_id; を挟む）のが正解でした。データを上書きする変換では、残したいメタデータを属性側に逃がしておくとよさそうです。
古いメッセージも新しい変換を通る
UDFチェーン版にupdateする前にpublishしていたメッセージが、update後の配信で新しいチェーンを通って出てきました（UDFが想定しないJSONだったため「レビュー: undefined」というプロンプトになって推論されていました）。変換が実行されるのは「配信時」であり、サブスクリプションに溜まっているメッセージも定義変更後は新定義で処理される、という挙動の裏付けです。本番で変換を差し替えるときは、滞留メッセージの存在を頭に入れておく必要があります。
使う前に知っておきたい制約
実際に組み込む前に押さえておきたい点がいくつかあります。
まず、1メッセージにつき1推論リクエストで、バッチングはされません。大量のメッセージが流れるトピックに雑に仕込むと、そのままモデル呼び出し回数、つまり請求額に跳ねます。秒間1,000メッセージのトピックにノリで設定するのはやめましょう。料金はモデルの利用料に加えて、推論オーケストレーションのインフラ料金が別途かかる2階建てなので、流量の多いトピックでは事前に試算しておいたほうが安全です。
推論には60秒の制限があります。超えると配信試行がタイムアウトしてリトライされ、リトライしきれなければ（設定していれば）デッドレタートピックに送られます。ここで注意したいのがリトライ時の再推論です。同じメッセージに対して推論が複数回走り得るので、LLMの出力が毎回同じとは限らない前提で、下流を冪等に作っておく必要があります。
まとめ
AI Inference SMTは、「推論して結果をくっつけるだけ」の中継コンポーネントをPub/Subに畳み込める機能です。分類・抽出・要約・埋め込み生成のような、1メッセージ完結の軽い推論なら、パイプラインを一段シンプルにできます。
一方で、メッセージング層というのは本来、軽くて挙動が予測しやすいべき場所です。そこにLLMという「遅いし、たまに聞いてないことを喋り出す」同僚みたいなコンポーネントを持ち込むわけなので、レイテンシ・コスト・冪等性の設計は従来より一段気を使う必要があります。便利さと引き換えに責務がメッセージ基盤に寄る、という構図は意識しておきたいところです。
実際に触ってみて、UDFとのチェーンで前処理・後処理まで組めるとわかったのは収穫でした。「publisherはそのまま、真ん中で全部やる」が現実的に成立します。今後、「推論して結果をくっつけるだけ」の要件が来たら、Functionを書き始める前に、まずSMTで済まないかを考えるようになりそうです。
※2026年7月時点の情報です。料金・仕様は変わる可能性があるので、最新の公式ドキュメントを確認してください。