笙亿网络策划

FLUX 3 Image Klein 批量出图别一张张点 / 用本地接口跑队列更省事

约 10 分钟读完 5 阅
文章导读
一次要出几十张图，手工改提示词、点生成、等出图、再重命名，时间大多耗在重复动作上。更省事的路径是：先确认本地服务暴露的 HTTP 提交入口，把提示词列表读成队列，脚本逐条 POST 提交，记录任务编号，失败按退避重试，最后按编号把图落盘。下面给的骨架不假定某一家固定接口，字段名和路径都需替换成你本机服务实际提供的那一套。
📋 目录
  1. 壹 先确认本地服务暴露的提交地址与鉴权方式
  2. 贰 写一个最小请求骨架，只提交一条任务
  3. 叁 把提示词列表读成队列逐条提交
  4. 肆 处理排队失败与超时，加退避重试
  5. 伍 用任务编号把结果和提示词对应起来
A A

一次要出几十张图，手工改提示词、点生成、等出图、再重命名，时间大多耗在重复动作上。更省事的路径是：先确认本地服务暴露的 HTTP 提交入口，把提示词列表读成队列，脚本逐条 POST 提交，记录任务编号，失败按退避重试，最后按编号把图落盘。下面给的骨架不假定某一家固定接口，字段名和路径都需替换成你本机服务实际提供的那一套。

批量出图的瓶颈通常不在模型速度，而在提交流程。建议先用一条任务打通本地接口，确认返回体里的任务编号字段，再扩成队列脚本；队列脚本至少要管三件事：提交前记下 job 编号与提示词的对应关系、失败任务单独落盘、重试有次数上限。接口路径、字段名、是否需要密钥都以本机服务的启动日志和配置文件为准，别人的示例只能当形状参考。风险边界：参数错误类返回不重试，否则会把一次配置错误放大成一批失败。

先确认本地服务暴露的提交地址与鉴权方式

不要凭教程里的路径直接写脚本。先拿到本机真实可用的请求入口，通常有三处可以确认。

  • 启动输出：服务在前台启动时一般会打印监听地址和端口，形如 Uvicorn running on http://127.0.0.1:8000 或 Listening on 0.0.0.0:7860。把 host 和 port 记下来，同时注意如果跑在容器里，端口是否做过映射。
  • 配置文件与启动脚本：翻 config.yaml、.env、start.sh 一类文件，找 host、port、api_key、auth_token 这类字段。这几项决定了脚本里写什么。
  • 自描述文档：很多本地服务会暴露 /docs 或 /openapi.json。请求一次能列出路径和字段，就等于拿到了真实的接口清单，比猜字段名可靠。

是否需要密钥，可以用一次无凭据请求来判定：返回 401 或 403 说明入口在但缺鉴权；返回 200、202 或 422（参数校验失败）说明入口是通的，只是字段还没对上。需要密钥时通常放在 Authorization: Bearer <token> 请求头里，具体头名仍以配置文件为准。

curl -s http://127.0.0.1:8000/openapi.json | head -c 800
curl -s -o /dev/null -w '%{http_code}\n' http://127.0.0.1:8000/generate

写一个最小请求骨架，只提交一条任务

先用一条任务把链路跑通，重点是看清返回体里哪个字段是任务编号。下面的路径、字段名都是占位，请替换成你本机接口文档里列出的名字。

import json
import requests

BASE = 'http://127.0.0.1:8000'
HEADERS = {'Content-Type': 'application/json'}
# 服务需要密钥时改成 {'Authorization': 'Bearer <token>'}

def submit(prompt, seed=None):
    payload = {
        'prompt': prompt,
        'width': 1024,
        'height': 1024,
        'steps': 28,
    }
    if seed is not None:
        payload['seed'] = seed
    r = requests.post(BASE + '/generate', json=payload, headers=HEADERS, timeout=60)
    open('raw_resp.json', 'w', encoding='utf-8').write(r.text)
    if r.status_code not in (200, 202):
        raise RuntimeError('HTTP %s: %s' % (r.status_code, r.text[:200]))
    body = r.json()
    task_id = body.get('task_id') or body.get('id') or body.get('prompt_id')
    if not task_id:
        raise RuntimeError('返回体缺少任务编号: %s' % body)
    return task_id

检查分两步：先看 status_code 是否在预期范围，再看返回体结构。把原始返回体写进 raw_resp.json，字段对不上时可以直接翻文件比对，不用反复改脚本打印。如果接口是异步的，提交只返回任务编号，需要再请求一次状态或结果路径，这一步也要在单条任务阶段确认清楚。

把提示词列表读成队列逐条提交

单条调通之后再扩成循环。提示词建议放纯文本，一行一条，# 开头当注释；需要固定随机种子或多列参数时改用 CSV，表头写 prompt,seed,negative 这类明确列名。

FLUX 3 Image Klein 批量出图别一张张点 / 用本地接口跑队列更省事
import csv
from pathlib import Path

def read_queue(path):
    path = Path(path)
    if path.suffix.lower() == '.csv':
        with path.open(encoding='utf-8', newline='') as f:
            for row in csv.DictReader(f):
                yield row
    else:
        for line in path.read_text(encoding='utf-8').splitlines():
            line = line.strip()
            if line and not line.startswith('#'):
                yield {'prompt': line}

records = []
for i, item in enumerate(read_queue('prompts.txt'), start=1):
    job_id = 'job_%04d' % i
    prompt = item['prompt']
    task_id = submit(prompt, item.get('seed'))
    records.append({'job_id': job_id, 'task_id': task_id, 'prompt': prompt})
    with open('queue.jsonl', 'a', encoding='utf-8') as f:
        f.write(json.dumps(records[-1], ensure_ascii=False) + '\n')

命名规则固定为 job_0001 这种补零格式，好处是文件排序稳定、日志里好检索。编号在提交前生成，不要等出图后再按顺序猜，否则一旦中间有任务失败，后面全部错位。提交成功的记录立刻追加到 queue.jsonl，脚本中途中断也能从这份记录恢复。

处理排队失败与超时，加退避重试

批量任务的失败主要分两类：网络层的超时与连接中断，接口层的错误返回。前者通常可重试，后者要看状态码。429 表示请求过快，5xx 表示服务端问题，这两类可以重试；4xx 中的参数错误和鉴权失败不要重试，重试只会把一次配置错误复制成一批失败。

def submit_with_retry(prompt, job_id, retries=3, base_wait=5):
    last_err = None
    for attempt in range(1, retries + 1):
        try:
            return submit(prompt)
        except (requests.Timeout, requests.ConnectionError) as e:
            last_err = 'network: %s' % e
        except RuntimeError as e:
            last_err = str(e)
            if 'HTTP 4' in last_err and 'HTTP 429' not in last_err:
                break
        time.sleep(base_wait * attempt)
    with open('failed.jsonl', 'a', encoding='utf-8') as f:
        f.write(json.dumps({'job_id': job_id, 'prompt': prompt,
                            'error': last_err, 'attempts': attempt},
                           ensure_ascii=False) + '\n')
    return None

超时时间建议把连接超时和读取超时分开设置，排队型接口的读取超时要给得比单张出图耗时更长。重试次数可以放在 3 次左右，间隔按 5 秒、10 秒、20 秒递增，避免失败时连续打满服务。失败任务单独写进 failed.jsonl，字段包含 job 编号、提示词原文、最后一次错误、尝试次数，修好参数后按这个文件单独重跑，不必整批重来。

用任务编号把结果和提示词对应起来

出图之后最难的是回答“这张图对应哪条描述”。目录结构按 job 编号建立，一个任务一个文件夹，里面放图片、提交参数和原始返回体，对照关系就不依赖记忆。

out/
  job_0001/
    image_0.png
    image_1.png
    request.json
    response.json
  job_0002/
    image_0.png
    request.json
    response.json
  queue.jsonl
  failed.jsonl
  mapping.csv

mapping.csv 是三列对照表，每完成一个任务追加一行，字段建议为 job_id,task_id,prompt。同一提示词要出多张时，用 image_0、image_1 区分序号，并把 seed 写进 request.json，之后想复现某一张，直接从对应文件夹取参数重提即可。这样一套保存方式下来，排查问题看 failed.jsonl，找图看 mapping.csv，核对参数看各 job 文件夹，三条线互不干扰。