Google Chat Bot で cardsV2 のプログレッシブ UX を実装したら壁だらけだった話

ユーザー: 「社内規定について教えて」
     ↓
（5〜10秒の沈黙）
     ↓
ボット: 「回答テキスト」

1. HTTP レスポンスで {} を返す（即座にレスポンスを返してタイムアウトを回避）
2. バックグラウンドスレッドで Chat API を呼び、「処理中...」カードを作成
3. パイプラインが進むたびにカードをパッチして進捗を表示
4. 完了時に最終結果でカードをパッチ

from googleapiclient.discovery import build
import google.auth

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

# メッセージ作成
response = service.spaces().messages().create(
    parent="spaces/SPACE_ID",
    body={"cardsV2": [{"cardId": "my-card", "card": {...}}]}
).execute()

# メッセージ更新（パッチ）
service.spaces().messages().patch(
    name=response["name"],
    updateMask="cardsV2",
    body={"cardsV2": [{"cardId": "my-card", "card": {...}}]}
).execute()

{
    "collapsible": True,
    "uncollapsibleWidgetsCount": 1,  # 最初の1つは常に表示
    "widgets": [
        # ↓ 常に表示（ステータス）
        {"decoratedText": {"text": "回答を生成中..."}},
        # ↓ 折りたたみ内（ステップ履歴）
        {"decoratedText": {"text": '<font color="#00C853">✅ 問い合わせを解析中</font>'}},
        {"decoratedText": {"text": '<font color="#00C853">✅ 検索クエリを作成中</font>'}},
        {"decoratedText": {"text": '<font color="#2979FF">⏳ 回答を生成中</font>'}},
    ]
}

class ThrottledPatcher:
    def __init__(self, chat_client, message_name, min_interval=1.0):
        self._chat_client = chat_client
        self._message_name = message_name
        self._min_interval = min_interval
        self._last_patch_time = 0.0
        self._buffered_body = None

    def patch(self, body, force=False):
        now = time.monotonic()
        elapsed = now - self._last_patch_time
        if force or elapsed >= self._min_interval:
            if force and elapsed < self._min_interval:
                time.sleep(self._min_interval - elapsed)
            self._chat_client.patch_message(
                self._message_name, body, "cardsV2"
            )
            self._last_patch_time = time.monotonic()
            self._buffered_body = None
        else:
            self._buffered_body = body  # latest-wins バッファ

    def flush(self):
        if self._buffered_body is not None:
            remaining = self._min_interval - (
                time.monotonic() - self._last_patch_time
            )
            if remaining > 0:
                time.sleep(remaining)
            self._chat_client.patch_message(
                self._message_name, self._buffered_body, "cardsV2"
            )
            self._buffered_body = None

HTTP レスポンス返却（即座） → バックグラウンドスレッド開始
     → build("chat", "v1") で Discovery Document をダウンロード（~2分）
     → カード作成 API 呼び出し
     → ユーザーにカードが表示される

@functions_framework.http
def handle_chat(request):
    body = request.get_json(silent=True)
    # ...
    if not is_warm():
        # コールドスタート: 同期でテキストメッセージを返す
        thread = threading.Thread(target=process_message, args=(...,))
        thread.start()
        return create_message("処理を開始しました。少々お待ちください...")
    else:
        # ウォーム: バックグラウンドでカードを作成
        thread = threading.Thread(target=process_message, args=(...,))
        thread.start()
        return {}

gcloud run services describe google-chat-bot --region=asia-northeast1 \
  --format="yaml(spec.template.metadata.annotations)"

annotations:
  run.googleapis.com/cpu-throttling: "true"  # ← これが原因

# Cloud Function のはずが、Cloud Run サービスとして見える
gcloud run services list --region=asia-northeast1
# NAME: google-chat-bot  ← 同じ名前で Cloud Run サービスが存在する

1. HTTP リクエスト受信
2. バックグラウンドスレッドを起動
3. HTTP レスポンス {} を即座に返却  ← ここで CPU 割り当てが激減
4. バックグラウンドスレッドで Chat API を呼ぶ  ← CPU がほぼない状態

# Step 1: CPU とメモリを増やしてデプロイ
gcloud functions deploy google-chat-bot \
  --gen2 \
  --runtime=python314 \
  --region=asia-northeast1 \
  --source=. \
  --entry-point=handle_chat \
  --trigger-http \
  --no-allow-unauthenticated \
  --memory=512Mi \
  --cpu=1

# Step 2: CPU スロットリングを無効化
gcloud run services update google-chat-bot \
  --region=asia-northeast1 \
  --no-cpu-throttling

# Discovery Document をダウンロード
curl -o chat_discovery.json \
  'https://chat.googleapis.com/$discovery/rest?version=v1'

import json
from pathlib import Path
from googleapiclient.discovery import build_from_document

_DISCOVERY_DOC_PATH = Path(__file__).parent / "chat_discovery.json"

def _get_default_service():
    credentials, _ = google.auth.default(scopes=SCOPES)
    doc = json.loads(_DISCOVERY_DOC_PATH.read_text())
    return build_from_document(doc, credentials=credentials)

# 定期的に再ダウンロード
curl -o chat_discovery.json \
  'https://chat.googleapis.com/$discovery/rest?version=v1'

# ❌ 動かない — function に関数名を指定
{
    "onClick": {
        "action": {
            "function": "feedback",
            "parameters": [
                {"key": "vote", "value": "up"},
            ]
        }
    }
}

# ✅ 正しい — function に完全な HTTPS URL を指定
{
    "onClick": {
        "action": {
            "function": "https://asia-northeast1-PROJECT.cloudfunctions.net/google-chat-bot",
            "parameters": [
                {"key": "action", "value": "feedback"},
                {"key": "vote", "value": "up"},
            ]
        }
    }
}

# ❌ request.base_url は内部 URL を返す
endpoint_url = request.base_url
# → "http://localhost:8080/" （Cloud Run の内部プロキシの URL）

# ❌ request.path は "/" を返す
host = request.headers.get("X-Forwarded-Host")  # "asia-northeast1-PROJECT.cloudfunctions.net"
scheme = request.headers.get("X-Forwarded-Proto")  # "https"
path = request.path  # "/" ← "/google-chat-bot" ではない！

import os

host = request.headers.get("X-Forwarded-Host") or request.headers.get("Host", "")
scheme = request.headers.get("X-Forwarded-Proto", "https")
service = os.environ.get("K_SERVICE", "")
endpoint_url = f"{scheme}://{host}/{service}" if host else ""
# → "https://asia-northeast1-PROJECT.cloudfunctions.net/google-chat-bot"

// ❌ renderActions はダイアログ用
{"renderActions": {"action": {"navigations": [{"updateCard": {...}}]}}}

// ❌ updateMessageAction は同期レスポンスのメッセージ更新用
{"hostAppDataAction": {"chatDataAction": {"updateMessageAction": {"message": {...}}}}}

// ✅ CARD_CLICKED に対する正しいレスポンス
{
  "actionResponse": {"type": "UPDATE_MESSAGE"},
  "cardsV2": [{
    "cardId": "progressive-card",
    "card": {
      "sections": [{
        "widgets": [{
          "textParagraph": {"text": "フィードバックありがとうございます！"}
        }]
      }]
    }
  }]
}

main.py              → HTTP ハンドラ（{} を返却、スレッド起動、CARD_CLICKED ルーティング）
worker.py            → パイプラインオーケストレーション（ステップ追跡付き）
cards.py             → cardsV2 ビルダー（プログレッシブカード）
models.py            → パイプラインデータモデル（StepStatus, PipelineState）
throttle.py          → レートリミット対応パッチャー（1 write/sec/space）
feedback.py          → CARD_CLICKED イベントハンドラ
chat_api.py          → Chat API ラッパー（静的 Discovery Document）
chat_discovery.json  → バンドル済み Chat API v1 Discovery Document

1. HTTP リクエスト受信 → {} を即座に返却
2. バックグラウンドスレッドでパイプライン開始
3. 初期カード作成（4ステップ表示、すべて PENDING）
4. 各ステップの進行に合わせてカードをパッチ
   - 問い合わせを解析中 → ✅
   - 検索クエリを作成中 → ✅
   - ナレッジベースを検索中 → ✅
   - 回答を生成中 → ✅
5. レスポンス段落を追加しながらパッチ
6. 完了時にフィードバックボタンを表示

# Step 1: 関数デプロイ
gcloud functions deploy google-chat-bot \
  --gen2 \
  --runtime=python314 \
  --region=asia-northeast1 \
  --source=. \
  --entry-point=handle_chat \
  --trigger-http \
  --no-allow-unauthenticated \
  --memory=512Mi \
  --cpu=1

# Step 2: CPU スロットリング無効化（バックグラウンドスレッドに必須）
gcloud run services update google-chat-bot \
  --region=asia-northeast1 \
  --no-cpu-throttling

現在:  HTTP リクエスト → {} 返却 → 同じインスタンスでバックグラウンドスレッド実行
将来:  HTTP リクエスト → Cloud Tasks にエンキュー → 別リクエストとしてワーカーが処理