背景:
? ? ? vllm推理框架啟動模型不具備api-key驗證。需借助fastapi可以實現該功能
代碼實現:
rom fastapi import FastAPI, Header, HTTPException, Request,Response
import httpx
import logging# 創建 FastAPI 應用
app = FastAPI()
logging.basicConfig(level=logging.DEBUG)
# 配置 vLLM 的服務地址
VLLM_BASE_URL = "http://localhost:25010"# 定義合法的 API Key 列表(可以根據需要擴展為數據庫或配置文件)
VALID_API_KEYS = {"zml_123456789", "zml_1234567890"}# 中間件:驗證 API Key
# 驗證 API Key
async def verify_api_key(authorization: str = Header(None)):# 打印接收到的 Authorization 字段logging.debug(f"Received Authorization header: {authorization}")# 檢查 Authorization 是否存在且以 "Bearer " 開頭if not authorization or not isinstance(authorization, str) or not authorization.startswith("Bearer "):raise HTTPException(status_code=403, detail="Invalid Authorization Header")# 提取 API Keytry:api_key = authorization.split(" ")[1] # 提取 "Bearer " 后的部分except IndexError:raise HTTPException(status_code=403, detail="Malformed Authorization Header")# 驗證 API Key 是否合法if api_key not in VALID_API_KEYS:raise HTTPException(status_code=403, detail="Invalid API Key")# 代理路由:轉發請求到 vLLM
@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy(request: Request, path: str, authorization: str = Header(None)):# 驗證 API Keylogging.debug(f"Received request with path: {path}")logging.debug(f"Received headers: {request.headers}")await verify_api_key(authorization)# 構造轉發的目標 URLtarget_url = f"{VLLM_BASE_URL}/{path}"# 獲取請求體和查詢參數body = await request.body()query_params = request.query_params# 使用 httpx 轉發請求async with httpx.AsyncClient() as client:# 根據請求方法轉發response = await client.request(method=request.method,url=target_url,params=query_params,data=body,headers={key: value for key, value in request.headers.items() if key != "host"})# 返回 vLLM 的響應return Response(content=response.content, status_code=response.status_code, headers=dict(response.headers))
三、啟動
uvicorn my_fastapi:app --host=0.0.0.0 --port=12345
# my_fastapi 為腳本名稱
通過訪問fastapi提供的12345即可實現改功能