高度な画像生成: ワークフローとリアリズム

import os
import requests

BASE = "https://api.apiyi.com/v1"
API_KEY = os.environ["APIYI_API_KEY"]          # never hard-code the key

REWRITE_SYSTEM = """You are an image prompt engineer. Rewrite the user's casual brief
into one structured image prompt.

Fill in all six elements. Supply whatever is missing; never ask the user:
1 Subject: material, colour, count, state
2 Environment: what the background is, what is sharp and what is blurred
3 Light: direction, hardness, fill or no fill — there must be one identifiable key light
4 Lens and angle: focal length, aperture, camera height, tilt
5 Grading and medium: white balance bias, saturation, film or digital character
6 Composition: where the subject sits in the frame, where the negative space is

Rules:
- Output only the prompt body: no explanation, no bullet points, no heading
- No brand names, logos, or legible text unless the user asked for them
- Never use vague quality words such as 8K, ultra HD, masterpiece, perfect
- Keep any element the user specified exactly as written"""


def rewrite(user_prompt: str) -> str:
    r = requests.post(
        f"{BASE}/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "gemini-3.5-flash",
            "messages": [
                {"role": "system", "content": REWRITE_SYSTEM},
                {"role": "user", "content": user_prompt},
            ],
        },
        timeout=60,
    )
    r.raise_for_status()
    return r.json()["choices"][0]["message"]["content"].strip()

import base64
import json
import os
from concurrent.futures import ThreadPoolExecutor

import requests

BASE = "https://api.apiyi.com"
API_KEY = os.environ["APIYI_API_KEY"]
HEAD = {"Authorization": f"Bearer {API_KEY}"}

STYLE_CONST = "Cool neutral white balance, low saturation, clean frame with generous negative space."


def draw(prompt: str, size: str = "2K", aspect: str = "1:1") -> bytes:
    """Generate one image (Nano Banana Pro, native Gemini endpoint)."""
    url = f"{BASE}/v1beta/models/gemini-3-pro-image:generateContent"
    body = {
        "contents": [{"parts": [{"text": prompt}]}],
        "generationConfig": {
            "responseModalities": ["IMAGE"],
            "imageConfig": {"aspectRatio": aspect, "imageSize": size},
        },
    }
    r = requests.post(url, headers=HEAD, json=body, timeout=600)   # headroom for 4K
    r.raise_for_status()
    parts = r.json()["candidates"][0]["content"]["parts"]
    part = next((p for p in parts if p.get("inlineData")), None)
    if part is None:                            # HTTP 200 with no image usually means moderation
        raise RuntimeError("no image returned: " + json.dumps(parts)[:300])
    return base64.b64decode(part["inlineData"]["data"])


def score(image: bytes, prompt: str) -> dict:
    """Score a candidate with a vision model; returns per-dimension scores and one issue line."""
    data_url = "data:image/png;base64," + base64.b64encode(image).decode()
    rubric = (
        "Score this image and return strict JSON: "
        '{"instruction":0-10,"anatomy":0-10,"text":0-10,"texture":0-10,'
        '"composition":0-10,"total":0-50,"issue":"one sentence"}. '
        "instruction = does it satisfy the brief below; anatomy = errors in hands, limbs, object structure; "
        "text = is any text in the image correct (score 10 if there is none); "
        "texture = does it read as a real photograph rather than a render; "
        "composition = is the framing usable. The brief:\n" + prompt
    )
    r = requests.post(
        f"{BASE}/v1/chat/completions",
        headers=HEAD,
        json={
            "model": "gemini-3.5-flash",
            "messages": [{"role": "user", "content": [
                {"type": "text", "text": rubric},
                {"type": "image_url", "image_url": {"url": data_url}},
            ]}],
            "response_format": {"type": "json_object"},
        },
        timeout=120,
    )
    r.raise_for_status()
    return json.loads(r.json()["choices"][0]["message"]["content"])


def best_of(user_input: str, n: int = 4) -> bytes:
    prompt = rewrite(user_input) + "\n" + STYLE_CONST         # steps 1 and 2
    with ThreadPoolExecutor(max_workers=n) as pool:           # step 3: client-side fan-out
        results = list(pool.map(lambda _: _safe(draw, prompt), range(n)))

    cands = [img for ok, img in results if ok]
    if not cands:
        raise RuntimeError("all candidates failed; check moderation or fall back to another model")

    with ThreadPoolExecutor(max_workers=len(cands)) as pool:  # step 4: score in parallel
        scores = list(pool.map(lambda im: score(im, prompt), cands))

    ranked = sorted(zip(cands, scores), key=lambda x: x[1]["total"], reverse=True)
    return ranked[0][0]                                       # step 5 retouch/rehost hooks in here


def _safe(fn, *args):
    try:
        return True, fn(*args)
    except Exception as e:                # one failure must not sink the batch
        return False, str(e)