"" "
Procesamiento por lotes continuo = programación a nivel de iteración + procesamiento por lotes irregular (empaquetado).
Se comparan dos enfoques (ambos ejecutan secuencias BATCH_SIZE simultáneamente, por lo que el
la comparación es justa entre tragamonedas):
1. Lotes estáticos (línea de base):
Las solicitudes se procesan BATCH_SIZE a la vez. Cada ola se rellena hasta un
longitud común y ejecutar juntos hasta la solicitud MÁS LARGA en esa ola
acabados; una dura " barrera de lotes " debe desaparecer antes de la próxima ola
comienza. Las solicitudes breves permanecen inactivas detrás de la barrera.
2. Lotes continuos (alineados con la producción):
Se combinan dos ideas para mantener ocupada la GPU.
(a) Programación a nivel de iteración: en el momento en que termina una secuencia, se libera
su ranura y el siguiente mensaje en cola se admite en el MISMO paso:
No hay que esperar al resto del lote.
(b) Lote irregular/empacado: la parte que lo hace verdaderamente " continuo ":
en lugar de rellenar cada secuencia en un rectángulo [B, max_len]
tensor, TODOS los tokens en vuelo se concatenan en un único sin relleno
[1, total_tokens] fila y corre en UN pase hacia adelante. Una diagonal de bloque
La máscara de atención causal impide que los tokens atiendan a través de la secuencia.
límites, por lo que empaquetar es matemáticamente idéntico a ejecutar cada
secuencia por sí sola (verificado: la salida codiciosa coincide por mensaje
generación token por token).
Como la atención se rige enteramente por la mascarilla, un recién ingresado
El PREFILL de múltiples tokens del mensaje avanza en el mismo pase hacia adelante que
el paso DECODE de un solo token de cada otra secuencia. El prellenado y la decodificación son
fusionado: sin relleno, sin pase de precarga separado.
Caché KV: cada secuencia mantiene su propio DynamicCache; cada paso los cachés
se concatenan a lo largo del eje de tiempo en un caché empaquetado, y el nuevo
El KV calculado se distribuye por secuencia. (Los motores reales almacenan el
caché en páginas de tamaño fijo – " atención paginada " – para evitar este paso por paso
reensamblaje, pero la lógica de atención/enmascaramiento es exactamente lo que ve aquí).
" ""
tiempo de importación
importar antorcha
desde clases de datos importar clase de datos , campo
al escribir importar Opcional
desde transformadores importa AutoTokenizer , AutoModelForCausalLM , caché dinámica
de transformadores . cache_utils importar capa dinámica
MODELO_ID = "comunidad-openai/gpt2" # intercambio por cualquier LM causal
BATCH_SIZE = 3 # máximo de secuencias concurrentes (ranuras)
def _device_sync ( modelo ) -> Ninguno :
"" "Bloquear hasta que finalice el trabajo de la GPU en cola, para que los tiempos sean precisos." ""
si modelo . dispositivo . tipo == "cuda" :
antorcha . cuda . sincronizar ( )
modelo elif . dispositivo . tipo == "mps" :
antorcha . diputados . sincronizar ( )
def static_batching ( solicitudes : lista [ tupla [ cadena , int ] ] , tokenizador , modelo ) -> lista [ cadena ] :
"" "Línea de base. Procesa solicitudes BATCH_SIZE a la vez; cada oleada se ejecuta en conjunto
hasta que finalice su solicitud MÁS LARGA, luego se elimina una barrera por lotes antes de que
comienza la siguiente ola.
Desventaja: las solicitudes cortas en una ola están inactivas hasta que finaliza la más larga de la ola.
y no se podrá rellenar ningún espacio hasta que toda la ola supere la barrera.
" ""
si no solicitudes :
devolver [ ]
tokenizador . lado_relleno = "izquierda"
resultados : dict [ int , cadena ] = { }
indexado = lista ( enumerar ( solicitudes ) ) # (req_id, (mensaje, límite))
para inicio_ola en rango ( 0 , len ( indexado ) , BATCH_SIZE ) :
ola = indexado [ wave_start : inicio_ola + BATCH_SIZE ]
onda_max = máx ( límite para _ , ( _ , gorra ) en ola )
# Muestra qué solicitud ocupa cada espacio en esta ola.
para ranura , ( req_id , ( inmediato , gorra ) ) en enumerar ( ola ) :
print ( f " ++ ranura {ranura} <- req {req_id} ({cap} tok cap): {prompt!r}" , color = Verdadero )
indicaciones = [ pag para _ , ( pag , _ ) en ola ]
entradas = tokenizador (
indicaciones , return_tensors = "pt" , relleno = Verdadero , truncamiento = Verdadero
) . a ( modelo . dispositivo )
con antorcha . no_graduación ( ) :
identificadores_salida = modelo . generar (
* * entradas ,
max_new_tokens = onda_max , # onda completa decodifica al más largo
pad_token_id = tokenizador . eos_token_id ,
hacer_muestra = Falso ,
)
ancho = entradas . identificadores_entrada . forma [ 1 ]
imprimir (
f " *** barrera por lotes: todas las ranuras {len(wave)} esperan más tiempo "
f "({wave_max} tokens) ***" ,
rubor = Verdadero ,
)
para ranura , ( ( req_id , ( inmediato , gorra ) ) , fila ) en enumerar ( zip ( ola , ids_salida ) ) :
texto = inmediato + tokenizador . decodificar ( fila [ ancho : ancho + gorra ] , skip_special_tokens = Verdadero )
resultados [ req_id ] = texto
imprimir (
f " — ranura {ranura} hecha solicitud {req_id} (tokens {cap}/{wave_max}): {text[:90]}" ,
rubor = Verdadero ,
)
devolver [ resultados [ k ] para k en ordenados ( resultados ) ]
@ clase de datos
clase Secuencia :
"" "Estado para una única secuencia en vuelo." ""
req_id : entero # índice de solicitud original (para ordenar resultados)
inmediato : cadena
max_new_tokens : entero # límite por solicitud para que las solicitudes breves finalicen antes
# Tokens para alimentar en el SIGUIENTE paso: todo el aviso inmediatamente después de la admisión
# (relleno previo), luego un único token por paso (decodificación).
pendientes_ids : lista [ int ]
# Caché KV por secuencia; Ninguno hasta que esta secuencia se haya ejecutado una vez.
kv_cache : Opcional [ Caché dinámico ] = Ninguno
kv_len : entero = 0 # número de tokens almacenados en caché (mensaje + generado)
tokens_generados : entero = 0
identificadores de salida : lista [ int ] = campo ( default_factory = lista )
def _make_cache ( capas_kv : lista [ tupla [ antorcha . tensores , antorcha . Tensores ] ] ) -> caché dinámica :
"" "Construya un DynamicCache a partir de tensores explícitos por capa (claves, valores).
CONFIGURAMOS los tensores directamente en lugar de llamar a DynamicLayer.update() (que
agregaría), porque estamos ensamblando cachés desde cero en cada paso.
" ""
cache = Caché dinámico ( )
para k , v en capas_kv :
capa = Capa dinámica ( )
capa . inicialización_perezosa ( k , v )
capa . llaves = k
capa . valores = v
caché . capas . añadir ( capa )
devolver cache
def _paso_ragged ( seqs : lista [ Secuencia ] , modelo , dispositivo , tipo ) -> lista [ int ] :
"" "Ejecute UNA pasada completa sobre cada secuencia activa.
Todas las secuencias se aplanan en una sola fila (lote tenue = 1):
input_ids [1, total_q] – tokens pendientes de cada secuencia
position_ids [1, total_q] – la posición de cada token en su secuencia
máscara_atención [1, 1, total_q, total_kv + total_q] – bloque-diagonal causal
past_key_values caché empaquetado [1, H, total_kv, D]
total_q = suma de tokens pendientes (1 por secuencia de decodificación, Prompt_len por secuencia nueva)
total_kv = suma de tokens ya almacenados en caché en todas las secuencias
Devuelve el siguiente token codicioso para cada secuencia (mismo orden que “seqs“).
" ""
q_lens = [ len ( s . id_pendientes ) para s en secuencias ]
total_q = suma ( q_lens )
total_kv = suma ( s . kv_len para s en secuencias )
# Entradas empaquetadas: concatena los tokens pendientes de cada secuencia en una fila.
ids_planos = [ t para s en secuencias para t en s . pendientes_ids ]
ids_entrada = antorcha . tensor ( [ id_planos ] , tipo d = antorcha . largo , dispositivo = dispositivo )
# Etiquete cada CLAVE y cada token de CONSULTA con (índice de secuencia, posición en secuencia).
# El espacio clave se presenta como [tokens almacenados en caché | los nuevos tokens de este paso], coincidentes
# cómo el modelo agrega un nuevo KV al final del caché empaquetado.
clave_seq , pos_clave = [ ] , [ ]
para sí , s en enumerar ( secuencias ) : # bloque en caché
para pag en rango ( s . kv_len ) :
clave_seq . añadir ( si )
pos_clave . añadir ( p )
q_seq , q_pos = [ ] , [ ]
para sí , s en enumerar ( secuencias ) : # nuevo bloque (también consultas)
para j en rango ( len ( s . id_pendientes ) ) :
posición = s . kv_len + j
q_seq . añadir ( si )
q_pos . agregar ( pos )
clave_seq . añadir ( si )
pos_clave . agregar ( pos )
q_seq_t = antorcha . tensor ( q_seq , dispositivo = dispositivo )
q_pos_t = antorcha . tensor ( q_pos , dispositivo = dispositivo )
clave_seq_t = antorcha . tensor ( key_seq , dispositivo = dispositivo )
clave_pos_t = antorcha . tensor ( key_pos , dispositivo = dispositivo )
# La incrustación posicional de cada token utiliza su propia posición de secuencia, no su
# desplazamiento en la fila empaquetada.
identificadores de posición = q_pos_t . descomprimir ( 0 ) # [1, total_q]
# Máscara causal de bloque diagonal: una consulta puede atender a una clave solo si pertenece
# a la MISMA secuencia (bloque-diagonal) y la clave no está en el futuro
# (causal). Este es el truco: hacer que empacar sea equivalente a correr
# cada secuencia por separado. 0,0 = asistir, negativo grande = bloqueado (aditivo).
mismo = q_seq_t [ : , Ninguno ] == key_seq_t [ Ninguno , : ]
causal = key_pos_t [ Ninguno , : ] <= q_pos_t [ : , Ninguno ]
permitido = mismo & causal # [total_q, total_kv + total_q]
atención_mascarilla = antorcha . ceros ( 1 , 1 , total_q , total_kv + total_q , tipod = tipod , dispositivo = dispositivo )
atención_máscara . masked_fill_ ( ~ permitido [ Ninguno , Ninguno ] , antorcha . finfo ( tipo d ) . mín .)
# Caché KV empaquetado: concatena el caché de cada secuencia a lo largo del eje de tiempo.
# Las secuencias recién admitidas (kv_len == 0) no aportan nada aquí.
almacenado en caché = [ s para s en secuencias si s . kv_len > 0 ]
si en caché :
num_capas = len ( en caché [ 0 ] . kv_cache . capas )
capas_kv = [ ]
para yo en rango ( núm_capas ) :
Kansas = antorcha . cat ( [ s . kv_cache . capas [ l ] . claves para s en almacenado en caché ] , tenue = 2 )
vs = antorcha . cat ( [ s . kv_cache . capas [ l ] . valores para s en almacenado en caché ] , tenue = 2 )
capas_kv . agregar ( ( ks , vs ) )
pasado = _make_cache ( capas_kv )
demás :
pasado = Caché dinámico ( )
con antorcha . no_graduación ( ) :
afuera = modelo (
ids_entrada = id_entrada ,
máscara_atención = máscara_atención ,
ids_posicion = ids_posicion ,
valores_clave_pasados = pasado ,
use_cache = Verdadero ,
)
# Siguiente token codicioso para cada secuencia: lee los logits en su ÚLTIMO pendiente
# token (para una secuencia de prellenado que es el token de aviso final).
logits = afuera . logits [ 0 ] # [total_q, vocabulario]
compensaciones , último_idx , apagado = [ ] , [ ] , 0
para ql en q_lens :
compensaciones . agregar ( desactivado )
último_idx . agregar ( desactivado) + ql – 1 )
apagado += ql
tokens_siguientes = [ int ( logits [ i ] . argmax ( ) ) para i en último_idx ]
# Distribuya el KV recién calculado en cada secuencia. La caché de salida es
# [bloque antiguo empaquetado | nuevo bloque empaquetado ]; cortar el nuevo bloque de este paso por
# secuencia y añádela al caché de esa secuencia.
salida_kv = afuera . valores_clave_pasados
num_capas = len ( out_kv . capas )
para sí , s en enumerar ( secuencias ) :
oh , ql = compensaciones [ si ] , q_lens [ si ]
capas_kv = [ ]
para yo en rango ( núm_capas ) :
sabía = salida_kv . capas [ l ] . llaves [ : , : , total_kv + o : total_kv + oh + ql , : ]
v_nuevo = salida_kv . capas [ l ] . valores [ : , : , total_kv + o : total_kv + oh + ql , : ]
si s . kv_cache es Ninguno :
capas_kv . agregar ( ( k_new , v_nuevo ) )
demás :
capas_kv . agregar ( (
antorcha . cat ( [ s.kv_cache.capas [ l ] .claves , sabía ] , tenue = 2 ) ,
antorcha . cat ( [ s.kv_cache.capas [ l ] .valores , v_nuevo ] , tenue = 2 ) ,
) )
s . kv_cache = _make_cache ( capas_kv )
s . kv_len += ql
devolver tokens_siguientes
def visualizar_ragged_step ( secuencias : lista [ Secuencia ] , tokenizador , título : cadena , ID_ranura : lista [ int ] ) -> Ninguno :
"" "Impresión ilustrativa de UN paso empaquetado: la fila de entrada concatenada y el
máscara de atención causal en bloque diagonal.
Esto refleja la lógica de enmascaramiento en _ragged_step (recalculada aquí como un valor booleano).
cuadrícula puramente para visualización) para que pueda VER que las secuencias están empaquetadas juntas
todavía aislado por la máscara. Cada secuencia recibe una letra A, B, C,…
# = una consulta puede atender a esa clave. = bloqueado
" ""
etiquetas = [ chr ( ord ( "A" ) + s . id_req ) para s en secuencias ]
q_lens = [ len ( s . id_pendientes ) para s en secuencias ]
total_q = suma ( q_lens )
total_kv = suma ( s . kv_len para s en secuencias )
imprimir ( f "n{'=' * 72}n {título}" )
print ( f "total_q={total_q} tokens alimentados en este paso | total_kv={total_kv} almacenados en caché" )
print ( f " {len(seqs)} secuencias empaquetadas en UNA fila de forma sin relleno [1, {total_q}]:n" )
# Los tokens concatenados, agrupados por secuencia (esta es la fila "irregular").
para i , s en enumerar ( secuencias ) :
amable = f "PRELLENAR({q_lens[i]})" si s . kv_len == 0 demás f "decodificar({q_lens[i]})"
toks = " " . unirse ( repr ( tokenizer . decodificar ( [ t ] ) ) para t en s . pendientes_ids )
si len ( toks ) > 66 :
toks = toks [ : 63 ] + "…"
imprimir ( f " {etiquetas[i]} = ranura {slot_ids[i]} {tipo:<11} {toks}" )
# Reconstruir la máscara causal de bloque diagonal como una cuadrícula booleana para su visualización.
clave_seq , pos_clave = [ ] , [ ]
para sí , s en enumerar ( secuencias ) : # claves almacenadas en caché
clave_seq += [ si ] * s . kv_len
pos_clave += lista ( rango ( s . kv_len ) )
q_seq , q_pos = [ ] , [ ]
para sí , s en enumerar ( secuencias ) : # nuevas claves/consultas
para j en rango ( q_lens [ si ] ) :
q_seq . añadir ( si )
q_pos . agregar ( s . kv_len + j )
clave_seq += q_seq
pos_clave += q_pos
q_seq_t , q_pos_t = antorcha . tensor ( q_seq ) , antorcha . tensor ( q_pos )
clave_seq_t , clave_pos_t = antorcha . tensor ( key_seq ) , antorcha . tensor ( key_pos )
permitido = ( q_seq_t [ : , Ninguno ] == key_seq_t [ Ninguno , : ] ) & ( key_pos_t [ Ninguno , : ] <= q_pos_t [ : , Ninguno ] )
k = len ( key_seq )
def fila_str ( celdas ) :
# Espacio entre grupos de secuencias; ' | ' en el caché -> división de nuevos tokens.
afuera = [ ]
para ki en rango ( K ) :
si total_kv > 0 y ki == total_kv :
afuera . agregar ( " | " )
elif ki > 0 y key_seq [ ki ] != key_seq [ ki – 1 ] :
afuera . agregar ( " " )
afuera . agregar ( celdas [ ki ] )
devolver "" . unirse ( fuera )
línea definida ( izquierda , células ) :
devolver f "{izquierda:>7} " + fila_str ( celdas )
print ( f "n máscara causal de bloque diagonal (fila = consulta, columna = clave) # atender. bloqueado" )
si total_kv > 0 :
print ( f "diseño de clave: [KV en caché | nuevos tokens de este paso]" )
imprimir ( línea ( "claves:" , [ etiquetas [ key_seq [ ki ] ] para ki en rango ( K ) ] ) )
para qi en rango ( total_q ) :
células = [ "#" si permitido [ qi , ki ] demás "." para ki en rango ( K ) ]
imprimir ( línea ( f "{labels[q_seq[qi]]} p{q_pos[qi]}" , células ) )
def continuo_batching ( solicitudes : lista [ tupla [ cadena , int ] ] , tokenizador , modelo ) -> lista [ cadena ] :
"" "Loteo continuo desigual: programación dinámica + precarga/decodificación empaquetada.
Política de programación:
– Se ejecutan hasta BATCH_SIZE secuencias simultáneamente.
– Una secuencia recién admitida se pone en cola con su mensaje completo como la siguiente
fichas para alimentar; su precarga luego pasa al siguiente paso
junto con la decodificación de todos los demás.
– Cada paso ejecuta UN pase hacia adelante lleno a través de todas las ranuras activas.
– Cuando finaliza una secuencia, se reemplaza inmediatamente por el siguiente mensaje.
El registro de admisión muestra los espacios que se reutilizan (programación a nivel de iteración).
Se visualizan dos pasos representativos: el primer paso (todas las indicaciones se
precargado a la vez) y el primer paso que fusiona el precargado de un nuevo mensaje con
tokens de decodificación de otras secuencias.
" ""
dispositivo = modelo . dispositivo
tipo d = siguiente ( modelo . parámetros ( ) ) . tipo d
cola = lista ( enumerar ( solicitudes ) ) # (req_id, (mensaje, max_new_tokens))
ranuras : lista [ Opcional [ Secuencia ] ] = [ Ninguno ] * BATCH_SIZE
resultados : dict [ int , cadena ] = { }
def _admit ( ranura_idx : int ) -> Ninguno :
si no cola :
ranuras [ slot_idx ] = Ninguno
devolver
id_req , ( inmediato , max_new_tokens ) = cola . pop ( 0 )
id_indicadores = tokenizador ( solicitud ) [ "input_ids" ]
ranuras [ slot_idx ] = Secuencia (
id_req = id_req ,
aviso = aviso ,
max_new_tokens = max_new_tokens ,
pendientes_ids = lista ( prompt_ids ) , # prefill monta el siguiente paso
)
imprimir (
f " ++ [paso {paso:3d}] ranura {slot_idx} <- admitir solicitud {req_id} "
f "({max_new_tokens} tok cap): {prompt!r}" ,
rubor = Verdadero ,
)
# Llene el grupo con el primer lote de indicaciones (paso 0 = antes de cualquier decodificación).
paso = 0
para i en rango ( BATCH_SIZE ) :
_admitir ( yo )
impreso_mixto = FALSO
mientras cualquiera ( s es no Ninguno para s en ranuras ) :
paso += 1
activo = [ ( i , s ) para i , s en enumerar ( espacios ) si s es no Ninguno ]
secuencias = [ s para _ , s en activo ]
ID_ranura = [ i para i , _ en activo ]
# Visualice un par de pasos representativos para que el embalaje sea visible.
# (imprimir cada paso sería demasiado resultado).
mezclado = cualquiera ( s . kv_len == 0 para s en secuencias ) y cualquiera ( s . kv_len > 0 para s en secuencias )
si paso == 1 :
visualizar_ragged_step (
secuencias , tokenizador , f "PASO {paso} – indicaciones empaquetadas juntas (todas PREFILL)" , ID_ranura )
elif mezclado y no impreso_mixto :
visualizar_ragged_step (
secuencias , tokenizador , f "PASO {paso} – PRELLENADO + DECODIFICACIÓN fusionados en una sola pasada" , ID_ranura )
impreso_mixto = Verdadero
# UN pase hacia adelante empaquetado (prellenado + decodificación fusionada, sin relleno).
tokens_siguientes = _paso_ragged ( secuencias , modelo , dispositivo , tipo )
para ( ranura_idx , siguiente ) , tomar en zip ( activo , tokens_siguientes ) :
siguiente . identificadores_salida . añadir ( tok )
siguiente . tokens_generados += 1
siguiente . pendientes_ids = [ tok ] # siguiente paso: un único token de decodificación
si tok == tokenizador . eos_token_id o siguiente . tokens_generados >= siguiente . max_new_tokens :
texto_resultado = siguiente . inmediato +
tokenizador . decodificar ( seq.output_ids , skip_special_tokens = Verdadero )
resultados [ sec . id_req ] = texto_resultado
imprimir (
f " — paso {paso:3d}] ranura {slot_idx} hecho req {seq.req_id} "
f "({seq.tokens_generated}/{seq.max_new_tokens} tokens): {result_text[:90]}" ,
rubor = Verdadero ,
)
_admit ( ranura_idx )
devolver [ resultados [ k ] para k en ordenados ( resultados ) ]
definición principal ( ) :
imprimir ( f "Cargando {MODEL_ID}" )
tokenizador = AutoTokenizador . from_pretrained ( MODELO_ID )
tokenizador . pad_token = tokenizador . eos_token
# Elija el dispositivo disponible más rápido. En Apple Silicon (M1/M2/…) esto es
# la GPU MPS. Mantenemos float32 en MPS a propósito: float16 voltea algunos
# lazos codiciosos, que romperían el "estático == continuo, token por token"
# propiedad en la que se basa esta demostración.
si antorcha . cuda . está_disponible ( ) :
dispositivo , tipo d = "cuda" , antorcha . flotador16
antorcha elif . backends . diputados . está_disponible ( ) :
dispositivo , tipo d = "diputados" , antorcha . flotador32
demás :
dispositivo , tipo d = "UPC" , antorcha . flotador32
modelo = AutoModelForCausalLM . from_pretrained (
MODELO_ID ,
tipod = tipod ,
attn_implementation = "ansioso" , # usa nuestra máscara 4D personalizada directamente
)
modelo . evaluar ( )
modelo . a ( dispositivo )
print ( f "Ejecutando en {dispositivo} ({dtype})n" )
solicitudes = [
( "La capital de Francia es" , 6 ) ,
( "El clima de hoy es así" , 50 ) ,
( "En el aprendizaje automático, un transformador es" , 300 ) ,
( "Érase una vez en una tierra lejana", , 30 ) ,
( "La computación cuántica se diferencia de la computación clásica porque" , 180 ) ,
( "Comenzó la historia del Imperio Romano" , 45 ) ,
]
print ( "=== Lotes estáticos ===" )
_device_sync ( modelo )
comenzar = tiempo . contador_perf ( )
static_batching ( solicitudes , tokenizador , modelo )
_device_sync ( modelo )
estática_transcurrida = tiempo . contador_perf ( ) – comenzar
print ( f "nLoteo estático transcurrido: {static_elapsed:.2f}sn" )
print ( "=== Lotes continuos (irregulares) ===" )
_device_sync ( modelo )
comenzar = tiempo . contador_perf ( )
procesamiento por lotes continuo ( solicitudes , tokenizador , modelo )
_device_sync ( modelo )
continuo_transcurrido = tiempo . contador_perf ( ) – comenzar
print ( f "nTranscurrió el procesamiento por lotes continuo: {continuous_elapsed:.2f}s" )
si __nombre__ == "__principal__" :
principal ( )