FLUX, Python және диффузорлары бар AI-мен жұмыс істейтін Image Generation API қызметі: жылдам нұсқаулық

Бұл мақалада біз сізге Python көмегімен жеке FLUX серверін жасау арқылы таныстырамыз. Бұл сервер қарапайым API арқылы мәтіндік шақыруларға негізделген кескіндерді жасауға мүмкіндік береді. Бұл серверді жеке пайдалану үшін іске қосып жатсаңыз немесе оны өндірістік қолданбаның бөлігі ретінде қолданасыз ба, бұл нұсқаулық сізге бастауға көмектеседі.

FLUX ( Black Forest Labs ) соңғы бірнеше айда AI кескінін жасау әлемін жаулап алды. Ол көптеген көрсеткіштер бойынша тұрақты диффузияны (бұрынғы ашық бастапқы король) жеңіп қана қоймай, сонымен қатар кейбір көрсеткіштер бойынша Dall-E немесе Midjourney сияқты меншікті үлгілерден асып түсті.

Бірақ FLUX қолданбасын қолданбаларыңыздың бірінде қалай пайдалану керек еді? Replicate және басқалары сияқты серверсіз хосттарды пайдалану туралы ойлауға болады, бірақ олар өте тез қымбатқа түсуі мүмкін және сізге қажетті икемділікті қамтамасыз етпеуі мүмкін. Міне, өзіңіздің жеке FLUX серверіңізді жасау ыңғайлы болады.

Алғы шарттар

Кодқа кіріспес бұрын, сізде қажетті құралдар мен кітапханалар орнатылғанына көз жеткізейік:

• Python: Құрылғыда орнатылған Python 3, жақсырақ 3.10 нұсқасы қажет.
• torch : FLUX іске қосу үшін біз қолданатын терең оқыту жүйесі.
• diffusers : FLUX үлгісіне қол жеткізуді қамтамасыз етеді.
• transformers : диффузорлардың қажетті тәуелділігі.
• sentencepiece : FLUX токенизаторын іске қосу үшін қажет
• protobuf : FLUX іске қосу үшін қажет
• accelerate : кейбір жағдайларда FLUX үлгісін тиімдірек жүктеуге көмектеседі.
• fastapi : кескін жасау сұрауларын қабылдай алатын веб-серверді жасауға арналған жақтау.
• uvicorn : FastAPI серверін іске қосу үшін қажет.
• 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")

cpu , cuda (NVIDIA GPU үшін) немесе mps (Apple металл өнімділігі шейдерлері үшін) көрсетуге болады. Содан кейін сценарий таңдалған құрылғының қолжетімділігін тексереді және ол жоқ болса, ерекше жағдайды тудырады.

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 портында іске қосады, бұл оны тек http://localhost:8000 сайтынан ғана емес, сонымен қатар 0.0.0.0 байланыстыруының арқасында басты компьютердің IP мекенжайын пайдаланатын сол желідегі басқа құрылғылардан да қолжетімді етеді.

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

Бұл пәрмен curl , jq және base64 утилиталары орнатылған UNIX негізіндегі жүйелерде ғана жұмыс істейді. Сондай-ақ, FLUX серверін орналастыратын аппараттық құралға байланысты аяқтауға бірнеше минут кетуі мүмкін.

Қорытынды

Құттықтаймыз! 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)