第九章：生产部署与监控

原始权重 W ∈ [-α, α]
量化过程：W_q = round(W × s / α) 
反量化：W' = W_q × α / s

其中 s = 2^(b-1) - 1，b 为量化位数

前向传播：使用量化权重
反向传播：使用全精度梯度更新

伪量化操作：
W_fake_quant = dequant(quant(W))

损失函数：
L = α × L_CE(y_student, y_true) + 
    β × L_KL(σ(z_student/T), σ(z_teacher/T))

其中：

- L_CE：交叉熵损失（硬标签）
- L_KL：KL 散度（软标签）
- T：温度参数，控制分布平滑度
- α, β：损失权重

重要性评分：

- 权重幅度：|W|
- 梯度幅度：|∂L/∂W|
- Taylor 展开：ΔL ≈ |W × ∂L/∂W|

剪枝流程：

1. 训练基线模型
2. 计算重要性分数
3. 移除低分神经元/通道/层
4. 微调恢复性能

稀疏掩码：M ∈ {0,1}^d
稀疏权重：W_sparse = W ⊙ M

动态稀疏训练：

- 周期性更新掩码
- 保持固定稀疏度

标准注意力：O(N²) 内存
Flash Attention：O(N) 内存

通过分块计算和融合 kernel 实现：

- 减少 HBM 访问
- 提高 SRAM 利用率

策略：

1. 多查询注意力（MQA）：共享 K、V 投影
2. 分组查询注意力（GQA）：K、V 头数 < Q 头数
3. 滑动窗口：限制注意力范围

内存计算：
Cache_size = batch × seq_len × n_layers × 
             (n_kv_heads × d_head) × 2 × dtype_size

动态批处理：

- 连续批处理（Continuous Batching）
- 填充优化（Padding Optimization）
- 序列并行（Sequence Parallelism）

吞吐量优化：
Throughput = batch_size / latency
找到最优 batch_size 平衡延迟和吞吐

     ┌─────────────┐
     │   Gateway   │
     └──────┬──────┘
            │
    ┌───────┼───────┐
    │       │       │
┌───▼───┐ ┌▼────┐ ┌▼─────┐
│Router │ │Auth │ │Rate  │
│       │ │     │ │Limit │
└───┬───┘ └─────┘ └──────┘
    │
    ├──────────┬─────────┬──────────┐
    │          │         │          │
┌───▼───┐ ┌───▼──┐ ┌───▼───┐ ┌────▼────┐
│Model  │ │Model │ │Cache  │ │Monitor  │
│Server │ │Pool  │ │Service│ │Service  │
└───────┘ └──────┘ └───────┘ └─────────┘

# 模型处理器示例
class LLMHandler(BaseHandler):
    def initialize(self, context):
        self.model = load_model(context.model_dir)
        self.tokenizer = load_tokenizer()

    def preprocess(self, requests):
        # 批处理预处理
        texts = [req.get("text") for req in requests]
        return self.tokenizer(texts, return_tensors="pt")

    def inference(self, inputs):
        with torch.no_grad():
            outputs = self.model.generate(**inputs)
        return outputs

    def postprocess(self, outputs):
        # 解码和格式化
        return self.tokenizer.batch_decode(outputs)

模型配置：
name: "llm_model"
platform: "pytorch_libtorch"
max_batch_size: 32
input [
  {
    name: "input_ids"
    data_type: TYPE_INT64
    dims: [-1]
  }
]
output [
  {
    name: "logits"
    data_type: TYPE_FP32
    dims: [-1, -1]
  }
]
instance_group [
  {
    count: 2
    kind: KIND_GPU
  }
]
dynamic_batching {
  max_queue_delay_microseconds: 100
}

# SSE (Server-Sent Events) 实现
async def stream_generate(request):
    async def generate():
        for token in model.generate_stream(request.text):
            yield f"data: {json.dumps({'token': token})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

# WebSocket 实现
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            async for token in model.generate_stream(data):
                await websocket.send_json({"token": token})
    except WebSocketDisconnect:
        pass

# Kubernetes 部署配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-service
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    spec:
      containers:

      - name: model-server
        resources:
          requests:
            nvidia.com/gpu: 1
          limits:
            nvidia.com/gpu: 1
        livenessProbe:
          httpGet:
            path: /health
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /ready
          periodSeconds: 10

主备模式：
Primary ──┐
          ├──> Load Balancer ──> Client
Secondary ┘    (Health Check)

负载均衡算法：

- 轮询（Round Robin）
- 最少连接（Least Connections）
- 一致性哈希（Consistent Hashing）
- 基于延迟的路由

# Prometheus 指标定义
from prometheus_client import Counter, Histogram, Gauge

# 请求指标
request_count = Counter(
    'llm_requests_total',
    'Total requests',
    ['model', 'status']
)

request_latency = Histogram(
    'llm_request_duration_seconds',
    'Request latency',
    ['model', 'operation']
)

# 资源指标
gpu_utilization = Gauge(
    'llm_gpu_utilization_percent',
    'GPU utilization',
    ['device_id']
)

memory_usage = Gauge(
    'llm_memory_usage_bytes',
    'Memory usage',
    ['type']  # gpu_memory, cpu_memory
)

# 质量指标
output_quality_score = Histogram(
    'llm_output_quality_score',
    'Output quality score from feedback',
    ['task_type']
)

# Token 统计
token_usage = Counter(
    'llm_tokens_total',
    'Total tokens processed',
    ['type']  # input, output
)

# 成本指标
api_cost = Counter(
    'llm_api_cost_dollars',
    'API cost in dollars',
    ['model', 'customer']
)

import structlog

logger = structlog.get_logger()

# 请求日志
logger.info(
    "request_received",
    request_id=request_id,
    user_id=user_id,
    model_version=model_version,
    input_tokens=len(tokens),
    timestamp=time.time()
)

# 推理日志
logger.info(
    "inference_completed",
    request_id=request_id,
    latency_ms=latency * 1000,
    output_tokens=len(output_tokens),
    gpu_memory_mb=gpu_memory,
    cache_hit=cache_hit
)

# 错误日志
logger.error(
    "inference_failed",
    request_id=request_id,
    error_type=type(e).__name__,
    error_message=str(e),
    traceback=traceback.format_exc()
)

日志流水线：
Application ──> Filebeat ──> Logstash ──> Elasticsearch ──> Kibana
                              │
                              ├─> 解析和增强
                              ├─> 过滤和路由
                              └─> 告警触发

from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

tracer = trace.get_tracer(__name__)

@app.post("/generate")
async def generate(request: GenerateRequest):
    with tracer.start_as_current_span("generate_request") as span:
        span.set_attribute("user_id", request.user_id)
        span.set_attribute("model", request.model)

        with tracer.start_as_current_span("tokenize"):
            tokens = tokenizer.encode(request.text)

        with tracer.start_as_current_span("inference"):
            output = model.generate(tokens)

        with tracer.start_as_current_span("decode"):
            result = tokenizer.decode(output)

        return result

# AlertManager 配置
route:
  group_by: ['alertname', 'cluster']
  group_wait: 10s
  group_interval: 10s
  repeat_interval: 1h
  receiver: 'web.hook'
  routes:

  - match:
      severity: critical
    receiver: pagerduty
    continue: true

  - match:
      severity: warning
    receiver: slack

# Prometheus 告警规则
groups:

- name: llm_alerts
  rules:

  - alert: HighLatency
    expr: histogram_quantile(0.99, llm_request_duration_seconds) > 5
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "P99 延迟超过 5 秒"

  - alert: GPUMemoryLeak
    expr: rate(llm_gpu_memory_bytes[5m]) > 100000000
    for: 10m
    labels:
      severity: critical
    annotations:
      summary: "GPU 内存持续增长"

class AlertThrottler:
    def __init__(self, window_seconds=300, max_alerts=10):
        self.window = window_seconds
        self.max_alerts = max_alerts
        self.alert_times = defaultdict(deque)

    def should_alert(self, alert_key):
        now = time.time()
        times = self.alert_times[alert_key]

        # 清理过期时间
        while times and times[0] < now - self.window:
            times.popleft()

        # 检查是否超过阈值
        if len(times) >= self.max_alerts:
            return False

        times.append(now)
        return True

## 9.4 模型漂移检测

模型漂移是指模型在生产环境中性能随时间下降的现象，可能由数据分布变化、用户行为演变或外部环境改变引起。

### 9.4.1 漂移类型与检测方法

#### 数据漂移（Data Drift）
输入数据分布的变化：

```python
# KL 散度检测
def kl_divergence(p, q, epsilon=1e-10):
    """计算两个分布的 KL 散度"""
    p = p + epsilon
    q = q + epsilon
    return np.sum(p * np.log(p / q))

# Kolmogorov-Smirnov 检验
from scipy.stats import ks_2samp

def ks_test_drift(reference_data, current_data, threshold=0.05):
    """使用 KS 检验检测分布变化"""
    statistic, p_value = ks_2samp(reference_data, current_data)
    return p_value < threshold, statistic

# 特征分布监控
class FeatureDriftMonitor:
    def __init__(self, reference_window=1000):
        self.reference_window = reference_window
        self.feature_buffers = defaultdict(deque)
        self.reference_stats = {}

    def update(self, features):
        for name, value in features.items():
            buffer = self.feature_buffers[name]
            buffer.append(value)
            if len(buffer) > self.reference_window:
                buffer.popleft()

    def detect_drift(self, sensitivity=2.0):
        drifts = {}
        for name, buffer in self.feature_buffers.items():
            if len(buffer) < 100:
                continue

            current_mean = np.mean(buffer)
            current_std = np.std(buffer)

            if name in self.reference_stats:
                ref_mean, ref_std = self.reference_stats[name]
                z_score = abs(current_mean - ref_mean) / (ref_std + 1e-10)
                drifts[name] = z_score > sensitivity
            else:
                self.reference_stats[name] = (current_mean, current_std)
                drifts[name] = False

        return drifts

# 滑动窗口性能监控
class ConceptDriftDetector:
    def __init__(self, window_size=100, threshold=0.1):
        self.window_size = window_size
        self.threshold = threshold
        self.performance_window = deque(maxlen=window_size)
        self.baseline_performance = None

    def update(self, prediction, ground_truth):
        """更新性能指标"""
        correct = (prediction == ground_truth)
        self.performance_window.append(correct)

        if len(self.performance_window) == self.window_size:
            current_accuracy = np.mean(self.performance_window)

            if self.baseline_performance is None:
                self.baseline_performance = current_accuracy
                return False

            drift = abs(current_accuracy - self.baseline_performance)
            return drift > self.threshold

        return False

# ADWIN (Adaptive Windowing) 算法
class ADWIN:
    def __init__(self, delta=0.002):
        self.delta = delta
        self.window = []
        self.total = 0
        self.variance = 0
        self.width = 0

    def update(self, value):
        """检测概念漂移"""
        self.window.append(value)
        self.total += value
        self.width += 1

        if self.width > 1:
            mean = self.total / self.width
            self.variance += (value - mean) ** 2

            # 检测两个子窗口的显著差异
            if self._detect_change():
                # 丢弃旧数据
                self._shrink_window()
                return True

        return False

    def _detect_change(self):
        """使用 Hoeffding 界检测变化"""
        if self.width < 4:
            return False

        for split_point in range(1, self.width):
            n0 = split_point
            n1 = self.width - split_point

            sum0 = sum(self.window[:split_point])
            sum1 = sum(self.window[split_point:])

            mean0 = sum0 / n0
            mean1 = sum1 / n1

            epsilon = np.sqrt(
                (1/(2*n0) + 1/(2*n1)) * 
                np.log(2/self.delta)
            )

            if abs(mean0 - mean1) > epsilon:
                return True

        return False

class ConfidenceMonitor:
    def __init__(self, calibration_bins=10):
        self.calibration_bins = calibration_bins
        self.confidence_scores = []
        self.accuracies = []

    def update_batch(self, probabilities, predictions, labels):
        """更新置信度统计"""
        max_probs = np.max(probabilities, axis=1)
        correct = (predictions == labels)

        self.confidence_scores.extend(max_probs)
        self.accuracies.extend(correct)

    def compute_ece(self):
        """计算期望校准误差 (ECE)"""
        bin_boundaries = np.linspace(0, 1, self.calibration_bins + 1)
        bin_lowers = bin_boundaries[:-1]
        bin_uppers = bin_boundaries[1:]

        ece = 0
        for bin_lower, bin_upper in zip(bin_lowers, bin_uppers):
            in_bin = [
                i for i, conf in enumerate(self.confidence_scores)
                if bin_lower <= conf < bin_upper
            ]

            if len(in_bin) > 0:
                bin_acc = np.mean([self.accuracies[i] for i in in_bin])
                bin_conf = np.mean([self.confidence_scores[i] for i in in_bin])
                ece += len(in_bin) * abs(bin_acc - bin_conf)

        return ece / len(self.confidence_scores)

# 文本生成质量指标
class TextGenerationMonitor:
    def __init__(self):
        self.metrics_history = defaultdict(list)

    def compute_metrics(self, generated_text):
        """计算文本质量指标"""
        metrics = {}

        # 多样性指标
        tokens = generated_text.split()
        metrics['unique_tokens'] = len(set(tokens)) / len(tokens)

        # 重复度
        bigrams = zip(tokens[:-1], tokens[1:])
        metrics['unique_bigrams'] = len(set(bigrams)) / (len(tokens) - 1)

        # 长度分布
        metrics['avg_sentence_length'] = np.mean([
            len(sent.split()) 
            for sent in generated_text.split('.')
        ])

        # 困惑度代理指标
        metrics['entropy'] = self._compute_entropy(tokens)

        return metrics

    def detect_quality_degradation(self, metrics, window=100):
        """检测生成质量下降"""
        alerts = []

        for metric_name, value in metrics.items():
            history = self.metrics_history[metric_name]
            history.append(value)

            if len(history) > window:
                history.pop(0)

                # 计算移动平均和标准差
                mean = np.mean(history[:-10])
                std = np.std(history[:-10])
                recent_mean = np.mean(history[-10:])

                # Z-score 检验
                if abs(recent_mean - mean) > 2 * std:
                    alerts.append({
                        'metric': metric_name,
                        'baseline': mean,
                        'current': recent_mean,
                        'severity': 'high' if abs(recent_mean - mean) > 3 * std else 'medium'
                    })

        return alerts

class DriftResponseOrchestrator:
    def __init__(self):
        self.drift_detectors = {}
        self.response_strategies = {}
        self.alert_manager = AlertManager()

    def register_detector(self, name, detector, strategy):
        """注册漂移检测器和响应策略"""
        self.drift_detectors[name] = detector
        self.response_strategies[name] = strategy

    def monitor_and_respond(self, data):
        """监控并自动响应"""
        for name, detector in self.drift_detectors.items():
            if detector.detect(data):
                self._handle_drift(name, data)

    def _handle_drift(self, detector_name, data):
        """处理检测到的漂移"""
        strategy = self.response_strategies[detector_name]

        if strategy == 'alert':
            self.alert_manager.send_alert(
                f"Drift detected by {detector_name}",
                severity='warning'
            )

        elif strategy == 'retrain':
            self._trigger_retraining(data)

        elif strategy == 'fallback':
            self._switch_to_fallback_model()

        elif strategy == 'recalibrate':
            self._recalibrate_model(data)

    def _trigger_retraining(self, data):
        """触发模型重训练"""
        # 收集新数据
        # 启动训练任务
        # 验证新模型
        pass

    def _recalibrate_model(self, data):
        """重新校准模型输出"""
        # 使用温度缩放
        # 更新后处理参数
        pass

# Kubernetes 蓝绿部署配置
apiVersion: v1
kind: Service
metadata:
  name: llm-service
spec:
  selector:
    app: llm
    version: green  # 切换到 blue 或 green
  ports:

  - port: 80
    targetPort: 8080

---
# Blue 部署
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-blue
spec:
  replicas: 3
  selector:
    matchLabels:
      app: llm
      version: blue
  template:
    metadata:
      labels:
        app: llm
        version: blue
    spec:
      containers:

      - name: model-server
        image: llm:v1.0

---
# Green 部署
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-green
spec:
  replicas: 3
  selector:
    matchLabels:
      app: llm
      version: green
  template:
    metadata:
      labels:
        app: llm
        version: green
    spec:
      containers:

      - name: model-server
        image: llm:v2.0

class CanaryDeployment:
    def __init__(self, initial_traffic_percent=5):
        self.canary_percent = initial_traffic_percent
        self.metrics_collector = MetricsCollector()
        self.decision_threshold = 0.95  # 成功率阈值

    def route_request(self, request):
        """路由请求到金丝雀或稳定版本"""
        if random.random() < self.canary_percent / 100:
            response = self.canary_model.predict(request)
            self.metrics_collector.record('canary', response)
            return response
        else:
            response = self.stable_model.predict(request)
            self.metrics_collector.record('stable', response)
            return response

    def progressive_rollout(self):
        """渐进式发布"""
        stages = [5, 10, 25, 50, 100]  # 流量百分比阶段

        for target_percent in stages:
            self.canary_percent = target_percent
            time.sleep(300)  # 每阶段观察 5 分钟

            if not self._check_health():
                self._rollback()
                return False

        return True

    def _check_health(self):
        """检查金丝雀版本健康状态"""
        canary_metrics = self.metrics_collector.get_metrics('canary')
        stable_metrics = self.metrics_collector.get_metrics('stable')

        # 比较关键指标
        canary_success_rate = canary_metrics['success_rate']
        stable_success_rate = stable_metrics['success_rate']

        if canary_success_rate < self.decision_threshold:
            return False

        # 统计显著性检验
        p_value = self._statistical_test(
            canary_metrics['latencies'],
            stable_metrics['latencies']
        )

        return p_value > 0.05  # 无显著差异

    def _rollback(self):
        """回滚到稳定版本"""
        self.canary_percent = 0
        logger.warning("Canary rollback triggered")

class FeatureFlags:
    def __init__(self, config_source='redis'):
        self.flags = {}
        self.config_source = config_source
        self._load_flags()

    def _load_flags(self):
        """从配置源加载特征开关"""
        if self.config_source == 'redis':
            self.flags = {
                'new_tokenizer': {'enabled': True, 'rollout': 100},
                'attention_optimization': {'enabled': True, 'rollout': 50},
                'experimental_sampler': {'enabled': False, 'rollout': 0}
            }

    def is_enabled(self, flag_name, user_id=None):
        """检查特征是否启用"""
        if flag_name not in self.flags:
            return False

        flag = self.flags[flag_name]
        if not flag['enabled']:
            return False

        if user_id:
            # 基于用户 ID 的一致性哈希
            hash_value = hash(f"{flag_name}:{user_id}") % 100
            return hash_value < flag['rollout']

        return flag['rollout'] == 100

    def with_feature(self, flag_name):
        """装饰器模式"""
        def decorator(func):
            def wrapper(*args, **kwargs):
                if self.is_enabled(flag_name):
                    return func(*args, **kwargs)
                else:
                    # 返回默认行为
                    return self._default_behavior(flag_name)
            return wrapper
        return decorator

class RollbackManager:
    def __init__(self):
        self.deployment_history = []
        self.health_checker = HealthChecker()
        self.max_rollback_window = 3600  # 1 小时

    def deploy(self, version, config):
        """部署新版本"""
        deployment = {
            'version': version,
            'config': config,
            'timestamp': time.time(),
            'status': 'deploying'
        }

        self.deployment_history.append(deployment)

        try:
            # 执行部署
            self._execute_deployment(version, config)

            # 健康检查
            if self._post_deployment_check(version):
                deployment['status'] = 'healthy'
                return True
            else:
                deployment['status'] = 'unhealthy'
                self.automatic_rollback()
                return False

        except Exception as e:
            deployment['status'] = 'failed'
            deployment['error'] = str(e)
            self.automatic_rollback()
            raise

    def automatic_rollback(self):
        """自动回滚到上一个健康版本"""
        for deployment in reversed(self.deployment_history[:-1]):
            if deployment['status'] == 'healthy':
                logger.info(f"Rolling back to version {deployment['version']}")
                self._execute_deployment(
                    deployment['version'],
                    deployment['config']
                )
                return True

        logger.error("No healthy version found for rollback")
        return False

    def _post_deployment_check(self, version):
        """部署后健康检查"""
        checks = [
            self.health_checker.check_endpoint_health(),
            self.health_checker.check_model_loading(),
            self.health_checker.check_inference_latency(),
            self.health_checker.check_error_rate()
        ]

        return all(checks)

    def _execute_deployment(self, version, config):
        """执行实际部署操作"""
        # 更新模型文件
        # 重启服务
        # 更新路由配置
        pass