Примечание
Для доступа к этой странице требуется авторизация. Вы можете попробовать войти или изменить каталоги.
Для доступа к этой странице требуется авторизация. Вы можете попробовать изменить каталоги.
Используйте эту статью, чтобы узнать, как работать с использованием компьютера в Azure OpenAI. Использование компьютеров — это специализированное средство искусственного интеллекта, которое использует специализированную модель, которая может выполнять задачи, взаимодействуя с компьютерными системами и приложениями через свои UIs. С помощью компьютера можно создать агент, который может обрабатывать сложные задачи и принимать решения, интерпретируя визуальные элементы и выполняя действия на основе содержимого на экране.
Использование компьютера предоставляет следующие возможности:
- Автономная навигация: например, открывает приложения, нажимает кнопки, заполняет формы и перемещает многостраничные рабочие процессы.
- Динамическая адаптация: интерпретирует изменения пользовательского интерфейса и корректирует действия соответствующим образом.
- Выполнение задачи между приложениями: работает в веб-приложениях и классических приложениях.
- Интерфейс естественного языка: пользователи могут описать задачу на простом языке, а модель использования компьютера определяет правильное взаимодействие пользовательского интерфейса для выполнения.
Запрос на доступ
Для доступа к computer-use-preview модели требуется регистрация и доступ будет предоставлен на основе критериев соответствия корпорации Майкрософт. Клиентам, имеющим доступ к другим моделям ограниченного доступа, по-прежнему потребуется запросить доступ к этой модели.
Запрос доступа: приложение модели ограниченного доступа для предварительного просмотра на компьютере
После предоставления доступа необходимо создать деплоймент для модели.
Региональная поддержка
Использование компьютера доступно в следующих регионах:
eastus2swedencentralsouthindia
Отправка вызова API в модель использования компьютера с помощью API ответов
Средство "Использование компьютера" доступно через API ответа. Средство работает в непрерывном цикле, который отправляет такие действия, как ввод текста или выполнение щелчка. Код выполняет эти действия на компьютере и отправляет снимки экрана результата в модель.
Таким образом, код имитирует действия человека с помощью компьютерного интерфейса, а модель использует снимки экрана, чтобы понять состояние среды и предложить следующие действия.
В следующих примерах показан базовый вызов API.
Замечание
Требуется ресурс Azure OpenAI с развертыванием computer-use-preview модели в поддерживаемом регионе.
Чтобы отправить запросы, необходимо установить следующие пакеты Python.
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)
Выходные данные
[
ResponseComputerToolCall(
id='cu_67d841873c1081908bfc88b90a8555e0',
action=ActionScreenshot(type='screenshot'),
call_id='call_wwEnfFDqQr1Z4Edk62Fyo7Nh',
pending_safety_checks=[],
status='completed',
type='computer_call'
)
]
После отправки первоначального запроса API выполняется цикл, в котором указанное действие выполняется в коде приложения, отправляя снимок экрана с каждым поворотом, чтобы модель может оценить обновленное состояние среды.
## 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"
)
Общие сведения об интеграции с использованием компьютера
При работе с средством "Использование компьютера" обычно выполняется следующее, чтобы интегрировать его в приложение.
- Отправьте запрос в модель, включающую вызов средства использования компьютера, а также размер дисплея и среду. Вы также можете включить снимок экрана начального состояния среды в первом запросе API.
- Получите ответ от модели. Если ответ содержит
actionэлементы, эти элементы содержат предлагаемые действия для достижения прогресса в достижении указанной цели. Например, действие может бытьscreenshotтаким, чтобы модель могла оценить текущее состояние с обновленным снимком экрана илиclickкоординатами X/Y, указывающими, где нужно переместить мышь. - Выполните действие с помощью кода приложения в среде компьютера или браузера.
- После выполнения действия зафиксировать обновленное состояние среды в виде снимка экрана.
- Отправьте новый запрос с обновленным состоянием в виде
computer_call_outputи повторите этот цикл, пока модель не перестанет запрашивать действия или вы решили остановиться.
Обработка истории переписки
Параметр можно использовать previous_response_id для связывания текущего запроса с предыдущим ответом. Этот параметр рекомендуется использовать, если вы не хотите управлять журналом бесед.
Если вы не используете этот параметр, необходимо убедиться, что все элементы, возвращаемые как выходные данные ответа предыдущего запроса, включены в массив входных данных. Это включает в себя аргументы, если они присутствуют.
Проверки безопасности
API имеет проверки безопасности для защиты от внедрения команд и ошибок моделей. Эти проверки включают в себя:
- Обнаружение вредоносных инструкций: система оценивает изображение снимка экрана и проверяет, содержит ли он состязательный контент, который может изменить поведение модели.
-
Обнаружение неуместных доменов: система оценивает
current_url(если предоставлено) и проверяет, считается ли текущий домен соответствующим с учетом журнала бесед. -
Обнаружение конфиденциального домена: система проверяет
current_url(если указано) и вызывает предупреждение при обнаружении пользователя в конфиденциальном домене.
Если запускается одна или несколько указанных выше проверок, проверка безопасности возникает при возвращении следующей computer_callмодели с параметром 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"
}
]
Чтобы продолжить, необходимо вернуть проверки безопасности как acknowledged_safety_checks в следующем запросе.
"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>"
}
}
],
Управление проверкой безопасности
Во всех случаях, когда pending_safety_checks возвращаются, действия должны быть переданы конечному пользователю, чтобы подтвердить правильное поведение модели и точность.
-
malicious_instructionsиirrelevant_domain: конечные пользователи должны просматривать действия модели и проверять, что модель ведет себя должным образом. -
sensitive_domain: убедитесь, что конечный пользователь активно отслеживает действия модели на этих сайтах. Точная реализация этого режима наблюдения может отличаться в зависимости от приложения, но потенциальный пример может представлять собой сбор данных о впечатлениях пользователей на сайте, чтобы убедиться, что с приложением происходит активное взаимодействие пользователей.
Интеграция Playwright
В этом разделе представлен простой пример сценария, который интегрирует модель Azure OpenAI computer-use-preview с Playwright для автоматизации основных взаимодействий с браузером. Объединение модели с Playwright позволяет модели просматривать экран браузера, принимать решения и выполнять такие действия, как щелчки, ввод и навигация веб-сайтов. При выполнении этого примера кода следует соблюдать осторожность. Этот код предназначен для локального выполнения, но должен выполняться только в тестовой среде. Используйте человека, чтобы подтвердить решения и не предоставлять модели доступ к конфиденциальным данным.
Сначала необходимо установить библиотеку Python для Playwright.
pip install playwright
После того, как пакет установлен, вам также нужно будет запустить
playwright install
Импорт и настройка
Во-первых, мы импортируем необходимые библиотеки и определяем параметры конфигурации. Поскольку мы используем asyncio, мы будем выполнять этот код вне среды ноутбуков Jupyter. Сначала рассмотрим код в блоках, а затем покажем, как его использовать.
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
Назначение клавиш для взаимодействия с браузером
Далее мы настраиваем сопоставления для специальных клавиш, которые модели может понадобиться передать в Playwright. В конечном счете модель никогда не выполняет действия, она передает представления команд, и необходимо предоставить окончательный уровень интеграции, который может выполнять эти команды и выполнять их в выбранной среде.
Это не исчерпывающий список возможных сопоставлений ключей. Этот список можно развернуть по мере необходимости. Этот словарь предназначен для интеграции модели с инструментом автоматизации тестирования Playwright. Если вы интегрируете модель с альтернативной библиотекой для предоставления доступа API к клавиатуре или мыши операционных систем, вам потребуется предоставить сопоставление, относящееся к этой библиотеке.
# 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"
}
Этот словарь преобразует удобочитаемые имена клавиш в формат, ожидаемый API клавиатуры Playwright.
Функция проверки координат
Чтобы убедиться, что все действия мыши, передаваемые из модели, остаются в границах окна браузера, мы добавим следующую функцию служебной программы:
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}")
Эта функция пытается обрабатывать различные типы действий. Нам нужно перевести команды computer-use-preview, которые будут создаваться, и библиотеку Playwright, которая будет выполнять действия. Для получения дополнительной информации обратитесь к справочной документации для ComputerAction.
Снимок экрана
Чтобы модель могла видеть, с чем она взаимодействует, ей нужен способ делать скриншоты. Для этого кода мы используем Playwright для записи снимка экрана, и мы ограничиваем представление только содержимым в окне браузера. Снимок экрана не включает строку URL-адресов или другие аспекты графического интерфейса браузера. Если вам нужна модель для просмотра за пределами главного окна браузера, вы можете расширить модель, создав собственную функцию снимка экрана.
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
Эта функция записывает текущее состояние браузера в виде изображения и возвращает его в виде строки в кодировке Base64, готовой к отправке в модель. Мы постоянно делаем это в цикле после каждого шага, позволяющего модели видеть, была ли команда, которую она пыталась выполнить успешно или нет, что позволяет настроить ее на основе содержимого снимка экрана. Мы могли бы позволить модели решить, нужно ли сделать снимок экрана, но для простоты мы принудим сделать снимок экрана для каждой итерации.
Обработка ответов модели
Эта функция обрабатывает ответы модели и выполняет запрошенные действия:
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.")
В этом разделе мы добавили код, который:
- Извлекает текст и аргументацию из модели и отображает их.
- Обрабатывает вызовы действий компьютерной программы.
- Обрабатывает потенциальные проверки безопасности, требующие подтверждения пользователя.
- Выполняет запрошенное действие.
- Записывает новый снимок экрана.
- Отправляет обновленное состояние обратно в модель и определяет
ComputerTool. - Повторяет этот процесс несколько раз.
Основная функция
Основная функция координирует весь процесс:
# 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())
Основная функция:
- Инициализирует клиент AzureOpenAI.
- Настраивает браузер Playwright.
- Начинается с Bing.com.
- Вводит цикл для принятия пользовательских задач.
- Записывает начальное состояние.
- Отправляет задачу и снимок экрана в модель.
- Обрабатывает ответ модели.
- Повторяется до тех пор, пока пользователь не завершит работу.
- Гарантирует правильность закрытия браузера.
Полный скрипт
Осторожность
Этот код является экспериментальным и только для демонстрационных целей. Он предназначен только для иллюстрации базового потока API ответов и computer-use-preview модели. Хотя этот код можно выполнить на локальном компьютере, настоятельно рекомендуется запускать этот код на виртуальной машине с низким уровнем привилегий без доступа к конфиденциальным данным. Этот код предназначен только для базовых целей тестирования.
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())