单条音频能跑通，扩到几百条时要解决的不是模型本身，而是三件事：把推理封成一个统一入口，避免参数散落在各处；用清单区分哪些已处理、哪些待处理；把失败样本单独落盘，别让一条坏音频中断整批任务。MiDashengLM-Spatial 的加载方式、输入格式和输出结构要以你的部署为准，下面这套骨架只约束“批量任务怎么组织”，模型调用部分留成一个可替换的函数。

适用场景是一次几十到几百条音频的离线推理。处理方向是：先封装 infer_one 作为唯一入口，扫描目录生成只追加的清单，结果按行写 jsonl（含音频路径、运行参数、输出内容、耗时），失败样本单独记到错误文件，再用 `--only` failed 重跑。验证方式是跑到第 10 条后中断再续跑，看结果文件行数与去重路径数是否一致。边界：并发数、显存占用和超时策略需要结合机器实际情况确认，没做过中断验证前不建议直接跑全量。

先把单文件推理封装成一个函数

批量任务里最容易被写坏的地方，是把模型加载、参数设置、后处理散落在循环体内。建议先写一个只处理单条音频的函数，循环只负责调用它和处理返回值。加载模型这类重操作放在函数外面只做一次，函数内部只做解码、推理、后处理。

def infer_one(audio_path, cfg, model=None):
    """单文件推理统一入口。load_audio / model_generate / postprocess
    三处按你的实际部署替换，其余逻辑不用动。"""
    t0 = time.time()
    out = {
        "audio_path": audio_path,
        "status": "failed",
        "text": None,
        "raw": None,
        "params": {
            "model_path": cfg["model_path"],
            "device": cfg["device"],
            "sample_rate": cfg["sample_rate"],
            "max_new_tokens": cfg["max_new_tokens"],
            "prompt_id": cfg["prompt_id"],
        },
        "elapsed_ms": None,
        "error": None,
    }
    try:
        audio = load_audio(audio_path, sr=cfg["sample_rate"])
        raw = model_generate(model, audio, cfg)
        out["raw"] = raw
        out["text"] = postprocess(raw)
        out["status"] = "done"
    except Exception as e:
        out["error"] = {"type": type(e).__name__, "msg": str(e)[:500]}
    out["elapsed_ms"] = int((time.time() - t0) * 1000)
    return out

参数集中在 cfg 里，好处是结果文件里能原样带出一份参数快照；后续发现某批结果不对，可以直接从结果里看出当时用的是哪个模型路径、哪组生成参数。params 字段建议只放会影响输出的项，别把 API key、绝对路径里的用户名这类东西写进去。

扫描目录并生成待处理清单

清单的作用是回答两个问题：这批文件里哪些已经出过结果、哪些还没跑。建议用 jsonl 追加写，每次运行前先把清单合并成内存里的状态字典，避免整份文件重写一半时被打断导致清单损坏。

AUDIO_EXT = {".wav", ".flac", ".mp3", ".m4a"}

def scan(root, manifest_path):
    root = Path(root)
    state = {}
    if Path(manifest_path).exists():
        for line in open(manifest_path, encoding="utf-8"):
            rec = json.loads(line)
            state[rec["path"]] = rec          # 后写的记录覆盖先写的
    pending = []
    with open(manifest_path, "a", encoding="utf-8") as f:
        for p in sorted(root.rglob("*")):
            if not p.is_file() or p.suffix.lower() not in AUDIO_EXT:
                continue
            key = str(p.resolve())
            st = p.stat()
            if state.get(key, {}).get("status") == "done" \
               and state[key].get("mtime") == int(st.st_mtime):
                continue
            rec = {"path": key, "size": st.st_size,
                   "mtime": int(st.st_mtime),
                   "status": "pending", "ts": int(time.time())}
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
            pending.append(rec)
    return pending

清单每行的最小字段是 path、size、mtime、status、ts。status 取 pending / done / failed 三态即可。带上 mtime 是为了处理“同名文件被替换过”的情况，只比路径会漏掉重新录制的样本。

逐条推理并记录失败原因

循环体里一条音频失败不应该抛到外层。做法是 infer_one 不向外抛异常（如上），循环再加一层兜底捕获，遇到失败就写一条错误记录并 continue。文件损坏、采样率异常、显存不足、生成超时，这几类问题在批量场景里都会零星出现。

for i, item in enumerate(pending, 1):
    try:
        res = infer_one(item["path"], cfg, model)
    except Exception as e:
        res = {"audio_path": item["path"], "status": "failed",
               "error": {"type": type(e).__name__, "msg": str(e)[:500]}}
    write_result(res)          # 追加写 results.jsonl
    if res["status"] != "done":
        write_error({
            "run_id": RUN_ID, "audio_path": item["path"],
            "stage": res.get("error", {}).get("stage", "infer"),
            "exc_type": res["error"]["type"],
            "exc_msg": res["error"]["msg"],
            "ts": int(time.time()),
        })
    if i % 20 == 0:
        print(f"{i}/{len(pending)} done")

错误日志的字段建议固定为 run_id / audio_path / stage / exc_type / exc_msg / ts，stage 用来区分是解码、推理还是写盘阶段出的问题，比只记一条异常信息更好定位。只重跑失败项时，可以从错误日志或结果文件筛出失败路径，再传给脚本：

jq -r 'select(.status=="failed") | .audio_path' results.jsonl \
  | sort -u > retry.txt
python batch.py `--root` ./audio `--manifest` manifest.jsonl `--only` failed

没有 jq 的环境用一小段 Python 读 jsonl 筛 status == "failed" 即可，效果一样。`--only` failed 的实现就是跳过清单里 status 不是 failed 的项，不用改别的逻辑。

确定结果落盘格式与字段

结果建议一行一条 jsonl，不用单个大 JSON 或每文件一个小文件。追加写的特性意味着中途被打断时，已经写出的行仍然有效，而整份 JSON 一旦截断就需要重来。下游用 pandas.read_json(path, lines=True) 或 DuckDB 直接读都不需要二次清洗。

{"run_id": "r2024-0611-01", "audio_path": "/data/audio/a001.wav",
 "status": "done", "text": "模型输出的文本结果",
 "raw": {"tokens": [123, 456], "finish_reason": "stop"},
 "params": {"model_path": "/models/dasheng-spatial", "device": "cuda:0",
            "sample_rate": 16000, "max_new_tokens": 256, "prompt_id": "v1"},
 "elapsed_ms": 1840, "ts": 1718000000, "error": null}

字段约定：audio_path 用解析后的绝对路径，便于和清单对齐；params 是本次运行实际生效的参数快照，不是配置文件的全部内容；raw 存模型原始输出，text 存清洗后的结果，两者都保留，后面改后处理规则时不用重跑；elapsed_ms 用来估算总耗时和找出异常慢的样本；status 和 error 保证结果文件自己就能筛出失败项。

做一次小批量断点续跑验证

续跑的核心判断是“这条音频是否已经有一条 status 为 done 的结果”。实际实现里，以结果文件为准比只看清单更稳：每次启动先扫一遍 results.jsonl 建 done_paths 集合，命中就跳过，同时把清单里对应的记录补成 done。这样即使清单写入落后于结果文件，也不会重复推理。

1. 准备一个只放了 10 条以上音频的测试目录，用 `--limit` 10 跑完，记下 results.jsonl 的行数。
2. 删掉 `--limit` 再跑，跑到控制台出现几条进度后按 Ctrl-C 中断，检查文件末尾是否只是少写了几行、没有出现半行残缺。
3. 把上限设为 30 条左右继续跑，观察启动日志里 skipped 的数量是否等于上一次已经写完的条数。
4. 跑完后用一段脚本核对 len(paths) 与 len(set(paths)) 是否相等，不相等说明有重复推理或重复写行，需要检查跳过逻辑是否用了绝对路径做 key。

最后的去重核对命令可以这样写：

python -c "import json;p=[json.loads(l)['audio_path'] for l in open('results.jsonl',encoding='utf-8')];print(len(p),len(set(p)))"

两个数字一致，说明续跑没有重复处理已完成项，这套流程可以放大到全量；不一致时不要去删行凑数，先查路径归一化和跳过判断。并发数、单条超时和显存上限需要结合自己的机器确认，在没做过这段中断验证之前，不建议直接对全量目录开跑。