vLLM 高吞吐量服务

什么是 vLLM？

vLLM 是一个基于 PagedAttention 的 GPU 推理框架。它通过提供对原始 LLM API 的高级抽象，简化了 AI 应用的构建。

最适合：服务部署

安装

bash
pip install vllm
或使用 uv:
uv add vllm

核心概念

vLLM 围绕几个关键思想构建：

快速开始

python
最小工作示例
import os
os.environ["OPENAI_API_KEY"] = "sk-..."
导入 vLLM
（具体导入方式请参考框架相关文档）
基于 PagedAttention 的 GPU 推理基本使用模式
def create_pipeline():
    """创建用于服务的 vLLM 管道。"""
    # 1. 初始化框架
    # 2. 配置你的 LLM（GPT-4o、Claude 等）
    # 3. 定义管道逻辑
    # 4. 返回配置好的管道
    pass
pipeline = create_pipeline()
result = pipeline.run("你的输入")
print(result)

实际示例：服务部署

python
from openai import OpenAI
import json
class vLLMPipeline:
    """
    用于服务部署的 vLLM 实现。
    
    架构：
    - 输入验证
    - vLLM 处理
    - 输出结构化
    """
    
    def __init__(self, model: str = "gpt-4o-mini"):
        self.client = OpenAI()
        self.model = model
        self.system_prompt = f"""你是一个专注于 {specialty} 的 AI 助手。
        利用你的专业知识提供准确、有用的回答。
        回答要简洁且结构化。"""
    
    def process(self, user_input: str, context: dict = None) -> dict:
        """通过 vLLM 管道处理输入。"""
        
        # 构建上下文感知的提示
        context_str = json.dumps(context, indent=2) if context else "None"
        
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": f"上下文：\n{context_str}\n\n请求：\n{user_input}"}
        ]
        
        # 执行 LLM 调用
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.2,
            max_tokens=2000
        )
        
        content = response.choices[0].message.content
        
        return {
            "result": content,
            "model": self.model,
            "framework": "vLLM",
            "tokens_used": response.usage.total_tokens
        }
    
    def batch_process(self, inputs: list[str]) -> list[dict]:
        """高效处理多个输入。"""
        return [self.process(inp) for inp in inputs]
使用示例
pipeline = vLLMPipeline()
result = pipeline.process("用代码示例解释服务部署")
print(result["result"])
print(f"使用的 token 数：{result['tokens_used']}")

高级模式

流式响应

python
def stream_response(self, user_input: str):
    """流式输出 token，实现实时响应。"""
    stream = self.client.chat.completions.create(
        model=self.model,
        messages=[{"role": "user", "content": user_input}],
        stream=True
    )
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            yield delta.content

错误处理与重试

python
import time
from openai import RateLimitError, APIError
def process_with_retry(self, input_text: str, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            return self.process(input_text)
        except RateLimitError:
            wait_time = 2 ** attempt
            print(f"触发速率限制，等待 {wait_time} 秒...")
            time.sleep(wait_time)
        except APIError as e:
            if attempt == max_retries - 1:
                raise
            print(f"API 错误：{e}，正在重试...")
    raise Exception("超过最大重试次数")

测试

python
import pytest
@pytest.fixture
def pipeline():
    return vLLMPipeline(model="gpt-4o-mini")
def test_basic_processing(pipeline):
    result = pipeline.process("什么是服务部署？")
    assert "result" in result
    assert len(result["result"]) > 10
def test_batch_processing(pipeline):
    inputs = ["问题 1", "问题 2", "问题 3"]
    results = pipeline.batch_process(inputs)
    assert len(results) == len(inputs)

生产部署

python
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="vLLM API")
pipeline = vLLMPipeline()
class ProcessRequest(BaseModel):
    input: str
    context: dict = {}
@app.get("/health")
async def health():
    return {"status": "ok", "framework": "vLLM"}

最佳实践

资源

官方 vLLM 文档

包含示例的 GitHub 仓库

社区 Discord/Slack 支持

包含实际模式的 Cookbook