الاستدلال الموزّع

يمكن تصنيف الاستدلال الموزّع في فئتين:

1. تحميل النموذج بأكمله على كل وحدة معالجة رسومية (GPU) وإرسال أجزاء من الدفعة (batch) عبر نسخة النموذج الموجودة على كل وحدة معالجة رسومية في كل مرة
2. تحميل أجزاء من النموذج على كل وحدة معالجة رسومية ومعالجة إدخال واحد في كل مرة

يتناول هذا الدليل القوس الأول. أما بالنسبة للقوس الثاني، فراجع Big Model Inference.

إرسال أجزاء الدفعة تلقائيًا إلى كل نموذج محمّل

هذا هو الحل الأكثر استهلاكًا للذاكرة، إذ يتطلب من كل وحدة معالجة رسومية (GPU) الاحتفاظ بنسخة كاملة من النموذج في الذاكرة في أي وقت معيّن.

عادةً، عند القيام بذلك، يرسل المستخدمون النموذج إلى جهاز محدد لتحميله من وحدة المعالجة المركزية (CPU)، ثم ينقلون كل موجه (prompt) إلى جهاز مختلف.

قد يبدو خط أنابيب أساسي باستخدام مكتبة diffusers على النحو التالي:

import torch
import torch.distributed as dist
from diffusers import DiffusionPipeline

pipe = DiffusionPipeline.from_pretrained("runwayml/stable-diffusion-v1-5", torch_dtype=torch.float16)

ثم يتم تنفيذ الاستدلال بناءً على الموجه المحدد:

def run_inference(rank, world_size):
    dist.init_process_group("nccl", rank=rank, world_size=world_size)
    pipe.to(rank)

    if torch.distributed.get_rank() == 0:
        prompt = "a dog"
    elif torch.distributed.get_rank() == 1:
        prompt = "a cat"

    result = pipe(prompt).images[0]
    result.save(f"result_{rank}.png")

سيلاحظ المرء أنه يتعين علينا التحقق من الرتبة (rank) لمعرفة أي موجه (prompt) يجب إرساله، وهو ما قد يكون مُتعِبًا بعض الشيء.

قد يظن المستخدم أيضًا أنه باستخدام Accelerate، فإن استخدام Accelerator لتحضير مُحمِّل بيانات لمهمة كهذه قد يكون أيضًا طريقة بسيطة لإدارة هذا الأمر. (لتعلم المزيد، راجع القسم ذي الصلة في الجولة السريعة)

هل يمكنه إدارتها؟ نعم. لكن هل يضيف كودًا إضافيًا غير ضروري؟ نعم أيضًا.

باستخدام Accelerate، يمكننا تبسيط هذه العملية من خلال استخدام مدير السياق Accelerator.split_between_processes() (والذي يوجد أيضاً في PartialState وAcceleratorState). ستقوم هذه الدالة تلقائياً بتقسيم أي بيانات تمررها إليها (سواء كانت موجه، أو مجموعة من الموترات، أو قاموس للبيانات السابقة، إلخ) عبر جميع العمليات (مع إمكانية إضافة حشوة) لتتمكن من استخدامها فوراً.

لنعيد كتابة المثال أعلاه باستخدام مدير السياق هذا:

import torch
from accelerate import PartialState  # Can also be Accelerator or AcceleratorState
from diffusers import DiffusionPipeline

pipe = DiffusionPipeline.from_pretrained("runwayml/stable-diffusion-v1-5", torch_dtype=torch.float16)
distributed_state = PartialState()
pipe.to(distributed_state.device)

# Assume two processes
with distributed_state.split_between_processes(["a dog", "a cat"]) as prompt:
    result = pipe(prompt).images[0]
    result.save(f"result_{distributed_state.process_index}.png")

ثم لتشغيل الكود، يمكننا استخدام Accelerate:

إذا كنت قد أنشأت ملف إعدادات للاستخدام عبر accelerate config:

accelerate launch distributed_inference.py

إذا كان لديك ملف تكوين محدد تريد استخدامه:

accelerate launch --config_file my_config.json distributed_inference.py

أو إذا كنت لا ترغب في إنشاء أي ملفات إعدادات والإطلاق على بطاقتَي GPU:

ملاحظة: ستظهر لك بعض التحذيرات حول القيم التي يتم تخمينها بناءً على نظامك. لإزالة هذه التحذيرات، يمكنك تنفيذ accelerate config default أو المرور عبر accelerate config لإنشاء ملف إعدادات.

accelerate launch --num_processes 2 distributed_inference.py

لقد قلّلنا الآن الكود القياسي المطلوب لتقسيم هذه البيانات إلى بضعة أسطر من الكود بسهولة كبيرة.

لكن ماذا لو كان لدينا توزيع غير متساوٍ للموجّهات على وحدات معالجة الرسوميات؟ على سبيل المثال، ماذا لو كان لدينا 3 موجّهات، لكن 2 وحدات معالجة رسوميات فقط؟

في إطار مدير السياق، تستقبل وحدة المعالجة الرسومية الأولى أولَ طلبين، بينما تستقبل وحدة المعالجة الرسومية الثانية الطلبَ الثالث، مما يضمن تقسيم جميع الطلبات دون الحاجة إلى أي عبء إضافي.

ومع ذلك، ماذا لو أردنا بعد ذلك القيام بشيء ما بنتائج جميع وحدات معالجة الرسومات؟ (على سبيل المثال، جمعها جميعًا وإجراء نوع من المعالجة اللاحقة) يمكنك تمرير apply_padding=True للتأكد من أن قوائم الأوامر مُستكملة حتى الطول نفسه، مع أخذ البيانات الإضافية من العينة الأخيرة. وبهذه الطريقة ستحتوي جميع وحدات معالجة الرسومات على نفس عدد الأوامر، ويمكنك بعد ذلك جمع النتائج.

لا يُطلب هذا إلا عند محاولة تنفيذ إجراء مثل جمع النتائج، حيث يجب أن تكون البيانات على كل جهاز بنفس الطول. لا يتطلب الاستدلال الأساسي هذا.

على سبيل المثال:

import torch
from accelerate import PartialState  # Can also be Accelerator or AcceleratorState
from diffusers import DiffusionPipeline

pipe = DiffusionPipeline.from_pretrained("runwayml/stable-diffusion-v1-5", torch_dtype=torch.float16)
distributed_state = PartialState()
pipe.to(distributed_state.device)

# Assume two processes
with distributed_state.split_between_processes(["a dog", "a cat", "a chicken"], apply_padding=True) as prompt:
    result = pipe(prompt).images

على وحدة معالجة الرسوميات الأولى، ستكون المحفّزات ["a dog", "a cat"]، وعلى وحدة معالجة الرسوميات الثانية ستكون ["a chicken", "a chicken"]. تأكد من حذف العينة الأخيرة، لأنها ستكون نسخة مكررة من العينة السابقة.

يمكنك العثور على أمثلة أكثر تعقيدًا هنا مثل كيفية استخدامه مع نماذج اللغة الكبيرة (LLMs).