From c099af6bc71289e4cb5d0bef2e5f5f99978c4b7c Mon Sep 17 00:00:00 2001 From: yumoqing Date: Fri, 24 Jul 2026 11:01:48 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=88=A0=E9=99=A4load=5Ftest.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- load_test.py | 163 --------------------------------------------------- 1 file changed, 163 deletions(-) delete mode 100644 load_test.py diff --git a/load_test.py b/load_test.py deleted file mode 100644 index 9f1cfdf..0000000 --- a/load_test.py +++ /dev/null @@ -1,163 +0,0 @@ -#!/usr/bin/env python3 -"""并发压力测试 llmage /v1/chat/completions，统计 TTFB / 完成时间 / QPM""" - -import asyncio -import aiohttp -import time -import json -import sys -import statistics -from dataclasses import dataclass, field -from typing import List - -URL = "https://token.opencomputing.cn/llmage/v1/chat/completions" -TOKEN = "V9J41PngWBUU6gdHWJWDJ" -MODEL = "qwen3.6-35b-a3b" -DURATION = 180 # 3 分钟 -CONCURRENCIES = [10, 50, 100, 200] - - -@dataclass -class ReqStat: - prompt_idx: int - start_ts: float - first_byte_ts: float | None = None - end_ts: float | None = None - - -async def worker(session: aiohttp.ClientSession, idx: int, stats_out: list): - """单个请求：发送 stream 请求，记录首字时间和完成时间""" - prompt = f"请用一句话介绍你自己，编号{idx}" - payload = { - "model": MODEL, - "stream": True, - "messages": [{"role": "user", "content": prompt}], - } - stat = ReqStat(prompt_idx=idx, start_ts=time.monotonic()) - try: - async with session.post( - URL, - json=payload, - headers={ - "Content-Type": "application/json", - "Authorization": f"Bearer {TOKEN}", - }, - timeout=aiohttp.ClientTimeout(total=120), - ) as resp: - first = True - async for line in resp.content: - if first: - stat.first_byte_ts = time.monotonic() - first = False - # 读完所有 chunk 才算完成 - stat.end_ts = time.monotonic() - except Exception as e: - # 异常请求也记录（TTFB=None 表示失败） - stat.end_ts = time.monotonic() - stats_out.append(stat) - - -async def run_concurrency(concurrency: int): - """以固定并发运行 DURATION 秒，持续发起新请求""" - stats: List[ReqStat] = [] - idx = 0 - stop_at = time.monotonic() + DURATION - - connector = aiohttp.TCPConnector(limit=concurrency + 20, force_close=True) - async with aiohttp.ClientSession(connector=connector) as session: - tasks: list[asyncio.Task] = [] - - while time.monotonic() < stop_at: - # 保持并发数：补满到 concurrency - while len(tasks) < concurrency and time.monotonic() < stop_at: - idx += 1 - tasks.append( - asyncio.create_task(worker(session, idx, stats)) - ) - - if not tasks: - break - - # 等待任意一个完成，腾出槽位 - done, tasks = await asyncio.wait( - tasks, return_when=asyncio.FIRST_COMPLETED, timeout=0.5 - ) - # 清理已完成的 - tasks = list(tasks) - - # 时间到，等待所有进行中的请求完成 - if tasks: - await asyncio.wait(tasks) - - return stats - - -def analyze(name: str, stats: List[ReqStat]): - """分析并打印统计""" - ttfb_list = [s.first_byte_ts - s.start_ts for s in stats if s.first_byte_ts] - total_list = [s.end_ts - s.start_ts for s in stats if s.end_ts and s.first_byte_ts] - failed = sum(1 for s in stats if s.first_byte_ts is None) - total_req = len(stats) - elapsed = DURATION - qpm = total_req / (elapsed / 60) - - print(f"\n{'='*60}") - print(f" 并发={name} | 运行{DURATION}s | 总请求={total_req} | 失败={failed}") - print(f"{'='*60}") - if ttfb_list: - print(f" TTFB (s): min={min(ttfb_list):.3f} avg={statistics.mean(ttfb_list):.3f} " - f"p50={statistics.median(ttfb_list):.3f} p95={_pct(ttfb_list, 95):.3f} p99={_pct(ttfb_list, 99):.3f}") - if total_list: - print(f" 完成 (s): min={min(total_list):.3f} avg={statistics.mean(total_list):.3f} " - f"p50={statistics.median(total_list):.3f} p95={_pct(total_list, 95):.3f} p99={_pct(total_list, 99):.3f}") - print(f" QPM: {qpm:.1f}") - print(f" QPS: {total_req / elapsed:.1f}") - - # 按分钟分段统计 - for minute in range(int(elapsed / 60)): - win_start = minute * 60 - win_end = (minute + 1) * 60 - cnt = sum(1 for s in stats if s.end_ts and (s.end_ts - s.start_ts) >= 0 - and win_start <= (s.start_ts - stats[0].start_ts) < win_end) - print(f" 第{minute+1}分钟完成请求数: {cnt}") - - return { - "concurrency": name, - "total": total_req, - "failed": failed, - "ttfb_avg": statistics.mean(ttfb_list) if ttfb_list else None, - "ttfb_p50": statistics.median(ttfb_list) if ttfb_list else None, - "ttfb_p95": _pct(ttfb_list, 95) if ttfb_list else None, - "total_avg": statistics.mean(total_list) if total_list else None, - "total_p50": statistics.median(total_list) if total_list else None, - "qpm": qpm, - } - - -def _pct(data, p): - return sorted(data)[int(len(data) * p / 100)] - - -async def main(): - results = [] - for c in CONCURRENCIES: - print(f"\n>>> 开始测试 并发={c} ...") - stats = await run_concurrency(c) - r = analyze(str(c), stats) - results.append(r) - - # 汇总表格 - print(f"\n{'='*60}") - print(" 汇总对比") - print(f"{'='*60}") - print(f" {'并发':>6} {'总请求':>8} {'失败':>5} {'TTFB_avg':>9} {'TTFB_p50':>9} {'TTFB_p95':>9} {'完成_avg':>9} {'QPM':>8}") - for r in results: - print(f" {r['concurrency']:>6} {r['total']:>8} {r['failed']:>5} " - f"{r['ttfb_avg']:.3f}s" if r['ttfb_avg'] else "N/A".rjust(9) + " " - f"{(r['ttfb_p50'] or 0):.3f}s".rjust(9) + " " - f"{(r['ttfb_p95'] or 0):.3f}s".rjust(9) + " " - f"{r['total_avg']:.3f}s".rjust(9) if r['total_avg'] else "N/A".rjust(9)) - - -if __name__ == "__main__": - asyncio.run(main())