Anteckning
Åtkomst till den här sidan kräver auktorisering. Du kan prova att logga in eller ändra kataloger.
Åtkomst till den här sidan kräver auktorisering. Du kan prova att ändra kataloger.
Använd den här artikeln om du vill lära dig hur du arbetar med datoranvändning i Azure OpenAI. Datoranvändning är ett specialiserat AI-verktyg som använder en specialiserad modell som kan utföra uppgifter genom att interagera med datorsystem och program via deras UIs. Med Datoranvändning kan du skapa en agent som kan hantera komplexa uppgifter och fatta beslut genom att tolka visuella element och vidta åtgärder baserat på skärminnehåll.
Datoranvändning tillhandahåller:
- autonom navigering: Till exempel öppnar program, klickar på knappar, fyller i formulär och navigerar i arbetsflöden för flera sidor.
- Dynamisk anpassning: Tolkar ändringar i användargränssnittet och justerar åtgärder i enlighet med detta.
- Uppgiftskörning mellan program: Fungerar på webbaserade och skrivbordsapplikationer.
- gränssnitt för naturligt språk: Användare kan beskriva en uppgift på vanligt språk och modellen Datoranvändning avgör rätt gränssnittsinteraktioner som ska köras.
Begär åtkomst
För åtkomst till computer-use-preview-modellen krävs registrering och åtkomst beviljas baserat på Microsofts berättigandekriterier. Kunder som har åtkomst till andra modeller för begränsad åtkomst måste fortfarande begära åtkomst för den här modellen.
Begär åtkomst: datoranvändning-förhandsvisning begränsad åtkomst modellansökan
När åtkomsten har beviljats måste du skapa en distribution för modellen.
Regionalt stöd
Datoranvändning är tillgängligt i följande regioner:
eastus2swedencentralsouthindia
Skicka ett API-anrop till modellen Datoranvändning med hjälp av svars-API:et
Verktyget för datoranvändning nås genom API-svar . Verktyget körs i en kontinuerlig loop som skickar åtgärder som att skriva text eller utföra ett klick. Koden kör dessa åtgärder på en dator och skickar skärmbilder av resultatet till modellen.
På så sätt simulerar koden en människas åtgärder med hjälp av ett datorgränssnitt, medan modellen använder skärmbilderna för att förstå miljöns tillstånd och föreslå nästa åtgärder.
I följande exempel visas ett grundläggande API-anrop.
Anmärkning
Du behöver en Azure OpenAI-resurs med en computer-use-preview modellimplementering i en stödd region.
Om du vill skicka begäranden måste du installera följande Python-paket.
pip install openai
pip install azure-identity
import os
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import OpenAI
#from openai import OpenAI
token_provider = get_bearer_token_provider(DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default")
client = OpenAI(
base_url = "https://YOUR-RESOURCE-NAME.openai.azure.com/openai/v1/",
api_key=token_provider,
)
response = client.responses.create(
model="computer-use-preview", # set this to your model deployment name
tools=[{
"type": "computer_use_preview",
"display_width": 1024,
"display_height": 768,
"environment": "browser" # other possible values: "mac", "windows", "ubuntu"
}],
input=[
{
"role": "user",
"content": "Check the latest AI news on bing.com."
}
],
truncation="auto"
)
print(response.output)
Utgång
[
ResponseComputerToolCall(
id='cu_67d841873c1081908bfc88b90a8555e0',
action=ActionScreenshot(type='screenshot'),
call_id='call_wwEnfFDqQr1Z4Edk62Fyo7Nh',
pending_safety_checks=[],
status='completed',
type='computer_call'
)
]
När den första API-begäran har skickats utför du en loop där den angivna åtgärden utförs i programkoden och skickar en skärmbild med varje tur så att modellen kan utvärdera det uppdaterade tillståndet för miljön.
## response.output is the previous response from the model
computer_calls = [item for item in response.output if item.type == "computer_call"]
if not computer_calls:
print("No computer call found. Output from model:")
for item in response.output:
print(item)
computer_call = computer_calls[0]
last_call_id = computer_call.call_id
action = computer_call.action
# Your application would now perform the action suggested by the model
# And create a screenshot of the updated state of the environment before sending another response
response_2 = client.responses.create(
model="computer-use-preview",
previous_response_id=response.id,
tools=[{
"type": "computer_use_preview",
"display_width": 1024,
"display_height": 768,
"environment": "browser" # other possible values: "mac", "windows", "ubuntu"
}],
input=[
{
"call_id": last_call_id,
"type": "computer_call_output",
"output": {
"type": "input_image",
# Image should be in base64
"image_url": f"data:image/png;base64,{<base64_string>}"
}
}
],
truncation="auto"
)
Förstå integrering av datoranvändning
När du arbetar med verktyget Datoranvändning utför du vanligtvis följande för att integrera det i ditt program.
- Skicka en begäran till modellen som innehåller ett anrop till datoranvändningsverktyget och visningsstorleken och miljön. Du kan också ta med en skärmbild av det ursprungliga tillståndet för miljön i den första API-begäran.
- Ta emot ett svar från modellen. Om svaret har
actionobjekt innehåller dessa objekt föreslagna åtgärder för att göra framsteg mot det angivna målet. En åtgärd kan till exempel varascreenshotså att modellen kan utvärdera det aktuella tillståndet med en uppdaterad skärmbild, ellerclickmed X/Y-koordinater som anger var musen ska flyttas. - Kör åtgärden med hjälp av programkoden på datorn eller webbläsarmiljön.
- När du har kört åtgärden avbildar du det uppdaterade tillståndet för miljön som en skärmbild.
- Skicka en ny begäran med det uppdaterade tillståndet som en
computer_call_outputoch upprepa den här loopen tills modellen slutar begära åtgärder eller så bestämmer du dig för att sluta.
Hantera konversationshistorik
Du kan använda parametern previous_response_id för att länka den aktuella begäran till föregående svar. Använd den här parametern om du inte vill hantera konversationshistoriken.
Om du inte använder den här parametern bör du se till att inkludera alla objekt som returneras i svarsutdata från den tidigare begäran i indatamatrisen. Detta inkluderar resonemangsobjekt om de finns.
Säkerhetskontroller
API:et har säkerhetskontroller för att skydda mot snabba inmatnings- och modellmisstag. Dessa kontroller omfattar följande:
- Identifiering av skadliga instruktioner: Systemet utvärderar skärmbilden och kontrollerar om den innehåller skadligt innehåll som kan ändra modellens beteende.
-
Irrelevant domänidentifiering: Systemet utvärderar
current_url(om det tillhandahålls) och kontrollerar om den aktuella domänen anses relevant med tanke på konversationshistoriken. -
Känslig domänidentifiering: Systemet kontrollerar
current_url(om det tillhandahålls) och genererar en varning när den identifierar att användaren finns på en känslig domän.
Om en eller flera av ovanstående kontroller utlöses, utlöses en säkerhetskontroll när modellen returnerar nästa computer_call, med parametern pending_safety_checks.
"output": [
{
"type": "reasoning",
"id": "rs_67cb...",
"summary": [
{
"type": "summary_text",
"text": "Exploring 'File' menu option."
}
]
},
{
"type": "computer_call",
"id": "cu_67cb...",
"call_id": "call_nEJ...",
"action": {
"type": "click",
"button": "left",
"x": 135,
"y": 193
},
"pending_safety_checks": [
{
"id": "cu_sc_67cb...",
"code": "malicious_instructions",
"message": "We've detected instructions that may cause your application to perform malicious or unauthorized actions. Please acknowledge this warning if you'd like to proceed."
}
],
"status": "completed"
}
]
Du måste skicka tillbaka säkerhetskontrollerna som acknowledged_safety_checks i nästa förfrågan för att kunna fortsätta.
"input":[
{
"type": "computer_call_output",
"call_id": "<call_id>",
"acknowledged_safety_checks": [
{
"id": "<safety_check_id>",
"code": "malicious_instructions",
"message": "We've detected instructions that may cause your application to perform malicious or unauthorized actions. Please acknowledge this warning if you'd like to proceed."
}
],
"output": {
"type": "computer_screenshot",
"image_url": "<image_url>"
}
}
],
Hantering av säkerhetskontroller
I alla fall där pending_safety_checks returneras bör åtgärder överlämnas till slutanvändaren för att bekräfta korrekt modellbeteende och korrekthet.
-
malicious_instructionsochirrelevant_domain: Slutanvändare bör granska modellåtgärder och bekräfta att modellen beter sig som avsett. -
sensitive_domain: se till att en slutanvändare aktivt övervakar modellåtgärderna på dessa platser. Exakt implementering av det här "klockläget" kan variera beroende på applikation, men ett potentiellt exempel kan vara att samla in data om användarupplevelse på webbplatsen för att säkerställa att det finns ett aktivt slutanvändarengagemang med applikationen.
Dramatikerintegrering
I det här avsnittet tillhandahåller vi ett enkelt exempelskript som integrerar Azure OpenAI:s computer-use-preview modell med Playwright för att automatisera grundläggande webbläsarinteraktioner. Genom att kombinera modellen med Playwright kan modellen se webbläsarskärmen, fatta beslut och utföra åtgärder som att klicka, skriva och navigera på webbplatser. Du bör vara försiktig när du kör den här exempelkoden. Den här koden är utformad för att köras lokalt men bör endast köras i en testmiljö. Använd en människa för att bekräfta beslut och ge inte modellen åtkomst till känsliga data.
Först måste du installera Python-biblioteket för Playwright.
pip install playwright
När paketet har installerats måste du också köra
playwright install
Importer och konfiguration
Först importerar vi nödvändiga bibliotek och definierar våra konfigurationsparametrar. Eftersom vi använder asyncio kommer vi att köra den här koden utanför Jupyter Notebooks. Vi går först igenom koden i segment och visar sedan hur du använder den.
import os
import asyncio
import base64
from openai import OpenAI
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from playwright.async_api import async_playwright, TimeoutError
token_provider = get_bearer_token_provider(
DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default"
)
# Configuration
BASE_URL = "https://YOUR-RESOURCE-NAME.openai.azure.com/openai/v1/"
MODEL = "computer-use-preview" # Set to model deployment name
DISPLAY_WIDTH = 1024
DISPLAY_HEIGHT = 768
ITERATIONS = 5 # Max number of iterations before returning control to human supervisor
Nyckelmappning för webbläsarinteraktion
Därefter konfigurerar vi mappningar för särskilda nycklar som modellen kan behöva skicka till Playwright. I slutändan utför modellen aldrig själva åtgärderna, den skickar representationer av kommandon och du måste ange det slutliga integrationsskiktet som kan ta dessa kommandon och köra dem i din valda miljö.
Det här är inte en fullständig lista över möjliga nyckelmappningar. Du kan expandera den här listan efter behov. Denna ordbok är specifik för att integrera modellen med Playwright. Om du integrerar modellen med ett alternativt bibliotek för att ge API-åtkomst till ditt operativsystems tangentbord/mus skulle du behöva ange en mappning som är specifik för biblioteket.
# Key mapping for special keys in Playwright
KEY_MAPPING = {
"/": "Slash", "\\": "Backslash", "alt": "Alt", "arrowdown": "ArrowDown",
"arrowleft": "ArrowLeft", "arrowright": "ArrowRight", "arrowup": "ArrowUp",
"backspace": "Backspace", "ctrl": "Control", "delete": "Delete",
"enter": "Enter", "esc": "Escape", "shift": "Shift", "space": " ",
"tab": "Tab", "win": "Meta", "cmd": "Meta", "super": "Meta", "option": "Alt"
}
Den här ordlistan översätter användarvänliga nyckelnamn till det format som förväntas av Playwrights tangentbords-API.
Koordinatverifieringsfunktion
För att säkerställa att alla musåtgärder som skickas från modellen håller sig inom webbläsarfönstrets gränser lägger vi till följande verktygsfunktion:
def validate_coordinates(x, y):
"""Ensure coordinates are within display bounds."""
return max(0, min(x, DISPLAY_WIDTH)), max(0, min(y, DISPLAY_HEIGHT))
Det här enkla verktyget försöker förhindra gränsöverskridande fel genom att begränsa koordinater till fönsterdimensionerna.
Hantering av åtgärder
Kärnan i vår webbläsarautomatisering är åtgärdshanteraren som bearbetar olika typer av användarinteraktioner och konverterar dem till åtgärder i webbläsaren.
async def handle_action(page, action):
"""Handle different action types from the model."""
action_type = action.type
if action_type == "drag":
print("Drag action is not supported in this implementation. Skipping.")
return
elif action_type == "click":
button = getattr(action, "button", "left")
# Validate coordinates
x, y = validate_coordinates(action.x, action.y)
print(f"\tAction: click at ({x}, {y}) with button '{button}'")
if button == "back":
await page.go_back()
elif button == "forward":
await page.go_forward()
elif button == "wheel":
await page.mouse.wheel(x, y)
else:
button_type = {"left": "left", "right": "right", "middle": "middle"}.get(button, "left")
await page.mouse.click(x, y, button=button_type)
try:
await page.wait_for_load_state("domcontentloaded", timeout=3000)
except TimeoutError:
pass
elif action_type == "double_click":
# Validate coordinates
x, y = validate_coordinates(action.x, action.y)
print(f"\tAction: double click at ({x}, {y})")
await page.mouse.dblclick(x, y)
elif action_type == "scroll":
scroll_x = getattr(action, "scroll_x", 0)
scroll_y = getattr(action, "scroll_y", 0)
# Validate coordinates
x, y = validate_coordinates(action.x, action.y)
print(f"\tAction: scroll at ({x}, {y}) with offsets ({scroll_x}, {scroll_y})")
await page.mouse.move(x, y)
await page.evaluate(f"window.scrollBy({{left: {scroll_x}, top: {scroll_y}, behavior: 'smooth'}});")
elif action_type == "keypress":
keys = getattr(action, "keys", [])
print(f"\tAction: keypress {keys}")
mapped_keys = [KEY_MAPPING.get(key.lower(), key) for key in keys]
if len(mapped_keys) > 1:
# For key combinations (like Ctrl+C)
for key in mapped_keys:
await page.keyboard.down(key)
await asyncio.sleep(0.1)
for key in reversed(mapped_keys):
await page.keyboard.up(key)
else:
for key in mapped_keys:
await page.keyboard.press(key)
elif action_type == "type":
text = getattr(action, "text", "")
print(f"\tAction: type text: {text}")
await page.keyboard.type(text, delay=20)
elif action_type == "wait":
ms = getattr(action, "ms", 1000)
print(f"\tAction: wait {ms}ms")
await asyncio.sleep(ms / 1000)
elif action_type == "screenshot":
print("\tAction: screenshot")
else:
print(f"\tUnrecognized action: {action_type}")
Den här funktionen försöker hantera olika typer av åtgärder. Vi måste översätta mellan kommandona som computer-use-preview kommer att generera och Playwright-biblioteket som kommer att utföra åtgärderna. Mer information finns i referensdokumentationen för ComputerAction.
Skärmdump
För att modellen ska kunna se vad den interagerar med, behöver modellen ett sätt att samla in skärmbilder. För den här koden använder vi Playwright för att avbilda skärmbilderna och vi begränsar vyn till enbart innehållet i webbläsarfönstret. Skärmbilden innehåller inte URL-fältet eller andra aspekter av webbläsarens GUI. Om du behöver modellen för att se utanför huvudwebbläsarens fönster kan du utöka modellen genom att skapa en egen skärmbildsfunktion.
async def take_screenshot(page):
"""Take a screenshot and return base64 encoding with caching for failures."""
global last_successful_screenshot
try:
screenshot_bytes = await page.screenshot(full_page=False)
last_successful_screenshot = base64.b64encode(screenshot_bytes).decode("utf-8")
return last_successful_screenshot
except Exception as e:
print(f"Screenshot failed: {e}")
print(f"Using cached screenshot from previous successful capture")
if last_successful_screenshot:
return last_successful_screenshot
Den här funktionen avbildar det aktuella webbläsartillståndet som en bild och returnerar det som en base64-kodad sträng som är redo att skickas till modellen. Vi kommer ständigt att göra detta i en loop efter varje steg så att modellen kan se om kommandot som den försökte köra lyckades eller inte, vilket sedan gör att den kan justeras baserat på innehållet i skärmbilden. Vi kan låta modellen avgöra om den behöver ta en skärmbild, men för enkelhetens skull framtvingar vi att en skärmbild tas för varje iteration.
Bearbetning av modellsvar
Den här funktionen bearbetar modellens svar och kör de begärda åtgärderna:
async def process_model_response(client, response, page, max_iterations=ITERATIONS):
"""Process the model's response and execute actions."""
for iteration in range(max_iterations):
if not hasattr(response, 'output') or not response.output:
print("No output from model.")
break
# Safely access response id
response_id = getattr(response, 'id', 'unknown')
print(f"\nIteration {iteration + 1} - Response ID: {response_id}\n")
# Print text responses and reasoning
for item in response.output:
# Handle text output
if hasattr(item, 'type') and item.type == "text":
print(f"\nModel message: {item.text}\n")
# Handle reasoning output
if hasattr(item, 'type') and item.type == "reasoning":
# Extract meaningful content from the reasoning
meaningful_content = []
if hasattr(item, 'summary') and item.summary:
for summary in item.summary:
# Handle different potential formats of summary content
if isinstance(summary, str) and summary.strip():
meaningful_content.append(summary)
elif hasattr(summary, 'text') and summary.text.strip():
meaningful_content.append(summary.text)
# Only print reasoning section if there's actual content
if meaningful_content:
print("=== Model Reasoning ===")
for idx, content in enumerate(meaningful_content, 1):
print(f"{content}")
print("=====================\n")
# Extract computer calls
computer_calls = [item for item in response.output
if hasattr(item, 'type') and item.type == "computer_call"]
if not computer_calls:
print("No computer call found in response. Reverting control to human operator")
break
computer_call = computer_calls[0]
if not hasattr(computer_call, 'call_id') or not hasattr(computer_call, 'action'):
print("Computer call is missing required attributes.")
break
call_id = computer_call.call_id
action = computer_call.action
# Handle safety checks
acknowledged_checks = []
if hasattr(computer_call, 'pending_safety_checks') and computer_call.pending_safety_checks:
pending_checks = computer_call.pending_safety_checks
print("\nSafety checks required:")
for check in pending_checks:
print(f"- {check.code}: {check.message}")
if input("\nDo you want to proceed? (y/n): ").lower() != 'y':
print("Operation cancelled by user.")
break
acknowledged_checks = pending_checks
# Execute the action
try:
await page.bring_to_front()
await handle_action(page, action)
# Check if a new page was created after the action
if action.type in ["click"]:
await asyncio.sleep(1.5)
# Get all pages in the context
all_pages = page.context.pages
# If we have multiple pages, check if there's a newer one
if len(all_pages) > 1:
newest_page = all_pages[-1] # Last page is usually the newest
if newest_page != page and newest_page.url not in ["about:blank", ""]:
print(f"\tSwitching to new tab: {newest_page.url}")
page = newest_page # Update our page reference
elif action.type != "wait":
await asyncio.sleep(0.5)
except Exception as e:
print(f"Error handling action {action.type}: {e}")
import traceback
traceback.print_exc()
# Take a screenshot after the action
screenshot_base64 = await take_screenshot(page)
print("\tNew screenshot taken")
# Prepare input for the next request
input_content = [{
"type": "computer_call_output",
"call_id": call_id,
"output": {
"type": "input_image",
"image_url": f"data:image/png;base64,{screenshot_base64}"
}
}]
# Add acknowledged safety checks if any
if acknowledged_checks:
acknowledged_checks_dicts = []
for check in acknowledged_checks:
acknowledged_checks_dicts.append({
"id": check.id,
"code": check.code,
"message": check.message
})
input_content[0]["acknowledged_safety_checks"] = acknowledged_checks_dicts
# Add current URL for context
try:
current_url = page.url
if current_url and current_url != "about:blank":
input_content[0]["current_url"] = current_url
print(f"\tCurrent URL: {current_url}")
except Exception as e:
print(f"Error getting URL: {e}")
# Send the screenshot back for the next step
try:
response = client.responses.create(
model=MODEL,
previous_response_id=response_id,
tools=[{
"type": "computer_use_preview",
"display_width": DISPLAY_WIDTH,
"display_height": DISPLAY_HEIGHT,
"environment": "browser"
}],
input=input_content,
truncation="auto"
)
print("\tModel processing screenshot")
except Exception as e:
print(f"Error in API call: {e}")
import traceback
traceback.print_exc()
break
if iteration >= max_iterations - 1:
print("Reached maximum number of iterations. Stopping.")
I det här avsnittet har vi lagt till kod som:
- Extraherar och visar text och resonemang från modellen.
- Bearbetar datoråtgärdsanrop.
- Hanterar potentiella säkerhetskontroller som kräver användarbekräftelse.
- Kör den begärda åtgärden.
- Tar en ny skärmbild.
- Skickar tillbaka det uppdaterade tillståndet till modellen och definierar
ComputerTool. - Upprepar den här processen för flera iterationer.
Huvudfunktion
Huvudfunktionen samordnar hela processen:
# Initialize OpenAI client
client = OpenAI(
base_url=BASE_URL,
api_key=token_provider,
)
# Initialize Playwright
async with async_playwright() as playwright:
browser = await playwright.chromium.launch(
headless=False,
args=[f"--window-size={DISPLAY_WIDTH},{DISPLAY_HEIGHT}", "--disable-extensions"]
)
context = await browser.new_context(
viewport={"width": DISPLAY_WIDTH, "height": DISPLAY_HEIGHT},
accept_downloads=True
)
page = await context.new_page()
# Navigate to starting page
await page.goto("https://www.bing.com", wait_until="domcontentloaded")
print("Browser initialized to Bing.com")
# Main interaction loop
try:
while True:
print("\n" + "="*50)
user_input = input("Enter a task to perform (or 'exit' to quit): ")
if user_input.lower() in ('exit', 'quit'):
break
if not user_input.strip():
continue
# Take initial screenshot
screenshot_base64 = await take_screenshot(page)
print("\nTake initial screenshot")
# Initial request to the model
response = client.responses.create(
model=MODEL,
tools=[{
"type": "computer_use_preview",
"display_width": DISPLAY_WIDTH,
"display_height": DISPLAY_HEIGHT,
"environment": "browser"
}],
instructions = "You are an AI agent with the ability to control a browser. You can control the keyboard and mouse. You take a screenshot after each action to check if your action was successful. Once you have completed the requested task you should stop running and pass back control to your human operator.",
input=[{
"role": "user",
"content": [{
"type": "input_text",
"text": user_input
}, {
"type": "input_image",
"image_url": f"data:image/png;base64,{screenshot_base64}"
}]
}],
reasoning={"generate_summary": "concise"},
truncation="auto"
)
print("\nSending model initial screenshot and instructions")
# Process model actions
await process_model_response(client, response, page)
except Exception as e:
print(f"An error occurred: {e}")
import traceback
traceback.print_exc()
finally:
# Close browser
await context.close()
await browser.close()
print("Browser closed.")
if __name__ == "__main__":
asyncio.run(main())
Huvudfunktionen:
- Initierar OpenAI-klienten.
- Konfigurerar webbläsaren Playwright.
- Börjar vid Bing.com.
- Går in i en loop för att acceptera användaruppgifter.
- Avbildar det inledande tillståndet.
- Skickar uppgiften och skärmbilden till modellen.
- Bearbetar modellens svar.
- Upprepas tills användaren avslutar.
- Ser till att webbläsaren stängs korrekt.
Fullständigt skript
Försiktighet
Den här koden är experimentell och endast i demonstrationssyfte. Det är bara avsett att illustrera det grundläggande flödet för svars-API:et computer-use-preview och modellen. Även om du kan köra den här koden på din lokala dator rekommenderar vi starkt att du kör den här koden på en virtuell dator med låg behörighet utan åtkomst till känsliga data. Den här koden är endast avsedd för grundläggande testning.
import os
import asyncio
import base64
from openai import OpenAI
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from playwright.async_api import async_playwright, TimeoutError
token_provider = get_bearer_token_provider(
DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default"
)
# Configuration
BASE_URL = "https://YOUR-RESOURCE-NAME.openai.azure.com/openai/v1/"
MODEL = "computer-use-preview"
DISPLAY_WIDTH = 1024
DISPLAY_HEIGHT = 768
ITERATIONS = 5 # Max number of iterations before forcing the model to return control to the human supervisor
# Key mapping for special keys in Playwright
KEY_MAPPING = {
"/": "Slash", "\\": "Backslash", "alt": "Alt", "arrowdown": "ArrowDown",
"arrowleft": "ArrowLeft", "arrowright": "ArrowRight", "arrowup": "ArrowUp",
"backspace": "Backspace", "ctrl": "Control", "delete": "Delete",
"enter": "Enter", "esc": "Escape", "shift": "Shift", "space": " ",
"tab": "Tab", "win": "Meta", "cmd": "Meta", "super": "Meta", "option": "Alt"
}
def validate_coordinates(x, y):
"""Ensure coordinates are within display bounds."""
return max(0, min(x, DISPLAY_WIDTH)), max(0, min(y, DISPLAY_HEIGHT))
async def handle_action(page, action):
"""Handle different action types from the model."""
action_type = action.type
if action_type == "drag":
print("Drag action is not supported in this implementation. Skipping.")
return
elif action_type == "click":
button = getattr(action, "button", "left")
# Validate coordinates
x, y = validate_coordinates(action.x, action.y)
print(f"\tAction: click at ({x}, {y}) with button '{button}'")
if button == "back":
await page.go_back()
elif button == "forward":
await page.go_forward()
elif button == "wheel":
await page.mouse.wheel(x, y)
else:
button_type = {"left": "left", "right": "right", "middle": "middle"}.get(button, "left")
await page.mouse.click(x, y, button=button_type)
try:
await page.wait_for_load_state("domcontentloaded", timeout=3000)
except TimeoutError:
pass
elif action_type == "double_click":
# Validate coordinates
x, y = validate_coordinates(action.x, action.y)
print(f"\tAction: double click at ({x}, {y})")
await page.mouse.dblclick(x, y)
elif action_type == "scroll":
scroll_x = getattr(action, "scroll_x", 0)
scroll_y = getattr(action, "scroll_y", 0)
# Validate coordinates
x, y = validate_coordinates(action.x, action.y)
print(f"\tAction: scroll at ({x}, {y}) with offsets ({scroll_x}, {scroll_y})")
await page.mouse.move(x, y)
await page.evaluate(f"window.scrollBy({{left: {scroll_x}, top: {scroll_y}, behavior: 'smooth'}});")
elif action_type == "keypress":
keys = getattr(action, "keys", [])
print(f"\tAction: keypress {keys}")
mapped_keys = [KEY_MAPPING.get(key.lower(), key) for key in keys]
if len(mapped_keys) > 1:
# For key combinations (like Ctrl+C)
for key in mapped_keys:
await page.keyboard.down(key)
await asyncio.sleep(0.1)
for key in reversed(mapped_keys):
await page.keyboard.up(key)
else:
for key in mapped_keys:
await page.keyboard.press(key)
elif action_type == "type":
text = getattr(action, "text", "")
print(f"\tAction: type text: {text}")
await page.keyboard.type(text, delay=20)
elif action_type == "wait":
ms = getattr(action, "ms", 1000)
print(f"\tAction: wait {ms}ms")
await asyncio.sleep(ms / 1000)
elif action_type == "screenshot":
print("\tAction: screenshot")
else:
print(f"\tUnrecognized action: {action_type}")
async def take_screenshot(page):
"""Take a screenshot and return base64 encoding with caching for failures."""
global last_successful_screenshot
try:
screenshot_bytes = await page.screenshot(full_page=False)
last_successful_screenshot = base64.b64encode(screenshot_bytes).decode("utf-8")
return last_successful_screenshot
except Exception as e:
print(f"Screenshot failed: {e}")
print(f"Using cached screenshot from previous successful capture")
if last_successful_screenshot:
return last_successful_screenshot
async def process_model_response(client, response, page, max_iterations=ITERATIONS):
"""Process the model's response and execute actions."""
for iteration in range(max_iterations):
if not hasattr(response, 'output') or not response.output:
print("No output from model.")
break
# Safely access response id
response_id = getattr(response, 'id', 'unknown')
print(f"\nIteration {iteration + 1} - Response ID: {response_id}\n")
# Print text responses and reasoning
for item in response.output:
# Handle text output
if hasattr(item, 'type') and item.type == "text":
print(f"\nModel message: {item.text}\n")
# Handle reasoning output
if hasattr(item, 'type') and item.type == "reasoning":
# Extract meaningful content from the reasoning
meaningful_content = []
if hasattr(item, 'summary') and item.summary:
for summary in item.summary:
# Handle different potential formats of summary content
if isinstance(summary, str) and summary.strip():
meaningful_content.append(summary)
elif hasattr(summary, 'text') and summary.text.strip():
meaningful_content.append(summary.text)
# Only print reasoning section if there's actual content
if meaningful_content:
print("=== Model Reasoning ===")
for idx, content in enumerate(meaningful_content, 1):
print(f"{content}")
print("=====================\n")
# Extract computer calls
computer_calls = [item for item in response.output
if hasattr(item, 'type') and item.type == "computer_call"]
if not computer_calls:
print("No computer call found in response. Reverting control to human supervisor")
break
computer_call = computer_calls[0]
if not hasattr(computer_call, 'call_id') or not hasattr(computer_call, 'action'):
print("Computer call is missing required attributes.")
break
call_id = computer_call.call_id
action = computer_call.action
# Handle safety checks
acknowledged_checks = []
if hasattr(computer_call, 'pending_safety_checks') and computer_call.pending_safety_checks:
pending_checks = computer_call.pending_safety_checks
print("\nSafety checks required:")
for check in pending_checks:
print(f"- {check.code}: {check.message}")
if input("\nDo you want to proceed? (y/n): ").lower() != 'y':
print("Operation cancelled by user.")
break
acknowledged_checks = pending_checks
# Execute the action
try:
await page.bring_to_front()
await handle_action(page, action)
# Check if a new page was created after the action
if action.type in ["click"]:
await asyncio.sleep(1.5)
# Get all pages in the context
all_pages = page.context.pages
# If we have multiple pages, check if there's a newer one
if len(all_pages) > 1:
newest_page = all_pages[-1] # Last page is usually the newest
if newest_page != page and newest_page.url not in ["about:blank", ""]:
print(f"\tSwitching to new tab: {newest_page.url}")
page = newest_page # Update our page reference
elif action.type != "wait":
await asyncio.sleep(0.5)
except Exception as e:
print(f"Error handling action {action.type}: {e}")
import traceback
traceback.print_exc()
# Take a screenshot after the action
screenshot_base64 = await take_screenshot(page)
print("\tNew screenshot taken")
# Prepare input for the next request
input_content = [{
"type": "computer_call_output",
"call_id": call_id,
"output": {
"type": "input_image",
"image_url": f"data:image/png;base64,{screenshot_base64}"
}
}]
# Add acknowledged safety checks if any
if acknowledged_checks:
acknowledged_checks_dicts = []
for check in acknowledged_checks:
acknowledged_checks_dicts.append({
"id": check.id,
"code": check.code,
"message": check.message
})
input_content[0]["acknowledged_safety_checks"] = acknowledged_checks_dicts
# Add current URL for context
try:
current_url = page.url
if current_url and current_url != "about:blank":
input_content[0]["current_url"] = current_url
print(f"\tCurrent URL: {current_url}")
except Exception as e:
print(f"Error getting URL: {e}")
# Send the screenshot back for the next step
try:
response = client.responses.create(
model=MODEL,
previous_response_id=response_id,
tools=[{
"type": "computer_use_preview",
"display_width": DISPLAY_WIDTH,
"display_height": DISPLAY_HEIGHT,
"environment": "browser"
}],
input=input_content,
truncation="auto"
)
print("\tModel processing screenshot")
except Exception as e:
print(f"Error in API call: {e}")
import traceback
traceback.print_exc()
break
if iteration >= max_iterations - 1:
print("Reached maximum number of iterations. Stopping.")
async def main():
# Initialize OpenAI client
client = OpenAI(
base_url=BASE_URL,
api_key=token_provider
)
# Initialize Playwright
async with async_playwright() as playwright:
browser = await playwright.chromium.launch(
headless=False,
args=[f"--window-size={DISPLAY_WIDTH},{DISPLAY_HEIGHT}", "--disable-extensions"]
)
context = await browser.new_context(
viewport={"width": DISPLAY_WIDTH, "height": DISPLAY_HEIGHT},
accept_downloads=True
)
page = await context.new_page()
# Navigate to starting page
await page.goto("https://www.bing.com", wait_until="domcontentloaded")
print("Browser initialized to Bing.com")
# Main interaction loop
try:
while True:
print("\n" + "="*50)
user_input = input("Enter a task to perform (or 'exit' to quit): ")
if user_input.lower() in ('exit', 'quit'):
break
if not user_input.strip():
continue
# Take initial screenshot
screenshot_base64 = await take_screenshot(page)
print("\nTake initial screenshot")
# Initial request to the model
response = client.responses.create(
model=MODEL,
tools=[{
"type": "computer_use_preview",
"display_width": DISPLAY_WIDTH,
"display_height": DISPLAY_HEIGHT,
"environment": "browser"
}],
instructions = "You are an AI agent with the ability to control a browser. You can control the keyboard and mouse. You take a screenshot after each action to check if your action was successful. Once you have completed the requested task you should stop running and pass back control to your human supervisor.",
input=[{
"role": "user",
"content": [{
"type": "input_text",
"text": user_input
}, {
"type": "input_image",
"image_url": f"data:image/png;base64,{screenshot_base64}"
}]
}],
reasoning={"generate_summary": "concise"},
truncation="auto"
)
print("\nSending model initial screenshot and instructions")
# Process model actions
await process_model_response(client, response, page)
except Exception as e:
print(f"An error occurred: {e}")
import traceback
traceback.print_exc()
finally:
# Close browser
await context.close()
await browser.close()
print("Browser closed.")
if __name__ == "__main__":
asyncio.run(main())