一，#本機(jī)環(huán)境檢查
執(zhí)行nvidia-smi，查看右上角。驗(yàn)證顯卡驅(qū)動已安裝最高支持的版本。
nvidia-smi
#在調(diào)試時(shí)，為了實(shí)時(shí)觀察GPU利用率，一般新開一個命令窗口，執(zhí)行以下命令，一秒刷新一次。
watch -n 1 nvidia-smi
執(zhí)行nvcc -V驗(yàn)證cuda
nvcc -V
執(zhí)行conda --version驗(yàn)證conda版本
conda --version
#列出所有已創(chuàng)建的Conda 環(huán)境??：
conda env list 或 conda info --envs
#若存在，先刪除已存在環(huán)境
conda env remove -n conda_qwen_image
#創(chuàng)建新環(huán)境
conda create -n conda_qwen_image python=3.10
#激活環(huán)境
conda activate conda_qwen_image
二，依賴庫安裝
根據(jù)CUDA版本安裝PyTorch??：
??CUDA 12.1??:
pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
??CUDA 12.2??:
pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu122
#驗(yàn)證PyTorch是否能正確識別GPU
python3 -c "import torch; print('PyTorch版本:', torch.__version__); print('CUDA可用:', torch.cuda.is_available()); print('CUDA版本:', torch.version.cuda); print('GPU設(shè)備:', torch.cuda.get_device_name(0) if torch.cuda.is_available() else 'None'); print('GPU數(shù)量:', torch.cuda.device_count());"
#魔搭modelscope庫安裝
pip3 install modelscope
#相關(guān)庫
pip3 install --upgrade transformers peft diffusers fastapi uvicorn
#新建man.py文件，加入代碼
from modelscope import DiffusionPipeline, FlowMatchEulerDiscreteScheduler, snapshot_download
import torch
import math
import os
from pathlib import Path
from datetime import datetime
import base64
from io import BytesIO
from typing import Optional, List, Union
import time
import uuid
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
# 設(shè)置HF_ENDPOINT使用國內(nèi)鏡像加速模型下載
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
# 減少顯存碎片化
os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "expandable_segments:True"
# 啟用推理模式，減少顯存占用
torch.inference_mode()
# 定義請求和響應(yīng)模型
class ImageGenerationRequest(BaseModel):
prompt: str
negative_prompt: Optional[str] = ""
width: Optional[int] = 1024
height: Optional[int] = 1024
num_inference_steps: Optional[int] = 8
seed: Optional[int] = 0
class ImageData(BaseModel):
b64_json: str
revised_prompt: str
class ImageGenerationResponse(BaseModel):
created: int
data: List[ImageData]
# 初始化模型
def initialize_model():
print("=" * 50)
print("開始初始化模型...")
print("=" * 50)
scheduler_config = {
'base_image_seq_len': 256,
'base_shift': math.log(3),
'invert_sigmas': False,
'max_image_seq_len': 8192,
'max_shift': math.log(3),
'num_train_timesteps': 1000,
'shift': 1.0,
'shift_terminal': None,
'stochastic_sampling': False,
'time_shift_type': 'exponential',
'use_beta_sigmas': False,
'use_dynamic_shifting': True,
'use_exponential_sigmas': False,
'use_karras_sigmas': False,
}
print("正在配置調(diào)度器...")
scheduler = FlowMatchEulerDiscreteScheduler.from_config(scheduler_config)
print("調(diào)度器配置完成")
# 檢測可用GPU數(shù)量
num_gpus = torch.cuda.device_count()
print(f"檢測到 {num_gpus} 個GPU設(shè)備")
# 根據(jù)項(xiàng)目規(guī)范，對于DiffusionPipeline模型，我們使用max_memory參數(shù)進(jìn)行顯存分配
# 而不是手動指定每層的設(shè)備映射
if num_gpus > 1:
print("檢測到多個GPU，正在進(jìn)行顯存分配...")
# 獲取每個GPU的顯存信息并計(jì)算可分配的顯存量
max_memory = {}
for i in range(num_gpus):
free_mem, total_mem = torch.cuda.mem_get_info(i)
# 按70%的空閑顯存計(jì)算分配量，同時(shí)確保不超過22GB
allocated_mem = min(int(free_mem * 0.7), 22 * 1024**3, free_mem)
max_memory[i] = allocated_mem
print(f"GPU {i}: 分配 {(allocated_mem / 1024**3):.2f} GB 顯存")
# 加載模型并指定顯存分配
print("正在加載模型到多個GPU...")
pipe = DiffusionPipeline.from_pretrained(
'Qwen/Qwen-Image',
scheduler=scheduler,
torch_dtype=torch.bfloat16,
max_memory=max_memory, # 為每個GPU分配顯存
)
print("多GPU模型加載完成")
else:
# 單GPU情況
print("檢測到單個GPU，正在加載模型...")
pipe = DiffusionPipeline.from_pretrained(
'Qwen/Qwen-Image',
scheduler=scheduler,
torch_dtype=torch.bfloat16,
)
pipe = pipe.to("cuda")
print("單GPU模型加載完成")
print(f"模型已分配到{num_gpus}個GPU設(shè)備上")
print("=" * 50)
print("模型初始化完成")
print("=" * 50)
return pipe
# 提前下載LoRA權(quán)重
def download_lora_weights():
print("=" * 50)
print("開始下載LoRA權(quán)重...")
print("=" * 50)
# 使用ModelScope的snapshot_download下載LoRA權(quán)重
model_dir = snapshot_download('lightx2v/Qwen-Image-Lightning')
print(f"LoRA權(quán)重已下載到: {model_dir}")
# 查找.pt或.safetensors文件
lora_files = list(Path(model_dir).glob("*.safetensors")) + list(Path(model_dir).glob("*.pt"))
if not lora_files:
raise FileNotFoundError("在下載的LoRA權(quán)重目錄中未找到.safetensors或.pt文件")
lora_file_path = lora_files[0] # 使用找到的第一個文件
print(f"使用LoRA文件: {lora_file_path}")
print("=" * 50)
print("LoRA權(quán)重下載完成")
print("=" * 50)
return str(lora_file_path)
# 圖像生成函數(shù)
def generate_image(request: ImageGenerationRequest, pipe):
"""生成圖像"""
print("=" * 50)
print("開始生成圖像...")
print("=" * 50)
print(f"輸入?yún)?shù):")
print(f" - 提示詞: {request.prompt}")
print(f" - 負(fù)面提示詞: {request.negative_prompt}")
print(f" - 圖像尺寸: {request.width}x{request.height}")
print(f" - 推理步數(shù): {request.num_inference_steps}")
print(f" - 隨機(jī)種子: {request.seed}")
start_time = time.time()
print("開始圖像生成過程...")
print("正在調(diào)用模型生成圖像...")
image = pipe(
prompt=request.prompt,
negative_prompt=request.negative_prompt,
width=request.width,
height=request.height,
num_inference_steps=request.num_inference_steps,
true_cfg_scale=1.0,
generator=torch.manual_seed(request.seed)
).images[0]
generation_time = time.time() - start_time
print(f"圖像生成完成，耗時(shí): {generation_time:.2f} 秒")
print("正在將圖像轉(zhuǎn)換為base64編碼...")
# 將圖像轉(zhuǎn)換為base64編碼
buffered = BytesIO()
image.save(buffered, format="PNG")
img_str = base64.b64encode(buffered.getvalue()).decode()
print("圖像轉(zhuǎn)換完成")
print("=" * 50)
print("圖像生成流程結(jié)束")
print("=" * 50)
return img_str
# 初始化FastAPI應(yīng)用
app = FastAPI(
title="Qwen-Image API",
description="基于Qwen-Image模型的OpenAI兼容圖像生成API",
version="1.0.0"
)
# 全局模型實(shí)例
pipe = None
@app.on_event("startup")
async def startup_event():
global pipe
# 初始化模型
pipe = initialize_model()
# 下載LoRA權(quán)重
lora_file_path = download_lora_weights()
print("正在加載LoRA權(quán)重...")
pipe.load_lora_weights(lora_file_path)
print("LoRA權(quán)重加載完成")
@app.post("/v1/images/generations", response_model=ImageGenerationResponse)
async def create_image(request: ImageGenerationRequest):
try:
# 生成圖像
image_data = generate_image(request, pipe)
# 構(gòu)造響應(yīng)
response = ImageGenerationResponse(
created=int(time.time()),
data=[
ImageData(
b64_json=image_data,
revised_prompt=request.prompt
)
]
)
return response
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8800)
運(yùn)行服務(wù)端API。 CUDA_VISIBLE_DEVICES 指定顯卡編號
CUDA_VISIBLE_DEVICES=0,1 python3 main.py
測試調(diào)用
curl -X POST http://localhost:8800/v1/images/generations -H "Content-Type: application/json" -d '{"prompt": "a lovely cat"}' | grep -q '"data"' && echo "API 工作正常" || echo "API 出現(xiàn)問題"
調(diào)用時(shí)控制臺截圖：
#安裝低代碼gradio庫
pip3 install gradio
#新建app.py 使用低代碼，創(chuàng)建文生圖用戶端。
import gradio as gr import requests import json import base64 from io import BytesIO from PIL import Image import traceback # 本地API配置 API_URL = "http://localhost:8800/v1/images/generations" # 示例提示詞 EXAMPLE_PROMPTS = [ "一只可愛的熊貓坐在竹子上吃竹子，高清攝影", "未來城市的夜景，有飛行汽車和霓虹燈，賽博朋克風(fēng)格", "寧靜的山水畫，有小橋流水和青山環(huán)繞，中國風(fēng)" ] def generate_image(prompt, negative_prompt="", width=1024, height=1024, steps=8, seed=0): """ 調(diào)用本地API生成圖片 Args: prompt (str): 正向提示詞 negative_prompt (str): 負(fù)向提示詞 width (int): 圖片寬度 height (int): 圖片高度 steps (int): 推理步數(shù) seed (int): 隨機(jī)種子 Returns: tuple: (生成的圖片, 狀態(tài)信息) """ if not prompt or prompt.strip() == "": return None, "請輸入提示詞" try: # 準(zhǔn)備請求數(shù)據(jù) payload = { "prompt": prompt, "negative_prompt": negative_prompt, "width": width, "height": height, "num_inference_steps": steps, "seed": seed } # 調(diào)用本地API print(f"正在調(diào)用本地API: {API_URL}") response = requests.post(API_URL, json=payload, timeout=1800) # 30分鐘超時(shí) if response.status_code == 200: result = response.json() # 解析返回的base64圖片數(shù)據(jù) if "data" in result and len(result["data"]) > 0: image_data = result["data"][0]["b64_json"] # 解碼base64圖片數(shù)據(jù) image_bytes = base64.b64decode(image_data) image = Image.open(BytesIO(image_bytes)) return image, "圖片生成成功" else: return None, "API返回?cái)?shù)據(jù)格式錯誤" else: return None, f"API調(diào)用失敗，狀態(tài)碼: {response.status_code}，響應(yīng): {response.text}" except requests.exceptions.ConnectionError: return None, "連接API失敗，請確保本地API服務(wù)已啟動" except requests.exceptions.Timeout: return None, "API調(diào)用超時(shí)，請稍后重試" except Exception as e: error_msg = f"生成圖片時(shí)出錯: {str(e)}" print(error_msg) traceback.print_exc() return None, error_msg # 創(chuàng)建Gradio界面 with gr.Blocks(title="文生圖應(yīng)用") as demo: gr.Markdown("# 文生圖應(yīng)用") gr.Markdown("輸入提示詞，調(diào)用本地API生成對應(yīng)的圖片") with gr.Row(): with gr.Column(): prompt_input = gr.Textbox( label="提示詞", placeholder="請輸入圖片描述...", lines=3 ) negative_prompt_input = gr.Textbox( label="負(fù)向提示詞", placeholder="請輸入不希望出現(xiàn)在圖片中的內(nèi)容（可選）...", lines=2 ) with gr.Accordion("高級參數(shù)", open=False): width_input = gr.Slider( minimum=256, maximum=2048, value=1024, step=64, label="圖片寬度" ) height_input = gr.Slider( minimum=256, maximum=2048, value=1024, step=64, label="圖片高度" ) steps_input = gr.Slider( minimum=1, maximum=50, value=8, step=1, label="推理步數(shù)" ) seed_input = gr.Number( label="隨機(jī)種子 (0表示隨機(jī))", value=0, precision=0 ) # 添加示例提示詞按鈕 gr.Markdown("### 示例提示詞") for i, example in enumerate(EXAMPLE_PROMPTS, 1): btn = gr.Button(f"示例 {i}: {example[:30]}..." if len(example) > 30 else f"示例 {i}: {example}") btn.click( fn=lambda ex=example: ex, outputs=prompt_input ) generate_btn = gr.Button("生成圖片", variant="primary") with gr.Column(): image_output = gr.Image(label="生成的圖片") status_output = gr.Textbox(label="狀態(tài)信息", max_lines=3) # 綁定生成按鈕事件 generate_btn.click( fn=generate_image, inputs=[ prompt_input, negative_prompt_input, width_input, height_input, steps_input, seed_input ], outputs=[image_output, status_output] ) if __name__ == "__main__": print("啟動文生圖Gradio界面...") print(f"請確保本地API服務(wù)已在 {API_URL} 啟動") demo.launch(server_name="0.0.0.0", server_port=7860, share=False)
#運(yùn)行應(yīng)用
python3 app.py
#使用效果圖
浙公網(wǎng)安備 33010602011771號