Qwen3-32B-Chat实战教程：基于FastAPI封装Qwen3 API并添加请求审计日志

Qwen3-32B-Chat实战教程基于FastAPI封装Qwen3 API并添加请求审计日志1. 教程概述本教程将指导您如何基于FastAPI框架封装Qwen3-32B-Chat模型的API服务并添加请求审计日志功能。通过本教程您将学会如何快速部署Qwen3-32B-Chat私有镜像使用FastAPI构建RESTful API服务实现API请求的审计日志记录优化大模型API服务的性能与稳定性本教程特别针对RTX 4090D 24GB显存环境优化采用CUDA 12.4和驱动550.90.07确保最佳推理性能。2. 环境准备与快速部署2.1 硬件与镜像要求确保您的环境满足以下要求GPURTX 4090D 24GB显存必须内存≥120GBCPU10核心以上系统盘50GB数据盘40GB2.2 一键启动API服务使用预置镜像中的启动脚本快速部署# 进入工作目录 cd /workspace # 启动API服务 bash start_api.sh服务启动后您可以通过以下地址访问API文档http://localhost:8001/docs默认端口80013. FastAPI基础封装3.1 创建FastAPI应用首先创建一个基础的FastAPI应用来封装Qwen3模型from fastapi import FastAPI from pydantic import BaseModel from transformers import AutoModelForCausalLM, AutoTokenizer app FastAPI() # 加载模型 model_path /workspace/models/Qwen3-32B tokenizer AutoTokenizer.from_pretrained(model_path) model AutoModelForCausalLM.from_pretrained( model_path, torch_dtypeauto, device_mapauto, trust_remote_codeTrue ) class ChatRequest(BaseModel): prompt: str max_length: int 512 temperature: float 0.7 app.post(/chat) async def chat_completion(request: ChatRequest): inputs tokenizer(request.prompt, return_tensorspt).to(cuda) outputs model.generate( **inputs, max_lengthrequest.max_length, temperaturerequest.temperature ) response tokenizer.decode(outputs[0], skip_special_tokensTrue) return {response: response}3.2 测试API接口启动服务后您可以使用curl测试APIcurl -X POST http://localhost:8001/chat \ -H Content-Type: application/json \ -d {prompt:你好介绍一下你自己,max_length:200}4. 添加审计日志功能4.1 实现日志中间件为API添加请求审计日志功能import time from fastapi import Request import logging # 配置日志 logging.basicConfig( filenameapi_audit.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) app.middleware(http) async def audit_log(request: Request, call_next): start_time time.time() # 记录请求信息 client_ip request.client.host method request.method path request.url.path response await call_next(request) # 计算处理时间 process_time time.time() - start_time # 记录审计日志 log_data { client_ip: client_ip, method: method, path: path, status_code: response.status_code, process_time: f{process_time:.3f}s } logging.info(fAPI请求审计 - {log_data}) return response4.2 增强的日志记录在聊天接口中添加详细的请求/响应日志app.post(/chat) async def chat_completion(request: ChatRequest): # 记录请求 logging.info(f请求内容: {request.prompt[:100]}... (长度:{len(request.prompt)})) inputs tokenizer(request.prompt, return_tensorspt).to(cuda) outputs model.generate( **inputs, max_lengthrequest.max_length, temperaturerequest.temperature ) response tokenizer.decode(outputs[0], skip_special_tokensTrue) # 记录响应 logging.info(f响应内容: {response[:100]}... (长度:{len(response)})) return {response: response}5. 性能优化与生产部署5.1 启用批处理支持修改API以支持批处理请求from typing import List class BatchChatRequest(BaseModel): prompts: List[str] max_length: int 512 temperature: float 0.7 app.post(/batch_chat) async def batch_chat_completion(request: BatchChatRequest): inputs tokenizer(request.prompts, return_tensorspt, paddingTrue).to(cuda) outputs model.generate( **inputs, max_lengthrequest.max_length, temperaturerequest.temperature ) responses [tokenizer.decode(output, skip_special_tokensTrue) for output in outputs] return {responses: responses}5.2 添加速率限制使用FastAPI的中间件添加API速率限制from fastapi.middleware import Middleware from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware from slowapi import Limiter from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) app.state.limiter limiter app.post(/chat) limiter.limit(10/minute) async def chat_completion(request: ChatRequest): # 原有实现...5.3 生产环境部署建议对于生产环境建议使用NGINX作为反向代理配置SSL/TLS加密启用Gunicorn或多进程部署监控GPU显存使用情况定期轮转审计日志示例Gunicorn启动命令gunicorn -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8001 main:app6. 总结与进阶建议通过本教程您已经学会了如何快速部署Qwen3-32B-Chat私有镜像使用FastAPI构建RESTful API服务实现API请求的审计日志功能进行性能优化和生产部署准备进阶建议考虑添加用户认证和授权实现API密钥管理添加更详细的性能监控考虑模型量化以降低显存占用定期更新模型和依赖库获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。