用 Replicate API 运行开源模型：同步预测、状态轮询与结果下载

Replicate 是一个用 API 运行开源模型的平台：图像、语音、视频、语言模型都有，按运行时间付费。 这篇做 predict.py：调用官方模型（例如 FLUX.1 schnell 文生图），先用同步模式等最多 60 秒，没等到就轮询，成功后马上把输出文件下载到本地，并保存一份元数据 JSON。API 生成的文件 1 小时后会被删除，所以“马上下载”是必须的。

先看清成品

• 用无效 Token 真实调用：两个接口返回的 401 内容不同（一个说 Unauthenticated，一个说 Authentication required），脚本都能识别。
• 对照文档发现两处会让轮询停不下来的问题：创建预测页的示例写的是 "successful"，而 HTTP 参考里的状态是 "succeeded"；预测生命周期页还有一个 aborted 状态，HTTP 参考没列。两者都已处理。
• 8 项 pytest 通过（mock）。

合规提醒：只用 Replicate 官方 HTTP API 和你自己的 API Token（replicate.com/account/api-tokens），Token 只放服务端，不共享账号。按运行时间计费，价格以 replicate.com/pricing 为准，本文不写价格。每个模型有自己的许可证和使用限制（在模型页面查看），生成内容要遵守 Replicate 服务条款，不生成侵权或违法内容，不冒用真人肖像。

运行说明：我没有 Replicate Token，没有跑过真实预测（也就没有产生费用）。真实跑过的是：用无效 Token 调用官方 API（2026-10-21 21:09–21:40，UTC+8，返回原样贴在文中），以及全部本地代码。正常的预测返回来自我写的 mock，字段照官方 HTTP 参考里的 prediction 对象。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

85-replicate-A-src/
├── predict.py
├── input.json
└── test_predict.py

第 1 步：示例输入

{"prompt": "一只橘猫坐在窗台上看雨，水彩画风格", "num_outputs": 1, "aspect_ratio": "1:1", "output_format": "webp"}

第 2 步：脚本

"""用 Replicate HTTP API 跑一次官方模型预测，并在 1 小时过期前把输出文件存到本地。

只用官方端点：
  POST /v1/models/{owner}/{name}/predictions   （官方模型，带 Prefer: wait）
  GET  urls.get                                （没等到结果时轮询）
"""
import argparse, json, os, pathlib, sys, time
import requests

API = "https://api.replicate.com/v1"
UA = "binbho-tutorial/1.0"
# HTTP 参考和 webhook 文档只列了前三个；“预测生命周期”页还有 aborted（超过期限、尚未开始运行）
TERMINAL = {"succeeded", "failed", "canceled", "aborted"}


def session(token):
    token = (token or "").strip()
    if not token:
        raise SystemExit("缺少 REPLICATE_API_TOKEN（replicate.com/account/api-tokens 创建）")
    s = requests.Session()
    s.headers.update({"Authorization": f"Bearer {token}", "User-Agent": UA})
    return s


def check(r):
    if r.status_code == 429:
        raise RuntimeError(f"429 限流：{r.json().get('detail')}")
    if r.status_code >= 400:
        try:
            d = r.json()
        except ValueError:
            d = {"detail": r.text[:200]}
        raise RuntimeError(f"HTTP {r.status_code} {d.get('title', '')}: {d.get('detail')}")
    return r.json()


def create(s, model, inp, wait=60):
    r = s.post(f"{API}/models/{model}/predictions", json={"input": inp},
               headers={"Prefer": f"wait={wait}"}, timeout=wait + 30)
    return check(r)


def poll(s, pred, interval=2, max_wait=600, sleep=time.sleep):
    waited = 0
    while pred["status"] not in TERMINAL:
        if waited >= max_wait:
            raise TimeoutError(f"{max_wait}s 未完成：{pred['urls']['web']}")
        sleep(interval); waited += interval
        pred = check(s.get(pred["urls"]["get"], timeout=30))
    return pred


def output_urls(output):
    if output is None:
        return []
    if isinstance(output, str):
        return [output] if output.startswith("https://") else []
    if isinstance(output, list):
        return [u for o in output for u in output_urls(o)]
    if isinstance(output, dict):
        return [u for o in output.values() for u in output_urls(o)]
    return []


def save_outputs(s, pred, outdir):
    out = pathlib.Path(outdir); out.mkdir(parents=True, exist_ok=True)
    saved = []
    for i, url in enumerate(output_urls(pred.get("output"))):
        name = f"{pred['id']}_{i}{pathlib.Path(url.split('?')[0]).suffix or '.bin'}"
        r = requests.get(url, timeout=120, headers={"User-Agent": UA})   # 文件链接不需要 token
        r.raise_for_status()
        (out / name).write_bytes(r.content)
        saved.append(name)
    meta = {k: pred.get(k) for k in ("id", "model", "version", "status", "input", "error", "metrics", "created_at", "completed_at")}
    meta["web_url"] = pred.get("urls", {}).get("web")
    meta["saved_files"] = saved
    (out / f"{pred['id']}.json").write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
    return saved


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("model", help="例如 black-forest-labs/flux-schnell")
    ap.add_argument("--input", required=True, help="JSON 字符串或 @文件")
    ap.add_argument("--out", default="outputs")
    a = ap.parse_args(argv)
    inp = json.load(open(a.input[1:], encoding="utf-8")) if a.input.startswith("@") else json.loads(a.input)
    s = session(os.environ.get("REPLICATE_API_TOKEN"))
    try:
        pred = create(s, a.model, inp)
        print(f"预测 {pred['id']} 状态 {pred['status']}  {pred['urls']['web']}")
        pred = poll(s, pred)
    except RuntimeError as e:
        print(e, file=sys.stderr); return 1
    if pred["status"] != "succeeded":
        print(f"未成功：{pred['status']} {pred.get('error')}", file=sys.stderr); return 2
    files = save_outputs(s, pred, a.out)
    print(f"完成，用时 {pred.get('metrics', {}).get('predict_time')}s，已保存：{files or '（无文件输出）'}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

第 3 步：测试（mock）

"""假 Replicate（我写的 mock，字段按官方 HTTP 参考文档的 prediction 对象）。"""
import json
import pytest
import predict as p


class R:
    def __init__(self, code, body=None, content=b""):
        self.status_code, self._b, self.content, self.text = code, body, content, json.dumps(body)
    def json(self):
        return self._b
    def raise_for_status(self):
        if self.status_code >= 400:
            raise RuntimeError(self.status_code)


def pred(status, output=None):
    return {"id": "abc", "model": "black-forest-labs/flux-schnell", "version": "v", "status": status,
            "input": {"prompt": "x"}, "output": output, "error": None, "metrics": {"predict_time": 1.2},
            "urls": {"get": "https://api.replicate.com/v1/predictions/abc", "web": "https://replicate.com/p/abc",
                     "cancel": "https://api.replicate.com/v1/predictions/abc/cancel"}}


class S:
    def __init__(self, gets):
        self.gets, self.headers = list(gets), {}
    def get(self, url, timeout=None):
        return self.gets.pop(0)


def test_poll_until_succeeded():
    s = S([R(200, pred("processing")), R(200, pred("succeeded", ["https://replicate.delivery/x/out-0.webp"]))])
    assert p.poll(s, pred("starting"), sleep=lambda x: None)["status"] == "succeeded"


def test_successful_is_not_terminal():
    # 文档“创建预测”页的示例写的是 "successful"，HTTP 参考里的状态是 "succeeded"
    assert "successful" not in p.TERMINAL and "succeeded" in p.TERMINAL


def test_poll_timeout():
    s = S([R(200, pred("processing"))] * 10)
    with pytest.raises(TimeoutError):
        p.poll(s, pred("starting"), interval=2, max_wait=4, sleep=lambda x: None)


def test_output_urls_shapes():
    assert p.output_urls("https://replicate.delivery/a.png") == ["https://replicate.delivery/a.png"]
    assert p.output_urls("Hello Alice") == []
    assert p.output_urls(["https://a/1.webp", "https://a/2.webp"]) == ["https://a/1.webp", "https://a/2.webp"]
    assert p.output_urls({"audio": "https://a/x.wav", "n": 3}) == ["https://a/x.wav"]


def test_save_outputs(tmp_path, monkeypatch):
    monkeypatch.setattr(p.requests, "get", lambda url, timeout=None, headers=None: R(200, content=b"RIFF....WEBP"))
    files = p.save_outputs(None, pred("succeeded", ["https://replicate.delivery/x/out-0.webp?sig=1"]), tmp_path)
    assert files == ["abc_0.webp"] and (tmp_path / "abc_0.webp").read_bytes().startswith(b"RIFF")
    meta = json.loads((tmp_path / "abc.json").read_text(encoding="utf-8"))
    assert meta["web_url"] == "https://replicate.com/p/abc" and meta["saved_files"] == ["abc_0.webp"]


def test_errors():
    with pytest.raises(RuntimeError, match="401"):
        p.check(R(401, {"status": 401, "detail": "Invalid or expired API token.", "title": "Authentication required"}))
    with pytest.raises(RuntimeError, match="resets"):
        p.check(R(429, {"detail": "Request was throttled. Your rate limit resets in ~30s."}))


def test_token_required():
    with pytest.raises(SystemExit):
        p.session("  \n")


def test_aborted_stops_polling():
    s = S([R(200, pred("aborted"))])
    assert p.poll(s, pred("starting"), sleep=lambda x: None)["status"] == "aborted"

第 4 步：运行

# 真实请求（2026-10-21 21:09 UTC+8）
$ curl -s https://api.replicate.com/v1/account -H 'Authorization: Bearer r8_invalid'
{"title":"Unauthenticated","detail":"You did not pass a valid authentication token","status":401}   ← HTTP 401
$ curl -s -X POST https://api.replicate.com/v1/models/black-forest-labs/flux-schnell/predictions -H 'Authorization: Bearer r8_invalid' -H 'Content-Type: application/json' -d '{"input":{"prompt":"x"}}'
{"status":401,"detail":"Invalid or expired API token.","title":"Authentication required"}   ← HTTP 401

$ REPLICATE_API_TOKEN=r8_invalid_demo python predict.py black-forest-labs/flux-schnell --input @input.json   # 21:17 UTC+8
HTTP 401 Authentication required: Invalid or expired API token.
$ echo $?
1
$ python -m pytest -q
8 passed

# 有 Token 时的预期输出格式（我没跑过）：
预测 <id> 状态 succeeded  https://replicate.com/p/<id>
完成，用时 <秒>s，已保存：['<id>_0.webp']

怎么检查结果？

• outputs/<id>.json 里有输入、状态、用时和网页链接，方便以后查是哪次生成的。
• 在 replicate.com 的 dashboard 能看到这次预测和费用。
• 输入参数以模型页面的 API schema 为准；input.json 里的参数取自 Replicate 官方 llms.txt 中 flux-schnell 的示例。

调试示例

轮询停不下来的两个坑，都来自文档。

1. "successful" 还是 "succeeded"？ 我先读的是“创建预测”页，它的同步模式响应示例是 "status": "successful"。但 HTTP 参考里状态列表是 succeeded，webhook 页和输出示例也都是 succeeded。如果照第一个示例写 while status != "successful"，成功的预测会一直轮询到超时。我以 HTTP 参考为准，并加了测试 test_successful_is_not_terminal 把这一点固定下来。

2. 漏了 aborted。 我最初的终态集合是 {"succeeded", "failed", "canceled"}，这是 HTTP 参考和 webhook 文档的写法。后来读“预测生命周期”页，发现还有 aborted：预测设置了期限、但还没开始运行就超时，状态就是 aborted。没把它算作终态的话，轮询要等到 max_wait 才结束。已加入终态，并加了测试 test_aborted_stops_polling。

另外，两个 401 的返回内容不同（见运行输出），所以脚本只看状态码和 detail 字段，不匹配 title 文本。

常见问题

模板规则

• 终态以 HTTP 参考为准，并包含 aborted。
• 成功后立即下载输出。
• 保存输入和预测 id。
• 不匹配错误文本，只看状态码。

项目资源

先跑一次，看清耗时和费用，再写成批量任务。