操作電腦

pip install --upgrade google-genai playwright termcolor
playwright install chromium

export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
export GOOGLE_CLOUD_LOCATION="global"
export GOOGLE_GENAI_USE_ENTERPRISE=True

from playwright.sync_api import sync_playwright

SCREEN_WIDTH = 1440
SCREEN_HEIGHT = 900

playwright = sync_playwright().start()
browser = playwright.chromium.launch(headless=False)

context = browser.new_context(
    viewport={"width": SCREEN_WIDTH, "height": SCREEN_HEIGHT}
)
page = context.new_page()
page.goto("https://www.google.com")

{
  "content": {
    "parts": [
      {
        "function_call": {
          "name": "type",
          "args": {
            "x": 450,
            "y": 320,
            "text": "smart fridge touchscreen 2 doors 25 cu ft under $4000",
            "press_enter": true,
            "intent": "Type the search query into the Google search box and submit."
          }
        }
      }
    ]
  }
}

{
  "content": {
    "parts": [
      {
        "function_call": {
          "name": "click",
          "args": {
            "x": 60,
            "y": 100,
            "intent": "Click the 'I'm not a robot' verification checkbox.",
            "safety_decision": {
              "explanation": "I have encountered a CAPTCHA challenge that requires interaction. I need you to complete the challenge by clicking the 'I'm not a robot' checkbox and any subsequent verification steps.",
              "decision": "require_confirmation"
            }
          }
        }
      }
    ]
  }
}

import time
from typing import Any, Dict, List, Tuple
import termcolor


def denormalize_x(x: int, screen_width: int) -> int:
    """Convert normalized x coordinate (0-1000) to actual pixel coordinate."""
    return int(x / 1000 * screen_width)


def denormalize_y(y: int, screen_height: int) -> int:
    """Convert normalized y coordinate (0-1000) to actual pixel coordinate."""
    return int(y / 1000 * screen_height)


def get_safety_confirmation(safety_decision: Dict[str, Any]) -> str:
    """Prompt user for confirmation when safety check is triggered."""
    termcolor.cprint("Safety service requires explicit confirmation!", color="red")
    print(safety_decision.get("explanation", ""))

    decision = ""
    while decision.lower() not in ("y", "n", "ye", "yes", "no"):
        decision = input("Do you wish to proceed? [Y]es/[N]o\n")

    if decision.lower() in ("n", "no"):
        return "TERMINATE"
    return "CONTINUE"


def execute_function_calls(
    response, page, screen_width: int, screen_height: int
) -> List[Tuple[str, Dict[str, Any]]]:
    """Extract and execute modern and legacy function calls from Gemini response."""
    candidate = response.candidates[0]
    function_calls = []
    thoughts = []

    for part in candidate.content.parts:
        if hasattr(part, "function_call") and part.function_call:
            function_calls.append(part.function_call)
        elif hasattr(part, "text") and part.text:
            thoughts.append(part.text)

    if thoughts:
        print(f"Model Reasoning: {' '.join(thoughts)}")

    results: List[Tuple[str, Dict[str, Any]]] = []
    for function_call in function_calls:
        fname = function_call.name
        args = function_call.args or {}
        action_result: Dict[str, Any] = {}

        print(f" -> Executing: {fname} (Intent: {args.get('intent', 'N/A')})")

        # Check for safety decision before executing
        if "safety_decision" in args:
            decision = get_safety_confirmation(args["safety_decision"])
            if decision == "TERMINATE":
                print("Terminating agent loop")
                break
            action_result["safety_acknowledgement"] = "true"

        try:
            if fname in ("open_web_browser", "open_app", "take_screenshot"):
                pass

            elif fname in (
                "click",
                "click_at",
                "double_click",
                "triple_click",
                "middle_click",
                "right_click",
                "mouse_down",
                "mouse_up",
                "move",
                "hover_at",
            ):
                actual_x = denormalize_x(args["x"], screen_width)
                actual_y = denormalize_y(args["y"], screen_height)

                if fname in ("click", "click_at"):
                    page.mouse.click(actual_x, actual_y)
                elif fname == "double_click":
                    page.mouse.dblclick(actual_x, actual_y)
                elif fname == "triple_click":
                    page.mouse.click(actual_x, actual_y, click_count=3)
                elif fname == "middle_click":
                    page.mouse.click(actual_x, actual_y, button="middle")
                elif fname == "right_click":
                    page.mouse.click(actual_x, actual_y, button="right")
                elif fname == "mouse_down":
                    page.mouse.move(actual_x, actual_y)
                    page.mouse.down()
                elif fname == "mouse_up":
                    page.mouse.move(actual_x, actual_y)
                    page.mouse.up()
                elif fname in ("move", "hover_at"):
                    page.mouse.move(actual_x, actual_y)

            elif fname in ("type", "type_text_at"):
                actual_x = denormalize_x(args["x"], screen_width) if "x" in args else None
                actual_y = denormalize_y(args["y"], screen_height) if "y" in args else None
                text = args["text"]
                press_enter = args.get("press_enter", False)
                clear_before_typing = args.get("clear_before_typing", True)

                if actual_x is not None and actual_y is not None:
                    page.mouse.click(actual_x, actual_y)
                    time.sleep(0.1)

                if clear_before_typing:
                    page.keyboard.press("ControlOrMeta+A")
                    page.keyboard.press("Backspace")

                page.keyboard.type(text)
                if press_enter:
                    page.keyboard.press("Enter")

            elif fname in ("scroll", "scroll_at", "scroll_document"):
                if "x" in args and "y" in args:
                    page.mouse.move(denormalize_x(args["x"], screen_width), denormalize_y(args["y"], screen_height))
                direction = args.get("direction", "down")
                magnitude = args.get("magnitude_in_pixels", args.get("magnitude", 300))
                dx, dy = 0, 0
                if direction == "down":
                    dy = magnitude
                elif direction == "up":
                    dy = -magnitude
                elif direction == "right":
                    dx = magnitude
                elif direction == "left":
                    dx = -magnitude
                page.mouse.wheel(dx, dy)

            elif fname == "drag_and_drop":
                start_x = denormalize_x(args.get("start_x", args.get("x", 0)), screen_width)
                start_y = denormalize_y(args.get("start_y", args.get("y", 0)), screen_height)
                end_x = denormalize_x(args.get("end_x", args.get("destination_x", 0)), screen_width)
                end_y = denormalize_y(args.get("end_y", args.get("destination_y", 0)), screen_height)
                page.mouse.move(start_x, start_y)
                page.mouse.down()
                page.mouse.move(end_x, end_y)
                page.mouse.up()

            elif fname == "press_key":
                page.keyboard.press(args["key"])
            elif fname == "key_down":
                page.keyboard.down(args["key"])
            elif fname == "key_up":
                page.keyboard.up(args["key"])
            elif fname == "hotkey":
                page.keyboard.press("+".join(args.get("keys", [])))
            elif fname == "key_combination":
                page.keyboard.press(args["keys"])
            elif fname == "navigate":
                page.goto(args["url"])
            elif fname == "search":
                page.goto("https://www.google.com")
            elif fname == "go_back":
                page.go_back()
            elif fname == "go_forward":
                page.go_forward()
            elif fname in ("wait", "wait_5_seconds"):
                time.sleep(5 if fname == "wait_5_seconds" else args.get("seconds", 1))
            else:
                action_result["warning"] = f"unhandled_function: {fname}"

            page.wait_for_load_state(timeout=5000)
            time.sleep(1)

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

        results.append((fname, action_result))

    return results

from google.genai import types
from google.genai.types import Content, FunctionResponse, Part


def get_function_responses(
    page, results: List[Tuple[str, Dict[str, Any]]]
) -> Content:
    """Capture screenshot and URL and package them as FunctionResponses."""
    screenshot = page.screenshot(type="png")
    current_url = page.url
    function_response_parts = []

    for name, action_result in results:
        function_response_parts.append(
            Part(
                function_response=FunctionResponse(
                    name=name,
                    response={
                        "url": current_url,
                        **action_result,
                    },
                    parts=[
                        types.FunctionResponsePart(
                            inline_data=types.FunctionResponseBlob(
                                mime_type="image/png",
                                data=screenshot,
                            )
                        )
                    ],
                )
            )
        )

    return Content(
        role="user",
        parts=function_response_parts,
    )

{
  "content": {
    "parts": [
      {
        "text": "I have evaluated step 2. It seems Google detected unusual traffic and is asking me to verify I'm not a robot. I need to click the 'I'm not a robot' checkbox located near the top left (y=98, x=95)."
      },
      {
        "function_call": {
          "name": "click_at",
          "args": {
            "x": 60,
            "y": 100,
            "safety_decision": {
              "explanation": "I have encountered a CAPTCHA challenge that requires interaction. I need you to complete the challenge by clicking the 'I'm not a robot' checkbox and any subsequent verification steps.",
              "decision": "require_confirmation"
            }
          }
        }
      }
    ]
  }
}

import termcolor

def get_safety_confirmation(safety_decision):
    """Prompt user for confirmation when safety check is triggered."""
    termcolor.cprint("Safety service requires explicit confirmation!", color="red")
    print(safety_decision["explanation"])

    decision = ""
    while decision.lower() not in ("y", "n", "ye", "yes", "no"):
        decision = input("Do you wish to proceed? [Y]es/[N]o\n")

    if decision.lower() in ("n", "no"):
        return "TERMINATE"
    return "CONTINUE"

def execute_function_calls(response, page, screen_width: int, screen_height: int):

    # ... Extract function calls from response ...

    for function_call in function_calls:
        extra_fr_fields = {}

        # Check for safety decision
        if 'safety_decision' in function_call.args:
            decision = get_safety_confirmation(function_call.args['safety_decision'])
            if decision == "TERMINATE":
                print("Terminating agent loop")
                break
            extra_fr_fields["safety_acknowledgement"] = "true" # Safety acknowledgement

        # ... Execute function call and append to results ...

function_response_parts.append(
    FunctionResponse(
        name=name,
        response={"url": current_url,
                  **extra_fr_fields},  # Include safety acknowledgement
        parts=[
            types.FunctionResponsePart(
                inline_data=types.FunctionResponseBlob(
                    mime_type="image/png", data=screenshot
                )
             )
           ]
         )
       )