FLUX, Python ve Difüzörlerle Yapay Zeka Destekli Görüntü Oluşturma API Hizmeti: Hızlı Bir Kılavuz

Bu makalede, Python kullanarak kendi FLUX sunucunuzu oluşturma konusunda size yol göstereceğiz. Bu sunucu, basit bir API aracılığıyla metin istemlerine dayalı görseller oluşturmanıza olanak tanır. Bu sunucuyu kişisel kullanım için çalıştırıyor veya bir üretim uygulamasının parçası olarak dağıtıyor olun, bu kılavuz başlamanıza yardımcı olacaktır.

FLUX ( Black Forest Labs tarafından) son birkaç ayda AI görüntü oluşturma dünyasını kasıp kavurdu. Sadece birçok kıyaslamada Stable Diffusion'ı (önceki açık kaynak kralı) yenmekle kalmadı, aynı zamanda bazı metriklerde Dall-E veya Midjourney gibi tescilli modelleri de geride bıraktı.

Peki FLUX'u uygulamalarınızdan birinde nasıl kullanırsınız? Replicate ve diğerleri gibi sunucusuz ana bilgisayarlar kullanmayı düşünebilirsiniz, ancak bunlar çok hızlı bir şekilde çok pahalı hale gelebilir ve ihtiyacınız olan esnekliği sağlamayabilir. İşte tam bu noktada kendi özel FLUX sunucunuzu oluşturmanız işe yarar.

Ön koşullar

Koda dalmadan önce gerekli araçların ve kütüphanelerin kurulu olduğundan emin olalım:

• Python: Makinenizde Python 3'ün, tercihen 3.10 sürümünün yüklü olması gerekir.
• torch : FLUX'u çalıştırmak için kullanacağımız derin öğrenme çerçevesi.
• diffusers : FLUX modeline erişim sağlar.
• transformers : Difüzörlerin zorunlu bağımlılığı.
• sentencepiece : FLUX belirteçleyicisini çalıştırmak için gereklidir
• protobuf : FLUX'u çalıştırmak için gereklidir
• accelerate : Bazı durumlarda FLUX modelinin daha verimli bir şekilde yüklenmesine yardımcı olur.
• fastapi : Görüntü oluşturma isteklerini kabul edebilen bir web sunucusu oluşturmak için kullanılan çerçeve.
• uvicorn : FastAPI sunucusunu çalıştırmak için gereklidir.
• psutil : Bilgisayarımızda ne kadar RAM olduğunu kontrol etmemizi sağlar.

Aşağıdaki komutu çalıştırarak tüm kütüphaneleri kurabilirsiniz: pip install torch diffusers transformers sentencepiece protobuf accelerate fastapi uvicorn .

M1 veya M2 çipli bir Mac kullanıyorsanız, optimum performans için PyTorch with Metal'i kurmalısınız. Devam etmeden önce resmi PyTorch with Metal rehberini takip edin.

Ayrıca, FLUX'u bir GPU aygıtında çalıştırmayı planlıyorsanız en az 12 GB VRAM'e sahip olduğunuzdan emin olmanız gerekir. Veya CPU/MPS'de (daha yavaş olacaktır) çalıştırmak için en az 12 GB RAM.

Adım 1: Ortamı Kurma

Kullandığımız donanıma göre çıkarımı çalıştırmak için doğru cihazı seçerek senaryoya başlayalım.

 device = 'cuda' # can also be 'cpu' or 'mps' import os # MPS support in PyTorch is not yet fully implemented if device == 'mps': os.environ["PYTORCH_ENABLE_MPS_FALLBACK"] = "1" import torch 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 GPU'ları için) veya mps (Apple'ın Metal Performans Gölgelendiricileri için) belirtebilirsiniz. Daha sonra komut dosyası seçili aygıtın kullanılabilir olup olmadığını kontrol eder ve kullanılabilir değilse bir istisna oluşturur.

Adım 2: FLUX Modelini Yükleme

Sonra, FLUX modelini yükleyeceğiz. Modeli fp16 hassasiyetinde yükleyeceğiz, bu da kalitede çok fazla kayıp olmadan bize biraz bellek kazandıracak.

Bu noktada, FLUX modeli kapalı olduğundan, HuggingFace ile kimlik doğrulaması yapmanız istenebilir. Başarılı bir şekilde kimlik doğrulaması yapmak için bir HuggingFace hesabı oluşturmanız, model sayfasına gitmeniz, şartları kabul etmeniz ve ardından hesap ayarlarınızdan bir HuggingFace belirteci oluşturmanız ve bunu makinenize HF_TOKEN ortam değişkeni olarak eklemeniz gerekir.

 from diffusers import FlowMatchEulerDiscreteScheduler, FluxPipeline import psutil model_name = "black-forest-labs/FLUX.1-dev" print(f"Loading {model_name} on {device}") pipeline = FluxPipeline.from_pretrained( model_name, # Diffusion models are generally trained on fp32, but fp16 # gets us 99% there in terms of quality, with just half the (V)RAM torch_dtype=torch.float16, # Ensure we don't load any dangerous binary code use_safetensors=True # We are using Euler here, but you can also use other samplers scheduler=FlowMatchEulerDiscreteScheduler() ).to(device)

Burada, difüzör kütüphanesini kullanarak FLUX modelini yüklüyoruz. Kullandığımız model black-forest-labs/FLUX.1-dev , fp16 hassasiyetinde yüklendi.

Ayrıca, daha hızlı çıkarımlara sahip ancak daha az ayrıntılı görüntüler üreten FLUX Schnell adlı zaman adımlı bir model ve kapalı kaynaklı bir FLUX Pro modeli de vardır. Burada Euler zamanlayıcısını kullanacağız ancak bunu deneyebilirsiniz. Zamanlayıcılar hakkında daha fazla bilgiyi burada okuyabilirsiniz. Görüntü oluşturma kaynak yoğun olabileceğinden, özellikle sınırlı belleğe sahip bir CPU veya cihazda çalışırken bellek kullanımını optimize etmek çok önemlidir.

 # 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()

Bu kod toplam kullanılabilir belleği kontrol eder ve sistemde 64 GB'den az RAM varsa dikkat dilimlemeyi etkinleştirir. Dikkat dilimleme, sınırlı kaynaklara sahip cihazlar için önemli olan görüntü oluşturma sırasında bellek kullanımını azaltır.

Adım 3: FastAPI ile API'yi oluşturma

Daha sonra, görselleri oluşturmak için bir API sağlayacak olan FastAPI sunucusunu kuracağız.

 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() # We will be returning the image as a base64 encoded string # which we will want compressed app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=7)

FastAPI, Python ile web API'leri oluşturmak için popüler bir çerçevedir. Bu durumda, görüntü oluşturma isteklerini kabul edebilen bir sunucu oluşturmak için kullanıyoruz. Ayrıca, özellikle görüntüleri base64 biçiminde geri gönderirken yararlı olan yanıtı sıkıştırmak için GZip ara yazılımını kullanıyoruz.

Üretim ortamında, CDN ve diğer iyileştirmelerden yararlanmak için, oluşturulan görüntüleri bir S3 kovasında veya başka bir bulut depolama alanında depolamak ve base64 kodlu dizeler yerine URL'leri döndürmek isteyebilirsiniz.

Adım 4: İstek Modelini Tanımlama

Şimdi API'mizin kabul edeceği istekler için bir model tanımlamamız gerekiyor.

 class GenerateRequest(BaseModel): 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")

Bu GenerateRequest modeli bir görüntü oluşturmak için gereken parametreleri tanımlar. prompt alanı, oluşturmak istediğiniz görüntünün metin açıklamasıdır. Diğer alanlar görüntü boyutlarını, çıkarım adımlarının sayısını ve toplu boyutu içerir.

Adım 5: Görüntü Oluşturma Son Noktasını Oluşturma

Şimdi, görüntü oluşturma isteklerini karşılayacak uç noktayı oluşturalım.

 @app.post("/") async def generate_image(request: GenerateRequest): # Validate that height and width are multiples of 8 # as required by FLUX 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") # Always calculate the seed on CPU for deterministic RNG # For a batch of images, seeds will be sequential like n, n+1, n+2, ... 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, generator=generator, num_inference_steps=request.steps, guidance_scale=request.cfg, num_images_per_prompt=request.batch_size ).images # Convert images to base64 strings # (for a production app, you might want to store the # images in an S3 bucket and return the URLs instead) 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, }

Bu uç nokta görüntü oluşturma sürecini yönetir. İlk olarak, FLUX'un gerektirdiği gibi yükseklik ve genişliğin 8'in katları olduğunu doğrular. Daha sonra sağlanan istemi temel alarak görüntüler oluşturur ve bunları base64 kodlu dizeler olarak döndürür.

Adım 6: Sunucuyu Başlatma

Son olarak script çalıştırıldığında sunucunun başlatılmasını sağlayacak bir kod ekleyelim.

 @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)

Bu kod, FastAPI sunucusunu 8000 numaralı portta başlatır ve yalnızca http://localhost:8000 adresinden değil, aynı zamanda 0.0.0.0 bağlaması sayesinde ana makinenin IP adresini kullanan aynı ağdaki diğer cihazlardan da erişilebilir hale getirir.

Adım 7: Sunucunuzu Yerel Olarak Test Etme

Artık FLUX sunucunuz çalışır durumda olduğuna göre, onu test etme zamanı geldi. HTTP istekleri yapmak için bir komut satırı aracı olan curl kullanarak sunucunuzla etkileşime girebilirsiniz:

 curl -X POST "http://localhost:8000/" \ -H "Content-Type: application/json" \ -d '{ "prompt": "A futuristic cityscape at sunset", "seed": 42, "height": 1024, "width": 1024, "cfg": 3.5, "steps": 50, "batch_size": 1 }' | jq -r '.images[0]' | base64 -d > test.png

Bu komut yalnızca curl , jq ve base64 yardımcı programlarının yüklü olduğu UNIX tabanlı sistemlerde çalışacaktır. Ayrıca FLUX sunucusunu barındıran donanıma bağlı olarak tamamlanması birkaç dakika sürebilir.

Çözüm

Tebrikler! Python kullanarak kendi FLUX sunucunuzu başarıyla oluşturdunuz. Bu kurulum, basit bir API aracılığıyla metin istemlerine dayalı görseller oluşturmanıza olanak tanır. Temel FLUX modelinin sonuçlarından memnun değilseniz, belirli kullanım durumlarında daha iyi performans için modeli ince ayarlamayı düşünebilirsiniz .

Tam kod

Bu kılavuzda kullanılan kodun tamamını aşağıda bulabilirsiniz:

 device = 'cuda' # can also be 'cpu' or 'mps' import os # MPS support in PyTorch is not yet fully implemented if device == 'mps': os.environ["PYTORCH_ENABLE_MPS_FALLBACK"] = "1" import torch 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 FlowMatchEulerDiscreteScheduler, FluxPipeline import psutil model_name = "black-forest-labs/FLUX.1-dev" print(f"Loading {model_name} on {device}") pipeline = FluxPipeline.from_pretrained( model_name, # Diffusion models are generally trained on fp32, but fp16 # gets us 99% there in terms of quality, with just half the (V)RAM torch_dtype=torch.float16, # Ensure we don't load any dangerous binary code use_safetensors=True, # We are using Euler here, but you can also use other samplers scheduler=FlowMatchEulerDiscreteScheduler() ).to(device) # 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() 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() # We will be returning the image as a base64 encoded string # which we will want compressed app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=7) class GenerateRequest(BaseModel): 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): # Validate that height and width are multiples of 8 # as required by FLUX 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") # Always calculate the seed on CPU for deterministic RNG # For a batch of images, seeds will be sequential like n, n+1, n+2, ... 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, generator=generator, num_inference_steps=request.steps, guidance_scale=request.cfg, num_images_per_prompt=request.batch_size ).images # Convert images to base64 strings # (for a production app, you might want to store the # images in an S3 bucket and return the URL's instead) 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)