Introduzione a TorchRL per il Deep Reinforcement Learning

Configurare TorchRL

In questa sezione ti mostro come installare e iniziare a usare TorchRL. 

Prerequisiti

Prima di installare e usare TorchRL, ti servono alcuni pacchetti software di dipendenza:

• PyTorch: TorchRL si basa su PyTorch, quindi è un prerequisito. 
• Gymnasium: Ti serve il pacchetto Gymnasium per importare ambienti RL. A gennaio 2025, l’ultima versione di Gymnasium non è compatibile con TorchRL, come spiegato in questa pagina Git Discussions. Quindi installa la versione precedente 0.29.1. 
• PyGame: È un pacchetto Python per videogiochi.  TorchRL lo usa per simulare ambienti RL in stile videogioco, come CartPole. 
• TensorDict: Usare una struttura dati tipo dizionario per memorizzare input e output delle reti neurali rende comodo lavorare con i tensori in TorchRL. TensorDict fornisce un contenitore che memorizza tensori come coppie chiave-valore. 

Installa i pacchetti prerequisiti:

!pip install torch tensordict  gymnasium==0.29.1 pygame

Installare TorchRL

Installa TorchRL usando pip. Consiglio vivamente di installare questi pacchetti in un ambiente Conda se lavori su un computer personale o un server. 

!pip install torchrl

Verificare l’installazione

Dopo l’installazione, verifica che TorchRL sia stato installato correttamente. Prova a importare torchrl in una shell Python (o notebook). Usa il metodo check_env_specs() per controllare che un ambiente standard (come CartPole) corrisponda alle specifiche di torchrl:

import torchrl 
from torchrl.envs import GymEnv 
from torchrl.envs.utils import check_env_specs

check_env_specs(GymEnv("CartPole-v1"))

L’output dovrebbe indicare che l’ambiente è stato creato con successo e funziona con TorchRL. 

[torchrl][INFO] check_env_specs succeeded!

Componenti chiave di TorchRL

Prima di costruire il tuo primo agente RL, vediamo i blocchi fondamentali di TorchRL.

Ambienti

TorchRL fornisce una API uniforme per interfacciarsi con diversi ambienti. Lo fa incapsulando le funzionalità specifiche di ciascun ambiente in una serie standard di classi e funzioni wrapper. Passi i parametri appropriati al wrapper e questo mappa internamente il tuo comando alla chiamata corrispondente dell’ambiente specifico. In particolare:

• TorchRL converte stati (osservazioni), azioni e ricompense dell’ambiente in oggetti tensore PyTorch, che possono essere usati direttamente dai moduli che implementano gli algoritmi RL.
• Rende possibile applicare passaggi di pre e post-processing, ad esempio per normalizzare e scalare i tensori di input o presentare i tensori di output in un formato specifico.

Per esempio, per creare un ambiente da Gymnasium, usa il modulo GymEnv:

env = GymEnv("CartPole-v1")

Transform

Estendere l’ambiente di base con funzionalità aggiuntive e trasformazioni che altrimenti dovresti implementare tu è un’operazione comune. Per esempio, puoi aggiungere un modulo contatore di passi invece di codificarlo da zero. Il contatore tiene traccia del numero di passi in ogni episodio. Il modulo transformedEnv aiuta in questo:

from torchrl.envs import GymEnv, StepCounter, TransformedEnv
env = TransformedEnv(GymEnv("CartPole-v1"), StepCounter())

Allo stesso modo, normalizzare i tensori prima di operarci sopra è un passaggio di pre-processing comune. Una trasformazione di normalizzazione, usando il modulo ObservationNorm, normalizza i tensori. La documentazione di TransformedEnv discute vari tipi di trasformazioni. 

Puoi anche combinare più trasformazioni usando il parametro compose():

base_env = GymEnv('CartPole-v1', device=device) 

env = TransformedEnv( 
    base_env, 
    Compose(
        ObservationNorm(in_keys=["observation"]), 
        StepCounter()
    )
)

Agenti e policy

In RL, l’agente decide le azioni usando la policy e in base allo stato osservato dell’ambiente. L’obiettivo dell’agente è massimizzare le ricompense cumulative dall’ambiente. Riceve ricompense quando sceglie l’azione giusta (come fermarsi a un semaforo rosso). 

Per esempio, in un’auto a guida autonoma, l’agente guida l’auto. Le sue decisioni (come sterzare in una certa direzione) si basano sulle osservazioni dello stato dell’ambiente (traffico, posizione dell’auto, ecc.) e sulla policy (evitare pedoni e ostacoli, fermarsi al rosso, ecc.). 

La policy più semplice è scegliere un’azione a caso. L’actor sceglie casualmente una delle azioni dallo spazio delle possibili azioni. Una policy random può essere usata per generare un dataset iniziale di interazioni prima di iniziare ad addestrare il modello. 

Usa il modulo RandomPolicy per creare una policy casuale. Questa policy accetta un parametro action_spec che specifica lo spazio delle azioni. Nell’esempio sotto, lo spazio delle azioni consiste in numeri (continui) tra -1 e +1. Quindi, la policy casuale sceglie un’azione rappresentata da un numero casuale tra -1 e +1. 

import torchrl 
import torch
from tensordict import TensorDict
from torchrl.data.tensor_specs import Bounded

action_spec = Bounded(-torch.ones(1), torch.ones(1))
actor = torchrl.envs.utils.RandomPolicy(action_spec=action_spec) 
td = actor(TensorDict({}, batch_size=[])) 
print(td.get("action"))

L’output dovrebbe essere un tensore, come mostrato sotto:

tensor([0.9258])

Controlla che la policy sia effettivamente casuale eseguendo di nuovo l’actor:

td = actor(TensorDict({}, batch_size=[])) 
print(td.get("action"))

Dovrebbe stampare un tensore casuale diverso.

Se ti serve un ripasso dei concetti RL, dai un’occhiata al percorso Reinforcement Learning in Python su DataCamp!

Costruire il tuo primo agente RL con TorchRL

In questa sezione ti mostro come implementare un semplice agente di Reinforcement Learning usando TorchRL. 

Prima di iniziare, importa in Python i pacchetti software necessari:

• time per misurare il tempo di addestramento.
• GymEnv, StepCounter e TransformedEnv per lavorare con ambienti Gymnasium.
• MLP per creare una semplice rete neurale MLP (multi-layer perceptron). 
• EGreedyModule per bilanciare esplorazione ed exploitation.
• QValueModule e DQNLoss per implementare l’algoritmo Deep Q-Learning.
• SoftUpdate per aggiornare la rete neurale.
• SyncDataCollector per raccogliere i dati dalle interazioni dell’agente. 
• ReplayBuffer per memorizzare i dati delle interazioni.
• Adam per la backpropagation.
• matplotlib per visualizzare l’andamento del training. 
• torchrl_logger per fare logging della sessione di training.

import time
import matplotlib.pyplot as plt
from torchrl.envs import GymEnv, StepCounter, TransformedEnv
from tensordict.nn import TensorDictModule as TensorDict, TensorDictSequential as Seq
from torchrl.modules import EGreedyModule, MLP, QValueModule
from torchrl.objectives import DQNLoss, SoftUpdate
from torchrl.collectors import SyncDataCollector
from torchrl.data import LazyTensorStorage, ReplayBuffer
from torch.optim import Adam
from torchrl._utils import logger as torchrl_logger

Passo 1: definire l’ambiente

In questo esempio risolviamo l’ambiente CartPole. Importa questo ambiente da Gymnasium insieme a un contatore di passi per tenere traccia del numero di step di training:

env = TransformedEnv(GymEnv("CartPole-v1"), StepCounter())

Imposta il seed per Python e per l’ambiente RL per replicare risultati simili tra sessioni di training. 

torch.manual_seed(0)
env = TransformedEnv(GymEnv("CartPole-v1"), StepCounter())
env.set_seed(0)

Definisci parametri e iperparametri per il training: 

• INIT_RAND_STEPS: Numero di passi in cui l’agente agisce casualmente prima di usare la policy. Questi passi iniziali servono a raccogliere i dati per avviare l’addestramento della policy.
• FRAMES_PER_BATCH: Numero di datapoint (uno per interazione o timestep) in un batch di training. 
• OPTIM_STEPS: Numero di passi per accumulare le loss prima di eseguire un passaggio di backward. 
• EPS_0: Valore iniziale di epsilon, il coefficiente di esplorazione. 
• BUFFER_LEN: Dimensione del replay buffer.
• ALPHA: Learning rate.
• TARGET_UPDATE_EPS: Fattore di decadimento per aggiornare la rete target usando il modulo di soft update.
• REPLAY_BUFFER_SAMPLE: Dimensione del campione casuale da prelevare dal replay buffer in ogni iterazione di training. Il replay buffer memorizza i risultati delle interazioni dell’agente con l’ambiente a ogni timestep. 
• LOG_EVERY: Numero di passi dopo i quali stampare l’avanzamento del training.
• MLP_SIZE: Dimensione della rete neurale.

INIT_RAND_STEPS = 5000 
FRAMES_PER_BATCH = 100
OPTIM_STEPS = 10
EPS_0 = 0.5
BUFFER_LEN = 100_000
ALPHA = 0.05
TARGET_UPDATE_EPS = 0.95
REPLAY_BUFFER_SAMPLE = 128
LOG_EVERY = 1000
MLP_SIZE = 64

Passo 2: creare la policy

Definisci una semplice rete neurale per implementare la policy: 

• Definisci l’MLP (rete neurale). Data un’osservazione, l’MLP produce il valore dell’azione, svolgendo di fatto il ruolo della funzione Q. 
• Definisci un dizionario di tensori usando il modulo TensorDict sull’MLP sopra. Questo dizionario mappa le osservazioni di stato dell’ambiente (chiavi) ai valori di probabilità delle azioni (valori) corrispondenti a quello stato. 
• Usa il QValueModule per implementare la funzione Q. Dato un tensore con i valori delle azioni, restituisce l’azione corrispondente al valore più alto. Implementa la strategia greedy (exploitation). 
• Combina (usando til modulo TensorDictSequential) il dizionario di tensori dell’MLP e il QValueModule per definire la policy. 

value_mlp = MLP(out_features=env.action_spec.shape[-1], num_cells=[MLP_SIZE, MLP_SIZE])
value_net = TensorDict(value_mlp, in_keys=["observation"], out_keys=["action_value"])
policy = Seq(value_net, QValueModule(spec=env.action_spec))

• Definisci la funzione di esplorazione usando EGreedyModule: accetta in input lo spazio delle azioni dell’ambiente, la lunghezza del replay buffer e il coefficiente di esplorazione epsilon. Collega in sequenza (con il modulo TensorDictSequential) questa funzione di esplorazione con la policy definita sopra per ottenere la policy finale:

exploration_module = EGreedyModule(
    env.action_spec, annealing_num_steps=BUFFER_LEN, eps_init=EPS_0
)
policy_explore = Seq(policy, exploration_module)

Passo 3: addestrare l’agente

Il primo passo nell’addestramento è raccogliere i dati dalle interazioni dell’agente con l’ambiente. Usa il SyncDataCollector per costruire un collector che esegue la policy e raccoglie i risultati delle interazioni: 

collector = SyncDataCollector(
    env,
    policy_explore,
    frames_per_batch=FRAMES_PER_BATCH,
    total_frames=-1,
    init_random_frames=INIT_RAND_STEPS,
)

Crea un replay buffer per memorizzare i risultati delle interazioni:

rb = ReplayBuffer(storage=LazyTensorStorage(BUFFER_LEN))

Devi anche dichiarare i moduli specifici per il training, come la funzione di loss (per usare il calcolo della loss basato su DQN), l’ottimizzatore (il classico algoritmo Adam) e le funzioni di aggiornamento (per aggiornare la rete neurale). Sono tutti basati su moduli TorchRL predefiniti:

loss = DQNLoss(value_network=policy, action_space=env.action_spec, delay_value=True)
optim = Adam(loss.parameters(), lr=ALPHA)
updater = SoftUpdate(loss, eps=TARGET_UPDATE_EPS)

Inizializza i contatori per tenere traccia del numero totale di passi ed episodi, dei passi riusciti per episodio e del tempo di esecuzione:

total_count = 0
total_episodes = 0
t0 = time.time()
success_steps = []

La funzione di training consiste in 2 cicli for: 

• Il ciclo più esterno esegue la policy e aggiunge i risultati delle interazioni dell’agente al replay buffer.
• Il ciclo interno suddivide i campioni di training in batch. Elabora ogni batch come segue: 
• Estrai un campione casuale dal replay buffer.
• Calcola le loss usando la funzione di loss.
• Esegui ottimizzatore e updater.
• Aggiorna i contatori.

Ogni passo prevede la chiamata a moduli TorchRL predefiniti senza scrivere codice da zero. Il codice seguente mostra come implementare questi passaggi:

for i, data in enumerate(collector):
    rb.extend(data)
    max_length = rb[:]["next", "step_count"].max()
    if len(rb) > INIT_RAND_STEPS:
        for _ in range(OPTIM_STEPS):
            sample = rb.sample(REPLAY_BUFFER_SAMPLE)
            loss_vals = loss(sample)
            loss_vals["loss"].backward()
            optim.step()

            optim.zero_grad()
            # Update exploration factor
            exploration_module.step(data.numel())

            # Update target params
            updater.step()
            total_count += data.numel()
            total_episodes += data["next", "done"].sum()
    success_steps.append(max_length)

Passo 4: valutare l’agente

Il ciclo nella sezione precedente addestra continuamente la policy. Dobbiamo stabilire i criteri per valutare le prestazioni e decidere quando considerare l’addestramento riuscito. Vogliamo anche mostrare l’avanzamento del training. 

Stampiamo l’avanzamento del training a intervalli periodici usando il logger di TorchRL:

if total_count > 0 and total_count % LOG_EVERY == 0:
    torchrl_logger.info(f"Successful steps in the last episode: {max_length}, rb length {len(rb)}, Number of episodes: {total_episodes}")

Usiamo il numero massimo di passi raggiunto dall’agente nell’ultimo episodio per determinare se il training è riuscito. L’ambiente CartPole-v1 limita il numero massimo di passi e la ricompensa totale per episodio a 500. Convenzionalmente, si considera una policy riuscita se supera 475 passi:

if max_length > 475:
    print("TRAINING COMPLETE")
    break

I due snippet sopra vanno aggiunti alla fine del ciclo di training (mostrato prima). Il seguente snippet mostra il ciclo di training con il codice per valutare l’agente: 

for i, data in enumerate(collector):
    # Write data in replay buffer
    rb.extend(data)
    max_length = rb[:]["next", "step_count"].max()
    if len(rb) > INIT_RAND_STEPS:
        for _ in range(OPTIM_STEPS):
            sample = rb.sample(REPLAY_BUFFER_SAMPLE)
            loss_vals = loss(sample)
            loss_vals["loss"].backward()
            optim.step()

            optim.zero_grad()
            # Update exploration factor
            exploration_module.step(data.numel())

            # Update target params
            updater.step()
            total_count += data.numel()
            total_episodes += data["next", "done"].sum()
    success_steps.append(max_length)

    if total_count > 0 and total_count % LOG_EVERY == 0:
        torchrl_logger.info(f"Successful steps in the last episode: {max_length}, rb length {len(rb)}, Number of episodes: {total_episodes}")

    if max_length > 475:
        print("TRAINING COMPLETE")
        break

Infine, dopo training e valutazione, stampa il tempo totale di training e traccia l’andamento:

t1 = time.time()

torchrl_logger.info(
    f"solved after {total_count} steps, {total_episodes} episodes and in {t1-t0}s."
)

def plot_steps():
    plt.plot(success_steps)
    plt.title('Successful steps over training episodes')
    plt.xlabel('Training episodes')
    plt.ylabel('Steps')
    plt.show()

plot_steps()

Puoi trovare ed eseguire il codice completo per implementare DQN con TorchRL in questo DataLab workbook. 

In questa sezione abbiamo costruito e addestrato un agente RL usando il semplice algoritmo DQN. Nella prossima sezione vediamo come usare i moduli predefiniti di TorchRL per implementare un algoritmo più complesso come PPO. 

Esplorare gli algoritmi predefiniti in TorchRL

Come accennato in precedenza, TorchRL include alcuni algoritmi predefiniti. Vediamo quali sono e come funzionano.

Algoritmi supportati

TorchRL include moduli predefiniti per molti algoritmi di Deep Reinforcement Learning comuni, tra cui: 

• Deep Q-Networks (DQN)
• Deep Deterministic Policy Gradient (DDPG)
• Soft Actor-Critic (SAC)
• Randomized Ensembled Double Q-Learning (REDQ)
• CrossQ
• Implicit Q-Learning (IQL)
• Conservative/Continuous Q-Learning (CQL)
• Generative Adversarial Imitation Learning (GAIL)
• Decision Transformer (DT)
• Twin-Delayed DDPG (TD3) 
• Actor-Advantage Critic (A2C)
• Proximal Policy Optimization (PPO)
• REINFORCE
• e altri 

Questo rende efficiente sperimentare diversi tipi di algoritmi e studiarne le prestazioni per risolvere un dato problema. 

Esempio: PPO in TorchRL

Proximal Policy Optimization usa una funzione obiettivo surrogate con clipping per ottenere un training più stabile che bilancia esplorazione ed exploitation. 

In questa sezione mostro come usare il modulo PPO predefinito in TorchRL. Come prima, importa i moduli necessari. Oltre ai moduli (come ReplayBuffer, SyncDataCollector, ecc.) usati in precedenza, ti servono alcuni pacchetti aggiuntivi per PPO:

• ProbabilisticActor, per scegliere un’azione in modo stocastico. Scegliere azioni stocasticamente (invece di seguire sempre la policy e massimizzare la ricompensa) favorisce l’esplorazione dell’ambiente e la scoperta di percorsi più ottimali durante il training.
• OneHotCategorical, per generare un one-hot encoding del tensore che denota le log-probabilità delle azioni. 
• ValueOperator, per costruire un modulo TorchRL basato sulla rete neurale che implementa la funzione di valore.
• GAE, per implementare la stima generalizzata del vantaggio. PPO usa una funzione di vantaggio come proxy della funzione di valore. Il modulo valore (sopra) è un input per il GAE. 
• ClipPPOLoss, per costruire la funzione obiettivo con clipping per implementare PPO. 

import torch
from torch import nn

from torchrl.envs import Compose, ObservationNorm, DoubleToFloat, StepCounter, TransformedEnv
from torchrl.envs.libs.gym import GymEnv
from torchrl.envs.utils import check_env_specs
from torchrl.modules import ProbabilisticActor, OneHotCategorical, ValueOperator
from torchrl.collectors import SyncDataCollector
from torchrl.data.replay_buffers import ReplayBuffer
from torchrl.data.replay_buffers.storages import LazyTensorStorage
from torchrl.data.replay_buffers.samplers import SamplerWithoutReplacement
from torchrl.objectives.value import GAE
from torchrl.objectives import ClipPPOLoss
from tensordict.nn import TensorDictModule

Dichiara parametri e iperparametri per il training:

• FRAMES_PER_BATCH: Numero di frame (timestep) in ogni batch di dati risultante dalle interazioni dell’agente con l’ambiente. 
• SUB_BATCH_SIZE: Numero di timestep in ogni sotto-batch. Nel training PPO, ogni batch è diviso in sotto-batch. 
• TOTAL_FRAMES: Numero totale di timestep per cui eseguire il training. 
• GAMMA: Fattore di sconto per scontare il valore della ricompensa dai timestep futuri.
• CLIP_EPSILON: Definisce la trust region entro cui la policy può cambiare in ogni iterazione di training successiva. È implementato tramite clipping della policy aggiornata in modo che il rapporto tra probabilità nuova e vecchia resti entro un limite. 
• ALPHA: Learning rate.
• ENTROPY_EPS: Iperparametro che controlla il rapporto tra esplorazione ed exploitation. 
• OPTIM_STEPS: Numero di volte in cui eseguire l’ottimizzatore su ogni sotto-batch di dati. 
• LOG_EVERY: Numero di passi dopo i quali valutare la policy e stampare le ricompense. 

FRAMES_PER_BATCH = 1024
TOTAL_FRAMES = 1048576
GAMMA = 0.99
LAMBDA = 0.95
CLIP_EPSILON = 0.2
ALPHA = 1e-4
ENTROPY_EPS = 5e-4 
SUB_BATCH_SIZE = 64
OPTIM_STEPS = 8
LOG_EVERY = 16

Importa l’ambiente CartPole di base da Gymnasium:

device="cpu"
base_env = GymEnv('CartPole-v1', device=device) 

Dichiara l’ambiente TorchRL importando l’ambiente di base e aggiungendo moduli per normalizzare i tensori di osservazione e contare i passi:

env = TransformedEnv( 
    base_env, 
    Compose(
        ObservationNorm(in_keys=["observation"]), 
        DoubleToFloat(), 
        StepCounter()
    )
)

Inizializza l’ambiente e imposta i generatori di numeri casuali:

env.transform[0].init_stats(1024) 
torch.manual_seed(0)
env.set_seed(0)
check_env_specs(env) 

Dichiara l’actor come una rete neurale con 1 livello nascosto con 16 pesi. Riceve in input l’osservazione dall’ambiente e produce la probabilità di ciascuna azione nello spazio delle azioni. 

actor_net = nn.Sequential(
    nn.Linear(env.observation_spec["observation"].shape[-1], 32, device=device),
    nn.ReLU(),
    nn.Linear(32, 32, device=device),
    nn.ReLU(),
    nn.Linear(32, env.action_spec.shape[-1], device=device),
    nn.ReLU()
)

Crea un dizionario di tensori (basato sulla rete neurale sopra) con chiavi e valori che mappano stati a probabilità delle azioni. 

actor_module = TensorDictModule(actor_net, in_keys=["observation"], out_keys=["logits"])

Per addestrare PPO (e molti altri algoritmi RL), non scegli sempre l’azione che massimizza la ricompensa nel passo successivo. Introdurre un certo grado di casualità nella scelta dell’azione consente all’agente di esplorare l’ambiente e scoprire percorsi migliori che possono portare a ritorni a lungo termine più elevati. Crea un actor probabilistico per farlo:

actor = ProbabilisticActor(
    module = actor_module,
    spec = env.action_spec,
    in_keys = ["logits"],
    distribution_class = OneHotCategorical, 
    return_log_prob = True
)

Crea la rete per implementare la funzione di valore. Questa rete ha 16 pesi. Riceve in input lo stato (osservazione) dell’ambiente e produce il valore atteso di quello stato. 

value_net = nn.Sequential(
    nn.Linear(env.observation_spec["observation"].shape[-1], 16, device=device),
    nn.ReLU(),
    nn.Linear(16, 1, device=device),
    nn.ReLU()
)

Crea un modulo TorchRL che incapsula la funzione di valore (implementata sopra) usando ValueOperator(). Questo modulo si interfaccia con gli altri componenti TorchRL. 

value_module = ValueOperator(
    module = value_net,
    in_keys = ["observation"]
)

Crea il replay buffer per memorizzare i risultati delle interazioni dell’agente con l’ambiente: 

replay_buffer = ReplayBuffer(
    storage = LazyTensorStorage(max_size=FRAMES_PER_BATCH),
    sampler = SamplerWithoutReplacement()
)

Crea un data collector per eseguire la policy sull’ambiente e raccogliere i risultati delle interazioni dell’agente a ogni timestep (frame):

collector = SyncDataCollector(
    env,
    actor,
    frames_per_batch = FRAMES_PER_BATCH,
    total_frames = TOTAL_FRAMES,
    split_trajs = True,
    reset_at_each_iter = True,
    device=device
)

Usa il modulo GAE per implementare la funzione di vantaggio per PPO. La funzione di vantaggio si basa sulla funzione di valore. 

advantage_module = GAE(
    gamma = GAMMA, 
    lmbda = GAMMA, 
    value_network = value_module,
    average_gae = True
) 

Usa il modulo integrato ClipPPOLoss per implementare la funzione di loss secondo l’algoritmo PPO:

loss_module = ClipPPOLoss(
    actor_network = actor,
    critic_network = value_module,
    clip_epsilon = CLIP_EPSILON,
    entropy_bonus = bool(ENTROPY_EPS),
    entropy_coef = ENTROPY_EPS
)

Dichiara l’ottimizzatore Adam: 

optim = torch.optim.Adam(loss_module.parameters(), lr=ALPHA)

Crea uno scheduler per ridurre gradualmente il learning rate man mano che il training avanza: 

scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optim, TOTAL_FRAMES // FRAMES_PER_BATCH)

Costruisci il ciclo di training usando i componenti creati sopra. La funzione di training consiste in tre cicli for:

• Il ciclo più esterno scorre i dati raccolti dalle interazioni dell’agente con l’ambiente. Il parametro TOTAL_FRAMES decide per quanti timestep eseguire l’agente. 
• Il ciclo intermedio fa quanto segue: 
• Esegue il ciclo più interno per un numero fisso di iterazioni (tipicamente tra 5 e 10), come definito dall’iperparametro OPTIM_STEPS. 
• Aggiorna il learning rate dopo ogni iterazione di training. 
• Calcola anche il valore della funzione di vantaggio per ogni timestep. 
• Valuta periodicamente la policy e stampa le ricompense. 
• Il ciclo più interno esegue il training loop:
• Campiona un batch di dati di training dal replay buffer.
• Calcola la loss usando il modulo di loss. 
• Esegui la backpropagation sulla loss.
• Usa l’ottimizzatore per fare il gradient descent. 

Il codice seguente implementa il training loop:

for i, tensordict_data in enumerate(collector): 
    for _ in range(OPTIM_STEPS): 
        advantage_module(tensordict_data)
        replay_buffer.extend(tensordict_data.reshape(-1).cpu())
        for _ in range(FRAMES_PER_BATCH // SUB_BATCH_SIZE): 
            data = replay_buffer.sample(SUB_BATCH_SIZE)
            loss = loss_module(data.to(device))
            loss_value = loss["loss_objective"] + loss["loss_critic"] + loss["loss_entropy"]
            loss_value.backward()
            optim.step()
            optim.zero_grad()
    scheduler.step()

    if i % LOG_EVERY == 0:
        with torch.no_grad():
            rollout = env.rollout(FRAMES_PER_BATCH, actor)
            reward_eval = rollout["next","reward"].sum()
            #print(rollout["next","reward"].sum())
            print(reward_eval)
            rewards.append(reward_eval)
            del rollout

Questo DataLab workbook contiene il codice (come sopra) per addestrare un agente RL usando l’algoritmo PPO di TorchRL. Usalo come punto di partenza per ottimizzare i parametri e migliorare le prestazioni dell’agente. 

Personalizzare gli algoritmi

TorchRL ha un design modulare che consente di personalizzare le soluzioni RL. Il framework gestisce l’interfacciamento tra i diversi componenti. Per testarne e confrontarne le prestazioni, puoi collegare e combinare ambienti, policy, replay buffer e algoritmi RL. Grazie a questa modularità:

• Puoi sostituire componenti diversi senza dover riscrivere grandi parti del programma. 
• I singoli componenti possono anche essere modificati indipendentemente per creare soluzioni personalizzate. 

• Puoi sperimentare con moduli diversi e costruire una soluzione ottimizzata. 

Alcuni esempi di personalizzazione in TorchRL:

• I training loop possono essere personalizzati con scheduler del learning rate, logger e metriche personalizzate. 
• Si possono usare lunghezze diverse dei replay buffer per regolare la durata dell’addestramento.
• L’ambiente può essere personalizzato per includere funzioni di scaling e normalizzazione, contatori di passi, ecc. 

Visualizzare e fare debug del training RL

Le sezioni precedenti hanno mostrato come usare i moduli TorchRL per addestrare agenti RL. In questa sezione, discutiamo modi per monitorare e visualizzare l’andamento del training. 

Monitorare l’avanzamento del training

È utile registrare varie metriche durante il training per monitorarne l’andamento. Pacchetti come TensorBoard rendono semplice registrare direttamente i risultati durante il training. Per esempio, SummaryWriter di TensorBoard ti permette di chiamarlo all’interno del training loop per loggare le metriche. Il seguente pseudo-codice mostra come:

from torch.utils.tensorboard import SummaryWriter
writer = SummaryWriter(log_dir="training_logs")

for _ in training_data (): # training loop
    loss = … 
    loss.backprop() 
    writer.add_scalar(“Loss - “, loss.item(), step)
   … 

writer.close() 

Dopo aver terminato training e logging, puoi tracciare e visualizzare i risultati. Per esempio, usa il comando tensorboard per visualizzare l’andamento del training. 

tensorboard --logdir=”training_logs”

Debug con TorchRL

Il debug è essenziale per verificare l’interazione dell’agente con l’ambiente. Come primo passo, vuoi controllare lo spazio delle azioni e quello delle osservazioni. La documentazione dell’ambiente dovrebbe includerne le specifiche. Per esempio, la documentazione di CartPole mostra che: 

• Lo spazio delle azioni include due valori discreti: 0 (spingere il carrello a sinistra) e 1 (spingere a destra) 
• Lo spazio delle osservazioni ha 4 valori: 
• Posizione del carrello, con valore tra 
• -4.8 e +4.8. 
• Velocità del carrello, con qualunque valore reale
• Angolo dell’asta, con valore tra -24 ° e  +24 °
• Velocità angolare dell’asta, con qualunque valore reale

Ispeziona spazi di osservazione e azione per verificare che corrispondano ai valori attesi. Per esempio:

print("Observation space: ", env.observation_spec)
print("Action space: ", env.action_spec)

Inoltre, puoi estrarre campioni casuali dagli spazi di osservazione e azione e controllarne i valori: 

print("Sample Observation:", base_env.observation_spec.sample().get("observation"))
print("Sample Action:", base_env.action_spec.sample())

L’output sarà simile al seguente:

Sample Observation: tensor([-4.5960e+00,  3.4028e+38,  2.2261e-02,  3.4028e+38])
Sample Action: tensor([1, 0])

Nota che i 4 valori di stato rientrano negli intervalli specificati in precedenza. Ripeti i comandi qui sopra qualche volta e nota che i diversi valori (campionati casualmente) rientrano negli intervalli indicati. 

I due comandi sopra si basano sull’ambiente di base, importato direttamente da Gymnasium. In pratica, usiamo l’ambiente trasformato, che applica trasformazioni, come la normalizzazione, ai valori tensoriali originali. Quindi, quando usi l’ambiente TorchRL trasformato per campionare osservazioni, i loro valori potrebbero non rientrare più nello stesso intervallo dell’ambiente Gymnasium originale. 

Per esempio, estrai un campione dall’ambiente trasformato:  

print("Sample Observation:", env.observation_spec.sample().get("observation"))

L’output sarà simile: 

Sample Observation: tensor([-57.5869,      nan,   4.5276,      nan])

Nota che i valori di posizione del carrello e angolo dell’asta sono fuori dagli intervalli. Questo perché i valori dei tensori sono stati normalizzati. Inoltre, la velocità del carrello e la velocità angolare dell’asta hanno valore nan perché l’agente non ha ancora iniziato a interagire con l’ambiente. 

Visualizzare le prestazioni dell’agente

Oltre a tracciare l’andamento del training, può essere utile renderizzare l’ambiente e osservare visivamente le interazioni dell’agente. Questo può fornire insight sulle prestazioni dell’agente. 

Il modo più pragmatico per visualizzare l’ambiente è renderizzare un video. Ti servono alcuni pacchetti aggiuntivi: 

• torchvision, per lavorare con file multimediali
• av, per incapsulare il pacchetto ffmpeg in modo che possa interfacciarsi con PyTorch. 

pip install torchvision
pip install av==12.0.0

Dopo aver concluso il training, segui questi passaggi preparatori per renderizzare il video:

• Dichiara il percorso (relativo) in cui salvare il video.
• Inizializza il logger per salvare i risultati delle interazioni dell’agente in formato CSV.
• Inizializza il video recorder per usare i log CSV e generare un video.
• Trasforma l’ambiente TorchRL per includere il video recorder.

…
# training loop
# for _ in …
…
…

from torchrl.record import CSVLogger, VideoRecorder 

path = "./training_loop"
logger = CSVLogger(exp_name="dqn", log_dir=path, video_format="mp4")
video_recorder = VideoRecorder(logger, tag="video")
record_env = TransformedEnv(
    GymEnv("CartPole-v1", from_pixels=True, pixels_only=False), video_recorder
)

Esegui la policy (addestrata) sull’ambiente e scarica i rendering nel file video:

record_env.rollout(max_steps=1000, policy=policy)
video_recorder.dump()

Dopo l’esecuzione, troverai il video nel percorso (./training_loop nello snippet sopra) impostato in precedenza. Nota che il DataLab workbook per implementare DQN con TorchRL non include il codice per creare il video perché esportare file video da un notebook online è complicato. Esegui il programma (e aggiungi i passaggi sopra) su un computer o server locale per registrare i video.

Best practice per usare TorchRL

Infine, vediamo alcune best practice. Ecco le mie raccomandazioni.

Inizia da ambienti semplici

Nonostante la facilità di sviluppo offerta da TorchRL, addestrare agenti RL a performare bene in ambienti complessi resta impegnativo. Dunque, prima di affrontare problemi difficili, è essenziale risolvere ambienti semplici come CartPole. 

La modularità di TorchRL permette di sperimentare con diversi algoritmi e parametri. Prima di usare in pratica algoritmi alternativi e opzioni di personalizzazione, è necessario capirne il funzionamento. Questo è più facile in un ambiente semplice, dove gli effetti dei singoli cambiamenti sono facilmente osservabili visivamente. 

Sperimenta con gli iperparametri

Le prestazioni di training dei modelli RL sono sensibili a vari iperparametri, come learning rate, tasso di esplorazione, tasso di sconto e altri iperparametri specifici dell’algoritmo, come il clipping ratio per PPO. Per esempio, un learning rate troppo alto rende il training instabile, mentre uno troppo basso richiede troppo tempo per convergere. 

Allo stesso modo, un tasso di esplorazione troppo alto impedisce la convergenza del training, mentre uno troppo basso può impedire all’agente di scoprire il percorso ottimale. 

Non esiste una formula per determinare i valori giusti. Esistono linee guida e valori raccomandati per ambienti standard, ma potrebbero non applicarsi a tutti i problemi. È quindi necessario sperimentare con vari metodi, come grid search o random search, per determinare i migliori valori degli iperparametri. 

È anche possibile usare librerie automatiche come Weights & Biases Sweeps o Optuna per testare le prestazioni del training su varie combinazioni di iperparametri. 

Sfrutta gli algoritmi predefiniti

Come visto negli esempi precedenti, usare i moduli predefiniti di TorchRL fa risparmiare molto sforzo di sviluppo. L’alternativa è costruire tutte le funzionalità da zero, con tempi e costi significativamente maggiori per sviluppare e testare. 

Poiché molti sviluppatori usano i moduli di TorchRL, questi beneficiano anche dell’essere stati testati in vari scenari. Puoi quindi aspettarti che siano più privi di bug rispetto a un modulo costruito ad hoc. Per la maggior parte dei casi d’uso standard, è fortemente consigliato usare moduli predefiniti. 

Conclusione

In questo articolo abbiamo coperto i concetti base di TorchRL, un framework basato su PyTorch per implementare algoritmi RL. Abbiamo anche visto esempi pratici di uso di TorchRL per implementare soluzioni più semplici come il Deep Q-Learning e algoritmi più complessi come PPO. 

Come passo successivo, puoi usare questi programmi come base per sperimentare con altri ambienti e algoritmi. 

Se vuoi approfondire i fondamenti dell’RL e le applicazioni pratiche, dai un’occhiata a Reinforcement Learning with Gymnasium in Python per fare più pratica con ambienti basati su Gymnasium.