AI推理基础设施与GPU集群优化完全教程

前言

随着大语言模型（LLM）的爆发式增长，如何高效、低成本地部署和运行AI推理服务已成为工程团队面临的核心挑战。一个70B参数的模型，如果部署不当，可能浪费90%的GPU算力；而经过系统优化后，同样的硬件可以支撑10倍以上的吞吐量。

本教程将从GPU选型、推理框架、核心优化技术、生产部署架构四个维度，系统讲解构建高性能AI推理平台的全部知识。

第一章：GPU选型与硬件对比

1.1 主流推理GPU对比

选择合适的GPU是推理优化的第一步。以下是最常见的推理GPU对比：

1.2 选型决策树

def select_gpu(model_size_b: float, budget: str, is_production: bool) -> str:
    """
    根据模型大小和预算推荐GPU
    
    Args:
        model_size_b: 模型参数量（单位：B，即十亿）
        budget: "low" / "medium" / "high" / "unlimited"
        is_production: 是否生产环境
    """
    # 估算模型在FP16下需要的显存（GB）
    # 简化公式：参数量(B) * 2字节 * 1.2（开销系数）
    fp16_memory = model_size_b * 2 * 1.2
    
    # INT8量化后显存需求约为FP16的一半
    int8_memory = fp16_memory / 2
    
    recommendations = []
    
    if model_size_b <= 7:
        # 7B模型，INT8约需8.4GB
        if budget == "low":
            recommendations.append("RTX 3090 (24GB) - 足够运行INT8/FP16，性价比最高")
            recommendations.append("RTX 4090 (24GB) - 更快的推理速度")
        elif budget == "medium":
            recommendations.append("L40S (48GB) - 可同时运行多个模型实例")
        else:
            recommendations.append("A100 40GB - 生产级稳定性")
    
    elif model_size_b <= 13:
        # 13B模型，FP16约需31GB
        if budget == "low":
            recommendations.append("RTX 4090 (24GB) - 必须INT8/INT4量化")
            recommendations.append("2x RTX 3090 - 张量并行部署")
        elif budget == "medium":
            recommendations.append("L40S (48GB) - FP16刚好放下")
            recommendations.append("A100 40GB - INT8轻松运行")
        else:
            recommendations.append("A100 80GB - FP16全精度，余量充足")
    
    elif model_size_b <= 34:
        # 34B模型，FP16约需82GB
        recommendations.append("A100 80GB (INT8) - 单卡INT8可运行")
        if budget in ("high", "unlimited"):
            recommendations.append("H100 80GB - 最佳性能")
            recommendations.append("2x A100 40GB - 张量并行FP16")
    
    elif model_size_b <= 70:
        # 70B模型
        if budget == "low":
            recommendations.append("2x RTX 4090 INT4 - 最低成本方案")
        elif budget == "medium":
            recommendations.append("2x A100 80GB INT8 - 生产级方案")
            recommendations.append("4x L40S INT8 - 成本优化方案")
        else:
            recommendations.append("4x H100 SXM - 最高性能")
            recommendations.append("4x A100 80GB - 成本性能平衡")
    
    else:
        # 超大模型 100B+
        recommendations.append(f"需要至少 {int(fp16_memory/80)+1} 张 A100/H100 80GB")
        if budget in ("high", "unlimited"):
            recommendations.append("8x H100 SXM (NVLINK) - 推荐方案")
    
    return recommendations


# 使用示例
for rec in select_gpu(model_size_b=70, budget="medium", is_production=True):
    print(f"  → {rec}")

1.3 显存需求精确计算

def estimate_vram(
    params_billions: float,
    precision: str = "fp16",
    seq_length: int = 4096,
    batch_size: int = 1,
    kv_cache_layers: int = 80,
    kv_cache_heads: int = 8,
    head_dim: int = 128,
) -> dict:
    """
    精确估算模型推理所需显存
    
    Returns:
        dict: 包含模型权重、KV Cache、激活值等各部分显存需求
    """
    bytes_per_param = {
        "fp32": 4, "fp16": 2, "bf16": 2,
        "int8": 1, "int4": 0.5, "fp8": 1,
    }
    
    # 1. 模型权重显存
    weight_bytes = params_billions * 1e9 * bytes_per_param[precision]
    weight_gb = weight_bytes / (1024**3)
    
    # 2. KV Cache显存
    # 每层每个token的KV Cache = 2(K+V) * num_heads * head_dim * bytes
    kv_per_token = 2 * kv_cache_heads * head_dim * bytes_per_param[precision]
    total_kv = kv_per_token * kv_cache_layers * seq_length * batch_size
    kv_gb = total_kv / (1024**3)
    
    # 3. 激活值和临时缓冲区（通常为权重的5-10%）
    activation_gb = weight_gb * 0.08
    
    # 4. CUDA运行时开销
    cuda_overhead_gb = 0.5
    
    total_gb = weight_gb + kv_gb + activation_gb + cuda_overhead_gb
    
    result = {
        "模型权重": f"{weight_gb:.2f} GB",
        "KV Cache": f"{kv_gb:.2f} GB",
        "激活值缓冲": f"{activation_gb:.2f} GB",
        "CUDA开销": f"{cuda_overhead_gb:.2f} GB",
        "总计": f"{total_gb:.2f} GB",
        "推荐GPU": recommend_gpu(total_gb),
    }
    
    for k, v in result.items():
        print(f"  {k}: {v}")
    
    return result


def recommend_gpu(gb_needed: float) -> str:
    gpus = [
        ("RTX 4090", 24), ("L40S", 48), ("A100 40GB", 40),
        ("A100 80GB", 80), ("H100", 80),
    ]
    for name, mem in gpus:
        if mem >= gb_needed * 1.1:  # 预留10%余量
            return f"推荐 {name} ({mem}GB)"
    return f"需要多卡并行，最低 {gb_needed:.0f}GB"


# 计算70B模型FP16推理显存
estimate_vram(params_billions=70, precision="fp16", batch_size=16)

第二章：推理框架深度对比

2.1 主流推理框架概览

当前最主流的三个推理框架各有侧重：

2.2 vLLM部署实战

vLLM是最受欢迎的推理框架，部署简单且性能优秀：

# 安装 vLLM
# pip install vllm

# === 方式1：Python API 直接调用 ===
from vllm import LLM, SamplingParams

# 初始化模型
llm = LLM(
    model="meta-llama/Llama-3-70B-Instruct",
    tensor_parallel_size=4,        # 4卡张量并行
    gpu_memory_utilization=0.90,   # GPU显存使用率
    max_model_len=8192,            # 最大上下文长度
    quantization="awq",            # 量化方式
    dtype="auto",
    trust_remote_code=True,
)

# 设置采样参数
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=1024,
    repetition_penalty=1.1,
)

# 批量推理
prompts = [
    "请解释什么是Transformer架构",
    "写一个Python快速排序算法",
    "解释量子计算的基本原理",
]

outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    prompt = output.prompt
    generated = output.outputs[0].text
    print(f"Prompt: {prompt[:50]}...")
    print(f"Generated: {generated[:200]}...")
    print(f"Tokens/s: {len(output.outputs[0].token_ids) / output.metrics.finished_time:.1f}")
    print("---")

# === 方式2：启动 OpenAI 兼容 API 服务 ===
# 命令行启动：
# python -m vllm.entrypoints.openai.api_server \
#     --model meta-llama/Llama-3-70B-Instruct \
#     --tensor-parallel-size 4 \
#     --gpu-memory-utilization 0.9 \
#     --max-model-len 8192 \
#     --quantization awq \
#     --port 8000 \
#     --served-model-name llama-70b

# Python客户端调用
from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="not-needed",  # vLLM默认不需要API key
)

# 非流式调用
response = client.chat.completions.create(
    model="llama-70b",
    messages=[
        {"role": "system", "content": "你是一个专业的AI助手"},
        {"role": "user", "content": "用Python实现一个简单的LRU缓存"},
    ],
    temperature=0.7,
    max_tokens=1024,
)
print(response.choices[0].message.content)

# 流式调用
stream = client.chat.completions.create(
    model="llama-70b",
    messages=[{"role": "user", "content": "讲一个笑话"}],
    stream=True,
)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

2.3 SGLang部署实战

SGLang在结构化生成和多轮对话场景下性能尤为突出：

# 安装 SGLang
# pip install "sglang[all]"

# === 方式1：启动API服务 ===
# python -m sglang.launch_server \
#     --model-path meta-llama/Llama-3-70B-Instruct \
#     --tp 4 \
#     --port 30000 \
#     --mem-fraction-static 0.9

# === 方式2：Python API ===
import sglang as sgl

@sgl.function
def multi_turn_chat(s, question1, question2):
    s += sgl.system("你是一个专业的AI助手")
    s += sgl.user(question1)
    s += sgl.assistant(sgl.gen("answer1", max_tokens=512))
    s += sgl.user(question2)
    s += sgl.assistant(sgl.gen("answer2", max_tokens=512))

# SGLang 的 RadixAttention 技术对多轮对话的KV Cache复用
# 可以将多轮对话的推理速度提升 2-5 倍
state = multi_turn_chat.run(
    question1="什么是深度学习？",
    question2="它和传统机器学习有什么区别？",
)
print(state["answer1"])
print(state["answer2"])

# === 方式3：批量结构化生成 ===
@sgl.function
def extract_info(s, text):
    s += sgl.user(f"从以下文本中提取人名和地点：\n{text}")
    s += sgl.assistant(
        "人名：" + sgl.gen("names", regex=r"[^\n]+", max_tokens=100) + "\n"
        "地点：" + sgl.gen("places", regex=r"[^\n]+", max_tokens=100)
    )

# 正则约束确保输出格式一致
texts = [
    "张三昨天去了北京出差",
    "李四在上海参加了会议",
]
states = extract_info.run_batch([{"text": t} for t in texts])
for state in states:
    print(f"人名: {state['names']}, 地点: {state['places']}")

2.4 TensorRT-LLM 部署实战

TensorRT-LLM 是 NVIDIA 官方的高性能推理引擎，性能最优但部署复杂度较高：

# TensorRT-LLM 部署流程（简化版）
# 1. 安装
# pip install tensorrt-llm -U --pre --extra-index-url https://pypi.nvidia.com

# 2. 模型转换（以 Llama 为例）
# python convert_checkpoint.py \
#     --model_dir ./Llama-3-8B-Instruct \
#     --output_dir ./trt_ckpt \
#     --dtype float16 \
#     --tp_size 1

# 3. 构建 TensorRT 引擎
# trtllm-build \
#     --checkpoint_dir ./trt_ckpt \
#     --output_dir ./trt_engine \
#     --gemm_plugin float16 \
#     --max_batch_size 64 \
#     --max_input_len 4096 \
#     --max_seq_len 8192

# 4. Python 推理
from tensorrt_llm.runtime import ModelRunner
from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("./Llama-3-8B-Instruct")

runner = ModelRunner(
    engine_dir="./trt_engine",
    rank=0,  # 当前GPU编号
)

# 编码输入
prompt = "用Python实现二分查找"
input_ids = tokenizer.encode(prompt, return_tensors="np")

# 推理
outputs = runner.generate(
    batch_input_ids=input_ids,
    max_new_tokens=512,
    temperature=0.7,
    top_k=50,
    top_p=0.9,
)

# 解码输出
output_text = tokenizer.decode(outputs[0][0])
print(output_text)

第三章：核心优化技术

3.1 PagedAttention 与 KV Cache 优化

PagedAttention 是 vLLM 提出的核心技术，它将 KV Cache 的管理从连续内存改为分页管理，类似于操作系统的虚拟内存机制：

"""
PagedAttention 原理演示

传统方式：每个请求预分配 max_seq_len 的连续内存 → 大量浪费
PagedAttention：按需分配固定大小的 Block → 内存利用率接近100%
"""

class SimplePagedKVCache:
    """简化的 PagedAttention KV Cache 管理器"""
    
    def __init__(self, num_blocks: int, block_size: int, num_heads: int, head_dim: int):
        self.block_size = block_size
        self.num_heads = num_heads
        self.head_dim = head_dim
        
        # 物理 Block 池（模拟GPU显存中的Block）
        # shape: [num_blocks, 2, block_size, num_heads, head_dim]
        # 2 表示 K 和 V
        self.kv_pool = np.zeros(
            (num_blocks, 2, block_size, num_heads, head_dim), 
            dtype=np.float16
        )
        
        # Block 管理
        self.free_blocks = list(range(num_blocks))  # 空闲Block列表
        self.block_tables = {}  # request_id -> [block_indices]
    
    def allocate(self, request_id: str, num_tokens: int) -> list:
        """为请求分配所需数量的Block"""
        num_blocks_needed = (num_tokens + self.block_size - 1) // self.block_size
        
        if len(self.free_blocks) < num_blocks_needed:
            raise MemoryError(f"显存不足，需要{num_blocks_needed}个Block，"
                            f"剩余{len(self.free_blocks)}个")
        
        allocated = []
        for _ in range(num_blocks_needed):
            block_idx = self.free_blocks.pop(0)
            allocated.append(block_idx)
        
        self.block_tables[request_id] = allocated
        return allocated
    
    def write(self, request_id: str, position: int, k: np.ndarray, v: np.ndarray):
        """将 KV 值写入指定位置"""
        blocks = self.block_tables[request_id]
        block_idx = position // self.block_size
        block_offset = position % self.block_size
        
        actual_block = blocks[block_idx]
        self.kv_pool[actual_block, 0, block_offset] = k  # Key
        self.kv_pool[actual_block, 1, block_offset] = v  # Value
    
    def free(self, request_id: str):
        """释放请求占用的所有Block"""
        blocks = self.block_tables.pop(request_id, [])
        self.free_blocks.extend(blocks)
    
    def get_stats(self) -> dict:
        """获取内存使用统计"""
        total = len(self.free_blocks) + sum(
            len(v) for v in self.block_tables.values()
        )
        used = total - len(self.free_blocks)
        return {
            "总Block数": total,
            "已使用": used,
            "空闲": len(self.free_blocks),
            "使用率": f"{used/total*100:.1f}%",
            "活跃请求数": len(self.block_tables),
        }


# 演示使用
import numpy as np

cache = SimplePagedKVCache(
    num_blocks=100,   # 100个物理Block
    block_size=16,     # 每个Block存16个token的KV
    num_heads=32,      # 32个注意力头
    head_dim=128,      # 每个头128维
)

# 模拟3个并发请求
cache.allocate("req_1", num_tokens=50)   # 需要4个Block
cache.allocate("req_2", num_tokens=100)  # 需要7个Block
cache.allocate("req_3", num_tokens=30)   # 需要2个Block

print("分配后:", cache.get_stats())

# 释放请求2
cache.free("req_2")
print("释放后:", cache.get_stats())

3.2 连续批处理（Continuous Batching）

连续批处理是提升GPU利用率的关键技术。传统的静态批处理中，所有请求必须同时开始、同时结束，短请求要等长请求完成，GPU利用率极低：

"""
连续批处理 vs 静态批处理 对比演示
"""

import time
from dataclasses import dataclass, field
from typing import List, Optional
from collections import deque


@dataclass
class Request:
    id: str
    prompt_tokens: int
    max_new_tokens: int
    generated_tokens: int = 0
    is_complete: bool = False
    
    @property
    def remaining_tokens(self) -> int:
        return self.max_new_tokens - self.generated_tokens


class StaticBatchScheduler:
    """静态批处理调度器 - 低效但简单"""
    
    def __init__(self, batch_size: int = 8):
        self.batch_size = batch_size
        self.queue = deque()
    
    def add_request(self, req: Request):
        self.queue.append(req)
    
    def process(self) -> dict:
        total_steps = 0
        processed = 0
        
        while self.queue:
            # 取一个批次
            batch = []
            for _ in range(min(self.batch_size, len(self.queue))):
                batch.append(self.queue.popleft())
            
            # 模拟处理：批次中所有请求一起处理，直到最长的完成
            max_tokens = max(r.max_new_tokens for r in batch)
            steps = max_tokens  # 必须等最长的完成
            total_steps += steps
            
            for req in batch:
                req.generated_tokens = req.max_new_tokens
                req.is_complete = True
                processed += 1
        
        return {"processed": processed, "total_steps": total_steps}


class ContinuousBatchScheduler:
    """连续批处理调度器 - 高效"""
    
    def __init__(self, batch_size: int = 8):
        self.batch_size = batch_size
        self.queue = deque()
        self.active: List[Request] = []
    
    def add_request(self, req: Request):
        self.queue.append(req)
    
    def process(self) -> dict:
        total_steps = 0
        processed = 0
        
        while self.queue or self.active:
            # 填充空位
            while self.active is not None and len(self.active) < self.batch_size and self.queue:
                self.active.append(self.queue.popleft())
            
            if not self.active:
                break
            
            # 每个请求生成一个token
            total_steps += 1
            completed = []
            for req in self.active:
                req.generated_tokens += 1
                if req.generated_tokens >= req.max_new_tokens:
                    req.is_complete = True
                    completed.append(req)
                    processed += 1
            
            # 立即移除完成的请求，腾出位置给新请求
            for req in completed:
                self.active.remove(req)
        
        return {"processed": processed, "total_steps": total_steps}


# 对比测试
requests = [
    Request(f"req_{i}", prompt_tokens=10, max_new_tokens=tokens)
    for i, tokens in enumerate([10, 50, 5, 100, 3, 20, 8, 60, 15, 40])
]

# 静态批处理
static = StaticBatchScheduler(batch_size=4)
for r in requests:
    static.add_request(Request(r.id, r.prompt_tokens, r.max_new_tokens))
static_result = static.process()

# 连续批处理
continuous = ContinuousBatchScheduler(batch_size=4)
for r in requests:
    continuous.add_request(Request(r.id, r.prompt_tokens, r.max_new_tokens))
cont_result = continuous.process()

print(f"静态批处理: 处理{static_result['processed']}个请求，总步数{static_result['total_steps']}")
print(f"连续批处理: 处理{cont_result['processed']}个请求，总步数{cont_result['total_steps']}")
print(f"效率提升: {static_result['total_steps'] / cont_result['total_steps']:.1f}x")

3.3 模型并行与张量并行

对于大模型，单卡放不下时需要多卡并行：

"""
张量并行（Tensor Parallelism）原理演示

将模型的权重矩阵按列或行切分到多张GPU上，
每张GPU只计算矩阵乘法的一部分，最后通过AllReduce汇总结果。
"""

import numpy as np

def demonstrate_tensor_parallelism():
    """演示张量并行的矩阵计算过程"""
    
    # 模拟一个大的线性层: Y = X @ W + bias
    # 输入: X [batch=2, hidden=8]
    # 权重: W [hidden=8, output=4]
    # 输出: Y [batch=2, output=4]
    
    X = np.random.randn(2, 8).astype(np.float32)
    W = np.random.randn(8, 4).astype(np.float32)
    bias = np.zeros(4, dtype=np.float32)
    
    # 原始单卡计算
    Y_original = X @ W + bias
    print("原始输出 shape:", Y_original.shape)
    
    # === 列并行（Column Parallel）===
    # 将 W 按列切分为两部分，每张GPU持有 W[:, :2] 和 W[:, 2:]
    # GPU 0: Y0 = X @ W[:, :2]
    # GPU 1: Y1 = X @ W[:, 2:]
    
    W_gpu0 = W[:, :2]  # [8, 2]
    W_gpu1 = W[:, 2:]  # [8, 2]
    
    Y0 = X @ W_gpu0  # [2, 2]
    Y1 = X @ W_gpu1  # [2, 2]
    
    # 拼接结果（在实际分布式中是AllGather操作）
    Y_column = np.concatenate([Y0, Y1], axis=1)
    print("列并行输出 shape:", Y_column.shape)
    print("列并行结果一致:", np.allclose(Y_original, Y_column))
    
    # === 行并行（Row Parallel）===
    # 将 W 按行切分，每张GPU持有 W[:4, :] 和 W[4:, :]
    # 输入 X 也需要相应切分
    # GPU 0: Y0 = X[:, :4] @ W[:4, :]
    # GPU 1: Y1 = X[:, 4:] @ W[4:, :]
    # 最终: Y = Y0 + Y1（AllReduce求和）
    
    W_gpu0 = W[:4, :]  # [4, 4]
    W_gpu1 = W[4:, :]  # [4, 4]
    
    Y0 = X[:, :4] @ W_gpu0  # [2, 4]
    Y1 = X[:, 4:] @ W_gpu1  # [2, 4]
    
    # AllReduce: 求和
    Y_row = Y0 + Y1
    print("行并行输出 shape:", Y_row.shape)
    print("行并行结果一致:", np.allclose(Y_original, Y_row))

demonstrate_tensor_parallelism()

# 使用 vLLM 进行张量并行部署（实际代码）
from vllm import LLM

# 单机4卡张量并行
llm = LLM(
    model="meta-llama/Llama-3-70B-Instruct",
    tensor_parallel_size=4,  # 使用4张GPU
    # vLLM 内部自动处理张量切分、通信等
)

# 多机张量并行（需要配置 Ray 集群）
# 在每个节点上运行:
# ray start --address=<head-node-ip>:6379
# 然后:
# llm = LLM(
#     model="meta-llama/Llama-3-70B-Instruct",
#     tensor_parallel_size=8,  # 跨2个节点，每节点4卡
#     distributed_executor_backend="ray",
# )

第四章：量化部署策略

4.1 量化技术对比

4.2 AWQ 量化实战

# === AWQ 量化（推荐的4bit量化方案）===

# 1. 安装
# pip install autoawq

from awq import AutoAWQForCausalLM
from transformers import AutoTokenizer

model_path = "meta-llama/Llama-3-8B-Instruct"
quant_path = "./Llama-3-8B-Instruct-AWQ"

# 加载模型
model = AutoAWQForCausalLM.from_pretrained(
    model_path,
    safetensors=True,
)
tokenizer = AutoTokenizer.from_pretrained(model_path)

# 量化配置
quant_config = {
    "zero_point": True,     # 使用零点量化
    "q_group_size": 128,    # 量化分组大小
    "w_bit": 4,             # 权重位数
    "version": "GEMM",      # GEMM kernel 更快
}

# 执行量化（需要校准数据集）
model.quantize(
    tokenizer,
    quant_config=quant_config,
    calib_data="pileval",   # 使用 Pile 验证集校准
)

# 保存量化模型
model.save_quantized(quant_path)
tokenizer.save_pretrained(quant_path)
print(f"量化模型已保存到 {quant_path}")

# === 使用量化模型推理 ===
from vllm import LLM, SamplingParams

llm = LLM(
    model=quant_path,
    quantization="awq",
    gpu_memory_utilization=0.9,
)

sampling_params = SamplingParams(temperature=0.7, max_tokens=512)
outputs = llm.generate(["解释什么是量子纠缠"], sampling_params)
print(outputs[0].outputs[0].text)

4.3 GPTQ 量化实战

# === GPTQ 量化 ===

# pip install auto-gptq optimum

from auto_gptq import AutoGPTQForCausalLM, BaseQuantizeConfig
from transformers import AutoTokenizer

model_path = "meta-llama/Llama-3-8B-Instruct"
quant_path = "./Llama-3-8B-Instruct-GPTQ-4bit"

# 量化配置
quantize_config = BaseQuantizeConfig(
    bits=4,              # 4-bit量化
    group_size=128,      # 量化分组
    damp_percent=0.01,   # 阻尼系数
    desc_act=True,       # 按激活值大小排序（更精确）
    sym=False,           # 非对称量化
)

# 加载模型和分词器
model = AutoGPTQForCausalLM.from_pretrained(model_path, quantize_config)
tokenizer = AutoTokenizer.from_pretrained(model_path)

# 准备校准数据
examples = [
    tokenizer("深度学习是人工智能的一个子领域", return_tensors="pt"),
    tokenizer("Python是最流行的编程语言之一", return_tensors="pt"),
    # 更多校准样本可以提高量化质量...
]

# 执行量化
model.quantize(examples)

# 保存
model.save_quantized(quant_path)
tokenizer.save_pretrained(quant_path)

4.4 FP8 量化（H100/H200专用）

# FP8 量化在 H100 上性能最优
# vLLM 直接支持 FP8 推理

from vllm import LLM

# 方法1：使用已有的FP8模型
llm = LLM(
    model="meta-llama/Llama-3-70B-Instruct",
    quantization="fp8",          # FP8量化
    tensor_parallel_size=4,
    gpu_memory_utilization=0.95,
)

# 方法2：在线动态FP8量化（不需要校准数据）
llm = LLM(
    model="meta-llama/Llama-3-70B-Instruct",
    quantization="fp8",
    dtype="auto",
    # vLLM会自动进行动态量化
)

第五章：负载均衡与自动扩缩容

5.1 推理服务架构设计

"""
AI推理服务的典型架构设计

客户端 → 负载均衡器 → 推理网关 → GPU推理实例（多个）
                                ↓
                          模型注册中心（管理模型版本和路由）
"""

# 推理网关示例代码
import asyncio
import time
import random
from dataclasses import dataclass, field
from typing import Dict, List
from collections import defaultdict


@dataclass
class InferenceInstance:
    """GPU推理实例"""
    id: str
    model: str
    host: str
    port: int
    max_concurrent: int = 64
    current_load: int = 0
    total_requests: int = 0
    avg_latency_ms: float = 0.0
    is_healthy: bool = True
    
    @property
    def load_ratio(self) -> float:
        return self.current_load / self.max_concurrent
    
    def update_latency(self, latency_ms: float):
        """指数移动平均更新延迟"""
        alpha = 0.1
        self.avg_latency_ms = alpha * latency_ms + (1 - alpha) * self.avg_latency_ms


class InferenceGateway:
    """推理网关 - 负责请求路由和负载均衡"""
    
    def __init__(self):
        self.instances: Dict[str, List[InferenceInstance]] = defaultdict(list)
        self.request_count = 0
    
    def register_instance(self, instance: InferenceInstance):
        self.instances[instance.model].append(instance)
        print(f"注册实例: {instance.id} (模型: {instance.model})")
    
    def select_instance(self, model: str, strategy: str = "least_load") -> InferenceInstance:
        """负载均衡策略选择实例"""
        available = [
            i for i in self.instances[model] 
            if i.is_healthy and i.current_load < i.max_concurrent
        ]
        
        if not available:
            raise RuntimeError(f"模型 {model} 没有可用实例")
        
        if strategy == "round_robin":
            # 轮询
            idx = self.request_count % len(available)
            return available[idx]
        
        elif strategy == "least_load":
            # 最少连接数
            return min(available, key=lambda i: i.current_load)
        
        elif strategy == "least_latency":
            # 最低延迟
            return min(available, key=lambda i: i.avg_latency_ms)
        
        elif strategy == "weighted":
            # 加权：综合考虑负载和延迟
            def score(inst):
                return inst.load_ratio * 0.6 + (inst.avg_latency_ms / 1000) * 0.4
            return min(available, key=score)
        
        return available[0]
    
    async def inference(self, model: str, prompt: str) -> dict:
        """处理推理请求"""
        self.request_count += 1
        instance = self.select_instance(model)
        
        instance.current_load += 1
        instance.total_requests += 1
        start = time.time()
        
        try:
            # 模拟推理调用
            await asyncio.sleep(random.uniform(0.1, 0.5))
            latency_ms = (time.time() - start) * 1000
            instance.update_latency(latency_ms)
            
            return {
                "instance": instance.id,
                "latency_ms": latency_ms,
                "result": f"Generated response for: {prompt[:50]}",
            }
        finally:
            instance.current_load -= 1


# 使用示例
async def main():
    gateway = InferenceGateway()
    
    # 注册多个推理实例
    for i in range(4):
        gateway.register_instance(InferenceInstance(
            id=f"gpu-{i}",
            model="llama-70b",
            host=f"10.0.1.{i+1}",
            port=8000,
            max_concurrent=64,
        ))
    
    # 模拟并发请求
    tasks = []
    for i in range(100):
        tasks.append(gateway.inference("llama-70b", f"请求 {i}"))
    
    results = await asyncio.gather(*tasks)
    
    # 统计
    latencies = [r["latency_ms"] for r in results]
    print(f"总请求: {len(results)}")
    print(f"平均延迟: {sum(latencies)/len(latencies):.1f}ms")
    print(f"P99延迟: {sorted(latencies)[int(len(latencies)*0.99)]:.1f}ms")

# asyncio.run(main())

5.2 自动扩缩容策略

"""
基于指标的自动扩缩容控制器
"""

import time
from dataclasses import dataclass
from typing import List


@dataclass
class GPUMetric:
    timestamp: float
    gpu_util: float          # GPU利用率 0-100
    memory_util: float       # 显存利用率 0-100
    queue_depth: int         # 等待队列长度
    avg_latency_ms: float    # 平均延迟
    requests_per_second: float


class AutoScaler:
    """GPU推理服务自动扩缩容"""
    
    def __init__(
        self,
        min_replicas: int = 1,
        max_replicas: int = 10,
        scale_up_threshold: float = 80.0,    # GPU利用率>80%扩容
        scale_down_threshold: float = 30.0,   # GPU利用率<30%缩容
        cooldown_seconds: int = 300,          # 冷却时间5分钟
        queue_threshold: int = 50,            # 队列长度>50扩容
        latency_threshold_ms: float = 2000,   # 延迟>2s扩容
    ):
        self.min_replicas = min_replicas
        self.max_replicas = max_replicas
        self.scale_up_threshold = scale_up_threshold
        self.scale_down_threshold = scale_down_threshold
        self.cooldown_seconds = cooldown_seconds
        self.queue_threshold = queue_threshold
        self.latency_threshold_ms = latency_threshold_ms
        
        self.current_replicas = min_replicas
        self.last_scale_time = 0
        self.metrics_history: List[GPUMetric] = []
    
    def add_metric(self, metric: GPUMetric):
        self.metrics_history.append(metric)
        # 只保留最近5分钟的指标
        cutoff = time.time() - 300
        self.metrics_history = [
            m for m in self.metrics_history if m.timestamp > cutoff
        ]
    
    def evaluate(self) -> dict:
        """评估是否需要扩缩容"""
        if not self.metrics_history:
            return {"action": "none", "reason": "无指标数据"}
        
        # 计算近期平均指标
        recent = self.metrics_history[-10:]  # 最近10个数据点
        avg_gpu = sum(m.gpu_util for m in recent) / len(recent)
        avg_queue = sum(m.queue_depth for m in recent) / len(recent)
        avg_latency = sum(m.avg_latency_ms for m in recent) / len(recent)
        
        # 检查冷却期
        in_cooldown = (time.time() - self.last_scale_time) < self.cooldown_seconds
        
        # 扩容判断
        should_scale_up = (
            avg_gpu > self.scale_up_threshold
            or avg_queue > self.queue_threshold
            or avg_latency > self.latency_threshold_ms
        )
        
        # 缩容判断
        should_scale_down = (
            avg_gpu < self.scale_down_threshold
            and avg_queue < 10
            and avg_latency < self.latency_threshold_ms * 0.3
        )
        
        if should_scale_up and not in_cooldown:
            new_replicas = min(self.current_replicas + 1, self.max_replicas)
            if new_replicas > self.current_replicas:
                self.current_replicas = new_replicas
                self.last_scale_time = time.time()
                return {
                    "action": "scale_up",
                    "replicas": self.current_replicas,
                    "reason": f"GPU利用率{avg_gpu:.0f}% 队列{avg_queue:.0f} 延迟{avg_latency:.0f}ms",
                }
        
        elif should_scale_down and not in_cooldown:
            new_replicas = max(self.current_replicas - 1, self.min_replicas)
            if new_replicas < self.current_replicas:
                self.current_replicas = new_replicas
                self.last_scale_time = time.time()
                return {
                    "action": "scale_down",
                    "replicas": self.current_replicas,
                    "reason": f"GPU利用率{avg_gpu:.0f}%，负载较低",
                }
        
        return {
            "action": "none",
            "replicas": self.current_replicas,
            "reason": f"GPU利用率{avg_gpu:.0f}%，状态稳定",
        }


# 模拟扩缩容过程
scaler = AutoScaler(min_replicas=2, max_replicas=8)

# 模拟负载上升
for i in range(20):
    metric = GPUMetric(
        timestamp=time.time() - (20 - i) * 10,
        gpu_util=40 + i * 3,  # 从40%逐渐上升到97%
        memory_util=50 + i * 2,
        queue_depth=max(0, i * 5 - 30),
        avg_latency_ms=500 + i * 100,
        requests_per_second=10 + i * 2,
    )
    scaler.add_metric(metric)
    result = scaler.evaluate()
    if result["action"] != "none":
        print(f"Step {i}: {result['action']} → {result['replicas']} replicas ({result['reason']})")

第六章：推理成本优化

6.1 成本分析模型

"""
AI推理成本分析与优化计算器
"""

class InferenceCostCalculator:
    """推理成本计算器"""
    
    # GPU每小时成本（云端，美元）
    GPU_COST_PER_HOUR = {
        "H100-SXM": 3.50,     # 云厂商价格
        "H100-PCIe": 2.80,
        "A100-80GB": 1.80,
        "A100-40GB": 1.20,
        "L40S": 0.90,
        "RTX-4090": 0.40,     # 自建折旧估算
        "RTX-3090": 0.25,
    }
    
    def __init__(self, gpu_type: str, num_gpus: int):
        self.gpu_type = gpu_type
        self.num_gpus = num_gpus
        self.hourly_cost = self.GPU_COST_PER_HOUR.get(gpu_type, 1.0) * num_gpus
    
    def estimate_monthly_cost(self, utilization: float = 0.7) -> float:
        """估算月度成本
        utilization: GPU利用率 (0-1)，云端通常按使用时间计费
        """
        hours_per_month = 24 * 30
        return self.hourly_cost * hours_per_month * utilization
    
    def cost_per_million_tokens(
        self,
        throughput_tokens_per_sec: float,
        utilization: float = 0.7,
    ) -> float:
        """计算每百万token的成本"""
        # 每秒成本
        cost_per_sec = self.hourly_cost / 3600
        
        # 每token成本
        cost_per_token = cost_per_sec / throughput_tokens_per_sec
        
        # 每百万token成本
        return cost_per_token * 1_000_000
    
    def compare_deployment(
        self,
        model_params_b: float,
        target_rps: float = 10,  # 目标每秒请求数
        avg_input_tokens: int = 500,
        avg_output_tokens: int = 200,
    ):
        """对比不同部署方案的成本"""
        
        print(f"\n{'='*70}")
        print(f"模型: {model_params_b}B 参数 | 目标: {target_rps} req/s")
        print(f"平均输入: {avg_input_tokens} tokens | 平均输出: {avg_output_tokens} tokens")
        print(f"{'='*70}")
        
        # 各方案的估算吞吐量（tokens/s/GPU）
        throughput_estimates = {
            "H100-SXM": {"fp16": 3000, "int8": 5000, "int4": 8000},
            "A100-80GB": {"fp16": 1500, "int8": 2500, "int4": 4000},
            "L40S": {"fp16": 800, "int8": 1500, "int4": 2500},
            "RTX-4090": {"fp16": 600, "int8": 1000, "int4": 1800},
        }
        
        print(f"\n{'GPU':<15} {'精度':<8} {'GPU数':<8} {'吞吐(tok/s)':<15} "
              f"{'月成本($)':<12} {'$/1M tokens':<12}")
        print("-" * 70)
        
        target_throughput = target_rps * (avg_input_tokens + avg_output_tokens)
        
        for gpu, throughputs in throughput_estimates.items():
            for precision, tps in throughputs.items():
                # 所需GPU数量（考虑70B模型在小显存卡上需要更多并行）
                base_gpus = max(1, int(model_params_b / 13))  # 粗略估算
                if precision == "int4":
                    base_gpus = max(1, base_gpus // 2)
                
                total_throughput = tps * base_gpus
                if total_throughput < target_throughput:
                    base_gpus = int(target_throughput / tps) + 1
                    total_throughput = tps * base_gpus
                
                calculator = InferenceCostCalculator(gpu, base_gpus)
                monthly = calculator.estimate_monthly_cost(0.7)
                cost_per_1m = calculator.cost_per_million_tokens(total_throughput / base_gpus, 0.7)
                
                print(f"{gpu:<15} {precision:<8} {base_gpus:<8} {total_throughput:<15} "
                      f"${monthly:<11,.0f} ${cost_per_1m:<11.2f}")


# 对比70B模型的部署成本
calc = InferenceCostCalculator("H100-SXM", 4)
calc.compare_deployment(model_params_b=70, target_rps=20)

6.2 多模型共享GPU

"""
多模型共享GPU方案

当多个小模型需要部署时，可以让它们共享GPU资源，
而不是每个模型独占一张卡。
"""

from vllm import LLM, SamplingParams

class MultiModelGPUShare:
    """在同一GPU上运行多个模型"""
    
    def __init__(self, gpu_memory_utilization: float = 0.9):
        self.models = {}
        self.gpu_util = gpu_memory_utilization
    
    def add_model(
        self,
        name: str,
        model_path: str,
        max_model_len: int = 4096,
        quantization: str = None,
        memory_fraction: float = None,
    ):
        """添加模型到共享GPU池"""
        llm = LLM(
            model=model_path,
            gpu_memory_utilization=self.gpu_util * (memory_fraction or 0.5),
            max_model_len=max_model_len,
            quantization=quantization,
            enforce_eager=True,  # 多模型共享时建议关闭CUDA Graph
        )
        self.models[name] = llm
        print(f"已加载模型: {name} ({model_path})")
    
    def generate(self, model_name: str, prompts: list, **kwargs) -> list:
        """使用指定模型生成"""
        if model_name not in self.models:
            raise ValueError(f"模型 {model_name} 未注册")
        
        sampling_params = SamplingParams(**kwargs)
        outputs = self.models[model_name].generate(prompts, sampling_params)
        return [o.outputs[0].text for o in outputs]


# 使用示例：在同一张A100上运行两个模型
# share = MultiModelGPUShare(gpu_memory_utilization=0.95)
# 
# # 模型1：代码生成模型（INT4量化，显存占用约10GB）
# share.add_model(
#     name="coder",
#     model_path="codellama/CodeLlama-34b-Instruct-hf",
#     quantization="awq",
#     memory_fraction=0.45,
# )
# 
# # 模型2：通用对话模型（INT4量化，显存占用约10GB）
# share.add_model(
#     name="chat",
#     model_path="meta-llama/Llama-3-8B-Instruct",
#     quantization="awq",
#     memory_fraction=0.45,
# )
# 
# # 按需路由到不同模型
# code_result = share.generate("coder", ["写一个快速排序"], temperature=0.2, max_tokens=512)
# chat_result = share.generate("chat", ["讲个笑话"], temperature=0.8, max_tokens=256)

第七章：云端 vs 自建集群

7.1 方案对比

7.2 成本对比计算

def compare_cloud_vs_selfhost(
    num_gpus: int = 8,
    gpu_type: str = "H100-SXM",
    monthly_utilization: float = 0.7,
    months: int = 24,
):
    """云端 vs 自建成本对比"""
    
    # 云端成本
    cloud_hourly = {
        "H100-SXM": 3.50, "A100-80GB": 1.80, "L40S": 0.90,
    }
    
    # 自建成本（美元）
    selfhost_gpu_cost = {
        "H100-SXM": 30000, "A100-80GB": 15000, "L40S": 7000,
    }
    
    server_cost_per_gpu = 3000  # 服务器配套成本（CPU、内存、网络等）
    monthly_power_per_gpu = 150  # 每张卡每月电费
    monthly_maintenance = 500    # 每月运维成本
    datacenter_monthly = 2000    # 机房租金/托管费
    
    hourly = cloud_hourly.get(gpu_type, 2.0)
    gpu_price = selfhost_gpu_cost.get(gpu_type, 15000)
    
    # 云端总成本
    cloud_monthly = hourly * num_gpus * 24 * 30 * monthly_utilization
    cloud_total = cloud_monthly * months
    
    # 自建总成本
    hardware = (gpu_price + server_cost_per_gpu) * num_gpus
    monthly_ops = (monthly_power_per_gpu * num_gpus + monthly_maintenance + datacenter_monthly)
    selfhost_total = hardware + monthly_ops * months
    
    print(f"\n{'='*60}")
    print(f"方案对比: {num_gpus}x {gpu_type} | {months}个月")
    print(f"{'='*60}")
    print(f"\n☁️  云端方案:")
    print(f"  月成本: ${cloud_monthly:,.0f}")
    print(f"  总成本: ${cloud_total:,.0f}")
    print(f"\n🏠 自建方案:")
    print(f"  硬件投入: ${hardware:,.0f}")
    print(f"  月运营成本: ${monthly_ops:,.0f}")
    print(f"  总成本: ${selfhost_total:,.0f}")
    print(f"\n📊 分析:")
    if selfhost_total < cloud_total:
        savings = cloud_total - selfhost_total
        print(f"  自建节省: ${savings:,.0f} ({savings/cloud_total*100:.1f}%)")
        # 回本周期
        breakeven = hardware / (cloud_monthly - monthly_ops)
        print(f"  回本周期: {breakeven:.1f}个月")
    else:
        extra = selfhost_total - cloud_total
        print(f"  云端节省: ${extra:,.0f}")
        print(f"  建议: 使用云端方案")


compare_cloud_vs_selfhost(num_gpus=8, gpu_type="H100-SXM", months=24)

第八章：生产部署最佳实践

8.1 完整部署脚本

#!/bin/bash
# deploy_vllm.sh - vLLM 生产环境部署脚本

MODEL_NAME="meta-llama/Llama-3-70B-Instruct"
MODEL_ALIAS="llama-70b"
TP_SIZE=4
PORT=8000
MAX_MODEL_LEN=8192
GPU_MEM_UTIL=0.90

# 启动 vLLM 推理服务
python -m vllm.entrypoints.openai.api_server \
    --model $MODEL_NAME \
    --served-model-name $MODEL_ALIAS \
    --tensor-parallel-size $TP_SIZE \
    --port $PORT \
    --max-model-len $MAX_MODEL_LEN \
    --gpu-memory-utilization $GPU_MEM_UTIL \
    --quantization awq \
    --dtype auto \
    --trust-remote-code \
    --disable-log-requests \
    --max-num-seqs 64 \
    --max-num-batched-tokens $MAX_MODEL_LEN \
    --swap-space 4 \
    --enforce-eager \
    2>&1 | tee /var/log/vllm/${MODEL_ALIAS}.log

8.2 健康检查与监控

"""
推理服务健康检查与监控
"""

import asyncio
import aiohttp
import time
from dataclasses import dataclass
from typing import List, Callable


@dataclass
class HealthCheckResult:
    instance_id: str
    is_healthy: bool
    latency_ms: float
    gpu_memory_used: float
    gpu_utilization: float
    queue_depth: int
    error: str = None


class InferenceMonitor:
    """推理服务监控"""
    
    def __init__(self, check_interval: int = 30):
        self.check_interval = check_interval
        self.instances: List[dict] = []
        self.alert_callbacks: List[Callable] = []
    
    def add_instance(self, instance_id: str, health_url: str):
        self.instances.append({
            "id": instance_id,
            "health_url": health_url,
            "consecutive_failures": 0,
        })
    
    def on_alert(self, callback: Callable):
        self.alert_callbacks.append(callback)
    
    async def check_health(self, instance: dict) -> HealthCheckResult:
        """检查单个实例健康状态"""
        try:
            start = time.time()
            async with aiohttp.ClientSession() as session:
                # vLLM 健康检查端点
                async with session.get(
                    instance["health_url"],
                    timeout=aiohttp.ClientTimeout(total=10),
                ) as resp:
                    latency = (time.time() - start) * 1000
                    
                    if resp.status == 200:
                        # 尝试获取GPU指标
                        metrics_url = instance["health_url"].replace("/health", "/metrics")
                        try:
                            async with session.get(metrics_url) as m:
                                # 解析Prometheus格式指标
                                text = await m.text()
                                gpu_util = self._parse_metric(text, "gpu_utilization")
                                gpu_mem = self._parse_metric(text, "gpu_memory_used_bytes")
                        except Exception:
                            gpu_util = 0
                            gpu_mem = 0
                        
                        instance["consecutive_failures"] = 0
                        return HealthCheckResult(
                            instance_id=instance["id"],
                            is_healthy=True,
                            latency_ms=latency,
                            gpu_memory_used=gpu_mem,
                            gpu_utilization=gpu_util,
                            queue_depth=0,
                        )
                    else:
                        raise Exception(f"HTTP {resp.status}")
        
        except Exception as e:
            instance["consecutive_failures"] += 1
            return HealthCheckResult(
                instance_id=instance["id"],
                is_healthy=False,
                latency_ms=0,
                gpu_memory_used=0,
                gpu_utilization=0,
                queue_depth=0,
                error=str(e),
            )
    
    def _parse_metric(self, text: str, metric_name: str) -> float:
        for line in text.split("\n"):
            if metric_name in line and not line.startswith("#"):
                parts = line.split()
                if len(parts) >= 2:
                    try:
                        return float(parts[-1])
                    except ValueError:
                        pass
        return 0.0
    
    async def run_checks(self):
        """执行所有实例的健康检查"""
        tasks = [self.check_health(inst) for inst in self.instances]
        results = await asyncio.gather(*tasks)
        
        for result in results:
            if not result.is_healthy:
                inst = next(i for i in self.instances if i["id"] == result.instance_id)
                if inst["consecutive_failures"] >= 3:
                    for callback in self.alert_callbacks:
                        await callback(f"⚠️ 实例 {result.instance_id} 连续失败 "
                                     f"{inst['consecutive_failures']}次: {result.error}")
            
            status = "✅" if result.is_healthy else "❌"
            print(f"{status} {result.instance_id}: "
                  f"延迟={result.latency_ms:.0f}ms "
                  f"GPU={result.gpu_utilization:.0f}%")


# 使用示例
# monitor = InferenceMonitor(check_interval=30)
# monitor.add_instance("gpu-0", "http://10.0.1.1:8000/health")
# monitor.add_instance("gpu-1", "http://10.0.1.2:8000/health")
# asyncio.run(monitor.run_checks())

总结

构建高效的AI推理基础设施需要从多个层面进行系统优化：

1. 硬件选型：根据模型大小、预算、性能需求选择合适的GPU
2. 推理框架：vLLM适合快速部署，SGLang适合结构化生成，TensorRT-LLM追求极致性能
3. 核心优化：PagedAttention、连续批处理、张量并行是必须掌握的三大技术
4. 量化部署：AWQ/INT4是性价比最高的量化方案，FP8是H100上的最优选择
5. 架构设计：负载均衡、自动扩缩容、多模型共享是生产环境必备能力
6. 成本优化：通过量化、批处理优化、合理的硬件选型可以将推理成本降低5-10倍

记住：没有银弹，只有适合场景的最优解。从你的具体需求出发，逐步优化每一个环节，才能构建出真正高效的AI推理平台。

本教程持续更新中，涵盖最新的推理优化技术和最佳实践。