第1章:vLLM基础入门 本章导读:通过本章学习,读者将掌握vLLM的核心概念、安装方法和基础使用,为后续深入学习打下坚实基础。从理论到实践,全面了解vLLM如何实现高性能LLM推理。 学习目标 深入理解vLLM的架构价值与核心优势 熟练掌握vLLM的环境准备、安装流程和配置选项 能够编写生产级别的vLLM推理代码和API服务 全面了解vLLM与传统LLM框架的性能差异 掌握vLLM在不同场景下的最佳实践 核心概念 vLLM(Very Large Language Model serving)是一个专为高效LLM推理而设计的开源服务引擎。它由UC Berkeley Sky Computing Lab于2023年首次提出,通过创新的内存管理和调度算法,实现了传统方案数倍的性能提升。
本章导读:通过本章学习,读者将掌握vLLM的核心概念、安装方法和基础使用,为后续深入学习打下坚实基础。从理论到实践,全面了解vLLM如何实现高性能LLM推理。
vLLM(Very Large Language Model serving)是一个专为高效LLM推理而设计的开源服务引擎。它由UC Berkeley Sky Computing Lab于2023年首次提出,通过创新的内存管理和调度算法,实现了传统方案数倍的性能提升。其核心价值体现在:更高的吞吐量、更低的内存占用、更智能的批处理、更灵活的扩展性。
vLLM填补了传统LLM推理方案与生产级服务之间的鸿沟。相比Hugging Face Transformers的原始实现,vLLM通过以下技术创新实现了质的飞跃:
# 安装uv包管理器 curl -LsSf https://astral.sh/uv/install.sh | sh # 使用uv安装vLLM(推荐) uv pip install vllm --torch-backend auto # 指定CUDA版本安装 uv pip install vllm --torch-backend cuda --index-url https://download.pytorch.org/whl/cu121
# CPU版本安装 pip install vllm # GPU版本安装(需预先安装PyTorch + CUDA) pip install vllm torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121 # 开发版本安装 pip install vllm --pre --extra-index-url https://pypi.org/simple/
# 验证安装 python -c "import vllm; print(vllm.__version__)" # 测试GPU可用性 python -c "import torch; print(f'CUDA available: {torch.cuda.is_available()}')"
| 版本类型 | 适用场景 | 优势 | 劣势 |
|---|---|---|---|
| Stable | 生产环境 | 稳定可靠 | 功能相对固定 |
| Nightly | 开发测试 | 最新功能 | 可能存在bug |
| CPU | 无GPU环境 | 部署简单 | 性能受限 |
| Source | 定制需求 | 完全控制 | 编译复杂 |
#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 基础vLLM推理示例 演示vLLM的核心功能和基本使用方法 """ import time from vllm import LLM, SamplingParams import torch def initialize_model(): """初始化vLLM模型""" print("=== 初始化vLLM模型 ===") # 基础初始化 llm = LLM( model="Qwen/Qwen1.5-7B-Chat", tensor_parallel_size=1, # 单GPU gpu_memory_utilization=0.9, # GPU内存利用率 max_model_len=8192, # 最大序列长度 max_num_batched_tokens=8192 # 最大批处理token数 ) print(f"模型初始化完成") print(f"可用GPU数量: {torch.cuda.device_count()}") print(f"当前GPU: {torch.cuda.get_device_name(0) if torch.cuda.is_available() else 'CPU'}") return llm def run_inference(llm): """执行推理""" print("\n=== 执行推理 ===") # 配置采样参数 sampling_params = SamplingParams( temperature=0.8, # 温度控制 top_p=0.95, # 核心采样概率 max_tokens=1000, # 最大生成长度 repetition_penalty=1.1 # 重复惩罚 ) # 测试提示词 test_prompts = [ "请详细解释vLLM的核心技术创新点,以及它相比传统LLM推理方案的优势。", "在生产环境中部署vLLM时,应该如何选择合适的硬件配置和参数设置?", "vLLM在吞吐量、延迟、内存使用方面有哪些具体的性能数据?" ] results = [] for prompt in test_prompts: print(f"\n--- 推理测试 ---") print(f"Prompt: {prompt[:50]}...") start_time = time.time() outputs = llm.generate([prompt], sampling_params) end_time = time.time() response = outputs[0].outputs[0].text result = { 'response_length': len(response), 'inference_time': end_time - start_time, 'tokens_per_second': len(response.split()) / (end_time - start_time) if response else 0 } results.append(result) print(f"响应长度: {result['response_length']} 字符") print(f"推理时间: {result['inference_time']:.3f} 秒") print(f"处理速度: {result['tokens_per_second']:.1f} tokens/s") print(f"响应预览: {response[:100]}...") return results def main(): print("=== vLLM基础推理完整示例 ===") # 初始化模型 llm = initialize_model() # 执行推理 results = run_inference(llm) # 输出总结 print("\n=== 推理性能总结 ===") avg_time = sum(r['inference_time'] for r in results) / len(results) avg_throughput = sum(r['tokens_per_second'] for r in results) / len(results) print(f"平均推理时间: {avg_time:.3f} 秒") print(f"平均处理速度: {avg_throughput:.1f} tokens/s") print("\n=== 测试完成 ===") if __name__ == "__main__": main()
#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ vLLM OpenAI兼容API服务 提供生产级的HTTP API服务,支持OpenAI兼容接口 """ from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any import uvicorn import time from vllm import LLM, SamplingParams import logging # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Pydantic模型定义 class CompletionRequest(BaseModel): prompt: str = Field(..., description="输入提示词") temperature: float = Field(default=0.7, ge=0.0, le=2.0) top_p: float = Field(default=0.9, ge=0.0, le=1.0) max_tokens: int = Field(default=1000, ge=1, le=4000) stream: bool = Field(default=False) n: int = Field(default=1, ge=1, le=4) stop: Optional[List[str]] = Field(default=None) class CompletionResponse(BaseModel): id: str object: str = "text_completion" created: int model: str choices: List[Dict[str, Any]] usage: Dict[str, int] class VLLMService: """vLLM服务类""" def __init__(self, model_name: str = "Qwen/Qwen1.5-7B-Chat"): self.model_name = model_name self.llm = None self.initialize_model() self.request_count = 0 self.start_time = time.time() def initialize_model(self): """初始化vLLM模型""" logger.info(f"正在初始化模型: {self.model_name}") try: self.llm = LLM( model=self.model_name, tensor_parallel_size=1, gpu_memory_utilization=0.9, max_model_len=8192, max_num_batched_tokens=8192, enable_lora=False, enforce_eager=True ) logger.info("模型初始化成功") except Exception as e: logger.error(f"模型初始化失败: {e}") raise def get_sampling_params(self, request: CompletionRequest) -> SamplingParams: """获取采样参数""" return SamplingParams( temperature=request.temperature, top_p=request.top_p, max_tokens=request.max_tokens, stop=request.stop, repetition_penalty=1.1 ) async def create_completion(self, request: CompletionRequest) -> CompletionResponse: """创建补全响应""" self.request_count += 1 try: # 生成提示词 prompt = request.prompt # 执行推理 start_time = time.time() outputs = self.llm.generate([prompt], self.get_sampling_params(request)) end_time = time.time() # 处理输出 generated_text = outputs[0].outputs[0].text # 构建响应 choice = { "text": generated_text, "index": 0, "logprobs": None, "finish_reason": "stop" } response = CompletionResponse( id=f"cmpl-{self.request_count}", created=int(time.time()), model=self.model_name, choices=[choice], usage={ "prompt_tokens": len(prompt.split()), "completion_tokens": len(generated_text.split()), "total_tokens": len(prompt.split()) + len(generated_text.split()) } ) # 记录日志 logger.info(f"请求 #{self.request_count} 耗时: {end_time - start_time:.3f}s") return response except Exception as e: logger.error(f"推理失败: {e}") raise HTTPException(status_code=500, detail=f"推理失败: {e}") # FastAPI应用 app = FastAPI( title="vLLM OpenAI兼容API", description="提供OpenAI兼容的vLLM推理服务", version="1.0.0" ) # 添加CORS中间件 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 初始化服务 vllm_service = VLLMService() @app.get("/") async def root(): """根路径""" return { "message": "vLLM OpenAI兼容API服务", "version": "1.0.0", "status": "running" } @app.get("/health") async def health_check(): """健康检查""" return { "status": "healthy", "model": vllm_service.model_name, "requests_processed": vllm_service.request_count } @app.post("/v1/completions") async def create_completion(request: CompletionRequest): """创建补全""" return await vllm_service.create_completion(request) if __name__ == "__main__": # 配置启动参数 uvicorn.run( app, host="0.0.0.0", port=8000, workers=1, log_level="info" )
A:vLLM主要优势在于内存效率和推理吞吐量。传统方案每次推理都需要重新计算,而vLLM通过PagedAttention和连续批处理技术,可以复用计算结果,减少70-80%的内存占用,吞吐量提升3-5倍。此外,vLLM还提供了生产级的API服务、动态批处理和智能调度等特性。
A:vLLM支持所有主流LLM模型,包括:
A:选择GPU时需综合考虑以下因素:
本章深入介绍了vLLM的基础知识,从技术概念到实际应用,全面涵盖了vLLM的核心特性和使用方法。通过详细的代码示例和性能测试,读者应该已经掌握了vLLM的基本使用、API服务和性能评估方法。下一章将深入探讨vLLM的核心架构原理,重点关注PagedAttention和连续批处理等关键技术。
关键词:vLLM, PagedAttention, 推理优化, 性能提升, 基础使用
难度:入门
预计阅读:40分钟