私有化部署选型：把模型搬回自己的机器

"""成本测算：私有化部署 vs 调云端 API，到底几个月才回本。

写这个脚本的原因：
    立项会上最常见的说法是「自己部署省钱」。
    但真实账单里除了电费，还有显卡采购、机房托管、运维人力、升级回归测试。
    很多低频场景算下来，私有化比调 API 贵好几倍 —— 只是没人算过。

这个脚本不替你做决定，它只做一件事：
    把两条路的钱都摊到「每月」，然后告诉你回本点在第几个月。
    如果回本点超过显卡的服役年限，那「省钱」这个理由就不成立，
    该老实承认真正的理由是「数据不能出网」。

用法：
    python3 cost_compare.py                      # 用内置的默认参数跑一遍
    python3 cost_compare.py --calls-per-day 5000 # 换成你自己的调用量
"""
import argparse
import sys

# --------------------------------------------------------------------------
# 默认参数：全部按 TODO 换成你们自己的真实数字，别直接拿去汇报
# --------------------------------------------------------------------------
DEFAULTS = {
    # ---- 业务量 ----
    "calls_per_day": 500,          # 每天调用次数
    "tokens_in": 800,              # 单次输入 token 数（含提示词和历史）
    "tokens_out": 400,             # 单次输出 token 数

    # ---- 云端 API 侧（单位：元 / 百万 token）----
    # 各家价格差别很大，且经常调整，务必以当天官网报价为准
    "api_price_in": 2.0,
    "api_price_out": 8.0,

    # ---- 私有化侧：一次性投入（元）----
    "gpu_cost": 16000,             # 显卡采购
    "server_cost": 8000,           # 主机其余部分
    "setup_hours": 24,             # 首次部署投入的人力工时

    # ---- 私有化侧：每月固定支出（元）----
    "power_watt": 450,             # 整机平均功耗（瓦）
    "power_price": 0.8,            # 电价，元 / 度
    "hosting_month": 0,            # 机房托管费；放办公室就是 0
    "ops_hours_month": 8,          # 每月运维工时

    # ---- 人力单价与服役年限 ----
    "hourly_rate": 150,            # 工程师时薪（元）
    "service_months": 36,          # 显卡预计服役月数
}


# --------------------------------------------------------------------------
def api_monthly(p):
    """云端 API 的每月支出：纯粹按量计费，用多少付多少。"""
    calls_month = p["calls_per_day"] * 30
    tin = calls_month * p["tokens_in"] / 1e6
    tout = calls_month * p["tokens_out"] / 1e6
    return tin * p["api_price_in"] + tout * p["api_price_out"]


def private_upfront(p):
    """私有化的一次性投入：硬件 + 首次部署人力。"""
    return p["gpu_cost"] + p["server_cost"] + p["setup_hours"] * p["hourly_rate"]


def private_monthly(p):
    """私有化的每月固定支出。

    注意这里和调用量无关：卡买了就在那儿，跑不跑都要交电费和运维。
    这正是低频场景吃亏的根源 —— 调用量再低，月支出也降不下来。
    """
    # 按 7×24 全天开机估电费：功率(W) ÷ 1000 × 24 × 30 = 每月度数
    power = p["power_watt"] / 1000 * 24 * 30 * p["power_price"]
    ops = p["ops_hours_month"] * p["hourly_rate"]
    return power + p["hosting_month"] + ops


def breakeven_month(p):
    """回本点：第几个月开始，私有化的累计成本低于 API 的累计成本。

    返回 None 表示在服役期内始终不回本。
    """
    up = private_upfront(p)
    pm = private_monthly(p)
    am = api_monthly(p)
    # 每月省下来的钱；不为正说明连月支出都打不过 API，永远回不了本
    saved = am - pm
    if saved <= 0:
        return None
    months = up / saved
    return months if months <= p["service_months"] else None


# --------------------------------------------------------------------------
def report(p):
    am = api_monthly(p)
    up = private_upfront(p)
    pm = private_monthly(p)
    calls_month = p["calls_per_day"] * 30

    print("=" * 58)
    print("业务量：每天 %d 次，每月约 %d 次" % (p["calls_per_day"], calls_month))
    print("        单次 %d token 进 / %d token 出"
          % (p["tokens_in"], p["tokens_out"]))
    print("-" * 58)
    print("【云端 API】")
    print("  一次性投入      0 元")
    print("  每月支出        %8.0f 元（随调用量涨落）" % am)
    print("【私有化部署】")
    print("  一次性投入      %8.0f 元（显卡 %d + 主机 %d + 部署人力 %d）"
          % (up, p["gpu_cost"], p["server_cost"],
             p["setup_hours"] * p["hourly_rate"]))
    print("  每月支出        %8.0f 元（电费 + 托管 + 运维，与调用量无关）" % pm)
    print("-" * 58)

    be = breakeven_month(p)
    if be is None:
        if am - pm <= 0:
            print("结论：私有化【永远不回本】。")
            print("      光是每月固定支出（%.0f 元）就已经超过 API 的月账单（%.0f 元）。"
                  % (pm, am))
        else:
            print("结论：私有化在 %d 个月服役期内【回不了本】。" % p["service_months"])
        print()
        print("  ⚠️ 这不代表不该私有化，只代表【省钱】这个理由不成立。")
        print("     如果数据确实不能出网，那就用这个理由立项，别用省钱。")
    else:
        print("结论：约第 %.1f 个月回本（服役期 %d 个月）。"
              % (be, p["service_months"]))
        total_saved = (am - pm) * p["service_months"] - up
        print("      整个服役期预计净省 %.0f 元。" % total_saved)
        print()
        print("  ⚠️ 前提是调用量能稳定维持。量掉下来，回本点会立刻往后推。")

    print("=" * 58)
    print("没有计入的成本，评估时请自行补上：")
    print("  · 模型升级时的回归测试人力")
    print("  · 服务故障期间的业务损失")
    print("  · 显卡残值与更换周期")
    print("  · 云端 API 侧的限流重试、以及涨价风险")


def main():
    ap = argparse.ArgumentParser()
    for k, v in DEFAULTS.items():
        ap.add_argument("--" + k.replace("_", "-"), type=type(v), default=v)
    args = ap.parse_args()
    report({k: getattr(args, k) for k in DEFAULTS})
    return 0


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

"""选型估算：这块卡到底放得下多大的模型。

估算口径（务必理解，别当成精确值）：
    权重占用 ≈ 参数量 × 每参数字节数
    每参数字节数由量化等级决定：FP16 约 2.0，Q8 约 1.0，Q4 约 0.55
    实际还要再加 KV cache 与运行时开销，这里按经验留 1.2 倍余量

真实占用请以 `ollama ps` 的 SIZE 列和 PROCESSOR 列为准：
PROCESSOR 显示 100% GPU 才是全部装进显存，出现 CPU 就说明溢出了。
"""

# 每个参数在不同量化等级下大致占多少字节
BYTES_PER_PARAM = {
    "FP16": 2.0,
    "Q8": 1.0,
    "Q4": 0.55,
}

# 运行时还要放 KV cache、激活值等，按 1.2 倍留余量
OVERHEAD = 1.2


def estimate_gb(params_b, quant):
    """params_b：参数量，单位 B（十亿）。返回预估显存占用 GB。"""
    per = BYTES_PER_PARAM[quant]
    # params_b × 1e9 个参数 × per 字节 ÷ 1024^3 换成 GiB
    weights = params_b * 1e9 * per / 1024 ** 3
    return weights * OVERHEAD


def fits(vram_gb, params_b, quant):
    need = estimate_gb(params_b, quant)
    return need <= vram_gb, need


def main():
    # 常见显卡显存档位
    cards = [("笔记本核显/无卡(纯CPU)", 0), ("8GB 显卡", 8), ("16GB 显卡", 16), ("24GB 显卡", 24)]
    # 常见模型规模
    models = [("0.5B", 0.5), ("1.5B", 1.5), ("7B", 7), ("14B", 14), ("32B", 32)]

    for quant in ("FP16", "Q4"):
        print("\n量化等级：%s（每参数约 %.2f 字节）" % (quant, BYTES_PER_PARAM[quant]))
        header = "%-12s" % "模型" + "".join("%-22s" % c[0] for c in cards)
        print(header)
        for name, pb in models:
            row = "%-12s" % name
            for _cname, vram in cards:
                ok, need = fits(vram, pb, quant)
                if vram == 0:
                    # 没有独立显卡时跑在内存里，能跑但速度是数量级的差距
                    row += "%-22s" % ("约%.1fGB 内存/慢" % need)
                else:
                    row += "%-22s" % ("%.1fGB %s" % (need, "放得下" if ok else "放不下"))
            print(row)

    print("\n同一个模型，Q4 相对 FP16 大约省下：")
    for name, pb in models:
        a = estimate_gb(pb, "FP16")
        b = estimate_gb(pb, "Q4")
        print("  %-6s %5.1fGB → %5.1fGB  省 %.0f%%" % (name, a, b, (1 - b / a) * 100))


if __name__ == "__main__":
    main()

"""选型验收：拿自己的真实样本，把候选模型逐个压一遍。

为什么必须有这一步：
    跑分榜测的是通用能力，和你的业务样本关系有限。
    「能装进显存」只说明跑得起来，不说明答得好。
    真正能拍板的证据只有一个 —— 你自己的三十条数据，各模型跑一遍的结果。

这个脚本负责把**可量化的部分**测出来（首 token 延迟、生成速度、并发下的退化），
并把每个模型的答案原样落盘，方便人工逐条对比质量。
质量这件事没法自动打分，必须人看。

用法：
    # 先把你的真实问题写进 samples.txt，一行一条
    python3 bench_local.py --models qwen2:1.5b,qwen2:7b --samples samples.txt
    python3 bench_local.py --models qwen2:7b --concurrency 4

依赖：只用标准库，不需要 pip 装任何东西。
"""
import argparse
import json
import os
import statistics
import sys
import threading
import time
import urllib.error
import urllib.request

BASE = os.environ.get("OLLAMA_BASE", "http://127.0.0.1:11434").rstrip("/")


# --------------------------------------------------------------------------
def call_once(model, prompt, timeout=300):
    """发一次非流式请求，返回 (答案, 指标字典)。

    Ollama 的响应里自带耗时统计，单位是纳秒：
        total_duration   整次请求耗时
        load_duration    加载模型耗时（模型已在内存里时接近 0）
        prompt_eval_count/duration   读题阶段
        eval_count/eval_duration     生成阶段
    生成速度 = eval_count / (eval_duration / 1e9)，这是最该关注的数字。
    """
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "stream": False,
        "options": {"temperature": 0},   # 压测固定为 0，保证可复现
    }
    req = urllib.request.Request(
        BASE + "/api/chat",
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
    )
    t0 = time.time()
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    wall = time.time() - t0

    eval_count = data.get("eval_count", 0)
    eval_dur = data.get("eval_duration", 0) / 1e9
    metrics = {
        "wall": wall,
        "load_s": data.get("load_duration", 0) / 1e9,
        "prompt_tokens": data.get("prompt_eval_count", 0),
        "gen_tokens": eval_count,
        "tps": eval_count / eval_dur if eval_dur else 0,
    }
    return data["message"]["content"], metrics


# --------------------------------------------------------------------------
def warmup(model):
    """先空跑一次，把模型加载进显存。

    不预热的话，第一条样本会把几十秒的加载时间算进成绩，
    整组数据直接废掉 —— 这是压测里最常见的错误。
    """
    try:
        call_once(model, "hi", timeout=600)
        return True
    except Exception as exc:
        print("  预热失败：%s" % exc)
        return False


def bench_serial(model, samples, outdir):
    """串行跑完所有样本，统计速度并把答案落盘。"""
    rows, answers = [], []
    for i, s in enumerate(samples, 1):
        try:
            ans, m = call_once(model, s)
        except Exception as exc:
            print("  样本 %d 失败：%s" % (i, exc))
            continue
        rows.append(m)
        answers.append({"index": i, "prompt": s, "answer": ans, "metrics": m})
        print("  [%2d/%d] %5.1fs  %5.1f token/s  生成 %d token"
              % (i, len(samples), m["wall"], m["tps"], m["gen_tokens"]))

    if not rows:
        return None

    # 答案落盘，人工逐条看质量。自动打分不可信，这一步不能省。
    os.makedirs(outdir, exist_ok=True)
    path = os.path.join(outdir, model.replace(":", "_").replace("/", "_") + ".json")
    with open(path, "w", encoding="utf-8") as f:
        json.dump(answers, f, ensure_ascii=False, indent=2)

    tps = [r["tps"] for r in rows]
    walls = [r["wall"] for r in rows]
    return {
        "样本数": len(rows),
        "生成速度中位数": "%.1f token/s" % statistics.median(tps),
        "生成速度最低": "%.1f token/s" % min(tps),
        "单次耗时中位数": "%.1fs" % statistics.median(walls),
        # P95 用来看长尾：中位数好看、P95 很差的服务，用户照样骂
        "单次耗时 P95": "%.1fs" % sorted(walls)[int(len(walls) * 0.95) - 1],
        "答案存档": path,
    }


def bench_parallel(model, prompt, n):
    """并发压测：n 个线程同时发同一条请求。

    单人测着飞快、多人一起用就超时，差别就出在这里。
    并发数超过服务端 OLLAMA_NUM_PARALLEL 时，多出来的请求会排队，
    表现为单次耗时线性变长，而不是报错。
    """
    results, lock = [], threading.Lock()

    def worker():
        try:
            _ans, m = call_once(model, prompt)
        except Exception as exc:
            with lock:
                results.append({"error": str(exc)})
            return
        with lock:
            results.append(m)

    threads = [threading.Thread(target=worker) for _ in range(n)]
    t0 = time.time()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    total = time.time() - t0

    ok = [r for r in results if "error" not in r]
    failed = len(results) - len(ok)
    if not ok:
        return {"并发数": n, "全部失败": failed}
    return {
        "并发数": n,
        "成功": len(ok),
        "失败": failed,
        "墙钟总耗时": "%.1fs" % total,
        "单请求耗时中位数": "%.1fs" % statistics.median([r["wall"] for r in ok]),
        "单请求速度中位数": "%.1f token/s" % statistics.median([r["tps"] for r in ok]),
    }


# --------------------------------------------------------------------------
DEFAULT_SAMPLES = [
    "用一句话解释什么是私有化部署。",
    "把下面这句话翻译成英文：数据不出内网。",
    "列出三条本地部署大模型的注意事项。",
]


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--models", required=True,
                    help="逗号分隔的模型全名，例如 qwen2:1.5b,qwen2:7b")
    ap.add_argument("--samples", help="真实业务样本文件，一行一条；不给就用内置示例")
    ap.add_argument("--concurrency", type=int, default=0,
                    help="大于 0 时额外做一轮并发压测")
    ap.add_argument("--outdir", default="./bench_out", help="答案存档目录")
    args = ap.parse_args()

    if args.samples:
        with open(args.samples, encoding="utf-8") as f:
            samples = [ln.strip() for ln in f if ln.strip()]
    else:
        samples = DEFAULT_SAMPLES
        print("⚠️  用的是内置示例样本，结论只能参考。")
        print("    真正拍板前请换成你自己的三十条真实业务数据。\n")

    for model in [m.strip() for m in args.models.split(",") if m.strip()]:
        print("=" * 60)
        print("模型：%s" % model)
        print("预热中（首次加载可能要几十秒）…")
        if not warmup(model):
            continue

        print("串行跑 %d 条样本：" % len(samples))
        summary = bench_serial(model, samples, args.outdir)
        if summary:
            print("  —— 汇总 ——")
            for k, v in summary.items():
                print("  %-16s %s" % (k, v))

        if args.concurrency > 0:
            print("并发 %d 压测：" % args.concurrency)
            for k, v in bench_parallel(model, samples[0], args.concurrency).items():
                print("  %-16s %s" % (k, v))

    print("=" * 60)
    print("速度数据已打印，答案存档在 %s" % args.outdir)
    print("⚠️  质量必须人工逐条看，脚本不替你判断哪个模型答得好。")
    return 0


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

"""骨架模板：私有化部署上线前的自检脚本。

复制这个文件到你的运维仓库，把 TODO 填成自己环境的值，
上线前跑一次，全绿再把地址发给同事。

用法：
    OLLAMA_BASE=http://10.0.0.20:11434 python3 deploy_probe_skeleton.py
"""
import json
import os
import urllib.error
import urllib.request

# TODO: 换成你自己的服务地址；口令/令牌一律走环境变量，不要写进源码
BASE = os.environ.get("OLLAMA_BASE", "http://127.0.0.1:11434")
# TODO: 如果前面挂了带鉴权的网关，把令牌放进环境变量再读出来
TOKEN = os.environ.get("OLLAMA_GATEWAY_TOKEN", "")

# TODO: 换成你们生产上真正要用的模型全名（必须带冒号版本）
REQUIRED_MODELS = [
    "qwen2:1.5b",
]

TIMEOUT = 10


def call(path, payload=None):
    """统一的请求封装：带上鉴权头，GET 与 POST 都走这里。"""
    url = BASE.rstrip("/") + path
    headers = {"Accept": "application/json"}
    if TOKEN:
        headers["Authorization"] = "Bearer " + TOKEN
    data = None
    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers)
    with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
        return json.loads(resp.read().decode("utf-8"))


def check_alive():
    """① 服务在不在。"""
    info = call("/api/version")
    return True, "版本 %s" % info.get("version", "未知")


def check_models():
    """② 生产要用的模型是不是都已经拉好。"""
    have = {m["name"] for m in call("/api/tags").get("models", [])}
    missing = [m for m in REQUIRED_MODELS if m not in have]
    if missing:
        return False, "缺模型：%s（先 ollama pull）" % "、".join(missing)
    return True, "模型齐全：%s" % "、".join(REQUIRED_MODELS)


def check_infer():
    """③ 真发一次推理，只有拿到非空回答才算通。"""
    payload = {
        "model": REQUIRED_MODELS[0],
        "messages": [{"role": "user", "content": "只回复两个字：正常"}],
        "stream": False,
        # TODO: 探活不需要长回答，按需要调小
        "options": {"num_predict": 16},
    }
    text = call("/api/chat", payload).get("message", {}).get("content", "")
    if not text.strip():
        return False, "模型返回空内容"
    return True, "推理正常，返回 %d 字" % len(text.strip())


# TODO: 按需要继续补自己的检查项，例如磁盘余量、显存占用、网关证书有效期
CHECKS = [
    ("服务存活", check_alive),
    ("模型就绪", check_models),
    ("推理可用", check_infer),
]


def main():
    print("探活目标：%s" % BASE)
    failed = 0
    for name, fn in CHECKS:
        try:
            ok, detail = fn()
        except urllib.error.HTTPError as exc:
            ok, detail = False, "HTTP %s" % exc.code
        except urllib.error.URLError as exc:
            ok, detail = False, "连不上：%s" % exc.reason
        print("[%s] %-8s %s" % ("通过" if ok else "失败", name, detail))
        if not ok:
            failed += 1
    print("\n结论：%s" % ("全部通过" if failed == 0 else "有 %d 项没过，别上线" % failed))
    return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(main())