vLLM推理引擎深度实战：从PagedAttention原理到高吞吐LLM服务部署与量化加速全流程解析

引言

vLLM是当前最高效的开源LLM推理引擎之一，其核心创新PagedAttention技术通过借鉴操作系统的虚拟内存分页机制，极大地优化了KV缓存的内存管理，将LLM推理吞吐量提升了2-4倍。2026年的vLLM已经支持广泛的模型（Llama、Qwen、Mistral、DeepSeek等）、多种量化方案（AWQ、GPTQ、INT8、FP8）、张量并行、流水线并行、连续批处理等高级特性。本文将从vLLM的核心架构出发，系统性地讲解PagedAttention原理、连续批处理、量化部署、分布式推理、API服务构建、性能调优等关键内容，通过大量可运行的Python代码和部署配置帮助读者掌握生产级LLM推理服务部署。

一、vLLM核心架构与PagedAttention原理

1.1 KV缓存问题

LLM推理的核心瓶颈在于KV缓存（Key-Value Cache）的内存管理。在自回归生成中，模型需要缓存之前所有token的Key和Value向量，以避免重复计算。传统方法为每个请求预分配连续的内存空间，但这种方法存在严重的内存浪费。由于生成长度不确定，预分配通常按最大长度分配，但实际使用率可能只有20-30%。此外，连续内存分配导致碎片化问题，限制了系统的并发能力。

1.2 PagedAttention原理

vLLM的PagedAttention将KV缓存划分为固定大小的"块"（blocks），每个块包含固定数量token的KV向量。这些块不需要在物理内存中连续存储，通过一个块表（block table）映射逻辑块到物理块。这种设计带来了多个优势。内存利用率从20-30%提升到接近100%，因为块是按需分配的。支持非连续内存分配，消除了碎片化。共享前缀的请求可以共享KV缓存块，显著减少内存使用。以下是PagedAttention的简化概念模型：

# PagedAttention概念演示（非实际实现，用于理解原理）

class PagedAttentionConcept:
    """PagedAttention的简化概念模型。"""

    BLOCK_SIZE = 16  # 每个块包含16个token的KV缓存

    def __init__(self, num_blocks: int = 256, num_heads: int = 32, head_dim: int = 128):
        self.num_blocks = num_blocks
        self.num_heads = num_heads
        self.head_dim = head_dim

        # 物理块池：所有可用的KV缓存块
        # 每个块: [BLOCK_SIZE, num_heads, head_dim * 2] (K和V)
        self.physical_blocks = [
            {"free": True, "data": None}
            for _ in range(num_blocks)
        ]

        # 每个序列的块表：逻辑块ID -> 物理块ID
        self.block_tables: dict[str, list[int]] = {}

        # 空闲块列表
        self.free_blocks = list(range(num_blocks))

    def allocate_sequence(self, seq_id: str, prompt_len: int):
        """为新序列分配KV缓存块。"""
        num_blocks_needed = (prompt_len + self.BLOCK_SIZE - 1) // self.BLOCK_SIZE
        if num_blocks_needed > len(self.free_blocks):
            raise MemoryError(f"Insufficient blocks: need {num_blocks_needed}, have {len(self.free_blocks)}")

        allocated = []
        for _ in range(num_blocks_needed):
            block_id = self.free_blocks.pop()
            self.physical_blocks[block_id]["free"] = False
            allocated.append(block_id)

        self.block_tables[seq_id] = allocated
        return allocated

    def append_token(self, seq_id: str):
        """为序列追加一个token的KV缓存。"""
        block_table = self.block_tables[seq_id]
        last_block = block_table[-1]

        # 检查最后一个块是否已满
        # 如果满了，分配新块
        # （实际实现中通过逻辑位置判断）
        pass

    def free_sequence(self, seq_id: str):
        """释放序列的所有块。"""
        for block_id in self.block_tables[seq_id]:
            self.physical_blocks[block_id]["free"] = True
            self.free_blocks.append(block_id)
        del self.block_tables[seq_id]

    def get_memory_usage(self) -> dict:
        """获取内存使用情况。"""
        used = self.num_blocks - len(self.free_blocks)
        return {
            "total_blocks": self.num_blocks,
            "used_blocks": used,
            "free_blocks": len(self.free_blocks),
            "utilization": used / self.num_blocks,
            "active_sequences": len(self.block_tables),
        }

    def can_share_prefix(self, seq_ids: list[str]) -> dict:
        """检查序列间是否可以共享前缀块。"""
        if len(seq_ids) < 2:
            return {}

        # 比较块表的前缀
        tables = [self.block_tables[sid] for sid in seq_ids]
        min_len = min(len(t) for t in tables)

        shared_count = 0
        for i in range(min_len):
            if all(t[i] == tables[0][i] for t in tables):
                shared_count += 1
            else:
                break

        return {
            "shared_blocks": shared_count,
            "shared_tokens": shared_count * self.BLOCK_SIZE,
            "memory_saved": shared_count * (len(seq_ids) - 1),
        }


# 演示
pa = PagedAttentionConcept(num_blocks=256)
pa.allocate_sequence("seq1", prompt_len=100)
pa.allocate_sequence("seq2", prompt_len=50)
pa.allocate_sequence("seq3", prompt_len=200)

print("内存使用:", pa.get_memory_usage())

1.3 连续批处理

传统LLM服务的批处理是静态的——等待一批请求到齐后一起处理，处理完再处理下一批。这种方式的问题在于，不同请求的生成长度不同，短请求完成后需要等待长请求完成才能释放资源。vLLM的连续批处理（Continuous Batching）在每次迭代时动态调整批次：已完成的请求立即移出批次，新的请求在下一个迭代立即加入。这大幅提升了GPU利用率：

# vLLM连续批处理概念演示

class ContinuousBatchingScheduler:
    """连续批处理调度器概念模型。"""

    def __init__(self, max_batch_size: int = 32):
        self.max_batch_size = max_batch_size
        self.running_queue: list[dict] = []
        self.waiting_queue: list[dict] = []
        self.completed: list[dict] = []

    def add_request(self, req_id: str, prompt: str, max_tokens: int = 512):
        """添加新请求到等待队列。"""
        self.waiting_queue.append({
            "id": req_id,
            "prompt": prompt,
            "max_tokens": max_tokens,
            "generated": [],
            "status": "waiting",
        })

    def schedule(self) -> list[dict]:
        """调度一轮迭代。"""
        # 1. 移除已完成的请求
        still_running = []
        for req in self.running_queue:
            if req["status"] == "finished":
                self.completed.append(req)
            elif len(req["generated"]) >= req["max_tokens"]:
                req["status"] = "finished"
                self.completed.append(req)
            else:
                still_running.append(req)
        self.running_queue = still_running

        # 2. 从等待队列补充新请求
        available_slots = self.max_batch_size - len(self.running_queue)
        while available_slots > 0 and self.waiting_queue:
            req = self.waiting_queue.pop(0)
            req["status"] = "running"
            self.running_queue.append(req)
            available_slots -= 1

        # 3. 返回当前批次的请求
        return self.running_queue

    def step(self):
        """执行一步生成（概念演示）。"""
        batch = self.schedule()
        if not batch:
            return

        # 模拟生成一个token
        for req in batch:
            token = f"token_{len(req['generated'])}"
            req["generated"].append(token)

        stats = {
            "running": len(self.running_queue),
            "waiting": len(self.waiting_queue),
            "completed": len(self.completed),
        }
        return stats


# 演示连续批处理
scheduler = ContinuousBatchingScheduler(max_batch_size=4)

# 添加不同长度的请求
for i in range(8):
    scheduler.add_request(f"req_{i}", f"Prompt {i}", max_tokens=(i+1)*2)

# 模拟执行
for step in range(20):
    stats = scheduler.step()
    if stats:
        print(f"Step {step:2d}: running={stats['running']}, waiting={stats['waiting']}, completed={stats['completed']}")
    if stats and stats["completed"] == 8:
        print("All requests completed!")
        break

二、vLLM安装与基本使用

2.1 安装

# 基础安装
pip install vllm

# 验证安装
python -c "import vllm; print(vllm.__version__)"

# 查看支持的模型
python -c "from vllm import LLM; help(LLM.__init__)"

2.2 离线批量推理

以下是一个使用vLLM进行批量推理的完整示例：

from vllm import LLM, SamplingParams
from typing import List, Dict

# 加载模型
llm = LLM(
    model="Qwen/Qwen2.5-72B-Instruct",  # 模型路径或HuggingFace ID
    tensor_parallel_size=2,              # 张量并行GPU数
    gpu_memory_utilization=0.9,          # GPU内存使用率
    max_model_len=8192,                  # 最大序列长度
    trust_remote_code=True,              # 信任远程代码
    dtype="auto",                        # 自动选择精度
    enforce_eager=False,                 # 使用CUDA Graph优化
)

# 配置采样参数
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    top_k=50,
    max_tokens=512,
    n=1,                    # 每个prompt生成几个回复
    frequency_penalty=0.0,
    presence_penalty=0.0,
    stop=["<|im_end|>"],    # 停止token
)

# 批量推理
prompts = [
    "请解释什么是PagedAttention，为什么它能提升LLM推理效率？",
    "写一个Python函数，实现快速排序算法，包含详细注释。",
    "比较PostgreSQL和MySQL在JSON处理方面的差异。",
    "设计一个高并发的Web API架构，要求支持10000 QPS。",
]

outputs = llm.generate(prompts, sampling_params)

# 处理输出
for output in outputs:
    prompt = output.prompt
    generated_text = output.outputs[0].text
    print(f"Prompt: {prompt[:80]}...")
    print(f"Generated: {generated_text[:200]}...")
    print(f"Tokens: {len(output.outputs[0].token_ids)}")
    print(f"Finish reason: {output.outputs[0].finish_reason}")
    print("---")

2.3 Chat模型推理

对于经过指令微调的Chat模型，需要使用对话模板：

from vllm import LLM, SamplingParams
from transformers import AutoTokenizer

model_name = "Qwen/Qwen2.5-72B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)

llm = LLM(
    model=model_name,
    tensor_parallel_size=2,
    max_model_len=8192,
    trust_remote_code=True,
)

# 使用对话模板
messages_list = [
    [
        {"role": "system", "content": "你是一个专业的Python技术顾问。"},
        {"role": "user", "content": "如何优化asyncio事件循环的性能？"},
    ],
    [
        {"role": "system", "content": "你是一个数据库架构师。"},
        {"role": "user", "content": "设计一个支持十亿级用户的高可用数据库架构。"},
    ],
]

# 应用对话模板
prompts = [
    tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
    for messages in messages_list
]

sampling_params = SamplingParams(temperature=0.7, max_tokens=1024)
outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    print(output.outputs[0].text)
    print("---")

三、OpenAI兼容API服务

3.1 启动API服务

vLLM内置了与OpenAI API兼容的服务器，可以直接替换OpenAI API使用：

# 启动API服务
python -m vllm.entrypoints.openai.api_server \
    --model Qwen/Qwen2.5-72B-Instruct \
    --tensor-parallel-size 2 \
    --port 8000 \
    --gpu-memory-utilization 0.9 \
    --max-model-len 8192 \
    --trust-remote-code \
    --served-model-name qwen-72b \
    --chat-template /path/to/chat_template.jinja

3.2 使用API

启动后可以使用标准OpenAI SDK调用：

from openai import OpenAI

# 指向vLLM服务
client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="not-needed",  # vLLM默认不验证API key
)

# Chat Completions
response = client.chat.completions.create(
    model="qwen-72b",
    messages=[
        {"role": "system", "content": "你是一个技术文档生成器。"},
        {"role": "user", "content": "写一篇关于Docker容器网络的技术文档。"},
    ],
    temperature=0.7,
    max_tokens=2000,
)
print(response.choices[0].message.content)

# 流式输出
stream = client.chat.completions.create(
    model="qwen-72b",
    messages=[{"role": "user", "content": "解释Kubernetes的调度算法。"}],
    stream=True,
    max_tokens=1024,
)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

# Embeddings（如果模型支持）
embed_response = client.embeddings.create(
    model="qwen-72b",
    input="vLLM是一个高性能LLM推理引擎",
)
print(f"Embedding dimension: {len(embed_response.data[0].embedding)}")

3.3 自定义API服务

以下是一个基于vLLM构建的更完整的API服务，包含负载监控和请求队列：

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from vllm import LLM, SamplingParams
from typing import Optional, List
import asyncio
import time
import json

app = FastAPI(title="vLLM Production API")

# 全局LLM实例（只加载一次）
llm: Optional[LLM] = None

class ChatRequest(BaseModel):
    model: str
    messages: List[dict]
    temperature: float = 0.7
    max_tokens: int = 1024
    top_p: float = 0.9
    stream: bool = False

@app.on_event("startup")
async def load_model():
    global llm
    llm = LLM(
        model="Qwen/Qwen2.5-72B-Instruct",
        tensor_parallel_size=2,
        gpu_memory_utilization=0.9,
        max_model_len=8192,
        trust_remote_code=True,
    )

@app.post("/v1/chat/completions")
async def chat_completions(req: ChatRequest):
    if llm is None:
        raise HTTPException(status_code=503, detail="Model not loaded")

    from transformers import AutoTokenizer
    tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-72B-Instruct", trust_remote_code=True)
    prompt = tokenizer.apply_chat_template(req.messages, tokenize=False, add_generation_prompt=True)

    sampling_params = SamplingParams(
        temperature=req.temperature,
        top_p=req.top_p,
        max_tokens=req.max_tokens,
    )

    if req.stream:
        # 流式输出
        def generate():
            outputs = llm.generate([prompt], sampling_params)
            for output in outputs:
                text = output.outputs[0].text
                chunk = {
                    "id": f"chatcmpl-{int(time.time())}",
                    "object": "chat.completion.chunk",
                    "model": req.model,
                    "choices": [{"delta": {"content": text}, "index": 0}],
                }
                yield f"data: {json.dumps(chunk)}\n\n"
            yield "data: [DONE]\n\n"

        return StreamingResponse(generate(), media_type="text/event-stream")
    else:
        # 非流式
        outputs = llm.generate([prompt], sampling_params)
        text = outputs[0].outputs[0].text

        return {
            "id": f"chatcmpl-{int(time.time())}",
            "object": "chat.completion",
            "model": req.model,
            "choices": [{
                "index": 0,
                "message": {"role": "assistant", "content": text},
                "finish_reason": "stop",
            }],
            "usage": {
                "prompt_tokens": len(tokenizer.encode(prompt)),
                "completion_tokens": len(tokenizer.encode(text)),
                "total_tokens": len(tokenizer.encode(prompt)) + len(tokenizer.encode(text)),
            },
        }

@app.get("/health")
async def health():
    if llm is None:
        return {"status": "loading"}
    return {"status": "healthy", "model": "Qwen2.5-72B-Instruct"}

@app.get("/metrics")
async def metrics():
    """返回推理引擎的运行指标。"""
    if llm is None:
        raise HTTPException(status_code=503, detail="Model not loaded")

    engine = llm.llm_engine
    stats = {
        "num_running_requests": len(engine.scheduler.running),
        "num_waiting_requests": len(engine.scheduler.waiting),
        "max_num_seqs": engine.scheduler.max_num_seqs,
        "gpu_memory_utilization": engine.cache_config.gpu_memory_utilization,
    }
    return stats

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

四、量化部署

4.1 AWQ量化

AWQ（Activation-aware Weight Quantization）是一种后训练量化方法，能将模型权重从FP16压缩到INT4，大幅减少显存占用且几乎不损失精度：

# 启动AWQ量化模型
python -m vllm.entrypoints.openai.api_server \
    --model TheBloke/Qwen2.5-72B-Instruct-AWQ \
    --quantization awq \
    --tensor-parallel-size 1 \
    --gpu-memory-utilization 0.85 \
    --max-model-len 4096 \
    --port 8000

# 使用AWQ量化模型
from vllm import LLM, SamplingParams

llm = LLM(
    model="TheBloke/Qwen2.5-72B-Instruct-AWQ",
    quantization="awq",
    tensor_parallel_size=1,  # AWQ后72B模型可在单张A100上运行
    gpu_memory_utilization=0.85,
    max_model_len=4096,
    trust_remote_code=True,
)

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

4.2 GPTQ量化

# GPTQ量化模型
llm = LLM(
    model="TheBloke/Qwen2.5-72B-Instruct-GPTQ",
    quantization="gptq",
    tensor_parallel_size=1,
    gpu_memory_utilization=0.9,
    max_model_len=4096,
    trust_remote_code=True,
)

4.3 量化对比

以下是一个量化方案对比的基准测试脚本：

import time
import json
from vllm import LLM, SamplingParams

def benchmark_model(model_name: str, quantization: str = None, tp_size: int = 1):
    """基准测试不同量化方案的性能。"""
    print(f"\n=== Benchmarking {model_name} (quantization={quantization}) ===")

    kwargs = {
        "model": model_name,
        "tensor_parallel_size": tp_size,
        "gpu_memory_utilization": 0.9,
        "max_model_len": 4096,
        "trust_remote_code": True,
        "enforce_eager": False,
    }
    if quantization:
        kwargs["quantization"] = quantization

    llm = LLM(**kwargs)

    # 准备测试数据
    prompts = [
        f"请写一篇关于主题{i}的短文。" for i in range(50)
    ]
    sampling_params = SamplingParams(temperature=0.7, max_tokens=512)

    # 预热
    llm.generate(prompts[:5], sampling_params)

    # 正式测试
    start_time = time.time()
    outputs = llm.generate(prompts, sampling_params)
    end_time = time.time()

    # 统计
    total_tokens = sum(len(o.outputs[0].token_ids) for o in outputs)
    duration = end_time - start_time
    throughput = total_tokens / duration

    # 内存使用
    import torch
    gpu_memory = torch.cuda.memory_allocated() / 1024**3  # GB

    result = {
        "model": model_name,
        "quantization": quantization or "none",
        "num_prompts": len(prompts),
        "total_tokens": total_tokens,
        "duration_seconds": round(duration, 2),
        "throughput_tokens_per_sec": round(throughput, 1),
        "gpu_memory_gb": round(gpu_memory, 2),
    }
    print(json.dumps(result, indent=2))

    # 清理
    del llm
    torch.cuda.empty_cache()

    return result

# 运行基准测试
# results = [
#     benchmark_model("Qwen/Qwen2.5-7B-Instruct", tp_size=1),
#     benchmark_model("Qwen/Qwen2.5-7B-Instruct-AWQ", quantization="awq", tp_size=1),
#     benchmark_model("Qwen/Qwen2.5-7B-Instruct-GPTQ", quantization="gptq", tp_size=1),
# ]

五、分布式推理

5.1 张量并行

张量并行将模型的权重矩阵切分到多张GPU上，每张GPU负责一部分计算。这对于大模型（70B+）在单GPU上无法加载的情况至关重要：

from vllm import LLM, SamplingParams

# 2-way张量并行（需要2张GPU）
llm = LLM(
    model="Qwen/Qwen2.5-72B-Instruct",
    tensor_parallel_size=2,           # 将模型切分到2张GPU
    pipeline_parallel_size=1,         # 流水线并行（通常不需要）
    gpu_memory_utilization=0.9,
    max_model_len=8192,
    trust_remote_code=True,
)

# 使用方式与单GPU完全相同
outputs = llm.generate(["Hello, world!"], SamplingParams(max_tokens=100))

5.2 多节点分布式推理

对于超大规模模型，可以使用多节点分布式推理：

# 节点0（主节点）
python -m vllm.entrypoints.openai.api_server \
    --model Qwen/Qwen2.5-72B-Instruct \
    --tensor-parallel-size 4 \
    --pipeline-parallel-size 2 \
    --distributed-executor-backend ray \
    --ray-address auto \
    --port 8000

# 需要先启动Ray集群
# 节点0: ray start --head --port=6379
# 节点1: ray start --address=<node0-ip>:6379

5.3 推理优化配置

llm = LLM(
    model="Qwen/Qwen2.5-72B-Instruct",
    tensor_parallel_size=2,

    # 内存优化
    gpu_memory_utilization=0.9,       # GPU内存使用率上限
    swap_space=4,                     # CPU交换空间(GB)
    max_model_len=8192,               # 最大序列长度

    # 批处理优化
    max_num_seqs=256,                 # 最大并发序列数
    max_num_batched_tokens=8192,      # 每批次最大token数

    # CUDA Graph优化
    enforce_eager=False,              # 启用CUDA Graph
    cuda_graph_sizes=[1, 2, 4, 8, 16, 32, 64, 128],  # CUDA Graph批次大小

    # 注意力后端
    attention_backend="FLASHINFER",   # 使用FlashInfer注意力

    # 其他优化
    use_v2_block_manager=True,        # V2块管理器
    enable_chunked_prefill=True,      # 分块预填充
    trust_remote_code=True,
)

六、vLLM与RAG集成

6.1 自托管RAG服务

以下是一个完整的自托管RAG系统，使用vLLM作为推理引擎：

from fastapi import FastAPI
from pydantic import BaseModel
from vllm import LLM, SamplingParams
from qdrant_client import QdrantClient
from sentence_transformers import CrossEncoder
from rank_bm25 import BM25Okapi
import numpy as np
from typing import List, Optional
import jieba

app = FastAPI(title="Self-hosted RAG with vLLM")

# 初始化组件
llm = LLM(
    model="Qwen/Qwen2.5-72B-Instruct",
    tensor_parallel_size=2,
    max_model_len=8192,
    trust_remote_code=True,
)

qdrant = QdrantClient(host="localhost", port=6333)
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")

class RAGQuery(BaseModel):
    question: str
    top_k: int = 10
    rerank_top_k: int = 5
    temperature: float = 0.3
    max_tokens: int = 1024

def embed_text(text: str) -> np.ndarray:
    """使用本地嵌入模型。"""
    from sentence_transformers import SentenceTransformer
    embedder = SentenceTransformer("BAAI/bge-m3")
    return embedder.encode(text)

def retrieve(question: str, top_k: int) -> List[dict]:
    """向量检索。"""
    query_vector = embed_text(question).tolist()
    results = qdrant.search(
        collection_name="documents",
        query_vector=query_vector,
        limit=top_k,
    )
    return [
        {"text": hit.payload["text"], "score": hit.score, "metadata": hit.payload}
        for hit in results
    ]

def rerank(question: str, docs: List[dict], top_k: int) -> List[dict]:
    """重排序。"""
    pairs = [(question, doc["text"]) for doc in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, docs), key=lambda x: x[0], reverse=True)
    return [{**doc, "rerank_score": float(score)} for score, doc in ranked[:top_k]]

@app.post("/rag")
async def rag_query(req: RAGQuery):
    # 检索
    retrieved = retrieve(req.question, req.top_k)
    # 重排序
    reranked = rerank(req.question, retrieved, req.rerank_top_k)

    # 构建Prompt
    context = "\n\n".join([f"[文档{i+1}] {d['text']}" for i, d in enumerate(reranked)])
    prompt = f"基于以下文档回答问题。\n\n文档：\n{context}\n\n问题：{req.question}\n\n回答："

    # vLLM推理
    sampling = SamplingParams(
        temperature=req.temperature,
        max_tokens=req.max_tokens,
    )
    outputs = llm.generate([prompt], sampling)
    answer = outputs[0].outputs[0].text

    return {
        "answer": answer,
        "sources": [
            {"text": d["text"][:200], "score": d.get("rerank_score", d["score"])}
            for d in reranked
        ],
    }

七、性能调优与监控

7.1 性能调优指南

# 性能调优基准测试脚本
import time
import json
from vllm import LLM, SamplingParams

def tune_performance():
    """测试不同配置下的性能。"""
    configs = [
        {"name": "default", "kwargs": {}},
        {"name": "high_batch", "kwargs": {"max_num_seqs": 512, "max_num_batched_tokens": 16384}},
        {"name": "chunked_prefill", "kwargs": {"enable_chunked_prefill": True}},
        {"name": "cuda_graph", "kwargs": {"enforce_eager": False}},
        {"name": "all_optimizations", "kwargs": {
            "max_num_seqs": 512,
            "max_num_batched_tokens": 16384,
            "enable_chunked_prefill": True,
            "enforce_eager": False,
            "use_v2_block_manager": True,
        }},
    ]

    prompts = [f"Write a short story about topic {i}." for i in range(100)]
    sampling = SamplingParams(temperature=0.7, max_tokens=256)

    results = []
    for config in configs:
        print(f"\nTesting: {config['name']}")
        llm = LLM(
            model="Qwen/Qwen2.5-7B-Instruct",
            tensor_parallel_size=1,
            gpu_memory_utilization=0.9,
            max_model_len=4096,
            trust_remote_code=True,
            **config["kwargs"],
        )

        # 预热
        llm.generate(prompts[:5], sampling)

        # 测试
        start = time.time()
        outputs = llm.generate(prompts, sampling)
        duration = time.time() - start

        total_tokens = sum(len(o.outputs[0].token_ids) for o in outputs)
        result = {
            "config": config["name"],
            "duration": round(duration, 2),
            "throughput": round(total_tokens / duration, 1),
            "total_tokens": total_tokens,
        }
        results.append(result)
        print(json.dumps(result, indent=2))

        del llm
        import torch
        torch.cuda.empty_cache()

    print("\n=== Summary ===")
    for r in results:
        print(f"{r['config']:25s}: {r['throughput']:>10.1f} tokens/s")

tune_performance()

7.2 Prometheus监控

from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import Response

# 定义指标
REQUEST_COUNT = Counter('vllm_requests_total', 'Total requests', ['model'])
REQUEST_LATENCY = Histogram('vllm_request_latency_seconds', 'Request latency')
GPU_MEMORY = Gauge('vllm_gpu_memory_bytes', 'GPU memory usage')
RUNNING_REQUESTS = Gauge('vllm_running_requests', 'Currently running requests')
WAITING_REQUESTS = Gauge('vllm_waiting_requests', 'Waiting requests')

@app.get("/metrics")
async def prometheus_metrics():
    # 更新指标
    if llm:
        engine = llm.llm_engine
        RUNNING_REQUESTS.set(len(engine.scheduler.running))
        WAITING_REQUESTS.set(len(engine.scheduler.waiting))
        import torch
        GPU_MEMORY.set(torch.cuda.memory_allocated())

    return Response(generate_latest(), media_type="text/plain")

总结

vLLM凭借PagedAttention和连续批处理两大核心创新，成为当前最高效的开源LLM推理引擎。本文系统性地覆盖了PagedAttention原理、连续批处理、离线推理和API服务、AWQ/GPTQ量化部署、张量并行分布式推理、与RAG系统集成、性能调优和监控等核心内容。关键要点包括：PagedAttention通过分式KV缓存将内存利用率从30%提升到接近100%；连续批处理动态调整批次使GPU利用率最大化；量化部署（AWQ/GPTQ）可将大模型显存需求降低4倍；张量并行使得70B+模型可以在多GPU上高效运行；生产部署需要合理配置max_num_seqs、enable_chunked_prefill和CUDA Graph等参数。随着开源大模型能力的持续提升，基于vLLM构建自托管LLM服务将成为企业AI基础设施的标准方案，掌握vLLM的部署和调优能力对于AI工程师至关重要。