只有两天的模型内测：如何评估 DeepSeek V4.1 Flash 临时版本

import json
import os
import time
import urllib.error
import urllib.request
from pathlib import Path

API_BASE = os.environ["API_BASE"].rstrip("/")
API_KEY = os.environ["API_KEY"]
MODELS = [m.strip() for m in os.environ["MODELS"].split(",") if m.strip()]
OUTPUT = Path(os.getenv("OUTPUT", "model-results.jsonl"))

CASES = [
    {
        "id": "json-extraction",
        "prompt": (
            "从下面句子提取订单号和金额，只输出合法 JSON，"
            "字段为 order_id 和 amount：订单 A-1042 已支付 318.50 元。"
        ),
    },
    {
        "id": "code-generation",
        "prompt": (
            "编写一个 Python 函数 deduplicate(items)，保持原顺序去重。"
            "只使用标准库，并给出两个 assert 测试。"
        ),
    },
    {
        "id": "instruction-following",
        "prompt": "用恰好三条项目符号解释什么是幂等 API，每条不超过 25 个汉字。",
    },
]


def call_model(model, prompt):
    body = json.dumps(
        {
            "model": model,
            "messages": [
                {"role": "system", "content": "准确执行任务，不要补充无关内容。"},
                {"role": "user", "content": prompt},
            ],
            "temperature": 0,
        }
    ).encode("utf-8")

    request = urllib.request.Request(
        f"{API_BASE}/chat/completions",
        data=body,
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        method="POST",
    )

    started = time.perf_counter()
    try:
        with urllib.request.urlopen(request, timeout=90) as response:
            payload = json.loads(response.read().decode("utf-8"))
        return {
            "ok": True,
            "latency_ms": round((time.perf_counter() - started) * 1000, 1),
            "answer": payload["choices"][0]["message"]["content"],
            "usage": payload.get("usage"),
        }
    except urllib.error.HTTPError as exc:
        return {
            "ok": False,
            "latency_ms": round((time.perf_counter() - started) * 1000, 1),
            "status": exc.code,
            "error": exc.read().decode("utf-8", errors="replace"),
        }
    except Exception as exc:
        return {
            "ok": False,
            "latency_ms": round((time.perf_counter() - started) * 1000, 1),
            "error": repr(exc),
        }


with OUTPUT.open("w", encoding="utf-8") as output_file:
    for case in CASES:
        for model in MODELS:
            result = {
                "case_id": case["id"],
                "model": model,
                "prompt": case["prompt"],
                **call_model(model, case["prompt"]),
            }
            output_file.write(json.dumps(result, ensure_ascii=False) + "\n")
            print(
                f'{case["id"]:24} {model:48} '
                f'ok={result["ok"]} latency_ms={result["latency_ms"]}'
            )

print(f"Results written to {OUTPUT}")

export API_BASE="https://your-api.example.com/v1"
export API_KEY="replace-with-your-key"
export MODELS="deepseek-v4.1-flash-expires-on-0910,your-stable-model-id"
python benchmark.py

import os

PRIMARY_MODEL = os.getenv("PRIMARY_MODEL", "your-stable-model-id")
FALLBACK_MODEL = os.getenv("FALLBACK_MODEL", "your-stable-model-id")


def generate(client, messages):
    try:
        return client.chat(model=PRIMARY_MODEL, messages=messages, timeout=30)
    except client.RetryableError:
        if PRIMARY_MODEL == FALLBACK_MODEL:
            raise
        return client.chat(model=FALLBACK_MODEL, messages=messages, timeout=30)