Поделиться через

Использование компьютера в Azure OpenAI (классическая модель)

Применяется только к:Портал Foundry (классический). Эта статья недоступна для нового портала Foundry. Дополнительные сведения о новом портале.

Используйте эту статью, чтобы узнать, как работать с использованием компьютера в Azure OpenAI. Использование компьютеров — это специализированное средство искусственного интеллекта, которое использует специализированную модель, которая может выполнять задачи, взаимодействуя с компьютерными системами и приложениями через свои UIs. С помощью компьютера можно создать агент, который может обрабатывать сложные задачи и принимать решения, интерпретируя визуальные элементы и выполняя действия на основе содержимого на экране.

Использование компьютера предоставляет следующие возможности:

  • Автономная навигация: например, открывает приложения, нажимает кнопки, заполняет формы и перемещает многостраничные рабочие процессы.
  • Динамическая адаптация: интерпретирует изменения пользовательского интерфейса и корректирует действия соответствующим образом.
  • Выполнение задачи между приложениями: работает в веб-приложениях и классических приложениях.
  • Интерфейс естественного языка: пользователи могут описать задачу на простом языке, а модель использования компьютера определяет правильное взаимодействие пользовательского интерфейса для выполнения.

Запросить доступ

Для доступа к модели gpt-5.4 требуется регистрация, и доступ будет предоставлен на основе критериев соответствия требованиям Майкрософт. Клиенты, которым предоставлен доступ к другим моделям с ограниченным доступом, всё равно должны будут запросить доступ к этой модели.

Запрос доступа: приложение модели ограниченного доступа gpt-5.4

После предоставления доступа необходимо создать развертывание модели.

Отправка вызова API в модель использования компьютера с помощью API ответов

Средство "Использование компьютера" доступно через API ответа. Средство работает в непрерывном цикле, который выполняет такие действия, как ввод текста или выполнение щелчка. Код выполняет эти действия на компьютере и отправляет снимки экрана результата в модель.

Таким образом, код имитирует действия человека с помощью компьютерного интерфейса, а модель использует снимки экрана, чтобы понять состояние среды и предложить следующие действия.

В следующих примерах показан базовый вызов API:

Чтобы отправить запросы, необходимо установить следующие пакеты Python.

pip install openai
pip install azure-identity

import os
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import OpenAI
import json

token_provider = get_bearer_token_provider(DefaultAzureCredential(), "https://ai.azure.com/.default")

client = OpenAI(  
  base_url = "https://YOUR-RESOURCE-NAME.openai.azure.com/openai/v1/",  
  api_key=token_provider,
)

response = client.responses.create(
    model="gpt-5.4", # set this to your model deployment name
    tools=[{"type": "computer"}],
    input=[
        {
            "role": "user",
            "content": "Check the latest AI news on bing.com."
        }
    ],
)

print(json.dumps(response.model_dump(), indent=2))

Выходные данные

{
  "id": "resp_068b0022b159a6710069b0d44d97848195911e2ff69ff500fa",
  "created_at": 1773196365.0,
  "error": null,
  "incomplete_details": null,
  "instructions": null,
  "metadata": {},
  "model": "gpt-5.4",
  "object": "response",
  "output": [
    {
      "id": "msg_068b0022b159a6710069b0d44ede1881959e2d1deefe9f8676",
      "content": [
        {
          "annotations": [],
          "text": "I\u2019ll open Bing, look for current AI news, and summarize the latest headlines I find.",
          "type": "output_text",
          "logprobs": []
        }
      ],
      "role": "assistant",
      "status": "completed",
      "type": "message",
      "phase": "commentary"
    },
    {
      "id": "cu_068b0022b159a6710069b0d45008448195980f77beaa9cec83",
      "action": null,
      "call_id": "call_4y94crSZe0elpGhdiiwjLpa0",
      "pending_safety_checks": null,
      "status": "completed",
      "type": "computer_call",
      "actions": [
        {
          "type": "screenshot"
        }
      ]
    }
  ],
  "parallel_tool_calls": true,
  "temperature": 1.0,
  "tool_choice": "auto",
  "tools": [
    {
      "name": null,
      "parameters": null,
      "strict": null,
      "type": "computer",
      "description": null
    }
  ],
  "top_p": 0.98,
  "background": false,
  "conversation": null,
  "max_output_tokens": null,
  "max_tool_calls": null,
  "previous_response_id": null,
  "prompt": null,
  "prompt_cache_key": null,
  "reasoning": {
    "effort": "none",
    "generate_summary": null,
    "summary": null
  },
  "safety_identifier": null,
  "service_tier": "default",
  "status": "completed",
  "text": {
    "format": {
      "type": "text"
    },
    "verbosity": "medium"
  },
  "top_logprobs": 0,
  "truncation": "disabled",
  "usage": {
    "input_tokens": 820,
    "input_tokens_details": {
      "cached_tokens": 0
    },
    "output_tokens": 40,
    "output_tokens_details": {
      "reasoning_tokens": 16
    },
    "total_tokens": 860
  },
  "user": null,
  "completed_at": 1773196368,
  "content_filters": [
    {Removed from example output}
  ],
  "frequency_penalty": 0.0,
  "presence_penalty": 0.0,
  "prompt_cache_retention": null,
  "store": true
}

После отправки первоначального запроса 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
actions = computer_call.actions  # actions is now a batched array

# Your application would now perform each action in the actions[] array, in order
# And create a screenshot of the updated state of the environment before sending another response

response_2 = client.responses.create(
    model="gpt-5.4",
    previous_response_id=response.id,
    tools=[{"type": "computer"}],
    input=[
        {
            "call_id": last_call_id,
            "type": "computer_call_output",
            "output": {
                "type": "computer_screenshot",
                # Image should be in base64
                "image_url": f"data:image/png;base64,{<base64_string>}",
                "detail": "original"
            }
        }
    ],
)

Общие сведения об интеграции с использованием компьютера

При работе с средством "Использование компьютера" обычно выполняется следующее, чтобы интегрировать его в приложение.

  1. Отправьте запрос к модели, который включает вызов средства использования компьютера. Вы также можете включить снимок экрана начального состояния среды в первом запросе API.
  2. Получите ответ от модели. Если ответ содержит actions массив, эти элементы содержат предлагаемые действия для достижения прогресса в достижении указанной цели. Например, действие может быть screenshot таким, чтобы модель могла оценить текущее состояние с обновленным снимком экрана или click координатами X/Y, указывающими, где нужно переместить мышь.
  3. Выполните действие с помощью кода приложения в среде компьютера или браузера.
  4. После выполнения действия зафиксировать обновленное состояние среды в виде снимка экрана.
  5. Отправьте новый запрос с обновленным состоянием в виде computer_call_output и повторите этот loop, пока модель не перестанет запрашивать действия или вы решили остановиться.

Обработка истории переписки

Параметр можно использовать previous_response_id для связывания текущего запроса с предыдущим ответом. Этот параметр рекомендуется использовать, если вы не хотите управлять журналом бесед.

Если вы не используете этот параметр, необходимо убедиться, что все элементы, возвращаемые как выходные данные ответа предыдущего запроса, включены в массив входных данных. Это включает в себя аргументы, если они присутствуют.

Проверки безопасности

API имеет проверки безопасности для защиты от внедрения команд и ошибок моделей. К таким проверкам относятся:

  • Обнаружение вредоносных инструкций: система оценивает изображение снимка экрана и проверяет, содержит ли он состязательный контент, который может изменить поведение модели.
  • Обнаружение неуместных доменов: система оценивает текущий домен и проверяет, считается ли он соответствующим с учетом журнала бесед.
  • Обнаружение конфиденциального домена: система проверяет текущий домен и вызывает предупреждение при обнаружении пользователя в конфиденциальном домене.

Если запускается одна или несколько указанных выше проверок, проверка безопасности возникает при возвращении следующей 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...",
        "actions": [
            {
                "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 gpt-5.4 с Playwright для автоматизации базовых взаимодействий с браузером. Объединение модели с Playwright позволяет модели просматривать экран браузера, принимать решения и выполнять такие действия, как щелчки, ввод и навигация веб-сайтов. При выполнении этого примера кода следует соблюдать осторожность. Этот код предназначен для локального выполнения, но должен выполняться только в тестовой среде. Используйте человека для подтверждения решений и не предоставляйте модели доступ к конфиденциальным данным.

Анимированный gif-файл модели предпросмотра использования компьютера, интегрированной с Playwright.

Сначала необходимо установить библиотеку Python для Playwright.

pip install playwright

После того, как пакет установлен, вам также нужно будет запустить

playwright install

Импорт и настройка

Во-первых, мы импортируем необходимые библиотеки и определяем параметры конфигурации. Поскольку мы используем asyncio, мы будем выполнять этот код вне среды ноутбуков Jupyter. Сначала рассмотрим код в блоках, а затем покажем, как его использовать.

import os
import asyncio
import base64
from openai import OpenAI
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from playwright.async_api import async_playwright, TimeoutError

token_provider = get_bearer_token_provider(
    DefaultAzureCredential(), "https://ai.azure.com/.default"
)

# Configuration

BASE_URL = "https://YOUR-RESOURCE-NAME.openai.azure.com/openai/v1/"
MODEL = "gpt-5.4" # Set to model deployment name
DISPLAY_WIDTH = 1440
DISPLAY_HEIGHT = 900
ITERATIONS = 5 # Max number of iterations before returning control to human supervisor

Замечание

Разрешение дисплея 1440x900 или 1600x900 рекомендуется для оптимальной точности щелчка с моделью использования компьютера.

Назначение клавиш для взаимодействия с браузером

Далее мы настраиваем сопоставления для специальных клавиш, которые модели может понадобиться передать в Playwright. В конечном счете модель никогда не выполняет действия, она передает представления команд, и необходимо предоставить окончательный уровень интеграции, который может выполнять эти команды и выполнять их в выбранной среде.

Это не исчерпывающий список возможных сопоставлений ключей. Этот список можно развернуть по мере необходимости. Этот словарь предназначен для интеграции модели с инструментом автоматизации тестирования Playwright. Если вы интегрируете модель с альтернативной библиотекой для предоставления API-доступа к клавиатуре или мыши операционной системы, вам потребуется предоставить отображение, специфичное для этой библиотеки.

# Key mapping for special keys in Playwright
# Supports multiple common spellings for each key (case-insensitive)
KEY_MAPPING = {
    "/": "Slash", "\\": "Backslash",
    "alt": "Alt", "option": "Alt",
    "arrowdown": "ArrowDown", "down": "ArrowDown",
    "arrowleft": "ArrowLeft", "left": "ArrowLeft",
    "arrowright": "ArrowRight", "right": "ArrowRight",
    "arrowup": "ArrowUp", "up": "ArrowUp",
    "backspace": "Backspace",
    "ctrl": "Control", "control": "Control",
    "cmd": "Meta", "command": "Meta", "meta": "Meta", "win": "Meta", "super": "Meta",
    "delete": "Delete",
    "enter": "Enter", "return": "Return",
    "esc": "Escape", "escape": "Escape",
    "shift": "Shift",
    "space": " ",
    "tab": "Tab",
    "pagedown": "PageDown", "pageup": "PageUp",
    "home": "Home", "end": "End",
    "insert": "Insert",
    "f1": "F1", "f2": "F2", "f3": "F3", "f4": "F4",
    "f5": "F5", "f6": "F6", "f7": "F7", "f8": "F8",
    "f9": "F9", "f10": "F10", "f11": "F11", "f12": "F12"
}

Этот словарь преобразует имена ключей в формат, ожидаемый API клавиатуры Playwright. Для каждого ключа поддерживается несколько распространенных орфографических значений (например, CTRL и CONTROL сопоставляются с Control).

Проверка координат

Чтобы убедиться, что все действия мыши, передаваемые из модели, остаются в границах окна браузера, мы добавим следующую функцию служебной программы:

def validate_coordinates(x, y):
    """Ensure coordinates are within display bounds."""
    return max(0, min(x, DISPLAY_WIDTH)), max(0, min(y, DISPLAY_HEIGHT))

Обработка действий

Основной частью нашей автоматизации браузера является обработчик действий, который обрабатывает различные типы взаимодействия пользователей и преобразует их в действия в браузере. Действия в массиве actions[] возвращаются в виде простых словарей.

async def handle_action(page, action):
    """Handle different action types from the model."""
    action_type = action.get("type")

    if action_type == "click":
        button = action.get("button", "left")
        x, y = validate_coordinates(action.get("x"), action.get("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":
        x, y = validate_coordinates(action.get("x"), action.get("y"))
        print(f"\tAction: double click at ({x}, {y})")
        await page.mouse.dblclick(x, y)

    elif action_type == "drag":
        path = action.get("path", [])
        if len(path) < 2:
            print("\tAction: drag requires at least 2 points. Skipping.")
            return
        start = path[0]
        sx, sy = validate_coordinates(start.get("x", 0), start.get("y", 0))
        print(f"\tAction: drag from ({sx}, {sy}) through {len(path) - 1} points")
        await page.mouse.move(sx, sy)
        await page.mouse.down()
        for point in path[1:]:
            px, py = validate_coordinates(point.get("x", 0), point.get("y", 0))
            await page.mouse.move(px, py)
        await page.mouse.up()

    elif action_type == "move":
        x, y = validate_coordinates(action.get("x"), action.get("y"))
        print(f"\tAction: move to ({x}, {y})")
        await page.mouse.move(x, y)

    elif action_type == "scroll":
        scroll_x = action.get("scroll_x", 0)
        scroll_y = action.get("scroll_y", 0)
        x, y = validate_coordinates(action.get("x"), action.get("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 = action.get("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 = action.get("text", "")
        print(f"\tAction: type text: {text}")
        await page.keyboard.type(text, delay=20)

    elif action_type == "wait":
        ms = action.get("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}")

Снимок экрана

Чтобы модель могла видеть, с чем она взаимодействует, ей нужен способ делать скриншоты. Для этого кода мы используем 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 response.output:
            print("No output from model.")
            break

        response_id = response.id
        print(f"\nIteration {iteration + 1} - Response ID: {response_id}\n")

        # Print text responses and reasoning
        for item in response.output:
            if item.type == "text":
                print(f"\nModel message: {item.text}\n")

            if item.type == "reasoning" and item.summary:
                print("=== Model Reasoning ===")
                for summary in item.summary:
                    if hasattr(summary, 'text') and summary.text.strip():
                        print(summary.text)
                print("=====================\n")

        # Extract computer calls
        computer_calls = [item for item in response.output if 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]
        call_id = computer_call.call_id
        actions = computer_call.actions  # actions is a batched array of dicts

        # Handle safety checks
        acknowledged_checks = []
        if 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 all actions in the batch, in order
        try:
            await page.bring_to_front()
            for action in actions:
                await handle_action(page, action)

                # Check if a new page was created after a click action
                if action.get("type") == "click":
                    await asyncio.sleep(1.5)
                    all_pages = page.context.pages
                    if len(all_pages) > 1:
                        newest_page = all_pages[-1]
                        if newest_page != page and newest_page.url not in ["about:blank", ""]:
                            print(f"\tSwitching to new tab: {newest_page.url}")
                            page = newest_page
                elif action.get("type") != "wait":
                    await asyncio.sleep(0.5)

        except Exception as e:
            print(f"Error handling action: {e}")
            import traceback
            traceback.print_exc()

        # Take a screenshot after the actions
        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": "computer_screenshot",
                "image_url": f"data:image/png;base64,{screenshot_base64}",
                "detail": "original"
            }
        }]

        # Add acknowledged safety checks if any
        if acknowledged_checks:
            input_content[0]["acknowledged_safety_checks"] = [
                {"id": check.id, "code": check.code, "message": check.message}
                for check in acknowledged_checks
            ]

        # Send the screenshot back for the next step
        try:
            response = client.responses.create(
                model=MODEL,
                previous_response_id=response_id,
                tools=[{"type": "computer"}],
                input=input_content,
            )
            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.")

В этом разделе мы добавили код, который:

  • Извлекает текст и аргументацию из модели и отображает их.
  • Обрабатывает вызовы действий компьютерной программы.
  • Обрабатывает потенциальные проверки безопасности, требующие подтверждения пользователя.
  • Выполняет запрошенные действия (пакетные в массиве диктовок).
  • Записывает новый снимок экрана.
  • Отправляет обновленное состояние обратно в модель и определяет computer инструмент.
  • Повторяет этот процесс несколько раз.

Основная функция

Основная функция координирует весь процесс:

    # Initialize OpenAI client
    client = OpenAI(
        base_url=BASE_URL,
        api_key=token_provider,
    )
    
    # Initialize Playwright
    async with async_playwright() as playwright:
        browser = await playwright.chromium.launch(
            headless=False,
            args=[f"--window-size={DISPLAY_WIDTH},{DISPLAY_HEIGHT}", "--disable-extensions"]
        )
        
        context = await browser.new_context(
            viewport={"width": DISPLAY_WIDTH, "height": DISPLAY_HEIGHT},
            accept_downloads=True
        )
        
        page = await context.new_page()
        
        # Navigate to starting page
        await page.goto("https://www.bing.com", wait_until="domcontentloaded")
        print("Browser initialized to Bing.com")
        
        # Main interaction loop
        try:
            while True:
                print("\n" + "="*50)
                user_input = input("Enter a task to perform (or 'exit' to quit): ")
                
                if user_input.lower() in ('exit', 'quit'):
                    break
                
                if not user_input.strip():
                    continue
                
                # Take initial screenshot
                screenshot_base64 = await take_screenshot(page)
                print("\nTake initial screenshot")
                
                # Initial request to the model
                response = client.responses.create(
                    model=MODEL,
                    tools=[{"type": "computer"}],
                    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}",
                            "detail": "original"
                        }]
                    }],
                    reasoning={"summary": "concise"},
                )
                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())

Основная функция:

  • Инициализирует клиент OpenAI.
  • Настраивает браузер Playwright.
  • Начинается с Bing.com.
  • Вводит loop для принятия пользовательских задач.
  • Записывает начальное состояние.
  • Отправляет задачу и снимок экрана в модель.
  • Обрабатывает ответ модели.
  • Повторяется до тех пор, пока пользователь не завершит работу.
  • Гарантирует правильность закрытия браузера.

Полный скрипт

Предостережение

Этот код является экспериментальным и только для демонстрационных целей. Он предназначен только для иллюстрации базового потока API ответов и gpt-5.4 модели. Хотя этот код можно выполнить на локальном компьютере, настоятельно рекомендуется запускать этот код на виртуальной машине с низким уровнем привилегий без access конфиденциальных данных. Этот код предназначен только для базовых целей тестирования.

import os
import asyncio
import base64
from openai import OpenAI
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from playwright.async_api import async_playwright, TimeoutError

token_provider = get_bearer_token_provider(
    DefaultAzureCredential(), "https://ai.azure.com/.default"
)

# Configuration

BASE_URL = "https://YOUR-RESOURCE-NAME.openai.azure.com/openai/v1/"
MODEL = "gpt-5.4"
DISPLAY_WIDTH = 1440
DISPLAY_HEIGHT = 900
ITERATIONS = 5 # Max number of iterations before forcing the model to return control to the human supervisor

# Key mapping for special keys in Playwright
# Supports multiple common spellings for each key (case-insensitive)
KEY_MAPPING = {
    "/": "Slash", "\\": "Backslash",
    "alt": "Alt", "option": "Alt",
    "arrowdown": "ArrowDown", "down": "ArrowDown",
    "arrowleft": "ArrowLeft", "left": "ArrowLeft",
    "arrowright": "ArrowRight", "right": "ArrowRight",
    "arrowup": "ArrowUp", "up": "ArrowUp",
    "backspace": "Backspace",
    "ctrl": "Control", "control": "Control",
    "cmd": "Meta", "command": "Meta", "meta": "Meta", "win": "Meta", "super": "Meta",
    "delete": "Delete",
    "enter": "Enter", "return": "Return",
    "esc": "Escape", "escape": "Escape",
    "shift": "Shift",
    "space": " ",
    "tab": "Tab",
    "pagedown": "PageDown", "pageup": "PageUp",
    "home": "Home", "end": "End",
    "insert": "Insert",
    "f1": "F1", "f2": "F2", "f3": "F3", "f4": "F4",
    "f5": "F5", "f6": "F6", "f7": "F7", "f8": "F8",
    "f9": "F9", "f10": "F10", "f11": "F11", "f12": "F12"
}

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.get("type")

    if action_type == "click":
        button = action.get("button", "left")
        x, y = validate_coordinates(action.get("x"), action.get("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":
        x, y = validate_coordinates(action.get("x"), action.get("y"))
        print(f"\tAction: double click at ({x}, {y})")
        await page.mouse.dblclick(x, y)

    elif action_type == "drag":
        path = action.get("path", [])
        if len(path) < 2:
            print("\tAction: drag requires at least 2 points. Skipping.")
            return
        start = path[0]
        sx, sy = validate_coordinates(start.get("x", 0), start.get("y", 0))
        print(f"\tAction: drag from ({sx}, {sy}) through {len(path) - 1} points")
        await page.mouse.move(sx, sy)
        await page.mouse.down()
        for point in path[1:]:
            px, py = validate_coordinates(point.get("x", 0), point.get("y", 0))
            await page.mouse.move(px, py)
        await page.mouse.up()

    elif action_type == "move":
        x, y = validate_coordinates(action.get("x"), action.get("y"))
        print(f"\tAction: move to ({x}, {y})")
        await page.mouse.move(x, y)

    elif action_type == "scroll":
        scroll_x = action.get("scroll_x", 0)
        scroll_y = action.get("scroll_y", 0)
        x, y = validate_coordinates(action.get("x"), action.get("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 = action.get("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 = action.get("text", "")
        print(f"\tAction: type text: {text}")
        await page.keyboard.type(text, delay=20)

    elif action_type == "wait":
        ms = action.get("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}")
        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 response.output:
            print("No output from model.")
            break
        
        response_id = response.id
        print(f"\nIteration {iteration + 1} - Response ID: {response_id}\n")
        
        # Print text responses and reasoning
        for item in response.output:
            # Handle text output
            if item.type == "text":
                print(f"\nModel message: {item.text}\n")
                
            if item.type == "reasoning" and item.summary:
                print("=== Model Reasoning ===")
                for summary in item.summary:
                    if hasattr(summary, 'text') and summary.text.strip():
                        print(summary.text)
                print("=====================\n")
        
        # Extract computer calls
        computer_calls = [item for item in response.output if 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]
        call_id = computer_call.call_id
        actions = computer_call.actions  # actions is a batched array of dicts

        # Handle safety checks
        acknowledged_checks = []
        if 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 all actions in the batch, in order
        try:
            await page.bring_to_front()
            for action in actions:
                await handle_action(page, action)

                # Check if a new page was created after a click action
                if action.get("type") == "click":
                    await asyncio.sleep(1.5)
                    all_pages = page.context.pages
                    if len(all_pages) > 1:
                        newest_page = all_pages[-1]
                        if newest_page != page and newest_page.url not in ["about:blank", ""]:
                            print(f"\tSwitching to new tab: {newest_page.url}")
                            page = newest_page
                elif action.get("type") != "wait":
                    await asyncio.sleep(0.5)

        except Exception as e:
            print(f"Error handling action: {e}")
            import traceback
            traceback.print_exc()

        # Take a screenshot after the actions
        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": "computer_screenshot",
                "image_url": f"data:image/png;base64,{screenshot_base64}",
                "detail": "original"
            }
        }]

        # Add acknowledged safety checks if any
        if acknowledged_checks:
            input_content[0]["acknowledged_safety_checks"] = [
                {"id": check.id, "code": check.code, "message": check.message}
                for check in acknowledged_checks
            ]

        # Send the screenshot back for the next step
        try:
            response = client.responses.create(
                model=MODEL,
                previous_response_id=response_id,
                tools=[{"type": "computer"}],
                input=input_content,
            )
            print("\tModel processing screenshot")
        except Exception as e:
            print(f"Error in API call: {e}")
            import traceback
            traceback.print_exc()
            break

    if iteration >= max_iterations - 1:
        print("Reached maximum number of iterations. Stopping.")

async def main():    
    # Initialize OpenAI client
    client = OpenAI(
        base_url=BASE_URL,
        api_key=token_provider
    )
    
    # Initialize Playwright
    async with async_playwright() as playwright:
        browser = await playwright.chromium.launch(
            headless=False,
            args=[f"--window-size={DISPLAY_WIDTH},{DISPLAY_HEIGHT}", "--disable-extensions"]
        )
        
        context = await browser.new_context(
            viewport={"width": DISPLAY_WIDTH, "height": DISPLAY_HEIGHT},
            accept_downloads=True
        )
        
        page = await context.new_page()
        
        # Navigate to starting page
        await page.goto("https://www.bing.com", wait_until="domcontentloaded")
        print("Browser initialized to Bing.com")
        
        # Main interaction loop
        try:
            while True:
                print("\n" + "="*50)
                user_input = input("Enter a task to perform (or 'exit' to quit): ")
                
                if user_input.lower() in ('exit', 'quit'):
                    break
                
                if not user_input.strip():
                    continue
                
                # Take initial screenshot
                screenshot_base64 = await take_screenshot(page)
                print("\nTake initial screenshot")
                
                # Initial request to the model
                response = client.responses.create(
                    model=MODEL,
                    tools=[{"type": "computer"}],
                    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}",
                            "detail": "original"
                        }]
                    }],
                    reasoning={"summary": "concise"},
                )
                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())

См. также

  • Last updated on