Omni Flash の会話編集は「アングルを30度右へ」をどこまで聞き取るか — 指示が効く境界を実測した記録

from dataclasses import dataclass, field
from pathlib import Path
 
OMNI_FLASH_MODEL = "gemini-omni-flash"   # 公開プレビュー
PRICE_PER_OUTPUT_SEC_USD = 0.10          # 出力1秒あたり0.10ドル
 
@dataclass
class EditResult:
    instruction: str
    out_path: Path
    out_seconds: float
    landed: bool = False
    confidence: float = 0.0
    note: str = ""
 
def omni_flash_edit(client, base_video: Path, instruction: str,
                    out_dir: Path) -> EditResult:
    """会話編集を1回かけて、出力動画のパスと尺を返す薄いラッパ。
    ※ 編集APIの正確な呼び出しは公開プレビューで変わり得るため関数内に隔離する。
    """
    resp = client.videos.edit(              # preview: 名称は変わり得ます
        model=OMNI_FLASH_MODEL,
        video=str(base_video),
        instruction=instruction,
    )
    out = out_dir / f"edit_{abs(hash(instruction)) % 10_000}.mp4"
    out.write_bytes(resp.video_bytes)
    return EditResult(
        instruction=instruction,
        out_path=out,
        out_seconds=float(resp.duration_seconds),
    )

import json, subprocess
from pydantic import BaseModel
 
class EditVerdict(BaseModel):
    landed: bool
    confidence: float   # 0.0〜1.0
    note: str
 
def grab_frame(video: Path, at_sec: float, out: Path) -> Path:
    # 代表フレームを1枚だけ取り出す（中央付近を既定にする）
    subprocess.run(
        ["ffmpeg", "-y", "-ss", str(at_sec), "-i", str(video),
         "-frames:v", "1", str(out)],
        check=True, capture_output=True,
    )
    return out
 
def verify_edit(client, before_png: Path, after_png: Path,
                instruction: str) -> EditVerdict:
    prompt = (
        "2枚の画像は、ある動画の編集前と編集後の代表フレームです。\n"
        f"要求した編集: 「{instruction}」\n"
        "この編集が後者に実際に反映されているかを判定してください。\n"
        "確信が持てない場合は landed=false 側に寄せてください。"
    )
    resp = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=[
            prompt,
            {"inline_data": {"mime_type": "image/png",
                             "data": before_png.read_bytes()}},
            {"inline_data": {"mime_type": "image/png",
                             "data": after_png.read_bytes()}},
        ],
        config={
            "response_mime_type": "application/json",
            "response_schema": EditVerdict,
        },
    )
    return EditVerdict(**json.loads(resp.text))

# Before: 目視で「良くなるまで」重ねる — コストが読めない
def edit_until_ok(client, base, instructions, out_dir):
    cur = base
    for ins in instructions:
        r = omni_flash_edit(client, cur, ins, out_dir)
        cur = r.out_path           # 効いたか確かめずに次へ積む
        # 人が動画を再生して目視で判断 … 外れも課金されている
    return cur

# After: 検証ゲートで「効いた編集だけ」採用し、費用を台帳に積む
@dataclass
class CostLedger:
    spent_usd: float = 0.0
    wasted_usd: float = 0.0            # 未反映に消えた分
    kept: list = field(default_factory=list)
 
    def charge(self, r: EditResult):
        cost = r.out_seconds * PRICE_PER_OUTPUT_SEC_USD
        self.spent_usd += cost
        if r.landed:
            self.kept.append(r)
        else:
            self.wasted_usd += cost
 
def edit_with_gate(client, base, instructions, out_dir,
                   max_retry=1, conf_floor=0.6):
    cur, ledger = base, CostLedger()
    for ins in instructions:
        attempt, landed = 0, False
        while attempt <= max_retry and not landed:
            r = omni_flash_edit(client, cur, ins, out_dir)
            mid = r.out_seconds / 2
            before = grab_frame(cur, mid, out_dir / "b.png")
            after = grab_frame(r.out_path, mid, out_dir / "a.png")
            v = verify_edit(client, before, after, ins)
            r.landed = v.landed and v.confidence >= conf_floor
            r.confidence, r.note = v.confidence, v.note
            ledger.charge(r)
            if r.landed:
                cur, landed = r.out_path, True
            attempt += 1
        if not landed:
            print(f"見送り: {ins}（最終確信度 {r.confidence:.2f}）")
    return cur, ledger