ModelScope API-Inference：兼容 OpenAI 的批量模型调用

魔搭的 API-Inference 提供 OpenAI 兼容接口（https://api-inference.modelscope.cn/v1），可以直接用 openai SDK 调用。免费服务有额度和并发限制，所以批处理脚本要做到三点：

• 调用前先确认模型在推理服务上可用；
• 遇到 429 时停止并保存进度，而不是一直重试；
• 下次运行时只处理剩下的部分。

先看清成品（box 上真实运行）

• 公开的 /v1/models 不需要令牌就能访问。我凭印象写的 Qwen/Qwen3-8B 不在推理服务上，脚本列出了同一组织下可用的 Qwen/Qwen3.5-27B 等模型，退出码为 1。
• 用无效令牌调用 Qwen/Qwen3.5-27B：返回 401 Authentication failed, please make sure that a valid ModelScope token is supplied.，脚本立即停止，退出码为 2。
• 本地替身上：第 2 次调用遇到 429 后停止，只调用了 2 次，没有重试；续跑后补完剩下的部分，没有重复处理。pytest 5 项通过。

合规提醒：从魔搭下载的每个模型都有自己的许可证（本文用的 Qwen2.5-0.5B-Instruct 仓库里是 Apache License），商用前先读 LICENSE 和模型卡。访问令牌（ms- 开头）只放在环境变量里，不要共享账号或令牌，也不要用多个账号轮换来绕过限额。免费推理服务的额度、并发和使用条件以魔搭官方页面为准，平台可能随时调整，本文不写具体数字和价格。

运行说明：box 上没有魔搭令牌（令牌需要账号满足官方的使用条件），所以没有跑通真实的推理。成功路径用的是我自己写的 OpenAI 兼容替身，客户端用的是真实的 openai SDK。

做完之后你会得到什么？

需要准备哪些环境？

Python 3.9 以上；pip install openai pytest（box 上 openai 的版本是 3.35.1）。真实调用需要魔搭访问令牌，放在环境变量 MODELSCOPE_API_KEY 中。

文件结构

61-modelscope-B-src/
├── ms_batch.py
├── fake_ms.py
├── prompts.jsonl
└── test_ms_batch.py

第 1 步：批处理脚本

"""用魔搭 API-Inference（OpenAI 兼容接口）批量处理 JSONL，并支持断点续跑。
- 调用前先查公开的 /v1/models，确认模型 ID 在推理服务上可用（模型库里有，不等于推理服务开放）
- 401：令牌无效，或者账号没满足使用条件，立即停止
- 429：额度或并发用完，**停止并保存进度**，不重试（重试只会继续消耗额度），之后再续跑
- 5xx 和网络错误：最多重试 2 次
- 如果响应头里有 modelscope-ratelimit-* 字段，就原样打印出来（字段名以官方限额页为准）"""
import argparse, json, os, pathlib, sys, time
import openai
from openai import OpenAI

BASE = "https://api-inference.modelscope.cn/v1"

class Stop(Exception):
    pass

def available_models(client):
    return {m.id for m in client.models.list()}

def call(client, model, prompt, sleep=time.sleep):
    for attempt in range(3):
        try:
            raw = client.chat.completions.with_raw_response.create(
                model=model, messages=[{"role": "user", "content": prompt}], temperature=0)
            limits = {k: v for k, v in raw.headers.items() if k.lower().startswith("modelscope-ratelimit")}
            return raw.parse().choices[0].message.content, limits
        except openai.AuthenticationError as e:
            raise Stop(f"401 鉴权失败：{e.message}")
        except openai.RateLimitError as e:
            raise Stop(f"429 额度或并发受限，已停止，稍后续跑：{e.message}")
        except (openai.APIConnectionError, openai.InternalServerError) as e:
            if attempt == 2:
                raise Stop(f"多次重试仍失败：{e}")
            sleep(2 ** attempt)

def run(src, out, model, client, sleep=time.sleep):
    done = set()
    if pathlib.Path(out).exists():
        done = {json.loads(l)["id"] for l in open(out, encoding="utf-8") if l.strip()}
    todo = [r for r in map(json.loads, open(src, encoding="utf-8")) if r["id"] not in done]
    print(f"共 {len(done) + len(todo)} 条，已完成 {len(done)}，本次待处理 {len(todo)}")
    with open(out, "a", encoding="utf-8") as f:
        for r in todo:
            try:
                text, limits = call(client, model, r["prompt"], sleep)
            except Stop as e:
                print("■", e); return 2
            f.write(json.dumps({"id": r["id"], "model": model, "output": text}, ensure_ascii=False) + "\n"); f.flush()
            print(f"✓ {r['id']}", limits or "")
    return 0

def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("src"); ap.add_argument("out"); ap.add_argument("--model", required=True)
    ap.add_argument("--base-url", default=BASE)
    a = ap.parse_args(argv)
    client = OpenAI(api_key=os.environ.get("MODELSCOPE_API_KEY", "missing"), base_url=a.base_url,
                    max_retries=0, default_headers={"User-Agent": "binbho-tutorial/1.0"})
    models = available_models(client)
    if a.model not in models:
        org = a.model.split("/")[0].lower()
        near = sorted(m for m in models if m.lower().startswith(org + "/"))[:8]     # 同一组织下可用的模型
        print(f"✗ 推理服务上没有 {a.model}。相近的有：{near}"); return 1
    return run(a.src, a.out, a.model, client)

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

第 2 步：替身、数据和测试

"""本地替身（我自己写的，不是魔搭官方程序）：模拟 OpenAI 兼容的 /v1/models 和 /v1/chat/completions。
401 时的错误体照搬真实 api-inference.modelscope.cn 返回的格式；429 在第 N 次调用时触发。"""
import json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

STATE = {"calls": 0, "limit_at": None, "fail_5xx": 0}
MODELS = ["Qwen/Qwen3-8B", "Qwen/Qwen2.5-7B-Instruct"]

class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass
    def _j(self, code, obj, headers=None):
        raw = json.dumps(obj, ensure_ascii=False).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json")
        for k, v in (headers or {}).items():
            self.send_header(k, v)
        self.end_headers(); self.wfile.write(raw)
    def do_GET(self):
        self._j(200, {"object": "list", "data": [{"id": m, "object": "", "owned_by": "system", "created": 0} for m in MODELS]})
    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        if self.headers.get("Authorization") != "Bearer ms-good":
            return self._j(401, {"error": {"message": "Authentication failed, please make sure that a valid ModelScope token is supplied.", "request_id": "fake"}})
        STATE["calls"] += 1
        if STATE["fail_5xx"]:
            STATE["fail_5xx"] -= 1; return self._j(503, {"error": {"message": "busy"}})
        if STATE["limit_at"] and STATE["calls"] >= STATE["limit_at"]:
            return self._j(429, {"error": {"message": "rate limited (fake)"}})
        text = body["messages"][-1]["content"]
        self._j(200, {"id": "x", "object": "chat.completion", "created": 0, "model": body["model"],
                      "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": f"摘要：{text[:6]}"}}]},
                {"modelscope-ratelimit-requests-remaining": "99"})

def start():
    s = ThreadingHTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=s.serve_forever, daemon=True).start()
    return s, f"http://127.0.0.1:{s.server_address[1]}/v1"

{"id": "p1", "prompt": "用一句话总结：魔搭社区是一个开源模型社区。"}
{"id": "p2", "prompt": "用一句话总结：断点续跑可以避免重复消耗额度。"}
{"id": "p3", "prompt": "用一句话总结：429 表示请求被限流。"}

import json
import pytest
from openai import OpenAI
import fake_ms, ms_batch

@pytest.fixture
def base():
    fake_ms.STATE.update(calls=0, limit_at=None, fail_5xx=0)
    s, u = fake_ms.start(); yield u; s.shutdown()

def client(u, key="ms-good"):
    return OpenAI(api_key=key, base_url=u, max_retries=0)

def lines(p):
    return [json.loads(l) for l in open(p, encoding="utf-8")]

def test_all_ok(base, tmp_path):
    out = tmp_path / "o.jsonl"
    assert ms_batch.run("prompts.jsonl", out, "Qwen/Qwen3-8B", client(base)) == 0
    assert [r["id"] for r in lines(out)] == ["p1", "p2", "p3"]

def test_429_stops_then_resume(base, tmp_path):
    out = tmp_path / "o.jsonl"
    fake_ms.STATE["limit_at"] = 2
    assert ms_batch.run("prompts.jsonl", out, "Qwen/Qwen3-8B", client(base)) == 2
    assert [r["id"] for r in lines(out)] == ["p1"] and fake_ms.STATE["calls"] == 2   # 没有重试
    fake_ms.STATE.update(limit_at=None)
    assert ms_batch.run("prompts.jsonl", out, "Qwen/Qwen3-8B", client(base)) == 0
    assert [r["id"] for r in lines(out)] == ["p1", "p2", "p3"]                       # 没有重复

def test_401_stops(base, tmp_path):
    assert ms_batch.run("prompts.jsonl", tmp_path / "o.jsonl", "Qwen/Qwen3-8B", client(base, "ms-bad")) == 2

def test_5xx_retried(base):
    fake_ms.STATE["fail_5xx"] = 2
    text, limits = ms_batch.call(client(base), "Qwen/Qwen3-8B", "你好世界", sleep=lambda s: None)
    assert text.startswith("摘要") and limits == {"modelscope-ratelimit-requests-remaining": "99"}

def test_unknown_model(base, capsys):
    assert ms_batch.main(["prompts.jsonl", "/tmp/x.jsonl", "--model", "Qwen/Qwen9-1B", "--base-url", base]) == 1
    assert "Qwen/Qwen3-8B" in capsys.readouterr().out

第 3 步：运行

$ MODELSCOPE_API_KEY=ms-invalid python ms_batch.py prompts.jsonl out.jsonl --model Qwen/Qwen3-8B
✗ 推理服务上没有 Qwen/Qwen3-8B。相近的有：['Qwen/Qwen-Image-Edit', 'Qwen/Qwen3.5-122B-A10B', 'Qwen/Qwen3.5-27B', ...]
$ MODELSCOPE_API_KEY=ms-invalid python ms_batch.py prompts.jsonl out.jsonl --model Qwen/Qwen3.5-27B
共 3 条，已完成 0，本次待处理 3
■ 401 鉴权失败：Error code: 401 - {'error': {'message': 'Authentication failed, please make sure that a valid ModelScope token is supplied.', 'request_id': 'f453854f-...'}}
$ python -m pytest -q
5 passed

可用模型列表会随时变化，以你运行时 /v1/models 的返回为准。

调试示例

1. 模型名是凭印象写的。 我先用 Qwen/Qwen3-8B 测试，预检直接拦下了：推理服务上的 Qwen 模型是 Qwen3.5-27B、Qwen3.8-27B 等。模型库里有某个模型，不代表推理服务开放了它，所以预检是必要的。

2. “相近模型”的提示是空的。 第一版是用模型名中第一个 - 之前的部分去匹配，Qwen/Qwen9-1B 得到的是 qwen9，什么都匹配不到，提示列表为空。改成列出同一组织（Qwen/）下的模型后，test_unknown_model 通过。

3. 429 不能当作临时错误来重试。 openai SDK 默认会自动重试 429。我在创建客户端时设置 max_retries=0，并让 429 直接停止。测试断言替身只被调用了 2 次，并且续跑后的输出中没有重复的 id。

常见问题

模板规则

• 调用前先预检模型。
• 429 时停止，不重试，保存进度。
• 输出逐行写入并 flush。
• 不要把额度数字写死在代码里。

项目资源

免费推理适合开发、测试和学习；在额度限制下，最重要的是能停下来，也能接着跑。