Muse Wonder 批量任务跑到一半报错——先查限流还是先查请求体？

单条请求能通、批量跑到几十条才报错，通常不是单一原因：可能是频率限制被触发，也可能是批量组装请求体时字段被改动，还可能是连接层抖动。判断顺序建议是先看原始响应，再按类型分档，最后才决定是改节奏还是改代码——直接调并发或直接改请求体，都是在没有证据的情况下覆盖症状。

批量中断时，先存下每条请求的原始响应和时间戳，再按频率类、格式类、网络类分开计数。频率类改节奏（降并发、加退避、错峰），格式类改代码（比对单条与批量的请求体差异），网络类只加重试。在拿到原始响应之前不要动并发，也不要凭报错文案猜是限流。适用场景是批量调用中断但单条可通；验证方式是分类计数后重跑同一样本，看失败类型是否收敛。

先把失败请求的原始响应完整存下来

很多排查卡住，是因为只看到界面上一句笼统报错，抓不到状态码、响应体和服务端返回头。建议在发起请求的那一层统一落盘，成功和失败都记，字段至少覆盖下表内容。落盘文件建议用 JSON Lines，一行一条，方便后续用 grep、jq 或脚本统计。

字段	含义	记录要求
ts	请求发起时间	ISO8601 格式，精确到毫秒，便于看时间窗分布
seq	批次内自增序号	从 1 开始，用于判断失败是集中在前段还是均匀分布
req_digest	请求体摘要	关键字段值 + 请求体字节长度 + 内容哈希，避免整段落盘敏感内容
resp_body	返回内容	原始响应体全文；过长时截取前若干字符，但要保留开头
http_status	HTTP 状态码	没有状态码时记录 SDK 或命令行返回的错误类型
elapsed_ms	耗时	毫秒，用于区分超时类失败和快速失败
concurrency	当前并发档位	分档测试时必填，否则无法归因

# 以下为字段格式示例，数值均为占位
{"ts":"<ISO8601>","seq":37,"req_digest":{"model":"...","prompt_len":812,"sha1":"..."},"resp_body":"...","http_status":429,"elapsed_ms":0,"concurrency":2}
{"ts":"<ISO8601>","seq":38,"req_digest":{"model":"...","prompt_len":0,"sha1":"..."},"resp_body":"...","http_status":400,"elapsed_ms":0,"concurrency":2}

注意第二个示例：请求体长度摘要为 0，这类信息在只看“报错”时经常被忽略，但它往往就是格式类问题的入口。落盘位置建议放在批量脚本自身目录下，按批次号分文件，重跑时不要覆盖旧文件。

把失败样本分成频率类、格式类、网络类

分类的目的很直接：频率类要改节奏，格式类要改代码，网络类只需要加重试。三类判定依据和典型返回特征如下表，具体状态码与错误文案需要结合你实际调用方式的返回确认。

类型	判定依据	典型返回特征	后续动作
频率类	失败集中在某个时间窗；并发越高失败越多；单条间隔调用不失败	HTTP 429、响应体含限流相关文案、带 Retry-After 类头部	降并发、加指数退避、错峰启动
格式类	同一条请求单独跑成功、进批量就失败；失败序号对应某些固定条目	HTTP 400/422、字段校验失败、缺少必填字段、负载过长	比对单条与批量的请求体，改组装逻辑
网络类	失败序号分散；与并发档位关系不大；原样重试常能过	连接超时、读超时、连接被重置、DNS 或 TLS 握手失败	保留重试，必要时调超时时间

分类完成后做两件事：一是按类型计数，二是按序号看分布。失败集中在批次开头，更像是启动瞬间并发过高；失败集中在某些固定序号，更像是这些请求体本身有问题；失败均匀散布，更偏向网络抖动。统计表可以这样记，数字自己填：

批次号：____  总条数：____  并发档位：____
频率类失败：____ 条   序号范围：____
格式类失败：____ 条   序号列表：____
网络类失败：____ 条   序号范围：____
重试后仍失败：____ 条

写带退避与并发上限的通用重试骨架

下面的骨架只依赖标准库和一个并发执行方式，具体调用替换成你自己的 HTTP 请求、SDK 或命令行封装即可。关键点是四点：指数退避、最大重试次数、并发上限、成功即写断点续跑标记。格式类失败不重试，直接记录，避免无意义地反复消耗次数。

import json, os, random, time
from concurrent.futures import ThreadPoolExecutor, as_completed

DONE_FILE   = "done_ids.txt"   # 断点续跑标记：每成功一条追加一行
LOG_FILE    = "batch_log.jsonl" # 原始响应落盘
MAX_RETRY   = 4                # 最大重试次数
CONCURRENCY = 2                # 并发上限，先取分档测试得到的稳定值
BASE_DELAY  = 1.0              # 退避基数（秒）
REQUEST_TIMEOUT = 30           # 单次请求超时（秒）

def load_done():
    if not os.path.exists(DONE_FILE):
        return set()
    with open(DONE_FILE, encoding="utf-8") as f:
        return {line.strip() for line in f if line.strip()}

def mark_done(task_id):
    with open(DONE_FILE, "a", encoding="utf-8") as f:
        f.write(str(task_id) + "\n")

def log_one(record):
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

def is_fatal(status, raw):
    # 格式类：客户端参数问题，重试无意义；按实际返回调整这里
    if status in (400, 422):
        return True
    return False

def call_once(task):
    # 替换为实际调用，返回 (ok, raw_text, http_status)
    # 例如 requests.post(url, json=task, timeout=REQUEST_TIMEOUT)
    raise NotImplementedError

def run_task(task):
    task_id = task["id"]
    last_status, last_raw = None, ""
    for attempt in range(MAX_RETRY + 1):
        try:
            ok, raw, status = call_once(task)
        except Exception as e:
            ok, raw, status = False, repr(e), None
        last_status, last_raw = status, raw
        if ok:
            mark_done(task_id)
            return task_id, True, status, raw
        if is_fatal(status, raw):
            return task_id, False, status, raw
        if attempt == MAX_RETRY:
            break
        delay = BASE_DELAY * (2 ** attempt) + random.uniform(0, 0.5)
        time.sleep(delay)
    return task_id, False, last_status, last_raw

def main(tasks):
    done = load_done()
    pending = [t for t in tasks if str(t["id"]) not in done]
    with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool:
        futures = [pool.submit(run_task, t) for t in pending]
        for fut in as_completed(futures):
            task_id, ok, status, raw = fut.result()
            log_one({"id": task_id, "ok": ok,
                     "http_status": status, "resp_body": raw[:2000]})
            print(task_id, ok, status)

if __name__ == "__main__":
    main(load_your_tasks())  # 替换为你的任务列表读取方式

断点续跑标记只记录成功项，中断后重跑会自动跳过已完成部分，不会整批重来。日志文件用于事后分类统计，不要和续跑标记混在一起。重试间隔加了随机抖动，是为了避免多个线程在同一时刻再次同时发起。

用小批量分档测试找出稳定并发数

并发数不要凭感觉定。选一批固定样本（比如同一批几十条，条数控制在你愿意重跑的范围内），在请求体完全相同的条件下，把并发依次设为 1、2、4，每档跑完整批并记录成功与失败条数，失败还要按上文的频率类、格式类、网络类分别标注。

档位   样本数   成功   频率类失败   格式类失败   网络类失败   备注
并发1  ____     ____   ____         ____         ____         ____
并发2  ____     ____   ____         ____         ____         ____
并发4  ____     ____   ____         ____         ____         ____

稳定值选取依据建议这样判断：从低档位往上找，取“该档位没有出现频率类失败、且上一档位开始出现频率类失败”的那个低档位，再往下留一点余量。如果各档位都只在格式类或网络类上失败，说明瓶颈不在并发，回头改请求体或加重试，而不是继续压低并发。每档建议至少完整跑一轮，样本量太小容易把偶然抖动当成规律；改动过请求体或重试参数后，需要重新分档确认，旧结论不要直接沿用。