出图进阶：工作流编排与去 AI 味

import os
import requests

BASE = "https://api.apiyi.com/v1"
API_KEY = os.environ["APIYI_API_KEY"]          # 绝不要把 Key 写死在代码里

REWRITE_SYSTEM = """你是图片提示词工程师。把用户的口语需求改写成一段结构化的中文出图提示词。

必须补齐六要素，缺什么补什么，不要询问用户：
1 主体：具体到材质、颜色、数量、状态
2 环境：背景是什么、虚实关系
3 光线：光源方向、软硬、有没有补光，必须是可指认的单一主光
4 镜头与视角：焦段、光圈、机位高度、俯仰角
5 色调与介质：白平衡倾向、饱和度、胶片或数码质感
6 构图：主体在画面什么位置，留白在哪

规则：
- 只输出提示词正文，不要解释、不要分点、不要加标题
- 不要出现品牌名、商标、可识别的文字内容，除非用户明确要求
- 不要使用 8K、超高清、杰作、完美 这类空泛的质量词
- 用户明确指定过的要素原样保留，不要改写"""


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 = "冷调中性白平衡，整体偏低饱和，画面干净留白充足。"   # 你的 preset


def draw(prompt: str, size: str = "2K", aspect: str = "1:1") -> bytes:
    """出一张图（Nano Banana Pro 原生 Gemini 端点）"""
    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)   # 4K 按 600s 留余量
    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 但没图 = 多半被审核拦了
        raise RuntimeError("未返回图片：" + json.dumps(parts, ensure_ascii=False)[:300])
    return base64.b64decode(part["inlineData"]["data"])


def score(image: bytes, prompt: str) -> dict:
    """用视觉模型给候选图打分，返回 {总分, 各维度分, 一句话讲问题}"""
    data_url = "data:image/png;base64," + base64.b64encode(image).decode()
    rubric = (
        "给这张图打分，严格返回 JSON："
        '{"instruction":0-10,"anatomy":0-10,"text":0-10,"texture":0-10,'
        '"composition":0-10,"total":0-50,"issue":"一句话"}。'
        "instruction=是否满足下面这条需求；anatomy=手指、肢体、物体结构有无错误；"
        "text=画面内文字是否正确（没有文字给 10）；texture=质感是否像真实拍摄而非渲染；"
        "composition=构图是否可用。需求如下：\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        # Step 1 + Step 2
    with ThreadPoolExecutor(max_workers=n) as pool:          # Step 3：客户端并发，不用 n 参数
        results = list(pool.map(lambda _: _safe(draw, prompt), range(n)))

    cands = [img for ok, img in results if ok]
    if not cands:
        raise RuntimeError("全部失败，检查审核拦截或降级到别的模型")

    with ThreadPoolExecutor(max_workers=len(cands)) as pool:  # Step 4：并发打分
        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 的精修与转存按业务接上


def _safe(fn, *args):
    try:
        return True, fn(*args)
    except Exception as e:                # 单张失败不拖垮整批
        return False, str(e)