Qwen2.5-72B-Instruct并发性能测试：基于vLLM 0.7.3的QwQ-32B/Qwen2.5-72B/32B三模型8卡A100评测

发表于 | 分类于 | 1.6k | 8 分钟

vLLM 推理框架—Qwen2.5-72B-Instruct并发测试

vLLM 是一种高性能的 GPU 推理框架，通过 PagedAttention 技术优化 GPU 内存效率和吞吐量，适用于大规模高并发场景。

QwQ-32B

基于 llm_management_20250224:py311（vllm==0.7.3）首次部署

模型名称：QwQ-32B

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 创建容器（llm_management_20250224:py311中vllm==0.7.3）
sudo docker run --gpus all -it -d --network llm-management-network --name llm_management --ipc=host -P -p 8900-8997:8900-8997 -v /data/models:/models llm_management_20250224:py311 /bin/bash

# 开启v1加速
export VLLM_USE_V1=1
export VLLM_WORKER_MULTIPROC_METHOD=spawn

# 8卡启动QwQ-32B
CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 python -m vllm.entrypoints.openai.api_server \
		--served-model-name QwQ-32B \
		--port 8996 \
		--model /models/QwQ-32B \
		--tensor-parallel-size 8 \
		--max-model-len 16384 \
		--gpu-memory-utilization 0.80 \
		--disable-log-stats \
		--api-key <YOUR_API_KEY>
		

# 测试脚本（feature_v2_sp3）
python test_qwq_batch_concurrent.py

QwQ-32B性能测试结果详情

img_01

QwQ-32B性能测试结果

img_02

img_03

Qwen2.5-72B-Instruct

基于 llm_management_20250224:py311（vllm==0.7.3）首次部署

模型名称：Qwen2.5-72B-Instruct

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 创建容器（llm_management_20250224:py311中vllm==0.7.3）
sudo docker run --gpus all -it -d --network llm-management-network --name llm_management --ipc=host -P -p 8900-8997:8900-8997 -v /data/models:/models llm_management_20250224:py311 /bin/bash

# 开启v1加速
export VLLM_USE_V1=1
export VLLM_WORKER_MULTIPROC_METHOD=spawn

# 8卡启动Qwen2.5-72B-Instruct
CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 python -m vllm.entrypoints.openai.api_server \
		--served-model-name Qwen2.5-72B-Instruct \
		--port 8996 \
		--model /models/Qwen2.5-72B-Instruct \
		--tensor-parallel-size 8 \
		--max-model-len 16384 \
		--gpu-memory-utilization 0.80 \
		--disable-log-stats \
		--api-key <YOUR_API_KEY>

Qwen2.5-72B-Instruct性能测试结果详情

img_04

Qwen2.5-72B-Instruct性能测试结果

img_05

img_06

Qwen2.5-32B-Instruct

基于 llm_management_20250224:py311（vllm==0.7.3）首次部署

模型名称：Qwen2.5-32B-Instruct

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 创建容器（llm_management_20250224:py311中vllm==0.7.3）
sudo docker run --gpus all -it -d --network llm-management-network --name llm_management --ipc=host -P -p 8900-8997:8900-8997 -v /data/models:/models llm_management_20250224:py311 /bin/bash

# 开启v1加速
export VLLM_USE_V1=1
export VLLM_WORKER_MULTIPROC_METHOD=spawn

# 8卡启动Qwen2.5-32B-Instruct
CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 python -m vllm.entrypoints.openai.api_server \
		--served-model-name Qwen2.5-32B-Instruct \
		--port 8996 \
		--model /models/Qwen2.5-32B-Instruct \
		--tensor-parallel-size 8 \
		--max-model-len 16384 \
		--gpu-memory-utilization 0.80 \
		--disable-log-stats \
		--api-key <YOUR_API_KEY>

Qwen2.5-32B-Instruct性能测试结果详情

img_07

Qwen2.5-32B-Instruct性能测试结果

img_08

img_09

评测基准

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
# 以vllm方式启用，openai client调用，测试qwq模型并发效果
import concurrent.futures
import os
import time
from datetime import datetime

import pandas as pd
from loguru import logger
from openai import OpenAI
from transformers import AutoTokenizer

from utils.generate_prompt_util import sample_sonnet_requests


# Initialize OpenAI client
client = OpenAI(
    api_key="<YOUR_API_KEY>",
    base_url="http://10.39.214.107:8996/v1",
)


# 初始化结果存储
results = []

def send_post_request(prompt, model, stream, max_tokens):
    """
    发送POST请求
    """
    response_start_time = datetime.now()
    collected_messages = []
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            stream=stream,
            max_tokens=max_tokens
        )

        if stream:
            first_token_return_time = None
            # 首token标志位
            ttft_flag = True
            for chunk in response:
                if not chunk.choices:
                    continue
                chunk_message = chunk.choices[0].delta
                if not chunk_message:
                    continue

                if chunk_message.content is not None:
                    if ttft_flag:
                        logger.info(chunk_message.content)
                        first_token_return_time = datetime.now()
                        logger.info(f"First token return time: {first_token_return_time - response_start_time}")
                        ttft_flag = False
                    collected_messages.append(chunk_message.content)

            response_end_time = datetime.now()
            logger.info(f"Response running time: {response_end_time - response_start_time}")
            logger.info(f"Collected messages: {''.join(collected_messages)}")

            return {
                "first_token_time": (first_token_return_time - response_start_time).total_seconds() if first_token_return_time else None,
                "total_time": (response_end_time - response_start_time).total_seconds()
            }
    except Exception as e:
        logger.error(f"Exception: {e}")
        return {"first_token_time": None, "total_time": None}
        

def ensure_directory_exists(directory):
    """
    确保目录存在，如果不存在则创建它

    Args:
        directory (str): 要确保存在的目录路径
    """
    if not os.path.exists(directory):
        os.makedirs(directory)


def run_concurrent_requests(backend, num_requests, input_length, model, stream, max_tokens, dataset_path, tokenizer):
    """
    执行并发请求，并发地向指定的后端发送请求，并计算相关性能指标。

    Args:
        backend (str): 后端服务名，用于区分不同的后端环境。
        num_requests (int): 并发请求的数量。
        input_length (int): 每个请求的输入长度。
        model (str): 用于生成响应的模型名。
        stream (bool): 是否以流的方式接收响应。
        max_tokens (int): 每个响应的最大token数。
        dataset_path (str): 数据集路径，从中采样请求输入。
        tokenizer (obj): 用于文本分词的tokenizer对象。

    Returns:
        tuple: 包含以下四个元素的元组：
            request_success_rate (float): 请求成功率，成功完成的请求数占总请求数的比例。
            avg_first_token_time (float): 平均首token返回时间，单位秒，表示从请求发送到收到第一个token的平均时间。
            avg_total_time (float): 平均完成对话时间，单位秒，表示从请求发送到收到完整响应的平均时间。
            avg_token_generation_speed (float): 平均token生成速度，单位tokens/s，表示每秒生成的token数。
    """
    logger.info(f"Running {num_requests} concurrent requests with input length {input_length}...")
    input_requests = sample_sonnet_requests(
        dataset_path=dataset_path,
        num_requests=num_requests,
        input_len=input_length,
        output_len=max_tokens,
        prefix_len=50,
        tokenizer=tokenizer
    )
    test_prompts = [prompt_formatted for prompt, prompt_formatted, prompt_len, output_len in input_requests]

    with concurrent.futures.ThreadPoolExecutor(max_workers=num_requests) as executor:
        futures = [executor.submit(send_post_request, prompt, model, stream, max_tokens) for prompt in test_prompts]
        results = [future.result() for future in concurrent.futures.as_completed(futures)]

    first_token_times = [result["first_token_time"] for result in results if result["first_token_time"] is not None]
    total_times = [result["total_time"] for result in results if result["total_time"] is not None]

    avg_first_token_time = sum(first_token_times) / len(first_token_times) if first_token_times else None
    avg_total_time = sum(total_times) / len(total_times) if total_times else None
    avg_token_generation_speed = max_tokens / (avg_total_time - avg_first_token_time) if avg_total_time - avg_first_token_time else None
    request_success_rate = len(total_times) / num_requests if num_requests else None

    # Ensure the results directory exists
    results_dir = f"./results/{backend}"
    ensure_directory_exists(results_dir)

    # Data storage
    with open(
            f"{results_dir}/{backend}_{model}_{num_requests}_requests_{input_length}_{max_tokens}_tokens.txt",
            "w", encoding="utf-8") as f:
        # 总体数据
        f.write(f"请求数: {num_requests}\n")
        f.write(f"平均首token返回时间: {avg_first_token_time}\n")
        f.write(f"平均完成对话时间: {avg_total_time}\n")
        f.write(f"平均token生成速度: {avg_token_generation_speed}\n\n\n")

        # 请求数据
        if num_requests >= 5:
            # 前5个请求的ttft
            ttft_data = first_token_times[: 5]
            ttft_str = ", ".join(map(str, ttft_data))
            f.write(f"前五个请求的首token返回时间为:\n {ttft_str}\n")
            # 前5个请求的完成时间
            response_end_time_data = total_times[: 5]
            response_end_time_str = ", ".join(map(str, response_end_time_data))
            f.write(f"前五个请求的完成时间为:\n {response_end_time_str}\n")
            # 所有请求数据
            ttft_data = first_token_times
            ttft_str = ", ".join(map(str, ttft_data))
            f.write(f"所有请求的首token返回时间为:\n {ttft_str}\n")
            # 前5个请求的完成时间
            response_end_time_data = total_times
            response_end_time_str = ", ".join(map(str, response_end_time_data))
            f.write(f"所有请求的完成时间为:\n {response_end_time_str}\n")


if __name__ == "__main__":
    # 定义测试参数
    # 使用框架
    test_backend = 'vllm' # sglang, lmdeploy
    test_concurrent_levels = [1, 2, 4, 8, 16, 32, 64, 128, 256]  # Concurrency levels
    test_input_lengths = [256, 512, 1024, 2048, 4096, 8192, 16000]  # Input lengths
    test_output_lengths = [128]  # Output lengths
    test_model_name = "QwQ-32B"  # Model name
    test_stream = True  # Stream output
    test_tokenizer_dir = "./models/QwQ-32B"  # Tokenizer path
    test_dataset_path = "./datasets/sonnet_4x.txt"  # Dataset path

    # 初始化Tokenizer
    test_tokenizer = AutoTokenizer.from_pretrained(test_tokenizer_dir, trust_remote_code=True)

    for test_output_length in test_output_lengths:
        for test_concurrent_level in test_concurrent_levels:
            for test_input_length in test_input_lengths:
                test_request_success_rate, test_avg_first_token_time, test_avg_total_time, test_avg_token_generation_speed = run_concurrent_requests(
                    backend=test_backend,
                    num_requests=test_concurrent_level,
                    input_length=test_input_length,
                    model=test_model_name,
                    stream=test_stream,
                    max_tokens=test_output_length,
                    dataset_path=test_dataset_path,
                    tokenizer=test_tokenizer
                )
                results.append({
                    "Concurrency": test_concurrent_level,
                    "Input Length": test_input_length,
                    "Output Length": test_output_length,
                    "Request Success Rate (%)": f"{test_request_success_rate * 100:.1f}%",
                    "Avg First Token Time (s)": f"{test_avg_first_token_time:.6f}",
                    "Avg Total Time (s)": f"{test_avg_total_time:.6f}",
                    "Avg Token Generation Speed (token/s)": f"{test_avg_token_generation_speed:.6f}",
                    "Output Throughput(token/s)":  f"{test_concurrent_level * test_avg_token_generation_speed:.6f}"
                })

            # 每次执行完一个并发数后暂停10秒
            logger.info(f"Completed {test_concurrent_level} concurrent requests. Pausing for 10 seconds...")
            time.sleep(10)
   

    # 将结果保存到Excel文件
    df = pd.DataFrame(results)
    df.to_excel(f"./results/{test_backend}/{test_backend}_{test_model_name}.xlsx", index=False)
    logger.info("Test results saved to test_results.xlsx")
本文结束 感谢您的阅读