Tao-8k模型部署压力测试：使用Python脚本模拟高并发请求

最近在星图GPU平台上部署了Tao-8k模型，服务跑起来之后，一个很实际的问题就摆在了面前：这服务到底能扛住多少用户同时访问？性能瓶颈在哪里？为了心里有底，也为了给后续的生产环境扩容提供数据支撑，我决定写个压力测试脚本，好好“压榨”一下这个服务。

这篇文章，我就手把手带你用Python写一个轻量但实用的压力测试工具。我们会用到asyncio和aiohttp这两个库来模拟高并发请求，然后收集关键的性能指标，比如每秒查询率（QPS）、响应时间分布以及错误率。整个过程不复杂，但得出的数据对于评估服务能力和做容量规划非常有价值。

1. 环境准备与工具选择

在开始写代码之前，我们先快速过一下需要准备的东西。你不需要是Python专家，只要有一些基础就能跟上。

你需要准备的环境：

• Python 3.7+：这是必须的，因为我们会用到一些较新的异步特性。你可以通过命令行输入 python --version 来确认。
• 安装必要的库：我们主要依赖 aiohttp 来发送异步HTTP请求，用 asyncio 来管理并发任务。通过pip安装即可：

  pip install aiohttp
  

  asyncio 是Python标准库的一部分，通常无需额外安装。

为什么选择asyncio + aiohttp？ 简单来说，就是为了高效。传统的多线程或多进程方式在模拟成百上千个并发用户时，会消耗大量的系统资源（内存、CPU上下文切换）。而asyncio基于事件循环，可以在单个线程内处理大量并发的网络I/O操作，非常适合压力测试这种I/O密集型场景。aiohttp则是一个基于asyncio的HTTP客户端/服务器库，用起来非常顺手。

你的Tao-8k服务地址：确保你的模型服务已经在星图GPU平台上部署并正常运行，并且你知道它的API访问端点（Endpoint），例如 http://your-service-address/v1/chat/completions。

2. 编写核心压力测试脚本

接下来，我们一步步构建脚本。我会把完整的代码拆解开讲解，你可以先通读一遍，再动手实践。

2.1 构建异步请求函数

这是脚本的心脏，负责向Tao-8k服务发送单个请求。我们把它定义为一个异步函数。

import aiohttp
import asyncio
import time
import logging
from typing import Dict, Any

async def send_request(session: aiohttp.ClientSession, url: str, payload: Dict[str, Any], request_id: int):
    """
    发送单个异步请求到Tao-8k服务。
    
    Args:
        session: aiohttp客户端会话，用于连接复用。
        url: 模型服务的API地址。
        payload: 请求的JSON数据体。
        request_id: 当前请求的ID，用于日志追踪。
    
    Returns:
        tuple: (请求ID, 状态码, 耗时(秒), 是否成功)
    """
    start_time = time.time()
    status_code = 0
    success = False
    
    try:
        # 设置一个合理的超时时间，例如30秒
        timeout = aiohttp.ClientTimeout(total=30)
        async with session.post(url, json=payload, timeout=timeout) as response:
            status_code = response.status
            # 对于压力测试，我们通常只关心HTTP状态码，不一定需要解析返回的完整内容
            # 如果状态码是2xx，我们认为请求成功
            if 200 <= status_code < 300:
                success = True
                # 可选：快速读取并丢弃响应体，确保连接正确关闭
                await response.read()
            else:
                # 记录非成功响应
                logging.warning(f"请求 {request_id} 失败，状态码: {status_code}")
    except asyncio.TimeoutError:
        logging.error(f"请求 {request_id} 超时")
        status_code = -1  # 用-1表示超时
    except Exception as e:
        logging.error(f"请求 {request_id} 发生异常: {e}")
        status_code = -2  # 用-2表示其他异常
    
    elapsed_time = time.time() - start_time
    return request_id, status_code, elapsed_time, success

代码解读：

1. async def：这定义了一个协程（coroutine），是asyncio并发的基础单元。
2. aiohttp.ClientSession：强烈建议使用会话（Session），因为它会自动管理连接池，复用TCP连接，能极大提升高并发下的请求效率。
3. ClientTimeout：设置超时非常重要，防止某些请求卡住，影响整个测试进程。
4. 异常处理：我们捕获了超时和其他通用异常，确保单个请求的失败不会导致整个测试崩溃，并将不同的错误类型用状态码区分。

2.2 创建并发测试任务

现在我们需要一个函数来创建和管理一大批并发请求任务。

async def run_load_test(url: str, payload: Dict[str, Any], total_requests: int, concurrency: int):
    """
    运行压力测试。
    
    Args:
        url: 目标服务URL。
        payload: 请求负载。
        total_requests: 总共要发送的请求数。
        concurrency: 并发数（同时进行的最大请求数）。
    """
    # 用于收集所有请求的结果
    results = []
    
    # 创建TCP连接器，限制连接池大小，模拟真实并发
    connector = aiohttp.TCPConnector(limit=concurrency, force_close=False)
    
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = []
        request_sent = 0
        
        # 使用信号量（Semaphore）控制最大并发数
        semaphore = asyncio.Semaphore(concurrency)
        
        async def sem_task(req_id):
            async with semaphore:  # 获取信号量，控制并发
                return await send_request(session, url, payload, req_id)
        
        print(f"开始压力测试，目标请求数: {total_requests}, 并发数: {concurrency}")
        test_start_time = time.time()
        
        # 分批创建任务，避免一次性创建过多任务消耗内存
        while request_sent < total_requests:
            # 准备下一批任务
            batch_size = min(concurrency * 2, total_requests - request_sent)
            for i in range(batch_size):
                task = asyncio.create_task(sem_task(request_sent + i + 1))
                tasks.append(task)
            request_sent += batch_size
            
            # 等待当前批次的部分任务完成，再继续添加，实现平滑的压力施加
            if len(tasks) >= concurrency * 3:
                done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
                for done_task in done:
                    results.append(done_task.result())
                # 更新tasks列表，只保留未完成的任务
                tasks = list(pending)
        
        # 等待所有剩余任务完成
        if tasks:
            batch_results = await asyncio.gather(*tasks, return_exceptions=False)
            results.extend(batch_results)
        
        test_duration = time.time() - test_start_time
        print(f"所有请求发送完毕，总耗时: {test_duration:.2f} 秒")
        
    return results, test_duration

关键点说明：

1. TCPConnector(limit=concurrency)：这很关键。它将HTTP客户端的最大同时连接数限制为我们设定的并发数，更真实地模拟并发用户。
2. asyncio.Semaphore：信号量是控制并发度的经典工具。async with semaphore: 确保同一时间只有 concurrency 个协程在执行 send_request。
3. 分批任务管理：我们没有一次性创建 total_requests 个任务，而是分批创建和管理。这样做的好处是内存使用更平稳，并且可以更早地开始收集已完成任务的结果，实现“边发送边回收”。
4. asyncio.create_task：将协程包装成任务（Task），调度其并发执行。
5. asyncio.gather：用于等待一批任务全部完成。

2.3 准备测试数据与配置

我们需要定义要发送给Tao-8k模型的请求内容，并设置测试参数。

def get_test_payload():
    """构造发送给Tao-8k模型的请求数据。"""
    # 这是一个简单的对话请求示例，你可以根据你的模型API文档进行调整
    payload = {
        "model": "Tao-8B",  # 模型名称
        "messages": [
            {"role": "user", "content": "请用一句话介绍一下你自己。"}
        ],
        "max_tokens": 100,
        "temperature": 0.7,
        "stream": False  # 压力测试建议关闭流式输出，简化处理
    }
    return payload

# 测试配置
TARGET_URL = "http://替换为你的实际服务地址和端口/v1/chat/completions"  # TODO: 替换成你的服务地址
TOTAL_REQUESTS = 100  # 计划发送的总请求数
CONCURRENCY_LEVEL = 10  # 并发用户数

请注意：

• 务必将 TARGET_URL 替换为你部署在星图GPU平台上的Tao-8k服务的真实API地址。
• TOTAL_REQUESTS 和 CONCURRENCY_LEVEL 是核心参数。建议从小开始，比如先设置 TOTAL_REQUESTS=50, CONCURRENCY_LEVEL=5，观察服务状态和脚本运行情况，再逐步调高。

3. 运行测试并分析结果

脚本写好了，数据也配置了，现在我们来运行它，并看看如何解读结果。

3.1 主函数与结果分析

我们添加一个主函数来组织整个流程，并编写一个函数来统计分析收集到的数据。

import statistics

def analyze_results(results, test_duration):
    """分析压力测试结果，计算关键指标。"""
    if not results:
        print("未收集到任何结果。")
        return
    
    successful_reqs = [r for r in results if r[3]]  # 成功的请求
    failed_reqs = [r for r in results if not r[3]]  # 失败的请求
    
    total_count = len(results)
    success_count = len(successful_reqs)
    failure_count = len(failed_reqs)
    
    # 计算QPS (Queries Per Second)
    qps = total_count / test_duration if test_duration > 0 else 0
    
    # 提取所有成功请求的耗时
    latencies = [r[2] for r in successful_reqs]
    
    if latencies:
        avg_latency = statistics.mean(latencies)
        p50_latency = statistics.median(latencies)
        # 计算P90和P95（需要排序）
        sorted_latencies = sorted(latencies)
        p90_index = int(len(sorted_latencies) * 0.9)
        p95_index = int(len(sorted_latencies) * 0.95)
        p90_latency = sorted_latencies[p90_index] if p90_index < len(sorted_latencies) else sorted_latencies[-1]
        p95_latency = sorted_latencies[p95_index] if p95_index < len(sorted_latencies) else sorted_latencies[-1]
        max_latency = max(latencies)
        min_latency = min(latencies)
    else:
        avg_latency = p50_latency = p90_latency = p95_latency = max_latency = min_latency = 0
    
    # 错误率
    error_rate = (failure_count / total_count) * 100 if total_count > 0 else 0
    
    # 打印报告
    print("\n" + "="*50)
    print("压力测试结果报告")
    print("="*50)
    print(f"总请求数: {total_count}")
    print(f"成功请求: {success_count}")
    print(f"失败请求: {failure_count}")
    print(f"错误率: {error_rate:.2f}%")
    print(f"测试总时长: {test_duration:.2f} 秒")
    print(f"平均QPS: {qps:.2f}")
    print("\n--- 响应时间 (秒) ---")
    print(f"平均响应时间: {avg_latency:.3f}")
    print(f"最小响应时间: {min_latency:.3f}")
    print(f"最大响应时间: {max_latency:.3f}")
    print(f"P50 (中位数): {p50_latency:.3f}")
    print(f"P90: {p90_latency:.3f}")
    print(f"P95: {p95_latency:.3f}")
    print("="*50)
    
    # 简单分析
    if error_rate > 5:
        print("警告：错误率较高，服务可能不稳定或配置不足。")
    if p95_latency > 5.0:  # 假设5秒为可接受上限，可根据业务调整
        print("提示：P95响应时间较长，部分用户体验可能受影响。")
    
    # 打印失败请求详情（前10个）
    if failed_reqs:
        print(f"\n失败请求详情 (前{min(10, len(failed_reqs))}个):")
        for req in failed_reqs[:10]:
            print(f"  请求ID: {req[0]}, 状态码: {req[1]}, 耗时: {req[2]:.3f}s")

async def main():
    """主函数，组织整个测试流程。"""
    payload = get_test_payload()
    print(f"测试目标URL: {TARGET_URL}")
    print(f"测试负载: {payload}")
    
    results, duration = await run_load_test(TARGET_URL, payload, TOTAL_REQUESTS, CONCURRENCY_LEVEL)
    analyze_results(results, duration)

if __name__ == "__main__":
    # 设置日志级别，避免过多调试信息
    logging.basicConfig(level=logging.WARNING)
    # 运行主异步函数
    asyncio.run(main())

关键指标解读：

• QPS：系统每秒能成功处理的请求数。这是衡量吞吐量的核心指标。越高越好。
• 错误率：失败请求占总请求的比例。生产环境通常要求低于1%或0.1%。
• 平均响应时间：所有成功请求耗时的平均值。但平均值容易被极端值影响。
• P50/P90/P95：百分位响应时间。例如P95=2.5秒，表示95%的请求响应时间在2.5秒以内。P95/P99比平均值更能反映用户体验，因为它们代表了大多数用户的感受。
• 最大/最小响应时间：了解响应时间的波动范围。

3.2 运行你的第一个测试

1. 将上述所有代码块按顺序保存到一个文件中，例如 tao8k_stress_test.py。
2. 修改 TARGET_URL 为你的真实服务地址。
3. 打开终端，导航到脚本所在目录。
4. 运行命令：

   python tao8k_stress_test.py
   

如果一切正常，你会看到脚本开始打印发送请求的日志，最后输出一份格式化的结果报告。

4. 进阶技巧与瓶颈分析初探

跑通基础测试后，我们可以让脚本变得更强大，并尝试做一些初步的性能瓶颈推断。

4.1 让测试更贴近真实场景

单一的固定请求负载可能无法反映真实情况。我们可以改进脚本：

• 使用多个不同的请求负载：创建一个负载列表，在发送请求时随机选取。

  payload_list = [
      {"model": "Tao-8B", "messages": [{"role": "user", "content": "问题A"}]},
      {"model": "Tao-8B", "messages": [{"role": "user", "content": "问题B"}]},
      # ... 更多负载
  ]
  # 在send_request中随机选择 payload
  

• 模拟思考时间（Think Time）：真实用户操作间有间隔。可以在并发任务中随机加入 await asyncio.sleep(random.uniform(0.5, 2)) 来模拟。
• 阶梯式增压（Ramp-up）：不是一开始就达到最大并发，而是逐渐增加并发用户数，观察系统在不同压力下的表现。这需要更复杂的任务调度逻辑。

4.2 从结果推断可能的瓶颈

压力测试的结果数据是发现系统弱点的第一手资料。这里有一些简单的分析思路：

1. QPS上不去，响应时间剧增：当并发数增加，QPS达到一个平台后不再增长，甚至下降，而响应时间却直线上升。这通常是服务端处理能力达到极限的标志。瓶颈可能在：

   • GPU算力：Tao-8k模型推理本身非常消耗GPU资源。观察星图平台提供的GPU监控（如GPU利用率、显存使用率），如果持续接近100%，说明GPU是瓶颈。
   • 服务框架/后端：可能是Python Web框架（如FastAPI）的工作进程/线程数不足，或者模型加载、预处理/后处理逻辑存在性能问题。

2. 错误率随并发升高：如果错误率（特别是超时错误、5xx错误）随着并发数增加而显著升高，可能意味着：

   • 服务端资源耗尽：内存不足、连接数满、线程池耗尽。
   • 客户端配置问题：我们脚本中的 TCPConnector 限制或操作系统本身的文件描述符限制可能太低。

3. 响应时间P95远大于平均值：说明有少量请求特别慢，拖了后腿。这可能是因为：

   • 服务端资源竞争：某个时刻大量请求同时到达，导致部分请求排队等待GPU或CPU。
   • “冷启动”或缓存未命中：第一个请求或某些特定请求触发了较慢的路径（如加载新数据）。

给你的行动建议：

• 监控是关键：在运行压力测试时，同时观察服务部署平台的监控面板（CPU、内存、GPU、网络IO）。
• 控制变量法：一次只改变一个参数（如并发数、请求负载大小），观察指标变化，更容易定位问题。
• 从简到繁：先用简单的请求测试出基准性能，再逐步增加复杂度。

5. 总结

通过这个教程，我们完成了一个从零开始的、针对Tao-8k模型服务的Python压力测试脚本。整个过程其实并不复杂，核心就是利用asyncio和aiohttp来高效地模拟大量并发用户，然后系统地收集和分析响应时间、成功率、吞吐量这些关键指标。

实际用下来，这个脚本能很快帮你摸清服务的“底细”。比如，在星图GPU平台的某个配置下，你的服务可能能轻松应对20个并发用户，QPS达到15，响应时间也很稳定。但当你把并发调到50，可能就会发现响应时间开始飙升，错误率也上来了，这时候你就知道当前配置的瓶颈大概在哪里。

这些数据对于生产环境规划特别有用。你知道在什么压力下服务会开始变慢、出错，就能以此为依据来决定是否需要升级GPU规格、增加服务实例数量，或者优化一下服务代码。当然，这只是一个起点，更复杂的测试场景，比如混合读写、长时间稳定性测试（耐力测试），都可以在这个基础上进行扩展。

最后提醒一下，压力测试本身也会对服务造成负担，最好在业务低峰期进行，并且提前做好预案。希望这个工具能帮你更自信地部署和管理你的AI服务。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。