当前位置：首页 > news >正文

你的大模型服务如何压测：首 Token 延迟、并发与 QPS

news 来源：原创 2025/7/23 2:01:48

在这里插入图片描述

写在前面

大型语言模型（LLM）API，特别是遵循 OpenAI 规范的接口（无论是 OpenAI 官方、Azure OpenAI，还是 DeepSeek、Moonshot 等众多兼容服务），已成为驱动下一代 AI 应用的核心引擎。然而，随着应用规模的扩大和用户量的增长，仅仅关注模型的功能是不够的，API 的性能表现成为决定用户体验和系统稳定性的关键因素。

开发者和运维团队常常需要回答以下问题：

用户发送请求后，需要多久才能看到第一个字的响应？（首 Token 延迟 - Time To First Token, TTFT）
我的 API 服务同时能处理多少个用户的请求而不会崩溃或严重延迟？（最大并发数 - Max Concurrency）
在稳定运行状态下，API 每秒钟能成功处理多少个请求？（每秒查询率 - Queries Per Second, QPS）

了解这些性能指标对于容量规划、成本估算、服务等级协议（SLA）设定以及优化用户体验至关重要。幸运的是，我们可以利用 Python 脚本，结合异步处理、并发控制等技术，对这些 OpenAI 类接口进行较为精确的压力测试（Stress Testing）和基准测试（Benchmarking）。

本篇博客将深入探讨如何使用 Python 脚本来测量 LLM API 的 TTFT、最大并发和 QPS，涵盖测试原理、关键库选择、脚本设计、示例代码、结果分析以及注意事项。

1. 核心性能指标解读：TTFT, Concurrency, QPS

在开始压测之前，我们必须清晰地理解我们要测量的目标：

首 Token 延迟 (Time To First Token, TTFT):
- 定义: 从客户端发送 API 请求开始，到接收到第一个由模型生成的有效内容 Token 所经过的时间。
- 重要性: 直接影响用户的感知响应速度。对于交互式应用（如聊天机器人），低 TTFT 意味着用户能更快地看到反馈，感觉更流畅。高 TTFT 则会让用户觉得“卡顿”或无响应。
- 测量难点: 需要使用流式接口 (Streaming API)，并在接收到第一个包含实际 content 的 delta 块时记录时间戳。
并发数 (Concurrency):
- 定义: 系统同时能够处理的活动请求的数量。注意，并发数不等于用户总数，而是指在任意时刻有多少请求正在被服务器处理（从接收到请求到响应完全结束）。
- 重要性: 决定了系统能同时服务多少“活跃”用户。达到或超过最大并发数通常会导致请求排队、延迟急剧增加甚至请求失败（如返回 429 Too Many Requests 或 503 Service Unavailable）。
- 测量方式: 通过逐步增加同时发起的请求数量，观察 API 的响应时间、成功率等指标的变化，找到系统开始不稳定的临界点。
每秒查询率 (Queries Per Second, QPS) / 吞吐量 (Throughput):
- 定义: 系统在单位时间（通常是秒）内成功处理的请求数量。
- 重要性: 反映了系统的整体处理能力。QPS 越高，系统能支持的总请求量越大。
- 与并发的关系: QPS 和并发数通常是相关的，但不完全等同。QPS ≈ Concurrency / Average_Request_Latency。提高并发数可以提高 QPS，但当系统达到瓶颈时，进一步增加并发可能导致延迟增加，反而降低 QPS。
- 测量方式: 在一段持续时间内，以一定的并发数（或速率）发送请求，统计单位时间内成功完成的请求总数。

其他相关指标:

Token 生成速率 (Tokens Per Second, TPS): 对于流式响应，指模型每秒生成多少个 Token。TPS ≈ (Total_Output_Tokens / Request_Duration) - TTFT (近似)。
请求总耗时 (End-to-End Latency): 从发送请求到接收到完整响应所花费的时间。
成功率 (Success Rate): 成功返回结果的请求占总请求的比例。压测时需要密切关注成功率，低于某个阈值（如 99%）通常意味着系统过载。
错误率 (Error Rate): 请求失败（如 4xx, 5xx 错误）的比例。

我们的 Python 压测脚本需要能够测量并记录这些关键指标。

2. 技术选型：Python 库的选择

为了有效地模拟并发请求并精确测量时间，我们需要合适的 Python 库：

HTTP 客户端:
- requests (同步): 简单易用，适合低并发或单线程测试。但在高并发场景下，同步阻塞 IO 会成为瓶颈。
- aiohttp (异步): 基于 asyncio，实现非阻塞 IO，是高并发压测的首选。能够用较少的线程/进程处理大量的并发连接。
- httpx (同步/异步): 一个现代的 HTTP 客户端，同时支持同步和异步操作，API 设计友好，也是一个很好的选择。
异步框架:
- asyncio: Python 内置的异步 IO 框架，是 aiohttp 和 httpx (异步模式) 的基础。需要掌握 async/await 语法。
并发控制:
- asyncio.Semaphore: 用于限制同时进行的异步任务数量，是控制并发数的关键工具。
- multiprocessing: 如果需要利用多核 CPU 且任务是 CPU 密集型（虽然 API 调用主要是 IO 密集型，但大量数据处理或复杂逻辑可能需要），可以考虑多进程。但进程间通信和状态共享更复杂。对于 IO 密集的 API 压测，asyncio 通常更高效。
OpenAI 客户端:
- openai Python 库: 官方库，支持同步和异步客户端 (AsyncOpenAI)，并且内置了对流式响应的处理逻辑。推荐使用官方库，特别是其异步版本，可以简化与 asyncio 的集成。
数据处理与统计:
- time: 用于精确计时。time.perf_counter() 是测量短时间间隔的首选。
- statistics / numpy: 用于计算平均值、中位数、百分位数（P90, P99）等统计指标。
- pandas (可选): 用于更方便地存储、处理和分析压测结果。

本文后续示例将主要使用 asyncio 和 openai 库的异步客户端 (AsyncOpenAI)，因为这是实现高并发测量 TTFT 的最自然方式。

3. 压测脚本设计：核心逻辑与考量

一个好的压测脚本需要考虑以下方面：

配置化:
- API Endpoint URL (base_url)。
- API Key。
- 目标模型名称 (model)。
- 请求 Payload (包括 messages, max_tokens, temperature 等，可以支持从文件加载多个不同的 Payload 以模拟真实场景)。
- 压测参数：并发数 (concurrency)、总请求数 (total_requests) 或压测持续时间 (duration)。
- 是否启用流式 (stream=True)。
核心请求函数 (make_request):
- 负责构造请求数据。
- 使用 AsyncOpenAI 客户端发起流式 API 调用 (client.chat.completions.create(..., stream=True))。
- 精确计时 TTFT: 在 await client.chat.completions.create 之前记录开始时间 t_start。在 async for chunk in stream: 循环中，检查 chunk.choices[0].delta.content 是否首次非空，如果是，记录此刻时间 t_first_token，计算 ttft = t_first_token - t_start。
- 记录 Token 生成速率（可选）。
- 记录请求总耗时：在循环结束后记录 t_end，计算 total_latency = t_end - t_start。
- 记录成功/失败状态和错误信息。
- 返回包含所有测量指标的字典或对象。
并发控制器 (run_test):
- 使用 asyncio.Semaphore(concurrency) 创建信号量，限制并发数量。
- 创建 total_requests 个 make_request 协程任务。
- 使用 asyncio.gather 或循环配合 semaphore.acquire() 和 semaphore.release() 来并发地运行这些任务。
- 收集所有任务的结果。
结果聚合与统计:
- 从所有请求结果中提取 TTFT、总延迟、成功/失败次数等。
- 计算关键统计指标：
  - TTFT: 平均值 (Avg), 中位数 (Median/P50), P90, P99。
  - 总延迟: Avg, Median, P90, P99。
  - QPS: 成功请求数 / 总测试时间。
  - 成功率: 成功请求数 / 总请求数。
  - 错误率 & 错误类型分布。
逐步加压 (可选，用于找最大并发):
- 可以编写一个循环，逐步增加 concurrency 的值，每次运行一轮压测，记录下不同并发数对应的 QPS、延迟和成功率。
- 观察指标变化：通常，随着并发增加，QPS 会先上升然后趋于平稳或下降，而延迟（尤其是 P99 延迟）和错误率会急剧上升。最大并发数通常定义为在满足可接受延迟（如 P99 TTFT < 1s）和高成功率（如 >99%）前提下的最高并发水平。

4. Python 压测代码

import asyncio
import time
import os
import statistics
import json
from openai import AsyncOpenAI # 使用异步客户端
from dotenv import load_dotenv
import numpy as np # 用于计算百分位数# --- 配置 ---
load_dotenv()
API_KEY = os.getenv("DEEPSEEK_API_KEY") or "YOUR_API_KEY" # 替换或确保环境变量设置
BASE_URL = "https://api.deepseek.com/v1" # DeepSeek API 地址
MODEL_NAME = "deepseek-chat" # 或其他模型
# MODEL_NAME = "deepseek-coder"# 压测参数
CONCURRENCY = 10      # 同时发起的请求数
TOTAL_REQUESTS = 100 # 总共要发送的请求数
MAX_TOKENS = 512       # 限制生成长度
TEMPERATURE = 0.5# 示例 Payload (可以扩展为从文件加载多个)
DEFAULT_PAYLOAD = {"model": MODEL_NAME,"messages": [{"role":

写在前面

1. 核心性能指标解读：TTFT, Concurrency, QPS

2. 技术选型：Python 库的选择

3. 压测脚本设计：核心逻辑与考量

4. Python 压测代码

相关文章：