AI生成工具压力测试方法与性能优化技术实现路径

AI生成工具压力测试基础框架

AI生成工具压力测试是评估系统在高负载情况下表现的关键手段。当你面对大量并发请求时，系统是否能够稳定运行，响应时间是否在可接受范围内，资源占用是否合理，这些都是压力测试需要回答的问题。

一个完整的压力测试框架通常包含以下几个核心组件：

AI生成工具并发请求测试实现

并发请求测试是评估AI生成工具处理能力的重要环节。当你需要验证系统在多用户同时访问时的表现时，可以采用以下方法：

import asyncio
import aiohttp
import time
import statistics

async def make_request(session, url, payload):
    start_time = time.time()
    try:
        async with session.post(url, json=payload) as response:
            data = await response.json()
            latency = time.time() - start_time
            return {
                "status": "success",
                "latency": latency,
                "response_length": len(str(data))
            }
    except Exception as e:
        latency = time.time() - start_time
        return {
            "status": "error",
            "latency": latency,
            "error": str(e)
        }

async def run_load_test(url, payload, concurrent_users, duration_seconds):
    results = []
    connector = aiohttp.TCPConnector(limit=concurrent_users)
    
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = []
        start_time = time.time()
        
        for _ in range(concurrent_users):
            task = asyncio.create_task(make_request(session, url, payload))
            tasks.append(task)
        
        while time.time() - start_time < duration_seconds:
            done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
            results.extend([task.result() for task in done])
            
            for task in done:
                tasks.remove(task)
                new_task = asyncio.create_task(make_request(session, url, payload))
                tasks.append(new_task)
        
         等待剩余任务完成
        results.extend([task.result() for task in await asyncio.gather(tasks)])
    
    return results

 使用示例
url = "https://api.example.com/ai-generate"
payload = {"prompt": "生成一段关于人工智能的文字", "max_tokens": 500}
results = await run_load_test(url, payload, concurrent_users=100, duration_seconds=60)

 分析结果
successful_requests = [r for r in results if r["status"] == "success"]
failed_requests = [r for r in results if r["status"] == "error"]

if successful_requests:
    latencies = [r["latency"] for r in successful_requests]
    avg_latency = statistics.mean(latencies)
    p95_latency = statistics.quantiles(latencies, n=20)[18]   95th percentile
    max_latency = max(latencies)
    
    print(f"成功请求数: {len(successful_requests)}")
    print(f"失败请求数: {len(failed_requests)}")
    print(f"平均响应时间: {avg_latency:.2f}秒")
    print(f"95%响应时间: {p95_latency:.2f}秒")
    print(f"最大响应时间: {max_latency:.2f}秒")
else:
    print("所有请求均失败")

这段代码使用Python的asyncio和aiohttp库实现了一个简单的并发请求测试。你可以根据需要调整并发用户数和测试持续时间。测试完成后，代码会计算并显示关键性能指标，如平均响应时间、95%响应时间和最大响应时间。

AI模型资源占用监控与分析

在压力测试过程中，监控AI模型的资源占用情况至关重要。你需要特别关注CPU使用率、内存消耗、GPU利用率和网络带宽等指标。以下是一个使用Prometheus和Grafana进行监控的配置示例：

 prometheus.yml 配置文件
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'ai_model_server'
    static_configs:
      - targets: ['localhost:9090']
    metrics_path: '/metrics'
    scrape_interval: 5s

  - job_name: 'node_exporter'
    static_configs:
      - targets: ['localhost:9100']

  - job_name: 'nvidia_gpu_exporter'
    static_configs:
      - targets: ['localhost:9835']

在AI模型服务器端，你需要集成Prometheus客户端来暴露指标：

from prometheus_client import start_http_server, Gauge, Counter, Histogram
import time
import psutil
import GPUtil

 定义指标
REQUEST_COUNT = Counter('ai_model_requests_total', 'Total requests', ['model', 'status'])
REQUEST_DURATION = Histogram('ai_model_request_duration_seconds', 'Request duration')
CPU_USAGE = Gauge('ai_model_cpu_usage_percent', 'CPU usage percent')
MEMORY_USAGE = Gauge('ai_model_memory_usage_bytes', 'Memory usage bytes')
GPU_USAGE = Gauge('ai_model_gpu_usage_percent', 'GPU usage percent', ['gpu_id'])
GPU_MEMORY = Gauge('ai_model_gpu_memory_usage_bytes', 'GPU memory usage bytes', ['gpu_id'])

def update_system_metrics():
    """更新系统资源使用指标"""
     CPU使用率
    cpu_percent = psutil.cpu_percent(interval=1)
    CPU_USAGE.set(cpu_percent)
    
     内存使用情况
    memory = psutil.virtual_memory()
    MEMORY_USAGE.set(memory.used)
    
     GPU使用情况
    gpus = GPUtil.getGPUs()
    for i, gpu in enumerate(gpus):
        GPU_USAGE.labels(gpu_id=i).set(gpu.load  100)
        GPU_MEMORY.labels(gpu_id=i).set(gpu.memoryUsed  1024  1024)   转换为字节

def process_request(model_name, prompt):
    """处理请求并记录指标"""
    start_time = time.time()
    try:
         模拟AI模型处理
        result = f"Generated content for: {prompt}"
        
         记录成功请求
        REQUEST_COUNT.labels(model=model_name, status='success').inc()
        REQUEST_DURATION.observe(time.time() - start_time)
        return result
    except Exception as e:
         记录失败请求
        REQUEST_COUNT.labels(model=model_name, status='error').inc()
        REQUEST_DURATION.observe(time.time() - start_time)
        raise e

if __name__ == '__main__':
     启动Prometheus指标服务器
    start_http_server(9090)
    
     定期更新系统指标
    while True:
        update_system_metrics()
        time.sleep(5)

通过这种监控方式，你可以实时了解AI模型在压力测试过程中的资源使用情况，并据此进行性能优化。

AI生成工具性能瓶颈分析与优化

在完成压力测试后，你可能会发现一些性能瓶颈。以下是常见的瓶颈及其优化方法：

模型推理优化

模型推理是AI生成工具的核心环节，也是最常见的性能瓶颈。你可以采用以下方法进行优化：

 使用TensorRT进行模型优化
import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit
import numpy as np

def build_engine(onnx_file_path, engine_file_path):
    """构建TensorRT引擎"""
    logger = trt.Logger(trt.Logger.WARNING)
    builder = trt.Builder(logger)
    network = builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH))
    parser = trt.OnnxParser(network, logger)
    
     解析ONNX模型
    with open(onnx_file_path, 'rb') as model:
        if not parser.parse(model.read()):
            for error in range(parser.num_errors):
                print(parser.get_error(error))
            return None
    
     配置构建参数
    config = builder.create_builder_config()
    config.max_workspace_size = 1 << 30   1GB
    
     启用FP16精度
    if builder.platform_has_fast_fp16:
        config.set_flag(trt.BuilderFlag.FP16)
    
     构建引擎
    engine = builder.build_engine(network, config)
    with open(engine_file_path, 'wb') as f:
        f.write(engine.serialize())
    
    return engine

def run_inference(engine, input_data):
    """使用TensorRT引擎进行推理"""
    context = engine.create_execution_context()
    
     分配输入输出内存
    input_binding_idx = engine.get_binding_index('input')
    output_binding_idx = engine.get_binding_index('output')
    
    input_size = trt.volume(engine.get_binding_shape(input_binding_idx))  engine.get_binding_dtype(input_binding_idx).itemsize
    output_size = trt.volume(engine.get_binding_shape(output_binding_idx))  engine.get_binding_dtype(output_binding_idx).itemsize
    
    d_input = cuda.mem_alloc(input_size)
    d_output = cuda.mem_alloc(output_size)
    
     创建流
    stream = cuda.Stream()
    
     将输入数据复制到设备
    cuda.memcpy_htod_async(d_input, input_data.astype(np.float32), stream)
    
     执行推理
    context.execute_async_v2(bindings=[int(d_input), int(d_output)], stream_handle=stream.handle)
    
     将结果从设备复制回主机
    h_output = np.empty(engine.get_binding_shape(output_binding_idx), dtype=np.float32)
    cuda.memcpy_dtoh_async(h_output, d_output, stream)
    
     同步流
    stream.synchronize()
    
    return h_output

 使用示例
engine = build_engine('model.onnx', 'model.engine')
input_data = np.random.randn(1, 3, 224, 224).astype(np.float32)
output = run_inference(engine, input_data)
print(output)

批处理优化

批处理是提高AI生成工具吞吐量的有效方法。通过将多个请求合并处理，可以显著提高资源利用率：

import queue
import threading
import time
from typing import List, Any

class BatchProcessor:
    def __init__(self, model, batch_size=8, max_wait_time=0.1):
        self.model = model
        self.batch_size = batch_size
        self.max_wait_time = max_wait_time
        self.request_queue = queue.Queue()
        self.result_dict = {}
        self.lock = threading.Lock()
        self.counter = 0
        self.running = True
        
         启动处理线程
        self.process_thread = threading.Thread(target=self._process_requests)
        self.process_thread.daemon = True
        self.process_thread.start()
    
    def _process_requests(self):
        """处理批处理请求的后台线程"""
        while self.running:
            batch = []
            request_ids = []
            start_time = time.time()
            
             收集请求，直到达到批大小或超时
            while len(batch) < self.batch_size and (time.time() - start_time)  int:
        """提交请求并返回请求ID"""
        with self.lock:
            request_id = self.counter
            self.counter += 1
        
        self.request_queue.put((request_id, request_data))
        return request_id
    
    def get_result(self, request_id: int, timeout: float = 10.0) -> Any:
        """获取请求结果"""
        start_time = time.time()
        while time.time() - start_time < timeout:
            with self.lock:
                if request_id in self.result_dict:
                    result = self.result_dict[request_id]
                    del self.result_dict[request_id]
                    return result
            
            time.sleep(0.01)
        
        raise TimeoutError(f"Request {request_id} timed out")
    
    def shutdown(self):
        """关闭批处理器"""
        self.running = False
        self.process_thread.join()

 使用示例
class MockModel:
    def process_batch(self, batch):
         模拟批处理推理
        time.sleep(0.1)   模拟推理时间
        return [f"Result for {data}" for data in batch]

model = MockModel()
processor = BatchProcessor(model, batch_size=4, max_wait_time=0.05)

 提交多个请求
request_ids = [processor.submit_request(f"Request {i}") for i in range(10)]

 获取结果
for request_id in request_ids:
    result = processor.get_result(request_id)
    print(f"Request {request_id}: {result}")

processor.shutdown()

缓存策略优化

对于重复或相似的请求，使用缓存可以显著减少计算负担：

import hashlib
import json
import time
from functools import wraps
from typing import Any, Dict, Optional

class LRUCache:
    def __init__(self, capacity: int, ttl: int = 3600):
        self.capacity = capacity
        self.ttl = ttl   缓存生存时间（秒）
        self.cache: Dict[str, Dict[str, Any]] = {}
        self.usage_order = []
    
    def _get_key(self, args, kwargs) -> str:
        """生成缓存键"""
         将参数转换为字符串并哈希
        args_str = json.dumps(args, sort_keys=True)
        kwargs_str = json.dumps(kwargs, sort_keys=True)
        key_str = f"{args_str}:{kwargs_str}"
        return hashlib.md5(key_str.encode()).hexdigest()
    
    def get(self, args, kwargs) -> Optional[Any]:
        """获取缓存值"""
        key = self._get_key(args, kwargs)
        
        if key in self.cache:
             检查是否过期
            if time.time() - self.cache[key]["timestamp"] > self.ttl:
                del self.cache[key]
                self.usage_order.remove(key)
                return None
            
             更新使用顺序
            self.usage_order.remove(key)
            self.usage_order.append(key)
            
            return self.cache[key]["value"]
        
        return None
    
    def set(self, args, kwargs, value: Any) -> None:
        """设置缓存值"""
        key = self._get_key(args, kwargs)
        
         如果缓存已满，删除最久未使用的项
        if len(self.cache) >= self.capacity and key not in self.cache:
            oldest_key = self.usage_order.pop(0)
            del self.cache[oldest_key]
        
         添加新值
        self.cache[key] = {
            "value": value,
            "timestamp": time.time()
        }
        
        if key in self.usage_order:
            self.usage_order.remove(key)
        self.usage_order.append(key)
    
    def clear(self) -> None:
        """清空缓存"""
        self.cache.clear()
        self.usage_order.clear()

def ai_cache(capacity: int = 1000, ttl: int = 3600):
    """AI模型缓存装饰器"""
    cache = LRUCache(capacity, ttl)
    
    def decorator(func):
        @wraps(func)
        def wrapper(args, kwargs):
             尝试从缓存获取结果
            cached_result = cache.get(args, kwargs)
            if cached_result is not None:
                return cached_result
            
             执行函数并缓存结果
            result = func(args, kwargs)
            cache.set(args, kwargs, result)
            return result
        
        def clear_cache():
            """清空缓存"""
            cache.clear()
        
        wrapper.clear_cache = clear_cache
        return wrapper
    
    return decorator

 使用示例
@ai_cache(capacity=100, ttl=600)   缓存100个结果，每个缓存10分钟
def generate_text(prompt, max_tokens=100):
    print(f"Generating text for: {prompt}")
     模拟AI生成过程
    time.sleep(1)   模拟处理时间
    return f"Generated text based on: {prompt[:20]}..."

 第一次调用，会执行实际生成
result1 = generate_text("人工智能的未来发展", max_tokens=50)
print(result1)

 第二次相同参数调用，会从缓存获取
result2 = generate_text("人工智能的未来发展", max_tokens=50)
print(result2)

 不同参数调用，会执行实际生成
result3 = generate_text("机器学习的基本概念", max_tokens=50)
print(result3)

 清空缓存
generate_text.clear_cache()

AI生成工具负载均衡与扩展策略

当单个AI生成工具实例无法满足需求时，负载均衡和水平扩展是必要的解决方案。以下是一个基于Docker和Kubernetes的部署方案：

 docker-compose.yml
version: '3.8'

services:
  ai-model-server:
    image: your-ai-model-server:latest
    ports:
      - "8000:8000"
    environment:
      - MODEL_PATH=/models/model.bin
      - MAX_BATCH_SIZE=8
      - MAX_WORKERS=4
    volumes:
      - ./models:/models
    deploy:
      replicas: 3
      resources:
        limits:
          cpus: '2.0'
          memory: 4G
        reservations:
          cpus: '1.0'
          memory: 2G
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      
  nginx:
    image: nginx:latest
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - ai-model-server

 nginx.conf
events {
    worker_connections 1024;
}

http {
    upstream ai_model_servers {
        least_conn;   使用最少连接数策略
        server ai-model-server:8000;
        server ai-model-server:8000;
        server ai-model-server:8000;
    }
    
    server {
        listen 80;
        
        location / {
            proxy_pass http://ai_model_servers;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
            
             超时设置
            proxy_connect_timeout 5s;
            proxy_send_timeout 60s;
            proxy_read_timeout 60s;
            
             缓冲设置
            proxy_buffering on;
            proxy_buffer_size 4k;
            proxy_buffers 8 4k;
        }
        
        location /health {
            access_log off;
            return 200 "healthyn";
        }
    }
}

对于更复杂的部署场景，你可以使用Kubernetes进行容器编排：

 ai-model-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-model-server
spec:
  replicas: 5
  selector:
    matchLabels:
      app: ai-model-server
  template:
    metadata:
      labels:
        app: ai-model-server
    spec:
      containers:
      - name: ai-model-server
        image: your-ai-model-server:latest
        ports:
        - containerPort: 8000
        env:
        - name: MODEL_PATH
          value: "/models/model.bin"
        - name: MAX_BATCH_SIZE
          value: "8"
        - name: MAX_WORKERS
          value: "4"
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
          limits:
            memory: "4Gi"
            cpu: "2"
        volumeMounts:
        - name: model-storage
          mountPath: /models
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
      volumes:
      - name: model-storage
        persistentVolumeClaim:
          claimName: ai-model-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: ai-model-service
spec:
  selector:
    app: ai-model-server
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: ai-model-ingress
  annotations:
    nginx.ingress.kubernetes.io/load-balance: "least_conn"
    nginx.ingress.kubernetes.io/proxy-connect-timeout: "5"
    nginx.ingress.kubernetes.io/proxy-send-timeout: "60"
    nginx.ingress.kubernetes.io/proxy-read-timeout: "60"
spec:
  rules:
  - http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: ai-model-service
            port:
              number: 80

通过这种部署方式，你可以根据负载情况动态调整AI生成工具的实例数量，确保系统在高并发情况下仍能保持良好的性能表现。

AI生成工具压力测试自动化与持续集成

将压力测试集成到CI/CD流程中，可以帮助你及时发现性能回归。以下是一个使用GitHub Actions实现的自动化压力测试流程：

 .github/workflows/performance-test.yml
name: Performance Test

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  performance-test:
    runs-on: ubuntu-latest
    
    services:
      ai-model-server:
        image: your-ai-model-server:latest
        ports:
          - 8000:8000
        options: >-
          --health-cmd "curl -f http://localhost:8000/health || exit 1"
          --health-interval 30s
          --health-timeout 10s
          --health-retries 3
    
    steps:
    - uses: actions/checkout@v2
    
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.9'
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
    
    - name: Wait for server to be ready
      run: |
        for i in {1..30}; do
          if curl -f http://localhost:8000/health; then
            echo "Server is ready"
            break
          fi
          echo "Waiting for server to be ready..."
          sleep 2
        done
    
    - name: Run performance test
      run: |
        python performance_test.py --url http://localhost:8000 --concurrent 50 --duration 60 --report performance_report.json
    
    - name: Analyze performance results
      run: |
        python analyze_performance.py performance_report.json --thresholds thresholds.json
    
    - name: Upload performance report
      uses: actions/upload-artifact@v2
      with:
        name: performance-report
        path: performance_report.json
    
    - name: Comment PR with performance results
      if: github.event_name == 'pull_request'
      uses: actions/github-script@v3
      with:
        github-token: ${{ secrets.GITHUB_TOKEN }}
        script: |
          const fs = require('fs');
          const report = JSON.parse(fs.readFileSync('performance_report.json', 'utf8'));
          
          const comment = `
           性能测试结果
          
          - 平均响应时间: ${report.avg_latency.toFixed(2)}秒
          - 95%响应时间: ${report.p95_latency.toFixed(2)}秒
          - 最大响应时间: ${report.max_latency.toFixed(2)}秒
          - 请求成功率: ${(report.success_rate  100).toFixed(2)}%
          - 吞吐量: ${report.throughput.toFixed(2)}请求/秒
          
          ${report.regression ? '⚠️ 检测到性能回归！' : '✅ 性能测试通过'}
          `;
          
          github.issues.createComment({
            issue_number: context.issue.number,
            owner: context.repo.owner,
            repo: context.repo.repo,
            body: comment
          });

对应的性能测试脚本可能如下所示：

 performance_test.py
import argparse
import asyncio
import aiohttp
import json
import statistics
import time
from datetime import datetime

async def make_request(session, url, payload):
    start_time = time.time()
    try:
        async with session.post(url, json=payload) as response:
            data = await response.json()
            latency = time.time() - start_time
            return {
                "status": "success",
                "latency": latency,
                "status_code": response.status,
                "response_length": len(str(data))
            }
    except Exception as e:
        latency = time.time() - start_time
        return {
            "status": "error",
            "latency": latency,
            "error": str(e)
        }

async def run_load_test(url, payload, concurrent_users, duration_seconds):
    results = []
    connector = aiohttp.TCPConnector(limit=concurrent_users)
    
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = []
        start_time = time.time()
        total_requests = 0
        
        for _ in range(concurrent_users):
            task = asyncio.create_task(make_request(session, url, payload))
            tasks.append(task)
            total_requests += 1
        
        while time.time() - start_time < duration_seconds:
            done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
            results.extend([task.result() for task in done])
            
            for task in done:
                tasks.remove(task)
                new_task = asyncio.create_task(make_request(session, url, payload))
                tasks.append(new_task)
                total_requests += 1
        
         等待剩余任务完成
        remaining_results = await asyncio.gather(tasks)
        results.extend(remaining_results)
    
    return results, total_requests

def analyze_results(results, total_requests, duration_seconds):
    successful_requests = [r for r in results if r["status"] == "success"]
    failed_requests = [r for r in results if r["status"] == "error"]
    
    if successful_requests:
        latencies = [r["latency"] for r in successful_requests]
        avg_latency = statistics.mean(latencies)
        p95_latency = statistics.quantiles(latencies, n=20)[18]   95th percentile
        max_latency = max(latencies)
        min_latency = min(latencies)
        
        throughput = len(successful_requests) / duration_seconds
        success_rate = len(successful_requests) / total_requests
        
        return {
            "timestamp": datetime.now().isoformat(),
            "duration_seconds": duration_seconds,
            "total_requests": total_requests,
            "successful_requests": len(successful_requests),
            "failed_requests": len(failed_requests),
            "success_rate": success_rate,
            "avg_latency": avg_latency,
            "p95_latency": p95_latency,
            "max_latency": max_latency,
            "min_latency": min_latency,
            "throughput": throughput,
            "errors": [r["error"] for r in failed_requests[:10]]   只记录前10个错误
        }
    else:
        return {
            "timestamp": datetime.now().isoformat(),
            "duration_seconds": duration_seconds,
            "total_requests": total_requests,
            "successful_requests": 0,
            "failed_requests": len(failed_requests),
            "success_rate": 0,
            "avg_latency": 0,
            "p95_latency": 0,
            "max_latency": 0,
            "min_latency": 0,
            "throughput": 0,
            "errors": [r["error"] for r in failed_requests]
        }

def main():
    parser = argparse.ArgumentParser(description="AI Model Performance Test")
    parser.add_argument("--url", required=True, help="AI model server URL")
    parser.add_argument("--concurrent", type=int, default=10, help="Number of concurrent users")
    parser.add_argument("--duration", type=int, default=60, help="Test duration in seconds")
    parser.add_argument("--report", required=True, help="Output report file path")
    args = parser.parse_args()
    
     测试负载
    payload = {
        "prompt": "生成一段关于人工智能的文字",
        "max_tokens": 500
    }
    
    print(f"Starting performance test against {args.url}")
    print(f"Concurrent users: {args.concurrent}")
    print(f"Duration: {args.duration} seconds")
    
     运行测试
    results, total_requests = asyncio.run(run_load_test(
        args.url, payload, args.concurrent, args.duration
    ))
    
     分析结果
    report = analyze_results(results, total_requests, args.duration)
    
     保存报告
    with open(args.report, 'w') as f:
        json.dump(report, f, indent=2)
    
    print(f"Performance test completed. Report saved to {args.report}")
    print(f"Average latency: {report['avg_latency']:.2f} seconds")
    print(f"95th percentile latency: {report['p95_latency']:.2f} seconds")
    print(f"Success rate: {report['success_rate']  100:.2f}%")
    print(f"Throughput: {report['throughput']:.2f} requests/second")

if __name__ == "__main__":
    main()

通过这种自动化测试流程，你可以在每次代码变更时自动运行性能测试，及时发现性能问题，确保AI生成工具的性能不会随着代码迭代而下降。