当前位置：首页 > news >正文

Qwen-Image保姆级教程：如何将Qwen-VL封装为FastAPI服务并添加JWT身份认证

news 2026/5/12 10:00:18

Qwen-Image保姆级教程：如何将Qwen-VL封装为FastAPI服务并添加JWT身份认证

1. 环境准备与快速部署

在开始之前，请确保您已经获取了Qwen-Image定制镜像并成功启动实例。这个镜像已经预装了所有必要的依赖，包括：

CUDA 12.4和对应GPU驱动
Python 3.x环境
PyTorch GPU版本
Qwen-VL模型推理所需的所有库

要验证环境是否准备就绪，可以运行以下命令：

nvidia-smi # 查看GPU状态 nvcc -V # 验证CUDA版本 python -c "import torch; print(torch.cuda.is_available())" # 检查PyTorch是否能使用GPU

2. 基础概念快速入门

2.1 什么是FastAPI？

FastAPI是一个现代、快速（高性能）的Web框架，用于构建API。它特别适合机器学习模型的部署，因为：

自动生成交互式API文档
基于Python类型提示
性能接近NodeJS和Go
简单易学但功能强大

2.2 JWT身份认证简介

JWT（JSON Web Token）是一种开放标准，用于在各方之间安全地传输信息。在我们的场景中，它将用于：

用户登录后生成令牌
后续请求携带令牌进行身份验证
服务器验证令牌有效性

3. 分步实践操作

3.1 安装额外依赖

虽然镜像已经预装了很多工具，但我们还需要安装FastAPI和相关库：

pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart

3.2 创建FastAPI应用结构

建议按以下结构组织项目：

/qwen-vl-api/ ├── app/ │ ├── __init__.py │ ├── main.py # 主应用文件 │ ├── models.py # 数据模型 │ ├── schemas.py # Pydantic模型 │ ├── auth.py # 认证相关 │ ├── config.py # 配置 │ └── qwen_vl.py # Qwen-VL模型封装 ├── requirements.txt └── README.md

3.3 封装Qwen-VL模型

首先创建qwen_vl.py文件，封装模型推理逻辑：

from transformers import AutoModelForCausalLM, AutoTokenizer from PIL import Image import torch class QwenVLWrapper: def __init__(self, model_path="Qwen/Qwen-VL"): self.device = "cuda" if torch.cuda.is_available() else "cpu" self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True) self.model = AutoModelForCausalLM.from_pretrained( model_path, device_map="auto", trust_remote_code=True ).eval() def generate_response(self, image_path: str, question: str): image = Image.open(image_path) query = self.tokenizer.from_list_format([ {'image': image_path}, {'text': question}, ]) inputs = self.tokenizer(query, return_tensors='pt').to(self.device) with torch.no_grad(): outputs = self.model.generate(**inputs) response = self.tokenizer.decode(outputs[0]) return response

3.4 实现JWT认证

在auth.py中添加认证逻辑：

from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from fastapi.security import OAuth2PasswordBearer from fastapi import Depends, HTTPException, status # 配置信息 SECRET_KEY = "your-secret-key-here" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") def verify_password(plain_password: str, hashed_password: str): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str): return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt async def get_current_user(token: str = Depends(oauth2_scheme)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception except JWTError: raise credentials_exception return username

3.5 创建主应用文件

在main.py中整合所有组件：

from fastapi import FastAPI, Depends, HTTPException, UploadFile, File from fastapi.middleware.cors import CORSMiddleware from typing import List import os from .qwen_vl import QwenVLWrapper from .auth import get_current_user, create_access_token from .schemas import Token, UserCreate, UserInDB app = FastAPI() # 允许跨域请求 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 初始化模型 model = QwenVLWrapper() # 模拟用户数据库 fake_users_db = { "testuser": { "username": "testuser", "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", } } @app.post("/token", response_model=Token) async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()): user = fake_users_db.get(form_data.username) if not user or not verify_password(form_data.password, user["hashed_password"]): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user["username"]}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @app.post("/query") async def query_image( image: UploadFile = File(...), question: str, current_user: str = Depends(get_current_user) ): # 保存上传的图片 image_path = f"/tmp/{image.filename}" with open(image_path, "wb") as buffer: buffer.write(image.file.read()) try: response = model.generate_response(image_path, question) return {"response": response} finally: # 清理临时文件 if os.path.exists(image_path): os.remove(image_path)

4. 快速上手示例

4.1 启动服务

使用以下命令启动FastAPI服务：

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

服务启动后，您可以访问http://localhost:8000/docs查看自动生成的API文档。

4.2 测试API

首先获取访问令牌：

curl -X 'POST' \ 'http://localhost:8000/token' \ -H 'accept: application/json' \ -H 'Content-Type: application/x-www-form-urlencoded' \ -d 'username=testuser&password=secret'

使用令牌查询图片：

curl -X 'POST' \ 'http://localhost:8000/query?question=这张图片里有什么' \ -H 'accept: application/json' \ -H 'Authorization: Bearer YOUR_TOKEN' \ -F 'image=@"/path/to/your/image.jpg"'