| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443344434453446344734483449345034513452345334543455345634573458345934603461346234633464346534663467346834693470347134723473347434753476347734783479348034813482348334843485348634873488348934903491349234933494349534963497349834993500350135023503350435053506350735083509351035113512351335143515351635173518351935203521352235233524352535263527352835293530353135323533353435353536353735383539354035413542354335443545354635473548354935503551355235533554355535563557355835593560356135623563356435653566356735683569357035713572357335743575357635773578357935803581358235833584358535863587358835893590359135923593359435953596359735983599360036013602360336043605360636073608360936103611361236133614361536163617361836193620362136223623362436253626362736283629363036313632363336343635363636373638363936403641364236433644364536463647364836493650365136523653365436553656365736583659366036613662366336643665366636673668366936703671367236733674367536763677367836793680368136823683368436853686368736883689369036913692369336943695369636973698369937003701370237033704370537063707370837093710371137123713371437153716371737183719372037213722372337243725372637273728372937303731373237333734373537363737373837393740374137423743374437453746374737483749375037513752375337543755375637573758375937603761376237633764376537663767376837693770377137723773377437753776377737783779378037813782378337843785378637873788378937903791379237933794379537963797379837993800380138023803380438053806380738083809381038113812381338143815381638173818381938203821382238233824382538263827382838293830383138323833383438353836383738383839384038413842384338443845384638473848384938503851385238533854385538563857385838593860386138623863386438653866386738683869387038713872387338743875387638773878387938803881388238833884388538863887388838893890389138923893389438953896389738983899390039013902390339043905390639073908390939103911391239133914391539163917391839193920392139223923392439253926392739283929393039313932393339343935393639373938393939403941394239433944394539463947394839493950395139523953395439553956395739583959396039613962396339643965396639673968396939703971397239733974397539763977397839793980398139823983398439853986398739883989399039913992399339943995399639973998399940004001400240034004400540064007400840094010401140124013401440154016401740184019402040214022402340244025402640274028402940304031403240334034403540364037403840394040404140424043404440454046404740484049405040514052405340544055405640574058405940604061406240634064406540664067406840694070407140724073407440754076407740784079408040814082408340844085408640874088408940904091409240934094409540964097409840994100410141024103410441054106410741084109411041114112411341144115411641174118411941204121412241234124412541264127412841294130413141324133413441354136413741384139414041414142414341444145414641474148414941504151415241534154415541564157415841594160416141624163416441654166416741684169417041714172417341744175417641774178417941804181418241834184418541864187418841894190419141924193419441954196419741984199420042014202420342044205420642074208420942104211421242134214421542164217421842194220422142224223422442254226422742284229423042314232423342344235423642374238423942404241424242434244424542464247424842494250425142524253425442554256425742584259426042614262426342644265426642674268426942704271427242734274427542764277427842794280428142824283428442854286428742884289429042914292429342944295429642974298429943004301430243034304430543064307430843094310431143124313431443154316431743184319432043214322432343244325432643274328432943304331433243334334433543364337433843394340434143424343434443454346434743484349435043514352435343544355435643574358435943604361436243634364436543664367436843694370437143724373437443754376437743784379438043814382438343844385438643874388438943904391439243934394439543964397439843994400440144024403440444054406440744084409441044114412441344144415441644174418441944204421442244234424442544264427442844294430443144324433443444354436443744384439444044414442444344444445444644474448444944504451445244534454445544564457445844594460446144624463446444654466446744684469447044714472447344744475447644774478447944804481448244834484448544864487448844894490449144924493449444954496449744984499450045014502450345044505450645074508450945104511451245134514451545164517451845194520452145224523452445254526452745284529453045314532453345344535453645374538453945404541454245434544454545464547454845494550455145524553455445554556455745584559456045614562456345644565456645674568456945704571457245734574457545764577457845794580458145824583458445854586458745884589459045914592459345944595459645974598459946004601460246034604460546064607460846094610461146124613461446154616461746184619462046214622462346244625462646274628462946304631463246334634463546364637463846394640464146424643464446454646464746484649465046514652465346544655465646574658465946604661466246634664466546664667466846694670467146724673467446754676467746784679468046814682468346844685468646874688468946904691469246934694469546964697469846994700470147024703470447054706470747084709471047114712471347144715471647174718471947204721472247234724472547264727472847294730473147324733473447354736473747384739474047414742474347444745474647474748474947504751475247534754475547564757475847594760476147624763476447654766476747684769477047714772477347744775477647774778477947804781478247834784478547864787478847894790479147924793479447954796479747984799480048014802480348044805480648074808480948104811481248134814481548164817481848194820482148224823482448254826482748284829483048314832483348344835483648374838483948404841484248434844484548464847484848494850485148524853485448554856485748584859486048614862486348644865486648674868486948704871487248734874487548764877487848794880488148824883488448854886488748884889489048914892489348944895489648974898489949004901490249034904490549064907490849094910491149124913491449154916491749184919492049214922492349244925492649274928492949304931493249334934493549364937493849394940494149424943494449454946494749484949495049514952495349544955495649574958495949604961496249634964496549664967496849694970497149724973497449754976497749784979498049814982498349844985498649874988498949904991499249934994499549964997499849995000500150025003500450055006500750085009501050115012501350145015501650175018501950205021502250235024502550265027502850295030503150325033503450355036503750385039504050415042504350445045504650475048504950505051505250535054505550565057505850595060506150625063506450655066506750685069507050715072507350745075507650775078507950805081508250835084508550865087508850895090509150925093509450955096509750985099510051015102510351045105510651075108510951105111511251135114511551165117511851195120512151225123512451255126512751285129513051315132513351345135513651375138513951405141514251435144514551465147514851495150515151525153515451555156515751585159516051615162516351645165516651675168516951705171517251735174517551765177517851795180518151825183518451855186518751885189519051915192519351945195519651975198519952005201520252035204520552065207520852095210521152125213521452155216521752185219522052215222522352245225522652275228522952305231523252335234523552365237523852395240524152425243524452455246524752485249525052515252525352545255525652575258525952605261526252635264526552665267526852695270527152725273527452755276527752785279528052815282528352845285528652875288528952905291529252935294529552965297529852995300530153025303530453055306530753085309531053115312531353145315531653175318531953205321532253235324532553265327532853295330533153325333533453355336533753385339534053415342534353445345534653475348534953505351535253535354535553565357535853595360536153625363536453655366536753685369537053715372537353745375537653775378537953805381538253835384538553865387538853895390539153925393539453955396539753985399540054015402540354045405540654075408540954105411541254135414541554165417541854195420542154225423542454255426542754285429543054315432543354345435543654375438543954405441544254435444544554465447544854495450545154525453545454555456545754585459546054615462546354645465546654675468546954705471547254735474547554765477547854795480548154825483548454855486548754885489549054915492549354945495549654975498549955005501550255035504550555065507550855095510551155125513551455155516551755185519552055215522552355245525552655275528552955305531553255335534553555365537553855395540554155425543554455455546554755485549555055515552555355545555555655575558555955605561556255635564556555665567556855695570557155725573557455755576557755785579558055815582558355845585558655875588558955905591559255935594559555965597559855995600560156025603560456055606560756085609561056115612561356145615561656175618561956205621562256235624562556265627562856295630563156325633563456355636563756385639564056415642564356445645564656475648564956505651565256535654565556565657565856595660566156625663566456655666566756685669567056715672567356745675567656775678567956805681568256835684568556865687568856895690569156925693569456955696569756985699570057015702570357045705570657075708570957105711571257135714571557165717571857195720572157225723 |
- # Copyright 2020-present the HuggingFace Inc. team.
- #
- # Licensed under the Apache License, Version 2.0 (the "License");
- # you may not use this file except in compliance with the License.
- # You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- """
- The Trainer class, to easily train a 🤗 Transformers from scratch or finetune it on a new task.
- """
- import contextlib
- import copy
- import functools
- import glob
- import importlib.metadata
- import inspect
- import json
- import math
- import os
- import random
- import re
- import shutil
- import sys
- import tempfile
- import time
- import warnings
- from collections.abc import Iterator, Mapping
- from functools import partial
- from pathlib import Path
- from typing import TYPE_CHECKING, Any, Callable, Optional, Union
- # Integrations must be imported before ML frameworks:
- # ruff: isort: off
- from .integrations import (
- get_reporting_integration_callbacks,
- )
- # ruff: isort: on
- import huggingface_hub.utils as hf_hub_utils
- import numpy as np
- import safetensors.torch
- import torch
- import torch.distributed as dist
- from huggingface_hub import ModelCard, create_repo, upload_folder
- from packaging import version
- from torch import nn
- from torch.utils.data import DataLoader, Dataset, IterableDataset, RandomSampler, SequentialSampler
- from . import __version__
- from .configuration_utils import PretrainedConfig
- from .data.data_collator import DataCollator, DataCollatorWithPadding, default_data_collator
- from .debug_utils import DebugOption, DebugUnderflowOverflow
- from .feature_extraction_sequence_utils import SequenceFeatureExtractor
- from .feature_extraction_utils import FeatureExtractionMixin
- from .hyperparameter_search import ALL_HYPERPARAMETER_SEARCH_BACKENDS, default_hp_search_backend
- from .image_processing_utils import BaseImageProcessor
- from .integrations.deepspeed import deepspeed_init, deepspeed_load_checkpoint, is_deepspeed_available
- from .integrations.tpu import tpu_spmd_dataloader
- from .modelcard import TrainingSummary
- from .modeling_utils import PreTrainedModel, load_sharded_checkpoint, unwrap_model
- from .models.auto.modeling_auto import (
- MODEL_FOR_CAUSAL_LM_MAPPING_NAMES,
- MODEL_MAPPING_NAMES,
- )
- from .optimization import Adafactor, get_scheduler
- from .processing_utils import ProcessorMixin
- from .pytorch_utils import (
- is_torch_greater_or_equal_than_2_3,
- )
- from .tokenization_utils_base import PreTrainedTokenizerBase
- from .trainer_callback import (
- CallbackHandler,
- DefaultFlowCallback,
- ExportableState,
- PrinterCallback,
- ProgressCallback,
- TrainerCallback,
- TrainerControl,
- TrainerState,
- )
- from .trainer_pt_utils import (
- DistributedTensorGatherer,
- EvalLoopContainer,
- IterableDatasetShard,
- LabelSmoother,
- LayerWiseDummyOptimizer,
- LengthGroupedSampler,
- SequentialDistributedSampler,
- distributed_broadcast_scalars,
- distributed_concat,
- find_batch_size,
- get_model_param_count,
- get_module_class_from_name,
- get_parameter_names,
- nested_concat,
- nested_detach,
- nested_numpify,
- nested_xla_mesh_reduce,
- reissue_pt_warnings,
- remove_dummy_checkpoint,
- set_rng_state_for_device,
- )
- from .trainer_utils import (
- PREFIX_CHECKPOINT_DIR,
- BestRun,
- EvalLoopOutput,
- EvalPrediction,
- HPSearchBackend,
- HubStrategy,
- PredictionOutput,
- RemoveColumnsCollator,
- SaveStrategy,
- TrainerMemoryTracker,
- TrainOutput,
- check_target_module_exists,
- default_compute_objective,
- denumpify_detensorize,
- enable_full_determinism,
- find_executable_batch_size,
- get_last_checkpoint,
- has_length,
- neftune_post_forward_hook,
- number_of_arguments,
- seed_worker,
- set_seed,
- speed_metrics,
- )
- from .training_args import OptimizerNames, ParallelMode, TrainingArguments
- from .utils import (
- ADAPTER_CONFIG_NAME,
- ADAPTER_SAFE_WEIGHTS_NAME,
- ADAPTER_WEIGHTS_NAME,
- CONFIG_NAME,
- GENERATION_CONFIG_NAME,
- SAFE_WEIGHTS_INDEX_NAME,
- SAFE_WEIGHTS_NAME,
- WEIGHTS_INDEX_NAME,
- WEIGHTS_NAME,
- XLA_FSDPV2_MIN_VERSION,
- PushInProgress,
- PushToHubMixin,
- can_return_loss,
- check_torch_load_is_safe,
- find_labels,
- is_accelerate_available,
- is_apollo_torch_available,
- is_bitsandbytes_available,
- is_datasets_available,
- is_galore_torch_available,
- is_grokadamw_available,
- is_in_notebook,
- is_liger_kernel_available,
- is_lomo_available,
- is_peft_available,
- is_sagemaker_dp_enabled,
- is_sagemaker_mp_enabled,
- is_schedulefree_available,
- is_torch_hpu_available,
- is_torch_mlu_available,
- is_torch_mps_available,
- is_torch_musa_available,
- is_torch_neuroncore_available,
- is_torch_npu_available,
- is_torch_optimi_available,
- is_torch_xla_available,
- is_torch_xpu_available,
- is_torchao_available,
- logging,
- strtobool,
- )
- from .utils.deprecation import deprecate_kwarg
- from .utils.import_utils import requires
- from .utils.quantization_config import QuantizationMethod
- DEFAULT_CALLBACKS = [DefaultFlowCallback]
- DEFAULT_PROGRESS_CALLBACK = ProgressCallback
- if is_in_notebook():
- from .utils.notebook import NotebookProgressCallback
- DEFAULT_PROGRESS_CALLBACK = NotebookProgressCallback
- if is_datasets_available():
- import datasets
- if is_torch_xla_available():
- import torch_xla.core.xla_model as xm
- import torch_xla.debug.metrics as met
- import torch_xla.runtime as xr
- from torch_xla import __version__ as XLA_VERSION
- IS_XLA_FSDPV2_POST_2_2 = version.parse(XLA_VERSION) >= version.parse(XLA_FSDPV2_MIN_VERSION)
- if IS_XLA_FSDPV2_POST_2_2:
- import torch_xla.distributed.spmd as xs
- else:
- IS_XLA_FSDPV2_POST_2_2 = False
- if is_sagemaker_mp_enabled():
- import smdistributed.modelparallel.torch as smp
- from smdistributed.modelparallel import __version__ as SMP_VERSION
- IS_SAGEMAKER_MP_POST_1_10 = version.parse(SMP_VERSION) >= version.parse("1.10")
- from .trainer_pt_utils import smp_forward_backward, smp_forward_only, smp_gather, smp_nested_concat
- else:
- IS_SAGEMAKER_MP_POST_1_10 = False
- if is_peft_available():
- from peft import PeftModel
- if is_accelerate_available():
- from accelerate import Accelerator, skip_first_batches
- from accelerate import __version__ as accelerate_version
- from accelerate.state import AcceleratorState
- from accelerate.utils import (
- AutocastKwargs,
- DistributedDataParallelKwargs,
- DistributedType,
- load_fsdp_model,
- load_fsdp_optimizer,
- save_fsdp_model,
- save_fsdp_optimizer,
- )
- DATA_SAMPLERS = [RandomSampler]
- if version.parse(accelerate_version) > version.parse("1.3.0"):
- from accelerate.utils import TorchTensorParallelPlugin
- from accelerate.data_loader import SeedableRandomSampler
- DATA_SAMPLERS += [SeedableRandomSampler]
- if is_deepspeed_available():
- from accelerate.utils import DeepSpeedSchedulerWrapper
- if is_accelerate_available("0.28.0"):
- from accelerate.utils import DataLoaderConfiguration
- def _is_peft_model(model):
- if is_peft_available():
- classes_to_check = (PeftModel,)
- # Here we also check if the model is an instance of `PeftMixedModel` introduced in peft>=0.7.0: https://github.com/huggingface/transformers/pull/28321
- if version.parse(importlib.metadata.version("peft")) >= version.parse("0.7.0"):
- from peft import PeftMixedModel
- classes_to_check = (*classes_to_check, PeftMixedModel)
- return isinstance(model, classes_to_check)
- return False
- def _get_fsdp_ckpt_kwargs():
- # TODO: @AjayP13, @younesbelkada replace this check with version check at the next `accelerate` release
- if is_accelerate_available() and "adapter_only" in list(inspect.signature(save_fsdp_model).parameters):
- return {"adapter_only": True}
- else:
- return {}
- def safe_globals():
- # Starting from version 2.4 PyTorch introduces a check for the objects loaded
- # with torch.load(weights_only=True). Starting from 2.6 weights_only=True becomes
- # a default and requires allowlisting of objects being loaded.
- # See: https://github.com/pytorch/pytorch/pull/137602
- # See: https://pytorch.org/docs/stable/notes/serialization.html#torch.serialization.add_safe_globals
- # See: https://github.com/huggingface/accelerate/pull/3036
- if version.parse(torch.__version__).release < version.parse("2.6").release:
- return contextlib.nullcontext()
- np_core = np._core if version.parse(np.__version__) >= version.parse("2.0.0") else np.core
- allowlist = [np_core.multiarray._reconstruct, np.ndarray, np.dtype]
- # numpy >1.25 defines numpy.dtypes.UInt32DType, but below works for
- # all versions of numpy
- allowlist += [type(np.dtype(np.uint32))]
- return torch.serialization.safe_globals(allowlist)
- if TYPE_CHECKING:
- import optuna
- logger = logging.get_logger(__name__)
- # Name of the files used for checkpointing
- TRAINING_ARGS_NAME = "training_args.bin"
- TRAINER_STATE_NAME = "trainer_state.json"
- OPTIMIZER_NAME = "optimizer.pt"
- SCALER_NAME = "scaler.pt"
- OPTIMIZER_NAME_BIN = "optimizer.bin"
- SCHEDULER_NAME = "scheduler.pt"
- FSDP_MODEL_NAME = "pytorch_model_fsdp"
- @requires(
- backends=(
- "torch",
- "accelerate",
- )
- )
- class Trainer:
- """
- Trainer is a simple but feature-complete training and eval loop for PyTorch, optimized for 🤗 Transformers.
- Args:
- model ([`PreTrainedModel`] or `torch.nn.Module`, *optional*):
- The model to train, evaluate or use for predictions. If not provided, a `model_init` must be passed.
- <Tip>
- [`Trainer`] is optimized to work with the [`PreTrainedModel`] provided by the library. You can still use
- your own models defined as `torch.nn.Module` as long as they work the same way as the 🤗 Transformers
- models.
- </Tip>
- args ([`TrainingArguments`], *optional*):
- The arguments to tweak for training. Will default to a basic instance of [`TrainingArguments`] with the
- `output_dir` set to a directory named *tmp_trainer* in the current directory if not provided.
- data_collator (`DataCollator`, *optional*):
- The function to use to form a batch from a list of elements of `train_dataset` or `eval_dataset`. Will
- default to [`default_data_collator`] if no `processing_class` is provided, an instance of
- [`DataCollatorWithPadding`] otherwise if the processing_class is a feature extractor or tokenizer.
- train_dataset (Union[`torch.utils.data.Dataset`, `torch.utils.data.IterableDataset`, `datasets.Dataset`], *optional*):
- The dataset to use for training. If it is a [`~datasets.Dataset`], columns not accepted by the
- `model.forward()` method are automatically removed.
- Note that if it's a `torch.utils.data.IterableDataset` with some randomization and you are training in a
- distributed fashion, your iterable dataset should either use a internal attribute `generator` that is a
- `torch.Generator` for the randomization that must be identical on all processes (and the Trainer will
- manually set the seed of this `generator` at each epoch) or have a `set_epoch()` method that internally
- sets the seed of the RNGs used.
- eval_dataset (Union[`torch.utils.data.Dataset`, dict[str, `torch.utils.data.Dataset`, `datasets.Dataset`]), *optional*):
- The dataset to use for evaluation. If it is a [`~datasets.Dataset`], columns not accepted by the
- `model.forward()` method are automatically removed. If it is a dictionary, it will evaluate on each
- dataset prepending the dictionary key to the metric name.
- processing_class (`PreTrainedTokenizerBase` or `BaseImageProcessor` or `FeatureExtractionMixin` or `ProcessorMixin`, *optional*):
- Processing class used to process the data. If provided, will be used to automatically process the inputs
- for the model, and it will be saved along the model to make it easier to rerun an interrupted training or
- reuse the fine-tuned model.
- This supersedes the `tokenizer` argument, which is now deprecated.
- model_init (`Callable[[], PreTrainedModel]`, *optional*):
- A function that instantiates the model to be used. If provided, each call to [`~Trainer.train`] will start
- from a new instance of the model as given by this function.
- The function may have zero argument, or a single one containing the optuna/Ray Tune/SigOpt trial object, to
- be able to choose different architectures according to hyper parameters (such as layer count, sizes of
- inner layers, dropout probabilities etc).
- compute_loss_func (`Callable`, *optional*):
- A function that accepts the raw model outputs, labels, and the number of items in the entire accumulated
- batch (batch_size * gradient_accumulation_steps) and returns the loss. For example, see the default [loss function](https://github.com/huggingface/transformers/blob/052e652d6d53c2b26ffde87e039b723949a53493/src/transformers/trainer.py#L3618) used by [`Trainer`].
- compute_metrics (`Callable[[EvalPrediction], Dict]`, *optional*):
- The function that will be used to compute metrics at evaluation. Must take a [`EvalPrediction`] and return
- a dictionary string to metric values. *Note* When passing TrainingArgs with `batch_eval_metrics` set to
- `True`, your compute_metrics function must take a boolean `compute_result` argument. This will be triggered
- after the last eval batch to signal that the function needs to calculate and return the global summary
- statistics rather than accumulating the batch-level statistics
- callbacks (List of [`TrainerCallback`], *optional*):
- A list of callbacks to customize the training loop. Will add those to the list of default callbacks
- detailed in [here](callback).
- If you want to remove one of the default callbacks used, use the [`Trainer.remove_callback`] method.
- optimizers (`tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR]`, *optional*, defaults to `(None, None)`):
- A tuple containing the optimizer and the scheduler to use. Will default to an instance of [`AdamW`] on your
- model and a scheduler given by [`get_linear_schedule_with_warmup`] controlled by `args`.
- optimizer_cls_and_kwargs (`tuple[Type[torch.optim.Optimizer], dict[str, Any]]`, *optional*):
- A tuple containing the optimizer class and keyword arguments to use.
- Overrides `optim` and `optim_args` in `args`. Incompatible with the `optimizers` argument.
- Unlike `optimizers`, this argument avoids the need to place model parameters on the correct devices before initializing the Trainer.
- preprocess_logits_for_metrics (`Callable[[torch.Tensor, torch.Tensor], torch.Tensor]`, *optional*):
- A function that preprocess the logits right before caching them at each evaluation step. Must take two
- tensors, the logits and the labels, and return the logits once processed as desired. The modifications made
- by this function will be reflected in the predictions received by `compute_metrics`.
- Note that the labels (second parameter) will be `None` if the dataset does not have them.
- Important attributes:
- - **model** -- Always points to the core model. If using a transformers model, it will be a [`PreTrainedModel`]
- subclass.
- - **model_wrapped** -- Always points to the most external model in case one or more other modules wrap the
- original model. This is the model that should be used for the forward pass. For example, under `DeepSpeed`,
- the inner model is wrapped in `DeepSpeed` and then again in `torch.nn.DistributedDataParallel`. If the inner
- model hasn't been wrapped, then `self.model_wrapped` is the same as `self.model`.
- - **is_model_parallel** -- Whether or not a model has been switched to a model parallel mode (different from
- data parallelism, this means some of the model layers are split on different GPUs).
- - **place_model_on_device** -- Whether or not to automatically place the model on the device - it will be set
- to `False` if model parallel or deepspeed is used, or if the default
- `TrainingArguments.place_model_on_device` is overridden to return `False` .
- - **is_in_train** -- Whether or not a model is currently running `train` (e.g. when `evaluate` is called while
- in `train`)
- """
- # Those are used as methods of the Trainer in examples.
- from .trainer_pt_utils import _get_learning_rate, log_metrics, metrics_format, save_metrics, save_state
- @deprecate_kwarg("tokenizer", new_name="processing_class", version="5.0.0", raise_if_both_names=True)
- def __init__(
- self,
- model: Union[PreTrainedModel, nn.Module, None] = None,
- args: Optional[TrainingArguments] = None,
- data_collator: Optional[DataCollator] = None,
- train_dataset: Optional[Union[Dataset, IterableDataset, "datasets.Dataset"]] = None,
- eval_dataset: Optional[Union[Dataset, dict[str, Dataset], "datasets.Dataset"]] = None,
- processing_class: Optional[
- Union[PreTrainedTokenizerBase, BaseImageProcessor, FeatureExtractionMixin, ProcessorMixin]
- ] = None,
- model_init: Optional[Callable[..., PreTrainedModel]] = None,
- compute_loss_func: Optional[Callable] = None,
- compute_metrics: Optional[Callable[[EvalPrediction], dict]] = None,
- callbacks: Optional[list[TrainerCallback]] = None,
- optimizers: tuple[Optional[torch.optim.Optimizer], Optional[torch.optim.lr_scheduler.LambdaLR]] = (None, None),
- optimizer_cls_and_kwargs: Optional[tuple[type[torch.optim.Optimizer], dict[str, Any]]] = None,
- preprocess_logits_for_metrics: Optional[Callable[[torch.Tensor, torch.Tensor], torch.Tensor]] = None,
- ):
- if args is None:
- output_dir = "tmp_trainer"
- logger.info(f"No `TrainingArguments` passed, using `output_dir={output_dir}`.")
- args = TrainingArguments(output_dir=output_dir)
- if args.batch_eval_metrics and compute_metrics is not None:
- if "compute_result" not in inspect.signature(compute_metrics).parameters:
- raise ValueError(
- "When using `batch_eval_metrics`, your `compute_metrics` function must take a `compute_result`"
- " boolean argument which will be triggered after the last batch of the eval set to signal that the"
- " summary statistics should be returned by the function."
- )
- if args.eval_strategy is not None and args.eval_strategy != "no" and eval_dataset is None:
- raise ValueError(
- f"You have set `args.eval_strategy` to {args.eval_strategy} but you didn't pass an `eval_dataset` to `Trainer`. Either set `args.eval_strategy` to `no` or pass an `eval_dataset`. "
- )
- if args.save_strategy == SaveStrategy.BEST or args.load_best_model_at_end:
- if args.metric_for_best_model is None:
- raise ValueError(
- "`args.metric_for_best_model` must be provided when using 'best' save_strategy or if `args.load_best_model_at_end` is set to `True`."
- )
- self.args = args
- self.compute_loss_func = compute_loss_func
- # Seed must be set before instantiating the model when using model
- enable_full_determinism(self.args.seed) if self.args.full_determinism else set_seed(self.args.seed)
- self.hp_name = None
- self.deepspeed = None
- self.is_in_train = False
- self.model = model
- self.create_accelerator_and_postprocess()
- # memory metrics - must set up as early as possible
- self._memory_tracker = TrainerMemoryTracker(self.args.skip_memory_metrics)
- self._memory_tracker.start()
- # set the correct log level depending on the node
- log_level = args.get_process_log_level()
- logging.set_verbosity(log_level)
- # force device and distributed setup init explicitly
- args._setup_devices
- if model is None:
- if model_init is not None:
- self.model_init = model_init
- model = self.call_model_init()
- else:
- raise RuntimeError("`Trainer` requires either a `model` or `model_init` argument")
- else:
- if model_init is not None:
- warnings.warn(
- "`Trainer` requires either a `model` or `model_init` argument, but not both. `model_init` will"
- " overwrite your model when calling the `train` method. This will become a fatal error in the next"
- " release.",
- FutureWarning,
- )
- self.model_init = model_init
- if model.__class__.__name__ in MODEL_MAPPING_NAMES:
- raise ValueError(
- f"The model you have picked ({model.__class__.__name__}) cannot be used as is for training: it only "
- "computes hidden states and does not accept any labels. You should choose a model with a head "
- "suitable for your task like any of the `AutoModelForXxx` listed at "
- "https://huggingface.co/docs/transformers/model_doc/auto"
- )
- if getattr(model, "is_parallelizable", False) and getattr(model, "model_parallel", False):
- self.is_model_parallel = True
- else:
- self.is_model_parallel = False
- if getattr(model, "hf_device_map", None) is not None:
- devices = [device for device in set(model.hf_device_map.values()) if device not in ["cpu", "disk"]]
- if len(devices) > 1:
- self.is_model_parallel = True
- elif len(devices) == 1:
- self.is_model_parallel = self.args.device != torch.device(devices[0])
- else:
- self.is_model_parallel = False
- # warn users
- if self.is_model_parallel:
- logger.info(
- "You have loaded a model on multiple GPUs. `is_model_parallel` attribute will be force-set"
- " to `True` to avoid any unexpected behavior such as device placement mismatching."
- )
- if self.args.use_liger_kernel:
- if is_liger_kernel_available():
- from liger_kernel.transformers import _apply_liger_kernel_to_instance
- # Prepare kernel config - use provided config or default (empty dict for default behavior)
- kernel_config = self.args.liger_kernel_config if self.args.liger_kernel_config is not None else {}
- if isinstance(model, PreTrainedModel):
- # Patch the model with liger kernels. Use the specified or default kernel configurations.
- _apply_liger_kernel_to_instance(model=model, **kernel_config)
- elif hasattr(model, "get_base_model") and isinstance(model.get_base_model(), PreTrainedModel):
- # Patch the base model with liger kernels where model is a PeftModel. Use the specified or default kernel configurations.
- _apply_liger_kernel_to_instance(model=model.get_base_model(), **kernel_config)
- else:
- logger.warning(
- "The model is not an instance of PreTrainedModel. No liger kernels will be applied."
- )
- else:
- raise ImportError(
- "You have set `use_liger_kernel` to `True` but liger-kernel >= 0.3.0 is not available. "
- "Please install it with `pip install liger-kernel`"
- )
- _is_quantized_and_base_model = getattr(model, "is_quantized", False) and not getattr(
- model, "_hf_peft_config_loaded", False
- )
- _quantization_method_supports_training = (
- getattr(model, "hf_quantizer", None) is not None and model.hf_quantizer.is_trainable
- )
- _is_model_quantized_and_qat_trainable = getattr(model, "hf_quantizer", None) is not None and getattr(
- model.hf_quantizer, "is_qat_trainable", False
- )
- # Filter out quantized + compiled models
- if _is_quantized_and_base_model and hasattr(model, "_orig_mod"):
- raise ValueError(
- "You cannot fine-tune quantized model with `torch.compile()` make sure to pass a non-compiled model when fine-tuning a quantized model with PEFT"
- )
- # At this stage the model is already loaded
- if _is_quantized_and_base_model and not _is_peft_model(model) and not _is_model_quantized_and_qat_trainable:
- raise ValueError(
- "You cannot perform fine-tuning on purely quantized models. Please attach trainable adapters on top of"
- " the quantized model to correctly perform fine-tuning. Please see: https://huggingface.co/docs/transformers/peft"
- " for more details"
- )
- elif _is_quantized_and_base_model and not _quantization_method_supports_training:
- raise ValueError(
- f"The model you are trying to fine-tune is quantized with {model.hf_quantizer.quantization_config.quant_method}"
- " but that quantization method do not support training. Please open an issue on GitHub: https://github.com/huggingface/transformers"
- f" to request the support for training support for {model.hf_quantizer.quantization_config.quant_method}"
- )
- self.is_fsdp_xla_enabled = args.fsdp_config["xla"]
- if len(args.fsdp) > 0:
- if self.is_deepspeed_enabled:
- raise ValueError(
- "Using --fsdp xxx together with --deepspeed is not possible, deactivate one of those flags."
- )
- if not args.fsdp_config["xla"] and args.parallel_mode != ParallelMode.DISTRIBUTED:
- raise ValueError("Using fsdp only works in distributed training.")
- # one place to sort out whether to place the model on device or not
- # postpone switching model to cuda when:
- # 1. MP - since we are trying to fit a much bigger than 1 gpu model
- # 2. fp16-enabled DeepSpeed loads the model in half the size and it doesn't need .to() anyway,
- # and we only use deepspeed for training at the moment
- # 3. full bf16 or fp16 eval - since the model needs to be cast to the right dtype first
- # 4. FSDP - same as MP
- self.place_model_on_device = args.place_model_on_device
- if (
- self.is_model_parallel
- or self.is_deepspeed_enabled
- or ((args.fp16_full_eval or args.bf16_full_eval) and not args.do_train)
- or self.is_fsdp_xla_enabled
- or self.is_fsdp_enabled
- ):
- self.place_model_on_device = False
- default_collator = (
- DataCollatorWithPadding(processing_class)
- if processing_class is not None
- and isinstance(processing_class, (PreTrainedTokenizerBase, SequenceFeatureExtractor))
- else default_data_collator
- )
- self.data_collator = data_collator if data_collator is not None else default_collator
- self.train_dataset = train_dataset
- self.eval_dataset = eval_dataset
- self.processing_class = processing_class
- # Bnb Quantized models doesn't support `.to` operation.
- if (
- self.place_model_on_device
- and getattr(model, "quantization_method", None) != QuantizationMethod.BITS_AND_BYTES
- ):
- self._move_model_to_device(model, args.device)
- # Force n_gpu to 1 to avoid DataParallel as MP will manage the GPUs
- if self.is_model_parallel:
- self.args._n_gpu = 1
- # later use `self.model is self.model_wrapped` to check if it's wrapped or not
- self.model_wrapped = model
- self.model = model
- # Just in case the model was wrapped outside of the `Trainer`
- unwrapped_model = self.accelerator.unwrap_model(model)
- # We also unwrap peft model
- if _is_peft_model(unwrapped_model):
- if hasattr(unwrapped_model, "get_base_model"):
- unwrapped_model = unwrapped_model.get_base_model()
- elif hasattr(unwrapped_model, "base_model") and hasattr(unwrapped_model.base_model, "model"):
- unwrapped_model = unwrapped_model.base_model.model
- else:
- raise AttributeError("Cannot extract base model safely from this PEFT wrapper.")
- # Check if the model has explicit setup for loss kwargs,
- # if not, check if `**kwargs` are in model.forward
- if hasattr(unwrapped_model, "accepts_loss_kwargs"):
- self.model_accepts_loss_kwargs = unwrapped_model.accepts_loss_kwargs
- else:
- forward_params = inspect.signature(unwrapped_model.forward).parameters
- self.model_accepts_loss_kwargs = any(
- k.kind == inspect.Parameter.VAR_KEYWORD for k in forward_params.values()
- )
- self.neftune_noise_alpha = args.neftune_noise_alpha
- self.compute_metrics = compute_metrics
- self.preprocess_logits_for_metrics = preprocess_logits_for_metrics
- self.optimizer, self.lr_scheduler = optimizers
- self.optimizer_cls_and_kwargs = optimizer_cls_and_kwargs
- if self.optimizer_cls_and_kwargs is not None and self.optimizer is not None:
- raise RuntimeError("Passing both `optimizers` and `optimizer_cls_and_kwargs` arguments is incompatible.")
- if model_init is not None and (self.optimizer is not None or self.lr_scheduler is not None):
- raise RuntimeError(
- "Passing a `model_init` is incompatible with providing the `optimizers` argument. "
- "You should subclass `Trainer` and override the `create_optimizer_and_scheduler` method."
- )
- if is_torch_xla_available() and self.optimizer is not None:
- for param in self.model.parameters():
- model_device = param.device
- break
- for param_group in self.optimizer.param_groups:
- if len(param_group["params"]) > 0:
- optimizer_device = param_group["params"][0].device
- break
- if model_device != optimizer_device:
- raise ValueError(
- "The model and the optimizer parameters are not on the same device, which probably means you"
- " created an optimizer around your model **before** putting on the device and passing it to the"
- " `Trainer`. Make sure the lines `import torch_xla.core.xla_model as xm` and"
- " `model.to(xm.xla_device())` is performed before the optimizer creation in your script."
- )
- if (self.is_fsdp_xla_enabled or self.is_fsdp_enabled) and (
- self.optimizer is not None or self.lr_scheduler is not None
- ):
- raise RuntimeError(
- "Passing `optimizers` is not allowed if PyTorch FSDP is enabled. "
- "You should subclass `Trainer` and override the `create_optimizer_and_scheduler` method."
- )
- default_callbacks = DEFAULT_CALLBACKS + get_reporting_integration_callbacks(self.args.report_to)
- callbacks = default_callbacks if callbacks is None else default_callbacks + callbacks
- self.callback_handler = CallbackHandler(
- callbacks, self.model, self.processing_class, self.optimizer, self.lr_scheduler
- )
- self.add_callback(PrinterCallback if self.args.disable_tqdm else DEFAULT_PROGRESS_CALLBACK)
- # Will be set to True by `self._setup_loggers()` on first call to `self.log()`.
- self._loggers_initialized = False
- # Create distant repo and output directory if needed
- self.hub_model_id = None
- if self.args.push_to_hub:
- self.init_hf_repo()
- if self.args.should_save:
- os.makedirs(self.args.output_dir, exist_ok=True)
- if not callable(self.data_collator) and callable(getattr(self.data_collator, "collate_batch", None)):
- raise TypeError("The `data_collator` should be a simple callable (function, class with `__call__`).")
- if args.max_steps > 0 and args.num_train_epochs > 0:
- logger.info("max_steps is given, it will override any value given in num_train_epochs")
- if train_dataset is not None and not has_length(train_dataset) and args.max_steps <= 0:
- raise ValueError(
- "The train_dataset does not implement __len__, max_steps has to be specified. "
- "The number of steps needs to be known in advance for the learning rate scheduler."
- )
- if (
- train_dataset is not None
- and isinstance(train_dataset, torch.utils.data.IterableDataset)
- and args.group_by_length
- ):
- raise ValueError("the `--group_by_length` option is only available for `Dataset`, not `IterableDataset")
- self._signature_columns = None
- # Mixed precision setup
- self.use_apex = False
- self.use_cpu_amp = False
- # Mixed precision setup for SageMaker Model Parallel
- if is_sagemaker_mp_enabled():
- # BF16 + model parallelism in SageMaker: currently not supported, raise an error
- if args.bf16:
- raise ValueError("SageMaker Model Parallelism does not support BF16 yet. Please use FP16 instead ")
- if IS_SAGEMAKER_MP_POST_1_10:
- # When there's mismatch between SMP config and trainer argument, use SMP config as truth
- if args.fp16 != smp.state.cfg.fp16:
- logger.warning(
- f"FP16 provided in SM_HP_MP_PARAMETERS is {smp.state.cfg.fp16}, "
- f"but FP16 provided in trainer argument is {args.fp16}, "
- f"setting to {smp.state.cfg.fp16}"
- )
- args.fp16 = smp.state.cfg.fp16
- else:
- # smp < 1.10 does not support fp16 in trainer.
- if hasattr(smp.state.cfg, "fp16"):
- logger.warning(
- f"FP16 provided in SM_HP_MP_PARAMETERS is {smp.state.cfg.fp16}, "
- "but SageMaker Model Parallelism < 1.10 does not support FP16 in trainer."
- )
- if (args.fp16 or args.bf16) and args.half_precision_backend == "auto":
- if args.device == torch.device("cpu"):
- if args.fp16:
- if not is_torch_greater_or_equal_than_2_3:
- raise ValueError("Tried to use `fp16` but it is not supported on cpu")
- else:
- args.half_precision_backend = "cpu_amp"
- logger.info(f"Using {args.half_precision_backend} half precision backend")
- if (args.fp16 or args.bf16) and not (self.is_deepspeed_enabled or is_sagemaker_mp_enabled()):
- # deepspeed and SageMaker Model Parallel manage their own half precision
- if args.half_precision_backend == "cpu_amp":
- self.use_cpu_amp = True
- self.amp_dtype = torch.bfloat16
- elif args.half_precision_backend == "apex":
- self.use_apex = True
- # Label smoothing
- if self.args.label_smoothing_factor != 0:
- self.label_smoother = LabelSmoother(epsilon=self.args.label_smoothing_factor)
- else:
- self.label_smoother = None
- # Check for multi-label classification incompatibility
- if self.args.label_smoothing_factor > 0:
- if getattr(self.model.config, "problem_type", None) == "multi_label_classification":
- warnings.warn(
- "Label smoothing is not compatible with multi-label classification. "
- "Disabling label smoothing for this training run.",
- UserWarning,
- )
- self.label_smoother = None
- self.control = TrainerControl()
- self.state = TrainerState(
- is_local_process_zero=self.is_local_process_zero(),
- is_world_process_zero=self.is_world_process_zero(),
- stateful_callbacks=[
- cb for cb in self.callback_handler.callbacks + [self.control] if isinstance(cb, ExportableState)
- ],
- )
- # Internal variable to count flos in each process, will be accumulated in `self.state.total_flos` then
- # returned to 0 every time flos need to be logged
- self.current_flos = 0
- self.hp_search_backend = None
- model_to_inspect = self.model
- if _is_peft_model(self.model):
- if hasattr(self.model, "get_base_model"):
- model_to_inspect = self.model.get_base_model()
- else:
- # PeftMixedModel do not provide a `get_base_model` method
- model_to_inspect = self.model.base_model.model
- default_label_names = find_labels(model_to_inspect.__class__)
- self.label_names = default_label_names if self.args.label_names is None else self.args.label_names
- self.can_return_loss = can_return_loss(model_to_inspect.__class__)
- self.control = self.callback_handler.on_init_end(self.args, self.state, self.control)
- # Internal variables to help with automatic batch size reduction
- self._train_batch_size = args.train_batch_size
- self._created_lr_scheduler = False
- # very last
- self._memory_tracker.stop_and_update_metrics()
- self.is_fsdp_xla_v2_enabled = args.fsdp_config.get("xla_fsdp_v2", False)
- if self.is_fsdp_xla_v2_enabled:
- if not IS_XLA_FSDPV2_POST_2_2:
- raise ValueError("FSDPv2 requires `torch_xla` 2.2 or higher.")
- # Prepare the SPMD mesh that is going to be used by the data loader and the FSDPv2 wrapper.
- # Tensor axis is just a placeholder where it will not be used in FSDPv2.
- num_devices = xr.global_runtime_device_count()
- xs.set_global_mesh(xs.Mesh(np.array(range(num_devices)), (num_devices, 1), axis_names=("fsdp", "tensor")))
- self.is_fsdp_xla_v1_enabled = self.is_fsdp_xla_enabled and not self.is_fsdp_xla_v2_enabled
- @property
- def tokenizer(self) -> Optional[PreTrainedTokenizerBase]:
- logger.warning("Trainer.tokenizer is now deprecated. You should use Trainer.processing_class instead.")
- return self.processing_class
- @tokenizer.setter
- def tokenizer(self, processing_class) -> None:
- logger.warning(
- "Trainer.tokenizer is now deprecated. You should use `Trainer.processing_class = processing_class` instead."
- )
- self.processing_class = processing_class
- def _activate_neftune(self, model):
- r"""
- Activates the neftune as presented in this code: https://github.com/neelsjain/NEFTune and paper:
- https://huggingface.co/papers/2310.05914
- """
- unwrapped_model = self.accelerator.unwrap_model(model)
- if _is_peft_model(unwrapped_model):
- embeddings = unwrapped_model.base_model.model.get_input_embeddings()
- else:
- embeddings = unwrapped_model.get_input_embeddings()
- del unwrapped_model
- embeddings.neftune_noise_alpha = self.neftune_noise_alpha
- hook_handle = embeddings.register_forward_hook(neftune_post_forward_hook)
- self.neftune_hook_handle = hook_handle
- return model
- def _deactivate_neftune(self, model):
- """
- Deactivates the neftune method. Make sure to call `_activate_neftune` first.
- """
- if not hasattr(self, "neftune_hook_handle"):
- raise ValueError("Neftune is not activated make sure to call `trainer._activate_neftune()` first")
- unwrapped_model = self.accelerator.unwrap_model(model)
- if _is_peft_model(unwrapped_model):
- embeddings = unwrapped_model.base_model.model.get_input_embeddings()
- else:
- embeddings = unwrapped_model.get_input_embeddings()
- self.neftune_hook_handle.remove()
- del embeddings.neftune_noise_alpha, unwrapped_model
- def add_callback(self, callback):
- """
- Add a callback to the current list of [`~transformers.TrainerCallback`].
- Args:
- callback (`type` or [`~transformers.TrainerCallback]`):
- A [`~transformers.TrainerCallback`] class or an instance of a [`~transformers.TrainerCallback`]. In the
- first case, will instantiate a member of that class.
- """
- self.callback_handler.add_callback(callback)
- def pop_callback(self, callback):
- """
- Remove a callback from the current list of [`~transformers.TrainerCallback`] and returns it.
- If the callback is not found, returns `None` (and no error is raised).
- Args:
- callback (`type` or [`~transformers.TrainerCallback]`):
- A [`~transformers.TrainerCallback`] class or an instance of a [`~transformers.TrainerCallback`]. In the
- first case, will pop the first member of that class found in the list of callbacks.
- Returns:
- [`~transformers.TrainerCallback`]: The callback removed, if found.
- """
- return self.callback_handler.pop_callback(callback)
- def remove_callback(self, callback):
- """
- Remove a callback from the current list of [`~transformers.TrainerCallback`].
- Args:
- callback (`type` or [`~transformers.TrainerCallback]`):
- A [`~transformers.TrainerCallback`] class or an instance of a [`~transformers.TrainerCallback`]. In the
- first case, will remove the first member of that class found in the list of callbacks.
- """
- self.callback_handler.remove_callback(callback)
- def _move_model_to_device(self, model, device):
- if getattr(model, "hf_device_map", None) is not None:
- logger.warning(
- "The model is already on multiple devices. Skipping the move to device specified in `args`."
- )
- return
- model = model.to(device)
- # Moving a model to an XLA device disconnects the tied weights, so we have to retie them.
- if self.args.parallel_mode == ParallelMode.TPU and hasattr(model, "tie_weights"):
- model.tie_weights()
- def _align_special_tokens(self):
- """
- Aligns the special tokens of the tokenizer with the model configs.
- A new tokens may be defined in the tokenizer for fine-tuning purposes, e.g. an "end of turn" token may be
- added on chat models. In that case, we want the model configs to be aligned with the tokenizer, so that all
- downstream uses work as expected. This alignment should happen before training, to ensure the prediction step
- uses the new tokens as well.
- """
- if isinstance(self.processing_class, ProcessorMixin):
- tokenizer: PreTrainedTokenizerBase = self.processing_class.tokenizer
- else:
- tokenizer = self.processing_class
- model_has_generation_config = (
- hasattr(self.model, "generation_config") and self.model.generation_config is not None
- )
- updated_tokens = {}
- # 1 - Align EOS token. EOS is more complex than the others, as `generation_config` may hold more than one EOS
- # token.
- tokenizer_has_new_eos = tokenizer.eos_token_id != self.model.config.eos_token_id
- if model_has_generation_config:
- # `generation_config.eos_token_id` is None: direct comparison
- if self.model.generation_config.eos_token_id is None:
- tokenizer_has_new_eos |= tokenizer.eos_token_id != self.model.generation_config.eos_token_id
- else:
- # `generation_config.eos_token_id` is an `int`: convert it to list (and continue below)
- if isinstance(self.model.generation_config.eos_token_id, int):
- self.model.generation_config.eos_token_id = [self.model.generation_config.eos_token_id]
- # `generation_config.eos_token_id` is a `list`: check if the tokenizer's EOS token is in the list
- tokenizer_has_new_eos |= tokenizer.eos_token_id not in self.model.generation_config.eos_token_id
- if tokenizer_has_new_eos:
- updated_tokens["eos_token_id"] = tokenizer.eos_token_id
- self.model.config.eos_token_id = tokenizer.eos_token_id
- # The generation config may hold more than one EOS token. We preserve the original EOS tokens: any of the
- # EOS tokens defined here will halt generation.
- if model_has_generation_config:
- all_eos_tokens = [tokenizer.eos_token_id]
- if self.model.generation_config.eos_token_id is not None:
- all_eos_tokens += list(self.model.generation_config.eos_token_id)
- self.model.generation_config.eos_token_id = [token for token in all_eos_tokens if token is not None]
- # 2 - Align BOS
- tokenizer_has_new_bos = tokenizer.bos_token_id != self.model.config.bos_token_id
- if model_has_generation_config:
- tokenizer_has_new_bos |= tokenizer.bos_token_id != self.model.generation_config.bos_token_id
- if tokenizer_has_new_bos:
- updated_tokens["bos_token_id"] = tokenizer.bos_token_id
- self.model.config.bos_token_id = tokenizer.bos_token_id
- if model_has_generation_config:
- self.model.generation_config.bos_token_id = tokenizer.bos_token_id
- # 3 - Align PAD
- tokenizer_has_new_pad = tokenizer.pad_token_id != self.model.config.pad_token_id
- if model_has_generation_config:
- tokenizer_has_new_pad |= tokenizer.pad_token_id != self.model.generation_config.pad_token_id
- if tokenizer_has_new_pad:
- updated_tokens["pad_token_id"] = tokenizer.pad_token_id
- self.model.config.pad_token_id = tokenizer.pad_token_id
- if model_has_generation_config:
- self.model.generation_config.pad_token_id = tokenizer.pad_token_id
- # 4 - Warn users about the changes
- if len(updated_tokens) > 0:
- logger.warning(
- "The tokenizer has new PAD/BOS/EOS tokens that differ from the model config and generation config. "
- "The model config and generation config were aligned accordingly, being updated with the tokenizer's "
- f"values. Updated tokens: {updated_tokens}."
- )
- def _set_signature_columns_if_needed(self):
- if self._signature_columns is None:
- # Inspect model forward signature to keep only the arguments it accepts.
- model_to_inspect = self.model
- if _is_peft_model(self.model):
- if hasattr(self.model, "get_base_model"):
- model_to_inspect = self.model.get_base_model()
- else:
- # PeftMixedModel do not provide a `get_base_model` method
- model_to_inspect = self.model.base_model.model
- signature = inspect.signature(model_to_inspect.forward)
- self._signature_columns = list(signature.parameters.keys())
- # Labels may be named label or label_ids, the default data collator handles that.
- self._signature_columns += list(set(["label", "label_ids"] + self.label_names))
- def _remove_unused_columns(self, dataset: "datasets.Dataset", description: Optional[str] = None):
- if not self.args.remove_unused_columns:
- return dataset
- self._set_signature_columns_if_needed()
- signature_columns = self._signature_columns
- ignored_columns = list(set(dataset.column_names) - set(signature_columns))
- if len(ignored_columns) > 0:
- dset_description = "" if description is None else f"in the {description} set"
- logger.info(
- f"The following columns {dset_description} don't have a corresponding argument in "
- f"`{self.model.__class__.__name__}.forward` and have been ignored: {', '.join(ignored_columns)}."
- f" If {', '.join(ignored_columns)} are not expected by `{self.model.__class__.__name__}.forward`, "
- " you can safely ignore this message."
- )
- columns = [k for k in signature_columns if k in dataset.column_names]
- if len(columns) == 0:
- raise ValueError(
- f"No columns in the dataset match the model's forward method signature: ({', '.join(signature_columns)}). "
- f"The following columns have been ignored: [{', '.join(ignored_columns)}]. "
- "Please check the dataset and model. You may need to set `remove_unused_columns=False` in `TrainingArguments`."
- )
- if version.parse(datasets.__version__) < version.parse("1.4.0"):
- dataset.set_format(
- type=dataset.format["type"], columns=columns, format_kwargs=dataset.format["format_kwargs"]
- )
- return dataset
- else:
- return dataset.remove_columns(ignored_columns)
- def _get_collator_with_removed_columns(
- self, data_collator: Callable, description: Optional[str] = None
- ) -> Callable:
- """Wrap the data collator in a callable removing unused columns."""
- if not self.args.remove_unused_columns:
- return data_collator
- self._set_signature_columns_if_needed()
- signature_columns = self._signature_columns
- remove_columns_collator = RemoveColumnsCollator(
- data_collator=data_collator,
- signature_columns=signature_columns,
- logger=logger,
- description=description,
- model_name=self.model.__class__.__name__,
- )
- return remove_columns_collator
- def _get_train_sampler(self, train_dataset: Optional[Dataset] = None) -> Optional[torch.utils.data.Sampler]:
- if train_dataset is None:
- train_dataset = self.train_dataset
- if train_dataset is None or not has_length(train_dataset):
- return None
- # Build the sampler.
- if self.args.group_by_length:
- if is_datasets_available() and isinstance(train_dataset, datasets.Dataset):
- lengths = (
- train_dataset[self.args.length_column_name]
- if self.args.length_column_name in train_dataset.column_names
- else None
- )
- else:
- lengths = None
- model_input_name = (
- self.processing_class.model_input_names[0] if self.processing_class is not None else None
- )
- return LengthGroupedSampler(
- self.args.train_batch_size * self.args.gradient_accumulation_steps,
- dataset=train_dataset,
- lengths=lengths,
- model_input_name=model_input_name,
- )
- else:
- return RandomSampler(train_dataset)
- def _get_dataloader(
- self,
- dataset: Dataset,
- description: str,
- batch_size: int,
- sampler_fn: Optional[Callable[[Dataset], torch.utils.data.Sampler]] = None,
- is_training: bool = False,
- dataloader_key: Optional[str] = None,
- ) -> DataLoader:
- """Create a [`~torch.utils.data.DataLoader`] from the given dataset."""
- data_collator = self.data_collator
- if is_datasets_available() and isinstance(dataset, datasets.Dataset):
- dataset = self._remove_unused_columns(dataset, description=description)
- else:
- data_collator = self._get_collator_with_removed_columns(self.data_collator, description=description)
- dataloader_params = {
- "batch_size": batch_size,
- "collate_fn": data_collator,
- "num_workers": self.args.dataloader_num_workers,
- "pin_memory": self.args.dataloader_pin_memory,
- "persistent_workers": self.args.dataloader_persistent_workers,
- }
- if not isinstance(dataset, torch.utils.data.IterableDataset):
- if sampler_fn is not None:
- dataloader_params["sampler"] = sampler_fn(dataset)
- dataloader_params["drop_last"] = self.args.dataloader_drop_last
- dataloader_params["prefetch_factor"] = self.args.dataloader_prefetch_factor
- if is_training:
- dataloader_params["worker_init_fn"] = partial(
- seed_worker, num_workers=self.args.dataloader_num_workers, rank=self.args.process_index
- )
- dataloader = self.accelerator.prepare(DataLoader(dataset, **dataloader_params))
- # Store the prepared dataloader for subsequent evaluations if using persistent workers.
- if dataloader_key is not None and self.args.dataloader_persistent_workers:
- if hasattr(self, "_eval_dataloaders"):
- self._eval_dataloaders[dataloader_key] = dataloader
- else:
- self._eval_dataloaders = {dataloader_key: dataloader}
- return dataloader
- def get_train_dataloader(self) -> DataLoader:
- """
- Returns the training [`~torch.utils.data.DataLoader`].
- Will use no sampler if `train_dataset` does not implement `__len__`, a random sampler (adapted to distributed
- training if necessary) otherwise.
- Subclass and override this method if you want to inject some custom behavior.
- """
- if self.train_dataset is None:
- raise ValueError("Trainer: training requires a train_dataset.")
- return self._get_dataloader(
- dataset=self.train_dataset,
- description="Training",
- batch_size=self._train_batch_size,
- sampler_fn=self._get_train_sampler,
- is_training=True,
- )
- def _get_eval_sampler(self, eval_dataset: Dataset) -> Optional[torch.utils.data.Sampler]:
- if eval_dataset is None or not has_length(eval_dataset):
- return None
- # Build the sampler.
- # Deprecated code
- if self.args.use_legacy_prediction_loop:
- if is_torch_xla_available():
- return SequentialDistributedSampler(
- eval_dataset, num_replicas=xr.world_size(), rank=xr.global_ordinal()
- )
- elif is_sagemaker_mp_enabled():
- return SequentialDistributedSampler(
- eval_dataset,
- num_replicas=smp.dp_size(),
- rank=smp.dp_rank(),
- batch_size=self.args.per_device_eval_batch_size,
- )
- else:
- return SequentialSampler(eval_dataset)
- if self.args.group_by_length:
- if is_datasets_available() and isinstance(eval_dataset, datasets.Dataset):
- lengths = (
- eval_dataset[self.args.length_column_name]
- if self.args.length_column_name in eval_dataset.column_names
- else None
- )
- else:
- lengths = None
- model_input_name = (
- self.processing_class.model_input_names[0] if self.processing_class is not None else None
- )
- return LengthGroupedSampler(
- self.args.eval_batch_size,
- dataset=eval_dataset,
- lengths=lengths,
- model_input_name=model_input_name,
- )
- if self.args.world_size <= 1:
- return SequentialSampler(eval_dataset)
- else:
- return None
- def get_eval_dataloader(self, eval_dataset: Optional[Union[str, Dataset]] = None) -> DataLoader:
- """
- Returns the evaluation [`~torch.utils.data.DataLoader`].
- Subclass and override this method if you want to inject some custom behavior.
- Args:
- eval_dataset (`str` or `torch.utils.data.Dataset`, *optional*):
- If a `str`, will use `self.eval_dataset[eval_dataset]` as the evaluation dataset. If a `Dataset`, will override `self.eval_dataset` and must implement `__len__`. If it is a [`~datasets.Dataset`], columns not accepted by the `model.forward()` method are automatically removed.
- """
- if eval_dataset is None and self.eval_dataset is None:
- raise ValueError("Trainer: evaluation requires an eval_dataset.")
- # If we have persistent workers, don't do a fork bomb especially as eval datasets
- # don't change during training
- dataloader_key = eval_dataset if isinstance(eval_dataset, str) else "eval"
- if (
- hasattr(self, "_eval_dataloaders")
- and dataloader_key in self._eval_dataloaders
- and self.args.dataloader_persistent_workers
- ):
- return self._eval_dataloaders[dataloader_key]
- eval_dataset = (
- self.eval_dataset[eval_dataset]
- if isinstance(eval_dataset, str)
- else eval_dataset
- if eval_dataset is not None
- else self.eval_dataset
- )
- return self._get_dataloader(
- dataset=eval_dataset,
- description="Evaluation",
- batch_size=self.args.eval_batch_size,
- sampler_fn=self._get_eval_sampler,
- dataloader_key=dataloader_key,
- )
- def get_test_dataloader(self, test_dataset: Dataset) -> DataLoader:
- """
- Returns the test [`~torch.utils.data.DataLoader`].
- Subclass and override this method if you want to inject some custom behavior.
- Args:
- test_dataset (`torch.utils.data.Dataset`, *optional*):
- The test dataset to use. If it is a [`~datasets.Dataset`], columns not accepted by the
- `model.forward()` method are automatically removed. It must implement `__len__`.
- """
- return self._get_dataloader(
- dataset=test_dataset,
- description="test",
- batch_size=self.args.eval_batch_size,
- sampler_fn=self._get_eval_sampler,
- )
- def create_optimizer_and_scheduler(self, num_training_steps: int):
- """
- Setup the optimizer and the learning rate scheduler.
- We provide a reasonable default that works well. If you want to use something else, you can pass a tuple in the
- Trainer's init through `optimizers`, or subclass and override this method (or `create_optimizer` and/or
- `create_scheduler`) in a subclass.
- """
- self.create_optimizer()
- if IS_SAGEMAKER_MP_POST_1_10 and smp.state.cfg.fp16:
- # If smp >= 1.10 and fp16 is enabled, we unwrap the optimizer
- optimizer = self.optimizer.optimizer
- else:
- optimizer = self.optimizer
- self.create_scheduler(num_training_steps=num_training_steps, optimizer=optimizer)
- def get_decay_parameter_names(self, model) -> list[str]:
- """
- Get all parameter names that weight decay will be applied to.
- This function filters out parameters in two ways:
- 1. By layer type (instances of layers specified in ALL_LAYERNORM_LAYERS)
- 2. By parameter name patterns (containing 'bias', or variation of 'norm')
- """
- forbidden_name_patterns = [r"bias", r"layernorm", r"rmsnorm", r"(?:^|\.)norm(?:$|\.)", r"_norm(?:$|\.)"]
- decay_parameters = get_parameter_names(model, [nn.LayerNorm], forbidden_name_patterns)
- return decay_parameters
- def create_optimizer(self):
- """
- Setup the optimizer.
- We provide a reasonable default that works well. If you want to use something else, you can pass a tuple in the
- Trainer's init through `optimizers`, or subclass and override this method in a subclass.
- """
- opt_model = self.model_wrapped if is_sagemaker_mp_enabled() else self.model
- if self.optimizer is None:
- decay_parameters = self.get_decay_parameter_names(opt_model)
- optimizer_grouped_parameters = [
- {
- "params": [
- p for n, p in opt_model.named_parameters() if (n in decay_parameters and p.requires_grad)
- ],
- "weight_decay": self.args.weight_decay,
- },
- {
- "params": [
- p for n, p in opt_model.named_parameters() if (n not in decay_parameters and p.requires_grad)
- ],
- "weight_decay": 0.0,
- },
- ]
- if self.optimizer_cls_and_kwargs is not None:
- optimizer_cls, optimizer_kwargs = self.optimizer_cls_and_kwargs
- else:
- optimizer_cls, optimizer_kwargs = self.get_optimizer_cls_and_kwargs(self.args, opt_model)
- # Overwrite `params` in case it's created by `get_optimizer_cls_and_kwargs`
- # e.g. for GaLore optimizer.
- if "params" in optimizer_kwargs:
- optimizer_grouped_parameters = optimizer_kwargs.pop("params")
- # Overwrite `model` in case it's created by `get_optimizer_cls_and_kwargs`
- # e.g. for LOMO optimizer.
- if "model" in optimizer_kwargs:
- optimizer_grouped_parameters = optimizer_kwargs.pop("model")
- # For layer-wise dummy optimizers we overwrite optimizer_grouped_parameters with `optimizer_dict`
- # to avoid arguments conflicts.
- if "optimizer_dict" in optimizer_kwargs:
- optimizer_grouped_parameters = optimizer_kwargs.pop("optimizer_dict")
- self.optimizer = optimizer_cls(optimizer_grouped_parameters, **optimizer_kwargs)
- if "bitsandbytes" in str(optimizer_cls) and optimizer_kwargs.get("optim_bits", None) == 8:
- import bitsandbytes
- manager = bitsandbytes.optim.GlobalOptimManager.get_instance()
- skipped = 0
- for module in opt_model.modules():
- if isinstance(module, nn.Embedding):
- skipped += sum({p.data_ptr(): p.numel() for p in module.parameters()}.values())
- logger.info(f"skipped {module}: {skipped / 2**20}M params")
- manager.register_module_override(module, "weight", {"optim_bits": 32})
- logger.debug(f"bitsandbytes: will optimize {module} in fp32")
- logger.info(f"skipped: {skipped / 2**20}M params")
- if is_sagemaker_mp_enabled():
- self.optimizer = smp.DistributedOptimizer(self.optimizer)
- return self.optimizer
- def get_num_trainable_parameters(self):
- """
- Get the number of trainable parameters.
- """
- return sum(p.numel() for p in self.model.parameters() if p.requires_grad)
- def get_learning_rates(self):
- """
- Returns the learning rate of each parameter from self.optimizer.
- """
- if self.optimizer is None:
- raise ValueError("Trainer optimizer is None, please make sure you have setup the optimizer before.")
- return [group["lr"] for group in self.optimizer.param_groups]
- def get_optimizer_group(self, param: Optional[Union[str, torch.nn.parameter.Parameter]] = None):
- """
- Returns optimizer group for a parameter if given, else returns all optimizer groups for params.
- Args:
- param (`str` or `torch.nn.parameter.Parameter`, *optional*):
- The parameter for which optimizer group needs to be returned.
- """
- if self.optimizer is None:
- raise ValueError("Trainer optimizer is None, please make sure you have setup the optimizer before.")
- if param is not None:
- for group in self.optimizer.param_groups:
- if param in group["params"]:
- return group
- return [group["params"] for group in self.optimizer.param_groups]
- @staticmethod
- def get_optimizer_cls_and_kwargs(
- args: TrainingArguments, model: Optional[PreTrainedModel] = None
- ) -> tuple[Any, Any]:
- """
- Returns the optimizer class and optimizer parameters based on the training arguments.
- Args:
- args (`transformers.training_args.TrainingArguments`):
- The training arguments for the training session.
- """
- # parse args.optim_args
- optim_args = {}
- if args.optim_args:
- for mapping in args.optim_args.replace(" ", "").split(","):
- key, value = mapping.split("=")
- optim_args[key] = value
- optimizer_kwargs = {"lr": args.learning_rate}
- adam_kwargs = {
- "betas": (args.adam_beta1, args.adam_beta2),
- "eps": args.adam_epsilon,
- }
- def setup_low_rank_optimizer(
- optimizer_name: str,
- optimizer_mapping: dict[str, Any],
- optim_kwargs: dict[str, Any],
- is_layerwise_supported: bool = True,
- ) -> tuple[Any, Any]:
- """
- Helper function to set up low-rank optimizers like GaLore and Apollo.
- Args:
- optimizer_name (str): Name of the optimizer.
- optimizer_mapping (dict): Mapping of optimizer names to their classes.
- optim_kwargs (dict): Keyword arguments for the optimizer.
- is_layerwise_supported (bool): Whether layerwise optimization is supported.
- Returns:
- tuple[Any, Any]: Optimizer class and updated optimizer kwargs.
- """
- is_layerwise = optimizer_name.lower().endswith("layerwise")
- if is_layerwise and args.parallel_mode == ParallelMode.DISTRIBUTED and is_layerwise_supported:
- raise NotImplementedError(f"Layer-wise {optimizer_name} does not support DDP at this time")
- optimizer_cls = optimizer_mapping[optimizer_name]
- if args.optim_target_modules is None:
- raise ValueError(f"You need to define `optim_target_modules` to use {optimizer_name} optimizers")
- if not isinstance(args.optim_target_modules, (list, str)):
- raise TypeError(
- f"`optim_target_modules` must be a list of strings, a regex string, or 'all-linear'. Got: {args.optim_target_modules}"
- )
- if model is None:
- raise ValueError(f"You need to pass a model to initialize {optimizer_name} optimizer.")
- all_linear = (
- isinstance(args.optim_target_modules, str)
- and args.optim_target_modules.replace("_", "-") == "all-linear"
- )
- target_params_names = []
- for module_name, module in model.named_modules():
- target_module_exists, is_regex = check_target_module_exists(
- args.optim_target_modules, module_name, return_is_regex=True
- )
- if not isinstance(module, nn.Linear):
- if target_module_exists and not is_regex:
- logger.warning(
- f"{module_name} matched but ignored. {optimizer_name} only supports linear layers."
- )
- continue
- if not target_module_exists and not all_linear:
- continue
- target_params_names.append(module_name + ".weight")
- if len(target_params_names) == 0:
- raise ValueError(f"No target modules found for {optimizer_name} ({args.optim_target_modules}).")
- target_params = [p for n, p in model.named_parameters() if n in target_params_names]
- non_target_params = [p for n, p in model.named_parameters() if n not in target_params_names]
- optim_kwargs.update(optim_args)
- param_groups = [
- {"params": non_target_params},
- {"params": target_params, **optim_kwargs},
- ]
- if is_layerwise:
- if args.gradient_accumulation_steps != 1:
- raise ValueError(f"Layerwise {optimizer_name} does not support gradient accumulation!")
- optimizer_dict = {}
- for param in non_target_params:
- optimizer_dict[param] = optimizer_cls([{"params": [param]}], **optimizer_kwargs)
- for param in target_params:
- optimizer_dict[param] = optimizer_cls([{"params": [param], **optim_kwargs}], **optimizer_kwargs)
- def optimizer_hook(param):
- if param.grad is not None:
- optimizer_dict[param].step()
- optimizer_dict[param].zero_grad()
- for param in model.parameters():
- if param.requires_grad:
- param.register_post_accumulate_grad_hook(optimizer_hook)
- optimizer_cls = LayerWiseDummyOptimizer
- optimizer_kwargs.update({"optimizer_dict": optimizer_dict})
- optimizer_kwargs.update({"params": param_groups})
- return optimizer_cls, optimizer_kwargs
- if args.optim == OptimizerNames.ADAFACTOR:
- optimizer_cls = Adafactor
- optimizer_kwargs.update({"scale_parameter": False, "relative_step": False})
- elif args.optim in [OptimizerNames.ADAMW_TORCH, OptimizerNames.ADAMW_TORCH_FUSED]:
- from torch.optim import AdamW
- optimizer_cls = AdamW
- optimizer_kwargs.update(adam_kwargs)
- if args.optim == OptimizerNames.ADAMW_TORCH_FUSED:
- optimizer_kwargs.update({"fused": True})
- elif args.optim == OptimizerNames.ADAMW_TORCH_XLA:
- try:
- from torch_xla.amp.syncfree import AdamW
- optimizer_cls = AdamW
- optimizer_kwargs.update(adam_kwargs)
- except ImportError:
- raise ValueError("Trainer failed to import syncfree AdamW from torch_xla.")
- elif args.optim == OptimizerNames.ADAMW_TORCH_NPU_FUSED:
- try:
- from torch_npu.optim import NpuFusedAdamW
- optimizer_cls = NpuFusedAdamW
- optimizer_kwargs.update(adam_kwargs)
- except ImportError:
- raise ValueError("Trainer failed to import FusedAdamW from torch_npu.")
- elif args.optim == OptimizerNames.ADAMW_APEX_FUSED:
- try:
- from apex.optimizers import FusedAdam
- optimizer_cls = FusedAdam
- optimizer_kwargs.update(adam_kwargs)
- except ImportError:
- raise ValueError("Trainer tried to instantiate apex FusedAdam but apex is not installed!")
- elif args.optim in [
- OptimizerNames.ADAMW_BNB,
- OptimizerNames.ADAMW_8BIT,
- OptimizerNames.PAGED_ADAMW,
- OptimizerNames.PAGED_ADAMW_8BIT,
- OptimizerNames.ADEMAMIX,
- OptimizerNames.ADEMAMIX_8BIT,
- OptimizerNames.PAGED_ADEMAMIX,
- OptimizerNames.PAGED_ADEMAMIX_8BIT,
- OptimizerNames.LION,
- OptimizerNames.LION_8BIT,
- OptimizerNames.PAGED_LION,
- OptimizerNames.PAGED_LION_8BIT,
- OptimizerNames.RMSPROP_BNB,
- OptimizerNames.RMSPROP_8BIT,
- OptimizerNames.RMSPROP_32BIT,
- ]:
- try:
- from bitsandbytes.optim import AdamW, Lion, RMSprop
- is_paged = False
- optim_bits = 32
- optimizer_cls = None
- additional_optim_kwargs = adam_kwargs
- if "paged" in args.optim:
- is_paged = True
- if "8bit" in args.optim:
- optim_bits = 8
- if "adam" in args.optim:
- optimizer_cls = AdamW
- elif "lion" in args.optim:
- optimizer_cls = Lion
- additional_optim_kwargs = {"betas": (args.adam_beta1, args.adam_beta2)}
- elif "rmsprop" in args.optim:
- optimizer_cls = RMSprop
- # Above we pass all `adam_kwargs` to the optimizer, here
- # we only pass `optim_args` which can be passed by the user.
- additional_optim_kwargs = optim_args
- elif "ademamix" in args.optim:
- if is_bitsandbytes_available() and version.parse(
- importlib.metadata.version("bitsandbytes")
- ) < version.parse("0.44.0"):
- raise ValueError(
- "The AdEMAMix optimizer is not supported by your current version of `bitsandbytes`. "
- "Please install `bitsandbytes` >= 0.44.0."
- )
- from bitsandbytes.optim import AdEMAMix
- optimizer_cls = AdEMAMix
- additional_optim_kwargs = {
- "betas": (
- float(optim_args.get("beta1", args.adam_beta1)),
- float(optim_args.get("beta2", args.adam_beta2)),
- float(optim_args.get("beta3", 0.9999)),
- ),
- "alpha": float(optim_args.get("alpha", 5.0)),
- "eps": float(optim_args.get("eps", args.adam_epsilon)),
- }
- if "t_alpha" in optim_args:
- additional_optim_kwargs["t_alpha"] = int(optim_args["t_alpha"])
- if "t_beta3" in optim_args:
- additional_optim_kwargs["t_beta3"] = int(optim_args["t_beta3"])
- bnb_kwargs = {"optim_bits": optim_bits}
- if "rmsprop" not in args.optim:
- bnb_kwargs["is_paged"] = is_paged
- optimizer_kwargs.update(additional_optim_kwargs)
- optimizer_kwargs.update(bnb_kwargs)
- except ImportError:
- raise ValueError("Trainer tried to instantiate bnb optimizer but `bitsandbytes` is not installed!")
- if is_bitsandbytes_available() and version.parse(
- importlib.metadata.version("bitsandbytes")
- ) < version.parse("0.41.1"):
- logger.warning(
- "You are using 8-bit optimizers with a version of `bitsandbytes` < 0.41.1. "
- "It is recommended to update your version as a major bug has been fixed in 8-bit optimizers."
- )
- elif args.optim == OptimizerNames.ADAMW_ANYPRECISION:
- try:
- from torchdistx.optimizers import AnyPrecisionAdamW
- optimizer_cls = AnyPrecisionAdamW
- optimizer_kwargs.update(adam_kwargs)
- # TODO Change dtypes back to M=FP32, Var = BF16, Kahan = False once they can be cast together in torchdistx.
- optimizer_kwargs.update(
- {
- "use_kahan_summation": strtobool(optim_args.get("use_kahan_summation", "False")),
- "momentum_dtype": getattr(torch, optim_args.get("momentum_dtype", "float32")),
- "variance_dtype": getattr(torch, optim_args.get("variance_dtype", "float32")),
- "compensation_buffer_dtype": getattr(
- torch, optim_args.get("compensation_buffer_dtype", "bfloat16")
- ),
- }
- )
- except ImportError:
- raise ValueError("Please install https://github.com/pytorch/torchdistx")
- elif args.optim == OptimizerNames.SGD:
- optimizer_cls = torch.optim.SGD
- elif args.optim == OptimizerNames.ADAGRAD:
- optimizer_cls = torch.optim.Adagrad
- elif args.optim == OptimizerNames.RMSPROP:
- optimizer_cls = torch.optim.RMSprop
- elif args.optim in [
- OptimizerNames.GALORE_ADAMW,
- OptimizerNames.GALORE_ADAMW_8BIT,
- OptimizerNames.GALORE_ADAFACTOR,
- OptimizerNames.GALORE_ADAMW_LAYERWISE,
- OptimizerNames.GALORE_ADAMW_8BIT_LAYERWISE,
- OptimizerNames.GALORE_ADAFACTOR_LAYERWISE,
- ]:
- if not is_galore_torch_available():
- raise ImportError(
- "You need to install `galore_torch` in order to use GaLore optimizers"
- " install it with `pip install git+https://github.com/jiaweizzhao/GaLore`"
- )
- from galore_torch import GaLoreAdafactor, GaLoreAdamW, GaLoreAdamW8bit
- optimizer_mapping = {
- OptimizerNames.GALORE_ADAMW: GaLoreAdamW,
- OptimizerNames.GALORE_ADAMW_8BIT: GaLoreAdamW8bit,
- OptimizerNames.GALORE_ADAFACTOR: GaLoreAdafactor,
- OptimizerNames.GALORE_ADAMW_LAYERWISE: GaLoreAdamW,
- OptimizerNames.GALORE_ADAMW_8BIT_LAYERWISE: GaLoreAdamW8bit,
- OptimizerNames.GALORE_ADAFACTOR_LAYERWISE: GaLoreAdafactor,
- }
- galore_optim_kwargs = {
- "rank": int(optim_args.pop("rank", 128)),
- "update_proj_gap": int(optim_args.pop("update_proj_gap", 200)),
- "scale": float(optim_args.pop("scale", 0.25)),
- "proj_type": optim_args.pop("proj_type", "std"),
- }
- optimizer_cls, optimizer_kwargs = setup_low_rank_optimizer(
- args.optim, optimizer_mapping, galore_optim_kwargs
- )
- if args.optim == OptimizerNames.GALORE_ADAFACTOR:
- optimizer_kwargs.update({"scale_parameter": False, "relative_step": False})
- elif args.optim in [
- OptimizerNames.APOLLO_ADAMW,
- OptimizerNames.APOLLO_ADAMW_LAYERWISE,
- ]:
- if not is_apollo_torch_available():
- raise ImportError(
- "You need to install `apollo_torch` in order to use APOLLO optimizers"
- " install it with `pip install git+https://github.com/zhuhanqing/APOLLO`"
- )
- from apollo_torch import APOLLOAdamW
- optimizer_mapping = {
- OptimizerNames.APOLLO_ADAMW: APOLLOAdamW,
- OptimizerNames.APOLLO_ADAMW_LAYERWISE: APOLLOAdamW,
- }
- apollo_optim_kwargs = {
- "rank": int(optim_args.pop("rank", 128)),
- "proj": optim_args.pop("proj", "random"),
- "scale_type": optim_args.pop("scale_type", "channel"),
- "update_proj_gap": int(optim_args.pop("update_proj_gap", 200)),
- "scale": float(optim_args.pop("scale", 1.0)),
- "proj_type": optim_args.pop("proj_type", "std"),
- }
- apollo_optim_kwargs.update(adam_kwargs)
- optimizer_cls, optimizer_kwargs = setup_low_rank_optimizer(
- args.optim, optimizer_mapping, apollo_optim_kwargs
- )
- elif args.optim in [OptimizerNames.LOMO, OptimizerNames.ADALOMO]:
- if not is_lomo_available():
- raise ImportError(
- "You need to install `lomo_optim` in order to use LOMO optimizers"
- " install it with `pip install lomo-optim`"
- )
- if not is_accelerate_available("0.30.0"):
- raise ImportError("You need to have `accelerate>=0.30.0` to be able to use LOMO optimizers")
- if model is None:
- raise ValueError("You need to pass a `model` in order to correctly initialize a LOMO optimizer.")
- from lomo_optim import AdaLomo, Lomo
- if "ada" in args.optim:
- optimizer_cls = AdaLomo
- else:
- optimizer_cls = Lomo
- optimizer_kwargs.update({"model": model})
- elif args.optim == OptimizerNames.GROKADAMW:
- if not is_grokadamw_available():
- raise ValueError("Please install grokadamw with `pip install grokadamw`")
- from grokadamw import GrokAdamW
- optimizer_cls = GrokAdamW
- optimizer_kwargs.update(
- {
- "alpha_init": float(optim_args.get("alpha_init", 0.98)),
- "lamb": float(optim_args.get("lamb", 2.0)),
- "gamma": float(optim_args.get("gamma", 0.1)),
- "grokking_signal_decay_rate": float(optim_args.get("grokking_signal_decay_rate", 0.1)),
- "gradient_clipping": float(optim_args.get("gradient_clipping", 1.0)),
- }
- )
- elif args.optim in [
- OptimizerNames.ADAMW_TORCH_4BIT,
- OptimizerNames.ADAMW_TORCH_8BIT,
- ]:
- if not is_torchao_available() or version.parse(importlib.metadata.version("torchao")) < version.parse(
- "0.4.0"
- ):
- raise ImportError(
- "You need to have `torchao>=0.4.0` in order to use torch 4-bit optimizers."
- "Install it with `pip install torchao` or follow the instructions here: https://github.com/pytorch/ao"
- )
- if version.parse(importlib.metadata.version("torch")) <= version.parse("2.4"):
- raise ImportError(
- "You need to have `torch>2.4` in order to use torch 4-bit optimizers. "
- "Install it with `pip install --upgrade torch` it is available on pipy. Otherwise, you need to install torch nightly."
- )
- if version.parse(importlib.metadata.version("torchao")) >= version.parse("0.11.0"):
- # https://github.com/pytorch/ao/pull/2159
- from torchao.optim import AdamW4bit, AdamW8bit
- else:
- from torchao.prototype.low_bit_optim import AdamW4bit, AdamW8bit
- if args.optim == OptimizerNames.ADAMW_TORCH_4BIT:
- optimizer_cls = AdamW4bit
- elif args.optim == OptimizerNames.ADAMW_TORCH_8BIT:
- optimizer_cls = AdamW8bit
- else:
- raise ValueError("Invalid optimizer")
- optimizer_kwargs.update(adam_kwargs)
- elif args.optim in [
- OptimizerNames.SCHEDULE_FREE_RADAM,
- OptimizerNames.SCHEDULE_FREE_ADAMW,
- OptimizerNames.SCHEDULE_FREE_SGD,
- ]:
- if not is_schedulefree_available():
- raise ImportError(
- "You need to install `schedulefree` in order to use schedulefree optimizers. "
- "Install it with `pip install schedulefree.`"
- )
- if not is_accelerate_available("0.30.0"):
- raise ImportError("You need to have `accelerate>=0.30.0` to be able to use schedulefree optimizers")
- from schedulefree import AdamWScheduleFree, SGDScheduleFree
- additional_optim_kwargs = {}
- require_warmup = True
- if args.optim == OptimizerNames.SCHEDULE_FREE_RADAM:
- if not is_schedulefree_available("1.4.0"):
- raise ImportError(
- "You need to install `schedulefree>=1.4.0` in order to use RAdamScheduleFree optimizer. "
- "Install it with `pip install schedulefree.`"
- )
- from schedulefree import RAdamScheduleFree
- optimizer_cls = RAdamScheduleFree
- additional_optim_kwargs = adam_kwargs
- require_warmup = False
- elif args.optim == OptimizerNames.SCHEDULE_FREE_ADAMW:
- optimizer_cls = AdamWScheduleFree
- additional_optim_kwargs = adam_kwargs
- elif args.optim == OptimizerNames.SCHEDULE_FREE_SGD:
- optimizer_cls = SGDScheduleFree
- else:
- raise ValueError("Invalid schedulefree optimizer")
- additional_optim_kwargs["weight_decay"] = args.weight_decay
- if require_warmup:
- additional_optim_kwargs["warmup_steps"] = args.warmup_steps
- additional_optim_kwargs.update(
- {
- "weight_lr_power": float(optim_args.get("weight_lr_power", 2.0)),
- "r": float(optim_args.get("r", 0.0)),
- }
- )
- optimizer_kwargs.update(additional_optim_kwargs)
- elif args.optim == OptimizerNames.STABLE_ADAMW:
- if not is_torch_optimi_available():
- raise ImportError(
- "You need to install `torch-optimi` in order to use stable_adamw optimizers. "
- "Install it with `pip install torch-optimi`."
- )
- from optimi import StableAdamW
- max_lr = optim_args.pop("max_lr", None)
- if max_lr is not None:
- max_lr = float(max_lr)
- kahan_sum = optim_args.pop("kahan_sum", None)
- if kahan_sum is not None:
- kahan_sum = bool(kahan_sum)
- adam_kwargs["weight_decay"] = args.weight_decay
- stable_adamw_kwargs = {
- "decouple_lr": bool(optim_args.pop("decouple_lr", False)),
- "max_lr": max_lr,
- "kahan_sum": kahan_sum,
- }
- optimizer_cls = StableAdamW
- optimizer_kwargs.update(adam_kwargs)
- optimizer_kwargs.update(stable_adamw_kwargs)
- else:
- raise ValueError(f"Trainer cannot instantiate unsupported optimizer: {args.optim}")
- return optimizer_cls, optimizer_kwargs
- def create_scheduler(self, num_training_steps: int, optimizer: torch.optim.Optimizer = None):
- """
- Setup the scheduler. The optimizer of the trainer must have been set up either before this method is called or
- passed as an argument.
- Args:
- num_training_steps (int): The number of training steps to do.
- """
- if self.lr_scheduler is None:
- self.lr_scheduler = get_scheduler(
- self.args.lr_scheduler_type,
- optimizer=self.optimizer if optimizer is None else optimizer,
- num_warmup_steps=self.args.get_warmup_steps(num_training_steps),
- num_training_steps=num_training_steps,
- scheduler_specific_kwargs=self.args.lr_scheduler_kwargs,
- )
- self._created_lr_scheduler = True
- return self.lr_scheduler
- def num_examples(self, dataloader: DataLoader) -> int:
- """
- Helper to get number of samples in a [`~torch.utils.data.DataLoader`] by accessing its dataset. When
- dataloader.dataset does not exist or has no length, estimates as best it can
- """
- try:
- dataset = dataloader.dataset
- # Special case for IterableDatasetShard, we need to dig deeper
- if isinstance(dataset, IterableDatasetShard):
- return len(dataloader.dataset.dataset)
- return len(dataloader.dataset)
- except (NameError, AttributeError, TypeError): # no dataset or length, estimate by length of dataloader
- return len(dataloader) * self.args.per_device_train_batch_size
- @staticmethod
- def num_tokens(train_dl: DataLoader, max_steps: Optional[int] = None) -> int:
- """
- Helper to get number of tokens in a [`~torch.utils.data.DataLoader`] by enumerating dataloader.
- """
- train_tokens = 0
- try:
- for batch in train_dl:
- tokens = batch["input_ids"].numel()
- if max_steps is not None:
- return tokens * max_steps
- train_tokens += tokens
- except KeyError:
- logger.warning("Cannot get num_tokens from dataloader")
- return train_tokens
- def _hp_search_setup(self, trial: Union["optuna.Trial", dict[str, Any]]):
- """HP search setup code"""
- self._trial = trial
- if self.hp_search_backend is None or trial is None:
- return
- if self.hp_search_backend == HPSearchBackend.OPTUNA:
- params = self.hp_space(trial)
- elif self.hp_search_backend == HPSearchBackend.RAY:
- params = trial
- params.pop("wandb", None)
- elif self.hp_search_backend == HPSearchBackend.SIGOPT:
- params = {k: int(v) if isinstance(v, str) else v for k, v in trial.assignments.items()}
- elif self.hp_search_backend == HPSearchBackend.WANDB:
- params = trial
- for key, value in params.items():
- if not hasattr(self.args, key):
- logger.warning(
- f"Trying to set {key} in the hyperparameter search but there is no corresponding field in"
- " `TrainingArguments`."
- )
- continue
- old_attr = getattr(self.args, key, None)
- # Casting value to the proper type
- if old_attr is not None:
- value = type(old_attr)(value)
- setattr(self.args, key, value)
- if self.hp_search_backend == HPSearchBackend.OPTUNA:
- logger.info(f"Trial: {trial.params}")
- if self.hp_search_backend == HPSearchBackend.SIGOPT:
- logger.info(f"SigOpt Assignments: {trial.assignments}")
- if self.hp_search_backend == HPSearchBackend.WANDB:
- logger.info(f"W&B Sweep parameters: {trial}")
- if self.is_deepspeed_enabled:
- if self.args.deepspeed is None:
- raise ValueError("For sweeps with deepspeed, `args.deepspeed` must be set")
- self.accelerator.free_memory()
- # Rebuild the deepspeed config to reflect the updated training parameters
- from accelerate.utils import DeepSpeedPlugin
- from transformers.integrations.deepspeed import HfTrainerDeepSpeedConfig
- self.args.hf_deepspeed_config = HfTrainerDeepSpeedConfig(self.args.deepspeed)
- self.args.hf_deepspeed_config.trainer_config_process(self.args)
- self.args.deepspeed_plugin = DeepSpeedPlugin(hf_ds_config=self.args.hf_deepspeed_config)
- # From 1.0 on, we need to fully wipe the DS plugin when doing sweeps.
- # Simply calling `_reset_state` is enough and doesn't need a version pin.
- AcceleratorState()._reset_state()
- self.create_accelerator_and_postprocess()
- def _report_to_hp_search(self, trial: Union["optuna.Trial", dict[str, Any]], step: int, metrics: dict[str, float]):
- if self.hp_search_backend is None or trial is None:
- return
- metrics = metrics.copy()
- self.objective = self.compute_objective(metrics)
- if self.hp_search_backend == HPSearchBackend.OPTUNA:
- import optuna
- if hasattr(trial, "study") and not trial.study._is_multi_objective():
- trial.report(self.objective, step)
- if trial.should_prune():
- self.callback_handler.on_train_end(self.args, self.state, self.control)
- raise optuna.TrialPruned()
- elif self.hp_search_backend == HPSearchBackend.RAY:
- import ray.train
- with tempfile.TemporaryDirectory() as temp_checkpoint_dir:
- checkpoint = None
- if self.control.should_save:
- self._tune_save_checkpoint(checkpoint_dir=temp_checkpoint_dir)
- checkpoint = ray.train.Checkpoint.from_directory(temp_checkpoint_dir)
- metrics["objective"] = self.objective
- ray.train.report(metrics, checkpoint=checkpoint)
- def _tune_save_checkpoint(self, checkpoint_dir: str):
- output_dir = os.path.join(checkpoint_dir, f"{PREFIX_CHECKPOINT_DIR}-{self.state.global_step}")
- self.save_model(output_dir, _internal_call=True)
- if self.args.should_save:
- # Update the `TrainerControl` state to where we are currently
- self.state.stateful_callbacks["TrainerControl"] = self.control.state()
- self.state.save_to_json(os.path.join(output_dir, TRAINER_STATE_NAME))
- torch.save(self.optimizer.state_dict(), os.path.join(output_dir, OPTIMIZER_NAME))
- torch.save(self.lr_scheduler.state_dict(), os.path.join(output_dir, SCHEDULER_NAME))
- def call_model_init(self, trial=None):
- model_init_argcount = number_of_arguments(self.model_init)
- if model_init_argcount == 0:
- model = self.model_init()
- elif model_init_argcount == 1:
- model = self.model_init(trial)
- else:
- raise RuntimeError("model_init should have 0 or 1 argument.")
- if model is None:
- raise RuntimeError("model_init should not return None.")
- return model
- def torch_jit_model_eval(self, model, dataloader, training=False):
- if not training:
- if dataloader is None:
- logger.warning("failed to use PyTorch jit mode due to current dataloader is none.")
- return model
- example_batch = next(iter(dataloader))
- example_batch = self._prepare_inputs(example_batch)
- try:
- jit_model = copy.copy(model)
- jit_model.eval()
- original_forward = jit_model.__dict__.pop("_original_forward", None)
- # remove mixed precision hooks from the model
- if original_forward:
- jit_model.forward = original_forward
- autocast_handler = AutocastKwargs(cache_enabled=False)
- with self.accelerator.autocast(autocast_handler=autocast_handler), torch.no_grad():
- if isinstance(example_batch, dict):
- jit_model = torch.jit.trace(jit_model, example_kwarg_inputs=example_batch, strict=False)
- else:
- jit_model = torch.jit.trace(
- jit_model,
- example_kwarg_inputs={key: example_batch[key] for key in example_batch},
- strict=False,
- )
- jit_model = torch.jit.freeze(jit_model)
- with torch.no_grad():
- jit_model(**example_batch)
- jit_model(**example_batch)
- model = jit_model
- self.use_cpu_amp = False
- except (RuntimeError, TypeError, ValueError, NameError, IndexError) as e:
- logger.warning(f"failed to use PyTorch jit mode due to: {e}.")
- return model
- def compare_trainer_and_checkpoint_args(self, training_args, trainer_state):
- attributes_map = {
- "logging_steps": "logging_steps",
- "eval_steps": "eval_steps",
- "save_steps": "save_steps",
- }
- has_warning = False
- warning_str = "Warning: The following arguments do not match the ones in the `trainer_state.json` within the checkpoint directory: "
- for arg_attr, state_attr in attributes_map.items():
- arg_value = getattr(training_args, arg_attr, None)
- state_value = getattr(trainer_state, state_attr, None)
- if arg_value is not None and state_value is not None and arg_value != state_value:
- warning_str += f"\n\t{arg_attr}: {arg_value} (from args) != {state_value} (from trainer_state.json)"
- has_warning = True
- # train bs is special as we need to account for multi-GPU
- train_bs_args = training_args.per_device_train_batch_size
- train_bs_state = trainer_state.train_batch_size // max(1, training_args.n_gpu)
- if train_bs_args != train_bs_state:
- warning_str += f"\n\tper_device_train_batch_size: {train_bs_args} (from args) != {train_bs_state} (from trainer_state.json)"
- has_warning = True
- if has_warning:
- logger.warning_once(warning_str)
- def _wrap_model(self, model, training=True, dataloader=None):
- if is_sagemaker_mp_enabled():
- # Wrapping the base model twice in a DistributedModel will raise an error.
- if isinstance(self.model_wrapped, smp.model.DistributedModel):
- return self.model_wrapped
- return smp.DistributedModel(model, backward_passes_per_step=self.args.gradient_accumulation_steps)
- # train/eval could be run multiple-times - if already wrapped, don't re-wrap it again
- if self.accelerator.unwrap_model(model, keep_torch_compile=False) is not model:
- return model
- # Mixed precision training with apex
- if self.use_apex and training:
- from apex import amp
- model, self.optimizer = amp.initialize(model, self.optimizer, opt_level=self.args.fp16_opt_level)
- # Multi-gpu training (should be after apex fp16 initialization) / 8bit models does not support DDP
- if self.args.n_gpu > 1 and not getattr(model, "is_loaded_in_8bit", False):
- model = nn.DataParallel(model)
- if self.args.jit_mode_eval:
- start_time = time.time()
- model = self.torch_jit_model_eval(model, dataloader, training)
- self.jit_compilation_time = round(time.time() - start_time, 4)
- # Note: in torch.distributed mode, there's no point in wrapping the model
- # inside a DistributedDataParallel as we'll be under `no_grad` anyways.
- if not training:
- return model
- # Distributed training (should be after apex fp16 initialization)
- # Distributed training using PyTorch FSDP
- if self.is_fsdp_xla_enabled:
- try:
- from torch_xla.distributed.fsdp import XlaFullyShardedDataParallel as FSDP
- from torch_xla.distributed.fsdp import checkpoint_module
- from torch_xla.distributed.fsdp.wrap import (
- size_based_auto_wrap_policy,
- transformer_auto_wrap_policy,
- )
- if self.is_fsdp_xla_v2_enabled:
- from torch_xla.experimental.spmd_fully_sharded_data_parallel import (
- SpmdFullyShardedDataParallel as FSDPv2,
- )
- except ImportError:
- raise ImportError("Missing XLA FSDP related module; please make sure to use torch-xla >= 2.0.")
- auto_wrap_policy = None
- auto_wrapper_callable = None
- default_transformer_cls_names_to_wrap = getattr(model, "_no_split_modules", None)
- fsdp_transformer_layer_cls_to_wrap = self.args.fsdp_config.get(
- "transformer_layer_cls_to_wrap", default_transformer_cls_names_to_wrap
- )
- if self.args.fsdp_config["min_num_params"] > 0:
- auto_wrap_policy = functools.partial(
- size_based_auto_wrap_policy, min_num_params=self.args.fsdp_config["min_num_params"]
- )
- elif fsdp_transformer_layer_cls_to_wrap is not None:
- transformer_cls_to_wrap = set()
- for layer_class in fsdp_transformer_layer_cls_to_wrap:
- transformer_cls = get_module_class_from_name(model, layer_class)
- if transformer_cls is None:
- raise Exception("Could not find the transformer layer class to wrap in the model.")
- else:
- transformer_cls_to_wrap.add(transformer_cls)
- auto_wrap_policy = functools.partial(
- transformer_auto_wrap_policy,
- # Transformer layer class to wrap
- transformer_layer_cls=transformer_cls_to_wrap,
- )
- fsdp_kwargs = self.args.xla_fsdp_config
- if self.args.fsdp_config["xla_fsdp_grad_ckpt"]:
- if model.config.use_cache:
- logger.warning_once(
- "`use_cache=True` is incompatible with gradient checkpointing. Setting `use_cache=False`."
- )
- model.config.use_cache = False
- # Apply gradient checkpointing to auto-wrapped sub-modules if specified
- def auto_wrapper_callable(m, *args, **kwargs):
- target_cls = FSDP if not self.is_fsdp_xla_v2_enabled else FSDPv2
- return target_cls(checkpoint_module(m), *args, **kwargs)
- # Wrap the base model with an outer FSDP wrapper
- if self.is_fsdp_xla_v2_enabled:
- def shard_output(output, mesh):
- from .modeling_outputs import CausalLMOutputWithPast
- real_output = None
- if isinstance(output, torch.Tensor):
- real_output = output
- elif isinstance(output, tuple):
- real_output = output[0]
- elif isinstance(output, CausalLMOutputWithPast):
- real_output = output.logits
- if real_output is None:
- raise ValueError("Something went wrong, the output of the model shouldn't be `None`")
- xs.mark_sharding(real_output, mesh, ("fsdp", None, None))
- self.model = model = FSDPv2(
- model,
- shard_output=shard_output,
- auto_wrap_policy=auto_wrap_policy,
- auto_wrapper_callable=auto_wrapper_callable,
- )
- else:
- self.model = model = FSDP(
- model,
- auto_wrap_policy=auto_wrap_policy,
- auto_wrapper_callable=auto_wrapper_callable,
- **fsdp_kwargs,
- )
- # Patch `xm.optimizer_step` should not reduce gradients in this case,
- # as FSDP does not need gradient reduction over sharded parameters.
- def patched_optimizer_step(optimizer, barrier=False, optimizer_args={}):
- loss = optimizer.step(**optimizer_args)
- if barrier:
- xm.mark_step()
- return loss
- xm.optimizer_step = patched_optimizer_step
- elif is_sagemaker_dp_enabled():
- model = nn.parallel.DistributedDataParallel(
- model, device_ids=[int(os.getenv("SMDATAPARALLEL_LOCAL_RANK"))]
- )
- elif self.args.parallel_mode == ParallelMode.DISTRIBUTED:
- if is_torch_neuroncore_available():
- return model
- kwargs = {}
- if self.args.ddp_find_unused_parameters is not None:
- kwargs["find_unused_parameters"] = self.args.ddp_find_unused_parameters
- elif isinstance(model, PreTrainedModel):
- # find_unused_parameters breaks checkpointing as per
- # https://github.com/huggingface/transformers/pull/4659#issuecomment-643356021
- kwargs["find_unused_parameters"] = not model.is_gradient_checkpointing
- else:
- kwargs["find_unused_parameters"] = True
- if self.args.ddp_bucket_cap_mb is not None:
- kwargs["bucket_cap_mb"] = self.args.ddp_bucket_cap_mb
- if self.args.ddp_broadcast_buffers is not None:
- kwargs["broadcast_buffers"] = self.args.ddp_broadcast_buffers
- self.accelerator.ddp_handler = DistributedDataParallelKwargs(**kwargs)
- return model
- def train(
- self,
- resume_from_checkpoint: Optional[Union[str, bool]] = None,
- trial: Union["optuna.Trial", dict[str, Any], None] = None,
- ignore_keys_for_eval: Optional[list[str]] = None,
- **kwargs: Any,
- ):
- """
- Main training entry point.
- Args:
- resume_from_checkpoint (`str` or `bool`, *optional*):
- If a `str`, local path to a saved checkpoint as saved by a previous instance of [`Trainer`]. If a
- `bool` and equals `True`, load the last checkpoint in *args.output_dir* as saved by a previous instance
- of [`Trainer`]. If present, training will resume from the model/optimizer/scheduler states loaded here.
- trial (`optuna.Trial` or `dict[str, Any]`, *optional*):
- The trial run or the hyperparameter dictionary for hyperparameter search.
- ignore_keys_for_eval (`list[str]`, *optional*)
- A list of keys in the output of your model (if it is a dictionary) that should be ignored when
- gathering predictions for evaluation during the training.
- kwargs (`dict[str, Any]`, *optional*):
- Additional keyword arguments used to hide deprecated arguments
- """
- if resume_from_checkpoint is False:
- resume_from_checkpoint = None
- # memory metrics - must set up as early as possible
- self._memory_tracker.start()
- args = self.args
- self.is_in_train = True
- # If the model uses a tokenizer, it may have a new tokens for fine-tuning purposes.
- if isinstance(self.processing_class, (PreTrainedTokenizerBase, ProcessorMixin)) and hasattr(
- self.model, "config"
- ):
- self._align_special_tokens()
- # Attach NEFTune hooks if necessary
- if self.neftune_noise_alpha is not None:
- self.model = self._activate_neftune(self.model)
- # do_train is not a reliable argument, as it might not be set and .train() still called, so
- # the following is a workaround:
- if (
- (args.fp16_full_eval or args.bf16_full_eval)
- and not args.do_train
- and not self.is_model_parallel
- and self.model_init is None
- ):
- self._move_model_to_device(self.model, args.device)
- if "model_path" in kwargs:
- resume_from_checkpoint = kwargs.pop("model_path")
- warnings.warn(
- "`model_path` is deprecated and will be removed in a future version. Use `resume_from_checkpoint` "
- "instead.",
- FutureWarning,
- )
- if len(kwargs) > 0:
- raise TypeError(f"train() got unexpected keyword arguments: {', '.join(list(kwargs.keys()))}.")
- # This might change the seed so needs to run first.
- self._hp_search_setup(trial)
- self._train_batch_size = self.args.train_batch_size
- # Model re-init
- model_reloaded = False
- if self.model_init is not None:
- # Seed must be set before instantiating the model when using model_init.
- enable_full_determinism(self.args.seed) if self.args.full_determinism else set_seed(self.args.seed)
- self.model = self.call_model_init(trial)
- model_reloaded = True
- # Reinitializes optimizer and scheduler
- self.optimizer, self.lr_scheduler = None, None
- # Load potential model checkpoint
- if isinstance(resume_from_checkpoint, bool) and resume_from_checkpoint:
- resume_from_checkpoint = get_last_checkpoint(args.output_dir)
- if resume_from_checkpoint is None:
- raise ValueError(f"No valid checkpoint found in output directory ({args.output_dir})")
- if resume_from_checkpoint is not None:
- if not is_sagemaker_mp_enabled() and not self.is_deepspeed_enabled and not self.is_fsdp_enabled:
- self._load_from_checkpoint(resume_from_checkpoint)
- # In case of repeating the find_executable_batch_size, set `self._train_batch_size` properly
- state = TrainerState.load_from_json(os.path.join(resume_from_checkpoint, TRAINER_STATE_NAME))
- if state.train_batch_size is not None:
- self._train_batch_size = state.train_batch_size
- # If model was re-initialized, put it on the right device and update self.model_wrapped
- if model_reloaded:
- if self.place_model_on_device:
- self._move_model_to_device(self.model, args.device)
- self.model_wrapped = self.model
- inner_training_loop = find_executable_batch_size(
- self._inner_training_loop, self._train_batch_size, args.auto_find_batch_size
- )
- if args.push_to_hub:
- try:
- # Disable progress bars when uploading models during checkpoints to avoid polluting stdout
- hf_hub_utils.disable_progress_bars()
- return inner_training_loop(
- args=args,
- resume_from_checkpoint=resume_from_checkpoint,
- trial=trial,
- ignore_keys_for_eval=ignore_keys_for_eval,
- )
- finally:
- hf_hub_utils.enable_progress_bars()
- else:
- return inner_training_loop(
- args=args,
- resume_from_checkpoint=resume_from_checkpoint,
- trial=trial,
- ignore_keys_for_eval=ignore_keys_for_eval,
- )
- def get_tp_size(self) -> int:
- """Get the tensor parallel size from either the model or DeepSpeed config."""
- # 1. Check model.tp_size first
- if (model_tp := getattr(self.model, "_tp_size", None)) is not None:
- return model_tp
- # 2. Fall back to DeepSpeed config if enabled
- if self.is_deepspeed_enabled and (deepspeed_config := getattr(self.args, "hf_deepspeed_config", None)):
- return deepspeed_config.config.get("tensor_parallel", {}).get("autotp_size", 1)
- # 3. Default fallback
- return 1
- def get_total_train_batch_size(self, args) -> int:
- """Calculates total batch size (micro_batch * grad_accum * dp_world_size).
- Note: Only considers DP and TP (dp_world_size = world_size // tp_size)."""
- dp_world_size = args.world_size // self.get_tp_size()
- return self._train_batch_size * args.gradient_accumulation_steps * dp_world_size
- def _inner_training_loop(
- self, batch_size=None, args=None, resume_from_checkpoint=None, trial=None, ignore_keys_for_eval=None
- ):
- self.accelerator.free_memory()
- self._train_batch_size = batch_size
- if self.args.auto_find_batch_size:
- if self.state.train_batch_size != self._train_batch_size:
- from accelerate.utils import release_memory
- (self.model_wrapped,) = release_memory(self.model_wrapped)
- self.model_wrapped = self.model
- # Check for DeepSpeed *after* the initial pass and modify the config
- if self.is_deepspeed_enabled:
- # Temporarily unset `self.args.train_batch_size`
- original_bs = self.args.per_device_train_batch_size
- self.args.per_device_train_batch_size = self._train_batch_size // max(1, self.args.n_gpu)
- self.propagate_args_to_deepspeed(True)
- self.args.per_device_train_batch_size = original_bs
- self.state.train_batch_size = self._train_batch_size
- logger.debug(f"Currently training with a batch size of: {self._train_batch_size}")
- # Data loader and number of training steps
- train_dataloader = self.get_train_dataloader()
- if self.is_fsdp_xla_v2_enabled:
- train_dataloader = tpu_spmd_dataloader(train_dataloader)
- # Setting up training control variables:
- # number of training epochs: num_train_epochs
- # number of training steps per epoch: num_update_steps_per_epoch
- # total number of training steps to execute: max_steps
- total_train_batch_size = self.get_total_train_batch_size(args)
- (
- num_train_epochs,
- num_update_steps_per_epoch,
- num_examples,
- num_train_samples,
- epoch_based,
- len_dataloader,
- max_steps,
- ) = self.set_initial_training_values(args, train_dataloader, total_train_batch_size)
- num_train_tokens = None
- if self.args.include_tokens_per_second:
- num_train_tokens = self.num_tokens(train_dataloader, None if epoch_based else max_steps)
- # If going by epochs, multiply tokens linearly
- if len_dataloader is not None and epoch_based:
- num_train_tokens *= args.num_train_epochs
- # Otherwise since its steps, we just multiply by grad accum
- else:
- num_train_tokens *= args.gradient_accumulation_steps
- if DebugOption.UNDERFLOW_OVERFLOW in self.args.debug:
- if self.args.n_gpu > 1:
- # nn.DataParallel(model) replicates the model, creating new variables and module
- # references registered here no longer work on other gpus, breaking the module
- raise ValueError(
- "Currently --debug underflow_overflow is not supported under DP. Please use DDP"
- " (torchrun or torch.distributed.launch (deprecated))."
- )
- else:
- DebugUnderflowOverflow(self.model)
- delay_optimizer_creation = is_sagemaker_mp_enabled() or self.is_fsdp_xla_enabled or self.is_fsdp_enabled
- # Can't delay optimizer creation when using FSDP2: https://github.com/huggingface/accelerate/blob/3f636d626063ffcf9a337c7d3624d61b7d187d59/src/accelerate/accelerator.py#L1404
- is_fsdp2 = self.is_fsdp_enabled and (getattr(self.accelerator.state.fsdp_plugin, "fsdp_version", 1) == 2)
- if is_fsdp2:
- delay_optimizer_creation = False
- # We need to reset the scheduler, as its parameters may be different on subsequent calls
- if self._created_lr_scheduler:
- self.lr_scheduler = None
- self._created_lr_scheduler = False
- if self.is_deepspeed_enabled:
- self.optimizer, self.lr_scheduler = deepspeed_init(self, num_training_steps=max_steps)
- if not delay_optimizer_creation:
- self.create_optimizer_and_scheduler(num_training_steps=max_steps)
- self.state = TrainerState(
- stateful_callbacks=[
- cb for cb in self.callback_handler.callbacks + [self.control] if isinstance(cb, ExportableState)
- ]
- )
- self.state.is_hyper_param_search = trial is not None
- self.state.train_batch_size = self._train_batch_size
- # Compute absolute values for logging, eval, and save if given as ratio
- self.state.compute_steps(args, max_steps)
- # Activate gradient checkpointing if needed
- if args.gradient_checkpointing:
- self.model.gradient_checkpointing_enable(gradient_checkpointing_kwargs=args.gradient_checkpointing_kwargs)
- model = self._wrap_model(self.model_wrapped)
- # as the model is wrapped, don't use `accelerator.prepare`
- # this is for unhandled cases such as
- # FSDP-XLA, SageMaker MP/DP, DataParallel, IPEX
- use_accelerator_prepare = model is self.model
- if use_accelerator_prepare and self.is_fsdp_enabled:
- # In case of auto_find_batch_size=True
- # Remove FSDP wrapping from sub-models.
- self.model = unwrap_model(self.model, recursive=True)
- if delay_optimizer_creation:
- if use_accelerator_prepare:
- # configure fsdp plugin for qlora if any
- self._fsdp_qlora_plugin_updates()
- if self.accelerator.mixed_precision != "fp8":
- self.model = self.accelerator.prepare(self.model)
- self.create_optimizer_and_scheduler(num_training_steps=max_steps)
- # prepare using `accelerator` prepare
- if use_accelerator_prepare:
- self.model.train()
- if hasattr(self.lr_scheduler, "step"):
- if self.use_apex:
- model = self.accelerator.prepare(self.model)
- else:
- # We should avoid accelerate preparing the model in TP case since we dont need it as it is handled by transformers from_pretrained and also it goes into DDP based preparation.
- if self.is_tp_enabled:
- self.optimizer = self.accelerator.prepare(self.optimizer)
- else:
- model, self.optimizer = self.accelerator.prepare(self.model, self.optimizer)
- else:
- # to handle cases wherein we pass "DummyScheduler" such as when it is specified in DeepSpeed config.
- model, self.optimizer, self.lr_scheduler = self.accelerator.prepare(
- self.model, self.optimizer, self.lr_scheduler
- )
- else:
- self.optimizer = self.accelerator.prepare(self.optimizer)
- if self.is_fsdp_enabled:
- self.model = self.model_wrapped = model
- # for the rest of this function `model` is the outside model, whether it was wrapped or not
- if model is not self.model:
- self.model_wrapped = model
- # backward compatibility
- if self.is_deepspeed_enabled:
- self.deepspeed = self.model_wrapped
- # ckpt loading
- if resume_from_checkpoint is not None:
- if self.is_deepspeed_enabled:
- deepspeed_load_checkpoint(
- self.model_wrapped, resume_from_checkpoint, load_module_strict=not _is_peft_model(self.model)
- )
- elif is_sagemaker_mp_enabled() or self.is_fsdp_enabled:
- self._load_from_checkpoint(resume_from_checkpoint, self.model_wrapped)
- # Check if saved optimizer or scheduler states exist
- self._load_optimizer_and_scheduler(resume_from_checkpoint)
- self._load_scaler(resume_from_checkpoint)
- # important: at this point:
- # self.model is the Transformers Model
- # self.model_wrapped is DDP(Transformers Model), Deepspeed(Transformers Model),
- # FSDP(Transformers Model), Dynamo Optimized Module(Transformers Model) etc.
- # Train!
- logger.info("***** Running training *****")
- logger.info(f" Num examples = {num_examples:,}")
- logger.info(f" Num Epochs = {num_train_epochs:,}")
- logger.info(f" Instantaneous batch size per device = {self.args.per_device_train_batch_size:,}")
- if self.args.per_device_train_batch_size != self._train_batch_size:
- logger.info(f" Training with DataParallel so batch size has been adjusted to: {self._train_batch_size:,}")
- logger.info(f" Total train batch size (w. parallel, distributed & accumulation) = {total_train_batch_size:,}")
- logger.info(f" Gradient Accumulation steps = {args.gradient_accumulation_steps}")
- logger.info(f" Total optimization steps = {max_steps:,}")
- logger.info(f" Number of trainable parameters = {get_model_param_count(model, trainable_only=True):,}")
- self.state.epoch = 0
- start_time = time.time()
- epochs_trained = 0
- steps_trained_in_current_epoch = 0
- # Check if continuing training from a checkpoint
- if resume_from_checkpoint is not None and os.path.isfile(
- os.path.join(resume_from_checkpoint, TRAINER_STATE_NAME)
- ):
- self.state = TrainerState.load_from_json(os.path.join(resume_from_checkpoint, TRAINER_STATE_NAME))
- self.compare_trainer_and_checkpoint_args(self.args, self.state)
- self._load_callback_state()
- epochs_trained = int(self.state.global_step // num_update_steps_per_epoch)
- if not args.ignore_data_skip:
- steps_trained_in_current_epoch = self.state.global_step % (num_update_steps_per_epoch)
- steps_trained_in_current_epoch *= args.gradient_accumulation_steps
- else:
- steps_trained_in_current_epoch = 0
- logger.info(" Continuing training from checkpoint, will skip to saved global_step")
- logger.info(f" Continuing training from epoch {epochs_trained}")
- logger.info(f" Continuing training from global step {self.state.global_step}")
- if not args.ignore_data_skip:
- logger.info(
- f" Will skip the first {epochs_trained} epochs then the first"
- f" {steps_trained_in_current_epoch} batches in the first epoch."
- )
- # Update the references
- for attr in ("model", "optimizer", "lr_scheduler"):
- setattr(self.callback_handler, attr, getattr(self, attr))
- self.callback_handler.train_dataloader = train_dataloader
- self.state.init_training_references(self, max_steps, num_train_epochs, trial)
- # tr_loss is a tensor to avoid synchronization of TPUs through .item()
- tr_loss = torch.tensor(0.0, device=args.device)
- # _total_loss_scalar is updated everytime .item() has to be called on tr_loss and stores the sum of all losses
- self._total_loss_scalar = 0.0
- self._globalstep_last_logged = self.state.global_step
- model.zero_grad()
- grad_norm: Optional[float] = None
- learning_rate = None
- self.control = self.callback_handler.on_train_begin(args, self.state, self.control)
- if args.eval_on_start:
- self._evaluate(trial, ignore_keys_for_eval, skip_scheduler=True)
- for epoch in range(epochs_trained, num_train_epochs):
- epoch_dataloader = train_dataloader
- if hasattr(epoch_dataloader, "set_epoch"):
- epoch_dataloader.set_epoch(epoch)
- # Reset the past mems state at the beginning of each epoch if necessary.
- if args.past_index >= 0:
- self._past = None
- steps_in_epoch = (
- len(epoch_dataloader)
- if len_dataloader is not None
- else args.max_steps * args.gradient_accumulation_steps
- )
- self.control = self.callback_handler.on_epoch_begin(args, self.state, self.control)
- step = -1
- rng_to_sync = False
- # Handle resumption from checkpoint
- if epoch == epochs_trained and resume_from_checkpoint is not None:
- if steps_trained_in_current_epoch > 0 and not args.ignore_data_skip:
- epoch_dataloader = skip_first_batches(epoch_dataloader, steps_trained_in_current_epoch)
- step = steps_trained_in_current_epoch - 1
- rng_to_sync = True
- elif steps_trained_in_current_epoch == 0:
- self._load_rng_state(resume_from_checkpoint)
- epoch_iterator = iter(epoch_dataloader)
- # We chunkify the epoch iterator into gradient accumulation steps `n` batches
- remainder = steps_in_epoch % args.gradient_accumulation_steps
- if remainder == 0:
- remainder = args.gradient_accumulation_steps
- update_step = -1
- total_updates = steps_in_epoch // args.gradient_accumulation_steps + int(
- remainder < args.gradient_accumulation_steps
- )
- for _ in range(total_updates):
- update_step += 1
- num_batches = args.gradient_accumulation_steps if update_step != (total_updates - 1) else remainder
- batch_samples, num_items_in_batch = self.get_batch_samples(epoch_iterator, num_batches, args.device)
- # Store the number of batches for current gradient accumulation
- # This is used to correctly scale the loss when the last accumulation step has fewer batches
- self.current_gradient_accumulation_steps = len(batch_samples)
- for i, inputs in enumerate(batch_samples):
- step += 1
- do_sync_step = (step + 1) % args.gradient_accumulation_steps == 0 or (step + 1) == steps_in_epoch
- # Since we perform prefetching, we need to manually set sync_gradients
- self.accelerator.gradient_state._set_sync_gradients(do_sync_step)
- if self.args.include_num_input_tokens_seen not in ["no", False]:
- main_input_name = getattr(self.model, "main_input_name", "input_ids")
- if main_input_name not in inputs:
- logger.warning(
- "Tried to track the number of tokens seen, however the current model is "
- "not configured properly to know what item is the input. To fix this, add "
- "a `main_input_name` attribute to the model class you are using."
- )
- else:
- if self.args.include_num_input_tokens_seen == "non_padding":
- if "attention_mask" in inputs:
- input_tokens = inputs["attention_mask"].sum()
- elif (
- self.processing_class is not None
- and hasattr(self.processing_class, "pad_token_id")
- and self.processing_class.pad_token_id is not None
- ):
- input_tokens = (
- inputs[main_input_name] != self.processing_class.pad_token_id
- ).sum()
- else:
- logger.warning(
- "Could not determine method to count non-padding tokens, falling back to counting all tokens."
- )
- input_tokens = inputs[main_input_name].numel()
- else:
- input_tokens = inputs[main_input_name].numel()
- input_tokens = torch.tensor(input_tokens, device=self.args.device, dtype=torch.int64)
- self.state.num_input_tokens_seen += self.accelerator.gather(input_tokens).sum().item()
- if rng_to_sync:
- self._load_rng_state(resume_from_checkpoint)
- rng_to_sync = False
- if step % args.gradient_accumulation_steps == 0:
- self.control = self.callback_handler.on_step_begin(args, self.state, self.control)
- # We explicitly want to avoid relying on `accelerator.accumulate` for generation training
- context = (
- functools.partial(self.accelerator.no_sync, model=model)
- if i != len(batch_samples) - 1
- and self.accelerator.distributed_type != DistributedType.DEEPSPEED
- else contextlib.nullcontext
- )
- with context():
- tr_loss_step = self.training_step(model, inputs, num_items_in_batch)
- if (
- args.logging_nan_inf_filter
- and not is_torch_xla_available()
- and (torch.isnan(tr_loss_step) or torch.isinf(tr_loss_step))
- ):
- # if loss is nan or inf simply add the average of previous logged losses
- tr_loss = tr_loss + tr_loss / (1 + self.state.global_step - self._globalstep_last_logged)
- else:
- if tr_loss.device != tr_loss_step.device:
- raise ValueError(
- f"Calculated loss must be on the original device: {tr_loss.device} but device in use is {tr_loss_step.device}"
- )
- tr_loss = tr_loss + tr_loss_step
- self.current_flos += float(self.floating_point_ops(inputs))
- if do_sync_step:
- # Since we perform prefetching, we need to manually set sync_gradients to True
- self.accelerator.gradient_state._set_sync_gradients(True)
- # Gradient clipping
- if args.max_grad_norm is not None and args.max_grad_norm > 0:
- if is_sagemaker_mp_enabled() and args.fp16:
- _grad_norm = self.optimizer.clip_master_grads(args.max_grad_norm)
- elif self.use_apex:
- from apex import amp
- # Revert to normal clipping otherwise, handling Apex or full precision
- _grad_norm = nn.utils.clip_grad_norm_(
- amp.master_params(self.optimizer),
- args.max_grad_norm,
- )
- else:
- grad_norm_context = contextlib.nullcontext
- if self.is_tp_enabled:
- from torch.distributed._tensor.experimental import implicit_replication
- grad_norm_context = implicit_replication
- with grad_norm_context():
- _grad_norm = self.accelerator.clip_grad_norm_(
- model.parameters(),
- args.max_grad_norm,
- )
- if (
- is_accelerate_available()
- and self.accelerator.distributed_type == DistributedType.DEEPSPEED
- ):
- grad_norm = model.get_global_grad_norm()
- # In some cases the grad norm may not return a float
- if hasattr(grad_norm, "item"):
- grad_norm = grad_norm.item()
- else:
- grad_norm = _grad_norm
- self.control = self.callback_handler.on_pre_optimizer_step(args, self.state, self.control)
- context = contextlib.nullcontext
- if self.is_tp_enabled:
- from torch.distributed._tensor.experimental import implicit_replication
- context = implicit_replication
- with context():
- self.optimizer.step()
- self.control = self.callback_handler.on_optimizer_step(args, self.state, self.control)
- # get leaning rate before update
- learning_rate = self._get_learning_rate()
- if not self.accelerator.optimizer_step_was_skipped:
- # Delay optimizer scheduling until metrics are generated
- if not isinstance(self.lr_scheduler, torch.optim.lr_scheduler.ReduceLROnPlateau):
- self.lr_scheduler.step()
- model.zero_grad()
- self.state.global_step += 1
- self.state.epoch = epoch + (step + 1) / steps_in_epoch
- self.control = self.callback_handler.on_step_end(args, self.state, self.control)
- self._maybe_log_save_evaluate(
- tr_loss,
- grad_norm,
- model,
- trial,
- epoch,
- ignore_keys_for_eval,
- start_time,
- learning_rate=learning_rate,
- )
- else:
- self.control = self.callback_handler.on_substep_end(args, self.state, self.control)
- # PyTorch/XLA relies on the data loader to insert the mark_step for
- # each step. Since we are breaking the loop early, we need to manually
- # insert the mark_step here.
- if self.control.should_epoch_stop or self.control.should_training_stop:
- if is_torch_xla_available():
- xm.mark_step()
- break
- # We also need to break out of the nested loop
- if self.control.should_epoch_stop or self.control.should_training_stop:
- if is_torch_xla_available():
- xm.mark_step()
- break
- if step < 0:
- logger.warning(
- "There seems not to be a single sample in your epoch_iterator, stopping training at step"
- f" {self.state.global_step}! This is expected if you're using an IterableDataset and set"
- f" num_steps ({max_steps}) higher than the number of available samples."
- )
- self.control.should_training_stop = True
- self.control = self.callback_handler.on_epoch_end(args, self.state, self.control)
- self._maybe_log_save_evaluate(
- tr_loss, grad_norm, model, trial, epoch, ignore_keys_for_eval, start_time, learning_rate=learning_rate
- )
- if DebugOption.TPU_METRICS_DEBUG in self.args.debug:
- if is_torch_xla_available():
- # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
- xm.master_print(met.metrics_report())
- else:
- logger.warning(
- "You enabled PyTorch/XLA debug metrics but you don't have a TPU "
- "configured. Check your training configuration if this is unexpected."
- )
- if self.control.should_training_stop:
- break
- if args.past_index and hasattr(self, "_past"):
- # Clean the state at the end of training
- delattr(self, "_past")
- logger.info("\n\nTraining completed. Do not forget to share your model on huggingface.co/models =)\n\n")
- if args.load_best_model_at_end and self.state.best_model_checkpoint is not None:
- self._load_best_model()
- # add remaining tr_loss
- self._total_loss_scalar += tr_loss.item()
- effective_global_step = max(self.state.global_step, 0.001) # Avoid ZeroDivisionError
- train_loss = self._total_loss_scalar / effective_global_step
- metrics = speed_metrics(
- "train",
- start_time,
- num_samples=num_train_samples,
- num_steps=self.state.max_steps,
- num_tokens=num_train_tokens,
- )
- self.store_flos()
- metrics["total_flos"] = self.state.total_flos
- metrics["train_loss"] = train_loss
- self.is_in_train = False
- self._memory_tracker.stop_and_update_metrics(metrics)
- self.log(metrics)
- run_dir = self._get_output_dir(trial)
- checkpoints_sorted = self._sorted_checkpoints(use_mtime=False, output_dir=run_dir)
- # Delete the last checkpoint when save_total_limit=1 if it's different from the best checkpoint and process allowed to save.
- if self.args.should_save and self.state.best_model_checkpoint is not None and self.args.save_total_limit == 1:
- for checkpoint in checkpoints_sorted:
- if not os.path.samefile(checkpoint, self.state.best_model_checkpoint):
- logger.info(f"Deleting older checkpoint [{checkpoint}] due to args.save_total_limit")
- shutil.rmtree(checkpoint, ignore_errors=True)
- self.control = self.callback_handler.on_train_end(args, self.state, self.control)
- # Wait for the checkpoint to be uploaded.
- self._finish_current_push()
- # After training we make sure to retrieve back the original forward pass method
- # for the embedding layer by removing the forward post hook.
- if self.neftune_noise_alpha is not None:
- self._deactivate_neftune(self.model)
- return TrainOutput(self.state.global_step, train_loss, metrics)
- def _get_output_dir(self, trial):
- if self.hp_search_backend is not None and trial is not None:
- if self.hp_search_backend == HPSearchBackend.OPTUNA:
- run_id = trial.number
- elif self.hp_search_backend == HPSearchBackend.RAY:
- import ray.train
- run_id = ray.train.get_context().get_trial_id()
- elif self.hp_search_backend == HPSearchBackend.SIGOPT:
- run_id = trial.id
- elif self.hp_search_backend == HPSearchBackend.WANDB:
- import wandb
- run_id = wandb.run.id
- run_name = self.hp_name(trial) if self.hp_name is not None else f"run-{run_id}"
- run_dir = os.path.join(self.args.output_dir, run_name)
- else:
- run_dir = self.args.output_dir
- return run_dir
- def _load_from_checkpoint(self, resume_from_checkpoint, model=None):
- if model is None:
- model = self.model
- config_file = os.path.join(resume_from_checkpoint, CONFIG_NAME)
- adapter_weights_file = os.path.join(resume_from_checkpoint, ADAPTER_WEIGHTS_NAME)
- adapter_safe_weights_file = os.path.join(resume_from_checkpoint, ADAPTER_SAFE_WEIGHTS_NAME)
- weights_file = os.path.join(resume_from_checkpoint, WEIGHTS_NAME)
- weights_index_file = os.path.join(resume_from_checkpoint, WEIGHTS_INDEX_NAME)
- safe_weights_file = os.path.join(resume_from_checkpoint, SAFE_WEIGHTS_NAME)
- safe_weights_index_file = os.path.join(resume_from_checkpoint, SAFE_WEIGHTS_INDEX_NAME)
- is_fsdp_ckpt = os.path.isdir(resume_from_checkpoint) and (
- # this checks the FSDP state dict when `SHARDED_STATE_DICT` is used
- any(
- FSDP_MODEL_NAME in folder_name
- for folder_name in os.listdir(resume_from_checkpoint)
- if os.path.isdir(os.path.join(resume_from_checkpoint, folder_name))
- )
- # this checks the FSDP state dict when `FULL_STATE_DICT` is used
- or os.path.isfile(os.path.join(resume_from_checkpoint, f"{FSDP_MODEL_NAME}.bin"))
- )
- # if multiple adapters exist, they get saved in sub directories
- adapter_subdirs = (
- [
- folder_name
- for folder_name in os.listdir(resume_from_checkpoint)
- if os.path.isdir(os.path.join(resume_from_checkpoint, folder_name))
- and (
- os.path.isfile(os.path.join(resume_from_checkpoint, folder_name, ADAPTER_WEIGHTS_NAME))
- or os.path.isfile(os.path.join(resume_from_checkpoint, folder_name, ADAPTER_SAFE_WEIGHTS_NAME))
- )
- ]
- if os.path.isdir(resume_from_checkpoint)
- else []
- )
- if is_fsdp_ckpt and not self.is_fsdp_enabled:
- raise ValueError(f"Checkpoint found at {resume_from_checkpoint} is only supported when using PyTorch FSDP")
- if not (
- any(
- os.path.isfile(f)
- for f in [
- weights_file,
- safe_weights_file,
- weights_index_file,
- safe_weights_index_file,
- adapter_weights_file,
- adapter_safe_weights_file,
- ]
- )
- or is_fsdp_ckpt
- or adapter_subdirs
- ):
- raise ValueError(f"Can't find a valid checkpoint at {resume_from_checkpoint}")
- logger.info(f"Loading model from {resume_from_checkpoint}.")
- if os.path.isfile(config_file):
- config = PretrainedConfig.from_json_file(config_file)
- checkpoint_version = config.transformers_version
- if checkpoint_version is not None and checkpoint_version != __version__:
- logger.warning(
- f"You are resuming training from a checkpoint trained with {checkpoint_version} of "
- f"Transformers but your current version is {__version__}. This is not recommended and could "
- "yield to errors or unwanted behaviors."
- )
- if os.path.isfile(weights_file) or os.path.isfile(safe_weights_file) or is_fsdp_ckpt:
- # If the model is on the GPU, it still works!
- if is_sagemaker_mp_enabled():
- if os.path.isfile(os.path.join(resume_from_checkpoint, "user_content.pt")):
- # If the 'user_content.pt' file exists, load with the new smp api.
- # Checkpoint must have been saved with the new smp api.
- smp.resume_from_checkpoint(
- path=resume_from_checkpoint, tag=WEIGHTS_NAME, partial=False, load_optimizer=False
- )
- else:
- # If the 'user_content.pt' file does NOT exist, load with the old smp api.
- # Checkpoint must have been saved with the old smp api.
- if hasattr(self.args, "fp16") and self.args.fp16 is True:
- logger.warning(
- "Enabling FP16 and loading from smp < 1.10 checkpoint together is not supported."
- )
- check_torch_load_is_safe()
- state_dict = torch.load(weights_file, map_location="cpu", weights_only=True)
- # Required for smp to not auto-translate state_dict from hf to smp (is already smp).
- state_dict["_smp_is_partial"] = False
- load_result = model.load_state_dict(state_dict, strict=True)
- # release memory
- del state_dict
- elif self.is_fsdp_enabled:
- load_fsdp_model(
- self.accelerator.state.fsdp_plugin,
- self.accelerator,
- model,
- resume_from_checkpoint,
- **_get_fsdp_ckpt_kwargs(),
- )
- else:
- # We load the model state dict on the CPU to avoid an OOM error.
- if self.args.save_safetensors and os.path.isfile(safe_weights_file):
- state_dict = safetensors.torch.load_file(safe_weights_file, device="cpu")
- else:
- check_torch_load_is_safe()
- state_dict = torch.load(weights_file, map_location="cpu", weights_only=True)
- # workaround for FSDP bug https://github.com/pytorch/pytorch/issues/82963
- # which takes *args instead of **kwargs
- load_result = model.load_state_dict(state_dict, False)
- # release memory
- del state_dict
- self._issue_warnings_after_load(load_result)
- # Load adapters following PR # 24096
- elif _is_peft_model(model):
- # If train a model using PEFT & LoRA, assume that adapter have been saved properly.
- # TODO: in the future support only specific min PEFT versions
- if (hasattr(model, "active_adapter") or hasattr(model, "active_adapters")) and hasattr(
- model, "load_adapter"
- ):
- if os.path.exists(resume_from_checkpoint):
- # For BC for older PEFT versions
- if hasattr(model, "active_adapters"):
- active_adapters = model.active_adapters
- if len(active_adapters) > 1:
- logger.warning("Multiple active adapters detected will only consider the first adapter")
- active_adapter = active_adapters[0]
- else:
- active_adapter = model.active_adapter
- if adapter_subdirs:
- for subdir_name in adapter_subdirs:
- peft_id = os.path.join(resume_from_checkpoint, subdir_name)
- model.load_adapter(peft_id, subdir_name, is_trainable=(subdir_name == active_adapter))
- model.set_adapter(active_adapter)
- else:
- model.load_adapter(resume_from_checkpoint, active_adapter, is_trainable=True)
- else:
- logger.warning(
- "The intermediate checkpoints of PEFT may not be saved correctly, "
- f"consider using a custom callback to save {ADAPTER_WEIGHTS_NAME} in corresponding saving folders. "
- "Check some examples here: https://github.com/huggingface/peft/issues/96"
- )
- else:
- logger.warning("Could not load adapter model, make sure to have `peft>=0.3.0` installed")
- else:
- # We load the sharded checkpoint
- load_result = load_sharded_checkpoint(
- model, resume_from_checkpoint, strict=is_sagemaker_mp_enabled(), prefer_safe=self.args.save_safetensors
- )
- if not is_sagemaker_mp_enabled():
- self._issue_warnings_after_load(load_result)
- def _load_best_model(self):
- logger.info(f"Loading best model from {self.state.best_model_checkpoint} (score: {self.state.best_metric}).")
- best_model_path = os.path.join(self.state.best_model_checkpoint, WEIGHTS_NAME)
- best_safe_model_path = os.path.join(self.state.best_model_checkpoint, SAFE_WEIGHTS_NAME)
- best_adapter_model_path = os.path.join(self.state.best_model_checkpoint, ADAPTER_WEIGHTS_NAME)
- best_safe_adapter_model_path = os.path.join(self.state.best_model_checkpoint, ADAPTER_SAFE_WEIGHTS_NAME)
- model = self.model_wrapped if is_sagemaker_mp_enabled() else self.model
- if self.is_deepspeed_enabled:
- deepspeed_load_checkpoint(
- self.model_wrapped,
- self.state.best_model_checkpoint,
- load_module_strict=not _is_peft_model(self.model),
- )
- elif self.is_fsdp_enabled:
- load_result = load_fsdp_model(
- self.accelerator.state.fsdp_plugin,
- self.accelerator,
- model,
- self.state.best_model_checkpoint,
- **_get_fsdp_ckpt_kwargs(),
- )
- elif (
- os.path.exists(best_model_path)
- or os.path.exists(best_safe_model_path)
- or os.path.exists(best_adapter_model_path)
- or os.path.exists(best_safe_adapter_model_path)
- ):
- has_been_loaded = True
- if is_sagemaker_mp_enabled():
- if os.path.isfile(os.path.join(self.state.best_model_checkpoint, "user_content.pt")):
- # If the 'user_content.pt' file exists, load with the new smp api.
- # Checkpoint must have been saved with the new smp api.
- smp.resume_from_checkpoint(
- path=self.state.best_model_checkpoint,
- tag=WEIGHTS_NAME,
- partial=False,
- load_optimizer=False,
- )
- else:
- # If the 'user_content.pt' file does NOT exist, load with the old smp api.
- # Checkpoint must have been saved with the old smp api.
- if self.args.save_safetensors and os.path.isfile(best_safe_model_path):
- state_dict = safetensors.torch.load_file(best_safe_model_path, device="cpu")
- else:
- check_torch_load_is_safe()
- state_dict = torch.load(best_model_path, map_location="cpu", weights_only=True)
- state_dict["_smp_is_partial"] = False
- load_result = model.load_state_dict(state_dict, strict=True)
- else:
- if _is_peft_model(model):
- # If train a model using PEFT & LoRA, assume that adapter have been saved properly.
- # TODO: in the future support only specific min PEFT versions
- if (hasattr(model, "active_adapter") or hasattr(model, "active_adapters")) and hasattr(
- model, "load_adapter"
- ):
- # For BC for older PEFT versions
- if hasattr(model, "active_adapters"):
- active_adapter = model.active_adapters[0]
- if len(model.active_adapters) > 1:
- logger.warning("Detected multiple active adapters, will only consider the first one")
- else:
- active_adapter = model.active_adapter
- if os.path.exists(best_adapter_model_path) or os.path.exists(best_safe_adapter_model_path):
- try:
- model.load_adapter(self.state.best_model_checkpoint, active_adapter)
- except RuntimeError as exc:
- if model.peft_config[active_adapter].is_prompt_learning:
- # for context: https://github.com/huggingface/peft/issues/2256
- msg = (
- "When using prompt learning PEFT methods such as "
- f"{model.peft_config[active_adapter].peft_type.value}, setting "
- "load_best_model_at_end=True can lead to errors, it is recommended "
- "to set this to False and to load the model manually from the checkpoint "
- "directory using PeftModel.from_pretrained(base_model, <path>) after training "
- "has finished."
- )
- raise RuntimeError(msg) from exc
- else:
- raise
- # Load_adapter has no return value present, modify it when appropriate.
- from torch.nn.modules.module import _IncompatibleKeys
- load_result = _IncompatibleKeys([], [])
- else:
- logger.warning(
- "The intermediate checkpoints of PEFT may not be saved correctly, "
- f"consider using a custom callback to save {ADAPTER_WEIGHTS_NAME} in corresponding saving folders. "
- "Check some examples here: https://github.com/huggingface/peft/issues/96"
- )
- has_been_loaded = False
- else:
- logger.warning("Could not load adapter model, make sure to have `peft>=0.3.0` installed")
- has_been_loaded = False
- else:
- # We load the model state dict on the CPU to avoid an OOM error.
- if self.args.save_safetensors and os.path.isfile(best_safe_model_path):
- state_dict = safetensors.torch.load_file(best_safe_model_path, device="cpu")
- else:
- check_torch_load_is_safe()
- state_dict = torch.load(best_model_path, map_location="cpu", weights_only=True)
- # If the model is on the GPU, it still works!
- # workaround for FSDP bug https://github.com/pytorch/pytorch/issues/82963
- # which takes *args instead of **kwargs
- load_result = model.load_state_dict(state_dict, False)
- if not is_sagemaker_mp_enabled() and has_been_loaded:
- self._issue_warnings_after_load(load_result)
- elif os.path.exists(os.path.join(self.state.best_model_checkpoint, SAFE_WEIGHTS_INDEX_NAME)) or os.path.exists(
- os.path.join(self.state.best_model_checkpoint, WEIGHTS_INDEX_NAME)
- ):
- load_result = load_sharded_checkpoint(
- model, self.state.best_model_checkpoint, strict=is_sagemaker_mp_enabled()
- )
- if not is_sagemaker_mp_enabled():
- self._issue_warnings_after_load(load_result)
- else:
- logger.warning(
- f"Could not locate the best model at {best_model_path}, if you are running a distributed training "
- "on multiple nodes, you should activate `--save_on_each_node`."
- )
- def _issue_warnings_after_load(self, load_result):
- if len(load_result.missing_keys) != 0:
- if self.model._keys_to_ignore_on_save is not None and set(load_result.missing_keys) == set(
- self.model._keys_to_ignore_on_save
- ):
- self.model.tie_weights()
- else:
- logger.warning(f"There were missing keys in the checkpoint model loaded: {load_result.missing_keys}.")
- if len(load_result.unexpected_keys) != 0:
- logger.warning(
- f"There were unexpected keys in the checkpoint model loaded: {load_result.unexpected_keys}."
- )
- def _evaluate(self, trial, ignore_keys_for_eval, skip_scheduler=False):
- metrics = self.evaluate(ignore_keys=ignore_keys_for_eval)
- self._report_to_hp_search(trial, self.state.global_step, metrics)
- # Run delayed LR scheduler now that metrics are populated
- if isinstance(self.lr_scheduler, torch.optim.lr_scheduler.ReduceLROnPlateau) and not skip_scheduler:
- metric_to_check = self.args.metric_for_best_model
- if not metric_to_check.startswith("eval_"):
- metric_to_check = f"eval_{metric_to_check}"
- try:
- self.lr_scheduler.step(metrics[metric_to_check])
- except KeyError as exc:
- raise KeyError(
- f"The `metric_for_best_model` training argument is set to '{metric_to_check}', "
- f"which is not found in the evaluation metrics. "
- f"The available evaluation metrics are: {list(metrics.keys())}. "
- f"Please ensure that the `compute_metrics` function returns a dictionary that includes '{metric_to_check}' or "
- f"consider changing the `metric_for_best_model` via the TrainingArguments."
- ) from exc
- return metrics
- def _maybe_log_save_evaluate(
- self, tr_loss, grad_norm, model, trial, epoch, ignore_keys_for_eval, start_time, learning_rate=None
- ):
- if self.control.should_log and self.state.global_step > self._globalstep_last_logged:
- if is_torch_xla_available():
- xm.mark_step()
- logs: dict[str, float] = {}
- # all_gather + mean() to get average loss over all processes
- tr_loss_scalar = self._nested_gather(tr_loss).mean().item()
- # reset tr_loss to zero
- tr_loss -= tr_loss
- logs["loss"] = round(tr_loss_scalar / (self.state.global_step - self._globalstep_last_logged), 4)
- if grad_norm is not None:
- logs["grad_norm"] = grad_norm.item() if isinstance(grad_norm, torch.Tensor) else grad_norm
- if learning_rate is not None:
- logs["learning_rate"] = learning_rate
- else:
- logs["learning_rate"] = self._get_learning_rate()
- self._total_loss_scalar += tr_loss_scalar
- self._globalstep_last_logged = self.state.global_step
- self.store_flos()
- self.log(logs, start_time)
- metrics = None
- if self.control.should_evaluate:
- metrics = self._evaluate(trial, ignore_keys_for_eval)
- is_new_best_metric = self._determine_best_metric(metrics=metrics, trial=trial)
- if self.args.save_strategy == SaveStrategy.BEST:
- self.control.should_save = is_new_best_metric
- if self.control.should_save:
- self._save_checkpoint(model, trial)
- self.control = self.callback_handler.on_save(self.args, self.state, self.control)
- def _load_rng_state(self, checkpoint):
- # Load RNG states from `checkpoint`
- if checkpoint is None:
- return
- if self.args.world_size > 1:
- process_index = self.args.process_index
- rng_file = os.path.join(checkpoint, f"rng_state_{process_index}.pth")
- if not os.path.isfile(rng_file):
- logger.info(
- f"Didn't find an RNG file for process {process_index}, if you are resuming a training that "
- "wasn't launched in a distributed fashion, reproducibility is not guaranteed."
- )
- return
- else:
- rng_file = os.path.join(checkpoint, "rng_state.pth")
- if not os.path.isfile(rng_file):
- logger.info(
- "Didn't find an RNG file, if you are resuming a training that was launched in a distributed "
- "fashion, reproducibility is not guaranteed."
- )
- return
- with safe_globals():
- checkpoint_rng_state = torch.load(rng_file)
- random.setstate(checkpoint_rng_state["python"])
- np.random.set_state(checkpoint_rng_state["numpy"])
- torch.random.set_rng_state(checkpoint_rng_state["cpu"])
- if is_torch_xla_available():
- xm.set_rng_state(checkpoint_rng_state["xla"])
- is_distributed = self.args.parallel_mode == ParallelMode.DISTRIBUTED
- if torch.cuda.is_available():
- set_rng_state_for_device("CUDA", torch.cuda, checkpoint_rng_state, is_distributed)
- if is_torch_npu_available():
- set_rng_state_for_device("NPU", torch.npu, checkpoint_rng_state, is_distributed)
- if is_torch_hpu_available():
- set_rng_state_for_device("HPU", torch.hpu, checkpoint_rng_state, is_distributed)
- if is_torch_mlu_available():
- set_rng_state_for_device("MLU", torch.mlu, checkpoint_rng_state, is_distributed)
- if is_torch_musa_available():
- set_rng_state_for_device("MUSA", torch.musa, checkpoint_rng_state, is_distributed)
- def _determine_best_metric(self, metrics, trial):
- """
- Determine if the model should be saved based on the evaluation metrics.
- Returns:
- bool: True if a new best metric was found, else False
- """
- is_new_best_metric = False
- if self.args.metric_for_best_model is not None:
- metric_to_check = self.args.metric_for_best_model
- if not metric_to_check.startswith("eval_"):
- metric_to_check = f"eval_{metric_to_check}"
- try:
- metric_value = metrics[metric_to_check]
- except KeyError as exc:
- raise KeyError(
- f"The `metric_for_best_model` training argument is set to '{metric_to_check}', which is not found in the evaluation metrics. "
- f"The available evaluation metrics are: {list(metrics.keys())}. Consider changing the `metric_for_best_model` via the TrainingArguments."
- ) from exc
- operator = np.greater if self.args.greater_is_better else np.less
- if self.state.best_metric is None:
- self.state.best_metric = float("-inf") if self.args.greater_is_better else float("inf")
- if operator(metric_value, self.state.best_metric):
- self.state.best_metric = metric_value
- if self.args.save_strategy in [SaveStrategy.STEPS, SaveStrategy.EPOCH]:
- self.state.best_global_step = self.state.global_step
- is_new_best_metric = True
- return is_new_best_metric
- def _save_checkpoint(self, model, trial):
- # In all cases, including ddp/dp/deepspeed, self.model is always a reference to the model we
- # want to save except FullyShardedDDP.
- # assert unwrap_model(model) is self.model, "internal model should be a reference to self.model"
- # Save model checkpoint
- checkpoint_folder = f"{PREFIX_CHECKPOINT_DIR}-{self.state.global_step}"
- if self.hp_search_backend is None and trial is None:
- self.store_flos()
- run_dir = self._get_output_dir(trial=trial)
- output_dir = os.path.join(run_dir, checkpoint_folder)
- self.save_model(output_dir, _internal_call=True)
- if self.args.save_strategy in [SaveStrategy.STEPS, SaveStrategy.EPOCH] and self.state.best_global_step:
- # Wait for everyone to get here so we are sure the model has been saved by process 0
- # before we check if the best_checkpoint_dir exists
- if is_torch_xla_available():
- xm.rendezvous("load_best_model_at_end")
- elif self.args.parallel_mode == ParallelMode.DISTRIBUTED:
- dist.barrier()
- elif is_sagemaker_mp_enabled():
- smp.barrier()
- best_checkpoint_folder = f"{PREFIX_CHECKPOINT_DIR}-{self.state.best_global_step}"
- best_checkpoint_dir = os.path.join(run_dir, best_checkpoint_folder)
- if os.path.exists(best_checkpoint_dir):
- self.state.best_model_checkpoint = best_checkpoint_dir
- if not self.args.save_only_model:
- # Save optimizer and scheduler
- self._save_optimizer_and_scheduler(output_dir)
- self._save_scaler(output_dir)
- # Save RNG state
- self._save_rng_state(output_dir)
- # Save the Trainer state
- if self.args.should_save:
- # Update `ExportableState` callbacks and `TrainerControl` state to where we are currently
- for cb in [
- cb for cb in self.callback_handler.callbacks + [self.control] if isinstance(cb, ExportableState)
- ]:
- cb_name = cb.__class__.__name__
- cb_state = cb.state()
- if isinstance(self.state.stateful_callbacks[cb_name], list):
- self.state.stateful_callbacks[cb_name].append(cb_state)
- else:
- self.state.stateful_callbacks[cb_name] = cb_state
- self.state.save_to_json(os.path.join(output_dir, TRAINER_STATE_NAME))
- if self.args.push_to_hub:
- self._push_from_checkpoint(output_dir)
- # Maybe delete some older checkpoints.
- if self.args.should_save:
- # we use mtime as default, filesystems without mtime support will be detected in `_sorted_checkpoints`
- self._rotate_checkpoints(use_mtime=True, output_dir=run_dir)
- def _save_rng_state(self, output_dir):
- # Save RNG state in non-distributed training
- rng_states = {
- "python": random.getstate(),
- "numpy": np.random.get_state(),
- "cpu": torch.random.get_rng_state(),
- }
- if torch.cuda.is_available():
- if self.args.parallel_mode == ParallelMode.DISTRIBUTED:
- # In non distributed, we save the global CUDA RNG state (will take care of DataParallel)
- rng_states["cuda"] = torch.cuda.random.get_rng_state_all()
- else:
- rng_states["cuda"] = torch.cuda.random.get_rng_state()
- if is_torch_xla_available():
- rng_states["xla"] = xm.get_rng_state()
- if is_torch_npu_available():
- if self.args.parallel_mode == ParallelMode.DISTRIBUTED:
- rng_states["npu"] = torch.npu.random.get_rng_state_all()
- else:
- rng_states["npu"] = torch.npu.random.get_rng_state()
- if is_torch_hpu_available():
- if self.args.parallel_mode == ParallelMode.DISTRIBUTED:
- rng_states["hpu"] = torch.hpu.random.get_rng_state_all()
- else:
- rng_states["hpu"] = torch.hpu.random.get_rng_state()
- if is_torch_mlu_available():
- if self.args.parallel_mode == ParallelMode.DISTRIBUTED:
- rng_states["mlu"] = torch.mlu.random.get_rng_state_all()
- else:
- rng_states["mlu"] = torch.mlu.random.get_rng_state()
- if is_torch_musa_available():
- if self.args.parallel_mode == ParallelMode.DISTRIBUTED:
- rng_states["musa"] = torch.musa.get_rng_state_all()
- else:
- rng_states["musa"] = torch.musa.get_rng_state()
- # A process can arrive here before the process 0 has a chance to save the model, in which case output_dir may
- # not yet exist.
- os.makedirs(output_dir, exist_ok=True)
- if self.args.world_size <= 1:
- torch.save(rng_states, os.path.join(output_dir, "rng_state.pth"))
- else:
- torch.save(rng_states, os.path.join(output_dir, f"rng_state_{self.args.process_index}.pth"))
- def _save_optimizer_and_scheduler(self, output_dir):
- if is_torch_xla_available():
- xm.rendezvous("saving_optimizer_states")
- if self.is_fsdp_xla_v1_enabled:
- optm = {
- "optimizer": self.optimizer.state_dict(),
- "shard_metadata": self.model.get_shard_metadata(),
- }
- xm.save(
- optm,
- os.path.join(
- output_dir, f"rank{self.args.process_index}-of-{self.args.world_size}-{OPTIMIZER_NAME}"
- ),
- master_only=False,
- )
- else:
- xm.save(self.optimizer.state_dict(), os.path.join(output_dir, OPTIMIZER_NAME))
- with warnings.catch_warnings(record=True) as caught_warnings:
- xm.save(self.lr_scheduler.state_dict(), os.path.join(output_dir, SCHEDULER_NAME))
- reissue_pt_warnings(caught_warnings)
- elif is_sagemaker_mp_enabled():
- opt_state_dict = self.optimizer.local_state_dict(gather_if_shard=False)
- smp.barrier()
- if smp.rdp_rank() == 0 or smp.state.cfg.shard_optimizer_state:
- smp.save(
- opt_state_dict,
- os.path.join(output_dir, OPTIMIZER_NAME),
- partial=True,
- v3=smp.state.cfg.shard_optimizer_state,
- )
- elif self.is_deepspeed_enabled:
- # under zero3 model file itself doesn't get saved since it's bogus! Unless deepspeed
- # config `stage3_gather_16bit_weights_on_model_save` is True
- accept_exclude_frozen_parameters = "exclude_frozen_parameters" in set(
- inspect.signature(self.model_wrapped.save_checkpoint).parameters.keys()
- )
- if accept_exclude_frozen_parameters and _is_peft_model(self.model):
- self.model_wrapped.save_checkpoint(output_dir, exclude_frozen_parameters=True)
- else:
- self.model_wrapped.save_checkpoint(output_dir)
- elif self.is_fsdp_enabled:
- # save fsdp specific ckpt for resuming from ckpt
- save_fsdp_model(
- self.accelerator.state.fsdp_plugin, self.accelerator, self.model, output_dir, **_get_fsdp_ckpt_kwargs()
- )
- save_fsdp_optimizer(
- self.accelerator.state.fsdp_plugin, self.accelerator, self.optimizer, self.model, output_dir
- )
- elif self.args.should_save:
- # deepspeed.save_checkpoint above saves model/optim/sched
- torch.save(self.optimizer.state_dict(), os.path.join(output_dir, OPTIMIZER_NAME))
- # Save SCHEDULER & SCALER
- is_deepspeed_custom_scheduler = self.is_deepspeed_enabled and not isinstance(
- self.lr_scheduler, DeepSpeedSchedulerWrapper
- )
- if (
- self.args.should_save
- and (not self.is_deepspeed_enabled or is_deepspeed_custom_scheduler)
- and not is_torch_xla_available()
- ):
- with warnings.catch_warnings(record=True) as caught_warnings:
- torch.save(self.lr_scheduler.state_dict(), os.path.join(output_dir, SCHEDULER_NAME))
- reissue_pt_warnings(caught_warnings)
- def _load_optimizer_and_scheduler(self, checkpoint):
- """If optimizer and scheduler states exist, load them."""
- if checkpoint is None:
- return
- if self.is_deepspeed_enabled:
- # deepspeed loads optimizer/lr_scheduler together with the model in deepspeed_init
- if not isinstance(self.lr_scheduler, DeepSpeedSchedulerWrapper):
- with warnings.catch_warnings(record=True) as caught_warnings:
- check_torch_load_is_safe()
- self.lr_scheduler.load_state_dict(
- torch.load(os.path.join(checkpoint, SCHEDULER_NAME), weights_only=True)
- )
- reissue_pt_warnings(caught_warnings)
- return
- checkpoint_file_exists = (
- glob.glob(os.path.join(checkpoint, OPTIMIZER_NAME) + "_*")
- if is_sagemaker_mp_enabled()
- else (
- os.path.isfile(os.path.join(checkpoint, OPTIMIZER_NAME))
- or os.path.isfile(os.path.join(checkpoint, OPTIMIZER_NAME_BIN))
- or (
- os.path.isdir(checkpoint)
- and any(
- OPTIMIZER_NAME_BIN.split(".")[0] in folder_name
- for folder_name in os.listdir(checkpoint)
- if os.path.isdir(os.path.join(checkpoint, folder_name))
- )
- )
- )
- )
- checkpoint_file_exists = (
- glob.glob(os.path.join(checkpoint, f"rank*-of-{self.args.world_size}-{OPTIMIZER_NAME}"))
- if self.is_fsdp_xla_v1_enabled
- else checkpoint_file_exists
- )
- if checkpoint_file_exists and os.path.isfile(os.path.join(checkpoint, SCHEDULER_NAME)):
- # Load in optimizer and scheduler states
- if is_torch_xla_available():
- # On TPU we have to take some extra precautions to properly load the states on the right device.
- if self.is_fsdp_xla_v1_enabled:
- check_torch_load_is_safe()
- optimizer_state = torch.load(
- os.path.join(
- checkpoint, f"rank{self.args.process_index}-of-{self.args.world_size}-{OPTIMIZER_NAME}"
- ),
- map_location="cpu",
- weights_only=True,
- )
- # We only need `optimizer` when resuming from checkpoint
- optimizer_state = optimizer_state["optimizer"]
- else:
- check_torch_load_is_safe()
- optimizer_state = torch.load(
- os.path.join(checkpoint, OPTIMIZER_NAME), map_location="cpu", weights_only=True
- )
- with warnings.catch_warnings(record=True) as caught_warnings:
- check_torch_load_is_safe()
- lr_scheduler_state = torch.load(
- os.path.join(checkpoint, SCHEDULER_NAME), map_location="cpu", weights_only=True
- )
- reissue_pt_warnings(caught_warnings)
- xm.send_cpu_data_to_device(optimizer_state, self.args.device)
- xm.send_cpu_data_to_device(lr_scheduler_state, self.args.device)
- self.optimizer.load_state_dict(optimizer_state)
- self.lr_scheduler.load_state_dict(lr_scheduler_state)
- else:
- if is_sagemaker_mp_enabled():
- if os.path.isfile(os.path.join(checkpoint, "user_content.pt")):
- # Optimizer checkpoint was saved with smp >= 1.10
- def opt_load_hook(mod, opt):
- opt.load_state_dict(smp.load(os.path.join(checkpoint, OPTIMIZER_NAME), partial=True))
- else:
- # Optimizer checkpoint was saved with smp < 1.10
- def opt_load_hook(mod, opt):
- if IS_SAGEMAKER_MP_POST_1_10:
- opt.load_state_dict(
- smp.load(os.path.join(checkpoint, OPTIMIZER_NAME), partial=True, back_compat=True)
- )
- else:
- opt.load_state_dict(smp.load(os.path.join(checkpoint, OPTIMIZER_NAME), partial=True))
- self.model_wrapped.register_post_step_hook(opt_load_hook)
- else:
- # We use the CPU when training on one GPU to avoid OOM for GPU RAM when training big models.
- # In distributed training however, we load directly on each GPU and risk the GPU OOM as it's more
- # likely to get OOM on CPU (since we load num_gpu times the optimizer state
- map_location = self.args.device if self.args.world_size > 1 else "cpu"
- if self.is_fsdp_enabled:
- load_fsdp_optimizer(
- self.accelerator.state.fsdp_plugin,
- self.accelerator,
- self.optimizer,
- self.model,
- checkpoint,
- **_get_fsdp_ckpt_kwargs(),
- )
- else:
- check_torch_load_is_safe()
- self.optimizer.load_state_dict(
- torch.load(
- os.path.join(checkpoint, OPTIMIZER_NAME), map_location=map_location, weights_only=True
- )
- )
- with warnings.catch_warnings(record=True) as caught_warnings:
- check_torch_load_is_safe()
- self.lr_scheduler.load_state_dict(
- torch.load(os.path.join(checkpoint, SCHEDULER_NAME), weights_only=True)
- )
- reissue_pt_warnings(caught_warnings)
- def _save_scaler(self, output_dir):
- # See if there is a scaler attribute
- try:
- scaler = self.accelerator.scaler
- except AttributeError:
- return
- if scaler is None:
- return
- if is_torch_xla_available():
- xm.rendezvous("saving_scaler_state")
- with warnings.catch_warnings(record=True) as caught_warnings:
- xm.save(self.accelerator.scaler.state_dict(), os.path.join(output_dir, SCALER_NAME))
- reissue_pt_warnings(caught_warnings)
- # Save SCALER
- if self.args.should_save and not is_torch_xla_available():
- with warnings.catch_warnings(record=True) as caught_warnings:
- torch.save(self.accelerator.scaler.state_dict(), os.path.join(output_dir, SCALER_NAME))
- reissue_pt_warnings(caught_warnings)
- def _load_scaler(self, checkpoint):
- """If scaler state exists, load it."""
- if checkpoint is None:
- return
- checkpoint_file_exists = os.path.isfile(os.path.join(checkpoint, SCALER_NAME))
- if checkpoint_file_exists:
- # On TPU we have to take some extra precautions to properly load the states on the right device.
- # Load in scaler states
- if is_torch_xla_available():
- with warnings.catch_warnings(record=True) as caught_warnings:
- check_torch_load_is_safe()
- scaler_state = torch.load(
- os.path.join(checkpoint, SCALER_NAME), map_location="cpu", weights_only=True
- )
- reissue_pt_warnings(caught_warnings)
- xm.send_cpu_data_to_device(scaler_state, self.args.device)
- self.accelerator.scaler.load_state_dict(scaler_state)
- else:
- with warnings.catch_warnings(record=True) as caught_warnings:
- check_torch_load_is_safe()
- self.accelerator.scaler.load_state_dict(
- torch.load(os.path.join(checkpoint, SCALER_NAME), weights_only=True)
- )
- reissue_pt_warnings(caught_warnings)
- def _load_callback_state(self):
- """If callback states exist and were passed in, restore their states if enabled"""
- if not self.args.restore_callback_states_from_checkpoint:
- return
- # Callback states are stored in stateful_callbacks
- not_found = []
- new_callbacks = []
- original_callbacks = self.callback_handler.callbacks + [self.control]
- for stored_callback, data in self.state.stateful_callbacks.items():
- if not isinstance(data, list):
- data = [data]
- if any(callback.__class__.__name__ == stored_callback for callback in original_callbacks):
- # We can load/restore from multiple callbacks of the same type.
- duplicates = [
- callback for callback in original_callbacks if callback.__class__.__name__ == stored_callback
- ]
- for callback, callback_data in zip(duplicates, data):
- args = callback_data.get("args", {})
- attributes = callback_data.get("attributes", {})
- new_callback = type(callback)(**args)
- for attribute, value in attributes.items():
- setattr(new_callback, attribute, value)
- if isinstance(callback, TrainerControl):
- # Specifically for restoring the `control` state
- self.control = new_callback
- else:
- new_callbacks.append(new_callback)
- # We remove the existing callback and add it to the list of new callbacks
- self.callback_handler.remove_callback(type(new_callback))
- logger.info("Continuing training from checkpoint, restoring any callbacks that were passed in")
- else:
- not_found.append(stored_callback)
- if len(not_found) > 0:
- logger.warning(
- f"Checkpoint included callbacks not included in current configuration. Ignoring. ({', '.join(not_found)})"
- )
- for callback in new_callbacks:
- self.callback_handler.add_callback(callback)
- def hyperparameter_search(
- self,
- hp_space: Optional[Callable[["optuna.Trial"], dict[str, float]]] = None,
- compute_objective: Optional[Callable[[dict[str, float]], float]] = None,
- n_trials: int = 20,
- direction: Union[str, list[str]] = "minimize",
- backend: Optional[Union["str", HPSearchBackend]] = None,
- hp_name: Optional[Callable[["optuna.Trial"], str]] = None,
- **kwargs,
- ) -> Union[BestRun, list[BestRun]]:
- """
- Launch an hyperparameter search using `optuna` or `Ray Tune` or `SigOpt`. The optimized quantity is determined
- by `compute_objective`, which defaults to a function returning the evaluation loss when no metric is provided,
- the sum of all metrics otherwise.
- <Tip warning={true}>
- To use this method, you need to have provided a `model_init` when initializing your [`Trainer`]: we need to
- reinitialize the model at each new run. This is incompatible with the `optimizers` argument, so you need to
- subclass [`Trainer`] and override the method [`~Trainer.create_optimizer_and_scheduler`] for custom
- optimizer/scheduler.
- </Tip>
- Args:
- hp_space (`Callable[["optuna.Trial"], dict[str, float]]`, *optional*):
- A function that defines the hyperparameter search space. Will default to
- [`~trainer_utils.default_hp_space_optuna`] or [`~trainer_utils.default_hp_space_ray`] or
- [`~trainer_utils.default_hp_space_sigopt`] depending on your backend.
- compute_objective (`Callable[[dict[str, float]], float]`, *optional*):
- A function computing the objective to minimize or maximize from the metrics returned by the `evaluate`
- method. Will default to [`~trainer_utils.default_compute_objective`].
- n_trials (`int`, *optional*, defaults to 100):
- The number of trial runs to test.
- direction (`str` or `list[str]`, *optional*, defaults to `"minimize"`):
- If it's single objective optimization, direction is `str`, can be `"minimize"` or `"maximize"`, you
- should pick `"minimize"` when optimizing the validation loss, `"maximize"` when optimizing one or
- several metrics. If it's multi objectives optimization, direction is `list[str]`, can be List of
- `"minimize"` and `"maximize"`, you should pick `"minimize"` when optimizing the validation loss,
- `"maximize"` when optimizing one or several metrics.
- backend (`str` or [`~training_utils.HPSearchBackend`], *optional*):
- The backend to use for hyperparameter search. Will default to optuna or Ray Tune or SigOpt, depending
- on which one is installed. If all are installed, will default to optuna.
- hp_name (`Callable[["optuna.Trial"], str]]`, *optional*):
- A function that defines the trial/run name. Will default to None.
- kwargs (`dict[str, Any]`, *optional*):
- Additional keyword arguments for each backend:
- - `optuna`: parameters from
- [optuna.study.create_study](https://optuna.readthedocs.io/en/stable/reference/generated/optuna.study.create_study.html)
- and also the parameters `timeout`, `n_jobs` and `gc_after_trial` from
- [optuna.study.Study.optimize](https://optuna.readthedocs.io/en/stable/reference/generated/optuna.study.Study.html#optuna.study.Study.optimize)
- - `ray`: parameters from [tune.run](https://docs.ray.io/en/latest/tune/api_docs/execution.html#tune-run).
- If `resources_per_trial` is not set in the `kwargs`, it defaults to 1 CPU core and 1 GPU (if available).
- If `progress_reporter` is not set in the `kwargs`,
- [ray.tune.CLIReporter](https://docs.ray.io/en/latest/tune/api/doc/ray.tune.CLIReporter.html) is used.
- - `sigopt`: the parameter `proxies` from
- [sigopt.Connection.set_proxies](https://docs.sigopt.com/support/faq#how-do-i-use-sigopt-with-a-proxy).
- Returns:
- [`trainer_utils.BestRun` or `list[trainer_utils.BestRun]`]: All the information about the best run or best
- runs for multi-objective optimization. Experiment summary can be found in `run_summary` attribute for Ray
- backend.
- """
- if backend is None:
- backend = default_hp_search_backend()
- backend = HPSearchBackend(backend)
- backend_obj = ALL_HYPERPARAMETER_SEARCH_BACKENDS[backend]()
- backend_obj.ensure_available()
- self.hp_search_backend = backend
- if self.model_init is None:
- raise RuntimeError(
- "To use hyperparameter search, you need to pass your model through a model_init function."
- )
- self.hp_space = backend_obj.default_hp_space if hp_space is None else hp_space
- self.hp_name = hp_name
- self.compute_objective = default_compute_objective if compute_objective is None else compute_objective
- best_run = backend_obj.run(self, n_trials, direction, **kwargs)
- self.hp_search_backend = None
- return best_run
- def log(self, logs: dict[str, float], start_time: Optional[float] = None) -> None:
- """
- Log `logs` on the various objects watching training.
- Subclass and override this method to inject custom behavior.
- Args:
- logs (`dict[str, float]`):
- The values to log.
- start_time (`Optional[float]`):
- The start of training.
- """
- if self.state.epoch is not None:
- logs["epoch"] = self.state.epoch
- if self.args.include_num_input_tokens_seen != "no":
- logs["num_input_tokens_seen"] = self.state.num_input_tokens_seen
- if start_time is not None:
- logs.update(speed_metrics("train", start_time, num_tokens=self.state.num_input_tokens_seen))
- output = {**logs, **{"step": self.state.global_step}}
- self.state.log_history.append(output)
- self.control = self.callback_handler.on_log(self.args, self.state, self.control, logs)
- def _prepare_input(self, data: Union[torch.Tensor, Any]) -> Union[torch.Tensor, Any]:
- """
- Prepares one `data` before feeding it to the model, be it a tensor or a nested list/dictionary of tensors.
- """
- if isinstance(data, Mapping):
- return type(data)({k: self._prepare_input(v) for k, v in data.items()})
- elif isinstance(data, (tuple, list)):
- return type(data)(self._prepare_input(v) for v in data)
- elif isinstance(data, torch.Tensor):
- kwargs = {"device": self.args.device}
- if self.is_deepspeed_enabled and (torch.is_floating_point(data) or torch.is_complex(data)):
- # NLP models inputs are int/uint and those get adjusted to the right dtype of the
- # embedding. Other models such as wav2vec2's inputs are already float and thus
- # may need special handling to match the dtypes of the model
- kwargs.update({"dtype": self.accelerator.state.deepspeed_plugin.hf_ds_config.dtype()})
- return data.to(**kwargs)
- return data
- def _prepare_inputs(self, inputs: dict[str, Union[torch.Tensor, Any]]) -> dict[str, Union[torch.Tensor, Any]]:
- """
- Prepare `inputs` before feeding them to the model, converting them to tensors if they are not already and
- handling potential state.
- """
- inputs = self._prepare_input(inputs)
- if len(inputs) == 0:
- raise ValueError(
- "The batch received was empty, your model won't be able to train on it. Double-check that your "
- f"training dataset contains keys expected by the model: {','.join(self._signature_columns)}."
- )
- if self.args.past_index >= 0 and self._past is not None:
- inputs["mems"] = self._past
- return inputs
- def _is_attention_mask_causal(self, attention_mask):
- """
- Check if an attention mask is causal (compatible with causal attention).
- Context parallelism only supports causal attention patterns. This function
- checks if the provided attention mask is compatible.
- Args:
- attention_mask (torch.Tensor): The attention mask to check
- Returns:
- bool: True if the mask is causal or compatible with causal attention
- """
- if attention_mask is None:
- return True # No mask is considered causal (model uses default causal masking)
- # Handle different mask dimensions
- if attention_mask.dim() == 2:
- # (batch_size, seq_len) - standard padding mask, compatible with causal attention
- return True
- elif attention_mask.dim() in [3, 4]:
- # (batch_size, seq_len, seq_len) or (batch_size, num_heads, seq_len, seq_len)
- # Check if it's lower triangular (causal)
- seq_len = attention_mask.shape[-1]
- if seq_len <= 1:
- return True # Single token or empty is always causal
- # Take first batch and head (if 4D) for checking pattern
- if attention_mask.dim() == 4:
- mask = attention_mask[0, 0] # First batch, first head
- else:
- mask = attention_mask[0] # First batch
- # Check if upper triangular part is masked (should be 0 or very negative for causal)
- upper_triangular = torch.triu(mask, diagonal=1)
- # For causal masks, upper triangular should be 0 or very negative (like -inf)
- # Use a reasonable threshold to handle float precision issues
- is_causal = torch.all(upper_triangular <= 1e-6) or torch.all(upper_triangular < -1e4)
- return is_causal.item() if isinstance(is_causal, torch.Tensor) else is_causal
- # For unknown dimensions, be conservative and reject
- return False
- def _prepare_context_parallel_inputs(self, model, inputs: dict[str, Union[torch.Tensor, Any]]):
- """
- Prepare inputs for context parallelism by setting up buffers and validation.
- Args:
- model: The model being trained
- inputs: Input tensors to prepare
- Returns:
- tuple: (context_manager, prepared_inputs) where context_manager is either
- the context parallelism wrapper or a no-op context
- """
- if (
- getattr(self.accelerator, "parallelism_config", None) is not None
- and self.accelerator.parallelism_config.cp_enabled
- ):
- if hasattr(model, "config"):
- if model.config._attn_implementation != "sdpa":
- raise ValueError(
- f"Context parallelism is supported only with SDPA attention, you are using {model.config._attn_implementation}."
- )
- if "position_ids" not in inputs:
- logger.warning_once("Position IDs not found in the inputs, generating manually")
- inputs["position_ids"] = torch.arange(
- inputs["input_ids"].size(1), device=inputs["input_ids"].device
- ).expand(inputs["input_ids"].size(0), -1)
- if "shift_labels" not in inputs:
- logger.warning_once("Shift labels not found in the inputs, shifting manually")
- if "labels" in inputs:
- _ignore_index = -100
- labels = nn.functional.pad(inputs["labels"], (0, 1), value=_ignore_index)
- inputs["shift_labels"] = labels[:, 1:].contiguous()
- buffers = []
- buffer_seq_dims = []
- if "input_ids" in inputs:
- buffers.append(inputs["input_ids"])
- buffer_seq_dims.append(1) # Sequence dimension
- if "labels" in inputs:
- buffers.append(inputs["labels"])
- buffer_seq_dims.append(1)
- if "shift_labels" in inputs:
- buffers.append(inputs["shift_labels"])
- buffer_seq_dims.append(1)
- # Add attention_mask to buffers for context parallel splitting (only if causal)
- if "attention_mask" in inputs:
- # Only validate causal mask once for performance
- if not getattr(self, "_attn_mask_causal_checked", False):
- # Context parallel currently doesn't support other masks than causal
- # Accelerate applies hooks to replace mask with is_causal arg in SDPA
- # Check if the mask is really causal and if not throw an error
- attention_mask = inputs["attention_mask"]
- if not self._is_attention_mask_causal(attention_mask):
- raise ValueError(
- "Context parallelism only supports causal attention masks. "
- "The provided attention_mask is not causal. "
- "Please ensure your data uses causal masking (lower triangular) "
- "or remove the attention_mask to use the model's default causal masking."
- )
- self._attn_mask_causal_checked = True
- if self._attn_mask_causal_checked:
- # Add to buffers only after validation (or if validation already passed)
- attention_mask = inputs["attention_mask"]
- if attention_mask.dim() == 2:
- buffers.append(attention_mask)
- buffer_seq_dims.append(1)
- else:
- # Other dimensionality; keep as-is without sharding to avoid incorrect splits
- pass
- # Include position_ids in context parallelism splitting
- if "position_ids" in inputs and inputs["position_ids"] is not None:
- buffers.append(inputs["position_ids"])
- buffer_seq_dims.append(1)
- return partial(
- self.accelerator.maybe_context_parallel,
- buffers=buffers,
- buffer_seq_dims=buffer_seq_dims,
- no_restore_buffers=set(buffers),
- ), inputs
- return contextlib.nullcontext, inputs
- def compute_loss_context_manager(self):
- """
- A helper wrapper to group together context managers.
- """
- ctx_stack = contextlib.ExitStack()
- autocast_ctx = self.autocast_smart_context_manager()
- if not isinstance(autocast_ctx, contextlib.nullcontext):
- ctx_stack.enter_context(autocast_ctx)
- return ctx_stack
- def autocast_smart_context_manager(self, cache_enabled: Optional[bool] = True):
- """
- A helper wrapper that creates an appropriate context manager for `autocast` while feeding it the desired
- arguments, depending on the situation.
- """
- if self.use_cpu_amp:
- ctx_manager = torch.autocast(device_type="cpu", cache_enabled=cache_enabled, dtype=self.amp_dtype)
- else:
- ctx_manager = contextlib.nullcontext()
- return ctx_manager
- def training_step(
- self,
- model: nn.Module,
- inputs: dict[str, Union[torch.Tensor, Any]],
- num_items_in_batch: Optional[torch.Tensor] = None,
- ) -> torch.Tensor:
- """
- Perform a training step on a batch of inputs.
- Subclass and override to inject custom behavior.
- Args:
- model (`nn.Module`):
- The model to train.
- inputs (`dict[str, Union[torch.Tensor, Any]]`):
- The inputs and targets of the model.
- The dictionary will be unpacked before being fed to the model. Most models expect the targets under the
- argument `labels`. Check your model's documentation for all accepted arguments.
- Return:
- `torch.Tensor`: The tensor with training loss on this batch.
- """
- # Prepare buffers for context parallelism
- cp_context, inputs = self._prepare_context_parallel_inputs(model, inputs)
- # Context manager is no-op if CP isn't enabled
- with cp_context():
- model.train()
- if hasattr(self.optimizer, "train") and callable(self.optimizer.train):
- self.optimizer.train()
- inputs = self._prepare_inputs(inputs)
- if is_sagemaker_mp_enabled():
- loss_mb = smp_forward_backward(model, inputs, self.args.gradient_accumulation_steps)
- return loss_mb.reduce_mean().detach().to(self.args.device)
- with self.compute_loss_context_manager():
- loss = self.compute_loss(model, inputs, num_items_in_batch=num_items_in_batch)
- del inputs
- if (
- self.args.torch_empty_cache_steps is not None
- and self.state.global_step % self.args.torch_empty_cache_steps == 0
- ):
- if is_torch_xpu_available():
- torch.xpu.empty_cache()
- elif is_torch_mlu_available():
- torch.mlu.empty_cache()
- elif is_torch_musa_available():
- torch.musa.empty_cache()
- elif is_torch_npu_available():
- torch.npu.empty_cache()
- elif is_torch_mps_available():
- torch.mps.empty_cache()
- elif is_torch_hpu_available():
- logger.warning(
- "`torch_empty_cache_steps` is set but HPU device/backend does not support empty_cache()."
- )
- else:
- torch.cuda.empty_cache()
- kwargs = {}
- # For LOMO optimizers you need to explicitly use the learning rate
- if self.args.optim in [OptimizerNames.LOMO, OptimizerNames.ADALOMO]:
- kwargs["learning_rate"] = self._get_learning_rate()
- if self.args.n_gpu > 1:
- loss = loss.mean() # mean() to average on multi-gpu parallel training
- if self.use_apex:
- from apex import amp
- with amp.scale_loss(loss, self.optimizer) as scaled_loss:
- scaled_loss.backward()
- else:
- # Finally we need to normalize the loss for reporting if GA loss bug is not fixed during compute loss
- if (
- not self.model_accepts_loss_kwargs or num_items_in_batch is None
- ) and self.compute_loss_func is None:
- # If the model does not accept loss kwargs, we need to normalize the loss by the number of gradient accumulation steps
- loss = loss / self.current_gradient_accumulation_steps
- # Turning off loss scaling w.r.t. gradient accumulation when DeepSpeed is enabled
- # https://github.com/huggingface/transformers/pull/35808
- if self.accelerator.distributed_type == DistributedType.DEEPSPEED:
- kwargs["scale_wrt_gas"] = False
- self.accelerator.backward(loss, **kwargs)
- return loss.detach()
- def compute_loss(
- self,
- model: nn.Module,
- inputs: dict[str, Union[torch.Tensor, Any]],
- return_outputs: bool = False,
- num_items_in_batch: Optional[torch.Tensor] = None,
- ):
- """
- How the loss is computed by Trainer. By default, all models return the loss in the first element.
- Args:
- model (`nn.Module`):
- The model to compute the loss for.
- inputs (`dict[str, Union[torch.Tensor, Any]]`):
- The input data for the model.
- return_outputs (`bool`, *optional*, defaults to `False`):
- Whether to return the model outputs along with the loss.
- num_items_in_batch (Optional[torch.Tensor], *optional*):
- The number of items in the batch. If num_items_in_batch is not passed,
- Returns:
- The loss of the model along with its output if return_outputs was set to True
- Subclass and override for custom behavior. If you are not using `num_items_in_batch` when computing your loss,
- make sure to overwrite `self.model_accepts_loss_kwargs` to `False`. Otherwise, the loss calculating might be slightly inaccurate when performing gradient accumulation.
- """
- if (self.label_smoother is not None or self.compute_loss_func is not None) and "labels" in inputs:
- labels = inputs.pop("labels")
- else:
- labels = None
- if self.model_accepts_loss_kwargs:
- kwargs = {}
- if num_items_in_batch is not None:
- kwargs["num_items_in_batch"] = num_items_in_batch
- inputs = {**inputs, **kwargs}
- outputs = model(**inputs)
- # Save past state if it exists
- # TODO: this needs to be fixed and made cleaner later.
- if self.args.past_index >= 0:
- self._past = outputs[self.args.past_index]
- # User-defined compute_loss function
- if self.compute_loss_func is not None:
- if labels is None:
- logger.warning(
- "Trainer: `compute_loss_func` is defined but `labels=None`. "
- "Your custom loss function will still be called with labels=None. "
- )
- loss = self.compute_loss_func(
- outputs,
- labels,
- num_items_in_batch=num_items_in_batch,
- )
- # Default HF loss handling (label smoothing) if no custom loss function
- elif labels is not None:
- unwrapped_model = self.accelerator.unwrap_model(model)
- model_name = (
- unwrapped_model.base_model.model._get_name()
- if _is_peft_model(unwrapped_model)
- else unwrapped_model._get_name()
- )
- if model_name in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES.values():
- loss = self.label_smoother(outputs, labels, shift_labels=True)
- else:
- loss = self.label_smoother(outputs, labels)
- else:
- if isinstance(outputs, dict) and "loss" not in outputs:
- raise ValueError(
- "The model did not return a loss from the inputs, only the following keys: "
- f"{','.join(outputs.keys())}. For reference, the inputs it received are {','.join(inputs.keys())}."
- )
- # We don't use .loss here since the model may return tuples instead of ModelOutput.
- loss = outputs["loss"] if isinstance(outputs, dict) else outputs[0]
- if (
- self.args.average_tokens_across_devices
- and (self.model_accepts_loss_kwargs or self.compute_loss_func)
- and num_items_in_batch is not None
- ):
- loss *= self.accelerator.num_processes if self.args.n_gpu <= 1 else self.args.n_gpu
- return (loss, outputs) if return_outputs else loss
- def is_local_process_zero(self) -> bool:
- """
- Whether or not this process is the local (e.g., on one machine if training in a distributed fashion on several
- machines) main process.
- """
- return self.args.local_process_index == 0
- def is_world_process_zero(self) -> bool:
- """
- Whether or not this process is the global main process (when training in a distributed fashion on several
- machines, this is only going to be `True` for one process).
- """
- # Special case for SageMaker ModelParallel since there process_index is dp_process_index, not the global
- # process index.
- if is_sagemaker_mp_enabled():
- return smp.rank() == 0
- else:
- return self.args.process_index == 0
- def save_model(self, output_dir: Optional[str] = None, _internal_call: bool = False):
- """
- Will save the model, so you can reload it using `from_pretrained()`.
- Will only save from the main process.
- """
- if output_dir is None:
- output_dir = self.args.output_dir
- if is_torch_xla_available():
- self._save_tpu(output_dir)
- elif is_sagemaker_mp_enabled():
- # Calling the state_dict needs to be done on the wrapped model and on all processes.
- os.makedirs(output_dir, exist_ok=True)
- state_dict = self.model_wrapped.state_dict()
- if self.args.should_save:
- self._save(output_dir, state_dict=state_dict)
- if IS_SAGEMAKER_MP_POST_1_10:
- # 'user_content.pt' indicates model state_dict saved with smp >= 1.10
- Path(os.path.join(output_dir, "user_content.pt")).touch()
- # We are in N-D parallelism if we have parallelism_config set, so we check accelerate if we're on a to_save rank
- elif getattr(self.accelerator, "parallelism_config", None) is not None:
- if self.accelerator.should_save_model:
- self._save(output_dir)
- # If we drop to here, we're in 1D parallelism, so all ranks need to go to `save_pretrained`
- elif (tp_size := getattr(self.model, "_tp_size", 0)) is not None and tp_size > 1:
- self._save(output_dir)
- elif self.is_fsdp_enabled:
- if "FULL_STATE_DICT" in str(self.accelerator.state.fsdp_plugin.state_dict_type):
- state_dict = self.accelerator.get_state_dict(self.model)
- if self.args.should_save:
- self._save(output_dir, state_dict=state_dict)
- elif self.is_deepspeed_enabled:
- try:
- state_dict = self.accelerator.get_state_dict(self.deepspeed)
- if self.args.should_save:
- self._save(output_dir, state_dict=state_dict)
- except ValueError:
- logger.warning(
- " stage3_gather_16bit_weights_on_model_save=false. Saving the full checkpoint instead, use"
- " zero_to_fp32.py to recover weights"
- )
- if self.args.should_save:
- self._save(output_dir, state_dict={})
- # remove the dummy state_dict
- remove_dummy_checkpoint(self.args.should_save, output_dir, [WEIGHTS_NAME, SAFE_WEIGHTS_NAME])
- self.model_wrapped.save_checkpoint(output_dir)
- elif self.args.should_save:
- self._save(output_dir)
- # Push to the Hub when `save_model` is called by the user.
- if self.args.push_to_hub and not _internal_call:
- self.push_to_hub(commit_message="Model save", revision=self.args.hub_revision)
- def _save_tpu(self, output_dir: Optional[str] = None):
- output_dir = output_dir if output_dir is not None else self.args.output_dir
- logger.info(f"Saving model checkpoint to {output_dir}")
- model = self.model
- xm.mark_step()
- if xm.is_master_ordinal(local=False):
- os.makedirs(output_dir, exist_ok=True)
- torch.save(self.args, os.path.join(output_dir, TRAINING_ARGS_NAME))
- # Save a trained model and configuration using `save_pretrained()`.
- # They can then be reloaded using `from_pretrained()`
- supported_classes = (PushToHubMixin,)
- xm.rendezvous("saving_checkpoint")
- if self.is_fsdp_xla_v1_enabled:
- ckpt = {
- "model": model.state_dict(),
- "shard_metadata": model.get_shard_metadata(),
- }
- ckpt_path = os.path.join(
- output_dir, f"rank{self.args.process_index}-of-{self.args.world_size}-{WEIGHTS_NAME}"
- )
- # All ranks save sharded checkpoint
- xm.save(ckpt, ckpt_path, master_only=False)
- # Make sure all ranks have saved checkpoints
- xm.rendezvous("save_full_checkpoints")
- # Master save full checkpoint
- if self.args.should_save:
- from torch_xla.distributed.fsdp import consolidate_sharded_model_checkpoints
- full_state_dict, _ = consolidate_sharded_model_checkpoints(
- ckpt_prefix=os.path.join(output_dir, ""),
- ckpt_suffix=f"rank*-of-*-{WEIGHTS_NAME}",
- save_model=False,
- )
- model = model.module.module
- unwrapped_model = self.accelerator.unwrap_model(model)
- if isinstance(unwrapped_model, supported_classes):
- unwrapped_model.save_pretrained(
- output_dir,
- state_dict=full_state_dict,
- save_function=xm.save,
- safe_serialization=self.args.save_safetensors,
- )
- else:
- logger.info("Trainer.model is not a `PreTrainedModel`, only saving its state dict.")
- xm.save(full_state_dict, os.path.join(output_dir, WEIGHTS_NAME))
- elif not isinstance(model, supported_classes):
- if isinstance(self.accelerator.unwrap_model(model), supported_classes):
- self.accelerator.unwrap_model(model).save_pretrained(
- output_dir,
- is_main_process=self.args.should_save,
- state_dict=xm._maybe_convert_to_cpu(model.state_dict()),
- save_function=xm.save,
- safe_serialization=self.args.save_safetensors,
- )
- else:
- logger.info("Trainer.model is not a `PreTrainedModel`, only saving its state dict.")
- state_dict = xm._maybe_convert_to_cpu(model.state_dict())
- xm.save(state_dict, os.path.join(output_dir, WEIGHTS_NAME))
- else:
- model.save_pretrained(
- output_dir,
- is_main_process=self.args.should_save,
- save_function=xm.save,
- safe_serialization=self.args.save_safetensors,
- state_dict=xm._maybe_convert_to_cpu(model.state_dict()),
- )
- if self.processing_class is not None and self.args.should_save:
- self.processing_class.save_pretrained(output_dir)
- def _save(self, output_dir: Optional[str] = None, state_dict=None):
- # If we are executing this function, we are the process zero, so we don't check for that.
- output_dir = output_dir if output_dir is not None else self.args.output_dir
- os.makedirs(output_dir, exist_ok=True)
- logger.info(f"Saving model checkpoint to {output_dir}")
- supported_classes = (PreTrainedModel,) if not is_peft_available() else (PreTrainedModel, PeftModel)
- # Save a trained model and configuration using `save_pretrained()`.
- # They can then be reloaded using `from_pretrained()`
- if not isinstance(self.model, supported_classes):
- if state_dict is None:
- state_dict = self.model.state_dict()
- if isinstance(self.accelerator.unwrap_model(self.model, keep_torch_compile=False), supported_classes):
- self.accelerator.unwrap_model(self.model, keep_torch_compile=False).save_pretrained(
- output_dir, state_dict=state_dict, safe_serialization=self.args.save_safetensors
- )
- else:
- logger.info("Trainer.model is not a `PreTrainedModel`, only saving its state dict.")
- if self.args.save_safetensors:
- safetensors.torch.save_file(
- state_dict, os.path.join(output_dir, SAFE_WEIGHTS_NAME), metadata={"format": "pt"}
- )
- else:
- torch.save(state_dict, os.path.join(output_dir, WEIGHTS_NAME))
- else:
- self.model.save_pretrained(
- output_dir, state_dict=state_dict, safe_serialization=self.args.save_safetensors
- )
- if self.processing_class is not None:
- self.processing_class.save_pretrained(output_dir)
- elif (
- self.data_collator is not None
- and hasattr(self.data_collator, "tokenizer")
- and self.data_collator.tokenizer is not None
- ):
- logger.info("Saving Trainer.data_collator.tokenizer by default as Trainer.processing_class is `None`")
- self.data_collator.tokenizer.save_pretrained(output_dir)
- # Good practice: save your training arguments together with the trained model
- torch.save(self.args, os.path.join(output_dir, TRAINING_ARGS_NAME))
- def store_flos(self):
- # Storing the number of floating-point operations that went into the model
- if self.args.parallel_mode == ParallelMode.DISTRIBUTED:
- self.state.total_flos += (
- distributed_broadcast_scalars([self.current_flos], device=self.args.device).sum().item()
- )
- self.current_flos = 0
- else:
- self.state.total_flos += self.current_flos
- self.current_flos = 0
- def _sorted_checkpoints(
- self, output_dir=None, checkpoint_prefix=PREFIX_CHECKPOINT_DIR, use_mtime=False
- ) -> list[str]:
- ordering_and_checkpoint_path = []
- glob_checkpoints = [str(x) for x in Path(output_dir).glob(f"{checkpoint_prefix}-*") if os.path.isdir(x)]
- for path in glob_checkpoints:
- if use_mtime:
- ordering_and_checkpoint_path.append((os.path.getmtime(path), path))
- else:
- regex_match = re.match(f".*{checkpoint_prefix}-([0-9]+)", path)
- if regex_match is not None and regex_match.groups() is not None:
- ordering_and_checkpoint_path.append((int(regex_match.groups()[0]), path))
- checkpoints_sorted = sorted(ordering_and_checkpoint_path)
- # mtime is not reliable on all filesystems, especially on some fuse fs in cloud environments
- # so we check if the mtime is fake and fallback to numerical ordering if needed
- if use_mtime and len(ordering_and_checkpoint_path) > 1:
- mtime_diff = checkpoints_sorted[-1][0] - checkpoints_sorted[0][0]
- if mtime_diff < 1.0: # less than 1 second, which is almost impossible when mtime works fine
- warnings.warn("mtime may not be reliable on this filesystem, falling back to numerical ordering")
- return self._sorted_checkpoints(
- use_mtime=False, output_dir=output_dir, checkpoint_prefix=checkpoint_prefix
- )
- checkpoints_sorted = [checkpoint[1] for checkpoint in checkpoints_sorted]
- # Make sure we don't delete the best model.
- if (
- self.state.best_model_checkpoint is not None
- and str(Path(self.state.best_model_checkpoint)) in checkpoints_sorted
- ):
- best_model_index = checkpoints_sorted.index(str(Path(self.state.best_model_checkpoint)))
- for i in range(best_model_index, len(checkpoints_sorted) - 2):
- checkpoints_sorted[i], checkpoints_sorted[i + 1] = checkpoints_sorted[i + 1], checkpoints_sorted[i]
- return checkpoints_sorted
- def _rotate_checkpoints(self, use_mtime=False, output_dir=None) -> None:
- if self.args.save_total_limit is None or self.args.save_total_limit <= 0:
- return
- # Check if we should delete older checkpoint(s)
- checkpoints_sorted = self._sorted_checkpoints(use_mtime=use_mtime, output_dir=output_dir)
- if len(checkpoints_sorted) <= self.args.save_total_limit:
- return
- # If save_total_limit=1 with load_best_model_at_end=True, we could end up deleting the last checkpoint, which
- # we don't do to allow resuming.
- save_total_limit = self.args.save_total_limit
- if (
- self.state.best_model_checkpoint is not None
- and self.args.save_total_limit == 1
- and checkpoints_sorted[-1] != self.state.best_model_checkpoint
- ):
- save_total_limit = 2
- number_of_checkpoints_to_delete = max(0, len(checkpoints_sorted) - save_total_limit)
- checkpoints_to_be_deleted = checkpoints_sorted[:number_of_checkpoints_to_delete]
- for checkpoint in checkpoints_to_be_deleted:
- logger.info(f"Deleting older checkpoint [{checkpoint}] due to args.save_total_limit")
- shutil.rmtree(checkpoint, ignore_errors=True)
- def evaluate(
- self,
- eval_dataset: Optional[Union[Dataset, dict[str, Dataset]]] = None,
- ignore_keys: Optional[list[str]] = None,
- metric_key_prefix: str = "eval",
- ) -> dict[str, float]:
- """
- Run evaluation and returns metrics.
- The calling script will be responsible for providing a method to compute metrics, as they are task-dependent
- (pass it to the init `compute_metrics` argument).
- You can also subclass and override this method to inject custom behavior.
- Args:
- eval_dataset (Union[`Dataset`, dict[str, `Dataset`]), *optional*):
- Pass a dataset if you wish to override `self.eval_dataset`. If it is a [`~datasets.Dataset`], columns
- not accepted by the `model.forward()` method are automatically removed. If it is a dictionary, it will
- evaluate on each dataset, prepending the dictionary key to the metric name. Datasets must implement the
- `__len__` method.
- <Tip>
- If you pass a dictionary with names of datasets as keys and datasets as values, evaluate will run
- separate evaluations on each dataset. This can be useful to monitor how training affects other
- datasets or simply to get a more fine-grained evaluation.
- When used with `load_best_model_at_end`, make sure `metric_for_best_model` references exactly one
- of the datasets. If you, for example, pass in `{"data1": data1, "data2": data2}` for two datasets
- `data1` and `data2`, you could specify `metric_for_best_model="eval_data1_loss"` for using the
- loss on `data1` and `metric_for_best_model="eval_data2_loss"` for the loss on `data2`.
- </Tip>
- ignore_keys (`list[str]`, *optional*):
- A list of keys in the output of your model (if it is a dictionary) that should be ignored when
- gathering predictions.
- metric_key_prefix (`str`, *optional*, defaults to `"eval"`):
- An optional prefix to be used as the metrics key prefix. For example the metrics "bleu" will be named
- "eval_bleu" if the prefix is "eval" (default)
- Returns:
- A dictionary containing the evaluation loss and the potential metrics computed from the predictions. The
- dictionary also contains the epoch number which comes from the training state.
- """
- # handle multiple eval datasets
- override = eval_dataset is not None
- eval_dataset = eval_dataset if override else self.eval_dataset
- if isinstance(eval_dataset, dict):
- metrics = {}
- for eval_dataset_name, _eval_dataset in eval_dataset.items():
- dataset_metrics = self.evaluate(
- eval_dataset=_eval_dataset if override else eval_dataset_name,
- ignore_keys=ignore_keys,
- metric_key_prefix=f"{metric_key_prefix}_{eval_dataset_name}",
- )
- metrics.update(dataset_metrics)
- return metrics
- # memory metrics - must set up as early as possible
- self._memory_tracker.start()
- eval_dataloader = self.get_eval_dataloader(eval_dataset)
- if self.is_fsdp_xla_v2_enabled:
- eval_dataloader = tpu_spmd_dataloader(eval_dataloader)
- start_time = time.time()
- eval_loop = self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
- output = eval_loop(
- eval_dataloader,
- description="Evaluation",
- # No point gathering the predictions if there are no metrics, otherwise we defer to
- # self.args.prediction_loss_only
- prediction_loss_only=True if self.compute_metrics is None else None,
- ignore_keys=ignore_keys,
- metric_key_prefix=metric_key_prefix,
- )
- total_batch_size = self.args.eval_batch_size * self.args.world_size
- if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
- start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
- if f"{metric_key_prefix}_model_preparation_time" in output.metrics:
- start_time += output.metrics[f"{metric_key_prefix}_model_preparation_time"]
- output.metrics.update(
- speed_metrics(
- metric_key_prefix,
- start_time,
- num_samples=output.num_samples,
- num_steps=math.ceil(output.num_samples / total_batch_size),
- )
- )
- self.log(output.metrics)
- if DebugOption.TPU_METRICS_DEBUG in self.args.debug:
- # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
- xm.master_print(met.metrics_report())
- self.control = self.callback_handler.on_evaluate(self.args, self.state, self.control, output.metrics)
- self._memory_tracker.stop_and_update_metrics(output.metrics)
- return output.metrics
- def predict(
- self, test_dataset: Dataset, ignore_keys: Optional[list[str]] = None, metric_key_prefix: str = "test"
- ) -> PredictionOutput:
- """
- Run prediction and returns predictions and potential metrics.
- Depending on the dataset and your use case, your test dataset may contain labels. In that case, this method
- will also return metrics, like in `evaluate()`.
- Args:
- test_dataset (`Dataset`):
- Dataset to run the predictions on. If it is an `datasets.Dataset`, columns not accepted by the
- `model.forward()` method are automatically removed. Has to implement the method `__len__`
- ignore_keys (`list[str]`, *optional*):
- A list of keys in the output of your model (if it is a dictionary) that should be ignored when
- gathering predictions.
- metric_key_prefix (`str`, *optional*, defaults to `"test"`):
- An optional prefix to be used as the metrics key prefix. For example the metrics "bleu" will be named
- "test_bleu" if the prefix is "test" (default)
- <Tip>
- If your predictions or labels have different sequence length (for instance because you're doing dynamic padding
- in a token classification task) the predictions will be padded (on the right) to allow for concatenation into
- one array. The padding index is -100.
- </Tip>
- Returns: *NamedTuple* A namedtuple with the following keys:
- - predictions (`np.ndarray`): The predictions on `test_dataset`.
- - label_ids (`np.ndarray`, *optional*): The labels (if the dataset contained some).
- - metrics (`dict[str, float]`, *optional*): The potential dictionary of metrics (if the dataset contained
- labels).
- """
- # memory metrics - must set up as early as possible
- self._memory_tracker.start()
- test_dataloader = self.get_test_dataloader(test_dataset)
- start_time = time.time()
- eval_loop = self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
- output = eval_loop(
- test_dataloader, description="Prediction", ignore_keys=ignore_keys, metric_key_prefix=metric_key_prefix
- )
- total_batch_size = self.args.eval_batch_size * self.args.world_size
- if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
- start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
- if f"{metric_key_prefix}_model_preparation_time" in output.metrics:
- start_time += output.metrics[f"{metric_key_prefix}_model_preparation_time"]
- output.metrics.update(
- speed_metrics(
- metric_key_prefix,
- start_time,
- num_samples=output.num_samples,
- num_steps=math.ceil(output.num_samples / total_batch_size),
- )
- )
- self.control = self.callback_handler.on_predict(self.args, self.state, self.control, output.metrics)
- self._memory_tracker.stop_and_update_metrics(output.metrics)
- return PredictionOutput(predictions=output.predictions, label_ids=output.label_ids, metrics=output.metrics)
- def evaluation_loop(
- self,
- dataloader: DataLoader,
- description: str,
- prediction_loss_only: Optional[bool] = None,
- ignore_keys: Optional[list[str]] = None,
- metric_key_prefix: str = "eval",
- ) -> EvalLoopOutput:
- """
- Prediction/evaluation loop, shared by `Trainer.evaluate()` and `Trainer.predict()`.
- Works both with or without labels.
- """
- args = self.args
- prediction_loss_only = prediction_loss_only if prediction_loss_only is not None else args.prediction_loss_only
- # if eval is called w/o train, handle model prep here
- if self.is_deepspeed_enabled and self.deepspeed is None:
- _, _ = deepspeed_init(self, num_training_steps=0, inference=True)
- model = self._wrap_model(self.model, training=False, dataloader=dataloader)
- if len(self.accelerator._models) == 0 and model is self.model:
- start_time = time.time()
- model = (
- self.accelerator.prepare(model)
- if self.is_deepspeed_enabled
- or (self.is_fsdp_enabled and self.accelerator.mixed_precision != "fp8" and not self.args.torch_compile)
- else self.accelerator.prepare_model(model, evaluation_mode=True)
- )
- self.model_preparation_time = round(time.time() - start_time, 4)
- if self.is_fsdp_enabled:
- self.model = model
- # for the rest of this function `model` is the outside model, whether it was wrapped or not
- if model is not self.model:
- self.model_wrapped = model
- # backward compatibility
- if self.is_deepspeed_enabled:
- self.deepspeed = self.model_wrapped
- # if full fp16 or bf16 eval is wanted and this ``evaluation`` or ``predict`` isn't called
- # while ``train`` is running, cast it to the right dtype first and then put on device
- if not self.is_in_train:
- if args.fp16_full_eval:
- model = model.to(dtype=torch.float16, device=args.device)
- elif args.bf16_full_eval:
- model = model.to(dtype=torch.bfloat16, device=args.device)
- batch_size = self.args.eval_batch_size
- logger.info(f"\n***** Running {description} *****")
- if has_length(dataloader):
- logger.info(f" Num examples = {self.num_examples(dataloader)}")
- else:
- logger.info(" Num examples: Unknown")
- logger.info(f" Batch size = {batch_size}")
- if hasattr(model, "eval") and callable(model.eval):
- model.eval()
- if hasattr(self.optimizer, "eval") and callable(self.optimizer.eval):
- self.optimizer.eval()
- self.callback_handler.eval_dataloader = dataloader
- # Do this before wrapping.
- eval_dataset = getattr(dataloader, "dataset", None)
- if args.past_index >= 0:
- self._past = None
- # Initialize containers
- all_losses = EvalLoopContainer(self.args.eval_do_concat_batches, padding_index=-100)
- all_preds = EvalLoopContainer(self.args.eval_do_concat_batches, padding_index=-100)
- all_labels = EvalLoopContainer(self.args.eval_do_concat_batches, padding_index=-100)
- all_inputs = EvalLoopContainer(self.args.eval_do_concat_batches, padding_index=-100)
- metrics = None
- eval_set_kwargs = {}
- # Will be useful when we have an iterable dataset so don't know its length.
- observed_num_examples = 0
- # Main evaluation loop
- for step, inputs in enumerate(dataloader):
- # Update the observed num examples
- observed_batch_size = find_batch_size(inputs)
- if observed_batch_size is not None:
- observed_num_examples += observed_batch_size
- # For batch samplers, batch_size is not known by the dataloader in advance.
- if batch_size is None:
- batch_size = observed_batch_size
- # Prediction step
- losses, logits, labels = self.prediction_step(model, inputs, prediction_loss_only, ignore_keys=ignore_keys)
- main_input_name = getattr(self.model, "main_input_name", "input_ids")
- inputs_decode = (
- self._prepare_input(inputs[main_input_name]) if "inputs" in args.include_for_metrics else None
- )
- if is_torch_xla_available():
- xm.mark_step()
- # Update containers
- if losses is not None:
- losses = self.gather_function(losses.repeat(batch_size))
- all_losses.add(losses)
- if inputs_decode is not None:
- inputs_decode = self.accelerator.pad_across_processes(inputs_decode, dim=1, pad_index=-100)
- inputs_decode = self.gather_function(inputs_decode)
- if not self.args.batch_eval_metrics or description == "Prediction":
- all_inputs.add(inputs_decode)
- if labels is not None:
- # Pad labels here, preparing for preprocess_logits_for_metrics in next logits block.
- labels = self.accelerator.pad_across_processes(labels, dim=1, pad_index=-100)
- if logits is not None:
- logits = self.accelerator.pad_across_processes(logits, dim=1, pad_index=-100)
- if self.preprocess_logits_for_metrics is not None:
- logits = self.preprocess_logits_for_metrics(logits, labels)
- logits = self.gather_function(logits)
- if not self.args.batch_eval_metrics or description == "Prediction":
- all_preds.add(logits)
- if labels is not None:
- labels = self.gather_function(labels)
- if not self.args.batch_eval_metrics or description == "Prediction":
- all_labels.add(labels)
- self.control = self.callback_handler.on_prediction_step(args, self.state, self.control)
- if self.args.batch_eval_metrics:
- if self.compute_metrics is not None and logits is not None and labels is not None:
- is_last_step = self.accelerator.gradient_state.end_of_dataloader
- batch_kwargs = {}
- batch_kwargs["losses"] = losses if "loss" in args.include_for_metrics else None
- batch_kwargs["inputs"] = inputs if "inputs" in args.include_for_metrics else None
- metrics = self.compute_metrics(
- EvalPrediction(predictions=logits, label_ids=labels, **batch_kwargs),
- compute_result=is_last_step,
- )
- del losses, logits, labels, inputs
- torch.cuda.empty_cache()
- # Gather all tensors and put them back on the CPU if we have done enough accumulation steps.
- elif args.eval_accumulation_steps is not None and (step + 1) % args.eval_accumulation_steps == 0:
- all_losses.to_cpu_and_numpy()
- all_preds.to_cpu_and_numpy()
- all_labels.to_cpu_and_numpy()
- all_inputs.to_cpu_and_numpy()
- del losses, logits, labels, inputs
- torch.cuda.empty_cache()
- # After all calls to `.gather_function`, reset to `gather_for_metrics`:
- self.gather_function = self.accelerator.gather_for_metrics
- if args.past_index and hasattr(self, "_past"):
- # Clean the state at the end of the evaluation loop
- delattr(self, "_past")
- # Gather all remaining tensors and put them back on the CPU
- all_losses = all_losses.get_arrays()
- all_preds = all_preds.get_arrays()
- all_labels = all_labels.get_arrays()
- all_inputs = all_inputs.get_arrays()
- # Number of samples
- if has_length(eval_dataset):
- num_samples = len(eval_dataset)
- # The instance check is weird and does not actually check for the type, but whether the dataset has the right
- # methods. Therefore we need to make sure it also has the attribute.
- elif isinstance(eval_dataset, IterableDatasetShard) and getattr(eval_dataset, "num_examples", 0) > 0:
- num_samples = eval_dataset.num_examples
- else:
- if has_length(dataloader):
- num_samples = self.num_examples(dataloader)
- else: # both len(dataloader.dataset) and len(dataloader) fail
- num_samples = observed_num_examples
- if num_samples == 0 and observed_num_examples > 0:
- num_samples = observed_num_examples
- # Metrics!
- if (
- self.compute_metrics is not None
- and all_preds is not None
- and all_labels is not None
- and not self.args.batch_eval_metrics
- ):
- eval_set_kwargs["losses"] = all_losses if "loss" in args.include_for_metrics else None
- eval_set_kwargs["inputs"] = all_inputs if "inputs" in args.include_for_metrics else None
- metrics = self.compute_metrics(
- EvalPrediction(predictions=all_preds, label_ids=all_labels, **eval_set_kwargs)
- )
- elif metrics is None:
- metrics = {}
- # To be JSON-serializable, we need to remove numpy types or zero-d tensors
- metrics = denumpify_detensorize(metrics)
- if isinstance(all_losses, list) and all_losses:
- metrics[f"{metric_key_prefix}_loss"] = np.concatenate(all_losses).mean().item()
- elif isinstance(all_losses, np.ndarray):
- metrics[f"{metric_key_prefix}_loss"] = all_losses.mean().item()
- if hasattr(self, "jit_compilation_time"):
- metrics[f"{metric_key_prefix}_jit_compilation_time"] = self.jit_compilation_time
- if hasattr(self, "model_preparation_time"):
- metrics[f"{metric_key_prefix}_model_preparation_time"] = self.model_preparation_time
- # Prefix all keys with metric_key_prefix + '_'
- for key in list(metrics.keys()):
- if not key.startswith(f"{metric_key_prefix}_"):
- metrics[f"{metric_key_prefix}_{key}"] = metrics.pop(key)
- return EvalLoopOutput(predictions=all_preds, label_ids=all_labels, metrics=metrics, num_samples=num_samples)
- def _nested_gather(self, tensors, name=None):
- """
- Gather value of `tensors` (tensor or list/tuple of nested tensors) and convert them to numpy before
- concatenating them to `gathered`
- """
- if tensors is None:
- return
- if is_torch_xla_available():
- if name is None:
- name = "nested_gather"
- tensors = nested_xla_mesh_reduce(tensors, name)
- elif is_sagemaker_mp_enabled():
- tensors = smp_gather(tensors)
- elif (self.args.distributed_state is not None and self.args.distributed_state.distributed_type != "NO") or (
- self.args.distributed_state is None and self.args.local_rank != -1
- ):
- tensors = distributed_concat(tensors)
- return tensors
- def prediction_step(
- self,
- model: nn.Module,
- inputs: dict[str, Union[torch.Tensor, Any]],
- prediction_loss_only: bool,
- ignore_keys: Optional[list[str]] = None,
- ) -> tuple[Optional[torch.Tensor], Optional[torch.Tensor], Optional[torch.Tensor]]:
- """
- Perform an evaluation step on `model` using `inputs`.
- Subclass and override to inject custom behavior.
- Args:
- model (`nn.Module`):
- The model to evaluate.
- inputs (`dict[str, Union[torch.Tensor, Any]]`):
- The inputs and targets of the model.
- The dictionary will be unpacked before being fed to the model. Most models expect the targets under the
- argument `labels`. Check your model's documentation for all accepted arguments.
- prediction_loss_only (`bool`):
- Whether or not to return the loss only.
- ignore_keys (`list[str]`, *optional*):
- A list of keys in the output of your model (if it is a dictionary) that should be ignored when
- gathering predictions.
- Return:
- tuple[Optional[torch.Tensor], Optional[torch.Tensor], Optional[torch.Tensor]]: A tuple with the loss,
- logits and labels (each being optional).
- """
- has_labels = False if len(self.label_names) == 0 else all(inputs.get(k) is not None for k in self.label_names)
- # For CLIP-like models capable of returning loss values.
- # If `return_loss` is not specified or being `None` in `inputs`, we check if the default value of `return_loss`
- # is `True` in `model.forward`.
- return_loss = inputs.get("return_loss")
- if return_loss is None:
- return_loss = self.can_return_loss
- loss_without_labels = len(self.label_names) == 0 and return_loss
- inputs = self._prepare_inputs(inputs)
- if ignore_keys is None:
- if hasattr(self.model, "config"):
- ignore_keys = getattr(self.model.config, "keys_to_ignore_at_inference", ["past_key_values"])
- else:
- ignore_keys = []
- # labels may be popped when computing the loss (label smoothing for instance) so we grab them first.
- if has_labels or loss_without_labels:
- labels = nested_detach(tuple(inputs.get(name) for name in self.label_names))
- if len(labels) == 1:
- labels = labels[0]
- else:
- labels = None
- with torch.no_grad():
- if is_sagemaker_mp_enabled():
- raw_outputs = smp_forward_only(model, inputs)
- if has_labels or loss_without_labels:
- if isinstance(raw_outputs, dict):
- loss_mb = raw_outputs["loss"]
- logits_mb = tuple(v for k, v in raw_outputs.items() if k not in ignore_keys + ["loss"])
- else:
- loss_mb = raw_outputs[0]
- logits_mb = raw_outputs[1:]
- loss = loss_mb.reduce_mean().detach().cpu()
- logits = smp_nested_concat(logits_mb)
- else:
- loss = None
- if isinstance(raw_outputs, dict):
- logits_mb = tuple(v for k, v in raw_outputs.items() if k not in ignore_keys)
- else:
- logits_mb = raw_outputs
- logits = smp_nested_concat(logits_mb)
- else:
- if has_labels or loss_without_labels:
- with self.compute_loss_context_manager():
- num_items_in_batch = self._get_num_items_in_batch([inputs], self.args.device)
- loss, outputs = self.compute_loss(
- model, inputs, return_outputs=True, num_items_in_batch=num_items_in_batch
- )
- loss = loss.detach().mean()
- if isinstance(outputs, dict):
- logits = tuple(v for k, v in outputs.items() if k not in ignore_keys + ["loss"])
- else:
- logits = outputs[1:]
- else:
- loss = None
- with self.compute_loss_context_manager():
- outputs = model(**inputs)
- if isinstance(outputs, dict):
- logits = tuple(v for k, v in outputs.items() if k not in ignore_keys)
- else:
- logits = outputs
- # TODO: this needs to be fixed and made cleaner later.
- if self.args.past_index >= 0:
- self._past = outputs[self.args.past_index - 1]
- if prediction_loss_only:
- return (loss, None, None)
- logits = nested_detach(logits)
- if len(logits) == 1:
- logits = logits[0]
- return (loss, logits, labels)
- def floating_point_ops(self, inputs: dict[str, Union[torch.Tensor, Any]]):
- """
- For models that inherit from [`PreTrainedModel`], uses that method to compute the number of floating point
- operations for every backward + forward pass. If using another model, either implement such a method in the
- model or subclass and override this method.
- Args:
- inputs (`dict[str, Union[torch.Tensor, Any]]`):
- The inputs and targets of the model.
- Returns:
- `int`: The number of floating-point operations.
- """
- if hasattr(self.model, "floating_point_ops"):
- return self.model.floating_point_ops(inputs)
- else:
- return 0
- def init_hf_repo(self, token: Optional[str] = None):
- """
- Initializes a git repo in `self.args.hub_model_id`.
- """
- # Only on process zero
- if not self.is_world_process_zero():
- return
- if self.args.hub_model_id is None:
- repo_name = Path(self.args.output_dir).absolute().name
- else:
- repo_name = self.args.hub_model_id
- token = token if token is not None else self.args.hub_token
- repo_url = create_repo(repo_name, token=token, private=self.args.hub_private_repo, exist_ok=True)
- self.hub_model_id = repo_url.repo_id
- self.push_in_progress = None
- def create_model_card(
- self,
- language: Optional[str] = None,
- license: Optional[str] = None,
- tags: Union[str, list[str], None] = None,
- model_name: Optional[str] = None,
- finetuned_from: Optional[str] = None,
- tasks: Union[str, list[str], None] = None,
- dataset_tags: Union[str, list[str], None] = None,
- dataset: Union[str, list[str], None] = None,
- dataset_args: Union[str, list[str], None] = None,
- ):
- """
- Creates a draft of a model card using the information available to the `Trainer`.
- Args:
- language (`str`, *optional*):
- The language of the model (if applicable)
- license (`str`, *optional*):
- The license of the model. Will default to the license of the pretrained model used, if the original
- model given to the `Trainer` comes from a repo on the Hub.
- tags (`str` or `list[str]`, *optional*):
- Some tags to be included in the metadata of the model card.
- model_name (`str`, *optional*):
- The name of the model.
- finetuned_from (`str`, *optional*):
- The name of the model used to fine-tune this one (if applicable). Will default to the name of the repo
- of the original model given to the `Trainer` (if it comes from the Hub).
- tasks (`str` or `list[str]`, *optional*):
- One or several task identifiers, to be included in the metadata of the model card.
- dataset_tags (`str` or `list[str]`, *optional*):
- One or several dataset tags, to be included in the metadata of the model card.
- dataset (`str` or `list[str]`, *optional*):
- One or several dataset identifiers, to be included in the metadata of the model card.
- dataset_args (`str` or `list[str]`, *optional*):
- One or several dataset arguments, to be included in the metadata of the model card.
- """
- if not self.is_world_process_zero():
- return
- model_card_filepath = os.path.join(self.args.output_dir, "README.md")
- is_peft_library = False
- if os.path.exists(model_card_filepath):
- library_name = ModelCard.load(model_card_filepath).data.get("library_name")
- is_peft_library = library_name == "peft"
- # Append existing tags in `tags`
- existing_tags = ModelCard.load(model_card_filepath).data.tags
- if tags is not None and existing_tags is not None:
- if isinstance(tags, str):
- tags = [tags]
- for tag in existing_tags:
- if tag not in tags:
- tags.append(tag)
- training_summary = TrainingSummary.from_trainer(
- self,
- language=language,
- license=license,
- tags=tags,
- model_name=model_name,
- finetuned_from=finetuned_from,
- tasks=tasks,
- dataset_tags=dataset_tags,
- dataset=dataset,
- dataset_args=dataset_args,
- )
- model_card = training_summary.to_model_card()
- with open(model_card_filepath, "w") as f:
- f.write(model_card)
- if is_peft_library:
- self.accelerator.unwrap_model(self.model).create_or_update_model_card(self.args.output_dir)
- def _push_from_checkpoint(self, checkpoint_folder):
- # Only push from one node.
- if not self.is_world_process_zero() or self.args.hub_strategy == HubStrategy.END:
- return
- # If we haven't finished the last push, we don't do this one unless args.hub_always_push=True.
- if not self.args.hub_always_push and self.push_in_progress is not None and not self.push_in_progress.is_done():
- return
- output_dir = self.args.output_dir
- # To avoid a new synchronization of all model weights, we just copy the file from the checkpoint folder
- modeling_files = [CONFIG_NAME, GENERATION_CONFIG_NAME, WEIGHTS_NAME, SAFE_WEIGHTS_NAME]
- # Add sharded checkpoints if we have an index
- for index_file in [WEIGHTS_INDEX_NAME, SAFE_WEIGHTS_INDEX_NAME]:
- index_path = os.path.join(checkpoint_folder, index_file)
- if os.path.isfile(index_path):
- modeling_files.append(index_file)
- with open(index_path) as f:
- index = json.loads(f.read())
- shard_files = list(set(index["weight_map"].values()))
- modeling_files.extend(shard_files)
- if is_peft_available():
- modeling_files.extend([ADAPTER_CONFIG_NAME, ADAPTER_WEIGHTS_NAME, ADAPTER_SAFE_WEIGHTS_NAME])
- for modeling_file in modeling_files:
- if os.path.isfile(os.path.join(checkpoint_folder, modeling_file)):
- shutil.copy(os.path.join(checkpoint_folder, modeling_file), os.path.join(output_dir, modeling_file))
- # Saving the processing class is fast and we don't know how many files it may have spawned, so we resave it to be sure.
- if self.processing_class is not None:
- self.processing_class.save_pretrained(output_dir)
- # Same for the training arguments
- torch.save(self.args, os.path.join(output_dir, TRAINING_ARGS_NAME))
- if self.args.save_strategy == SaveStrategy.STEPS:
- commit_message = f"Training in progress, step {self.state.global_step}"
- else:
- commit_message = f"Training in progress, epoch {int(self.state.epoch)}"
- model_push_job = upload_folder(
- repo_id=self.hub_model_id,
- folder_path=output_dir,
- commit_message=commit_message,
- token=self.args.hub_token,
- run_as_future=True,
- ignore_patterns=["_*", f"{PREFIX_CHECKPOINT_DIR}-*"],
- revision=self.args.hub_revision,
- )
- push_jobs = [model_push_job]
- if self.args.hub_strategy in [HubStrategy.CHECKPOINT, HubStrategy.ALL_CHECKPOINTS]:
- path_in_repo = (
- "last-checkpoint" if self.args.hub_strategy == HubStrategy.CHECKPOINT else Path(checkpoint_folder).name
- )
- checkpoint_push = upload_folder(
- repo_id=self.hub_model_id,
- folder_path=checkpoint_folder,
- path_in_repo=path_in_repo,
- commit_message=commit_message + ", checkpoint",
- token=self.args.hub_token,
- run_as_future=True,
- revision=self.args.hub_revision,
- )
- push_jobs.append(checkpoint_push)
- if self.push_in_progress is None or self.push_in_progress.is_done():
- self.push_in_progress = PushInProgress(push_jobs)
- else:
- self.push_in_progress.jobs.extend(push_jobs)
- def _finish_current_push(self):
- if not hasattr(self, "push_in_progress"):
- return
- if self.push_in_progress is not None and not self.push_in_progress.is_done():
- logger.info("Waiting for the current checkpoint push to be finished, this might take a couple of minutes.")
- self.push_in_progress.wait_until_done()
- def push_to_hub(
- self,
- commit_message: Optional[str] = "End of training",
- blocking: bool = True,
- token: Optional[str] = None,
- revision: Optional[str] = None,
- **kwargs,
- ) -> str:
- """
- Upload `self.model` and `self.processing_class` to the 🤗 model hub on the repo `self.args.hub_model_id`.
- Parameters:
- commit_message (`str`, *optional*, defaults to `"End of training"`):
- Message to commit while pushing.
- blocking (`bool`, *optional*, defaults to `True`):
- Whether the function should return only when the `git push` has finished.
- token (`str`, *optional*, defaults to `None`):
- Token with write permission to overwrite Trainer's original args.
- revision (`str`, *optional*):
- The git revision to commit from. Defaults to the head of the "main" branch.
- kwargs (`dict[str, Any]`, *optional*):
- Additional keyword arguments passed along to [`~Trainer.create_model_card`].
- Returns:
- The URL of the repository where the model was pushed if `blocking=False`, or a `Future` object tracking the
- progress of the commit if `blocking=True`.
- """
- model_name = kwargs.pop("model_name", None)
- if model_name is None and self.args.should_save:
- if self.args.hub_model_id is None:
- model_name = Path(self.args.output_dir).name
- else:
- model_name = self.args.hub_model_id.split("/")[-1]
- token = token if token is not None else self.args.hub_token
- # In case the user calls this method with args.push_to_hub = False
- if self.hub_model_id is None:
- self.init_hf_repo(token=token)
- # Needs to be executed on all processes for TPU training, but will only save on the processed determined by
- # self.args.should_save.
- self.save_model(_internal_call=True)
- # Only push from one node.
- if not self.is_world_process_zero():
- return
- # Add additional tags in the case the model has already some tags and users pass
- # "tags" argument to `push_to_hub` so that trainer automatically handles internal tags
- # from all models since Trainer does not call `model.push_to_hub`.
- if getattr(self.model, "model_tags", None) is not None:
- if "tags" not in kwargs:
- kwargs["tags"] = []
- # If it is a string, convert it to a list
- if isinstance(kwargs["tags"], str):
- kwargs["tags"] = [kwargs["tags"]]
- for model_tag in self.model.model_tags:
- if model_tag not in kwargs["tags"]:
- kwargs["tags"].append(model_tag)
- self.create_model_card(model_name=model_name, **kwargs)
- if revision is None:
- revision = self.args.hub_revision
- # Wait for the current upload to be finished.
- self._finish_current_push()
- return upload_folder(
- repo_id=self.hub_model_id,
- folder_path=self.args.output_dir,
- commit_message=commit_message,
- token=token,
- run_as_future=not blocking,
- ignore_patterns=["_*", f"{PREFIX_CHECKPOINT_DIR}-*"],
- revision=revision,
- )
- #
- # Deprecated code
- #
- def prediction_loop(
- self,
- dataloader: DataLoader,
- description: str,
- prediction_loss_only: Optional[bool] = None,
- ignore_keys: Optional[list[str]] = None,
- metric_key_prefix: str = "eval",
- ) -> EvalLoopOutput:
- """
- Prediction/evaluation loop, shared by `Trainer.evaluate()` and `Trainer.predict()`.
- Works both with or without labels.
- """
- args = self.args
- if not has_length(dataloader):
- raise ValueError("dataloader must implement a working __len__")
- prediction_loss_only = prediction_loss_only if prediction_loss_only is not None else args.prediction_loss_only
- # if eval is called w/o train, handle model prep here
- if self.is_deepspeed_enabled and self.deepspeed is None:
- _, _ = deepspeed_init(self, num_training_steps=0, inference=True)
- model = self._wrap_model(self.model, training=False, dataloader=dataloader)
- if len(self.accelerator._models) == 0 and model is self.model:
- model = (
- self.accelerator.prepare(model)
- if self.is_deepspeed_enabled or self.is_fsdp_enabled
- else self.accelerator.prepare_model(model, evaluation_mode=True)
- )
- if self.is_fsdp_enabled:
- self.model = model
- # for the rest of this function `model` is the outside model, whether it was wrapped or not
- if model is not self.model:
- self.model_wrapped = model
- # backward compatibility
- if self.is_deepspeed_enabled:
- self.deepspeed = self.model_wrapped
- # if full fp16 or bf16 eval is wanted and this ``evaluation`` or ``predict`` isn't called
- # while ``train`` is running, cast it to the right dtype first and then put on device
- if not self.is_in_train:
- if args.fp16_full_eval:
- model = model.to(dtype=torch.float16, device=args.device)
- elif args.bf16_full_eval:
- model = model.to(dtype=torch.bfloat16, device=args.device)
- batch_size = (
- dataloader.total_batch_size
- if getattr(dataloader, "_is_accelerate_prepared", False)
- else dataloader.batch_size
- )
- if batch_size is None:
- raise ValueError(
- "Batch size cannot be None. Ensure the dataloader has a valid batch_size or total_batch_size."
- )
- num_examples = self.num_examples(dataloader)
- logger.info(f"\n***** Running {description} *****")
- logger.info(f" Num examples = {num_examples}")
- logger.info(f" Batch size = {batch_size}")
- losses_host: Optional[torch.Tensor] = None
- preds_host: Union[torch.Tensor, list[torch.Tensor], None] = None
- labels_host: Union[torch.Tensor, list[torch.Tensor], None] = None
- inputs_host: Union[torch.Tensor, list[torch.Tensor], None] = None
- metrics: Optional[dict] = None
- eval_set_kwargs: dict = {}
- world_size = max(1, args.world_size)
- eval_losses_gatherer = DistributedTensorGatherer(world_size, num_examples, make_multiple_of=batch_size)
- if not prediction_loss_only:
- # The actual number of eval_sample can be greater than num_examples in distributed settings (when we pass
- # a batch size to the sampler)
- make_multiple_of = None
- if hasattr(dataloader, "sampler") and isinstance(dataloader.sampler, SequentialDistributedSampler):
- make_multiple_of = dataloader.sampler.batch_size
- preds_gatherer = DistributedTensorGatherer(world_size, num_examples, make_multiple_of=make_multiple_of)
- labels_gatherer = DistributedTensorGatherer(world_size, num_examples, make_multiple_of=make_multiple_of)
- inputs_gatherer = DistributedTensorGatherer(world_size, num_examples, make_multiple_of=make_multiple_of)
- model.eval()
- if hasattr(self.optimizer, "eval") and callable(self.optimizer.eval):
- self.optimizer.eval()
- if args.past_index >= 0:
- self._past = None
- self.callback_handler.eval_dataloader = dataloader
- for step, inputs in enumerate(dataloader):
- loss, logits, labels = self.prediction_step(model, inputs, prediction_loss_only, ignore_keys=ignore_keys)
- main_input_name = getattr(self.model, "main_input_name", "input_ids")
- inputs_decode = (
- self._prepare_input(inputs[main_input_name]) if "inputs" in args.include_for_metrics else None
- )
- if loss is not None:
- losses = loss.repeat(batch_size)
- losses_host = losses if losses_host is None else torch.cat((losses_host, losses), dim=0)
- if logits is not None:
- preds_host = logits if preds_host is None else nested_concat(preds_host, logits, padding_index=-100)
- if labels is not None:
- labels_host = labels if labels_host is None else nested_concat(labels_host, labels, padding_index=-100)
- if inputs_decode is not None:
- inputs_host = (
- inputs_decode
- if inputs_host is None
- else nested_concat(inputs_host, inputs_decode, padding_index=-100)
- )
- self.control = self.callback_handler.on_prediction_step(args, self.state, self.control)
- if self.args.batch_eval_metrics:
- if self.compute_metrics is not None and preds_host is not None and labels_host is not None:
- is_last_step = self.accelerator.gradient_state.end_of_dataloader
- batch_kwargs = {}
- batch_kwargs["losses"] = losses_host if "loss" in args.include_for_metrics else None
- batch_kwargs["inputs"] = inputs_host if "inputs" in args.include_for_metrics else None
- metrics = self.compute_metrics(
- EvalPrediction(predictions=preds_host, label_ids=labels_host, **batch_kwargs),
- compute_result=is_last_step,
- )
- if self.args.batch_eval_metrics or (
- args.eval_accumulation_steps is not None and (step + 1) % args.eval_accumulation_steps == 0
- ):
- # Gather all tensors and put them back on the CPU if we have done enough accumulation steps.
- eval_losses_gatherer.add_arrays(self._gather_and_numpify(losses_host, "eval_losses"))
- if not prediction_loss_only:
- preds_gatherer.add_arrays(self._gather_and_numpify(preds_host, "eval_preds"))
- labels_gatherer.add_arrays(self._gather_and_numpify(labels_host, "eval_label_ids"))
- inputs_gatherer.add_arrays(self._gather_and_numpify(inputs_host, "eval_inputs_ids"))
- # Set back to None to begin a new accumulation
- del losses_host, preds_host, labels_host, inputs_host
- torch.cuda.empty_cache()
- losses_host, preds_host, labels_host, inputs_host = None, None, None, None
- if args.past_index and hasattr(self, "_past"):
- # Clean the state at the end of the evaluation loop
- delattr(self, "_past")
- # Gather all remaining tensors and put them back on the CPU
- eval_losses_gatherer.add_arrays(self._gather_and_numpify(losses_host, "eval_losses"))
- if not prediction_loss_only:
- preds_gatherer.add_arrays(self._gather_and_numpify(preds_host, "eval_preds"))
- labels_gatherer.add_arrays(self._gather_and_numpify(labels_host, "eval_label_ids"))
- inputs_gatherer.add_arrays(self._gather_and_numpify(inputs_host, "eval_inputs_ids"))
- eval_loss = eval_losses_gatherer.finalize()
- preds = preds_gatherer.finalize() if not prediction_loss_only else None
- label_ids = labels_gatherer.finalize() if not prediction_loss_only else None
- inputs_ids = inputs_gatherer.finalize() if not prediction_loss_only else None
- if (
- self.compute_metrics is not None
- and preds is not None
- and label_ids is not None
- and not self.args.batch_eval_metrics
- ):
- eval_set_kwargs["losses"] = eval_loss if "loss" in args.include_for_metrics else None
- eval_set_kwargs["inputs"] = inputs_ids if "inputs" in args.include_for_metrics else None
- metrics = self.compute_metrics(EvalPrediction(predictions=preds, label_ids=label_ids, **eval_set_kwargs))
- elif metrics is None:
- metrics = {}
- # To be JSON-serializable, we need to remove numpy types or zero-d tensors
- metrics = denumpify_detensorize(metrics)
- if eval_loss is not None:
- metrics[f"{metric_key_prefix}_loss"] = eval_loss.mean().item()
- # Prefix all keys with metric_key_prefix + '_'
- for key in list(metrics.keys()):
- if not key.startswith(f"{metric_key_prefix}_"):
- metrics[f"{metric_key_prefix}_{key}"] = metrics.pop(key)
- return EvalLoopOutput(predictions=preds, label_ids=label_ids, metrics=metrics, num_samples=num_examples)
- def _gather_and_numpify(self, tensors, name):
- """
- Gather value of `tensors` (tensor or list/tuple of nested tensors) and convert them to numpy before
- concatenating them to `gathered`
- """
- if tensors is None:
- return
- if is_torch_xla_available():
- tensors = nested_xla_mesh_reduce(tensors, name)
- elif is_sagemaker_mp_enabled():
- tensors = smp_gather(tensors)
- elif self.args.parallel_mode == ParallelMode.DISTRIBUTED:
- tensors = distributed_concat(tensors)
- return nested_numpify(tensors)
- def _add_sm_patterns_to_gitignore(self) -> None:
- """Add SageMaker Checkpointing patterns to .gitignore file."""
- # Make sure we only do this on the main process
- if not self.is_world_process_zero():
- return
- patterns = ["*.sagemaker-uploading", "*.sagemaker-uploaded"]
- # Get current .gitignore content
- if os.path.exists(os.path.join(self.repo.local_dir, ".gitignore")):
- with open(os.path.join(self.repo.local_dir, ".gitignore")) as f:
- current_content = f.read()
- else:
- current_content = ""
- # Add the patterns to .gitignore
- content = current_content
- for pattern in patterns:
- if pattern not in content:
- if content.endswith("\n"):
- content += pattern
- else:
- content += f"\n{pattern}"
- # Write the .gitignore file if it has changed
- if content != current_content:
- with open(os.path.join(self.repo.local_dir, ".gitignore"), "w") as f:
- logger.debug(f"Writing .gitignore file. Content: {content}")
- f.write(content)
- self.repo.git_add(".gitignore")
- # avoid race condition with git status
- time.sleep(0.5)
- if not self.repo.is_repo_clean():
- self.repo.git_commit("Add *.sagemaker patterns to .gitignore.")
- self.repo.git_push()
- def create_accelerator_and_postprocess(self):
- # We explicitly don't rely on the `Accelerator` to do gradient accumulation
- grad_acc_kwargs = {}
- if is_accelerate_available("0.28.0") and self.args.accelerator_config.gradient_accumulation_kwargs is not None:
- grad_acc_kwargs = self.args.accelerator_config.gradient_accumulation_kwargs
- # check if num_steps is attempted to be passed in gradient_accumulation_kwargs
- if "num_steps" in grad_acc_kwargs:
- if self.args.gradient_accumulation_steps > 1:
- # raise because we do not know which setting is intended.
- raise ValueError(
- "The `AcceleratorConfig`'s `num_steps` is set but `gradient_accumulation_steps` is greater than 1 in the passed `TrainingArguments`"
- "If using the passed `AcceleratorConfig` is desired, do not set the `TrainingArguments` `gradient_accumulation_steps`."
- )
- else:
- self.args.gradient_accumulation_steps = grad_acc_kwargs["num_steps"]
- accelerator_config = self.args.accelerator_config.to_dict()
- if is_accelerate_available("0.28.0"):
- # Extract dataloader config params from accelerator config
- dataloader_params = ["split_batches", "dispatch_batches", "even_batches", "use_seedable_sampler"]
- dataloader_config = DataLoaderConfiguration(
- **{param: accelerator_config.pop(param) for param in dataloader_params}
- )
- if is_accelerate_available("1.1.0"):
- dataloader_config.data_seed = self.args.data_seed
- non_blocking = accelerator_config.pop("non_blocking")
- if not is_accelerate_available("0.30.0"):
- if non_blocking:
- raise ImportError(
- "`non_blocking` is only supported in accelerate v0.30.0 and above. Please upgrade accelerate to use this feature."
- )
- else:
- if non_blocking and not self.args.dataloader_pin_memory:
- logger.warning(
- "`non_blocking` is enabled but `dataloader_pin_memory` is not. For the best performance, it's recommended to enable both."
- )
- dataloader_config.non_blocking = non_blocking
- # this would have been updated above, no need for it anymore
- accelerator_config.pop("gradient_accumulation_kwargs")
- args = {
- "deepspeed_plugin": self.args.deepspeed_plugin,
- }
- # We defer compatibility checks to accelerator
- if self.args.parallelism_config is not None:
- if not is_accelerate_available("1.10.1"):
- raise ImportError(
- "ParallelismConfig requires accelerate v1.10.1 and above. Please upgrade accelerate to use this feature."
- )
- args["parallelism_config"] = self.args.parallelism_config
- if is_accelerate_available("0.28.0"):
- args["dataloader_config"] = dataloader_config
- else:
- args.update(accelerator_config)
- # tp is initialized at Accelerator init phase so
- # args should be prepared here
- if hasattr(self.model, "tp_size") and self.model.tp_size is not None and self.model.tp_size > 1:
- self.is_tp_enabled = True
- if version.parse(accelerate_version) > version.parse("1.3.0"):
- args["torch_tp_plugin"] = TorchTensorParallelPlugin(tp_size=self.model.tp_size)
- else:
- raise ValueError("Requires accelerate>1.3.0 to use Tensor Parallelism.")
- # create accelerator object
- self.accelerator = Accelerator(**args)
- # some Trainer classes need to use `gather` instead of `gather_for_metrics`, thus we store a flag
- self.gather_function = self.accelerator.gather_for_metrics
- if "use_gather_object" in inspect.signature(self.gather_function).parameters:
- self.gather_function = functools.partial(
- self.gather_function, use_gather_object=self.args.eval_use_gather_object
- )
- # deepspeed and accelerate flags covering both trainer args and accelerate launcher
- self.is_deepspeed_enabled = getattr(self.accelerator.state, "deepspeed_plugin", None) is not None
- self.is_fsdp_enabled = getattr(self.accelerator.state, "fsdp_plugin", None) is not None
- self.is_tp_enabled = getattr(self.accelerator.state, "torch_tp_plugin", None) is not None
- # post accelerator creation setup
- if self.is_fsdp_enabled:
- fsdp_plugin = self.accelerator.state.fsdp_plugin
- for param in ["limit_all_gathers", "activation_checkpointing"]:
- setattr(fsdp_plugin, param, self.args.fsdp_config.get(param, getattr(fsdp_plugin, param)))
- if fsdp_plugin.activation_checkpointing and self.args.gradient_checkpointing:
- raise ValueError(
- "The activation_checkpointing in FSDP config and the gradient_checkpointing in training arg "
- "can't be set to True simultaneously. Please use FSDP's activation_checkpointing logic "
- "when using FSDP."
- )
- if self.is_deepspeed_enabled and getattr(self.args, "hf_deepspeed_config", None) is None:
- self.propagate_args_to_deepspeed()
- # `save_only_model` can't be used with DeepSpeed/FSDP along with `load_best_model_at_end`
- if (
- self.args.save_only_model
- and (self.is_deepspeed_enabled or self.is_fsdp_enabled)
- and self.args.load_best_model_at_end
- ):
- wrapper = "DeepSpeed" if self.is_deepspeed_enabled else "FSDP"
- raise ValueError(f"{wrapper} can't be used with `save_only_model` along with `load_best_model_at_end`.")
- # `auto_find_batch_size` isn't supported yet with DeepSpeed Zero-3
- if (
- self.is_deepspeed_enabled
- and self.accelerator.state.deepspeed_plugin.zero_stage == 3
- and self.args.auto_find_batch_size
- ):
- raise ValueError(
- "`auto_find_batch_size` isn't supported yet with DeepSpeed Zero-3. Please consider using Zero-2, Zero-1, or FSDP"
- )
- if (
- self.args.save_only_model
- and self.is_fsdp_enabled
- and "SHARDED_STATE_DICT" in str(self.accelerator.state.fsdp_plugin.state_dict_type)
- ):
- raise ValueError("save_only_model option is not compatible with FSDP state dict type 'SHARDED_STATE_DICT'")
- def propagate_args_to_deepspeed(self, auto_find_batch_size=False):
- """
- Sets values in the deepspeed plugin based on the Trainer args
- """
- from transformers.integrations.deepspeed import HfTrainerDeepSpeedConfig
- ds_plugin = self.accelerator.state.deepspeed_plugin
- ds_plugin.hf_ds_config = HfTrainerDeepSpeedConfig(ds_plugin.hf_ds_config.config)
- ds_plugin.deepspeed_config = ds_plugin.hf_ds_config.config
- ds_plugin.hf_ds_config.trainer_config_process(self.args, auto_find_batch_size)
- def _fsdp_qlora_plugin_updates(self):
- if self.is_fsdp_enabled and _is_peft_model(self.model):
- from peft import PeftConfig
- from peft.utils.other import fsdp_auto_wrap_policy
- if isinstance(self.model.active_peft_config, PeftConfig):
- self.accelerator.state.fsdp_plugin.auto_wrap_policy = fsdp_auto_wrap_policy(self.model)
- if (
- getattr(self.model, "quantization_method", None) == QuantizationMethod.BITS_AND_BYTES
- and self.model.hf_quantizer.quantization_config.bnb_4bit_quant_storage.is_floating_point
- and version.parse(accelerate_version) > version.parse("0.27.0")
- ):
- self.accelerator.state.fsdp_plugin.set_mixed_precision(
- self.model.hf_quantizer.quantization_config.bnb_4bit_quant_storage, override=True
- )
- def _get_num_items_in_batch(self, batch_samples: list, device: torch.device) -> Optional[Union[torch.Tensor, int]]:
- """
- Counts the number of items in the batches to properly scale the loss.
- Args:
- batch_samples (`list`): List of batches
- device (`torch.device`): The device on which the number of items in the batch should be.
- Returns:
- None if the number of items in the batch doesn't need to be computed else the number of items in the batch
- """
- num_items_in_batch = None
- count_num_items_in_batch = (
- len(batch_samples) > 0
- and "labels" in batch_samples[0]
- and (
- # num_items_in_batch is passed to model forward
- # https://github.com/huggingface/transformers/blob/v4.49.0/src/transformers/trainer.py#L3757
- self.model_accepts_loss_kwargs
- # num_items_in_batch is passed to compute_loss_func
- # https://github.com/huggingface/transformers/blob/v4.49.0/src/transformers/trainer.py#L3773
- or self.compute_loss_func is not None
- # num_items_in_batch is also verified if (self.model_accepts_loss_kwargs or self.compute_loss_func)
- # https://github.com/huggingface/transformers/blob/v4.49.0/src/transformers/trainer.py#L3790
- )
- )
- if count_num_items_in_batch:
- # For now we don't support object detection
- try:
- num_items_in_batch = sum((batch["labels"].ne(-100)).sum() for batch in batch_samples)
- except (TypeError, AttributeError):
- pass
- if num_items_in_batch is not None:
- if self.args.average_tokens_across_devices and self.args.world_size >= 1:
- num_items_in_batch = self.accelerator.gather(num_items_in_batch.to(device)).sum()
- elif self.args.n_gpu >= 1:
- # In DP case, if we don't average, we need to divide by the number of gpu. This is the simplest approximation.
- # Otherwise, we would have to scatter labels and calculate num_items_in_batch for each gpu.
- num_items_in_batch = num_items_in_batch // self.args.n_gpu
- if torch.is_tensor(num_items_in_batch):
- num_items_in_batch = num_items_in_batch.to(device)
- if self.args.n_gpu > 1 and num_items_in_batch.dim() == 0:
- # In the DataParallel case, convert the scalar tensor into a 2-dim tensor with the same value repeated
- num_items_in_batch = num_items_in_batch.unsqueeze(0).expand(self.args.n_gpu, -1)
- # Divide by number of devices with the same batch
- if pc := getattr(self.accelerator, "parallelism_config", None):
- num_items_in_batch = num_items_in_batch // pc.non_data_parallel_size
- return num_items_in_batch
- def get_batch_samples(
- self, epoch_iterator: Iterator, num_batches: int, device: torch.device
- ) -> tuple[list, Optional[Union[torch.Tensor, int]]]:
- """
- Collects a specified number of batches from the epoch iterator and optionally counts the number of items in the batches to properly scale the loss.
- """
- batch_samples = []
- for _ in range(num_batches):
- try:
- batch_samples.append(next(epoch_iterator))
- except StopIteration:
- break
- num_items_in_batch = self._get_num_items_in_batch(batch_samples, device)
- return batch_samples, num_items_in_batch
- def set_initial_training_values(
- self, args: TrainingArguments, dataloader: DataLoader, total_train_batch_size: int
- ):
- """
- Calculates and returns the following values:
- - `num_train_epochs`
- - `num_update_steps_per_epoch`
- - `num_examples`
- - `num_train_samples`
- - `epoch_based`
- - `len_dataloader`
- - `max_steps`
- """
- # Case 1: we rely on `args.max_steps` first
- max_steps = args.max_steps
- # If max_steps is negative, we use the number of epochs to determine the number of total steps later
- epoch_based = max_steps < 0
- len_dataloader = len(dataloader) if has_length(dataloader) else None
- # Case 2: We have a dataloader length and can extrapolate
- if len_dataloader is not None:
- num_update_steps_per_epoch = max(
- len_dataloader // args.gradient_accumulation_steps
- + int(len_dataloader % args.gradient_accumulation_steps > 0),
- 1,
- )
- # Case 3: We have a length but are using epochs, we can extrapolate the number of steps
- if epoch_based:
- max_steps = math.ceil(args.num_train_epochs * num_update_steps_per_epoch)
- # Now we figure out `num_examples`, `num_train_epochs`, and `train_samples`
- if len_dataloader:
- num_examples = self.num_examples(dataloader)
- if args.max_steps > 0:
- num_train_epochs = max_steps // num_update_steps_per_epoch + int(
- max_steps % num_update_steps_per_epoch > 0
- )
- # May be slightly incorrect if the last batch in the training dataloader has a smaller size but it's
- # the best we can do.
- num_train_samples = max_steps * total_train_batch_size
- else:
- num_train_epochs = math.ceil(args.num_train_epochs)
- num_train_samples = self.num_examples(dataloader) * args.num_train_epochs
- elif args.max_steps > 0: # Rely on max_steps when dataloader does not have a working size
- # Setting a very large number of epochs so we go as many times as necessary over the iterator.
- num_train_epochs = sys.maxsize
- num_update_steps_per_epoch = max_steps
- num_examples = total_train_batch_size * args.max_steps
- num_train_samples = args.max_steps * total_train_batch_size
- else:
- raise ValueError(
- "args.max_steps must be set to a positive value if dataloader does not have a length, was"
- f" {args.max_steps}"
- )
- return (
- num_train_epochs,
- num_update_steps_per_epoch,
- num_examples,
- num_train_samples,
- epoch_based,
- len_dataloader,
- max_steps,
- )
|