Azure OpenAIでGPT-Image-1をPythonのopenaiライブラリから使用する方法まとめ【注意点・トークン消費検証】

'openai.BadRequestError: Error code: 400 - {\'error\': {\'message\': "Model not supported with Responses API. Supported models are: [\'gpt-image-1\', \'gpt-image-0721-mini-alpha\']", \'type\': \'invalid_request_error\', \'param\': None, \'code\': None}}\n']

unknown parameter: 'response_format'.", 'type': 'invalid_request_error', 'param': 'response_format', 'code': 'unknown_parameter'

### ストリーミングなしの場合

openai.BadRequestError: Error code: 400 - {'error': {'message': 'Your request was rejected by the safety system. If you believe this is an error, contact us at oai-support@microsoft.com and include the request ID xxxxxx. safety_violations=[illicit].', 'type': 'image_generation_user_error', 'param': None, 'code': 'moderation_blocked'}} 

### ストリーミングの場合
openai.APIError: Your request was rejected by the safety system. If you believe this is an error, contact us at oai-support@microsoft.com and include the request ID xxxxxx." 

azure-identity
aiohttp
openai

import asyncio
import base64
import traceback
import time
import os
from typing import Union
from openai import (
    AsyncAzureOpenAI,
    BadRequestError,
    APIError,
)
from azure.identity.aio import (
    DefaultAzureCredential,
    get_bearer_token_provider
)

SCOPES = "https://cognitiveservices.azure.com/.default"
OPENAI_RESOURCE_NAME = "TODO"
ENDPOINT = f"https://{OPENAI_RESOURCE_NAME}.openai.azure.com/"
MODEL_DEPLOYMENT_NAME = "gpt-image-1"
API_VERSION = "2025-04-01-preview"
API_KEY = "TODO"
OUTPUT = "./output"
BASENAME = "image"
PROMPT = "生成AIチャットボットのアイコンを作成せよ"
EDIT_PROMPT = "2つの画像を組み合わせて、背景の色を灰色にして"
MASK_PROMPT = "笑顔の犬"

async def generate_image(
        user_prompt: str,
        background: str = "auto",
        moderation: str = "low",
        output_compression: int = 100,
        output_format: str = "png",
        output_file_basename: str = "image",
        quality: str = "low",
        timeout: int = 600,
        n: int = 1,
        size: str = "1024x1024",
        stream_flag: bool = False,
        partial_images: int = 0
    ):
    """画像生成
    Args:
        user_prompt (str): ユーザープロンプト
        background (str, optional): 透明 or 不透明 or 自動. Defaults to "auto".
        moderation (str, optional): 生成画像のモデレーションの厳しさ. Defaults to "low".
        output_compression (int, optional): 0-100, jpegの場合のみ有効. Defaults to 100.
        output_format (str, optional): png or jpeg. Defaults to "png".
        output_file_basename (str, optional): 生成画像のファイル名ベース. Defaults to "image".
        quality (str, optional): low, medium, high, auto. Defaults to "low".
        timeout (int, optional): タイムアウト秒数. Defaults to 600.
        n (int, optional): 生成する画像の数: 1 - 10. Defaults to 1.
        size (str, optional): auto, 1024x1024, 1024x1536, 1536x1024. Defaults to "1024x1024".
        stream_flag (bool, optional): ストリーミングフラグ. Defaults to False.
        partial_images (int, optional): 生成する部分画像の数 <=3 (streamingの場合のみ有���). Defaults to 0.
    Returns:
        list: 生成された画像のファイル名リスト
    """
    print("Begin to generate image")

    image_file_names = []

    credential = DefaultAzureCredential()
    # noinspection PyTypeChecker
    token_provider = get_bearer_token_provider(
        credential,
        SCOPES
    )
    client = AsyncAzureOpenAI(
        azure_endpoint=ENDPOINT,
        api_key=API_KEY,
        api_version=API_VERSION
    )

    # 実行時間計測開始
    begin_time = time.time()
    total_tokens = 0

    try:
        if stream_flag:
            print("Streaming mode")
            file_basename = f"{output_file_basename}_streaming_{quality}"
            # noinspection PyTypeChecker
            response = await client.images.generate(
                model=MODEL_DEPLOYMENT_NAME,
                prompt=user_prompt,
                background=background,
                moderation=moderation,
                output_compression=output_compression,
                output_format=output_format,
                quality=quality,
                timeout=timeout,
                size=size,
                stream=stream_flag,
                n=n, # 生成する画像の数: streamingの場合は1のみ
                partial_images=partial_images # streamingの場合のみ生成する部分画像の数 <=3
            )
            print(f"response [{type(response)}], attributes: [{response.__dict__.keys()}]")
            image_size = 0
            # openai.AsyncStream
            async for event in response:
                created_at = event.created_at
                print(f"[{type(event)}]: event_type:[{event.type}]")

                # openai.types.image_gen_partial_image_event.ImageGenPartialImageEvent
                if event.type == "image_generation.partial_image":
                    idx = event.partial_image_index
                    image_base64 = event.b64_json
                    image_bytes = base64.b64decode(image_base64)
                    image_size = len(image_bytes)
                    file_name = f"{file_basename}_{idx}.{output_format}"
                    with open(file_name, "wb") as f:
                        f.write(image_bytes)
                        print(f"No [{idx}] image saved as [{file_name}], size: [{image_size}] bytes")
                    image_file_names.append(file_name)

                # openai.types.image_gen_completed_event.ImageGenCompletedEvent
                elif event.type == "image_generation.completed":
                    image_base64 = event.b64_json
                    image_bytes = base64.b64decode(image_base64)
                    image_size = len(image_bytes)
                    file_name = f"{file_basename}.{output_format}"
                    with open(file_name, "wb") as f:
                        f.write(image_bytes)
                        print(f"Last image saved as [{file_name}], size: [{image_size}] bytes")
                    image_file_names.append(file_name)

                    # トークン使用量
                    input_token = event.usage.input_tokens
                    input_image_token = event.usage.input_tokens_details.image_tokens
                    input_text_token = event.usage.input_tokens_details.text_tokens
                    output_token = event.usage.output_tokens
                    total_tokens = input_token + output_token
                    print(
                        f"[{created_at}], quality: [{quality}], size: [{size}], compression: [{output_compression}], input_token: [{input_token}], input_image_token: [{input_image_token}], input_text_token: [{input_text_token}], output_token: [{output_token}], total_tokens: [{total_tokens}], image_size: [{image_size}] bytes")
        else:
            print("Non streaming mode")
            file_basename = f"{output_file_basename}_non_streaming_{quality}"
            # openai.types.images_response.ImagesResponse
            # noinspection PyTypeChecker
            response = await client.images.generate(
                model=MODEL_DEPLOYMENT_NAME,
                prompt=user_prompt,
                background=background,
                moderation=moderation,
                output_compression=output_compression,
                output_format=output_format,
                quality=quality,
                timeout=timeout,
                n=n,
                size=size,
                stream=stream_flag
            )
            print(f"response [{type(response)}], attributes: [{response.__dict__.keys()}]")
            created = response.created
            input_token = response.usage.input_tokens
            input_image_token = response.usage.input_tokens_details.image_tokens
            input_text_token = response.usage.input_tokens_details.text_tokens
            output_token = response.usage.output_tokens
            total_tokens = input_token + output_token

            image_data_list = response.data
            for idx, image in enumerate(image_data_list):
                image_base64 = image.b64_json
                image_bytes = base64.b64decode(image_base64)
                image_size = len(image_bytes)
                file_name = f"{file_basename}_{idx}.{output_format}"
                with open(file_name, "wb") as f:
                    f.write(image_bytes)
                    print(f"Image [{idx}] saved as [{file_name}]")
                print(f"[{created}], No: [{idx}], quality: [{response.quality}], size: [{response.size}], compression: [{output_compression}], input_token: [{input_token}], input_image_token: [{input_image_token}], input_text_token: [{input_text_token}], output_token: [{output_token}], total_tokens: [{total_tokens}], image_size: [{image_size}] bytes")
                image_file_names.append(file_name)

    except BaseException as e:
        stack_traces = list(traceback.TracebackException.from_exception(e).format())
        exception_message = stack_traces[-1].replace("\n", "")

        if isinstance(e, (BadRequestError, APIError)):
            # コンテンツフィルターでエラーになった場合
            if exception_message.find("Your request was rejected by the safety system") != -1:
                print(f"Warning: Your request was rejected by the safety system. <{user_prompt}>")
            else:
                print(f"BadRequestError: [{exception_message}] <{stack_traces}>")
        else:
            print(f"Error: [{exception_message}] <{stack_traces}>")

    finally:
        # 実行時間計測終了
        end_time = time.time()
        execution_time = end_time - begin_time
        await client.close()
        await credential.close()
        print("Client closed")
        print(f"End to generate image. execution time: [{execution_time}] seconds, list [{image_file_names}], total_tokens: [{total_tokens}]")

# noinspection PyTypeChecker
async def edit_image(
        image: list,
        user_prompt: str,
        background: str = "auto",
        output_compression: int = 100,
        output_format: str = "png",
        output_file_basename: str = "image",
        input_fidelity: str ="low",
        quality: str = "low",
        timeout: int = 600,
        n: int = 1,
        size: str = "1024x1024",
        stream_flag: bool = False,
        partial_images: int = 0,
        mask: Union[tuple, str, bytes, bytearray, None] = None
):
    """画像編集
    Args:
        image: Union[FileTypes, SequenceNotStr[FileTypes]]: 画像ファイル or 画像ファイルリスト
        user_prompt (str): ユーザープロンプト
        background (str, optional): 透明 or 不透明 or 自動. Defaults to "auto".
        output_compression (int, optional): 0-100, jpegの場合のみ有効. Defaults to 100.
        output_format (str, optional): png or jpeg. Defaults to "png".
        output_file_basename (str, optional): 生成画像のファイル名ベース. Defaults to "image".
        input_fidelity (str, optional): 入力画像の史実度 low, high. Defaults to "low".
        quality (str, optional): low, medium, high, auto. Defaults to "low".
        timeout (int, optional): タイムアウト秒数. Defaults to 600.
        n (int, optional): 生成する画像の数: 1 - 10. Defaults to 1.
        size (str, optional): auto, 1024x1024, 1024x1536, 1536x1024. Defaults to "1024x1024".
        stream_flag (bool, optional): ストリーミングフラグ. Defaults to False.
        partial_images (int, optional): 生成する部分画像の数 <=3 (streamingの場合のみ有効). Defaults to 0.
        mask (FileTypes | NotGiven, optional): マスク画像ファイル. Defaults to NOT_GIVEN.
    Returns:
        list: 生成された画像のファイル名リスト
    """
    print("Begin to edit image")

    image_file_names = []

    credential = DefaultAzureCredential()
    # noinspection PyTypeChecker
    token_provider = get_bearer_token_provider(
        credential,
        SCOPES
    )
    client = AsyncAzureOpenAI(
        azure_endpoint=ENDPOINT,
        api_key=API_KEY,
        api_version=API_VERSION
    )

    # 実行時間計測開始
    begin_time = time.time()
    total_tokens = 0

    try:
        if stream_flag:
            print("Streaming mode")
            file_basename = f"{output_file_basename}_streaming_{quality}_{input_fidelity}"
            print(f"file_basename: {file_basename}, mask: {mask is not None}")
            if mask is None:
                # noinspection PyTypeChecker
                response = await client.images.edit(
                    model=MODEL_DEPLOYMENT_NAME,
                    prompt=user_prompt,
                    background=background,
                    image=image,
                    input_fidelity=input_fidelity,
                    output_compression=output_compression,
                    output_format=output_format,
                    quality=quality,
                    timeout=timeout,
                    size=size,
                    stream=stream_flag,
                    n=n, # 生成する画像の数: streamingの場合は1のみ
                    partial_images=partial_images # streamingの場合のみ生成する部分画像の数 <=3
                )
            else:
                # noinspection PyTypeChecker
                response = await client.images.edit(
                    model=MODEL_DEPLOYMENT_NAME,
                    prompt=user_prompt,
                    background=background,
                    image=image,
                    input_fidelity=input_fidelity,
                    output_compression=output_compression,
                    output_format=output_format,
                    quality=quality,
                    timeout=timeout,
                    size=size,
                    stream=stream_flag,
                    n=n, # 生成する画像の数: streamingの場合は1のみ
                    partial_images=partial_images, # streamingの場合のみ生成する部分画像の��� <=3
                    mask=mask
                )
            print(f"response [{type(response)}], attributes: [{response.__dict__.keys()}]")
            image_size = 0
            # openai.AsyncStream
            async for event in response:
                created_at = event.created_at
                print(f"[{type(event)}]: event_type:[{event.type}]")

                # openai.types.image_edit_partial_image_event.ImageGenPartialImageEvent
                if event.type == "image_edit.partial_image":
                    idx = event.partial_image_index
                    image_base64 = event.b64_json
                    image_bytes = base64.b64decode(image_base64)
                    image_size = len(image_bytes)
                    file_name = f"{file_basename}_{idx}.{output_format}"
                    with open(file_name, "wb") as f:
                        f.write(image_bytes)
                        print(f"No [{idx}] image saved as [{file_name}], size: [{image_size}] bytes")
                    image_file_names.append(file_name)

                # openai.types.image_edit_completed_event.ImageGenCompletedEvent
                elif event.type == "image_edit.completed":
                    image_base64 = event.b64_json
                    image_bytes = base64.b64decode(image_base64)
                    image_size = len(image_bytes)
                    file_name = f"{file_basename}.{output_format}"
                    with open(file_name, "wb") as f:
                        f.write(image_bytes)
                        print(f"Last image saved as [{file_name}], size: [{image_size}] bytes")
                    image_file_names.append(file_name)

                    # トークン使用量
                    input_token = event.usage.input_tokens
                    input_image_token = event.usage.input_tokens_details.image_tokens
                    input_text_token = event.usage.input_tokens_details.text_tokens
                    output_token = event.usage.output_tokens
                    total_tokens = input_token + output_token
                    print(
                        f"[{created_at}], quality: [{quality}], size: [{size}], compression: [{output_compression}], input_token: [{input_token}], input_image_token: [{input_image_token}], input_text_token: [{input_text_token}], output_token: [{output_token}], total_tokens: [{total_tokens}], image_size: [{image_size}] bytes")
        else:
            print("Non streaming mode")
            file_basename = f"{output_file_basename}_non_streaming_{quality}_{input_fidelity}"
            print(f"file_basename: {file_basename}, mask: {mask is not None}")
            # openai.types.images_response.ImagesResponse
            # noinspection PyTypeChecker
            if mask is None:
                response = await client.images.edit(
                    model=MODEL_DEPLOYMENT_NAME,
                    prompt=user_prompt,
                    background=background,
                    image=image,
                    input_fidelity=input_fidelity,
                    output_compression=output_compression,
                    output_format=output_format,
                    quality=quality,
                    timeout=timeout,
                    size=size,
                    stream=stream_flag,
                    n=n # 生成する画像の数: streamingの場合は1のみ
                )
            else:
                response = await client.images.edit(
                    model=MODEL_DEPLOYMENT_NAME,
                    prompt=user_prompt,
                    background=background,
                    image=image,
                    input_fidelity=input_fidelity,
                    output_compression=output_compression,
                    output_format=output_format,
                    quality=quality,
                    timeout=timeout,
                    size=size,
                    stream=stream_flag,
                    n=n, # 生成する画像の数: streamingの場合は1のみ
                    mask=mask
                )
            print(f"response [{type(response)}], attributes: [{response.__dict__.keys()}]")
            created = response.created
            input_token = response.usage.input_tokens
            input_image_token = response.usage.input_tokens_details.image_tokens
            input_text_token = response.usage.input_tokens_details.text_tokens
            output_token = response.usage.output_tokens
            total_tokens = input_token + output_token

            image_data_list = response.data
            for idx, image in enumerate(image_data_list):
                image_base64 = image.b64_json
                image_bytes = base64.b64decode(image_base64)
                image_size = len(image_bytes)
                file_name = f"{file_basename}_edit_{quality}_{idx}.{output_format}"
                with open(file_name, "wb") as f:
                    f.write(image_bytes)
                    print(f"Image [{idx}] saved as [{file_name}]")
                print(f"[{created}], No: [{idx}], quality: [{response.quality}], size: [{response.size}], compression: [{output_compression}], input_token: [{input_token}], input_image_token: [{input_image_token}], input_text_token: [{input_text_token}], output_token: [{output_token}], total_tokens: [{total_tokens}], image_size: [{image_size}] bytes")
                image_file_names.append(file_name)

    except BaseException as e:
        stack_traces = list(traceback.TracebackException.from_exception(e).format())
        exception_message = stack_traces[-1].replace("\n", "")

        if isinstance(e, (BadRequestError, APIError)):
            # コンテンツフィルターでエラーになった場合
            if exception_message.find("Your request was rejected by the safety system") != -1:
                print(f"Warning: Your request was rejected by the safety system. <{user_prompt}>")
            else:
                print(f"BadRequestError: [{exception_message}] <{stack_traces}>")
        else:
            print(f"Error: [{exception_message}] <{stack_traces}>")

    finally:
        # 実行時間計測終了
        end_time = time.time()
        execution_time = end_time - begin_time
        await client.close()
        await credential.close()
        print("Client closed")
        print(f"End to edit image. execution time: [{execution_time}] seconds, list [{image_file_names}], total_tokens: [{total_tokens}]")


if __name__ == "__main__":
    user_prompt = PROMPT
    images = []
    input_fidelity = "low"
    output_image_file_base = f"{BASENAME}"
    mask_file_type = None

    if not os.path.exists(OUTPUT):
        os.makedirs(OUTPUT)
        print(f"Directory {OUTPUT} created.")

    # ユーザー入力
    prompt_type = input("please input prompt type [generate, edit] (default: generate): ") or "generate"
    if prompt_type == "generate":
        user_prompt = input(f"please input prompt (default: {PROMPT}): ") or PROMPT
    else:
        input_fidelity = input(f"please input input quality [low, high] (default: low): ") or "low"
        if prompt_type == "edit":
            user_prompt = input(f"please input edit prompt (default: {EDIT_PROMPT}): ") or EDIT_PROMPT

            # デフォルト画像パス
            default_path = "./image_0.png"
            default_mask_path = "./mask.png"

            # ファイルパスをテキスト入力で受け取る
            continue_response = "y"
            while continue_response.lower() == 'y':
                exist_file_path = input(f"Please input file path (default: {default_path}): ") or default_path
                with open(exist_file_path, "rb") as image_file:
                    image_data = image_file.read()
                    file_type = (os.path.basename(exist_file_path), image_data, "image/png")
                    images.append(file_type)
                continue_response = input("Do you want to add another image? [y/N]: ") or "N"

            # デフォルト画像パス
            # ファイル選択ダイアログ
            mask_file_path = input(f"Please input mask file path (default: None): ") or None
            if not mask_file_path is None:
                with open(mask_file_path, "rb") as mask_file:
                    mask_data = mask_file.read()
                    mask_file_type = (os.path.basename(mask_file_path), mask_data, "image/png")

    stream_flag_str = input(f"please input stream flag [True, False] (default: False): ") or "False"
    stream_flag = bool(stream_flag_str == "True")
    if stream_flag:
        n = 1
        partial_images = int(input(f"please input number of partial images to generate (1-3) (streaming only) (default: 2): ") or 2)
    else:
        n = int(input(f"please input number of images to generate (1-10) (default: 1): ") or 1)
        partial_images = 0
    quality = input(f"please input quality [low, medium, high, auto] (default: low): ") or "low"
    background = input(f"please input background [transparent, opaque, auto] (default: auto): ") or "auto" # 透明 or 不透明 or 自動
    moderation = input(f"please input moderation [low, auto] (default: low): ") or "low" # 生成画像のモデレーションの厳しさ
    output_format = input(f"please input output format [png, jpeg] (default: png): ") or "png"
    if output_format == "jpeg":
        output_compression = int(input(f"please input output compression [0-100] (default: 100): ") or 100)
    else:
        output_compression = 100
    output_image_file_base = input(f"please input output image file base name (default: {BASENAME}): ") or BASENAME
    output_image_file_base = f"{OUTPUT}/{output_image_file_base}"
    size = input(f"please input image size [auto, 1024x1024, 1024x1536, 1536x1024] (default: 1024x1024): ") or "1024x1024"
    timeout = int(input(f"please input timeout in seconds (default: 600): ") or 600)

    # 画像生成
    if prompt_type == "generate":
        asyncio.run(generate_image(
            user_prompt=user_prompt,
            background=background,
            moderation=moderation,
            output_compression=output_compression,
            output_format=output_format,
            output_file_basename=output_image_file_base,
            quality=quality,
            timeout=timeout,
            n=n,
            size=size,
            stream_flag=stream_flag,
            partial_images=partial_images
        ))

    # エディットモード ----------------------------------------------------------------------------
    # 2つの画像を組み合わせ
    elif prompt_type == "edit":
        asyncio.run(edit_image(
            image=images,
            user_prompt=user_prompt,
            background=background,
            output_compression=output_compression,
            output_format=output_format,
            output_file_basename=output_image_file_base,
            input_fidelity=input_fidelity,
            quality=quality,
            timeout=timeout,
            n=n,
            size=size,
            stream_flag=stream_flag,
            partial_images=partial_images,
            mask=mask_file_type
        ))