从零搭一个 AI 批量出图平台：用 Agent 把出图变成流水线，中途接入 4sapi

一批出图需求（CSV/表格）
   ↓
① 提示词构造   每行需求 → 规范化的出图提示词（含统一风格）
   ↓
② 任务入队     几十上百个任务 → 排进队列，控制并发
   ↓
③ 批量生成     每个任务 → 调文生图/图生图模型出图
   ↓
④ 质检过滤     太糊/不合规的自动剔除或重抽
   ↓
⑤ 归档交付     按命名规则存好，打包给用户下载

   批量需求 ──→ ┌─────────────────────┐
                │  Producer 任务生产者  │  拆成一个个出图任务
                └──────────┬──────────┘
                           ↓ 推入队列
                ┌─────────────────────┐
                │     Task Queue       │  控制并发、失败重试
                └──────────┬──────────┘
              ┌────────────┼────────────┐
              ↓            ↓            ↓
        [出图Worker]  [出图Worker]  [出图Worker]   ← 多个并发消费
              └────────────┼────────────┘
                           ↓
                ┌─────────────────────┐
                │  统一 API 网关 (4sapi) │  ← 文生图/图生图都走这
                └─────────────────────┘

https://4sapi.com/v1

# .env
FOURSAPI_BASE_URL=https://4sapi.com/v1
FOURSAPI_KEY=sk-你的令牌

# gateway.py
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["FOURSAPI_KEY"],
    base_url=os.environ["FOURSAPI_BASE_URL"],   # 指向中转网关
)

def gen_image(prompt, size="1024x1024", seed=None):
    """文生图，走网关"""
    extra = {"seed": seed} if seed is not None else {}
    resp = client.images.generate(
        model="flux-dev", prompt=prompt, size=size, extra_body=extra,
    )
    return resp.data[0].url

if __name__ == "__main__":
    url = gen_image("a red apple on a white table, product photo")
    print(url)   # 能返回图片 URL 即网关接通

# 全局风格：所有图共用，保证视觉统一
STYLE_SUFFIX = "clean studio lighting, white background, product photography, high detail"

def build_prompt(item):
    # item 来自 CSV 的一行，比如 {"name": "陶瓷马克杯", "color": "米白色"}
    core = f"a {item['color']} {item['name']}"
    return f"{core}, {STYLE_SUFFIX}"

import asyncio

async def worker(name, queue, results):
    while True:
        item = await queue.get()
        if item is None:
            break
        try:
            prompt = build_prompt(item)
            url = gen_image(prompt, seed=item.get("seed"))
            results.append({"item": item, "url": url, "ok": True})
        except Exception as e:
            results.append({"item": item, "error": str(e), "ok": False})
        finally:
            queue.task_done()

async def run_batch(items, concurrency=4):
    queue = asyncio.Queue()
    for it in items:
        queue.put_nowait(it)
    results = []
    # 并发上限 = concurrency，别让请求一把涌出去
    workers = [asyncio.create_task(worker(f"w{i}", queue, results))
               for i in range(concurrency)]
    await queue.join()
    for _ in workers: queue.put_nowait(None)
    await asyncio.gather(*workers)
    return results

def quality_check(img_bytes):
    from PIL import Image
    import io
    img = Image.open(io.BytesIO(img_bytes))
    # 太小直接淘汰
    if img.width < 512 or img.height < 512:
        return False
    # 颜色过于单一（疑似生成失败）淘汰
    colors = img.getcolors(maxcolors=10000)
    if colors and len(colors) < 20:
        return False
    return True

def img2img(base_image_url, prompt, strength=0.6):
    resp = client.images.edit(           # 图生图，走网关
        model="flux-dev",
        image=base_image_url,
        prompt=prompt,
        extra_body={"strength": strength},  # 越低越贴近原图
    )
    return resp.data[0].url

import csv

async def batch_generate(csv_path):
    with open(csv_path, encoding="utf-8") as f:
        items = list(csv.DictReader(f))     # 读入批量需求

    results = await run_batch(items, concurrency=4)   # 并发出图

    ok = [r for r in results if r["ok"]]
    failed = [r for r in results if not r["ok"]]
    # 失败的自动重试一轮
    if failed:
        retry = await run_batch([r["item"] for r in failed], concurrency=2)
        ok += [r for r in retry if r["ok"]]

    return ok      # 交付：按命名规则归档打包

function BatchImageStudio() {
  const [progress, setProgress] = useState({ done: 0, total: 0 });

  async function upload(file) {
    const form = new FormData();
    form.append("csv", file);
    const res = await fetch("/api/batch", { method: "POST", body: form });
    // SSE 实时回报 done/total，渲染进度
  }

  return (
    <div>
      <FileUpload accept=".csv" onUpload={upload} />
      <StyleConfigPanel />     {/* 配全局风格 */}
      <Progress done={progress.done} total={progress.total} />
      <DownloadAll />          {/* 全部完成后打包下载 */}
    </div>
  );
}