gpt-oss 模型在 Azure A10 和单卡 H100 机型上的性能测评

本月初，我们发布了 gpt-oss 两款模型正式登陆 Azure AI Foundry （国际版），详细发布细节请参考 👉OpenAI 开源模型 gpt-oss 重磅登陆 Foundry 平台

gpt-oss-120b 在核心推理基准测试中表现与OpenAI o4-mini 几乎持平，并且可在单个 80 GB GPU (NC_H100单卡机型)上高效运行。

gpt-oss-20b 在常见基准测试的结果与 OpenAI o3-mini 相近，并且可以在仅有 16GB 内存的边缘设备上运行，非常适合本地推理、端侧部署或低成本的快速迭代场景。

这两款模型在工具调用、少样本函数调用（few-shot function calling）、CoT 推理（如 Tau-Bench 智能体评估套件上的测试）和 HealthBench （甚至优于 OpenAI o1 和 GPT 4o 等专有模型）方面也表现出色。

在这篇文章中，我将展示两款模型在 A10 和 H100 两款 GPU 上的性能，包括TTFT，每秒 tokens 数等。

假设我们有 8 个数（实际 MX 是 32，为了演示用小样本）：

[1.0, 0.9, 1.1, 0.95, 1.05, 1.0, 0.92, 100.0]

传统 INT4

• 找最大值 = 100

• INT4 最大整数 = 15 → scale ≈ 100 / 15 = 6.67

• 小数：1.0 / 6.67 ≈ 0.15 → 量化成 0 → 反量化回 0

• 结果：

[0, 0, 0, 0, 0, 0, 0, 100]

小值全没了。

MXFP4

• 元素最大可以表示 6.0

• X = 2^(floor(log2(100 / 6.0))) ≈ 16

• 除以 X 后：

[0.0625, ..., 6.25]

• 量化

P_i ≈ [0.0625, ..., 6.0 (saturated)]

• 反量化

[1.0, 1.0, ..., 96.0]

小值全部保留大致精度，只有大值轻微截断。

[ INT4 (无符号示例) ]                 [ E2M1 (FP4) ]
┌───┬───┬───┬───┐                   ┌───┬───┬───┬───┐│b3 │b2 │b1 │b0 │  ← 4 bits         │ S │ E1│ E0│ M0│  ← 4 bits└───┴───┴───┴───┘                   └───┴───┴───┴───┘b3..b0 → 整数值 (0~15)               S: 符号位 (0=正, 1=负)                                     E1E0: 2-bit 指数（Exponent）                                     M0: 1-bit 尾数（Mantissa）

在 gpt-oss 中，Sink Token 机制主要用于提升长上下文场景下的推理吞吐量与延迟表现，同时通过防止上下文丢失来保持准确性——需要注意的是，它并不会直接提升模型的推理能力。

• 通过将昂贵的上下文重扫描（re-scan）转化为更具成本效益的“全局内存查找”。Sink Token 在超长上下文（≥ 32k tokens）中，显著降低 TTFT（Time To First Token） 和整体吞吐量压力。

  
  

• FA3 不只是优化方案，在 vLLM 中，若想要高效运行 Sink Token，FA3 是必备条件。

  
  

• 在 Ampere GPU（A10 / A100）上运行

• 如果不需要 Sink Token 的长上下文加速 → 建议使用 Ollama（MXFP4）。

• 或者在 vLLM 中禁用 Sink Token，以避免 FA3 兼容性问题。

  
  

• 在Hopper GPU（H100 / L40S）上运行，必选 vLLM + FA3 + Sink Token，从而确保性能最大化。

ollama run gpt-oss:20b

 TTFT < 1s Throughput: 45~55 tokens/s

import requests, time, json
MODEL = "gpt-oss:20b"PROMPT = "Give me a 2000-word introduction to Ollama."
url = "http://localhost:11434/api/generate"payload = {"model": MODEL, "prompt": PROMPT, "stream": True}
t0 = time.time()first_token_time = Nonetoken_count = 0
with requests.post(url, json=payload, stream=True, timeout=600) as resp:    for line in resp.iter_lines():        ifnot line:            continue        data = json.loads(line)        if data.get("done"):            break        chunk = data.get("response", "")        ifnot chunk:            continue        if first_token_time is None:            first_token_time = time.time()        token_count += len(chunk.split())   # 简化统计token，可用tiktoken更精确
t1 = time.time()ttft = first_token_time - t0throughput = token_count/(t1-first_token_time) if first_token_time else0print(f"TTFT: {ttft:.3f}s, Tokens: {token_count}, Throughput: {throughput:.2f} tokens/s")

vllm serve openai/gpt-oss-20b

#!/usr/bin/env python3# stress_test.py"""Asynchronously stress-test a local vLLM OpenAI-compatible endpoint.Prerequisites:  pip install "httpx[http2]" tqdm orjsonAuthor: 2025-08-06"""import argparse, asyncio, time, statistics, osimport orjson, httpxfrom tqdm.asyncio import tqdm_asyncio as tqdm   # tqdm ≥ 4.66ENDPOINT = "http://127.0.0.1:8000/v1/chat/completions"HEADERS  = {"Content-Type": "application/json"}SYSTEM   = "You are a helpful assistant."def build_payload(prompt: str, max_tokens: int = 128, temp: float = 0.0):    return {        "model": "openai/gpt-oss-20b",      # 任意字符串也行，只要跟 serve 时一致        "messages": [            {"role": "system", "content": SYSTEM},            {"role": "user",   "content": prompt},        ],        "temperature": temp,        "max_tokens": max_tokens,        "stream": False                    # 如需压 TTFT 可设 True，但统计更复杂    }async def worker(    client: httpx.AsyncClient,    payload: dict,    latencies: list,    ttfts: list,    tokens: list,):    """Send a single request and record metrics."""    t0 = time.perf_counter()    resp = await client.post(ENDPOINT, headers=HEADERS, content=orjson.dumps(payload))    t1 = time.perf_counter()    if resp.status_code != 200:        raise RuntimeError(f"HTTP {resp.status_code}: {resp.text[:200]}")    out = resp.json()    # usage 字段遵循 OpenAI 规范    usage  = out.get("usage", {})    c_tok  = usage.get("completion_tokens", 0)    ttft   = out.get("ttft", 0)            # vLLM 0.10 起返回，若无自行估算    ifnot ttft:        # 估算：总时长*(prompt_tokens/全部tokens)   粗略近似        p_tok = usage.get("prompt_tokens", 1)        ttft  = (p_tok / (p_tok + c_tok + 1e-6)) * (t1 - t0)    latencies.append(t1 - t0)    ttfts.append(ttft)    tokens.append(c_tok)async def run(concurrency: int, total_requests: int, payload: dict):    latencies, ttfts, tokens = [], [], []    limits = httpx.Limits(max_connections=concurrency)    timeout = httpx.Timeout(60.0)          # 适当加大    async with httpx.AsyncClient(limits=limits, timeout=timeout, http2=True) as client:        sem = asyncio.Semaphore(concurrency)        async def _task(_):            async with sem:                await worker(client, payload, latencies, ttfts, tokens)        await tqdm.gather(*[_task(i) for i in range(total_requests)])    return latencies, ttfts, tokensdef main():    ap = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)    ap.add_argument("--concurrency", "-c", type=int, default=64,                    help="number of concurrent requests")    ap.add_argument("--requests", "-n", type=int, default=1024,                    help="total requests to send")    ap.add_argument("--prompt", type=str, default="Explain quantum mechanics in one sentence.",                    help="user prompt")    ap.add_argument("--max-tokens", type=int, default=128)    args = ap.parse_args()    payload = build_payload(args.prompt, max_tokens=args.max_tokens)    print(f"Start stress test: {args.requests} requests  | "          f"concurrency={args.concurrency} | max_tokens={args.max_tokens}")    st = time.perf_counter()    lat, ttft, toks = asyncio.run(run(args.concurrency, args.requests, payload))    et = time.perf_counter()    total_time = et - st    # ── Stats ────────────────────────────────────────────────────────────────    def pct(lst, p): return statistics.quantiles(lst, n=100)[p-1]    print("\nRESULTS")    print(f"Total wall-clock time : {total_time:8.2f}  s")    print(f"Requests / second     : {args.requests / total_time:8.1f}  req/s")    print(f"Tokens  / second      : {sum(toks) / total_time:8.1f}  tok/s")    for name, arr in [("Latency (s)", lat), ("TTFT (s)", ttft)]:        print(f"{name:<15} p50={statistics.median(arr):.3f} "              f"p90={pct(arr,90):.3f}  p99={pct(arr,99):.3f}")    print("\nDone.")if __name__ == "__main__":    main()(oss-20b-tgi) root@h100vm:~# (oss-20b-tgi) root@h100vm:~# (oss-20b-tgi) root@h100vm:~# (oss-20b-tgi) root@h100vm:~# (oss-20b-tgi) root@h100vm:~# cat stress_test.py #!/usr/bin/env python3# stress_test.py"""Asynchronously stress-test a local vLLM OpenAI-compatible endpoint.Prerequisites:  pip install "httpx[http2]" tqdm orjsonAuthor: 2025-08-06"""import argparse, asyncio, time, statistics, osimport orjson, httpxfrom tqdm.asyncio import tqdm_asyncio as tqdm   # tqdm ≥ 4.66ENDPOINT = "http://127.0.0.1:8000/v1/chat/completions"HEADERS  = {"Content-Type": "application/json"}SYSTEM   = "You are a helpful assistant."def build_payload(prompt: str, max_tokens: int = 128, temp: float = 0.0):    return {        "model": "openai/gpt-oss-20b",      # 任意字符串也行，只要跟 serve 时一致        "messages": [            {"role": "system", "content": SYSTEM},            {"role": "user",   "content": prompt},        ],        "temperature": temp,        "max_tokens": max_tokens,        "stream": False                    # 如需压 TTFT 可设 True，但统计更复杂    }async def worker(    client: httpx.AsyncClient,    payload: dict,    latencies: list,    ttfts: list,    tokens: list,):    """Send a single request and record metrics."""    t0 = time.perf_counter()    resp = await client.post(ENDPOINT, headers=HEADERS, content=orjson.dumps(payload))    t1 = time.perf_counter()    if resp.status_code != 200:        raise RuntimeError(f"HTTP {resp.status_code}: {resp.text[:200]}")    out = resp.json()    # usage 字段遵循 OpenAI 规范    usage  = out.get("usage", {})    c_tok  = usage.get("completion_tokens", 0)    ttft   = out.get("ttft", 0)            # vLLM 0.10 起返回，若无自行估算    ifnot ttft:        # 估算：总时长*(prompt_tokens/全部tokens)   粗略近似        p_tok = usage.get("prompt_tokens", 1)        ttft  = (p_tok / (p_tok + c_tok + 1e-6)) * (t1 - t0)    latencies.append(t1 - t0)    ttfts.append(ttft)    tokens.append(c_tok)async def run(concurrency: int, total_requests: int, payload: dict):    latencies, ttfts, tokens = [], [], []    limits = httpx.Limits(max_connections=concurrency)    timeout = httpx.Timeout(60.0)          # 适当加大    async with httpx.AsyncClient(limits=limits, timeout=timeout, http2=True) as client:        sem = asyncio.Semaphore(concurrency)        async def _task(_):            async with sem:                await worker(client, payload, latencies, ttfts, tokens)        await tqdm.gather(*[_task(i) for i in range(total_requests)])    return latencies, ttfts, tokensdef main():    ap = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)    ap.add_argument("--concurrency", "-c", type=int, default=64,                    help="number of concurrent requests")    ap.add_argument("--requests", "-n", type=int, default=1024,                    help="total requests to send")    ap.add_argument("--prompt", type=str, default="Explain quantum mechanics in one sentence.",                    help="user prompt")    ap.add_argument("--max-tokens", type=int, default=128)    args = ap.parse_args()    payload = build_payload(args.prompt, max_tokens=args.max_tokens)    print(f"Start stress test: {args.requests} requests  | "          f"concurrency={args.concurrency} | max_tokens={args.max_tokens}")    st = time.perf_counter()    lat, ttft, toks = asyncio.run(run(args.concurrency, args.requests, payload))    et = time.perf_counter()    total_time = et - st    # ── Stats ────────────────────────────────────────────────────────────────    def pct(lst, p): return statistics.quantiles(lst, n=100)[p-1]    print("\nRESULTS")    print(f"Total wall-clock time : {total_time:8.2f}  s")    print(f"Requests / second     : {args.requests / total_time:8.1f}  req/s")    print(f"Tokens  / second      : {sum(toks) / total_time:8.1f}  tok/s")    for name, arr in [("Latency (s)", lat), ("TTFT (s)", ttft)]:        print(f"{name:<15} p50={statistics.median(arr):.3f} "              f"p90={pct(arr,90):.3f}  p99={pct(arr,99):.3f}")    print("\nDone.")if __name__ == "__main__":    main()

x (oss-20b-tgi) root@h100vm:~# python stress_test.py --concurrency 256 --requests 2000     --prompt "Explain quantum mechanics in one paragraph."   --max-tokens 256

(oss-20b-tgi) root@h100vm:~# python stress_test.py --concurrency 256 --requests 2000     --prompt "Explain quantum mechanics in one paragraph."   --max-tokens 256

Start stress test: 2000 requests  | concurrency=256 | max_tokens=256100%|████████████████████████████████████████████████████████████████████████████████████████████████████████| 2000/2000 [01:06<00:00, 29.92it/s]
RESULTSTotal wall-clock time :    66.89  sRequests / second     :     29.9  req/sTokens  / second      :   7645.2  tok/sLatency (s)     p50=8.835 p90=11.235  p99=14.755TTFT (s)        p50=2.271 p90=2.874  p99=3.775
Done.

(gpt-oss) root@h100vm:~# vllm serve openai/gpt-oss-120b

(gpt-oss) root@h100vm:~# python stress_test-120b.py --concurrency 256 --requests 2000     --prompt "Explain quantum mechanics in one paragraph."   --max-tokens 128

RESULTSTotal wall-clock time :    60.73  sRequests / second     :     32.9  req/sTokens  / second      :   4215.6  tok/sLatency(s)     p50=8.254 p90=10.479  p99=11.782TTFT (s)        p50=3.363 p90=4.269  p99=4.800
Done.

(gpt-oss) root@h100vm:~# python run_local_llm.py "Please write me a Python program that can run directly in the terminal. This program should be a Tetris game with a colorful interface, and allow the player to control the direction of the blocks, game screen should has a clear border, run without any error."

cat run_local_llm.py

#!/usr/bin/env python3"""简易命令行调用本地 vLLM (OpenAI 兼容) 的脚本用法:    python run_llm.py "你的 prompt ..."可选:    -m/--model      指定模型名称          (默认: 自动探测)    -u/--url        指定服务地址          (默认: http://127.0.0.1:8000)    -v/--verbose    显示完整 JSON 响应"""
import argparseimport sysimport requestsfrom openai import OpenAIfrom openai.types.chat import ChatCompletion
def list_models(base_url: str):    """调用 /v1/models 获取当前加载的模型列表"""    try:        resp = requests.get(f"{base_url.rstrip('/')}/models", timeout=3)        resp.raise_for_status()        data = resp.json()        return [m["id"] for m in data.get("data", [])]    except Exception:        return []
def main() -> None:    parser = argparse.ArgumentParser(description="Run prompt on local vLLM server")    parser.add_argument("prompt", nargs="+", help="提示词")    parser.add_argument("-m", "--model", help="模型名称 (默认: 自动探测)")    parser.add_argument("-u", "--url", default="http://127.0.0.1:8000",                        help="服务器地址(不带 /v1)，默认 http://127.0.0.1:8000")    parser.add_argument("-v", "--verbose", action="store_true",                        help="打印完整 JSON")
    args = parser.parse_args()    base_url = args.url.rstrip("/") + "/v1"
    # 如果用户没指定模型，就到 /v1/models 去探测    model_name = args.model    if model_name is None:        models = list_models(base_url)        ifnot models:            print("❌ 无法从 /v1/models 获取模型列表，请检查 vLLM 是否在运行。", file=sys.stderr)            sys.exit(1)        if len(models) > 1:            print("⚠️  服务器上有多个模型，请用 -m 指定；当前可用：", ", ".join(models), file=sys.stderr)            sys.exit(1)        model_name = models[0]
    prompt_text = " ".join(args.prompt)
    client = OpenAI(base_url=base_url, api_key="EMPTY")
    try:        resp: ChatCompletion = client.chat.completions.create(            model=model_name,            messages=[{"role": "user", "content": prompt_text}],            temperature=0.7        )    except Exception as e:        print("❌ 调用失败:", e, file=sys.stderr)        sys.exit(1)
    # 输出    if args.verbose:        print("=== 完整 JSON ===")        print(resp.model_dump_json(indent=2, ensure_ascii=False))        print("\n=== 模型回答 ===")
    print(resp.choices[0].message.content.strip())
if __name__ == "__main__":    main()