【大模型技术学习】大模型压力测试全攻略：以Qwen3-32B为例

在大模型落地生产环境的过程中，压力测试是不可或缺的一环——它能帮我们验证模型在高并发场景下的稳定性、响应效率和资源利用率，避免上线后出现服务崩溃、响应超时等问题。本文以阿里通义千问的Qwen3-32B模型为例，从核心指标、环境搭建、代码实现到结果分析，完整讲解大模型压力测试的全流程。

一、压力测试核心指标

在开始实操前，先明确我们需要关注的核心指标，这些指标是评估模型服务能力的关键：

二、测试环境准备

2.1 硬件环境（核心）

Qwen3-32B属于大参数量模型，推理时对显存要求较高，推荐硬件配置：

• GPU：单卡NVIDIA A100 80G（或双卡RTX 4090 24G，需开启模型分片）
• CPU：16核以上（如Intel Xeon 8375C）
• 内存：128GB以上
• 磁盘：SSD（避免模型加载时IO瓶颈）

2.2 软件环境

本次实操基于Python生态，核心依赖库如下（建议用conda创建独立环境）：

# 基础依赖pipinstallpython==3.10# 大模型推理加速（vLLM比原生transformers推理效率高5-10倍，适合压测）pipinstallvllm==0.4.2# Web服务框架（提供推理接口）pipinstallfastapi==0.104.1uvicorn==0.24.0.post1# 压测工具（Locust是轻量、易扩展的分布式压测框架）pipinstalllocust==2.16.1# 资源监控+数据可视化pipinstallpsutil==5.9.6 nvidia-ml-py==12.535.108pandas==2.1.4matplotlib==3.8.2

三、实操环节：Qwen3-32B压力测试全流程

3.1 步骤1：基于vLLM部署Qwen3-32B推理服务

vLLM是专为大模型设计的高效推理框架，支持动态批处理、PagedAttention，能显著提升并发处理能力。我们先搭建一个支持高并发的Qwen3-32B推理API服务。

创建文件qwen3_32b_api.py，代码如下（含详细注释）：

importasynciofromfastapiimportFastAPI,Request,JSONResponsefromvllmimportAsyncLLMEngine,EngineArgs,SamplingParamsfromvllm.utilsimportrandom_uuid# 初始化FastAPI应用app=FastAPI(title="Qwen3-32B Inference API")# 配置vLLM引擎参数engine_args=EngineArgs(# 模型路径（本地路径或HuggingFace Hub地址，需先下载Qwen3-32B模型）model="Qwen/Qwen3-32B-Chat",# 显存优化：FP16精度（32B模型FP16约需64G显存，若显存不足可加参数：tensor_parallel_size=2 开启双卡分片）dtype="float16",# 最大并发批处理大小（根据GPU显存调整）max_num_batched_tokens=8192,# 最大等待批处理时间（毫秒），提升并发效率max_batch_delay=10,# 关闭日志冗余输出disable_log_requests=True,)# 初始化异步LLM引擎（支持高并发）engine=AsyncLLMEngine.from_engine_args(engine_args)# 定义采样参数（和模型生成效果相关）sampling_params=SamplingParams(temperature=0.7,# 随机性top_p=0.8,# 核采样max_tokens=512,# 最大生成token数stop=["<|endoftext|>"],# 停止符)@app.post("/chat/completions")asyncdefchat_completions(request:Request):""" 兼容OpenAI格式的聊天接口，用于接收压测请求 """try:# 解析请求参数data=awaitrequest.json()messages=data.get("messages",[])ifnotmessages:returnJSONResponse(status_code=400,content={"error":"messages不能为空"})# 构造Qwen3-32B的输入格式prompt=""formsginmessages:ifmsg["role"]=="user":prompt+=f"<|im_start|>user\n{msg['content']}<|im_end|>\n"elifmsg["role"]=="assistant":prompt+=f"<|im_start|>assistant\n{msg['content']}<|im_end|>\n"prompt+="<|im_start|>assistant\n"# 生成请求IDrequest_id=random_uuid()# 提交推理请求到vLLM引擎results_generator=engine.generate(prompt=prompt,sampling_params=sampling_params,request_id=request_id,)# 获取推理结果final_output=""asyncforoutputinresults_generator:ifoutput.outputs[0].finish_reasonisnotNone:final_output=output.outputs[0].textbreak# 返回响应（兼容OpenAI格式）returnJSONResponse(content={"id":request_id,"object":"chat.completion","created":int(asyncio.get_event_loop().time()),"choices":[{"message":{"role":"assistant","content":final_output},"finish_reason":"stop","index":0}],"usage":{"prompt_tokens":len(prompt.split()),"completion_tokens":len(final_output.split()),"total_tokens":len(prompt.split())+len(final_output.split())}})exceptExceptionase:# 捕获异常，返回错误信息returnJSONResponse(status_code=500,content={"error":str(e)})if__name__=="__main__":# 启动API服务（workers=4适配多核CPU，port可自定义）importuvicorn uvicorn.run(app="qwen3_32b_api:app",host="0.0.0.0",port=8000,workers=4,log_level="info")

启动推理服务：

# 直接运行（若显存不足，添加环境变量开启模型分片：CUDA_VISIBLE_DEVICES=0,1）python qwen3_32b_api.py

启动成功后，可通过curl测试接口是否可用：

curl-X POST http://localhost:8000/chat/completions\-H"Content-Type: application/json"\-d'{ "messages": [{"role": "user", "content": "介绍一下大模型压力测试的核心指标"}] }'

3.2 步骤2：编写Locust压测脚本

Locust是基于Python的分布式压测工具，支持自定义用户行为，能模拟高并发请求。创建locustfile.py，代码如下：

importtimeimportjsonfromlocustimportHttpUser,task,between,eventsimportpsutilimportnvidia_smiimportpandasaspd# 初始化NVIDIA监控（获取GPU使用率）nvidia_smi.nvmlInit()gpu_handle=nvidia_smi.nvmlDeviceGetHandleByIndex(0)# 0表示第一个GPU# 存储压测数据（用于后续分析）test_metrics=[]# 压测开始时的钩子@events.test_start.add_listenerdefon_test_start(environment,**kwargs):print("===== Qwen3-32B 压力测试开始 =====")globalstart_time start_time=time.time()# 压测结束时的钩子@events.test_stop.add_listenerdefon_test_stop(environment,**kwargs):print("===== Qwen3-32B 压力测试结束 =====")# 将压测数据保存为CSVdf=pd.DataFrame(test_metrics)df.to_csv("qwen3_32b_load_test_metrics.csv",index=False)print(f"压测数据已保存至: qwen3_32b_load_test_metrics.csv")# 关闭NVIDIA监控nvidia_smi.nvmlShutdown()# 定义压测用户行为classQwen32BUser(HttpUser):# 模拟用户请求间隔：1-3秒（可根据实际场景调整）wait_time=between(1,3)# 测试用的prompt（尽量贴近真实业务场景）test_prompt={"messages":[{"role":"user","content":"请解释大模型的注意力机制，要求通俗易懂，字数控制在300字以内"}]}# 核心压测任务：调用聊天接口@task(1)# task权重为1（表示主要任务）defchat_completion(self):# 记录请求开始时间start=time.time()try:# 发送POST请求到推理接口response=self.client.post(url="/chat/completions",headers={"Content-Type":"application/json"},data=json.dumps(self.test_prompt))# 记录请求耗时response_time=time.time()-start# 获取系统资源使用率cpu_usage=psutil.cpu_percent(interval=0.1)mem_usage=psutil.virtual_memory().percent# 获取GPU使用率和显存占用gpu_info=nvidia_smi.nvmlDeviceGetUtilizationRates(gpu_handle)gpu_usage=gpu_info.gpu gpu_mem_usage=gpu_info.memory# 记录所有指标test_metrics.append({"timestamp":time.strftime("%Y-%m-%d %H:%M:%S"),"response_time":round(response_time,3),"status_code":response.status_code,"cpu_usage":cpu_usage,"mem_usage":mem_usage,"gpu_usage":gpu_usage,"gpu_mem_usage":gpu_mem_usage,"concurrent_users":self.environment.runner.user_count# 当前并发用户数})# 打印实时指标（可选）print(f"并发数:{self.environment.runner.user_count}| 响应时间:{response_time:.3f}s | GPU使用率:{gpu_usage}%")exceptExceptionase:# 记录失败请求response_time=time.time()-start test_metrics.append({"timestamp":time.strftime("%Y-%m-%d %H:%M:%S"),"response_time":round(response_time,3),"status_code":500,"cpu_usage":psutil.cpu_percent(interval=0.1),"mem_usage":psutil.virtual_memory().percent,"gpu_usage":nvidia_smi.nvmlDeviceGetUtilizationRates(gpu_handle).gpu,"gpu_mem_usage":nvidia_smi.nvmlDeviceGetUtilizationRates(gpu_handle).memory,"concurrent_users":self.environment.runner.user_count,"error":str(e)})print(f"请求失败:{e}")

3.3 步骤3：执行压测并监控

3.3.1 启动Locust压测

在新的终端中运行以下命令启动Locust（需确保推理服务已正常运行）：

# 本地启动Locust（Web UI模式，便于可视化操作）locust -f locustfile.py --host=http://localhost:8000

启动成功后，访问http://localhost:8089进入Locust Web控制台，配置压测参数：

• Number of users (peak concurrency)：目标并发数（如50）
• Spawn rate (users per second)：用户孵化速率（如5，即每秒新增5个并发用户）
• Host：推理服务地址（已自动填充）

点击「Start swarming」开始压测，可实时在Web界面查看：

• 每秒请求数（RPS）
• 响应时间分布
• 错误率
• 并发用户数曲线

3.3.2 多轮压测建议

为了全面评估模型性能，建议分多轮压测不同并发数：

1. 低并发（10-20）：验证基础稳定性
2. 中并发（50-80）：模拟日常高峰
3. 高并发（100+）：验证极限承载能力

每轮压测持续5-10分钟，避免短时间压测导致结果失真。

3.4 步骤4：压测结果分析与可视化

压测结束后，会生成qwen3_32b_load_test_metrics.csv文件，我们可以编写脚本分析并可视化结果：

创建analyze_results.py：

importpandasaspdimportmatplotlib.pyplotaspltimportwarnings warnings.filterwarnings("ignore")# 设置中文字体（避免图表乱码）plt.rcParams["font.sans-serif"]=["SimHei"]plt.rcParams["axes.unicode_minus"]=False# 读取压测数据df=pd.read_csv("qwen3_32b_load_test_metrics.csv")# 1. 基础统计分析print("===== 压测基础统计 =====")# 过滤成功请求success_df=df[df["status_code"]==200]# 计算核心指标avg_rt=success_df["response_time"].mean()p95_rt=success_df["response_time"].quantile(0.95)p99_rt=success_df["response_time"].quantile(0.99)tps=len(success_df)/(df["timestamp"].nunique()/60)# 每分钟TPSerror_rate=len(df[df["status_code"]!=200])/len(df)*100print(f"平均响应时间:{avg_rt:.3f}s")print(f"P95响应时间:{p95_rt:.3f}s")print(f"P99响应时间:{p99_rt:.3f}s")print(f"吞吐量(TPS):{tps:.2f}req/min")print(f"错误率:{error_rate:.2f}%")print(f"平均CPU使用率:{df['cpu_usage'].mean():.2f}%")print(f"平均GPU使用率:{df['gpu_usage'].mean():.2f}%")print(f"平均GPU显存使用率:{df['gpu_mem_usage'].mean():.2f}%")# 2. 可视化：响应时间 vs 并发数plt.figure(figsize=(12,8))# 子图1：响应时间分布plt.subplot(2,2,1)plt.plot(success_df["concurrent_users"],success_df["response_time"],'b-',alpha=0.6,label="响应时间")plt.axhline(y=avg_rt,color='r',linestyle='--',label=f"平均RT:{avg_rt:.3f}s")plt.xlabel("并发数")plt.ylabel("响应时间（s）")plt.title("响应时间随并发数变化")plt.legend()plt.grid(True,alpha=0.3)# 子图2：GPU使用率plt.subplot(2,2,2)plt.plot(df["concurrent_users"],df["gpu_usage"],'g-',alpha=0.6,label="GPU使用率")plt.xlabel("并发数")plt.ylabel("GPU使用率（%）")plt.title("GPU使用率随并发数变化")plt.legend()plt.grid(True,alpha=0.3)# 子图3：CPU使用率plt.subplot(2,2,3)plt.plot(df["concurrent_users"],df["cpu_usage"],'orange',alpha=0.6,label="CPU使用率")plt.xlabel("并发数")plt.ylabel("CPU使用率（%）")plt.title("CPU使用率随并发数变化")plt.legend()plt.grid(True,alpha=0.3)# 子图4：错误率plt.subplot(2,2,4)error_df=df.groupby("concurrent_users")["status_code"].apply(lambdax:(x!=200).sum()/len(x)*100).reset_index()plt.bar(error_df["concurrent_users"],error_df["status_code"],color='red',alpha=0.6,label="错误率")plt.xlabel("并发数")plt.ylabel("错误率（%）")plt.title("错误率随并发数变化")plt.legend()plt.grid(True,alpha=0.3)plt.tight_layout()plt.savefig("qwen3_32b_load_test_results.png",dpi=300)print("压测结果可视化图表已保存至: qwen3_32b_load_test_results.png")plt.show()

运行分析脚本：

python analyze_results.py

执行后会输出统计指标，并生成可视化图表（示例效果如下）：

• 响应时间随并发数增长的趋势（并发越高，RT越长，需关注拐点）
• GPU/CPU使用率变化（若GPU使用率长期100%，说明硬件已到瓶颈）
• 错误率分布（高并发下若错误率飙升，需优化服务）

四、常见问题与优化建议

4.1 压测中常见问题

1. 显存不足：Qwen3-32B FP16推理需约64G显存，若显存不足，可开启tensor_parallel_size=2（双卡分片）或使用INT4/INT8量化（vLLM支持quantization=awq）。
2. 响应时间过长：可优化vLLM的max_num_batched_tokens（增大批处理大小）、减少max_tokens（缩短生成文本长度）。
3. 服务崩溃：检查uvicorn的workers数（建议等于CPU核心数），或使用Nginx做反向代理，限制并发连接数。

4.2 性能优化方向

1. 推理层优化：使用vLLM/TGI等高效推理框架，开启动态批处理、PagedAttention。
2. 硬件层优化：升级GPU（如A100→H100）、增加GPU数量（模型分片）。
3. 服务层优化：使用K8s做容器化部署，配置HPA（水平自动扩缩容），根据并发数动态调整实例数。
4. 业务层优化：对请求做限流、降级（如高并发时返回缓存结果），或拆分长请求为短请求。

五、总结

本文以Qwen3-32B为例，完整讲解了大模型压力测试的核心流程：从环境搭建、推理服务部署，到Locust压测脚本编写、结果分析与可视化。核心要点如下：

1. 压测前明确核心指标（并发、RT、TPS、资源利用率、错误率），避免无目标测试；
2. 大模型推理优先选择vLLM等高效框架，提升并发处理能力；
3. 压测需分多轮（低/中/高并发），结果需结合资源使用率分析瓶颈；
4. 压测后根据结果针对性优化（硬件/推理/服务层），确保生产环境稳定。

本文的代码可直接复用，只需根据实际场景调整模型路径、prompt、并发数等参数。希望能帮助大家高效完成大模型的压力测试，让大模型稳定落地生产。