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是否能使用GPU2. 基础概念快速入门2.1 什么是FastAPIFastAPI是一个现代、快速高性能的Web框架用于构建API。它特别适合机器学习模型的部署因为自动生成交互式API文档基于Python类型提示性能接近NodeJS和Go简单易学但功能强大2.2 JWT身份认证简介JWTJSON Web Token是一种开放标准用于在各方之间安全地传输信息。在我们的场景中它将用于用户登录后生成令牌后续请求携带令牌进行身份验证服务器验证令牌有效性3. 分步实践操作3.1 安装额外依赖虽然镜像已经预装了很多工具但我们还需要安装FastAPI和相关库pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart3.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.md3.3 封装Qwen-VL模型首先创建qwen_vl.py文件封装模型推理逻辑from transformers import AutoModelForCausalLM, AutoTokenizer from PIL import Image import torch class QwenVLWrapper: def __init__(self, model_pathQwen/Qwen-VL): self.device cuda if torch.cuda.is_available() else cpu self.tokenizer AutoTokenizer.from_pretrained(model_path, trust_remote_codeTrue) self.model AutoModelForCausalLM.from_pretrained( model_path, device_mapauto, trust_remote_codeTrue ).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_tensorspt).to(self.device) with torch.no_grad(): outputs self.model.generate(**inputs) response self.tokenizer.decode(outputs[0]) return response3.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], deprecatedauto) oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) 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(minutes15) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) return encoded_jwt async def get_current_user(token: str Depends(oauth2_scheme)): credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detailCould 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 username3.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_credentialsTrue, allow_methods[*], allow_headers[*], ) # 初始化模型 model QwenVLWrapper() # 模拟用户数据库 fake_users_db { testuser: { username: testuser, hashed_password: $2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW, } } app.post(/token, response_modelToken) 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_codestatus.HTTP_401_UNAUTHORIZED, detailIncorrect username or password, headers{WWW-Authenticate: Bearer}, ) access_token_expires timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) access_token create_access_token( data{sub: user[username]}, expires_deltaaccess_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 usernametestuserpasswordsecret使用令牌查询图片curl -X POST \ http://localhost:8000/query?question这张图片里有什么 \ -H accept: application/json \ -H Authorization: Bearer YOUR_TOKEN \ -F image/path/to/your/image.jpg5. 实用技巧与进阶5.1 性能优化建议模型预热在服务启动时预先加载模型避免第一次请求延迟过高批处理支持修改代码支持同时处理多个查询GPU内存管理添加显存监控防止内存泄漏5.2 安全增强使用环境变量存储敏感信息如JWT密钥实现速率限制防止滥用添加输入验证防止恶意请求5.3 生产部署对于生产环境建议使用Gunicorn作为WSGI服务器配置Nginx作为反向代理设置HTTPS加密实现日志记录和监控6. 总结通过本教程我们完成了以下工作在Qwen-Image定制镜像环境中部署了Qwen-VL模型使用FastAPI创建了RESTful API服务实现了JWT身份认证保护API端点提供了完整的代码示例和测试方法现在您已经拥有了一个安全、高效的Qwen-VL API服务可以轻松集成到各种应用中。下一步您可以考虑添加更多API端点支持不同功能实现用户管理系统部署到云服务器提供公开服务获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Qwen-Image保姆级教程：如何将Qwen-VL封装为FastAPI服务并添加JWT身份认证
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是否能使用GPU2. 基础概念快速入门2.1 什么是FastAPIFastAPI是一个现代、快速高性能的Web框架用于构建API。它特别适合机器学习模型的部署因为自动生成交互式API文档基于Python类型提示性能接近NodeJS和Go简单易学但功能强大2.2 JWT身份认证简介JWTJSON Web Token是一种开放标准用于在各方之间安全地传输信息。在我们的场景中它将用于用户登录后生成令牌后续请求携带令牌进行身份验证服务器验证令牌有效性3. 分步实践操作3.1 安装额外依赖虽然镜像已经预装了很多工具但我们还需要安装FastAPI和相关库pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart3.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.md3.3 封装Qwen-VL模型首先创建qwen_vl.py文件封装模型推理逻辑from transformers import AutoModelForCausalLM, AutoTokenizer from PIL import Image import torch class QwenVLWrapper: def __init__(self, model_pathQwen/Qwen-VL): self.device cuda if torch.cuda.is_available() else cpu self.tokenizer AutoTokenizer.from_pretrained(model_path, trust_remote_codeTrue) self.model AutoModelForCausalLM.from_pretrained( model_path, device_mapauto, trust_remote_codeTrue ).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_tensorspt).to(self.device) with torch.no_grad(): outputs self.model.generate(**inputs) response self.tokenizer.decode(outputs[0]) return response3.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], deprecatedauto) oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) 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(minutes15) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) return encoded_jwt async def get_current_user(token: str Depends(oauth2_scheme)): credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detailCould 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 username3.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_credentialsTrue, allow_methods[*], allow_headers[*], ) # 初始化模型 model QwenVLWrapper() # 模拟用户数据库 fake_users_db { testuser: { username: testuser, hashed_password: $2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW, } } app.post(/token, response_modelToken) 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_codestatus.HTTP_401_UNAUTHORIZED, detailIncorrect username or password, headers{WWW-Authenticate: Bearer}, ) access_token_expires timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) access_token create_access_token( data{sub: user[username]}, expires_deltaaccess_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 usernametestuserpasswordsecret使用令牌查询图片curl -X POST \ http://localhost:8000/query?question这张图片里有什么 \ -H accept: application/json \ -H Authorization: Bearer YOUR_TOKEN \ -F image/path/to/your/image.jpg5. 实用技巧与进阶5.1 性能优化建议模型预热在服务启动时预先加载模型避免第一次请求延迟过高批处理支持修改代码支持同时处理多个查询GPU内存管理添加显存监控防止内存泄漏5.2 安全增强使用环境变量存储敏感信息如JWT密钥实现速率限制防止滥用添加输入验证防止恶意请求5.3 生产部署对于生产环境建议使用Gunicorn作为WSGI服务器配置Nginx作为反向代理设置HTTPS加密实现日志记录和监控6. 总结通过本教程我们完成了以下工作在Qwen-Image定制镜像环境中部署了Qwen-VL模型使用FastAPI创建了RESTful API服务实现了JWT身份认证保护API端点提供了完整的代码示例和测试方法现在您已经拥有了一个安全、高效的Qwen-VL API服务可以轻松集成到各种应用中。下一步您可以考虑添加更多API端点支持不同功能实现用户管理系统部署到云服务器提供公开服务获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
