📌 本文部分内容由 AI 辅助整理，已经人工核对。文中脚本是我自己写的，所有输出都是真跑出来的。但有一点必须先说清楚：跑这些输出用的是文中那个假端点，不是真实模型——我手上没有对应的推理环境，所以凡是涉及真实模型速度的地方，我只给方法，不给数字。哪些数字能信、哪些不能，第四节会专门讲。

很多人本地部署是这么结束的：

装驱动、编译、调参数，折腾两个晚上，终于跑起来了。打开对话框问几句，感觉"还行，但不如在线的那个"。然后就没有然后了，模型在硬盘上躺着，再也没打开过。

问题不在模型不够好，在于用法选错了。

把本地模型当成一个更差的聊天窗口，它必输——那不是它的主场。它的主场是另一件事，而那件事恰好是最能省时间的。

---

一、本地模型的优势不是"更聪明"

拿本地小模型和在线大模型比聪明，方向就错了。参数量差着量级，比不过是正常的。

它真正比得过的是这几条：

调用次数不要钱。 你可以对同一批数据反复跑十遍、换五种 prompt 做对比，成本只有电费。按次计费的场景里，这种"随便试"是不敢想的。

数据不出网。 客户资料、内部文档、代码库——很多东西不是你不想用 AI 处理，是它不能离开这台机器。

没有限流，没有排队，不看别人脸色。 半夜三点它也是那个速度。

能嵌进流程里。 它就是本机的一个 HTTP 接口，可以塞进脚本、塞进定时任务、塞进 CI，不需要联网、不需要管密钥。

这四条凑在一起指向同一类场景：量大、重复、单条价值不高、但加起来很费时间的活。

---

二、真正划算的三类活

第一类：批量的格式化处理。

把两百份会议记录各自压成三句话；把一堆杂乱的产品描述统一成同一种句式；把日志里的报错按类型归档；给几千条数据打标签。

这类活的共同点是：每一条都不难，难在条数多。 人做一条要一分钟，做两百条就是一下午，而且做到后面质量还会掉。

第二类：不能出网的数据。

合同、病历、简历、内部代码、客户名单。这类数据的处理需求是真实存在的，卡点从来不是技术，是合规。本地模型在这里不是"更便宜的选择"——是根本没有别的选项。

第三类：需要嵌进自动化流程的环节。

监控日志里出现异常，自动让模型读一遍并归类；每天早上把昨天的更新自动整理成一份摘要；提交代码时自动生成一版说明草稿。

这类活的特点是触发频繁、单次输出不长、对延迟不敏感——正好是本地小模型能胜任的区间。

---

反过来，有两类活别指望它：

• 需要广博世界知识的。小模型的知识密度就那样，硬问会开始编。
• 一次性的复杂推理。你就问一次，那直接用在线大模型，省事。

判断标准很简单：这件事你要做一次，还是要做一千次？ 一次的用在线，一千次的用本地。

---

三、效率账该怎么算

这笔账其实很好算，只是很多人没算过。

假设一条任务的输出是 200 个 token，你的机器上模型跑 20 token/s，那么一条要 10 秒。

• 1000 条串着跑：接近 3 小时
• 人工做同样的事，一条一分钟：16 小时以上

3 小时对 16 小时，这个差距已经足够说明问题了。但 3 小时依然很长，而这 3 小时里机器大部分时间是在等——上一篇讲过，单请求解码是被内存带宽卡住的，计算单元一直在等数据搬过来。

并发就是把这段等待利用起来：权重读一遍，可以同时喂给多个请求。

所以这笔账的第二步是：并发之后能压到多少。

---

四、并发能提多少：先说哪些数字能信

先把话说在前面。下面这组数字，是我用文中那个假端点跑的——它每条固定睡 0.3 秒，不做任何真实推理。

并发数	24 条总用时	吞吐
1	7.3s	3.30 条/秒
4	2.0s	12.26 条/秒
8	1.2s	19.92 条/秒
16	1.5s	16.13 条/秒

这组数字能说明的：并发在调度层面是真的有效，而且存在一个拐点——到 8 是一路涨的，到 16 反而掉回去了。

这组数字不能说明的：你换成真实模型能提速几倍，以及你的拐点在第几档。真实模型的拐点由显存和算力决定：并发上去之后 KV cache 会成倍占显存，瓶颈也会从内存带宽转向算力。位置和这里完全不是一回事。

所以正确的用法不是抄我这张表，是用同一个脚本在你自己的机器上跑一遍，把并发数从 1、2、4、8 一档档往上试，看吞吐在哪一档不再涨。那个数才是你的。

顺便说一句，我在测的时候还撞到过 RuntimeError: can't start new thread——并发开太大，直接撞上系统的线程限制了。并发数不是越大越好，这件事有两个上限：一个是吞吐拐点，一个是系统扛不扛得住。

---

五、断点续跑为什么是刚需

批量任务跑得久，跑到一半出事是常态：网断了、服务重启了、手贱按了 Ctrl-C、或者干脆断电。

如果结果只在内存里攒着、最后一次性写出去，那这一趟就全白跑了。跑了 800 条挂掉，重来一遍还是从第 1 条开始。

所以规矩只有两条：

1. 每完成一条立刻落盘并 flush。 挂掉最多只丢正在跑的那几条。
2. 启动时先读结果文件，把已完成的 id 跳过。 重跑就是续跑，不需要额外的参数，也不需要记住上次跑到哪。

这样"失败"就不再是灾难——重新执行同一条命令就行，它自己知道从哪继续。

---

六、脚本

零依赖，只用标准库。对接任何 OpenAI 兼容接口。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
batch_run.py —— 把一批重复的活并发丢给本地模型，断点续跑。

本地模型真正省时间的地方不是"比在线大模型聪明"，而是**同一件事你可以
跑一万次，不计费、不出网、不看别人的限流脸色**。但要吃到这个好处，
得先解决三个很俗的工程问题：

  · 一条一条串着跑太慢 —— 单请求受内存带宽限制，机器大部分时间在等
  · 跑到第 800 条挂了，重跑一遍前 800 条 —— 时间全白花
  · 偶发超时、连接被重置，一条失败整批停

这个脚本就干这三件事：并发、断点续跑、失败重试。对接任何 OpenAI 兼容
接口（llama.cpp 的 server、Ollama、vLLM 都提供），只用标准库。

⚠️ 边界：
  · 它不管模型答得对不对 —— 那是 prompt 和模型选型的事
  · 并发数不是越大越好，超过拐点吞吐反而掉，拐点取决于你的硬件，
    自己用 --concurrency 试几档，别抄别人的数
  · 没有 GPU 也能先跑通全流程：--mock 会起一个假端点

    python3 batch_run.py --mock                       # 终端 A：起假端点
    python3 batch_run.py --input tasks.jsonl \\
        --out out.jsonl --prompt "一句话概括：" -c 4   # 终端 B：跑
"""

import argparse
import json
import os
import sys
import threading
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed


def load_tasks(path):
    """读 jsonl，每行一个对象，必须有 id 和 text。"""
    tasks = []
    with open(path, encoding="utf-8") as f:
        for n, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError as e:
                raise SystemExit(f"{path} 第 {n} 行不是合法 JSON：{e}")
            if "id" not in obj or "text" not in obj:
                raise SystemExit(f"{path} 第 {n} 行缺 id 或 text 字段")
            tasks.append(obj)
    return tasks


def done_ids(out_path):
    """
    已经跑完的 id。断点续跑就靠这个。

    故意逐行读、逐行判：结果文件可能因为上次被 Ctrl-C 掐断而
    最后一行只写了一半，这种残行跳过即可，不该让整批重跑。
    """
    if not os.path.exists(out_path):
        return set()
    ids = set()
    with open(out_path, encoding="utf-8") as f:
        for line in f:
            try:
                ids.add(json.loads(line)["id"])
            except (json.JSONDecodeError, KeyError):
                continue
    return ids


def ensure_trailing_newline(path):
    """
    追加写之前，先保证文件是以换行结尾的。

    上次进程被强杀（断电、kill -9）时，可能正停在写了一半的那行上。
    如果直接追加，新结果会和这条残行粘成一行 —— 粘出来的整行不是合法
    JSON，于是**那条本来已经跑完的记录也一起报废**，下次续跑会白白重跑它。
    """
    if not os.path.exists(path) or os.path.getsize(path) == 0:
        return
    with open(path, "rb+") as f:
        f.seek(-1, os.SEEK_END)
        if f.read(1) != b"\n":
            f.write(b"\n")


def call_once(url, model, prompt, text, timeout):
    """调一次 OpenAI 兼容的 /v1/chat/completions。"""
    body = json.dumps({
        "model": model,
        "messages": [{"role": "user", "content": f"{prompt}\n\n{text}"}],
        "stream": False,
    }).encode("utf-8")
    req = urllib.request.Request(
        url, data=body, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        data = json.loads(r.read().decode("utf-8"))
    return data["choices"][0]["message"]["content"]


def work(task, args, retries=3):
    """跑一条，失败退避重试。返回的 dict 一定带 ok 字段。"""
    for attempt in range(retries):
        t0 = time.monotonic()
        try:
            out = call_once(args.url, args.model, args.prompt,
                            task["text"], args.timeout)
            return {"id": task["id"], "ok": True, "output": out,
                    "seconds": round(time.monotonic() - t0, 2)}
        except (urllib.error.URLError, TimeoutError, KeyError,
                json.JSONDecodeError) as e:
            if attempt == retries - 1:
                return {"id": task["id"], "ok": False, "error": repr(e)}
            time.sleep(2 ** attempt)          # 1s、2s、4s……


def run(args):
    tasks = load_tasks(args.input)
    skip = done_ids(args.out)
    todo = [t for t in tasks if t["id"] not in skip]

    print(f"  共 {len(tasks)} 条，已完成 {len(skip)} 条，本次要跑 {len(todo)} 条")
    if not todo:
        print("  没有要跑的了。")
        return 0
    print(f"  并发 {args.concurrency}，端点 {args.url}\n")

    ensure_trailing_newline(args.out)
    lock = threading.Lock()
    t0, done, failed = time.monotonic(), 0, 0

    # 追加模式：每完成一条立刻落盘并 flush。跑到一半断电也只丢正在跑的那几条。
    with open(args.out, "a", encoding="utf-8") as fout, \
            ThreadPoolExecutor(max_workers=args.concurrency) as pool:
        futures = [pool.submit(work, t, args) for t in todo]
        for fut in as_completed(futures):
            res = fut.result()
            with lock:
                fout.write(json.dumps(res, ensure_ascii=False) + "\n")
                fout.flush()
                done += 1
                failed += 0 if res["ok"] else 1
                el = time.monotonic() - t0
                print(f"\r  {done}/{len(todo)}  失败 {failed}  "
                      f"用时 {el:.1f}s  均速 {done / el:.2f} 条/秒",
                      end="", flush=True)

    el = time.monotonic() - t0
    print(f"\n\n  完成 {done} 条，失败 {failed} 条，总用时 {el:.1f}s")
    if done:
        print(f"  平均每条 {el / done:.2f}s，整体吞吐 {done / el:.2f} 条/秒")
    if failed:
        print(f"  失败的重跑一次即可，已完成的会自动跳过：")
        print(f"    python3 {os.path.basename(__file__)} "
              f"--input {args.input} --out {args.out} ...")
    return 1 if failed else 0


def serve_mock(port):
    """
    一个假的 OpenAI 兼容端点。没有 GPU 也能把整条流程跑通、把并发和
    断点续跑验证清楚，再换成真的端点地址。
    """
    # 必须是 ThreadingHTTPServer。单线程的 HTTPServer 会把并发请求排成队，
    # 拿它测并发只会得出"并发没用"的错误结论——真实的推理服务是并发的。
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

    class H(BaseHTTPRequestHandler):
        def do_POST(self):
            n = int(self.headers.get("Content-Length", 0))
            req = json.loads(self.rfile.read(n).decode("utf-8"))
            time.sleep(0.3)          # 装作在推理
            body = json.dumps({"choices": [{"message": {
                "role": "assistant",
                "content": "[mock] " + req["messages"][0]["content"][:40],
            }}]}).encode("utf-8")
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def log_message(self, *a):
            pass                     # 别把进度条冲掉

    print(f"  假端点已起：http://127.0.0.1:{port}/v1/chat/completions")
    print("  每条固定 0.3 秒，Ctrl-C 退出")
    ThreadingHTTPServer(("127.0.0.1", port), H).serve_forever()


def main():
    p = argparse.ArgumentParser(description="并发批处理 + 断点续跑")
    p.add_argument("--input", help="jsonl，每行 {\"id\":..., \"text\":...}")
    p.add_argument("--out", default="out.jsonl", help="结果 jsonl，追加写")
    p.add_argument("--prompt", default="一句话概括：", help="每条都用的指令")
    p.add_argument("--url", default="http://127.0.0.1:8000/v1/chat/completions")
    p.add_argument("--model", default="local", help="有些端点不校验这个字段")
    p.add_argument("-c", "--concurrency", type=int, default=4)
    p.add_argument("--timeout", type=float, default=120)
    p.add_argument("--mock", action="store_true", help="起假端点，不跑任务")
    p.add_argument("--mock-port", type=int, default=8000)
    a = p.parse_args()

    if a.mock:
        try:
            serve_mock(a.mock_port)
        except KeyboardInterrupt:
            print("\n  假端点已停")
        return 0
    if not a.input:
        p.print_help()
        return 1
    return run(a)


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

---

七、跑给你看

没有 GPU 也能先把整条流程跑通——脚本内置了一个假端点。

开一个终端起假端点：

python3 batch_run.py --mock --mock-port 8323

另一个终端跑任务（tasks.jsonl 每行一个 {"id":..., "text":...}）：

python3 batch_run.py --input tasks.jsonl --out out.jsonl \
    --url http://127.0.0.1:8323/v1/chat/completions -c 4

断点续跑演示。 先跑 1.2 秒，然后直接 kill 掉，不给它任何收尾机会：

① 跑到一半 kill（模拟断电）
   → 已落盘 12 条
② 人为追加一条被掐断的残行
③ 续跑
  共 24 条，已完成 12 条，本次要跑 12 条
  并发 4，端点 http://127.0.0.1:8323/v1/chat/completions
  完成 12 条，失败 0 条，总用时 1.1s
  平均每条 0.09s，整体吞吐 11.17 条/秒
④ 全部跑完后再跑一次
  共 24 条，已完成 24 条，本次要跑 0 条
  没有要跑的了。
合法 24 条 / 去重 24 个 / 残行 1 条 / 零重复跑: True

第 ② 步是故意的：往结果文件尾巴上塞了半截不完整的 JSON，模拟"上次正写到一半就断电了"。续跑照常识别出 12 条已完成，没有把残行当成灾难，也没有重复跑任何一条。

换成你真实的端点，只要改 --url 就行。

---

八、批量场景的 prompt 和聊天不一样

这一节其实比脚本重要。脚本只解决"跑得动"，prompt 决定"跑出来的东西能不能用"。

聊天和批量的根本差别是：有没有人在看。

聊天时模型多说两句废话无所谓，你自己会挑；格式偶尔飘一下也无所谓，你会追问一句。批量跑一千条，没有人一条条看——只要格式飘了，后面的解析就全废，而你要到最后才发现。

所以批量场景的 prompt 有三条硬要求：

① 把"只输出什么"说死，并且明确禁止解释。 不要写"请概括一下"，要写"直接输出一句话概括，不要任何前言、解释或补充说明"。模型很爱在正文前面加一句"好的，以下是概括："，这一句会毁掉你的下游解析。

② 给例子，比给形容词管用。 "请简洁"这种要求每个模型的理解都不一样。直接给一个输入输出的样例，它照着学，稳定得多。小模型尤其吃这套。

③ 限死长度。 "不超过 30 字"比"简短一些"有效得多，而且顺带控制了耗时——输出长度直接决定单条要跑多久。

如果下游要机器解析，就让它输出 JSON，并且自己加一层校验。 模型偶尔会吐出带 ```json 包裹的、或者末尾多个逗号的东西。解析失败的那些单独挑出来重跑一遍，比事后人工修便宜得多。

温度调低。 批量任务要的是稳定不是创意，大部分端点都支持传 temperature，往低了调。

最后一条，也是最容易省掉又最不该省的：先拿十条跑。

跑这十条不是看内容写得好不好，是看三件机械的事：格式是不是每条都一样、有没有多余的开场白、长度有没有失控。这十条稳了再放开跑一千条。否则你花三个小时批量生产的，是一千份格式各异的垃圾——而且电费和时间都是真花掉了。

---

九、写这个脚本时踩的两个坑

坑一：假端点是单线程的，测并发全是假的。

http.server 里的 HTTPServer 一次只处理一个请求。用它测并发，请求会乖乖排队，你会看到并发 8 和并发 1 一样快，然后得出"并发没用"的结论。

改成 ThreadingHTTPServer 才对。真实的推理服务是并发处理的，测试用的假端点必须在这一点上保持一致，否则它验证的东西是错的。

坑二：残行会把一条已完成的记录一起毁掉。

这个更隐蔽。进程被强杀时可能正停在写了一半的那行上，文件末尾没有换行。下次续跑直接追加，新结果就和这条残行粘成了一行：

{"id": "doc-099", "ok": tr{"id": "doc-013", "ok": true, ...}

粘出来的整行不是合法 JSON，于是解析时被跳过——那条本来已经跑完的记录也一起报废了，下次续跑会白白重跑它。

我是怎么发现的：全部跑完之后又跑了一次，本该输出"没有要跑的了"，结果它跑了 1 条。一条，正好一条。

修法是追加前先补一个换行：

with open(path, "rb+") as f:
    f.seek(-1, os.SEEK_END)
    if f.read(1) != b"\n":
        f.write(b"\n")

这两个坑的共同点是：不看输出细节根本发现不了。 第一个会让你得出错误结论，第二个会安静地多花一次调用。批量任务跑上千条的时候，"安静地多花一点"累积起来就不是小数目。

---

十、它不解决什么

它不管模型答得对不对。 输出质量是 prompt 和模型选型的事，脚本只负责把活派出去、把结果收回来。批量跑之前一定先用十条试，确认输出格式稳定了再放开跑一千条，否则就是批量生产垃圾。

它没做限流。 并发数是你自己给的，给多大它就开多大。对着别人的付费 API 用这个脚本要小心，本地服务无所谓。

它没做结果校验。 返回什么就存什么。如果你要求模型输出 JSON，那还得自己加一层解析和重试。

失败的那些条它不会自动重跑。 单条内部会退避重试三次，三次都失败就记 ok: false 落盘。你重新执行同一条命令，它会把失败的那些再试一遍——因为它们不在"已完成"里。

---

小结

• 本地模型的价值不在"更聪明"，在不计次、不出网、能嵌进流程
• 判断标准：这件事你要做一次，还是一千次？一次的用在线，一千次的用本地
• 单请求慢是因为机器一直在等数据，并发就是把这段等待利用起来
• 并发有拐点，别抄别人的数，用同一个脚本在自己机器上一档档试
• 批量任务的两条铁规矩：逐条落盘 + 启动时跳过已完成
• 跑一千条之前先跑十条

最后留个问题。第四节那张表是假端点跑的，真实模型的拐点位置我没有环境验证。如果你在自己的机器上试过并发梯度，吞吐是在第几档不再涨的，你的显存和模型规模是多少——这个组合我很想知道。 这种数据没人整理过，多几个人报一下，大家选并发数就不用瞎试了。