RNNs, LSTMs et Transformers : Traiter des séquences avec PyTorch

# Problème : comment traiter des phrases de longueurs différentes ?
phrase1 = "Bonjour"  # 1 mot
phrase2 = "Le chat dort sur le tapis"  # 6 mots
# MLP nécessite une taille d'entrée fixe !

# Les CNNs voient une fenêtre locale, mais pas l'ordre global
# "Le film n'est pas mauvais" vs "Le film est mauvais"
# Le "n'est pas" inverse complètement le sens

x₁ → [RNN] → h₁ → y₁
      ↑ |
      └─┘ (récurrence)
x₂ → [RNN] → h₂ → y₂
      ↑ |
      └─┘
x₃ → [RNN] → h₃ → y₃

h_t = tanh(W_hh · h_{t-1} + W_xh · x_t + b_h)
y_t = W_hy · h_t + b_y

import torch
import torch.nn as nn

class VanillaRNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super().__init__()
        self.hidden_size = hidden_size

        # Couches de transformation
        self.i2h = nn.Linear(input_size + hidden_size, hidden_size)
        self.h2o = nn.Linear(hidden_size, output_size)
        self.tanh = nn.Tanh()

    def forward(self, x, hidden):
        """
        x: (batch, input_size)
        hidden: (batch, hidden_size)
        """
        # Concaténer entrée et état caché
        combined = torch.cat([x, hidden], dim=1)

        # Nouvel état caché
        hidden = self.tanh(self.i2h(combined))

        # Sortie
        output = self.h2o(hidden)

        return output, hidden

    def init_hidden(self, batch_size):
        """Initialiser l'état caché à zéro"""
        return torch.zeros(batch_size, self.hidden_size)

# Exemple d'utilisation
rnn = VanillaRNN(input_size=50, hidden_size=128, output_size=10)

# Séquence de 20 éléments
sequence_length = 20
batch_size = 32

hidden = rnn.init_hidden(batch_size)

for t in range(sequence_length):
    x_t = torch.randn(batch_size, 50)  # Entrée au temps t
    output, hidden = rnn(x_t, hidden)
    # hidden contient maintenant la "mémoire" des t premiers éléments

class SimpleRNNModel(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_size, num_classes):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        self.rnn = nn.RNN(
            input_size=embedding_dim,
            hidden_size=hidden_size,
            num_layers=1,
            batch_first=True  # (batch, seq, features)
        )
        self.fc = nn.Linear(hidden_size, num_classes)

    def forward(self, x):
        """
        x: (batch, seq_length) - indices de tokens
        """
        # Embedding: (batch, seq_length, embedding_dim)
        x = self.embedding(x)

        # RNN: output (batch, seq, hidden), h_n (1, batch, hidden)
        output, h_n = self.rnn(x)

        # Utiliser le dernier état caché pour la classification
        # h_n: (num_layers, batch, hidden) -> (batch, hidden)
        h_last = h_n.squeeze(0)

        # Classification
        logits = self.fc(h_last)
        return logits

# Exemple
model = SimpleRNNModel(vocab_size=10000, embedding_dim=100,
                       hidden_size=256, num_classes=5)

# Batch de phrases (indices de tokens)
batch = torch.randint(0, 10000, (32, 50))  # 32 phrases de 50 tokens

logits = model(batch)  # (32, 5) - scores pour 5 classes
print(logits.shape)  # torch.Size([32, 5])

class LSTMClassifier(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_size,
                 num_layers, num_classes, dropout=0.5):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        self.lstm = nn.LSTM(
            input_size=embedding_dim,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0,
            bidirectional=False
        )
        self.dropout = nn.Dropout(dropout)
        self.fc = nn.Linear(hidden_size, num_classes)

    def forward(self, x):
        """
        x: (batch, seq_length)
        """
        # Embedding
        x = self.embedding(x)  # (batch, seq, embedding_dim)

        # LSTM
        # output: (batch, seq, hidden)
        # (h_n, c_n): (num_layers, batch, hidden) chacun
        output, (h_n, c_n) = self.lstm(x)

        # Utiliser le dernier hidden state de la dernière couche
        h_last = h_n[-1]  # (batch, hidden)

        # Dropout et classification
        h_last = self.dropout(h_last)
        logits = self.fc(h_last)

        return logits

# Modèle pour analyse de sentiment (5 classes)
model = LSTMClassifier(
    vocab_size=10000,
    embedding_dim=128,
    hidden_size=256,
    num_layers=2,
    num_classes=5,
    dropout=0.5
)

# Exemple
x = torch.randint(0, 10000, (32, 100))  # 32 reviews de 100 tokens
logits = model(x)  # (32, 5)

class BiLSTMClassifier(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_size, num_classes):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        self.lstm = nn.LSTM(
            embedding_dim,
            hidden_size,
            num_layers=2,
            batch_first=True,
            bidirectional=True,  # Bidirectionnel !
            dropout=0.5
        )
        # hidden_size * 2 car bidirectionnel (forward + backward)
        self.fc = nn.Linear(hidden_size * 2, num_classes)

    def forward(self, x):
        x = self.embedding(x)
        output, (h_n, c_n) = self.lstm(x)

        # h_n shape: (num_layers * 2, batch, hidden)
        # Concaténer forward et backward de la dernière couche
        h_forward = h_n[-2]  # (batch, hidden)
        h_backward = h_n[-1]  # (batch, hidden)
        h_combined = torch.cat([h_forward, h_backward], dim=1)  # (batch, hidden*2)

        logits = self.fc(h_combined)
        return logits

Phrase longue (50 tokens) → LSTM → [h_last] → Prédiction
                                      ↑
                              Tout compressé dans 256 dimensions !

Tous les h_t → Attention → Combinaison pondérée → Prédiction
               ↑
         "Je regarde les parties importantes"

class AttentionLayer(nn.Module):
    def __init__(self, hidden_size):
        super().__init__()
        self.attention = nn.Linear(hidden_size, 1)

    def forward(self, lstm_output):
        """
        lstm_output: (batch, seq_length, hidden_size)
        """
        # Calculer les scores d'attention pour chaque timestep
        # (batch, seq_length, 1)
        attention_scores = self.attention(lstm_output)

        # Softmax pour obtenir des poids qui somment à 1
        # (batch, seq_length, 1)
        attention_weights = torch.softmax(attention_scores, dim=1)

        # Combinaison pondérée
        # (batch, seq_length, hidden) * (batch, seq_length, 1)
        # → sum over seq → (batch, hidden)
        context_vector = torch.sum(attention_weights * lstm_output, dim=1)

        return context_vector, attention_weights

class LSTMWithAttention(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_size, num_classes):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, hidden_size, batch_first=True)
        self.attention = AttentionLayer(hidden_size)
        self.fc = nn.Linear(hidden_size, num_classes)

    def forward(self, x):
        # (batch, seq, embedding)
        x = self.embedding(x)

        # (batch, seq, hidden)
        lstm_output, _ = self.lstm(x)

        # Attention : (batch, hidden)
        context, attn_weights = self.attention(lstm_output)

        # Classification
        logits = self.fc(context)

        return logits, attn_weights  # Retourner aussi les poids pour visualisation

# Exemple
model = LSTMWithAttention(10000, 128, 256, 5)
x = torch.randint(0, 10000, (32, 50))
logits, attn_weights = model(x)

# Visualiser quels mots sont importants
print(attn_weights.shape)  # (32, 50, 1)

Phrase : "Le chat qui était sur le tapis a mangé"

Quand on traite "mangé", l'attention permet de voir :
- "mangé" (lui-même)
- "chat" (sujet, forte attention)
- "a" (verbe auxiliaire, attention moyenne)
- "qui", "tapis" (faible attention)

Attention(Q, K, V) = softmax(Q·K^T / √d_k) · V

import math

class SelfAttention(nn.Module):
    def __init__(self, embed_size, heads):
        super().__init__()
        self.embed_size = embed_size
        self.heads = heads
        self.head_dim = embed_size // heads

        assert self.head_dim * heads == embed_size, "embed_size doit être divisible par heads"

        # Projections linéaires pour Q, K, V
        self.queries = nn.Linear(self.head_dim, self.head_dim, bias=False)
        self.keys = nn.Linear(self.head_dim, self.head_dim, bias=False)
        self.values = nn.Linear(self.head_dim, self.head_dim, bias=False)
        self.fc_out = nn.Linear(embed_size, embed_size)

    def forward(self, values, keys, queries, mask=None):
        """
        values, keys, queries: (batch, seq_length, embed_size)
        mask: (batch, 1, 1, seq_length) pour masquer certains tokens (padding)
        """
        batch_size = queries.shape[0]
        value_len, key_len, query_len = values.shape[1], keys.shape[1], queries.shape[1]

        # Séparer en plusieurs têtes (multi-head attention)
        # (batch, seq, embed) → (batch, seq, heads, head_dim)
        queries = queries.reshape(batch_size, query_len, self.heads, self.head_dim)
        keys = keys.reshape(batch_size, key_len, self.heads, self.head_dim)
        values = values.reshape(batch_size, value_len, self.heads, self.head_dim)

        # Appliquer les transformations linéaires
        queries = self.queries(queries)
        keys = self.keys(keys)
        values = self.values(values)

        # Calculer les scores d'attention
        # (batch, heads, query_len, key_len)
        energy = torch.einsum("bqhd,bkhd->bhqk", [queries, keys])

        # Masquer si nécessaire (pour le padding)
        if mask is not None:
            energy = energy.masked_fill(mask == 0, float("-1e20"))

        # Normalisation avec softmax et scaling
        attention = torch.softmax(energy / math.sqrt(self.head_dim), dim=3)

        # Appliquer l'attention aux valeurs
        # (batch, heads, query_len, key_len) x (batch, heads, value_len, head_dim)
        # → (batch, heads, query_len, head_dim)
        out = torch.einsum("bhql,blhd->bqhd", [attention, values])

        # Concaténer les têtes
        # (batch, query_len, heads, head_dim) → (batch, query_len, embed_size)
        out = out.reshape(batch_size, query_len, self.embed_size)

        # Projection finale
        out = self.fc_out(out)

        return out

class TransformerEncoderBlock(nn.Module):
    def __init__(self, embed_size, heads, dropout, forward_expansion):
        super().__init__()
        self.attention = SelfAttention(embed_size, heads)
        self.norm1 = nn.LayerNorm(embed_size)
        self.norm2 = nn.LayerNorm(embed_size)

        self.feed_forward = nn.Sequential(
            nn.Linear(embed_size, forward_expansion * embed_size),
            nn.GELU(),
            nn.Linear(forward_expansion * embed_size, embed_size)
        )

        self.dropout = nn.Dropout(dropout)

    def forward(self, value, key, query, mask):
        # Self-attention avec connexion résiduelle
        attention = self.attention(value, key, query, mask)
        x = self.dropout(self.norm1(attention + query))

        # Feed-forward avec connexion résiduelle
        forward = self.feed_forward(x)
        out = self.dropout(self.norm2(forward + x))

        return out

class TransformerEncoder(nn.Module):
    def __init__(self, vocab_size, embed_size, num_layers, heads,
                 device, forward_expansion, dropout, max_length):
        super().__init__()
        self.embed_size = embed_size
        self.device = device

        # Embeddings
        self.word_embedding = nn.Embedding(vocab_size, embed_size)
        self.position_embedding = nn.Embedding(max_length, embed_size)

        # Empilement de N blocs Transformer
        self.layers = nn.ModuleList([
            TransformerEncoderBlock(embed_size, heads, dropout, forward_expansion)
            for _ in range(num_layers)
        ])

        self.dropout = nn.Dropout(dropout)

    def forward(self, x, mask=None):
        """
        x: (batch, seq_length)
        """
        batch_size, seq_length = x.shape

        # Positional encoding
        positions = torch.arange(0, seq_length).expand(batch_size, seq_length).to(self.device)

        # Combiner word + positional embeddings
        out = self.dropout(
            self.word_embedding(x) + self.position_embedding(positions)
        )

        # Passer à travers tous les blocs
        for layer in self.layers:
            out = layer(out, out, out, mask)

        return out

# Exemple de classification avec Transformer
class TransformerClassifier(nn.Module):
    def __init__(self, vocab_size, embed_size, num_layers, heads,
                 forward_expansion, dropout, max_length, num_classes):
        super().__init__()
        self.encoder = TransformerEncoder(
            vocab_size, embed_size, num_layers, heads,
            device="cuda" if torch.cuda.is_available() else "cpu",
            forward_expansion, dropout, max_length
        )
        self.fc = nn.Linear(embed_size, num_classes)

    def forward(self, x):
        # Encoder
        encoded = self.encoder(x)  # (batch, seq, embed_size)

        # Pooling : moyenne sur la séquence (ou utiliser [CLS] token comme BERT)
        pooled = encoded.mean(dim=1)  # (batch, embed_size)

        # Classification
        logits = self.fc(pooled)
        return logits

# Modèle
model = TransformerClassifier(
    vocab_size=10000,
    embed_size=512,
    num_layers=6,
    heads=8,
    forward_expansion=4,
    dropout=0.1,
    max_length=512,
    num_classes=5
)

# Exemple
x = torch.randint(0, 10000, (32, 128))
logits = model(x)
print(logits.shape)  # torch.Size([32, 5])

pip install transformers datasets

from transformers import BertTokenizer, BertForSequenceClassification, Trainer, TrainingArguments
from datasets import load_dataset
import torch

# Charger un modèle BERT pré-entraîné
model_name = "bert-base-uncased"
tokenizer = BertTokenizer.from_pretrained(model_name)
model = BertForSequenceClassification.from_pretrained(
    model_name,
    num_labels=2  # Classification binaire (pos/neg)
)

# Charger un dataset (exemple : IMDb sentiment analysis)
dataset = load_dataset("imdb")

# Tokenizer les textes
def tokenize_function(examples):
    return tokenizer(
        examples["text"],
        padding="max_length",
        truncation=True,
        max_length=512
    )

tokenized_datasets = dataset.map(tokenize_function, batched=True)

# Préparer pour l'entraînement
tokenized_datasets = tokenized_datasets.rename_column("label", "labels")
tokenized_datasets.set_format("torch", columns=["input_ids", "attention_mask", "labels"])

# Configuration de l'entraînement
training_args = TrainingArguments(
    output_dir="./results",
    evaluation_strategy="epoch",
    learning_rate=2e-5,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=16,
    num_train_epochs=3,
    weight_decay=0.01,
    logging_dir="./logs",
    logging_steps=100,
    save_strategy="epoch",
    load_best_model_at_end=True,
)

# Trainer
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized_datasets["train"].select(range(1000)),  # Subset pour rapidité
    eval_dataset=tokenized_datasets["test"].select(range(200)),
)

# Fine-tuning
trainer.train()

# Évaluation
results = trainer.evaluate()
print(f"Accuracy: {results['eval_accuracy']:.4f}")

# Prédiction sur un nouveau texte
text = "This movie was absolutely fantastic! I loved every minute."
inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
outputs = model(**inputs)
prediction = torch.argmax(outputs.logits, dim=1)
print(f"Sentiment: {'Positive' if prediction.item() == 1 else 'Negative'}")

from transformers import GPT2LMHeadModel, GPT2Tokenizer

# Charger GPT-2
model_name = "gpt2"
tokenizer = GPT2Tokenizer.from_pretrained(model_name)
model = GPT2LMHeadModel.from_pretrained(model_name)

# Générer du texte
prompt = "Artificial intelligence will change the world by"
inputs = tokenizer(prompt, return_tensors="pt")

# Génération
outputs = model.generate(
    inputs["input_ids"],
    max_length=100,
    num_return_sequences=1,
    no_repeat_ngram_size=2,
    top_k=50,
    top_p=0.95,
    temperature=0.7,
    do_sample=True
)

generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
print(generated_text)

from transformers import MarianMTModel, MarianTokenizer

# Modèle anglais → français
model_name = "Helsinki-NLP/opus-mt-en-fr"
tokenizer = MarianTokenizer.from_pretrained(model_name)
model = MarianMTModel.from_pretrained(model_name)

# Traduire
text = "Transformers are revolutionizing natural language processing."
inputs = tokenizer(text, return_tensors="pt", padding=True)

translated = model.generate(**inputs)
translation = tokenizer.decode(translated[0], skip_special_tokens=True)
print(translation)
# "Les transformateurs révolutionnent le traitement du langage naturel."

class SentimentAnalyzer:
    def __init__(self, model_path="distilbert-base-uncased-finetuned-sst-2-english"):
        from transformers import pipeline
        self.classifier = pipeline("sentiment-analysis", model=model_path)

    def analyze(self, text):
        result = self.classifier(text)[0]
        return {
            "label": result["label"],
            "confidence": result["score"]
        }

# Utilisation
analyzer = SentimentAnalyzer()
print(analyzer.analyze("I love this product!"))
# {'label': 'POSITIVE', 'confidence': 0.9998}
print(analyzer.analyze("This is terrible."))
# {'label': 'NEGATIVE', 'confidence': 0.9991}

from transformers import pipeline

ner = pipeline("ner", model="dbmdz/bert-large-cased-finetuned-conll03-english", aggregation_strategy="simple")

text = "Hugging Face is based in New York City and was founded by Clément Delangue."
entities = ner(text)

for entity in entities:
    print(f"{entity['word']}: {entity['entity_group']} (score: {entity['score']:.4f})")

# Hugging Face: ORG (score: 0.9989)
# New York City: LOC (score: 0.9994)
# Clément Delangue: PER (score: 0.9997)

from transformers import pipeline

qa = pipeline("question-answering", model="distilbert-base-cased-distilled-squad")

context = """
PyTorch is an open-source machine learning library based on the Torch library.
It is primarily developed by Facebook's AI Research lab. PyTorch provides
two high-level features: tensor computation with strong GPU acceleration
and deep neural networks built on a tape-based autograd system.
"""

question = "Who develops PyTorch?"

answer = qa(question=question, context=context)
print(f"Answer: {answer['answer']}")
print(f"Confidence: {answer['score']:.4f}")
# Answer: Facebook's AI Research lab
# Confidence: 0.9856

from transformers import pipeline

summarizer = pipeline("summarization", model="facebook/bart-large-cnn")

long_text = """
The Transformer architecture has revolutionized natural language processing
since its introduction in 2017. Unlike recurrent neural networks, Transformers
process all tokens in parallel using self-attention mechanisms. This allows
them to capture long-range dependencies more effectively and train much faster
on modern hardware. Models like BERT, GPT, and T5 are all based on the
Transformer architecture. These models have achieved state-of-the-art results
on virtually every NLP benchmark. The key innovation is the attention mechanism,
which allows each token to attend to all other tokens in the sequence.
Multi-head attention enables the model to focus on different aspects of the
input simultaneously. Positional encodings provide information about token
order, since the architecture itself has no inherent notion of sequence.
"""

summary = summarizer(long_text, max_length=50, min_length=25, do_sample=False)
print(summary[0]["summary_text"])
# Transformers process all tokens in parallel using self-attention mechanisms.
# Models like BERT, GPT, and T5 are based on Transformer architecture.

class SentimentLSTM(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_size, num_classes):
        # TODO: Embedding, BiLSTM, Attention, FC
        pass

    def forward(self, x):
        # TODO: Implémenter le forward pass
        pass

# TODO: Entraînement et évaluation