<h1 id="GPT_corr">GPT_corr¶</h1><p>Objectif : repartir du TP 1 (régression logistique) pour implémenter GPT-2 sur les prénoms</p>
<p>Programme :</p>
- algèbre linéaire en Python pur ;
- implémentattion de la différentiation automatique par ordre topologique inverse ;
- réimplémenter la régression logistique avec la différentiation automatique ;
- implémentation GPT (attention naïve + KV cache) ;
- entraînement sur le même nombre de steps et comparaison (courbes + éval + générations).
In [1]:
import math
import os
import random
import time
import matplotlib.pyplot as plt
<h2 id="Donn%C3%A9es-+-tokens">Données + tokens¶</h2><p>Tokens = caractères + token
stop (début/fin). Donc (T = 27).</p>
In [2]:
random.seed(42)
# noms (dataset makemore)
if not os.path.exists('input.txt'):
import urllib.request
names_url = 'https://raw.githubusercontent.com/karpathy/makemore/988aa59/names.txt'
urllib.request.urlretrieve(names_url, 'input.txt')
names = [line.strip() for line in open('input.txt') if line.strip()]
random.shuffle(names)
print(f"num names: {len(names)}")
print(names[:8])
characters = sorted(set(''.join(names)))
stop_token = len(characters)
T = len(characters) + 1
print(f"vocab size: T = {T}")
def encode(name):
return [stop_token] + [characters.index(ch) for ch in name] + [stop_token]
print('anna ->', encode('anna'))
num names: 32033 ['yuheng', 'diondre', 'xavien', 'jori', 'juanluis', 'erandi', 'phia', 'samatha'] vocab size: T = 27 anna -> [26, 0, 13, 13, 0, 26]
<h2 id="0.-Alg%C3%A8bre-lin%C3%A9aire-en-Python-(snippets)">0. Algèbre linéaire en Python (snippets)¶</h2><p>Vecteurs = listes, matrices = listes de lignes.</p>
In [3]:
# vecteurs
rand_vec = lambda d, std=1: [random.gauss(0, std) for _ in range(d)]
one_hot = lambda t, T: [1.0 if i == t else 0.0 for i in range(T)]
ext_prod = lambda alpha, u: [alpha * ui for ui in u]
sum_vec = lambda u, v: [ui + vi for ui, vi in zip(u, v)]
dot_prod = lambda u, v: sum(ui * vi for ui, vi in zip(u, v))
norm = lambda v: math.sqrt(dot_prod(v, v))
# matrices
rand_mat = lambda n, m, std=1: [rand_vec(m, std) for _ in range(n)]
transpose = lambda M: list(zip(*M))
mat_vec = lambda M, v: [dot_prod(mi_, v) for mi_ in M]
mat_mat = lambda A, B: [[dot_prod(ai_, b_j) for b_j in zip(*B)] for ai_ in A]
def softmax(v):
# snippet du poly + stabilité numérique (translation invariance)
m = max(v)
exp_v = [math.exp(vk - m) for vk in v]
return ext_prod(1 / sum(exp_v), exp_v)
def moving_average(x, k=100):
out, s = [], 0.0
from collections import deque
win = deque()
for xi in x:
win.append(xi); s += xi
if len(win) > k: s -= win.popleft()
out.append(s / len(win))
return out
<h2 id="1.-Feature-vector-(r%C3%A9gression-logistique)">1. Feature vector (régression logistique)¶</h2><p>$v(t_0,\dots,t_{l-1}) = \frac{1}{l}\sum_{k=0}^{l-1} \mathbf{1}_{t_k}$.</p>
In [4]:
def feature_vector(tokens, T):
sum_tokens = [0.0] * T
for token in tokens:
sum_tokens[token] += 1.0
# sum_tokens = sum_vec(sum_tokens, one_hot(token, T)) # autre façon
return ext_prod(1 / len(tokens), sum_tokens)
<h2 id="2.-Diff%C3%A9rentiation-automatique-(ordre-topologique-inverse)">2. Différentiation automatique (ordre topologique inverse)¶</h2><p>Classe
Value inspirée du poly : opérations élémentaires, puis backward() avec build_topo + parcours en ordre topologique inverse.</p>
In [5]:
class Value:
# Noms alignés sur le poly : value, dloss_dself, parents, dself_dparents
def __init__(self, value, parents=(), dself_dparents=()):
self.value = value
self.dloss_dself = None
self.parents = parents
self.dself_dparents = dself_dparents
# operations a deux arguments (une constante est transformee en noeud)
def __add__(self, other):
other = other if isinstance(other, Value) else Value(other)
return Value(self.value + other.value, parents=(self, other), dself_dparents=(1, 1))
def __mul__(self, other):
other = other if isinstance(other, Value) else Value(other)
return Value(self.value * other.value, (self, other), dself_dparents=(other.value, self.value))
# operations d'une variable
def __pow__(self, n): return Value(self.value**n, (self,), (n * self.value**(n-1),))
def exp(self): return Value(math.exp(self.value), (self,), (math.exp(self.value),))
def log(self): return Value(math.log(self.value), (self,), (1 / self.value,))
def relu(self): return Value(max(0, self.value), (self,), (float(self.value > 0),))
# operations derivees
def __neg__(self): return self * -1
def __radd__(self, other): return self + other
def __sub__(self, other): return self + (-other)
def __rsub__(self, other): return other + (-self)
def __rmul__(self, other): return self * other
def __truediv__(self, other): return self * other ** -1
def __rtruediv__(self, other): return other * self ** -1
def build_topo(self, topo=None, visited=None):
if topo is None: topo = []
if visited is None: visited = set()
if self not in visited:
visited.add(self)
self.dloss_dself = 0
for parent in self.parents:
parent.build_topo(topo, visited)
topo.append(self)
return topo, visited
def backward(self):
topo, _ = self.build_topo()
self.dloss_dself = 1
for v in reversed(topo):
for parent, dself_dparent in zip(v.parents, v.dself_dparents):
parent.dloss_dself += dself_dparent * v.dloss_dself
<p>Premier exemple, fonction a une variable
$$\mathcal{L}(x) = \exp( \sqrt{x^2 + 1})$$</p>
In [6]:
# calcul explicite
x_ = math.sqrt(3)
loss_ = math.exp(math.sqrt(x_**2 + 1))
loss_prime_ = x_ / math.sqrt(x_**2 + 1) * loss_
print(f"loss = {loss_:.7f}, deriv = {loss_prime_:.7f}")
# differentiation automatique
x = Value(math.sqrt(3))
loss = ((x**2 + 1) ** 0.5).exp()
loss.backward()
print(f"loss = {loss.value:.7f}, deriv = {x.dloss_dself:.7f}")
loss = 7.3890561, deriv = 6.3991103 loss = 7.3890561, deriv = 6.3991103
<p>Deuxième exemple, fonction a deux variables
$$\mathcal{L}(x, y) = 2\exp(x + y) - 0.5\exp(x_- y) + 1.1\exp(-2x + y)$$</p>
In [7]:
# test numerique pour x=1, y=2
x_ = 1.0
y_ = 2.0
z_ = -1
# calcul explicite de la loss et de ses derivees partielles
loss_ = 2*math.exp(x_ + y_) - 0.5*math.exp(x_ - y_) + 1.1*math.exp(-2*x_ + y_)
dloss_dx_ = 2*math.exp(x_ + y_) - 0.5*math.exp(x_ - y_) - 2.2*math.exp(-2*x_ + y_)
dloss_dy_ = 2*math.exp(x_ + y_) + 0.5*math.exp(x_ - y_) + 1.1*math.exp(-2*x_ + y_)
# loss et ses derivees partielles par differentiation automatique
x = Value(x_)
y = Value(y_)
z = Value(z_)
loss = 2*(x + y).exp() - 0.5*(x - y).exp() + 1.1*(-2*x + y).exp()
loss.backward()
print(f"loss = {loss.value:.7f} (attendu {loss_:.7f})")
print(f"dl/dx = {x.dloss_dself:.7f} (attendu {dloss_dx_:.7f})")
print(f"dl/dy = {y.dloss_dself:.7f} (attendu {dloss_dy_:.7f})")
loss = 41.0871341 (attendu 41.0871341) dl/dx = 37.7871341 (attendu 37.7871341) dl/dy = 41.4550136 (attendu 41.4550136)
<h2 id="2-Retour-sur-la-r%C3%A9gression-logistique-avec-differentiation-automatique">2 Retour sur la régression logistique avec differentiation automatique¶</h2><p>Modèle :
$p = \mathrm{softmax}(M v)$, loss = moyenne des $-\log p_{t_l}$ sur les tokens.</p>
<p>Entraînement par descente de gradient stochastique</p>
In [8]:
# fonction softmax calculée avec value
def softmax_value(z):
m = max(zi.value for zi in z) # stabilite numerique
exp_z = [(zi - m).exp() for zi in z]
s = sum(exp_z)
return [e / s for e in exp_z]
In [9]:
def loss_logreg_autodiff(M, tokens):
losses = []
for l in range(len(tokens) - 1):
v = feature_vector(tokens[:l + 1], T) # float
z = mat_vec(M, v) # Value
p = softmax_value(z) # Value
losses.append(-p[tokens[l + 1]].log())
return sum(losses) * (1 / len(losses))
def generate_name_logreg(M_float, max_chars=100, temperature=1.0):
context = [stop_token]
out = []
for _ in range(max_chars):
v = feature_vector(context, T)
z = mat_vec(M_float, v)
p = softmax([zi / temperature for zi in z])
t = random.choices(range(T), weights=p)[0]
if t == stop_token:
break
out.append(characters[t])
context.append(t)
return ''.join(out)
learning_rate_logreg = 1.0
num_steps = 1000
print("Initializing matrix M to zero matrix...")
M = [[Value(0.0) for _ in range(T)] for _ in range(T)]
params_logreg = [p for row in M for p in row]
print(f"Number of parameters: {len(params_logreg)}")
hist_logreg = []
t0 = time.time()
for step in range(num_steps):
loss = loss_logreg_autodiff(M, encode(names[step]))
loss.backward()
for p in params_logreg:
p.value -= learning_rate_logreg * p.dloss_dself
p.dloss_dself = 0.0
hist_logreg.append(loss.value)
if step % 100 == 0 or step == num_steps - 1:
print(f"logreg | step {step + 1:5d}/{num_steps} | loss {loss.value:.4f} ({sum(hist_logreg[-100:])/100:.4f})", end='\r')
print(f"\nlogreg done in {time.time() - t0:.2f}s")
M_logreg = [[p.value for p in row] for row in M]
random.seed(42)
print([generate_name_logreg(M_logreg) for _ in range(10)])
Initializing matrix M to zero matrix... Number of parameters: 729 logreg | step 1000/1000 | loss 2.6635 (2.6027) logreg done in 11.55s ['laheyv', 'alainai', 'kbmsavrhb', 'ead', 'lrom', 'gltnyouagibgchskkeh', 'mjarbi', 'ljnxteaiig', 'seol', 'iddner']
<h2 id="3.-Le-transformer">3. Le transformer¶</h2><p>Architecture : GPT-2 simplifié (RMSNorm, pas de biais, ReLU). Implémentation naïve + KV cache (pour génération).</p>
In [10]:
n_layer = 1
n_embd = 16
block_size = 16
n_head = 4
head_dim = n_embd // n_head
random.seed(42)
param_mat = lambda n, m, std=0.08: [[Value(random.gauss(0, std)) for _ in range(m)] for _ in range(n)]
state_dict = {
'wte': param_mat(T, n_embd),
'wpe': param_mat(block_size, n_embd),
'lm_head': param_mat(T, n_embd),
}
for li in range(n_layer):
state_dict[f'layer{li}.attn_wq'] = param_mat(n_embd, n_embd)
state_dict[f'layer{li}.attn_wk'] = param_mat(n_embd, n_embd)
state_dict[f'layer{li}.attn_wv'] = param_mat(n_embd, n_embd)
state_dict[f'layer{li}.attn_wo'] = param_mat(n_embd, n_embd)
state_dict[f'layer{li}.mlp_fc1'] = param_mat(4 * n_embd, n_embd)
state_dict[f'layer{li}.mlp_fc2'] = param_mat(n_embd, 4 * n_embd)
params_gpt = [p for mat in state_dict.values() for row in mat for p in row]
print(f"nombre de parametres : {len(params_gpt)}")
def embed(token, pos):
return sum_vec(state_dict['wte'][token], state_dict['wpe'][pos])
def rmsnorm(x):
ms = sum(xi * xi for xi in x) / len(x)
scale = (ms + 1e-5) ** -0.5
return [xi * scale for xi in x]
def attention_head(q, ks, vs):
scores = [dot_prod(q, k) / math.sqrt(len(q)) for k in ks]
w = softmax_value(scores)
return mat_vec(transpose(vs), w)
def attention_naive_1head(X, Wq, Wk, Wv):
Q = [mat_vec(Wq, x) for x in X]
K = [mat_vec(Wk, x) for x in X]
V = [mat_vec(Wv, x) for x in X]
return [attention_head(Q[l], K[:l + 1], V[:l + 1]) for l in range(len(X))]
def multi_head(q, ks, vs, n_head):
d_h = len(q) // n_head
out = []
for h in range(n_head):
hs = slice(h * d_h, (h + 1) * d_h)
out += attention_head(q[hs], [k[hs] for k in ks], [v[hs] for v in vs])
return out
def attention_naive(X, Wq, Wk, Wv, Wo, n_head):
Q = [mat_vec(Wq, x) for x in X]
K = [mat_vec(Wk, x) for x in X]
V = [mat_vec(Wv, x) for x in X]
return [mat_vec(Wo, multi_head(Q[l], K[:l + 1], V[:l + 1], n_head)) for l in range(len(X))]
def mlp(x, W1, W2):
h = [hi.relu() for hi in mat_vec(W1, x)]
return mat_vec(W2, h)
def gpt_naive(tokens):
sd = state_dict
X = [rmsnorm(embed(t, l)) for l, t in enumerate(tokens)]
for li in range(n_layer):
A = attention_naive([rmsnorm(x) for x in X],
sd[f'layer{li}.attn_wq'], sd[f'layer{li}.attn_wk'],
sd[f'layer{li}.attn_wv'], sd[f'layer{li}.attn_wo'], n_head)
X = [sum_vec(x, a) for x, a in zip(X, A)]
X = [sum_vec(x, mlp(rmsnorm(x), sd[f'layer{li}.mlp_fc1'], sd[f'layer{li}.mlp_fc2'])) for x in X]
return [mat_vec(sd['lm_head'], x) for x in X]
def attention_kv(x, Wq, Wk, Wv, Wo, n_head, keys, values):
q, k, v = mat_vec(Wq, x), mat_vec(Wk, x), mat_vec(Wv, x)
keys.append(k)
values.append(v)
return mat_vec(Wo, multi_head(q, keys, values, n_head))
def gpt_kv(token, pos, keys, values):
sd = state_dict
x = rmsnorm(embed(token, pos))
for li in range(n_layer):
a = attention_kv(rmsnorm(x),
sd[f'layer{li}.attn_wq'], sd[f'layer{li}.attn_wk'],
sd[f'layer{li}.attn_wv'], sd[f'layer{li}.attn_wo'], n_head,
keys[li], values[li])
x = sum_vec(x, a)
x = sum_vec(x, mlp(rmsnorm(x), sd[f'layer{li}.mlp_fc1'], sd[f'layer{li}.mlp_fc2']))
return mat_vec(sd['lm_head'], x)
def sample(logits, temperature):
p = softmax_value([z / temperature for z in logits])
return random.choices(range(T), weights=[pi.value for pi in p])[0]
def generate_kv(temperature=0.8):
keys, values = [[] for _ in range(n_layer)], [[] for _ in range(n_layer)]
token, out = stop_token, []
for pos in range(block_size):
token = sample(gpt_kv(token, pos, keys, values), temperature)
if token == stop_token:
break
out.append(characters[token])
return ''.join(out)
nombre de parametres : 4192
<h2 id="4-Entra%C3%AEnement-GPT">4 Entraînement GPT¶</h2><p>Méthode ADAM</p>
<p>Même NLL que la logreg, gradient via
loss.backward().</p>
In [11]:
def loss_gpt_naive(tokens):
logits = gpt_naive(tokens)
# logits = feature_vector(tokens)
losses = []
for l in range(len(tokens) - 1):
p = softmax_value(logits[l])
losses.append(-p[tokens[l + 1]].log())
return sum(losses) * (1 / len(losses))
def loss_gpt_kv(tokens):
# version sequence + KV cache : on calcule position par position
keys, values = [[] for _ in range(n_layer)], [[] for _ in range(n_layer)]
losses = []
for pos, token in enumerate(tokens[:-1]):
logits = gpt_kv(token, pos, keys, values) # logits de la position courante
p = softmax_value(logits)
losses.append(-p[tokens[pos + 1]].log())
return sum(losses) * (1 / len(losses))
In [12]:
learning_rate_gpt, beta1, beta2, eps_adam = 0.01, 0.85, 0.99, 1e-8
m_adam = [0.0] * len(params_gpt)
v_adam = [0.0] * len(params_gpt)
loss_gpt = loss_gpt_kv
hist_gpt = []
t0 = time.time()
for step in range(num_steps):
# en entraînement, on utilise la version naïve : toute la séquence est connue
loss = loss_gpt(encode(names[step]))
loss.backward()
lr_t = learning_rate_gpt * (1 - step / num_steps) # decroissance lineaire
for i, p in enumerate(params_gpt):
if p.dloss_dself is None:
continue
g = p.dloss_dself
m_adam[i] = beta1 * m_adam[i] + (1 - beta1) * g
v_adam[i] = beta2 * v_adam[i] + (1 - beta2) * g * g
m_hat = m_adam[i] / (1 - beta1 ** (step + 1))
v_hat = v_adam[i] / (1 - beta2 ** (step + 1))
p.value -= lr_t * m_hat / (v_hat ** 0.5 + eps_adam)
p.dloss_dself = None
hist_gpt.append(loss.value)
if step % 50 == 0 or step == num_steps - 1:
print(f"gpt | step {step + 1:5d}/{num_steps} | loss {loss.value:.4f}", end='\r')
print(f"\ngpt done in {time.time() - t0:.2f}s")
random.seed(42)
print([generate_kv(temperature=0.8) for _ in range(10)])
gpt | step 1000/1000 | loss 2.5641 gpt done in 80.95s ['makar', 'sanalia', 'main', 'arlei', 'edasn', 'neyin', 'lonnan', 'akalin', 'beynon', 'alynn']
<h2 id="5-Comparaison-GPT-vs-log-reg">5 Comparaison GPT vs log-reg¶</h2><p>On compare les courbes NLL et une évaluation sur des prénoms non vus (les
eval_names après les num_steps premiers).</p>
In [13]:
avg_window = 10
plt.figure(figsize=(8, 4))
plt.plot(moving_average(hist_logreg, avg_window), label='logreg (autodiff)')
plt.plot(moving_average(hist_gpt, avg_window), label='gpt (adam)')
plt.axhline(math.log(T), color='gray', ls='--', label='log(T) (uniforme)')
plt.xlabel('etape')
plt.ylabel('NLL')
plt.title('Loss training curve')
plt.legend()
plt.tight_layout()
plt.show()
In [14]:
def eval_logreg_loss(M_float, names_list):
total, n = 0.0, 0
for nm in names_list:
toks = encode(nm)
for l in range(len(toks) - 1):
v = feature_vector(toks[:l + 1], T)
z = mat_vec(M_float, v)
p = softmax(z)
total += -math.log(p[toks[l + 1]])
n += 1
return total / n
def eval_gpt_loss(names_list):
total, n = 0.0, 0
for nm in names_list:
toks = encode(nm)
logits = gpt_naive(toks)
for l in range(len(toks) - 1):
p = softmax_value(logits[l])
total += -math.log(p[toks[l + 1]].value)
n += 1
return total / n
n_eval_names = 50
eval_names = names[-n_eval_names:]
logreg_eval = eval_logreg_loss(M_logreg, eval_names)
gpt_eval = eval_gpt_loss(eval_names)
print(f"logreg final train loss (avg {n_eval_names}): {sum(hist_logreg[-n_eval_names:])/n_eval_names:.4f} | eval NLL {n_eval_names} samples: {logreg_eval:.4f}")
print(f"gpt final train loss (avg {n_eval_names}): {sum(hist_gpt[-n_eval_names:])/n_eval_names:.4f} | eval NLL {n_eval_names} samples: {gpt_eval:.4f}")
logreg final train loss (avg 50): 2.6433 | eval NLL 50 samples: 2.6903 gpt final train loss (avg 50): 2.3373 | eval NLL 50 samples: 2.3201