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 ( 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 veya gibi tescilli modelleri de geride bıraktı. Black Forest Labs Dall-E Midjourney Peki FLUX'u uygulamalarınızdan birinde nasıl kullanırsınız? 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. Replicate Ö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. : FLUX'u çalıştırmak için kullanacağımız derin öğrenme çerçevesi. torch : FLUX modeline erişim sağlar. diffusers : Difüzörlerin zorunlu bağımlılığı. transformers : FLUX belirteçleyicisini çalıştırmak için gereklidir sentencepiece : FLUX'u çalıştırmak için gereklidir protobuf : Bazı durumlarda FLUX modelinin daha verimli bir şekilde yüklenmesine yardımcı olur. accelerate : Görüntü oluşturma isteklerini kabul edebilen bir web sunucusu oluşturmak için kullanılan çerçeve. fastapi : FastAPI sunucusunu çalıştırmak için gereklidir. uvicorn : Bilgisayarımızda ne kadar RAM olduğunu kontrol etmemizi sağlar. psutil 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 takip edin. resmi PyTorch with Metal rehberini 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") , (NVIDIA GPU'ları için) veya (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. cpu cuda mps 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 ortam değişkeni olarak eklemeniz gerekir. HF_TOKEN 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 , fp16 hassasiyetinde yüklendi. black-forest-labs/FLUX.1-dev 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 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. burada # 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 modeli bir görüntü oluşturmak için gereken parametreleri tanımlar. 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. GenerateRequest prompt 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 adresinden değil, aynı zamanda bağlaması sayesinde ana makinenin IP adresini kullanan aynı ağdaki diğer cihazlardan da erişilebilir hale getirir. http://localhost:8000 0.0.0.0 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 kullanarak sunucunuzla etkileşime girebilirsiniz: curl 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 , ve 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. curl jq base64 Çö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)
