""" LLM生成性能测试 本模块实现LLM生成性能测试，包括API响应时间、令牌使用、质量度量和 不同生成策略的比较。提供基准测试和性能分析功能。 """ import asyncio import time import json import statistics import psutil import threading from typing import List, Dict, Any, Optional, Tuple from dataclasses import dataclass, asdict from pathlib import Path import aiohttp import pytest import matplotlib.pyplot as plt import seaborn as sns import pandas as pd from src.spec.llm_generator import LLMGenerator, GenerationRequest, GenerationResult from src.parser.c_parser import CParserFactory from src.parser.ast_extractor import ASTExtractor from src.utils.logger import get_logger from test_data.simple_c_examples import get_performance_test_suite logger = get_logger(__name__) @dataclass class PerformanceMetrics: """性能指标数据类""" function_name: str generation_time: float tokens_used: int quality_score: float api_response_time: float token_rate: float memory_usage: float cpu_usage: float timestamp: float complexity_level: str verification_goals: List[str] @dataclass class BenchmarkResult: """基准测试结果""" test_name: str total_tests: int successful_tests: int average_generation_time: float average_tokens_used: int average_quality_score: float average_token_rate: float min_generation_time: float max_generation_time: float std_generation_time: float total_memory_usage: float total_cpu_usage: float metrics: List[PerformanceMetrics] class PerformanceMonitor: """性能监控器""" def __init__(self): self.process = psutil.Process() self.start_memory = None self.start_cpu = None self.start_time = None def start_monitoring(self): """开始监控""" self.start_memory = self.process.memory_info().rss / 1024 / 1024 # MB self.start_cpu = self.process.cpu_percent() self.start_time = time.time() def stop_monitoring(self) -> Tuple[float, float]: """停止监控并返回资源使用情况""" end_memory = self.process.memory_info().rss / 1024 / 1024 # MB end_cpu = self.process.cpu_percent() memory_usage = end_memory - self.start_memory if self.start_memory else 0 cpu_usage = end_cpu - self.start_cpu if self.start_cpu else 0 return memory_usage, cpu_usage class LLMPerformanceTester: """LLM性能测试器""" def __init__(self, api_key: str, base_url: str = "https://api.siliconflow.cn/v1"): self.api_key = api_key self.base_url = base_url self.monitor = PerformanceMonitor() self.parser = CParserFactory.create_parser() self.ast_extractor = ASTExtractor() self.logger = logger async def benchmark_api_performance(self, test_iterations: int = 10) -> BenchmarkResult: """基准测试API性能""" self.logger.info(f"Starting API performance benchmark with {test_iterations} iterations") # 创建简单的测试用例 test_request = GenerationRequest( function_name="test_function", function_info={ 'name': 'test_function', 'return_type': 'int', 'parameters': [ {'name': 'a', 'type': 'int'}, {'name': 'b', 'type': 'int'} ] }, verification_goals=['functional_correctness'], max_retries=1, validate=False, store=False ) metrics = [] async with LLMGenerator(api_key=self.api_key, base_url=self.base_url) as generator: for i in range(test_iterations): self.monitor.start_monitoring() try: start_time = time.time() result = await generator.generate_specification(test_request) end_time = time.time() memory_usage, cpu_usage = self.monitor.stop_monitoring() metric = PerformanceMetrics( function_name="test_function", generation_time=end_time - start_time, tokens_used=result.tokens_used, quality_score=result.quality_score, api_response_time=result.generation_time, token_rate=result.tokens_used / (end_time - start_time), memory_usage=memory_usage, cpu_usage=cpu_usage, timestamp=time.time(), complexity_level="basic", verification_goals=['functional_correctness'] ) metrics.append(metric) if (i + 1) % 5 == 0: self.logger.info(f"Completed {i + 1}/{test_iterations} iterations") except Exception as e: self.logger.error(f"Iteration {i + 1} failed: {e}") continue return self._create_benchmark_result("API Performance Benchmark", metrics) async def benchmark_by_complexity(self) -> Dict[str, BenchmarkResult]: """按复杂度进行基准测试""" self.logger.info("Starting complexity-based benchmarking") test_cases = get_performance_test_suite() complexity_groups = { 'basic': [tc for tc in test_cases if tc.complexity_level == 'basic'], 'intermediate': [tc for tc in test_cases if tc.complexity_level == 'intermediate'], 'advanced': [tc for tc in test_cases if tc.complexity_level == 'advanced'] } results = {} for complexity, cases in complexity_groups.items(): if cases: self.logger.info(f"Benchmarking {complexity} complexity functions") metrics = [] async with LLMGenerator(api_key=self.api_key, base_url=self.base_url) as generator: for test_case in cases: # 创建临时文件 import tempfile with tempfile.NamedTemporaryFile(mode='w', suffix='.c', delete=False) as f: f.write(test_case.source_code) temp_path = f.name try: # 解析文件 ast = self.parser.parse_file(temp_path) metadata = self.ast_extractor.extract_metadata(ast) if metadata.functions: func_name = list(metadata.functions.keys())[0] func_info = metadata.functions[func_name] request = GenerationRequest( function_name=func_name, function_info=func_info.to_dict(), verification_goals=test_case.verification_goals, max_retries=2, validate=True, store=False ) self.monitor.start_monitoring() start_time = time.time() result = await generator.generate_specification(request) end_time = time.time() memory_usage, cpu_usage = self.monitor.stop_monitoring() metric = PerformanceMetrics( function_name=func_name, generation_time=end_time - start_time, tokens_used=result.tokens_used, quality_score=result.quality_score, api_response_time=result.generation_time, token_rate=result.tokens_used / (end_time - start_time), memory_usage=memory_usage, cpu_usage=cpu_usage, timestamp=time.time(), complexity_level=complexity, verification_goals=test_case.verification_goals ) metrics.append(metric) except Exception as e: self.logger.error(f"Failed to benchmark {test_case.name}: {e}") finally: import os os.unlink(temp_path) results[complexity] = self._create_benchmark_result(f"{complexity.title()} Complexity Benchmark", metrics) return results async def benchmark_concurrent_requests(self, concurrent_count: int = 5, test_iterations: int = 3) -> BenchmarkResult: """基准测试并发请求性能""" self.logger.info(f"Starting concurrent request benchmark: {concurrent_count} concurrent requests") test_request = GenerationRequest( function_name="concurrent_test", function_info={ 'name': 'concurrent_test', 'return_type': 'int', 'parameters': [{'name': 'x', 'type': 'int'}] }, verification_goals=['functional_correctness'], max_retries=1, validate=False, store=False ) metrics = [] async with LLMGenerator(api_key=self.api_key, base_url=self.base_url) as generator: for iteration in range(test_iterations): self.logger.info(f"Concurrent iteration {iteration + 1}/{test_iterations}") # 创建并发请求 semaphore = asyncio.Semaphore(concurrent_count) async def make_request(): async with semaphore: self.monitor.start_monitoring() start_time = time.time() try: result = await generator.generate_specification(test_request) end_time = time.time() memory_usage, cpu_usage = self.monitor.stop_monitoring() metric = PerformanceMetrics( function_name="concurrent_test", generation_time=end_time - start_time, tokens_used=result.tokens_used, quality_score=result.quality_score, api_response_time=result.generation_time, token_rate=result.tokens_used / (end_time - start_time), memory_usage=memory_usage, cpu_usage=cpu_usage, timestamp=time.time(), complexity_level="basic", verification_goals=['functional_correctness'] ) return metric except Exception as e: self.logger.error(f"Concurrent request failed: {e}") return None # 执行并发请求 tasks = [make_request() for _ in range(concurrent_count)] results = await asyncio.gather(*tasks, return_exceptions=True) # 收集成功的指标 for result in results: if result and not isinstance(result, Exception): metrics.append(result) return self._create_benchmark_result(f"Concurrent Requests ({concurrent_count}) Benchmark", metrics) async def benchmark_retry_impact(self) -> BenchmarkResult: """基准测试重试机制的影响""" self.logger.info("Starting retry impact benchmark") test_request = GenerationRequest( function_name="retry_test", function_info={ 'name': 'retry_test', 'return_type': 'int', 'parameters': [{'name': 'x', 'type': 'int'}] }, verification_goals=['functional_correctness'], max_retries=3, validate=True, store=False ) metrics = [] async with LLMGenerator(api_key=self.api_key, base_url=self.base_url) as generator: for i in range(10): self.monitor.start_monitoring() start_time = time.time() try: result = await generator.generate_specification(test_request) end_time = time.time() memory_usage, cpu_usage = self.monitor.stop_monitoring() metric = PerformanceMetrics( function_name="retry_test", generation_time=end_time - start_time, tokens_used=result.tokens_used, quality_score=result.quality_score, api_response_time=result.generation_time, token_rate=result.tokens_used / (end_time - start_time), memory_usage=memory_usage, cpu_usage=cpu_usage, timestamp=time.time(), complexity_level="basic", verification_goals=['functional_correctness'] ) metrics.append(metric) except Exception as e: self.logger.error(f"Retry test {i + 1} failed: {e}") return self._create_benchmark_result("Retry Impact Benchmark", metrics) def _create_benchmark_result(self, test_name: str, metrics: List[PerformanceMetrics]) -> BenchmarkResult: """创建基准测试结果""" if not metrics: return BenchmarkResult( test_name=test_name, total_tests=0, successful_tests=0, average_generation_time=0, average_tokens_used=0, average_quality_score=0, average_token_rate=0, min_generation_time=0, max_generation_time=0, std_generation_time=0, total_memory_usage=0, total_cpu_usage=0, metrics=[] ) successful_metrics = [m for m in metrics if m.quality_score > 0] generation_times = [m.generation_time for m in successful_metrics] return BenchmarkResult( test_name=test_name, total_tests=len(metrics), successful_tests=len(successful_metrics), average_generation_time=statistics.mean(generation_times) if generation_times else 0, average_tokens_used=statistics.mean([m.tokens_used for m in successful_metrics]) if successful_metrics else 0, average_quality_score=statistics.mean([m.quality_score for m in successful_metrics]) if successful_metrics else 0, average_token_rate=statistics.mean([m.token_rate for m in successful_metrics]) if successful_metrics else 0, min_generation_time=min(generation_times) if generation_times else 0, max_generation_time=max(generation_times) if generation_times else 0, std_generation_time=statistics.stdev(generation_times) if len(generation_times) > 1 else 0, total_memory_usage=sum(m.memory_usage for m in metrics), total_cpu_usage=sum(m.cpu_usage for m in metrics), metrics=metrics ) def generate_performance_report(self, results: Dict[str, BenchmarkResult], output_dir: str = "performance_reports"): """生成性能报告""" output_path = Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) report = { 'timestamp': time.time(), 'summary': self._generate_performance_summary(results), 'detailed_results': {name: asdict(result) for name, result in results.items()}, 'recommendations': self._generate_recommendations(results) } # 保存JSON报告 json_file = output_path / f"performance_report_{int(time.time())}.json" with open(json_file, 'w') as f: json.dump(report, f, indent=2) # 生成可视化报告 self._generate_visualizations(results, output_path) self.logger.info(f"Performance report generated: {json_file}") return str(json_file) def _generate_performance_summary(self, results: Dict[str, BenchmarkResult]) -> Dict[str, Any]: """生成性能摘要""" all_metrics = [] for result in results.values(): all_metrics.extend(result.metrics) if not all_metrics: return {'error': 'No performance data available'} generation_times = [m.generation_time for m in all_metrics if m.generation_time > 0] quality_scores = [m.quality_score for m in all_metrics if m.quality_score > 0] token_rates = [m.token_rate for m in all_metrics if m.token_rate > 0] return { 'total_tests': sum(r.total_tests for r in results.values()), 'successful_tests': sum(r.successful_tests for r in results.values()), 'success_rate': sum(r.successful_tests for r in results.values()) / sum(r.total_tests for r in results.values()), 'average_generation_time': statistics.mean(generation_times) if generation_times else 0, 'average_quality_score': statistics.mean(quality_scores) if quality_scores else 0, 'average_token_rate': statistics.mean(token_rates) if token_rates else 0, 'total_memory_usage': sum(r.total_memory_usage for r in results.values()), 'total_cpu_usage': sum(r.total_cpu_usage for r in results.values()), 'best_performing_test': max(results.items(), key=lambda x: x[1].average_token_rate), 'slowest_test': max(results.items(), key=lambda x: x[1].average_generation_time) } def _generate_recommendations(self, results: Dict[str, BenchmarkResult]) -> List[str]: """生成性能建议""" recommendations = [] # 分析平均生成时间 avg_times = [r.average_generation_time for r in results.values() if r.average_generation_time > 0] if avg_times: avg_time = statistics.mean(avg_times) if avg_time > 30: recommendations.append("Average generation time is high (>30s). Consider optimizing prompts or using faster models.") # 分析成功率 total_tests = sum(r.total_tests for r in results.values()) successful_tests = sum(r.successful_tests for r in results.values()) if total_tests > 0: success_rate = successful_tests / total_tests if success_rate < 0.9: recommendations.append("Success rate is below 90%. Check API reliability and error handling.") # 分析令牌速率 token_rates = [r.average_token_rate for r in results.values() if r.average_token_rate > 0] if token_rates: avg_token_rate = statistics.mean(token_rates) if avg_token_rate < 10: recommendations.append("Token generation rate is low. Consider network optimization or model selection.") # 分析内存使用 total_memory = sum(r.total_memory_usage for r in results.values()) if total_memory > 1000: # 1GB recommendations.append("High memory usage detected. Consider optimizing memory management.") return recommendations def _generate_visualizations(self, results: Dict[str, BenchmarkResult], output_path: Path): """生成可视化图表""" # 设置图表样式 plt.style.use('seaborn-v0_8') # 1. 性能对比图 fig, axes = plt.subplots(2, 2, figsize=(15, 10)) fig.suptitle('LLM Generation Performance Analysis', fontsize=16) test_names = list(results.keys()) generation_times = [r.average_generation_time for r in results.values()] quality_scores = [r.average_quality_score for r in results.values()] token_rates = [r.average_token_rate for r in results.values()] # 生成时间对比 axes[0, 0].bar(test_names, generation_times, color='skyblue') axes[0, 0].set_title('Average Generation Time') axes[0, 0].set_ylabel('Time (seconds)') axes[0, 0].tick_params(axis='x', rotation=45) # 质量分数对比 axes[0, 1].bar(test_names, quality_scores, color='lightgreen') axes[0, 1].set_title('Average Quality Score') axes[0, 1].set_ylabel('Quality Score') axes[0, 1].tick_params(axis='x', rotation=45) # 令牌速率对比 axes[1, 0].bar(test_names, token_rates, color='lightcoral') axes[1, 0].set_title('Average Token Rate') axes[1, 0].set_ylabel('Tokens/second') axes[1, 0].tick_params(axis='x', rotation=45) # 成功率对比 success_rates = [r.successful_tests / r.total_tests if r.total_tests > 0 else 0 for r in results.values()] axes[1, 1].bar(test_names, success_rates, color='gold') axes[1, 1].set_title('Success Rate') axes[1, 1].set_ylabel('Success Rate') axes[1, 1].tick_params(axis='x', rotation=45) plt.tight_layout() plt.savefig(output_path / 'performance_comparison.png', dpi=300, bbox_inches='tight') plt.close() # 2. 复杂度对比图（如果有） complexity_results = {k: v for k, v in results.items() if 'complexity' in k.lower()} if complexity_results: fig, ax = plt.subplots(1, 1, figsize=(10, 6)) complexities = list(complexity_results.keys()) times = [complexity_results[c].average_generation_time for c in complexities] qualities = [complexity_results[c].average_quality_score for c in complexities] x = range(len(complexities)) width = 0.35 ax.bar([i - width/2 for i in x], times, width, label='Generation Time', color='skyblue') ax.bar([i + width/2 for i in x], qualities, width, label='Quality Score', color='lightgreen') ax.set_xlabel('Complexity Level') ax.set_ylabel('Value') ax.set_title('Performance vs Complexity') ax.set_xticks(x) ax.set_xticklabels(complexities) ax.legend() plt.tight_layout() plt.savefig(output_path / 'complexity_analysis.png', dpi=300, bbox_inches='tight') plt.close() self.logger.info(f"Performance visualizations saved to {output_path}") # 便捷函数 async def run_performance_benchmark(api_key: str, base_url: str = "https://api.siliconflow.cn/v1") -> Dict[str, BenchmarkResult]: """运行性能基准测试的便捷函数""" tester = LLMPerformanceTester(api_key, base_url) # 运行所有基准测试 results = {} # API性能基准测试 results['api_performance'] = await tester.benchmark_api_performance() # 复杂度基准测试 complexity_results = await tester.benchmark_by_complexity() results.update(complexity_results) # 并发请求基准测试 results['concurrent_5'] = await tester.benchmark_concurrent_requests(5) # 重试影响基准测试 results['retry_impact'] = await tester.benchmark_retry_impact() return results def save_performance_report(results: Dict[str, BenchmarkResult], output_dir: str = "performance_reports"): """保存性能报告的便捷函数""" tester = LLMPerformanceTester("dummy_key") # 只用于生成报告 return tester.generate_performance_report(results, output_dir) if __name__ == "__main__": # 运行性能测试 import os api_key = os.getenv('SILICONFLOW_API_KEY') if not api_key: print("Please set SILICONFLOW_API_KEY environment variable") exit(1) async def main(): results = await run_performance_benchmark(api_key) report_path = save_performance_report(results) print(f"Performance report saved to: {report_path}") asyncio.run(main())