参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

_PATTERNS = [
    r"(?:正确答案|答案|正确选项|应选|选择|选)\s*(?:是|为|:|：)?\s*[（(\[]?\s*([A-H])\s*[)）\]]?",
    r"(?:correct\s+answer|answer)\s*(?:is|:)?\s*[（(\[]?\s*([A-H])\b",
    r"\b([A-H])\s*[)）.、]\s*",
    r"[（(\[]\s*([A-H])\s*[)）\]]",
    r"^\s*[（(\[]?\s*([A-H])\s*[)）\]]?\s*[.。!！?？,，]*\s*$",
]

def parse_choice(text: str, valid: str = "ABCDEFGH") -> str | None:
    if not text:
        return None
    cleaned = re.sub(r"[*`_]", "", text.strip()).upper()
    for pattern in _COMPILED:
        m = pattern.search(cleaned)
        if m and m.group(1).upper() in valid:
            return m.group(1).upper()
    return None

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def accuracy(n_correct: int, n_total: int) -> float:
    return (n_correct / n_total) if n_total else float("nan")

def exact_match(prediction, golds) -> float:
    pred = normalize_text(prediction or "")
    if not pred:
        return 0.0
    pred_ns = pred.replace(" ", "")
    for g in golds:
        gold_ns = normalize_text(g)
        if pred == gold_ns or pred_ns == gold_ns.replace(" ", ""):
            return 1.0
    return 0.0

def f1_score(prediction, gold) -> float:
    pred_tokens = _tokenize(prediction or "")
    gold_tokens = _tokenize(gold)
    if not pred_tokens or not gold_tokens:
        return 0.0
    common, gold_count = 0, {}
    for t in gold_tokens:
        gold_count[t] = gold_count.get(t, 0) + 1
    for t in pred_tokens:
        if gold_count.get(t, 0) > 0:
            common += 1
            gold_count[t] -= 1
    if common == 0:
        return 0.0
    precision = common / len(pred_tokens)
    recall = common / len(gold_tokens)
    return 2 * precision * recall / (precision + recall)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

class C3Task(EvalTask):
    name = "c3"
    primary_metric = "accuracy"

    def parse_answer(self, raw: str) -> Optional[str]:
        return parse_choice(raw, valid="ABCD")

    def score(self, prediction, item) -> dict[str, float]:
        return {"accuracy": 1.0 if prediction == item.gold else 0.0}

    def _to_items(self, rows: list[dict]) -> list[EvalItem]:
        return [
            EvalItem(
                id=str(row["id"]),
                prompt=self.build_prompt(row),
                gold=row["answer"].strip().upper(),
                meta={"question": row["question"], "options": row["options"], "context": row.get("context", "")},
            )
            for row in rows
        ]

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def generate(self, prompts: list[str]) -> list[str]:
    self.calls += len(prompts)
    return [self.policy(p) for p in prompts]

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

async def evaluate_tasks(adapter, tasks, cfg) -> RunReport:
    run = RunReport(adapter=adapter.name, model=cfg.model, config=cfg.__dict__.copy())
    run.started_at = datetime.now().isoformat(timespec="seconds")
    t0 = time.time()
    cache = ResponseCache(cfg.cache_path) if cfg.use_cache else None
    try:
        for task in tasks:
            run.tasks.append(await evaluate_task(adapter, task, cfg, cache))
    finally:
        if cache:
            cache.close()
        aclose = getattr(adapter, "aclose", None)
        if aclose is not None:
            await aclose()
        else:
            adapter.close()
    run.duration_s = time.time() - t0
    run.finished_at = datetime.now().isoformat(timespec="seconds")
    return run

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

full_credit = all(v >= (1.0 - 1e-12) for v in scores.values())
if not full_credit:
    badcase = {
        "task": task.name,
        "id": item.id,
        "prompt": item.prompt,
        "gold": item.gold,
        "prediction": prediction,
        "raw_output": raw,
        "error_type": "api_error" if failed else task.error_type(raw, prediction, item),
        "scores": scores,
    }
    report.badcases.append(badcase)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def _generate_batch(self, prompts: list[str]) -> list[str]:
    chats = [
        self.tokenizer.apply_chat_template([{"role": "user", "content": p}],
                                           tokenize=False, add_generation_prompt=True)
        for p in prompts
    ]
    batch = self.tokenizer(chats, return_tensors="pt", padding=True, truncation=True, max_length=1024)
    batch = {k: v.to(self.model.device) for k, v in batch.items()}
    gen_kwargs = dict(max_new_tokens=self.max_new_tokens, pad_token_id=self.tokenizer.pad_token_id)
    if self.temperature and self.temperature > 0:
        gen_kwargs.update(do_sample=True, temperature=self.temperature)
    else:
        gen_kwargs.update(do_sample=False)
    with torch.no_grad():
        out = self.model.generate(**batch, **gen_kwargs)
    new_tokens = out[:, batch["input_ids"].shape[1]:]
    return self.tokenizer.batch_decode(new_tokens, skip_special_tokens=True)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

@staticmethod
def make_key(model: str, prompt: str, params: dict) -> str:
    payload = json.dumps({"model": model, "prompt": prompt, "params": params},
                         ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

def get(self, key: str) -> str | None:
    row = self.conn.execute("SELECT response FROM responses WHERE key = ?", (key,)).fetchone()
    if row is None:
        self.misses += 1
        return None
    self.hits += 1
    return row[0]

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

async def _one(self, client, prompt: str) -> str:
    try:
        resp = await client.post(self.base_url + "/v1/chat/completions",
                                 json=self._payload(prompt), headers=self._headers())
        if resp.status_code in (429, 500, 502, 503, 504):
            raise AdapterError(f"HTTP {resp.status_code}（可重试）")
        if resp.status_code in (400, 401):
            raise FatalAdapterError(f"HTTP {resp.status_code}（不可重试）")
        resp.raise_for_status()
        return resp.json()["choices"][0]["message"]["content"]
    except httpx.TimeoutException as e:
        raise AdapterError(f"请求超时：{e}") from e

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

async def _generate_missing(adapter, prompts, cfg):
    if not prompts:
        return []
    if getattr(adapter, "supports_concurrency", False):
        sem = asyncio.Semaphore(max(1, cfg.concurrency))
        async def one(prompt: str) -> str:
            async with sem:
                try:
                    return (await adapter.agenerate([prompt]))[0]
                except AdapterError as e:
                    logger.error("请求最终失败：%s", e)
                    return ""
        return list(await asyncio.gather(*(one(p) for p in prompts)))
    try:
        return await asyncio.to_thread(lambda: adapter.generate(prompts))
    except AdapterError as e:
        logger.error("批量生成最终失败：%s", e)
        return [""] * len(prompts)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

async def _async_with_retry(fn, cfg):
    last = None
    for attempt in range(cfg.retries + 1):
        try:
            return await fn()
        except FatalAdapterError:
            raise
        except AdapterError as e:
            last = e
            if attempt < cfg.retries:
                wait = cfg.retry_backoff_s * (2 ** attempt)
                logger.warning("调用失败（第 %d 次）：%s；%.1fs 后重试", attempt + 1, e, wait)
                await asyncio.sleep(wait)
    raise last

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def write_reports(run, cfg, out_dir: Path) -> dict[str, Path]:
    out_dir.mkdir(parents=True, exist_ok=True)
    results_path = out_dir / "results.json"
    results_path.write_text(json.dumps(run.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8")
    badcases_path = out_dir / "badcases.jsonl"
    with badcases_path.open("w", encoding="utf-8") as f:
        for t in run.tasks:
            for case in t.badcases:
                f.write(json.dumps(case, ensure_ascii=False) + "\n")
    ...  # summary.md 拼接（表头 + 每任务一行 + bad case 示例 + 复现命令）

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def build_adapter(cfg: EvalConfig) -> ModelAdapter:
    if cfg.adapter == "mock":
        return MockAdapter()
    if cfg.adapter == "hf":
        from llm_eval.adapters.hf import HuggingFaceAdapter
        return HuggingFaceAdapter(cfg.model, max_new_tokens=cfg.max_new_tokens,
                                  temperature=cfg.temperature, peft_adapter=cfg.peft_adapter or None)
    if cfg.adapter == "openai":
        from llm_eval.adapters.openai_compat import OpenAICompatibleAdapter
        return OpenAICompatibleAdapter(cfg.model, base_url=cfg.base_url, api_key=cfg.api_key,
                                       max_new_tokens=cfg.max_new_tokens, temperature=cfg.temperature)
    raise ValueError(f"未知 adapter：{cfg.adapter}")

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

完整实现：projects/llm-eval/（21 个测试函数 / 32 个用例 + 真实 C3/XCOPA 运行记录）。 对照重点：

1. runner.py 的「查缓存 → 生成缺失 → 回填 → 判分」主循环；
2. openai_compat.py 的状态码分类；
3. reports.py 的 summary.md 结构。