tran_mem_exp.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import gym\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import numpy as np\n",
    "import copy\n",
    "\n",
    "\n",
    "class TransformerExperienceReplay:\n",
    "    def __init__(self, capacity, input_dim):\n",
    "        self.capacity = capacity\n",
    "        self.buffer = [None] * capacity\n",
    "        self.position = 0\n",
    "        self.input_dim = input_dim\n",
    "        self.transformer = nn.Transformer(d_model=input_dim, nhead=4, \n",
    "                                          num_encoder_layers=2, num_decoder_layers=2)\n",
    "\n",
    "\n",
    "    def push(self, experience):\n",
    "        state, action, reward, next_state, done = experience\n",
    "        state_tensor = state.clone().detach()\n",
    "        next_state_tensor = next_state.clone().detach()\n",
    "\n",
    "        action_tensor = torch.tensor([action], dtype=torch.float32)\n",
    "        reward_tensor = torch.tensor([reward], dtype=torch.float32)\n",
    "        done_tensor = torch.tensor([done], dtype=torch.float32)\n",
    "        self.buffer[self.position] = (state_tensor, action_tensor, reward_tensor, next_state_tensor, done_tensor)\n",
    "        self.position = (self.position + 1) % self.capacity\n",
    "\n",
    "    def sample(self, batch_size, current_state):\n",
    "        if len(self.buffer) < batch_size:\n",
    "            return []\n",
    "\n",
    "        query = self._prepare_query(current_state)\n",
    "        keys_values = self._prepare_keys_values()\n",
    "\n",
    "        attention_output = self.transformer(query, keys_values, keys_values)\n",
    "        attention_weights = torch.softmax(attention_output.squeeze(), dim=0)\n",
    "\n",
    "        sampled_indices = torch.multinomial(attention_weights, batch_size, replacement=True)\n",
    "        return [self.buffer[i] for i in sampled_indices]\n",
    "\n",
    "    def _prepare_query(self, current_state, batch_size=1, seq_length=1):\n",
    "        # Flatten current_state and adjust its length to match d_model\n",
    "        current_state_flat = current_state.view(-1)\n",
    "        if current_state_flat.nelement() < self.input_dim:\n",
    "            current_state_padded = torch.nn.functional.pad(current_state_flat, (0, self.input_dim - current_state_flat.nelement()), \"constant\", 0)\n",
    "        else:\n",
    "            current_state_padded = current_state_flat[:self.input_dim]\n",
    "\n",
    "        # Reshape to [batch_size, seq_length, feature_dim]\n",
    "        query = current_state_padded.view(batch_size, seq_length, self.input_dim)\n",
    "        return query\n",
    "\n",
    "\n",
    "    def _prepare_keys_values(self, batch_size=1, seq_length=1):\n",
    "        processed_experiences = []\n",
    "        for exp in self.buffer:\n",
    "            if exp is not None:\n",
    "                state, action, reward, next_state, done = exp\n",
    "                exp_tensor = torch.cat([s.view(-1) for s in [state, next_state, action, reward, done]])\n",
    "                if exp_tensor.nelement() < self.input_dim:\n",
    "                    exp_tensor_padded = torch.nn.functional.pad(exp_tensor, (0, self.input_dim - exp_tensor.nelement()), \"constant\", 0)\n",
    "                else:\n",
    "                    exp_tensor_padded = exp_tensor[:self.input_dim]\n",
    "\n",
    "                exp_tensor_padded = exp_tensor_padded.view(batch_size, seq_length, self.input_dim)\n",
    "                processed_experiences.append(exp_tensor_padded)\n",
    "\n",
    "        if not processed_experiences:\n",
    "            return torch.zeros((batch_size, seq_length, self.input_dim))\n",
    "\n",
    "        keys_values = torch.cat(processed_experiences, dim=0)\n",
    "        return keys_values\n",
    "\n",
    "\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.buffer)\n",
    "\n",
    "    def count_parameters(self):\n",
    "        return sum(p.numel() for p in self.transformer.parameters() if p.requires_grad)\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}