importar clases de datos
importar conjuntos de datos
importar antorcha
importar antorcha.nn como nn
importar tqdm
@clases de datos.clase de datos
clase BertConfig:
“”“Configuración para el modelo BERT”.“”
tamaño_vocab: entero = 30522
num_capas: entero = 12
tamaño_oculto: entero = 768
numero_cabezas: entero = 12
problema_deserción: flotar = 0.1
pad_id: entero = 0
max_seq_len: entero = 512
núm_tipos: entero = 2
clase BertBlock(nn.Módulo):
“”“Un bloque transformador en BERT”.“”
definición __inicio__(ser, tamaño_oculto: entero, numero_cabezas: entero, problema_deserción: flotar):
súper().__inicio__()
ser.atención = nn.Atención multicabezal(tamaño_oculto, numero_cabezas,
Abandonar=problema_deserción, lote_primero=Verdadero)
ser.atención_norm = nn.Norma de capa(tamaño_oculto)
ser.ff_norm = nn.Norma de capa(tamaño_oculto)
ser.Abandonar = nn.Abandonar(problema_deserción)
ser.feed_forward = nn.Secuencial(
nn.Lineal(tamaño_oculto, 4 * tamaño_oculto),
nn.GELU(),
nn.Lineal(4 * tamaño_oculto, tamaño_oculto),
)
definición adelante(ser, incógnita: antorcha.Tensor, máscara_pad: antorcha.Tensor) -> antorcha.Tensor:
# autoatención con mascarilla acolchada y post-norma
atención_salida, _ = ser.atención(incógnita, incógnita, incógnita, máscara_padding_key=máscara_pad)
incógnita = ser.atención_norm(incógnita + atención_salida)
# feed-forward con activación GeLU y post-norma
ff_salida = ser.feed_forward(incógnita)
incógnita = ser.ff_norm(incógnita + ser.Abandonar(ff_salida))
devolver incógnita
clase BertPooler(nn.Módulo):
“”“Capa de pooler para que BERT procese el [CLS] salida de token.”“”
definición __inicio__(ser, tamaño_oculto: entero):
súper().__inicio__()
ser.denso = nn.Lineal(tamaño_oculto, tamaño_oculto)
ser.activación = nn.tanh()
definición adelante(ser, incógnita: antorcha.Tensor) -> antorcha.Tensor:
incógnita = ser.denso(incógnita)
incógnita = ser.activación(incógnita)
devolver incógnita
clase BertModel(nn.Módulo):
“”“La columna vertebral del modelo BERT”.“”
definición __inicio__(ser, configuración: BertConfig):
súper().__inicio__()
# capas de incrustación
ser.incrustaciones de palabras = nn.incrustar(configuración.tamaño_vocab, configuración.tamaño_oculto,
padding_idx=configuración.pad_id)
ser.tipo_incrustaciones = nn.incrustar(configuración.núm_tipos, configuración.tamaño_oculto)
ser.posiciones_incrustaciones = nn.incrustar(configuración.max_seq_len, configuración.tamaño_oculto)
ser.incrustaciones_norm = nn.Norma de capa(configuración.tamaño_oculto)
ser.incrustaciones_dropout = nn.Abandonar(configuración.problema_deserción)
# bloques transformadores
ser.bloques = nn.Lista de módulos([
BertBlock(config.hidden_size, config.num_heads, config.dropout_prob)
for _ in range(config.num_layers)
])
# [CLS] capa de pooler
ser.piscina = BertPooler(configuración.tamaño_oculto)
definición adelante(ser, ids_entrada: antorcha.Tensor, identificadores_tipo_token: antorcha.Tensor, pad_id: entero = 0
) -> tupla[torch.Tensor, torch.Tensor]:
# crear una máscara de atención para los tokens de relleno
máscara_pad = ids_entrada == almohadilla_identificación
# convertir tokens enteros en vectores integrados
tamaño_lote, seq_len = ids_entrada.forma
identificadores de posición = antorcha.organizar(seq_len, dispositivo=ids_entrada.dispositivo).descomprimir(0)
posiciones_incrustaciones = ser.posiciones_incrustaciones(identificadores de posición)
tipo_incrustaciones = ser.tipo_incrustaciones(identificadores_tipo_token)
incrustaciones_token = ser.incrustaciones de palabras(ids_entrada)
incógnita = incrustaciones_token + tipo_incrustaciones + posición_incrustaciones
incógnita = ser.incrustaciones_norm(incógnita)
incógnita = ser.incrustaciones_dropout(incógnita)
# procesar la secuencia con bloques transformadores
para bloquear en ser.bloques:
incógnita = bloquear(incógnita, máscara_pad)
# agrupar el estado oculto de `[CLS]` ficha
salida_agrupada = ser.piscina(incógnita[:, 0, :])
devolver incógnita, salida_agrupada
clase BertPreentrenamientoModelo(nn.Módulo):
definición __inicio__(ser, configuración: BertConfig):
súper().__inicio__()
ser.berto = BertModel(configuración)
ser.mlm_cabeza = nn.Secuencial(
nn.Lineal(configuración.tamaño_oculto, configuración.tamaño_oculto),
nn.GELU(),
nn.Norma de capa(configuración.tamaño_oculto),
nn.Lineal(configuración.tamaño_oculto, configuración.tamaño_vocab),
)
ser.nsp_cabeza = nn.Lineal(configuración.tamaño_oculto, 2)
definición adelante(ser, ids_entrada: antorcha.Tensor, identificadores_tipo_token: antorcha.Tensor, pad_id: entero = 0
) -> tupla[torch.Tensor, torch.Tensor]:
# Procesar la secuencia con la columna vertebral del modelo BERT
incógnita, salida_agrupada = ser.berto(ids_entrada, identificadores_tipo_token, pad_id)
# Predecir los tokens enmascarados para la tarea MLM y la clasificación para la tarea NSP
mlm_logits = ser.mlm_cabeza(incógnita)
nsp_logits = ser.nsp_cabeza(salida_agrupada)
devolver mlm_logits, nsp_logits
# Parámetros de entrenamiento
épocas = 10
tasa_de_aprendizaje = 1e–4
tamaño_lote = 32
# Cargar conjunto de datos y configurar el cargador de datos
conjunto de datos = conjuntos de datos.Conjunto de datos.desde_parquet(“wikitext-2_train_data.parquet”)
definición intercalar_fn(lote: lista[dict]):
“”“Función de clasificación personalizada para manejar secuencias de longitud variable en un conjunto de datos”.“”
# siempre con longitud máxima: tokens, segment_ids; siempre singleton: is_random_next
ids_entrada = antorcha.tensor([item[“tokens”] para artículo en lote])
identificadores_tipo_token = antorcha.tensor([item[“segment_ids”] para artículo en lote]).abdominales()
es_aleatorio_siguiente = antorcha.tensor([item[“is_random_next”] para artículo en lote]).a(entero)
# longitud variable: posiciones_enmascaradas, etiquetas_enmascaradas
pos_enmascarado = [(idx, pos) for idx, item in enumerate(batch) for pos in item[“masked_positions”]]
etiquetas_enmascaradas = antorcha.tensor([label for item in batch for label in item[“masked_labels”]])
devolver ids_entrada, identificadores_tipo_token, es_aleatorio_siguiente, pos_enmascarado, etiquetas_enmascaradas
cargador de datos = antorcha.utiles.datos.Cargador de datos(conjunto de datos, tamaño_lote=tamaño_lote, barajar=Verdadero,
intercalar_fn=intercalar_fn, numero_trabajadores=8)
# entrenar el modelo
dispositivo = antorcha.dispositivo(“cuda” si antorcha.cuda.está_disponible() demás “UPC”)
modelo = BertPreentrenamientoModelo(BertConfig()).a(dispositivo)
modelo.tren()
optimizador = antorcha.optimo.AdamW(modelo.parámetros(), lr=tasa_de_aprendizaje)
planificador = antorcha.optimo.lr_programador.PasoLR(optimizador, tamaño_paso=1, gama=0.1)
pérdida_fn = nn.Pérdida de entropía cruzada()
para época en rango(épocas):
pbar = tqdm.tqdm(cargador de datos, desc=F“Época {época+1}/{épocas}”)
para lote en pbar:
# obtener datos por lotes
ids_entrada, identificadores_tipo_token, es_aleatorio_siguiente, pos_enmascarado, etiquetas_enmascaradas = lote
ids_entrada = ids_entrada.a(dispositivo)
identificadores_tipo_token = identificadores_tipo_token.a(dispositivo)
es_aleatorio_siguiente = es_aleatorio_siguiente.a(dispositivo)
etiquetas_enmascaradas = etiquetas_enmascaradas.a(dispositivo)
# extraer la salida del modelo
mlm_logits, nsp_logits = modelo(ids_entrada, identificadores_tipo_token)
# Pérdida de MLM: posiciones_enmascaradas es una lista de tuplas de (B, S), extrae el
# logits correspondientes del tensor mlm_logits de forma (B, S, V)
índices_por lotes, posiciones_token = cremallera(*pos_enmascarado)
mlm_logits = mlm_logits[batch_indices, token_positions]
mlm_loss = pérdida_fn(mlm_logits, etiquetas_enmascaradas)
# Calcular la pérdida de la tarea NSP
nsp_loss = pérdida_fn(nsp_logits, es_aleatorio_siguiente)
# hacia atrás con pérdida total
pérdida_total = mlm_loss + nsp_loss
pbar.conjunto_postfix(multinivel=mlm_loss.artículo(), NSP=nsp_loss.artículo(), Total=pérdida_total.artículo())
optimizador.cero_grado()
pérdida_total.hacia atrás()
optimizador.paso()
planificador.paso()
pbar.actualizar(1)
pbar.cerca()
# Guardar el modelo
antorcha.ahorrar(modelo.dictado_estado(), “bert_pretraining_model.pth”)
antorcha.ahorrar(modelo.berto.dictado_estado(), “bert_model.pth”)