model-benchmark¶
第6章 · Agent 的评估 · 配套项目
chapter6/model-benchmark
项目说明¶
多维度模型性能基准测试（实验 6-8 配套代码）¶
对多个 OpenAI 兼容的 LLM API 提供商做横向基准测试，一条命令跑出
TTFT / 端到端延迟 / 吞吐 / 标准差 / p50 / p95 / p99 / 成功率 的多维度对比表，
为模型选型提供实测依据。还支持并发压测（逐档加压找限流点，看指标随并发的变化）
与离线自检（--mock 合成数据，无需 key/网络即可验证指标聚合）。
对应《深入理解 AI Agent》第 6 章 实验 6-8：多维度模型性能基准测试。
目的¶
书中实验 6-8 的完整版要求"一周内每小时探测、8K/32K/128K 上下文、 100+ 请求、MTTR/限流阈值/综合成本"等。本配套代码聚焦其中最核心、 可低成本本地复现的一环：用流式接口精确测量首 token 延迟， 在并发下测出延迟分位数与吞吐，并以成功率刻画可用性—— 让读者用几分钟、几分钱就能得到一张真实的多提供商对比表， 理解"选型是多维权衡而非单看排行榜"。
指标定义¶
| 指标 | 含义 | 怎么测的 |
|---|---|---|
| 成功率（可用性） | 成功请求数 / 总请求数 | 单次请求任何异常（超时/限流/网络错误/空响应）都计为失败，不中断整表 |
| TTFT | 首个 token 到达延迟 | 流式读取，记录第一个"有内容" chunk 到达的时刻 − 请求发出时刻 |
| 端到端延迟 | 请求发出到响应结束的总耗时 | 最后一个 chunk 时刻 − 请求发出时刻 |
| 吞吐（tokens/s） | 生成阶段的输出速度 | 输出 token 数 / (端到端 − TTFT)，剥离首 token 等待，反映纯解码速度 |
| p50 / p95 / p99 | 延迟的中位数 / 95 / 99 分位 | 对同一 (provider, model) 的多次成功请求排序后线性插值；p95、p99 高说明长尾重、体验不稳 |
| 标准差（std） | 延迟的离散程度 | 样本标准差；书中强调"高延迟方差意味着用户体验不稳定" |
| 聚合吞吐 / RPS | 整批的总吞吐 | 并发压测时：全部成功请求的输出 token 总数 / 整批墙钟耗时（RPS 为成功请求数 / 墙钟）；随并发上升先增后趋平，触及服务端上限即触顶 |
输出 token 数优先取服务端回传的精确
usage.completion_tokens； 若服务不返回 usage，则以流式 chunk 数近似计数（会略微偏高，已在代码注释标明）。
运行¶
cd chapter6/model-benchmark
pip install -r requirements.txt
# 配置 key：只需填手上有的，未设置的提供商会自动跳过
cp env.example .env # 然后编辑 .env
# 或直接 export OPENAI_API_KEY=... MOONSHOT_API_KEY=... ARK_API_KEY=...
python demo.py # 一条命令跑出对比表
常用参数：
python demo.py --list # 仅列出将测试的提供商
python demo.py --num-requests 20 --concurrency 5 # 加大样本与并发
python demo.py --serial # 串行发送（并发=1，看无竞争下的基线延迟）
python demo.py --max-tokens 256 # 生成更长响应，更充分地测吞吐
默认参数（N=10/家, 并发=3, max_tokens=64）单次全跑成本约几分钱。
要接近书中"每配置 ≥100 次请求"的统计口径，把 --num-requests 调到 100 即可
（注意成本与限流会同步上升）。
指定任意 OpenAI 兼容端点（不改代码测新模型/新提供商）¶
书中要求"对同一模型测试不同 API 提供商（如 DeepSeek 官方 vs SiliconFlow）"。
用 --base-url / --model / --api-key-env 即可直接指定单个端点，无需改 DEFAULT_PROVIDERS：
python demo.py --base-url https://api.deepseek.com --model deepseek-chat \
--api-key-env DEEPSEEK_API_KEY --name "DeepSeek官方/deepseek-chat"
# 换个 base_url、保持同一 model，即可对比"同模型不同提供商"
并发压测：逐步加压找限流点¶
书中实验 6-8 要求"通过逐步提升并发量来找到限流点，记录 RPM/TPM 上限"。
--concurrency-sweep 对同一模型逐档加压，产出一张随并发变化的指标表
（p50/p95/p99/std/成功率/RPS/聚合吞吐）：
随着并发上升，单请求延迟长尾（p95/p99/std）通常变差、可用性可能因限流下降， 而聚合吞吐（tokens/s）与 RPS 先升后趋平——趋平点即服务端的实际吞吐上限。
选择要显示的指标 / 导出结果¶
python demo.py --metrics ttft,throughput # 主表只看 TTFT 与吞吐（成功率始终显示）
python demo.py --output result.json # 完整结果（含 p50/p95/p99/std）写入 JSON
离线自检（--mock，无需 key/网络）¶
用合成（synthetic）数据跑通整条指标聚合链路，便于在没有 API key 或无网络时
验证 p50/p95/p99/std/可用性/聚合吞吐的计算是否正确。输出数字全部为伪随机合成，
带 [SYNTHETIC] 标注，绝非真实基准，切勿用于选型。
一次合成并发压测的输出（--mock --concurrency-sweep 1,2,4,8,16 --num-requests 100，
数字为合成，仅演示趋势）：
并发 | 成功率 | TTFT_p50 | TTFT_p95 | 端到端p50 | 端到端p95 | 端到端p99 | 端到端std | RPS | 聚合吞吐
-----+----------------+----------+----------+-----------+-----------+-----------+-----------+------+----------
1 | 99/100 (99%) | 301ms | 514ms | 0.73s | 1.04s | 1.16s | 0.13s | 1.3 | 49.8 t/s
2 | 100/100 (100%) | 335ms | 570ms | 0.79s | 1.07s | 1.19s | 0.15s | 2.5 | 94.4 t/s
4 | 98/100 (98%) | 381ms | 617ms | 0.83s | 1.11s | 1.19s | 0.16s | 4.7 | 180.0 t/s
8 | 92/100 (92%) | 523ms | 932ms | 0.96s | 1.53s | 1.67s | 0.25s | 8.0 | 305.3 t/s
16 | 97/100 (97%) | 878ms | 1487ms | 1.30s | 1.97s | 2.37s | 0.35s | 11.9 | 441.0 t/s
可见随并发上升：端到端 p95/p99 与 std 走高（长尾变差），聚合吞吐持续增长（尚未触顶）。 真实端点上这条曲线会在某个并发处趋平并伴随可用性下降——那就是限流点。
默认测试的提供商¶
代码里 DEFAULT_PROVIDERS 默认只跑手上有有效 key的提供商（OpenAI 一个 key 测多个模型）：
| 展示名 | 模型 | base_url | key 环境变量 |
|---|---|---|---|
| OpenAI/gpt-5.6-luna | gpt-5.6-luna | （官方默认，可回退 OpenRouter） | OPENAI_API_KEY |
| Moonshot/moonshot-v1-8k | moonshot-v1-8k | https://api.moonshot.cn/v1 | MOONSHOT_API_KEY |
| Doubao/doubao-1.5-pro-32k | doubao-1-5-pro-32k-250115 | https://ark.cn-beijing.volces.com/api/v3 | ARK_API_KEY |
OpenRouter 回退：
OpenAI/*这几条（base_url 为空的 OpenAI 原生条目）在未设置OPENAI_API_KEY时会自动改走 OpenRouter（OPENROUTER_API_KEY，模型名映射为openai/*）。gpt-5.x直连 OpenAI 需组织实名认证，因此只要设置了OPENROUTER_API_KEY就优先走 OpenRouter。带专属base_url的条目（Kimi/豆包）不参与回退。
提供商列表是可配置的：在 benchmark.py 的 DEFAULT_PROVIDERS 里追加
ProviderConfig(...) 即可扩展。所有提供商都走同一套 OpenAI 兼容协议，
只是 base_url 与 model 不同——这正是可以"同一模型对比不同提供商"
（如书中提到的 DeepSeek 官方 vs SiliconFlow）的原因。
真实运行结果（示例）¶
以下是一次真实运行的输出（python demo.py --num-requests 10 --concurrency 3，
测试机在中国大陆网络环境，2026-07）。数字为真实测得，非虚构；
不同网络/时段会有波动，请以自己跑出的结果为准。
Provider/Model | 成功率 | TTFT均值 | TTFT_p95 | 端到端均值 | 端到端p95 | 吞吐 | 输出tok
--------------------------+--------------+----------+----------+------------+-----------+-----------+--------
OpenAI/gpt-5.6-luna | 10/10 (100%) | 1360ms | 2334ms | 1.73s | 2.54s | 174.9 t/s | 26
Moonshot/moonshot-v1-8k | 10/10 (100%) | 530ms | 671ms | 0.89s | 1.07s | 92.1 t/s | 32
Doubao/doubao-1.5-pro-32k | 10/10 (100%) | 1097ms | 1409ms | 2.32s | 2.91s | 36.2 t/s | 44
结论（基于上面这次运行）¶
- 可用性：本次三家全部 10/10（100%）成功。可用性差异往往要在更大样本、 更高并发或更长时间窗口下才暴露——这正是书中强调"一周每小时探测"的原因。 代码已把单点失败设计成"记为可用性下降、不中断整表"，便于长时间挂机采样。
- 首 token 延迟（TTFT）：本测试机在国内网络下，Kimi 的 TTFT（~530ms）明显低于 跨境访问的 OpenAI/gpt-5.6-luna（~1.36s）；豆包 TTFT（~1.1s）略低于 OpenAI 但端到端更长。 TTFT 强依赖网络位置——同一份代码在美国机房跑，OpenAI 的 TTFT 会大幅下降。
- 吞吐：本次 gpt-5.6-luna（175 t/s）> Kimi（92 t/s）> 豆包（36 t/s）。 吞吐决定长响应的等待时间，与 TTFT 是两个独立维度。
- 稳定性（p95）：看 p95 与均值的差距。gpt-5.6-luna 跨境访问，TTFT p95(2.33s)/均值(1.36s) 拉开较大，长尾更重；Kimi 的 p95 与均值最接近，本次最稳。
- 选型启示：不存在"全面最优"的一家——延迟、吞吐、可用性、价格是多维权衡。 面向国内用户的实时交互场景，低 TTFT 的本地化服务体验更好； 批处理/长文本生成则更看重吞吐与单价。务必在你自己的部署网络环境下实测， 不要直接照搬第三方监测平台（如 Artificial Analysis）的数字。
文件说明¶
| 文件 | 作用 |
|---|---|
benchmark.py |
核心：提供商配置、单次流式测量、并发调度、指标聚合（含 p99/std/聚合吞吐）、并发扫描 sweep_concurrency、合成数据 synthetic_summary |
demo.py |
命令行入口：解析参数、跑测试（含并发压测 / --mock 离线自检）、打印对比表、导出 JSON |
requirements.txt |
依赖（openai SDK + 可选 python-dotenv） |
env.example |
key 配置模板 |
注意事项¶
- 成本控制：默认
max_tokens=64、N=10，全跑成本极低。调大参数前请留意计费。 - 限流：把并发或 N 调很大时可能触发提供商 RPM/TPM 限流，届时会以失败形式 计入可用性下降——这本身也是一种"实测限流阈值"的方式（书中实验 6-8 的一环）。
- TTFT 与网络强相关：跨境访问的服务 TTFT 会显著偏高，结论需结合部署地点解读。
- OpenRouter 回退：未设置
OPENAI_API_KEY时，OpenAI/*条目自动经 OpenRouter 路由 （需OPENROUTER_API_KEY，gpt-*映射为openai/*）；gpt-5.x只要有OPENROUTER_API_KEY即优先走 OpenRouter（直连需实名认证）。其它提供商（DEEPSEEK / SILICONFLOW 等）如需启用， 在DEFAULT_PROVIDERS中补充配置并设置对应环境变量即可。
源代码¶
benchmark.py¶
"""
多维度模型性能基准测试（实验 6-8 配套代码）
对多个 OpenAI 兼容的 LLM API 提供商，测量以下核心指标：
- TTFT（Time To First Token，首个 token 到达延迟）
- 端到端延迟（发出请求到接收完整响应）
- 吞吐（tokens/s，按生成的输出 token 计；并发下另给聚合吞吐 / RPS）
- 标准差 / p50 / p95 / p99 延迟分位数（方差大意味着体验不稳定）
- 可用性 / 成功率（失败即计入可用性下降，不中断整表）
支持两种模式：
- 单档位对比：多提供商横向对比表（默认）。
- 并发扫描（压测）：对同一模型逐步提升并发，观察延迟长尾与聚合吞吐随并发的变化。
实现要点：
- 使用 openai SDK 的流式接口（stream=True）来精确测量 TTFT。
- 通过 base_url 复用同一套 OpenAI 兼容协议，适配 Kimi / 豆包等国产 API。
- 单点请求失败被捕获并记录，不影响同一 (provider, model) 的其它请求，
也不影响其它 provider —— 这样一次运行就能测出"可用性"这一维度。
"""
from __future__ import annotations
import os
import time
import random
import statistics
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Optional
from openai import OpenAI
# ---------------------------------------------------------------------------
# OpenRouter 回退：对「OpenAI 原生」条目（base_url 为空）在缺主 key 时改走 OpenRouter。
# gpt-5.x 直连 OpenAI 需组织实名认证，只要有 OPENROUTER_API_KEY 就优先走 OpenRouter。
# ---------------------------------------------------------------------------
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
def _to_openrouter_model(model: str) -> str:
"""把模型名映射成 OpenRouter id：含 '/' 视为原生 id；gpt-* -> openai/*；
claude-* -> anthropic/claude-opus-4.8；其余回退到 openai/gpt-5.6-luna。"""
if "/" in model:
return model
if model.startswith("gpt-"):
return "openai/" + model
if model.startswith("claude-"):
return "anthropic/claude-opus-4.8"
return "openai/gpt-5.6-luna"
# ---------------------------------------------------------------------------
# 提供商配置
# ---------------------------------------------------------------------------
@dataclass
class ProviderConfig:
"""单个待测 (提供商, 模型) 配置。"""
name: str # 展示名，例如 "OpenAI/gpt-5.6-luna"
model: str # 传给 API 的模型名
api_key_env: str # 读取 API key 的环境变量名
base_url: Optional[str] = None # OpenAI 官方留空；其它填各自 base_url
def api_key(self) -> Optional[str]:
return os.environ.get(self.api_key_env)
def _openrouter_key(self) -> Optional[str]:
return os.environ.get("OPENROUTER_API_KEY", "").strip() or None
def resolve(self) -> tuple[Optional[str], Optional[str], str, bool]:
"""解析实际使用的 (api_key, base_url, model, 是否经 OpenRouter)。
仅「OpenAI 原生」条目（base_url 为空）参与回退；带专属 base_url 的条目
（如 Kimi/豆包）保持不变。回退规则：
- gpt-5.x 且有 OPENROUTER_API_KEY -> 优先走 OpenRouter（直连需实名认证）；
- 否则主 key 存在 -> 直连，模型名不变；
- 否则（OpenAI 原生 + 有 OPENROUTER_API_KEY）-> 走 OpenRouter，模型名映射。
"""
primary = self.api_key()
openai_native = self.base_url is None
orkey = self._openrouter_key() if openai_native else None
prefer_or = bool(orkey) and self.model.startswith("gpt-5")
if not prefer_or and primary:
return primary, self.base_url, self.model, False
if orkey:
return orkey, OPENROUTER_BASE_URL, _to_openrouter_model(self.model), True
return primary, self.base_url, self.model, False
def is_available(self) -> bool:
"""主 key 存在即可测；OpenAI 原生条目在缺主 key 时可回退 OpenRouter。"""
if self.api_key():
return True
return self.base_url is None and self._openrouter_key() is not None
# 默认只跑"手上有有效 key"的三家提供商。
# 需要扩展时，往这里追加 ProviderConfig 即可（例如 DeepSeek 官方 vs SiliconFlow 对比）。
DEFAULT_PROVIDERS: list[ProviderConfig] = [
# OpenAI 官方（一个 key 测多个模型，观察同厂不同规格的差异）
# gpt-5.6-luna 为当前廉价旗舰；无 OPENAI_API_KEY 时自动经 OpenRouter 路由
# （openai/gpt-5.6-luna），gpt-5.x 只要有 OPENROUTER_API_KEY 就优先走 OpenRouter。
ProviderConfig(
name="OpenAI/gpt-5.6-luna",
model="gpt-5.6-luna",
api_key_env="OPENAI_API_KEY",
),
# 月之暗面 Kimi（OpenAI 兼容）
ProviderConfig(
name="Moonshot/moonshot-v1-8k",
model="moonshot-v1-8k",
api_key_env="MOONSHOT_API_KEY",
base_url="https://api.moonshot.cn/v1",
),
# 字节豆包 / 火山方舟（OpenAI 兼容）
ProviderConfig(
name="Doubao/doubao-1.5-pro-32k",
model="doubao-1-5-pro-32k-250115",
api_key_env="ARK_API_KEY",
base_url="https://ark.cn-beijing.volces.com/api/v3",
),
]
# ---------------------------------------------------------------------------
# 单次请求测量
# ---------------------------------------------------------------------------
@dataclass
class RequestResult:
"""一次流式请求的测量结果。"""
ok: bool
ttft: Optional[float] = None # 首 token 延迟（秒）
latency: Optional[float] = None # 端到端延迟（秒）
completion_tokens: Optional[int] = None # 生成的输出 token 数
throughput: Optional[float] = None # 输出吞吐（tokens/s）
error: Optional[str] = None # 失败原因（可用性下降时记录）
def measure_once(
client: OpenAI,
model: str,
prompt: str,
max_tokens: int,
timeout: float,
) -> RequestResult:
"""
发起一次流式请求并测量各项指标。
任何异常都被捕获为一次"失败"，用于统计可用性 —— 绝不向上抛出，
以免单点故障中断整表测试。
"""
start = time.perf_counter()
first_token_at: Optional[float] = None
completion_tokens = 0
try:
stream = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens,
temperature=0.0,
stream=True,
# 请求用量统计（部分 OpenAI 兼容服务支持；不支持时下方回退到计数）
stream_options={"include_usage": True},
timeout=timeout,
)
reported_tokens: Optional[int] = None
for chunk in stream:
# 首个"有内容"的 chunk 到达时刻即 TTFT
if chunk.choices:
delta = chunk.choices[0].delta
content = getattr(delta, "content", None)
if content:
if first_token_at is None:
first_token_at = time.perf_counter()
completion_tokens += 1 # 回退计数：以流式 chunk 近似 token 数
# 若服务在末尾回传了精确 usage，则以其为准
usage = getattr(chunk, "usage", None)
if usage is not None:
reported_tokens = getattr(usage, "completion_tokens", None)
end = time.perf_counter()
if first_token_at is None:
# 拿到了响应但没有任何内容 token，视为失败
return RequestResult(ok=False, error="empty response (no content token)")
final_tokens = reported_tokens if reported_tokens else completion_tokens
latency = end - start
ttft = first_token_at - start
# 吞吐按"生成阶段"计：输出 token 数 / (端到端 - 首 token 延迟)
gen_time = max(latency - ttft, 1e-6)
throughput = final_tokens / gen_time if final_tokens else 0.0
return RequestResult(
ok=True,
ttft=ttft,
latency=latency,
completion_tokens=final_tokens,
throughput=throughput,
)
except Exception as exc: # noqa: BLE001 —— 故意兜底，任何错误都记为可用性下降
return RequestResult(ok=False, error=f"{type(exc).__name__}: {exc}")
# ---------------------------------------------------------------------------
# 聚合结果
# ---------------------------------------------------------------------------
@dataclass
class ProviderSummary:
provider: str
model: str
total: int
success: int
results: list[RequestResult] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
concurrency: int = 1 # 本次批次使用的并发数（并发扫描时用于标注行）
wall_time: float = 0.0 # 整批请求的墙钟耗时（秒），用于算聚合吞吐/RPS
@property
def availability(self) -> float:
return self.success / self.total if self.total else 0.0
@property
def rps(self) -> Optional[float]:
"""吞吐（请求/秒）：成功请求数 / 整批墙钟耗时。并发越高一般越大，直到触顶。"""
if self.wall_time <= 0:
return None
return self.success / self.wall_time
@property
def agg_throughput(self) -> Optional[float]:
"""聚合输出吞吐（tokens/s）：全部成功请求的输出 token 总数 / 整批墙钟耗时。"""
if self.wall_time <= 0:
return None
total_tokens = sum(
r.completion_tokens for r in self.results
if r.ok and r.completion_tokens
)
return total_tokens / self.wall_time if total_tokens else 0.0
def _vals(self, attr: str) -> list[float]:
return [getattr(r, attr) for r in self.results if r.ok and getattr(r, attr) is not None]
@staticmethod
def _pct(values: list[float], q: float) -> Optional[float]:
"""线性插值分位数；样本过少时退化为最大/最小值。"""
if not values:
return None
s = sorted(values)
if len(s) == 1:
return s[0]
pos = q * (len(s) - 1)
lo = int(pos)
hi = min(lo + 1, len(s) - 1)
frac = pos - lo
return s[lo] + (s[hi] - s[lo]) * frac
def stat(self, attr: str, kind: str) -> Optional[float]:
vals = self._vals(attr)
if not vals:
return None
if kind == "mean":
return statistics.mean(vals)
if kind == "std":
# 标准差：样本 <2 时无从谈起，返回 0 而非报错
return statistics.stdev(vals) if len(vals) >= 2 else 0.0
if kind == "p50":
return self._pct(vals, 0.50)
if kind == "p95":
return self._pct(vals, 0.95)
if kind == "p99":
return self._pct(vals, 0.99)
raise ValueError(kind)
def benchmark_provider(
cfg: ProviderConfig,
prompt: str,
num_requests: int,
concurrency: int,
max_tokens: int,
timeout: float,
) -> ProviderSummary:
"""对单个提供商发起 num_requests 次请求（并发 concurrency）。"""
# 这是延迟基准：显式关闭 SDK 自动重试（max_retries=0），让一次超时/挂起的
# 请求被如实记为「失败」（计入可用性下降），而不是被静默重试从而拉高延迟、
# 掩盖真实故障。每次请求仍带 per-call timeout（见 measure_once）。
# 再加一个客户端级 timeout 作为兜底，避免个别请求永久挂起拖死线程池。
# 解析实际使用的凭据/端点/模型（OpenAI 原生条目缺 key 时回退 OpenRouter）。
api_key, base_url, model, via_openrouter = cfg.resolve()
if via_openrouter:
print(f" （回退 OpenRouter：{cfg.model} -> {model}）", flush=True)
client = OpenAI(
api_key=api_key,
base_url=base_url,
timeout=timeout,
max_retries=0,
)
results: list[RequestResult] = []
batch_start = time.perf_counter()
if concurrency <= 1:
for _ in range(num_requests):
results.append(measure_once(client, model, prompt, max_tokens, timeout))
else:
with ThreadPoolExecutor(max_workers=concurrency) as pool:
futures = [
pool.submit(measure_once, client, model, prompt, max_tokens, timeout)
for _ in range(num_requests)
]
for fut in as_completed(futures):
results.append(fut.result())
wall_time = time.perf_counter() - batch_start
success = sum(1 for r in results if r.ok)
errors = [r.error for r in results if not r.ok and r.error]
return ProviderSummary(
provider=cfg.name,
model=model,
total=num_requests,
success=success,
results=results,
errors=errors,
concurrency=concurrency,
wall_time=wall_time,
)
def run_benchmark(
providers: list[ProviderConfig],
prompt: str,
num_requests: int,
concurrency: int,
max_tokens: int,
timeout: float,
) -> list[ProviderSummary]:
"""依次对每个提供商跑基准测试（提供商之间串行，单提供商内部并发）。"""
summaries: list[ProviderSummary] = []
for cfg in providers:
print(f" → 正在测试 {cfg.name} "
f"(model={cfg.model}, N={num_requests}, 并发={concurrency}) ...", flush=True)
summary = benchmark_provider(
cfg, prompt, num_requests, concurrency, max_tokens, timeout
)
print(f" 完成：成功 {summary.success}/{summary.total}", flush=True)
summaries.append(summary)
return summaries
def sweep_concurrency(
cfg: ProviderConfig,
prompt: str,
num_requests: int,
concurrency_levels: list[int],
max_tokens: int,
timeout: float,
) -> list[ProviderSummary]:
"""
压测：对同一 (provider, model) 逐步提升并发，返回每个并发档位的汇总。
对应书中"通过逐步提升并发量来找到限流点，记录 RPM/TPM 上限"——
随着并发上升，单请求延迟（p95）会变差、可用性可能因限流而下降，
而聚合吞吐（RPS / tokens·s⁻¹）会先升后平（触及服务端上限即触顶）。
"""
summaries: list[ProviderSummary] = []
for c in concurrency_levels:
print(f" → {cfg.name} @ 并发={c} (N={num_requests}) ...", flush=True)
summary = benchmark_provider(cfg, prompt, num_requests, c, max_tokens, timeout)
print(f" 完成：成功 {summary.success}/{summary.total}, "
f"墙钟 {summary.wall_time:.2f}s", flush=True)
summaries.append(summary)
return summaries
# ---------------------------------------------------------------------------
# 合成（synthetic）数据：仅供离线演示指标聚合，绝非真实基准
# ---------------------------------------------------------------------------
def synthetic_summary(
provider: str,
model: str,
num_requests: int,
concurrency: int,
*,
base_ttft: float = 0.30,
base_gen_throughput: float = 90.0,
fail_rate: float = 0.0,
seed: int = 0,
) -> ProviderSummary:
"""
用伪随机数生成一批"看起来像真实测量"的 RequestResult，用于：
1) 在没有 API key / 没有网络时验证指标聚合数学（p50/p95/p99/std/可用性）；
2) 演示并发上升时延迟长尾变差、可用性可能下降的趋势。
⚠️ 生成的所有数字都是合成的，不代表任何真实模型/提供商的性能。
并发越高，用一个简单的排队模型抬高 TTFT 与端到端延迟，仅为呈现趋势。
"""
rng = random.Random(seed + concurrency * 1000)
# 并发放大系数：并发越高，排队等待越久（简单线性 + 抖动模型）
contention = 1.0 + 0.12 * max(concurrency - 1, 0)
results: list[RequestResult] = []
total_tokens = 0
sum_latency = 0.0
for _ in range(num_requests):
# 高并发下失败率随之升高（模拟限流），封顶 60%
eff_fail = min(fail_rate * contention, 0.60)
if rng.random() < eff_fail:
results.append(RequestResult(ok=False, error="synthetic: rate_limited (429)"))
continue
# TTFT：对数正态形状，右偏（长尾），再乘并发放大
ttft = base_ttft * contention * rng.lognormvariate(0.0, 0.35)
gen_tp = max(base_gen_throughput * rng.uniform(0.75, 1.15), 1.0)
tokens = rng.randint(28, 48)
gen_time = tokens / gen_tp
latency = ttft + gen_time
total_tokens += tokens
sum_latency += latency
results.append(RequestResult(
ok=True,
ttft=ttft,
latency=latency,
completion_tokens=tokens,
throughput=gen_tp,
))
success = sum(1 for r in results if r.ok)
# 合成墙钟：把成功请求的总延迟按并发均摊，得到一个自洽的批次耗时
wall_time = max(sum_latency / max(concurrency, 1), 1e-6)
errors = [r.error for r in results if not r.ok and r.error]
return ProviderSummary(
provider=provider,
model=model,
total=num_requests,
success=success,
results=results,
errors=errors,
concurrency=concurrency,
wall_time=wall_time,
)
demo.py¶
"""
demo.py —— 一条命令跑出多提供商性能对比表 / 并发压测表。
用法：
python demo.py # 使用默认参数，多提供商横向对比
python demo.py --num-requests 20 --concurrency 5
python demo.py --serial # 串行发送（并发=1）
python demo.py --list # 仅列出将要测试的提供商
# 指定任意一个 OpenAI 兼容端点（不改代码即可测新模型/新提供商）：
python demo.py --base-url https://api.deepseek.com --model deepseek-chat \
--api-key-env DEEPSEEK_API_KEY
# 并发压测：对同一模型逐步提升并发，找限流点、看延迟长尾随并发的变化：
python demo.py --model gpt-5.6-luna --concurrency-sweep 1,2,4,8
# 离线自检（无需 key/网络）：用合成数据跑通指标聚合数学
python demo.py --mock
python demo.py --mock --concurrency-sweep 1,2,4,8,16
默认只测"手上有有效 key"的提供商（OpenAI / Kimi / 豆包）。
未设置对应环境变量的提供商会被自动跳过。
"""
from __future__ import annotations
import argparse
import json
import os
# 若安装了 python-dotenv 且存在 .env，则自动加载（可选，不强制）
try:
from dotenv import load_dotenv
load_dotenv()
except Exception: # noqa: BLE001
pass
from benchmark import (
DEFAULT_PROVIDERS,
ProviderConfig,
ProviderSummary,
run_benchmark,
sweep_concurrency,
synthetic_summary,
)
# 短 prompt：控制成本，同时保证有稳定的输出用于测吞吐。
DEFAULT_PROMPT = "用一句话解释什么是大语言模型。"
# 主对比表可选的指标族（成功率始终显示）。--metrics 用逗号选择子集。
METRIC_KEYS = ["ttft", "e2e", "throughput", "tokens"]
def _fmt(v, unit: str = "", scale: float = 1.0, digits: int = 1) -> str:
"""把可能为 None 的数值格式化为对齐的字符串。"""
if v is None:
return " N/A"
return f"{v * scale:.{digits}f}{unit}"
def _render_table(headers: list[str], rows: list[list[str]]) -> None:
"""按中文宽度对齐打印一张表。"""
def width(text: str) -> int:
return sum(2 if ord(c) > 127 else 1 for c in text)
cols = len(headers)
col_w = [width(headers[i]) for i in range(cols)]
for row in rows:
for i in range(cols):
col_w[i] = max(col_w[i], width(row[i]))
def pad(text: str, w: int) -> str:
return text + " " * (w - width(text))
sep = "-+-".join("-" * col_w[i] for i in range(cols))
print()
print(" | ".join(pad(headers[i], col_w[i]) for i in range(cols)))
print(sep)
for row in rows:
print(" | ".join(pad(row[i], col_w[i]) for i in range(cols)))
print()
def _print_errors(summaries: list[ProviderSummary]) -> None:
"""打印失败明细，便于定位可用性问题。"""
if not any(s.errors for s in summaries):
return
print("失败请求明细（可用性下降原因）：")
for s in summaries:
if s.errors:
for e in s.errors[:3]:
print(f" - {s.provider}: {e}")
if len(s.errors) > 3:
print(f" ... 以及另外 {len(s.errors) - 3} 条同类错误")
print()
def print_table(summaries: list[ProviderSummary], metrics: list[str]) -> None:
"""打印多提供商横向对比表（成功率 + 所选指标族）。"""
headers = ["Provider/Model", "成功率"]
for m in metrics:
if m == "ttft":
headers += ["TTFT均值", "TTFT_p95"]
elif m == "e2e":
headers += ["端到端均值", "端到端p95"]
elif m == "throughput":
headers += ["吞吐"]
elif m == "tokens":
headers += ["输出tok"]
rows: list[list[str]] = []
for s in summaries:
row = [
s.provider,
f"{s.success}/{s.total} ({s.availability * 100:.0f}%)",
]
for m in metrics:
if m == "ttft":
row += [_fmt(s.stat("ttft", "mean"), "ms", 1000, 0),
_fmt(s.stat("ttft", "p95"), "ms", 1000, 0)]
elif m == "e2e":
row += [_fmt(s.stat("latency", "mean"), "s", 1, 2),
_fmt(s.stat("latency", "p95"), "s", 1, 2)]
elif m == "throughput":
row += [_fmt(s.stat("throughput", "mean"), " t/s", 1, 1)]
elif m == "tokens":
row += [_fmt(s.stat("completion_tokens", "mean"), "", 1, 0)]
rows.append(row)
_render_table(headers, rows)
_print_errors(summaries)
def print_sweep_table(summaries: list[ProviderSummary]) -> None:
"""
打印并发压测表：每一行是一个并发档位，展示延迟长尾（p50/p95/p99/std）、
可用性与聚合吞吐（RPS / tokens·s⁻¹）随并发的变化。
"""
headers = [
"并发", "成功率", "TTFT_p50", "TTFT_p95",
"端到端p50", "端到端p95", "端到端p99", "端到端std",
"RPS", "聚合吞吐",
]
rows: list[list[str]] = []
for s in summaries:
rows.append([
str(s.concurrency),
f"{s.success}/{s.total} ({s.availability * 100:.0f}%)",
_fmt(s.stat("ttft", "p50"), "ms", 1000, 0),
_fmt(s.stat("ttft", "p95"), "ms", 1000, 0),
_fmt(s.stat("latency", "p50"), "s", 1, 2),
_fmt(s.stat("latency", "p95"), "s", 1, 2),
_fmt(s.stat("latency", "p99"), "s", 1, 2),
_fmt(s.stat("latency", "std"), "s", 1, 2),
_fmt(s.rps, "", 1, 1),
_fmt(s.agg_throughput, " t/s", 1, 1),
])
_render_table(headers, rows)
_print_errors(summaries)
def summary_to_dict(s: ProviderSummary) -> dict:
"""把一个汇总序列化为可 JSON 落盘的结构（供 --output 使用）。"""
def stats(attr: str) -> dict:
return {
k: s.stat(attr, k)
for k in ("mean", "std", "p50", "p95", "p99")
}
return {
"provider": s.provider,
"model": s.model,
"concurrency": s.concurrency,
"total": s.total,
"success": s.success,
"availability": s.availability,
"wall_time_s": s.wall_time,
"rps": s.rps,
"agg_throughput_tps": s.agg_throughput,
"ttft_s": stats("ttft"),
"latency_s": stats("latency"),
"throughput_tps": stats("throughput"),
"completion_tokens_mean": s.stat("completion_tokens", "mean"),
"errors": s.errors[:20],
}
def write_output(path: str, meta: dict, summaries: list[ProviderSummary]) -> None:
payload = {"meta": meta, "results": [summary_to_dict(s) for s in summaries]}
with open(path, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, indent=2)
print(f"结果已写入：{path}")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="多维度模型性能基准测试（实验 6-8）：TTFT / 端到端 / 吞吐 / p50·p95·p99·std / 可用性",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--num-requests", type=int, default=10,
help="每个档位的请求次数（默认 10，控制成本；书中口径 ≥100）")
parser.add_argument("--concurrency", type=int, default=3,
help="单档位并发数（默认 3；与 --concurrency-sweep 二选一）")
parser.add_argument("--serial", action="store_true",
help="串行发送（等价于 --concurrency 1，看无竞争下的基线延迟）")
parser.add_argument("--concurrency-sweep", type=str, default=None, metavar="1,2,4,8",
help="并发压测：逗号分隔的并发档位列表，对同一模型逐档加压找限流点")
parser.add_argument("--max-tokens", type=int, default=64,
help="每次请求生成的最大 token 数（默认 64，控制成本）")
parser.add_argument("--timeout", type=float, default=60.0,
help="单次请求超时（秒），超时记为可用性下降")
parser.add_argument("--prompt", type=str, default=DEFAULT_PROMPT,
help="测试用的短 prompt")
parser.add_argument("--metrics", type=str, default="all",
help="主对比表显示的指标族，逗号分隔，可选 "
"ttft/e2e/throughput/tokens 或 all（默认 all；成功率始终显示）")
parser.add_argument("--output", type=str, default=None, metavar="FILE.json",
help="把完整结果（含 p50/p95/p99/std）写入 JSON 文件")
parser.add_argument("--list", action="store_true",
help="仅列出将测试的提供商后退出")
# 指定任意单个 OpenAI 兼容端点（不改代码即可测新提供商/新模型）
grp = parser.add_argument_group("自定义端点（指定后只测这一个，忽略默认提供商列表）")
grp.add_argument("--base-url", type=str, default=None,
help="OpenAI 兼容端点的 base_url（OpenAI 官方留空）")
grp.add_argument("--model", type=str, default=None,
help="要测试的模型名（如 gpt-5.6-luna / deepseek-chat）")
grp.add_argument("--api-key-env", type=str, default="OPENAI_API_KEY",
help="读取 API key 的环境变量名（默认 OPENAI_API_KEY）")
grp.add_argument("--name", type=str, default=None,
help="该端点在表格中的展示名（默认用 model 名）")
parser.add_argument("--mock", action="store_true",
help="离线自检：用合成（synthetic）数据跑通指标聚合，"
"不发任何网络请求、不需要 key（数字为合成，非真实基准）")
return parser.parse_args()
def resolve_metrics(raw: str) -> list[str]:
if raw.strip().lower() == "all":
return list(METRIC_KEYS)
chosen = [m.strip() for m in raw.split(",") if m.strip()]
bad = [m for m in chosen if m not in METRIC_KEYS]
if bad:
raise SystemExit(f"未知指标：{', '.join(bad)}；可选：{', '.join(METRIC_KEYS)} 或 all")
return chosen
def build_providers(args: argparse.Namespace) -> tuple[list[ProviderConfig], list[ProviderConfig]]:
"""
返回 (available, skipped)。
若指定了 --base-url 或 --model，则构造单个自定义提供商（覆盖默认列表）。
"""
if args.base_url or args.model:
if not args.model:
raise SystemExit("使用自定义端点时必须提供 --model")
cfg = ProviderConfig(
name=args.name or f"custom/{args.model}",
model=args.model,
api_key_env=args.api_key_env,
base_url=args.base_url,
)
available = [cfg] if cfg.is_available() else []
skipped = [] if cfg.is_available() else [cfg]
return available, skipped
available = [p for p in DEFAULT_PROVIDERS if p.is_available()]
skipped = [p for p in DEFAULT_PROVIDERS if not p.is_available()]
return available, skipped
def run_mock(args: argparse.Namespace, metrics: list[str]) -> None:
"""用合成数据演示指标聚合，无需 key/网络。"""
print("=" * 72)
print("多维度模型性能基准测试（实验 6-8）—— 合成数据自检模式 [SYNTHETIC]")
print("=" * 72)
print("⚠️ 以下所有数字均为合成（伪随机）生成，仅用于验证指标聚合数学，")
print(" 不代表任何真实模型/提供商/网络环境的性能，切勿作为选型依据。")
print("-" * 72)
name = args.name or (args.model and f"custom/{args.model}") or "mock/demo-model"
model = args.model or "demo-model"
if args.concurrency_sweep:
levels = parse_sweep_levels(args.concurrency_sweep)
print(f"并发压测（合成）：{name} 档位={levels} N={args.num_requests}/档")
summaries = [
synthetic_summary(name, model, args.num_requests, c, fail_rate=0.02, seed=42)
for c in levels
]
print_sweep_table(summaries)
print("解读：并发上升 → 端到端 p95/p99 与 std 走高（长尾变差），")
print(" 可用性因限流下降，聚合吞吐先升后趋平（触及服务端上限即触顶）。")
else:
concurrency = 1 if args.serial else args.concurrency
print(f"单档位对比（合成）：并发={concurrency} N={args.num_requests}/家")
# 造三个"提供商"，参数不同以体现横向差异
summaries = [
synthetic_summary("mockA/fast-low-ttft", "fast", args.num_requests,
concurrency, base_ttft=0.20, base_gen_throughput=110, seed=1),
synthetic_summary("mockB/balanced", "balanced", args.num_requests,
concurrency, base_ttft=0.35, base_gen_throughput=85, seed=2),
synthetic_summary("mockC/high-throughput", "hi-tp", args.num_requests,
concurrency, base_ttft=0.55, base_gen_throughput=140,
fail_rate=0.05, seed=3),
]
print_table(summaries, metrics)
if args.output:
write_output(args.output, {"mode": "mock-synthetic", "note": "数字为合成，非真实基准"},
summaries)
def parse_sweep_levels(raw: str) -> list[int]:
try:
levels = [int(x) for x in raw.split(",") if x.strip()]
except ValueError:
raise SystemExit(f"--concurrency-sweep 需为逗号分隔的整数，如 1,2,4,8；收到：{raw!r}")
levels = [c for c in levels if c >= 1]
if not levels:
raise SystemExit("--concurrency-sweep 至少需要一个 ≥1 的并发档位")
return levels
def main() -> None:
args = parse_args()
metrics = resolve_metrics(args.metrics)
if args.mock:
run_mock(args, metrics)
return
available, skipped = build_providers(args)
print("=" * 72)
print("多维度模型性能基准测试（实验 6-8）")
print("=" * 72)
if skipped:
for p in skipped:
print(f"[跳过] {p.name} —— 未设置环境变量 {p.api_key_env}")
if not available:
print("没有任何可用提供商：请设置对应 API key 环境变量，")
print("或用 --mock 在无 key 情况下离线验证指标聚合。")
return
print(f"待测提供商：{', '.join(p.name for p in available)}")
# ---- 并发压测模式 ----
if args.concurrency_sweep:
levels = parse_sweep_levels(args.concurrency_sweep)
print(f"模式：并发压测（逐档加压找限流点） 档位={levels}")
print(f"参数：N={args.num_requests}/档, max_tokens={args.max_tokens}, "
f"timeout={args.timeout}s")
print(f"Prompt：{args.prompt!r}")
if args.list:
return
all_summaries: list[ProviderSummary] = []
for cfg in available:
print("-" * 72)
print(f"压测 {cfg.name}:")
summaries = sweep_concurrency(
cfg, args.prompt, args.num_requests, levels,
args.max_tokens, args.timeout,
)
print_sweep_table(summaries)
all_summaries.extend(summaries)
if args.output:
write_output(args.output,
{"mode": "concurrency-sweep", "levels": levels}, all_summaries)
return
# ---- 单档位横向对比模式（默认，保持原行为）----
concurrency = 1 if args.serial else args.concurrency
print(f"参数：N={args.num_requests}/家, 并发={concurrency}, "
f"max_tokens={args.max_tokens}, timeout={args.timeout}s")
print(f"Prompt：{args.prompt!r}")
if args.list:
return
print("-" * 72)
summaries = run_benchmark(
providers=available,
prompt=args.prompt,
num_requests=args.num_requests,
concurrency=concurrency,
max_tokens=args.max_tokens,
timeout=args.timeout,
)
print_table(summaries, metrics)
print("指标说明：")
print(" 成功率 = 成功请求数 / 总请求数（可用性维度）")
print(" TTFT = 首个 token 到达延迟（流式测得），越低越流畅")
print(" 端到端 = 请求发出到响应结束的总耗时")
print(" 吞吐 = 输出 token 数 / 生成阶段耗时（tokens/s）")
print(" p95 = 95 分位延迟，反映长尾/稳定性（方差大则体验不稳）")
print(" 提示 = 加 --concurrency-sweep 1,2,4,8 可做并发压测，看指标随并发的变化")
if args.output:
write_output(args.output,
{"mode": "single", "concurrency": concurrency}, summaries)
if __name__ == "__main__":
main()