vLLM-v0.17.1 API设计详解：从零构建类似OpenAI的接口服务

Shen Planck

168人浏览 · 2026-04-29 05:18:13

Shen Planck · 2026-04-29 05:18:13 发布

vLLM-v0.17.1 API设计详解：从零构建类似OpenAI的接口服务

1. 为什么需要兼容OpenAI的API接口

大模型服务生态中，OpenAI的API格式已经成为事实标准。许多第三方工具和应用都内置了对OpenAI API格式的支持。如果你的服务能兼容这种格式，开发者就能直接使用他们熟悉的工具链来调用你的服务，而不需要额外适配。

举个例子，像LangChain这样的流行框架默认就支持OpenAI的API格式。如果你的服务也采用相同格式，开发者就能无缝切换不同的模型服务提供商。

2. vLLM基础API快速入门

vLLM本身提供了一套原生的API接口，我们先快速了解一下它的基本用法。安装最新版vLLM很简单：

pip install vllm==0.17.1

启动一个基础服务只需要几行代码：

from vllm import LLM, SamplingParams

llm = LLM(model="meta-llama/Llama-2-7b-chat-hf")
sampling_params = SamplingParams(temperature=0.7, top_p=0.9)

outputs = llm.generate(["你好，介绍一下你自己"], sampling_params)
print(outputs[0].text)

这个原生API虽然简单直接，但缺少企业级服务需要的许多功能，比如标准化响应格式、身份验证等。

3. 设计兼容OpenAI的API接口

3.1 请求和响应格式

OpenAI的API主要使用JSON格式进行数据交换。一个典型的聊天补全请求如下：

{
  "model": "gpt-3.5-turbo",
  "messages": [
    {"role": "system", "content": "你是一个有帮助的助手"},
    {"role": "user", "content": "你好，你是谁？"}
  ],
  "temperature": 0.7
}

对应的响应格式应该是：

{
  "id": "chatcmpl-123",
  "object": "chat.completion",
  "created": 1677652288,
  "choices": [{
    "index": 0,
    "message": {
      "role": "assistant",
      "content": "我是一个AI助手"
    },
    "finish_reason": "stop"
  }],
  "usage": {
    "prompt_tokens": 9,
    "completion_tokens": 12,
    "total_tokens": 21
  }
}

3.2 使用FastAPI实现基础路由

我们可以用FastAPI来构建这个接口服务。首先创建一个基础路由：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ChatMessage(BaseModel):
    role: str
    content: str

class ChatCompletionRequest(BaseModel):
    model: str
    messages: list[ChatMessage]
    temperature: float = 0.7

@app.post("/v1/chat/completions")
async def create_chat_completion(request: ChatCompletionRequest):
    # 这里将实现vLLM调用逻辑
    pass

4. 集成vLLM引擎

现在我们需要把vLLM的推理能力集成到这个API中。关键是要把OpenAI格式的消息列表转换为vLLM能理解的提示词：

def convert_messages_to_prompt(messages):
    prompt = ""
    for msg in messages:
        if msg.role == "system":
            prompt += f"<|system|>\n{msg.content}</s>\n"
        elif msg.role == "user":
            prompt += f"<|user|>\n{msg.content}</s>\n"
        elif msg.role == "assistant":
            prompt += f"<|assistant|>\n{msg.content}</s>\n"
    prompt += "<|assistant|>\n"
    return prompt

然后完善我们的API端点：

from vllm import SamplingParams

@app.post("/v1/chat/completions")
async def create_chat_completion(request: ChatCompletionRequest):
    prompt = convert_messages_to_prompt(request.messages)
    sampling_params = SamplingParams(
        temperature=request.temperature,
        top_p=0.9,
        max_tokens=1024
    )
    
    outputs = llm.generate([prompt], sampling_params)
    completion = outputs[0].text
    
    return {
        "id": f"chatcmpl-{uuid.uuid4()}",
        "object": "chat.completion",
        "created": int(time.time()),
        "choices": [{
            "index": 0,
            "message": {
                "role": "assistant",
                "content": completion
            },
            "finish_reason": "length"
        }],
        "usage": {
            "prompt_tokens": len(outputs[0].prompt_token_ids),
            "completion_tokens": len(outputs[0].output_token_ids),
            "total_tokens": len(outputs[0].prompt_token_ids) + len(outputs[0].output_token_ids)
        }
    }

5. 添加企业级功能

5.1 身份验证

生产环境必须要有身份验证。我们可以使用API密钥：

from fastapi import Depends, Header

async def verify_api_key(api_key: str = Header(...)):
    if api_key != "your-secret-key":
        raise HTTPException(status_code=401, detail="Invalid API key")

@app.post("/v1/chat/completions", dependencies=[Depends(verify_api_key)])
async def create_chat_completion(request: ChatCompletionRequest):
    # 原有实现

5.2 速率限制

防止滥用需要添加速率限制。可以使用FastAPI的中间件：

from fastapi import Request
from fastapi.middleware import Middleware
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/v1/chat/completions")
@limiter.limit("10/minute")
async def create_chat_completion(request: ChatCompletionRequest, request: Request):
    # 原有实现

5.3 日志记录

记录请求和响应有助于监控和调试：

import logging

logging.basicConfig(filename='api.log', level=logging.INFO)

@app.post("/v1/chat/completions")
async def create_chat_completion(request: ChatCompletionRequest):
    start_time = time.time()
    
    # 原有实现
    
    duration = time.time() - start_time
    logging.info(f"Request: {request.model} | Duration: {duration:.2f}s | Tokens: {response['usage']['total_tokens']}")
    return response