Бұл мақалада біз сізге Python көмегімен жеке FLUX серверін жасау арқылы таныстырамыз. Бұл сервер қарапайым API арқылы мәтіндік шақыруларға негізделген кескіндерді жасауға мүмкіндік береді. Бұл серверді жеке пайдалану үшін іске қосып жатсаңыз немесе оны өндірістік қолданбаның бөлігі ретінде қолданасыз ба, бұл нұсқаулық сізге бастауға көмектеседі. FLUX ( ) соңғы бірнеше айда AI кескінін жасау әлемін жаулап алды. Ол көптеген көрсеткіштер бойынша тұрақты диффузияны (бұрынғы ашық бастапқы король) жеңіп қана қоймай, сонымен қатар кейбір көрсеткіштер бойынша немесе сияқты меншікті үлгілерден асып түсті. Black Forest Labs Dall-E Midjourney Бірақ FLUX қолданбасын қолданбаларыңыздың бірінде қалай пайдалану керек еді? және басқалары сияқты серверсіз хосттарды пайдалану туралы ойлауға болады, бірақ олар өте тез қымбатқа түсуі мүмкін және сізге қажетті икемділікті қамтамасыз етпеуі мүмкін. Міне, өзіңіздің жеке FLUX серверіңізді жасау ыңғайлы болады. Replicate Алғы шарттар Кодқа кіріспес бұрын, сізде қажетті құралдар мен кітапханалар орнатылғанына көз жеткізейік: Python: Құрылғыда орнатылған Python 3, жақсырақ 3.10 нұсқасы қажет. : FLUX іске қосу үшін біз қолданатын терең оқыту жүйесі. torch : FLUX үлгісіне қол жеткізуді қамтамасыз етеді. diffusers : диффузорлардың қажетті тәуелділігі. transformers : FLUX токенизаторын іске қосу үшін қажет sentencepiece : FLUX іске қосу үшін қажет protobuf : кейбір жағдайларда FLUX үлгісін тиімдірек жүктеуге көмектеседі. accelerate : кескін жасау сұрауларын қабылдай алатын веб-серверді жасауға арналған жақтау. fastapi : FastAPI серверін іске қосу үшін қажет. uvicorn : Біздің құрылғыда қанша ЖЖҚ бар екенін тексеруге мүмкіндік береді. psutil Барлық кітапханаларды келесі пәрменді орындау арқылы орнатуға болады: . pip install torch diffusers transformers sentencepiece protobuf accelerate fastapi uvicorn M1 немесе M2 чипі бар Mac қолдансаңыз, оңтайлы өнімділік үшін PyTorch with Metal орнату керек. Жалғастырмас бұрын орындаңыз. ресми PyTorch with Metal нұсқаулығын Сондай-ақ GPU құрылғысында FLUX іске қосуды жоспарласаңыз, сізде кемінде 12 ГБ VRAM бар екеніне көз жеткізуіңіз керек. Немесе CPU/MPS жүйесінде жұмыс істеу үшін кем дегенде 12 ГБ жедел жады (ол баяуырақ болады). 1-қадам: Ортаны орнату Сценарийді біз пайдаланып жатқан аппараттық құрал негізінде қорытынды шығару үшін дұрыс құрылғыны таңдау арқылы бастайық. 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 үшін) немесе (Apple металл өнімділігі шейдерлері үшін) көрсетуге болады. Содан кейін сценарий таңдалған құрылғының қолжетімділігін тексереді және ол жоқ болса, ерекше жағдайды тудырады. cpu cuda mps 2-қадам: FLUX үлгісін жүктеу Содан кейін біз FLUX үлгісін жүктейміз. Біз модельді fp16 дәлдігімен жүктейміз, бұл бізге жадты сапаны жоғалтпай үнемдейді. Осы сәтте сізден HuggingFace көмегімен аутентификация сұралуы мүмкін, себебі FLUX үлгісі жабық. Сәтті аутентификация үшін сізге HuggingFace тіркелгісін жасап, үлгі бетіне өтіп, шарттарды қабылдап, тіркелгі параметрлерінен HuggingFace таңбалауышын жасап, оны ортасының айнымалы мәні ретінде құрылғыңызға қосу керек. 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) Мұнда диффузорлар кітапханасы арқылы FLUX үлгісін жүктеп жатырмыз. Біз қолданып жатқан модель - fp16 дәлдігімен жүктелген . black-forest-labs/FLUX.1-dev Сондай-ақ, FLUX Schnell деп аталатын уақыт қадамы бойынша тазартылған модель бар, ол жылдамырақ қорытынды жасайды, бірақ азырақ егжей-тегжейлі кескіндер шығарады, сонымен қатар жабық бастапқы FLUX Pro моделі бар. Біз осы жерде Эйлер жоспарлаушысын қолданамыз, бірақ сіз онымен тәжірибе жасай аласыз. Жоспарлаушылар туралы толығырақ оқи аласыз. Кескінді жасау ресурсты көп қажет ететіндіктен, жадты пайдалануды оңтайландыру өте маңызды, әсіресе процессорда немесе жады шектеулі құрылғыда жұмыс істегенде. мына жерден # 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-қадам: FastAPI көмегімен API құру Содан кейін кескіндерді жасау үшін API қамтамасыз ететін FastAPI серверін орнатамыз. 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 көмегімен веб API құруға арналған танымал құрылым. Бұл жағдайда біз оны кескін жасау үшін сұрауларды қабылдай алатын сервер жасау үшін пайдаланамыз. Біз сондай-ақ жауапты сығу үшін GZip аралық бағдарламалық құралын қолданамыз, бұл әсіресе суреттерді base64 пішімінде кері жіберу кезінде пайдалы. Өндіріс ортасында жасалған кескіндерді S3 шелекінде немесе басқа бұлттық жадта сақтағыңыз және CDN және басқа оңтайландырулардың артықшылығын пайдалану үшін base64 кодталған жолдардың орнына URL мекенжайларын қайтарғыңыз келуі мүмкін. 4-қадам: Сұраныс үлгісін анықтау Енді API қабылдайтын сұраулар үлгісін анықтауымыз керек. 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") Бұл үлгісі кескінді жасау үшін қажетті параметрлерді анықтайды. өрісі жасағыңыз келетін кескіннің мәтіндік сипаттамасы болып табылады. Басқа өрістер кескін өлшемдерін, қорытынды қадамдарының санын және пакет өлшемін қамтиды. GenerateRequest prompt 5-қадам: Кескінді жасаудың соңғы нүктесін жасау Енді кескінді құру сұрауларын өңдейтін соңғы нүктені жасайық. @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, } Бұл соңғы нүкте кескінді жасау процесін басқарады. Ол алдымен FLUX талап еткендей биіктік пен ені 8-ге еселік екенін растайды. Содан кейін ол берілген шақыруға негізделген кескіндерді жасайды және оларды 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 портында іске қосады, бұл оны тек сайтынан ғана емес, сонымен қатар байланыстыруының арқасында басты компьютердің IP мекенжайын пайдаланатын сол желідегі басқа құрылғылардан да қолжетімді етеді. http://localhost:8000 0.0.0.0 7-қадам: Серверіңізді жергілікті сынау Енді сіздің FLUX серверіңіз жұмыс істеп тұр, оны тексеру уақыты келді. Сервермен өзара әрекеттесу үшін HTTP сұрауларын жасауға арналған пәрмен жолы құралын пайдалануға болады: 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 Бұл пәрмен , және утилиталары орнатылған UNIX негізіндегі жүйелерде ғана жұмыс істейді. Сондай-ақ, FLUX серверін орналастыратын аппараттық құралға байланысты аяқтауға бірнеше минут кетуі мүмкін. curl jq base64 Қорытынды Құттықтаймыз! Python арқылы өзіңіздің FLUX серверіңізді сәтті жасадыңыз. Бұл орнату қарапайым API арқылы мәтіндік шақыруларға негізделген кескіндерді жасауға мүмкіндік береді. Негізгі FLUX үлгісінің нәтижелеріне қанағаттанбасаңыз, . нақты пайдалану жағдайларында одан да жақсырақ жұмыс істеу үшін үлгіні дәл баптау мүмкіндігін қарастыруыңызға болады Толық код Төменде осы нұсқаулықта пайдаланылған толық кодты таба аласыз: 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)
