Geminiネイティブ（画像、ストリーミング）

curl --no-buffer --request POST "https://api.mixroute.ai/v1/models/gemini-nano-banana-2.1:streamGenerateContent?alt=sse" \
  --header "Authorization: Bearer $MIXROUTE_API_KEY" \
  --header "Content-Type: application/json" \
  --data '{
  "contents": [
    {
      "role": "user",
      "parts": [
        {
          "text": "A clean product photograph of a red ceramic mug on a white background."
        }
      ]
    }
  ],
  "generationConfig": {
    "responseModalities": [
      "TEXT",
      "IMAGE"
    ],
    "imageConfig": {
      "aspectRatio": "1:1",
      "imageSize": "1K"
    }
  }
}'

import base64
import json
import os
from pathlib import Path
import requests

payload = json.loads(r'''
{
  "contents": [
    {
      "role": "user",
      "parts": [
        {
          "text": "A clean product photograph of a red ceramic mug on a white background."
        }
      ]
    }
  ],
  "generationConfig": {
    "responseModalities": [
      "TEXT",
      "IMAGE"
    ],
    "imageConfig": {
      "aspectRatio": "1:1",
      "imageSize": "1K"
    }
  }
}
''')

def events(lines):
    data = []
    for line in lines:
        if line == "":
            if data:
                yield "\n".join(data)
                data.clear()
        elif line.startswith("data:"):
            value = line[5:]
            data.append(value[1:] if value.startswith(" ") else value)
    if data:
        yield "\n".join(data)

url = "https://api.mixroute.ai/v1/models/gemini-nano-banana-2.1:streamGenerateContent?alt=sse"
image_index = 0
with requests.post(url, headers={"Authorization": "Bearer " + os.environ["MIXROUTE_API_KEY"]}, json=payload, stream=True, timeout=(30, 180)) as response:
    response.raise_for_status()
    response.encoding = "utf-8"
    for data in events(response.iter_lines(decode_unicode=True)):
        if not data.strip():
            continue
        if data.strip() == "[DONE]":
            break
        chunk = json.loads(data)
        if "error" in chunk:
            raise RuntimeError(chunk["error"])
        for candidate in chunk.get("candidates", []):
            for part in candidate.get("content", {}).get("parts", []):
                if part.get("thought"):
                    continue
                if "inlineData" in part:
                    inline = part["inlineData"]
                    if inline.get("data"):
                        extension = {"image/png": "png", "image/jpeg": "jpg", "image/webp": "webp"}.get(inline.get("mimeType"), "bin")
                        Path(f"image_{image_index}.{extension}").write_bytes(base64.b64decode(inline["data"], validate=True))
                        image_index += 1
                elif "text" in part:
                    print(part["text"], end="")