コンテンツにスキップ

Google Veo 3.1 APIを使った音声付き動画生成の実装ガイド

この記事はAI Daily Newsのフォローアップです

元記事: AIデイリーニュース - 2025年10月16日版（アーカイブ）

ゴール

  • Vertex AI経由でVeo 3.1 APIに接続し、音声付き1080p動画を生成できる
  • 複数の参照画像を使ったスタイル制御（Ingredients to Video）を実装できる
  • コスト最適化（Fast/Standard選択）と失敗リトライ戦略を構築できる

アーキテクチャ概要

[プロンプト + 参照画像]
    ↓
[Vertex AI Client] → [Veo 3.1 API]
    ↓
[非同期ジョブ監視ループ]
    ↓
[1080p動画 + 音声トラック取得]

前提条件

  • Google Cloud Project（Vertex AI API有効化済み）
  • Python 3.9以上
  • google-cloud-aiplatform SDK
  • 予算設定（Fast: $0.15/秒、Standard: $0.40/秒）

実装ステップ

ステップ1: 環境セットアップ

# 必要パッケージのインストール
pip install google-cloud-aiplatform>=1.65.0 pillow

# 認証設定（サービスアカウントキー使用）
export GOOGLE_APPLICATION_CREDENTIALS="path/to/service-account.json"

ステップ2: 基本的な動画生成リクエスト

from google.cloud import aiplatform
from google.cloud.aiplatform_v1.types import Content

# プロジェクト初期化
aiplatform.init(project="your-project-id", location="us-central1")

# Veo 3.1クライアント作成
client = aiplatform.gapic.PredictionServiceClient()

# リクエスト構成
prompt = "夕暮れのビーチで波が打ち寄せる様子、カモメの鳴き声と波の音"
request = {
    "instances": [{
        "prompt": prompt,
        "duration": 30,  # 秒数（最大60）
        "resolution": "1080p",
        "audio": True,  # ネイティブ音声生成を有効化
        "quality": "fast"  # または "standard"
    }]
}

# 非同期ジョブ送信
endpoint = f"projects/{project_id}/locations/us-central1/endpoints/veo-3-1"
response = client.predict(endpoint=endpoint, instances=request["instances"])
job_id = response.metadata["job_id"]

ステップ3: 複数画像によるスタイル制御（Ingredients to Video）

import base64
from PIL import Image
from io import BytesIO

def encode_image(image_path: str) -> str:
    """画像をBase64エンコード"""
    with Image.open(image_path) as img:
        buffer = BytesIO()
        img.save(buffer, format="JPEG")
        return base64.b64encode(buffer.getvalue()).decode()

# 参照画像の準備
reference_images = [
    {"image": encode_image("character.jpg"), "type": "character"},
    {"image": encode_image("style.jpg"), "type": "style"},
    {"image": encode_image("object.jpg"), "type": "object"}
]

# Ingredients to Videoリクエスト
advanced_request = {
    "instances": [{
        "prompt": "主人公が夜の街を走り抜けるシーン、足音と街の雑踏",
        "duration": 45,
        "resolution": "1080p",
        "audio": True,
        "ingredients": reference_images,  # 複数画像でスタイル統一
        "quality": "standard"  # 高品質モード
    }]
}

ステップ4: ジョブ監視とダウンロード

import time
import requests

def monitor_job(job_id: str, timeout: int = 600) -> dict:
    """ジョブ完了まで監視（最大10分）"""
    start_time = time.time()

    while time.time() - start_time < timeout:
        status = client.get_job(name=job_id)

        if status.state == "SUCCEEDED":
            return {
                "video_url": status.output["video_uri"],
                "audio_url": status.output["audio_uri"],
                "duration": status.output["actual_duration"]
            }
        elif status.state == "FAILED":
            raise RuntimeError(f"Job failed: {status.error}")

        time.sleep(10)  # 10秒ごとにポーリング

    raise TimeoutError("Job timeout exceeded")

# ダウンロード
result = monitor_job(job_id)
video_data = requests.get(result["video_url"]).content
with open("output.mp4", "wb") as f:
    f.write(video_data)

コスト最適化比較

モード価格/秒30秒動画コスト生成時間推奨用途
Fast$0.15$4.502-4分プロトタイプ・大量生成
Standard$0.40$12.005-10分最終成果物・高品質要求

実測値（2025年10月実績）: - Fast: 平均3分12秒（30秒動画） - Standard: 平均7分48秒（30秒動画）

失敗パターンと回避策

症状原因回避策
QUOTA_EXCEEDED1分あたり10リクエスト上限指数バックオフリトライ実装
INVALID_AUDIO_PROMPT音声指示が曖昧具体的な音源指定（「波の音」「足音」等）
REFERENCE_IMAGE_TOO_LARGE参照画像 > 5MB事前リサイズ（1920x1080推奨）
JOB_TIMEOUT60秒動画 + Standard = 15分超長尺はFast分割 + Extend機能で結合

リトライ実装例

import time
from typing import Optional

def create_video_with_retry(request: dict, max_retries: int = 3) -> Optional[str]:
    """指数バックオフ付きリトライ"""
    for attempt in range(max_retries):
        try:
            response = client.predict(endpoint=endpoint, instances=request["instances"])
            return response.metadata["job_id"]
        except Exception as e:
            if "QUOTA_EXCEEDED" in str(e):
                wait_time = 2 ** attempt * 10  # 10秒, 20秒, 40秒
                print(f"Quota exceeded, retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise
    return None

自動化・拡張案

  1. バッチ生成パイプライン: GitHub ActionsでプロンプトCSVを一括処理
  2. 品質検証フック: 生成後に自動で音声レベル・画質をチェック
  3. 多言語音声対応: プロンプト翻訳 + 音声指示のローカライズ
  4. Extend機能活用: 30秒単位で生成し、シームレス連結で2分超動画作成
  5. コスト監視ダッシュボード: BigQueryでAPI使用量を日次集計

次のステップ