┌─────────────────────────────────────────────────────────────┐
│                    🌐 负载均衡层                            │
│              Nginx / ALB / CloudFlare                      │
├─────────────────────────────────────────────────────────────┤
│                    🐳 应用服务层                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  Web前端    │  │  API网关    │  │  AI服务     │        │
│  │  (Vue3)     │  │  (FastAPI)  │  │  (AutoGen)  │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    🧠 AI服务层                              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  模型服务   │  │  向量检索   │  │  智能体编排 │        │
│  │  (vLLM)     │  │  (ChromaDB) │  │  (AutoGen)  │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    💾 数据存储层                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  关系数据库 │  │  缓存数据库 │  │  文件存储   │        │
│  │ (PostgreSQL)│  │   (Redis)   │  │   (MinIO)   │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    📊 监控运维层                            │
│        Prometheus + Grafana + Jaeger + ELK                 │
└─────────────────────────────────────────────────────────────┘

技术选型：稳定可靠的工具链

容器化技术：

Docker：应用容器化，环境一致性保障
Docker Compose：本地开发和小规模部署
Kubernetes：大规模生产环境编排

AI服务优化：

vLLM：高性能的LLM推理服务
Ollama：本地模型部署方案
模型缓存：减少重复调用成本

监控告警：

Prometheus：指标收集和存储
Grafana：可视化监控面板
Jaeger：分布式链路追踪

🐳 容器化部署实战

1. AI服务容器化

构建高效的AI服务容器：

# AI服务Dockerfile
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 设置环境变量
ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    gcc \
    g++ \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件
COPY requirements.txt .

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建非root用户
RUN adduser --disabled-password --gecos '' appuser && \
    chown -R appuser:appuser /app
USER appuser

# 暴露端口
EXPOSE 8000

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

2. Docker Compose配置

完整的服务编排配置：

version: '3.8'

services:
  # 前端服务
  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile
    ports:
      - "80:80"
    depends_on:
      - api-gateway
    networks:
      - ai-platform
    restart: unless-stopped

  # API网关
  api-gateway:
    build:
      context: ./backend
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://postgres:password@postgres:5432/ai_platform
      - REDIS_URL=redis://redis:6379/0
      - AI_SERVICE_URL=http://ai-service:8001
    depends_on:
      - postgres
      - redis
      - ai-service
    networks:
      - ai-platform
    restart: unless-stopped
    volumes:
      - ./logs:/app/logs

  # AI服务
  ai-service:
    build:
      context: ./ai-service
      dockerfile: Dockerfile
    ports:
      - "8001:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY}
      - CHROMA_HOST=chromadb
      - CHROMA_PORT=8000
    depends_on:
      - chromadb
    networks:
      - ai-platform
    restart: unless-stopped
    volumes:
      - ./ai-models:/app/models
      - ./knowledge-base:/app/knowledge-base
    deploy:
      resources:
        limits:
          memory: 4G
          cpus: '2.0'

  # 向量数据库
  chromadb:
    image: chromadb/chroma:latest
    ports:
      - "8002:8000"
    environment:
      - CHROMA_SERVER_HOST=0.0.0.0
      - CHROMA_SERVER_HTTP_PORT=8000
    networks:
      - ai-platform
    restart: unless-stopped
    volumes:
      - chroma_data:/chroma/chroma

  # PostgreSQL数据库
  postgres:
    image: postgres:15
    environment:
      - POSTGRES_DB=ai_platform
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password
    ports:
      - "5432:5432"
    networks:
      - ai-platform
    restart: unless-stopped
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql

  # Redis缓存
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    networks:
      - ai-platform
    restart: unless-stopped
    volumes:
      - redis_data:/data
    command: redis-server --appendonly yes --maxmemory 1gb --maxmemory-policy allkeys-lru

  # Nginx负载均衡
  nginx:
    image: nginx:alpine
    ports:
      - "443:443"
      - "80:80"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf
      - ./nginx/ssl:/etc/nginx/ssl
    depends_on:
      - frontend
      - api-gateway
    networks:
      - ai-platform
    restart: unless-stopped

  # Prometheus监控
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    networks:
      - ai-platform
    restart: unless-stopped

  # Grafana可视化
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana_data:/var/lib/grafana
      - ./monitoring/grafana/dashboards:/etc/grafana/provisioning/dashboards
    networks:
      - ai-platform
    restart: unless-stopped

networks:
  ai-platform:
    driver: bridge

volumes:
  postgres_data:
  redis_data:
  chroma_data:
  prometheus_data:
  grafana_data:

⚡ 性能优化策略

1. AI服务优化

优化AI调用性能和成本：

import asyncio
import time
from typing import Dict, Any, Optional
from functools import lru_cache
import hashlib
import json

class AIServiceOptimizer:
    """AI服务优化器 - 性能的'加速器'"""
    
    def __init__(self):
        self.response_cache = {}
        self.rate_limiter = RateLimiter()
        self.model_pool = ModelPool()
        self.metrics = MetricsCollector()
    
    async def optimized_ai_call(self, prompt: str, model: str = "gpt-4", **kwargs) -> str:
        """优化的AI调用"""
        start_time = time.time()
        
        try:
            # 1. 检查缓存
            cache_key = self._generate_cache_key(prompt, model, kwargs)
            cached_response = self._get_cached_response(cache_key)
            if cached_response:
                self.metrics.record_cache_hit(model)
                return cached_response
            
            # 2. 速率限制
            await self.rate_limiter.acquire(model)
            
            # 3. 模型负载均衡
            available_model = await self.model_pool.get_available_model(model)
            
            # 4. 执行AI调用
            response = await self._execute_ai_call(prompt, available_model, **kwargs)
            
            # 5. 缓存响应
            self._cache_response(cache_key, response)
            
            # 6. 记录指标
            duration = time.time() - start_time
            self.metrics.record_ai_call(model, duration, len(prompt), len(response))
            
            return response
            
        except Exception as e:
            self.metrics.record_error(model, str(e))
            raise
    
    def _generate_cache_key(self, prompt: str, model: str, kwargs: Dict) -> str:
        """生成缓存键"""
        content = f"{model}:{prompt}:{json.dumps(kwargs, sort_keys=True)}"
        return hashlib.md5(content.encode()).hexdigest()
    
    def _get_cached_response(self, cache_key: str) -> Optional[str]:
        """获取缓存响应"""
        cache_entry = self.response_cache.get(cache_key)
        if cache_entry:
            # 检查缓存是否过期（1小时）
            if time.time() - cache_entry['timestamp'] < 3600:
                return cache_entry['response']
            else:
                del self.response_cache[cache_key]
        return None
    
    def _cache_response(self, cache_key: str, response: str):
        """缓存响应"""
        self.response_cache[cache_key] = {
            'response': response,
            'timestamp': time.time()
        }
        
        # 限制缓存大小
        if len(self.response_cache) > 1000:
            # 删除最旧的缓存项
            oldest_key = min(self.response_cache.keys(), 
                           key=lambda k: self.response_cache[k]['timestamp'])
            del self.response_cache[oldest_key]

class RateLimiter:
    """速率限制器"""
    
    def __init__(self):
        self.limits = {
            "gpt-4": {"requests_per_minute": 60, "tokens_per_minute": 40000},
            "gpt-3.5-turbo": {"requests_per_minute": 200, "tokens_per_minute": 90000},
            "deepseek": {"requests_per_minute": 100, "tokens_per_minute": 50000}
        }
        self.usage = {}
    
    async def acquire(self, model: str):
        """获取调用许可"""
        current_minute = int(time.time() // 60)
        
        if model not in self.usage:
            self.usage[model] = {}
        
        minute_usage = self.usage[model].get(current_minute, {"requests": 0, "tokens": 0})
        limit = self.limits.get(model, {"requests_per_minute": 60, "tokens_per_minute": 40000})
        
        if minute_usage["requests"] >= limit["requests_per_minute"]:
            # 等待到下一分钟
            wait_time = 60 - (time.time() % 60)
            await asyncio.sleep(wait_time)
        
        # 更新使用量
        minute_usage["requests"] += 1
        self.usage[model][current_minute] = minute_usage

class ModelPool:
    """模型池管理"""
    
    def __init__(self):
        self.models = {
            "gpt-4": {"endpoint": "openai", "status": "available", "load": 0},
            "deepseek": {"endpoint": "deepseek", "status": "available", "load": 0},
            "local-llm": {"endpoint": "ollama", "status": "available", "load": 0}
        }
    
    async def get_available_model(self, preferred_model: str) -> str:
        """获取可用模型"""
        # 检查首选模型是否可用
        if (preferred_model in self.models and 
            self.models[preferred_model]["status"] == "available" and
            self.models[preferred_model]["load"] < 0.8):
            return preferred_model
        
        # 选择负载最低的可用模型
        available_models = [
            (model, info) for model, info in self.models.items()
            if info["status"] == "available" and info["load"] < 0.9
        ]
        
        if available_models:
            return min(available_models, key=lambda x: x[1]["load"])[0]
        
        # 如果没有可用模型，等待并重试
        await asyncio.sleep(1)
        return await self.get_available_model(preferred_model)

class MetricsCollector:
    """指标收集器"""
    
    def __init__(self):
        self.metrics = {
            "ai_calls_total": 0,
            "ai_calls_duration": [],
            "cache_hits": 0,
            "cache_misses": 0,
            "errors": []
        }
    
    def record_ai_call(self, model: str, duration: float, prompt_length: int, response_length: int):
        """记录AI调用指标"""
        self.metrics["ai_calls_total"] += 1
        self.metrics["ai_calls_duration"].append(duration)
        
        # 保持最近1000次调用的记录
        if len(self.metrics["ai_calls_duration"]) > 1000:
            self.metrics["ai_calls_duration"] = self.metrics["ai_calls_duration"][-1000:]
    
    def record_cache_hit(self, model: str):
        """记录缓存命中"""
        self.metrics["cache_hits"] += 1
    
    def record_error(self, model: str, error: str):
        """记录错误"""
        self.metrics["errors"].append({
            "model": model,
            "error": error,
            "timestamp": time.time()
        })
    
    def get_performance_stats(self) -> Dict[str, Any]:
        """获取性能统计"""
        durations = self.metrics["ai_calls_duration"]
        
        return {
            "total_calls": self.metrics["ai_calls_total"],
            "avg_duration": sum(durations) / len(durations) if durations else 0,
            "p95_duration": sorted(durations)[int(len(durations) * 0.95)] if durations else 0,
            "cache_hit_rate": self.metrics["cache_hits"] / max(self.metrics["ai_calls_total"], 1),
            "error_rate": len(self.metrics["errors"]) / max(self.metrics["ai_calls_total"], 1)
        }

2. 数据库优化

优化数据库性能：

from sqlalchemy import create_engine, text
from sqlalchemy.pool import QueuePool
import asyncio

class DatabaseOptimizer:
    """数据库优化器"""
    
    def __init__(self, database_url: str):
        # 连接池优化
        self.engine = create_engine(
            database_url,
            poolclass=QueuePool,
            pool_size=20,
            max_overflow=30,
            pool_pre_ping=True,
            pool_recycle=3600
        )
        
        self.setup_database_optimization()
    
    def setup_database_optimization(self):
        """设置数据库优化"""
        with self.engine.connect() as conn:
            # 设置PostgreSQL优化参数
            optimizations = [
                "SET shared_buffers = '256MB'",
                "SET effective_cache_size = '1GB'",
                "SET maintenance_work_mem = '64MB'",
                "SET checkpoint_completion_target = 0.9",
                "SET wal_buffers = '16MB'",
                "SET default_statistics_target = 100"
            ]
            
            for sql in optimizations:
                try:
                    conn.execute(text(sql))
                except Exception as e:
                    print(f"优化设置失败: {sql}, 错误: {e}")
    
    def create_indexes(self):
        """创建性能索引"""
        indexes = [
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_test_cases_project_id ON test_cases(project_id)",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_test_cases_type ON test_cases(type)",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_test_cases_priority ON test_cases(priority)",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_execution_records_created_at ON execution_records(created_at)",
            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_chat_messages_session_id ON chat_messages(session_id)",
        ]
        
        with self.engine.connect() as conn:
            for index_sql in indexes:
                try:
                    conn.execute(text(index_sql))
                    print(f"索引创建成功: {index_sql}")
                except Exception as e:
                    print(f"索引创建失败: {index_sql}, 错误: {e}")

📊 监控告警系统

监控配置

# monitoring/prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - "alert_rules.yml"

scrape_configs:
  - job_name: 'ai-platform'
    static_configs:
      - targets: ['api-gateway:8000', 'ai-service:8000']
    metrics_path: /metrics
    scrape_interval: 10s

  - job_name: 'postgres'
    static_configs:
      - targets: ['postgres:5432']

  - job_name: 'redis'
    static_configs:
      - targets: ['redis:6379']

alerting:
  alertmanagers:
    - static_configs:
        - targets:
          - alertmanager:9093

告警规则

# monitoring/alert_rules.yml
groups:
  - name: ai_platform_alerts
    rules:
      - alert: HighAIServiceLatency
        expr: histogram_quantile(0.95, ai_service_duration_seconds) > 10
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "AI服务响应延迟过高"
          description: "AI服务95%分位延迟超过10秒"

      - alert: HighErrorRate
        expr: rate(ai_service_errors_total[5m]) > 0.1
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "AI服务错误率过高"
          description: "AI服务错误率超过10%"

      - alert: DatabaseConnectionHigh
        expr: postgres_connections_active / postgres_connections_max > 0.8
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "数据库连接数过高"
          description: "数据库连接使用率超过80%"

🎯 部署最佳实践

1. 环境配置管理

#!/bin/bash
# deploy.sh - 一键部署脚本

set -e

# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'

log_info() {
    echo -e "${GREEN}[INFO]${NC} $1"
}

log_warn() {
    echo -e "${YELLOW}[WARN]${NC} $1"
}

log_error() {
    echo -e "${RED}[ERROR]${NC} $1"
}

# 检查环境
check_environment() {
    log_info "检查部署环境..."
    
    if ! command -v docker &> /dev/null; then
        log_error "Docker未安装"
        exit 1
    fi
    
    if ! command -v docker-compose &> /dev/null; then
        log_error "Docker Compose未安装"
        exit 1
    fi
    
    log_info "环境检查通过"
}

# 生成配置文件
generate_config() {
    log_info "生成配置文件..."
    
    # 生成随机密钥
    DB_PASSWORD=$(openssl rand -base64 32)
    SECRET_KEY=$(openssl rand -base64 32)
    
    cat > .env << EOF
# 数据库配置
DATABASE_URL=postgresql://postgres:${DB_PASSWORD}@postgres:5432/ai_platform
POSTGRES_PASSWORD=${DB_PASSWORD}

# AI服务配置
OPENAI_API_KEY=${OPENAI_API_KEY:-your-openai-key}
DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY:-your-deepseek-key}

# 应用配置
SECRET_KEY=${SECRET_KEY}
DEBUG=false
ENVIRONMENT=production

# Redis配置
REDIS_URL=redis://redis:6379/0
EOF
    
    log_info "配置文件生成完成"
}

# 部署服务
deploy_services() {
    log_info "部署AI测试平台..."
    
    # 拉取最新镜像
    docker-compose pull
    
    # 启动服务
    docker-compose up -d
    
    # 等待服务启动
    log_info "等待服务启动..."
    sleep 30
    
    # 健康检查
    if curl -f http://localhost:8000/health > /dev/null 2>&1; then
        log_info "✅ AI测试平台部署成功！"
        echo ""
        echo "🎉 访问地址："
        echo "  前端界面: http://localhost"
        echo "  API文档: http://localhost:8000/docs"
        echo "  监控面板: http://localhost:3000"
        echo ""
    else
        log_error "❌ 服务启动失败"
        docker-compose logs
        exit 1
    fi
}

# 主函数
main() {
    log_info "开始部署AI测试平台..."
    
    check_environment
    generate_config
    deploy_services
    
    log_info "部署完成！"
}

main "$@"

🎉 总结

AI测试平台的部署与优化是一个系统工程，通过合理的架构设计、性能优化和监控告警，我们可以构建一个：

高可用：7x24小时稳定运行的AI测试服务
高性能：快速响应的AI调用和数据处理
可扩展：支持业务增长的弹性架构
可观测：全方位的监控和告警体系

至此，我们的AI测试平台开发系列就全部完成了！从入门到部署，从理论到实践，希望这个系列能够帮助你构建出真正有价值的AI测试工具。

💡 部署优化小贴士：部署不是终点，而是起点。持续监控、持续优化、持续改进，才能让AI测试平台真正发挥价值。记住，技术是手段，解决问题才是目的！