1. 项目概述为什么大模型API并发测试是门“必修课”最近在折腾几个AI应用项目从智能客服到内容生成核心都绕不开调用大模型API。项目上线前团队里总会有人问“咱们这系统到底能扛住多少用户同时提问” 这个问题看似简单背后却是一连串的技术焦虑API调用会不会突然变慢服务商有没有限流我们的钱包能不能撑住突发的流量洪峰更重要的是用户体验会不会在关键时刻“掉链子”这些问题靠猜是没用的必须靠数据说话。这就是为什么我们需要对大模型API进行并发压力测试而Locust正是我用来解决这个问题的“瑞士军刀”。你可能听说过JMeter、wrk这些压测工具为什么我独爱Locust原因很简单Python原生。大模型API的调用逻辑无论是使用OpenAI格式、还是国内智谱、DeepSeek的SDK本质上都是一段Python代码。用Locust我可以用写业务逻辑的思维来设计压测场景模拟真实的用户思考、等待、连续对话等复杂行为而不是在图形界面里拖拽一个个难以维护的元件。它能精准地告诉我在每秒50个、100个甚至500个并发请求下我的API响应时间P99延迟是多少失败率有多高从而清晰地找到系统的性能瓶颈和成本临界点。这篇文章我就以一个真实的“智能问答系统”接入某大模型API的场景为例带你从零开始用Locust完成一次完整的、可复现的并发极限评估实战。你会学到如何设计贴近真实的用户行为脚本、如何解读关键的压测指标、以及如何避开那些我踩过的“坑”。无论你是后端开发、算法工程师还是项目负责人这套方法都能帮你把“系统能扛多少并发”从一个模糊的疑问变成一个精确的、有数据支撑的答案。2. 核心思路与工具选型为什么是Locust异步在规划这次压测时我首先明确了几个核心目标第一测试场景必须高度模拟真实用户不能是简单的单次请求循环第二工具要能清晰展示性能瓶颈是网络、API限流还是我自身代码的问题第三整个流程要易于集成和自动化方便在每次API更新或业务逻辑调整后快速回归测试。基于这三点Locust几乎是唯一的选择。2.1 Locust的核心优势解析Locust是一个用Python编写的开源负载测试工具。它的核心哲学是“用代码定义用户行为”这带来了几个无可替代的优势测试即代码你的压测脚本locustfile.py就是一个标准的Python模块。你可以使用requests、aiohttp、openai等任何你熟悉的库来发起请求也可以方便地引入业务逻辑比如先登录获取token再发起对话。这使得测试脚本的维护和版本控制变得极其简单。分布式与可扩展性单机跑不动了Locust原生支持分布式运行。一台机器作为主节点master多台机器作为从节点worker可以轻松模拟数万甚至数十万的并发用户。这对于真正想探知大模型API全局限流阈值的情况至关重要。实时Web UI与详实数据启动Locust后可以通过浏览器访问一个本地Web界面实时查看当前RPS每秒请求数、响应时间、失败率等关键指标。测试结束后还能生成详细的HTML报告方便团队分享和归档。资源消耗相对较低相比于一些基于Java的GUI工具Locust特别是使用异步客户端时在资源利用上更为高效单机也能模拟较高的并发。2.2 同步与异步客户端的关键抉择这是使用Locust时第一个重要的技术决策点。Locust支持两种HTTP客户端默认的同步HttpUser和基于gevent的FastHttpUser以及需要手动实现的异步客户端如aiohttp。同步客户端HttpUser使用简单适合请求间隔较长或逻辑简单的场景。但在模拟高并发时每个虚拟用户User都是一个独立的greenlet微线程虽然比操作系统线程轻量但在发起网络IO等待时仍然会进行上下文切换。当并发数极高时例如数千其调度开销会变得明显。异步客户端aiohttp asyncio这是我强烈推荐用于大模型API压测的方案。大模型API的响应时间通常在1-10秒甚至更长这是一个典型的高延迟、低频率的IO密集型场景。使用异步可以用极少的操作系统线程甚至一个管理成千上万个并发连接。每个虚拟用户在一个事件循环中发起请求后便挂起让出控制权给其他用户直到收到响应再恢复。这能极大地提升单机模拟高并发的上限并且更准确地反映真实世界中大量用户同时等待长响应的场景。简单来说如果你想在个人电脑上就模拟出上千用户同时与大模型对话的场景异步方案是必由之路。接下来的实战部分我将基于异步方案展开。2.3 大模型API压测的特殊性与测试一个返回“Hello World”的普通HTTP接口不同测试大模型API需要特别注意以下几点令牌Token管理API Key是敏感信息不能硬编码在脚本中。需要安全地通过环境变量或外部配置文件传入。请求体构造大模型请求的Prompt、参数如max_tokens, temperature会显著影响响应时间和token消耗从而影响性能和成本。压测时应使用有代表性的、符合业务场景的Prompt。响应解析与断言除了检查HTTP状态码是否为200我们还需要验证响应体结构是否正确、是否包含了预期的“choices”或“message”字段甚至可以对返回的文本内容做简单校验如是否非空。尊重服务商限流盲目地进行极限压测可能导致你的API Key被临时封禁。务必先从低并发开始阶梯式增加并密切关注返回的429Too Many Requests或其他限流错误。我们的目标是找到稳定运行的“甜蜜点”而不是暴力攻击。3. 环境准备与Locustfile核心脚本编写理论说得再多不如一行代码。让我们开始动手搭建压测环境。我假设你已经在本地或测试服务器上准备好了Python环境3.7。3.1 依赖安装与项目结构首先创建一个干净的目录并安装必要的包。我们主要需要locust和异步HTTP客户端aiohttp。# 创建项目目录 mkdir mlloadtest cd mlloadtest # 创建虚拟环境推荐 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装核心依赖 pip install locust aiohttp # 可选安装你所用大模型官方的SDK例如OpenAI、智谱等方便构造请求。 # pip install openai项目目录结构可以这样规划mlloadtest/ ├── locustfile.py # 核心压测脚本 ├── prompts.json # 存储测试用的Prompt池 ├── .env # 存储API Key等环境变量务必加入.gitignore └── requirements.txt # 依赖列表3.2 编写异步Locustfile脚本这是整个压测的核心。我们将创建一个继承自HttpUser但使用aiohttp会话的异步用户类。# locustfile.py import os import random import asyncio from locust import HttpUser, task, between, events from locust.exception import StopUser import aiohttp import json from dotenv import load_dotenv # 用于加载.env文件 # 加载环境变量 load_dotenv() class AsyncAIOHttpSession: 一个简单的aiohttp会话包装器用于在Locust中记录请求统计。 def __init__(self, user): self.user user self.session None async def __aenter__(self): self.session aiohttp.ClientSession() return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self.session.close() async def post(self, url, **kwargs): 发起POST请求并自动集成到Locust的统计中。 # 为请求命名用于Locust报告中的区分 name kwargs.pop(name, url) start_time time.time() response None try: response await self.session.post(url, **kwargs) # 手动触发Locust的成功/失败事件 total_time int((time.time() - start_time) * 1000) # 转毫秒 if response.status 400: events.request.fire( request_typePOST, namename, response_timetotal_time, response_lengthlen(await response.text()), contextself.user._context(), exceptionNone, ) else: events.request.fire( request_typePOST, namename, response_timetotal_time, response_length0, contextself.user._context(), exceptionException(fHTTP Error {response.status}), ) return response except Exception as e: total_time int((time.time() - start_time) * 1000) events.request.fire( request_typePOST, namename, response_timetotal_time, response_length0, contextself.user._context(), exceptione, ) raise # 假设我们测试一个兼容OpenAI格式的API端点 API_BASE_URL os.getenv(API_BASE_URL, https://api.example.com/v1) API_KEY os.getenv(API_KEY) MODEL_NAME os.getenv(MODEL_NAME, gpt-3.5-turbo) # 加载测试Prompt池 with open(prompts.json, r, encodingutf-8) as f: PROMPT_POOL json.load(f)[prompts] class ChatAPIUser(HttpUser): 模拟一个与大模型进行对话的用户。 每个用户会随机从Prompt池中选择一个问题进行提问。 # 用户任务之间的等待时间范围秒模拟用户思考时间。 wait_time between(2, 5) def on_start(self): 当虚拟用户启动时执行可用于初始化会话、登录等。 # 初始化aiohttp会话 self.aiohttp_session None # 注意由于Locust的架构我们需要在任务中异步初始化这里先占位。 task def ask_question(self): 核心任务向大模型API发送一个提问请求。 由于Locust的task默认不支持async我们通过asyncio.run来运行异步函数。 在高版本Locust或特定模式下可以考虑使用task(weight)装饰异步函数。 这里采用一个兼容性较好的模式在同步任务中运行异步事件循环。 # 为每个任务创建一个新的事件循环在非WebUI模式下可行 loop asyncio.new_event_loop() asyncio.set_event_loop(loop) try: loop.run_until_complete(self._async_ask_question()) finally: loop.close() async def _async_ask_question(self): 真正的异步请求逻辑 if self.aiohttp_session is None: self.aiohttp_session AsyncAIOHttpSession(self) async with self.aiohttp_session as session: # 1. 随机选择一个Prompt prompt random.choice(PROMPT_POOL) # 2. 构造请求体 (OpenAI格式示例) payload { model: MODEL_NAME, messages: [{role: user, content: prompt}], max_tokens: 150, # 控制生成长度影响响应时间和成本 temperature: 0.7, } headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } # 3. 发起请求并使用‘name’参数标识此请求类型 url f{API_BASE_URL}/chat/completions try: response await session.post( url, name/chat/completions, jsonpayload, headersheaders, timeoutaiohttp.ClientTimeout(total30) # 设置超时 ) # 4. 检查响应 if response.status 200: data await response.json() # 可选验证响应结构 if not data.get(choices): raise ValueError(Response missing choices field) # 你可以在这里对回复内容做进一步检查例如长度、关键词等 # print(fUser got reply: {data[choices][0][message][content][:50]}...) elif response.status 429: # 遇到限流可以记录日志并让这个虚拟用户“休息”一会儿 print(fRate limited! Status: {response.status}) await asyncio.sleep(5) # 等待5秒后重试或继续 else: # 其他错误记录并可能停止该用户根据测试策略 error_text await response.text() print(fRequest failed with status {response.status}: {error_text}) except asyncio.TimeoutError: print(Request timed out!) except Exception as e: print(fAn error occurred: {e})脚本关键点解析与避坑指南异步会话封装AsyncAIOHttpSession类是我们的核心创新点。它包装了aiohttp.ClientSession并在每个请求前后手动触发了Locust的events.request事件。这是将异步请求集成到Locust统计系统的关键。没有这一步Locust的Web UI将无法正确显示你的请求数据。任务task装饰器Locust原生的task装饰器不支持async def。我们采用了一个变通方案在同步的ask_question方法内使用asyncio.run或loop.run_until_complete来执行真正的异步函数_async_ask_question。在高并发下为每个任务创建新的事件循环有一定开销但对于大模型API这种长耗时请求这个开销可以接受。更优雅的方案是使用locust-plugins等第三方库的异步支持但为了减少依赖本例采用基础方法。超时设置aiohttp.ClientTimeout(total30)至关重要。大模型API响应可能很慢但没有超时设置的请求会一直挂起耗尽系统资源。设置一个合理的总超时如30秒超时后触发asyncio.TimeoutErrorLocust会将其记录为失败请求。错误处理与限流我们特别处理了HTTP 429状态码限流。在真实压测中一旦频繁出现429意味着已经触达API提供方的当前限制。此时不应继续疯狂重试而是应该让虚拟用户“睡眠”一下或者停止增加并发数这更符合真实场景和“友好测试”的原则。Prompt池将测试用的Prompt放在外部的prompts.json文件中使得测试用例可以灵活扩展更贴近真实用户输入的多样性。文件内容类似{ prompts: [ 用简单的语言解释一下什么是机器学习, 写一首关于春天的五言绝句。, 计算一下15的阶乘是多少, 总结《三国演义》中赤壁之战的主要经过。, 将‘Hello, world!’翻译成法语和西班牙语。 ] }4. 压测执行策略与关键参数配置脚本准备好了接下来是如何执行它。Locust提供了命令行和Web UI两种交互方式对于探索性测试和结果展示Web UI非常直观对于自动化或CI/CD集成命令行模式更合适。4.1 通过Web UI执行与监控这是最常用的方式适合交互式地调整负载观察实时曲线。# 在项目根目录下执行 locust -f locustfile.py启动后打开浏览器访问http://localhost:8089你会看到Locust的Web控制台。控制台参数配置详解Number of users (peak concurrency)峰值并发用户数。这是Locust中最重要的概念之一。它不等于每秒请求数RPS。例如设置1000个用户且每个用户平均每5秒执行一个任务wait_time between(2, 5)那么理论上的RPS约为 1000 users / 5s 200 RPS。我们的目标是找到系统稳定运行下能支持的“最大峰值并发用户数”。Spawn rate (users started/second)孵化率即每秒启动多少个虚拟用户直到达到峰值用户数。设置为10意味着每秒增加10个用户慢慢给系统加压而不是瞬间将所有用户砸向系统。这有助于观察系统负载逐渐上升时的表现是一种更温和、更真实的测试方式。Host这里填写你的大模型API的基础URL例如https://api.example.com/v1。注意我们的脚本里已经定义了API_BASE_URL如果这里也填写了脚本中的URL会以此为准。执行流程建议第一步冒烟测试。设置Number of users 1,Spawn rate 1运行1分钟。确保单个用户能正常请求并得到响应验证脚本和环境无误。第二步阶梯加压。这是最关键的步骤。不要一上来就设置几千并发。从低并发开始例如Users50,Spawn rate5运行3-5分钟。观察响应时间特别是95和99分位值和失败率。如果一切正常失败率0.1%P99延迟在可接受范围如5秒内再逐步增加并发数例如Users100,150,200... 每次增加后稳定运行一段时间。密切关注失败请求。一旦失败率特别是因429限流导致的失败显著上升或者P99响应时间急剧增加出现“拐点”说明系统已经达到或接近当前配置下的极限。记录下这个临界点的并发用户数。第三步稳定性测试 soak test 。在找到的“临界点”之下选择一个相对安全的并发数例如临界点的80%运行较长时间如30分钟到1小时。观察系统在持续负载下的表现内存是否缓慢增长响应时间是否保持平稳。这能发现一些在短时压力下不明显的潜在问题如内存泄漏、连接池耗尽等。4.2 通过命令行无头模式执行对于自动化测试或集成到CI/CD流水线可以使用无头headless模式。# 基本命令运行30秒每秒启动2个用户直到达到100个用户然后关闭。 locust -f locustfile.py --headless --users 100 --spawn-rate 2 --run-time 30s --host https://api.example.com/v1 # 更实用的命令指定更多参数并生成报告 locust -f locustfile.py \ --headless \ --users 500 \ # 总用户数 --spawn-rate 10 \ # 孵化率 --run-time 5m \ # 运行时间 5分钟 --csvreport \ # 生成CSV报告前缀 --csv-full-history \ # 记录整个运行过程的历史数据 --htmlreport.html \ # 生成HTML报告 --hosthttps://api.example.com/v1命令行参数解析--headless: 启用无头模式不启动Web UI。--users和--spawn-rate: 等同于Web UI中的两个核心参数。--run-time: 测试运行的总时长格式如30s,5m,1h30m。--csv和--html: 生成数据报告。CSV文件便于用Excel或Python进行后续分析HTML报告则提供了可视化的图表和表格非常适合存档和分享。强烈建议每次压测都生成报告。5. 结果分析与性能瓶颈定位压测完成后面对Locust提供的海量数据我们该关注什么如何从数据中读出系统的“健康状况”和“极限所在”5.1 核心性能指标解读在Locust的Web UI或生成的报告中重点关注以下指标指标含义健康信号报警信号RPS (Requests/s)每秒成功完成的请求数。随着并发用户数增加而平稳上升。达到某个点后不再增长甚至下降说明系统吞吐量已达上限或出现瓶颈。响应时间 (Response Times)请求从发出到收到完整响应所花费的时间。重点关注平均响应时间、95分位值(P95)和99分位值(P99)。P95/P99响应时间增长平缓与平均响应时间差距不大。P95/P99响应时间急剧上升出现“长尾”远高于平均值。例如平均2秒P99达到20秒说明部分请求体验极差。失败率 (Failures/s Failure %)每秒失败的请求数及其占总请求数的百分比。接近于0%例如 0.1%。持续高于0.5%或突然飙升。需要立刻查看失败原因是429限流、超时还是5xx服务器错误。用户数 (Number of Users)当前活跃的虚拟用户数。平稳达到预设的峰值并发数。用户数无法达到预设值可能因为孵化过程就出现了大量失败。关键心法寻找“拐点”。性能测试的核心不是看最高能跑到多少RPS而是找到性能拐点。即当并发用户数或RPS增加到某个值时P99响应时间开始非线性地急剧增加和/或失败率开始显著上升。这个点就是系统在当前配置下的有效并发处理极限。你的系统应该运行在这个拐点之前并留有一定的安全余量比如拐点并发数的70%。5.2 常见瓶颈分析与排查思路当发现性能不佳时如何定位问题问题可能出在多个环节。客户端瓶颈你的压测机现象Locust的CPU或内存使用率接近100%但RPS很低。排查使用top或htop命令查看locust进程资源消耗。单机模拟过高并发如数千时可能受限于网络端口数、文件描述符数或Python GIL对于同步客户端。解决使用异步客户端如我们脚本中的aiohttp大幅提升单机能力。如果还不够采用Locust分布式模式增加压测从机worker。网络瓶颈现象响应时间中“连接建立时间TTFB前期”占比较高或者出现大量连接超时、重置错误。排查在压测脚本中记录更细粒度的耗时或使用curl -w或专业网络监控工具。检查压测机与API服务器之间的网络延迟和带宽。解决确保压测机与API服务器在同一区域网络或使用云服务商的内网地址进行测试以排除公网不稳定的影响。大模型API服务端瓶颈/限流现象这是最常见的情况。表现为大量HTTP 429Too Many Requests错误或者响应时间随着并发增加呈阶梯式上升触发了服务端的队列机制。排查仔细查看失败请求的响应体和响应头。许多API会在响应头中返回限流信息如X-RateLimit-Limit,X-RateLimit-Remaining,X-RateLimit-Reset。记录并分析这些信息。解决尊重限流。根据返回的限流策略调整你的压测策略。例如如果限流是每分钟N次请求那么你的并发用户数和任务间隔需要据此计算。你的测试目标应该是验证在限流范围内系统的表现而不是试图“攻破”限流。自身应用逻辑瓶颈现象如果你是在测试一个封装了大模型API的自家后端服务而非直接测API那么瓶颈可能出现在你的业务逻辑、数据库、缓存或外部依赖上。排查在Locust中为不同的请求步骤命名如/api/chat,/api/auth分别观察它们的响应时间。同时监控你的应用服务器如Nginx, uWSGI, Gunicorn和后端数据库的指标。解决优化慢查询引入缓存如Redis缓存常见问答使用连接池管理数据库和外部API连接考虑异步处理耗时任务等。5.3 生成与解读HTML报告使用--html参数生成的报告是复盘和分享的利器。报告主要包括Statistics统计概览所有请求类型的汇总数据包括我们最关心的RPS、响应时间分位数、失败数。Charts图表Total Requests per Second RPS随时间变化曲线看是否平稳。Response Times (ms) 响应时间平均、中位数、P95随时间变化曲线。理想状态是三条线贴近且平稳。如果P95线陡然上升就是“拐点”的视觉体现。Number of Users 并发用户数变化曲线。Failures失败详情 列出所有失败的请求包括错误类型和发生次数是定位问题的直接入口。Download Data下载数据 可以下载完整的CSV数据用于更深入的自定义分析。报告分析实战假设你运行了一次阶梯加压测试从报告图表中看到当并发用户数达到180时P95响应时间从稳定的2秒内突然跳涨到10秒以上同时失败率开始出现429错误。那么180就是这个API在当前Prompt长度、参数配置下的一个性能临界点。你的系统设计应该以此为依据。6. 高级技巧与实战避坑指南掌握了基础方法后下面这些来自实战的经验和技巧能让你把Locust用得更加得心应手测试结果也更加可靠。6.1 模拟更真实的用户行为我们的基础脚本是每个用户随机问一个问题。真实的对话场景要复杂得多多轮对话Session用户会进行连续追问。这需要你在虚拟用户中维护一个对话历史messages列表每次请求都将历史记录发送给API。class ConversationalUser(HttpUser): def on_start(self): self.conversation_history [ {role: system, content: 你是一个有帮助的助手。} ] task def multi_turn_chat(self): # 用户说 user_prompt random.choice(PROMPT_POOL) self.conversation_history.append({role: user, content: user_prompt}) # 发送整个历史 payload { model: MODEL_NAME, messages: self.conversation_history, max_tokens: 150, } # ... 发送请求 ... # 假设收到回复 # reply data[choices][0][message][content] # 助手说 (模拟) assistant_reply f模拟回复 for: {user_prompt} self.conversation_history.append({role: assistant, content: assistant_reply}) # 控制对话轮次避免历史无限增长 if len(self.conversation_history) 10: # 保留最近10轮 self.conversation_history [self.conversation_history[0]] self.conversation_history[-8:]思考时间与行为随机性使用between或constant来设置wait_time只是基础。更真实的模拟可以使用constant_pacing来确保每个用户至少间隔X秒执行一次任务或者用自定义函数实现更复杂的随机分布如正态分布。用户分组与差异化不是所有用户行为都一样。你可以定义多个User类赋予不同的weight权重和tasks。例如80%的用户只进行简单问答SimpleUser20%的用户进行复杂多轮对话ComplexUser。6.2 参数化与数据驱动动态Prompt与参数除了从文件读取Prompt池还可以连接数据库、调用其他API动态生成测试数据。例如测试一个摘要生成API可以从新闻网站RSS实时获取文章标题和内容作为输入。并发与Token消耗/成本估算大模型API按Token收费。在压测脚本中你可以解析响应估算每次请求消耗的Prompt Token和Completion Token。结合你的目标RPS就能粗略估算出在特定负载下每小时的API调用成本。这是一个非常重要的衍生洞察。# 在收到响应后 if response.status 200: data await response.json() prompt_tokens data.get(usage, {}).get(prompt_tokens, 0) completion_tokens data.get(usage, {}).get(completion_tokens, 0) total_tokens prompt_tokens completion_tokens # 可以在这里累加统计或者发送到外部监控系统 self.user.environment.events.request.fire( ... # 其他参数 meta{prompt_tokens: prompt_tokens, completion_tokens: completion_tokens} )6.3 我踩过的那些“坑”坑忘记设置超时。早期测试时脚本没有设置timeout。当API服务端偶尔无响应时请求会一直挂起快速耗尽压测机的可用端口和内存导致Locust主进程崩溃。务必为所有网络请求设置合理的超时。坑API Key硬编码或误提交。曾不小心将包含测试Key的脚本提交到了GitHub公共仓库。务必使用.env文件python-dotenv管理密钥并将.env加入.gitignore。坑对“成功”的定义过于简单。最初只检查HTTP状态码为200。后来发现有些请求返回200但响应体里是{error: model overloaded}。必须解析响应体验证业务逻辑是否成功。坑压测环境与生产环境网络差异。在办公室网络测试一切良好上线后用户反馈慢。原因是办公室到云服务商是优质线路而部分用户网络质量差。压测最好在贴近生产网络环境的位置进行比如使用与后端服务同区域的云服务器发起压测。坑忽略“预热”阶段。直接开始高并发压测前几秒的响应时间会非常长因为服务端冷启动、连接池建立等。在正式记录数据前先以一个较低的并发运行1-2分钟让系统进入稳定状态。Locust的--run-time包含了预热期分析数据时可以剔除前期的异常值。用Locust对大模型API进行并发极限评估本质上是一个通过科学实验逼近系统真相的过程。它不能保证线上百分百不出问题但能极大地消除不确定性让你对系统的承载能力、响应特性和成本结构有一个量化的、坚实的理解。从编写一个模拟真实用户行为的脚本开始到阶梯式加压观察拐点再到深入分析性能瓶颈每一步都需要耐心和细致。当你能够清晰地说出“我们的服务在200并发用户下P99响应时间能稳定在3秒以内预计每小时API成本约为X元”时你和你的团队就拥有了做出更优技术决策和产品规划的底气。