Implementazione di un semplice framework di rete neurale da zero
Nonostante abbia svolto per un po’ di tempo lavoro e ricerca nell’ecosistema dell’intelligenza artificiale, fino a poco tempo fa non mi ero veramente fermato a pensare alla propagazione all’indietro e agli aggiornamenti del gradiente all’interno delle reti neurali. Questo articolo cerca di rimediare a questo problema e, si spera, fornirà un’immersione approfondita ma facile da seguire nell’argomento implementando da zero un semplice (ma piuttosto potente) framework di rete neurale.
Fondamentalmente, una rete neurale è semplicemente una funzione matematica che collega il nostro spazio di input allo spazio di output desiderato. In effetti, possiamo effettivamente “sformare” qualsiasi rete neurale in una funzione. Consideriamo, ad esempio, la seguente semplice rete neurale con due strati e un input:
Possiamo ora costruire una funzione equivalente procedendo strato per strato, partendo dall’input. Seguiamo la nostra funzione finale strato per strato:
- All’input, iniziamo con la funzione identità pred(x) = x
- Al primo strato lineare, otteniamo pred(x) = w₁x+b₁
- La ReLU ci mette in rete pred(x) = max(0, w₁x+b₁)
- Allo strato finale, otteniamo pred(x) = w₂(max(0, w₁x+b₁)) + B₂
Con reti più complicate, queste funzioni ovviamente diventano ingombranti, ma il punto è che possiamo costruire tali rappresentazioni delle reti neurali.
Possiamo però fare un ulteriore passo avanti: le funzioni di questa forma non sono estremamente convenienti per il calcolo, ma possiamo analizzarle in una forma più utile, vale a dire un albero di sintassi. Per la nostra rete semplice, l’albero sarebbe simile a questo:
In questa forma ad albero, le nostre foglie sono parametri, costanti e input, mentre gli altri nodi lo sono operazioni elementari i cui argomenti sono i loro figli. Naturalmente, queste operazioni elementari non devono essere binarie: l’operazione sigmoide, ad esempio, è unaria (e lo è anche ReLU se non la rappresentiamo come un massimo di 0 e x), e possiamo scegliere di supportare moltiplicazione e addizione di più di un input.
Pensando alla nostra rete come un albero di queste operazioni elementari, ora possiamo fare molte cose molto facilmente con la ricorsione, che costituirà la base sia dei nostri algoritmi di propagazione all’indietro che di propagazione in avanti. Nel codice, possiamo definire una classe di rete neurale ricorsiva simile a questa:
from dataclasses import dataclass, field
from typing import List@dataclass
class NeuralNetNode:
"""A node in our neural network tree"""
children: List('NeuralNetNode') = field(default_factory=list)
def op(self, x: List(float)) -> float:
"""The operation that this node performs"""
raise NotImplementedError
def forward(self) -> float:
"""Evaluate this node on the given input"""
return self.op((child.forward() for child in self.children))
# This is just for convenience
def __call__(self) -> List(float):
return self.forward()
def __repr__(self):
return f'{self.__class__.__name__}({self.children})'
Supponiamo ora di avere una funzione di perdita differenziabile per la nostra rete neurale, ad esempio MSE. Ricordiamo che MSE (per un campione) è definito come segue:
Ora desideriamo aggiornare i nostri parametri (i cerchi verdi nella nostra rappresentazione ad albero) dato il valore della nostra perdita. Per fare ciò, abbiamo bisogno della derivata della nostra funzione di perdita rispetto a ciascun parametro. Tuttavia, calcolarlo direttamente dalla perdita è estremamente difficile: dopo tutto, il nostro MSE viene calcolato in termini di valore previsto dalla nostra rete neurale, che può essere una funzione straordinariamente complicata.
È qui che entra in gioco un elemento matematico molto utile: la regola della catena. Invece di essere costretti a calcolare le nostre derivate altamente complesse fin dall’inizio, possiamo invece calcolare una serie di derivate più semplici.
Si scopre che la regola della catena si adatta molto bene alla nostra struttura ad albero ricorsiva. Fondamentalmente l’idea funziona così: supponendo di avere operazioni elementari sufficientemente semplici, ciascuna operazione elementare conosce la sua derivata rispetto a tutti i suoi argomenti. Data la derivata dell’operazione madre, possiamo quindi calcolare la derivata di ciascuna operazione figlia rispetto alla funzione di perdita attraverso una semplice moltiplicazione. Per un semplice modello di regressione lineare che utilizza MSE, possiamo rappresentarlo come segue:
Naturalmente, alcuni dei nostri nodi non fanno nulla con i loro derivati, vale a dire che se ne preoccupano solo i nostri nodi foglia. Ma ora ogni nodo può ottenere la derivata del suo output rispetto alla funzione di perdita attraverso questo processo ricorsivo. Possiamo quindi aggiungere i seguenti metodi alla nostra classe NeuralNetNode:
def grad(self) -> List(float):
"""The gradient of this node with respect to its inputs"""
raise NotImplementedErrordef backward(self, derivative_from_parent: float):
"""Propagate the derivative from the parent to the children"""
self.on_backward(derivative_from_parent)
deriv_wrt_children = self.grad()
for child, derivative_wrt_child in zip(self.children, deriv_wrt_children):
child.backward(derivative_from_parent * derivative_wrt_child)
def on_backward(self, derivative_from_parent: float):
"""Hook for subclasses to override. Things like updating parameters"""
pass
Esercizio 1: Prova a creare uno di questi alberi per un modello di regressione lineare semplice ed esegui manualmente gli aggiornamenti del gradiente ricorsivo per un paio di passaggi.
Nota: per semplicità, richiediamo che i nostri nodi abbiano un solo genitore (o nessuno). Se a ciascun nodo è consentito avere più genitori, il nostro algoritmo reverse() diventa un po’ più complicato poiché ogni figlio deve sommare la derivata dei suoi genitori per calcolare la propria. Possiamo farlo iterativamente con un ordinamento topologico (es Qui) o ancora ricorsivamente, cioè con accumulo inverso (anche se in questo caso bisognerebbe fare un secondo passaggio per aggiornare effettivamente tutti i parametri). Questo non è particolarmente difficile, quindi lo lascerò come esercizio al lettore (ne parlerò più approfonditamente nella parte 2, restate sintonizzati).
Modelli di costruzione
Il resto del nostro codice riguarda in realtà solo l’implementazione di parametri, input e operazioni e, naturalmente, l’esecuzione della nostra formazione. I parametri e gli input sono costrutti abbastanza semplici:
import random@dataclass
class Input(NeuralNetNode):
"""A leaf node that represents an input to the network"""
value: float=0.0
def op(self, x):
return self.value
def grad(self) -> List(float):
return (1.0)
def __repr__(self):
return f'{self.__class__.__name__}({self.value})'
@dataclass
class Parameter(NeuralNetNode):
"""A leaf node that represents a parameter to the network"""
value: float=field(default_factory=lambda: random.uniform(-1, 1))
learning_rate: float=0.01
def op(self, x):
return self.value
def grad(self):
return (1.0)
def on_backward(self, derivative_from_parent: float):
self.value -= derivative_from_parent * self.learning_rate
def __repr__(self):
return f'{self.__class__.__name__}({self.value})'
Le operazioni sono leggermente più complicate, anche se non troppo: dobbiamo solo calcolare correttamente i loro gradienti. Di seguito sono riportate le implementazioni di alcune operazioni utili:
import math@dataclass
class Operation(NeuralNetNode):
"""A node that performs an operation on its inputs"""
pass
@dataclass
class Add(Operation):
"""A node that adds its inputs"""
def op(self, x):
return sum(x)
def grad(self):
return (1.0) * len(self.children)
@dataclass
class Multiply(Operation):
"""A node that multiplies its inputs"""
def op(self, x):
return math.prod(x)
def grad(self):
grads = ()
for i in range(len(self.children)):
cur_grad = 1
for j in range(len(self.children)):
if i == j:
continue
cur_grad *= self.children(j).forward()
grads.append(cur_grad)
return grads
@dataclass
class ReLU(Operation):
"""
A node that applies the ReLU function to its input.
Note that this should only have one child.
"""
def op(self, x):
return max(0, x(0))
def grad(self):
return (1.0 if self.children(0).forward() > 0 else 0.0)
@dataclass
class Sigmoid(Operation):
"""
A node that applies the sigmoid function to its input.
Note that this should only have one child.
"""
def op(self, x):
return 1 / (1 + math.exp(-x(0)))
def grad(self):
return (self.forward() * (1 - self.forward()))
La superclasse operativa qui non è ancora utile, anche se ne avremo bisogno per trovare più facilmente gli input del nostro modello in seguito.
Nota quanto spesso i gradienti delle funzioni richiedono i valori dai loro figli, quindi è necessario chiamare il metodo forward() del figlio. Ne parleremo più approfonditamente tra poco.
Definire una rete neurale nel nostro framework è un po’ prolisso ma è molto simile alla costruzione di un albero. Ecco, ad esempio, il codice per un semplice classificatore lineare nel nostro framework:
linear_classifier = Add((
Multiply((
Parameter(),
Input()
)),
Parameter()
))
Utilizzando i nostri modelli
Per eseguire una previsione con il nostro modello, dobbiamo prima popolare gli input nel nostro albero e poi chiamare forward() sul genitore. Per popolare gli input, però, dobbiamo prima trovarli, quindi aggiungiamo il seguente metodo al nostro Operazione class (non lo aggiungiamo alla nostra classe NeuralNetNode poiché il tipo di input non è ancora definito lì):
def find_input_nodes(self) -> List(Input):
"""Find all of the input nodes in the subtree rooted at this node"""
input_nodes = ()
for child in self.children:
if isinstance(child, Input):
input_nodes.append(child)
elif isinstance(child, Operation):
input_nodes.extend(child.find_input_nodes())
return input_nodes
Ora possiamo aggiungere il metodo predit() alla classe Operation:
def predict(self, inputs: List(float)) -> float:
"""Evaluate the network on the given inputs"""
input_nodes = self.find_input_nodes()
assert len(input_nodes) == len(inputs)
for input_node, value in zip(input_nodes, inputs):
input_node.value = value
return self.forward()
Esercizio 2: Il modo attuale in cui abbiamo implementato predic() è alquanto inefficiente poiché dobbiamo attraversare l’albero per trovare tutti gli input ogni volta che eseguiamo predic(). Scrivere un metodo compile() che memorizzi nella cache gli input dell’operazione quando viene eseguita.
L’addestramento dei nostri modelli ora è molto semplice:
from typing import Callable, Tupledef train_model(
model: Operation,
loss_fn: Callable((float, float), float),
loss_grad_fn: Callable((float, float), float),
data: List(Tuple(List(float), float)),
epochs: int=1000,
print_every: int=100
):
"""Train the given model on the given data"""
for epoch in range(epochs):
total_loss = 0.0
for x, y in data:
prediction = model.predict(x)
total_loss += loss_fn(y, prediction)
model.backward(loss_grad_fn(y, prediction))
if epoch % print_every == 0:
print(f'Epoch {epoch}: loss={total_loss/len(data)}')
Ecco, ad esempio, come addestreremmo un classificatore lineare da Fahrenheit a Celsius utilizzando il nostro framework:
def mse_loss(y_true: float, y_pred: float) -> float:
return (y_true - y_pred) ** 2def mse_loss_grad(y_true: float, y_pred: float) -> float:
return -2 * (y_true - y_pred)
def fahrenheit_to_celsius(x: float) -> float:
return (x - 32) * 5 / 9
def generate_f_to_c_data() -> List(List(float)):
data = ()
for _ in range(1000):
f = random.uniform(-1, 1)
data.append(((f), fahrenheit_to_celsius(f)))
return data
linear_classifier = Add((
Multiply((
Parameter(),
Input()
)),
Parameter()
))
train_model(linear_classifier, mse_loss, mse_loss_grad, generate_f_to_c_data())
Dopo aver eseguito questo, otteniamo
print(linear_classifier)
print(linear_classifier.predict((32)))>> Add(children=(Multiply(children=(Parameter(0.5555555555555556), Input(0.8930639016107234))), Parameter(-17.777777777777782)))
>> -1.7763568394002505e-14
Che corrisponde correttamente a un classificatore lineare con peso 0,56, bias -17,78 (che è la formula da Fahrenheit a Celsius)
Naturalmente possiamo anche addestrare modelli molto più complessi, ad esempio eccone uno per prevedere se un punto (x, y) è sopra o sotto la linea y = x:
def bce_loss(y_true: float, y_pred: float, eps: float=0.00000001) -> float:
y_pred = min(max(y_pred, eps), 1 - eps)
return -y_true * math.log(y_pred) - (1 - y_true) * math.log(1 - y_pred)def bce_loss_grad(y_true: float, y_pred: float, eps: float=0.00000001) -> float:
y_pred = min(max(y_pred, eps), 1 - eps)
return (y_pred - y_true) / (y_pred * (1 - y_pred))
def generate_binary_data():
data = ()
for _ in range(1000):
x = random.uniform(-1, 1)
y = random.uniform(-1, 1)
data.append(((x, y), 1 if y > x else 0))
return data
model_binary = Sigmoid(
(
Add(
(
Multiply(
(
Parameter(),
ReLU(
(
Add(
(
Multiply(
(
Parameter(),
Input()
)
),
Multiply(
(
Parameter(),
Input()
)
),
Parameter()
)
)
)
)
)
),
Parameter()
)
)
)
)
train_model(model_binary, bce_loss, bce_loss_grad, generate_binary_data())
Quindi otteniamo ragionevolmente
print(model_binary.predict((1, 0)))
print(model_binary.predict((0, 1)))
print(model_binary.predict((0, 1000)))
print(model_binary.predict((-5, 3)))
print(model_binary.predict((0, 0)))>> 3.7310797619230176e-66
>> 0.9997781079343139
>> 0.9997781079343139
>> 0.9997781079343139
>> 0.23791579184662365
Sebbene abbia un’autonomia ragionevole, è un po’ più lento di quanto ci aspetteremmo. Questo perché dobbiamo chiamare forward() e ricalcolare gli input del modello molto nella chiamata a previous(). Pertanto, esegui il seguente esercizio:
Esercizio 3: aggiungi la memorizzazione nella cache alla nostra rete. Cioè, alla chiamata a forward(), il modello dovrebbe restituire il valore memorizzato nella cache dalla chiamata precedente a forward() se e solo se gli input non sono cambiati dall’ultima chiamata. Assicurati di eseguire nuovamente forward() se gli input sono cambiati.
E questo è tutto! Ora disponiamo di una struttura di rete neurale funzionante in cui possiamo addestrare molti modelli interessanti (sebbene non reti con nodi che alimentano più altri nodi. Questo non è troppo difficile da aggiungere: vedere la nota nella discussione della catena regola), anche se è un po’ prolisso. Se desideri migliorarlo, prova alcuni dei seguenti:
Esercizio 4: Se ci pensi, i nodi più “complessi” nella nostra rete (ad esempio i livelli lineari) sono in realtà solo “macro” in un certo senso – cioè, se avessimo un albero della rete neurale che assomigliasse, diciamo, come segue:
quello che stai realmente facendo è questo:
In altre parole, Lineare(inp) è in realtà solo una macro per un albero che contiene |inp| +1 parametri, il primo dei quali sono pesi moltiplicati e l’ultimo è un bias. Ogni volta che vediamo Lineare(inp)possiamo quindi sostituirlo con un albero equivalente composto solo da operazioni elementari.
Per questo esercizio, il tuo compito è quindi implementare il Macro classe. La classe dovrebbe essere un Operazione che ricorsivamente si sostituisce con operazioni elementari
Nota: questo passaggio può essere eseguito in qualsiasi momento, anche se probabilmente è più semplice aggiungere un metodo compile() alla classe Operation che devi chiamare prima dell’addestramento (o aggiungerlo al metodo esistente dall’Esercizio 2). Possiamo ovviamente implementare nodi più complessi anche in altri modi (magari più efficienti), ma questo è comunque un buon esercizio.
Esercizio 5: Anche se in realtà non abbiamo mai bisogno che i nodi interni producano qualcosa di diverso da un numero come output, a volte è carino che la radice del nostro albero (cioè il nostro livello di output) produca qualcos’altro (ad esempio un elenco di numeri in il caso di un Softmax). Implementare il Produzione class e consentirgli di produrre un Listof(float) anziché solo un float. Come bonus, prova a implementare l’output SoftMax.
Nota: ci sono alcuni modi per farlo. È possibile fare in modo che l’output estenda l’operazione e quindi modificare il metodo op() della classe NeuralNetNode per restituire un List(float) anziché solo un float. In alternativa, potresti creare una nuova superclasse Nodo che estendono sia Output che Operazione. Questo è probabilmente più facile.
Si noti inoltre che sebbene questi output possano produrre elenchi, otterranno comunque solo una derivata dalla funzione di perdita: la funzione di perdita prenderà semplicemente una lista di float invece di un float (ad esempio la perdita di entropia incrociata categoriale).
Esercizio 6: Ricordi che in precedenza nell’articolo abbiamo detto che le reti neurali sono solo funzioni matematiche costituite da operazioni elementari? Aggiungi il funcificare() metodo alla classe NeuralNetNode che lo trasforma in una tale funzione scritta in notazione leggibile dall’uomo (aggiungi parentesi a tuo piacimento). Ad esempio, la rete neurale Aggiungi((Parametro(0.1), Parametro(0.2))) dovrebbe collassare a “0,1 + 0,2” (o “(0,1 + 0,2)”).
Nota: affinché funzioni, gli input dovrebbero essere nominati. Se hai eseguito l’esercizio 2, dai un nome ai tuoi input nella funzione compile(). In caso contrario, dovrai trovare un modo per nominare i tuoi input: scrivere una funzione compile() è probabilmente ancora il modo più semplice.
Esercizio 7: Modifica il nostro framework per consentire ai nodi di avere più genitori. Lo risolverò nella parte 2.
È tutto per ora! Se desideri controllare il codice, puoi guardare questo Google Colab che ha tutto (tranne le soluzioni per tutti gli esercizi tranne il n. 6, anche se potrei aggiungerle nella parte 2).
Contattami a mchak@calpoly.edu per qualsiasi richiesta.
Se non diversamente specificato, tutte le immagini sono dell’autore.
Fonte: towardsdatascience.com
