diff --git a/llmage/utils.py b/llmage/utils.py index 709399e..9c2655b 100644 --- a/llmage/utils.py +++ b/llmage/utils.py @@ -364,7 +364,7 @@ where 1=1 elif catelogid: sql += " and m.llmcatelogid = ${catelogid}$" ns['catelogid'] = catelogid - sql += " order by m.llmcatelogid, a.id" + sql += " order by m.llmcatelogid, a.id, a.name" recs = await sor.sqlExe(sql, ns) if llmid: return recs[0] if recs else None @@ -458,8 +458,7 @@ async def get_llms_by_catelog(catelogid=None, orderby='providerid'): if catelogid: sql += " and m.llmcatelogid = ${catelogid}$" params['catelogid'] = catelogid - - sql += " order by m.llmcatelogid, a.id" + sql += " order by m.llmcatelogid, a.id, a.name" recs = await sor.sqlExe(sql, params) # 批量查询所有模型的 ppid 映射（避免 N+1 查询） diff --git a/load_test.py b/load_test.py new file mode 100644 index 0000000..9f1cfdf --- /dev/null +++ b/load_test.py @@ -0,0 +1,163 @@ +#!/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())