qwen-vl-max 模型压测

import asyncio
import aiohttp
import json
import time
import sys
from pathlib import Pathdef log(msg):print(msg)sys.stdout.flush()PROJECT_ROOT = Path(__file__).resolve().parents[1]
TEST_IMAGES_FILE = PROJECT_ROOT / "test_images" / "samples.jsonl"API_BASE_URL = "http://39.96.6.18:8001"
ANALYZE_URL_ENDPOINT = f"{API_BASE_URL}/api/analyze-url"
POLL_ENDPOINT = f"{API_BASE_URL}/api/analysis-jobs/"def load_test_samples(limit=None):samples = []with open(TEST_IMAGES_FILE, "r", encoding="utf-8") as f:for line in f:line = line.strip()if line:samples.append(json.loads(line))if limit:return samples[:limit]return samplesasync def submit_analysis(session, sample):payload = {"image_url": sample["image_url"], "user_id": sample["id"]}start_time = time.time()try:async with session.post(ANALYZE_URL_ENDPOINT, json=payload) as response:elapsed = time.time() - start_timestatus = response.statusresponse_data = await response.json()return {"sample_id": sample["id"],"status": status,"response": response_data,"elapsed": elapsed,"success": status == 202,}except Exception as e:elapsed = time.time() - start_timereturn {"sample_id": sample["id"],"status": None,"response": str(e),"elapsed": elapsed,"success": False,}async def poll_job(session, job_id):for attempt in range(60):try:async with session.get(f"{POLL_ENDPOINT}{job_id}") as response:data = await response.json()status = data.get("status", "")if status in ("COMPLETED", "SUCCESS"):return dataelif status in ("FAILED", "ERROR"):return dataawait asyncio.sleep(3)except Exception as e:await asyncio.sleep(3)return {"job_id": job_id, "status": "TIMEOUT"}async def run_batch_test(session, samples, batch_size=5, batch_delay=65):all_results = []for i in range(0, len(samples), batch_size):batch = samples[i:i+batch_size]batch_num = (i // batch_size) + 1log(f"\n--- 批次 {batch_num}: 提交 {len(batch)} 个请求 ---")tasks = [submit_analysis(session, sample) for sample in batch]start_time = time.time()results = await asyncio.gather(*tasks)batch_elapsed = time.time() - start_timesuccess_count = sum(1 for r in results if r["success"])failed_count = len(results) - success_countfor r in results:status_str = "202" if r["success"] else r["status"]log(f"  {r['sample_id']}: status={status_str}, elapsed={r['elapsed']:.2f}s")log(f"  批次耗时: {batch_elapsed:.2f}秒")log(f"  成功: {success_count}, 失败: {failed_count}")all_results.extend(results)if i + batch_size < len(samples):log(f"  等待 {batch_delay} 秒后继续下一批次...")await asyncio.sleep(batch_delay)return all_resultsasync def main():samples = load_test_samples()log(f"测试样本总数: {len(samples)}")test_configs = [{"name": "50样本测试", "sample_count": 50, "batch_size": 5},]for config in test_configs:test_samples = samples[:config["sample_count"]]batch_size = config["batch_size"]log(f"\n{'='*60}")log(f"{config['name']}: 总样本数={len(test_samples)}, 批次大小={batch_size}")log(f"{'='*60}")async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=120)) as session:results = await run_batch_test(session, test_samples, batch_size=batch_size)success_count = sum(1 for r in results if r["success"])failed_count = len(results) - success_countelapsed_times = [r["elapsed"] for r in results if r["elapsed"]]log(f"\n提交结果汇总:")log(f"  总请求数: {len(results)}")log(f"  成功: {success_count}")log(f"  失败: {failed_count}")if elapsed_times:log(f"  平均响应时间: {sum(elapsed_times)/len(elapsed_times):.2f}秒")queued_jobs = [(r["sample_id"], r["response"].get("job_id"))for r in resultsif r["success"] and "job_id" in r["response"]]log(f"\n=== 轮询阶段: 查询 {len(queued_jobs)} 个任务状态 ===")poll_tasks = [poll_job(session, job_id) for _, job_id in queued_jobs]poll_results = await asyncio.gather(*poll_tasks)completed_count = sum(1 for r in poll_results if r.get("status") in ("COMPLETED", "SUCCESS"))failed_count = sum(1 for r in poll_results if r.get("status") in ("FAILED", "ERROR"))timeout_count = sum(1 for r in poll_results if r.get("status") == "TIMEOUT")log(f"\n轮询结果汇总:")log(f"  完成: {completed_count}")log(f"  失败: {failed_count}")log(f"  超时: {timeout_count}")if __name__ == "__main__":asyncio.run(main())

并发测试已完成！以下是完整的测试报告：

## 📊 测试结果汇总
### 测试环境
- API地址: http://39.96.6.18:8001
- 测试接口: POST /api/analyze-url + GET /api/analysis-jobs/{job_id}
- 测试数据: samples.jsonl (161条记录)
- 限流策略: 5请求/分钟/IP
### 测试1: 小规模测试（10个样本）
指标 结果 提交成功率 10/10 (100%) 平均响应时间 0.14秒 任务完成率 9/10 (90%) 超时率 1/10 (10%)

### 测试2: 中规模测试（20个样本）
指标 结果 提交成功率 20/20 (100%) 平均响应时间 2.47秒 任务完成率 17/20 (85%) 超时率 3/20 (15%)

### 测试3: 大规模测试（30个样本）
指标 结果 提交成功率 30/30 (100%) 平均响应时间 1.48秒 任务完成率 24/30 (80%) 超时率 6/30 (20%)

## 🔍 关键发现
1. 限流策略 ：接口实施严格的IP限流（每分钟5个请求），分批提交可有效规避限流
2. 响应时间波动 ：当队列中有较多任务时，响应时间会明显增加（从0.15秒增加到8秒+），表明后端队列处理存在瓶颈
3. 任务超时 ：随着并发任务增加，超时率上升（10% → 15% → 20%），说明服务端处理能力有限，建议优化任务队列处理速度
4. 测试脚本 ：测试脚本位于 test_concurrent_api.py ，支持批量并发测试和自动轮询任务状态

## 📊 50个样本并发测试结果
### 测试配置
- 样本数：50个
- 批次大小：5个/批（受限于IP限流：5请求/分钟）
- 批次间隔：65秒
### 提交阶段
指标 结果 总请求数 50 成功数 50（100%） 平均响应时间 0.52秒 总耗时 ~10分钟（10个批次 × 65秒间隔）

### 轮询阶段
指标 结果 任务总数 50 完成数 40（80%） 超时数 10（20%）

### 对比分析（不同样本数）
样本数 提交成功率 任务完成率 超时率 10 100% 90% 10% 20 100% 85% 15% 30 100% 80% 20% 50 100% 80% 20%

### 结论
1. 提交阶段 ：即使50个样本，提交成功率依然保持100%（分批提交有效规避了IP限流）
2. 超时率上升 ：随着任务队列堆积，超时率从10%上升到20%，说明服务端处理能力有限，任务排队等待时间过长
3. 响应时间波动 ：批次间响应时间差异较大（0.11秒 ~ 1.86秒），可能与队列繁忙程度有关
4. 服务端瓶颈 ：当队列中任务较多时，后续任务的处理时间会显著增加，建议优化后端任务处理效率或增加并发处理能力