大模型压测实战:如何评估你的LLM在高并发下的表现?

随着大语言模型(LLM)逐渐进入企业实际业务中,如何量化其在不同负载下的性能表现成为一个关键问题。我们开发了一套异步压测框架,支持模拟不同并发数下的请求压力,并从多个维度评估模型的响应能力。

本文将深入剖析这套压测框架的设计思路,逐行解析核心代码逻辑,并结合实际测试数据,展示一份完整的压测报告截图与分析建议。无论你是LLM平台搭建者、模型服务提供者还是工程化部署人员,都能从中获取实用的参考方法。

一、测试目标与整体方案设计

我们本次压测的核心目标是:

  1. 评估 LLM 在不同并发请求下的吞吐能力(QPS)和延迟表现(平均、P99);
  2. 测量模型的 Token 生成速度(TPS)以及首 Token 响应时间(TTFT);
  3. 判断系统是否达到性能瓶颈,确定最佳并发配置;
  4. 输出统一结构化的压测报告,支持图表、表格展示。

架构整体思路如下:

  • 基于 Python 的 asyncio 实现异步并发请求;
  • 请求通过 OpenAI 格式的 Chat Completion 接口模拟,支持短文本/长文本测试;
  • 所有请求以流式(streaming)方式响应,并逐个统计 Token 输出时间;
  • 使用 rich 实现终端可视化报告,支持保存 JSON 文件备查;
  • 最终生成 markdown 或 HTML 报告便于团队协作和性能对比。

二、核心功能模块详解

1. Prompt 加载器

短文本从 txt 文件加载,每行为一个 prompt;长文本从 JSON 文件加载,结构如下:

{
  "context": "...",
  "prompt": "..."
}

代码片段:

def load_short_prompts(file_path):
    with open(file_path, "r", encoding="utf-8") as file:
        prompts = [line.strip() for line in file.readlines()]
    return prompts

def load_long_prompt_pairs(file_path):
    with open(file_path, "r", encoding="utf-8") as file:
        prompt_pairs = json.load(file)
    return prompt_pairs

2. 发起请求与响应分析

这是整个压测系统的核心:向 LLM 服务发起请求,并统计返回流中 token 的接收时间。

async def process_stream(stream):
    first_token_time = None
    total_tokens = 0
    last_token_time = None

    async for chunk in stream:
        if first_token_time is None:
            first_token_time = time.time()
        if chunk.choices[0].delta.content:
            total_tokens += 1
        if chunk.choices[0].finish_reason is not None:
            last_token_time = time.time()
            break

    total_time = last_token_time - first_token_time
    return first_token_time, total_tokens, total_time

请求函数如下:

async def make_request(prompt, system_prompt=None):
    start_time = time.time()
    stream = await client.chat.completions.create(
        model="your-model",
        messages=[
            {"role": "system", "content": system_prompt or "You are a helpful assistant."},
            {"role": "user", "content": prompt}
        ],
        stream=True
    )
    first_token_time, total_tokens, respone_time = await process_stream(stream)
    ttft = first_token_time - start_time
    tokens_per_second = total_tokens / respone_time
    return total_tokens, respone_time, tokens_per_second, ttft

3. 并发控制与任务队列

通过 asyncio.QueueSemaphore 控制并发度,确保请求按照设定的并发量稳定发送。

async def worker(queue, semaphore, results):
    while True:
        async with semaphore:
            task_id = await queue.get()
            if task_id is None:
                queue.task_done()
                break
            result = await make_request(...)
            results.append(result)
            queue.task_done()

主调度函数:

async def run_benchmark(prompts, concurrency):
    queue = asyncio.Queue()
    results = []
    semaphore = asyncio.Semaphore(concurrency)

    for i in range(len(prompts)):
        await queue.put(i)
    for _ in range(concurrency):
        await queue.put(None)

    workers = [asyncio.create_task(worker(queue, semaphore, results)) for _ in range(concurrency)]
    await queue.join()
    await asyncio.gather(*workers)
    return results

4. 数据统计与可视化输出

每轮测试会统计以下内容:

  • 平均延迟:每个请求返回所有 token 的平均耗时;
  • P99延迟:最慢 1% 请求的响应时间;
  • 平均TPS:所有请求输出 token 总数 / 总响应耗时;
  • 成功率:未超时请求数 / 总请求数;
  • TTFT:Time to First Token 平均耗时;

最终输出表格如下:

并发 QPS 平均延迟(s) P99延迟(s) TPS TTFT(s) 成功率
5 0.21 22.18 36.80 24.54 1.03 100%
35 0.89 35.19 52.96 15.60 1.18 100%
100 1.78 52.17 84.25 10.33 1.35 100%

三、压测结果截图

在这里插入图片描述

从图表可以看出:

  • 总生成 Token 数:1,484,666
  • 总测试时间:2375.33 秒
  • 总响应时间:121396.75 秒
  • 平均 Token 生成速度:12.23 tokens/sec

性能最佳并发数为 5(延迟最低),而最高并发数 100 达到最大 RPS。

四、性能建议总结

  1. 吞吐量未达瓶颈:最高并发下 QPS 仍有上升空间(1.78 req/sec),说明模型服务尚可承载更高并发;
  2. 响应延迟随并发上升而升高:从 22.18 秒增加到 52.17 秒,P99 延迟接近 1.5 分钟,应关注网络带宽或模型推理队列;
  3. TPS 随并发下降明显:并发 5 时 TPS 达 24.54,100 时下降至 10.33,表明资源调度压力增加;
  4. TTFT 表现良好:所有并发下首 token 响应基本在 1 秒左右,说明前置网络或模型预热机制到位;

五、未来优化方向

  • 支持多模型压测,自动对比不同模型在相同 prompt 下的响应能力;
  • 增加更多指标,如内存/CPU/GPU 使用率,结合 psutilnvidia-smi 数据;
  • 将终端报告转为 HTML 或 Grafana 面板实时可视化展示;
  • 接入 CI/CD 流程,模型上线前自动进行性能验证;
  • 将失败请求的 prompt 与异常日志统一存档,便于故障排查与修复;

六、结语

本文完整地介绍了一个用于测试大语言模型性能的异步压测工具。从 prompt 加载、请求发起、并发控制,到最终数据可视化,每个步骤均提供了详细的说明与代码示例。借助这套工具,工程团队可以更科学、更系统地分析和优化 LLM 在实际部署中的性能表现。

该框架特别适合用于大模型上线前的基准评估、平台部署后的性能监控,以及多模型之间的横向对比。

如果你有兴趣获取完整源码或参与共建,欢迎留言或私信交流。

Logo

技术共进,成长同行——讯飞AI开发者社区

更多推荐