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

Où :
- QK^T : calcule les scores de similarité (produit scalaire)
- √d_k : facteur d'échelle (dimension des clés)
- softmax : normalise les scores en probabilités
- × V : pondère les valeurs selon les probabilités
                

Matrice d'attention sans masque :
  Le  chat mange
Le  ✓   ✓    ✓
chat ✓  ✓    ✓
mange ✓  ✓   ✓

Avec masque causal :
  Le  chat mange
Le  ✓   ✗    ✗
chat ✓  ✓    ✗
mange ✓  ✓   ✓
                

import torch
import torch.nn as nn
import math

class SelfAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super().__init__()
        assert d_model % num_heads == 0

        self.d_model = d_model
        self.num_heads = num_heads
        self.d_k = d_model // num_heads

        # Projections linéaires pour Q, K, V
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)

    def split_heads(self, x):
        batch_size, seq_len, d_model = x.size()
        return x.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)

    def forward(self, x, mask=None):
        batch_size, seq_len, d_model = x.size()

        # Projections Q, K, V
        Q = self.split_heads(self.W_q(x))  # (batch, heads, seq, d_k)
        K = self.split_heads(self.W_k(x))
        V = self.split_heads(self.W_v(x))

        # Scaled dot-product attention
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)

        # Appliquer le masque causal si nécessaire
        if mask is not None:
            scores = scores.masked_fill(mask == 0, float('-inf'))

        attention_weights = torch.softmax(scores, dim=-1)
        attention_output = torch.matmul(attention_weights, V)

        # Recombiner les têtes
        attention_output = attention_output.transpose(1, 2).contiguous()
        attention_output = attention_output.view(batch_size, seq_len, d_model)

        # Projection finale
        output = self.W_o(attention_output)

        return output, attention_weights

# Exemple d'utilisation
d_model = 512
num_heads = 8
seq_len = 10
batch_size = 2

attention = SelfAttention(d_model, num_heads)
x = torch.randn(batch_size, seq_len, d_model)

# Créer un masque causal
mask = torch.tril(torch.ones(seq_len, seq_len)).unsqueeze(0).unsqueeze(0)

output, weights = attention(x, mask)
print(f"Output shape: {output.shape}")  # (2, 10, 512)
print(f"Attention weights shape: {weights.shape}")  # (2, 8, 10, 10)
                

import matplotlib.pyplot as plt
import seaborn as sns

def visualize_attention(attention_weights, sentence):
    """
    Visualise les poids d'attention pour une tête
    attention_weights: (seq_len, seq_len)
    sentence: liste de tokens
    """
    plt.figure(figsize=(10, 8))
    sns.heatmap(
        attention_weights.detach().numpy(),
        xticklabels=sentence,
        yticklabels=sentence,
        cmap='viridis',
        cbar=True,
        square=True
    )
    plt.xlabel('Keys')
    plt.ylabel('Queries')
    plt.title('Attention Weights Heatmap')
    plt.tight_layout()
    plt.savefig('attention_viz.png', dpi=300)
    plt.show()

# Exemple
sentence = ["Le", "chat", "noir", "dort", "paisiblement"]
# weights[0, 0] = premier batch, première tête
visualize_attention(weights[0, 0], sentence)
                

PE(pos, 2i)   = sin(pos / 10000^(2i/d_model))
PE(pos, 2i+1) = cos(pos / 10000^(2i/d_model))

Où :
- pos : position du token (0, 1, 2, ...)
- i : dimension de l'embedding
- d_model : taille totale de l'embedding (ex: 512)
                

import torch
import numpy as np

def sinusoidal_positional_encoding(max_len, d_model):
    """
    Crée les positional encodings sinusoïdaux
    """
    pe = torch.zeros(max_len, d_model)
    position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)

    div_term = torch.exp(
        torch.arange(0, d_model, 2).float() *
        (-np.log(10000.0) / d_model)
    )

    pe[:, 0::2] = torch.sin(position * div_term)
    pe[:, 1::2] = torch.cos(position * div_term)

    return pe

# Exemple
max_len = 512
d_model = 512
pe = sinusoidal_positional_encoding(max_len, d_model)

print(f"PE shape: {pe.shape}")  # (512, 512)

# Visualisation
import matplotlib.pyplot as plt
plt.figure(figsize=(12, 6))
plt.imshow(pe[:50, :50], cmap='RdBu', aspect='auto')
plt.xlabel('Dimension')
plt.ylabel('Position')
plt.title('Sinusoidal Positional Encoding')
plt.colorbar()
plt.savefig('sinusoidal_pe.png')
                

import torch
import torch.nn as nn

class RotaryPositionalEmbedding(nn.Module):
    def __init__(self, dim, max_seq_len=2048, base=10000):
        super().__init__()
        self.dim = dim
        self.max_seq_len = max_seq_len
        self.base = base

        # Pré-calculer les fréquences
        inv_freq = 1.0 / (base ** (torch.arange(0, dim, 2).float() / dim))
        self.register_buffer('inv_freq', inv_freq)

        # Cache pour les positions
        self._seq_len_cached = None
        self._cos_cached = None
        self._sin_cached = None

    def _update_cache(self, seq_len, device):
        if seq_len != self._seq_len_cached:
            self._seq_len_cached = seq_len
            t = torch.arange(seq_len, device=device).type_as(self.inv_freq)
            freqs = torch.einsum('i,j->ij', t, self.inv_freq)
            emb = torch.cat((freqs, freqs), dim=-1)
            self._cos_cached = emb.cos()
            self._sin_cached = emb.sin()

    def rotate_half(self, x):
        """Rotation par paires (x1, x2) → (-x2, x1)"""
        x1, x2 = x[..., :x.shape[-1]//2], x[..., x.shape[-1]//2:]
        return torch.cat((-x2, x1), dim=-1)

    def forward(self, q, k):
        seq_len = q.shape[1]
        self._update_cache(seq_len, q.device)

        # Appliquer la rotation
        q_rotated = (q * self._cos_cached) + (self.rotate_half(q) * self._sin_cached)
        k_rotated = (k * self._cos_cached) + (self.rotate_half(k) * self._sin_cached)

        return q_rotated, k_rotated

# Exemple d'utilisation
batch_size = 2
seq_len = 128
num_heads = 8
head_dim = 64

rope = RotaryPositionalEmbedding(dim=head_dim)

q = torch.randn(batch_size, seq_len, num_heads, head_dim)
k = torch.randn(batch_size, seq_len, num_heads, head_dim)

q_rot, k_rot = rope(q, k)
print(f"Q rotated shape: {q_rot.shape}")
                

scores = Q @ K^T + ALiBi_bias

ALiBi_bias pour distance d :
Head 1: -1 * d
Head 2: -2 * d
Head 3: -4 * d
...

Effet : pénalise les tokens distants de manière progressive
                

# RoPE standard (entraîné sur 4k tokens)
base_freq = 10000

# Linear Scaling (8k tokens)
scaled_freq = base_freq * 2

# NTK-Aware Scaling (meilleure qualité)
alpha = 2  # facteur d'extension
new_base = base_freq * alpha^(d/(d-2))

# Dynamic NTK (utilisé par Llama 2 Long)
# Ajuste automatiquement selon la longueur
                

class MultiHeadAttentionWithRoPE(nn.Module):
    def __init__(self, d_model, num_heads, max_seq_len=2048):
        super().__init__()
        self.num_heads = num_heads
        self.d_k = d_model // num_heads

        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)

        self.rope = RotaryPositionalEmbedding(self.d_k, max_seq_len)

    def forward(self, x, mask=None):
        batch_size, seq_len, d_model = x.size()

        # Projections Q, K, V
        Q = self.W_q(x).view(batch_size, seq_len, self.num_heads, self.d_k)
        K = self.W_k(x).view(batch_size, seq_len, self.num_heads, self.d_k)
        V = self.W_v(x).view(batch_size, seq_len, self.num_heads, self.d_k)

        # Appliquer RoPE sur Q et K
        Q, K = self.rope(Q, K)

        # Transpose pour attention
        Q = Q.transpose(1, 2)  # (batch, heads, seq, d_k)
        K = K.transpose(1, 2)
        V = V.transpose(1, 2)

        # Scaled dot-product attention
        scores = torch.matmul(Q, K.transpose(-2, -1)) / (self.d_k ** 0.5)

        if mask is not None:
            scores = scores.masked_fill(mask == 0, float('-inf'))

        attn_weights = torch.softmax(scores, dim=-1)
        output = torch.matmul(attn_weights, V)

        # Recombiner
        output = output.transpose(1, 2).contiguous()
        output = output.view(batch_size, seq_len, d_model)

        return self.W_o(output)
                

FFN(x) = max(0, xW₁ + b₁)W₂ + b₂

Dimensions :
- Input: d_model (ex: 512)
- Hidden: d_ff = 4 × d_model (ex: 2048)
- Output: d_model (ex: 512)

Activation : ReLU, GELU, ou SwiGLU
                

import torch.nn as nn
import torch.nn.functional as F

class FeedForward(nn.Module):
    """FFN classique (GPT-2, BERT style)"""
    def __init__(self, d_model, d_ff, dropout=0.1):
        super().__init__()
        self.w1 = nn.Linear(d_model, d_ff)
        self.w2 = nn.Linear(d_ff, d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):
        # x: (batch, seq_len, d_model)
        x = self.w1(x)              # (batch, seq_len, d_ff)
        x = F.gelu(x)               # Activation GELU
        x = self.dropout(x)
        x = self.w2(x)              # (batch, seq_len, d_model)
        x = self.dropout(x)
        return x

# Exemple
d_model = 768
d_ff = 3072  # 4 × d_model
ffn = FeedForward(d_model, d_ff)

x = torch.randn(2, 128, d_model)
output = ffn(x)
print(f"Output shape: {output.shape}")  # (2, 128, 768)
                

class SwiGLU(nn.Module):
    """FFN avec SwiGLU (Llama style)"""
    def __init__(self, d_model, d_ff=None):
        super().__init__()
        if d_ff is None:
            # Llama utilise ~2.7x au lieu de 4x
            d_ff = int(2 * (4 * d_model) / 3)
            # Arrondir au multiple de 256 supérieur
            d_ff = 256 * ((d_ff + 256 - 1) // 256)

        self.w_gate = nn.Linear(d_model, d_ff, bias=False)
        self.w_up = nn.Linear(d_model, d_ff, bias=False)
        self.w_down = nn.Linear(d_ff, d_model, bias=False)

    def forward(self, x):
        # Swish activation (aussi appelé SiLU)
        gate = F.silu(self.w_gate(x))  # (batch, seq, d_ff)
        up = self.w_up(x)               # (batch, seq, d_ff)

        # Gating : multiplication élément par élément
        hidden = gate * up

        # Projection vers d_model
        output = self.w_down(hidden)
        return output

# Exemple Llama-like
d_model = 4096
swiglu = SwiGLU(d_model)

x = torch.randn(1, 128, d_model)
output = swiglu(x)
print(f"Output shape: {output.shape}")  # (1, 128, 4096)

# Vérifier la dimension intermédiaire
print(f"FFN hidden dim: {swiglu.w_gate.out_features}")  # ~11008 pour Llama 7B
                

LayerNorm(x) = γ × (x - μ) / √(σ² + ε) + β

Où :
- μ = moyenne sur les features (d_model)
- σ² = variance sur les features
- γ, β = paramètres apprenables (scale & shift)
- ε = constante pour stabilité numérique (1e-5)
                

class LayerNorm(nn.Module):
    """LayerNorm classique"""
    def __init__(self, d_model, eps=1e-5):
        super().__init__()
        self.eps = eps
        self.gamma = nn.Parameter(torch.ones(d_model))
        self.beta = nn.Parameter(torch.zeros(d_model))

    def forward(self, x):
        # x: (batch, seq_len, d_model)
        mean = x.mean(dim=-1, keepdim=True)
        var = x.var(dim=-1, keepdim=True, unbiased=False)

        # Normalisation
        x_norm = (x - mean) / torch.sqrt(var + self.eps)

        # Scale and shift
        return self.gamma * x_norm + self.beta
                

RMSNorm(x) = γ × x / RMS(x)

Où RMS(x) = √(1/n × Σx²)

Différences avec LayerNorm :
- Pas de soustraction de la moyenne (μ)
- Pas de paramètre β (shift)
- Plus rapide et plus stable
                

class RMSNorm(nn.Module):
    """RMSNorm (Llama style)"""
    def __init__(self, d_model, eps=1e-6):
        super().__init__()
        self.eps = eps
        self.weight = nn.Parameter(torch.ones(d_model))

    def forward(self, x):
        # x: (batch, seq_len, d_model)
        # Calculer RMS
        rms = torch.sqrt(torch.mean(x ** 2, dim=-1, keepdim=True) + self.eps)

        # Normaliser et scale
        x_norm = x / rms
        return self.weight * x_norm

# Comparaison performance
import time

d_model = 4096
batch_size = 8
seq_len = 2048

x = torch.randn(batch_size, seq_len, d_model).cuda()

# LayerNorm
ln = nn.LayerNorm(d_model).cuda()
start = time.time()
for _ in range(100):
    _ = ln(x)
ln_time = time.time() - start

# RMSNorm
rms = RMSNorm(d_model).cuda()
start = time.time()
for _ in range(100):
    _ = rms(x)
rms_time = time.time() - start

print(f"LayerNorm: {ln_time:.4f}s")
print(f"RMSNorm: {rms_time:.4f}s")
print(f"Speedup: {ln_time/rms_time:.2f}x")
                

class TransformerBlock(nn.Module):
    """Bloc Transformer complet (Llama-style)"""
    def __init__(self, d_model, num_heads, d_ff=None, dropout=0.1):
        super().__init__()

        # Attention
        self.attention = MultiHeadAttentionWithRoPE(d_model, num_heads)
        self.attn_norm = RMSNorm(d_model)

        # Feed-Forward
        self.ffn = SwiGLU(d_model, d_ff)
        self.ffn_norm = RMSNorm(d_model)

        self.dropout = nn.Dropout(dropout)

    def forward(self, x, mask=None):
        # Attention block (Pre-Norm)
        residual = x
        x = self.attn_norm(x)
        x = self.attention(x, mask)
        x = self.dropout(x)
        x = residual + x  # Residual connection

        # FFN block (Pre-Norm)
        residual = x
        x = self.ffn_norm(x)
        x = self.ffn(x)
        x = self.dropout(x)
        x = residual + x  # Residual connection

        return x

# Stack de N blocs
class TransformerStack(nn.Module):
    def __init__(self, num_layers, d_model, num_heads):
        super().__init__()
        self.layers = nn.ModuleList([
            TransformerBlock(d_model, num_heads)
            for _ in range(num_layers)
        ])
        self.final_norm = RMSNorm(d_model)

    def forward(self, x, mask=None):
        for layer in self.layers:
            x = layer(x, mask)
        x = self.final_norm(x)
        return x

# Exemple : Llama-7B like
num_layers = 32
d_model = 4096
num_heads = 32

model = TransformerStack(num_layers, d_model, num_heads)
print(f"Total parameters: {sum(p.numel() for p in model.parameters()) / 1e9:.2f}B")
                

Algorithme BPE :

1. Initialisation : chaque byte/caractère est un token
   "low" → ['l', 'o', 'w']

2. Compter les paires consécutives
   "low low low lower" →
   ('l','o'): 4 fois
   ('o','w'): 4 fois
   ('w',' '): 3 fois
   ...

3. Merger la paire la plus fréquente
   ('l','o') → 'lo'
   "low" → ['lo', 'w']

4. Répéter jusqu'à atteindre la taille de vocab désirée
                

# Installation
# pip install tokenizers

from tokenizers import Tokenizer
from tokenizers.models import BPE
from tokenizers.trainers import BpeTrainer
from tokenizers.pre_tokenizers import Whitespace

# Créer et entraîner un tokenizer BPE
tokenizer = Tokenizer(BPE(unk_token="[UNK]"))
tokenizer.pre_tokenizer = Whitespace()

trainer = BpeTrainer(
    vocab_size=30000,
    min_frequency=2,
    special_tokens=["[UNK]", "[PAD]", "[CLS]", "[SEP]", "[MASK]"]
)

# Entraîner sur des fichiers texte
files = ["data/train.txt"]
tokenizer.train(files, trainer)

# Sauvegarder
tokenizer.save("my_bpe_tokenizer.json")

# Utiliser
output = tokenizer.encode("Hello, how are you?")
print(f"Tokens: {output.tokens}")
print(f"IDs: {output.ids}")

# Exemple de sortie :
# Tokens: ['Hello', ',', 'how', 'are', 'you', '?']
# IDs: [15496, 11, 703, 389, 345, 30]
                

# pip install sentencepiece

import sentencepiece as spm

# Entraîner un tokenizer
spm.SentencePieceTrainer.train(
    input='data/train.txt',
    model_prefix='my_sp_model',
    vocab_size=32000,
    model_type='bpe',  # ou 'unigram'
    character_coverage=0.9995,  # couvre 99.95% des caractères
    num_threads=16,
    train_extremely_large_corpus=True,
    # Tokens spéciaux
    pad_id=0,
    unk_id=1,
    bos_id=2,  # beginning of sequence
    eos_id=3,  # end of sequence
    user_defined_symbols=['<|im_start|>', '<|im_end|>']  # pour chat format
)

# Charger et utiliser
sp = spm.SentencePieceProcessor()
sp.load('my_sp_model.model')

# Encoder
text = "Hello world! 你好世界"
ids = sp.encode(text, out_type=int)
tokens = sp.encode(text, out_type=str)

print(f"IDs: {ids}")
print(f"Tokens: {tokens}")

# Decoder
decoded = sp.decode(ids)
print(f"Decoded: {decoded}")

# Info sur le vocab
print(f"Vocab size: {sp.vocab_size()}")
print(f"Token 1000: {sp.id_to_piece(1000)}")
                

# pip install tiktoken

import tiktoken

# Charger un tokenizer pré-entraîné
# GPT-4
enc = tiktoken.get_encoding("cl100k_base")

# Encoder
text = "Hello world! This is a test."
tokens = enc.encode(text)
print(f"Tokens: {tokens}")
print(f"Num tokens: {len(tokens)}")

# Decoder
decoded = enc.decode(tokens)
print(f"Decoded: {decoded}")

# Voir les tokens individuels
for token_id in tokens:
    token_bytes = enc.decode_single_token_bytes(token_id)
    print(f"{token_id}: {token_bytes}")

# Compter les tokens (utile pour API)
def count_tokens(text, model="gpt-4"):
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

prompt = "Explain quantum computing in simple terms."
num_tokens = count_tokens(prompt)
print(f"Prompt tokens: {num_tokens}")
                

Exemple : "International cooperation"

Vocab 10k :  ['Int', 'ernation', 'al', 'co', 'oper', 'ation'] → 6 tokens
Vocab 32k :  ['International', 'cooperation'] → 2 tokens
Vocab 100k : ['International', 'cooperation'] → 2 tokens

Trade-offs :
✓ Vocab plus grand → Moins de tokens → Contexte plus long
✗ Vocab plus grand → Embedding matrix plus grande → Plus de paramètres
✗ Vocab plus grand → Tokens rares mal appris
                

import tiktoken

enc = tiktoken.get_encoding("cl100k_base")

# Comparer l'efficacité par langue
texts = {
    "English": "The quick brown fox jumps over the lazy dog.",
    "French": "Le renard brun rapide saute par-dessus le chien paresseux.",
    "Chinese": "快速的棕色狐狸跳过懒狗。",
    "Arabic": "الثعلب البني السريع يقفز فوق الكلب الكسول.",
    "Code": "def factorial(n):\n    return 1 if n == 0 else n * factorial(n-1)"
}

for lang, text in texts.items():
    tokens = enc.encode(text)
    ratio = len(tokens) / len(text.split())
    print(f"{lang:10} | Chars: {len(text):3} | Tokens: {len(tokens):3} | Ratio: {ratio:.2f}")

# Résultat typique :
# English    | Chars:  44 | Tokens:  10 | Ratio: 1.11
# French     | Chars:  62 | Tokens:  15 | Ratio: 1.67
# Chinese    | Chars:  15 | Tokens:  14 | Ratio: 14.00
# Arabic     | Chars:  50 | Tokens:  23 | Ratio: 2.88
# Code       | Chars:  70 | Tokens:  24 | Ratio: 3.43
                

Llama 2 Chat :
[INST] <>
You are a helpful assistant.
<>

What is the capital of France? [/INST] The capital of France is Paris.

GPT-4 (ChatML format) :
<|im_start|>system
You are a helpful assistant.
<|im_end|>
<|im_start|>user
What is the capital of France?
<|im_end|>
<|im_start|>assistant
The capital of France is Paris.
<|im_end|>

Mistral Instruct :
[INST] What is the capital of France? [/INST] The capital of France is Paris.
                

from tokenizers import Tokenizer, models, trainers, pre_tokenizers, processors

# Créer un tokenizer BPE from scratch
tokenizer = Tokenizer(models.BPE())

# Pré-tokenization (split sur espaces et ponctuation)
tokenizer.pre_tokenizer = pre_tokenizers.ByteLevel(add_prefix_space=False)

# Entraîner
trainer = trainers.BpeTrainer(
    vocab_size=32000,
    min_frequency=2,
    special_tokens=["<|endoftext|>", "<|padding|>"],
    show_progress=True
)

files = ["data/my_corpus.txt"]
tokenizer.train(files, trainer)

# Post-processing (ajouter tokens spéciaux)
tokenizer.post_processor = processors.ByteLevel(trim_offsets=False)

# Sauvegarder
tokenizer.save("my_custom_tokenizer.json")

# Tester
text = "This is my domain-specific tokenizer!"
output = tokenizer.encode(text)
print(f"Tokens: {output.tokens}")
print(f"IDs: {output.ids}")
                

L(N, D) = (Nc / N)^αN + (Dc / D)^αD + E

Où :
- L : perte (loss) du modèle
- N : nombre de paramètres (non-embedding)
- D : nombre de tokens d'entraînement
- Nc, Dc, E : constantes empiriques
- αN ≈ 0.076, αD ≈ 0.095

Conclusion : Loss diminue avec N et D selon des power laws
                

FLOPs ≈ 6 × N × D

Où :
- N : nombre de paramètres
- D : nombre de tokens d'entraînement

Le facteur 6 vient de :
- 2 FLOPs pour forward pass (1 multiply-add = 2 FLOPs)
- 4 FLOPs pour backward pass (2× forward pour gradients)
                

def compute_flops(num_params, num_tokens):
    """
    Calcule les FLOPs nécessaires pour entraîner un modèle
    """
    flops = 6 * num_params * num_tokens
    return flops

def flops_to_gpu_days(flops, gpu_tflops):
    """
    Convertit FLOPs en jours-GPU

    Args:
        flops: nombre de FLOPs
        gpu_tflops: TFLOPs de la GPU (ex: A100 = 312 TFLOPs en FP16)
    """
    seconds = flops / (gpu_tflops * 1e12)
    days = seconds / 86400
    return days

# Exemple : Llama 2 7B
num_params = 7e9
num_tokens = 2e12

total_flops = compute_flops(num_params, num_tokens)
print(f"Total FLOPs: {total_flops:.2e}")
print(f"Total PFLOPs: {total_flops/1e15:.2f}")

# Sur A100 (312 TFLOPs en FP16)
gpu_days = flops_to_gpu_days(total_flops, 312)
print(f"GPU-days (A100): {gpu_days:.0f}")

# Sur H100 (989 TFLOPs en FP16)
gpu_days_h100 = flops_to_gpu_days(total_flops, 989)
print(f"GPU-days (H100): {gpu_days_h100:.0f}")

# Coût approximatif (A100 ≈ $3/heure sur cloud)
cost_usd = (gpu_days / 1) * 24 * 3
print(f"Coût estimé (A100 cloud): ${cost_usd:,.0f}")

# Résultats :
# Total FLOPs: 8.40e+22
# Total PFLOPs: 84.00
# GPU-days (A100): 3122
# GPU-days (H100): 985
# Coût estimé (A100 cloud): $224,784
                

import numpy as np
import matplotlib.pyplot as plt

def chinchilla_optimal(compute_budget_flops):
    """
    Calcule N et D optimaux pour un budget compute donné
    selon la loi de Chinchilla
    """
    # D = 20 * N (Chinchilla)
    # FLOPs = 6 * N * D = 6 * N * 20N = 120N²
    # N = sqrt(FLOPs / 120)

    N = np.sqrt(compute_budget_flops / 120)
    D = 20 * N

    return N, D

# Exemple : Budget de 1e23 FLOPs (≈ Llama 2 70B)
budget = 1e23

N_opt, D_opt = chinchilla_optimal(budget)
print(f"Optimal selon Chinchilla :")
print(f"  Paramètres : {N_opt/1e9:.1f}B")
print(f"  Tokens : {D_opt/1e12:.2f}T")

# Comparer avec différentes stratégies
strategies = {
    "Kaplan (sous-entraîné)": (175e9, 300e9),
    "Chinchilla (optimal)": (N_opt, D_opt),
    "Llama 2 (sur-entraîné)": (70e9, 2e12),
}

for name, (N, D) in strategies.items():
    flops = 6 * N * D
    ratio = D / N
    print(f"\n{name}:")
    print(f"  N: {N/1e9:.1f}B, D: {D/1e12:.2f}T")
    print(f"  Ratio D/N: {ratio:.1f}×")
    print(f"  FLOPs: {flops:.2e}")

# Visualisation
plt.figure(figsize=(10, 6))
params = np.logspace(9, 12, 100)  # 1B to 1T

for budget_name, budget_flops in [("1e22 (7B optimal)", 1e22),
                                    ("1e23 (70B optimal)", 1e23)]:
    tokens = budget_flops / (6 * params)
    plt.loglog(params/1e9, tokens/1e12, label=budget_name)

# Chinchilla line
plt.loglog(params/1e9, 20*params/1e12, 'r--', label='Chinchilla (D=20N)', linewidth=2)

plt.xlabel('Paramètres (B)')
plt.ylabel('Tokens (T)')
plt.title('Scaling Laws: Paramètres vs Tokens')
plt.legend()
plt.grid(True, alpha=0.3)
plt.savefig('scaling_laws.png', dpi=300)
                

# Télécharger Common Crawl
import requests
from warcio.archiveiterator import ArchiveIterator

def download_common_crawl(warc_url):
    response = requests.get(warc_url, stream=True)
    for record in ArchiveIterator(response.raw):
        if record.rec_type == 'response':
            url = record.rec_headers.get_header('WARC-Target-URI')
            content = record.content_stream().read()
            yield url, content

# Exemple
warc_url = "https://data.commoncrawl.org/crawl-data/CC-MAIN-2024-10/segments/.../warc/..."
for url, html in download_common_crawl(warc_url):
    print(f"Downloaded: {url}")
                

C - Capacity and Role : "Agis en tant qu'expert Python..."
R - Insight : "Je veux créer une API REST..."
I - Statement : "Génère le code avec FastAPI..."
S - Personality : "Utilise un style professionnel..."
P - Parameters : "Format JSON, 50 lignes max..."
E - Experiment : "Propose 2 alternatives..."
                

# Exemple de prompt CRISPE complet
prompt = """
Agis en tant qu'expert en architecture logicielle avec 15 ans d'expérience (Capacity).

Je développe une plateforme e-commerce qui doit gérer 10000 requêtes/seconde
avec une latence <100ms. J'hésite entre architecture microservices et monolithe (Insight).

Analyse ces deux approches et recommande la meilleure solution pour mon cas (Statement).

Sois pragmatique et concis, avec des arguments chiffrés (Personality).

Format :
- Avantages/Inconvénients de chaque approche (tableau)
- Recommandation finale justifiée
- Maximum 300 mots (Parameters)

Si pertinent, propose une architecture hybride (Experiment).
"""