第5章：MVP开发与技术架构

┌────────────────────────────────────┐
│        MVP开发挑战矩阵              │
├────────────────────────────────────┤
│                                    │
│  高 ┤ 实时渲染  │  物理模拟        │
│     │           │                  │
│  技 │-----------┼──────────────    │
│  术 │           │                  │
│  复 │ 3D生成    │  纹理优化        │
│  杂 │           │                  │
│  度 │-----------┼──────────────    │
│     │           │                  │
│  低 ┤ 格式转换  │  简单变形        │
│     └───────────┴──────────────    │
│       低        用户价值        高   │
└────────────────────────────────────┘

Priority Score = (Reach × Impact × Confidence) / Effort

        是否是核心价值主张？
              │
      ┌───────┴───────┐
      是              否
      │               │
  技术可行性？    是否有竞争优势？
      │               │
  ┌───┴───┐      ┌───┴───┐
  高      低      是      否
  │       │       │       │
 P0     延后     P1     舍弃

┌─────────────────────────────────────────┐
│            API Gateway                   │
│         (Kong/Envoy)                     │
└─────────┬───────────────────────────────┘
          │
    ┌─────┴─────┬─────────┬──────────┐
    │           │         │          │
┌───▼───┐ ┌────▼────┐ ┌──▼───┐ ┌───▼────┐
│Auth   │ │3D Upload│ │AI    │ │Render  │
│Service│ │Service  │ │Service│ │Service │
└───────┘ └─────────┘ └──────┘ └────────┘
    │           │         │          │
    └───────────┴─────────┴──────────┘
                │
        ┌───────▼────────┐
        │  Message Queue │
        │  (RabbitMQ/   │
        │   Kafka)       │
        └────────────────┘

# 使用对象存储（S3/OSS）作为数据交换层
class Asset3DService:
    def process_model(self, model_id):
        # 1. 从对象存储获取模型URL
        model_url = self.storage.get_presigned_url(model_id)

        # 2. 传递URL而非数据
        job_id = self.queue.submit({
            'model_url': model_url,
            'operation': 'optimize',
            'params': {...}
        })

        # 3. 异步处理
        return {'job_id': job_id, 'status': 'processing'}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-inference-service
spec:
  replicas: 2
  template:
    spec:
      containers:

      - name: inference
        image: your-registry/3d-ai-inference:v1.0
        resources:
          limits:
            nvidia.com/gpu: 1  # 请求1个GPU
            memory: 32Gi
            cpu: 8
          requests:
            nvidia.com/gpu: 1
            memory: 16Gi
            cpu: 4
        volumeMounts:

        - name: model-cache
          mountPath: /models
      nodeSelector:
        gpu-type: "tesla-t4"  # 指定GPU型号

┌────────────────────────────────┐
│     热数据（NVMe SSD）          │ <- 正在处理的模型
├────────────────────────────────┤
│     温数据（SSD）               │ <- 最近访问的缓存
├────────────────────────────────┤
│     冷数据（对象存储）           │ <- 归档的原始数据
└────────────────────────────────┘

// 客户端代码示例
class ModelLoader {
  async loadModel(modelId) {
    // 1. 获取CDN URL（带区域路由）
    const cdnUrl = await this.getCDNUrl(modelId);

    // 2. 并行下载几何和纹理
    const [geometry, textures] = await Promise.all([
      this.fetchCompressed(cdnUrl + '/geometry.draco'),
      this.fetchTextures(cdnUrl + '/textures/')
    ]);

    // 3. 本地解压和组装
    return this.assembleModel(geometry, textures);
  }
}

# NVIDIA MPS (Multi-Process Service) 配置
# 允许多个进程共享GPU
export CUDA_VISIBLE_DEVICES=0
nvidia-smi -i 0 -c EXCLUSIVE_PROCESS
nvidia-cuda-mps-control -d

# 创建GPU实例
nvidia-smi mig -cgi 2g.10gb,3g.20gb -i 0
# 为不同负载分配不同规格的GPU实例

class BatchInferenceService:
    def __init__(self, batch_size=8, wait_time_ms=50):
        self.batch_queue = []
        self.batch_size = batch_size
        self.wait_time = wait_time_ms

    async def add_request(self, request):
        self.batch_queue.append(request)

        if len(self.batch_queue) >= self.batch_size:
            return await self.process_batch()

        # 等待更多请求或超时
        await asyncio.sleep(self.wait_time / 1000)
        if self.batch_queue:
            return await self.process_batch()

┌─────────────────────────────────┐
│      请求分发层                  │
└──────┬──────────────────────────┘
       │
   ┌───┴───┐
   │Router │
   └───┬───┘
       │
┌──────┴──────┬──────────┬──────────┐
│             │          │          │
▼             ▼          ▼          ▼
自有GPU    Spot实例   Reserved   按需实例
(基础负载)  (批处理)   (预测负载)  (峰值)

class SpotInstanceManager:
    def __init__(self):
        self.spot_pool = []
        self.on_demand_pool = []

    def handle_interruption(self, instance_id):
        # 1. 将任务迁移到按需实例
        tasks = self.get_running_tasks(instance_id)
        self.migrate_tasks(tasks, self.on_demand_pool)

        # 2. 请求新的Spot实例
        self.request_spot_instance()

    def cost_optimizer(self):
        # 根据价格动态调整实例组合
        spot_price = self.get_spot_price()
        on_demand_price = self.get_on_demand_price()

        if spot_price < on_demand_price * 0.3:
            self.increase_spot_ratio()
        else:
            self.increase_on_demand_ratio()

# 量化示例（INT8）
import torch
from torch.quantization import quantize_dynamic

# 动态量化
quantized_model = quantize_dynamic(
    original_model,
    {torch.nn.Linear, torch.nn.Conv3d},
    dtype=torch.qint8
)

# 性能对比
# 原始模型：100ms/推理，16GB显存
# 量化模型：30ms/推理，4GB显存
# 精度损失：<2%

class InferenceCache:
    def __init__(self):
        self.l1_cache = {}  # 内存缓存（LRU）
        self.l2_cache = Redis()  # 分布式缓存
        self.l3_cache = S3()  # 持久化存储

    async def get_or_compute(self, key, compute_fn):
        # L1查找
        if key in self.l1_cache:
            return self.l1_cache[key]

        # L2查找
        result = await self.l2_cache.get(key)
        if result:
            self.l1_cache[key] = result
            return result

        # L3查找
        result = await self.l3_cache.get(key)
        if result:
            await self.l2_cache.set(key, result)
            self.l1_cache[key] = result
            return result

        # 计算并存储
        result = await compute_fn()
        await self.store_all_levels(key, result)
        return result

技术债务评分 = Σ(影响范围 × 严重程度 × 修复成本)

影响范围（1-5分）：
1 - 影响单个模块
2 - 影响2-3个模块
3 - 影响整个服务
4 - 影响多个服务
5 - 影响整个系统

严重程度（1-5分）：
1 - 代码可读性问题
2 - 性能次优
3 - 扩展性受限
4 - 安全隐患
5 - 系统稳定性风险

修复成本（人天）：
直接转换为数值

class TechDebtTracker:
    def __init__(self):
        self.debt_items = []

    def add_debt(self, description, impact, severity, cost):
        debt_score = impact * severity * cost
        self.debt_items.append({
            'description': description,
            'score': debt_score,
            'added_date': datetime.now(),
            'status': 'open'
        })

    def prioritize_debts(self):
        # 按债务分数和时间加权排序
        return sorted(self.debt_items, 
                     key=lambda x: x['score'] * 
                     (1 + (datetime.now() - x['added_date']).days / 365))

# 示例：记录渲染管线的技术债务
tracker = TechDebtTracker()
tracker.add_debt(
    "使用CPU进行法线计算而非GPU",
    impact=3,  # 影响整个渲染服务
    severity=2,  # 性能次优
    cost=5  # 5人天修复
)

         高 ┬───────────────────────────┐
            │   立即重构  │  计划重构    │
     新     │            │              │
     功     ├────────────┼──────────────┤
     能     │            │              │
     开     │  延后重构  │  技术破产    │
     发     │            │              │
     速     └────────────┴──────────────┘
     度低      低      技术债务程度      高

class RefactoringStrategy:
    def __init__(self):
        self.strategies = {
            'strangler_fig': self.strangler_pattern,
            'branch_by_abstraction': self.branch_abstraction,
            'parallel_run': self.parallel_implementation
        }

    def strangler_pattern(self, old_service, new_service):
        """
        逐步用新服务替代旧服务
        """
        router = APIRouter()

        # 阶段1：所有流量到旧服务
        router.add_route('/api/*', old_service, weight=100)

        # 阶段2：部分流量到新服务
        router.add_route('/api/v2/*', new_service, weight=10)

        # 阶段3：逐步增加新服务流量
        # 阶段4：完全切换到新服务
        return router

    def parallel_implementation(self, feature):
        """
        新旧实现并行运行，对比结果
        """
        old_result = old_implementation(feature)
        new_result = new_implementation(feature)

        # 对比并记录差异
        if not self.results_match(old_result, new_result):
            self.log_discrepancy(feature, old_result, new_result)

        # 返回旧实现结果（保证稳定性）
        return old_result

阶段1：MVP（0-6月）
├── 单体应用
├── 单机GPU
└── 文件存储

阶段2：产品市场契合（6-18月）
├── 服务拆分
├── GPU集群
└── 对象存储

阶段3：规模化（18-36月）
├── 微服务架构
├── 混合云GPU
└── 分布式存储

阶段4：平台化（36月+）
├── 服务网格
├── 边缘计算
└── 数据湖

# ADR-001: 选择PyTorch作为深度学习框架

## 状态
已接受

## 背景
需要选择3D AI模型的训练和推理框架。

## 决策
选择PyTorch 2.0作为主要深度学习框架。

## 理由

1. 3D视觉社区活跃（PyTorch3D生态）
2. 动态图便于调试
3. TorchScript支持生产部署
4. 团队熟悉度高

## 后果

- 正面：快速开发，社区支持
- 负面：推理性能不如TensorRT
- 缓解：关键路径使用ONNX转换

## 替代方案

- TensorFlow：生态成熟但3D支持较弱
- JAX：性能优秀但生态不成熟

# 监控GPU利用率
nvidia-smi dmon -s u -c 10

# 分析显存使用
nvidia-smi --query-gpu=memory.used,memory.free --format=csv -l 1

# 识别性能瓶颈
nsys profile --stats=true python inference.py

// 错误：同步加载
const model = await loadModel(url);  // 可能需要30秒

// 正确：渐进式加载
const loader = new ProgressiveLoader();
loader.loadLOD(url, level=0);  // 低模，1秒
loader.on('progress', (lod) => render(lod));
loader.loadFullModel(url);  // 后台加载高模

业务指标:

  - 3D生成成功率
  - 平均处理时间
  - 队列积压量

技术指标:

  - GPU利用率和温度
  - 内存和显存使用
  - API响应时间分布

成本指标:

  - 单位推理成本
  - 云资源使用率
  - 数据传输费用