KaLM-Jev 装进本地:下载、部署、跑分全记录

我把 KaLM-Jev 下下来,在一台没有独显的 Windows 机器上跑通了。官方对它的定位是一个「本地判断引擎」。

我拿 8 条餐饮评价问了它 48 道判断题,它答对 30 道。

下面是完整的下载、部署、实测过程,每一步都给了可以照着跑的代码,也包括官方参数表里没写的那部分。

01  先说结论

02  它是什么,以及参数表里没写的东西

03  下载:国内网络的第一道坎

set HF_ENDPOINT=https://hf-mirror.com

mkdir -p kalmjev-lab/models/KaLM-Reranker-V1-Nano-R2

# dl_par.py —— 8 段并行分块下载,再按顺序合并

import concurrent.futures as cf, os, subprocess

URL = ("https://hf-mirror.com/KaLM-Embedding/KaLM-Reranker-V1-Nano-R2"

       "/resolve/main/model.safetensors")

TOTAL, N = 1572182264, 8

DEST = "models/KaLM-Reranker-V1-Nano-R2/model.safetensors"

TMP = "_parts"

os.makedirs(TMP, exist_ok=True)

C = TOTAL // N

jobs = []

for i in range(N):

    s = i * C

    e = TOTAL - 1 if i == N - 1 else (i + 1) * C - 1

    jobs.append((i, s, e))

def grab(job):

    i, s, e = job

    p = os.path.join(TMP, "part%02d" % i)

    want = e - s + 1

    if os.path.exists(p) and os.path.getsize(p) == want:

        return "part%02d skip" % i

    subprocess.run(["curl", "-sL", "--max-time", "3600", "--retry", "3",

                    "-r", "%d-%d" % (s, e), URL, "-o", p], check=True)

    return "part%02d got %d/%d" % (i, os.path.getsize(p), want)

with cf.ThreadPoolExecutor(max_workers=N) as ex:

    for r in ex.map(grab, jobs):

        print(r, flush=True)

with open(DEST, "wb") as out:

    for i in range(N):

        out.write(open(os.path.join(TMP, "part%02d" % i), "rb").read())

size = os.path.getsize(DEST)

print("merged %d expect %d %s" % (size, TOTAL, "OK" if size == TOTAL else "MISMATCH"))

python -c "import hashlib;print(hashlib.sha256(open('models/KaLM-Reranker-V1-Nano-R2/model.safetensors','rb').read()).hexdigest())"

04  部署:一套钉死的依赖

python -m venv envs/kalmjev

envs/kalmjev/Scripts/python.exe -m pip install -U pip

pip install torch==2.8.0 --index-url https://download.pytorch.org/whl/cpu

pip install "numpy>=2,<3" transformers==5.3.0

# fetch_wheels.py 骨架:挑轮子 -> 抓下来 -> 让 pip 离线装

import json, os, re, subprocess

WHEELS = "wheels"

INDEX = "https://pypi.tuna.tsinghua.edu.cn/simple"

REQS = ["transformers==5.3.0", "tokenizers==0.22.2", "safetensors==0.7.0",

        "numpy>=2,<3", "huggingface_hub>=1.3,<2", "fastapi>=0.115,<1",

        "uvicorn>=0.30,<1", "pydantic>=2.10,<3", "httpx>=0.27,<1"]

os.makedirs(WHEELS, exist_ok=True)

def curl(url, out=None):

    cmd = ["curl", "-sL", "--max-time", "180", "--retry", "3", url]

    if out:

        subprocess.run(cmd + ["-o", out], check=True)

        return b""

    return subprocess.run(cmd, capture_output=True).stdout

def pick(files):

    """只留 cp313 / py3 / abi3 且 win_amd64 或 any 的正式版,取版本号最大的一个"""

    best = None

    for f in files:

        fn = f["filename"]

        if not fn.endswith(".whl") or len(fn[:-4].split("-")) < 5:

            continue

        py, abi, plat = fn[:-4].split("-")[-3:]

        if abi not in ("cp313", "abi3", "none") or plat not in ("win_amd64", "any"):

            continue

        ver = fn[:-4].split("-")[1]

        if not re.fullmatch(r"[0-9]+(\.[0-9]+)*", ver):   # 排除 rc / dev

            continue

        key = [int(x) if x.isdigit() else 0 for x in re.split(r"[.\-+]", ver)]

        if best is None or key > best[0]:

            best = (key, f)

    return best[1] if best else None

for it in range(1, 41):

    p = subprocess.run(["python", "-m", "pip", "install", "--no-index",

                        "--find-links", WHEELS] + REQS, capture_output=True)

    text = (p.stdout + p.stderr).decode("utf-8", "replace")

    if p.returncode == 0:

        print("iter %02d SUCCESS" % it)

        break

    miss = sorted(set(re.findall(

        r"Could not find a version that satisfies the requirement ([^\s<]+)", text)))

    if not miss:

        print("iter %02d 无进展,停" % it)

        break

    print("iter %02d 缺 %s" % (it, ",".join(miss)))

    for n in miss:

        raw = curl("%s/%s/" % (INDEX, n)).decode("utf-8", "replace")

        f = pick(json.loads(raw).get("files", []))

        if f is None:

            print("   没有匹配的轮子: " + n)

            continue

        dest = os.path.join(WHEELS, f["filename"])

        if not os.path.exists(dest):

            curl(f["url"], out=dest)

        print("   got " + f["filename"])

python -c "from transformers import T5Gemma2ForConditionalGeneration; print('ok')"

kalm-jev serve \

    --model kalm-jev-nano \

    --model-path ./models/KaLM-Reranker-V1-Nano-R2 \

    --device cpu --dtype float32 --batch-size 4

set PYTHONPATH=repo/src

python -m kalm_jev.server serve --model kalm-jev-nano \

    --model-path ./models/KaLM-Reranker-V1-Nano-R2 \

    --device cpu --dtype float32

# 读当前进程的驻留内存与峰值内存

import ctypes, ctypes.wintypes as wt

class PMC(ctypes.Structure):

    _fields_ = [("cb", wt.DWORD), ("PageFaultCount", wt.DWORD),

        ("PeakWorkingSetSize", ctypes.c_size_t),

        ("WorkingSetSize", ctypes.c_size_t),

        ("QuotaPeakPagedPoolUsage", ctypes.c_size_t),

        ("QuotaPagedPoolUsage", ctypes.c_size_t),

        ("QuotaPeakNonPagedPoolUsage", ctypes.c_size_t),

        ("QuotaNonPagedPoolUsage", ctypes.c_size_t),

        ("PagefileUsage", ctypes.c_size_t),

        ("PeakPagefileUsage", ctypes.c_size_t)]

_k32 = ctypes.windll.kernel32

_k32.GetCurrentProcess.restype = ctypes.c_void_p

_psapi = ctypes.windll.psapi

_psapi.GetProcessMemoryInfo.argtypes = [ctypes.c_void_p, ctypes.POINTER(PMC), wt.DWORD]

_psapi.GetProcessMemoryInfo.restype = wt.BOOL

def rss_mb():

    c = PMC()

    c.cb = ctypes.sizeof(PMC)

    _psapi.GetProcessMemoryInfo(_k32.GetCurrentProcess(), ctypes.byref(c), c.cb)

    return c.WorkingSetSize / 1048576.0, c.PeakWorkingSetSize / 1048576.0

05  跑通第一个请求

curl http://127.0.0.1:8000/health

curl -X POST http://127.0.0.1:8000/v1/systemone \

    -H "Content-Type: application/json" \

    --data-binary @repo/examples/mixed.json

{

  "department": {

    "probabilities": {

      "returns": 0.5007, "shipping": 0.3650, "billing": 0.1342

    },

    "confidence": 0.1046,

    "choice": "returns"

  },

  "bug_severity": {

    "probabilities": { "0": 0.0463, "1": 0.4773, "2": 0.4764 },

    "confidence": 0.2277,

    "score": 1.4301

  },

  "is_human_escalation": { "noul": 0.9862 },

  "is_repeat_contact":    { "noul": 0.1111 }

}

"calibration": "none",

"confidence_method": "normalized_entropy_v1",

"usage": { "input_tokens": 1165, "output_tokens": 0 }

dup = ('{"state":"x","questions":{"q":{"type":"noul","instructions":"Is it urgent?"},'

'"q":{"type":"noul","instructions":"Is it urgent?"}}}')

eng.evaluate(dup)                                  # 报 Duplicate JSON key: q

eng.evaluate('{"state":"x","questions":{"q":{"type":"noul","instructions":"i"}},"bogus":1}')

eng.evaluate('{"state":"NaN","questions":{}}')     # 空 questions 也被拒

06  测试一:官方自带的 36 条双语题

cases = json.load(open("repo/tests/semantic_cases.json", encoding="utf-8"))

stat = {}

for c in cases:

    resp = eng.evaluate(json.dumps(c["request"], ensure_ascii=False))

    a = resp["answers"][c["question_id"]]

    if c["kind"] == "choice":

        got = a.get("choice")

    elif c["kind"] == "score":

        got = int(round(a.get("score")))

    else:

        got = 1 if a.get("noul", 0) >= 0.5 else 0

    s = stat.setdefault((c["language"], c["kind"]), [0, 0])

    s[0] += 1 if got == c["expected"] else 0

    s[1] += 1

07  测试二:餐饮评价,48 个判断

REVIEWS = [

 {"id": "R1", "label": "好评",

  "text": "招牌红烧肉肥而不腻，配的梅干菜吸饱了汤汁，两个人点三个菜刚好。"

          "服务员看我们带了小孩，主动换到了靠墙的位置。人均80，在这个地段不算贵，下次还来。",

  "gold": {"overall_sentiment": "positive", "main_aspect": "taste", "star_rating": 5,

           "needs_human": 0, "food_safety_risk": 0, "is_suspicious_review": 0}},

 {"id": "R4", "label": "食安风险-菜里有异物",

  "text": "吃到一半在青菜里发现一根头发，还是卷的那种。叫服务员过来，她说「可能是不小心掉的」，"

          "然后把这盘菜退掉了，其他菜照收钱。回家之后拉了两次肚子。",

  "gold": {"overall_sentiment": "negative", "main_aspect": "food_safety", "star_rating": 1,

           "needs_human": 1, "food_safety_risk": 1, "is_suspicious_review": 0}},

 # R2 差评 / R3 中评 / R5 外卖 / R6 空话 / R7 索赔 / R8 反讽，结构相同

]

QUESTIONS = {

"star_rating": {"type": "score",

"instructions": "What star rating would this reviewer give?",

"criteria": ["1 star: severely dissatisfied, a serious problem occurred",

"2 stars: dissatisfied, several complaints",

"3 stars: average, neither good nor bad",

"4 stars: satisfied, minor issues only",

"5 stars: very satisfied, no complaint at all"]},

"overall_sentiment": {"type": "choice",

"instructions": "What is the overall sentiment of this restaurant review?",

"criteria": {"positive": "Overall praise; the reviewer recommends or would return",

"neutral": "Mixed or balanced; praise and complaint both present",

"negative": "Overall complaint; the reviewer will not return"}},

"needs_human": {"type": "noul",

"instructions": "Does this review need a human staff member to follow up?",

"criteria": {"true": "Demands compensation, refund, apology, or reports harm",

"false": "Ordinary feedback that no one has to act on"}},

# main_aspect / food_safety_risk / is_suspicious_review 结构相同

}

def grade(resp, gold):

    rows = []

    for qid, want in gold.items():

        a = resp["answers"].get(qid, {})

        if a.get("type") == "choice":

            got = a.get("choice")

        elif a.get("type") == "score":

            got = int(round(a.get("score")))           # 期望档位四舍五入

        else:

            got = 1 if a.get("noul", 0) >= 0.5 else 0  # 0.5 切二值

        rows.append((qid, got, want, got == want, a.get("confidence")))

    return rows

"overall_sentiment": {

"probabilities": { "positive": 0.3607, "neutral": 0.0232, "negative": 0.6161 },

"confidence": 0.3141,

"choice": "negative"

}

08  置信度比答案更有用

buckets = [(0, 0.1), (0.1, 0.2), (0.2, 0.3), (0.3, 0.4), (0.4, 1.01)]

for lo, hi in buckets:

    sel = [r for r in rows if r["conf"] is not None and lo <= r["conf"] < hi]

    if not sel:

        continue

    n_ok = sum(1 for r in sel if r["ok"])

    print("%.1f-%.1f  %d/%d = %.0f%%" % (lo, hi, n_ok, len(sel), 100.0 * n_ok / len(sel)))

09  五个必须先知道的坑

10  餐饮评价这个场景,我的结论