google-gemini-omni-video

Google Gemini Omni の動画

ファーストパーティの Google モデル gemini-omni-flash-preview には、このスキルを使用してください。以下の提供状況、価格、制限、プレビュー動作は 2026-07-10 時点で確認済みとして扱い、本番環境へのリリース前に、引用されているファーストパーティのページを再確認してください。

作業を始める前に範囲を明確にする

対象のバックエンド、データリージョン、ソースメディア、編集モード、アスペクト比、希望する長さ、音声／会話の有無、想定する視聴者、権利の状況、費用の承認について確認してください。Omni のリクエストを、別の製品向けのリクエストに黙って置き換えないでください。

サーフェス	対象となる仕様	混同しないもの
Gemini Developer API	POST https://generativelanguage.googleapis.com/v1beta/interactions；API／認証キー；Omni は有料ティアのみ	Gemini コンシューマーアプリ、AI Studio の UI 動作、Google Cloud のデータ管理
Gemini Enterprise Agent Platform	POST .../v1beta1/projects/{project}/locations/{location}/interactions；OAuth／IAM；Agent Platform API；正確なホストは確認済みのロケーション仕様によって異なります	Developer API のそのまま置き換え可能なエンドポイント、または従来の Veo 操作スキーマ
Gemini Omni Flash	ネイティブの短尺動画生成、参照、音声生成、対話形式の編集	Veo。Veo は、対応バージョンでの動画延長／補間など、エンドポイントや機能が異なる特化型動画ファミリーです
Gemini ネイティブ画像（“Nano Banana”）	個別の画像生成／編集モデル	Omni の動画出力
Gemini アプリ、Flow、YouTube	独自のアクセス、設定、利用規約を持つコンシューマー／クリエイター向け製品	API のデータ保持、スキーマ、提供状況、来歴メタデータに関する保証
サードパーティのゲートウェイ／再販業者	このスキルの対象外	Google ファーストパーティの課金、プライバシー、安全性、再試行の保証

事実： Gemini Omni Flash はプレビューモデルですが、Interactions API 自体は GA です。Developer API のモデルページでは、テキスト／画像／動画の入力、動画の出力、1,048,576 トークンのコンテキスト、3–10 秒の 720p／24 FPS 出力が指定されています。Cloud のモデルページでは、これとは異なる 131,072 トークンの入力上限と 57,920 トークンの出力上限が公開されています。各バックエンドについて、そのバックエンドの仕様に記載された制限のみ適用してください。

事実： Cloud のプレビューページでは、2027 年 6 月 30 日の提供終了日が公開されています。Developer API のページでは、同じライフサイクルは保証されていません。モデル ID を固定し、リリースノートを監視して、プロバイダーに紐づいた移行テストを維持してください。

バックエンドを意図的に選択する

API キーを使った、最短のファーストパーティ統合と会話状態の管理には、Developer API を使用します。新しい AI Studio キーはサービスアカウントに紐付けられた認証キーです。制限のない標準キーは 2026 年 6 月 19 日に使用できなくなり、すべての標準キーは 2026 年 9 月に拒否される予定です。キーはサーバー側に保持して制限を設定し、GEMINI_API_KEY または GOOGLE_API_KEY を使用します（両方が存在する場合は GOOGLE_API_KEY が優先されます）。

Cloud IAM、Cloud 契約、監査／セキュリティ制御、GCS 配信、または Cloud データガバナンスが必要な場合は、Gemini Enterprise Agent Platform を使用します。Application Default Credentials または有効期間の短い OAuth アクセストークンで認証します。専用の Omni 動画ガイドでは現在 location=global が示されていますが、このエンドポイントではリージョン単位の処理を制御できません。一方、新しい Cloud デプロイページには Omni が eu および us のマルチリージョンで利用可能と記載されており、これらのエンドポイントでは、選択した管轄区域内で ML 処理が行われると説明されています。これはファーストパーティのルートに関するドキュメントの矛盾として扱ってください。規制対象のワークロードでは、プロジェクトコンソール、SDK のスモークテスト、契約、またはサポートで該当する Omni ルートが確認できた場合に限り、aiplatform.eu.rep.googleapis.com または aiplatform.us.rep.googleapis.com の Interactions ルートを使用してください。確認できない場合は、global にフォールバックせず中止してください。

Cloud の利用可否を Developer の利用可否から推測したり、その逆を行ったりしないでください。検証日時点で、Cloud モデルのページ自体に矛盾があります。従量課金はサポート対象外、固定クォータはサポート対象と記載する一方で、global の Standard PayGo も掲載されています。プロジェクトコンソールまたは Google サポートで確認できるまで、Cloud の課金方式、クォータ割り当て、料金は 不明 として扱ってください。Developer の価格だけを根拠に、Cloud の作成リクエストを送信しないでください。

現在の機能仕様に合わせる

開発者向け API の仕様：

• 音声付きの 720p 動画を 1 本出力します。長さは 3〜10 秒、フレームレートは 24 FPS です。対応するアスペクト比は 16:9 と 9:16 です（デフォルトは横向き）。
• generation_config.video_config.task では、タスクに text_to_video、image_to_video、reference_to_video、edit を指定します。省略すると、モデルが推測します。
• previous_interaction_id を使った編集のたびに新しい動画が作成されます。親のインタラクションを保存しておく必要があります。
• REST を直接使う場合は、steps[].type == "model_output" から動画コンテンツを読み取ります。output_video は SDK の便宜的なフィールドです。
• 生成された動画には、目に見えない SynthID が含まれます。開発者向け API ガイドでは C2PA は保証されていません。Cloud のドキュメントでは、Omni の出力に対する自動 C2PA 署名が別途保証されています。C2PA に対応していない後処理を行うと、署名が無効になる場合があります。
• 英語で評価されています。他の言語は評価されていません。生成時間は変動します。
• 安全フィルターはプロンプトと出力に適用され、地域によって異なります。Interactions ではカスタム安全設定はサポートされていません。

現在の開発者向け API の非対応機能：

• 音声リファレンスのアップロード、音声の編集、動画の延長、最初と最後のフレーム間の補間、複数動画の推論、システム指示、temperature、top_p、停止シーケンス、専用の negative_prompts、プロビジョンド スループット、YouTube メディアソースには対応していません。
• 3 秒以下の動画リファレンスはスキーマを通過しますが、正しく処理されません。使用しないでください。
• EEA、スイス、英国では、アップロードした動画の編集は利用できません。ただし、モデルが生成した動画はこれらの地域でも編集できます。
• EEA、スイス、英国では、未成年者を含む画像のアップロードや編集はサポートされていません。認識可能な人物の一部は、さらに広い地域でサポート対象外です。

ドキュメントの矛盾： Google のモデルカードでは、音声を受け付けるネイティブなマルチモーダルモデルと説明されています。また、開発者向け料金表には音声入力の料金が記載されていますが、現在の開発者向け Omni ガイドでは、アップロードした音声リファレンスは明示的にサポート対象外とされています。API ガイドに従い、希望する音声、音楽、環境音、効果音はテキストで説明してください。Cloud のモデルページでは音声は入力専用とされていますが、送信前に Cloud の正確なリクエストガイドを確認してください。

ドキュメントの矛盾： 汎用 Interactions モデル表（2026-07-09 更新）には Omni が記載されていませんが、専用の Omni ガイドと変更履歴では Interactions が必須とされています。専用ガイドとモデルのエンドポイントを使用し、新しいプレビューであるためスモークテストを継続してください。

コスト、クォータ、承認の計画

Developer Omni に無料枠はありません。公開されている標準料金は、テキスト／画像／動画／音声の入力トークン 100 万トークンあたり $1.50、テキスト出力トークン 100 万トークンあたり $9、思考処理を含む動画出力トークン 100 万トークンあたり $17.50 です。Google は 720p 出力を 1 秒あたり 5,792 トークンとして課金しており、料金は約 $0.10／秒です。動画出力のみの正確な計算では $0.10136／秒です。10 秒の出力は、入力とテキスト出力を除き、$1.0136 です。

作成または編集のたびに、次の手順を実施してください。

1. 要求されたバリエーション数とターン数を数えます。編集のたびに出力が追加され、追加料金が発生する場合があります。
2. 保守的な見積もりを提示します。Developer API では turns × ($1.0136 maximum video output + estimated inputs/text output) を使います。これは見積もりであり、プロバイダーが定める上限ではないことを明記してください。
3. バックエンド、有料の作成回数、推定最大額、保持方法、ソースメディアを明示したうえで、明確な承認を得ます。
4. AI Studio でプロジェクトの現在の制限を確認します。制限はプロジェクトごとに設定され、ティアやモデルによって異なります。プレビューモデルではより厳しく、保証もされません。RPM 値をハードコードしないでください。
5. 429 RESOURCE_EXHAUSTED は、容量またはクォータに関する通知として扱います。再試行のヒントがあれば従い、ポーリングや読み取り操作ではバックオフしてください。結果が不明な作成リクエストを自動で再実行しないでください。

ソースと成果物の安全な取り扱い

• ソース、ブランド、パフォーマンスに関する権利と、肖像の使用に関する文書化された同意を必須とします。モデルによる受け入れは、権利処理の完了を意味しません。親密な画像を悪用する行為、なりすまし、生体情報やプライバシーの侵害、児童の搾取、監視、禁止されている用途は拒否します。広告、著名人、政治、センシティブな特性、重要な意思決定に関わる内容はエスカレーションします。
• ソースにハッシュを付与し、ソースと出力の台帳にMIMEタイプとバイト数を記録します。コンテンツを検証し、読み込みやデコードの前に上限を設定し、アクティブファイルやポリグロットファイルを拒否して、アップロードを非公開に保ちます。キー、base64、署名付きURI、秘密情報を含むプロンプト、思考は決してログに記録しないでください。
• Developer Files APIへのアップロードは、1ファイルあたり最大2 GB、プロジェクトあたり最大20 GBで、48時間保持されます。このAPIではダウンロードできず、手動で削除できます。使用後は早めに削除してください。
• Developerの有料Interactionsでは、デフォルトの保存期間は55日で、7／14／28／55日に設定できます。IDを指定して削除してください。store=falseにすると、親の状態とバックグラウンド処理が無効になります。別途、悪用に関するデータは55日間保持され、認可されたレビューが行われる場合があります。有料コンテンツは製品の改善には使用されませんが、悪用のログ記録は引き続き行われます。コンシューマー向け／無料の利用規約は適用されません。
• Cloudの非同期データは最大14日間保持されます。学習に使用しない利用規約にも、悪用、Advanced AI、ログ記録に関する例外があります。分離されたメモリーキャッシュは、無効にしない限りデフォルトで24時間保持されます。契約内容、例外、キャッシュ、GCSのIAM／リージョン／暗号化／ログ記録／ライフサイクルを確認してください。
• リリース前に、MP4の長さ、寸法、FPS、音声、全体のデコード、サイズ、ハッシュを検証し、フレーム、台詞、テキスト、ロゴ、顔、連続性、安全性、事実やブランドに関する主張を確認してください。

テスト可能なプロンプトを作成する

被写体、動作、環境、ショット、レンズ／カメラの動き、照明、色調、テンポ、物理的な動き、音声を指定します。ワンショットの場合は、single continuous unbroken shot; no scene cutsと指定します。除外事項は通常のテキストで記述します。例：No dialogue. No extra sound effects.

自然なタイミングまたはタイムコードを使用します。

Single continuous unbroken 9:16 shot; no scene cuts.
[0-3s] A paper kite rises above a quiet dawn beach; slow handheld push-in.
[3-6s] The kite crosses the sun and the camera tilts upward.
Audio: soft surf and light wind. No dialogue, no music.
Keep all lettering and logos out of frame.

タイムコードはプロンプトによる指示であり、フレーム単位で正確に編集される保証ではありません。結果を確認してください。

画像の役割を指定する場合は、メディアの順序を安定させ、明示的に割り当てます。

• <FIRST_FRAME>は、開始フレームとして画像を使用します。
• <IMAGE_REF_N>は参照画像を指定します。番号は0から始まります。
• 曖昧さを避けるには、[# Sources <FIRST_FRAME>@Image1]や[# References <IMAGE_REF_0>@Image1]など、Googleが指定する明示的なソース／参照プレフィックスを使用します。

Googleは複数の参照画像に対応していますが、Developerガイドには数値での上限が記載されていません。上限を推測で設定しないでください。Cloudモデルのページには、1つのプロンプトにつき最大10枚の画像と個別に記載されています。

編集では、1回につき1つだけ変更します。例：Make the room lighting warmer. Keep everything else the same.編集を重ねると、アイデンティティ、レイアウト、動き、テキスト、音声の一貫性が損なわれる場合があります。モデルカードでは、編集をまたいだ一貫性、複雑な動き、完璧なテキスト表現を既知の制約として挙げています。維持されると断言せず、確認してください。

安全な Developer API ワークフローを使用する

4 MB を超える出力には、URI での配信を優先してください。最初のやり取り ID と URI をすぐに保存してください。後続の GET /interactions/{id} は、作成リクエストで URI を指定していても、現在はインラインの base64 を返すため、URI が保証されるのは最初のレスポンスまたは SSE ストリームのみです。対応する File リソースを ACTIVE になるまでポーリングし、FAILED または期限超過の場合は失敗として扱います。その後、認証を使用して一時ファイルにダウンロードし、検証してからアトミックに名前を変更してください。

Omni の作成に関する冪等性キーは文書化されていません。POST の前に、クライアント試行 UUID、リクエストハッシュ、承認情報、posting 状態を含む試行レコードを永続化してください。送信後に接続が切断され、やり取り ID を取得できない場合は、create_outcome_unknown としてマークし、処理を停止して、オペレーターとプロジェクトログおよび課金を照合してください。「念のため再試行」は決して行わないでください。GET、ファイル状態のポーリング、ダウンロードは読み取り専用であり、上限を設けたジッター付き再試行を使用できます。

以下の完全な Python の例では、テキスト、状態を保持する会話、別途ステージングするアップロード動画の編集を扱います。デフォルトではオフラインのドライランを行い、有料 API 呼び出しは行いません。実行する場合に限り、google-genai>=2.3.0 をインストールしてください。

#!/usr/bin/env python3
import argparse, base64, hashlib, http.client, ipaddress, json, os, random; import socket, ssl, subprocess, tempfile, threading, time, uuid; from decimal import Decimal, InvalidOperation; from fractions import Fraction
from pathlib import Path; from urllib.parse import parse_qs, urlsplit; MODEL = "gemini-omni-flash-preview"; ESTIMATED_MAX_USD = 1.10  # one 10s Developer turn; estimate, not a provider cap
MAX_VIDEO_BYTES = 80 * 1024 * 1024; MAX_TOOL_BYTES = 1024 * 1024; DOWNLOAD_HOST = "generativelanguage.googleapis.com"
def canonical(value) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
def atomic_json(path: Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True); fd, tmp = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path)
    finally:
        if os.path.exists(tmp): os.unlink(tmp)
def claim_once(path: Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True); fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
def read_record(path: Path) -> dict:
    if path.stat().st_size > MAX_TOOL_BYTES: raise RuntimeError("Attempt record size rejected")
    with path.open("rb") as source: raw = source.read(MAX_TOOL_BYTES + 1)
    if not raw or len(raw) > MAX_TOOL_BYTES: raise RuntimeError("Attempt record size rejected")
    value = json.loads(raw)
    if not isinstance(value, dict): raise RuntimeError("Attempt record must be an object")
    return value
def run_bounded(argv, timeout, cap=MAX_TOOL_BYTES):
    proc = subprocess.Popen(argv, shell=False, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE); out, err, lock, overflow = bytearray(), bytearray(), threading.Lock(), threading.Event()
    total = 0
    def drain(stream, bucket):
        nonlocal total
        while True:
            chunk = stream.read1(65536) if hasattr(stream, "read1") else stream.read(65536)
            if not chunk: break
            with lock:
                room = max(0, cap - total); bucket.extend(chunk[:room]); total += len(chunk)
                if total > cap: overflow.set()
            if overflow.is_set():
                try: proc.kill()
                except OSError: pass
    threads = [threading.Thread(target=drain, args=(proc.stdout, out), daemon=True), threading.Thread(target=drain, args=(proc.stderr, err), daemon=True)]
    for thread in threads: thread.start()
    timed_out = False
    try: code = proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True; proc.kill(); code = proc.wait()
    for thread in threads: thread.join()
    return {"code": code, "stdout": bytes(out), "stderr": bytes(err), "timeout": timed_out, "overflow": overflow.is_set()}
class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host, ip, timeout=30):
        super().__init__(host, 443, timeout=timeout, context=ssl.create_default_context()); self.ip = ip
    def connect(self):
        sock = socket.create_connection((self.ip, self.port), self.timeout); self.sock = self._context.wrap_socket(sock, server_hostname=self.host)
def generated_file_name(uri: str) -> str:
    parsed = urlsplit(uri)
    if (parsed.scheme != "https" or parsed.hostname != DOWNLOAD_HOST or parsed.username or parsed.password or parsed.fragment or parsed.port not in (None, 443)):
        raise RuntimeError("Generated file URI policy rejected")
    prefix = "/v1beta/files/"
    if not parsed.path.startswith(prefix) or not parsed.path.endswith(":download"):
        raise RuntimeError("Generated file URI path rejected")
    name = parsed.path[len(prefix):-len(":download")]
    if not name or not name.isalnum() or parse_qs(parsed.query) != {"alt": ["media"]}:
        raise RuntimeError("Generated file URI identifier/query rejected")
    return "files/" + name
def download_generated(uri: str, api_key: str, parent: Path) -> Path:
    name = generated_file_name(uri).split("/", 1)[1]; addresses = sorted({item[4][0] for item in socket.getaddrinfo(DOWNLOAD_HOST, 443, type=socket.SOCK_STREAM)})
    if not addresses or any(not ipaddress.ip_address(item).is_global for item in addresses):
        raise RuntimeError("Generated file host did not resolve publicly")
    conn, temp_path = PinnedHTTPSConnection(DOWNLOAD_HOST, addresses[0], timeout=60), None
    try:
        conn.request("GET", f"/v1beta/files/{name}:download?alt=media", headers={"x-goog-api-key": api_key, "Accept": "video/mp4"}); response = conn.getresponse()
        if 300 <= response.status < 400: raise RuntimeError("Generated file redirects are rejected")
        if response.status != 200: raise RuntimeError(f"Generated file HTTP {response.status}")
        mime = response.getheader("Content-Type", "").split(";", 1)[0].lower()
        if mime != "video/mp4": raise RuntimeError("Generated file MIME rejected")
        length = response.getheader("Content-Length")
        if length is not None and (not length.isdigit() or int(length) > MAX_VIDEO_BYTES):
            raise RuntimeError("Generated file declared size rejected")
        fd, name_on_disk = tempfile.mkstemp(prefix=".omni-", suffix=".mp4", dir=parent); temp_path = Path(name_on_disk); total = 0
        with os.fdopen(fd, "wb") as f:
            while True:
                chunk = response.read(1024 * 1024)
                if not chunk: break
                total += len(chunk)
                if total > MAX_VIDEO_BYTES: raise RuntimeError("Generated file crossed byte cap")
                f.write(chunk)
            f.flush(); os.fsync(f.fileno())
        return temp_path
    except BaseException:
        if temp_path is not None and temp_path.exists(): temp_path.unlink()
        raise
    finally:
        conn.close()
def inline_generated(data: str, parent: Path) -> Path:
    if len(data) > ((MAX_VIDEO_BYTES + 2) // 3) * 4 + 4:
        raise RuntimeError("Inline video encoding exceeds byte cap")
    blob = base64.b64decode(data, validate=True)
    if not blob or len(blob) > MAX_VIDEO_BYTES: raise RuntimeError("Inline video size rejected")
    fd, name = tempfile.mkstemp(prefix=".omni-", suffix=".mp4", dir=parent)
    with os.fdopen(fd, "wb") as f: f.write(blob); f.flush(); os.fsync(f.fileno())
    return Path(name)
def validate_and_publish(temp_path: Path, destination: Path) -> dict:
    try:
        with temp_path.open("rb") as f: header = f.read(12)
        if len(header) < 12 or header[4:8] != b"ftyp": raise RuntimeError("MP4 signature rejected")
        probe = run_bounded(["ffprobe", "-v", "error", "-show_entries", "format=duration,size:stream=index,codec_type,codec_name,width,height,r_frame_rate", "-of", "json", str(temp_path)], 30)
        if probe["timeout"] or probe["overflow"] or probe["code"] != 0: raise RuntimeError("ffprobe failed")
        metadata = json.loads(probe["stdout"]); streams = metadata.get("streams", []); videos = [item for item in streams if item.get("codec_type") == "video"]
        if len(videos) != 1 or not any(item.get("codec_type") == "audio" for item in streams):
            raise RuntimeError("Expected one video stream and generated audio")
        width, height = videos[0].get("width"), videos[0].get("height")
        if sorted((width, height)) != [720, 1280]: raise RuntimeError("Expected 720p dimensions")
        if abs(float(Fraction(videos[0].get("r_frame_rate", "0/1"))) - 24.0) > 0.1:
            raise RuntimeError("Expected 24 FPS")
        duration = float(metadata.get("format", {}).get("duration", 0))
        if not 2.9 <= duration <= 10.1: raise RuntimeError("Video duration outside 3-10 seconds")
        decode = run_bounded(["ffmpeg", "-v", "error", "-i", str(temp_path), "-f", "null", "-"], 300)
        if decode["timeout"] or decode["overflow"] or decode["code"] != 0: raise RuntimeError("Full decode failed")
        if destination.exists(): raise RuntimeError("Output already exists")
        digest = hashlib.sha256()
        with temp_path.open("rb") as f:
            for chunk in iter(lambda: f.read(1024 * 1024), b""): digest.update(chunk)
        artifact = {"path": str(destination), "bytes": temp_path.stat().st_size, "sha256": digest.hexdigest(), "probe": metadata}; os.replace(temp_path, destination); temp_path = None
        return artifact
    finally:
        if temp_path is not None and temp_path.exists(): temp_path.unlink()
def retry_after(exc) -> float | None:
    response = getattr(exc, "response", None); headers = getattr(response, "headers", {}) or {}; raw = headers.get("retry-after") or headers.get("Retry-After")
    try: value = float(raw)
    except (TypeError, ValueError): return None
    return value if 0 <= value <= 60 else None
def safe_read(call, deadline, attempts=5):
    for attempt in range(attempts):
        try: return call()
        except Exception as exc:
            if attempt + 1 == attempts or time.monotonic() >= deadline: raise
            delay = retry_after(exc) or min(20.0, 2 ** attempt) * (1 + random.random() * .25)
            if time.monotonic() + delay >= deadline: raise
            time.sleep(delay)
def find_video(interaction):
    convenience = getattr(interaction, "output_video", None)
    if convenience: return convenience
    for step in getattr(interaction, "steps", []) or []:
        if getattr(step, "type", None) == "model_output":
            for item in getattr(step, "content", []) or []:
                if getattr(item, "type", None) == "video": return item
    raise RuntimeError("No video output in completed interaction")
def delete_remote_files(client, items) -> list[dict]:
    results = []
    for role, name in items:
        try:
            client.files.delete(name=name)  # one cleanup attempt; ambiguity is recorded for an operator; results.append({"role": role, "name": name, "status": "deleted"})
        except Exception as exc:
            results.append({"role": role, "name": name, "status": "delete_outcome_unknown", "errorType": type(exc).__name__})
    return results
def load_source_manifest(path: Path, jurisdiction: str | None) -> dict:
    if jurisdiction in {"eea", "ch", "uk"}:
        raise RuntimeError("Uploaded-video editing is unavailable in this jurisdiction")
    if jurisdiction != "other":
        raise RuntimeError("--source-jurisdiction other is required for an uploaded-video edit")
    if path.is_symlink() or not path.is_file(): raise RuntimeError("Source manifest path rejected")
    if path.stat().st_size > MAX_TOOL_BYTES: raise RuntimeError("Source manifest size rejected")
    with path.open("rb") as source_file: raw = source_file.read(MAX_TOOL_BYTES + 1)
    if not raw or len(raw) > MAX_TOOL_BYTES: raise RuntimeError("Source manifest size rejected")
    value = json.loads(raw)
    if not isinstance(value, dict): raise RuntimeError("Source manifest must be an object")
    required = {"name", "uri", "sha256", "bytes", "mime", "state", "deleteAfterUse", "durationSeconds", "localFullDecodePassed", "rightsApproved"}
    if set(value) != required: raise RuntimeError("Source manifest schema rejected")
    name = value["name"]
    if not isinstance(name, str) or not name.startswith("files/") or not name[6:].isalnum():
        raise RuntimeError("Source File resource name rejected")
    parsed = urlsplit(value["uri"] if isinstance(value["uri"], str) else "")
    if (parsed.scheme != "https" or parsed.hostname != DOWNLOAD_HOST or parsed.username or parsed.password or parsed.port not in (None, 443) or parsed.query or parsed.fragment or parsed.path != "/v1beta/" + name):
        raise RuntimeError("Source File URI rejected")
    sha = value["sha256"]
    if not isinstance(sha, str) or len(sha) != 64 or any(c not in "0123456789abcdef" for c in sha):
        raise RuntimeError("Source SHA-256 rejected")
    if (type(value["bytes"]) is not int or not 0 < value["bytes"] <= 2 * 1024 * 1024 * 1024 or value["mime"] != "video/mp4" or value["state"] != "ACTIVE" or value["deleteAfterUse"] is not True or value["localFullDecodePassed"] is not True or value["rightsApproved"] is not True):
        raise RuntimeError("Source validation/lifecycle assertions rejected")
    try: duration = Decimal(str(value["durationSeconds"]))
    except InvalidOperation as exc: raise RuntimeError("Source duration rejected") from exc
    if not duration.is_finite() or not Decimal("0") < duration <= Decimal("10"):
        raise RuntimeError("Uploaded source must be no longer than 10 seconds")
    return value
def main() -> int:
    p = argparse.ArgumentParser(); p.add_argument("--prompt", required=True); p.add_argument("--aspect", choices=("16:9", "9:16"), default="16:9")
    p.add_argument("--stateful", action="store_true", help="store this turn so it can be edited later"); p.add_argument("--previous-id", help="stored parent interaction for a paid edit turn")
    p.add_argument("--source-manifest", type=Path, help="bounded manifest for an already ACTIVE, locally validated Developer File upload"); p.add_argument("--source-jurisdiction", choices=("other", "eea", "ch", "uk"))
    p.add_argument("--output", type=Path, default=Path("omni-output.mp4")); p.add_argument("--attempt-log", type=Path, default=Path("omni-attempt.json"))
    p.add_argument("--attempt-id", required=True, help="fresh canonical UUIDv4 used for this one paid turn"); p.add_argument("--execute", action="store_true")
    p.add_argument("--resume", action="store_true", help="read/query a claimed attempt; never create"); p.add_argument("--max-usd", default="1.10"); p.add_argument("--approval-sha256")
    p.add_argument("--show-full-plan", action="store_true"); a = p.parse_args()
    if not a.prompt.strip() or len(a.prompt.encode("utf-8")) > 8000:
        raise SystemExit("Prompt must be 1..8000 UTF-8 bytes")
    if a.output.suffix.lower() != ".mp4": raise SystemExit("Output must end in .mp4")
    try: attempt = uuid.UUID(a.attempt_id); maximum = Decimal(a.max_usd)
    except (ValueError, InvalidOperation) as exc: raise SystemExit("Attempt ID or maximum USD is invalid") from exc
    if attempt.version != 4 or str(attempt) != a.attempt_id.lower(): raise SystemExit("Attempt ID must be canonical UUIDv4")
    if not maximum.is_finite() or maximum <= 0 or maximum < Decimal("1.10"):
        raise SystemExit("Maximum USD must be finite, positive, and cover 1.10")
    output, attempt_log = a.output.resolve(), a.attempt_log.resolve()
    if output == attempt_log: raise SystemExit("Output and attempt record must differ")
    if a.source_jurisdiction and not a.source_manifest:
        raise SystemExit("--source-jurisdiction requires --source-manifest")
    if a.source_manifest and a.previous_id:
        raise SystemExit("Choose either an uploaded-video edit or a previous-interaction edit")
    source = load_source_manifest(a.source_manifest.resolve(), a.source_jurisdiction) if a.source_manifest else None
    request = { "model": MODEL, "input": a.prompt, "response_format": {"type": "video", "delivery": "uri", "aspect_ratio": a.aspect}, "background": False, "stream": False, "store": a.stateful or bool(a.previous_id), }
    if source:
        request["input"] = [{"type": "document", "uri": source["uri"]}, {"type": "text", "text": a.prompt}]; request["generation_config"] = {"video_config": {"task": "edit"}}
    if a.previous_id: request["previous_interaction_id"] = a.previous_id
    request_hash = hashlib.sha256(canonical(request)).hexdigest()
    envelope = {"backend": "gemini-developer-api", "endpoint": "/v1beta/interactions", "model": MODEL, "request": request, "sourceHashes": ([] if not source else [{"sha256": source["sha256"], "bytes": source["bytes"], "mime": source["mime"], "fileName": source["name"], "jurisdiction": a.source_jurisdiction, "deleteAfterUse": True}]), "attemptId": str(attempt), "attemptRecord": str(attempt_log), "outputPolicy": {"path": str(output), "mime": "video/mp4", "maxBytes": MAX_VIDEO_BYTES, "ffprobe": True, "fullDecode": True, "requireAudio": True}, "cost": {"estimatedMaxUsd": "1.10", "approvedMaxUsd": format(maximum, "f"), "priceAsOf": "2026-07-10", "paidCreates": 1}, "retention": {"storeInteraction": request["store"], "previousInteractionId": a.previous_id}}
    approval = hashlib.sha256(canonical(envelope)).hexdigest(); redacted_input = f"sha256:{hashlib.sha256(a.prompt.encode()).hexdigest()}"
    if source:
        redacted_input = [{"type": "document", "uriSha256": hashlib.sha256(source["uri"].encode()).hexdigest()}, {"type": "text", "sha256": hashlib.sha256(a.prompt.encode()).hexdigest()}]
    safe_envelope = {**envelope, "request": {**request, "input": redacted_input}}; print(json.dumps({"dry_run": not a.execute, "approval_sha256": approval, "plan": safe_envelope}, indent=2))
    if a.show_full_plan: print(json.dumps({"protected_full_envelope": envelope}, indent=2))
    if not a.execute: return 0
    if a.approval_sha256 != approval: raise SystemExit("Exact approval hash mismatch")
    api_key = os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY")
    if not api_key:
        raise SystemExit("Set GEMINI_API_KEY or GOOGLE_API_KEY server-side")
    from google import genai; from google.genai import types; create_client = genai.Client(api_key=api_key, http_options=types.HttpOptions( timeout=600000, retry_options=types.HttpRetryOptions(attempts=1)))
    read_client = genai.Client(api_key=api_key, http_options=types.HttpOptions( timeout=30000, retry_options=types.HttpRetryOptions(attempts=1)))
    if source:
        info = safe_read(lambda: read_client.files.get(name=source["name"]), time.monotonic() + 120); state = getattr(getattr(info, "state", None), "name", getattr(info, "state", None))
        if state != "ACTIVE" or getattr(info, "uri", None) != source["uri"]:
            raise RuntimeError("Approved source upload is not the same ACTIVE File resource")
    if a.resume:
        if not attempt_log.exists(): raise SystemExit("Resume record does not exist")
        record = read_record(attempt_log)
        if (record.get("attempt_id") != str(attempt) or record.get("approval_sha256") != approval or record.get("request_sha256") != request_hash):
            raise SystemExit("Resume record does not match this exact approval")
        interaction_id = record.get("interaction_id")
        if not isinstance(interaction_id, str) or not interaction_id:
            raise SystemExit("Create outcome is unknown without an interaction ID; do not replay")
        interaction = safe_read(lambda: read_client.interactions.get(interaction_id), time.monotonic() + 120)
    else:
        record = {"attempt_id": str(attempt), "backend": "gemini-developer-api", "model": MODEL, "approval_sha256": approval, "request_sha256": request_hash, "status": "posting", "approved_max_usd": format(maximum, "f"), "interaction_id": None, "source_file_name": source["name"] if source else None, "output_uri_sha256": None, "created_unix": int(time.time())}
        try: claim_once(attempt_log, record)
        except FileExistsError as exc: raise SystemExit("Attempt is already claimed; use --resume, never replay") from exc
        try:
            interaction = create_client.interactions.create(**request)  # exactly one network attempt
        except Exception as exc:
            record.update(status="create_outcome_unknown", error_type=type(exc).__name__, updated_unix=int(time.time())); atomic_json(attempt_log, record)
            raise RuntimeError("Create outcome unknown; reconcile logs/billing, do not retry") from exc
        interaction_id = getattr(interaction, "id", None)
        if not isinstance(interaction_id, str) or not interaction_id:
            record.update(status="create_outcome_unknown", error_type="MissingInteractionId"); atomic_json(attempt_log, record)
            raise RuntimeError("Create returned no interaction ID; do not retry")
        record.update(interaction_id=interaction_id, status="interaction_id_saved", updated_unix=int(time.time())); atomic_json(attempt_log, record)  # first action after create: preserve provider identity
    video = find_video(interaction); uri = getattr(video, "uri", None); data = getattr(video, "data", None)
    if uri:
        file_name = generated_file_name(uri); record.update(status="output_uri_saved", output_uri_sha256=hashlib.sha256(uri.encode()).hexdigest(), generated_file_name=file_name, updated_unix=int(time.time()))
        atomic_json(attempt_log, record)
    else: file_name = None
    if uri:
        deadline = time.monotonic() + 900
        while True:
            info = safe_read(lambda: read_client.files.get(name=file_name), deadline); state = getattr(getattr(info, "state", None), "name", getattr(info, "state", None))
            if state == "ACTIVE": break
            if state == "FAILED": raise RuntimeError("Provider file processing failed")
            if state not in {"PROCESSING", "PENDING", None}: raise RuntimeError(f"Unknown file state {state!r}")
            if time.monotonic() >= deadline: raise TimeoutError("File processing exceeded 15 minutes")
            time.sleep(5 + random.random())
        output.parent.mkdir(parents=True, exist_ok=True); temp_path = download_generated(uri, api_key, output.parent)
    elif data:
        output.parent.mkdir(parents=True, exist_ok=True); temp_path = inline_generated(data, output.parent)
    else:
        raise RuntimeError("Video has neither URI nor data")
    try:
        artifact = validate_and_publish(temp_path, output)
    except BaseException as exc:
        cleanup_items = ([] if not file_name else [("generated_output", file_name)])
        if source: cleanup_items.append(("uploaded_source", source["name"]))
        cleanup_results = delete_remote_files(read_client, cleanup_items); cleanup_unknown = any(item["status"] != "deleted" for item in cleanup_results)
        record.update(status="artifact_validation_failed_cleanup_pending" if cleanup_unknown else "artifact_validation_failed", error_type=type(exc).__name__, remote_file_cleanup=cleanup_results, cleanup_action=("Inspect/delete the recorded File resources; do not replay create" if cleanup_unknown else None), updated_unix=int(time.time()))
        atomic_json(attempt_log, record); raise
    cleanup_items = ([] if not file_name else [("generated_output", file_name)])
    if source: cleanup_items.append(("uploaded_source", source["name"]))
    record.update(status="artifact_saved_cleanup_pending" if cleanup_items else "artifact_saved", artifact=artifact, retention=envelope["retention"], rights_review="required", creative_qa="pending", synthetic_disclosure="required", synthid_expected=True, developer_c2pa="not_documented", updated_unix=int(time.time()))
    atomic_json(attempt_log, record)
    if cleanup_items:
        cleanup_results = delete_remote_files(read_client, cleanup_items); cleanup_unknown = any(item["status"] != "deleted" for item in cleanup_results)
        record.update(status="artifact_saved_cleanup_pending" if cleanup_unknown else "artifact_saved", remote_file_cleanup=cleanup_results, cleanup_action=("Inspect/delete the recorded File resources; do not replay create" if cleanup_unknown else None), updated_unix=int(time.time()))
        atomic_json(attempt_log, record)
        if cleanup_unknown:
            raise RuntimeError("Artifact saved, but remote File cleanup needs operator reconciliation")
    print(json.dumps(record, indent=2)); return 0
if __name__ == "__main__": raise SystemExit(main())

まずドライランを行います。

python omni_generate.py --attempt-id "7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4" --prompt "Single continuous shot of rain on a greenhouse roof. Audio: rain only. No dialogue."

承認後、有料のターンを 1 回だけ実行します。

python omni_generate.py --execute --approval-sha256 "<exact unchanged digest>" --max-usd 1.10 --attempt-id "7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4" --prompt "Single continuous shot of rain on a greenhouse roof. Audio: rain only. No dialogue."

編集する予定の最初のターンでは、ドライランの前に --stateful を追加し、実行時にもその設定を変更しないでください。デフォルトの単独結果はステートレスであり、後から親にはできません。承認時に保持期間の選択を記録してください。既知のやり取りを再開するには、同じ引数に --resume --execute を付けてください。再開処理が作成ブランチに入ることはありません。

画像入力の場合は、input を順序付きのアイテムに置き換え、タスクを明示的に設定してください。

request["input"] = [
    {"type": "image", "data": base64.b64encode(image_bytes).decode(), "mime_type": "image/png"},
    {"type": "text", "text": "<FIRST_FRAME> single unbroken shot; the paper boat drifts forward. Keep its design unchanged. Audio: creek only."},
]
request["generation_config"] = {"video_config": {"task": "image_to_video"}}

画像を差し替える前に、バイト数の上限を設定し、内容を検証してハッシュを計算し、使用権を承認してください。アップロード動画を編集する場合は、ステージングを別途承認してください。シンボリックリンクとポリグロットを拒否し、サイズを 2 GB 以下に制限し、MP4 かつ 10 秒以下であることを必須とし、上限付きプローブ／全体デコード／ハッシュ計算を行ってから、1 回だけアップロードしてください。その後、PROCESSING → ACTIVE|FAILED をポーリングします。Google の正確な値を記録してください。

{"name":"files/abc123","uri":"https://generativelanguage.googleapis.com/v1beta/files/abc123","sha256":"<64 lowercase hex>","bytes":1234567,"mime":"video/mp4","state":"ACTIVE","deleteAfterUse":true,"durationSeconds":8.0,"localFullDecodePassed":true,"rightsApproved":true}

--source-manifest source.json --source-jurisdiction other は変更せずにドライランと実行の両方に渡してください。保護された --show-full-plan を使用して、正確なプロンプト／URI を確認してください。実行時には ACTIVE の URI を再確認し、ドキュメント入力を使用して、検証後にソースと生成された Files を削除します。作成またはクリーンアップの結果が不明な場合は、照合のために記録し、決して再実行しないでください。eea、ch、uk では処理を停止してください。壊れている 3 秒以下の参照パスは決して使用しないでください。

Cloud の契約は個別に扱う

Cloud の文書化された同期リクエストでは、3s から 10s までの明示的な duration 文字列、GCS URI による配信、リスト形式の response_format がサポートされています。Developer のコードにこれらのフィールドをコピーする前に、必ずそのリファレンスを確認してください。

正確なプロジェクトのルート、クォータモード、価格を確認した後に限り、このフェイルクローズのトランザクション契約を使用してください。それまではドライランを準備し、汎用的な本番リクエストを提示しないでください。

1. 正規形式の UUIDv4 試行 ID、正確なプロジェクト／ロケーション／オリジン、非公開の GCS プレフィックス、完全なリクエスト JSON、出力パス／ポリシー、現在の価格を示す証拠、有限かつ正の見積もり、見積もりをカバーする有限の上限を必須とします。これらのフィールドを正規化し、SHA-256 を出力してください。実行には、ブール値の承認ではなく、変更されていないダイジェストが必要です。
2. デフォルトはオフラインのドライランです。ADC の検索、ディレクトリの作成、GCS 呼び出し、ネットワーク通信は行いません。実行する場合は、OAuth／ADC を使用し、モード 0700 の新しい試行ディレクトリをアトミックに作成します。ディレクトリがすでに存在する場合は失敗させます。別途選択した再開ブランチでは、永続化済みの既知のインタラクション ID に対する GET のみを許可します。
3. 自動再試行なしで、上限を設けた POST を 1 回送信します。プロンプト／メディアの詳細がプロセス一覧に露出しないよう、保護対象のリクエスト本文をモード 0600 の試行ローカル一時ファイルに格納します。返された ID は直ちに永続化します。タイムアウトまたは解析失敗により永続化可能な ID を取得できなかった場合は create_outcome_unknown とし、再送せずに照合します。
4. GET の再試行回数に上限を設け、Retry-After を尊重します。保護されたインタラクション ID、リクエスト／承認ハッシュ、正確なクリーンアップ URI、レスポンスハッシュ、サニタイズ済みの状態のみを永続化します。生のレスポンスやモデルの思考は決して保存しないでください。承認済みの GCS プレフィックス配下に返される動画 URI が正確に 1 つであることを必須とします。
5. メタデータ、ダウンロード、ツールの時間／出力に上限を設けます。video/mp4、宣言済みおよび実測のバイト数上限、MP4 の ftyp マジック、720p／約 24 FPS の動画ストリームが正確に 1 つであること、生成音声、3〜10 秒の長さ、上限を設けた ffprobe、上限内の完全な ffmpeg デコード、最新の c2patool による検証成功を必須とします。既存の保存先は拒否し、アトミックな名前変更とハッシュ計算によって公開します。公開変換を行った後は C2PA を再検証します。
6. ローカルの成果物と保護されたマニフェストが永続化された後、上限を設けた呼び出しで GCS オブジェクトを削除します。削除の結果が不明確な場合は、正確な URI を保持したまま artifact_saved_cleanup_pending とします。クリーンアップを解決するために再作成してはなりません。

background:true の場合も、同じ作成／識別ルールと、上限を設けた読み取り専用のポーリングを適用します。Cloud の文書では、非同期保持期間は最大 14 日とされています。適用される Cloud 契約、乱用監視の例外、任意のリクエスト／レスポンスロギング、24 時間のインメモリキャッシュ設定、バケットの IAM／データ所在地／保持期間、インタラクション／オブジェクトの削除は、それぞれ別個の承認項目です。

フェーズごとに障害に対処する

• 事前確認／スキーマ／認証／リージョン： 設定を修正します。サービスが受け付ける前に明確に拒否した場合は、作成は行われていません。
• 作成時に ID が返された場合： ID を永続化し、そのインタラクションをポーリング／読み取りします。生成に時間がかかるという理由だけで、代替を作成してはなりません。
• ID を伴わない作成時の接続障害： 結果は不明です。停止して照合します。文書化された冪等性キーはありません。
• 安全性ブロック： 制御を弱めたり回避したりしないでください。リクエストが正当かつ準拠している場合に限り修正し、肖像の類似性／権利に関する判断が不明確なケースはエスカレーションします。
• ファイルが PROCESSING の場合： ジッターを加え、上限を設けてポーリングします。FAILED は終端状態です。GET interaction で URI が返ると決めつけても、失われた URI は復旧できません。
• 成果物の検証に失敗した場合： 出力を隔離してメタデータを保持し、公開しないでください。新たに料金が発生するターンには、新しい承認が必要です。
• プレビューとの差異、または 404／未対応フィールド： モデルのページ、Omni ガイド、変更履歴、バックエンドのリファレンスを再確認し、代替を推測しないでください。

根拠と判断を分ける

計画やレポートでは、次のラベルを使用してください。

• FACT: 現在の API の動作、モデルの制限、価格、ポリシー、または規約として直接文書化されている内容です。
• PROVIDER CLAIM: Google による定性的な説明、または社内で報告されたベンチマーク／モデルカードの結果です。独立した証拠として提示しないでください。
• OBSERVATION: 日付が明記されたローカルでのドライランまたは有料テストの結果です。バックエンド、モデル ID、SDK バージョン、リージョン、リクエストハッシュを含めてください。
• HEURISTIC: 1 回のやり取りにつき編集は 1 つにする、または「それ以外はすべて同じにする」といったプロンプトに関するアドバイスです。
• UNKNOWN: Developer API での参照数の上限、厳格なレイテンシ SLA、作成の冪等性、Developer C2PA、Cloud の消費モードの競合、または未検証の SDK の動作など、文書化されていない内容です。

まず次の一次情報を再確認してください

• Developer ガイドとプロンプト構文：https://ai.google.dev/gemini-api/docs/omni
• Developer のモデル詳細：https://ai.google.dev/gemini-api/docs/models/gemini-omni-flash
• Interactions の状態／保持／SDK の最低要件：https://ai.google.dev/gemini-api/docs/interactions-overview
• 料金と最新の上限：https://ai.google.dev/gemini-api/docs/pricing and https://ai.google.dev/gemini-api/docs/rate-limits
• Files API：https://ai.google.dev/gemini-api/docs/files
• API キー：https://ai.google.dev/gemini-api/docs/api-key
• リリースノート：https://ai.google.dev/gemini-api/docs/changelog
• 利用規約、リージョン、悪用の監視：https://ai.google.dev/gemini-api/terms, https://ai.google.dev/gemini-api/docs/available-regions, and https://ai.google.dev/gemini-api/docs/usage-policies
• 禁止されている用途：https://policies.google.com/terms/generative-ai/use-policy
• DeepMind の製品情報とモデルカード：https://deepmind.google/models/gemini-omni/ and https://deepmind.google/models/model-cards/gemini-omni-flash/
• Cloud のモデル／バックエンド：https://docs.cloud.google.com/gemini-enterprise-agent-platform/models/gemini/omni-flash-preview and https://docs.cloud.google.com/gemini-enterprise-agent-platform/models/video/generate-videos-from-text
• Cloud C2PA とエンドポイントのデータ所在地：https://docs.cloud.google.com/gemini-enterprise-agent-platform/models/content-credentials and https://docs.cloud.google.com/gemini-enterprise-agent-platform/resources/locations