Accelerate documentation
Accelerate ile gradyan biriktirme
Accelerate ile gradyan biriktirme
Gradient accumulation, makinenizin belleğine sığamayacak kadar büyük batch boyutlarında eğitim yapmanızı sağlayan bir tekniktir. Bu, gradientlerin birkaç batch boyunca biriktirilmesi ve belirli sayıda batch tamamlandıktan sonra optimizer'ın adım atmasıyla gerçekleştirilir.
Teknik olarak standart gradyan biriktirme kodu dağıtık bir kurulumda sorunsuz çalışsa da, bunu yapmanın en verimli yöntemi değildir ve önemli yavaşlamalarla karşılaşabilirsiniz!
Bu eğitimde, Accelerate içinde sağlanan yardımcılarla gradyan biriktirmeyi (gradient accumulation) nasıl hızlıca kurup uygulayacağınızı göreceksiniz; bu, yalnızca tek bir yeni kod satırı eklemeye denk gelebilir!
Bu örnek, her iki batch'te bir gradyan biriktirme (gradient accumulation) yapan son derece basit bir PyTorch eğitim döngüsü kullanacaktır:
device = "cuda"
model.to(device)
gradient_accumulation_steps = 2
for index, batch in enumerate(training_dataloader):
inputs, targets = batch
inputs = inputs.to(device)
targets = targets.to(device)
outputs = model(inputs)
loss = loss_function(outputs, targets)
loss = loss / gradient_accumulation_steps
loss.backward()
if (index + 1) % gradient_accumulation_steps == 0:
optimizer.step()
scheduler.step()
optimizer.zero_grad()Bunu Accelerate'e dönüştürme
Önce daha önce gösterilen kod, özel gradyan biriktirme yardımcısı olmadan Accelerate kullanacak şekilde dönüştürülecektir:
+ from accelerate import Accelerator
+ accelerator = Accelerator()
+ model, optimizer, training_dataloader, scheduler = accelerator.prepare(
+ model, optimizer, training_dataloader, scheduler
+ )
for index, batch in enumerate(training_dataloader):
inputs, targets = batch
- inputs = inputs.to(device)
- targets = targets.to(device)
outputs = model(inputs)
loss = loss_function(outputs, targets)
loss = loss / gradient_accumulation_steps
+ accelerator.backward(loss)
if (index+1) % gradient_accumulation_steps == 0:
optimizer.step()
scheduler.step()
optimizer.zero_grad()Mevcut haliyle bu kod, Kavramlar öğreticisinde daha fazla bilgi edinebileceğiniz gradyan senkronizasyonu adı verilen bir süreç nedeniyle gradyan biriktirmeyi verimli bir şekilde gerçekleştirmeyecektir.
Gradient birikimini Accelerate'in yönetmesi
Artık geriye kalan tek şey, gradyan birikimini bizim için Accelerator'a devretmektir. Bunu yapmak için, her step() çağrısından önce kaç adım yapılacağını ve backward() çağrısı sırasında kaybın nasıl otomatik olarak ayarlanacağını belirleyen gradient_accumulation_steps parametresini iletmeniz gerekir:
from accelerate import Accelerator
- accelerator = Accelerator()
+ accelerator = Accelerator(gradient_accumulation_steps=2)Alternatif olarak, Accelerator nesnesinin __init__ metoduna gradient_accumulation_plugin parametresini ileterek gradyan biriktirme davranışını daha da özelleştirebilirsiniz.
Buna dair daha fazla bilgi için GradientAccumulationPlugin belgelerine göz atın.
Buradan itibaren, eğitim döngünüzün içinden accumulate() bağlam yöneticisini kullanarak gradyan birikimlerini sizin için otomatik olarak gerçekleştirebilirsiniz! Bunu, kodumuzun eğitim kısmının tamamının etrafına sarmalmanız yeterlidir:
- for index, batch in enumerate(training_dataloader):
+ for batch in training_dataloader:
+ with accelerator.accumulate(model):
inputs, targets = batch
outputs = model(inputs)Adım sayısı ve kayıp ayarlaması için tüm özel kontrolleri kaldırabilirsiniz:
- loss = loss / gradient_accumulation_steps
accelerator.backward(loss)
- if (index+1) % gradient_accumulation_steps == 0:
optimizer.step()
scheduler.step()
optimizer.zero_grad()Gördüğünüz gibi Accelerator, üzerinde olduğunuz parti numarasını takip edebilir ve hazırlanmış optimizasyonu adım adım ilerletip ilerletmeyeceğini ve kaybı nasıl ayarlayacağını otomatik olarak bilir.
Genellikle gradyan biriktirme ile, eğitim yaptığınız toplam batch sayısındaki değişikliği yansıtacak şekilde adım sayısını ayarlamanız gerekir. Accelerate bunu varsayılan olarak sizin için otomatik olarak yapar. Arka planda, bunu yapmak üzere yapılandırılmış bir
GradientAccumulationPluginörneği oluşturulur.
state.GradientState, üzerinde iterasyon yapılan aktif dataloader ile senkronize edilir. Bu nedenle, dataloaderın sonuna ulaşıldığında her şeyin senkronize olacağı ve bir adımın gerçekleştirileceği varsayılır. Bunu devre dışı bırakmak için
GradientAccumulationPluginiçindesync_with_dataloaderdeğeriniFalseolarak ayarlayın:from accelerate import Accelerator from accelerate.utils import GradientAccumulationPlugin plugin = GradientAccumulationPlugin(sync_with_dataloader=False) accelerator = Accelerator(..., gradient_accumulation_plugin=plugin)
Tamamlanan kod
Aşağıda, Accelerate ile gradyan biriktirme (gradient accumulation) gerçekleştirmek için tamamlanmış uygulama yer almaktadır
from accelerate import Accelerator
accelerator = Accelerator(gradient_accumulation_steps=2)
model, optimizer, training_dataloader, scheduler = accelerator.prepare(
model, optimizer, training_dataloader, scheduler
)
for batch in training_dataloader:
with accelerator.accumulate(model):
inputs, targets = batch
outputs = model(inputs)
loss = loss_function(outputs, targets)
accelerator.backward(loss)
optimizer.step()
scheduler.step()
optimizer.zero_grad()sadece bir ileri/geri adımın bağlam yöneticisi
with accelerator.accumulate(model)içinde yapılması önemlidir.
Bu sihrin arkasındaki mekanizmayı daha ayrıntılı öğrenmek için Gradient Synchronization kavram kılavuzunu okuyun.
Kendi içinde tam örnek
Aşağıda, Accelerate ile gradyan birikiminin nasıl çalıştığını görmek için çalıştırabileceğiniz bağımsız bir örnek bulunmaktadır:
import torch
import copy
from accelerate import Accelerator
from accelerate.utils import set_seed
from torch.utils.data import TensorDataset, DataLoader
# seed
set_seed(0)
# define toy inputs and labels
x = torch.tensor([1., 2., 3., 4., 5., 6., 7., 8.])
y = torch.tensor([2., 4., 6., 8., 10., 12., 14., 16.])
gradient_accumulation_steps = 4
per_device_batch_size = len(x) // gradient_accumulation_steps
# define dataset and dataloader
dataset = TensorDataset(x, y)
dataloader = DataLoader(dataset, batch_size=per_device_batch_size)
# define model, optimizer and loss function
class SimpleLinearModel(torch.nn.Module):
def __init__(self):
super(SimpleLinearModel, self).__init__()
self.weight = torch.nn.Parameter(torch.zeros((1, 1)))
def forward(self, inputs):
return inputs @ self.weight
model = SimpleLinearModel()
model_clone = copy.deepcopy(model)
criterion = torch.nn.MSELoss()
model_optimizer = torch.optim.SGD(model.parameters(), lr=0.02)
accelerator = Accelerator(gradient_accumulation_steps=gradient_accumulation_steps)
model, model_optimizer, dataloader = accelerator.prepare(model, model_optimizer, dataloader)
model_clone_optimizer = torch.optim.SGD(model_clone.parameters(), lr=0.02)
print(f"initial model weight is {model.weight.mean().item():.5f}")
print(f"initial model weight is {model_clone.weight.mean().item():.5f}")
for i, (inputs, labels) in enumerate(dataloader):
with accelerator.accumulate(model):
inputs = inputs.view(-1, 1)
print(i, inputs.flatten())
labels = labels.view(-1, 1)
outputs = model(inputs)
loss = criterion(outputs, labels)
accelerator.backward(loss)
model_optimizer.step()
model_optimizer.zero_grad()
loss = criterion(x.view(-1, 1) @ model_clone.weight, y.view(-1, 1))
model_clone_optimizer.zero_grad()
loss.backward()
model_clone_optimizer.step()
print(f"w/ accumulation, the final model weight is {model.weight.mean().item():.5f}")
print(f"w/o accumulation, the final model weight is {model_clone.weight.mean().item():.5f}")initial model weight is 0.00000
initial model weight is 0.00000
0 tensor([1., 2.])
1 tensor([3., 4.])
2 tensor([5., 6.])
3 tensor([7., 8.])
w/ accumulation, the final model weight is 2.04000
w/o accumulation, the final model weight is 2.04000Değişken boyutlu eğitim örneklerinde gradyan biriktirme
Bu blog yazısında da belirtildiği gibi, değişken boyutlu eğitim örneklerinde gradyan biriktirme yapılırken ortaya çıkan yaygın bir hata:
[…] token düzeyindeki görevlerde (ör. nedensel LM eğitimi) gradyan biriktirme için, doğru kayıp, gradyan biriktirme adımındaki tüm toplulardaki toplam kayıp değerinin, o toplulardaki tüm dolgu olmayan tokenların toplam sayısı değerine bölünmesiyle hesaplanmalıdır. Bu, topluluk başına kayıp değerlerinin ortalamasıyla aynı değildir.
Diğer bir deyişle, token düzeyinde çalışan kayıplarda bazı ayarlamalar yapılmalıdır.
Kayba gerçekten katkıda bulunan hedefleri sayın. Etiketleri içsel olarak kaydıran Transformers nedensel dil modelleri için, kaydırılmamış etiketleri iletin ve (labels[..., 1:] != -100).sum() ifadesini sayın: ilk etiketin öncesi bir tahmin yoktur. Bu etiketleri veri toplayıcıda (data collator) bir kez daha kaydırmayın. otoregresif birikim örneği bu kuralı kullanır; diğer kayıp fonksiyonları farklı hedef hizalamaları kullanabilir.
İskelet kod
from accelerate import Accelerator
import math
import contextlib
gradient_accumulation_steps = 2
accelerator = Accelerator(gradient_accumulation_steps=gradient_accumulation_steps)
model, optimizer, training_dataloader, scheduler = accelerator.prepare(
model, optimizer, training_dataloader, scheduler
)
training_iterator = iter(training_dataloader)
num_samples_in_epoch = len(training_dataloader)
remainder = num_samples_in_epoch % gradient_accumulation_steps
remainder = remainder if remainder != 0 else gradient_accumulation_steps
total_updates = math.ceil(num_samples_in_epoch / gradient_accumulation_steps)
total_batched_samples = 0
for update_step in range(total_updates):
# In order to correctly the total number of non-padded tokens on which we'll compute the cross-entropy loss
# we need to pre-load the full local batch - i.e the next per_device_batch_size * accumulation_steps samples
batch_samples = []
num_batches_in_step = gradient_accumulation_steps if update_step != (total_updates - 1) else remainder
for _ in range(num_batches_in_step):
batch_samples += [next(training_iterator)]
# get local num items in batch
num_items_in_batch = sum([(batch["labels"].ne(-100)).sum() for batch in batch_samples])
# to compute it correctly in a multi-device DDP training, we need to gather the total number of items in the full batch.
num_items_in_batch = accelerator.gather(num_items_in_batch).sum().item()
for i, batch in enumerate(batch_samples):
# if we perform gradient accumulation in a multi-devices set-up, we want to avoid unnecessary communications when accumulating
# cf: https://muellerzr.github.io/blog/gradient_accumulation.html
if (i < len(batch_samples) - 1 and accelerator.num_processes > 1):
ctx = model.no_sync
else:
ctx = contextlib.nullcontext
total_batched_samples += 1
with ctx():
inputs, targets = batch
outputs = model(inputs)
loss = loss_function(outputs, targets) # the loss function should sum over samples rather than averaging
# We multiply by num_processes because the DDP calculates the average gradient across all devices whereas dividing by num_items_in_batch already takes into account all devices
# Same reason for gradient_accumulation_steps, but this times it's Accelerate that calculate the average gradient across the accumulated steps
loss = (loss * gradient_accumulation_steps * accelerator.num_processes) / num_items_in_batch
accelerator.backward(loss)
# Sync gradients and perform optimization steps once every gradient_accumulation_steps
optimizer.step()
scheduler.step()
optimizer.zero_grad()Kendi içinde tamamlanmış nedensel LM örneği
import torch
import copy
from accelerate import Accelerator
from accelerate.utils import set_seed
from accelerate.logging import get_logger
from torch.utils.data import Dataset, DataLoader
import math
import contexlib
# seed
set_seed(0)
logger = get_logger(__name__)
class MyDataset(Dataset):
def __init__(self, num_samples):
super().__init__()
self.len = num_samples
def __getitem__(self, index):
input_ids = torch.arange(1, index+2, dtype=torch.float32)
labels = torch.remainder(input_ids, 2)
return {"input_ids": input_ids, "labels": labels}
def __len__(self):
return self.len
def collate_fn(features):
input_ids = torch.nn.utils.rnn.pad_sequence([f["input_ids"] for f in features], batch_first=True, padding_value=-100)
labels = torch.nn.utils.rnn.pad_sequence([f["labels"] for f in features], batch_first=True, padding_value=-100)
return {"input_ids": input_ids[..., None], "labels": labels[..., None]}
# define toy inputs and labels
gradient_accumulation_steps = 2
per_device_batch_size = 4
# define accelerator
accelerator = Accelerator(gradient_accumulation_steps=gradient_accumulation_steps)
# define dataset and dataloader
# for this toy example, we'll compute gradient descent over one single global batch
dataset = MyDataset(per_device_batch_size*gradient_accumulation_steps*accelerator.num_processes)
dataloader = DataLoader(dataset, batch_size=per_device_batch_size, collate_fn=collate_fn)
# define model, model_optimizer and loss function
model = torch.nn.Linear(1, 2, bias=False)
model_clone = copy.deepcopy(model)
criterion = torch.nn.CrossEntropyLoss(reduction="sum") # must sum over samples rather than averaging
model_optimizer = torch.optim.SGD(model.parameters(), lr=0.08)
logger.warning(f"initial model weight is {model.weight.detach().cpu().squeeze()}")
logger.warning(f"initial model clone weight is {model_clone.weight.detach().cpu().squeeze()}")
# prepare artifacts - accelerator handles device placement and dataloader splitting
model, model_optimizer = accelerator.prepare(model, model_optimizer)
dataloader = accelerator.prepare_data_loader(dataloader, device_placement=True)
training_iterator = iter(dataloader)
num_samples_in_epoch = len(dataloader)
remainder = num_samples_in_epoch % gradient_accumulation_steps
remainder = remainder if remainder != 0 else gradient_accumulation_steps
total_gradient_updates = math.ceil(num_samples_in_epoch / gradient_accumulation_steps)
total_batched_samples = 0
for update_step in range(total_gradient_updates):
# In order to correctly the total number of non-padded tokens on which we'll compute the cross-entropy loss
# we need to pre-load the full local batch - i.e the next per_device_batch_size * accumulation_steps samples
batch_samples = []
num_batches_in_step = gradient_accumulation_steps if update_step != (total_gradient_updates - 1) else remainder
for _ in range(num_batches_in_step):
batch_samples += [next(training_iterator)]
# get local num items in batch
local_num_items_in_batch = sum([(batch["labels"].ne(-100)).sum() for batch in batch_samples])
logger.warning(f"Step {update_step} - Device {accelerator.process_index} - num items in the local batch {local_num_items_in_batch}", main_process_only=False)
# to compute it correctly in a multi-device DDP training, we need to gather the total number of items in the full batch.
num_items_in_batch = accelerator.gather(local_num_items_in_batch).sum().item()
logger.warning(f"Total num items {num_items_in_batch}")
for i, batch in enumerate(batch_samples):
inputs, labels = batch["input_ids"], batch["labels"]
total_batched_samples += 1
# if we perform gradient accumulation in a multi-devices set-up, we want to avoid unnecessary communications when accumulating
# cf: https://muellerzr.github.io/blog/gradient_accumulation.html
if (i < len(batch_samples) - 1 and accelerator.num_processes > 1):
ctx = model.no_sync
else:
ctx = contextlib.nullcontext
with ctx():
outputs = model(inputs)
loss = criterion(outputs.view(-1, 2), labels.view(-1).to(torch.int64))
# We multiply by num_processes because the DDP calculates the average gradient across all devices whereas dividing by num_items_in_batch already takes into account all devices
# Same reason for gradient_accumulation_steps, but this times it's Accelerate that calculate the average gradient across the accumulated steps
loss = (loss * gradient_accumulation_steps * accelerator.num_processes) / num_items_in_batch
accelerator.backward(loss)
model_optimizer.step()
model_optimizer.zero_grad()
logger.warning(f"Device {accelerator.process_index} - w/ accumulation, the final model weight is {accelerator.unwrap_model(model).weight.detach().cpu().squeeze()}", main_process_only=False)
# We know do the same operation but on a single device and without gradient accumulation
if accelerator.is_main_process:
# prepare one single entire batch
dataloader = DataLoader(dataset, batch_size=len(dataset), collate_fn=collate_fn)
full_batch_without_accum = next(iter(dataloader))
total_inputs, total_labels = full_batch_without_accum["input_ids"], full_batch_without_accum["labels"]
model_clone_optimizer = torch.optim.SGD(model_clone.parameters(), lr=0.08)
# train the cloned model
loss = torch.nn.CrossEntropyLoss(reduction="mean")(model_clone(total_inputs).view(-1, 2), total_labels.view(-1).to(torch.int64))
model_clone_optimizer.zero_grad()
loss.backward()
model_clone_optimizer.step()
# We should have the same final weights.
logger.warning(f"w/o accumulation, the final model weight is {model_clone.weight.detach().cpu().squeeze()}")
Tek bir cihazda sonuçlar - gradient accumulation steps 1 ve batch_size 8 olarak ayarlandı:
initial model weight is tensor([-0.0075, 0.5364])
initial model clone weight is tensor([-0.0075, 0.5364])
Step 0 - Device 0 - num items in the local batch 36
Total num items 36
Device 0 - w/ accumulation, the final model weight is tensor([0.0953, 0.4337])
w/o accumulation, the final model weight is tensor([0.0953, 0.4337])İki cihazlı bir kurulumda sonuçlar - gradyan birikim adımları 2 olarak ve batch_size 4 olarak ayarlandı.
initial model weight is tensor([-0.0075, 0.5364])
initial model clone weight is tensor([-0.0075, 0.5364])
Step 0 - Device 0 - num items in the local batch 52
Step 0 - Device 1 - num items in the local batch 84
Total num items 136
Device 1 - w/ accumulation, the final model weight is tensor([0.2117, 0.3172])
Device 0 - w/ accumulation, the final model weight is tensor([0.2117, 0.3172])
w/o accumulation, the final model weight is tensor([0.2117, 0.3172])Daha ileri gitmek için:
Gerçek dünya eğitim çalışması için tam bir örnek betiği, accelerate/examples/by_feature/gradient_accumulation_for_autoregressive_models.py yolundaki examples klasöründe bulabilirsiniz.
Sabit küresel parti boyutu 32 olacak şekilde birkaç eğitim yapılandırmasında çalıştırıldığında aşağıdaki grafik elde edilir:
Eğitim kayıplarının 20. eğitim adımına kadar tamamen aynı olduğunu not edin. Bu eğitim adımından sonraki küçük sapma, varsayılan olarak toplam parti boyutu veri setini tam olarak bölmediğinde veri yükleyicinin veri setinin başındaki örnekleri kopyaladığı için ilk dönemin en sonunda gerçekleşir.
Update on GitHub