Hinweis
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, sich anzumelden oder das Verzeichnis zu wechseln.
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, das Verzeichnis zu wechseln.
In diesem Artikel erfahren Sie, wie Sie mit der Nutzung des Computers in Azure OpenAI umgehen. Computer Use ist ein spezielles KI-Tool, das ein spezielles Modell verwendet, das Aufgaben ausführen kann, indem sie mit Computersystemen und Anwendungen über ihre UIs interagieren. Mithilfe der Computerverwendung können Sie einen Agent erstellen, der komplexe Aufgaben verarbeiten und Entscheidungen treffen kann, indem Sie visuelle Elemente interpretieren und aktionen basierend auf Bildschirminhalten ergreifen.
Die Computerverwendung bietet Folgendes:
- Autonome Navigation: Öffnet z. B. Anwendungen, klickt auf Schaltflächen, füllt Formulare aus und navigiert in mehrseitigen Workflows.
- Dynamische Anpassung: Interpretiert Ui-Änderungen und passt Aktionen entsprechend an.
- Anwendungsübergreifende Aufgabenausführung: Arbeitet über webbasierte und Desktopanwendungen hinweg.
- Benutzeroberfläche für natürliche Sprache: Benutzer können eine Aufgabe in einfacher Sprache beschreiben, und das Computer use-Modell bestimmt die richtigen Benutzeroberflächeninteraktionen, die ausgeführt werden sollen.
Zugriff anfordern
Für den Zugriff auf das computer-use-preview Modell ist die Registrierung erforderlich, und der Zugriff wird basierend auf den Berechtigungskriterien von Microsoft gewährt. Kunden, die Zugriff auf andere Modelle mit eingeschränktem Zugriff haben, müssen weiterhin Zugriff für dieses Modell anfordern.
Anfordern des Zugriffs: Anwendung für eingeschränkten Zugriff auf computer-use-preview
Nachdem der Zugriff gewährt wurde, müssen Sie eine Bereitstellung für das Modell erstellen.
Regionsunterstützung
Die Computerverwendung ist in den folgenden Regionen verfügbar:
eastus2swedencentralsouthindia
Senden eines API-Aufrufs an das Computerverwendungsmodell mithilfe der Antwort-API
Der Zugriff auf das Tool "Computerverwendung" erfolgt über die Antwort-API. Das Tool wird in einer fortlaufenden Schleife ausgeführt, die Aktionen sendet, z. B. Text eingeben oder einen Klick ausführen. Ihr Code führt diese Aktionen auf einem Computer aus und sendet Screenshots des Ergebnisses an das Modell.
Auf diese Weise simuliert Ihr Code die Aktionen eines Menschen mithilfe einer Computerschnittstelle, während das Modell die Screenshots verwendet, um den Zustand der Umgebung zu verstehen und nächste Aktionen vorzuschlagen.
Die folgenden Beispiele zeigen einen einfachen API-Aufruf.
Hinweis
Sie benötigen eine Azure OpenAI-Ressource mit einer computer-use-preview Modellbereitstellung in einer unterstützten Region.
Zum Senden von Anforderungen müssen Sie die folgenden Python-Pakete installieren.
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)
Ausgabe
[
ResponseComputerToolCall(
id='cu_67d841873c1081908bfc88b90a8555e0',
action=ActionScreenshot(type='screenshot'),
call_id='call_wwEnfFDqQr1Z4Edk62Fyo7Nh',
pending_safety_checks=[],
status='completed',
type='computer_call'
)
]
Sobald die anfängliche API-Anforderung gesendet wurde, führen Sie eine Schleife aus, in der die angegebene Aktion im Anwendungscode ausgeführt wird, und senden einen Screenshot mit jeder Wendung, damit das Modell den aktualisierten Zustand der Umgebung auswerten kann.
## 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"
)
Grundlegendes zur Integration der Computerverwendung
Wenn Sie mit dem Tool "Computerverwendung" arbeiten, führen Sie in der Regel folgendes aus, um es in Ihre Anwendung zu integrieren.
- Senden Sie eine Anforderung an das Modell, das einen Aufruf des Computerverwendungstools sowie die Anzeigegröße und -umgebung enthält. Sie können auch einen Screenshot des anfänglichen Zustands der Umgebung in die erste API-Anforderung einschließen.
- Empfangen einer Antwort vom Modell. Wenn die Antwort Elemente enthält
action, enthalten diese Elemente vorgeschlagene Aktionen, um fortschritte in Richtung des angegebenen Ziels zu erzielen. Eine Aktion kann beispielsweise seinscreenshot, damit das Modell den aktuellen Zustand mit einem aktualisierten Screenshot bewerten kann, oderclickmit X/Y-Koordinaten, die angeben, wo die Maus verschoben werden soll. - Führen Sie die Aktion mithilfe des Anwendungscodes auf Ihrem Computer oder in Ihrer Browserumgebung aus.
- Erfassen Sie nach dem Ausführen der Aktion den aktualisierten Zustand der Umgebung als Screenshot.
- Senden Sie eine neue Anforderung mit dem aktualisierten Zustand als ,
computer_call_outputund wiederholen Sie diese Schleife, bis das Modell keine Aktionen anfordert, oder Sie entscheiden, den Vorgang zu beenden.
Aufgezeichnete Unterhaltungen verarbeiten
Sie können den previous_response_id Parameter verwenden, um die aktuelle Anforderung mit der vorherigen Antwort zu verknüpfen. Sie sollten diesen Parameter verwenden, wenn Sie aufgezeichnete Unterhaltungen nicht verwalten möchten.
Wenn Sie diesen Parameter nicht verwenden, sollten Sie sicherstellen, dass alle Elemente, die in der Antwortausgabe der vorherigen Anforderung in Ihr Eingabearray zurückgegeben werden, einbezogen werden. Dies schließt die Begründung von Elementen ein, falls vorhanden.
Sicherheitskontrollen
Die API verfügt über Sicherheitsprüfungen, um den Schutz vor eingabeaufforderungs- und Modellfehlern zu gewährleisten. Zu diesen Prüfungen gehören:
- Erkennung bösartiger Anweisungen: Das System wertet das Screenshotbild aus und überprüft, ob es gegnerische Inhalte enthält, die das Verhalten des Modells ändern können.
- Irrelevante Domänenerkennung: Das System wertet die
current_url(sofern angegeben) aus und überprüft, ob die aktuelle Domäne aufgrund des Unterhaltungsverlaufs als relevant eingestuft wird. - Erkennung vertraulicher Domänen: Das System überprüft (
current_urlsofern angegeben) und löst eine Warnung aus, wenn erkannt wird, dass sich der Benutzer in einer vertraulichen Domäne befindet.
Wenn eine oder mehrere der oben genannten Überprüfungen ausgelöst werden, wird eine Sicherheitsüberprüfung aktiviert, sobald das Modell das nächste computer_call mit dem pending_safety_checks-Parameter zurückgibt.
"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"
}
]
Sie müssen die Sicherheitsüberprüfungen bei der nächsten Anforderung als acknowledged_safety_checks zurückgeben, um fortzufahren.
"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>"
}
}
],
Verarbeitung der Sicherheitsüberprüfung
In allen Fällen, in denen pending_safety_checks zurückgegeben wird, sollten Aktionen an den Endbenutzer übergeben werden, um das richtige Modellverhalten und die Richtige Genauigkeit zu bestätigen.
malicious_instructionsundirrelevant_domain: Endbenutzer sollten Modellaktionen überprüfen und bestätigen, dass das Modell wie beabsichtigt verhält.sensitive_domain: Stellen Sie sicher, dass ein Endbenutzer die Modellaktionen auf diesen Websites aktiv überwacht. Die genaue Implementierung dieses "Überwachungsmodus" kann je nach Anwendung variieren, aber ein mögliches Beispiel könnte das Sammeln von Benutzereindruckdaten auf der Website sein, um sicherzustellen, dass eine aktive Endbenutzerbindung mit der Anwendung besteht.
Playwright-Integration
In diesem Abschnitt stellen wir ein einfaches Beispielskript bereit, das das Modell von computer-use-preview Azure OpenAI in Playwright integriert, um grundlegende Browserinteraktionen zu automatisieren. Durch die Kombination des Modells mit Playwright kann das Modell den Browserbildschirm anzeigen, Entscheidungen treffen und Aktionen wie Klicken, Eingeben und Navigieren in Websites ausführen. Beim Ausführen dieses Beispielcodes sollten Sie Vorsicht walten lassen. Dieser Code ist für die lokale Ausführung konzipiert, sollte jedoch nur in einer Testumgebung ausgeführt werden. Verwenden Sie einen Menschen, um Entscheidungen zu bestätigen und dem Modell keinen Zugriff auf vertrauliche Daten zu gewähren.
Zuerst müssen Sie die Python-Bibliothek für Playwright installieren.
pip install playwright
Nachdem das Paket installiert wurde, müssen Sie auch Folgendes ausführen:
playwright install
Importe und Konfiguration
Zunächst importieren wir die erforderlichen Bibliotheken und definieren unsere Konfigurationsparameter. Da wir asyncio verwenden, werden wir diesen Code außerhalb von Jupyter-Notizbüchern ausführen. Wir werden den Code zuerst in Abschnitten durchgehen und dann veranschaulichen, wie er verwendet wird.
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
Schlüsselzuordnung für Browserinteraktionen
Als Nächstes richten wir Zuordnungen für spezielle Schlüssel ein, die das Modell möglicherweise an Playwright übergeben muss. Letztendlich führt das Modell nie selbst Aktionen aus, er übergibt Darstellungen von Befehlen, und Sie müssen die endgültige Integrationsebene bereitstellen, die diese Befehle ausführen und in Ihrer ausgewählten Umgebung ausführen kann.
Dies ist keine vollständige Liste möglicher Schlüsselzuordnungen. Sie können diese Liste nach Bedarf erweitern. Dieses Wörterbuch ist spezifisch für die Integration des Modells in Playwright. Wenn Sie das Modell in eine alternative Bibliothek integrieren würden, um API-Zugriff auf die Tastatur/Maus des Betriebssystems bereitzustellen, müssten Sie eine zuordnungsspezifische Abbildung für diese Bibliothek gewährleisten.
# 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"
}
Dieses Wörterbuch übersetzt benutzerfreundliche Schlüsselnamen in das format, das von der Tastatur-API von Playwright erwartet wird.
Koordinatenüberprüfungsfunktion
Um sicherzustellen, dass alle Mausaktionen, die vom Modell übergeben werden, innerhalb der Browserfenstergrenzen bleiben, fügen wir die folgende Hilfsfunktion hinzu:
def validate_coordinates(x, y):
"""Ensure coordinates are within display bounds."""
return max(0, min(x, DISPLAY_WIDTH)), max(0, min(y, DISPLAY_HEIGHT))
Dieses einfache Hilfsprogramm versucht, Fehler außerhalb der Begrenzung zu vermeiden, indem es die Koordinaten an die Fensterabmessungen anpasst.
Aktionsverwaltung
Der Kern unserer Browserautomatisierung ist der Aktionshandler, der verschiedene Arten von Benutzerinteraktionen verarbeitet und in Aktionen im Browser konvertiert.
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}")
Diese Funktion versucht, verschiedene Arten von Aktionen zu behandeln. Wir müssen zwischen den Befehlen übersetzen, die von der computer-use-preview Datei generiert werden, und der Playwright-Bibliothek, die die Aktionen ausführt. Weitere Informationen finden Sie in der Referenzdokumentation für ComputerAction.
Screenshotaufnahme
Damit das Modell sehen kann, was mit dem Modell interagiert, muss eine Möglichkeit zum Erfassen von Screenshots erforderlich sein. Für diesen Code verwenden wir Playwright, um die Screenshots zu erfassen, und wir beschränken die Ansicht auf den Inhalt im Browserfenster. Der Screenshot enthält nicht die URL-Leiste oder andere Aspekte der Browser-GUI. Wenn Sie das Modell benötigen, um außerhalb des Hauptbrowserfensters zu sehen, können Sie das Modell erweitern, indem Sie eine eigene Screenshotfunktion erstellen.
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
Diese Funktion erfasst den aktuellen Browserstatus als Bild und gibt ihn als base64-codierte Zeichenfolge zurück, die an das Modell gesendet werden kann. Wir werden dies ständig in einer Schleife nach jedem Schritt ausführen, damit das Modell prüfen kann, ob der Befehl, den es auszuführen versuchte, erfolgreich war. Dadurch kann es Anpassungen basierend auf dem Inhalt des Screenshots vornehmen. Wir könnten das Modell entscheiden lassen, ob es einen Screenshot erstellen muss, aber aus Gründen der Einfachheit erzwingen wir, dass ein Screenshot für jede Iteration erstellt wird.
Modellantwortverarbeitung
Diese Funktion verarbeitet die Antworten des Modells und führt die angeforderten Aktionen aus:
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.")
In diesem Abschnitt haben wir Code hinzugefügt, der:
- Extrahiert und zeigt Text und Begründungen aus dem Modell an.
- Verarbeitet Computeraktionsaufrufe.
- Behandelt potenzielle Sicherheitsprüfungen, die eine Benutzerbestätigung erfordern.
- Führt die angeforderte Aktion aus.
- Erfasst einen neuen Screenshot.
- Sendet den aktualisierten Zustand zurück an das Modell und definiert die
ComputerTool. - Wiederholt diesen Vorgang für mehrere Iterationen.
Hauptfunktion
Die Hauptfunktion koordiniert den gesamten Prozess:
# 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())
Die Hauptfunktion:
- Initialisiert den AzureOpenAI-Client.
- Richtet den Playwright-Browser ein.
- Beginnt bei Bing.com.
- Tritt in eine Schleife ein, um Benutzeraufgaben zu akzeptieren.
- Erfasst den Anfangszustand.
- Sendet die Aufgabe und den Screenshot an das Modell.
- Verarbeitet die Antwort des Modells.
- Wiederholt sich, bis der Benutzer das Programm beendet.
- Stellt sicher, dass der Browser ordnungsgemäß geschlossen ist.
Vollständiges Skript
Vorsicht
Dieser Code ist experimentell und dient nur zu Demonstrationszwecken. Es ist nur vorgesehen, den grundlegenden Fluss der Antwort-API und des computer-use-preview Modells zu veranschaulichen. Sie können diesen Code zwar auf Ihrem lokalen Computer ausführen, es wird jedoch dringend empfohlen, diesen Code auf einem virtuellen Computer mit geringen Rechten ohne Zugriff auf vertrauliche Daten auszuführen. Dieser Code dient nur zu grundlegenden Testzwecken.
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())