用 Python 给大模型跑分：dataclass、异步重试和代理配置一个不少

from dataclasses import dataclass
from typing import AsyncIterator
import httpx
import logging
import asyncio
import random

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("llm_bench")


@dataclass
class ModelConfig:
    name: str
    api_url: str
    api_key: str
    # 走亿牛云代理时填这里，海外模型 API 从国内机器直连经常卡在握手阶段
    proxy: str | None = None
    timeout: float = 30.0
    max_retries: int = 3


@dataclass
class BenchResult:
    model: str
    question_id: str
    answer: str
    latency_ms: int
    ok: bool = True
    error: str | None = None


async def call_model(
    client: httpx.AsyncClient,
    cfg: ModelConfig,
    question: str,
    qid: str,
) -> BenchResult:
    payload = {
        "model": cfg.name,
        "messages": [{"role": "user", "content": question}],
        "temperature": 0,
    }
    for attempt in range(cfg.max_retries):
        try:
            resp = await client.post(
                cfg.api_url,
                json=payload,
                headers={"Authorization": f"Bearer {cfg.api_key}"},
                timeout=cfg.timeout,
            )
            resp.raise_for_status()
            data = resp.json()
            text = data["choices"][0]["message"]["content"]
            return BenchResult(
                model=cfg.name,
                question_id=qid,
                answer=text,
                latency_ms=int(resp.elapsed.total_seconds() * 1000),
            )
        except Exception as e:
            wait = (2 ** attempt) + random.random()
            logger.warning("模型 %s 第 %d 次失败: %s，%.1fs 后重试", cfg.name, attempt + 1, e, wait)
            await asyncio.sleep(wait)
    return BenchResult(cfg.name, qid, "", 0, ok=False, error="retries exhausted")


async def run_bench(
    configs: list[ModelConfig],
    questions: list[tuple[str, str]],
) -> AsyncIterator[BenchResult]:
    limits = httpx.Limits(max_connections=8)
    for cfg in configs:
        async with httpx.AsyncClient(proxy=cfg.proxy, limits=limits) as client:
            for qid, q in questions:
                yield await call_model(client, cfg, q, qid)


models = [
    ModelConfig(
        name="deepseek-4.1",
        api_url="https://api.deepseek.com/v1/chat/completions",
        api_key="YOUR_KEY",
        proxy="http://user:pass@proxy.yiniu.cn:7788",
    ),
    ModelConfig(
        name="opus-5",
        api_url="https://api.anthropic.com/v1/messages",
        api_key="YOUR_KEY",
    ),
    ModelConfig(
        name="gpt-5.6",
        api_url="https://api.openai.com/v1/chat/completions",
        api_key="YOUR_KEY",
        proxy="http://user:pass@proxy.yiniu.cn:7788",
    ),
]

questions = [("q1", "用 Python 写一个快速排序，并解释时间复杂度。")]