适用场景
这套方案适合需要稳定产出大量配图的小团队：内容运营要给文章批量出封面、电商要给商品图换风格、设计团队要先跑一批草图再挑。它解决的问题不是"怎么调一次接口"，而是把提示词、风格参数、重试、落盘和断点续跑固化成一条可重复执行的流水线，让同一批任务今天跑和下周跑都能得到结构一致的产物。
环境与前置条件
- 操作系统：Linux（Ubuntu 系或 Debian 系均可）或 macOS；Windows 建议在 WSL2 里操作，避免路径与编码问题。
- Python：3.10 及以上，用虚拟环境隔离依赖。
- 磁盘：单张 1024 级别图片约 1～3 MB，按每天 500 张估算，预留 50 GB 以上并单独挂一个数据盘更稳妥。
- 内存：走 API 接入时 4 GB 足够；本地部署推理服务时内存与显存要求差别很大，请以官方仓库 README 给出的显存建议为准，低于建议值会直接 OOM。
- 网络：能访问接口域名；本地部署时确认端口未被占用。
- 凭据：一个可用的 API Key（走 API 路线），或已下载的模型权重（走本地部署路线）。
下文所有版本号、模型名、接口路径、支持的尺寸枚举，一律以官方文档当前版本为准，不要照抄记忆里的旧值。
分步骤部署
第 1 步：建项目骨架
```bash
mkdir -p mm-h3-pipeline/{out,logs,prompts}
cd mm-h3-pipeline
python3 -m venv .venv
source .venv/bin/activate
```
requirements.txt：
```text
requests
pyyaml
python-dotenv
tenacity
tqdm
```
```bash
pip install -r requirements.txt
pip freeze > requirements.lock.txt
```
成功标志：pip install 无报错，python -c "import requests, yaml, tenacity" 静默退出。
第 2 步：把凭据放进环境变量
不要写进代码。建 .env：
```bash
MINIMAX_API_KEY=你的key
```
.gitignore 里加一行 .env。加载方式在脚本里用 python-dotenv，或者启动前 export $(grep -v '^#' .env | xargs)。
第 3 步：写配置文件，把易变的东西集中
config.yaml：
```yaml
api:
base_url: "https://<官方文档给出的接口域名>"
path: "/<官方文档给出的生成接口路径>"
model: "<官方文档当前的模型名>"
timeout: 120
max_retries: 4
concurrency: 4
defaults:
size: "1024x1024" # 取值以官方文档支持的尺寸枚举为准
num_images: 1
negative_prompt: "低分辨率, 模糊, 文字水印, 肢体畸变, 多余手指"
styles:
flat_illustration:
prefix: "扁平化矢量插画，"
suffix: "，干净的色块，统一配色，无渐变"
strength: 0.6
cinematic_photo:
prefix: "电影感摄影，"
suffix: "，35mm 镜头，浅景深，自然光，细腻颗粒"
strength: 0.5
```
这一步的意义：把"接口地址、模型名、尺寸、风格词"从代码里剥离出来。以后官方换模型名，只改一行 YAML，不用动脚本。
第 4 步：打通最小可用调用
generate.py 的核心部分：
```python
import base64, json, os, pathlib, time
import requests, yaml
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
ROOT = pathlib.Path(__file__).parent
CFG = yaml.safe_load((ROOT / "config.yaml").read_text(encoding="utf-8"))
API_KEY = os.environ["MINIMAX_API_KEY"]
URL = CFG["api"]["base_url"].rstrip("/") + CFG["api"]["path"]
class Retryable(Exception):
pass
def build_prompt(subject: str, style: str | None = None, extra: str = "") -> str:
st = CFG["styles"].get(style)
if not st:
return subject + extra
return f"{st['prefix']}{subject}{st['suffix']}{extra}"
@retry(retry=retry_if_exception_type(Retryable),
stop=stop_after_attempt(CFG["api"]["max_retries"]),
wait=wait_exponential(multiplier=2, min=2, max=60))
def text_to_image(prompt, negative=None, size=None, n=None, seed=None):
payload = {
"model": CFG["api"]["model"],
"prompt": prompt,
"negative_prompt": negative or CFG["defaults"]["negative_prompt"],
"size": size or CFG["defaults"]["size"],
"n": n or CFG["defaults"]["num_images"],
}
if seed is not None:
payload["seed"] = seed # 字段名以官方文档为准，用于复现
r = requests.post(URL,
headers={"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"},
json=payload, timeout=CFG["api"]["timeout"])
if r.status_code == 429 or r.status_code >= 500:
raise Retryable(f"{r.status_code} {r.text[:200]}")
r.raise_for_status()
return r.json()
```
返回体结构以官方文档为准，通常要么给图片 URL，要么给 base64。写一个兼容两种形态的提取函数：
```python
def save_images(data, out_dir: pathlib.Path, stem: str):
items = data.get("images") or data.get("data") or []
paths = []
for i, it in enumerate(items):
p = out_dir / f"{stem}_{i}.png"
if isinstance(it, str) and it.startswith("http"):
p.write_bytes(requests.get(it, timeout=120).content)
elif isinstance(it, str):
p.write_bytes(base64.b64decode(it))
elif isinstance(it, dict) and it.get("b64_json"):
p.write_bytes(base64.b64decode(it["b64_json"]))
elif isinstance(it, dict) and it.get("url"):
p.write_bytes(requests.get(it["url"], timeout=120).content)
else:
continue
paths.append(p)
return paths
```
成功标志：脚本跑完，out/ 下出现非零字节的 PNG，用 file out/*.png 看到 PNG image data。
第 5 步：提示词与风格参数工程化
提示词写成固定槽位，比自由发挥稳定得多：
```text
{风格前缀} + {主体} + {动作或构图} + {环境} + {光线} + {镜头或材质} + {风格后缀} + {画质修饰}
```
具体对比例子：
- 松散写法：
一只猫在街上 - 槽位写法：
扁平化矢量插画，一只橘猫蹲在便利店门口避雨，微微抬头看向招牌，夜晚街道，湿地面反光，霓虹冷光打亮侧脸，干净的色块，统一配色，无渐变
风格控制靠三件事：
1. 风格词前后缀（prefix / suffix），控制整体视觉语言；
2. strength 之类的风格强度参数，值越高越贴风格、越偏离原始描述；参数是否支持、取值范围以官方文档为准；
3. 负向提示词，把"水印、畸变、多余手指"这类高频问题挡掉。
如果接口支持参考图（图生图 / 风格参考），把参考图的权重字段也放进 styles 里统一管理，不要每次手写。同一批出图务必把 seed 记进任务清单，否则复现不了。
第 6 步：批量出图与断点续跑
任务清单 tasks.csv：
```csv
id,prompt,style,size,n,seed
t001,一只橘猫蹲在便利店门口避雨,cinematic_photo,1024x1024,1,12345
t002,一杯手冲咖啡放在木桌上,flat_illustration,1024x1024,1,
t003,城市天际线剪影,flat_illustration,1024x1024,4,
```
batch.py 的关键逻辑：
```python
import csv, json, pathlib, time
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
from generate import CFG, build_prompt, text_to_image, save_images
ROOT = pathlib.Path(__file__).parent
MANIFEST = ROOT / "logs" / "manifest.jsonl"
def done_ids():
if not MANIFEST.exists():
return set()
ids = set()
for line in MANIFEST.read_text(encoding="utf-8").splitlines():
try:
rec = json.loads(line)
if rec.get("status") == "ok":
ids.add(rec["id"])
except json.JSONDecodeError:
continue
return ids
def run_one(row, out_dir):
prompt = build_prompt(row["prompt"], row.get("style") or None)
t0 = time.time()
data = text_to_image(prompt, size=row.get("size") or None,
n=int(row.get("n") or 1),
seed=int(row["seed"]) if row.get("seed") else None)
paths = save_images(data, out_dir, row["id"])
return {"id": row["id"], "status": "ok", "prompt": prompt,
"files": [str(p) for p in paths],
"elapsed": round(time.time() - t0, 2), "ts": int(time.time())}
def main():
out_dir = ROOT / "out" / time.strftime("%Y%m%d")
out_dir.mkdir(parents=True, exist_ok=True)
skip = done_ids()
rows = [r for r in csv.DictReader(open(ROOT / "tasks.csv", encoding="utf-8"))
if r["id"] not in skip]
if not rows:
print("所有任务已完成")
return
with ThreadPoolExecutor(max_workers=CFG["api"]["concurrency"]) as pool, \
MANIFEST.open("a", encoding="utf-8") as fp:
futs = {pool.submit(run_one, r, out_dir): r for r in rows}
for f in tqdm(as_completed(futs), total=len(futs)):
row = futs[f]
try:
rec = f.result()
except Exception as e:
rec = {"id": row["id"], "status": "fail",
"error": str(e)[:300], "ts": int(time.time())}
fp.write(json.dumps(rec, ensure_ascii=False) + "\n")
fp.flush()
if __name__ == "__main__":
main()
```
要点：并发数从 4 起步，跑稳了再往上加；每条任务的结果都追加写入 manifest.jsonl 并立即 flush，中途 Ctrl+C 也不会丢记录；重跑时自动跳过成功的任务，失败的会被重新执行。
第 7 步：本地部署路线（可选）
如果打算自己托管推理服务，流程是：按官方仓库 README 建环境、装推理依赖、下载权重、起服务。
```bash
python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt # 以官方仓库给出的依赖清单为准
按 README 下载权重到本地目录，路径写进启动参数
python -m <官方仓库提供的启动入口> --host 0.0.0.0 --port 8000
```
服务起来后，如果它对外暴露的是兼容 OpenAI 风格的接口，直接把 config.yaml 里的 base_url 改成 http://127.0.0.1:8000、path 改成对应路径，前面的 generate.py 一行不用动；如果不是，就按仓库 README 封装一层 HTTP 接口再复用同一套客户端代码。本地部署的好处是数据不出内网、单张边际成本可控，代价是要自己管显存、排队和升级。
验证部署是否成功
按顺序做四件事：
1. 配置是否被正确读到：
```bash
python -c "from generate import CFG, URL; print(URL, CFG['api']['model'])"
```
预期输出接口完整地址和模型名，没有 KeyError。
2. 单张出图：
```bash
python generate.py --prompt "一只白瓷杯放在木桌上，清晨侧光" --out out/smoke
```
预期：终端打印保存路径与耗时，out/smoke/ 下有 PNG。
3. 校验文件真的是图片：
```bash
file out/smoke/*.png
ls -lh out/smoke/
```
预期出现 PNG image data，单张体积在几百 KB 到几 MB 之间。如果只有几百字节，基本是错误信息被当成图片存下来了。
4. 批量跑通：
```bash
python batch.py
python - <<'PY'
import json, pathlib
recs = [json.loads(l) for l in pathlib.Path("logs/manifest.jsonl").read_text().splitlines() if l.strip()]
ok = [r for r in recs if r.get("status") == "ok"]
print("总数", len(recs), "成功", len(ok), "平均耗时", round(sum(r.get("elapsed",0) for r in ok)/max(len(ok),1), 2), "秒")
PY
```
预期：成功率 100%（不稳定的网络下允许个别失败，重跑 batch.py 会自动补跑）；再次执行 python batch.py 应直接输出"所有任务已完成"，说明断点续跑生效。
常见报错与解决
报错：401 Unauthorized / invalid api key
→ 原因：环境变量没加载、key 前后带空格引号，或 .env 没被读取。
→ 解决：
```bash
echo -n "$MINIMAX_API_KEY" | wc -c # 对比 key 实际长度，多出字符就是引号或空格
set -a && source .env && set +a
```
报错：429 Too Many Requests 或返回 rate limit exceeded
→ 原因：并发开太高，短时间请求数超过配额。
→ 解决：把 config.yaml 里 concurrency 从 4 降到 2，重试退避已在 tenacity 里配置，直接重跑：
```bash
sed -i 's/concurrency: 4/concurrency: 2/' config.yaml
python batch.py
```
报错：400 Bad Request / invalid parameter: size
→ 原因：尺寸或参数名不在接口支持范围内，比如自己拼了 1200x800。
→ 解决：改成官方文档列出的枚举值，并确认参数字段名拼写一致。
```bash
sed -i 's/1200x800/1024x1024/' tasks.csv
python -c "from generate import CFG; print(CFG['defaults']['size'])"
```
报错：requests.exceptions.ReadTimeout
→ 原因：单张生成耗时长于 timeout，或网络抖动。
→ 解决：把超时提到 180 秒并重跑，未完成的任务会被自动补跑。
```bash
sed -i 's/timeout: 120/timeout: 180/' config.yaml
python batch.py
```
报错：本地部署时出现 CUDA out of memory
→ 原因：显存不够，分辨率或并发批大小超出硬件。
→ 解决：降分辨率、把单次生成张数 n 调回 1、降低并发，必要时按官方文档启用显存优化选项（如分批或 CPU 卸载）。
```bash
sed -i 's/concurrency: 4/concurrency: 1/' config.yaml
```
后续维护
备份：需要备份的只有四样——config.yaml、styles 配置、tasks.csv、logs/manifest.jsonl。产物图片按 out/YYYYMMDD/ 分目录，定期打包归档到对象存储即可。建议每天把配置与清单做一次增量备份，这对复现问题比图片本身更值钱。
升级：把依赖锁在 requirements.lock.txt，升级前先在测试目录跑一遍固定的十张"回归集"（挑难度高、易出畸变的提示词），对比新旧效果再决定是否全量切换。模型名或接口路径变更时只改 config.yaml，不要动业务脚本。
日志与监控：manifest.jsonl 就是天然的监控数据源，每天统计三个指标——成功率、P95 耗时、失败错误码分布。成功率跌破 95% 或 P95 耗时明显上涨时先看是不是配额或服务端排队。再加两条兜底：磁盘使用率告警（图片很容易把盘写满），以及按天统计生成张数，用来估算配额消耗节奏。
成本与配额：批量任务前先用 --dry-run 打印将要发送的完整提示词，确认没有重复任务和错别字，避免把配额浪费在写错的提示词上。