mirror of
https://github.com/wassname/kair_algorithms_draft.git
synced 2026-08-24 05:47:53 +08:00
Seungjae Ryan Lee
ca5c99bc41
* Format repository * Clone files from medipixel repo * Fix DDPGfDAgent.update_model() * Fix bug on _initialize() * Add demo-path parameter and demo data * Rename init_priority to _max_priority for PER This makes PER and PERfD consistent. * Make i_episode attribute of DDPGAgent * Clone SAC code from medipixel repo * Fix update_model() for SACfD * Fix _initialize() for SACfD * Add is_discrete attribute to AbstractAgent for SACfD * Add i_episode attribute to SACAgent for SACfD * Modularize DDPGAgent and SACAgent * Modify hyperparameters for DDPGfD and SACfD * Add NStepBuffer * Add n-step to DDPGfD * Add n-step to SACfD * Add TD3fD without n-step * Attempt to tune hyperparameters * Remove discrete environment check in SAC * Implement n-step on TD3fD * Fix step function of TD3 No done check, and _add_transition_to_memory was not called. * Fix actor loss calculation for TD3fD * Attempt to tune hyperparameters * Print both critic losses * Fix typo bug * Attempt to tune hyperparameters * Fix bug in n-step demo retrieval * Fix bug in n-step transition addition
80 lines
2.2 KiB
Python
80 lines
2.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Common util functions for all algorithms.
|
|
|
|
- Author: Curt Park
|
|
- Contact: curt.park@medipixel.io
|
|
"""
|
|
|
|
import random
|
|
from collections import deque
|
|
from typing import Deque, List, Tuple
|
|
|
|
import gym
|
|
import numpy as np
|
|
import torch
|
|
import torch.nn as nn
|
|
|
|
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
|
|
|
|
|
|
def identity(x: torch.Tensor) -> torch.Tensor:
|
|
"""Return input without any change."""
|
|
return x
|
|
|
|
|
|
def soft_update(local: nn.Module, target: nn.Module, tau: float):
|
|
"""Soft-update: target = tau*local + (1-tau)*target."""
|
|
for t_param, l_param in zip(target.parameters(), local.parameters()):
|
|
t_param.data.copy_(tau * l_param.data + (1.0 - tau) * t_param.data)
|
|
|
|
|
|
def set_random_seed(seed: int, env: gym.Env):
|
|
"""Set random seed"""
|
|
env.seed(seed)
|
|
torch.manual_seed(seed)
|
|
np.random.seed(seed)
|
|
random.seed(seed)
|
|
|
|
|
|
def get_n_step_info_from_demo(
|
|
demo: List, n_step: int, gamma: float
|
|
) -> Tuple[List, List]:
|
|
"""Return 1 step and n step demos."""
|
|
assert demo
|
|
assert n_step > 1
|
|
|
|
demos_1_step = list()
|
|
demos_n_step = list()
|
|
n_step_buffer: Deque = deque(maxlen=n_step)
|
|
|
|
for transition in demo:
|
|
n_step_buffer.append(transition)
|
|
|
|
if len(n_step_buffer) == n_step:
|
|
# add a single step transition
|
|
demos_1_step.append(n_step_buffer[0])
|
|
|
|
# add a multi step transition
|
|
curr_state, action = n_step_buffer[0][:2]
|
|
reward, next_state, done = get_n_step_info(n_step_buffer, gamma)
|
|
transition = (curr_state, action, reward, next_state, done)
|
|
demos_n_step.append(transition)
|
|
|
|
return demos_1_step, demos_n_step
|
|
|
|
|
|
def get_n_step_info(
|
|
n_step_buffer: Deque, gamma: float
|
|
) -> Tuple[np.int64, np.ndarray, bool]:
|
|
"""Return n step reward, next state, and done."""
|
|
# info of the last transition
|
|
reward, next_state, done = n_step_buffer[-1][-3:]
|
|
|
|
for transition in reversed(list(n_step_buffer)[:-1]):
|
|
r, n_s, d = transition[-3:]
|
|
|
|
reward = r + gamma * reward * (1 - d)
|
|
next_state, done = (n_s, d) if d else (next_state, done)
|
|
|
|
return reward, next_state, done
|