主流大模型全景与选型

"""探测一个 OpenAI 兼容服务：它到底有哪些模型、哪个真能调通。

为什么要探测：模型列表里有，不代表你的账号这一刻调得动。
只有真发一次请求、看响应里的 model 字段，才算「调得通」。

运行前配置：
    export LLM_API_KEY="你的key"
    export LLM_BASE_URL="https://api.deepseek.com"
依赖：pip install "openai>=1.0"
"""
import os
import sys

from openai import OpenAI

BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.deepseek.com")
client = OpenAI(api_key=os.environ["LLM_API_KEY"], base_url=BASE_URL, timeout=30.0)


def list_models():
    """第一步：把服务端声明支持的模型列出来。"""
    try:
        models = client.models.list()
    except Exception as err:
        print("[x] 列举模型失败:", type(err).__name__, err)
        return []
    names = sorted(m.id for m in models.data)
    print("服务 %s 声明了 %d 个模型:" % (BASE_URL, len(names)))
    for name in names:
        print("   ", name)
    return names


def probe(model):
    """第二步：真发一次最小请求，看它是否返回、返回的 model 是哪个。

    只要 1 个 token 的输出，成本可以忽略不计。
    """
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": "hi"}],
            max_tokens=1,
        )
    except Exception as err:
        return False, "%s: %s" % (type(err).__name__, str(err)[:80])

    # 注意：响应里的 model 可能与你请求的名字不同（别名、灰度、降级都会改写它）
    served = response.model
    note = "实际服务模型 %s" % served
    if served != model:
        note += "（与请求名 %s 不一致）" % model
    return True, note


def main():
    names = sys.argv[1:] or list_models()
    if not names:
        print("没有可探测的模型名，可在命令行直接传入，如：")
        print("   python3 probe_endpoint.py deepseek-flash")
        return

    print("\n逐个实测（每个只花 1 个输出 token）:")
    print("%-34s %-6s %s" % ("模型", "可用", "说明"))
    print("-" * 78)
    for name in names:
        ok, note = probe(name)
        print("%-34s %-6s %s" % (name, "✓" if ok else "✗", note))


if __name__ == "__main__":
    main()

$ export LLM_API_KEY="你的key"
$ export LLM_BASE_URL="https://api.deepseek.com"

# 用法一：不带参数 —— 先列举服务声明支持的模型，再逐个实测
$ python3 probe_endpoint.py

服务 https://api.deepseek.com 声明了 N 个模型:
    <模型名 1>
    <模型名 2>

逐个实测（每个只花 1 个输出 token）:
模型                               可用   说明
------------------------------------------------------------------------------
<模型名 1>                         ✓     实际服务模型 <响应里的 model 字段>
<模型名 2>                         ✗     NotFoundError: ...

# 用法二：只测你关心的那几个，跳过列举
$ python3 probe_endpoint.py deepseek-flash deepseek-v4-pro

# 换一家服务，只改 base_url，脚本一个字都不用动
$ export LLM_BASE_URL="https://open.bigmodel.cn/api/paas/v4/"
$ python3 probe_endpoint.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""显存估算——选型时最先要回答的问题：这张卡装不装得下。

只用标准库，直接 python3 vram_estimate.py 就能跑。
算的是量级，不是精确值；真实占用还受推理框架、批大小、显存碎片影响，
但用它筛掉明显装不下的候选，足够了。
"""

# 每个参数在不同精度下占多少字节
BYTES_PER_PARAM = {
    "FP32": 4,
    "FP16": 2,
    "BF16": 2,
    "INT8": 1,
    "INT4": 0.5,
}

# 推理时除权重外的额外开销系数（激活值、框架常驻、显存碎片）
# 经验值：按权重的 1.2 倍留余量
OVERHEAD = 1.2


def weight_gb(params_b, dtype):
    """权重占用（GB）。params_b 单位是十亿（B）。"""
    return params_b * 1e9 * BYTES_PER_PARAM[dtype] / (1024 ** 3)


def kv_cache_gb(layers, kv_heads, head_dim, seq_len, batch, dtype="FP16"):
    """KV cache 占用（GB）。

    每层要为 K 和 V 各存一份，所以乘 2。
    GQA 的关键就在 kv_heads——它比 query 头少，KV cache 直接按比例缩小。
    """
    per_token = 2 * layers * kv_heads * head_dim * BYTES_PER_PARAM[dtype]
    return per_token * seq_len * batch / (1024 ** 3)


def fits(need_gb, card_gb):
    return "✅ 装得下" if need_gb <= card_gb else "❌ 装不下"


def main():
    print("=" * 68)
    print("一、同一个 7B 模型，不同精度下的权重占用")
    print("=" * 68)
    print("%-8s %-12s %-14s %s" % ("精度", "权重(GB)", "含开销(GB)", "24GB 卡"))
    print("-" * 68)
    for dtype in ["FP32", "FP16", "INT8", "INT4"]:
        w = weight_gb(7, dtype)
        total = w * OVERHEAD
        print("%-8s %-12.2f %-14.2f %s" % (dtype, w, total, fits(total, 24)))
    print()
    print("结论：量化是把大模型塞进小卡的第一手段。")
    print("      FP16 到 INT4，显存直接降到四分之一。")
    print()

    print("=" * 68)
    print("二、不同规模模型在 FP16 下要几张 80GB 卡")
    print("=" * 68)
    print("%-16s %-14s %-12s %s" % ("模型规模", "权重(GB)", "含开销(GB)", "80GB 卡数"))
    print("-" * 68)
    for params in [7, 13, 32, 70, 235]:
        w = weight_gb(params, "FP16")
        total = w * OVERHEAD
        cards = int(total // 80) + (1 if total % 80 else 0)
        print("%-16s %-14.2f %-12.2f %d 张" % ("%dB" % params, w, total, cards))
    print()
    print("注意：235B 这一行对 MoE 模型同样成立——")
    print("      权重要全部加载，省的是计算量，不是显存。")
    print()

    print("=" * 68)
    print("三、KV cache：长上下文的隐形成本")
    print("=" * 68)
    # 参数取自公开的 7B 级模型常见配置
    layers, head_dim = 28, 128
    print("配置：28 层，head_dim=128，FP16，batch=1")
    print("%-14s %-12s %-14s %s" % ("上下文长度", "MHA(28头)", "GQA(4头)", "GQA 省了"))
    print("-" * 68)
    for seq in [2048, 8192, 32768, 131072]:
        mha = kv_cache_gb(layers, 28, head_dim, seq, 1)
        gqa = kv_cache_gb(layers, 4, head_dim, seq, 1)
        save = (1 - gqa / mha) * 100
        print("%-14s %-12.2f %-14.2f %.0f%%" % ("%d" % seq, mha, gqa, save))
    print()
    print("这就是 GQA 被所有新模型采用的原因：")
    print("  KV 头从 28 降到 4，KV cache 直接降到 1/7，长上下文才跑得起。")
    print()

    print("=" * 68)
    print("四、一个完整的选型判断示例")
    print("=" * 68)
    scenarios = [
        ("单张 4090 24GB 跑 7B", 7, "INT4", 24, 28, 4, 8192),
        ("单张 A100 80GB 跑 32B", 32, "FP16", 80, 64, 8, 8192),
        ("单张 A100 80GB 跑 70B", 70, "FP16", 80, 80, 8, 8192),
    ]
    for name, params, dtype, card, layers_n, kv_h, seq in scenarios:
        w = weight_gb(params, dtype) * OVERHEAD
        kv = kv_cache_gb(layers_n, kv_h, 128, seq, 1)
        total = w + kv
        print("%s：" % name)
        print("   权重+开销 %.1f GB + KV cache %.1f GB = %.1f GB  ->  %s"
              % (w, kv, total, fits(total, card)))
    print()
    print("70B FP16 单卡装不下，只有三条路：上多卡、上量化、换小模型。")
    print("选型阶段就要算清楚，不要等部署时才发现。")


if __name__ == "__main__":
    main()

====================================================================
一、同一个 7B 模型，不同精度下的权重占用
====================================================================
精度       权重(GB)       含开销(GB)        24GB 卡
--------------------------------------------------------------------
FP32     26.08        31.29          ❌ 装不下
FP16     13.04        15.65          ✅ 装得下
INT8     6.52         7.82           ✅ 装得下
INT4     3.26         3.91           ✅ 装得下

结论：量化是把大模型塞进小卡的第一手段。
      FP16 到 INT4，显存直接降到四分之一。

====================================================================
二、不同规模模型在 FP16 下要几张 80GB 卡
====================================================================
模型规模             权重(GB)         含开销(GB)      80GB 卡数
--------------------------------------------------------------------
7B               13.04          15.65        1 张
13B              24.21          29.06        1 张
32B              59.60          71.53        1 张
70B              130.39         156.46       2 张
235B             437.72         525.27       7 张

注意：235B 这一行对 MoE 模型同样成立——
      权重要全部加载，省的是计算量，不是显存。

====================================================================
三、KV cache：长上下文的隐形成本
====================================================================
配置：28 层，head_dim=128，FP16，batch=1
上下文长度          MHA(28头)     GQA(4头)        GQA 省了
--------------------------------------------------------------------
2048           0.77         0.11           86%
8192           3.06         0.44           86%
32768          12.25        1.75           86%
131072         49.00        7.00           86%

这就是 GQA 被所有新模型采用的原因：
  KV 头从 28 降到 4，KV cache 直接降到 1/7，长上下文才跑得起。

====================================================================
四、一个完整的选型判断示例
====================================================================
单张 4090 24GB 跑 7B：
   权重+开销 3.9 GB + KV cache 0.4 GB = 4.3 GB  ->  ✅ 装得下
单张 A100 80GB 跑 32B：
   权重+开销 71.5 GB + KV cache 2.0 GB = 73.5 GB  ->  ✅ 装得下
单张 A100 80GB 跑 70B：
   权重+开销 156.5 GB + KV cache 2.5 GB = 159.0 GB  ->  ❌ 装不下

70B FP16 单卡装不下，只有三条路：上多卡、上量化、换小模型。
选型阶段就要算清楚，不要等部署时才发现。

"""选型打分：把「哪个模型更合适」从拍脑袋变成一张可复算的表。

两步走，顺序不能反：
  第一步 硬门槛淘汰——不满足业务底线的直接出局，多便宜都不要；
  第二步 加权排序——在活下来的候选里，按业务权重算总分。

只做第二步是典型错误：一个效果不及格但极便宜的模型会把总分刷上去。
纯标准库，直接 python3 select_model.py 就能跑。
"""
# 候选模型的属性。分数一律 0~10，含义见 CRITERIA。
# ⚠️ 这些数字必须由你自己的实测填入，不要抄榜单。
CANDIDATES = [
    {"name": "闭源旗舰 API",   "quality": 10, "cost": 3, "latency": 6, "privacy": 2, "controllable": 3},
    {"name": "闭源轻量 API",   "quality": 7,  "cost": 8, "latency": 9, "privacy": 2, "controllable": 3},
    {"name": "开源权重 · 云端", "quality": 8,  "cost": 6, "latency": 7, "privacy": 5, "controllable": 8},
    {"name": "开源权重 · 内网", "quality": 7,  "cost": 5, "latency": 5, "privacy": 10, "controllable": 10},
    {"name": "小参数本地部署",  "quality": 4,  "cost": 10, "latency": 8, "privacy": 10, "controllable": 10},
]

CRITERIA = {
    "quality":      "任务效果：用你自己的测试集测出来的准确率/通过率",
    "cost":         "性价比：同样一批请求花多少钱，分越高越便宜",
    "latency":      "响应速度：首 token 延迟与整体吞吐",
    "privacy":      "数据可控：数据是否离开内网",
    "controllable": "可改造性：能不能微调、能不能锁版本",
}

# 每个场景两部分：gates 是一票否决的硬门槛，weights 是排序权重。
SCENARIOS = {
    "对外客服（量大、要便宜、要快）": {
        "gates": {"quality": 6},                 # 答错了要赔钱，效果有底线
        "weights": {"quality": 3, "cost": 5, "latency": 5, "privacy": 1, "controllable": 1},
    },
    "内部知识库（数据不能出内网）": {
        "gates": {"privacy": 8, "quality": 6},   # 合规是红线，不参与权衡
        "weights": {"quality": 3, "cost": 2, "latency": 2, "privacy": 5, "controllable": 4},
    },
    "复杂推理（效果压倒一切）": {
        "gates": {"quality": 8},
        "weights": {"quality": 8, "cost": 1, "latency": 1, "privacy": 1, "controllable": 1},
    },
}


def passes(candidate, gates):
    """硬门槛检查：返回 (是否通过, 被哪一项卡住)。"""
    for key, floor in gates.items():
        if candidate[key] < floor:
            return False, "%s=%d < %d" % (key, candidate[key], floor)
    return True, ""


def score(candidate, weights):
    """加权求和后归一化到 0~10，方便横向比较。"""
    total = sum(candidate[k] * w for k, w in weights.items())
    return total / sum(weights.values())


def main():
    print("评分维度：")
    for key, desc in CRITERIA.items():
        print("  %-13s %s" % (key, desc))

    for scenario, conf in SCENARIOS.items():
        gates, weights = conf["gates"], conf["weights"]
        print("\n" + "=" * 64)
        print("场景：%s" % scenario)
        print("硬门槛：%s" % "  ".join("%s>=%d" % (k, v) for k, v in gates.items()))
        print("权重：  %s" % "  ".join("%s=%d" % (k, v) for k, v in weights.items()))
        print("-" * 64)

        survivors, rejected = [], []
        for candidate in CANDIDATES:
            ok, reason = passes(candidate, gates)
            (survivors if ok else rejected).append((candidate, reason))

        for candidate, reason in rejected:
            print("  ✗ %-16s 出局（%s）" % (candidate["name"], reason))

        ranked = sorted(survivors, key=lambda pair: -score(pair[0], weights))
        for rank, (candidate, _) in enumerate(ranked, 1):
            print("  %d. %-16s 得分 %.2f" % (rank, candidate["name"],
                                            score(candidate, weights)))
        if ranked:
            print("  => 首选：%s" % ranked[0][0]["name"])
        else:
            print("  => 无人通过门槛，说明门槛定得过高或候选池太小")

    print("\n" + "=" * 64)
    print("同一批候选，换一组门槛和权重，冠军就换人——")
    print("所以「哪个模型最好」这个问题不成立，成立的是「在我的约束下哪个最好」。")


if __name__ == "__main__":
    main()

评分维度：
  quality       任务效果：用你自己的测试集测出来的准确率/通过率
  cost          性价比：同样一批请求花多少钱，分越高越便宜
  latency       响应速度：首 token 延迟与整体吞吐
  privacy       数据可控：数据是否离开内网
  controllable  可改造性：能不能微调、能不能锁版本

================================================================
场景：对外客服（量大、要便宜、要快）
硬门槛：quality>=6
权重：  quality=3  cost=5  latency=5  privacy=1  controllable=1
----------------------------------------------------------------
  ✗ 小参数本地部署          出局（quality=4 < 6）
  1. 闭源轻量 API         得分 7.40
  2. 开源权重 · 云端        得分 6.80
  3. 开源权重 · 内网        得分 6.07
  4. 闭源旗舰 API         得分 5.33
  => 首选：闭源轻量 API

================================================================
场景：内部知识库（数据不能出内网）
硬门槛：privacy>=8  quality>=6
权重：  quality=3  cost=2  latency=2  privacy=5  controllable=4
----------------------------------------------------------------
  ✗ 闭源旗舰 API         出局（privacy=2 < 8）
  ✗ 闭源轻量 API         出局（privacy=2 < 8）
  ✗ 开源权重 · 云端        出局（privacy=5 < 8）
  ✗ 小参数本地部署          出局（quality=4 < 6）
  1. 开源权重 · 内网        得分 8.19
  => 首选：开源权重 · 内网

================================================================
场景：复杂推理（效果压倒一切）
硬门槛：quality>=8
权重：  quality=8  cost=1  latency=1  privacy=1  controllable=1
----------------------------------------------------------------
  ✗ 闭源轻量 API         出局（quality=7 < 8）
  ✗ 开源权重 · 内网        出局（quality=7 < 8）
  ✗ 小参数本地部署          出局（quality=4 < 8）
  1. 闭源旗舰 API         得分 7.83
  2. 开源权重 · 云端        得分 7.50
  => 首选：闭源旗舰 API

================================================================
同一批候选，换一组门槛和权重，冠军就换人——
所以「哪个模型最好」这个问题不成立，成立的是「在我的约束下哪个最好」。

"""选型实测骨架：拿自己的题目去考候选模型，出一张可复算的对比表。

榜单只能用来圈定候选，最终拍板必须靠你自己的测试集。
这份骨架把「跑一批题 → 记录耗时与 token → 判对错 → 出表」串起来。
复制后改 5 处 TODO。

运行前配置各家的 key，例如：
    export DEEPSEEK_API_KEY=...
依赖：pip install "openai>=1.0"
"""
import json
import os
import time

from openai import OpenAI

# TODO 1：登记参与对比的候选。base_url 与模型名以各家官方文档为准。
CANDIDATES = [
    {"label": "A 家轻量", "base_url": "https://api.deepseek.com",
     "key_env": "DEEPSEEK_API_KEY", "model": "deepseek-flash"},
    # {"label": "B 家旗舰", "base_url": "...", "key_env": "...", "model": "..."},
]

# TODO 2：换成你自己的题目。20~50 道覆盖真实业务分布，比 1000 道通用题有用。
TASKS = [
    {"id": "cls-1", "prompt": "判断这句话的情感，只回答 正面 或 负面：这家店服务真差。",
     "expect": "负面"},
    {"id": "cls-2", "prompt": "判断这句话的情感，只回答 正面 或 负面：物流很快，包装完好。",
     "expect": "正面"},
]

# TODO 3：换成你的判分逻辑。分类看精确匹配，生成类可换成 ROUGE 或人工抽检。
def judge(output, expect):
    return expect in output


def ask(candidate, prompt):
    """发一次请求，返回 (文本, 耗时秒, 输出token数)。失败返回文本 None。"""
    api_key = os.environ.get(candidate["key_env"])
    if not api_key:
        return None, 0.0, 0
    client = OpenAI(api_key=api_key, base_url=candidate["base_url"], timeout=60.0)

    start = time.time()
    try:
        response = client.chat.completions.create(
            model=candidate["model"],
            messages=[{"role": "user", "content": prompt}],
            # TODO 4：判定类任务把 temperature 压到 0，让结果可复现
            temperature=0,
            max_tokens=64,
        )
    except Exception as err:
        print("   [x] %s: %s" % (candidate["label"], type(err).__name__))
        return None, time.time() - start, 0

    elapsed = time.time() - start
    return (response.choices[0].message.content.strip(),
            elapsed,
            response.usage.completion_tokens)


def main():
    report = []
    for candidate in CANDIDATES:
        print("测试 %s（%s）" % (candidate["label"], candidate["model"]))
        correct, total_time, total_tokens, done = 0, 0.0, 0, 0

        for task in TASKS:
            output, elapsed, tokens = ask(candidate, task["prompt"])
            total_time += elapsed
            total_tokens += tokens
            if output is None:
                continue
            done += 1
            ok = judge(output, task["expect"])
            correct += int(ok)
            print("   %-8s %s  %.2fs  %s" % (task["id"], "✓" if ok else "✗",
                                             elapsed, output[:24]))

        row = {
            "模型": candidate["label"],
            "完成数": done,
            "正确率": round(correct / done, 4) if done else None,
            "平均耗时秒": round(total_time / done, 3) if done else None,
            "输出token合计": total_tokens,
        }
        report.append(row)

    print("\n对比表：")
    if report:
        cols = list(report[0].keys())
        print("  " + "  ".join("%-12s" % c for c in cols))
        for row in report:
            print("  " + "  ".join("%-12s" % row[c] for c in cols))

    # TODO 5：落盘存档，下次换版本时直接对比，避免「感觉变差了」这种无证据判断
    with open("bench_report.json", "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    print("\n已写入 bench_report.json")


if __name__ == "__main__":
    main()