第 12 章 LLM 服务的生产化部署

# Ubuntu 22.04
curl -fsSL https://nvidia.github.io/libnvidia-container/gpgkey | \
  sudo gpg --dearmor -o /usr/share/keyrings/nvidia-container-toolkit-keyring.gpg
 
curl -s -L https://nvidia.github.io/libnvidia-container/stable/deb/nvidia-container-toolkit.list | \
  sed 's#deb https://#deb [signed-by=/usr/share/keyrings/nvidia-container-toolkit-keyring.gpg] https://#g' | \
  sudo tee /etc/apt/sources.list.d/nvidia-container-toolkit.list
 
sudo apt-get update && sudo apt-get install -y nvidia-container-toolkit
sudo nvidia-ctk runtime configure --runtime=docker
sudo systemctl restart docker

docker run --rm --gpus all nvidia/cuda:12.4.0-base-ubuntu22.04 nvidia-smi

FROM vllm/vllm-openai:v0.6.6.post1
 
# 安装额外依赖（如果有自定义逻辑）
COPY requirements.txt /app/requirements.txt
RUN pip install --no-cache-dir -r /app/requirements.txt
 
# 模型权重通常不打进镜像（太大了），通过 volume mount 挂载
# 这里只放配置文件
COPY model_config.json /app/model_config.json
 
ENV MODEL_NAME="Qwen/Qwen2.5-7B-Instruct"
ENV TENSOR_PARALLEL_SIZE=1
ENV MAX_MODEL_LEN=8192
ENV GPU_MEMORY_UTILIZATION=0.90
 
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1
 
EXPOSE 8000
 
ENTRYPOINT python -m vllm.entrypoints.openai.api_server \
  --model /models/${MODEL_NAME} \
  --tensor-parallel-size ${TENSOR_PARALLEL_SIZE} \
  --max-model-len ${MAX_MODEL_LEN} \
  --gpu-memory-utilization ${GPU_MEMORY_UTILIZATION} \
  --host 0.0.0.0 \
  --port 8000

docker run -d --gpus all \
  -v /data/models:/models \
  -p 8000:8000 \
  --name vllm-server \
  --shm-size=4g \
  my-vllm-server:latest

# 登录
docker login --username=<阿里云账号> registry.cn-hangzhou.aliyuncs.com
 
# 打 tag 并推送
docker tag my-vllm-server:latest \
  registry.cn-hangzhou.aliyuncs.com/<命名空间>/vllm-server:v1.0
docker push registry.cn-hangzhou.aliyuncs.com/<命名空间>/vllm-server:v1.0

docker login ccr.ccs.tencentyun.com --username=<腾讯云账号ID>
 
docker tag my-vllm-server:latest \
  ccr.ccs.tencentyun.com/<命名空间>/vllm-server:v1.0
docker push ccr.ccs.tencentyun.com/<命名空间>/vllm-server:v1.0

# 安装 NVIDIA device plugin（DaemonSet 方式）
kubectl apply -f https://raw.githubusercontent.com/NVIDIA/k8s-device-plugin/v0.16.1/deployments/static/nvidia-device-plugin.yml
 
# 验证 GPU 资源
kubectl describe node <gpu-node> | grep nvidia.com/gpu
# nvidia.com/gpu:  4
# nvidia.com/gpu:  4

resources:
  limits:
    nvidia.com/gpu: 1  # 请求 1 张 GPU
  requests:
    nvidia.com/gpu: 1

# 在 GPU 节点上启用 MIG
sudo nvidia-smi -i 0 -mig 1
 
# 创建 MIG 实例（以 A100 为例，切成 3 个 3g.40gb 实例）
sudo nvidia-smi mig -i 0 -cgi 9,9,9 -C

resources:
  limits:
    nvidia.com/mig-3g.40gb: 1

# ConfigMap for time-slicing
apiVersion: v1
kind: ConfigMap
metadata:
  name: nvidia-device-plugin
  namespace: kube-system
data:
  config: |
    version: v1
    sharing:
      timeSlicing:
        resources:
        - name: nvidia.com/gpu
          replicas: 4  # 每张 GPU 虚拟化为 4 份

# 使用 aliyun CLI 创建 GPU 节点池
aliyun cs POST /clusters/<cluster-id>/nodepools --body '{
  "nodepool_info": {
    "name": "gpu-pool-a10"
  },
  "scaling_group": {
    "instance_types": ["ecs.gn7i-c8g1.2xlarge"],  # A10, 1卡24GB
    "system_disk_category": "cloud_essd",
    "system_disk_size": 200,
    "data_disks": [{
      "category": "cloud_essd",
      "size": 500  # 模型权重需要大磁盘
    }],
    "desired_size": 2
  },
  "kubernetes_config": {
    "labels": [
      {"key": "gpu-type", "value": "a10"}
    ],
    "taints": [
      {"key": "nvidia.com/gpu", "value": "present", "effect": "NoSchedule"}
    ]
  }
}'

apiVersion: apps/v1
kind: Deployment
metadata:
  name: vllm-qwen-7b
  labels:
    app: vllm-qwen-7b
spec:
  replicas: 2
  selector:
    matchLabels:
      app: vllm-qwen-7b
  template:
    metadata:
      labels:
        app: vllm-qwen-7b
    spec:
      tolerations:
      - key: "nvidia.com/gpu"
        operator: "Exists"
        effect: "NoSchedule"
      nodeSelector:
        gpu-type: a10
      containers:
      - name: vllm
        image: registry.cn-hangzhou.aliyuncs.com/my-ns/vllm-server:v1.0
        ports:
        - containerPort: 8000
        env:
        - name: MODEL_NAME
          value: "Qwen/Qwen2.5-7B-Instruct"
        - name: TENSOR_PARALLEL_SIZE
          value: "1"
        - name: MAX_MODEL_LEN
          value: "8192"
        resources:
          limits:
            nvidia.com/gpu: 1
            memory: "32Gi"
            cpu: "8"
          requests:
            nvidia.com/gpu: 1
            memory: "24Gi"
            cpu: "4"
        volumeMounts:
        - name: model-cache
          mountPath: /models
        - name: shm
          mountPath: /dev/shm
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 120  # 模型加载需要时间
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 120
          periodSeconds: 10
        lifecycle:
          preStop:
            exec:
              # 收到 SIGTERM 后先停止接流量，sleep 5s 让 Service 把这个 Pod 摘出来
              command: ["/bin/sh", "-c", "sleep 5"]
      terminationGracePeriodSeconds: 120  # 留足时间给 streaming 请求收尾
      volumes:
      - name: model-cache
        hostPath:
          path: /data/models
          type: DirectoryOrCreate
      - name: shm
        emptyDir:
          medium: Memory
          sizeLimit: "4Gi"

apiVersion: v1
kind: Service
metadata:
  name: vllm-qwen-7b
spec:
  selector:
    app: vllm-qwen-7b  # 匹配 Deployment 的 Pod label
  ports:
  - name: http
    port: 8000         # Service 端口
    targetPort: 8000   # Pod 内的容器端口
  type: ClusterIP      # 集群内访问，默认值

#!/bin/bash
# init-model.sh - Pod 的 initContainer 脚本
MODEL_PATH="/models/${MODEL_NAME}"
OSS_PATH="oss://my-models/${MODEL_NAME}"
 
if [ -d "$MODEL_PATH" ] && [ -f "$MODEL_PATH/config.json" ]; then
    echo "Model already cached locally, skipping download"
else
    echo "Downloading model from OSS..."
    # ossutil 是阿里云 OSS 的官方命令行工具，支持并发上传/下载
    ossutil cp -r "$OSS_PATH" "$MODEL_PATH" --parallel 10
fi

initContainers:
- name: model-downloader
  image: registry.cn-hangzhou.aliyuncs.com/my-ns/ossutil:latest
  command: ["/bin/bash", "/scripts/init-model.sh"]
  env:
  - name: MODEL_NAME
    value: "Qwen2.5-7B-Instruct"
  volumeMounts:
  - name: model-cache
    mountPath: /models
  - name: scripts
    mountPath: /scripts

models/
├── Qwen2.5-7B-Instruct/
│   ├── v1.0_20250101/    # 基础版本
│   ├── v1.1_20250115/    # 微调版本
│   └── current -> v1.1_20250115  # 软链接指向当前版本

# v1 版本 - 9 个副本
apiVersion: apps/v1
kind: Deployment
metadata:
  name: vllm-qwen-v1
spec:
  replicas: 9
  selector:
    matchLabels:
      app: vllm-qwen
      version: v1
  template:
    metadata:
      labels:
        app: vllm-qwen   # 公共 label,供 Service 选中
        version: v1      # 区分版本的 label
    # ...
 
---
# v2 版本 - 1 个副本
apiVersion: apps/v1
kind: Deployment
metadata:
  name: vllm-qwen-v2
spec:
  replicas: 1
  selector:
    matchLabels:
      app: vllm-qwen
      version: v2
  template:
    metadata:
      labels:
        app: vllm-qwen
        version: v2
    # ...
 
---
# 共享同一个 Service,selector 只匹配 app,两个版本的 Pod 都会被选中
apiVersion: v1
kind: Service
metadata:
  name: vllm-qwen
spec:
  selector:
    app: vllm-qwen
  ports:
  - port: 8000

第一层：请求级限流（简单快速）
  - 每个 API Key 每分钟最多 60 个请求（RPM, Requests Per Minute）

第二层：Token 级限流（精确但有延迟）
  - 每个 API Key 每分钟最多 100K tokens（TPM, Tokens Per Minute）
  - 需要在请求完成后扣减，不能预扣（因为不知道 completion 会生成多少 token）

import redis
import time
 
r = redis.Redis()
 
def check_rate_limit(api_key: str, rpm_limit: int = 60) -> bool:
    """滑动窗口限流"""
    now = time.time()
    window_start = now - 60
    key = f"rate_limit:{api_key}"
 
    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)  # 清理过期记录
    pipe.zcard(key)                               # 当前窗口请求数
    pipe.zadd(key, {str(now): now})               # 添加当前请求
    pipe.expire(key, 120)                         # 设置过期时间
    _, count, _, _ = pipe.execute()
 
    if count >= rpm_limit:
        r.zrem(key, str(now))  # 超限，移除刚添加的
        return False
    return True

class LeastPendingRouter:
    def __init__(self, backends: list[str]):
        self.backends = backends
        self.pending = {b: 0 for b in backends}
        # 仅做演示。生产环境多协程并发读写 self.pending 有竞态:
        # 两个请求几乎同时调 route(),可能拿到同一个 backend,导致计数偏移。
        # 真实部署建议用以下任一方式:
        #   1) asyncio.Lock 保护 route/release(单进程内)
        #   2) Redis INCR/DECR(跨进程、跨实例)
        #   3) 直接读 vLLM 的 /metrics 里的 num_requests_running 做决策
 
    async def route(self) -> str:
        backend = min(self.pending, key=self.pending.get)
        self.pending[backend] += 1
        return backend
 
    async def release(self, backend: str):
        self.pending[backend] = max(0, self.pending[backend] - 1)

# 错误做法
timeout = aiohttp.ClientTimeout(total=30)  # 30 秒后整个请求超时
 
# 正确做法：分段超时
timeout = aiohttp.ClientTimeout(
    connect=5,        # 连接超时 5 秒
    sock_read=30,     # 单次读取超时 30 秒（两个 chunk 之间的间隔）
    total=300,        # 总超时 5 分钟
)

def classify_request(messages: list[dict]) -> str:
    """简单的请求分类"""
    last_msg = messages[-1]["content"].lower()
    total_tokens = sum(len(m["content"]) for m in messages) // 4  # 粗估 token 数
 
    # 长上下文或复杂任务 -> 大模型
    if total_tokens > 2000:
        return "qwen-72b"
    if any(kw in last_msg for kw in ["代码", "code", "分析", "推理", "debug"]):
        return "qwen-72b"
 
    # 简单任务 -> 小模型
    return "qwen-7b"

class ModelRouter:
    def __init__(self):
        self.routes = {
            "qwen-72b": {
                "primary": "http://vllm-72b:8000",
                "fallback": ["http://vllm-7b:8000"],  # 降级到小模型
            },
            "qwen-7b": {
                "primary": "http://vllm-7b:8000",
                "fallback": ["https://api.siliconflow.cn/v1"],  # 降级到云 API
            },
        }
 
    async def call(self, model: str, request: dict) -> dict:
        route = self.routes[model]
 
        # 先尝试主模型
        try:
            return await self._call_backend(route["primary"], request)
        except Exception as e:
            logger.warning(f"Primary backend failed: {e}")
 
        # 依次尝试 fallback
        for fallback in route["fallback"]:
            try:
                return await self._call_backend(fallback, request)
            except Exception:
                continue
 
        raise Exception("All backends failed")

async def cost_aware_route(self, request: dict) -> str:
    """优先用自建实例，满载时溢出到云 API"""
    self_hosted_pending = await self.get_pending_count("self-hosted")
    threshold = 50  # 队列深度阈值
 
    if self_hosted_pending < threshold:
        return "self-hosted"
    else:
        logger.info(f"Self-hosted queue depth {self_hosted_pending}, "
                    f"routing to cloud API")
        return "cloud-api"