性能优化指南

# 混合精度推理优化
def optimize_model_loading(self, model_path: str, lora_path: Optional[str] = None):
    """优化的模型加载方法"""
    # 使用半精度加载模型
    self.model = AutoModelForSeq2SeqLM.from_pretrained(
        model_path,
        torch_dtype=torch.float16,  # 使用半精度
        device_map="auto",
        low_cpu_mem_usage=True,     # 减少CPU内存使用
        max_memory={0: "10GB"}      # 限制GPU内存使用
    )
    
    # 启用内存优化
    if torch.cuda.is_available():
        torch.cuda.empty_cache()
        torch.backends.cudnn.benchmark = True

# 批处理推理实现
def batch_inference(self, prompts: List[str], batch_size: int = 4):
    """批处理推理优化"""
    results = []
    
    for i in range(0, len(prompts), batch_size):
        batch_prompts = prompts[i:i + batch_size]
        
        # 批处理编码
        inputs = self.tokenizer(
            batch_prompts,
            max_length=self.config["model"]["max_length"],
            padding=True,
            truncation=True,
            return_tensors="pt"
        )
        
        # 批处理推理
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_length=self.config["model"]["max_length"],
                temperature=self.config["model"]["temperature"],
                do_sample=True,
                num_return_sequences=1
            )
        
        # 批处理解码
        batch_results = [self.tokenizer.decode(output, skip_special_tokens=True) 
                        for output in outputs]
        results.extend(batch_results)
    
    return results

# 模型缓存实现
class ModelCache:
    def __init__(self, max_size: int = 3):
        self.cache = {}
        self.max_size = max_size
        self.access_order = []
    
    def get_model(self, model_path: str, lora_path: str = None):
        """获取缓存的模型"""
        cache_key = f"{model_path}_{lora_path or 'base'}"
        
        if cache_key in self.cache:
            # 更新访问顺序
            self.access_order.remove(cache_key)
            self.access_order.append(cache_key)
            return self.cache[cache_key]
        
        return None
    
    def cache_model(self, model_path: str, lora_path: str, model):
        """缓存模型"""
        cache_key = f"{model_path}_{lora_path or 'base'}"
        
        # 如果缓存已满，移除最久未使用的模型
        if len(self.cache) >= self.max_size:
            oldest_key = self.access_order.pop(0)
            del self.cache[oldest_key]
        
        self.cache[cache_key] = model
        self.access_order.append(cache_key)

# 多进程数据生成优化
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor

class OptimizedUIDataGenerator(UIDataGenerator):
    def __init__(self, config_path: str = "config/model_config.yaml", 
                 num_workers: int = None):
        super().__init__(config_path)
        self.num_workers = num_workers or mp.cpu_count()
    
    def generate_dataset_parallel(self, num_samples: int = 1000) -> List[Dict[str, Any]]:
        """并行生成数据集"""
        # 计算每个进程的样本数
        samples_per_worker = num_samples // self.num_workers
        
        with ProcessPoolExecutor(max_workers=self.num_workers) as executor:
            # 提交任务
            futures = []
            for i in range(self.num_workers):
                start_idx = i * samples_per_worker
                end_idx = start_idx + samples_per_worker if i < self.num_workers - 1 else num_samples
                
                future = executor.submit(
                    self._generate_batch, 
                    start_idx, 
                    end_idx - start_idx
                )
                futures.append(future)
            
            # 收集结果
            all_data = []
            for future in futures:
                batch_data = future.result()
                all_data.extend(batch_data)
        
        return all_data
    
    def _generate_batch(self, start_idx: int, batch_size: int) -> List[Dict[str, Any]]:
        """生成一批数据"""
        batch_data = []
        for i in range(batch_size):
            # 生成单个样本的逻辑
            sample = self._generate_single_sample(start_idx + i)
            batch_data.append(sample)
        return batch_data

# 内存映射文件优化
import mmap

class MemoryMappedDataset:
    def __init__(self, file_path: str):
        self.file_path = file_path
        self.file = None
        self.mmap = None
    
    def __enter__(self):
        self.file = open(self.file_path, 'r+b')
        self.mmap = mmap.mmap(self.file.fileno(), 0)
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.mmap:
            self.mmap.close()
        if self.file:
            self.file.close()
    
    def read_line(self, offset: int) -> str:
        """从指定偏移量读取一行"""
        self.mmap.seek(offset)
        line = self.mmap.readline()
        return line.decode('utf-8')

# 异步处理优化
import asyncio
from asyncio import Queue
from concurrent.futures import ThreadPoolExecutor

class AsyncUIGenerator:
    def __init__(self, max_workers: int = 4):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.request_queue = Queue(maxsize=100)
        self.result_cache = {}
    
    async def generate_ui_async(self, request: GenerateUIRequest):
        """异步UI生成"""
        # 检查缓存
        cache_key = f"{request.prompt}_{request.output_format}"
        if cache_key in self.result_cache:
            return self.result_cache[cache_key]
        
        # 提交到线程池执行
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(
            self.executor, 
            self._generate_ui_sync, 
            request
        )
        
        # 缓存结果
        self.result_cache[cache_key] = result
        return result
    
    def _generate_ui_sync(self, request: GenerateUIRequest):
        """同步UI生成（在线程池中执行）"""
        # 实际的UI生成逻辑
        pass

# 数据库连接池优化
import asyncpg
from asyncpg import Pool

class DatabaseManager:
    def __init__(self, database_url: str, min_connections: int = 5, 
                 max_connections: int = 20):
        self.database_url = database_url
        self.pool: Pool = None
        self.min_connections = min_connections
        self.max_connections = max_connections
    
    async def initialize(self):
        """初始化连接池"""
        self.pool = await asyncpg.create_pool(
            self.database_url,
            min_size=self.min_connections,
            max_size=self.max_connections,
            command_timeout=60,
            server_settings={
                'jit': 'off',  # 关闭JIT编译，减少内存使用
                'shared_preload_libraries': 'pg_stat_statements'
            }
        )
    
    async def execute_query(self, query: str, *args):
        """执行查询"""
        async with self.pool.acquire() as connection:
            return await connection.fetch(query, *args)

# 响应压缩优化
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

# 添加Gzip压缩中间件
app.add_middleware(
    GZipMiddleware, 
    minimum_size=1000,  # 只压缩大于1KB的响应
    compresslevel=6     # 压缩级别（1-9，6是平衡点）
)

# 自定义压缩中间件
from starlette.middleware.base import BaseHTTPMiddleware

class CustomCompressionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        
        # 只对特定类型的响应进行压缩
        if (response.headers.get("content-type", "").startswith("application/json") and
            len(response.body) > 1024):
            response.headers["content-encoding"] = "gzip"
            # 实际的压缩逻辑
        
        return response

# 图片渲染优化
class OptimizedUIRenderer(UIRenderer):
    def __init__(self, config_path: str = "config/model_config.yaml", 
                 tokens_path: str = "config/ui_tokens.json"):
        super().__init__(config_path, tokens_path)
        
        # 预加载字体
        self._preload_fonts()
        
        # 组件渲染缓存
        self.component_cache = {}
    
    def _preload_fonts(self):
        """预加载字体，避免重复加载"""
        font_paths = [
            "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
            "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"
        ]
        
        self.font_cache = {}
        for font_path in font_paths:
            try:
                self.font_cache[font_path] = {
                    "small": ImageFont.truetype(font_path, 12),
                    "medium": ImageFont.truetype(font_path, 16),
                    "large": ImageFont.truetype(font_path, 20),
                    "title": ImageFont.truetype(font_path, 24)
                }
            except:
                continue
    
    def render_with_cache(self, dsl: Dict[str, Any]) -> Image.Image:
        """带缓存的渲染"""
        # 生成缓存键
        cache_key = self._generate_cache_key(dsl)
        
        if cache_key in self.component_cache:
            return self.component_cache[cache_key]
        
        # 执行渲染
        image = self.render(dsl)
        
        # 缓存结果
        self.component_cache[cache_key] = image
        return image
    
    def _generate_cache_key(self, dsl: Dict[str, Any]) -> str:
        """生成缓存键"""
        import hashlib
        dsl_str = json.dumps(dsl, sort_keys=True)
        return hashlib.md5(dsl_str.encode()).hexdigest()

# 并行渲染实现
import concurrent.futures
from typing import List

class ParallelRenderer:
    def __init__(self, max_workers: int = 4):
        self.max_workers = max_workers
        self.image_renderer = UIRenderer()
        self.vue_renderer = VueRenderer()
    
    def render_multiple(self, dsl_list: List[Dict[str, Any]], 
                       output_formats: List[str]) -> Dict[str, List]:
        """并行渲染多个DSL"""
        results = {"images": [], "vue": []}
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 提交图片渲染任务
            if "png" in output_formats:
                image_futures = [
                    executor.submit(self.image_renderer.render, dsl) 
                    for dsl in dsl_list
                ]
                results["images"] = [future.result() for future in image_futures]
            
            # 提交Vue渲染任务
            if "vue" in output_formats:
                vue_futures = [
                    executor.submit(self.vue_renderer.render, dsl) 
                    for dsl in dsl_list
                ]
                results["vue"] = [future.result() for future in vue_futures]
        
        return results

# 性能监控实现
import psutil
import time
from prometheus_client import Counter, Histogram, Gauge, start_http_server

class PerformanceMonitor:
    def __init__(self):
        # Prometheus指标
        self.request_count = Counter('ui_requests_total', 'Total UI requests')
        self.request_duration = Histogram('ui_request_duration_seconds', 'Request duration')
        self.active_connections = Gauge('ui_active_connections', 'Active connections')
        self.gpu_memory_usage = Gauge('ui_gpu_memory_usage_bytes', 'GPU memory usage')
        self.cpu_usage = Gauge('ui_cpu_usage_percent', 'CPU usage percentage')
        
        # 启动监控服务器
        start_http_server(8001)
    
    def record_request(self, duration: float):
        """记录请求指标"""
        self.request_count.inc()
        self.request_duration.observe(duration)
    
    def update_system_metrics(self):
        """更新系统指标"""
        # CPU使用率
        cpu_percent = psutil.cpu_percent(interval=1)
        self.cpu_usage.set(cpu_percent)
        
        # GPU内存使用（如果可用）
        if torch.cuda.is_available():
            gpu_memory = torch.cuda.memory_allocated()
            self.gpu_memory_usage.set(gpu_memory)
    
    def get_performance_report(self) -> Dict[str, Any]:
        """获取性能报告"""
        return {
            "cpu_usage": psutil.cpu_percent(),
            "memory_usage": psutil.virtual_memory().percent,
            "disk_usage": psutil.disk_usage('/').percent,
            "gpu_memory": torch.cuda.memory_allocated() if torch.cuda.is_available() else 0,
            "active_connections": self.active_connections._value._value
        }

# 性能测试实现
import asyncio
import aiohttp
import time
import statistics

class PerformanceTester:
    def __init__(self, base_url: str = "http://localhost:8000"):
        self.base_url = base_url
        self.test_prompts = [
            "黑金风格的电商首页，顶部搜索，中间两列商品卡，底部导航",
            "简约风格的商品详情页，顶部轮播图，展示价格和卖家信息",
            "白银风格的搜索页面，搜索框和筛选条件，单列商品列表"
        ]
    
    async def test_single_request(self, session: aiohttp.ClientSession, 
                                 prompt: str) -> Dict[str, Any]:
        """测试单个请求"""
        start_time = time.time()
        
        async with session.post(
            f"{self.base_url}/generate-ui",
            json={"prompt": prompt, "output_format": "json"}
        ) as response:
            result = await response.json()
            duration = time.time() - start_time
            
            return {
                "duration": duration,
                "success": result.get("success", False),
                "status_code": response.status
            }
    
    async def test_concurrent_requests(self, num_requests: int = 100, 
                                     concurrency: int = 10) -> Dict[str, Any]:
        """测试并发请求"""
        semaphore = asyncio.Semaphore(concurrency)
        
        async def limited_request(session, prompt):
            async with semaphore:
                return await self.test_single_request(session, prompt)
        
        async with aiohttp.ClientSession() as session:
            tasks = [
                limited_request(session, self.test_prompts[i % len(self.test_prompts)])
                for i in range(num_requests)
            ]
            
            results = await asyncio.gather(*tasks)
        
        # 分析结果
        durations = [r["duration"] for r in results if r["success"]]
        success_rate = sum(1 for r in results if r["success"]) / len(results)
        
        return {
            "total_requests": num_requests,
            "success_rate": success_rate,
            "avg_duration": statistics.mean(durations) if durations else 0,
            "p95_duration": statistics.quantiles(durations, n=20)[18] if durations else 0,
            "max_duration": max(durations) if durations else 0,
            "min_duration": min(durations) if durations else 0
        }
    
    def run_benchmark(self):
        """运行性能基准测试"""
        print("开始性能基准测试...")
        
        # 测试不同并发级别
        concurrency_levels = [1, 5, 10, 20, 50]
        results = {}
        
        for concurrency in concurrency_levels:
            print(f"测试并发级别: {concurrency}")
            result = asyncio.run(self.test_concurrent_requests(100, concurrency))
            results[concurrency] = result
            
            print(f"  成功率: {result['success_rate']:.2%}")
            print(f"  平均响应时间: {result['avg_duration']:.2f}s")
            print(f"  P95响应时间: {result['p95_duration']:.2f}s")
        
        return results