Qwen1.5-72B 模型的完整性能测试脚本(Python),涵盖:

✅ 推理延迟 & 吞吐量
✅ Token 生成速度
✅ 显存占用监控
✅ 多轮对话 & 长文本支持
✅ 自动化质量评估(可选)
✅ 压力并发测试(使用 Locust)


🧰 环境准备

1. 硬件要求(推荐)

  • GPU:至少 2× A100 80GB 或 H100,显存不足会OOM
  • vLLM 支持 PagedAttention,可降低显存需求

2. 安装依赖

pip install torch transformers accelerate vllm datasets locust psutil tabulate tqdm

✅ 强烈推荐使用 vLLM —— 目前推理 Qwen1.5-72B 最高效的引擎


📜 脚本一:基础性能测试(单请求)

# qwen1.5_72b_benchmark.py
import time
import torch
from vllm import LLM, SamplingParams
from transformers import AutoTokenizer
import psutil
import GPUtil

MODEL_PATH = "Qwen/Qwen1.5-72B"  # 或本地路径 /path/to/qwen1.5-72b

def get_gpu_memory_usage():
    gpus = GPUtil.getGPUs()
    if gpus:
        return sum([gpu.memoryUsed for gpu in gpus])
    return 0

def run_inference_test():
    print("🚀 加载 Qwen1.5-72B 模型(使用 vLLM)...")
    start_load = time.time()

    # 初始化 vLLM 引擎(启用 tensor_parallel 支持多卡)
    llm = LLM(
        model=MODEL_PATH,
        tensor_parallel_size=2,      # 根据GPU数量调整
        dtype="bfloat16",            # 或 "float16"
        max_model_len=32768,         # 支持长上下文
        gpu_memory_utilization=0.95, # 显存利用率
        enforce_eager=False,         # 启用 CUDA Graph 提升性能
    )

    load_time = time.time() - start_load
    tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH)

    print(f"✅ 模型加载完成,耗时: {load_time:.2f} 秒")
    print(f"📊 初始显存占用: {get_gpu_memory_usage():.1f} MB")

    # 测试 prompt
    prompts = [
        "解释量子纠缠的基本原理。",
        "写一个 Python 快速排序函数,并添加注释。",
        "请总结《三体》第一部的主要情节,不超过200字。"
    ]

    sampling_params = SamplingParams(
        temperature=0.7,
        top_p=0.95,
        max_tokens=512,
        stop_token_ids=[tokenizer.eos_token_id]
    )

    for i, prompt in enumerate(prompts):
        print(f"\n--- 测试 Prompt {i+1} ---")
        print(f"📥 输入: {prompt[:100]}...")

        # 记录开始时间
        start_time = time.time()
        outputs = llm.generate([prompt], sampling_params)
        end_time = time.time()

        generated_text = outputs[0].outputs[0].text
        token_ids = outputs[0].outputs[0].token_ids
        num_tokens = len(token_ids)
        latency = end_time - start_time
        tokens_per_sec = num_tokens / latency

        print(f"📤 输出 (前200字符): {generated_text[:200]}...")
        print(f"⏱️  总延迟: {latency:.3f} 秒")
        print(f"⚡ Token 生成速度: {tokens_per_sec:.2f} tokens/s")
        print(f"🔢 生成 Token 数: {num_tokens}")
        print(f"📈 当前显存占用: {get_gpu_memory_usage():.1f} MB")

    return llm, tokenizer

if __name__ == "__main__":
    llm, tokenizer = run_inference_test()

🧪 脚本二:压力测试(Locust 并发模拟)

创建 locustfile.py

# locustfile.py
from locust import HttpUser, task, between
import json

class QwenUser(HttpUser):
    wait_time = between(1, 3)
    host = "http://localhost:8000"  # vLLM OpenAI API 兼容服务地址

    @task
    def generate_text(self):
        payload = {
            "model": "qwen1.5-72b",
            "messages": [
                {"role": "user", "content": "请用通俗语言解释相对论。"}
            ],
            "max_tokens": 256,
            "temperature": 0.7
        }
        headers = {"Content-Type": "application/json"}
        with self.client.post("/v1/chat/completions", json=payload, headers=headers, catch_response=True) as response:
            if response.status_code != 200:
                response.failure(f"HTTP {response.status_code}")
            else:
                try:
                    data = response.json()
                    usage = data["usage"]
                    total_tokens = usage["total_tokens"]
                    print(f"✅ 生成 {total_tokens} tokens")
                except Exception as e:
                    response.failure(f"解析失败: {e}")

启动 vLLM OpenAI API 服务(终端1)

python -m vllm.entrypoints.openai.api_server \
  --model Qwen/Qwen1.5-72B \
  --tensor-parallel-size 2 \
  --dtype bfloat16 \
  --max-model-len 32768 \
  --port 8000

启动 Locust 压测(终端2)

locust -f locustfile.py --users 50 --spawn-rate 5 --run-time 5m

访问 http://localhost:8089 查看实时压测报告。


📊 脚本三:质量评估(使用 TruthfulQA + 人工评分模拟)

# quality_eval.py
from datasets import load_dataset
from vllm import LLM, SamplingParams
from rouge_score import rouge_scorer
import re

def normalize_answer(s):
    """Lower text and remove punctuation, articles and extra whitespace."""
    def remove_articles(text):
        return re.sub(r'\b(a|an|the)\b', ' ', text)
    def white_space_fix(text):
        return ' '.join(text.split())
    def remove_punc(text):
        exclude = set(string.punctuation)
        return ''.join(ch for ch in text if ch not in exclude)
    def lower(text):
        return text.lower()
    return white_space_fix(remove_articles(remove_punc(lower(s))))

def exact_match_score(prediction, ground_truth):
    return normalize_answer(prediction) == normalize_answer(ground_truth)

# 加载评测集(示例:TruthfulQA)
dataset = load_dataset("truthful_qa", "generation", split="validation[:50]")  # 取50条快速测试

llm = LLM(model="Qwen/Qwen1.5-72B", tensor_parallel_size=2, dtype="bfloat16")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen1.5-72B")

sampling_params = SamplingParams(temperature=0, max_tokens=256)  # 温度=0确保确定性输出

correct = 0
total = 0
rougeL_scores = []

scorer = rouge_scorer.RougeScorer(['rougeL'], use_stemmer=True)

for item in dataset:
    question = item['question']
    correct_answers = item['correct_answers']  # list

    prompt = f"问题:{question}\n请直接简要回答:"
    outputs = llm.generate([prompt], sampling_params)
    prediction = outputs[0].outputs[0].text.strip()

    # 判断是否匹配任一正确答案
    is_correct = any(exact_match_score(prediction, ans) for ans in correct_answers)
    if is_correct:
        correct += 1
    total += 1

    # 计算 ROUGE-L(与第一个标准答案比)
    if correct_answers:
        scores = scorer.score(correct_answers[0], prediction)
        rougeL_scores.append(scores['rougeL'].fmeasure)

    print(f"[{total}/50] Q: {question[:50]}... | Pred: {prediction[:60]}... | 正确: {'✅' if is_correct else '❌'}")

print(f"\n🎯 准确率: {correct}/{total} = {correct/total*100:.2f}%")
print(f"📈 平均 ROUGE-L F1: {sum(rougeL_scores)/len(rougeL_scores):.4f}")

📈 脚本四:生成性能报告(整合版)

# report_generator.py
from tabulate import tabulate
import json

results = {
    "model": "Qwen1.5-72B",
    "hardware": "2×A100 80GB",
    "framework": "vLLM 0.4.0",
    "metrics": {
        "load_time_sec": 42.3,
        "first_token_latency_ms": 310,
        "tokens_per_second": 89.5,
        "max_concurrent_requests": 45,
        "gpu_memory_MB": 142000,
        "accuracy_truthfulqa": 76.8,
        "rougeL_avg": 0.6241,
        "hallucination_rate_est": 9.2  # 需人工或工具辅助评估
    }
}

def print_report(data):
    table = [
        ["模型", data["model"]],
        ["硬件", data["hardware"]],
        ["推理框架", data["framework"]],
        ["---", "---"],
        ["加载时间", f"{data['metrics']['load_time_sec']:.1f} s"],
        ["首Token延迟", f"{data['metrics']['first_token_latency_ms']} ms"],
        ["Token生成速度", f"{data['metrics']['tokens_per_second']:.1f} token/s"],
        ["最大并发数", f"{data['metrics']['max_concurrent_requests']} req/s"],
        ["显存占用", f"{data['metrics']['gpu_memory_MB']/1024:.1f} GB"],
        ["TruthfulQA准确率", f"{data['metrics']['accuracy_truthfulqa']:.1f}%"],
        ["平均ROUGE-L", f"{data['metrics']['rougeL_avg']:.4f}"],
        ["预估幻觉率", f"{data['metrics']['hallucination_rate_est']:.1f}%"],
    ]
    print("\n📊 Qwen1.5-72B 性能测试报告")
    print(tabulate(table, headers=["指标", "值"], tablefmt="pretty"))

print_report(results)

输出示例:

📊 Qwen1.5-72B 性能测试报告
+---------------------+-----------+
|        指标         |    值     |
+---------------------+-----------+
|        模型         | Qwen1.5-72B |
|        硬件         | 2×A100 80GB |
|      推理框架       | vLLM 0.4.0 |
|         ---         |    ---    |
|      加载时间       |   42.3 s  |
|    首Token延迟      |   310 ms  |
|   Token生成速度     | 89.5 token/s |
|     最大并发数      | 45 req/s  |
|      显存占用       |  138.7 GB |
| TruthfulQA准确率    |   76.8%   |
|     平均ROUGE-L     |  0.6241   |
|    预估幻觉率       |   9.2%    |
+---------------------+-----------+

⚙️ 优化建议(针对 Qwen1.5-72B)

  1. 量化部署(降低显存/提升速度):

    # 使用 AWQ 4bit 量化版(如果官方提供)
    llm = LLM(model="Qwen/Qwen1.5-72B-AWQ", quantization="AWQ")
    
  2. 启用 Prefix Caching(重复Prompt加速):

    llm = LLM(..., enable_prefix_caching=True)
    
  3. 调整 Block Size(平衡显存与吞吐):

    llm = LLM(..., block_size=32)  # 默认16,增大可提升长文本性能
    
  4. 使用 Continuous Batching(vLLM默认开启,大幅提升并发效率)


📁 目录结构建议

qwen1.5-72b-test/
├── benchmark.py          # 基础性能测试
├── locustfile.py         # 压力测试
├── quality_eval.py       # 质量评估
├── report_generator.py   # 报告生成
├── prompts/              # 自定义测试Prompt集
├── results/              # 保存测试结果JSON/CSV
└── README.md             # 测试说明文档

✅ 执行流程

  1. 启动 vLLM 服务或直接运行 benchmark.py
  2. 运行质量评估脚本(可选)
  3. 启动 Locust 压测(可选)
  4. 生成并导出报告
  5. 对比不同参数/量化版本 → 选择最优配置
Logo

码道开发者社区,聚焦华为云码道 CodeArts 代码智能体,沉淀 Agent、Skill、鸿蒙开发实战内容,供开发者查阅资料、交流技术、分享工程实践

更多推荐