#!/usr/bin/env python3 """全量能力评测（10 个基准），用于 Qwen3.8-27B W4A16 两个产品的对照。设计要点: - 抽样索引只由 --seed 决定,与模型无关 => 两个模型严格可比 - 思维链默认开满(enable_thinking=true),正文取 message.content, 思考取 message.reasoning_content(vLLM 的 --reasoning-parser qwen3 会分离) - 每个数据集独立 checkpoint 落盘,中途挂掉可续跑 - 判分器分五类: mc(选择) / num(数值) / exact(精确) / code(执行单测) / ifeval(规则) 用法: python3 capability_eval_full.py --base-url http://127.0.0.1:9411 \ --model bench-model --label UNC --tasks all """ import argparse import json import os import re import subprocess import sys import tempfile import time from concurrent.futures import ThreadPoolExecutor from random import Random import requests MC_LETTERS = "ABCDE" MC_LETTERS_LONG = "ABCDEFGHIJKLMNOP" # TruthfulQA mc1 选项常超 5 个 CMMLU_DIR = os.path.expanduser(os.environ.get("CMMLU_DIR", "~/cmmlu-data")) # 每个任务的默认抽样量与 max_tokens(思考模式下要给足) TASK_SPEC = { "mmlu": {"n": 150, "kind": "mc", "max_tokens": 4096}, "cmmlu": {"n": 150, "kind": "mc", "max_tokens": 4096}, # 2026-08-22 加。理由:MMLU 在 9B 上已接近饱和(基座 80.67%),区分度不够; # Pro 版难度上一档、选项最多 10 个(实读 12032 题),9B 上仍有区分度。 # 成本只是多一项,不引入 agent 脚手架 —— 那类(SWE-Bench/Terminal-Bench/τ3) # 工程量超过整条消融线,HLE 在 9B 上是地板,都不做。 "mmlu_pro": {"n": 150, "kind": "mc", "max_tokens": 4096}, "ceval": {"n": 150, "kind": "mc", "max_tokens": 4096}, "arc": {"n": 150, "kind": "mc", "max_tokens": 4096}, "truthfulqa": {"n": 150, "kind": "mc", "max_tokens": 4096}, "gsm8k": {"n": 100, "kind": "num", "max_tokens": 4096}, "math500": {"n": 100, "kind": "num", "max_tokens": 8192}, "bbh": {"n": 150, "kind": "exact", "max_tokens": 4096}, "humaneval": {"n": 164, "kind": "code", "max_tokens": 4096}, "ifeval": {"n": 100, "kind": "ifeval", "max_tokens": 3072}, } ALL_TASKS = list(TASK_SPEC) # ------------------------------------------------------------------ 数据加载 def _shuffled(n_total, k, seed): idx = list(range(n_total)) Random(seed).shuffle(idx) return idx[:k] def _mc_item(item_id, question, choices, gold, instruction, letters=MC_LETTERS): body = "\n".join(f"{letters[j]}. {c}" for j, c in enumerate(choices)) if isinstance(gold, int): gold_letter = letters[gold] else: g = str(gold).strip() gold_letter = g.upper() if g.upper() in letters else letters[int(g)] return { "id": item_id, "kind": "mc", "gold": gold_letter, # 2026-08-22 加:记录本题的字母上界。原先 score_mc 只按 gold 自适应 —— # 十选一时若 gold 是 A-E,匹配范围就被缩成 A-E,模型答 H 匹配不到, # 可能回落到正文里随便捡个 A-E 字母误判为对。 "hi": letters[len(choices) - 1], "prompt": f"{question.strip()}\n{body}\n\n{instruction}", } def load_items(task, n, seed): from datasets import load_dataset if task == "mmlu": ds = load_dataset("cais/mmlu", "all", split="test") return [_mc_item(f"mmlu-{i}", ds[i]["question"], ds[i]["choices"], ds[i]["answer"], "Answer with a single letter (A, B, C or D) on the last line, " "formatted exactly as: ANSWER: ") for i in _shuffled(len(ds), n, seed)] if task == "mmlu_pro": ds = load_dataset("TIGER-Lab/MMLU-Pro", split="test") out = [] for i in _shuffled(len(ds), n, seed): r = ds[i] opts = list(r["options"]) # 实读:最多 10 个,故用 MC_LETTERS_LONG gold = r["answer"] # 已经是字母 # 必须传 MC_LETTERS_LONG:默认的 MC_LETTERS 只有 ABCDE 五个字母, # 而 MMLU-Pro 最多 10 个选项 —— 不传会 IndexError,或把 F-J 的正确答案判错。 out.append(_mc_item(f"mmlupro-{r['question_id']}", r["question"], opts, gold, "Answer with a single letter on the last line, " "formatted exactly as: ANSWER: ", letters=MC_LETTERS_LONG)) return out if task == "cmmlu": import glob import pandas as pd files = sorted(glob.glob(os.path.join(CMMLU_DIR, "test", "*.csv"))) rows = [] for path in files: subject = os.path.basename(path)[:-4] df = pd.read_csv(path) for k, row in df.iterrows(): rows.append((f"cmmlu-{subject}-{k}", row)) return [_mc_item(rid, str(row["Question"]), [row["A"], row["B"], row["C"], row["D"]], row["Answer"], "最后一行只输出答案字母,格式严格为: ANSWER: <字母>") for rid, row in (rows[i] for i in _shuffled(len(rows), n, seed))] if task == "ceval": # 离线模式 get_dataset_config_names 会失败,直接从 hub 缓存扫 config 目录 base = os.path.expanduser( "~/.cache/huggingface/datasets/ceval___ceval-exam") cfgs = sorted(d for d in (os.listdir(base) if os.path.isdir(base) else []) if os.path.isdir(os.path.join(base, d))) rows = [] for cfg in cfgs: try: ds = load_dataset("ceval/ceval-exam", cfg, split="val") except Exception: continue for k in range(len(ds)): r = ds[k] if not str(r.get("answer", "")).strip(): continue rows.append((f"ceval-{cfg}-{k}", r)) return [_mc_item(rid, str(r["question"]), [r["A"], r["B"], r["C"], r["D"]], r["answer"], "最后一行只输出答案字母,格式严格为: ANSWER: <字母>") for rid, r in (rows[i] for i in _shuffled(len(rows), n, seed))] if task == "arc": ds = load_dataset("allenai/ai2_arc", "ARC-Challenge", split="test") out = [] for i in _shuffled(len(ds), n, seed): r = ds[i] labels = list(r["choices"]["label"]) texts = list(r["choices"]["text"]) gold = r["answerKey"].strip() if gold not in labels: continue body = "\n".join(f"{lab}. {t}" for lab, t in zip(labels, texts)) out.append({ "id": f"arc-{r['id']}", "kind": "mc", "gold": gold, "prompt": f"{r['question'].strip()}\n{body}\n\n" "Answer with a single letter on the last line, " "formatted exactly as: ANSWER: ", }) return out if task == "truthfulqa": ds = load_dataset("truthfulqa/truthful_qa", "multiple_choice", split="validation") out = [] for i in _shuffled(len(ds), n, seed): r = ds[i] mc1 = r["mc1_targets"] choices = list(mc1["choices"]) labels = list(mc1["labels"]) if 1 not in labels or len(choices) > len(MC_LETTERS_LONG): continue gold_idx = labels.index(1) # 固定顺序打散,但用同一 seed+题号保证两模型一致 order = list(range(len(choices))) Random(seed * 1000 + i).shuffle(order) shuffled = [choices[j] for j in order] gold_letter = MC_LETTERS_LONG[order.index(gold_idx)] body = "\n".join(f"{MC_LETTERS_LONG[j]}. {c}" for j, c in enumerate(shuffled)) out.append({ "id": f"tqa-{i}", "kind": "mc", "gold": gold_letter, "prompt": f"{r['question'].strip()}\n{body}\n\n" "Pick the single most truthful answer. Output the letter on the " "last line, formatted exactly as: ANSWER: ", }) return out if task == "gsm8k": ds = load_dataset("openai/gsm8k", "main", split="test") out = [] for i in _shuffled(len(ds), n, seed): r = ds[i] gold = r["answer"].split("####")[-1].strip().replace(",", "") out.append({ "id": f"gsm8k-{i}", "kind": "num", "gold": gold, "prompt": r["question"].strip() + "\n\nSolve it. Put the final numeric answer on the last line, " "formatted exactly as: ANSWER: ", }) return out if task == "math500": ds = load_dataset("HuggingFaceH4/MATH-500", split="test") out = [] for i in _shuffled(len(ds), n, seed): r = ds[i] out.append({ "id": f"math-{r.get('unique_id', i)}", "kind": "num", "gold": str(r["answer"]).strip(), "prompt": r["problem"].strip() + "\n\nSolve it. Put the final answer on the last line, " "formatted exactly as: ANSWER: ", }) return out if task == "bbh": base = os.path.expanduser("~/.cache/huggingface/datasets/lukaemon___bbh") cfgs = sorted(d for d in (os.listdir(base) if os.path.isdir(base) else []) if os.path.isdir(os.path.join(base, d))) rows = [] for cfg in cfgs: try: ds = load_dataset("lukaemon/bbh", cfg, split="test") except Exception: continue for k in range(len(ds)): rows.append((f"bbh-{cfg}-{k}", ds[k])) return [{ "id": rid, "kind": "exact", "gold": str(r["target"]).strip(), "prompt": str(r["input"]).strip() + "\n\nThink it through, then put the final answer on the last line, " "formatted exactly as: ANSWER: ", } for rid, r in (rows[i] for i in _shuffled(len(rows), n, seed))] if task == "humaneval": ds = load_dataset("openai/openai_humaneval", split="test") idx = list(range(len(ds))) if n >= len(ds) else _shuffled(len(ds), n, seed) out = [] for i in idx: r = ds[i] out.append({ "id": r["task_id"], "kind": "code", "gold": {"test": r["test"], "entry_point": r["entry_point"]}, "prompt": "Complete the following Python function. Output the COMPLETE " "function (including the signature) inside a single ```python " "code block. No explanation after the code block.\n\n" f"```python\n{r['prompt']}```", }) return out if task == "ifeval": ds = load_dataset("HuggingFaceH4/ifeval", split="train") return [{ "id": f"ifeval-{i}", "kind": "ifeval", "gold": {"ids": list(ds[i]["instruction_id_list"]), "kwargs": ds[i]["kwargs"]}, "prompt": ds[i]["prompt"], } for i in _shuffled(len(ds), n, seed)] raise ValueError(task) # ---------------------------------------------------------------------- 判分 def _tail_answer(text): """取 'ANSWER: xxx' 之后的内容;没有就返回全文尾部。""" m = re.findall(r"ANSWER\s*[:：]\s*(.+)", text or "", re.I) if m: return m[-1].strip() return (text or "").strip() def score_mc(text, gold, hi=None): """gold 是单个字母;匹配范围优先用题目自带的 hi(选项数决定),没有才按 gold 自适应。""" if not hi: hi = "P" if gold.upper() > "E" else "E" hi = hi.upper() pat_w = rf"\b([A-{hi}])\b" pat_b = rf"([A-{hi}])" tail = _tail_answer(text) m = re.findall(pat_w, tail.upper()) or re.findall(pat_b, tail.upper()) if m: return m[0] == gold m2 = re.findall(pat_w, (text or "").upper()) return bool(m2) and m2[-1] == gold def _norm_num(s): s = str(s).replace(",", "").replace("$", "").replace("\\!", "").strip() s = re.sub(r"\\(?:text|mbox|mathrm)\{([^}]*)\}", r"\1", s) s = s.replace("\\%", "").replace("%", "").replace("^\\circ", "").replace("\\circ", "") s = re.sub(r"\\left|\\right", "", s) s = s.strip().rstrip(".").strip() if s.startswith("\\boxed{") and s.endswith("}"): s = s[7:-1].strip() return s def score_num(text, gold): tail = _tail_answer(text) g = _norm_num(gold) cand = _norm_num(tail) if cand == g: return True # 数值等价 try: if abs(float(cand) - float(g)) < 1e-6: return True except ValueError: pass # 从尾部抓最后一个数字 nums = re.findall(r"-?\d+(?:\.\d+)?(?:/\d+)?", cand.replace(" ", "")) if nums: last = nums[-1] try: lv = eval(last) if "/" in last else float(last) # noqa: S307 - 仅数字/分数 gv = eval(g) if "/" in g and re.fullmatch(r"-?\d+/\d+", g) else float(g) if abs(lv - gv) < 1e-6: return True except Exception: pass # LaTeX 片段直接比对(去空格) return cand.replace(" ", "") == g.replace(" ", "") def score_exact(text, gold): tail = _tail_answer(text).strip() g = str(gold).strip() if tail.lower() == g.lower(): return True tail_clean = tail.strip("().,'\"[]{} ").lower() g_clean = g.strip("().,'\"[]{} ").lower() if tail_clean == g_clean: return True # BBH 的 (A)/(B) 风格 m = re.findall(r"$([A-Z])$", tail.upper()) gm = re.findall(r"$([A-Z])$", g.upper()) if m and gm: return m[0] == gm[0] return False def _extract_code(text): blocks = re.findall(r"```(?:python)?\s*\n(.*?)```", text or "", re.S) if blocks: return max(blocks, key=len) return text or "" def score_code(text, gold, timeout=12): code = _extract_code(text) if not code.strip(): return False program = ( "import math, re, sys, itertools, collections, heapq, functools, string\n" f"{code}\n\n{gold['test']}\n\ncheck({gold['entry_point']})\n" ) with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as fh: fh.write(program) path = fh.name try: r = subprocess.run([sys.executable, path], capture_output=True, timeout=timeout, text=True) return r.returncode == 0 except Exception: return False finally: try: os.unlink(path) except OSError: pass # --- IFEval: 实现常见验证器,未覆盖的 instruction 记为 skipped(不计分母) --- def _words(t): return re.findall(r"\b[\w']+\b", t or "") def _sentences(t): return [s for s in re.split(r"(?<=[.!?。！？])\s+", (t or "").strip()) if s.strip()] def _check_one_instruction(iid, kw, resp): """返回 True/False/None(None = 未实现,跳过)。""" kw = kw or {} r = resp or "" low = r.lower() if iid == "keywords:existence": return all(str(k).lower() in low for k in (kw.get("keywords") or [])) if iid == "keywords:frequency": k = str(kw.get("keyword", "")).lower() n = kw.get("frequency") rel = kw.get("relation", "at least") if not k or n is None: return None c = low.count(k) return c >= n if rel == "at least" else c <= n if iid == "keywords:forbidden_words": return not any(str(k).lower() in low for k in (kw.get("forbidden_words") or [])) if iid == "keywords:letter_frequency": letter = str(kw.get("letter", "")).lower() n = kw.get("let_frequency") rel = kw.get("let_relation", "at least") if not letter or n is None: return None c = low.count(letter) return c >= n if rel == "at least" else c <= n if iid == "length_constraints:number_words": n, rel = kw.get("num_words"), kw.get("relation", "at least") if n is None: return None c = len(_words(r)) return c >= n if rel == "at least" else c <= n if iid == "length_constraints:number_sentences": n, rel = kw.get("num_sentences"), kw.get("relation", "at least") if n is None: return None c = len(_sentences(r)) return c >= n if rel == "at least" else c <= n if iid == "length_constraints:number_paragraphs": n = kw.get("num_paragraphs") if n is None: return None return len([p for p in re.split(r"\n\s*\*\*\*\s*\n|\n\n+", r.strip()) if p.strip()]) == n if iid == "length_constraints:nth_paragraph_first_word": n, idx, fw = kw.get("num_paragraphs"), kw.get("nth_paragraph"), kw.get("first_word") if None in (n, idx, fw): return None paras = [p for p in re.split(r"\n\n+", r.strip()) if p.strip()] if len(paras) < idx: return False first = _words(paras[idx - 1]) return bool(first) and first[0].lower() == str(fw).lower() if iid == "detectable_format:number_bullet_lists": n = kw.get("num_bullets") if n is None: return None return len(re.findall(r"^\s*[\*\-]\s+", r, re.M)) == n if iid == "detectable_format:json_format": try: json.loads(re.sub(r"^```(?:json)?|```$", "", r.strip(), flags=re.M).strip()) return True except Exception: return False if iid == "detectable_format:title": return bool(re.search(r"<<[^>]+>>", r)) if iid == "detectable_format:constrained_response": return any(o in r for o in ("My answer is yes.", "My answer is no.", "My answer is maybe.")) if iid == "detectable_format:number_highlighted_sections": n = kw.get("num_highlights") if n is None: return None return len(re.findall(r"\*[^*\n]+\*", r)) >= n if iid == "detectable_format:multiple_sections": spl, n = kw.get("section_spliter"), kw.get("num_sections") if not spl or n is None: return None return len(re.findall(re.escape(str(spl)), r)) >= n if iid == "change_case:english_lowercase": return r == r.lower() if iid == "change_case:english_capital": return r == r.upper() if iid == "change_case:capital_word_frequency": n, rel = kw.get("capital_frequency"), kw.get("capital_relation", "at least") if n is None: return None c = len([w for w in _words(r) if w.isupper() and len(w) > 1]) return c >= n if rel == "at least" else c <= n if iid == "startend:end_checker": end = str(kw.get("end_phrase", "")).strip() return bool(end) and r.strip().endswith(end) if iid == "startend:quotation": s = r.strip() return s.startswith('"') and s.endswith('"') if iid == "punctuation:no_comma": return "," not in r and "，" not in r if iid == "detectable_content:number_placeholders": n = kw.get("num_placeholders") if n is None: return None return len(re.findall(r"\[[^\]]+\]", r)) >= n if iid == "detectable_content:postscript": marker = str(kw.get("postscript_marker", "P.S.")) return marker.lower() in low if iid == "combination:two_responses": return "******" in r if iid == "combination:repeat_prompt": pr = str(kw.get("prompt_to_repeat", "")).strip() return bool(pr) and r.strip().lower().startswith(pr.lower()[:60]) if iid == "language:response_language": lang = str(kw.get("language", "")).lower() if lang in ("zh", "chinese"): return bool(re.search(r"[\u4e00-\u9fff]", r)) if lang in ("en", "english"): return not re.search(r"[\u4e00-\u9fff]", r) return None return None def score_ifeval(text, gold): """返回 (strict_ok, checked, passed)。checked=0 表示全部未实现。""" ids = gold.get("ids") or [] kwargs_list = gold.get("kwargs") or [] results = [] for i, iid in enumerate(ids): kw = kwargs_list[i] if i < len(kwargs_list) else {} if isinstance(kw, str): try: kw = json.loads(kw) except Exception: kw = {} res = _check_one_instruction(iid, kw, text) if res is not None: results.append(bool(res)) if not results: return None, 0, 0 return all(results), len(results), sum(results) # ---------------------------------------------------------------------- 调用 def call(session, args, item, spec): payload = { "model": args.model, "messages": [{"role": "user", "content": item["prompt"]}], "temperature": args.temperature, "top_p": args.top_p, "max_tokens": args.max_tokens_override or spec["max_tokens"], "seed": args.seed, } if args.thinking: payload["chat_template_kwargs"] = {"enable_thinking": True} else: payload["chat_template_kwargs"] = {"enable_thinking": False} last = None for attempt in range(args.retries): try: r = session.post(f"{args.base_url.rstrip('/')}/v1/chat/completions", json=payload, timeout=args.timeout) if r.status_code == 400 and "chat_template_kwargs" in payload: payload.pop("chat_template_kwargs") continue r.raise_for_status() d = r.json() msg = d["choices"][0]["message"] return (msg.get("content") or "", msg.get("reasoning_content") or "", d.get("usage", {}), None) except Exception as exc: # noqa: BLE001 last = f"{type(exc).__name__}: {str(exc)[:160]}" time.sleep(min(2.0 * (attempt + 1), 8)) return "", "", {}, last def _pct(xs, q): """分位数。None 一律剔除;空集返回 None,不返回 0 —— 0 会被误读成「真的是 0」。""" v = sorted(x for x in xs if x is not None) if not v: return None i = min(len(v) - 1, int(round((q / 100.0) * (len(v) - 1)))) return v[i] def run_task(args, task): spec = TASK_SPEC[task] n = args.n_override or spec["n"] items = load_items(task, n, args.seed) session = requests.Session() session.headers.update({"Connection": "keep-alive"}) results = [None] * len(items) ifeval_acc = {"checked": 0, "passed": 0} def work(k): item = items[k] content, reasoning, usage, err = call(session, args, item, spec) kind = item["kind"] detail = {} if kind == "mc": ok = score_mc(content, item["gold"], item.get("hi")) elif kind == "num": ok = score_num(content, item["gold"]) elif kind == "exact": ok = score_exact(content, item["gold"]) elif kind == "code": ok = score_code(content, item["gold"]) elif kind == "ifeval": strict, checked, passed = score_ifeval(content, item["gold"]) ok = bool(strict) if strict is not None else False detail = {"checked": checked, "passed": passed, "skipped": strict is None} ifeval_acc["checked"] += checked ifeval_acc["passed"] += passed else: ok = False ctoks = usage.get("completion_tokens") results[k] = { "id": item["id"], "ok": bool(ok), "error": err, "gold": item["gold"] if kind != "code" else item["gold"]["entry_point"], "completion_tokens": ctoks, "reasoning_chars": len(reasoning), "empty": not bool(content.strip()), "finish_reason": usage.get("finish_reason"), "content_mode": usage.get("content_mode"), "length_limited": usage.get("finish_reason")=="length", # 打满 max_tokens 且正文为空 => 思考未收尾被截断,答案根本没写出来 "truncated": bool(ctoks and ctoks >= (args.max_tokens_override or spec["max_tokens"]) - 8 and not (content or "").strip()), "reply_tail": (content or "").strip()[-200:], **detail, } t0 = time.time() with ThreadPoolExecutor(max_workers=args.concurrency) as pool: list(pool.map(work, range(len(items)))) elapsed = time.time() - t0 valid = [r for r in results if r] hits = sum(1 for r in valid if r["ok"]) errs = sum(1 for r in valid if r["error"]) trunc = sum(1 for r in valid if r.get("truncated")) gen_tokens = sum(r["completion_tokens"] or 0 for r in valid) scored = [r for r in valid if not r.get("truncated")] out = { "task": task, "n": len(valid), "hits": hits, "accuracy_pct": round(100.0 * hits / len(valid), 2) if valid else 0.0, # 排除思考截断题后的准确率(截断是采样预算问题,不是能力问题) "accuracy_ex_trunc_pct": (round(100.0 * sum(1 for r in scored if r["ok"]) / len(scored), 2) if scored else None), "truncated": trunc, # 2026-08-22 加:长度分布。预算给足之后,「答多长」本身就是要报的数字 —— # 它是消融副作用最直观的证据(消融版思考更短这件事,原先只能从截断数反推)。 "completion_tokens_p50": _pct([r["completion_tokens"] for r in valid], 50), "completion_tokens_p90": _pct([r["completion_tokens"] for r in valid], 90), "completion_tokens_max": max([r["completion_tokens"] or 0 for r in valid], default=0), "reasoning_chars_p50": _pct([r["reasoning_chars"] for r in valid], 50), "reasoning_chars_p90": _pct([r["reasoning_chars"] for r in valid], 90), "reasoning_leaked": sum(r["reasoning_chars"]>0 for r in valid) if not args.thinking else None, "empty": sum(r.get("empty",False) for r in valid), "length_limited": sum(r.get("length_limited",False) for r in valid), "errors": errs, "elapsed_s": round(elapsed, 1), "gen_tokens": gen_tokens, "tok_per_s": round(gen_tokens / elapsed, 1) if elapsed else 0, "details": valid, } if task == "ifeval": out["instruction_level_pct"] = ( round(100.0 * ifeval_acc["passed"] / ifeval_acc["checked"], 2) if ifeval_acc["checked"] else None) out["instructions_checked"] = ifeval_acc["checked"] return out def main(): p = argparse.ArgumentParser() p.add_argument("--base-url", required=True) p.add_argument("--model", required=True) p.add_argument("--label", required=True) p.add_argument("--tasks", default="all") p.add_argument("--seed", type=int, default=20260816) p.add_argument("--concurrency", type=int, default=16) p.add_argument("--timeout", type=int, default=900) p.add_argument("--retries", type=int, default=3) p.add_argument("--temperature", type=float, default=0.0) p.add_argument("--top-p", type=float, default=1.0) p.add_argument("--n-override", type=int, default=0) # 2026-08-22 加:全局覆盖每任务的 max_tokens。 # 起因是那轮固定预算(4096/8192)下截断率高达 34-91 条,Δ 的符号几乎完全跟着截断差走 —— # 消融版思考更短、被截断更少,全量分因此更高,那不是变聪明是话变少(交接 §13)。 # 创始人 2026-08-22 定:要么关思考与别的模型同口径,要么开思考但把预算给足到全部答完。 p.add_argument("--max-tokens", dest="max_tokens_override", type=int, default=0, help="覆盖所有任务的 max_tokens;0=用 TASK_SPEC 的默认值") p.add_argument("--no-thinking", dest="thinking", action="store_false") p.add_argument("--outdir", default=os.path.expanduser("~/capeval")) args = p.parse_args() os.environ.setdefault("HF_HUB_OFFLINE", "1") os.makedirs(args.outdir, exist_ok=True) tasks = ALL_TASKS if args.tasks == "all" else [t.strip() for t in args.tasks.split(",")] summary = { "label": args.label, "model": args.model, "base_url": args.base_url, "seed": args.seed, "thinking": args.thinking, "started": time.strftime("%F %T"), "tasks": {}, } for task in tasks: ckpt = os.path.join(args.outdir, f"{args.label}-{task}.json") if os.path.exists(ckpt): print(f"[{task}] 已有 checkpoint,跳过", flush=True) with open(ckpt, encoding="utf-8") as fh: res = json.load(fh) else: print(f"--- {task} 开始 ---", flush=True) try: res = run_task(args, task) except Exception as exc: # noqa: BLE001 print(f"[{task}] 整体失败: {type(exc).__name__}: {exc}", flush=True) continue # 2026-08-22 加的护栏:错误率过高就**不写 checkpoint**。 # 起因:kill 掉 vLLM 后 harness 仍在跑,150 条请求全部失败, # 而它把 acc=0.0 errors=150 当成正常结果写了盘 —— # 重启后「已有 checkpoint 就跳过」,那个 0.0% 会一直当成真结果留在表里, # 长得和正常结果一模一样。**失败必须失败掉,不能变成一个像样的产物。** err_rate = res["errors"] / max(res["n"], 1) if err_rate > 0.1: print(f"[{task}] ⚠️ 错误率 {err_rate:.0%}（{res['errors']}/{res['n']}）" f"过高，**不写 checkpoint**，本项需重跑", flush=True) continue with open(ckpt, "w", encoding="utf-8") as fh: json.dump(res, fh, ensure_ascii=False, indent=2) extra = "" if res.get("instruction_level_pct") is not None: extra = f" 指令级={res['instruction_level_pct']}%" if res.get("truncated"): extra += (f" 截断={res['truncated']}" f"(排除后={res.get('accuracy_ex_trunc_pct')}%)") if res.get("completion_tokens_p50") is not None: extra += (f" 输出token p50/p90/max=" f"{res['completion_tokens_p50']}/{res['completion_tokens_p90']}" f"/{res['completion_tokens_max']}") print(f"[{task}] {res['hits']}/{res['n']} = {res['accuracy_pct']}% " f"errors={res['errors']} {res['elapsed_s']}s " f"{res['tok_per_s']}tok/s{extra}", flush=True) summary["tasks"][task] = {k: v for k, v in res.items() if k != "details"} summary["finished"] = time.strftime("%F %T") spath = os.path.join(args.outdir, f"SUMMARY-{args.label}.json") with open(spath, "w", encoding="utf-8") as fh: json.dump(summary, fh, ensure_ascii=False, indent=2) print(f"\n==== SUMMARY {args.label} ====") for t, r in summary["tasks"].items(): print(f"{t:>11}: {r['accuracy_pct']:6.2f}% ({r['hits']}/{r['n']}, err={r['errors']})") print("summary:", spath) return 0 if __name__ == "__main__": sys.exit(main())