【python】「gemini-2.5-computer-use-preview-10-2025」で、ブラウザ操作の自動化を試してみた（

#【PowerShell】
cd “対象となるディレクトリの絶対パス”
git clone https://github.com/google/computer-use-preview.git

pip install termcolor
pip install pydantic
pip install google-genai
pip install playwright
pip install rich
pip install pytest
pip install python-dotenv

pip install -r requirements.txt

# 仮想環境なら
py -m pipenv install -r requirements.txt

playwright install chromium

# 仮想環境なら
py -m pipenv run python -m playwright install chromium

# main.py
import argparse
import os
import time
from typing import List, Tuple
from dotenv import load_dotenv
from playwright.sync_api import sync_playwright
from google import genai
from google.genai import types
from google.genai.types import Content, Part

# =========================
# 設定
# =========================
PLAYWRIGHT_SCREEN_SIZE = (1440, 900)  
AI_QUERY = "検索でhello　worldって検索して"
INITIAL_URL = "https://www.google.com/"
LLM_MODEL = "gemini-2.5-computer-use-preview-10-2025"  
TURN_LIMIT = 15  

# APIキー取得
load_dotenv()
API_KEY = os.getenv("Gemini_API_KEY")
if not API_KEY:
    raise RuntimeError("Gemini APIキーが見つかりません。GEMINI_API_KEY を設定してください。")
client = genai.Client(api_key=API_KEY)

# =========================
# Playwrightの初期化
# =========================
def start_browser(initial_url: str):
    width, height = PLAYWRIGHT_SCREEN_SIZE
    playwright_instance = sync_playwright().start()
    browser = playwright_instance.chromium.launch(headless=False)
    context = browser.new_context(viewport={"width": width, "height": height})
    page = context.new_page()
    page.goto(initial_url)
    return playwright_instance, browser, context, page

# =========================
# 座標の正規化↔実座標変換
# =========================
def denorm_x(x: int, screen_width: int) -> int:
    return int(x / 1000 * screen_width)

def denorm_y(y: int, screen_height: int) -> int:
    return int(y / 1000 * screen_height)

# =========================
# function_call を実行
# =========================
def execute_function_calls(candidate, page, screen_w: int, screen_h: int) -> List[Tuple[str, dict]]:
    """
    生成Aiからの回答（candidate）から、何をしたらいいか（function_call）を抽出し、
    それをPlaywrightのpageオブジェクトを使って実行する関数です。
    Args:
        candidate: 生成Aiからの回答オブジェクト。
        page: 画面操作用のPlaywrightページオブジェクト。
        screen_w (int): 画面の幅（ピクセル）。
        screen_h (int): 画面の高さ（ピクセル）。
    Returns:
        List[Tuple[str, dict]]: 実行したfunction_call名と、その結果（警告・エラー含む）のリスト。
    """
    results = []
    content = getattr(candidate, "content", None)
    parts = getattr(content, "parts", None) or []
    # 返答から function_call を抜き出す（並列呼び出しも想定）
    function_calls = []
    for p in parts:
        fc = getattr(p, "function_call", None)
        if fc:
            function_calls.append(fc)

    if not function_calls:
        print("[Info] 生成Aiからの指示はここで終了します。")
        time.sleep(5)
        os._exit(0)
        # return results

    # 生成Aiからの function_call を実行。
    # https://ai.google.dev/gemini-api/docs/computer-use　を参照。
    for fc in function_calls:
        name = fc.name
        args = fc.args or {}
        action_result = {}
        try:
            # open_web_browser: デフォルト検索ページへ移動
            if name == "open_web_browser":
                page.goto(INITIAL_URL)
                action_result = {"ok": True, "url": page.url}

            # click_at: x,y は 0-999 のグリッド座標
            elif name == "click_at":
                ax = args.get("x", args.get("X"))
                ay = args.get("y", args.get("Y"))
                if ax is None or ay is None:
                    raise ValueError("click_at requires x and y")
                x = denorm_x(int(ax), screen_w)
                y = denorm_y(int(ay), screen_h)
                page.mouse.click(x, y)
                action_result = {"ok": True, "x": x, "y": y}

            # hover_at: マウスを移動してホバー
            elif name == "hover_at":
                ax = args.get("x", args.get("X"))
                ay = args.get("y", args.get("Y"))
                if ax is None or ay is None:
                    raise ValueError("hover_at requires x and y")
                x = denorm_x(int(ax), screen_w)
                y = denorm_y(int(ay), screen_h)
                page.mouse.move(x, y)
                action_result = {"ok": True, "x": x, "y": y}

            # type_text_at: クリアの有無・Enter 押下の有無を引数で制御
            elif name == "type_text_at":
                ax = args.get("x", args.get("X"))
                ay = args.get("y", args.get("Y"))
                text = args.get("text", "")
                press_enter = args.get("press_enter", True)
                clear_before = args.get("clear_before_typing", True)
                if ax is None or ay is None:
                    raise ValueError("type_text_at requires x and y")
                x = denorm_x(int(ax), screen_w)
                y = denorm_y(int(ay), screen_h)
                page.mouse.click(x, y)
                if clear_before:
                    page.keyboard.press("Control+A")
                    page.keyboard.press("Backspace")
                if text:
                    page.keyboard.type(text)
                if press_enter:
                    page.keyboard.press("Enter")
                action_result = {"ok": True, "text": text, "x": x, "y": y}

            # go_back / go_forward / reload_page
            elif name == "go_back":
                page.go_back()
                action_result = {"ok": True, "url": page.url}
            elif name == "go_forward":
                page.go_forward()
                action_result = {"ok": True, "url": page.url}
            elif name == "reload_page":
                page.reload()
                action_result = {"ok": True, "url": page.url}

            # wait_5_seconds: 動的読み込み待ち
            elif name == "wait_5_seconds":
                time.sleep(5)
                action_result = {"ok": True, "waited": 5}

            # search: デフォルト検索エンジンへ移動
            elif name == "search":
                page.goto(INITIAL_URL)
                action_result = {"ok": True, "url": page.url}

            # navigate: 指定URLへ遷移
            elif name == "navigate":
                url = args.get("url")
                if not url:
                    raise ValueError("navigate requires url")
                page.goto(url)
                action_result = {"ok": True, "url": page.url}

            # key_combination: キー入力 / 組み合わせ（例: "Control+A"）
            elif name == "key_combination":
                keys = args.get("keys")
                if not keys:
                    raise ValueError("key_combination requires keys")
                page.keyboard.press(keys)
                action_result = {"ok": True, "keys": keys}

            # scroll_document: ページ全体をスクロール
            elif name == "scroll_document":
                direction = args.get("direction", "down").lower()
                if direction not in ("up", "down", "left", "right"):
                    raise ValueError("scroll_document direction must be up/down/left/right")
                # スクロール量は画面サイズに依存させる
                dx = 0
                dy = 0
                if direction in ("up", "down"):
                    amount = int(screen_h * 0.8)
                    dy = -amount if direction == "up" else amount
                else:
                    amount = int(screen_w * 0.8)
                    dx = -amount if direction == "left" else amount
                page.evaluate("window.scrollBy(arguments[0], arguments[1])", dx, dy)
                action_result = {"ok": True, "direction": direction, "dx": dx, "dy": dy}

            # scroll_at: 指定座標付近の要素をスクロール（マウス位置に移動してwheelを使う）
            elif name == "scroll_at":
                ax = args.get("x", args.get("X"))
                ay = args.get("y", args.get("Y"))
                direction = args.get("direction", "down").lower()
                magnitude = int(args.get("magnitude", 800))
                if ax is None or ay is None:
                    raise ValueError("scroll_at requires x and y")
                if direction not in ("up", "down", "left", "right"):
                    raise ValueError("scroll_at direction must be up/down/left/right")
                x = denorm_x(int(ax), screen_w)
                y = denorm_y(int(ay), screen_h)
                # magnitude は 0-999 を想定 → ピクセル換算
                mag_px_y = int(magnitude / 1000 * screen_h)
                mag_px_x = int(magnitude / 1000 * screen_w)
                # Playwright の mouse.wheel(delta_x, delta_y)
                dx = 0
                dy = 0
                if direction == "down":
                    dy = mag_px_y
                elif direction == "up":
                    dy = -mag_px_y
                elif direction == "right":
                    dx = mag_px_x
                elif direction == "left":
                    dx = -mag_px_x
                page.mouse.move(x, y)
                # wheel は相対スクロール
                try:
                    page.mouse.wheel(dx, dy)
                except Exception:
                    # 万一 wheel がサポート外なら window.scrollBy にフォールバック
                    page.evaluate(
                        "window.scrollBy(arguments[0], arguments[1])",
                        dx if dx else 0,
                        dy if dy else 0,
                    )
                action_result = {"ok": True, "x": x, "y": y, "dx": dx, "dy": dy}

            # drag_and_drop: 開始座標から終了座標へドラッグ
            elif name == "drag_and_drop":
                ax = args.get("x", args.get("X"))
                ay = args.get("y", args.get("Y"))
                dest_x = args.get("destination_x", args.get("destination_X"))
                dest_y = args.get("destination_y", args.get("destination_Y"))
                if None in (ax, ay, dest_x, dest_y):
                    raise ValueError("drag_and_drop requires x,y,destination_x,destination_y")
                sx = denorm_x(int(ax), screen_w)
                sy = denorm_y(int(ay), screen_h)
                dx = denorm_x(int(dest_x), screen_w)
                dy = denorm_y(int(dest_y), screen_h)
                page.mouse.move(sx, sy)
                page.mouse.down()
                time.sleep(0.12)
                page.mouse.move(dx, dy, steps=12)
                time.sleep(0.12)
                page.mouse.up()
                action_result = {"ok": True, "start": (sx, sy), "end": (dx, dy)}

            else:
                print(f"[警告] 生成Aiから未実装のアクションをようきゅうされました。アクション要求名は {name}です")
                action_result = {"warning": f"Unimplemented action: {name}"}

            # 可能ならページ遷移や描画待ち
            try:
                page.wait_for_load_state(timeout=5000)
            except Exception:
                pass
            time.sleep(0.6)

        except Exception as e:
            print(f"[Error] {name}: {e}")
            action_result = {"error": str(e)}

        results.append((name, action_result))
    return results

# =========================
# 実行結果を function_response に変換
# スクショとURLを返す
# =========================
def build_function_responses(page, results: List[Tuple[str, dict]]):
    """
    ページをスクリーンショットして、次に行うアクションを返します。
    引数:
        page: スクリーンショット取得やURL参照が可能なページオブジェクト。
        results (List[Tuple[str, dict]]): 関数名とその結果情報（辞書）のタプルのリスト。

    戻り値:
        List[types.FunctionResponse]: 各関数名に対応するFunctionResponseオブジェクトのリスト。各レスポンスにはページのURLと結果情報、PNG形式のスクリーンショット画像が含まれます。
    """
    # ページのスクリーンショットをPNG形式で取得
    screenshot_bytes = page.screenshot(type="png")
    
    # 現在のページURLを取得
    current_url = page.url
    function_responses = []
    
    # 各実行結果ごとにFunctionResponseを作成
    for name, result in results:
        payload = {"url": current_url}  
        payload.update(result)          
        function_responses.append(
            types.FunctionResponse(
                name=name,              
                response=payload,       
                parts=[
                    types.FunctionResponsePart(
                        inline_data=types.FunctionResponseBlob(
                            mime_type="image/png",  
                            data=screenshot_bytes   
                        )
                    )
                ],
            )
        )
    return function_responses

# =========================
# Computer Use 用の GenerateContentConfig を作成
# =========================
def make_generate_content_config(excluded: list | None = None):
    return genai.types.GenerateContentConfig(
        tools=[
            types.Tool(
                computer_use=types.ComputerUse(
                    environment=types.Environment.ENVIRONMENT_BROWSER,
                    excluded_predefined_functions=excluded or [],  # 必要に応じて制限
                )
            )
        ]
        # ここにカスタム関数（function_declarations）を追加可能
    )

# =========================
# エージェント本体
# =========================
def run_agent(query: str, initial_url: str) -> int:
    playwright_instance, browser, context, page = start_browser(initial_url)
    width, height = PLAYWRIGHT_SCREEN_SIZE

    try:
        config = make_generate_content_config()
        contents: List[Content | types.FunctionResponse] = [
            Content(role="user", parts=[Part(text=query)])
        ]

        for turn in range(1,TURN_LIMIT + 1):
            print(f"\n--- Turn {turn} ---")
            # 1) モデル呼び出し
            response = client.models.generate_content(
                model=LLM_MODEL, contents=contents, config=config
            )
            
            # 待機時間
            backoff = 1.0
            
            # 応答の解析
            cands = getattr(response, "candidates", None)
            if not cands:
                fb = getattr(response, "prompt_feedback", None)
                err = getattr(response, "error", None)
                print(f"[Warn] No candidates on turn {turn}. "
                    f"prompt_feedback={fb}, error={err}. backoff={backoff:.1f}s")
                time.sleep(backoff)
                backoff = min(backoff * 2, 8.0)  # 上限8秒
                continue

            candidate = cands[0]        
            
            
            contents.append(candidate.content)  

            # 2) safety_decision の確認
            safety = getattr(candidate, "safety_decision", None)
            if safety and getattr(safety, "require_confirmation", False):
                print("[Info] 要ユーザー確認のアクション → 本実装では実行せず次へ")

            # 3) function_call をPlaywrightで実行
            results = execute_function_calls(candidate, page, width, height)

            # 4) function_response を積んで次ターンへ
            func_responses = build_function_responses(page, results)
            contents.extend(func_responses)

            # 適宜終了判定（簡易）：検索結果画面に到達していれば抜けるなど
            if "search" in (page.url or "").lower():
                pass

        return 0
    finally:
        try:
            context.close()
            browser.close()
            playwright_instance.stop()
        except Exception:
            pass

# =========================
# CLI引数
# =========================
def parse_args_if_any():
    import sys
    if len(sys.argv) == 1:
        return None
    parser = argparse.ArgumentParser(description="Gemini Computer Use エージェント")
    parser.add_argument("--query", type=str, required=True, help="エージェントへの指示文")
    parser.add_argument("--initial_url", type=str, default=INITIAL_URL)
    parser.add_argument("--turns", type=int, default=6)
    return parser.parse_args()

def main() -> int:
    args = parse_args_if_any()
    if args is None:
        query = os.getenv("CU_QUERY", AI_QUERY)
        initial_url = os.getenv("CU_INITIAL_URL", INITIAL_URL)
        return run_agent(query, initial_url)
    else:
        return run_agent(args.query, args.initial_url, args.turns)

if __name__ == "__main__":
    raise SystemExit(main())