大模型評測與基準測試

import asyncio
import json
import os
import re
import time

from openai import AsyncOpenAI, BadRequestError

client = AsyncOpenAI(base_url="https://www.tutujin.com/v1", api_key=os.environ["TUTUJIN_API_KEY"], max_retries=4)
MODELS = ["gpt-5.5", "deepseek-v4.1-flash"]
JUDGE = "claude-sonnet-5-5"  # from a vendor that is not under test
REPEATS = 3
SYSTEM = "Answer the question. End with a last line of the form 'Answer: <answer>'."
limit = asyncio.Semaphore(8)  # requests in flight; raise it slowly while watching for 429


async def ask(model, messages, params):
    async with limit:
        started = time.time()
        try:
            resp = await client.chat.completions.create(model=model, messages=messages, **params)
        except BadRequestError:
            # many reasoning models reject temperature: retry without it and keep what was really sent
            params = {k: v for k, v in params.items() if k != "temperature"}
            resp = await client.chat.completions.create(model=model, messages=messages, **params)
        return resp, params, round(time.time() - started, 2)


async def judge(item, answer):
    """Open-ended answers: 0-5 against a written rubric, without the name of the model that wrote them."""
    prompt = (
        f"Question:\n{item['prompt']}\n\nReference answer:\n{item['reference']}\n\nRubric:\n{item['rubric']}\n\n"
        f"Answer to score:\n{answer}\n\n"
        'Score the answer from 0 to 5 by the rubric only. Reply with JSON only: {"score": 0, "reason": ""}'
    )
    resp, _, _ = await ask(JUDGE, [{"role": "user", "content": prompt}], {"temperature": 0})
    raw = resp.choices[0].message.content
    return json.loads(raw[raw.find("{"): raw.rfind("}") + 1])["score"] / 5


async def run_one(model, item, rep):
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": item["prompt"]}]
    resp, params, latency = await ask(model, messages, {"temperature": 0, "seed": 7})
    text = resp.choices[0].message.content or ""
    if "answer" in item:  # questions with a key are scored by rule
        m = re.search(r"Answer:\s*(.+?)\s*$", text)
        score = float(bool(m) and m.group(1).strip(" *.").lower() == item["answer"].lower())
    else:
        score = await judge(item, text)
    return {
        "item": item["id"], "repeat": rep, "model": model, "returned_model": resp.model, "params": params,
        "score": score, "scored_by": "rule" if "answer" in item else JUDGE, "latency_s": latency,
        "finish_reason": resp.choices[0].finish_reason, "usage": resp.usage.model_dump() if resp.usage else None,
        "called_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "output": text,
    }


async def main(path="testset.jsonl"):
    items = [json.loads(line) for line in open(path, encoding="utf-8")]
    jobs = [(m, it, r) for m in MODELS for it in items for r in range(REPEATS)]
    results = await asyncio.gather(*(run_one(*j) for j in jobs), return_exceptions=True)
    with open("results.jsonl", "a", encoding="utf-8") as f:
        for (m, it, r), res in zip(jobs, results):
            row = res if isinstance(res, dict) else {"item": it["id"], "repeat": r, "model": m, "error": repr(res)}
            f.write(json.dumps(row, ensure_ascii=False) + "\n")
    for m in MODELS:
        ok = [x for x in results if isinstance(x, dict) and x["model"] == m]
        failed = sum(1 for (jm, _, _), x in zip(jobs, results) if jm == m and not isinstance(x, dict))
        print(m, "mean score", round(sum(x["score"] for x in ok) / max(len(ok), 1), 3),
              "| mean latency", round(sum(x["latency_s"] for x in ok) / max(len(ok), 1), 2), "s",
              "| failed, rerun these:", failed)


asyncio.run(main())