Aracılığıyla paylaş

Azure OpenAI'de Bilgisayar Kullanımı (klasik)

Yalnızca şunlar için geçerlidir:Dökümhane (klasik) portalı. Bu makale yeni Dökümhane portalında kullanılamaz. Yeni portal hakkında daha fazla bilgi edinin.

Azure OpenAI'de Bilgisayar Kullanımı ile çalışmayı öğrenmek için bu makaleyi kullanın. Bilgisayar Kullanımı, bilgisayar sistemleri ve uygulamalarıyla kendi URI'leri aracılığıyla etkileşim kurarak görevleri gerçekleştirebilen özel bir model kullanan özel bir yapay zeka aracıdır. Bilgisayar Kullanımı ile, görsel öğeleri yorumlayarak ve ekran içeriğine göre eylem gerçekleştirerek karmaşık görevleri işleyebilen ve kararlar alabilen bir aracı oluşturabilirsiniz.

Bilgisayar Kullanımı sağlar:

  • Otonom gezinti: Örneğin, uygulamaları açar, düğmelere tıklar, formları doldurur ve çok sayfalı iş akışlarında geziner.
  • Dinamik uyarlama: Kullanıcı arabirimi değişikliklerini yorumlar ve eylemleri buna göre ayarlar.
  • Uygulamalar arası görev yürütme: Web tabanlı ve masaüstü uygulamalarında çalışır.
  • Doğal dil arabirimi: Kullanıcılar bir görevi düz dilde açıklayabilir ve Bilgisayar Kullanımı modeli yürütülecek doğru kullanıcı arabirimi etkileşimlerini belirler.

Erişim isteği

gpt-5.4 modeline erişim sağlamak için kayıt gereklidir ve erişim Microsoft'un uygunluk kriterlerine göre sağlanacaktır. Diğer sınırlı erişim modellerine erişimi olan müşterilerin, bu model için yine de erişim talep etmesi gerekmektedir.

Erişim isteme: gpt-5.4 sınırlı erişim modeli uygulaması

Erişim izni verildiğinde, model için bir dağıtım yapmanız gerekecek.

Yanıtlar API'sini kullanarak Bilgisayar Kullanımı modeline API çağrısı gönderme

Bilgisayar Kullanımı aracına yanıtlar API'si aracılığıyla erişilir. Araç, metin yazma veya tıklama gerçekleştirme gibi eylemler gönderen sürekli bir loop çalışır. Kodunuz bu eylemleri bir bilgisayarda yürütür ve sonucun ekran görüntülerini modele gönderir.

Bu şekilde kodunuz bilgisayar arabirimi kullanarak bir insanın eylemlerinin benzetimini yaparken model, ortamın durumunu anlamak ve sonraki eylemleri önermek için ekran görüntülerini kullanır.

Aşağıdaki örneklerde temel bir API çağrısı gösterilmektedir:

İstek göndermek için aşağıdaki Python paketlerini yüklemeniz gerekir.

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))

Çıktı

{
  "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
}

İlk API isteği gönderildikten sonra, belirtilen eylemin uygulama kodunuzda gerçekleştirildiği bir loop gerçekleştirirsiniz ve modelin ortamın güncelleştirilmiş durumunu değerlendirebilmesi için her turda bir ekran görüntüsü gönderirsiniz.


## 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"
            }
        }
    ],
)

Bilgisayar Kullanımının Entegrasyonu'nun Anlaşılması

Bilgisayar Kullanımı aracıyla çalışırken, bunu uygulamanızla tümleştirmek için genellikle aşağıdakileri gerçekleştirirsiniz.

  1. Bilgisayar kullanım aracına çağrı içeren modele bir istek gönderin. İlk API isteğine ortamın ilk durumunun ekran görüntüsünü de ekleyebilirsiniz.
  2. Modelden bir yanıt alın. Yanıtın bir actions dizisi varsa, bu öğeler belirtilen hedefe doğru ilerleme kaydetmek için önerilen eylemleri içerir. Örneğin, modelin, güncellenmiş bir ekran görüntüsü ile mevcut durumu değerlendirebilmesi için bir eylem screenshot veya farenin nereye taşınması gerektiğini belirten X/Y koordinatlarıyla olabilir click.
  3. Bilgisayarınızda veya tarayıcı ortamınızda uygulama kodunuzu kullanarak eylemi yürütür.
  4. Eylemi yürüttkten sonra, ortamın güncelleştirilmiş durumunu ekran görüntüsü olarak yakalayın.
  5. Güncellenmiş durumda yeni bir isteği computer_call_output olarak gönderin ve model eylemleri durdurana veya siz durmaya karar verene kadar döngüyü tekrarlayın.

Konuşma geçmişini işleme

Geçerli isteği önceki yanıta bağlamak için parametresini kullanabilirsiniz previous_response_id . Konuşma geçmişini yönetmek istemiyorsanız bu parametreyi kullanmanız önerilir.

Bu parametreyi kullanmıyorsanız, önceki isteğin yanıt çıkışında döndürülen tüm öğeleri girişler dizinize eklediğinizden emin olmanız gerekir. Bu, varsa akıl yürütme öğelerini içerir.

Güvenlik denetimleri

API' nin, istem ekleme ve model hatalarına karşı korumaya yardımcı olmak için güvenlik denetimleri vardır. Bu denetimler şunlardır:

  • Kötü amaçlı yönerge algılama: Sistem ekran görüntüsünü değerlendirir ve modelin davranışını değiştirebilecek saldırgan içerik içerip içermediğini denetler.
  • İlgisiz etki alanı algılama: Sistem, geçerli etki alanını değerlendirir ve konuşma geçmişine göre ilgili olup olmadığını denetler.
  • Hassas etki alanı algılama: Sistem geçerli etki alanını denetler ve kullanıcının hassas bir etki alanında olduğunu algıladığında bir uyarı oluşturur.

Yukarıdaki denetimlerden biri veya daha fazlası tetikleniyorsa, model sonraki computer_call değerini pending_safety_checks parametresi ile döndürdüğünde bir güvenlik denetimi başlatılır.

"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"
    }
]

Devam etmek için bir sonraki istekte acknowledged_safety_checks olarak güvenlik kontrollerini geri göndermeniz gerekiyor.

"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>"
            }
        }
    ],

Güvenlik denetimi yönetimi

Her durumda, pending_safety_checks döndürüldüğünde, modelin doğru çalıştığından ve doğruluğundan emin olmak için eylemlerin son kullanıcıya devredilmesi gerekir.

  • malicious_instructions ve irrelevant_domain: Son kullanıcılar model eylemlerini gözden geçirmeli ve modelin istenen şekilde davrandığını onaylamalıdır.
  • sensitive_domain: Son kullanıcının bu sitelerdeki model eylemlerini etkin bir şekilde izlediğinden emin olun. Bu "izleme modunun" tam olarak uygulanması uygulamaya göre farklılık gösterebilir, ancak olası bir örnek, uygulamayla etkin son kullanıcı etkileşimi olduğundan emin olmak için sitede kullanıcı gösterim verilerini toplamak olabilir.

Playwright tümleştirmesi

Bu bölümde, Azure OpenAI'nin gpt-5.4 modelini Playwright ile tümleştirerek temel tarayıcı etkileşimlerini otomatik hale getiren basit bir örnek betik sunuyoruz. Modeli Playwright ile birleştirmek, modelin tarayıcı ekranını görmesine, kararlar vermesine ve web sitelerine tıklama, yazma ve gezinme gibi eylemler gerçekleştirmesine olanak tanır. Bu örnek kodu çalıştırırken dikkatli olmanız gerekir. Bu kod yerel olarak çalıştırılacak şekilde tasarlanmıştır, ancak yalnızca bir test ortamında yürütülmelidir. Kararları onaylamak için bir insan kullanın ve modele hassas verilere access vermeyin.

Playwright ile tümleştirilen bilgisayar kullanım önizlemesi modelinin animasyonlu gif'i.

İlk olarak Playwright için Python kitaplığını yüklemeniz gerekir.

pip install playwright

Paket yüklendikten sonra, ayrıca çalıştırmanız gerekecek.

playwright install

İçeri aktarmalar ve yapılandırma

İlk olarak gerekli kitaplıkları içeri aktarır ve yapılandırma parametrelerimizi tanımlarız. asyncio kullandığımız için bu kodu Jupyter defterlerinin dışında çalıştıracağız. Kodu önce öbekler halinde inceleyeceğiz ve ardından nasıl kullanılacağını göstereceğiz.

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

Uyarı

Bilgisayar kullanım modelinde en uygun tıklama doğruluğu için 1440x900 veya 1600x900 ekran çözünürlüğü önerilir.

Tarayıcı etkileşimi için anahtar eşleme

Ardından, modelin Playwright'a iletmesi gerekebilecek özel anahtarlar için eşlemeler ayarlayacağız. Sonuç olarak model hiçbir zaman eylem gerçekleştirmez, komutların temsillerini geçirir ve bu komutları alıp seçtiğiniz ortamda yürütebilecek son tümleştirme katmanını sağlamanız gerekir.

Bu, olası anahtar eşlemelerinin kapsamlı bir listesi değildir. Bu listeyi gerektiği gibi genişletebilirsiniz. Bu sözlük, modeli Playwright ile entegre etmeye özeldir. İşletim sistemleri klavyenize/farenize API access sağlamak için modeli alternatif bir kitaplıkla tümleştirdiyseniz bu kitaplığa özgü bir eşleme sağlamanız gerekirdi.

# 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"
}

Bu sözlük, tuş adlarını Playwright'ın klavye API'sinin beklediği biçime çevirir. Her anahtar için birden çok ortak yazım desteklenir (örneğin, CTRL ve CONTROL her ikisi de Control ile eşlenir).

Koordinat doğrulama

Modelden geçirilen tüm fare eylemlerinin tarayıcı penceresi sınırları içinde kalmasını sağlamak için aşağıdaki yardımcı program işlevini ekleyeceğiz:

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

Eylem işleme

Tarayıcı otomasyonumuzun temeli, çeşitli kullanıcı etkileşimlerini işleyen ve bunları tarayıcıdaki eylemlere dönüştüren eylem işleyicidir. Dizideki actions[] eylemler düz sözlük olarak döndürülür.

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}")

Ekran görüntüsü yakalama

Modelin neyle etkileşimde olduğunu görebilmesi için ekran görüntülerini yakalayabileceği bir yol gerekir. Bu kod için ekran görüntülerini yakalamak için Playwright kullanıyoruz ve görünümü yalnızca tarayıcı penceresindeki içerikle sınırlandırıyoruz. Ekran görüntüsü url çubuğunu veya tarayıcı GUI'sinin diğer yönlerini içermez. Modelin ana tarayıcı penceresinin dışında görmesi gerekiyorsa kendi ekran görüntüsü işlevinizi oluşturarak modeli genişletebilirsiniz.

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

Bu işlev geçerli tarayıcı durumunu görüntü olarak yakalar ve modele gönderilmeye hazır base64 kodlu bir dize olarak döndürür. Bunu, modelin yürütmeye çalıştığı komutun başarılı olup olmadığını görmesine ve ardından ekran görüntüsünün içeriğine göre ayarlamalar yapmasına olanak tanıyan her adımdan sonra sürekli olarak bir döngüde yapacağız. Modelin ekran görüntüsü alması gerekip gerekmediğini belirlemesine izin verebiliriz, ancak kolaylık olması için her yineleme için bir ekran görüntüsü alınmasını zorlayacağız.

Model yanıt işlemesi

Bu işlev modelin yanıtlarını işler ve istenen eylemleri yürütür:

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.")

Bu bölümde şunları sağlayan kod ekledik:

  • Modelden metin ve akıl yürütme çıkarır ve görüntüler.
  • Bilgisayar eylem çağrılarını işler.
  • Kullanıcı onayı gerektiren olası güvenlik denetimlerini işler.
  • İstenen eylemleri yürütür (bir dizi dikte halinde toplu olarak).
  • Yeni bir ekran görüntüsü yakalar.
  • Güncelleştirilmiş durumu modele geri gönderir ve aracı tanımlar computer .
  • Bu işlemi birden çok yineleme için yineler.

Ana işlev

Ana işlev sürecin tamamını koordine eder:

    # 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())

Ana işlev:

  • OpenAI istemcisini başlatır.
  • Playwright tarayıcısını ayarlar.
  • Bing.com'da başlar.
  • Kullanıcı görevlerini kabul etmek için bir döngüye girer.
  • Başlangıç durumunu kaydeder.
  • Görevi ve ekran görüntüsünü modele gönderir.
  • Modelin yanıtını işler.
  • Kullanıcı çıkana kadar yineler.
  • Tarayıcının düzgün bir şekilde kapatılmasını sağlar.

Tam metin

Dikkat

Bu kod deneyseldir ve yalnızca tanıtım amaçlıdır. Yalnızca yanıtlar API'si ve gpt-5.4 modelinin temel akışını göstermeye yöneliktir. Bu kodu yerel bilgisayarınızda yürütebilirsiniz ancak bu kodu hassas verilere access olmayan düşük ayrıcalıklı bir sanal makinede çalıştırmanızı kesinlikle öneririz. Bu kod yalnızca temel test amaçlıdır.

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())

Ayrıca bakınız

  • Last updated on