OpenAI API で Computer use を実装する

export OPENAI_API_KEY=sk-...

set OPENAI_API_KEY=sk-...

git clone https://github.com/openai/openai-cua-sample-app.git
cd openai-cua-sample-app
pip install -r requirements.txt

playwright install

openai-cua-sample-app % python cli.py --computer local-playwright --input "ぼっち・ざ・ろっくのWebサイトを探してください。"

New page created
screenshot({})
click({'button': 'left', 'x': 393, 'y': 185})
type({'text': 'ぼっち・ざ・ろっく'})
keypress({'keys': ['ENTER']})
wait({})
scroll({'x': 524, 'y': 699, 'scroll_x': 0, 'scroll_y': 483})
click({'button': 'left', 'x': 346, 'y': 593})
wait({})
New page created
I found the official website for "ぼっち・ざ・ろっく!" listed in the search results. Should I visit it?
>

from agent.agent import Agent
from computers import LocalPlaywrightComputer

# メインの定義
def main(user_input=None):
    with LocalPlaywrightComputer() as computer:
        # Agentとメッセージリストの準備
        agent = Agent(computer=computer)
        items = []

        # メインループ
        while True:
            # ユーザー入力
            user_input = input("> ")
            items.append({"role": "user", "content": user_input})

            # フルターンの実行
            output_items = agent.run_full_turn(items, debug=True, show_images=True)
            items += output_items

# メインの実行
if __name__ == "__main__":
    main()

from computers import Computer
from utils import (
    create_response,
    show_image,
    pp,
    sanitize_message,
    check_blocklisted_url,
)
import json
from typing import Callable

# Agent
class Agent:
    # 初期化
    def __init__(
        self,
        model="computer-use-preview",
        computer: Computer = None,
        tools: list[dict] = [],
        acknowledge_safety_check_callback: Callable = lambda: False,
    ):
        self.model = model
        self.computer = computer
        self.tools = tools
        self.print_steps = True
        self.debug = False
        self.show_images = False
        self.acknowledge_safety_check_callback = acknowledge_safety_check_callback

        if computer:
            self.tools += [
                {
                    "type": "computer-preview",
                    "display_width": computer.dimensions[0],
                    "display_height": computer.dimensions[1],
                    "environment": computer.environment,
                },
            ]

    # デバッグ出力
    def debug_print(self, *args):
        if self.debug:
            pp(*args)

    # 各項目を処理
    def handle_item(self, item):
        # メッセージを標準出力
        if item["type"] == "message":
            if self.print_steps:
                print(item["content"][0]["text"])

        # Function Callingを実行してその結果を返す
        if item["type"] == "function_call":
            name, args = item["name"], json.loads(item["arguments"])
            if self.print_steps:
                print(f"{name}({args})")

            # FunctionがComputer上に存在する場合はそれを呼び出す
            if hasattr(self.computer, name):
                method = getattr(self.computer, name)
                method(**args)
            return [
                {
                    "type": "function_call_output",
                    "call_id": item["call_id"],
                    "output": "success",
                }
            ]

        # Computerを操作してスクリーンショットを返す
        if item["type"] == "computer_call":
            action = item["action"]
            action_type = action["type"]
            action_args = {k: v for k, v in action.items() if k != "type"}
            if self.print_steps:
                print(f"{action_type}({action_args})")

            method = getattr(self.computer, action_type)
            method(**action_args)

            screenshot_base64 = self.computer.screenshot()
            if self.show_images:
                show_image(screenshot_base64)

            # ユーザーがすべてのセーフティチェックに応答しない場合はエラーで終了
            pending_checks = item.get("pending_safety_checks", [])
            for check in pending_checks:
                message = check["message"]
                if not self.acknowledge_safety_check_callback(message):
                    raise ValueError(
                        f"Safety check failed: {message}. Cannot continue with unacknowledged safety checks."
                    )

            call_output = {
                "type": "computer_call_output",
                "call_id": item["call_id"],
                "acknowledged_safety_checks": pending_checks,
                "output": {
                    "type": "input_image",
                    "image_url": f"data:image/png;base64,{screenshot_base64}",
                },
            }

            # ブラウザ環境向けの追加のURLセーフティチェック
            if self.computer.environment == "browser":
                current_url = self.computer.get_current_url()
                check_blocklisted_url(current_url)
                call_output["output"]["current_url"] = current_url

            return [call_output]
        return []

    # フルターンの実行
    def run_full_turn(
        self, input_items, print_steps=True, debug=False, show_images=False
    ):
        self.print_steps = print_steps
        self.debug = debug
        self.show_images = show_images
        new_items = []

        # Assistantメッセージが得られるまでループ
        while new_items[-1].get("role") != "assistant" if new_items else True:
            self.debug_print([sanitize_message(msg) for msg in input_items + new_items])

            # 「computer-use-preview」の推論を実行
            response = create_response(
                model=self.model,
                input=input_items + new_items,
                tools=self.tools,
                truncation="auto",
            )
            self.debug_print(response)

            if "output" not in response and self.debug:
                print(response)
                raise ValueError("No output from model")
            else:
                # 推論結果をメッセージリストに追加
                new_items += response["output"]

                # Computerを操作してスクリーンショットをメッセージリストに追加
                for item in response["output"]:
                    new_items += self.handle_item(item)

        return new_items

import os
import requests
from dotenv import load_dotenv
import json
import base64
from PIL import Image
from io import BytesIO
import io
from urllib.parse import urlparse

load_dotenv(override=True)

# ブロックリスト
BLOCKED_DOMAINS = [
    "maliciousbook.com",
    "evilvideos.com",
    "darkwebforum.com",
    "shadytok.com",
    "suspiciouspins.com",
    "ilanbigio.com",
]

# JSON出力
def pp(obj):
    print(json.dumps(obj, indent=4))

# 画像の表示
def show_image(base_64_image):
    image_data = base64.b64decode(base_64_image)
    image = Image.open(BytesIO(image_data))
    image.show()

# 画像サイズの取得
def calculate_image_dimensions(base_64_image):
    image_data = base64.b64decode(base_64_image)
    image = Image.open(io.BytesIO(image_data))
    return image.size

# computer_call_output メッセージの場合、image_url を省略したメッセージのコピーを返す
def sanitize_message(msg: dict) -> dict:
    if msg.get("type") == "computer_call_output":
        output = msg.get("output", {})
        if isinstance(output, dict):
            sanitized = msg.copy()
            sanitized["output"] = {**output, "image_url": "[omitted]"}
            return sanitized
    return msg

# 推論の実行
def create_response(**kwargs):
    url = "https://api.openai.com/v1/responses"
    headers = {
        "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
        "Content-Type": "application/json"
    }

    openai_org = os.getenv("OPENAI_ORG")
    if openai_org:
        headers["Openai-Organization"] = openai_org

    response = requests.post(url, headers=headers, json=kwargs)

    if response.status_code != 200:
        print(f"Error: {response.status_code} {response.text}")

    return response.json()

# 指定された URL (サブドメインを含む) がブロックリストに含まれている場合はエラー
def check_blocklisted_url(url: str) -> None:
    hostname = urlparse(url).hostname or ""
    if any(
        hostname == blocked or hostname.endswith(f".{blocked}")
        for blocked in BLOCKED_DOMAINS
    ):
        raise ValueError(f"Blocked URL: {url}")