Google Cloud Workflows × Gemini API 本番オーケストレーション：タイムアウト・リトライ・コスト制御まで

# 必要なAPIを有効化
gcloud services enable workflows.googleapis.com
gcloud services enable cloudscheduler.googleapis.com
gcloud services enable aiplatform.googleapis.com
 
# ワークフロー実行用サービスアカウントを作成
gcloud iam service-accounts create gemini-workflow-runner \
    --display-name="Gemini Workflow Runner"
 
SA_EMAIL="gemini-workflow-runner@YOUR_PROJECT_ID.iam.gserviceaccount.com"
 
# 必要な権限を付与
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
    --member="serviceAccount:${SA_EMAIL}" \
    --role="roles/aiplatform.user"
 
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
    --member="serviceAccount:${SA_EMAIL}" \
    --role="roles/workflows.invoker"
 
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
    --member="serviceAccount:${SA_EMAIL}" \
    --role="roles/storage.objectViewer"

# workflow-gemini-summarize.yaml
main:
  params: [args]
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - location: "us-central1"
          - model: "gemini-2.5-pro"
          - input_text: ${args.text}
 
    - call_gemini_api:
        try:
          call: http.post
          args:
            url: ${"https://" + location + "-aiplatform.googleapis.com/v1/projects/" + project_id + "/locations/" + location + "/publishers/google/models/" + model + ":generateContent"}
            auth:
              type: OAuth2
            headers:
              Content-Type: "application/json"
            body:
              contents:
                - role: "user"
                  parts:
                    - text: ${"以下のテキストを200文字以内で要約してください:\n\n" + input_text}
              generationConfig:
                maxOutputTokens: 500
                temperature: 0.3
          result: gemini_response
        retry:
          predicate: ${http_retry_predicate}
          max_retries: 3
          backoff:
            initial_delay: 2
            max_delay: 60
            multiplier: 2
        except:
          as: e
          steps:
            - handle_api_error:
                raise: ${e}
 
    - extract_text:
        assign:
          - summary: ${gemini_response.body.candidates[0].content.parts[0].text}
 
    - return_result:
        return:
          summary: ${summary}
          input_tokens: ${gemini_response.body.usageMetadata.promptTokenCount}
          output_tokens: ${gemini_response.body.usageMetadata.candidatesTokenCount}
 
# リトライ条件（429: レート制限、503: サービス一時停止）
http_retry_predicate:
  params: [e]
  steps:
    - check_status:
        switch:
          - condition: ${e.code == 429 or e.code == 503}
            return: true
        return: false

# ワークフローをデプロイ
gcloud workflows deploy gemini-summarize \
    --source=workflow-gemini-summarize.yaml \
    --location=us-central1 \
    --service-account="${SA_EMAIL}"
 
# 実行（テスト）
gcloud workflows run gemini-summarize \
    --location=us-central1 \
    --data='{"text": "Google Cloud Workflowsは、HTTPベースのサービスを呼び出す複数のステップを定義できるサーバーレスオーケストレーターです。各ステップの実行状態は自動保存されます。"}'

{
  "summary": "Google Cloud Workflowsは、HTTPサービスを呼び出す複数ステップを定義できるサーバーレスオーケストレーターで、各ステップの実行状態を自動保存します。",
  "input_tokens": 87,
  "output_tokens": 52
}

# workflow-content-pipeline.yaml
main:
  params: [args]
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - location: "us-central1"
          - model: "gemini-2.5-pro"
          - topic: ${args.topic}
          - target_length: ${default(map.get(args, "target_length"), 2000)}
          - bucket: ${default(map.get(args, "output_bucket"), "gemini-workflow-outputs")}
          - run_id: ${sys.now()}
 
    # Step 1: 見出し構成を生成
    - generate_outline:
        call: googleapis.aiplatform.v1.projects.locations.publishers.models.generateContent
        args:
          model: ${"projects/" + project_id + "/locations/" + location + "/publishers/google/models/" + model}
          body:
            contents:
              - role: "user"
                parts:
                  - text: |
                      ${topic}について、${target_length}文字程度の記事を書くための
                      見出し構成（H2を4〜6個）を作成してください。
                      JSON配列形式で返してください：
                      ["見出し1", "見出し2", ...]
            generationConfig:
              responseMimeType: "application/json"
              maxOutputTokens: 300
        result: outline_response
 
    - parse_outline:
        assign:
          - outline_json: ${outline_response.candidates[0].content.parts[0].text}
          - headings: ${json.decode(outline_json)}
 
    # Step 2: 各見出しの本文を並列生成
    - generate_sections:
        parallel:
          for:
            value: heading
            in: ${headings}
            steps:
              - generate_section_content:
                  call: http.post
                  args:
                    url: ${"https://" + location + "-aiplatform.googleapis.com/v1/projects/" + project_id + "/locations/" + location + "/publishers/google/models/" + model + ":generateContent"}
                    auth:
                      type: OAuth2
                    headers:
                      Content-Type: "application/json"
                    body:
                      contents:
                        - role: "user"
                          parts:
                            - text: |
                                ${topic}に関する記事の「${heading}」セクションを書いてください。
                                400〜600文字程度で、丁寧語（です/ます調）で書いてください。
                      generationConfig:
                        maxOutputTokens: 1000
                        temperature: 0.7
                  result: section_response
 
    # Step 3: 品質チェック（ハルシネーション検出）
    - quality_check:
        call: http.post
        args:
          url: ${"https://" + location + "-aiplatform.googleapis.com/v1/projects/" + project_id + "/locations/" + location + "/publishers/google/models/" + model + ":generateContent"}
          auth:
            type: OAuth2
          headers:
            Content-Type: "application/json"
          body:
            contents:
              - role: "user"
                parts:
                  - text: |
                      以下のコンテンツに明らかな事実誤りや矛盾がないか確認してください。
                      問題があれば "ISSUES: [問題点]"、問題なければ "OK" と返してください。
 
                      トピック: ${topic}
            generationConfig:
              maxOutputTokens: 200
              temperature: 0
          result: quality_response
 
    - check_quality_result:
        assign:
          - quality_text: ${quality_response.body.candidates[0].content.parts[0].text}
 
    - validate_quality:
        switch:
          - condition: ${text.substring(quality_text, 0, 6) == "ISSUES"}
            steps:
              - quality_failed:
                  raise:
                    code: 400
                    message: ${"品質チェック失敗: " + quality_text}
 
    # Step 4: Cloud Storageに保存
    - save_to_storage:
        call: http.post
        args:
          url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + bucket + "/o?uploadType=media&name=articles/" + run_id + ".json"}
          auth:
            type: OAuth2
          headers:
            Content-Type: "application/json"
          body:
            topic: ${topic}
            headings: ${headings}
            quality_status: "OK"
            generated_at: ${run_id}
        result: storage_response
 
    - return_success:
        return:
          status: "SUCCESS"
          storage_path: ${"gs://" + bucket + "/articles/" + run_id + ".json"}
          headings_count: ${list.length(headings)}

# ❌ ネイティブコネクタの場合
# result.candidates[0].content.parts[0].text
 
# ✅ http.post の場合  
# result.body.candidates[0].content.parts[0].text

# 並列実行結果の正しい収集方法
- init_results:
    assign:
      - results: []
 
- parallel_generate:
    parallel:
      shared: [results]  # ← shared を明示
      for:
        value: item
        in: ${items}
        steps:
          - process:
              call: http.post
              # ...
              result: item_result
          - collect:
              assign:
                - results: ${list.concat(results, [item_result.body])}

# 大きなコンテンツは即座にGCSに保存
- save_large_content:
    call: http.post
    args:
      url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + bucket + "/o?uploadType=media&name=temp/" + item_id + ".txt"}
      auth:
        type: OAuth2
      body: ${large_content}
    result: gcs_result
 
# 後続ステップではURLのみ保持
- store_url_only:
    assign:
      - content_url: ${"gs://" + bucket + "/temp/" + item_id + ".txt"}

- call_gemini_with_timeout:
    call: http.post
    args:
      url: ${gemini_endpoint}
      timeout: 300  # ← 5分に設定（デフォルト30秒では不足することがある）
      auth:
        type: OAuth2
      body:
        # ...

# Cloud Schedulerジョブを作成（毎日午前6時JST）
gcloud scheduler jobs create http gemini-content-daily \
    --location=asia-northeast1 \
    --schedule="0 21 * * *" \
    --uri="https://workflowexecutions.googleapis.com/v1/projects/YOUR_PROJECT_ID/locations/us-central1/workflows/content-pipeline/executions" \
    --message-body='{"argument": "{\"topic\": \"Gemini APIの最新活用事例\", \"target_length\": 2000}"}' \
    --oauth-service-account-email="${SA_EMAIL}" \
    --oauth-token-scope="https://www.googleapis.com/auth/cloud-platform"

import json
import time
from google.cloud import workflows_v1
from google.cloud.workflows import executions_v1
 
def run_gemini_workflow(
    project_id: str,
    location: str,
    workflow_name: str,
    topic: str,
    target_length: int = 2000,
    poll_interval: int = 10,
    max_wait_seconds: int = 600
) -> dict:
    """
    Cloud Workflowsを起動し、完了まで待機して結果を返す。
    
    Args:
        project_id: GCPプロジェクトID
        location: ワークフローのリージョン（例: us-central1）
        workflow_name: ワークフロー名
        topic: コンテンツ生成のトピック
        target_length: 目標文字数
        poll_interval: ポーリング間隔（秒）
        max_wait_seconds: 最大待機時間（秒）
    
    Returns:
        ワークフローの実行結果
    
    Raises:
        TimeoutError: max_wait_seconds を超えても完了しない場合
        RuntimeError: ワークフローが失敗した場合
    """
    # Executionsクライアントを初期化
    exec_client = executions_v1.ExecutionsClient()
    
    # ワークフローのリソース名を構築
    parent = exec_client.workflow_path(project_id, location, workflow_name)
    
    # 実行を開始
    execution = exec_client.create_execution(
        request=executions_v1.CreateExecutionRequest(
            parent=parent,
            execution=executions_v1.Execution(
                argument=json.dumps({
                    "topic": topic,
                    "target_length": target_length,
                    "output_bucket": f"{project_id}-gemini-outputs"
                })
            )
        )
    )
    
    execution_name = execution.name
    print(f"ワークフロー実行開始: {execution_name}")
    
    # 完了までポーリング
    elapsed = 0
    while elapsed < max_wait_seconds:
        execution = exec_client.get_execution(
            request=executions_v1.GetExecutionRequest(name=execution_name)
        )
        
        state = execution.state
        
        if state == executions_v1.Execution.State.SUCCEEDED:
            result = json.loads(execution.result)
            print(f"✅ 完了: {result}")
            return result
            
        elif state == executions_v1.Execution.State.FAILED:
            error = execution.error
            raise RuntimeError(
                f"ワークフロー失敗: {error.message} (コード: {error.http_status_code})"
            )
            
        elif state == executions_v1.Execution.State.CANCELLED:
            raise RuntimeError("ワークフローがキャンセルされました")
        
        # まだ実行中
        print(f"  実行中... ({elapsed}秒経過)")
        time.sleep(poll_interval)
        elapsed += poll_interval
    
    raise TimeoutError(
        f"ワークフローが{max_wait_seconds}秒以内に完了しませんでした。"
        f"実行ID: {execution_name}"
    )
 
# 使用例
if __name__ == "__main__":
    try:
        result = run_gemini_workflow(
            project_id="your-project-id",
            location="us-central1",
            workflow_name="content-pipeline",
            topic="Gemini 2.5 Proの実践的な使い方",
            target_length=2000
        )
        print(f"保存先: {result['storage_path']}")
    except RuntimeError as e:
        print(f"エラー: {e}")
    except TimeoutError as e:
        print(f"タイムアウト: {e}")

from google.cloud import billing_budgets_v1
from google.protobuf import field_mask_pb2
 
def create_gemini_budget_alert(
    billing_account_id: str,
    project_id: str,
    monthly_budget_usd: float,
    alert_thresholds: list[float] = [0.5, 0.8, 1.0],
    notification_email: str = None
) -> str:
    """
    Gemini API利用のコスト予算アラートを設定する。
    
    Args:
        billing_account_id: 請求アカウントID（例: 012345-ABCDEF-789012）
        project_id: 監視対象のプロジェクトID
        monthly_budget_usd: 月間予算（USD）
        alert_thresholds: アラートを発火する閾値（0.0〜1.0のリスト）
        notification_email: 通知先メールアドレス
    
    Returns:
        作成された予算のリソース名
    """
    client = billing_budgets_v1.BudgetServiceClient()
    
    # 予算フィルター（Gemini APIの費用のみ対象）
    budget_filter = billing_budgets_v1.Filter(
        projects=[f"projects/{project_id}"],
        services=["services/F3B6-3D8E-295F"],  # AI Platform / Vertex AI のサービスID
        credit_types_treatment=billing_budgets_v1.Filter.CreditTypesTreatment.EXCLUDE_ALL_CREDITS
    )
    
    # アラート閾値の設定
    threshold_rules = [
        billing_budgets_v1.ThresholdRule(
            threshold_percent=threshold,
            spend_basis=billing_budgets_v1.ThresholdRule.Basis.CURRENT_SPEND
        )
        for threshold in alert_thresholds
    ]
    
    # 予算の作成
    budget = billing_budgets_v1.Budget(
        display_name=f"Gemini API Budget - {project_id}",
        budget_filter=budget_filter,
        amount=billing_budgets_v1.BudgetAmount(
            specified_amount={
                "currency_code": "USD",
                "units": int(monthly_budget_usd),
                "nanos": int((monthly_budget_usd % 1) * 1e9)
            }
        ),
        threshold_rules=threshold_rules
    )
    
    parent = f"billingAccounts/{billing_account_id}"
    created_budget = client.create_budget(parent=parent, budget=budget)
    
    print(f"予算アラート作成完了: {created_budget.name}")
    print(f"月間上限: ${monthly_budget_usd} USD")
    print(f"アラート閾値: {[f'{int(t*100)}%' for t in alert_thresholds]}")
    
    return created_budget.name
 
# 設定例：月$50のアラートを50%・80%・100%で発火
budget_name = create_gemini_budget_alert(
    billing_account_id="012345-ABCDEF-789012",
    project_id="my-gemini-project",
    monthly_budget_usd=50.0,
    alert_thresholds=[0.5, 0.8, 1.0],
    notification_email="admin@example.com"
)

# ワークフロー失敗率のアラートポリシーを作成
gcloud alpha monitoring policies create \
    --policy-from-file=- << 'EOF'
{
  "displayName": "Cloud Workflow Failure Alert",
  "conditions": [
    {
      "displayName": "Workflow execution failure rate",
      "conditionThreshold": {
        "filter": "resource.type=\"workflows.googleapis.com/Workflow\" AND metric.type=\"workflows.googleapis.com/finished_execution_count\" AND metric.labels.status=\"FAILED\"",
        "comparison": "COMPARISON_GT",
        "thresholdValue": 0,
        "duration": "60s",
        "aggregations": [
          {
            "alignmentPeriod": "300s",
            "perSeriesAligner": "ALIGN_SUM"
          }
        ]
      }
    }
  ],
  "alertStrategy": {
    "notificationRateLimit": {
      "period": "3600s"
    }
  }
}
EOF

# ワークフロー実行ログをBigQueryでクエリする例
from google.cloud import bigquery
 
def analyze_workflow_costs(project_id: str, days: int = 30) -> None:
    """直近N日間のワークフロー実行コストを分析する"""
    client = bigquery.Client()
    
    query = f"""
    SELECT
        DATE(timestamp) as date,
        COUNT(*) as total_executions,
        COUNTIF(JSON_VALUE(jsonPayload.state) = 'SUCCEEDED') as succeeded,
        COUNTIF(JSON_VALUE(jsonPayload.state) = 'FAILED') as failed,
        -- Gemini APIトークン使用量（各実行のログから集計）
        SUM(CAST(JSON_VALUE(jsonPayload.metadata.totalTokens) AS INT64)) as total_tokens
    FROM `{project_id}.global._Default._AllLogs`
    WHERE
        resource.type = 'workflows.googleapis.com/Workflow'
        AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {days} DAY)
    GROUP BY date
    ORDER BY date DESC
    """
    
    results = client.query(query)
    
    print(f"直近{days}日間のワークフロー実行サマリー:")
    print(f"{'日付':<12} {'実行数':>6} {'成功':>6} {'失敗':>6} {'総トークン':>12}")
    print("-" * 50)
    
    for row in results:
        total_tokens = row.total_tokens or 0
        # Gemini 2.5 Pro の料金（概算）: $7/1M input tokens
        estimated_cost = (total_tokens / 1_000_000) * 7
        print(
            f"{str(row.date):<12} "
            f"{row.total_executions:>6} "
            f"{row.succeeded:>6} "
            f"{row.failed:>6} "
            f"{total_tokens:>12,} "
            f"(推定 ${estimated_cost:.4f})"
        )