vLLM 高性能大模型推理部署教程
一、vLLM概述：为什么需要高性能推理引擎
在大语言模型（LLM）的实际应用中，训练只是第一步，真正的挑战在于如何将模型高效地部署到生产环境中为用户提供服务。当你的模型需要同时服务成百上千的用户请求时，推理性能就成为了决定产品成败的关键因素。
1.1 LLM推理的核心瓶颈
LLM推理面临的核心问题包括：
内存瓶颈：一个70B参数的模型，即使使用FP16精度，也需要约140GB的显存来存储模型权重。而推理过程中还需要大量的显存来存储KV Cache（键值缓存），这是Transformer架构自回归生成的本质需求。
延迟问题：用户期望实时响应，但LLM的自回归生成特性意味着每个token的生成都依赖于前面所有token的计算，无法简单并行化。
吞吐量限制：在高并发场景下，如何最大化GPU利用率，同时处理尽可能多的请求，是推理引擎必须解决的问题。
成本压力：GPU资源昂贵，如何在有限的硬件资源上实现最大的服务吞吐量，直接影响运营成本。
1.2 vLLM的诞生与核心价值
vLLM是由UC Berkeley的研究团队开发的开源高性能LLM推理引擎。它的核心创新在于提出了PagedAttention技术，从根本上解决了LLM推理中的内存管理问题。
vLLM的核心价值：
- 极致的吞吐量：相比传统的HuggingFace Transformers推理，vLLM可以实现2-24倍的吞吐量提升
- 高效的内存管理：通过PagedAttention技术，将KV Cache的内存利用率从传统的60-80%提升至95%以上
- OpenAI兼容API：提供与OpenAI API完全兼容的接口，现有应用可以无缝迁移
- 丰富的量化支持：支持GPTQ、AWQ、FP8等多种量化方案
- 灵活的并行策略：支持Tensor并行和Pipeline并行
- 动态LoRA加载：支持在不重启服务的情况下动态切换LoRA适配器
1.3 与其他推理框架的对比
| 特性 | vLLM | TGI | TensorRT-LLM | llama.cpp |
|---|---|---|---|---|
| 吞吐量 | 极高 | 高 | 极高 | 中等 |
| 易用性 | 高 | 中 | 低 | 高 |
| 量化支持 | GPTQ/AWQ/FP8 | GPTQ/AWQ | FP8/INT8/INT4 | GGUF多种 |
| 动态批处理 | ✅ | ✅ | ✅ | ❌ |
| LoRA支持 | ✅ 动态加载 | ✅ | ✅ | ✅ |
| OpenAI兼容API | ✅ | ✅ | 需适配 | ✅ |
| 社区活跃度 | 极高 | 高 | 中 | 高 |
| 适用场景 | 生产部署 | 生产部署 | 极致性能 | 边缘设备 |
二、PagedAttention核心原理：KV Cache优化
2.1 传统KV Cache的问题
在Transformer模型的自回归生成过程中，每生成一个新token，都需要计算该token与之前所有token之间的注意力分数。为了避免重复计算，我们会将之前计算过的Key和Value缓存下来，这就是KV Cache。
传统方法中，KV Cache的内存分配存在两个严重问题：
内存碎片化：每个请求需要预分配一个连续的内存块来存储最大可能长度的KV Cache。但实际上，大多数请求的实际生成长度远小于最大长度，导致大量内存被浪费。
内存浪费：为了保证请求能完成，系统必须按最大序列长度预分配内存。例如，如果最大序列长度是2048，但实际平均生成长度只有200，那么约90%的预分配内存都是浪费的。
传统KV Cache内存分配示意：
┌─────────────────────────────────────────────────────┐
│ 请求1: [实际使用的KV][预分配但未使用的空间............] │ ← 内存浪费
│ 请求2: [实际使用的KV][预分配但未使用的空间............] │ ← 内存浪费
│ 请求3: [实际使用的KV][预分配但未使用的空间............] │ ← 内存浪费
└─────────────────────────────────────────────────────┘
实际利用率: ~30%
2.2 PagedAttention的创新
PagedAttention借鉴了操作系统中虚拟内存和分页的思想，将KV Cache的管理从"连续分配"改为"分页管理"。
核心思想：
- 物理块（Physical Block）：将GPU显存划分为固定大小的块（Block），每个块可以存储固定数量token的KV Cache
- 逻辑块（Logical Block）：每个请求维护一个逻辑块表，记录其KV Cache存储在哪些物理块中
- 按需分配：只有当实际需要存储新的KV Cache时，才分配新的物理块
- 块共享：多个请求可以共享相同的物理块（用于Prompt相同的部分）
PagedAttention内存分配示意：
物理块池:
┌──────┬──────┬──────┬──────┬──────┬──────┬──────┬──────┐
│ 块0 │ 块1 │ 块2 │ 块3 │ 块4 │ 块5 │ 块6 │ 块7 │
│req1 │req1 │req2 │req2 │req3 │空闲 │空闲 │空闲 │
└──────┴──────┴──────┴──────┴──────┴──────┴──────┴──────┘
请求1的逻辑视图: [块0] → [块1]
请求2的逻辑视图: [块2] → [块3]
请求3的逻辑视图: [块4]
内存利用率: ~95%+
2.3 Copy-on-Write机制
vLLM还实现了Copy-on-Write（写时复制）机制来优化并行采样场景。当多个采样序列共享相同的Prompt前缀时，它们可以共享相同的物理块，只有当某个序列生成了不同的token时，才会复制一份新的物理块。
# 并行采样示例：3个采样结果共享Prompt的KV Cache
# Prompt: "请解释量子计算的基本原理"
#
# 物理块分配：
# Prompt部分: [Block_A, Block_B] ← 3个采样共享
# 采样1独有: [Block_C1]
# 采样2独有: [Block_C2]
# 采样3独有: [Block_C3]
#
# 内存节省: Prompt部分只存储一次，而非3次
2.4 性能收益
PagedAttention带来的性能提升：
- 内存利用率：从60-80%提升至95%以上
- 吞吐量提升：在相同硬件上可以服务2-4倍的并发请求
- 最大序列长度：可以支持更长的输出序列，因为不再需要预分配
- 批处理效率：动态批处理更加高效，因为内存分配更加灵活
三、vLLM安装与环境配置
3.1 系统要求
- 操作系统：Linux (Ubuntu 20.04+, CentOS 7+)
- Python：3.8 - 3.11
- CUDA：11.8 - 12.1
- GPU：NVIDIA GPU with compute capability ≥ 7.0 (V100, A10, A100, H100等)
- 内存：至少32GB系统内存
- 显存：根据模型大小，至少16GB
3.2 安装方法
方法一：pip安装（推荐）
# 创建虚拟环境
python -m venv vllm-env
source vllm-env/bin/activate
# 安装vLLM
pip install vllm
# 如果需要从源码安装最新版本
pip install git+https://github.com/vllm-project/vllm.git
方法二：Docker安装
# 拉取官方Docker镜像
docker pull vllm/vllm-openai:latest
# 运行容器
docker run --runtime nvidia --gpus all \
-v ~/.cache/huggingface:/root/.cache/huggingface \
-p 8000:8000 \
--ipc=host \
vllm/vllm-openai:latest \
--model meta-llama/Llama-2-7b-chat-hf
方法三：conda安装
# 创建conda环境
conda create -n vllm python=3.10
conda activate vllm
# 安装PyTorch (根据你的CUDA版本选择)
conda install pytorch pytorch-cuda=12.1 -c pytorch -c nvidia
# 安装vLLM
pip install vllm
3.3 环境配置
# 设置HuggingFace镜像（国内用户）
export HF_ENDPOINT=https://hf-mirror.com
# 设置模型缓存目录
export HF_HOME=/data/models/huggingface
# 设置NCCL环境变量（多GPU场景）
export NCCL_DEBUG=INFO
export NCCL_P2P_DISABLE=1 # 如果遇到P2P通信问题
# 设置vLLM日志级别
export VLLM_LOGGING_LEVEL=INFO
3.4 验证安装
# test_vllm.py
from vllm import LLM, SamplingParams
# 初始化模型
llm = LLM(model="facebook/opt-125m") # 使用小模型测试
# 设置采样参数
sampling_params = SamplingParams(temperature=0.8, top_p=0.95, max_tokens=100)
# 生成文本
prompts = ["Hello, my name is", "The capital of France is"]
outputs = llm.generate(prompts, sampling_params)
for output in outputs:
prompt = output.prompt
generated_text = output.outputs[0].text
print(f"Prompt: {prompt!r}")
print(f"Generated: {generated_text!r}")
print()
python test_vllm.py
四、模型加载与量化：GPTQ、AWQ、FP8
4.1 模型加载基础
from vllm import LLM
# 基础加载
llm = LLM(model="meta-llama/Llama-2-7b-chat-hf")
# 指定参数加载
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
dtype="float16", # 数据类型: float16, bfloat16, float32
max_model_len=4096, # 最大序列长度
gpu_memory_utilization=0.9, # GPU内存利用率
tensor_parallel_size=1, # Tensor并行数
trust_remote_code=True, # 信任远程代码
download_dir="/data/models", # 模型下载目录
)
4.2 GPTQ量化
GPTQ是一种训练后量化方法，可以将FP16模型量化为INT4/INT8，显著减少显存占用。
# 加载GPTQ量化模型
llm = LLM(
model="TheBloke/Llama-2-7B-Chat-GPTQ",
dtype="float16",
quantization="gptq",
gpu_memory_utilization=0.9,
)
# 使用示例
from vllm import SamplingParams
sampling_params = SamplingParams(temperature=0.7, max_tokens=512)
outputs = llm.generate(["请介绍一下人工智能的发展历史"], sampling_params)
print(outputs[0].outputs[0].text)
GPTQ量化的优势：
- INT4量化：模型大小减少约75%
- 推理速度：相比FP16有1.5-2倍提升
- 精度损失：在可接受范围内（通常<1%）
4.3 AWQ量化
AWQ（Activation-aware Weight Quantization）是一种更先进的量化方法，通过保护重要权重通道来实现更好的量化质量。
# 加载AWQ量化模型
llm = LLM(
model="TheBloke/Llama-2-7B-Chat-AWQ",
dtype="float16",
quantization="awq",
gpu_memory_utilization=0.9,
)
AWQ vs GPTQ：
- AWQ通常有更好的量化质量
- AWQ的量化速度更快
- AWQ对校准数据集的依赖更少
4.4 FP8量化
FP8是Hopper架构（H100/H200）和Ada Lovelace架构（L40S）支持的新数据类型，可以在几乎不损失精度的情况下提升推理速度。
# FP8量化（需要H100或L40S GPU）
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
dtype="float16",
quantization="fp8", # 使用FP8量化
gpu_memory_utilization=0.9,
)
# 或者使用FP8 E5M2格式
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
quantization="fp8_e5m2",
)
4.5 自动量化检测
vLLM可以自动检测模型的量化格式：
# vLLM会自动检测GPTQ/AWQ量化
llm = LLM(
model="TheBloke/Llama-2-7B-Chat-GPTQ",
# 不需要手动指定quantization参数
)
4.6 量化方案选择指南
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| H100/L40S GPU | FP8 | 硬件原生支持，性能最佳 |
| A100/A10 GPU | AWQ | 量化质量好，兼容性佳 |
| V100 GPU | GPTQ | 广泛支持，成熟稳定 |
| 显存极度紧张 | GPTQ INT4 | 最小显存占用 |
| 精度要求高 | AWQ 或 FP8 | 精度损失最小 |
五、API服务：OpenAI兼容API、流式输出
5.1 启动API服务
# 基础启动
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-2-7b-chat-hf \
--host 0.0.0.0 \
--port 8000
# 完整参数启动
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-2-7b-chat-hf \
--host 0.0.0.0 \
--port 8000 \
--dtype float16 \
--max-model-len 4096 \
--gpu-memory-utilization 0.9 \
--tensor-parallel-size 1 \
--served-model-name llama-7b \
--trust-remote-code \
--api-key your-api-key-here
5.2 使用OpenAI客户端调用
from openai import OpenAI
# 初始化客户端
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="your-api-key-here" # 与服务端配置一致
)
# 非流式调用
response = client.chat.completions.create(
model="llama-7b",
messages=[
{"role": "system", "content": "你是一个有帮助的AI助手。"},
{"role": "user", "content": "请介绍一下Python的装饰器是什么？"}
],
temperature=0.7,
max_tokens=1024,
)
print(response.choices[0].message.content)
5.3 流式输出
from openai import OpenAI
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="your-api-key-here"
)
# 流式调用
stream = client.chat.completions.create(
model="llama-7b",
messages=[
{"role": "user", "content": "写一篇关于机器学习的短文，500字左右。"}
],
temperature=0.7,
max_tokens=1024,
stream=True, # 启用流式输出
)
# 逐块接收输出
for chunk in stream:
if chunk.choices[0].delta.content is not None:
print(chunk.choices[0].delta.content, end="", flush=True)
print() # 换行
5.4 Embedding服务
# 启动Embedding服务
# python -m vllm.entrypoints.openai.api_server \
# --model BAAI/bge-base-en-v1.5 \
# --host 0.0.0.0 \
# --port 8000
from openai import OpenAI
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="your-api-key-here"
)
# 获取文本嵌入
response = client.embeddings.create(
model="bge-base-en-v1.5",
input=["Hello world", "How are you?"]
)
for data in response.data:
print(f"Embedding维度: {len(data.embedding)}")
print(f"前5个值: {data.embedding[:5]}")
5.5 完整的API服务器配置
# server_config.py
from vllm import AsyncLLMEngine, AsyncEngineArgs
from vllm.entrypoints.openai.api_server import run_server
from vllm.entrypoints.openai.serving_chat import OpenAIServingChat
import asyncio
# 自定义API服务器配置
engine_args = AsyncEngineArgs(
model="meta-llama/Llama-2-7b-chat-hf",
dtype="float16",
max_model_len=4096,
gpu_memory_utilization=0.9,
tensor_parallel_size=1,
quantization=None,
enforce_eager=False, # 使用CUDA Graph优化
max_context_len_to_capture=4096,
)
# 异步启动服务
async def main():
engine = AsyncLLMEngine.from_engine_args(engine_args)
await run_server(engine, host="0.0.0.0", port=8000)
if __name__ == "__main__":
asyncio.run(main())
六、批处理与调度策略优化
6.1 动态批处理
vLLM使用动态批处理（Continuous Batching）技术，不同于传统的静态批处理：
# 传统静态批处理
# - 等待一批请求全部完成才处理下一批
# - 短请求需要等待长请求完成
# - GPU利用率低
# vLLM动态批处理
# - 请求可以随时加入或离开批次
# - 新请求不必等待当前批次完成
# - GPU始终保持高利用率
6.2 调度策略配置
from vllm import LLM, SamplingParams
# 配置调度相关参数
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
max_num_batched_tokens=8192, # 每批次最大token数
max_num_seqs=256, # 最大并发序列数
max_model_len=4096, # 模型最大序列长度
gpu_memory_utilization=0.9, # GPU内存利用率
swap_space=4, # CPU交换空间(GB)
)
6.3 优先级调度
# 通过API设置请求优先级
import requests
response = requests.post(
"http://localhost:8000/v1/chat/completions",
json={
"model": "llama-7b",
"messages": [{"role": "user", "content": "紧急请求"}],
"priority": 1, # 优先级: 0(最低) - 9(最高)
}
)
6.4 前缀缓存
vLLM支持自动前缀缓存，当多个请求共享相同的Prompt前缀时，可以复用计算结果：
# 启用前缀缓存
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
enable_prefix_caching=True, # 启用前缀缓存
)
# 多个请求共享系统Prompt
system_prompt = "你是一个专业的法律顾问，精通中国法律体系。请用专业但易懂的语言回答用户的问题。"
# 请求1和请求2会共享system_prompt的计算结果
prompts = [
f"{system_prompt}\n\n用户：什么是合同法？",
f"{system_prompt}\n\n用户：什么是知识产权？",
]
6.5 批处理性能调优
# 性能调优参数
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
# 批处理相关
max_num_batched_tokens=16384, # 增大批次token数
max_num_seqs=512, # 增大并发数
# 内存相关
gpu_memory_utilization=0.95, # 提高GPU内存利用率
swap_space=8, # 增加交换空间
# 计算优化
enforce_eager=False, # 使用CUDA Graph
max_context_len_to_capture=8192, # CUDA Graph捕获的最大上下文
)
七、Tensor并行与Pipeline并行
7.1 Tensor并行
Tensor并行是将模型的每一层切分到多个GPU上，每个GPU计算该层的一部分。
# 2卡Tensor并行
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-2-70b-chat-hf \
--tensor-parallel-size 2 \
--host 0.0.0.0 \
--port 8000
# 4卡Tensor并行
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-2-70b-chat-hf \
--tensor-parallel-size 4 \
--host 0.0.0.0 \
--port 8000
# Python API
from vllm import LLM
llm = LLM(
model="meta-llama/Llama-2-70b-chat-hf",
tensor_parallel_size=4, # 使用4个GPU
)
Tensor并行的特点：
- 每层的计算被切分到多个GPU
- 需要高带宽的GPU间通信（NVLink最佳）
- 适合单机多卡场景
- 延迟最低，因为每层的计算都是并行的
7.2 Pipeline并行
Pipeline并行是将模型的不同层分配到不同GPU上。
# Pipeline并行（vLLM部分版本支持）
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-2-70b-chat-hf \
--pipeline-parallel-size 2 \
--host 0.0.0.0 \
--port 8000
Pipeline并行的特点：
- 不同层在不同GPU上顺序执行
- 通信需求较低
- 适合跨机器场景
- 可能有pipeline bubble导致的效率损失
7.3 多节点部署
# 节点1 (主节点)
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-2-70b-chat-hf \
--tensor-parallel-size 8 \
--distributed-executor-backend ray \
--host 0.0.0.0 \
--port 8000
# 或者使用Ray集群
# 先启动Ray集群
ray start --head --port=6379
# 在其他节点加入集群
ray start --address=<head-node-ip>:6379
# 然后启动vLLM
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-2-70b-chat-hf \
--tensor-parallel-size 8 \
--distributed-executor-backend ray \
--host 0.0.0.0 \
--port 8000
7.4 并行策略选择
# 选择指南
# 场景1: 单GPU，模型能放入显存
llm = LLM(model="meta-llama/Llama-2-7b-chat-hf") # 无需并行
# 场景2: 多GPU，模型太大无法放入单卡
llm = LLM(
model="meta-llama/Llama-2-70b-chat-hf",
tensor_parallel_size=4, # 4卡Tensor并行
)
# 场景3: 多机多卡
llm = LLM(
model="meta-llama/Llama-2-70b-chat-hf",
tensor_parallel_size=8,
distributed_executor_backend="ray",
)
八、LoRA动态加载与多模型服务
8.1 LoRA基础
LoRA（Low-Rank Adaptation）是一种高效的模型微调方法，通过在原始模型权重旁边添加低秩矩阵来实现微调，而不修改原始权重。
# 加载基础模型 + LoRA
from vllm import LLM, SamplingParams
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
enable_lora=True, # 启用LoRA支持
max_lora_rank=64, # 最大LoRA秩
max_loras=4, # 同时加载的最大LoRA数量
)
# 使用LoRA进行推理
from vllm.lora.request import LoRARequest
lora_request = LoRARequest(
lora_name="my_lora",
lora_int_id=1,
lora_path="/path/to/my_lora_adapter",
)
sampling_params = SamplingParams(temperature=0.7, max_tokens=512)
outputs = llm.generate(
["请用专业术语解释量子纠缠"],
sampling_params,
lora_request=lora_request,
)
print(outputs[0].outputs[0].text)
8.2 动态LoRA加载
vLLM支持在运行时动态加载和切换LoRA适配器，无需重启服务：
# API服务器启动时启用LoRA
# python -m vllm.entrypoints.openai.api_server \
# --model meta-llama/Llama-2-7b-chat-hf \
# --enable-lora \
# --max-lora-rank 64 \
# --max-loras 4 \
# --lora-modules lora1=/path/to/lora1 lora2=/path/to/lora2
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="token-abc")
# 使用不同的LoRA适配器
response1 = client.chat.completions.create(
model="meta-llama/Llama-2-7b-chat-hf",
messages=[{"role": "user", "content": "什么是深度学习？"}],
extra_body={"lora_request": {"lora_name": "lora1", "lora_int_id": 1}},
)
response2 = client.chat.completions.create(
model="meta-llama/Llama-2-7b-chat-hf",
messages=[{"role": "user", "content": "什么是深度学习？"}],
extra_body={"lora_request": {"lora_name": "lora2", "lora_int_id": 2}},
)
8.3 多LoRA服务架构
# 多LoRA服务配置
# 启动服务时预加载多个LoRA
# python -m vllm.entrypoints.openai.api_server \
# --model meta-llama/Llama-2-7b-chat-hf \
# --enable-lora \
# --max-lora-rank 128 \
# --max-loras 8 \
# --lora-modules \
# medical=/path/to/medical_lora \
# legal=/path/to/legal_lora \
# code=/path/to/code_lora \
# finance=/path/to/finance_lora
# 使用示例：根据用户类型选择不同的LoRA
def get_lora_for_user(user_type: str):
"""根据用户类型选择对应的LoRA适配器"""
lora_map = {
"medical": {"lora_name": "medical", "lora_int_id": 1},
"legal": {"lora_name": "legal", "lora_int_id": 2},
"code": {"lora_name": "code", "lora_int_id": 3},
"finance": {"lora_name": "finance", "lora_int_id": 4},
}
return lora_map.get(user_type, None)
# 路由请求到对应的LoRA
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="token-abc")
user_type = "medical"
query = "什么是高血压的诊断标准？"
response = client.chat.completions.create(
model="meta-llama/Llama-2-7b-chat-hf",
messages=[{"role": "user", "content": query}],
extra_body={"lora_request": get_lora_for_user(user_type)},
)
8.4 LoRA性能优化
# LoRA相关的性能优化参数
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
enable_lora=True,
max_lora_rank=64,
max_loras=4,
# LoRA计算优化
fully_sharded_loras=False, # 是否全分片LoRA
# 内存优化
max_cpu_loras=8, # CPU上缓存的LoRA数量
)
九、基准测试与性能调优
9.1 使用benchmark工具
vLLM内置了基准测试工具：
# 吞吐量基准测试
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-2-7b-chat-hf \
--host 0.0.0.0 --port 8000 &
# 等待服务启动
sleep 30
# 运行基准测试
python -m vllm.entrypoints.openai.bench \
--backend openai-chat \
--base-url http://localhost:8000/v1 \
--model meta-llama/Llama-2-7b-chat-hf \
--dataset-path ShareGPT_V3_unfiltered_cleaned_split.json \
--num-prompts 1000 \
--request-rate 10 \
--max-concurrency 100
9.2 自定义基准测试脚本
# benchmark.py
import time
import asyncio
import statistics
from openai import AsyncOpenAI
from dataclasses import dataclass
from typing import List
@dataclass
class BenchmarkResult:
total_requests: int
successful_requests: int
failed_requests: int
total_time: float
requests_per_second: float
avg_latency: float
p50_latency: float
p95_latency: float
p99_latency: float
avg_tokens_per_second: float
async def benchmark_concurrent(
base_url: str,
model: str,
prompts: List[str],
concurrency: int,
api_key: str = "token-abc"
) -> BenchmarkResult:
"""并发基准测试"""
client = AsyncOpenAI(base_url=base_url, api_key=api_key)
latencies = []
tokens_per_sec = []
successful = 0
failed = 0
semaphore = asyncio.Semaphore(concurrency)
async def single_request(prompt: str):
nonlocal successful, failed
async with semaphore:
try:
start = time.time()
response = await client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
max_tokens=256,
temperature=0.7,
)
latency = time.time() - start
latencies.append(latency)
# 计算tokens/s
tokens = len(response.choices[0].message.content) // 4 # 粗略估计
tokens_per_sec.append(tokens / latency)
successful += 1
except Exception as e:
failed += 1
print(f"Error: {e}")
total_start = time.time()
tasks = [single_request(p) for p in prompts]
await asyncio.gather(*tasks)
total_time = time.time() - total_start
return BenchmarkResult(
total_requests=len(prompts),
successful_requests=successful,
failed_requests=failed,
total_time=total_time,
requests_per_second=successful / total_time,
avg_latency=statistics.mean(latencies) if latencies else 0,
p50_latency=statistics.median(latencies) if latencies else 0,
p95_latency=sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0,
p99_latency=sorted(latencies)[int(len(latencies) * 0.99)] if latencies else 0,
avg_tokens_per_second=statistics.mean(tokens_per_sec) if tokens_per_sec else 0,
)
# 运行基准测试
async def main():
prompts = [
"请解释什么是机器学习",
"写一首关于春天的诗",
"Python和Java有什么区别？",
"如何学习深度学习？",
"什么是大语言模型？",
] * 200 # 1000个请求
result = await benchmark_concurrent(
base_url="http://localhost:8000/v1",
model="llama-7b",
prompts=prompts,
concurrency=50,
)
print(f"总请求数: {result.total_requests}")
print(f"成功请求数: {result.successful_requests}")
print(f"失败请求数: {result.failed_requests}")
print(f"总耗时: {result.total_time:.2f}s")
print(f"QPS: {result.requests_per_second:.2f}")
print(f"平均延迟: {result.avg_latency:.3f}s")
print(f"P50延迟: {result.p50_latency:.3f}s")
print(f"P95延迟: {result.p95_latency:.3f}s")
print(f"P99延迟: {result.p99_latency:.3f}s")
print(f"平均生成速度: {result.avg_tokens_per_second:.2f} tokens/s")
if __name__ == "__main__":
asyncio.run(main())
9.3 性能调优建议
# 性能调优清单
# 1. 启用CUDA Graph
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
enforce_eager=False, # 禁用eager模式，启用CUDA Graph
)
# 2. 调整GPU内存利用率
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
gpu_memory_utilization=0.95, # 提高到95%
)
# 3. 启用前缀缓存
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
enable_prefix_caching=True,
)
# 4. 使用合适的量化
llm = LLM(
model="TheBloke/Llama-2-7B-Chat-AWQ",
quantization="awq",
)
# 5. 调整批处理参数
llm = LLM(
model="meta-llama/Llama-2-7b-chat-hf",
max_num_batched_tokens=16384, # 增大批次大小
max_num_seqs=512, # 增大并发数
)
# 6. 使用Tensor并行（多GPU）
llm = LLM(
model="meta-llama/Llama-2-70b-chat-hf",
tensor_parallel_size=4,
)
十、生产部署：Docker、Kubernetes、负载均衡
10.1 Docker部署
Dockerfile：
FROM vllm/vllm-openai:latest
# 设置工作目录
WORKDIR /app
# 复制配置文件
COPY config.yaml /app/config.yaml
# 设置环境变量
ENV MODEL_NAME=meta-llama/Llama-2-7b-chat-hf
ENV HOST=0.0.0.0
ENV PORT=8000
# 暴露端口
EXPOSE 8000
# 启动命令
CMD ["python", "-m", "vllm.entrypoints.openai.api_server", \
"--model", "${MODEL_NAME}", \
"--host", "${HOST}", \
"--port", "${PORT}", \
"--dtype", "float16", \
"--gpu-memory-utilization", "0.9"]
docker-compose.yml：
version: '3.8'
services:
vllm:
image: vllm/vllm-openai:latest
runtime: nvidia
environment:
- NVIDIA_VISIBLE_DEVICES=all
- MODEL_NAME=meta-llama/Llama-2-7b-chat-hf
- HF_HOME=/root/.cache/huggingface
volumes:
- ./models:/root/.cache/huggingface
ports:
- "8000:8000"
ipc: host
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: all
capabilities: [gpu]
command: >
python -m vllm.entrypoints.openai.api_server
--model meta-llama/Llama-2-7b-chat-hf
--host 0.0.0.0
--port 8000
--dtype float16
--gpu-memory-utilization 0.9
--max-model-len 4096
nginx:
image: nginx:latest
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
depends_on:
- vllm
10.2 Kubernetes部署
deployment.yaml：
apiVersion: apps/v1
kind: Deployment
metadata:
name: vllm-server
labels:
app: vllm-server
spec:
replicas: 2
selector:
matchLabels:
app: vllm-server
template:
metadata:
labels:
app: vllm-server
spec:
containers:
- name: vllm
image: vllm/vllm-openai:latest
ports:
- containerPort: 8000
env:
- name: MODEL_NAME
value: "meta-llama/Llama-2-7b-chat-hf"
resources:
limits:
nvidia.com/gpu: 1
memory: "32Gi"
cpu: "8"
requests:
nvidia.com/gpu: 1
memory: "16Gi"
cpu: "4"
volumeMounts:
- name: model-cache
mountPath: /root/.cache/huggingface
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 60
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 60
periodSeconds: 10
command: ["python", "-m", "vllm.entrypoints.openai.api_server"]
args:
- "--model"
- "$(MODEL_NAME)"
- "--host"
- "0.0.0.0"
- "--port"
- "8000"
- "--dtype"
- "float16"
- "--gpu-memory-utilization"
- "0.9"
volumes:
- name: model-cache
persistentVolumeClaim:
claimName: model-pvc
---
apiVersion: v1
kind: Service
metadata:
name: vllm-service
spec:
selector:
app: vllm-server
ports:
- port: 8000
targetPort: 8000
type: ClusterIP
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: model-pvc
spec:
accessModes:
- ReadWriteOnce
resources:
requests:
storage: 100Gi
storageClassName: gp3
ingress.yaml：
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: vllm-ingress
annotations:
nginx.ingress.kubernetes.io/proxy-body-size: "10m"
nginx.ingress.kubernetes.io/proxy-read-timeout: "300"
spec:
rules:
- host: llm.example.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: vllm-service
port:
number: 8000
tls:
- hosts:
- llm.example.com
secretName: tls-secret
10.3 Nginx负载均衡
# nginx.conf
upstream vllm_backend {
least_conn; # 使用最少连接算法
server vllm-1:8000 weight=1;
server vllm-2:8000 weight=1;
server vllm-3:8000 weight=1;
keepalive 32; # 保持连接池
}
server {
listen 80;
server_name llm.example.com;
client_max_body_size 10m;
# 代理超时设置（LLM推理可能需要较长时间）
proxy_connect_timeout 60s;
proxy_send_timeout 300s;
proxy_read_timeout 300s;
# 请求限流
limit_req_zone $binary_remote_addr zone=api:10m rate=10r/s;
location /v1/ {
limit_req zone=api burst=20 nodelay;
proxy_pass http://vllm_backend;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
# 支持SSE流式输出
proxy_buffering off;
proxy_cache off;
# HTTP/1.1支持
proxy_http_version 1.1;
proxy_set_header Connection "";
}
location /health {
proxy_pass http://vllm_backend/health;
}
}
10.4 监控与告警
# Prometheus监控配置
# prometheus.yml
scrape_configs:
- job_name: 'vllm'
static_configs:
- targets: ['vllm-1:8000', 'vllm-2:8000']
metrics_path: /metrics
# 自定义监控指标
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
# 定义指标
REQUEST_COUNT = Counter(
'vllm_request_total',
'Total number of requests',
['method', 'endpoint', 'status']
)
REQUEST_LATENCY = Histogram(
'vllm_request_duration_seconds',
'Request latency in seconds',
['method', 'endpoint']
)
ACTIVE_REQUESTS = Gauge(
'vllm_active_requests',
'Number of active requests'
)
TOKENS_GENERATED = Counter(
'vllm_tokens_generated_total',
'Total tokens generated'
)
# 中间件示例
@app.middleware("http")
async def monitor_requests(request, call_next):
ACTIVE_REQUESTS.inc()
start_time = time.time()
response = await call_next(request)
latency = time.time() - start_time
REQUEST_COUNT.labels(
method=request.method,
endpoint=request.url.path,
status=response.status_code
).inc()
REQUEST_LATENCY.labels(
method=request.method,
endpoint=request.url.path
).observe(latency)
ACTIVE_REQUESTS.dec()
return response
十一、实战项目：搭建企业级LLM推理服务
11.1 项目概述
我们将搭建一个完整的企业级LLM推理服务，包括：
- 多模型支持：同时部署多个模型，支持动态路由
- LoRA热切换：支持在运行时切换不同的LoRA适配器
- 智能负载均衡：根据GPU负载动态分配请求
- 完整的监控系统：Prometheus + Grafana监控
- 高可用设计：自动故障转移和恢复
- API网关：统一的API入口，支持认证、限流、日志
11.2 项目结构
enterprise-llm-service/
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
├── config/
│ ├── models.yaml # 模型配置
│ ├── lora_adapters.yaml # LoRA适配器配置
│ └── nginx.conf # Nginx配置
├── src/
│ ├── api/
│ │ ├── __init__.py
│ │ ├── main.py # FastAPI应用
│ │ ├── routes/
│ │ │ ├── chat.py # 聊天API路由
│ │ │ ├── embeddings.py # Embedding API路由
│ │ │ └── models.py # 模型管理路由
│ │ └── middleware/
│ │ ├── auth.py # 认证中间件
│ │ ├── rate_limit.py # 限流中间件
│ │ └── logging.py # 日志中间件
│ ├── core/
│ │ ├── __init__.py
│ │ ├── engine.py # vLLM引擎管理
│ │ ├── router.py # 请求路由器
│ │ └── lora_manager.py # LoRA管理器
│ ├── monitoring/
│ │ ├── __init__.py
│ │ ├── metrics.py # 指标收集
│ │ └── health.py # 健康检查
│ └── utils/
│ ├── __init__.py
│ ├── config.py # 配置加载
│ └── logger.py # 日志配置
├── tests/
│ ├── test_api.py
│ ├── test_engine.py
│ └── test_router.py
├── monitoring/
│ ├── prometheus.yml
│ └── grafana/
│ └── dashboards/
│ └── vllm.json
└── scripts/
├── deploy.sh
├── benchmark.sh
└── health_check.sh
11.3 核心代码实现
models.yaml：
models:
- name: "llama-7b"
path: "meta-llama/Llama-2-7b-chat-hf"
dtype: "float16"
max_model_len: 4096
gpu_memory_utilization: 0.9
tensor_parallel_size: 1
enable_lora: true
max_loras: 4
max_lora_rank: 64
served_name: "llama-7b"
- name: "llama-13b"
path: "meta-llama/Llama-2-13b-chat-hf"
dtype: "float16"
max_model_len: 4096
gpu_memory_utilization: 0.9
tensor_parallel_size: 2
served_name: "llama-13b"
- name: "qwen-7b"
path: "Qwen/Qwen-7B-Chat"
dtype: "float16"
max_model_len: 8192
gpu_memory_utilization: 0.9
tensor_parallel_size: 1
served_name: "qwen-7b"
lora_adapters:
- name: "medical"
path: "/models/lora/medical"
base_model: "llama-7b"
- name: "legal"
path: "/models/lora/legal"
base_model: "llama-7b"
- name: "code"
path: "/models/lora/code"
base_model: "llama-7b"
engine.py（vLLM引擎管理）：
"""
vLLM引擎管理器
负责管理多个vLLM引擎实例的生命周期
"""
import asyncio
from typing import Dict, Optional, List
from vllm import AsyncLLMEngine, AsyncEngineArgs
from vllm.lora.request import LoRARequest
import yaml
import logging
logger = logging.getLogger(__name__)
class EngineManager:
"""vLLM引擎管理器"""
def __init__(self, config_path: str):
self.config = self._load_config(config_path)
self.engines: Dict[str, AsyncLLMEngine] = {}
self.lora_adapters: Dict[str, LoRARequest] = {}
def _load_config(self, path: str) -> dict:
with open(path, 'r') as f:
return yaml.safe_load(f)
async def initialize_engines(self):
"""初始化所有配置的引擎"""
for model_config in self.config['models']:
model_name = model_config['name']
logger.info(f"Initializing engine for {model_name}")
engine_args = AsyncEngineArgs(
model=model_config['path'],
dtype=model_config.get('dtype', 'float16'),
max_model_len=model_config.get('max_model_len', 4096),
gpu_memory_utilization=model_config.get('gpu_memory_utilization', 0.9),
tensor_parallel_size=model_config.get('tensor_parallel_size', 1),
enable_lora=model_config.get('enable_lora', False),
max_loras=model_config.get('max_loras', 4),
max_lora_rank=model_config.get('max_lora_rank', 64),
enforce_eager=False,
)
engine = AsyncLLMEngine.from_engine_args(engine_args)
self.engines[model_name] = engine
logger.info(f"Engine {model_name} initialized successfully")
# 加载LoRA适配器
for lora_config in self.config.get('lora_adapters', []):
lora_name = lora_config['name']
lora_path = lora_config['path']
self.lora_adapters[lora_name] = LoRARequest(
lora_name=lora_name,
lora_int_id=len(self.lora_adapters) + 1,
lora_path=lora_path,
)
logger.info(f"LoRA adapter {lora_name} loaded")
def get_engine(self, model_name: str) -> Optional[AsyncLLMEngine]:
"""获取指定模型的引擎"""
return self.engines.get(model_name)
def get_lora(self, lora_name: str) -> Optional[LoRARequest]:
"""获取指定的LoRA适配器"""
return self.lora_adapters.get(lora_name)
async def shutdown(self):
"""关闭所有引擎"""
for name, engine in self.engines.items():
logger.info(f"Shutting down engine {name}")
# vLLM的异步引擎会自动清理
self.engines.clear()
self.lora_adapters.clear()
router.py（请求路由器）：
"""
智能请求路由器
根据请求特征选择最优的模型和LoRA适配器
"""
from typing import Optional, Dict, Any
from dataclasses import dataclass
import re
@dataclass
class RoutingDecision:
model_name: str
lora_name: Optional[str] = None
priority: int = 5
estimated_tokens: int = 0
class RequestRouter:
"""请求路由器"""
def __init__(self, engine_manager):
self.engine_manager = engine_manager
self.routing_rules = self._load_routing_rules()
def _load_routing_rules(self) -> Dict[str, Any]:
"""加载路由规则"""
return {
"medical_keywords": ["疾病", "症状", "药物", "治疗", "诊断", "手术"],
"legal_keywords": ["法律", "合同", "诉讼", "法规", "判决", "律师"],
"code_keywords": ["代码", "编程", "函数", "算法", "bug", "调试"],
"model_preferences": {
"default": "llama-7b",
"complex": "llama-13b",
"chinese": "qwen-7b",
}
}
def route(self, request: Dict[str, Any]) -> RoutingDecision:
"""根据请求内容决定路由"""
messages = request.get("messages", [])
user_message = ""
for msg in messages:
if msg.get("role") == "user":
user_message += msg.get("content", "") + " "
# 检测是否需要LoRA
lora_name = self._detect_lora_need(user_message)
# 选择模型
model_name = self._select_model(user_message, request)
# 估算token数
estimated_tokens = len(user_message) * 2 # 粗略估计
return RoutingDecision(
model_name=model_name,
lora_name=lora_name,
priority=request.get("priority", 5),
estimated_tokens=estimated_tokens,
)
def _detect_lora_need(self, text: str) -> Optional[str]:
"""检测是否需要特定的LoRA适配器"""
text_lower = text.lower()
for keyword in self.routing_rules.get("medical_keywords", []):
if keyword in text_lower:
return "medical"
for keyword in self.routing_rules.get("legal_keywords", []):
if keyword in text_lower:
return "legal"
for keyword in self.routing_rules.get("code_keywords", []):
if keyword in text_lower:
return "code"
return None
def _select_model(self, text: str, request: Dict) -> str:
"""选择最合适的模型"""
# 如果用户指定了模型
if "model" in request:
requested_model = request["model"]
if requested_model in self.engine_manager.engines:
return requested_model
# 根据文本长度选择模型
if len(text) > 2000:
return self.routing_rules["model_preferences"]["complex"]
# 检测是否主要是中文
chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
if chinese_chars / max(len(text), 1) > 0.5:
return self.routing_rules["model_preferences"]["chinese"]
return self.routing_rules["model_preferences"]["default"]
main.py（FastAPI应用）：
"""
企业级LLM推理服务 - FastAPI应用
"""
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from contextlib import asynccontextmanager
import time
import uuid
import logging
from typing import Optional, List, Dict, Any
from .core.engine import EngineManager
from .core.router import RequestRouter
from .monitoring.metrics import MetricsCollector
from .middleware.auth import verify_api_key
from .middleware.rate_limit import RateLimiter
logger = logging.getLogger(__name__)
# 全局实例
engine_manager: Optional[EngineManager] = None
router: Optional[RequestRouter] = None
metrics: Optional[MetricsCollector] = None
rate_limiter: Optional[RateLimiter] = None
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理"""
global engine_manager, router, metrics, rate_limiter
logger.info("Starting LLM inference service...")
# 初始化引擎管理器
engine_manager = EngineManager("config/models.yaml")
await engine_manager.initialize_engines()
# 初始化路由器
router = RequestRouter(engine_manager)
# 初始化监控
metrics = MetricsCollector()
# 初始化限流器
rate_limiter = RateLimiter(max_requests_per_minute=60)
logger.info("Service started successfully")
yield
# 清理
logger.info("Shutting down service...")
await engine_manager.shutdown()
app = FastAPI(
title="Enterprise LLM Inference Service",
description="High-performance LLM inference service powered by vLLM",
version="1.0.0",
lifespan=lifespan,
)
# CORS中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.post("/v1/chat/completions")
async def chat_completions(
request: Request,
api_key: str = Depends(verify_api_key),
):
"""聊天完成API"""
request_id = str(uuid.uuid4())
start_time = time.time()
# 解析请求
body = await request.json()
# 路由决策
routing_decision = router.route(body)
model_name = routing_decision.model_name
lora_name = routing_decision.lora_name
# 获取引擎
engine = engine_manager.get_engine(model_name)
if not engine:
raise HTTPException(status_code=404, detail=f"Model {model_name} not found")
# 获取LoRA
lora_request = None
if lora_name:
lora_request = engine_manager.get_lora(lora_name)
# 构建采样参数
from vllm import SamplingParams
sampling_params = SamplingParams(
temperature=body.get("temperature", 0.7),
top_p=body.get("top_p", 0.9),
max_tokens=body.get("max_tokens", 512),
presence_penalty=body.get("presence_penalty", 0.0),
frequency_penalty=body.get("frequency_penalty", 0.0),
)
# 提取消息
messages = body.get("messages", [])
prompt = "\n".join([f"{m['role']}: {m['content']}" for m in messages])
# 流式输出
if body.get("stream", False):
async def generate_stream():
async for output in engine.generate(
prompt,
sampling_params,
request_id,
lora_request=lora_request,
):
chunk = {
"id": request_id,
"object": "chat.completion.chunk",
"choices": [{
"index": 0,
"delta": {"content": output.outputs[0].text},
"finish_reason": None,
}]
}
yield f"data: {json.dumps(chunk)}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate_stream(), media_type="text/event-stream")
# 非流式输出
else:
outputs = await engine.generate(
prompt,
sampling_params,
request_id,
lora_request=lora_request,
)
generated_text = outputs[0].outputs[0].text
# 记录指标
latency = time.time() - start_time
metrics.record_request(model_name, latency, len(generated_text))
return {
"id": request_id,
"object": "chat.completion",
"created": int(time.time()),
"model": model_name,
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": generated_text,
},
"finish_reason": "stop",
}],
"usage": {
"prompt_tokens": len(prompt) // 4,
"completion_tokens": len(generated_text) // 4,
"total_tokens": (len(prompt) + len(generated_text)) // 4,
}
}
@app.get("/v1/models")
async def list_models(api_key: str = Depends(verify_api_key)):
"""列出可用模型"""
models = []
for name, engine in engine_manager.engines.items():
models.append({
"id": name,
"object": "model",
"created": int(time.time()),
"owned_by": "organization",
})
return {"data": models}
@app.get("/health")
async def health_check():
"""健康检查"""
return {"status": "healthy", "timestamp": int(time.time())}
@app.get("/metrics")
async def get_metrics():
"""获取Prometheus指标"""
from prometheus_client import generate_latest
return Response(generate_latest(), media_type="text/plain")
11.4 部署脚本
deploy.sh：
#!/bin/bash
set -e
echo "=== Enterprise LLM Service Deployment ==="
# 检查Docker
if ! command -v docker &> /dev/null; then
echo "Error: Docker is not installed"
exit 1
fi
# 检查NVIDIA Docker
if ! docker info 2>/dev/null | grep -q "Runtimes.*nvidia"; then
echo "Error: NVIDIA Docker runtime is not installed"
exit 1
fi
# 检查GPU
echo "Checking GPU status..."
nvidia-smi
# 构建镜像
echo "Building Docker image..."
docker build -t enterprise-llm-service .
# 启动服务
echo "Starting services..."
docker-compose up -d
# 等待服务启动
echo "Waiting for service to start..."
sleep 30
# 健康检查
echo "Running health check..."
for i in {1..10}; do
if curl -s http://localhost:8000/health | grep -q "healthy"; then
echo "Service is healthy!"
break
fi
echo "Waiting... ($i/10)"
sleep 10
done
echo "=== Deployment Complete ==="
echo "Service URL: http://localhost:8000"
echo "API Docs: http://localhost:8000/docs"
echo "Metrics: http://localhost:8000/metrics"
11.5 测试脚本
# test_service.py
import requests
import json
import time
BASE_URL = "http://localhost:8000"
API_KEY = "your-api-key-here"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
def test_chat_completion():
"""测试聊天完成API"""
print("Testing chat completion...")
response = requests.post(
f"{BASE_URL}/v1/chat/completions",
headers=headers,
json={
"model": "llama-7b",
"messages": [
{"role": "system", "content": "你是一个有帮助的AI助手。"},
{"role": "user", "content": "请用简单的话解释什么是量子计算？"}
],
"temperature": 0.7,
"max_tokens": 512,
}
)
if response.status_code == 200:
result = response.json()
print(f"Response: {result['choices'][0]['message']['content'][:200]}...")
print("✓ Chat completion test passed")
else:
print(f"✗ Chat completion test failed: {response.status_code}")
print(response.text)
def test_streaming():
"""测试流式输出"""
print("\nTesting streaming...")
response = requests.post(
f"{BASE_URL}/v1/chat/completions",
headers=headers,
json={
"model": "llama-7b",
"messages": [
{"role": "user", "content": "写一首关于AI的短诗"}
],
"stream": True,
},
stream=True,
)
if response.status_code == 200:
print("Stream output: ", end="")
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: ') and line != 'data: [DONE]':
data = json.loads(line[6:])
content = data['choices'][0]['delta'].get('content', '')
print(content, end="", flush=True)
print()
print("✓ Streaming test passed")
else:
print(f"✗ Streaming test failed: {response.status_code}")
def test_lora_routing():
"""测试LoRA路由"""
print("\nTesting LoRA routing...")
# 医疗相关查询
response = requests.post(
f"{BASE_URL}/v1/chat/completions",
headers=headers,
json={
"model": "llama-7b",
"messages": [
{"role": "user", "content": "什么是高血压的诊断标准？"}
],
}
)
if response.status_code == 200:
result = response.json()
print(f"Medical query response: {result['choices'][0]['message']['content'][:100]}...")
print("✓ LoRA routing test passed")
else:
print(f"✗ LoRA routing test failed: {response.status_code}")
def test_health_check():
"""测试健康检查"""
print("\nTesting health check...")
response = requests.get(f"{BASE_URL}/health")
if response.status_code == 200:
print(f"Health: {response.json()}")
print("✓ Health check test passed")
else:
print(f"✗ Health check test failed: {response.status_code}")
def test_list_models():
"""测试模型列表"""
print("\nTesting list models...")
response = requests.get(f"{BASE_URL}/v1/models", headers=headers)
if response.status_code == 200:
models = response.json()
print(f"Available models: {[m['id'] for m in models['data']]}")
print("✓ List models test passed")
else:
print(f"✗ List models test failed: {response.status_code}")
if __name__ == "__main__":
print("=== Enterprise LLM Service Tests ===\n")
test_health_check()
test_list_models()
test_chat_completion()
test_streaming()
test_lora_routing()
print("\n=== All tests completed ===")
总结
vLLM作为当前最先进的LLM推理引擎，通过PagedAttention等创新技术，极大地提升了大模型推理的效率和经济性。本教程从原理到实践，系统地介绍了vLLM的方方面面。
关键要点回顾：
- PagedAttention是核心：理解其分页管理KV Cache的原理，是理解vLLM性能优势的基础
- 量化是成本利器：合理使用GPTQ/AWQ/FP8量化，可以在不显著损失精度的前提下大幅降低硬件成本
- API兼容性很重要：OpenAI兼容API让现有应用可以无缝迁移
- 并行策略要因地制宜：根据模型大小和硬件配置选择合适的Tensor/Pipeline并行方案
- LoRA实现了多租户：动态LoRA加载让一个基础模型可以服务多个专业领域
- 生产部署要全面：Docker/K8s部署、负载均衡、监控告警，缺一不可
随着大模型应用的持续爆发，高性能推理将成为AI基础设施的关键一环。掌握vLLM的部署和优化技能，将帮助你在AI工程化浪潮中占据先机。
本教程由 AI 教程助手生成，内容基于vLLM框架的核心设计理念和工程实践经验编写。
