Aprendizaje por Refuerzo

import numpy as np
import gymnasium as gym

# ═══════════════════════════════════════════════════
# Q-Learning tabular — FrozenLake
# ═══════════════════════════════════════════════════

env = gym.make("FrozenLake-v1", is_slippery=False)

# Hiperparámetros
alpha = 0.1          # Tasa de aprendizaje
gamma = 0.99         # Factor de descuento
epsilon = 1.0        # Exploración inicial
eps_min = 0.01       # Exploración mínima
eps_decay = 0.995    # Decay por episodio
n_episodes = 10000

# Inicializar Q-table: |S| × |A| = 16 × 4
n_states = env.observation_space.n     # 16
n_actions = env.action_space.n         # 4
Q = np.zeros((n_states, n_actions))

rewards_history = []

for episode in range(n_episodes):
    state, _ = env.reset()
    total_reward = 0
    done = False

    while not done:
        # ε-greedy
        if np.random.random() < epsilon:
            action = env.action_space.sample()   # Explorar
        else:
            action = np.argmax(Q[state])          # Explotar

        next_state, reward, terminated, truncated, _ = env.step(action)
        done = terminated or truncated

        # Actualización Q-Learning
        td_target = reward + gamma * np.max(Q[next_state]) * (1 - terminated)
        td_error = td_target - Q[state, action]
        Q[state, action] += alpha * td_error

        state = next_state
        total_reward += reward

    # Decay epsilon
    epsilon = max(eps_min, epsilon * eps_decay)
    rewards_history.append(total_reward)

# Mostrar Q-table final
print("Q-table aprendida:")
print(Q.round(2))
print(f"\nÉxito últimos 100 eps: {np.mean(rewards_history[-100:]):.0%}")

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import gymnasium as gym
from collections import deque
import random

# ═══════════════════════════════════════════════════
# 1. Red neuronal Q-Network
# ═══════════════════════════════════════════════════
class QNetwork(nn.Module):
    def __init__(self, state_dim, action_dim, hidden=128):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, hidden),
            nn.ReLU(),
            nn.Linear(hidden, hidden),
            nn.ReLU(),
            nn.Linear(hidden, action_dim)  # Sin activación
        )

    def forward(self, x):
        return self.net(x)

# ═══════════════════════════════════════════════════
# 2. Replay Buffer
# ═══════════════════════════════════════════════════
class ReplayBuffer:
    def __init__(self, capacity=100_000):
        self.buffer = deque(maxlen=capacity)

    def push(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (
            torch.FloatTensor(np.array(states)),
            torch.LongTensor(actions),
            torch.FloatTensor(rewards),
            torch.FloatTensor(np.array(next_states)),
            torch.FloatTensor(dones),
        )

    def __len__(self):
        return len(self.buffer)

# ═══════════════════════════════════════════════════
# 3. Agente DQN
# ═══════════════════════════════════════════════════
class DQNAgent:
    def __init__(self, state_dim, action_dim):
        self.action_dim = action_dim
        self.gamma = 0.99
        self.tau = 0.005
        self.epsilon = 1.0
        self.eps_min = 0.01
        self.eps_decay = 0.995
        self.batch_size = 64

        # Q-network y target network
        self.q_net = QNetwork(state_dim, action_dim)
        self.target_net = QNetwork(state_dim, action_dim)
        self.target_net.load_state_dict(self.q_net.state_dict())

        self.optimizer = optim.Adam(self.q_net.parameters(), lr=1e-3)
        self.buffer = ReplayBuffer()

    def select_action(self, state):
        if random.random() < self.epsilon:
            return random.randrange(self.action_dim)
        with torch.no_grad():
            q_values = self.q_net(torch.FloatTensor(state).unsqueeze(0))
            return q_values.argmax(dim=1).item()

    def train_step(self):
        if len(self.buffer) < self.batch_size:
            return None

        states, actions, rewards, next_states, dones = \
            self.buffer.sample(self.batch_size)

        # Q-values actuales para las acciones tomadas
        q_values = self.q_net(states).gather(1, actions.unsqueeze(1)).squeeze()

        # TD target con target network (sin gradiente)
        with torch.no_grad():
            next_q = self.target_net(next_states).max(dim=1)[0]
            targets = rewards + self.gamma * next_q * (1 - dones)

        # Loss y backprop
        loss = nn.SmoothL1Loss()(q_values, targets)  # Huber Loss
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        # Soft update de target network
        for p, tp in zip(self.q_net.parameters(),
                         self.target_net.parameters()):
            tp.data.copy_(self.tau * p.data + (1 - self.tau) * tp.data)

        return loss.item()

    def decay_epsilon(self):
        self.epsilon = max(self.eps_min, self.epsilon * self.eps_decay)

# ═══════════════════════════════════════════════════
# 4. Training loop
# ═══════════════════════════════════════════════════
env = gym.make("CartPole-v1")
agent = DQNAgent(state_dim=4, action_dim=2)

for episode in range(500):
    state, _ = env.reset()
    total_reward = 0

    while True:
        action = agent.select_action(state)
        next_state, reward, term, trunc, _ = env.step(action)
        done = term or trunc

        agent.buffer.push(state, action, reward, next_state, float(done))
        loss = agent.train_step()

        state = next_state
        total_reward += reward

        if done:
            break

    agent.decay_epsilon()

    if (episode + 1) % 50 == 0:
        print(f"Ep {episode+1} | Reward: {total_reward:.0f} | "
              f"ε: {agent.epsilon:.3f} | Buffer: {len(agent.buffer)}")

import tensorflow as tf
import numpy as np
import gymnasium as gym
from collections import deque
import random

# ═══════════════════════════════════════════════════
# DQN en TensorFlow/Keras — CartPole
# ═══════════════════════════════════════════════════

def build_q_network(state_dim, action_dim, hidden=128):
    return tf.keras.Sequential([
        tf.keras.layers.Dense(hidden, activation='relu', input_shape=(state_dim,)),
        tf.keras.layers.Dense(hidden, activation='relu'),
        tf.keras.layers.Dense(action_dim)  # Sin activación
    ])

# Crear redes
q_net = build_q_network(4, 2)
target_net = build_q_network(4, 2)
target_net.set_weights(q_net.get_weights())

optimizer = tf.keras.optimizers.Adam(1e-3)
buffer = deque(maxlen=100_000)
gamma, tau, batch_size = 0.99, 0.005, 64
epsilon, eps_min, eps_decay = 1.0, 0.01, 0.995

@tf.function
def train_step(states, actions, rewards, next_states, dones):
    # TD target
    next_q = tf.reduce_max(target_net(next_states), axis=1)
    targets = rewards + gamma * next_q * (1.0 - dones)

    with tf.GradientTape() as tape:
        q_values = q_net(states)
        indices = tf.stack([tf.range(batch_size), actions], axis=1)
        q_selected = tf.gather_nd(q_values, indices)
        loss = tf.keras.losses.Huber()(targets, q_selected)

    grads = tape.gradient(loss, q_net.trainable_variables)
    optimizer.apply_gradients(zip(grads, q_net.trainable_variables))

    # Soft update
    for w, tw in zip(q_net.weights, target_net.weights):
        tw.assign(tau * w + (1 - tau) * tw)

    return loss

env = gym.make("CartPole-v1")
for episode in range(500):
    state, _ = env.reset()
    total_reward = 0

    while True:
        if random.random() < epsilon:
            action = env.action_space.sample()
        else:
            q = q_net(tf.expand_dims(state, 0))
            action = int(tf.argmax(q, axis=1))

        ns, r, term, trunc, _ = env.step(action)
        buffer.append((state, action, r, ns, float(term or trunc)))
        state = ns
        total_reward += r

        if len(buffer) >= batch_size:
            batch = random.sample(buffer, batch_size)
            s, a, rw, ns2, d = map(np.array, zip(*batch))
            train_step(
                tf.constant(s, dtype=tf.float32),
                tf.constant(a, dtype=tf.int32),
                tf.constant(rw, dtype=tf.float32),
                tf.constant(ns2, dtype=tf.float32),
                tf.constant(d, dtype=tf.float32),
            )

        if term or trunc:
            break

    epsilon = max(eps_min, epsilon * eps_decay)

import torch
import torch.nn as nn
import torch.optim as optim
import gymnasium as gym

class PolicyNetwork(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, 128), nn.ReLU(),
            nn.Linear(128, action_dim),
            nn.Softmax(dim=-1)  # Distribución sobre acciones
        )

    def forward(self, x):
        return self.net(x)

env = gym.make("CartPole-v1")
policy = PolicyNetwork(4, 2)
optimizer = optim.Adam(policy.parameters(), lr=1e-3)

for episode in range(1000):
    states, actions, rewards = [], [], []
    state, _ = env.reset()

    # 1. Recoger episodio completo
    while True:
        probs = policy(torch.FloatTensor(state))
        dist = torch.distributions.Categorical(probs)
        action = dist.sample()

        next_state, reward, term, trunc, _ = env.step(action.item())
        states.append(state)
        actions.append(action)
        rewards.append(reward)

        state = next_state
        if term or trunc:
            break

    # 2. Calcular retornos descontados
    G, returns = 0, []
    for r in reversed(rewards):
        G = r + 0.99 * G
        returns.insert(0, G)
    returns = torch.FloatTensor(returns)
    returns = (returns - returns.mean()) / (returns.std() + 1e-8)  # Baseline

    # 3. Policy gradient update
    loss = 0
    for s, a, G_t in zip(states, actions, returns):
        probs = policy(torch.FloatTensor(s))
        log_prob = torch.log(probs[a])
        loss -= log_prob * G_t  # Negativo porque optimizer minimiza

    optimizer.zero_grad()
    loss.backward()
    optimizer.step()