wehub-resource-sync
290 lines
11 KiB
Python
290 lines
11 KiB
Python
# Copyright 2025 Google LLC
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
|
|
from google import genai
|
|
from google.genai.types import (
|
|
ComputerUse,
|
|
Content,
|
|
Environment,
|
|
FinishReason,
|
|
FunctionResponse,
|
|
FunctionResponseBlob,
|
|
GenerateContentConfig,
|
|
Part,
|
|
ThinkingConfig,
|
|
Tool,
|
|
)
|
|
from playwright.async_api import Page, async_playwright
|
|
|
|
logging.getLogger("google_genai._common").setLevel(logging.ERROR)
|
|
|
|
# --- CONFIGURATION ---
|
|
# Load configuration from environment variables for best practice.
|
|
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT")
|
|
LOCATION = os.environ.get("GOOGLE_LOCATION", "global")
|
|
MODEL_ID = os.environ.get("MODEL_ID", "gemini-3.5-flash")
|
|
|
|
|
|
# --- HELPER FUNCTIONS ---
|
|
|
|
|
|
def normalize_x(x: int, screen_width: int) -> int:
|
|
"""Convert normalized x coordinate (0-1000) to actual pixel coordinate."""
|
|
return int(x / 1000 * screen_width)
|
|
|
|
|
|
def normalize_y(y: int, screen_height: int) -> int:
|
|
"""Convert normalized y coordinate (0-1000) to actual pixel coordinate."""
|
|
return int(y / 1000 * screen_height)
|
|
|
|
|
|
async def execute_function_calls(
|
|
response, page: Page, screen_width: int, screen_height: int
|
|
) -> tuple[
|
|
str, list[tuple[str, str, bool]]
|
|
]: # <-- Note the added bool for safety status
|
|
"""Extracts and executes function calls from the model response."""
|
|
await asyncio.sleep(0.1)
|
|
|
|
function_calls = [
|
|
part.function_call
|
|
for part in response.candidates[0].content.parts
|
|
if hasattr(part, "function_call") and part.function_call
|
|
]
|
|
|
|
thoughts = [
|
|
part.text
|
|
for part in response.candidates[0].content.parts
|
|
if hasattr(part, "text") and part.text
|
|
]
|
|
|
|
if thoughts:
|
|
print(f"🤔 Model Reasoning: {' '.join(thoughts)}")
|
|
|
|
if not function_calls:
|
|
return "NO_ACTION", []
|
|
|
|
results = []
|
|
for function_call in function_calls:
|
|
result = None
|
|
safety_acknowledged = False
|
|
|
|
safety_decision = function_call.args.get("safety_decision")
|
|
if (
|
|
safety_decision
|
|
and safety_decision.get("decision") == "require_confirmation"
|
|
):
|
|
print(f"\n⚠️ SAFETY PROMPT: {safety_decision.get('explanation')}")
|
|
user_input = input(
|
|
f"Allow the agent to execute '{function_call.name}'? (y/n): "
|
|
)
|
|
|
|
if user_input.strip().lower() not in ["y", "yes"]:
|
|
print("🚫 Action denied by user.")
|
|
results.append((function_call.name, "user_denied", False))
|
|
continue # Skip execution and move to the next function call
|
|
|
|
print("✅ Action approved.")
|
|
safety_acknowledged = True
|
|
|
|
print(f"⚡ Executing Action: {function_call.name}")
|
|
try:
|
|
if function_call.name == "open_web_browser":
|
|
result = "success" # The browser is already open
|
|
elif function_call.name == "navigate":
|
|
await page.goto(function_call.args["url"])
|
|
result = "success"
|
|
elif function_call.name == "click_at":
|
|
actual_x = normalize_x(function_call.args["x"], screen_width)
|
|
actual_y = normalize_y(function_call.args["y"], screen_height)
|
|
await page.mouse.click(actual_x, actual_y)
|
|
result = "success"
|
|
elif function_call.name == "type_text_at":
|
|
text_to_type = function_call.args["text"]
|
|
print(f'[DEBUG] Typing text: "{text_to_type}"')
|
|
actual_x = normalize_x(function_call.args["x"], screen_width)
|
|
actual_y = normalize_y(function_call.args["y"], screen_height)
|
|
await page.mouse.click(actual_x, actual_y)
|
|
await asyncio.sleep(0.1)
|
|
await page.keyboard.type(text_to_type)
|
|
if function_call.args.get("press_enter", False):
|
|
await page.keyboard.press("Enter")
|
|
result = "success"
|
|
else:
|
|
result = "unknown_function"
|
|
except Exception as e:
|
|
print(f"❗️ Error executing {function_call.name}: {e}")
|
|
result = f"error: {e!s}"
|
|
|
|
results.append((function_call.name, result, safety_acknowledged))
|
|
|
|
return "CONTINUE", results
|
|
|
|
|
|
# --- THE AGENT LOOP ---
|
|
|
|
|
|
async def agent_loop(initial_prompt: str, max_turns: int = 20) -> None:
|
|
"""Main agent loop for local execution with a browser."""
|
|
if not PROJECT_ID:
|
|
raise ValueError("GOOGLE_PROJECT_ID environment variable not set.")
|
|
|
|
client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)
|
|
|
|
browser = None
|
|
try:
|
|
async with async_playwright() as p:
|
|
# MODIFIED: Launch browser in a try...finally block
|
|
browser = await p.chromium.launch(headless=False)
|
|
page = await browser.new_page()
|
|
sw, sh = 960, 1080
|
|
await page.set_viewport_size({"width": sw, "height": sh})
|
|
await page.goto("https://www.google.com")
|
|
|
|
print(f"🎬 Starting Agent Loop with prompt: '{initial_prompt}'")
|
|
|
|
# Configure Computer Use tool with browser environment
|
|
# Base configuration for the Computer Use tool
|
|
config_kwargs = {
|
|
"tools": [
|
|
Tool(
|
|
computer_use=ComputerUse(
|
|
environment=Environment.ENVIRONMENT_BROWSER,
|
|
# Optional: Exclude specific predefined functions
|
|
excluded_predefined_functions=["drag_and_drop"],
|
|
)
|
|
)
|
|
]
|
|
}
|
|
|
|
# Conditionally add thinking_config only for the Gemini 3 models
|
|
model_version = float(MODEL_ID.split("-")[1])
|
|
if model_version >= 3:
|
|
config_kwargs["thinking_config"] = ThinkingConfig(include_thoughts=True)
|
|
|
|
config = GenerateContentConfig(**config_kwargs)
|
|
|
|
screenshot = await page.screenshot()
|
|
contents = [
|
|
Content(
|
|
role="user",
|
|
parts=[
|
|
Part(text=initial_prompt),
|
|
Part.from_bytes(data=screenshot, mime_type="image/png"),
|
|
],
|
|
)
|
|
]
|
|
for turn in range(max_turns):
|
|
print(f"\n--- 🔁 Turn {turn + 1} ---")
|
|
print(f"[DEBUG] Current URL: {page.url}")
|
|
|
|
response = client.models.generate_content(
|
|
model=MODEL_ID, contents=contents, config=config
|
|
)
|
|
|
|
if not response.candidates:
|
|
print("❗️ Model returned no candidates. Terminating loop.")
|
|
print("Full Response:", response)
|
|
break
|
|
|
|
if response.candidates[0].finish_reason == FinishReason.SAFETY:
|
|
print(
|
|
"🛑 SAFETY TRIGGERED: The model halted execution due to safety policies."
|
|
)
|
|
print(f"Details: {response.candidates[0].safety_ratings}")
|
|
break
|
|
|
|
print(
|
|
f"[DEBUG] Model Finish Reason: {response.candidates[0].finish_reason}"
|
|
)
|
|
contents.append(response.candidates[0].content)
|
|
print("[DEBUG] Appended model response to history.")
|
|
|
|
# Check if the attribute exists AND is not None
|
|
active_function_calls = [
|
|
part.function_call
|
|
for part in response.candidates[0].content.parts
|
|
if hasattr(part, "function_call") and part.function_call
|
|
]
|
|
|
|
if not active_function_calls:
|
|
final_text = "".join(
|
|
part.text
|
|
for part in response.candidates[0].content.parts
|
|
if hasattr(part, "text") and part.text is not None
|
|
)
|
|
if final_text:
|
|
print(f"✅ Agent Finished: {final_text}")
|
|
break
|
|
|
|
status, execution_results = await execute_function_calls(
|
|
response, page, sw, sh
|
|
)
|
|
print(
|
|
f"[DEBUG] Execution Results: status='{status}', results={execution_results}"
|
|
)
|
|
|
|
if status == "NO_ACTION":
|
|
continue
|
|
|
|
function_response_parts = []
|
|
# Unpack the 3 variables returned by our updated function
|
|
for name, result, safety_acknowledged in execution_results:
|
|
screenshot = await page.screenshot()
|
|
current_url = page.url
|
|
|
|
# Prepare the response payload
|
|
response_payload = {"url": current_url}
|
|
|
|
# Handle the safety and denial states
|
|
if result == "user_denied":
|
|
response_payload["error"] = "user_denied"
|
|
elif safety_acknowledged:
|
|
# CRITICAL: Acknowledge the safety decision so the API doesn't throw an error
|
|
response_payload["safety_acknowledgement"] = True
|
|
|
|
function_response_parts.append(
|
|
Part(
|
|
function_response=FunctionResponse(
|
|
name=name,
|
|
response=response_payload,
|
|
parts=[
|
|
Part(
|
|
inline_data=FunctionResponseBlob(
|
|
mime_type="image/png", data=screenshot
|
|
)
|
|
)
|
|
],
|
|
)
|
|
)
|
|
)
|
|
|
|
contents.append(Content(role="user", parts=function_response_parts))
|
|
print(f"📝 State captured. History now has {len(contents)} messages.")
|
|
finally:
|
|
if browser:
|
|
await browser.close()
|
|
print("\n--- Browser closed. ---")
|
|
|
|
|
|
# --- SCRIPT ENTRY POINT ---
|
|
if __name__ == "__main__":
|
|
prompt = "Navigate to the Google Store and find the page of 'Pixel 10'."
|
|
|
|
asyncio.run(agent_loop(prompt))
|