use anyhow::{Result, anyhow};
use burn::{
module::{AutodiffModule, list_param_ids},
optim::{GradientsParams, Optimizer},
prelude::ToElement,
tensor::{Tensor, backend::AutodiffBackend},
};
use rand::Rng;
use super::{
config::DQNConfig,
loss::{compute_loss, compute_td_target, compute_td_target_double, gather_action_q},
};
use crate::{
buffer::replay::{ReplayBuffer, sample},
train::optimizer::{BackendOptimizer, BurnOptimizer},
};
#[derive(Debug, Clone, Copy)]
pub struct DQNStepStatsBurn {
pub td_loss: f64,
pub mean_q: f64,
pub epsilon: f64,
pub buffer_len: usize,
}
pub struct DQNTrainerBurn<B, Q, O>
where
B: AutodiffBackend,
Q: AutodiffModule<B> + Clone,
O: Optimizer<Q, B>,
{
config: DQNConfig,
n_actions: i64,
online: Option<Q>,
target: Q,
optimizer: BurnOptimizer<B, Q, O>,
buffer: ReplayBuffer,
device: B::Device,
total_env_steps: usize,
total_train_steps: usize,
total_episodes: usize,
last_epsilon: f64,
}
impl<B, Q, O> DQNTrainerBurn<B, Q, O>
where
B: AutodiffBackend,
Q: AutodiffModule<B> + Clone,
O: Optimizer<Q, B>,
{
pub fn new(
config: DQNConfig,
online: Q,
optimizer: BurnOptimizer<B, Q, O>,
obs_dim: usize,
n_actions: i64,
device: B::Device,
) -> Result<Self> {
config.validate()?;
if config.prioritized_replay {
return Err(anyhow!(
"DQNTrainerBurn does not yet implement prioritized replay (phase 3 \
scope on issue #80). Use the tch DQNTrainer or wait for phase 5."
));
}
let target = online.clone();
let buffer = ReplayBuffer::new(config.buffer_capacity, obs_dim);
let last_epsilon = config.epsilon_start;
Ok(Self {
config,
n_actions,
online: Some(online),
target,
optimizer,
buffer,
device,
total_env_steps: 0,
total_train_steps: 0,
total_episodes: 0,
last_epsilon,
})
}
pub fn n_actions(&self) -> i64 {
self.n_actions
}
pub fn config(&self) -> &DQNConfig {
&self.config
}
pub fn online(&self) -> &Q {
self.online.as_ref().expect("online network is None mid-step")
}
pub fn target(&self) -> &Q {
&self.target
}
pub fn buffer(&self) -> &ReplayBuffer {
&self.buffer
}
pub fn buffer_mut(&mut self) -> &mut ReplayBuffer {
&mut self.buffer
}
pub fn buffer_len(&self) -> usize {
self.buffer.len()
}
pub fn increment_env_step(&mut self) {
self.total_env_steps += 1;
}
pub fn increment_episodes(&mut self, n: usize) {
self.total_episodes += n;
}
pub fn total_env_steps(&self) -> usize {
self.total_env_steps
}
pub fn total_train_steps(&self) -> usize {
self.total_train_steps
}
pub fn total_episodes(&self) -> usize {
self.total_episodes
}
pub fn last_epsilon(&self) -> f64 {
self.last_epsilon
}
pub fn select_action<R: Rng, F>(&mut self, obs: &[f32], rng: &mut R, greedy_fn: F) -> i64
where
F: FnOnce(&Q, &[f32]) -> i64,
{
let eps = self.config.epsilon_at(self.total_env_steps);
self.last_epsilon = eps;
if rng.random::<f64>() < eps {
rng.random_range(0..self.n_actions)
} else {
greedy_fn(self.online(), obs)
}
}
pub fn maybe_sync_target<F>(&mut self, blend_fn: F) -> bool
where
F: FnOnce(&Q, Q, f64) -> Q,
{
match self.config.soft_update_tau {
Some(tau) => {
let online = self.online().clone();
let target = std::mem::replace(&mut self.target, online.clone());
self.target = blend_fn(&online, target, tau);
true
}
None => {
if self.total_env_steps > 0
&& self.total_env_steps.is_multiple_of(self.config.target_update_interval)
{
self.target = self.online().clone();
true
} else {
false
}
}
}
}
pub fn train_step<R: Rng, FOnline, FTarget>(
&mut self,
rng: &mut R,
forward_fn: FOnline,
forward_target_fn: FTarget,
) -> Result<Option<DQNStepStatsBurn>>
where
FOnline: Fn(&Q, Tensor<B, 2>) -> Tensor<B, 2>,
FTarget: Fn(&Q, Tensor<B, 2>) -> Tensor<B, 2>,
{
if !self.buffer.is_ready(self.config.min_buffer_size) {
return Ok(None);
}
let batch = sample(&self.buffer, self.config.batch_size, rng);
let buffer_len = self.buffer.len();
let t = batch.to_burn_tensors::<B>(&self.device);
let online = self
.online
.take()
.ok_or_else(|| anyhow!("online network is None; concurrent train_step calls?"))?;
let q_online_all = forward_fn(&online, t.observations);
let q_taken = gather_action_q(q_online_all.clone(), t.actions);
let next_q_online_all = forward_fn(&online, t.next_observations.clone());
let next_q_target_all = forward_target_fn(&self.target, t.next_observations);
let td_target = compute_td_target_double(
t.rewards,
t.dones,
next_q_online_all,
next_q_target_all,
self.config.gamma,
);
let td_loss = compute_loss(q_taken.clone(), td_target);
let td_loss_val: f64 = td_loss.clone().into_scalar().to_f64();
let mean_q_val: f64 = q_taken.mean().into_scalar().to_f64();
if !td_loss_val.is_finite() {
return Err(anyhow!("Non-finite TD loss: {}", td_loss_val));
}
let grads = td_loss.backward();
let grads = GradientsParams::from_grads(grads, &online);
let lr = self.optimizer.learning_rate();
let online = self.optimizer.inner_mut().step(lr, online, grads);
self.online = Some(online);
self.total_train_steps += 1;
Ok(Some(DQNStepStatsBurn {
td_loss: td_loss_val,
mean_q: mean_q_val,
epsilon: self.last_epsilon,
buffer_len,
}))
}
pub fn train_step_scaled<R: Rng, FOnline, FTarget>(
&mut self,
rng: &mut R,
loss_scale: f64,
forward_fn: FOnline,
forward_target_fn: FTarget,
) -> Result<Option<(DQNStepStatsBurn, bool)>>
where
FOnline: Fn(&Q, Tensor<B, 2>) -> Tensor<B, 2>,
FTarget: Fn(&Q, Tensor<B, 2>) -> Tensor<B, 2>,
{
if !self.buffer.is_ready(self.config.min_buffer_size) {
return Ok(None);
}
let batch = sample(&self.buffer, self.config.batch_size, rng);
let buffer_len = self.buffer.len();
let t = batch.to_burn_tensors::<B>(&self.device);
let online = self
.online
.take()
.ok_or_else(|| anyhow!("online network is None; concurrent train_step calls?"))?;
let q_online_all = forward_fn(&online, t.observations);
let q_taken = gather_action_q(q_online_all.clone(), t.actions);
let next_q_online_all = forward_fn(&online, t.next_observations.clone());
let next_q_target_all = forward_target_fn(&self.target, t.next_observations);
let td_target = compute_td_target_double(
t.rewards,
t.dones,
next_q_online_all,
next_q_target_all,
self.config.gamma,
);
let td_loss = compute_loss(q_taken.clone(), td_target);
let scaled_loss = td_loss.clone().mul_scalar(loss_scale as f32);
let td_loss_val: f64 = td_loss.into_scalar().to_f64();
let scaled_loss_val: f64 = scaled_loss.clone().into_scalar().to_f64();
let mean_q_val: f64 = q_taken.mean().into_scalar().to_f64();
let stats = DQNStepStatsBurn {
td_loss: td_loss_val,
mean_q: mean_q_val,
epsilon: self.last_epsilon,
buffer_len,
};
if !td_loss_val.is_finite() || !scaled_loss_val.is_finite() {
self.online = Some(online);
return Ok(Some((stats, false)));
}
let grads = scaled_loss.backward();
let grads = GradientsParams::from_grads(grads, &online);
let grads = unscale_grads::<B, Q>(grads, &online, loss_scale);
let lr = self.optimizer.learning_rate();
let online = self.optimizer.inner_mut().step(lr, online, grads);
self.online = Some(online);
self.total_train_steps += 1;
Ok(Some((stats, true)))
}
pub fn train_step_vanilla<R: Rng, FOnline, FTarget>(
&mut self,
rng: &mut R,
forward_fn: FOnline,
forward_target_fn: FTarget,
) -> Result<Option<DQNStepStatsBurn>>
where
FOnline: Fn(&Q, Tensor<B, 2>) -> Tensor<B, 2>,
FTarget: Fn(&Q, Tensor<B, 2>) -> Tensor<B, 2>,
{
if !self.buffer.is_ready(self.config.min_buffer_size) {
return Ok(None);
}
let batch = sample(&self.buffer, self.config.batch_size, rng);
let buffer_len = self.buffer.len();
let t = batch.to_burn_tensors::<B>(&self.device);
let online = self
.online
.take()
.ok_or_else(|| anyhow!("online network is None; concurrent train_step calls?"))?;
let q_online_all = forward_fn(&online, t.observations);
let q_taken = gather_action_q(q_online_all.clone(), t.actions);
let next_q_target_all = forward_target_fn(&self.target, t.next_observations);
let td_target = compute_td_target(t.rewards, t.dones, next_q_target_all, self.config.gamma);
let td_loss = compute_loss(q_taken.clone(), td_target);
let td_loss_val: f64 = td_loss.clone().into_scalar().to_f64();
let mean_q_val: f64 = q_taken.mean().into_scalar().to_f64();
if !td_loss_val.is_finite() {
return Err(anyhow!("Non-finite TD loss: {}", td_loss_val));
}
let grads = td_loss.backward();
let grads = GradientsParams::from_grads(grads, &online);
let lr = self.optimizer.learning_rate();
let online = self.optimizer.inner_mut().step(lr, online, grads);
self.online = Some(online);
self.total_train_steps += 1;
Ok(Some(DQNStepStatsBurn {
td_loss: td_loss_val,
mean_q: mean_q_val,
epsilon: self.last_epsilon,
buffer_len,
}))
}
}
fn unscale_grads<B, Q>(mut grads: GradientsParams, module: &Q, loss_scale: f64) -> GradientsParams
where
B: AutodiffBackend,
Q: AutodiffModule<B> + Clone,
{
type Inner<B> = <B as AutodiffBackend>::InnerBackend;
let inv = (1.0 / loss_scale) as f32;
for id in list_param_ids(module) {
if let Some(g) = grads.remove::<Inner<B>, 1>(id) {
grads.register::<Inner<B>, 1>(id, g.mul_scalar(inv));
} else if let Some(g) = grads.remove::<Inner<B>, 2>(id) {
grads.register::<Inner<B>, 2>(id, g.mul_scalar(inv));
} else if let Some(g) = grads.remove::<Inner<B>, 4>(id) {
grads.register::<Inner<B>, 4>(id, g.mul_scalar(inv));
}
}
grads
}
#[cfg(test)]
mod tests {
use burn::{
backend::{Autodiff, NdArray},
optim::AdamConfig,
};
use rand::SeedableRng;
use super::*;
use crate::{policy::mlp::MlpBurnPolicy, train::optimizer::BurnOptimizer};
type B = Autodiff<NdArray<f32>>;
fn small_config() -> DQNConfig {
DQNConfig::new()
.buffer_capacity(128)
.min_buffer_size(8)
.batch_size(8)
.target_update_interval(4)
.epsilon_decay_steps(100)
}
#[test]
fn dqn_trainer_burn_constructs() {
let device = Default::default();
let online = MlpBurnPolicy::<B>::new(4, 2, 16, &device);
let inner_opt = AdamConfig::new().init();
let burn_opt = BurnOptimizer::new(inner_opt, small_config().learning_rate);
let trainer = DQNTrainerBurn::new(small_config(), online, burn_opt, 4, 2, device).unwrap();
assert_eq!(trainer.total_env_steps(), 0);
assert_eq!(trainer.buffer_len(), 0);
}
#[test]
fn dqn_trainer_burn_rejects_prioritized_config() {
let device = Default::default();
let online = MlpBurnPolicy::<B>::new(4, 2, 16, &device);
let inner_opt = AdamConfig::new().init();
let burn_opt = BurnOptimizer::new(inner_opt, 1e-3);
let cfg = small_config().prioritized_replay(true);
assert!(DQNTrainerBurn::new(cfg, online, burn_opt, 4, 2, device).is_err());
}
#[test]
fn dqn_trainer_burn_train_step_runs() {
let device = Default::default();
let online = MlpBurnPolicy::<B>::new(4, 2, 16, &device);
let inner_opt = AdamConfig::new().init();
let burn_opt = BurnOptimizer::new(inner_opt, 1e-3);
let mut trainer =
DQNTrainerBurn::new(small_config(), online, burn_opt, 4, 2, device).unwrap();
for i in 0..32 {
let phase = (i as f32) * 0.1;
let obs = [phase.sin(), phase.cos(), phase * 0.5, phase * -0.3];
let next_obs = [(phase + 0.1).sin(), (phase + 0.1).cos(), phase * 0.5, phase * -0.3];
let action = (i % 2) as i64;
let reward = if action == 0 { 1.0 } else { -1.0 };
let done = i % 8 == 7;
trainer.buffer_mut().push(&obs, action, reward, &next_obs, done);
}
let mut rng = rand::rngs::StdRng::seed_from_u64(7);
let forward_fn = |q: &MlpBurnPolicy<B>, o: Tensor<B, 2>| -> Tensor<B, 2> {
let (logits, _) = q.forward(o);
logits
};
let stats = trainer.train_step(&mut rng, forward_fn, forward_fn).unwrap();
assert!(stats.is_some());
let s = stats.unwrap();
assert!(s.td_loss.is_finite());
}
#[test]
fn dqn_trainer_burn_train_step_scaled_runs() {
let device = Default::default();
let online = MlpBurnPolicy::<B>::new(4, 2, 16, &device);
let inner_opt = AdamConfig::new().init();
let burn_opt = BurnOptimizer::new(inner_opt, 1e-3);
let mut trainer =
DQNTrainerBurn::new(small_config(), online, burn_opt, 4, 2, device).unwrap();
for i in 0..32 {
let phase = (i as f32) * 0.1;
let obs = [phase.sin(), phase.cos(), phase * 0.5, phase * -0.3];
let next_obs = [(phase + 0.1).sin(), (phase + 0.1).cos(), phase * 0.5, phase * -0.3];
let action = (i % 2) as i64;
let reward = if action == 0 { 1.0 } else { -1.0 };
let done = i % 8 == 7;
trainer.buffer_mut().push(&obs, action, reward, &next_obs, done);
}
let mut rng = rand::rngs::StdRng::seed_from_u64(7);
let forward_fn = |q: &MlpBurnPolicy<B>, o: Tensor<B, 2>| -> Tensor<B, 2> {
let (logits, _) = q.forward(o);
logits
};
let out = trainer.train_step_scaled(&mut rng, 32_768.0, forward_fn, forward_fn).unwrap();
assert!(out.is_some());
let (s, applied) = out.unwrap();
assert!(applied, "scaled step should apply on finite loss");
assert!(s.td_loss.is_finite());
assert_eq!(trainer.total_train_steps(), 1);
}
}