性能测试培训 · AI专项篇
大模型 + AI应用性能测试 · 从指标定义到实战落地
1. AI接口 vs 传统接口
在开始 AI 性能测试之前,必须先理解 AI 接口和传统 REST 接口在本质上的差异。这些差异直接决定了测试方法、工具选型和评估标准。
1.1 核心差异对比
| 维度 | 传统接口 | AI 大模型接口 |
|---|---|---|
| 响应模式 | 一次性返回完整 JSON | 流式逐 Token 返回(SSE) |
| 响应时间 | 50~500ms | 3~30s(甚至更长) |
| 计费方式 | 按请求次数 / 不计费 | 按 Token 数量计费 |
| 幂等性 | 通常幂等(GET) | 不幂等,每次回答都不同 |
| 资源消耗 | CPU + 内存 | GPU 显存 + GPU 算力 |
| 并发天花板 | 数千~数万 QPS | 10~100 并发 |
| 超时阈值 | 3~10 秒 | 30~120 秒 |
| 限流方式 | 固定 QPS 限流 | RPM + TPM 多维限流 |
| 请求大小 | KB 级 | KB~MB 级(长上下文) |
| 结果验证 | 精确比对 | 只能做格式/完整性校验 |
1.2 响应时间数量级对比
📊 典型响应时间对比
传统接口 AI 大模型接口
━━━━━━━━━━━━━━━━━━━━━━ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Redis 查询 ~1ms TTFT(首Token) ~200ms~3s
内存缓存 ~5ms 短回答总时长 ~2~5s
数据库查询 ~50ms 中等回答总时长 ~5~15s
业务逻辑 ~100ms 长文生成总时长 ~15~60s
外部API ~300ms 多工具串行调用 ~10~45s
|----+----+----+----+----+----+----+----+----+---->
0 100ms 500ms 1s 3s 5s 10s 15s 30s 60s
[传统接口区间] [AI接口区间.................]传统接口的响应时间在毫秒级别,而 AI 接口在秒级甚至十秒级。这意味着:
- 传统压测 5 分钟可以产生 30,000+ 请求,AI 压测 5 分钟可能只有 300~1000 请求
- 统计意义上需要更长的压测时间才能得到可靠结论
- 每个并发用户占用连接的时间更长,系统需要维持更多长连接
1.3 计费模式的本质差异
| 维度 | 传统接口 | AI 接口 |
|---|---|---|
| 费用来源 | 服务器资源(CPU/内存/带宽) | Token 消耗 + GPU 算力 |
| 压测成本 | 服务器已购买,边际成本 ≈ 0 | 每次请求都产生 Token 费用 |
| 费用可控性 | 固定成本 | 与压测规模线性增长 |
| 典型压测费用 | 忽略不计 | 几十~几百元(第三方API) |
⚠️ 真实案例
某团队用 GPT-4 做压测,50 并发跑了 1 小时,产生了 ¥3,200 的 API 费用。后来改用 GPT-4o-mini + 短 Prompt,同等规模测试费用降到 ¥15。
1.4 流式返回对压测工具的挑战
传统压测工具(如 JMeter、wrk)基于"请求 → 等待 → 收到完整响应"模型设计,而 AI 接口的 SSE 流式返回打破了这个假设:
| 挑战 | 说明 |
|---|---|
| 响应时间定义模糊 | 是首 Token 时间?还是完整响应时间?传统工具无法区分 |
| 数据持续到达 | 响应不是一个完整 JSON,而是多个 data: 事件流 |
| 连接保持时间长 | 单个请求可能占用连接 30 秒以上 |
| 自定义指标 | TTFT、Token 吞吐等指标需要自行计算 |
课堂练习
- 列出你所在项目中 AI 接口和传统接口各 3 个,对比它们在上述维度的差异
- 计算:如果用 GPT-4o 做压测,10 并发 × 10 分钟 × 平均每请求 500 input + 200 output tokens,压测费用是多少?
- 讨论:你的传统压测工具(JMeter/k6/Locust)能直接用于 AI 接口吗?需要做哪些适配?
2. 流式响应原理(SSE)
几乎所有主流大模型 API(OpenAI、Claude、豆包、Kimi、通义千问)都采用 SSE(Server-Sent Events)协议来实现流式返回。理解 SSE 是做 AI 性能测试的基础。
2.1 SSE 协议详解
📖 什么是 SSE
SSE(Server-Sent Events)是一种基于 HTTP 的单向推送协议。客户端发起一个普通 HTTP 请求,服务端保持连接不关闭,持续向客户端推送事件(event),每个事件以 data: 开头,事件之间用空行分隔。
SSE 请求头特征
POST /v1/chat/completions HTTP/1.1
Content-Type: application/json
Authorization: Bearer sk-xxx
{
"model": "gpt-4o",
"messages": [{"role": "user", "content": "你好"}],
"stream": true
}SSE 响应头特征
HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
Transfer-Encoding: chunkedSSE 数据流格式
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"你"},"finish_reason":null}]}
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"好"},"finish_reason":null}]}
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"!"},"finish_reason":null}]}
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
data: [DONE]2.2 SSE vs WebSocket vs HTTP Long Polling
| 特性 | SSE | WebSocket | HTTP Long Polling |
|---|---|---|---|
| 通信方向 | 服务端 → 客户端(单向) | 双向 | 服务端 → 客户端 |
| 协议 | HTTP/1.1 或 HTTP/2 | 独立的 ws:// 协议 | HTTP |
| 连接方式 | 保持 HTTP 连接 | 升级到 WebSocket 连接 | 反复建立新连接 |
| 自动重连 | 浏览器原生支持 | 需手动实现 | 每次都是新连接 |
| 大模型场景 | 主流方案 | 部分厂商使用 | 基本不用 |
| 压测复杂度 | 中等 | 较高 | 简单 |
| 代理/CDN 兼容 | 好(标准 HTTP) | 需额外配置 | 好 |
💡 为什么大模型选择 SSE 而不是 WebSocket?
- 大模型对话本质上是"请求 → 流式响应"的单向模式,不需要双向通信
- SSE 基于标准 HTTP,对代理、负载均衡器、CDN 兼容性好
- SSE 有内置的重连机制,更可靠
- 对于 AI 场景,简单比功能全更重要
2.3 如何正确测量流式响应的各项指标
发送请求 t₀ → 收到首个 data: t₁ → 持续接收 Token → 收到 [DONE] t₂
| 指标 | 计算方式 |
|---|---|
| TTFT | t₁ - t₀ (发请求到收到第一个有内容的 data) |
| 端到端延迟 | t₂ - t₀ (发请求到收到 [DONE]) |
| Token 吞吐 | 总 Token 数 / (t₂ - t₁) (生成阶段的速率) |
| Token 间隔 | 相邻两个 data 事件的时间差,反映生成流畅度 |
2.4 Python 解析 SSE 流示例
import httpx
import json
import time
async def stream_chat(prompt: str) -> dict:
"""流式调用大模型并收集性能指标"""
url = "https://api.openai.com/v1/chat/completions"
headers = {
"Authorization": "Bearer sk-xxx",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"stream": True,
"max_tokens": 200
}
t_start = time.perf_counter()
t_first_token = None
token_count = 0
full_content = ""
async with httpx.AsyncClient(timeout=60) as client:
async with client.stream("POST", url, json=payload, headers=headers) as resp:
async for line in resp.aiter_lines():
if not line.startswith("data: "):
continue
data_str = line[6:] # 去掉 "data: " 前缀
if data_str == "[DONE]":
break
chunk = json.loads(data_str)
delta = chunk["choices"][0]["delta"]
content = delta.get("content", "")
if content and t_first_token is None:
t_first_token = time.perf_counter()
if content:
token_count += 1 # 简化:每个 chunk 算 1 个 token
full_content += content
t_end = time.perf_counter()
return {
"ttft_ms": round((t_first_token - t_start) * 1000, 2) if t_first_token else None,
"total_ms": round((t_end - t_start) * 1000, 2),
"token_count": token_count,
"tokens_per_sec": round(token_count / (t_end - t_first_token), 2) if t_first_token else 0,
"content": full_content
}示例
运行上述代码,发送 Prompt "用一句话解释什么是机器学习",典型输出:
{
"ttft_ms": 342.15,
"total_ms": 2156.78,
"token_count": 38,
"tokens_per_sec": 20.94,
"content": "机器学习是人工智能的一个分支,通过让计算机从数据中自动学习模式和规律,从而做出预测或决策。"
}课堂练习
- 修改上面的 Python 代码,添加"Token 间隔时间"的记录(记录每个 Token 到达的时间戳,计算相邻 Token 的时间差)
- 用
curl命令手动调用一个 SSE 接口,观察原始数据流格式 - 讨论:为什么
token_count用 chunk 数来近似?更精确的 Token 计数方法是什么?
3. TTFT — 首 Token 延迟
3.1 定义与计算公式
📐 TTFT = Time To First Token
从客户端发出请求到收到第一个有效 Token 的时间差。 TTFT = t(第一个Token到达) - t(请求发出)
3.2 为什么 TTFT 对 AI 产品特别重要
- **用户心理感知:**用户点击"发送"后,看到第一个字开始出现的等待时间。超过 3 秒用户开始焦虑,超过 5 秒开始怀疑系统是否卡死
- **流式体验基础:**TTFT 是流式输出的起点,TTFT 过长意味着用户面对空白等待
- **产品差异化指标:**不同模型的 TTFT 差异显著,直接影响用户选择
3.3 什么影响 TTFT
| 影响因素 | 影响程度 | 说明 |
|---|---|---|
| 模型大小 | ⭐⭐⭐⭐⭐ | 7B 模型 TTFT ~200ms,70B 模型 TTFT ~800ms |
| 输入 Token 长度 | ⭐⭐⭐⭐ | Prefill 阶段需要处理所有输入 Token,输入越长 TTFT 越大 |
| 并发数 | ⭐⭐⭐⭐ | 并发增加导致排队,TTFT 急剧上升 |
| GPU 显存 | ⭐⭐⭐ | 显存不足会触发 swap/卸载,大幅增加延迟 |
| KV Cache 命中 | ⭐⭐⭐ | 多轮对话如果前缀匹配,Prefill 可以跳过已缓存部分 |
| 量化精度 | ⭐⭐ | INT4/INT8 量化加速 Prefill,降低 TTFT |
| 网络延迟 | ⭐⭐ | 第三方 API 有网络往返,自部署可忽略 |
3.4 TTFT 测量代码示例
import httpx
import time
import json
import asyncio
from dataclasses import dataclass, field
@dataclass
class TTFTResult:
prompt: str
ttft_ms: float
status_code: int
error: str = ""
async def measure_ttft(prompt: str, api_url: str, api_key: str) -> TTFTResult:
"""精确测量单次请求的 TTFT"""
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"stream": True,
"max_tokens": 50
}
t_start = time.perf_counter()
try:
async with httpx.AsyncClient(timeout=30) as client:
async with client.stream("POST", api_url, json=payload, headers=headers) as resp:
if resp.status_code != 200:
return TTFTResult(prompt=prompt, ttft_ms=-1,
status_code=resp.status_code, error=f"HTTP {resp.status_code}")
async for line in resp.aiter_lines():
if line.startswith("data: ") and line[6:] != "[DONE]":
chunk = json.loads(line[6:])
content = chunk["choices"][0]["delta"].get("content", "")
if content:
ttft = (time.perf_counter() - t_start) * 1000
return TTFTResult(prompt=prompt, ttft_ms=round(ttft, 2),
status_code=200)
except Exception as e:
return TTFTResult(prompt=prompt, ttft_ms=-1, status_code=0, error=str(e))
return TTFTResult(prompt=prompt, ttft_ms=-1, status_code=200, error="No content received")3.5 TTFT 健康范围参考值
| 场景 | 优秀 | 合格 | 需优化 |
|---|---|---|---|
| 自部署 7B 模型(低并发) | < 200ms | < 500ms | > 1s |
| 自部署 70B 模型(低并发) | < 500ms | < 1.5s | > 3s |
| 第三方 API(GPT-4o) | < 500ms | < 1.5s | > 3s |
| 第三方 API(Claude 3.5) | < 600ms | < 2s | > 4s |
| RAG 全链路 TTFT | < 2s | < 4s | > 6s |
| 高并发场景(50+) | < 3s | < 5s | > 10s |
3.6 不同模型 TTFT 典型值对比
| 模型 | 1并发 TTFT | 10并发 TTFT | 50并发 TTFT |
|---|---|---|---|
| GPT-4o-mini | ~300ms | ~400ms | ~800ms |
| GPT-4o | ~500ms | ~700ms | ~2s |
| Claude 3.5 Sonnet | ~600ms | ~900ms | ~2.5s |
| 豆包 Pro | ~200ms | ~350ms | ~1s |
| 通义千问 Max | ~300ms | ~500ms | ~1.5s |
| Kimi | ~400ms | ~600ms | ~2s |
| 自部署 Qwen-7B (vLLM) | ~100ms | ~250ms | ~1.5s |
| 自部署 Qwen-72B (vLLM) | ~400ms | ~1s | ~5s |
注意:
以上数据为典型参考值,实际会受网络、输入长度、服务负载等因素影响。
课堂练习
- 用上面的代码测量你项目中 AI 接口的 TTFT,分别在 1/5/10 并发下测试
- 绘制"并发数-TTFT"折线图,找到 TTFT 急剧上升的拐点
- 对比不同长度的 Prompt(100 Token vs 1000 Token vs 5000 Token)对 TTFT 的影响
4. Token 吞吐速率
4.1 定义与计算公式
📐 Token Throughput = tokens/second
单位时间内模型生成的 Token 数量。可以从两个维度衡量:
- 单请求吞吐:
输出 Token 数 / 生成时间(s)— 衡量用户感知的"打字速度" - 系统整体吞吐:
所有请求的总输出 Token 数 / 总时间(s)— 衡量系统处理能力
4.2 为什么这个指标重要
Token 吞吐直接决定了用户的"阅读等待体验":
| 吞吐速率 | 用户体感 |
|---|---|
| > 60 tokens/s | 文字瞬间刷出,几乎无感等待 |
| 30~60 tokens/s | 流畅阅读,像人在快速打字 |
| 15~30 tokens/s | 可接受,但能感觉到速度 |
| 5~15 tokens/s | 明显感觉慢,像人在慢慢思考着打字 |
| < 5 tokens/s | 令人焦虑,用户可能放弃等待 |
4.3 影响因素
| 因素 | 影响 | 说明 |
|---|---|---|
| 模型架构 | ⭐⭐⭐⭐⭐ | MoE 架构(如 Mixtral)decode 更快 |
| 量化精度 | ⭐⭐⭐⭐ | FP16→INT8:吞吐提升 ~30%;FP16→INT4:提升 ~60% |
| Batch Size | ⭐⭐⭐⭐ | 连续批处理将多个请求合并,单请求吞吐下降但系统整体吞吐上升 |
| KV Cache | ⭐⭐⭐ | 避免重复计算注意力,长序列尤其重要 |
| GPU 型号 | ⭐⭐⭐ | A100 vs H100 的 decode 速度差异 ~2x |
| Speculative Decoding | ⭐⭐ | 投机解码可提升 2-3x 吞吐 |
4.4 Token 吞吐测量方法
import time
import json
import httpx
import tiktoken
async def measure_throughput(prompt: str, api_url: str, api_key: str) -> dict:
"""测量 Token 吞吐速率"""
enc = tiktoken.encoding_for_model("gpt-4o")
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"stream": True,
"max_tokens": 500
}
full_content = ""
t_first = None
t_last = None
chunk_timestamps = []
async with httpx.AsyncClient(timeout=60) as client:
async with client.stream("POST", api_url, json=payload, headers=headers) as resp:
async for line in resp.aiter_lines():
if not line.startswith("data: ") or line[6:] == "[DONE]":
continue
now = time.perf_counter()
chunk = json.loads(line[6:])
content = chunk["choices"][0]["delta"].get("content", "")
if content:
if t_first is None:
t_first = now
t_last = now
full_content += content
chunk_timestamps.append(now)
output_tokens = len(enc.encode(full_content))
gen_duration = t_last - t_first if t_first and t_last else 0
intervals = [chunk_timestamps[i] - chunk_timestamps[i-1]
for i in range(1, len(chunk_timestamps))]
return {
"output_tokens": output_tokens,
"gen_duration_s": round(gen_duration, 3),
"tokens_per_sec": round(output_tokens / gen_duration, 2) if gen_duration > 0 else 0,
"avg_interval_ms": round(sum(intervals) / len(intervals) * 1000, 2) if intervals else 0,
"max_interval_ms": round(max(intervals) * 1000, 2) if intervals else 0,
}课堂练习
- 分别测量你的模型在 1/5/10/20 并发下的单请求 Token 吞吐和系统总吞吐
- 绘制"并发数 — 单请求吞吐"和"并发数 — 系统总吞吐"双轴图
- 思考:为什么并发增加时,单请求吞吐下降但系统总吞吐反而上升?(提示:连续批处理)
5. 端到端延迟
5.1 定义与计算公式
📐 End-to-End Latency
E2E Latency = t(收到完整响应 / [DONE]) - t(请求发出) 这是用户发出问题到看到完整回答的总时间。对于非流式场景尤为关键。
5.2 完整链路分解
客户端发送 → 网络传输 → 服务端接收/鉴权 → 排队等待 → Prefill → Decode 生成 → 后处理 → 网络返回
5.3 各环节耗时占比(典型值)
| 环节 | 低并发 占比 | 高并发 占比 | 说明 |
|---|---|---|---|
| 网络往返 | 5% | 2% | 同机房 ~1ms,跨境 ~100ms |
| 鉴权/预处理 | 2% | 1% | Token 验证、参数校验 |
| 排队等待 | 0% | 30~50% | 高并发瓶颈所在 |
| Prefill(输入处理) | 10% | 5% | 与输入长度线性相关 |
| Decode(Token 生成) | 80% | 40% | 最耗时环节,与输出长度相关 |
| 后处理 | 3% | 2% | 内容审核、格式化 |
💡 关键洞察
低并发时瓶颈在 Decode(模型生成本身慢),高并发时瓶颈转移到排队等待。这意味着优化方向在不同阶段完全不同:低并发优化模型推理,高并发优化调度和扩容。
5.4 端到端延迟的分段测量
对于 RAG 或 Tool Calling 场景,端到端延迟需要做更精细的分段记录:
RAG 端到端延迟分解:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
环节 耗时 累计 占比
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Query 预处理 30ms 30ms 0.5%
Embedding 向量化 120ms 150ms 2%
向量检索 50ms 200ms 1%
Rerank 重排序 200ms 400ms 3%
Prompt 拼接 10ms 410ms 0.2%
大模型生成 5200ms 5610ms 89%
后处理/审核 150ms 5760ms 2.5%
网络传输 100ms 5860ms 1.8%
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
总端到端延迟 5860ms - 100%课堂练习
- 对你的 AI 系统做端到端延迟分解,找出各环节的耗时占比
- 分别在低并发和高并发下做分解,观察排队等待时间的变化
- 基于分解结果,列出优化优先级(哪个环节优化 ROI 最高)
6. Rate Limit 与配额
6.1 三种限流维度
| 维度 | 全称 | 含义 | 典型值 |
|---|---|---|---|
| RPM | Requests Per Minute | 每分钟最大请求数 | 60~10000 |
| RPD | Requests Per Day | 每天最大请求数 | 500~100000 |
| TPM | Tokens Per Minute | 每分钟最大 Token 数 | 40K~10M |
💡 注意:三个维度是"取交集"
即使你的 RPM 还有余量,但 TPM 超了,也会被限流。长上下文场景特别容易触发 TPM 限流。
6.2 各家限流策略对比表
| 厂商 | RPM | TPM | RPD | 限流响应 |
|---|---|---|---|---|
| OpenAI (Tier 1) | 500 | 30,000 | 10,000 | HTTP 429 + Retry-After |
| OpenAI (Tier 3) | 5,000 | 800,000 | — | HTTP 429 + Retry-After |
| OpenAI (Tier 5) | 10,000 | 10,000,000 | — | HTTP 429 + Retry-After |
| Claude (Free) | 5 | 20,000 | 300 | HTTP 429 |
| Claude (Team) | 4,000 | 400,000 | — | HTTP 429 + 过载时 529 |
| 豆包 | 按套餐 | 按套餐 | — | HTTP 429 |
| Kimi | 按套餐 | 按套餐 | 按套餐 | HTTP 429 |
| 通义千问 | 按套餐 | 按套餐 | — | HTTP 429 |
6.3 限流后的系统行为测试
限流不可避免,关键是系统在限流后如何表现。需要测试以下场景:
场景 A:限流后排队重试
→ 用户请求被 429 → 系统自动等待 Retry-After 后重试
→ 观察:用户等待时间?队列是否会堆积溢出?
场景 B:限流后切换模型
→ GPT-4o 被 429 → 自动切 GPT-4o-mini / 豆包 / 通义千问
→ 观察:切换延迟?回答质量下降多少?
场景 C:限流后直接降级
→ 所有模型都 429 → 返回预设回答 / 提示用户稍后重试
→ 观察:降级是否优雅?是否暴露了技术错误信息?
场景 D:限流恢复后的行为
→ 限流解除 → 积压的请求是否会"洪峰"发出?
→ 观察:是否有令牌桶/漏桶控制,避免恢复后再次触发限流?课堂练习
- 查询你使用的 AI API 的 Rate Limit 配额(通过 API 或控制台),记录 RPM/TPM/RPD
- 编写脚本,逐步增加并发直到触发 429,记录触发阈值和响应头中的 Retry-After 值
- 测试你的系统在被限流后的实际行为(是否有重试/降级/友好提示)
7. Token 消耗与成本
7.1 各家定价对比
| 模型 | Input (每百万Token) | Output (每百万Token) | 备注 |
|---|---|---|---|
| GPT-4o | $2.50 | $10.00 | 主力模型 |
| GPT-4o-mini | $0.15 | $0.60 | 性价比之选 |
| Claude 3.5 Sonnet | $3.00 | $15.00 | 代码能力强 |
| Claude 3.5 Haiku | $0.80 | $4.00 | 快速响应 |
| 豆包 Pro 128K | ¥5.00 | ¥9.00 | 国产性价比高 |
| 豆包 Lite 128K | ¥0.30 | ¥0.60 | 极致低价 |
| 通义千问 Max | ¥2.40 | ¥9.60 | 阿里云生态 |
| Kimi | ¥1.00 | ¥3.00 | 长上下文 |
7.2 压测成本预算计算
📐 压测费用公式
费用 = 总请求数 × (平均 Input Tokens × Input 单价 + 平均 Output Tokens × Output 单价)
总请求数 = 并发数 × 持续时间(秒) / 平均请求时间(秒)
示例:GPT-4o-mini, 10 并发, 10 分钟, 平均每请求 3 秒
总请求数 = 10 × 600 / 3 = 2,000 次
假设平均 input 500 tokens, output 200 tokens
Input 费用 = 2000 × 500 / 1M × $0.15 = $0.15
Output 费用 = 2000 × 200 / 1M × $0.60 = $0.24
总费用 ≈ $0.39 ≈ ¥2.8
如果换成 GPT-4o:
Input 费用 = 2000 × 500 / 1M × $2.50 = $2.50
Output 费用 = 2000 × 200 / 1M × $10.00 = $4.00
总费用 ≈ $6.50 ≈ ¥477.3 成本优化技巧
| 技巧 | 效果 | 说明 |
|---|---|---|
| 用最便宜的模型 | 节省 90%+ | 压测目标是测系统能力,不是测模型智能 |
| 缩短 Prompt | 节省 30~70% | 用 "回复OK" 代替真实长 Prompt |
| 限制 max_tokens | 节省 50~80% | 设为 10~50,不需要完整回答 |
| 控制压测时间 | 精确控制 | 每轮 3~5 分钟足够观察趋势 |
| 使用 Mock 先调通 | 避免浪费 | 脚本调试阶段用 Mock Server |
| 分阶段测试 | 避免浪费 | 先小规模验证,确认无误再放量 |
课堂练习
- 计算你的项目如果用 GPT-4o 做一次完整压测(10 并发 × 30 分钟),预计费用是多少
- 制定一个"低成本压测方案":选择哪个模型、Prompt 多长、max_tokens 设多少、跑多久
- 对比方案:相同并发下,用 Mock Server 跑一遍 vs 用真实 API 跑一遍,结果差异在哪里
8. 自研模型压测
模型部署在自己的 GPU 服务器上,没有第三方限制,最接近传统压测,但需要关注 GPU 这个特殊资源。
8.1 GPU 推理框架对比
| 框架 | 特点 | 适合场景 | 并发能力 |
|---|---|---|---|
| vLLM | PagedAttention + 连续批处理,开源生态好 | 通用生产部署 | 高 |
| TGI (Text Generation Inference) | HuggingFace 出品,Flash Attention 优化 | HF 生态模型 | 高 |
| Ollama | 简单易用,本地开发友好 | 开发调试/小模型 | 低 |
| TensorRT-LLM | NVIDIA 官方,极致性能优化 | NVIDIA GPU 生产部署 | 最高 |
| SGLang | RadixAttention,前缀缓存效率高 | 多轮对话/共享前缀 | 高 |
8.2 推理优化技术对性能的影响
| 技术 | 原理 | TTFT 影响 | 吞吐影响 | 显存影响 |
|---|---|---|---|---|
| KV Cache | 缓存已计算的 Key-Value,避免重复计算 | 多轮对话降低 | 提升 10~30% | 增加占用 |
| 连续批处理 | 动态合并不同阶段的请求一起推理 | 可能增加 | 系统吞吐提升 2~5x | 更高效利用 |
| INT8 量化 | 将权重从 FP16 转为 INT8 | 降低 ~20% | 提升 ~30% | 减半 |
| INT4 量化 (GPTQ/AWQ) | 4 位量化 | 降低 ~40% | 提升 ~60% | 减为 1/4 |
| Flash Attention | IO-aware 注意力计算 | 长序列显著降低 | 提升 10~20% | 减少 |
| Speculative Decoding | 小模型草稿 + 大模型验证 | 不变 | 提升 2~3x | 增加(额外小模型) |
8.3 自研模型压测典型数据
| 并发数 | 1 | 5 | 10 | 20 | 50 |
|---|---|---|---|---|---|
| TTFT(ms) | 200 | 250 | 400 | 1200 | 5000 |
| Tokens/s(单请求) | 50 | 48 | 42 | 30 | 15 |
| Tokens/s(系统总) | 50 | 240 | 420 | 600 | 750 |
| GPU 利用率 | 30% | 65% | 85% | 98% | 99% |
| GPU 显存 | 12GB | 14GB | 18GB | 22GB | 23.5GB(接近24GB上限) |
| 队列等待 | 0ms | 0ms | 100ms | 2s | 8s |
| 错误率 | 0% | 0% | 0% | 0.5% | 5% |
**结论:**这台 GPU(24GB 显存)最佳并发 ≈ 10,极限 ≈ 20。
8.4 GPU 显存分析
GPU 显存占用 = 模型权重 + KV Cache + 激活值 + 框架开销
以 Qwen-14B (FP16) 为例:
模型权重:14B × 2 bytes = 28GB → 需要至少 2 张 A100-40GB
KV Cache(每请求):~200MB(4K 上下文)
20 并发 KV Cache:20 × 200MB = 4GB
框架开销:~2GB
总计 ≈ 34GB → 刚好放在 2 张 A100-40GB 上
如果用 INT4 量化:
模型权重:14B × 0.5 bytes = 7GB → 1 张 A100-40GB 绰绰有余
同等条件下可以支撑更多并发8.5 完整自研模型压测方案模板
TIP
一、环境信息
├─ GPU 型号/数量:NVIDIA A100-80GB × 2
├─ 推理框架:vLLM v0.4.x
├─ 模型:Qwen2-72B-Instruct (INT4 AWQ)
└─ 系统:Ubuntu 22.04, CUDA 12.2, Python 3.11
二、固定变量
├─ Input Tokens:500 ± 50(控制 Prompt 长度)
├─ Max Output Tokens:1024
├─ Temperature:0.7, Top-P:0.9
└─ 测试数据:50 条不同 Prompt
三、测试阶段
├─ 基准(1 并发,3 分钟)→ 记录基准 TTFT/吞吐
├─ 阶梯加压(1→5→10→20→50→100,每级 5 分钟)
├─ 稳定性(最佳并发数,持续 2 小时)
└─ 极限(持续加压直到错误率 > 5%)
四、监控指标
├─ 客户端:TTFT / Token 吞吐 / E2E 延迟 / 错误率
└─ 服务端:GPU 利用率 / 显存 / 队列长度 / CPU / 内存
五、输出物
├─ 各并发级别的性能数据表
├─ 性能曲线图(并发数 vs TTFT/吞吐/GPU 利用率)
└─ 最佳并发数建议 + 扩容方案课堂练习
- 如果你的项目使用自部署模型,用
nvidia-smi查看当前 GPU 型号和显存 - 按上面的模板制定一份压测方案,填入实际的模型和环境信息
- 计算你的模型在 FP16 和 INT4 下分别需要多少 GPU 显存
9. 第三方 API 压测
压测第三方大模型 API,你的目标不是测模型本身的能力,而是测你的系统在真实调用链路下的表现。
9.1 核心测试点
- 你的后端服务 — 能不能高效转发和管理大模型请求
- 第三方 API 的限流策略 — 你的配额够不够用
- 你的重试/降级/排队机制 — 被限流后系统表现如何
9.2 配额查询方法
| 厂商 | 查询方式 | 说明 |
|---|---|---|
| OpenAI | 控制台 → Settings → Limits;或检查响应头 x-ratelimit-remaining-* | Tier 等级决定配额 |
| Claude | 响应头 anthropic-ratelimit-* | 包含 requests/tokens 剩余量 |
| 豆包 | 火山引擎控制台 → 模型管理 | 按套餐等级划分 |
| 通义千问 | 阿里云 DashScope 控制台 | 可申请提升配额 |
| Kimi | Moonshot 控制台 | 不同模型配额不同 |
// OpenAI 响应头中的限流信息示例
x-ratelimit-limit-requests: 500
x-ratelimit-limit-tokens: 30000
x-ratelimit-remaining-requests: 485
x-ratelimit-remaining-tokens: 28500
x-ratelimit-reset-requests: 12s
x-ratelimit-reset-tokens: 3s9.3 阶梯式加压策略
第一步:先搞清楚你的配额
→ 例如 OpenAI Tier 1: 500 RPM ≈ 8.3 RPS
第二步:在配额范围内做阶梯式加压
→ 1 → 3 → 5 → 8 → 10 → 15 并发
→ 每个阶段持续 3~5 分钟
→ 观察:429 开始出现的并发数、TTFT 变化趋势
第三步:专门测限流后的系统行为
→ 故意超过配额去压
→ 你的系统是直接报错?还是排队?还是切备用模型?9.4 多模型负载均衡方案
✅ 推荐:多模型路由架构
┌──→ OpenAI GPT-4o (主)
用户请求 → 路由层 ──┼──→ Claude 3.5 Sonnet (备)
├──→ 豆包 Pro (第二备)
└──→ 自部署模型 (兜底)
路由策略:
1. 默认走主模型
2. 主模型 429 → 切备用(带权重轮询)
3. 所有第三方都 429 → 走自部署
4. 自部署也满载 → 排队 + 提示用户9.5 限流后的自动降级策略
class ModelRouter:
def __init__(self):
self.models = [
{"name": "gpt-4o", "priority": 1, "status": "healthy"},
{"name": "claude-3.5", "priority": 2, "status": "healthy"},
{"name": "doubao-pro", "priority": 3, "status": "healthy"},
{"name": "qwen-local", "priority": 4, "status": "healthy"},
]
async def route(self, request):
for model in sorted(self.models, key=lambda m: m["priority"]):
if model["status"] == "healthy":
try:
resp = await call_model(model["name"], request)
return resp
except RateLimitError:
model["status"] = "rate_limited"
model["retry_after"] = time.time() + 60
continue
except TimeoutError:
model["status"] = "timeout"
continue
return fallback_response("所有模型暂时不可用,请稍后重试")9.6 费用控制
- 用最短的 Prompt("回复OK"这种)
- 限制
max_tokens到最小值(比如 10) - 用最便宜的模型(
gpt-4o-mini而不是gpt-4o) - 设定压测时间上限(比如总共只跑 10 分钟)
- 提前算好预算:10 分钟 × 8 QPS × 每次 ¥0.001 ≈ ¥5
课堂练习
- 查询你使用的 API 的剩余配额,用代码从响应头中提取限流信息
- 设计一个多模型路由方案:选 3 个模型,定义切换条件和优先级
- 故意用超过配额的并发数去压测,观察并记录系统的实际降级行为
10. RAG 链路压测
RAG(Retrieval-Augmented Generation)的链路比纯大模型调用长得多,每个环节都可能成为瓶颈。
10.1 完整 RAG 管道
用户提问 → Query 预处理 → Embedding 向量化 → 向量数据库检索 → Rerank 重排序 → 拼接 Prompt → 调大模型生成 → 后处理返回
10.2 分段压测方法
📖 核心原则:先分段找瓶颈,再整体验证
把 RAG 链路拆成独立环节,分别压测每个环节的并发能力,找出最弱的那个。然后做端到端整体压测验证。
环节 1:Embedding 模型
| 指标 | 测试方法 | 典型值 |
|---|---|---|
| 单请求延迟 | 固定文本长度,单次调用 | 50~200ms |
| 最大 QPS | 逐步加压到错误率 > 1% | 50~500 QPS |
| 批量处理能力 | batch_size=32 vs 1 对比 | 批量提升 5~10x |
环节 2:向量数据库检索
| 数据规模 | 典型检索延迟 | 说明 |
|---|---|---|
| 1 万条文档 | < 5ms | 几乎无瓶颈 |
| 10 万条文档 | 5~20ms | 性能良好 |
| 100 万条文档 | 20~100ms | 需要关注索引策略 |
| 1000 万条文档 | 100~500ms | 必须用 ANN 索引(HNSW/IVF) |
环节 3:Rerank 模型
Rerank 模型的并发能力测试:
→ 固定 top_k=20(每次重排 20 个候选文档)
→ 逐步加压:1→5→10→20→50 并发
→ 观察延迟变化和错误率
典型结果(bge-reranker-large, 单 GPU):
1 并发:120ms
5 并发:150ms
10 并发:250ms
20 并发:500ms
50 并发:1200ms(GPU 利用率 95%)→ 瓶颈10.3 Embedding 模型的性能瓶颈分析
| 瓶颈场景 | 表现 | 解决方案 |
|---|---|---|
| CPU 推理太慢 | 延迟 > 500ms,CPU 打满 | 切换到 GPU 推理 或 使用 API 服务 |
| 单次处理 | QPS 上不去 | 开启批量处理 batch_size=32 |
| 第三方 API 限流 | 429 错误 | 本地部署 Embedding 模型 |
| 输入文本过长 | 延迟波动大 | 文本分块控制在 512 Token 以内 |
10.4 向量数据库在不同数据规模下的性能曲线
检索延迟 (ms)
│
500 │ ╱ Flat (暴力搜索)
│ ╱
200 │ ╱╱
│ ╱╱
100 │ ────────────────────── HNSW
50 │ ────────╱──────────────────────── IVF-PQ
20 │ ────╱───
10 │╱
│
└──────────────────────────────────────────
1万 10万 50万 100万 500万 1000万 (文档数)
结论:
• 10 万以下:任何索引都够用
• 10~100 万:必须用 ANN 索引(HNSW 推荐)
• 100 万以上:考虑 IVF-PQ + 量化,或分片部署10.5 完整 RAG 性能报告示例
TIP
RAG 性能测试报告摘要
═══════════════════════════════════════
测试条件:知识库 50 万文档,top_k=10, rerank top_k=5
并发数:10
分段延迟:
Embedding: 85ms (占比 1.4%)
向量检索: 35ms (占比 0.6%)
Rerank: 210ms (占比 3.5%)
Prompt 拼接: 5ms (占比 0.1%)
大模型生成: 5500ms (占比 91.7%)
后处理/审核: 160ms (占比 2.7%)
端到端延迟:5995ms (P50) | 7200ms (P95) | 9100ms (P99)
Token 吞吐:32 tokens/s
错误率:0.2%
瓶颈分析:
主瓶颈 → 大模型生成 (91.7%)
次瓶颈 → Rerank 模型 (3.5%, 但并发 20+ 时急剧上升)
优化建议:
1. 大模型换用更快的模型或增加 GPU
2. Rerank 改为异步流水线,不阻塞主链路
3. 考虑跳过 Rerank 用向量检索 top-5 直出课堂练习
- 拆解你的 RAG 链路,分别测量每个环节的延迟和并发上限
- 制作一个类似上面的"分段延迟"表格,找出你的瓶颈环节
- 在不同文档数量(1 万 / 10 万 / 50 万)下测试向量检索延迟
11. Tool Calling 场景
11.1 工具调用放大系数
⚠️ 核心概念:工具调用放大系数
1 个用户请求 ≠ 1 个后端请求。工具调用会产生"放大效应":
放大系数 = 大模型调用次数 + 工具调用次数
单工具调用:放大系数 = 2 (决定调工具 + 根据结果回答) + 1 (工具) = 3
双工具串行:放大系数 = 3 (思考+中间+回答) + 2 (工具) = 5
三工具串行:放大系数 = 4 + 3 = 7
三工具并行:放大系数 = 2 (思考+回答) + 3 (工具) = 5
50 并发用户 × 放大系数 5 = 250 个内部请求同时在飞11.2 各类工具调用的典型放大系数
| 场景 | 大模型调用 | 工具调用 | 放大系数 | 50 并发的内部压力 |
|---|---|---|---|---|
| 无工具(纯对话) | 1 | 0 | 1 | 50 |
| 单工具(快工具) | 2 | 1 | 3 | 150 |
| 单工具(慢工具) | 2 | 1 | 3 | 150 + 连接占用时间长 |
| 3 工具串行 | 4 | 3 | 7 | 350 |
| 3 工具并行 | 2 | 3 | 5 | 250(瞬间峰值高) |
11.3 并行工具调用的正确性验证
并行工具调用需要验证以下几点:
验证项 1:是否真的并行了
→ 对比总耗时和各工具耗时之和
→ 并行时 total ≈ max(tool1, tool2, tool3)
→ 串行时 total ≈ sum(tool1, tool2, tool3)
→ 如果 total ≈ sum,说明系统实际是串行执行(这是 Bug)
验证项 2:并行工具的资源竞争
→ 3 个工具都查数据库 → 数据库连接池是否够用?
→ 3 个工具都调外部 API → 是否会同时触发限流?
→ 50 并发 × 3 并行 = 150 个工具请求瞬间发出
验证项 3:部分工具失败时的行为
→ 工具 A 成功, 工具 B 超时, 工具 C 成功
→ 系统是等 B 超时再返回?还是先返回 A+C 的结果?
→ 最差情况:B 超时 30 秒,整个请求被拖住 30 秒课堂练习
- 列出你项目中所有工具,计算每种调用组合的放大系数
- 设计一个测试用例验证"并行工具调用是否真的并行了"
- 模拟一个并行工具中有一个超时的场景,观察系统行为
12. MCP 场景
12.1 MCP 连接池管理
MCP(Model Context Protocol)使用长连接与外部服务通信,连接池管理至关重要:
| 问题 | 现象 | 解决方案 |
|---|---|---|
| 连接池耗尽 | 新请求等待连接,TTFT 急剧上升 | 增大连接池 / 设置最大等待时间 |
| 连接泄漏 | 长时间运行后可用连接数持续减少 | 添加连接健康检查 + 自动回收 |
| 连接创建慢 | 冷启动时延迟高 | 预热连接池 / 设置最小连接数 |
12.2 多 MCP Server 的资源竞争
场景:"查 Jira 上的 Bug,然后在 Slack 通知团队,最后更新 Notion 文档"
资源竞争点:
1. 每个 MCP Server 独立连接池 → 总连接数 = N × 池大小
2. 大模型串行调用每个 MCP → 单请求占用时间长
3. 如果某个 MCP Server 慢了 → 后续 MCP 调用被阻塞
压测要点:
→ 同时压 Jira + Slack + Notion 三个 MCP Server
→ 观察哪个 Server 先成为瓶颈
→ 一个 Server 挂了,是否影响其他 Server 的调用
→ 连接池总数是否会导致系统内存不足12.3 MCP 单服务调用
场景:"帮我在 Jira 上创建一个 Bug"
链路:大模型 → MCP Jira Server → Jira API → 返回 issue ID
观察:
→ MCP 协议本身的开销(建连、序列化)
→ 并发时 MCP 连接池是否够用
→ 外部服务挂了时的处理12.4 MCP 多服务串联
场景:"把 Slack 今天的讨论总结到 Notion"
链路:MCP Slack → 大模型总结 → MCP Notion 创建页面
降级:Slack 成功但 Notion 失败 →
应告诉用户"总结已生成但 Notion 创建失败,以下是总结内容"
而不是整个请求失败,把已获取的数据也丢了课堂练习
- 列出你项目中使用的所有 MCP Server,画出调用关系图
- 测试每个 MCP Server 的连接池配置和最大并发能力
- 模拟一个 MCP Server 挂掉,观察是否影响其他 Server 的调用
13. FAQ 场景
13.1 FAQ 命中与未命中
FAQ 命中链路:搜索 FAQ 库 → 直接返回预设答案,不调大模型
→ 响应时间:期望 < 500ms
→ QPS:应该很高(几百到几千)
FAQ 未命中链路:搜索 FAQ 库 → 未命中 → 走大模型兜底
→ 响应时间:取决于大模型
→ 额外延迟:FAQ 检索时间(通常 50~200ms)13.2 FAQ 命中率对系统性能的影响分析
📐 FAQ 命中率直接影响系统整体吞吐
假设:100 QPS 总请求, 大模型接口极限 = 10 QPS
命中率 90%: FAQ 处理 90 QPS, 大模型处理 10 QPS → 刚好可以承载
命中率 80%: FAQ 处理 80 QPS, 大模型需处理 20 QPS → 超载! 需要扩容
命中率 70%: FAQ 处理 70 QPS, 大模型需处理 30 QPS → 严重超载!
结论:FAQ 命中率每下降 10%,大模型压力可能翻倍
→ FAQ 命中率本身就是最有效的"性能优化手段"| FAQ 命中率 | 大模型压力 | 系统整体 QPS 上限 | 成本 |
|---|---|---|---|
| 95% | 极低 | 很高(FAQ 决定) | 极低 |
| 80% | 中等 | 中等 | 中等 |
| 60% | 高 | 低(模型瓶颈) | 高 |
| 30% | 极高 | 极低 | 极高 |
课堂练习
- 统计你的系统目前的 FAQ 命中率
- 用混合流量(70% 命中 + 30% 未命中)做压测,观察系统表现
- 计算:如果 FAQ 命中率从当前值提升到 95%,系统能多承载多少并发?
14. 纯大模型对话场景
用例 1:短输出对话
场景:用户问简单问题,大模型直接回答,输出很短
请求示例:
POST /api/chat
{ "message": "1+1等于几", "max_tokens": 50 }
测试数据(准备 30+ 个不重复的):
"1+1等于几" "中国的首都是哪里" "water 用中文怎么说"
"今年是哪一年" "HTTP 状态码 404 是什么意思" ...
观察指标:
→ TTFT:期望 < 500ms
→ 总响应时间:期望 < 2s
→ Token 吞吐:正常 30~80 tokens/s
→ 错误率:期望 < 0.5%
判定标准:
✅ PASS: TTFT P95 < 500ms, E2E P95 < 2s, 错误率 < 0.5%
⚠️ WARN: TTFT P95 500ms~1s, 或 E2E P95 2~4s
❌ FAIL: TTFT P95 > 1s, 或 E2E P95 > 4s, 或错误率 > 1%
意义:最轻量的场景,用来建立性能基准线用例 2:长输出对话
场景:用户要求生成长篇内容
请求示例:
POST /api/chat
{ "message": "写一篇 1000 字关于人工智能发展历史的文章", "max_tokens": 2000 }
观察指标:
→ TTFT:期望 < 1s
→ 总响应时间:可能 10~30s(正常)
→ Token 吞吐:重点看并发增加时是否大幅下降
→ GPU/内存占用(自部署模型)
判定标准:
✅ PASS: TTFT P95 < 1s, Token 吞吐 > 20 tokens/s
⚠️ WARN: Token 吞吐 10~20 tokens/s
❌ FAIL: TTFT P95 > 3s, 或 Token 吞吐 < 10 tokens/s
意义:长输出占用推理资源时间最久,最容易把系统压垮用例 3:多轮对话(上下文递增)
场景:用户连续对话多轮,每轮带着所有历史
随着轮数增加,发送给大模型的 Token 越来越大
测试数据(准备 10~20 组对话链):
对话链A:数据库设计→加字段→加索引→写SQL→优化查询(10轮)
对话链B:写营销文案→改语气→缩短→加标题→翻译成英文(8轮)
预期现象:
第1轮 TTFT 300ms → 第5轮 800ms → 第10轮 2s → 第15轮 5s+
判定标准:
✅ PASS: 第10轮 TTFT < 2s, Token 吞吐保持 > 15 tokens/s
⚠️ WARN: 第10轮 TTFT 2~5s
❌ FAIL: 第10轮 TTFT > 5s, 或系统 OOM
观察重点:
→ TTFT 和 Token 吞吐随轮数的退化曲线
→ 上下文窗口接近限制时的系统行为(截断 vs 报错 vs 自动总结)
意义:真实用户不会只聊一轮。上下文越长推理越慢,但单轮测试发现不了15. 工具调用场景
用例 4:单工具调用 — 快工具
完整链路:
用户发消息 → 大模型思考(决定调工具)→ 调用工具 → 工具返回
→ 大模型根据结果生成回答 = 2 次大模型推理 + 1 次工具调用
测试数据(准备 30+,覆盖不同工具):
"查订单 ORD-2024-001 的状态" → get_order_status
"我的账户余额是多少" → get_account_balance
"查一下北京今天的天气" → get_weather
观察指标:
→ 总 E2E 延迟 vs 纯对话的延迟差
→ 工具调用的延迟(应 < 200ms)
→ 大模型是否正确识别需要调用工具
判定标准:
✅ PASS: E2E P95 < 5s, 工具调用成功率 > 99%
❌ FAIL: E2E P95 > 10s, 或工具未被正确调用
观察:有工具调用 vs 无工具调用的总响应时间差多少用例 5:单工具调用 — 慢工具
场景:工具本身响应很慢(复杂数据库查询、第三方 API)
示例:
"分析一下上个月所有用户的活跃趋势"
→ 触发复杂数据库聚合查询 → 花 5 秒
观察指标:
→ 工具超时是否有合理的 timeout 控制
→ 慢工具是否阻塞了其他请求
→ 长连接占用情况
判定标准:
✅ PASS: 有超时控制(如 10s),超时后有合理降级
❌ FAIL: 无超时控制,连接被无限占用
意义:
快工具 100ms 返回,不影响整体
慢工具 5s 返回,10 并发 × 5s = 系统同时 hold 住 10 个长连接用例 6:多工具串行调用
场景:"帮我查客户张三的订单,然后把物流信息发邮件通知他"
背后发生的事:
1. 大模型 → get_customer("张三") → {id:"C001",email:"zhang@xx.com"}
2. 大模型 → get_latest_order("C001") → {order_id:"ORD-99",tracking:"SF123"}
3. 大模型 → send_email(to,subject,body) → {success:true}
4. 大模型生成最终回答
总计:3 次大模型推理 + 3 次工具调用 = 来回 6 次
观察指标:
→ 串行调用的总耗时(各步骤累加)
→ 中间步骤失败时的行为
→ 放大系数验证:50 并发 → 内部是否产生 300 次调用
判定标准:
✅ PASS: E2E P95 < 15s, 各步骤间正确传递参数
❌ FAIL: E2E P95 > 30s, 或中间步骤失败导致整个请求崩溃
意义:
50 个并发用户 = 300 次内部调用,远超表面的 50 并发用例 7:多工具并行调用
场景:"对比北京、上海、广州今天的天气"
背后:并行调用 3 次 get_weather
总耗时 = 大模型思考 + max(3s,2s,4s) + 大模型生成 ≈ 6s
(不是 3+2+4=9s,因为是并行的)
观察指标:
→ 是否真的并行了(total ≈ max 还是 ≈ sum)
→ 并行调用对下游服务的瞬间压力
→ 部分工具失败的处理
判定标准:
✅ PASS: 实际耗时 ≈ max(各工具耗时) + 模型推理时间
❌ FAIL: 实际耗时 ≈ sum(各工具耗时)(说明没有真正并行)
观察:
→ 系统是否真的并行了(还是串行了 = bug)
→ 50 并发 × 3 并行工具 = 瞬间 150 个工具请求课堂练习
- 用你的 AI 系统分别执行用例 4~7,记录 E2E 延迟和放大系数
- 验证并行工具调用是否真的是并行执行的(对比 total 和 sum)
- 模拟多工具串行中第 2 步失败的场景,观察系统是否会继续执行第 3 步
16. 降级场景(核心)
⚠️ 降级测试是 AI 性能测试中最重要的部分
大模型系统的故障不是"会不会发生",而是"什么时候发生"。限流、超时、服务不可用是常态。降级策略的好坏直接决定了用户体验的下限。
用例 8:工具超时
模拟:让工具后端人为延迟(sleep 30s / 60s)
期望的降级行为(按优先级):
方案 A — 超时后切换备用工具/缓存数据
方案 B — 让大模型用自身知识回答,注明"未获取到最新数据"
方案 C — 友好提示"查询服务响应较慢,请稍后再试"
❌ 最差情况:
→ 用户等 60 秒后收到 500 错误
→ 页面一直转圈然后什么都没有
→ 整个系统因等待而卡死,其他用户也受影响
判定标准:
✅ PASS: 超时 < 10s, 有友好降级回复, 不影响其他请求
❌ FAIL: 超时 > 30s, 或无降级, 或影响其他用户用例 9:工具返回错误
模拟:让工具返回 HTTP 500 / 格式错误 / 业务错误
期望:
业务错误(客户不存在)→ 正常告知用户
系统错误(500)→ 友好提示,不暴露技术细节
格式错误(JSON 解析失败)→ 捕获异常,不让整个对话崩溃
判定标准:
✅ PASS: 所有错误类型都有友好处理, 不暴露堆栈信息
❌ FAIL: 用户看到 "Internal Server Error" 或堆栈信息用例 10:工具限流(429)
场景:高并发下第三方 API 触发限流
降级方案:
A — 排队重试 + "预计等待 10 秒..."
B — 切换备用服务(高德限流 → 切百度)
C — 返回缓存结果(不是最新但至少有回答)
判定标准:
✅ PASS: 有至少一种降级方案, 用户等待 < 15s
❌ FAIL: 直接返回 429 给用户, 无任何降级用例 11:大模型自身限流
降级方案(多模型切换 — 最推荐):
GPT-4o 限流 → 自动切 GPT-4o-mini
GPT-4o-mini 也限流 → 切豆包
豆包也限流 → 切自研小模型
→ 每层效果差一点,但至少有回答
测试要点:
→ 切换延迟多少?对用户透明吗?
→ 所有模型都限流后的最终行为
→ 限流恢复后是否自动切回主模型
判定标准:
✅ PASS: 自动切换延迟 < 2s, 所有模型限流后有兜底
❌ FAIL: 无切换机制, 或切换后报错用例 12:MCP 单服务调用
场景:"帮我在 Jira 上创建一个 Bug"
链路:大模型 → MCP Jira Server → Jira API → 返回 issue ID
观察:
→ MCP 协议本身的开销(建连、序列化)
→ 并发时 MCP 连接池是否够用
→ 外部服务挂了时的处理
判定标准:
✅ PASS: MCP 开销 < 100ms, 连接池不耗尽
❌ FAIL: MCP 连接泄漏, 或外部服务挂了导致整个系统阻塞用例 13:MCP 多服务串联
场景:"把 Slack 今天的讨论总结到 Notion"
链路:MCP Slack → 大模型总结 → MCP Notion 创建页面
降级:Slack 成功但 Notion 失败 →
应告诉用户"总结已生成但 Notion 创建失败,以下是总结内容"
而不是整个请求失败,把已获取的数据也丢了
判定标准:
✅ PASS: 部分失败时保留已获取的数据
❌ FAIL: 一个 MCP 失败导致整个请求数据全丢课堂练习
- 对你的系统执行用例 8~13 的降级测试,记录每个场景的实际行为
- 与产品经理确认:每个降级场景的预期行为是什么?
- 制作一张"降级策略矩阵表":故障类型 × 降级方案 × 用户体验
17. 综合混合场景
用例 14:RAG 检索 + 大模型生成
链路:用户提问 → Embedding → 向量检索 → Rerank → 拼接Prompt → 大模型生成
观察指标(分段计时):
→ Embedding 耗时:通常 50~200ms
→ 向量检索耗时:通常 10~100ms
→ Rerank 耗时:通常 100~500ms
→ 大模型生成:通常 2~10s
→ 总端到端耗时
重点看并发增加时哪个环节先扛不住
降级场景:
→ 向量数据库挂了 → 走 FAQ 关键词匹配兜底
→ 检索不到相关文档 → 大模型用自身知识,注明"未找到文档"
→ Rerank 超时 → 跳过 Rerank,直接用向量检索的 top 结果
判定标准:
✅ PASS: E2E P95 < 10s, 各环节降级正常工作
❌ FAIL: 某环节故障导致整个链路崩溃用例 15:FAQ 命中
链路:搜索 FAQ 库 → 直接返回预设答案,不调大模型
观察:
→ 响应时间:期望 < 500ms
→ QPS:应该很高(几百到几千)
判定标准:
✅ PASS: P95 < 500ms, QPS > 100
❌ FAIL: P95 > 1s, 或 QPS < 50
意义:FAQ 命中率越高,系统整体压力越小 → FAQ 命中率本身就是性能优化手段用例 16:FAQ 未命中 → 大模型兜底
测试:混合 70% 命中 + 30% 未命中
观察指标:
→ 命中请求的 P95 延迟(应保持 < 500ms)
→ 未命中请求的 P95 延迟(取决于大模型)
→ 大模型兜底是否影响了 FAQ 命中请求的延迟
判定标准:
✅ PASS: FAQ 命中延迟不受未命中请求影响(隔离性好)
❌ FAIL: 大模型压力影响了 FAQ 命中请求的响应用例 17:真实流量模拟
把所有场景按真实比例混合:
FAQ 命中 40%
FAQ 未命中 10%
RAG 检索 15%
纯大模型短对话 10%
纯大模型长对话 5%
多轮对话 5%
单工具调用 10%
多工具调用 3%
MCP 调用 2%
判定标准:
✅ PASS: 整体错误率 < 0.5%, 各场景 P95 达标
⚠️ WARN: 整体错误率 0.5~2%
❌ FAIL: 整体错误率 > 2%
这是最终的"大考"——最接近上线后的真实情况用例 18:降级策略综合验证
步骤:
1. 先跑正常混合流量 5 分钟,记录基准
2. 关掉向量数据库 → 观察 RAG 降级
3. 恢复,跑 2 分钟
4. 让大模型 API 返回 429 → 观察模型切换
5. 恢复,跑 2 分钟
6. 让工具接口超时 → 观察是否影响其他请求
7. 同时关掉大模型 + 向量数据库 → 系统是否优雅降级
观察:
→ 故障期间的错误率
→ 非相关请求是否正常(隔离性)
→ 故障恢复后多久回到正常(恢复速度)
判定标准:
✅ PASS: 单点故障不导致全面崩溃, 恢复时间 < 30s
❌ FAIL: 单点故障导致雪崩, 或恢复时间 > 5min用例 19:WebSocket 长连接场景
场景:客户端通过 WebSocket 建立长连接进行多轮 AI 对话
链路:WebSocket 建连 → 发送消息 → 服务端处理 → 流式推送回答 → 保持连接
压测要点:
→ 最大 WebSocket 连接数(系统能维持多少条长连接)
→ 连接建立速率(每秒能建立多少新连接)
→ 长时间保持连接后的内存增长(1000 连接 × 24 小时)
→ 连接断开后的重连行为
→ 服务端重启后的连接恢复
测试步骤:
1. 建立 100 条 WebSocket 连接
2. 每条连接每分钟发送一次消息
3. 逐步增加到 500 → 1000 → 2000 连接
4. 持续运行 1 小时,观察内存趋势
5. 杀掉服务端进程,观察客户端重连
判定标准:
✅ PASS: 支持 1000+ 长连接, 内存增长线性且可预测
❌ FAIL: 连接数 < 500 即出现超时, 或内存指数增长用例 20:多模型 A/B 路由场景
场景:对 50% 用户使用模型 A,50% 使用模型 B,进行效果对比
压测要点:
→ 路由逻辑是否正确(流量分配比例准确吗?)
→ 两个模型的独立指标对比(TTFT/吞吐/错误率)
→ 一个模型出问题是否影响另一个模型的流量
→ 动态调整权重时的切换延迟
测试步骤:
1. 配置 A/B 路由规则(50/50)
2. 发起 1000 个请求,统计实际分配比例
3. 让模型 A 返回 429,观察是否自动把流量切到模型 B
4. 恢复模型 A,观察流量是否重新平衡
判定标准:
✅ PASS: 实际分配偏差 < 5%, 故障时流量自动切换
❌ FAIL: 分配偏差 > 10%, 或一个模型故障影响另一个课堂练习
- 设计一个最接近你项目真实流量的混合场景,确定各场景的占比
- 按用例 18 的步骤做一次完整的降级策略综合验证
- 如果你的系统使用 WebSocket,执行用例 19 的长连接测试
18. AI 压测工具选型
18.1 工具能力对比
| 能力 | JMeter | k6 | Locust | 自写脚本 |
|---|---|---|---|---|
| 普通 HTTP | 很好 | 很好 | 很好 | 完全控制 |
| SSE 流式 | 不支持 | 需手动 | 需手动 | 完全控制 |
| 计算 TTFT | 做不了 | 需手动 | 需手动 | 精确控制 |
| Token 吞吐 | 做不了 | 需自己算 | 需自己算 | 精确控制 |
| 学习成本 | 高 | 中 | 低 | 取决于水平 |
| 资源消耗 | 高(Java) | 低(Go) | 中 | 取决于实现 |
| 分布式支持 | 内置 | 内置 | 内置 | 需自行实现 |
| Web UI | 有 | Cloud | 内置 | 无 |
18.2 为什么推荐 Locust
✅ 大模型场景推荐:Locust
| 理由 | 说明 |
|---|---|
| Python 生态 | AI 生态(openai/httpx/tiktoken)都是 Python 库,直接引用 |
| 脚本灵活 | 纯 Python 代码,SSE 解析、TTFT 计算、Token 统计可完全自定义 |
| 并发模型 | 基于 gevent 协程,轻量;大模型并发 10~100 绰绰有余 |
| Web UI | 内置实时监控面板,无需额外搭建 |
| 分布式 | master-worker 模式,扩展方便 |
| 社区活跃 | 文档完善,教程多 |
18.3 Locust + AI 的示例框架
from locust import HttpUser, task, between, events
import json, time
class AIUser(HttpUser):
wait_time = between(2, 5)
host = "https://your-api.com"
@task(3)
def short_chat(self):
"""短对话 — 权重 3"""
self._stream_chat("1+1等于几", max_tokens=50)
@task(1)
def long_chat(self):
"""长对话 — 权重 1"""
self._stream_chat("写一首关于春天的诗", max_tokens=500)
def _stream_chat(self, prompt, max_tokens):
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"stream": True,
"max_tokens": max_tokens
}
t_start = time.perf_counter()
t_first_token = None
token_count = 0
with self.client.post(
"/v1/chat/completions",
json=payload,
stream=True,
catch_response=True,
headers={"Authorization": "Bearer sk-xxx"}
) as resp:
if resp.status_code != 200:
resp.failure(f"HTTP {resp.status_code}")
return
for line in resp.iter_lines():
if not line:
continue
line = line.decode("utf-8")
if line.startswith("data: ") and line[6:] != "[DONE]":
chunk = json.loads(line[6:])
content = chunk["choices"][0]["delta"].get("content", "")
if content:
if t_first_token is None:
t_first_token = time.perf_counter()
token_count += 1
t_end = time.perf_counter()
if t_first_token:
ttft_ms = (t_first_token - t_start) * 1000
total_ms = (t_end - t_start) * 1000
tps = token_count / (t_end - t_first_token) if t_end > t_first_token else 0
events.request.fire(
request_type="TTFT",
name=f"ttft_{prompt[:10]}",
response_time=ttft_ms,
response_length=0,
exception=None,
context={}
)
resp.success()课堂练习
- 安装 Locust(
pip install locust),用上面的框架代码测试你的 AI 接口 - 对比:用纯 Python 脚本 vs Locust 做压测,各自的优缺点是什么?
- 扩展 Locust 脚本,添加 FAQ、RAG、Tool Calling 等场景(按比例分配权重)
19. 流式接口压测脚本
以下是一个可以直接复制使用的完整 Python 压测脚本。它使用 asyncio + httpx 实现异步并发,自动计算 TTFT、Token 吞吐、端到端延迟等指标,并输出统计结果。
19.1 完整脚本模板
#!/usr/bin/env python3
"""AI 流式接口压测脚本 — 可直接复制使用"""
import asyncio
import httpx
import json
import time
import statistics
from dataclasses import dataclass, field
from typing import Optional
# 配置 ───────────────────────────────────────────
API_URL = "https://api.openai.com/v1/chat/completions"
API_KEY = "sk-your-key-here"
MODEL = "gpt-4o-mini"
MAX_TOKENS = 100
CONCURRENCY = 10 # 并发数
TOTAL_REQUESTS = 100 # 总请求数
TIMEOUT = 60 # 单请求超时(秒)
PROMPTS = [
"1+1等于几", "中国的首都是哪里", "water 用中文怎么说",
"HTTP 状态码 404 是什么意思", "今年是哪一年",
"什么是人工智能", "Python 的创始人是谁", "地球到月球的距离",
"写一句鼓励的话", "什么是 REST API",
]
# 数据结构 ────────────────────────────────────────
@dataclass
class RequestResult:
success: bool
ttft_ms: Optional[float] = None
total_ms: Optional[float] = None
token_count: int = 0
tokens_per_sec: float = 0
error: str = ""
status_code: int = 0
# 核心:单次流式请求 ──────────────────────────────
async def stream_request(client: httpx.AsyncClient, prompt: str) -> RequestResult:
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
payload = {
"model": MODEL,
"messages": [{"role": "user", "content": prompt}],
"stream": True,
"max_tokens": MAX_TOKENS
}
t_start = time.perf_counter()
t_first_token = None
token_count = 0
try:
async with client.stream("POST", API_URL, json=payload, headers=headers) as resp:
if resp.status_code != 200:
body = await resp.aread()
return RequestResult(success=False, status_code=resp.status_code,
error=body.decode()[:200])
async for line in resp.aiter_lines():
if not line.startswith("data: "):
continue
data_str = line[6:]
if data_str == "[DONE]":
break
try:
chunk = json.loads(data_str)
content = chunk["choices"][0]["delta"].get("content", "")
if content:
if t_first_token is None:
t_first_token = time.perf_counter()
token_count += 1
except (json.JSONDecodeError, KeyError, IndexError):
continue
t_end = time.perf_counter()
ttft = (t_first_token - t_start) * 1000 if t_first_token else None
total = (t_end - t_start) * 1000
gen_time = t_end - t_first_token if t_first_token else 0
tps = token_count / gen_time if gen_time > 0 else 0
return RequestResult(success=True, ttft_ms=ttft, total_ms=total,
token_count=token_count, tokens_per_sec=tps,
status_code=200)
except httpx.TimeoutException:
return RequestResult(success=False, error="Timeout", status_code=0)
except Exception as e:
return RequestResult(success=False, error=str(e)[:200], status_code=0)
# 并发控制 ────────────────────────────────────────
async def worker(sem: asyncio.Semaphore, client: httpx.AsyncClient,
prompt: str, results: list):
async with sem:
result = await stream_request(client, prompt)
results.append(result)
async def run_benchmark():
sem = asyncio.Semaphore(CONCURRENCY)
results: list[RequestResult] = []
prompts = [PROMPTS[i % len(PROMPTS)] for i in range(TOTAL_REQUESTS)]
print(f"开始压测: {CONCURRENCY} 并发 × {TOTAL_REQUESTS} 请求")
print(f"模型: {MODEL}, max_tokens: {MAX_TOKENS}")
print("=" * 60)
t_start = time.perf_counter()
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
tasks = [worker(sem, client, p, results) for p in prompts]
await asyncio.gather(*tasks)
t_total = time.perf_counter() - t_start
# ─── 统计 ─────────────────────────────────────────
success = [r for r in results if r.success]
failed = [r for r in results if not r.success]
ttfts = [r.ttft_ms for r in success if r.ttft_ms is not None]
totals = [r.total_ms for r in success if r.total_ms is not None]
tps_list = [r.tokens_per_sec for r in success if r.tokens_per_sec > 0]
token_counts = [r.token_count for r in success]
def percentile(data, p):
if not data:
return 0
sorted_d = sorted(data)
idx = int(len(sorted_d) * p / 100)
return sorted_d[min(idx, len(sorted_d) - 1)]
print(f"\n{'=' * 60}")
print(f"总耗时: {t_total:.1f}s")
print(f"成功/失败: {len(success)}/{len(failed)}")
print(f"错误率: {len(failed)/len(results)*100:.1f}%")
print(f"实际 RPS: {len(results)/t_total:.1f}")
print(f"\n--- TTFT (ms) ---")
if ttfts:
print(f" P50: {percentile(ttfts, 50):.0f}")
print(f" P95: {percentile(ttfts, 95):.0f}")
print(f" P99: {percentile(ttfts, 99):.0f}")
print(f" Avg: {statistics.mean(ttfts):.0f}")
print(f"\n--- 端到端延迟 (ms) ---")
if totals:
print(f" P50: {percentile(totals, 50):.0f}")
print(f" P95: {percentile(totals, 95):.0f}")
print(f" P99: {percentile(totals, 99):.0f}")
print(f"\n--- Token 吞吐 (tokens/s) ---")
if tps_list:
print(f" P50: {percentile(tps_list, 50):.1f}")
print(f" Avg: {statistics.mean(tps_list):.1f}")
print(f"\n--- Token 总量 ---")
if token_counts:
print(f" Total: {sum(token_counts)}")
print(f" Avg/req: {statistics.mean(token_counts):.0f}")
if failed:
print(f"\n--- 错误详情 ---")
error_types = {}
for r in failed:
key = r.error[:50] if r.error else f"HTTP {r.status_code}"
error_types[key] = error_types.get(key, 0) + 1
for err, cnt in sorted(error_types.items(), key=lambda x: -x[1]):
print(f" {err}: {cnt}次")
if __name__ == "__main__":
asyncio.run(run_benchmark())示例
运行结果示例:
开始压测: 10 并发 × 100 请求
模型: gpt-4o-mini, max_tokens: 100
============================================================
============================================================
总耗时: 48.3s
成功/失败: 98/2
错误率: 2.0%
实际 RPS: 2.1
--- TTFT (ms) ---
P50: 342
P95: 856
P99: 1203
Avg: 412
--- 端到端延迟 (ms) ---
P50: 2156
P95: 4523
P99: 6180
--- Token 吞吐 (tokens/s) ---
P50: 28.5
Avg: 31.2
--- Token 总量 ---
Total: 5840
Avg/req: 60
--- 错误详情 ---
Timeout: 2次课堂练习
- 复制上面的脚本,修改配置后在你的 AI 接口上运行
- 分别用 CONCURRENCY = 1, 5, 10, 20 跑一遍,对比结果
- 扩展脚本:将结果输出为 CSV 文件,方便用 Excel 做图表分析
20. 费用控制技巧
20.1 压测预算计算器
📐 压测预算公式
步骤 1: 计算总请求数
总请求数 = 并发数 × 持续时间(秒) / 平均单请求时间(秒)
步骤 2: 计算 Token 消耗
总 Input Tokens = 总请求数 × 平均 Input Tokens
总 Output Tokens = 总请求数 × 平均 Output Tokens
步骤 3: 计算费用
Input 费用 = 总 Input Tokens / 1,000,000 × Input 单价
Output 费用 = 总 Output Tokens / 1,000,000 × Output 单价
总费用 = Input 费用 + Output 费用不同模型的压测费用快速估算表
| 配置 | GPT-4o | GPT-4o-mini | 豆包 Pro | 豆包 Lite |
|---|---|---|---|---|
| 10 并发 × 10min (~200 请求) | ~¥10 | ~¥0.40 | ~¥1.50 | ~¥0.10 |
| 20 并发 × 30min (~1200 请求) | ~¥60 | ~¥2.40 | ~¥9.00 | ~¥0.60 |
| 50 并发 × 1h (~6000 请求) | ~¥300 | ~¥12 | ~¥45 | ~¥3.00 |
(假设平均 Input 500 Tokens, Output 200 Tokens)
20.2 各阶段的费用控制策略
| 阶段 | 策略 | 说明 |
|---|---|---|
| 脚本开发期 | Mock Server | 用本地 Mock 服务器模拟 SSE 响应,零成本调试脚本 |
| 基准测试期 | 最便宜模型 + 短 Prompt | GPT-4o-mini + "回复OK" + max_tokens=10 |
| 负载测试期 | 短周期 + 阶梯 | 每级 3 分钟足够看趋势,不需要 10 分钟 |
| 稳定性测试期 | 降低并发 + 延长时间 | 5 并发跑 2 小时,比 50 并发跑 2 小时便宜 10 倍 |
| 最终验证期 | 真实模型 + 真实 Prompt | 只在最后一轮使用生产配置,控制总时间 < 15 分钟 |
💡 Mock Server 快速搭建
# 用 Python 快速搭建一个 SSE Mock Server
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio, json
app = FastAPI()
@app.post("/v1/chat/completions")
async def chat(request: dict):
async def generate():
for char in "这是一个模拟回复用于测试":
chunk = {"choices": [{"delta": {"content": char}, "finish_reason": None}]}
yield f"data: {json.dumps(chunk)}\n\n"
await asyncio.sleep(0.05) # 模拟 20 tokens/s
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
# uvicorn mock_server:app --port 8000课堂练习
- 用上面的公式计算你下一次压测的预估费用
- 搭建一个 Mock Server,用压测脚本对接调试
- 对比 Mock Server 和真实 API 的压测结果,哪些指标可以从 Mock 得到,哪些必须用真实 API?
21. GPU 监控专项
21.1 nvidia-smi 常用命令
# 查看 GPU 基本信息
nvidia-smi
# 持续监控(每 1 秒刷新)
nvidia-smi -l 1
# 以 CSV 格式输出指定指标(适合脚本解析)
nvidia-smi --query-gpu=timestamp,name,utilization.gpu,utilization.memory,memory.used,memory.total,temperature.gpu,power.draw \
--format=csv -l 1
# 查看进程级 GPU 使用
nvidia-smi pmon -s um -d 1
# 查看显存使用详情
nvidia-smi --query-gpu=memory.used,memory.free,memory.total --format=csv21.2 关键 GPU 指标解读
| 指标 | 含义 | 健康范围 | 异常表现 |
|---|---|---|---|
| GPU Utilization | GPU 计算核心利用率 | 60~90% | > 95% 说明是瓶颈;< 30% 说明没充分利用 |
| Memory Used | 显存占用 | < 90% | > 95% 可能 OOM,触发 swap 极慢 |
| Temperature | GPU 温度 | < 80°C | > 85°C 开始降频,> 90°C 风险 |
| Power Draw | 功耗 | 取决于型号 | 接近 TDP 说明满载 |
| SM Clock | SM 时钟频率 | 接近 Boost 频率 | 低于基础频率说明在降频 |
21.3 DCGM Exporter + Prometheus + Grafana
# 部署 DCGM Exporter(Docker 方式)
docker run -d --gpus all --rm \
-p 9400:9400 \
nvcr.io/nvidia/k8s/dcgm-exporter:3.3.0-3.2.0-ubuntu22.04
# Prometheus 配置 (prometheus.yml)
scrape_configs:
- job_name: 'gpu'
static_configs:
- targets: ['localhost:9400']
scrape_interval: 5s
# 常用 DCGM 指标
DCGM_FI_DEV_GPU_UTIL # GPU 利用率
DCGM_FI_DEV_FB_USED # 已用显存 (MB)
DCGM_FI_DEV_FB_FREE # 可用显存 (MB)
DCGM_FI_DEV_GPU_TEMP # 温度
DCGM_FI_DEV_POWER_USAGE # 功耗
DCGM_FI_DEV_SM_CLOCK # SM 时钟
DCGM_FI_DEV_MEM_CLOCK # 显存时钟
DCGM_FI_DEV_PCIE_TX_THROUGHPUT # PCIe 发送吞吐
DCGM_FI_DEV_PCIE_RX_THROUGHPUT # PCIe 接收吞吐21.4 如何判断 GPU 是否是瓶颈
| 现象 | GPU 利用率 | 显存 | 判断 |
|---|---|---|---|
| TTFT 随并发线性增长 | > 90% | 稳定 | GPU 计算是瓶颈 → 需要更多 GPU 或量化 |
| 突然 OOM 崩溃 | 任意 | 接近 100% | 显存不足 → 减小 batch size 或量化 |
| TTFT 增长但 GPU 不高 | < 50% | 稳定 | 瓶颈不在 GPU → 查推理框架排队/CPU/网络 |
| Token 吞吐下降 | > 85% | 增长 | 连续批处理 batch 变大 → 正常现象 |
| GPU 温度过高降频 | 波动 | 稳定 | 散热问题 → 检查风扇/机房温度 |
示例
压测期间的 GPU 监控脚本:
#!/bin/bash
# gpu_monitor.sh — 压测期间持续记录 GPU 指标到 CSV
OUTPUT="gpu_metrics_$(date +%Y%m%d_%H%M%S).csv"
echo "timestamp,gpu_util,mem_used,mem_total,temp,power" > $OUTPUT
while true; do
nvidia-smi --query-gpu=timestamp,utilization.gpu,memory.used,memory.total,temperature.gpu,power.draw \
--format=csv,noheader,nounits >> $OUTPUT
sleep 1
done课堂练习
- 运行
nvidia-smi -l 1的同时执行压测,观察 GPU 指标变化 - 如果条件允许,部署 DCGM Exporter + Prometheus + Grafana 搭建 GPU 监控大盘
- 在压测过程中,判断你的系统瓶颈是否在 GPU(参考上面的判断表)
22. AI 性能报告模板
22.1 报告结构
TIP
1. 测试概述
→ 测试目的、时间、环境配置、系统版本
2. 测试场景
→ 场景列表、并发数、持续时间、流量分配比例
3. 测试结果数据
→ TPS、RT(P50/P90/P95/P99)、错误率
→ 大模型额外:TTFT、Token 吞吐
→ 服务端资源使用(CPU、内存、GPU)
→ 用图表展示
4. 性能拐点
→ 多少并发时性能开始下降
→ 多少并发时开始出错
→ 系统最大承载能力
5. 瓶颈分析
→ 瓶颈在哪、证据是什么
6. 降级测试结果
→ 故障注入后的表现
7. 结论和建议
→ 是否满足上线要求
→ 优化方向 / 扩容方案
8. 风险提示
→ 未覆盖的场景、已知风险点22.2 AI 专属指标的展示方式
除了传统的 TPS/RT/错误率,AI 性能报告需要额外展示以下指标:
| 指标 | 展示方式 | 重点展示内容 |
|---|---|---|
| TTFT | 折线图 + 表格 | P50/P95/P99 × 不同并发级别 |
| Token 吞吐 | 折线图 | 单请求吞吐 + 系统总吞吐 × 不同并发 |
| GPU 利用率 | 时序图 | 与并发数叠加展示 |
| GPU 显存 | 时序图 | 显存增长趋势,是否有泄漏 |
| Rate Limit | 饼图/统计 | 429 占比 + 触发时的并发数 |
| 降级命中 | 事件时间线 | 降级触发时间、降级方案、恢复时间 |
| Token 费用 | 汇总表 | 总消耗 Token 数 + 总费用 |
22.3 降级测试结论的写法
✅ 降级测试结论示例(好的写法)
降级场景 1:大模型 API 限流 (429)
触发条件:并发 > 15 时触发
降级方案:自动切换 GPT-4o → GPT-4o-mini → 豆包 Pro
切换延迟:< 500ms
用户感知:回答质量略降,响应速度略快
恢复行为:限流解除后 30s 内自动切回主模型
结论:✅ 降级机制有效,用户体验损失可接受
降级场景 2:向量数据库不可用
触发条件:向量数据库连接失败
降级方案:跳过 RAG,直接用大模型回答 + 提示"未引用文档"
切换延迟:< 200ms(超时检测)
用户感知:回答可能不够精确,但有明确提示
结论:✅ 降级机制有效
降级场景 3:所有模型均不可用
触发条件:所有 API 返回 429/500
降级方案:返回预设兜底回复 "系统繁忙,请稍后重试"
用户感知:无 AI 能力,但不报错
结论:⚠️ 需要优化,兜底回复太简单,建议区分不同问题类型22.4 完整报告范例
TIP
═══════════════════════════════════════════════
AI 性能测试报告 — XX 智能客服系统 v2.3
═══════════════════════════════════════════════
1. 测试概述
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
测试目的:验证系统在目标负载下的性能表现
测试时间:2025-01-15 14:00~18:00
测试环境:
应用服务器:4C8G × 3 台
GPU 服务器:A100-80GB × 1(自部署 Qwen-14B)
向量数据库:Milvus(50 万文档)
第三方 API:GPT-4o-mini(Tier 3, 5000 RPM)
测试工具:Python + httpx + asyncio 自研脚本
2. 测试场景
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
场景 占比 并发数 持续时间
FAQ 命中 40% 20 5min/级
FAQ 未命中 10% 5 5min/级
RAG 检索 15% 8 5min/级
纯对话(短) 10% 5 5min/级
纯对话(长) 5% 3 5min/级
Tool Calling 15% 8 5min/级
MCP 调用 5% 3 5min/级
阶梯加压:总并发 10 → 20 → 30 → 50 → 80
3. 测试结果
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
3.1 TTFT (ms)
并发 P50 P95 P99
10 320 580 850
20 410 720 1100
30 560 1200 2100
50 1200 3500 5600 ← 拐点
80 3800 8500 12000
3.2 Token 吞吐 (tokens/s - 单请求)
并发 P50 Avg
10 35 38
20 32 34
30 28 30
50 18 20 ← 明显下降
80 8 10 ← 不可接受
3.3 各场景端到端延迟 P95 (ms)
并发 FAQ命中 RAG 纯对话 Tool Call
10 120 3200 1800 4500
20 150 4100 2300 5800
30 180 5500 3100 7200
50 250 8200 5200 12000
80 400 15000 9800 20000+
3.4 错误率
并发 总错误率 429率 超时率
10 0.1% 0% 0.1%
20 0.3% 0% 0.3%
30 0.5% 0% 0.5%
50 2.1% 0.5% 1.6%
80 8.5% 3.2% 5.3%
3.5 GPU 指标(自部署 Qwen-14B)
并发 GPU利用率 显存 温度
10 45% 32GB 62°C
20 72% 38GB 68°C
30 88% 45GB 74°C
50 97% 55GB 79°C
80 99% 62GB 82°C ← 接近降频
4. 性能拐点
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
→ 30 并发:TTFT P95 突破 1s(用户感知明显)
→ 50 并发:Token 吞吐降到 20 tokens/s(体验变差)
→ 50 并发:错误率突破 2%(不可接受)
→ 最佳并发数:30(TTFT 合理 + 吞吐合理 + 错误率低)
5. 瓶颈分析
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
主瓶颈:自部署模型 GPU 显存(80GB 已用 55/62GB)
次瓶颈:Rerank 模型 CPU 推理(共享 GPU 时竞争)
应用层:CPU 40%, 内存 60% → 不是瓶颈
数据库:向量检索 P95 < 100ms → 不是瓶颈
6. 降级测试结果(详见 22.3 降级结论写法)
✅ 大模型限流降级:有效
✅ 向量数据库降级:有效
⚠️ 全部不可用降级:需优化
7. 结论和建议
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
结论:系统可支撑 30 并发用户,满足当前上线要求(预期 20 并发)
优化建议:
短期:Qwen-14B 开启 INT8 量化(预计并发提升至 50)
中期:Rerank 换用 GPU 推理 / 异步化
长期:增加 GPU 节点支撑更多并发
8. 风险提示
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
• 未测试 WebSocket 长连接场景
• 未测试 100+ 并发的极端场景(GPU 显存不足可能 OOM)
• 多轮对话超过 15 轮时 TTFT 可能超 10s课堂练习
- 按上面的模板,为你的 AI 系统编写一份完整的性能测试报告
- 重点:确保报告中有 TTFT、Token 吞吐、GPU 指标这些 AI 专属指标
- 让其他团队成员 review 你的报告,确认结论是否清晰、建议是否可操作
补充参考答案要点
- AI 性能练习的关键答案应同时覆盖 TTFT、TPS/TPOT、总延迟和单位请求成本,而不是只看 QPS。
- 链路拆解时要区分 FAQ 命中、RAG 检索、模型生成和工具调用,因为它们的瓶颈完全不同。
- 结论输出要能说明性能问题究竟来自排队、检索、推理还是前端流式消费,而不是只写“慢”。