Poznámka
Přístup k této stránce vyžaduje autorizaci. Můžete se zkusit přihlásit nebo změnit adresáře.
Přístup k této stránce vyžaduje autorizaci. Můžete zkusit změnit adresáře.
V tomto článku se dozvíte, jak pracovat s používáním počítače v Azure OpenAI. Použití počítače je specializovaný nástroj AI, který používá specializovaný model, který může provádět úlohy interakcí s počítačovými systémy a aplikacemi prostřednictvím jejich uživatelských rozhraní. Pomocí funkce Použití počítače můžete vytvořit agenta, který dokáže zpracovávat složité úlohy a rozhodovat se interpretací vizuálních prvků a prováděním akcí na základě obsahu na obrazovce.
Použití počítače poskytuje:
- Autonomní navigace: Například otevře aplikace, klikne na tlačítka, vyplní formuláře a přejde na vícestráňové pracovní postupy.
- Dynamická adaptace: Interpretuje změny uživatelského rozhraní a odpovídajícím způsobem upravuje akce.
- Provádění úloh mezi aplikacemi: Funguje napříč webovými a desktopovými aplikacemi.
- Rozhraní přirozeného jazyka: Uživatelé mohou popsat úlohu v prostém jazyce a model Použití počítače určuje správné interakce uživatelského rozhraní ke spuštění.
Žádost o přístup
Pro přístup k computer-use-preview modelu se vyžaduje registrace a přístup se udělí na základě kritérií způsobilosti microsoftu. Zákazníci, kteří mají přístup k jiným modelům omezeného přístupu, budou muset požádat o přístup k tomuto modelu.
Žádost o přístup: aplikace modelu omezeného přístupu ve verzi pro počítače - náhled
Po udělení přístupu budete muset pro model vytvořit nasazení.
Regionální podpora
Použití počítače je k dispozici v následujících oblastech:
eastus2swedencentralsouthindia
Odeslání volání rozhraní API do modelu používání počítače pomocí response API
K nástroji pro užívání počítače se přistupuje prostřednictvím rozhraní API pro odpovědi. Nástroj funguje v nepřetržité smyčce, která odesílá akce, jako je psaní textu nebo provádění kliknutí. Váš kód tyto akce provede na počítači a odešle snímky obrazovky výsledku do modelu.
Tímto způsobem váš kód simuluje akce člověka pomocí počítačového rozhraní, zatímco model pomocí snímků obrazovky rozumí stavu prostředí a navrhuje další akce.
Následující příklady ukazují základní volání rozhraní API.
Poznámka:
Potřebujete mít prostředek Azure OpenAI s nasazeným computer-use-preview modelem v podporované oblasti.
Pokud chcete odesílat požadavky, budete muset nainstalovat následující balíčky Pythonu.
pip install openai
pip install azure-identity
import os
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import AzureOpenAI
#from openai import OpenAI
token_provider = get_bearer_token_provider(DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default")
client = AzureOpenAI(
base_url = "https://YOUR-RESOURCE-NAME.openai.azure.com/openai/v1/",
azure_ad_token_provider=token_provider,
api_version="preview"
)
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)
Výstup
[
ResponseComputerToolCall(
id='cu_67d841873c1081908bfc88b90a8555e0',
action=ActionScreenshot(type='screenshot'),
call_id='call_wwEnfFDqQr1Z4Edk62Fyo7Nh',
pending_safety_checks=[],
status='completed',
type='computer_call'
)
]
Po odeslání počátečního požadavku rozhraní API provedete smyčku, ve které se v kódu aplikace provede zadaná akce, a odešlete snímek obrazovky s jednotlivými kroky, aby model mohl vyhodnotit aktualizovaný stav prostředí.
## 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"
)
Principy integrace použití počítače
Při práci s nástrojem Použití počítače byste obvykle provedli následující kroky, abyste ho integrovali do své aplikace.
- Odešlete do modelu požadavek, který zahrnuje volání nástroje pro využití počítače, velikost zobrazení a prostředí. Do prvního požadavku rozhraní API můžete zahrnout také snímek obrazovky s počátečním stavem prostředí.
- Obdrží odpověď od modelu. Pokud odpověď obsahuje
actionpoložky, obsahují tyto položky navrhované akce, které budou pokračovat směrem k zadanému cíli. Příkladem může býtscreenshotakce, aby model mohl vyhodnotit aktuální stav s aktualizovaným snímkem obrazovky neboclicksouřadnicemi X/Y označující, kam se má myš přesunout. - Spusťte akci pomocí kódu aplikace v počítači nebo v prostředí prohlížeče.
- Po provedení akce zachyťte aktualizovaný stav prostředí jako snímek obrazovky.
- Odešlete nový požadavek s aktualizovaným stavem jako
computer_call_outputa opakujte tuto smyčku, dokud model nepřestane žádat o akce nebo se nerozhodnete zastavit.
Zpracování historie konverzací
Pomocí parametru previous_response_id můžete propojit aktuální požadavek s předchozí odpovědí. Tento parametr se doporučuje, pokud nechcete spravovat historii konverzací.
Pokud tento parametr nepoužíváte, nezapomeňte do pole vstupů zahrnout všechny položky vrácené ve výstupu odpovědi předchozího požadavku. To zahrnuje odůvodnění položek, pokud jsou k dispozici.
Bezpečnostní kontroly
Rozhraní API má kontroly zabezpečení, které pomáhají chránit před útoky injekce a chybami modelu. Mezi tyto kontroly patří:
- Detekce škodlivých instrukcí: Systém vyhodnotí obrázek snímku obrazovky a zkontroluje, jestli obsahuje nežádoucí obsah, který by mohl změnit chování modelu.
-
Irelevantní detekce domény: Systém vyhodnotí
current_url(pokud je k dispozici) a zkontroluje, jestli je aktuální doména považována za relevantní vzhledem k historii konverzací. -
Detekce citlivých domén: Systém zkontroluje
current_url(pokud je k dispozici) a zobrazí upozornění, když zjistí, že je uživatel v citlivé doméně.
Pokud se aktivuje jedna nebo více výše uvedených kontrol, vyvolá se bezpečnostní kontrola, když model vrátí následující computer_call s parametrem 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"
}
]
Abyste mohli pokračovat, musíte bezpečnostní kontroly předat zpět jako acknowledged_safety_checks v další žádosti.
"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>"
}
}
],
Řízení bezpečnostní kontroly
Ve všech případech, kdy pending_safety_checks se vrátí, by měly být akce předány koncovému uživateli, aby potvrdil správné chování a přesnost modelu.
-
malicious_instructionsairrelevant_domain: Koncoví uživatelé by měli zkontrolovat akce modelu a ověřit, že se model chová podle očekávání. -
sensitive_domain: Zajistěte, aby koncový uživatel aktivně monitoruje akce modelu na těchto webech. Přesná implementace tohoto "režimu sledování" se může lišit podle aplikace, ale potenciálním příkladem může být shromažďování dat uživatelského zobrazení na webu, aby se zajistilo, že je aktivní zapojení koncových uživatelů s aplikací.
Integrace Playwright
V této části poskytujeme jednoduchý ukázkový skript, který integruje model Azure OpenAI computer-use-preview s playwrightem pro automatizaci základních interakcí prohlížeče. Kombinace modelu s Playwright umožňuje modelu zobrazit obrazovku prohlížeče, rozhodovat se a provádět akce, jako je kliknutí, psaní a navigace na webech. Při spuštění tohoto ukázku kódu byste měli postupovat opatrně. Tento kód je navržený tak, aby byl spuštěn místně, ale měl by být spuštěn pouze v testovacím prostředí. Pomocí člověka potvrďte rozhodnutí a neudělte modelu přístup k citlivým datům.
Nejdřív budete muset nainstalovat knihovnu Pythonu pro Playwright.
pip install playwright
Po instalaci balíčku budete také muset spustit
playwright install
Importy a konfigurace
Nejprve naimportujeme potřebné knihovny a definujeme parametry konfigurace. Vzhledem k tomu, že používáme asyncio , budeme tento kód spouštět mimo poznámkové bloky Jupyter. Nejprve si projdeme kód v blocích dat a ukážeme si, jak ho použít.
import os
import asyncio
import base64
from openai import AzureOpenAI
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
API_VERSION = "preview" #Use this API version or later
ITERATIONS = 5 # Max number of iterations before returning control to human supervisor
Mapování klíčů pro interakci s prohlížečem
Dále jsme nastavili mapování pro speciální klíče, které model může potřebovat předat Playwright. V konečném důsledku model nikdy neprovádí samotné akce, předává reprezentaci příkazů a musíte poskytnout konečnou integrační vrstvu, která může tyto příkazy provést a spouštět je ve zvoleném prostředí.
Nejedná se o vyčerpávající seznam možných mapování kláves. Podle potřeby můžete tento seznam rozbalit. Tento slovník je specifický pro integraci modelu se službou Playwright. Pokud byste model integrovali s alternativní knihovnou, která poskytuje přístup rozhraní API k vašim operačním systémům pomocí klávesnice nebo myši, museli byste poskytnout mapování specifické pro tuto knihovnu.
# 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"
}
Tento slovník překládá uživatelsky přívětivé názvy kláves do formátu očekávaného rozhraním API klávesnice Playwright.
Funkce ověřování souřadnic
Abychom měli jistotu, že všechny akce myši předávané z modelu zůstanou v hranicích okna prohlížeče, přidáme následující funkci nástroje:
def validate_coordinates(x, y):
"""Ensure coordinates are within display bounds."""
return max(0, min(x, DISPLAY_WIDTH)), max(0, min(y, DISPLAY_HEIGHT))
Tento jednoduchý nástroj se pokusí zabránit chybám mimo hranice tím, že omezí souřadnice na rozměry okna.
Zpracování akcí
Jádrem automatizace prohlížeče je obslužná rutina akcí, která zpracovává různé typy interakcí uživatelů a převádí je na akce v prohlížeči.
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}")
Tato funkce se pokusí zpracovat různé typy akcí. Musíme přeložit příkazy, které computer-use-preview vygeneruje, a knihovnu Playwright, která vykoná akce. Další informace najdete v referenční dokumentaci pro ComputerAction.
Snímek obrazovky
Aby model mohl vidět, s čím interaguje, potřebuje způsob, jak zachytit snímky obrazovky. Pro tento kód používáme Playwright k zachycení snímků obrazovky a omezujeme zobrazení jenom na obsah v okně prohlížeče. Snímek obrazovky nebude obsahovat panel adres URL ani jiné aspekty grafického uživatelského rozhraní prohlížeče. Pokud potřebujete, aby se model zobrazil mimo hlavní okno prohlížeče, můžete model rozšířit vytvořením vlastní funkce snímku obrazovky.
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
Tato funkce zachytí aktuální stav prohlížeče jako obrázek a vrátí ho jako řetězec kódovaný v base64, který je připravený k odeslání do modelu. Po každém kroku to budeme neustále provádět ve smyčce, což modelu umožní zjistit, jestli byl příkaz, který se pokusil provést, úspěšný nebo ne, a pak ho umožní upravit na základě obsahu snímku obrazovky. Model bychom mohli nechat rozhodnout, jestli potřebuje pořídit snímek obrazovky, ale pro jednoduchost vynutíme pořízení snímku obrazovky pro každou iteraci.
Zpracování odpovědi modelu
Tato funkce zpracuje odpovědi modelu a provede požadované akce:
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.")
V této části jsme přidali kód, který:
- Extrahuje a zobrazí text a odůvodnění z modelu.
- Zpracovává počítačové akční volání.
- Zpracovává potenciální bezpečnostní kontroly vyžadující potvrzení uživatele.
- Provede požadovanou akci.
- Zachytí nový snímek obrazovky.
- Odešle aktualizovaný stav zpět do modelu a definuje
ComputerTool. - Tento proces opakuje pro více iterací.
Hlavní funkce
Hlavní funkce koordinuje celý proces:
# Initialize OpenAI client
client = AzureOpenAI(
base_url=BASE_URL,
azure_ad_token_provider=token_provider,
api_version=API_VERSION
)
# 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())
Hlavní funkce:
- Inicializuje klienta AzureOpenAI.
- Nastavení prohlížeče Playwright.
- Začíná na Bing.com.
- Vstoupí do smyčky pro přijetí úkolů od uživatele.
- Zaznamená počáteční stav.
- Odešle úkol a snímek obrazovky k modelu.
- Zpracovává odpověď modelu.
- Opakuje se, dokud uživatel neukončí proces.
- Zajišťuje, že je prohlížeč řádně zavřený.
Kompletní skript
Upozornění
Tento kód je experimentální a pouze pro demonstrační účely. Účelem je znázornit pouze základní tok rozhraní API odpovědí a computer-use-preview modelu. I když tento kód můžete spustit na místním počítači, důrazně doporučujeme tento kód spustit na virtuálním počítači s nízkými oprávněními bez přístupu k citlivým datům. Tento kód je určen pouze pro základní účely testování.
import os
import asyncio
import base64
from openai import AzureOpenAI
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
API_VERSION = "preview"
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 = AzureOpenAI(
base_url=BASE_URL,
azure_ad_token_provider=token_provider,
api_version=API_VERSION
)
# 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())