DuoDuoProject/xiaozhi-esp32-server-0.7.1/main/xiaozhi-server/performance_tester_vllm.py

import time
import asyncio
import logging
import statistics
import base64
from typing import Dict
from tabulate import tabulate
from config.settings import load_config
from core.utils.vllm import create_instance

# 设置全局日志级别为WARNING，抑制INFO级别日志
logging.basicConfig(level=logging.WARNING)


class AsyncVisionPerformanceTester:
    def __init__(self):
        self.config = load_config()
        self.test_images = [
            "../../docs/images/demo1.png",
            "../../docs/images/demo2.png",
        ]
        self.test_questions = [
            "这张图片里有什么？",
            "请详细描述这张图片的内容",
        ]

        # 加载测试图片
        self.results = {"vllm": {}}

    async def _test_vllm(self, vllm_name: str, config: Dict) -> Dict:
        """异步测试单个视觉大模型性能"""
        try:
            # 检查API密钥配置
            if "api_key" in config and any(
                x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]
            ):
                print(f"⏭️  VLLM {vllm_name} 未配置api_key，已跳过")
                return {"name": vllm_name, "type": "vllm", "errors": 1}

            # 获取实际类型（兼容旧配置）
            module_type = config.get("type", vllm_name)
            vllm = create_instance(module_type, config)

            print(f"🖼️ 测试 VLLM: {vllm_name}")

            # 创建所有测试任务
            test_tasks = []
            for question in self.test_questions:
                for image in self.test_images:
                    test_tasks.append(
                        self._test_single_vision(vllm_name, vllm, question, image)
                    )

            # 并发执行所有测试
            test_results = await asyncio.gather(*test_tasks)

            # 处理结果
            valid_results = [r for r in test_results if r is not None]
            if not valid_results:
                print(f"⚠️  {vllm_name} 无有效数据，可能配置错误")
                return {"name": vllm_name, "type": "vllm", "errors": 1}

            response_times = [r["response_time"] for r in valid_results]

            # 过滤异常数据
            mean = statistics.mean(response_times)
            stdev = statistics.stdev(response_times) if len(response_times) > 1 else 0
            filtered_times = [t for t in response_times if t <= mean + 3 * stdev]

            if len(filtered_times) < len(test_tasks) * 0.5:
                print(f"⚠️  {vllm_name} 有效数据不足，可能网络不稳定")
                return {"name": vllm_name, "type": "vllm", "errors": 1}

            return {
                "name": vllm_name,
                "type": "vllm",
                "avg_response": sum(response_times) / len(response_times),
                "std_response": (
                    statistics.stdev(response_times) if len(response_times) > 1 else 0
                ),
                "errors": 0,
            }

        except Exception as e:
            print(f"⚠️ VLLM {vllm_name} 测试失败: {str(e)}")
            return {"name": vllm_name, "type": "vllm", "errors": 1}

    async def _test_single_vision(
        self, vllm_name: str, vllm, question: str, image: str
    ) -> Dict:
        """测试单个视觉问题的性能"""
        try:
            print(f"📝 {vllm_name} 开始测试: {question[:20]}...")
            start_time = time.time()

            # 读取图片并转换为base64
            with open(image, "rb") as image_file:
                image_data = image_file.read()
                image_base64 = base64.b64encode(image_data).decode("utf-8")

            # 直接获取响应
            response = vllm.response(question, image_base64)
            response_time = time.time() - start_time
            print(f"✓ {vllm_name} 完成响应: {response_time:.3f}s")

            return {
                "name": vllm_name,
                "type": "vllm",
                "response_time": response_time,
            }
        except Exception as e:
            print(f"⚠️ {vllm_name} 测试失败: {str(e)}")
            return None

    def _print_results(self):
        """打印测试结果"""
        vllm_table = []
        for name, data in self.results["vllm"].items():
            if data["errors"] == 0:
                stability = data["std_response"] / data["avg_response"]
                vllm_table.append(
                    [
                        name,
                        f"{data['avg_response']:.3f}秒",
                        f"{stability:.3f}",
                    ]
                )

        if vllm_table:
            print("\n视觉大模型性能排行:\n")
            print(
                tabulate(
                    vllm_table,
                    headers=["模型名称", "响应耗时", "稳定性"],
                    tablefmt="github",
                    colalign=("left", "right", "right"),
                    disable_numparse=True,
                )
            )
        else:
            print("\n⚠️ 没有可用的视觉大模型进行测试。")

    async def run(self):
        """执行全量异步测试"""
        print("🔍 开始筛选可用视觉大模型...")

        if not self.test_images:
            print(f"\n⚠️  {self.image_root} 路径下没有图片文件，无法进行测试")
            return

        # 创建所有测试任务
        all_tasks = []

        # VLLM测试任务
        if self.config.get("VLLM") is not None:
            for vllm_name, config in self.config.get("VLLM", {}).items():
                if "api_key" in config and any(
                    x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]
                ):
                    print(f"⏭️  VLLM {vllm_name} 未配置api_key，已跳过")
                    continue
                print(f"🖼️ 添加VLLM测试任务: {vllm_name}")
                all_tasks.append(self._test_vllm(vllm_name, config))

        print(f"\n✅ 找到 {len(all_tasks)} 个可用视觉大模型")
        print(f"✅ 使用 {len(self.test_images)} 张测试图片")
        print(f"✅ 使用 {len(self.test_questions)} 个测试问题")
        print("\n⏳ 开始并发测试所有模型...\n")

        # 并发执行所有测试任务
        all_results = await asyncio.gather(*all_tasks, return_exceptions=True)

        # 处理结果
        for result in all_results:
            if isinstance(result, dict) and result["errors"] == 0:
                self.results["vllm"][result["name"]] = result

        # 打印结果
        print("\n📊 生成测试报告...")
        self._print_results()


async def main():
    tester = AsyncVisionPerformanceTester()
    await tester.run()


if __name__ == "__main__":
    asyncio.run(main())