さくらのAI Engineのマルチモーダル：画像を送ってテキストを受け取る

export SAKURA_AI_ENGINE_ACCOUNT_TOKEN="あなたのアクセストークン"

#!/usr/bin/env python3
"""v1/chat/completions で画像入力 + テキスト送信し、SSE ストリーミング (stream=True) で応答を受け取りながら finish_reason を判定するサンプル。"""

import base64
import os
from pathlib import Path

from openai import OpenAI

# ── 設定 ──────────────────────────────────────────
BASE_URL = "https://api.ai.sakura.ad.jp/v1"
API_KEY = os.environ.get("SAKURA_AI_ENGINE_ACCOUNT_TOKEN", "")

# マルチモーダル対応モデル一覧
MULTIMODAL_MODELS = [
    "preview/gemma-4-31B-it",
    "preview/Kimi-K2.6",
    "preview/Phi-4-multimodal-instruct",
    "preview/Qwen3-VL-30B-A3B-Instruct",
    "preview/Qwen3.6-35B-A3B",
]

DEFAULT_MODEL = MULTIMODAL_MODELS[0]  # preview/gemma-4-31B-it


# ── 関数 ──────────────────────────────────────────
def get_client() -> OpenAI:
    """OpenAI 互換クライアントを初期化して返す。"""
    if not API_KEY:
        raise RuntimeError(
            "環境変数 SAKURA_AI_ENGINE_ACCOUNT_TOKEN が設定されていません。"
            "さくらのAI Engineのアクセストークンを設定してください。"
        )
    return OpenAI(base_url=BASE_URL, api_key=API_KEY)


def encode_image(path: str) -> str:
    """画像ファイルを base64 エンコードして data URI を返す。"""
    p = Path(path)
    data = p.read_bytes()
    b64 = base64.b64encode(data).decode("utf-8")
    ext = p.suffix.lower().lstrip(".")
    if ext == "jpg":
        ext = "jpeg"
    return f"data:image/{ext};base64,{b64}"


# ── メイン ─────────────────────────────────────────
def main() -> None:
    client = get_client()
    image_path = Path("data") / "ai-back.png"

    if not image_path.exists():
        print(f"画像ファイルが見つかりません: {image_path}")
        print("先に data/ai-back.png を用意してください。")
        return

    image_url = encode_image(str(image_path))

    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "この画像に写っているものを日本語で詳細に説明してください。"},
                {"type": "image_url", "image_url": {"url": image_url}},
            ],
        },
    ]

    print(f"モデル: {DEFAULT_MODEL}")
    print(f"エンドポイント: POST {client.base_url}chat/completions (stream=True)")
    print(f"画像: {image_path}")
    print("-" * 40)

    # stream=True で SSE ストリーミングレスポンスを取得
    stream = client.chat.completions.create(
        model=DEFAULT_MODEL,
        messages=messages,
        max_tokens=4096,
        temperature=0.4,
        stream=True,
    )

    finish_reason = None

    # リアルタイムでトークンを逐次表示
    for chunk in stream:
        if chunk.choices:
            choice = chunk.choices[0]
            if choice.delta and choice.delta.content:
                print(choice.delta.content, end="", flush=True)
            if choice.finish_reason is not None:
                finish_reason = choice.finish_reason

    print()  # 改行
    print("-" * 40)

    # finish_reason による安全なハンドリング
    if finish_reason == "length":
        print("⚠️  [警告] レスポンスが max_tokens の上限に達したため途中で切れました。")
    elif finish_reason == "stop":
        print("✅ レスポンスは正常に完了しました (finish_reason: stop)。")
    elif finish_reason:
        print(f"ℹ️  完了理由 (finish_reason): {finish_reason}")


if __name__ == "__main__":
    main()

[project]
name = "20260719-sakura-ai-multimodal"
version = "0.1.0"
description = "さくらのAI Engine マルチモーダル呼び出しサンプル"
readme = "README.md"
requires-python = ">=3.14"
dependencies = [
    "openai>=2.45.0",
]

[project.optional-dependencies]
dev = []

uv sync
uv run chat_completions_multimodal_stream.py

{"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,/9j/4AAQ..."}}

#!/usr/bin/env python3
"""v1/messages (Anthropic互換) で画像入力 + テキスト送信し、通常応答(非ストリーミング)で受け取るサンプル。"""
import base64
import os
from pathlib import Path
import anthropic

# ── 設定 ──────────────────────────────────────────
BASE_URL = "https://api.ai.sakura.ad.jp"
AUTH_TOKEN = os.environ.get("SAKURA_AI_ENGINE_ACCOUNT_TOKEN", "")

MULTIMODAL_MODELS = [
    "preview/gemma-4-31B-it",
    "preview/Kimi-K2.6",
    "preview/Phi-4-multimodal-instruct",
    "preview/Qwen3-VL-30B-A3B-Instruct",
    "preview/Qwen3.6-35B-A3B",
]
DEFAULT_MODEL = MULTIMODAL_MODELS[0]  # preview/gemma-4-31B-it


# ── 関数 ──────────────────────────────────────────
def get_client() -> anthropic.Anthropic:
    """Anthropic互換クライアントを初期化して返す。"""
    if not AUTH_TOKEN:
        raise RuntimeError(
            "環境変数 SAKURA_AI_ENGINE_ACCOUNT_TOKEN が設定されていません。"
            "さくらのAI Engineのアクセストークンを設定してください。"
        )
    # さくらのAI EngineはAuthorization: Bearer形式のため auth_token を使う (api_key ではない)
    return anthropic.Anthropic(base_url=BASE_URL, auth_token=AUTH_TOKEN)


def encode_image(path: str) -> tuple[str, str]:
    """画像ファイルを base64 エンコードし、(media_type, base64データ) を返す。"""
    p = Path(path)
    data = p.read_bytes()
    b64 = base64.b64encode(data).decode("utf-8")
    ext = p.suffix.lower().lstrip(".")
    if ext == "jpg":
        ext = "jpeg"
    return f"image/{ext}", b64


# ── メイン ─────────────────────────────────────────
def main() -> None:
    client = get_client()

    image_path = Path("data") / "ai-back.png"
    if not image_path.exists():
        print(f"画像ファイルが見つかりません: {image_path}")
        print("先に data/ai-back.png を用意してください。")
        return

    media_type, image_b64 = encode_image(str(image_path))
    messages = [
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": image_b64,
                    },
                },
                {"type": "text", "text": "この画像に写っているものを日本語で詳細に説明してください。"},
            ],
        },
    ]

    print(f"モデル: {DEFAULT_MODEL}")
    print(f"エンドポイント: POST {BASE_URL}/v1/messages")
    print(f"画像: {image_path}")
    print("-" * 40)

    response = client.messages.create(
        model=DEFAULT_MODEL,
        max_tokens=4096,
        temperature=0.4,
        messages=messages,
    )

    text = "".join(block.text for block in response.content if block.type == "text")
    print(text)
    print("-" * 40)

    stop_reason = response.stop_reason
    if stop_reason == "max_tokens":
        print("⚠️  [警告] レスポンスが max_tokens の上限に達したため途中で切れました。")
    elif stop_reason == "end_turn":
        print("✅ レスポンスは正常に完了しました (stop_reason: end_turn)。")
    elif stop_reason:
        print(f"ℹ️  完了理由 (stop_reason): {stop_reason}")


if __name__ == "__main__":
    main()