Gemini Computer Use が「古い画面」を操作して静かに事故るとき — 観測と検証で守る運用メモ

import time
from google import genai
from google.genai import types
 
client = genai.Client()
MODEL = "gemini-2.5-computer-use-preview"  # 実際のモデル名は最新のドキュメントで確認してください
 
def capture():
    """環境依存。スクリーンショットの bytes と論理サイズを返す。"""
    png, width, height = grab_screen()  # 各自の実装
    return png, width, height
 
def run_step(prev_png, goal, history):
    """1ステップ分の推論。モデルに直前の画面と目標を渡す。"""
    resp = client.models.generate_content(
        model=MODEL,
        contents=[
            types.Part.from_bytes(data=prev_png, mime_type="image/png"),
            types.Part.from_text(text=f"目標: {goal}\nこれまで: {history}\n次の1操作だけを返す"),
        ],
        config=types.GenerateContentConfig(
            tools=[types.Tool(computer_use=types.ComputerUse())],
            temperature=0.0,
        ),
    )
    return extract_action(resp)  # {"type": "click", "x":..,"y":..} 等を取り出す

def perceptual_hash(png):
    """画面の見た目をハッシュ化。stale 検出に使う軽量な指紋。"""
    return dhash(png)  # 画像差分ハッシュ。完全一致でなく近傍判定に使う
 
def execute_with_guard(action, observed_png, observed_hash):
    # 1) 実行直前にもう一度撮る
    fresh_png, w, h = capture()
    if hamming(observed_hash, perceptual_hash(fresh_png)) > STALE_THRESHOLD:
        # 観測時と画面が違う → 古いフレームへの操作は中止
        return {"status": "stale", "fresh_png": fresh_png}
 
    # 2) クリック対象の局所領域だけを比較（全画面より敏感）
    if action["type"] == "click":
        roi = crop(fresh_png, action["x"], action["y"], radius=48)
        roi_at_observe = crop(observed_png, action["x"], action["y"], radius=48)
        if local_changed(roi, roi_at_observe):
            return {"status": "stale_local", "fresh_png": fresh_png}
 
    # 3) 実行 → 直後を撮って「画面が変わったか」を確認
    do_action(action)
    time.sleep(SETTLE_MS / 1000)
    after_png, _, _ = capture()
    progressed = hamming(perceptual_hash(fresh_png), perceptual_hash(after_png)) > NOOP_THRESHOLD
    return {"status": "ok" if progressed else "noop", "after_png": after_png}

DESTRUCTIVE = {"delete", "submit", "purchase", "send"}
 
def guarded_destructive(action, fresh_png):
    label = action.get("label", "")
    if action["type"] not in DESTRUCTIVE:
        return execute_with_guard(action, fresh_png, perceptual_hash(fresh_png))
 
    # 実行前条件: ボタン文言が想定どおりか、確認ダイアログが出ているか
    text_here = ocr_near(fresh_png, action["x"], action["y"], radius=64)
    if label and label not in text_here:
        # モデルが「削除」を押すつもりが、実画面では別ラベル → 中止
        return {"status": "precondition_failed", "expected": label, "found": text_here}
 
    # 取り消し不能な操作はステップを2段に割り、確認画面を必須にする
    return {"status": "needs_confirm_screen", "action": action}

from collections import Counter
 
class RunMetrics:
    def __init__(self):
        self.counts = Counter()
        self.recent_hashes = []
 
    def record(self, result, screen_hash):
        self.counts[result["status"]] += 1
        self.recent_hashes.append(screen_hash)
        self.recent_hashes = self.recent_hashes[-5:]
 
    def stuck_len(self):
        if not self.recent_hashes:
            return 0
        last = self.recent_hashes[-1]
        return sum(1 for h in reversed(self.recent_hashes)
                   if hamming(h, last) <= NOOP_THRESHOLD)
 
    def success_rate(self):
        ok = self.counts["ok"]
        total = sum(self.counts.values()) or 1
        return ok / total

def run_task(goal, max_steps=40):
    m = RunMetrics()
    history = []
    for step in range(max_steps):
        png, w, h = capture()
        h_now = perceptual_hash(png)
        if m.stuck_len() >= 3:
            return escalate("stuck", goal, history)      # 同じ画面で足踏み
        if m.success_rate() < 0.5 and step > 8:
            return escalate("low_success", goal, history) # 早期に手応えがない
        action = run_step(png, goal, history)
        if action["type"] == "done":
            return {"status": "completed", "steps": step}
        result = (guarded_destructive(action, png)
                  if action["type"] in DESTRUCTIVE else
                  execute_with_guard(action, png, h_now))
        m.record(result, h_now)
        history.append({"step": step, "action": action["type"], "result": result["status"]})
    return escalate("budget_exhausted", goal, history)