專案實錄 · 12 分鐘閱讀

用 Python 跑批量 AI 任務：佇列、重試、斷點與成本

批量任務不是給 API 呼叫套一層 for 迴圈。這篇實錄搭出讓它跑得住、停得下、算得清的四件事：穩定任務 key、併發上限、斷點檔案、開跑前就能說清的價格。

先有任務清單，再有迴圈

一批任務的本質是一組帶穩定身份的輸入。把輸入文字做雜湊當 key，同一張工單永遠落到同一個 key：重跑安全、重複可見，都靠它。花錢之前先數一遍總行數與去重後的 key 數。

import hashlib

def job_key(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]

texts = ["first ticket", "second ticket", "first ticket"]
jobs = [{"key": job_key(t), "text": t} for t in texts]
print(len(jobs), len({job["key"] for job in jobs}))

key 讓重複暴露出來

把併發壓在配額之下

瓶頸不在你的 CPU，在伺服器端配額。給每次呼叫套一個訊號量，限制同時在途的請求數，再用 gather 等全部結束。這個數字要寫進設定：上午九點能扛住的併發，晚高峰通常扛不住。

import asyncio

async def call(text, gate):
    async with gate:
        await asyncio.sleep(0.01)   # 代替真實 API 呼叫
        return text.upper()

async def run(texts, limit=4):
    gate = asyncio.Semaphore(limit)
    return await asyncio.gather(*(call(t, gate) for t in texts))

print(asyncio.run(run(["a", "b", "c"])))

一個訊號量管住整批

429 是讓你慢一點，不是失敗

限流的意思是等一會兒再來。每次失敗把等待翻倍，再加一點隨機抖動，避免所有 worker 同時重試；失敗三四次就放棄。400、401 這類業務錯誤一次都不要重試——重試多少次結果都一樣。

import asyncio, random

async def with_backoff(call, attempts=4):
    delay = 0.5
    for attempt in range(attempts):
        try:
            return await call()
        except RuntimeError:            # 代替 HTTP 429
            if attempt == attempts - 1:
                raise
            await asyncio.sleep(delay + random.uniform(0, delay / 2))
            delay *= 2

有上限的退避加抖動

斷點續跑靠追加寫

每完成一條就追加一行 JSON 並立即落盤。重新啟動時先把這個檔案讀成「已完成 key 集合」，跳過其中的任務。追加寫就是全部訣竅：崩潰只損失目前這一條，不會丟掉前面的五百條。

import json, pathlib

def load_done(path):
    p = pathlib.Path(path)
    if not p.exists():
        return set()
    rows = [json.loads(line) for line in p.read_text(encoding="utf-8").splitlines() if line.strip()]
    return {row["key"] for row in rows if row.get("ok")}

def save(path, key, text):
    with open(path, "a", encoding="utf-8") as fh:
        fh.write(json.dumps({"key": key, "ok": True, "text": text}, ensure_ascii=False) + "\n")

JSONL 扛得住行程被殺

落盤之前先驗證

就算要求了 JSON 輸出，回傳也可能裹著程式碼圍欄，或者在半個物件處被截斷。剝掉圍欄再解析，失敗就降級成原文前 200 字，而不是在迴圈裡拋例外——一條壞資料不該終結整批 800 條。

import json

def parse(text, fallback_chars=200):
    raw = text.strip()
    if raw.startswith("```"):
        raw = raw.split("\n", 1)[-1].rsplit("```", 1)[0]
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return {"title": raw[:fallback_chars], "fallback": True}

print(parse('```json\n{"title": "ok"}\n```'))
print(parse("cut off mid")["fallback"])

解析失敗要有兜底

先看清一次請求，再管一千次

前面所有做法都預設你知道單次呼叫長什麼樣：一個 POST、兩個請求頭、一個 JSON 體——模型名、訊息串列、輸出契約。response_format 只保證輸出能被解析，上一節的 schema 驗證仍然要跑。

import os, requests

resp = requests.post(
    "https://api.example.com/v1/chat/completions",
    headers={
        "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
        "Content-Type": "application/json",
    },
    json={
        "model": "gpt-4o-mini",
        "messages": [{"role": "user", "content": "Return the ticket title as JSON."}],
        "response_format": {"type": "json_object"},
        "max_tokens": 120,
    },
    timeout=(3.05, 60),
)
resp.raise_for_status()
data = resp.json()
print(data["choices"][0]["message"]["content"], data["usage"]["total_tokens"])

一次請求，四個決定

開跑之前先把價算出來

按字數估一遍輸入 token，乘上單價列印出來；跑完再記實際用量做對照。差額通常來自你忘了計入的提示詞模板，而它會在每次重試時再算一遍。

PRICE_IN = 0.15 / 1_000_000
PRICE_OUT = 0.60 / 1_000_000

def batch_cost(prompt_tokens, completion_tokens):
    return round(prompt_tokens * PRICE_IN + completion_tokens * PRICE_OUT, 4)

print(batch_cost(1_200_000, 300_000))

說得出數字的成本

你的練習清單

每條任務都有內容決定的 key，重跑不會重複付費。
併發數寫在設定裡，並且低於伺服器端配額。
429 用翻倍退避加抖動；4xx 業務錯誤不重試。
結果按 JSONL 逐行追加，寫完即落盤。
結構化輸出有明確的解析失敗分支。
單次請求的四個要點寫得出來：認證頭、模型、輸出契約、超時。
收尾列印任務數、失敗數與實際花費。

常見問題

能不能直接用 ThreadPoolExecutor？

如果 SDK 是同步的、呼叫只是網路等待，用執行緒更簡單也夠用。需要單行程幾百個併發等待時 asyncio 更省記憶體。選一種，別把兩套混著巢狀。

結果檔案會多大？

每行約 200 位元組，一百萬條約 200 MB。按天或按來源切檔案；只需要「哪些做完了」的話，單獨維護一份 key 索引即可。

重試幾次合適？

三到四次，總等待時間控制在單條任務的超時預算之內。仍失敗就寫進 failed.jsonl 記下錯誤，繼續下一條。

為什麼用雜湊而不是行號？

輸入檔案一變行號就全錯位，也發現不了同一張工單來了兩次。內容雜湊兩件事一起解決。

練一練重試與結構化輸出

覆蓋重試預算、超時、流式分片與 JSON 驗證的題目。

打開練習

繼續閱讀：