FLUX (توسط Black Forest Labs) در چند ماه گذشته دنیای تولید تصاویر هوش مصنوعی را طوفانی کرده است. نه تنها در بسیاری از معیارها از Stable Diffusion (پادشاه منبع باز قبلی) پیشی گرفته است، بلکه در برخی معیارها از مدل های اختصاصی مانند Dall-E یا Midjourney نیز پیشی گرفته است.

اما چگونه می خواهید از FLUX در یکی از برنامه های خود استفاده کنید؟ ممکن است کسی به استفاده از هاست های بدون سرور مانند Replicate و دیگران فکر کند، اما این ها می توانند خیلی سریع گران شوند و ممکن است انعطاف پذیری مورد نیاز شما را فراهم نکنند. اینجاست که ایجاد سرور FLUX خودتان به کارتان می آید.

در این مقاله، شما را با ایجاد سرور FLUX خود با استفاده از پایتون راهنمایی می کنیم. این سرور به شما این امکان را می دهد که از طریق یک API ساده، تصاویر را بر اساس پیام های متنی تولید کنید. چه این سرور را برای استفاده شخصی اجرا کنید یا آن را به عنوان بخشی از یک برنامه تولیدی به کار ببرید، این راهنما به شما در شروع کار کمک می کند.

پیش نیازها

قبل از ورود به کد، مطمئن شویم که ابزارها و کتابخانه های لازم را تنظیم کرده اید:

پایتون: شما باید پایتون 3 را روی دستگاه خود نصب کنید، ترجیحاً نسخه 3.10.

torch : چارچوب یادگیری عمیقی که برای اجرای FLUX استفاده خواهیم کرد.

diffusers : دسترسی به مدل FLUX را فراهم می کند.

transformers : وابستگی مورد نیاز دیفیوزرها.

sentencepiece : برای اجرای توکنایزر FLUX مورد نیاز است

protobuf : برای اجرای FLUX مورد نیاز است

accelerate : در برخی موارد به بارگیری مدل FLUX با کارایی بیشتر کمک می کند.

fastapi : چارچوبی برای ایجاد یک وب سرور که می تواند درخواست های تولید تصویر را بپذیرد.

uvicorn : برای اجرای سرور fastapi مورد نیاز است.

psutil : به ما امکان می دهد تحلیل کنیم که چه مقدار RAM در دستگاه ما وجود دارد.

شما می توانید تمام کتابخانه ها را با اجرای دستور زیر نصب کنید: pip install torch diffusers transformers accelerate fastapi uvicorn psutil .

نکته برای کاربران MacOS: اگر از مک با تراشه M1 یا M2 استفاده می‌کنید، باید PyTorch را با Metal راه‌اندازی کنید تا عملکرد مطلوبی داشته باشد. قبل از ادامه، راهنمای رسمی PyTorch with Metal را دنبال کنید.

مرحله 1: تنظیم محیط

بیایید اسکریپت را با انتخاب دستگاه مناسب برای اجرای استنتاج بر اساس سخت افزاری که استفاده می کنیم شروع کنیم.

 import torch device = 'cuda ' # can also be 'cpu ' or 'mps ' if device == 'mps ' and not torch.backends.mps.is_available(): raise Exception ( "Device set to MPS, but MPS is not available" ) elif device == 'cuda ' and not torch.cuda.is_available(): raise Exception ( "Device set to CUDA, but CUDA is not available" )

می‌توانید cpu ، cuda (برای پردازنده‌های گرافیکی NVIDIA)، یا mps (برای متال پرفورمنس شیدرهای اپل) را مشخص کنید. سپس اسکریپت تحلیل می‌کند که آیا دستگاه انتخاب‌شده در دسترس است یا خیر و در صورت عدم وجود یک استثناء ایجاد می‌کند.

مرحله 2: بارگیری مدل FLUX

سپس مدل FLUX را بارگذاری می کنیم. ما مدل را با دقت fp16 بارگذاری می کنیم که بدون افت کیفیت، مقداری از حافظه ما را ذخیره می کند.

توجه: در این مرحله، ممکن است از شما خواسته شود که با HuggingFace احراز هویت کنید، زیرا مدل FLUX دارای دروازه است. برای احراز هویت با موفقیت، باید یک حساب HuggingFace ایجاد کنید، به صفحه مدل بروید، شرایط را بپذیرید و سپس یک نشانه HuggingFace از تنظیمات حساب خود ایجاد کنید و آن را به عنوان متغیر محیطی HF_TOKEN روی دستگاه خود اضافه کنید.

 from diffusers import DDIMScheduler, FluxPipeline import psutil model_name = "black-forest-labs/FLUX.1-dev" print(f "Loading {model_name} on {device}" ) pipeline = FluxPipeline.from_pretrained( model_name, torch_dtype=torch.float16, use_safetensors=True ).to(device) pipeline. scheduler = DDIMScheduler.from_config(pipeline.scheduler.config)

در اینجا، ما مدل FLUX را با استفاده از کتابخانه diffusers بارگذاری می کنیم. مدلی که ما از آن استفاده می‌کنیم black-forest-labs/FLUX.1-dev است که با دقت fp16 بارگیری شده است. همچنین یک مدل حرفه ای FLUX وجود دارد که قوی تر است، اما متاسفانه منبع باز نیست، پس نمی توان از آن استفاده کرد.

ما در اینجا از زمان‌بندی DDIM استفاده می‌کنیم، اما می‌توانید یکی دیگر مانند Euler یا UniPC را نیز انتخاب کنید. شما می توانید اطلاعات بیشتر در مورد زمانبندی را در اینجا بخوانید.

از آنجایی که تولید تصویر می‌تواند منابع زیادی داشته باشد، بهینه‌سازی استفاده از حافظه بسیار مهم است، به‌ویژه زمانی که روی CPU یا دستگاهی با حافظه محدود اجرا می‌شود.

 # Recommended if running on MPS or CPU with < 64 GB of RAM total_memory = psutil.virtual_memory().total total_memory_gb = total_memory / ( 1024 ** 3 ) if (device == 'cpu ' or device == 'mps ') and total_memory_gb < 64 : print( "Enabling attention slicing" ) pipeline.enable_attention_slicing()

این کد کل حافظه موجود را تحلیل می کند و اگر سیستم کمتر از 64 گیگابایت رم داشته باشد، برش توجه را فعال می کند. برش توجه، مصرف حافظه را در طول تولید تصویر کاهش می‌دهد، که برای دستگاه‌هایی با منابع محدود ضروری است.

مرحله 3: ایجاد API با FastAPI

در مرحله بعد، سرور FastAPI را راه اندازی می کنیم که یک API برای تولید تصاویر ارائه می دهد.

 from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field, conint, confloat from fastapi.middleware.gzip import GZipMiddleware from io import BytesIO import base64 app = FastAPI() app.add_middleware(GZipMiddleware, minimum_size= 1000 , compresslevel= 7 )

FastAPI یک چارچوب محبوب برای ساخت API های وب با پایتون است. در این مورد، ما از آن برای ایجاد سروری استفاده می‌کنیم که می‌تواند درخواست‌های تولید تصویر را بپذیرد. ما همچنین از میان‌افزار GZip برای فشرده‌سازی پاسخ استفاده می‌کنیم، که مخصوصاً هنگام ارسال تصاویر در قالب base64 مفید است.

توجه: در یک محیط تولید، ممکن است بخواهید تصاویر تولید شده را در یک سطل S3 یا دیگر فضای ذخیره سازی ابری ذخیره کنید و URL ها را به جای رشته های کدگذاری شده با base64 برگردانید تا از CDN و سایر بهینه سازی ها استفاده کنید.

مرحله 4: تعریف مدل درخواست

ما باید یک مدل برای درخواست هایی که API ما می پذیرد تعریف کنیم.

 class GenerateRequest(BaseModel): prompt: str negative_prompt: str seed: conint(ge= 0 ) = Field( ... , description= "Seed for random number generation" ) height: conint( gt = 0 ) = Field( ... , description= "Height of the generated image, must be a positive integer and a multiple of 8" ) width: conint( gt = 0 ) = Field( ... , description= "Width of the generated image, must be a positive integer and a multiple of 8" ) cfg: confloat( gt = 0 ) = Field( ... , description= "CFG (classifier-free guidance scale), must be a positive integer or 0" ) steps: conint(ge= 0 ) = Field( ... , description= "Number of steps" ) batch_size: conint( gt = 0 ) = Field( ... , description= "Number of images to generate in a batch" )

این مدل GenerateRequest پارامترهای مورد نیاز برای تولید یک تصویر را تعریف می کند. اعلان توضیحات متنی تصویری است که می خواهید ایجاد کنید. از negative_prompt می توان برای مشخص کردن مواردی که در تصویر نمی خواهید استفاده کرد. فیلدهای دیگر شامل ابعاد تصویر، تعداد مراحل استنتاج و اندازه دسته است.

مرحله 5: ایجاد نقطه پایانی تولید تصویر

اکنون، بیایید نقطه پایانی را ایجاد کنیم که درخواست های تولید تصویر را مدیریت می کند.

 @app.post("/") async def generate_image (request: GenerateRequest) : if request.height % 8 != 0 or request.width % 8 != 0 : raise HTTPException(status_code= 400 , detail= "Height and width must both be multiples of 8" ) generator = [torch.Generator(device= "cpu" ).manual_seed(i) for i in range(request.seed, request.seed + request.batch_size)] images = pipeline( height=request.height, width=request.width, prompt=request.prompt, negative_prompt=request.negative_prompt, generator=generator, num_inference_steps=request.steps, guidance_scale=request.cfg, num_images_per_prompt=request.batch_size ).images base64_images = [] for image in images: buffered = BytesIO() image.save(buffered, format= "PNG" ) img_str = base64.b64encode(buffered.getvalue()).decode( "utf-8" ) base64_images.append(img_str) return { "images" : base64_images, }

این نقطه پایانی فرآیند تولید تصویر را مدیریت می کند. ابتدا تأیید می کند که ارتفاع و عرض مضربی از 8 هستند، همانطور که توسط FLUX لازم است. سپس تصاویر را بر اساس دستور ارائه شده تولید می کند و آنها را به عنوان رشته های کدگذاری شده با base64 برمی گرداند.

مرحله 6: راه اندازی سرور

در نهایت، اجازه دهید مقداری کد برای راه اندازی سرور هنگام اجرای اسکریپت اضافه کنیم.

 @app.on_event("startup") async def startup_event () : print( "Image generation server running" ) if __name__ == "__main__" : import uvicorn uvicorn.run(app, host= "0.0.0.0" , port= 8000 )

این کد سرور FastAPI را در پورت 8000 راه اندازی می کند و از http://localhost:8000 قابل دسترسی است.

مرحله 7: سرور خود را به صورت محلی آزمایش کنید

اکنون که سرور FLUX شما راه اندازی شده است، زمان آن رسیده است که آن را آزمایش کنید. می توانید از curl، یک ابزار خط فرمان برای ایجاد درخواست های HTTP، برای تعامل با سرور خود استفاده کنید:

 curl -X POST "http://localhost:8000/" \ -H "Content-Type: application/json" \ -d '{ " prompt ": " A futuristic cityscape at sunset ", " negative_prompt ": " low quality, blurry ", " seed ": 42, " height ": 512, " width ": 512, " cfg ": 7.5, " steps ": 50, " batch_size ": 1 }'

نتیجه گیری

تبریک می گویم! شما با موفقیت سرور FLUX خود را با استفاده از پایتون ایجاد کردید. این تنظیمات به شما امکان می دهد تصاویر را بر اساس دستورات متنی از طریق یک API ساده تولید کنید. اگر از نتایج مدل پایه FLUX راضی نیستید، ممکن است مدل را برای عملکرد بهتر یا موارد استفاده خاص تنظیم کنید.

کد کامل

می توانید کد کامل مورد استفاده در این راهنما را در زیر بیابید:

 import torch device = 'cuda' if device == 'mps' and not torch.backends.mps.is_available(): raise Exception( "Device set to MPS, but MPS is not available" ) elif device == 'cuda' and not torch.cuda.is_available(): raise Exception( "Device set to CUDA, but CUDA is not available" ) from diffusers import DDIMScheduler, FluxPipeline import psutil model_name = "black-forest-labs/FLUX.1-dev" print(f "Loading {model_name} on {device}" ) pipeline = FluxPipeline.from_pretrained( model_name, torch_dtype=torch.float16, use_safetensors= True ).to(device) pipeline.scheduler = DDIMScheduler.from_config(pipeline.scheduler.config) total_memory = psutil.virtual_memory().total total_memory_gb = total_memory / ( 1024 ** 3 ) if (device == 'cpu' or device == 'mps' ) and total_memory_gb < 64 : print( "Enabling attention slicing" ) pipeline.enable_attention_slicing() from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field, conint, confloat from fastapi.middleware.gzip import GZipMiddleware from io import BytesIO import base64 app = FastAPI() app.add_middleware(GZipMiddleware, minimum_size= 1000 , compresslevel= 7 ) class GenerateRequest (BaseModel) : prompt: str negative_prompt: str seed: conint(ge= 0 ) = Field(..., description= "Seed for random number generation" ) height: conint(gt= 0 ) = Field(..., description= "Height of the generated image, must be a positive integer and a multiple of 8" ) width: conint(gt= 0 ) = Field(..., description= "Width of the generated image, must be a positive integer and a multiple of 8" ) cfg: confloat(gt= 0 ) = Field(..., description= "CFG (classifier-free guidance scale), must be a positive integer or 0" ) steps: conint(ge= 0 ) = Field(..., description= "Number of steps" ) batch_size: conint(gt= 0 ) = Field(..., description= "Number of images to generate in a batch" ) @app.post("/") async def generate_image (request: GenerateRequest) : if request.height % 8 != 0 or request.width % 8 != 0 : raise HTTPException(status_code= 400 , detail= "Height and width must both be multiples of 8" ) generator = [torch.Generator(device= "cpu" ).manual_seed(i) for i in range(request.seed, request.seed + request.batch_size)] images = pipeline( height=request.height, width=request.width, prompt=request.prompt, negative_prompt=request.negative_prompt, generator=generator, num_inference_steps=request.steps, guidance_scale=request.cfg, num_images_per_prompt=request.batch_size ).images base64_images = [] for image in images: buffered = BytesIO() image.save(buffered, format= "PNG" ) img_str = base64.b64encode(buffered.getvalue()).decode( "utf-8" ) base64_images.append(img_str) return { "images" : base64_images, } @app.on_event("startup") async def startup_event () : print( "Image generation server running" ) if __name__ == "__main__" : import uvicorn uvicorn.run(app, host= "0.0.0.0" , port= 8000 )

خبرکاو