推理性能优化实战:从毫秒到微秒的优化之旅
大语言模型的推理成本直接影响着 AI 应用的商业模式和用户体验。本文将深入探讨 LLM 推理优化的核心技术,从 KV Cache 管理到投机解码,帮助您将推理性能推向极致。
一、推理性能瓶颈分析
1.1 自回归生成的计算特点
LLM 推理分为两个阶段:
| 阶段 | 特点 | 计算瓶颈 |
|---|---|---|
| Prefill | 处理输入 prompt,计算量大 | 计算密集型,与 prompt 长度成正比 |
| Decode | 逐个生成 token,访存密集型 | 内存带宽限制,KV Cache 访问 |
1.2 性能公式
Decode Time = (num_tokens × model_params × precision_bytes) / memory_bandwidth
以 Llama-2-70B FP16 在 A100 80GB 上为例:
- 模型参数:140 GB (FP16)
- 内存带宽:2 TB/s
- 理论单 token 延迟:70ms
二、KV Cache 优化
2.1 PagedAttention 原理
传统 KV Cache 存在严重的内存碎片问题。PagedAttention 借鉴操作系统的虚拟内存管理,将 KV Cache 划分为固定大小的块(block)。
# PagedAttention 核心概念示意
class PagedAttention:
def __init__(self, block_size=16, num_blocks=10000):
self.block_size = block_size
self.num_blocks = num_blocks
# 空闲块列表
self.free_blocks = list(range(num_blocks))
# 块分配表:seq_id -> [block_ids]
self.block_tables = {}
def allocate(self, seq_id, num_tokens):
"""为序列分配 KV Cache 块"""
num_blocks_needed = (num_tokens + self.block_size - 1) // self.block_size
blocks = []
for _ in range(num_blocks_needed):
if self.free_blocks:
block_id = self.free_blocks.pop()
blocks.append(block_id)
self.block_tables[seq_id] = blocks
return blocks
def free(self, seq_id):
"""释放序列占用的块"""
if seq_id in self.block_tables:
self.free_blocks.extend(self.block_tables[seq_id])
del self.block_tables[seq_id]
2.2 vLLM 中的 KV Cache 配置
from vllm import LLM, SamplingParams
# 优化 KV Cache 配置
llm = LLM(
model="meta-llama/Llama-2-7b-hf",
# 块大小配置
block_size=16, # 可选: 8, 16, 32
# GPU 内存利用率
gpu_memory_utilization=0.9,
# 最大序列数
max_num_seqs=256,
# 最大模型长度
max_model_len=4096,
# 启用前缀缓存 (vLLM 0.3.0+)
enable_prefix_caching=True,
)
sampling_params = SamplingParams(
temperature=0.7,
top_p=0.9,
max_tokens=512
)
# 批量推理
prompts = ["你好"] * 100
outputs = llm.generate(prompts, sampling_params)
2.3 前缀缓存(Prefix Caching)
前缀缓存可以显著加速具有相同前缀的多次调用,如多轮对话、RAG 应用等场景。
# 前缀缓存实战:RAG 场景优化
from vllm import LLM, SamplingParams
llm = LLM(
model="meta-llama/Llama-2-7b-hf",
enable_prefix_caching=True, # 启用前缀缓存
)
# 共享的系统提示前缀
system_prefix = """你是一个专业的 AI 助手。请基于以下上下文回答问题:
上下文:
{context}
问题:"""
# 多个问题共享相同上下文
context = "...长上下文内容..."
prompts = [
system_prefix.format(context=context) + "什么是机器学习?",
system_prefix.format(context=context) + "深度学习有哪些应用?",
system_prefix.format(context=context) + "如何入门 NLP?",
]
# 第一次调用会缓存前缀,后续调用加速 30-50%
outputs = llm.generate(prompts, SamplingParams(max_tokens=256))
三、量化技术详解
3.1 量化方案对比
| 量化类型 | 精度损失 | 速度提升 | 显存节省 | 适用场景 |
|---|---|---|---|---|
| FP16 | 基准 | 1x | 1x | 通用 |
| FP8 (e4m3) | < 1% | 1.5-2x | 2x | NVIDIA H100 |
| INT8 (SmoothQuant) | 1-2% | 1.3-1.5x | 2x | 通用 |
| INT4 (GPTQ/AWQ) | 2-4% | 1.5-2x | 4x | 资源受限 |
| GGUF (Q4_K_M) | 3-5% | CPU 可用 | 4x | 边缘设备 |
3.2 AWQ 量化实战
AWQ (Activation-aware Weight Quantization) 是一种保护重要权重通道的量化方法。
# AWQ 量化步骤
from awq import AutoAWQForCausalLM
from transformers import AutoTokenizer
model_path = "meta-llama/Llama-2-7b-hf"
quant_path = "llama-2-7b-awq"
quant_config = {
"zero_point": True,
"q_group_size": 128,
"w_bit": 4,
"version": "GEMM"
}
# 加载模型
model = AutoAWQForCausalLM.from_pretrained(
model_path, **{"low_cpu_mem_usage": True}
)
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
# 准备校准数据
from awq.utils.calib_data import get_calib_dataset
calib_data = get_calib_dataset(
data="pileval", # 或 "wikitext", "c4"
tokenizer=tokenizer,
n_samples=128,
block_size=512
)
# 量化
model.quantize(
tokenizer,
quant_config=quant_config,
calib_data=calib_data,
)
# 保存
model.save_quantized(quant_path)
tokenizer.save_pretrained(quant_path)
3.3 GPTQ 量化与推理
# 使用 AutoGPTQ 进行量化
pip install auto-gptq
# 量化脚本
python << 'EOF'
from auto_gptq import AutoGPTQForCausalLM, BaseQuantizeConfig
from transformers import AutoTokenizer
model_id = "meta-llama/Llama-2-7b-hf"
quantize_config = BaseQuantizeConfig(
bits=4,
group_size=128,
desc_act=False, # 设为 True 提升精度但降低速度
)
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoGPTQForCausalLM.from_pretrained(
model_id, quantize_config
)
# 准备校准数据
examples = [
tokenizer("auto-gptq is an easy-to-use model quantization library",
return_tensors="pt").input_ids[0]
for _ in range(128)
]
# 量化
model.quantize(examples, batch_size=1)
# 保存
quantized_model_dir = "llama-2-7b-gptq"
model.save_quantized(quantized_model_dir)
tokenizer.save_pretrained(quantized_model_dir)
EOF
3.4 vLLM 加载量化模型
from vllm import LLM, SamplingParams
# 加载 AWQ/GPTQ 量化模型
llm = LLM(
model="llama-2-7b-awq", # 或 "llama-2-7b-gptq"
quantization="awq", # 或 "gptq"
dtype="auto",
gpu_memory_utilization=0.9,
)
# 正常推理
output = llm.generate("你好", SamplingParams(max_tokens=100))
print(output[0].outputs[0].text)
四、投机解码(Speculative Decoding)
4.1 原理概述
投机解码使用小模型(draft model)快速生成候选 token,大模型(target model)并行验证,实现 2-3 倍加速。
传统解码: [大模型] -> token1 -> [大模型] -> token2 -> [大模型] -> token3
投机解码: [小模型] -> t1,t2,t3,t4 -> [大模型验证] -> accept 3, reject 1
4.2 vLLM 投机解码实战
from vllm import LLM, SamplingParams
# 配置投机解码
llm = LLM(
model="meta-llama/Llama-2-70b-hf",
speculative_model="meta-llama/Llama-2-7b-hf", # draft model
num_speculative_tokens=5, # 每次推测的 token 数
gpu_memory_utilization=0.9,
)
sampling_params = SamplingParams(
temperature=0.7,
max_tokens=512,
)
# 推理速度提升 1.5-2.5 倍
output = llm.generate(
"请详细解释量子计算的原理和应用",
sampling_params
)
4.3 自定义 Draft Model
# 使用更小的同系列模型作为 draft
from vllm import LLM
llm = LLM(
model="codellama/CodeLlama-34b-Instruct-hf",
speculative_model="codellama/CodeLlama-7b-Instruct-hf",
num_speculative_tokens=4,
speculative_max_model_len=4096,
)
五、FlashAttention 与内存优化
5.1 FlashAttention-2 集成
FlashAttention 通过 IO-aware 的 attention 计算,显著减少 HBM 访问。
# vLLM 自动使用 FlashAttention-2
# 手动启用配置
import os
os.environ["VLLM_ATTENTION_BACKEND"] = "FLASH_ATTN" # 或 "XFORMERS", "FLASHINFER"
from vllm import LLM
llm = LLM(
model="meta-llama/Llama-2-7b-hf",
# FlashAttention 配置
dtype="half", # FP16
max_model_len=8192,
)
5.2 内存优化技巧
# 1. 梯度检查点(训练时)
from transformers import AutoModelForCausalLM
model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-2-7b-hf",
torch_dtype="auto",
use_cache=False, # 推理时不需要梯度检查点
)
model.gradient_checkpointing_enable()
# 2. 8-bit/4-bit 加载
from transformers import BitsAndBytesConfig
import torch
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16,
bnb_4bit_quant_type="nf4",
bnb_4bit_use_double_quant=True,
)
model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-2-7b-hf",
quantization_config=bnb_config,
device_map="auto",
)
六、Continuous Batching 优化
6.1 动态批处理原理
Continuous Batching(Inflight Batching)允许在生成过程中动态添加/移除请求,最大化 GPU 利用率。
# vLLM 的 Continuous Batching 自动启用
# 参数调优指南
from vllm import LLM
llm = LLM(
model="meta-llama/Llama-2-7b-hf",
# 关键参数
max_num_batched_tokens=4096, # 最大批处理 token 数
max_num_seqs=256, # 最大并发序列数
max_model_len=4096,
# 调度策略
scheduling_policy="fcfs", # 或 "priority"
)
6.2 请求调度优化
# 自定义调度策略
from vllm.core.scheduler import Scheduler
class PriorityScheduler(Scheduler):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.priority_queue = []
def add_request(self, request, priority=0):
"""添加带优先级的请求"""
request.priority = priority
self.priority_queue.append(request)
self.priority_queue.sort(key=lambda x: x.priority, reverse=True)
七、性能测试与监控
7.1 基准测试脚本
import time
import statistics
from vllm import LLM, SamplingParams
def benchmark_throughput(
model_path,
prompts,
max_tokens=256,
warmup=3
):
"""吞吐量基准测试"""
llm = LLM(
model=model_path,
gpu_memory_utilization=0.9,
)
sampling_params = SamplingParams(
temperature=0.7,
max_tokens=max_tokens,
)
# 预热
for _ in range(warmup):
llm.generate(prompts[:4], sampling_params)
# 正式测试
start = time.time()
outputs = llm.generate(prompts, sampling_params)
elapsed = time.time() - start
# 统计
total_tokens = sum(
len(o.outputs[0].token_ids) for o in outputs
)
throughput = total_tokens / elapsed
print(f"总时间: {elapsed:.2f}s")
print(f"总 token: {total_tokens}")
print(f"吞吐量: {throughput:.2f} tokens/s")
return throughput
# 测试不同并发数
prompts = ["你好,请介绍一下自己"] * 100
throughput = benchmark_throughput(
"meta-llama/Llama-2-7b-hf",
prompts,
max_tokens=256
)
7.2 延迟分析
import torch
import time
class LatencyProfiler:
def __init__(self):
self.timings = {}
def profile_module(self, model, input_ids):
"""分析各模块延迟"""
hooks = []
def hook_fn(name):
def fn(module, input, output):
if name not in self.timings:
self.timings[name] = []
self.timings[name].append(time.time())
return fn
# 注册钩子
for name, module in model.named_modules():
if "attention" in name or "mlp" in name:
hooks.append(
module.register_forward_hook(hook_fn(name))
)
# 推理
with torch.no_grad():
model(input_ids)
# 移除钩子
for h in hooks:
h.remove()
return self.timings
# 使用示例
# profiler = LatencyProfiler()
# timings = profiler.profile_module(model, input_ids)
八、综合优化案例
8.1 RAG 系统优化
# 完整的 RAG 推理优化配置
from vllm import LLM, SamplingParams
class OptimizedRAGPipeline:
def __init__(self):
# 1. 量化模型减少显存
self.llm = LLM(
model="llama-2-7b-awq",
quantization="awq",
# 2. 启用前缀缓存
enable_prefix_caching=True,
# 3. 配置批处理
max_num_seqs=64,
max_model_len=8192,
gpu_memory_utilization=0.9,
)
self.sampling_params = SamplingParams(
temperature=0.1, # RAG 需要确定性输出
top_p=0.95,
max_tokens=512,
)
# 4. 系统提示模板(用于前缀缓存)
self.system_template = """基于以下上下文回答问题:
{context}
问题:{question}
回答:"""
def batch_generate(self, contexts, questions):
"""批量生成,利用前缀缓存"""
prompts = [
self.system_template.format(context=c, question=q)
for c, q in zip(contexts, questions)
]
outputs = self.llm.generate(prompts, self.sampling_params)
return [o.outputs[0].text for o in outputs]
# 使用
rag = OptimizedRAGPipeline()
answers = rag.batch_generate(
contexts=["上下文1...", "上下文2..."],
questions=["问题1?", "问题2?"]
)
8.2 性能对比数据
在 A100 40GB 上测试 Llama-2-7B:
| 优化技术 | 吞吐量 (tok/s) | 延迟 (ms/token) | 显存占用 |
|---|---|---|---|
| 基线 (FP16) | 85 | 11.8 | 14.2 GB |
| + PagedAttention | 120 | 8.3 | 12.1 GB |
| + AWQ 量化 | 180 | 5.6 | 5.8 GB |
| + 前缀缓存 | 220 | 4.5 | 5.8 GB |
| + 投机解码 | 350 | 2.9 | 8.2 GB |
九、总结
LLM 推理优化是一个系统工程,需要根据具体场景组合多种技术:
- KV Cache 优化:使用 PagedAttention 和前缀缓存减少显存占用
- 量化技术:AWQ/GPTQ 在精度和速度间取得平衡
- 投机解码:适合延迟敏感场景,可提升 2-3 倍速度
- Continuous Batching:最大化 GPU 利用率
- FlashAttention:减少 attention 计算的内存访问
建议优化路径:
- 首先启用 PagedAttention 和 Continuous Batching
- 根据显存情况选择 AWQ 或 GPTQ 量化
- 对延迟敏感场景添加投机解码
- 多轮对话/RAG 场景启用前缀缓存
参考资源: