importar clases de datos
importar sistema operativo
importar conjuntos de datos
importar tqdm
importar tokenizadores
importar antorcha
antorcha de importación . nn como nn
antorcha de importación . nn . funcional como F
antorcha de importación . optimo . lr_scheduler como lr_programador
desde la importación de antorcha Tensor
# Cargar el tokenizador
tokenizador = tokenizadores . Tokenizador . from_file ( "bpe_50K.json" )
# Cargar el conjunto de datos
conjunto de datos = conjuntos de datos . load_dataset ( "HuggingFaceFW/fineweb" , "muestra-10BT" , dividir = "tren" )
# Construye el modelo
@ clases de datos . clase de datos
clase Configuración de llama :
"" "Definir los hiperparámetros del modelo Llama." ""
tamaño_vocab : entero = 50000 # Tamaño del vocabulario del tokenizador
max_position_embeddings : entero = 2048 # Longitud máxima de secuencia
tamaño_oculto : entero = 768 # Dimensión de capas ocultas
tamaño_intermedio : entero = 4 * 768 # Dimensión de la capa oculta de MLP
num_capas_ocultas : entero = 12 # Número de capas del transformador
num_attention_heads : entero = 12 # Número de cabezas de atención
num_key_value_heads : entero = 3 # Número de cabezas clave-valor para GQA
def rotar_mitad ( x : tensor ) -> Tensores :
"" "Gira la mitad de los oscurecimientos ocultos de la entrada.
Esta es una función auxiliar para incrustaciones de posiciones rotativas (RoPE).
Para un tensor de forma (…, d), devuelve un tensor donde el último
Las dimensiones d/2 se rotan intercambiando y negando.
Argumentos:
x: tensor de entrada de forma (…, d)
Devoluciones:
Tensor de la misma forma con la última dimensión girada
" ""
x1 , x2 = x . trozo ( 2 , tenue = – 1 )
devolver antorcha . gato ( ( – x2 , x1 ) , tenue = – 1 ) # Concatenar con rotación
clase RotaryPositionEncoding ( nn . Módulo ) :
"" "Codificación de posición giratoria." ""
def __init__ ( yo , tenue : entero , max_position_embeddings : int ) -> Ninguno :
"" "Inicializar el módulo RotaryPositionEncoding
Argumentos:
tenue: la dimensión oculta del tensor de entrada al que se aplica RoPE
max_position_embeddings: la longitud máxima de secuencia del tensor de entrada
" ""
súper ( ) . __inicio__ ( )
ser . oscuro = oscuro
ser . max_position_embeddings = max_position _ incrustaciones
# calcular una matriz de ntheta_i
norte = 10_000.0
frecuencia_inv = 1.0 / ( N * * ( antorcha . arange ( 0 , tenue , 2 ) . flotar ( ) / tenue ) )
frecuencia_inv = antorcha . gato ( ( inv_freq , inv_freq ) , tenue = – 1 )
posición = antorcha . arange ( max_position_embeddings ) . flotar ( )
sinusoide_inp = antorcha . exterior ( posición , frecuencia_inv )
# guardar matrices coseno y seno como buffers, no parámetros
ser . registro_buffer ( "cos" , sinusoid_inp . porque ( ) )
ser . registrarse_buffer ( "pecado" , sinusoid_inp . pecado ( ) )
def adelante ( yo , x : tensor ) -> Tensores :
"" "Aplicar RoPE al tensor x
Argumentos:
x: tensor de entrada de forma (batch_size, seq_length, num_heads, head_dim)
Devoluciones:
Tensor de forma de salida (batch_size, seq_length, num_heads, head_dim)
" ""
tamaño_lote , secuencia_len , numero_cabezas , cabeza_dim = x . forma
tipo d = x . tipo d
# transforma las matrices coseno y seno a tensor 4D y el mismo tipo d que x
porque = ser . porque . a ( tipod ) [ : seq_len ] . ver ( 1 , secuencia_len , 1 , – 1 )
pecado = ser . pecado . a ( tipod ) [ : seq_len ] . ver ( 1 , secuencia_len , 1 , – 1 )
# aplicar RoPE a x
producción = ( x * porque ) + ( rotar_mitad ( x ) * pecado )
devolver producción
clase LlamaAttention ( nn . Módulo ) :
"" "Atención de consultas agrupadas con incrustaciones rotativas." ""
def __init__ ( yo , configuración : LlamaConfig ) -> Ninguno :
súper ( ) . __inicio__ ( )
ser . tamaño_oculto = configuración . tamaño_oculto
ser . numero_cabezas = configuración . num_atención_cabezas
ser . cabeza_dim = ser . tamaño_oculto // self.num_heads
ser . num_kv_cabezas = configuración . num_key_value _ cabezas # GQA: H_kv < H_q
# tamaño_oculto debe ser divisible por num_heads
afirmar ( yo . head_dim * ser . núm_cabezas ) == ser . oculto _ tamaño
# Capas lineales para proyecciones Q, K, V
ser . q_proyecto = nn . Lineal ( self . tamaño_oculto , ser . numero_cabezas * ser . cabeza_tenue , sesgo = falso )
ser . k_proj = nn . Lineal ( self . tamaño_oculto , ser . num_kv_heads * ser . cabeza_tenue , sesgo = falso )
ser . v_proyecto = nn . Lineal ( self . tamaño_oculto , ser . num_kv_heads * ser . cabeza_tenue , sesgo = falso )
ser . o_proyecto = nn . Lineal ( self . num_heads * ser . cabeza_tenue , ser . tamaño_oculto , sesgo = falso )
def adelante ( yo , estados_ocultos : tensores , soga : Codificación de posición rotativa , attn_mask : tensor ) -> Tensores :
bs , secuencia_len , oscuro = estados_ocultos . tamaño ( )
# Entradas del proyecto a Q, K, V
estados_consulta = ser . q_proj ( estados_ocultos ) . ver ( bs , secuencia_len , ser . numero_cabezas , ser . cabeza_dim )
estados_clave = ser . k_proj ( estados_ocultos ) . ver ( bs , secuencia_len , ser . num_kv_heads , ser . cabeza_dim )
estados_valor = ser . v_proj ( estados_ocultos ) . ver ( bs , secuencia_len , ser . num_kv_heads , ser . cabeza_dim )
# Aplicar incrustaciones de posición rotatoria.
estados_consulta = cuerda ( query_states )
estados_clave = cuerda ( key_states )
# Transponer tensores de la dimensión BSHD a BHSD para scaled_dot_product_attention
estados_consulta = estados_consulta . transponer ( 1 , 2 )
estados_clave = estados_clave . transponer ( 1 , 2 )
estados_valor = estados_valor . transponer ( 1 , 2 )
# Utilice la implementación de atención optimizada de PyTorch
# configuración is_causal=True es incompatible con establecer una máscara de atención explícita
atención_salida = F. atención_producto_punto_escalado (
estados_consulta ,
estados_clave ,
estados_valor ,
máscara_atención = máscara_atención ,
abandono_p = 0.0 ,
enable_gqa = Verdadero ,
)
# Transponer el tensor de salida de BHSD a la dimensión BSHD, remodelar a 3D y luego proyectar la salida
atención_salida = atención_salida . transponer ( 1 , 2 ) . remodelar ( bs , secuencia_len , ser . tamaño_oculto )
atención_salida = ser . o_proj ( attn_output )
devolver atención_salida
clase LlamaMLP ( nn . Módulo ) :
"" "Red feed-forward con activación SwiGLU." ""
def __init__ ( yo , configuración : LlamaConfig ) -> Ninguno :
súper ( ) . __inicio__ ( )
# Dos proyecciones paralelas para SwiGLU
ser . puerta_proj = nn . Lineal ( config . tamaño_oculto , configuración . tamaño_intermedio , sesgo = falso )
ser . up_proyecto = nn . Lineal ( config . tamaño_oculto , configuración . tamaño_intermedio , sesgo = falso )
ser . actuar_fn = F. silú # Función de activación SwiGLU
# Proyectar de nuevo al tamaño oculto
ser . down_proj = nn . Lineal ( config . tamaño_intermedio , configuración . tamaño_oculto , sesgo = falso )
def adelante ( yo , x : tensor ) -> Tensores :
# Activación SwiGLU: puerta multiplicada y entradas proyectadas hacia arriba
puerta = ser . act_fn ( self . gate_proj ( x ) )
arriba = ser . up_proj ( x )
devolver ser . down_proj ( puerta * arriba )
clase LlamaDecoderLayer ( nn . Módulo ) :
"" "Capa de transformador única para un modelo Llama." ""
def __init__ ( yo , configuración : LlamaConfig ) -> Ninguno :
súper ( ) . __inicio__ ( )
ser . norma_capa_entrada = nn . RMSNorm ( config . tamaño_oculto , pasos = 1e – 5 )
ser . atención propia = LlamaAtención ( config )
ser . post_attention_layernorm = nn . RMSNorm ( config . tamaño_oculto , pasos = 1e – 5 )
ser . mlp = LlamaMLP ( configuración )
def adelante ( yo , estados_ocultos : tensores , soga : Codificación de posición rotativa , attn_mask : tensor ) -> Tensores :
# Primer bloque residual: Autoatención
residual = estados_ocultos
estados_ocultos = ser . input_layernorm ( estados_ocultos )
atención_salidas = ser . self_attn ( estados_ocultos , cuerda = cuerda , máscara_atención = máscara_atención )
estados_ocultos = atención_salidas + residual
# Segundo bloque residual: MLP
residual = estados_ocultos
estados_ocultos = ser . post_attention_layernorm ( estados_ocultos )
estados_ocultos = ser . mlp ( estados_ocultos ) + residual
devolver estados_ocultos
clase LlamaModel ( nn . Módulo ) :
"" "El modelo Llama completo sin cabezas de entrenamiento previo." ""
def __init__ ( yo , configuración : LlamaConfig ) -> Ninguno :
súper ( ) . __inicio__ ( )
ser . rotativo_emb = Codificación de posición rotativa (
configuración . tamaño_oculto // config.num_attention_heads,
configuración . max_position_embeddings ,
)
ser . tokens_incrustados = nn . Incrustar ( config.vocab_size , configuración . tamaño_oculto )
ser . capas = nn . ListaDeMódulos ( [ LlamaDecoderLayer ( configuración ) para _ en rango ( config . num_hidden_layers ) ] )
ser . norma = nn . RMSNorm ( config . tamaño_oculto , pasos = 1e – 5 )
def adelante ( yo , identificadores de entrada : tensores , attn_mask : tensor ) -> Tensores :
# Convertir ID de token de entrada en incrustaciones
estados_ocultos = ser . incrustar_tokens ( id_entrada )
# Procese a través de todas las capas del transformador, luego la capa de norma final
para capa en ser . capas :
estados_ocultos = capa ( estados_ocultos , cuerda = yo . rotativo_emb , máscara_atención = máscara_atención )
estados_ocultos = ser . norma ( estados_ocultos )
# Devolver los estados ocultos finales
devolver estados_ocultos
clase LlamaForPretraining ( nn . Módulo ) :
def __init__ ( yo , configuración : LlamaConfig ) -> Ninguno :
súper ( ) . __inicio__ ( )
ser . modelo_base = LlamaModel ( configuración )
ser . lm_cabeza = nn . Lineal ( config . tamaño_oculto , configuración . tamaño_vocab , sesgo = falso )
def adelante ( yo , identificadores de entrada : tensores , attn_mask : tensor ) -> Tensores :
estados_ocultos = ser . modelo_base ( id_entrada , atención_máscara )
devolver ser . lm_head ( estados_ocultos )
def create_causal_mask ( seq_len : entero , dispositivo : antorcha . dispositivo , tipo de : antorcha . tipo d = antorcha . flotador32 ) -> Tensores :
"" "Crea una máscara causal para la autoatención.
Argumentos:
seq_len: Longitud de la secuencia
dispositivo: Dispositivo para crear la máscara
dtype: tipo de datos de la máscara
Devoluciones:
Máscara causal de forma (seq_len, seq_len)
" ""
mascarilla = antorcha . completo ( ( seq_len , seq_len ) , flotador ( '-inf' ) , dispositivo = dispositivo , dtipo = dtipo )
. triu ( diagonal = 1 )
devolver mascarilla
def create_padding_mask ( lote , padding_token_id , dispositivo : antorcha . dispositivo , tipo de : antorcha . tipo d = antorcha . flotador32 ) -> Tensores :
"" "Cree una máscara de relleno para un lote de secuencias para la autoatención.
Argumentos:
lote: lote de secuencias, forma (batch_size, seq_len)
padding_token_id: ID del token de relleno
Devoluciones:
Máscara de relleno de forma (batch_size, 1, seq_len, seq_len)
" ""
acolchado = antorcha . zeros_like ( lote , dispositivo = dispositivo , dtipo = dtipo )
. relleno_enmascarado ( lote == padding_token_id , flotante ( '-inf' ) )
mascarilla = acolchado [ : , : , Ninguno ] + acolchado [ : , Ninguno , : ]
devolver mascarilla [ : , Ninguno , : , : ]
# Función generadora para crear secuencias acolchadas de longitud fija.
clase PretrainingBatchDataset ( antorcha . utils . data . Dataset ) :
def __init__ ( yo , conjunto de datos : conjuntos de datos . conjunto de datos , tokenizador : tokenizadores . tokenizador ,
longitud_secuencia : entero , tamaño_lote : entero , dispositivo : antorcha . dispositivo = Ninguno ) :
ser . conjunto de datos = conjunto de datos
ser . tokenizador = tokenizador
ser . dispositivo = dispositivo
ser . longitud_secuencia = longitud_secuencia
ser . tamaño_lote = tamaño_lote
ser . bot = tokenizador . token_to_id ( "[BOT]" )
ser . eot = tokenizador . token_to_id ( "[EOT]" )
ser . almohadilla = tokenizador . token_to_id ( "[PAD]" )
def __len__ ( yo ) :
devolver len ( self . conjunto de datos )
def __getitem__ ( yo , índice ) :
"" "Obtiene una secuencia de identificadores de tokens del conjunto de datos. Tokens [BOT] y [EOT]
se agregan. Recortado y acolchado según la longitud de la secuencia.
" ""
secuencia = ser . conjunto de datos [ índice ] [ "texto" ]
fichas : lista [ int ] = [ ser . bot ] + ser . tokenizador . codificar ( sec ) . identificaciones + [ ser . yot ]
# pad para apuntar a la longitud de la secuencia
toklen = len ( fichas )
si toklen < ser . longitud_secuencia + 1 :
longitud_almohadilla = ser . longitud_secuencia + 1 – toklen
fichas += [ ser . almohadilla ] * almohadilla _ longitud
# devolver la secuencia
incógnita = antorcha . tensor ( tokens [ : self.seq_length ] , tipo d = antorcha . int64 , dispositivo = yo . dispositivo )
y = antorcha . tensor ( tokens [ 1 : self . seq_length + 1 ] , tipo d = antorcha . int64 , dispositivo = yo . dispositivo )
devolver x , y
# Crear modelo de preentrenamiento con configuración predeterminada
prueba_config = LlamaConfig ( )
dispositivo = antorcha . dispositivo ( "cuda" ) si antorcha . cuda . está_disponible ( ) demás antorcha . dispositivo ( "cpu" )
modelo = LlamaForPretraining ( test_config ) . a ( dispositivo )
# imprimir el tamaño del modelo
print ( f "Tamaño de los parámetros del modelo: {sum(p.numel() for p in model.parameters()) / 1024**2:.2f} M" )
print ( f "Tamaño de los buffers del modelo: {sum(p.numel() for p in model.buffers()) / 1024**2:.2f} M" )
# Parámetros de entrenamiento
épocas = 3
tasa_de_aprendizaje = 1e – 3
tamaño_lote = 8
longitud_secuencia = 512
num_pasos_de_calentamiento = 1000
PAD_TOKEN_ID = tokenizador . token_to_id ( "[PAD]" )
# DataLoader, optimizador, programador y función de pérdida
modelo . tren ( )
cargador de datos = antorcha . utiles . datos . Cargador de datos (
PretrainingBatchDataset ( conjunto de datos , tokenizador , longitud_secuencia , tamaño_lote , dispositivo ) ,
tamaño_lote = tamaño_lote
)
optimizador = antorcha . optimo . AdánW (
modelo . parámetros ( ) , lr = tasa_de_aprendizaje , betas = ( 0,9 , 0,95 ) , pasos = 1e – 8 , decaimiento_peso = 0,01
)
num_pasos_de_formación = len ( cargador de datos ) * épocas
programador_calentamiento = lr_programador . LinealLR (
optimizador ,
factor_inicial = 0,1 , factor_final = 1.0 , total_iters = num_calentamiento _ pasos
)
programador_coseno = lr_programador . CosenoRecocidoLR (
optimizador ,
T_max = núm_pasos_de_entrenamiento – num_pasos_de calentamiento ,
eta_mín = 0
)
planificador = lr_programador . SecuencialLR (
optimizador ,
programadores = [ warmup_scheduler , programador_coseno ] ,
hitos = [ num_warmup_steps ]
)
pérdida_fn = nn . CrossEntropyLoss ( ignorar_index = PAD_TOKEN_ID )
# busca el último punto de control
si os . camino . existe ( "llama_pretraining_checkpoint.pth" ) :
control = antorcha . cargar ( "llama_pretraining_checkpoint.pth" )
comenzar_época = punto de control [ "época" ]
comenzar_lote = punto de control [ "lote" ]
modelo . load_state_dict ( punto de control [ "modelo" ] )
optimizador . load_state_dict ( punto de control [ "optimizador" ] )
planificador . load_state_dict ( punto de control [ "programador" ] )
del puesto de control
print ( f "Reanudando el entrenamiento desde la época {begin_epoch} y el lote {begin_batch}" )
demás :
comenzar_época = 0
comenzar_lote = 0
# empezar a entrenar
para época en rango ( empezar_época , épocas ) :
cargador de datos = antorcha . utiles . datos . Cargador de datos (
Conjunto de datos por lotes previo al entrenamiento (
conjunto de datos . omitir ( comenzar_lote * tamaño_lote ) ,
tokenizador ,
longitud_secuencia ,
tamaño_lote ,
dispositivo ,
) ,
tamaño_lote = tamaño_lote
)
pbar = tqdm . tqdm ( cargador de datos , desc = f "Época {época+1}/{épocas}" )
para id_lote , lote en enumerar ( pbar ) :
si ( comenzar_lote + id_lote ) % 1000 == 0 :
# controla el modelo y el estado del optimizador
antorcha . ahorrar ( {
"modelo" : modelo . dictado_de_estado ( ) ,
"optimizador" : optimizador . dictado_de_estado ( ) ,
"programador" : planificador . dictado_de_estado ( ) ,
"época" : época ,
"lote" : id_lote + comenzar_lote ,
} , f "llama_pretraining_checkpoint.pth" )
# obtener datos por lotes
identificadores de entrada , identificadores_objetivos = lote
# crear máscara de atención: máscara causal + máscara de relleno
atención_mascarilla = create_causal_mask ( input_ids . forma [ 1 ] , dispositivo ) +
crear_padding_mask ( entrada_ids , PAD_TOKEN_ID , dispositivo )
# extraer la salida del modelo
logits = modelo ( input_ids , atención_máscara )
# pérdida de cálculo: entropía cruzada entre logits y el objetivo, ignorando los tokens de relleno
pérdida = loss_fn ( logits . ver ( – 1 , logits . tamaño ( -1 ) ) , identificadores_objetivo . ver ( – 1 ) )
# hacia atrás con pérdida
optimizador . graduación_cero ( )
pérdida . hacia atrás ( )
optimizador . paso ( )
planificador . paso ( )
pbar . set_postfix ( pérdida = pérdida . artículo ( ) )
pbar . actualizar ( 1 )
comenzar_lote = 0 # restablecer para la próxima época
pbar . cerca ( )
# Guardar el modelo
antorcha . guardar ( modelo . state_dict ( ) , "llama_pretraining_model.pth" )
antorcha . guardar ( modelo.base_model.state_dict ( ) , "llama_model.pth" )