附录B 关键性能指标公式

附录B 关键性能指标公式

你无法优化你无法衡量的东西。AI Infra 工程师的核心能力之一是精确地量化系统性能。

本附录给出 AI Infra 最关键的性能指标定义、公式和 Python 计算代码。每个指标都配有实际数值示例。

B.1 训练效率指标

B.1.1 MFU（Model FLOPS Utilization）

定义：实际训练吞吐量与硬件理论峰值的比值。这是衡量训练效率最重要的指标。

import torch

def compute_mfu(
    model_params: int,
    batch_size: int,
    seq_len: int,
    step_time: float,  # 秒
    num_gpus: int,
    gpu_flops: float,  # 单 GPU 理论 FLOPS
    forward_only: bool = False,
):
    """
    计算 MFU（Model FLOPS Utilization）
    
    公式：
    MFU = 实际 FLOPS / 理论 FLOPS
    
    其中：
    - 实际 FLOPS = 训练一步的总 FLOP / step_time
    - 训练一步的总 FLOP ≈ 6 * N * B * S （前向+反向）
    - N = 参数量, B = batch size, S = 序列长度
    """
    # 训练一步的计算量（3x 参数量 × tokens，前向 1x + 反向 2x）
    factor = 2 if forward_only else 6  # 推理 2x，训练 6x
    total_flop = factor * model_params * batch_size * seq_len
    
    actual_flops = total_flop / step_time
    theoretical_flops = num_gpus * gpu_flops
    
    mfu = actual_flops / theoretical_flops
    
    print(f"=== MFU 计算 ===")
    print(f"参数量: {model_params / 1e9:.1f}B")
    print(f"Batch: {batch_size}, Seq: {seq_len}")
    print(f"Step time: {step_time:.3f}s")
    print(f"总 FLOP: {total_flop / 1e12:.1f} TFLOP")
    print(f"实际: {actual_flops / 1e12:.1f} TFLOPS")
    print(f"理论: {theoretical_flops / 1e12:.1f} TFLOPS")
    print(f"MFU: {mfu * 100:.1f}%")
    
    return mfu


# 示例：7B 模型在 8x A100 上训练
compute_mfu(
    model_params=7e9,
    batch_size=8 * 8,  # 8 GPU × micro-batch 8
    seq_len=4096,
    step_time=2.5,  # 秒
    num_gpus=8,
    gpu_flops=312e12,  # A100 BF16: 312 TFLOPS
)
# 输出: MFU ≈ 44.5%（对于 7B Dense 模型这是合理的）

B.1.2 HFU（Hardware FLOPS Utilization）

定义：与 MFU 类似，但考虑了重计算（gradient checkpointing）的额外 FLOP。

def compute_hfu(
    model_params: int,
    batch_size: int,
    seq_len: int,
    step_time: float,
    num_gpus: int,
    gpu_flops: float,
    recomputation_factor: float = 1.0,  # 无重计算=1.0, 全重计算≈1.33
):
    """
    HFU 考虑了重计算的额外计算开销。
    
    公式：
    HFU = (6 * N * B * S * (1 + α)) / (step_time * num_gpus * gpu_flops)
    
    其中 α 是重计算因子：
    - 无重计算：α = 0
    - 部分重计算：α ≈ 0.15-0.33
    - 全重计算：α ≈ 0.33
    """
    total_flop = 6 * model_params * batch_size * seq_len * (1 + recomputation_factor)
    actual_flops = total_flop / step_time
    hfu = actual_flops / (num_gpus * gpu_flops)
    
    print(f"HFU: {hfu * 100:.1f}%（重计算因子: {recomputation_factor}）")
    return hfu


# 对比 MFU 和 HFU
print("=== 无重计算 ===")
compute_hfu(7e9, 64, 4096, 2.5, 8, 312e12, 0.0)

print("\n=== 全重计算 ===")
compute_hfu(7e9, 64, 4096, 2.5, 8, 312e12, 0.33)

B.1.3 训练吞吐量（Tokens/sec）

def compute_training_throughput(
    batch_size: int,
    seq_len: int,
    step_time: float,
    gradient_accumulation_steps: int = 1,
):
    """
    训练吞吐量：每秒处理的 token 数
    
    公式：
    Throughput = (batch_size × seq_len × grad_accum_steps) / step_time
    """
    tokens_per_step = batch_size * seq_len * gradient_accumulation_steps
    throughput = tokens_per_step / step_time
    
    print(f"Tokens/step: {tokens_per_step:,}")
    print(f"Throughput: {throughput:,.0f} tokens/sec")
    print(f"Throughput: {throughput / 1e6:.2f}M tokens/sec")
    return throughput


# 示例
compute_training_throughput(
    batch_size=64,  # 全局 batch
    seq_len=4096,
    step_time=2.5,
)

B.1.4 Wall-clock Training Time

def estimate_wall_clock_time(
    total_tokens: int,       # 训练总 token 数
    throughput: float,       # tokens/sec
    overhead_factor: float = 1.1,  # 额外开销（10%）
):
    """
    估算训练总时间
    
    公式：
    Wall-clock = total_tokens × overhead / throughput
    """
    total_seconds = total_tokens * overhead_factor / throughput
    days = total_seconds / 86400
    
    print(f"总 token 数: {total_tokens / 1e9:.0f}B")
    print(f"吞吐量: {throughput / 1e6:.2f}M tokens/sec")
    print(f"预估时间: {days:.1f} 天 ({total_seconds / 3600:.0f} 小时)")
    return days


# 示例：训练一个 1T token 的模型
estimate_wall_clock_time(
    total_tokens=1e12,       # 1T tokens
    throughput=200_000,      # 200K tokens/sec
)
# 输出: 约 63.4 天

B.2 推理性能指标

B.2.1 TTFT（Time To First Token）

import time
from dataclasses import dataclass

@dataclass
class InferenceMetrics:
    ttft: float       # 首 token 延迟（秒）
    tpot: float       # 每输出 token 延迟（秒）
    total_tokens: int
    input_tokens: int


def measure_inference(llm, prompt, max_tokens=256):
    """测量推理性能"""
    # 模拟流式推理
    start = time.time()
    first_token_time = None
    
    tokens = []
    for token in llm.generate_stream(prompt, max_tokens=max_tokens):
        if first_token_time is None:
            first_token_time = time.time()
        tokens.append(token)
    
    end = time.time()
    
    metrics = InferenceMetrics(
        ttft=first_token_time - start,
        tpot=(end - first_token_time) / (len(tokens) - 1) if len(tokens) > 1 else 0,
        total_tokens=len(tokens),
        input_tokens=len(prompt.split()),
    )
    
    print(f"=== 推理性能 ===")
    print(f"TTFT: {metrics.ttot * 1000:.1f}ms")
    print(f"TPOT: {metrics.tpot * 1000:.1f}ms")
    print(f"总延迟: {(end - start) * 1000:.1f}ms")
    print(f"输出 tokens: {metrics.total_tokens}")
    return metrics


# 目标值参考
ttft_targets = {
    "实时对话": 0.2,     # 200ms
    "交互式应用": 0.5,   # 500ms
    "后台任务": 5.0,     # 5s
    "批处理": 30.0,      # 30s
}

B.2.2 TPOT（Time Per Output Token）

def compute_tpot(total_output_time, num_output_tokens):
    """
    TPOT = 总输出时间 / 输出 token 数
    
    注意：总输出时间 = 总时间 - TTFT
    """
    tpot = total_output_time / max(num_output_tokens - 1, 1)
    
    print(f"TPOT: {tpot * 1000:.1f}ms/token")
    print(f"生成速度: {1 / tpot:.1f} tokens/sec")
    return tpot


# TPOT 目标值
tpot_targets = {
    "人类阅读速度": 0.015,    # ~65 tokens/sec
    "实时交互": 0.05,         # 20 tokens/sec
    "可接受延迟": 0.1,        # 10 tokens/sec
}

B.2.3 推理吞吐量（Throughput）

def compute_inference_throughput(
    num_requests: int,
    avg_output_tokens: int,
    total_time: float,
):
    """
    推理吞吐量：每秒生成的 token 总数
    
    公式：
    Throughput = (num_requests × avg_output_tokens) / total_time
    """
    total_tokens = num_requests * avg_output_tokens
    throughput = total_tokens / total_time
    
    print(f"=== 推理吞吐量 ===")
    print(f"请求数: {num_requests}")
    print(f"总输出 tokens: {total_tokens:,}")
    print(f"总时间: {total_time:.1f}s")
    print(f"吞吐量: {throughput:,.0f} tokens/sec")
    return throughput


# 对比不同 batch size 下的吞吐量
print("--- Batch Size = 1 ---")
compute_inference_throughput(1, 256, 12.8)

print("\n--- Batch Size = 32 ---")
compute_inference_throughput(32, 256, 30.5)

B.2.4 Goodput

定义：满足 SLO 要求的请求吞吐量。这是真正有业务意义的指标。

def compute_goodput(
    requests: list,       # 每个请求的延迟记录
    slo_ttft: float,     # TTFT SLO（秒）
    slo_e2e: float,      # 端到端延迟 SLO（秒）
    total_time: float,
):
    """
    Goodput = 满足 SLO 的请求数 / 总时间
    
    只有同时满足 TTFT 和 端到端延迟 的请求才计入。
    """
    good_requests = [
        r for r in requests
        if r['ttft'] <= slo_ttft and r['e2e_latency'] <= slo_e2e
    ]
    
    goodput = len(good_requests) / total_time
    total_throughput = len(requests) / total_time
    slo_compliance = len(good_requests) / len(requests)
    
    print(f"=== Goodput ===")
    print(f"总请求数: {len(requests)}")
    print(f"满足 SLO: {len(good_requests)} ({slo_compliance * 100:.1f}%)")
    print(f"总吞吐量: {total_throughput:.1f} req/s")
    print(f"Goodput: {goodput:.1f} req/s")
    print(f"SLO 目标: TTFT < {slo_ttft*1000:.0f}ms, E2E < {slo_e2e:.1f}s")
    return goodput

B.2.5 端到端延迟分解

def breakdown_inference_latency(
    model_params: int,
    input_tokens: int,
    output_tokens: int,
    batch_size: int,
    gpu_memory_bandwidth: float,  # bytes/sec
    gpu_flops: float,
    num_gpus: int,
):
    """
    端到端延迟的近似分解：
    Total = Prefill 时间 + Decode 时间 + 通信开销
    
    Prefill（预填充）：处理输入 prompt，计算密集型
    Decode（解码）：逐 token 生成，内存密集型
    """
    # 简化估算
    bytes_per_param = 2  # FP16
    
    # Prefill: ~6 * N * input_tokens / (num_gpus * gpu_flops)
    prefill_flops = 2 * model_params * input_tokens  # 前向 only
    prefill_time = prefill_flops / (num_gpus * gpu_flops)
    
    # Decode: 逐 token，主要受内存带宽限制
    # 每步需要读取全部权重 + KV Cache
    weight_bytes = model_params * bytes_per_param
    decode_per_token = weight_bytes / (num_gpus * gpu_memory_bandwidth)
    decode_time = decode_per_token * output_tokens
    
    # KV Cache 读写
    kv_cache_per_token = 2 * 32 * 32 * 128 * 2  # layers × heads × head_dim × fp16
    kv_total = kv_cache_per_token * (input_tokens + output_tokens) * batch_size
    kv_time = kv_total / (num_gpus * gpu_memory_bandwidth) * output_tokens
    
    total = prefill_time + decode_time + kv_time
    
    print(f"=== 延迟分解（近似）===")
    print(f"Prefill: {prefill_time * 1000:.1f}ms")
    print(f"Decode ({output_tokens} tokens): {decode_time * 1000:.1f}ms")
    print(f"  - 每步: {decode_per_token * 1000:.1f}ms")
    print(f"KV Cache 访问: {kv_time * 1000:.1f}ms")
    print(f"总计: {total * 1000:.1f}ms")
    return total


# 示例：7B 模型在单 A100 上
breakdown_inference_latency(
    model_params=7e9,
    input_tokens=512,
    output_tokens=256,
    batch_size=1,
    gpu_memory_bandwidth=2e12,  # A100: 2TB/s
    gpu_flops=312e12,
    num_gpus=1,
)

B.3 集群利用率指标

B.3.1 GPU 利用率

def compute_gpu_utilization(
    busy_seconds: float,  # GPU 执行 kernel 的时间
    total_seconds: float,  # 总观察时间
):
    """
    GPU 利用率 = GPU 执行 kernel 的时间 / 总时间
    
    注意：nvidia-smi 显示的 utilization 反映的是"有 kernel 在执行"的
    时间百分比，而不是 FLOPS 利用率。GPU 利用率 100% 不等于 MFU 100%。
    """
    utilization = busy_seconds / total_seconds
    print(f"GPU 利用率: {utilization * 100:.1f}%")
    print(f"⚠️ 注意：GPU 利用率 ≠ MFU")
    print(f"   GPU 利用率 100% 可能 MFU 只有 40%")
    return utilization


# 从 nvidia-smi 获取
# nvidia-smi --query-gpu=utilization.gpu,memory.used --format=csv -l 1

B.3.2 集群空闲率与排队延迟

def compute_cluster_metrics(
    gpu_hours_total: float,    # 总 GPU 小时
    gpu_hours_used: float,     # 实际使用 GPU 小时
    gpu_hours_idle: float,     # 空闲 GPU 小时
    gpu_hours_queue: float,    # 排队等待 GPU 小时
):
    """
    集群利用率指标
    """
    utilization = gpu_hours_used / gpu_hours_total
    idle_rate = gpu_hours_idle / gpu_hours_total
    queue_rate = gpu_hours_queue / gpu_hours_total
    wasted_rate = idle_rate  # 空闲 = 浪费
    
    print(f"=== 集群利用率 ===")
    print(f"总 GPU 小时: {gpu_hours_total:,.0f}")
    print(f"使用率: {utilization * 100:.1f}%")
    print(f"空闲率: {idle_rate * 100:.1f}%")
    print(f"排队占比: {queue_rate * 100:.1f}%")
    
    # 健康度评估
    if utilization > 0.85:
        print("✅ 使用率高，考虑扩容")
    elif utilization > 0.6:
        print("✅ 使用率健康")
    elif utilization > 0.3:
        print("⚠️ 使用率偏低，考虑优化调度")
    else:
        print("❌ 使用率过低，资源浪费")
    
    return utilization


# 月度集群统计
compute_cluster_metrics(
    gpu_hours_total=8 * 24 * 30,  # 8 GPU × 30 天
    gpu_hours_used=3850,
    gpu_hours_idle=1250,
    gpu_hours_queue=240,
)

B.4 成本指标

B.4.1 每千 Token 成本（$/1M tokens）

def compute_cost_per_token(
    # 硬件成本
    gpu_hourly_cost: float,    # $/GPU/hour
    num_gpus: int,
    # 性能数据
    throughput: float,         # tokens/sec
    # 额外成本
    overhead_factor: float = 1.3,  # 电力+冷却+网络等（30%）
):
    """
    计算每百万 token 的推理成本
    
    公式：
    Cost_per_1M_tokens = (gpu_cost_per_hour × overhead) / (throughput × 3600) × 1e6
    """
    gpu_cost_per_hour = gpu_hourly_cost * num_gpus
    total_cost_per_hour = gpu_cost_per_hour * overhead_factor
    tokens_per_hour = throughput * 3600
    cost_per_1m = total_cost_per_hour / tokens_per_hour * 1e6
    
    print(f"=== 推理成本 ===")
    print(f"硬件成本: ${gpu_cost_per_hour:.2f}/hour")
    print(f"总成本（含开销）: ${total_cost_per_hour:.2f}/hour")
    print(f"吞吐量: {throughput:,.0f} tokens/sec ({throughput * 3600 / 1e6:.1f}M/hour)")
    print(f"成本: ${cost_per_1m:.4f} / 1M tokens")
    return cost_per_1m


# 示例对比
print("=== A100 推理 7B 模型 ===")
compute_cost_per_token(
    gpu_hourly_cost=2.5,   # A100 按需约 $2.5/hour
    num_gpus=1,
    throughput=3000,        # ~3000 tokens/sec
)

print("\n=== H100 推理 70B 模型 ===")
compute_cost_per_token(
    gpu_hourly_cost=4.0,   # H100 按需约 $4/hour
    num_gpus=4,
    throughput=1500,
)

B.4.2 每 FLOP 成本（$/FLOP for Training）

def compute_training_cost(
    gpu_hourly_cost: float,
    num_gpus: int,
    training_days: float,
    model_params: int,
    training_tokens: int,
    mfu: float,
    gpu_flops: float,
):
    """
    训练成本分析
    """
    total_hours = training_days * 24
    total_cost = gpu_hourly_cost * num_gpus * total_hours
    
    # 理论 FLOP
    theoretical_flops = gpu_flops * num_gpus * total_hours * 3600
    actual_flops = theoretical_flops * mfu
    
    # 每 token 成本
    cost_per_token = total_cost / training_tokens
    
    print(f"=== 训练成本 ===")
    print(f"GPU: {num_gpus} × ${gpu_hourly_cost}/hour × {training_days}天")
    print(f"总成本: ${total_cost:,.0f}")
    print(f"训练 tokens: {training_tokens / 1e9:.1f}B")
    print(f"MFU: {mfu * 100:.1f}%")
    print(f"实际 FLOP: {actual_flops / 1e21:.2f} ZFLOP")
    print(f"成本/token: ${cost_per_token:.2e}")
    
    return total_cost


# 示例：训练 70B 模型
compute_training_cost(
    gpu_hourly_cost=2.5,    # A100
    num_gpus=256,
    training_days=30,
    model_params=70e9,
    training_tokens=1.5e12,  # 1.5T tokens
    mfu=0.45,
    gpu_flops=312e12,
)
# 总成本约 $460,800

B.4.3 TCO（Total Cost of Ownership）

def compute_tco(
    # 硬件采购
    gpu_unit_price: float,       # 单 GPU 采购价
    num_gpus: int,
    server_unit_price: float,    # 每台服务器（不含 GPU）
    num_servers: int,
    network_cost: float,         # 网络设备总价
    # 数据中心
    power_per_gpu: float,        # W
    pue: float,                  # 数据中心 PUE
    electricity_rate: float,     # $/kWh
    # 运营
    years: float,                # 使用年限
    ops_team_cost: float,        # 运维团队年成本
    facility_cost: float,        # 机房租赁/建设年成本
):
    """
    GPU 集群的 TCO（总拥有成本）
    """
    # 资本支出（CapEx）
    gpu_capex = gpu_unit_price * num_gpus
    server_capex = server_unit_price * num_servers
    capex = gpu_capex + server_capex + network_cost
    
    # 运营支出（OpEx）
    total_power_kw = (power_per_gpu * num_gpus / 1000) * pue
    annual_electricity = total_power_kw * 24 * 365 * electricity_rate
    opex_per_year = annual_electricity + ops_team_cost + facility_cost
    total_opex = opex_per_year * years
    
    # 折旧后的硬件残值（假设 3 年后 20%）
    residual = capex * 0.2
    
    tco = capex + total_opex - residual
    
    # 单 GPU 月成本
    monthly_cost_per_gpu = tco / (years * 12) / num_gpus
    
    print(f"=== TCO 分析（{years}年）===")
    print(f"\nCapEx（资本支出）:")
    print(f"  GPU: ${gpu_capex:,.0f}")
    print(f"  服务器: ${server_capex:,.0f}")
    print(f"  网络: ${network_cost:,.0f}")
    print(f"  小计: ${capex:,.0f}")
    
    print(f"\nOpEx（运营支出/年）:")
    print(f"  电费: ${annual_electricity:,.0f}")
    print(f"  运维: ${ops_team_cost:,.0f}")
    print(f"  机房: ${facility_cost:,.0f}")
    print(f"  小计: ${opex_per_year:,.0f}/年")
    
    print(f"\n残值: ${residual:,.0f}")
    print(f"\nTCO（{years}年）: ${tco:,.0f}")
    print(f"每 GPU 月成本: ${monthly_cost_per_gpu:,.0f}")
    print(f"每 GPU 小时成本: ${monthly_cost_per_gpu / (30*24):.2f}")
    
    return tco


# 示例：256 GPU 集群 3 年 TCO
compute_tco(
    gpu_unit_price=30_000,    # H100
    num_gpus=256,
    server_unit_price=50_000,  # 每台 8-GPU 服务器
    num_servers=32,
    network_cost=500_000,      # IB 交换机 + 线缆
    power_per_gpu=700,         # H100 SXM 700W
    pue=1.2,                   # 液冷数据中心
    electricity_rate=0.08,     # 工业电价
    years=3,
    ops_team_cost=500_000,
    facility_cost=300_000,
)

Tip成本优化经验法则

1. 推理成本主要由首 token 延迟目标决定——越低的 TTFT 要求越高的 GPU 配比。
2. 训练成本与 MFU 直接挂钩——MFU 从 40% 提升到 50%，训练成本降 20%。
3. 自建 vs 云：GPU 利用率 > 70% 时自建更划算，< 40% 时用云更便宜。
4. 折旧周期：GPU 性能每 2 年翻倍，建议按 3 年折旧计算，而非传统的 4-5 年。

小结

本附录的核心公式汇总：

指标	公式	用途
MFU	6NS / (T × GPU × FLOPS)	训练效率
HFU	6NS(1+α) / (T × GPU × FLOPS)	含重计算的训练效率
训练吞吐量	B×S / T	tokens/sec
TTFT	首 token 生成时间	用户体验
TPOT	(总时间-TTFT) / (N-1)	解码速度
Goodput	满足 SLO 请求数 / 总时间	业务有效吞吐
成本/1M token	$/hour / (TPS × 3600) × 1e6	推理经济性
TCO	CapEx + OpEx × Years - Residual	总拥有成本

记住一点：指标本身没有意义，与目标值的对比和趋势变化才有意义。 建立你的指标基线，持续追踪。

延伸阅读

• PaLM 训练论文（MFU 的经典应用案例）
• LLM-Perf benchmark: https://github.com/ray-project/llmperf
• MLPerf: https://mlcommons.org/benchmarks/
• vLLM benchmark suite: 内置的性能测试工具
• NVIDIA Nsight 文档: 性能分析的标准工具