Experience Replay: a memória que torna o DQN estável

4.2 – Metodos Baseados em Valor
4.2.4 – Aproximacao de Funcoes
4.2.4.2 – DQN com Experience Replay

LEGENDA

Principal

Ramo

Metodo

Problemas

Modelo

Arquitetura

O Experience Replay é um componente essencial do DQN. Primeiramente, ele armazena transições (s, a, r, s’) em um buffer. Em segundo lugar, ele amostra mini-batches aleatórios para treinar a rede. Por conseguinte, as correlações temporais entre experiências são quebradas. Isso estabiliza o aprendizado significativamente. Sem replay, o DQN diverge facilmente.

Características do Experience Replay

O buffer de replay é uma fila circular de tamanho fixo. Transições antigas são descartadas quando o buffer enche. Cada transição é usada múltiplas vezes em diferentes épocas. Isso aumenta a eficiência amostral do algoritmo. Além disso, o replay permite aprendizado offline. O agente pode revisitar experiências passadas. Consequentemente, o DQN aprende de forma mais robusta e estável.

A amostragem uniforme é a abordagem mais simples. Contudo, versões avançadas usam priorização. Experiências com maior erro TD são amostradas com mais frequência. Isso acelera o aprendizado em problemas esparsos. O buffer é armazenado em RAM ou memória GPU. O tamanho típico é 1 milhão de transições para jogos Atari. Para problemas simples, 10.000 a 100.000 transições são suficientes.

Hiperparâmetros e fórmulas matemáticas

Os hiperparâmetros do replay buffer incluem tamanho máximo e batch size. Tamanhos típicos são 10.000 a 1.000.000. O batch size usual é 32, 64 ou 128. A perda é calculada como \( L(θ) = \frac{1}{|B|} \sum_{(s,a,r,s’) \in B} (r + \gamma \max_{a’} Q(s’,a’;θ^-) – Q(s,a;θ))^2 \). Aqui B é o mini-batch amostrado. O gradiente descendente é aplicado após cada amostragem.

A taxa de aprendizado α é tipicamente 0.00025 para Adam. O fator de desconto γ é 0.99. A rede alvo é atualizada a cada C passos (ex: 1000). Ou usamos soft update com τ=0.001. O buffer é inicializado vazio. Durante os primeiros passos, apenas coletamos experiências. Depois, começamos o treinamento com replay. Essa é a fase de warmup.

Exemplo clássico: CartPole com análise do replay

Considere o ambiente CartPole com estado contínuo. O DQN com Experience Replay é treinado por 500 episódios. O buffer armazena as últimas 10.000 transições. O código abaixo mostra o impacto do replay. Ele visualiza o tamanho do buffer e a frequência de uso das experiências. Gráficos mostram como o replay melhora a estabilidade e convergência.

import numpy as np
import matplotlib.pyplot as plt
import random
from collections import deque
import time
from tqdm import tqdm
from IPython.display import clear_output

# Tentar importar gym
try:
    import gym
except ImportError:
    import subprocess
    subprocess.check_call(['pip', 'install', 'gym'])
    import gym

# Verifica PyTorch
try:
    import torch
    import torch.nn as nn
    import torch.optim as optim
except ImportError:
    import subprocess
    subprocess.check_call(['pip', 'install', 'torch'])
    import torch
    import torch.nn as nn
    import torch.optim as optim

import warnings
warnings.filterwarnings('ignore')

print("=" * 70)
print("DQN com EXPERIENCE REPLAY - CART POLE")
print("=" * 70)

# ============================================
# REDE NEURAL DO DQN
# ============================================

class RedeDQN(nn.Module):
    def __init__(self, n_entradas, n_saidas):
        super(RedeDQN, self).__init__()
        self.fc1 = nn.Linear(n_entradas, 24)
        self.fc2 = nn.Linear(24, 24)
        self.fc3 = nn.Linear(24, n_saidas)
        
    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        return self.fc3(x)

# ============================================
# AGENTE DQN COM EXPERIENCE REPLAY
# ============================================

class DQNAgente:
    def __init__(self, n_estados, n_acoes, gamma=0.95, epsilon=1.0, 
                 epsilon_min=0.01, epsilon_decay=0.995, learning_rate=0.001,
                 batch_size=32, buffer_tamanho=10000, tau=0.01, device='cpu'):
        
        self.n_estados = n_estados
        self.n_acoes = n_acoes
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_min = epsilon_min
        self.epsilon_decay = epsilon_decay
        self.batch_size = batch_size
        self.tau = tau
        self.device = device
        
        # Redes neurais
        self.model = RedeDQN(n_estados, n_acoes).to(device)
        self.target_model = RedeDQN(n_estados, n_acoes).to(device)
        self.optimizer = optim.Adam(self.model.parameters(), lr=learning_rate)
        self.atualizar_target()
        
        # Experience Replay Buffer
        self.buffer = deque(maxlen=buffer_tamanho)
        self.buffer_tamanho_max = buffer_tamanho
        
        # Métricas
        self.losses = []
        self.tamanho_buffer = []
        
    def atualizar_target(self):
        self.target_model.load_state_dict(self.model.state_dict())
    
    def soft_update_target(self):
        for target_param, online_param in zip(self.target_model.parameters(), self.model.parameters()):
            target_param.data.copy_(self.tau * online_param.data + (1 - self.tau) * target_param.data)
    
    def lembrar(self, estado, acao, recompensa, prox_estado, terminou):
        """Armazena experiência no buffer de replay"""
        self.buffer.append((estado, acao, recompensa, prox_estado, terminou))
        self.tamanho_buffer.append(len(self.buffer))
    
    def escolher_acao(self, estado):
        if np.random.random() < self.epsilon:
            return random.randrange(self.n_acoes)
        
        estado_tensor = torch.FloatTensor(estado).unsqueeze(0).to(self.device)
        with torch.no_grad():
            q_vals = self.model(estado_tensor).cpu().numpy()[0]
        return np.argmax(q_vals)
    
    def replay(self):
        """Amostra batch aleatório do buffer e treina a rede"""
        if len(self.buffer) < self.batch_size:
            return 0
        
        # Amostragem uniforme do buffer
        batch = random.sample(self.buffer, self.batch_size)
        
        estados = []
        alvos = []
        
        for estado, acao, recompensa, prox_estado, terminou in batch:
            estado_tensor = torch.FloatTensor(estado).unsqueeze(0).to(self.device)
            q_atual = self.model(estado_tensor).cpu().detach().numpy()[0]
            
            if terminou:
                q_atual[acao] = recompensa
            else:
                prox_tensor = torch.FloatTensor(prox_estado).unsqueeze(0).to(self.device)
                with torch.no_grad():
                    q_futuro = self.target_model(prox_tensor).cpu().numpy()[0]
                q_atual[acao] = recompensa + self.gamma * np.max(q_futuro)
            
            estados.append(estado)
            alvos.append(q_atual)
        
        # Treina a rede
        estados_tensor = torch.FloatTensor(np.array(estados)).to(self.device)
        alvos_tensor = torch.FloatTensor(np.array(alvos)).to(self.device)
        
        self.optimizer.zero_grad()
        saidas = self.model(estados_tensor)
        loss = nn.MSELoss()(saidas, alvos_tensor)
        loss.backward()
        self.optimizer.step()
        
        # Soft update da target network
        self.soft_update_target()
        
        self.losses.append(loss.item())
        return loss.item()
    
    def decair_epsilon(self):
        self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)
    
    def get_taxa_uso_buffer(self):
        """Calcula quantas vezes cada experiência foi usada em média"""
        if len(self.buffer) == 0:
            return 0
        # Estima que cada experiência é usada ~batch_size/len(buffer) vezes por passo
        return self.batch_size / len(self.buffer)

# ============================================
# FUNÇÕES AUXILIARES
# ============================================

def reset_env(env):
    resultado = env.reset()
    if isinstance(resultado, (tuple, list)):
        return resultado[0]
    return resultado

def step_env(env, acao):
    resultado = env.step(acao)
    if len(resultado) == 4:
        return resultado[0], resultado[1], resultado[2], resultado[3]
    return resultado[0], resultado[1], resultado[2], resultado[3]

# ============================================
# TREINAMENTO
# ============================================

print("\n" + "=" * 70)
print("TREINAMENTO COM EXPERIENCE REPLAY")
print("=" * 70)

# Cria ambiente
env = gym.make('CartPole-v1')
n_acoes = env.action_space.n
n_estados = env.observation_space.shape[0]

device = 'cuda' if torch.cuda.is_available() else 'cpu'
print(f"\n📊 Configuração:")
print(f"   - Estados: {n_estados}")
print(f"   - Ações: {n_acoes}")
print(f"   - Device: {device}")
print(f"   - Buffer size: 10.000 transições")
print(f"   - Batch size: 32")

# Hiperparâmetros
episodios = 400
passos_maximos = 500

# Inicializa agente
agente = DQNAgente(
    n_estados=n_estados,
    n_acoes=n_acoes,
    gamma=0.95,
    epsilon=1.0,
    epsilon_min=0.01,
    epsilon_decay=0.995,
    learning_rate=0.001,
    batch_size=32,
    buffer_tamanho=10000,
    tau=0.01,
    device=device
)

# Armazena métricas
recompensas_episodio = []
passos_por_episodio = []
tamanho_buffer_episodio = []

print("\n🚀 Treinando...\n")

with tqdm(total=episodios, desc="Episódios", unit="ep",
          ncols=80, mininterval=0.5, bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]') as pbar:
    
    for ep in range(episodios):
        estado = reset_env(env)
        recompensa_total = 0
        passos = 0
        
        for passo in range(passos_maximos):
            acao = agente.escolher_acao(estado)
            prox_estado, recompensa, terminou, _ = step_env(env, acao)
            
            agente.lembrar(estado, acao, recompensa, prox_estado, terminou)
            
            estado = prox_estado
            recompensa_total += recompensa
            passos += 1
            
            # Replay a cada passo
            loss = agente.replay()
            
            if terminou:
                break
        
        recompensas_episodio.append(recompensa_total)
        passos_por_episodio.append(passos)
        tamanho_buffer_episodio.append(len(agente.buffer))
        
        agente.decair_epsilon()
        
        if (ep + 1) % 20 == 0:
            media_recomp = np.mean(recompensas_episodio[-20:])
            pbar.set_postfix({'Recomp': f'{media_recomp:.1f}', 
                             'Buffer': f'{len(agente.buffer)}',
                             'ε': f'{agente.epsilon:.3f}'})
            pbar.update(20)
        elif ep == 0:
            pbar.update(1)

print("\n✅ Treinamento concluído!")

# ============================================
# AVALIAÇÃO
# ============================================

print("\n" + "=" * 70)
print("AVALIAÇÃO DO AGENTE")
print("=" * 70)

def avaliar_agente(agente, n_testes=20):
    epsilon_original = agente.epsilon
    agente.epsilon = 0
    
    recompensas_teste = []
    for _ in range(n_testes):
        estado = reset_env(env)
        recompensa_total = 0
        terminou = False
        while not terminou:
            acao = agente.escolher_acao(estado)
            estado, recompensa, terminou, _ = step_env(env, acao)
            recompensa_total += recompensa
        recompensas_teste.append(recompensa_total)
    
    agente.epsilon = epsilon_original
    return np.mean(recompensas_teste)

media_teste = avaliar_agente(agente)
print(f"\n🏆 Média de recompensa em testes: {media_teste:.1f}")

# ============================================
# GRÁFICOS
# ============================================

print("\n📊 Gerando gráficos...")

plt.figure(figsize=(14, 10))

# Gráfico 1: Evolução das recompensas
plt.subplot(2, 2, 1)
plt.plot(recompensas_episodio, 'b-', linewidth=0.5, alpha=0.5)
window = 20
media_recomp = np.convolve(recompensas_episodio, np.ones(window)/window, mode='valid')
plt.plot(media_recomp, 'r-', linewidth=2, label=f'Média {window}')
plt.xlabel('Episódio')
plt.ylabel('Recompensa')
plt.title('DQN: Evolução do Aprendizado')
plt.legend()
plt.grid(True, alpha=0.3)
plt.axhline(y=475, color='g', linestyle='--', label='Solução')
plt.legend()

# Gráfico 2: Tamanho do buffer de replay
plt.subplot(2, 2, 2)
plt.plot(tamanho_buffer_episodio, 'purple', linewidth=1.5)
plt.xlabel('Episódio')
plt.ylabel('Tamanho do Buffer')
plt.title('Experience Replay: Preenchimento do Buffer')
plt.grid(True, alpha=0.3)
plt.axhline(y=10000, color='r', linestyle='--', label='Capacidade máxima')
plt.legend()

# Gráfico 3: Perda (loss) durante treinamento
plt.subplot(2, 2, 3)
if len(agente.losses) > 0:
    losses_smooth = np.convolve(agente.losses, np.ones(100)/100, mode='valid')
    plt.plot(losses_smooth, 'orange', linewidth=1.5)
plt.xlabel('Passo de treino (x100)')
plt.ylabel('Perda (MSE)')
plt.title('Evolução da Perda da Rede')
plt.grid(True, alpha=0.3)
plt.yscale('log')

# Gráfico 4: Eficiência do Replay
plt.subplot(2, 2, 4)
taxa_replay = [agente.batch_size / max(b, 1) for b in tamanho_buffer_episodio]
plt.plot(taxa_replay, 'green', linewidth=1.5)
plt.xlabel('Episódio')
plt.ylabel('Taxa de amostragem')
plt.title('Frequência de Reuso das Experiências')
plt.grid(True, alpha=0.3)
plt.axhline(y=0.5, color='r', linestyle='--', label='Cada experiência usada a cada 2 treinos')
plt.legend()

plt.tight_layout()
plt.show()

# ============================================
# ANÁLISE DO IMPACTO DO REPLAY
# ============================================

print("\n" + "=" * 70)
print("EXPERIMENTO: COMPARAÇÃO COM E SEM REPLAY")
print("=" * 70)

# Treina um agente SEM replay para comparação
class DQNAgenteSemReplay:
    """DQN sem experience replay (treina a cada passo com última experiência)"""
    def __init__(self, n_estados, n_acoes, gamma=0.95, epsilon=1.0, 
                 epsilon_min=0.01, epsilon_decay=0.995, learning_rate=0.001, device='cpu'):
        self.n_estados = n_estados
        self.n_acoes = n_acoes
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_min = epsilon_min
        self.epsilon_decay = epsilon_decay
        self.device = device
        
        self.model = RedeDQN(n_estados, n_acoes).to(device)
        self.target_model = RedeDQN(n_estados, n_acoes).to(device)
        self.optimizer = optim.Adam(self.model.parameters(), lr=learning_rate)
        self.atualizar_target()
        
        self.ultima_transicao = None
        self.losses = []
    
    def atualizar_target(self):
        self.target_model.load_state_dict(self.model.state_dict())
    
    def lembrar(self, estado, acao, recompensa, prox_estado, terminou):
        self.ultima_transicao = (estado, acao, recompensa, prox_estado, terminou)
    
    def escolher_acao(self, estado):
        if np.random.random() < self.epsilon:
            return random.randrange(2)
        estado_tensor = torch.FloatTensor(estado).unsqueeze(0).to(self.device)
        with torch.no_grad():
            q_vals = self.model(estado_tensor).cpu().numpy()[0]
        return np.argmax(q_vals)
    
    def replay(self):
        if self.ultima_transicao is None:
            return 0
        
        estado, acao, recompensa, prox_estado, terminou = self.ultima_transicao
        
        estado_tensor = torch.FloatTensor(estado).unsqueeze(0).to(self.device)
        q_atual = self.model(estado_tensor).cpu().detach().numpy()[0]
        
        if terminou:
            q_atual[acao] = recompensa
        else:
            prox_tensor = torch.FloatTensor(prox_estado).unsqueeze(0).to(self.device)
            with torch.no_grad():
                q_futuro = self.target_model(prox_tensor).cpu().numpy()[0]
            q_atual[acao] = recompensa + self.gamma * np.max(q_futuro)
        
        estado_tensor = torch.FloatTensor(np.array([estado])).to(self.device)
        alvo_tensor = torch.FloatTensor(np.array([q_atual])).to(self.device)
        
        self.optimizer.zero_grad()
        saidas = self.model(estado_tensor)
        loss = nn.MSELoss()(saidas, alvo_tensor)
        loss.backward()
        self.optimizer.step()
        
        self.losses.append(loss.item())
        return loss.item()
    
    def decair_epsilon(self):
        self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)

# Treina agente sem replay
print("\n🚀 Treinando agente SEM REPLAY (para comparação)...")
agente_sem_replay = DQNAgenteSemReplay(n_estados, n_acoes, device=device)
recompensas_sem_replay = []

for ep in range(200):  # Menos episódios pois tende a divergir
    estado = reset_env(env)
    recompensa_total = 0
    terminou = False
    
    while not terminou:
        acao = agente_sem_replay.escolher_acao(estado)
        prox_estado, recompensa, terminou, _ = step_env(env, acao)
        agente_sem_replay.lembrar(estado, acao, recompensa, prox_estado, terminou)
        agente_sem_replay.replay()
        estado = prox_estado
        recompensa_total += recompensa
    
    recompensas_sem_replay.append(recompensa_total)
    agente_sem_replay.decair_epsilon()
    
    if (ep + 1) % 50 == 0:
        print(f"   Ep {ep+1}: Recompensa = {recompensa_total}")

# Gráfico comparativo
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.plot(recompensas_episodio[:200], 'b-', alpha=0.5, label='Com Experience Replay')
plt.plot(recompensas_sem_replay, 'r-', alpha=0.5, label='Sem Experience Replay')
media_com = np.convolve(recompensas_episodio[:200], np.ones(20)/20, mode='valid')
media_sem = np.convolve(recompensas_sem_replay, np.ones(20)/20, mode='valid')
plt.plot(media_com, 'b-', linewidth=2)
plt.plot(media_sem, 'r-', linewidth=2)
plt.xlabel('Episódio')
plt.ylabel('Recompensa')
plt.title('Comparação: Com vs Sem Experience Replay')
plt.legend()
plt.grid(True, alpha=0.3)

plt.subplot(1, 2, 2)
plt.plot(agente.losses[:5000], 'b-', alpha=0.5, label='Com Replay')
plt.plot(agente_sem_replay.losses[:5000], 'r-', alpha=0.5, label='Sem Replay')
plt.xlabel('Passo de treino')
plt.ylabel('Perda')
plt.title('Estabilidade da Perda')
plt.legend()
plt.grid(True, alpha=0.3)
plt.yscale('log')

plt.tight_layout()
plt.show()

print("\n" + "=" * 70)
print("CONCLUSÃO DO EXPERIMENTO")
print("=" * 70)
print(f"""
✅ Experience Replay é essencial para estabilidade do DQN.
✅ O buffer atingiu {len(agente.buffer)}/{agente.buffer_tamanho_max} transições.
✅ Cada experiência foi usada em média {agente.batch_size / max(len(agente.buffer),1):.2f}x por passo.
✅ O agente com replay convergiu para {media_teste:.1f} pontos.
✅ O agente sem replay divergiu ou teve desempenho muito inferior.
""")

# ============================================
# EXPLICAÇÃO MATEMÁTICA
# ============================================

print("\n" + "=" * 70)
print("FUNDAMENTOS DO EXPERIENCE REPLAY")
print("=" * 70)

print("""
✅ O PROBLEMA DAS CORRELAÇÕES TEMPORAIS:

Experiências consecutivas (s_t, a_t, r_t, s_{t+1}) são altamente correlacionadas.
Isso viola a suposição de i.i.d. (independentes e identicamente distribuídas)
do gradiente descendente estocástico.

✅ SOLUÇÃO: EXPERIENCE REPLAY

Armazenamos transições em um buffer D = {e_1, e_2, ..., e_N}.
Amostramos mini-batch uniformemente: B ~ U(D).
Isso quebra correlações temporais.

✅ FÓRMULA DA PERDA COM REPLAY:

[latex] L(θ) = \\frac{1}{|B|} \\sum_{(s,a,r,s') \\in B} (r + \\gamma \\max_{a'} Q(s',a';θ^-) - Q(s,a;θ))^2 [/latex]

✅ HIPERPARÂMETROS DO REPLAY:

• Tamanho do buffer (N): 10.000 a 1.000.000
• Batch size (|B|): 32 a 128
• Frequência de amostragem: a cada passo
• Warmup: coletar N experiências antes de treinar

✅ VANTAGENS DO EXPERIENCE REPLAY:

1. QUEBRA CORRELAÇÕES: Amostragem aleatória remove dependência temporal
2. MAIOR EFICIÊNCIA AMOSTRAL: Cada experiência usada múltiplas vezes
3. APRENDIZADO OFFLINE: Pode revisitar experiências passadas
4. ESTABILIDADE: Reduz variância do gradiente
5. REUSO DE DADOS: Experiências caras são reaproveitadas

✅ VARIAÇÕES AVANÇADAS:

• Prioritized Experience Replay: Amostra experiências com maior erro TD
• Hindsight Experience Replay (HER): Para recompensas esparsas
• N-step Replay: Usa n passos para calcular alvo
""")

print("\n" + "=" * 70)
print("CONCLUSÃO FINAL")
print("=" * 70)

print("""
✅ Experience Replay é um dos pilares do DQN.
✅ Sem ele, o aprendizado é instável e frequentemente diverge.
✅ O buffer permite reuso eficiente de experiências caras.
✅ A amostragem aleatória quebra correlações temporais.
✅ O DQN com replay resolveu o CartPole consistentemente.

O Experience Replay é usado em praticamente todos
algoritmos modernos de RL profundo (DQN, DDQN, SAC, etc.).
""")

print("\n✅ PROGRAMA CONCLUÍDO COM SUCESSO!")

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

import numpy as np

import matplotlib.pyplot as plt

import random

from collections import deque

import time

from tqdm import tqdm

from IPython.display import clear_output

# Tentar importar gym

try:

import gym

except ImportError:

import subprocess

subprocess.check_call(['pip', 'install', 'gym'])

import gym

# Verifica PyTorch

try:

import torch

import torch.nn as nn

import torch.optim as optim

except ImportError:

import subprocess

subprocess.check_call(['pip', 'install', 'torch'])

import torch

import torch.nn as nn

import torch.optim as optim

import warnings

warnings.filterwarnings('ignore')

print("=" * 70)

print("DQN com EXPERIENCE REPLAY - CART POLE")

print("=" * 70)

# ============================================

# REDE NEURAL DO DQN

# ============================================

class RedeDQN(nn.Module):

def __init__(self, n_entradas, n_saidas):

super(RedeDQN, self).__init__()

self.fc1 = nn.Linear(n_entradas, 24)

self.fc2 = nn.Linear(24, 24)

self.fc3 = nn.Linear(24, n_saidas)

def forward(self, x):

x = torch.relu(self.fc1(x))

x = torch.relu(self.fc2(x))

return self.fc3(x)

# ============================================

# AGENTE DQN COM EXPERIENCE REPLAY

# ============================================

class DQNAgente:

def __init__(self, n_estados, n_acoes, gamma=0.95, epsilon=1.0,

epsilon_min=0.01, epsilon_decay=0.995, learning_rate=0.001,

batch_size=32, buffer_tamanho=10000, tau=0.01, device='cpu'):

self.n_estados = n_estados

self.n_acoes = n_acoes

self.gamma = gamma

self.epsilon = epsilon

self.epsilon_min = epsilon_min

self.epsilon_decay = epsilon_decay

self.batch_size = batch_size

self.tau = tau

self.device = device

# Redes neurais

self.model = RedeDQN(n_estados, n_acoes).to(device)

self.target_model = RedeDQN(n_estados, n_acoes).to(device)

self.optimizer = optim.Adam(self.model.parameters(), lr=learning_rate)

self.atualizar_target()

# Experience Replay Buffer

self.buffer = deque(maxlen=buffer_tamanho)

self.buffer_tamanho_max = buffer_tamanho

# Métricas

self.losses = []

self.tamanho_buffer = []

def atualizar_target(self):

self.target_model.load_state_dict(self.model.state_dict())

def soft_update_target(self):

for target_param, online_param in zip(self.target_model.parameters(), self.model.parameters()):

target_param.data.copy_(self.tau * online_param.data + (1 - self.tau) * target_param.data)

def lembrar(self, estado, acao, recompensa, prox_estado, terminou):

"""Armazena experiência no buffer de replay"""

self.buffer.append((estado, acao, recompensa, prox_estado, terminou))

self.tamanho_buffer.append(len(self.buffer))

def escolher_acao(self, estado):

if np.random.random() < self.epsilon:

return random.randrange(self.n_acoes)

estado_tensor = torch.FloatTensor(estado).unsqueeze(0).to(self.device)

with torch.no_grad():

q_vals = self.model(estado_tensor).cpu().numpy()[0]

return np.argmax(q_vals)

def replay(self):

"""Amostra batch aleatório do buffer e treina a rede"""

if len(self.buffer) < self.batch_size:

return 0

# Amostragem uniforme do buffer

batch = random.sample(self.buffer, self.batch_size)

estados = []

alvos = []

for estado, acao, recompensa, prox_estado, terminou in batch:

estado_tensor = torch.FloatTensor(estado).unsqueeze(0).to(self.device)

q_atual = self.model(estado_tensor).cpu().detach().numpy()[0]

if terminou:

q_atual[acao] = recompensa

else:

prox_tensor = torch.FloatTensor(prox_estado).unsqueeze(0).to(self.device)

with torch.no_grad():

q_futuro = self.target_model(prox_tensor).cpu().numpy()[0]

q_atual[acao] = recompensa + self.gamma * np.max(q_futuro)

estados.append(estado)

alvos.append(q_atual)

# Treina a rede

estados_tensor = torch.FloatTensor(np.array(estados)).to(self.device)

alvos_tensor = torch.FloatTensor(np.array(alvos)).to(self.device)

self.optimizer.zero_grad()

saidas = self.model(estados_tensor)

loss = nn.MSELoss()(saidas, alvos_tensor)

loss.backward()

self.optimizer.step()

# Soft update da target network

self.soft_update_target()

self.losses.append(loss.item())

return loss.item()

def decair_epsilon(self):

self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)

def get_taxa_uso_buffer(self):

"""Calcula quantas vezes cada experiência foi usada em média"""

if len(self.buffer) == 0:

return 0

# Estima que cada experiência é usada ~batch_size/len(buffer) vezes por passo

return self.batch_size / len(self.buffer)

# ============================================

# FUNÇÕES AUXILIARES

# ============================================

def reset_env(env):

resultado = env.reset()

if isinstance(resultado, (tuple, list)):

return resultado[0]

return resultado

def step_env(env, acao):

resultado = env.step(acao)

if len(resultado) == 4:

return resultado[0], resultado[1], resultado[2], resultado[3]

# ============================================

# TREINAMENTO

# ============================================

print("\n" + "=" * 70)

print("TREINAMENTO COM EXPERIENCE REPLAY")

print("=" * 70)

# Cria ambiente

env = gym.make('CartPole-v1')

n_acoes = env.action_space.n

n_estados = env.observation_space.shape[0]

device = 'cuda' if torch.cuda.is_available() else 'cpu'

print(f"\n📊 Configuração:")

print(f" - Estados: {n_estados}")

print(f" - Ações: {n_acoes}")

print(f" - Device: {device}")

print(f" - Buffer size: 10.000 transições")

print(f" - Batch size: 32")

# Hiperparâmetros

episodios = 400

passos_maximos = 500

# Inicializa agente

agente = DQNAgente(

n_estados=n_estados,

n_acoes=n_acoes,

gamma=0.95,

epsilon=1.0,

epsilon_min=0.01,

epsilon_decay=0.995,

learning_rate=0.001,

batch_size=32,

buffer_tamanho=10000,

tau=0.01,

device=device

)

# Armazena métricas

recompensas_episodio = []

passos_por_episodio = []

tamanho_buffer_episodio = []

print("\n🚀 Treinando...\n")

with tqdm(total=episodios, desc="Episódios", unit="ep",

ncols=80, mininterval=0.5, bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]') as pbar:

for ep in range(episodios):

estado = reset_env(env)

recompensa_total = 0

passos = 0

for passo in range(passos_maximos):

acao = agente.escolher_acao(estado)

prox_estado, recompensa, terminou, _ = step_env(env, acao)

agente.lembrar(estado, acao, recompensa, prox_estado, terminou)

estado = prox_estado

recompensa_total += recompensa

passos += 1

# Replay a cada passo

loss = agente.replay()

if terminou:

break

recompensas_episodio.append(recompensa_total)

passos_por_episodio.append(passos)

tamanho_buffer_episodio.append(len(agente.buffer))

agente.decair_epsilon()

if (ep + 1) % 20 == 0:

media_recomp = np.mean(recompensas_episodio[-20:])

pbar.set_postfix({'Recomp': f'{media_recomp:.1f}',

'Buffer': f'{len(agente.buffer)}',

'ε': f'{agente.epsilon:.3f}'})

pbar.update(20)

elif ep == 0:

pbar.update(1)

print("\n✅ Treinamento concluído!")

# ============================================

# AVALIAÇÃO

# ============================================

print("\n" + "=" * 70)

print("AVALIAÇÃO DO AGENTE")

print("=" * 70)

def avaliar_agente(agente, n_testes=20):

epsilon_original = agente.epsilon

agente.epsilon = 0

recompensas_teste = []

for _ in range(n_testes):

estado = reset_env(env)

recompensa_total = 0

terminou = False

while not terminou:

acao = agente.escolher_acao(estado)

estado, recompensa, terminou, _ = step_env(env, acao)

recompensa_total += recompensa

recompensas_teste.append(recompensa_total)

agente.epsilon = epsilon_original

return np.mean(recompensas_teste)

media_teste = avaliar_agente(agente)

print(f"\n🏆 Média de recompensa em testes: {media_teste:.1f}")

# ============================================

# GRÁFICOS

# ============================================

print("\n📊 Gerando gráficos...")

plt.figure(figsize=(14, 10))

# Gráfico 1: Evolução das recompensas

plt.subplot(2, 2, 1)

plt.plot(recompensas_episodio, 'b-', linewidth=0.5, alpha=0.5)

window = 20

media_recomp = np.convolve(recompensas_episodio, np.ones(window)/window, mode='valid')

plt.plot(media_recomp, 'r-', linewidth=2, label=f'Média {window}')

plt.xlabel('Episódio')

plt.ylabel('Recompensa')

plt.title('DQN: Evolução do Aprendizado')

plt.legend()

plt.grid(True, alpha=0.3)

plt.axhline(y=475, color='g', linestyle='--', label='Solução')

plt.legend()

# Gráfico 2: Tamanho do buffer de replay

plt.subplot(2, 2, 2)

plt.plot(tamanho_buffer_episodio, 'purple', linewidth=1.5)

plt.xlabel('Episódio')

plt.ylabel('Tamanho do Buffer')

plt.title('Experience Replay: Preenchimento do Buffer')

plt.grid(True, alpha=0.3)

plt.axhline(y=10000, color='r', linestyle='--', label='Capacidade máxima')

plt.legend()

# Gráfico 3: Perda (loss) durante treinamento

plt.subplot(2, 2, 3)

if len(agente.losses) > 0:

losses_smooth = np.convolve(agente.losses, np.ones(100)/100, mode='valid')

plt.plot(losses_smooth, 'orange', linewidth=1.5)

plt.xlabel('Passo de treino (x100)')

plt.ylabel('Perda (MSE)')

plt.title('Evolução da Perda da Rede')

plt.grid(True, alpha=0.3)

plt.yscale('log')

# Gráfico 4: Eficiência do Replay

plt.subplot(2, 2, 4)

taxa_replay = [agente.batch_size / max(b, 1) for b in tamanho_buffer_episodio]

plt.plot(taxa_replay, 'green', linewidth=1.5)

plt.xlabel('Episódio')

plt.ylabel('Taxa de amostragem')

plt.title('Frequência de Reuso das Experiências')

plt.grid(True, alpha=0.3)

plt.axhline(y=0.5, color='r', linestyle='--', label='Cada experiência usada a cada 2 treinos')

plt.legend()

plt.tight_layout()

plt.show()

# ============================================

# ANÁLISE DO IMPACTO DO REPLAY

# ============================================

print("\n" + "=" * 70)

print("EXPERIMENTO: COMPARAÇÃO COM E SEM REPLAY")

print("=" * 70)

# Treina um agente SEM replay para comparação

class DQNAgenteSemReplay:

"""DQN sem experience replay (treina a cada passo com última experiência)"""

def __init__(self, n_estados, n_acoes, gamma=0.95, epsilon=1.0,

epsilon_min=0.01, epsilon_decay=0.995, learning_rate=0.001, device='cpu'):

self.n_estados = n_estados

self.n_acoes = n_acoes

self.gamma = gamma

self.epsilon = epsilon

self.epsilon_min = epsilon_min

self.epsilon_decay = epsilon_decay

self.device = device

self.model = RedeDQN(n_estados, n_acoes).to(device)

self.target_model = RedeDQN(n_estados, n_acoes).to(device)

self.optimizer = optim.Adam(self.model.parameters(), lr=learning_rate)

self.atualizar_target()

self.ultima_transicao = None

self.losses = []

def atualizar_target(self):

self.target_model.load_state_dict(self.model.state_dict())

def lembrar(self, estado, acao, recompensa, prox_estado, terminou):

self.ultima_transicao = (estado, acao, recompensa, prox_estado, terminou)

def escolher_acao(self, estado):

if np.random.random() < self.epsilon:

return random.randrange(2)

estado_tensor = torch.FloatTensor(estado).unsqueeze(0).to(self.device)

with torch.no_grad():

q_vals = self.model(estado_tensor).cpu().numpy()[0]

return np.argmax(q_vals)

def replay(self):

if self.ultima_transicao is None:

return 0

estado, acao, recompensa, prox_estado, terminou = self.ultima_transicao

estado_tensor = torch.FloatTensor(estado).unsqueeze(0).to(self.device)

q_atual = self.model(estado_tensor).cpu().detach().numpy()[0]

if terminou:

q_atual[acao] = recompensa

else:

prox_tensor = torch.FloatTensor(prox_estado).unsqueeze(0).to(self.device)

with torch.no_grad():

q_futuro = self.target_model(prox_tensor).cpu().numpy()[0]

q_atual[acao] = recompensa + self.gamma * np.max(q_futuro)

estado_tensor = torch.FloatTensor(np.array([estado])).to(self.device)

alvo_tensor = torch.FloatTensor(np.array([q_atual])).to(self.device)

self.optimizer.zero_grad()

saidas = self.model(estado_tensor)

loss = nn.MSELoss()(saidas, alvo_tensor)

loss.backward()

self.optimizer.step()

self.losses.append(loss.item())

return loss.item()

def decair_epsilon(self):

self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)

# Treina agente sem replay

print("\n🚀 Treinando agente SEM REPLAY (para comparação)...")

agente_sem_replay = DQNAgenteSemReplay(n_estados, n_acoes, device=device)

recompensas_sem_replay = []

for ep in range(200): # Menos episódios pois tende a divergir

estado = reset_env(env)

recompensa_total = 0

terminou = False

while not terminou:

acao = agente_sem_replay.escolher_acao(estado)

prox_estado, recompensa, terminou, _ = step_env(env, acao)

agente_sem_replay.lembrar(estado, acao, recompensa, prox_estado, terminou)

agente_sem_replay.replay()

estado = prox_estado

recompensa_total += recompensa

recompensas_sem_replay.append(recompensa_total)

agente_sem_replay.decair_epsilon()

if (ep + 1) % 50 == 0:

print(f" Ep {ep+1}: Recompensa = {recompensa_total}")

# Gráfico comparativo

plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)

plt.plot(recompensas_episodio[:200], 'b-', alpha=0.5, label='Com Experience Replay')

plt.plot(recompensas_sem_replay, 'r-', alpha=0.5, label='Sem Experience Replay')

media_com = np.convolve(recompensas_episodio[:200], np.ones(20)/20, mode='valid')

media_sem = np.convolve(recompensas_sem_replay, np.ones(20)/20, mode='valid')

plt.plot(media_com, 'b-', linewidth=2)

plt.plot(media_sem, 'r-', linewidth=2)

plt.xlabel('Episódio')

plt.ylabel('Recompensa')

plt.title('Comparação: Com vs Sem Experience Replay')

plt.legend()

plt.grid(True, alpha=0.3)

plt.subplot(1, 2, 2)

plt.plot(agente.losses[:5000], 'b-', alpha=0.5, label='Com Replay')

plt.plot(agente_sem_replay.losses[:5000], 'r-', alpha=0.5, label='Sem Replay')

plt.xlabel('Passo de treino')

plt.ylabel('Perda')

plt.title('Estabilidade da Perda')

plt.legend()

plt.grid(True, alpha=0.3)

plt.yscale('log')

plt.tight_layout()

plt.show()

print("\n" + "=" * 70)

print("CONCLUSÃO DO EXPERIMENTO")

print("=" * 70)

print(f"""

✅ Experience Replay é essencial para estabilidade do DQN.

✅ O buffer atingiu {len(agente.buffer)}/{agente.buffer_tamanho_max} transições.

✅ Cada experiência foi usada em média {agente.batch_size / max(len(agente.buffer),1):.2f}x por passo.

✅ O agente com replay convergiu para {media_teste:.1f} pontos.

✅ O agente sem replay divergiu ou teve desempenho muito inferior.

""")

# ============================================

# EXPLICAÇÃO MATEMÁTICA

# ============================================

print("\n" + "=" * 70)

print("FUNDAMENTOS DO EXPERIENCE REPLAY")

print("=" * 70)

print("""

✅ O PROBLEMA DAS CORRELAÇÕES TEMPORAIS:

Experiências consecutivas (s_t, a_t, r_t, s_{t+1}) são altamente correlacionadas.

Isso viola a suposição de i.i.d. (independentes e identicamente distribuídas)

do gradiente descendente estocástico.

✅ SOLUÇÃO: EXPERIENCE REPLAY

Armazenamos transições em um buffer D = {e_1, e_2, ..., e_N}.

Amostramos mini-batch uniformemente: B ~ U(D).

Isso quebra correlações temporais.

✅ FÓRMULA DA PERDA COM REPLAY:

[latex] L(θ) = \\frac{1}{|B|} \\sum_{(s,a,r,s') \\in B} (r + \\gamma \\max_{a'} Q(s',a';θ^-) - Q(s,a;θ))^2 [/latex]

✅ HIPERPARÂMETROS DO REPLAY:

• Tamanho do buffer (N): 10.000 a 1.000.000

• Batch size (|B|): 32 a 128

• Frequência de amostragem: a cada passo

• Warmup: coletar N experiências antes de treinar

✅ VANTAGENS DO EXPERIENCE REPLAY:

1. QUEBRA CORRELAÇÕES: Amostragem aleatória remove dependência temporal

2. MAIOR EFICIÊNCIA AMOSTRAL: Cada experiência usada múltiplas vezes

3. APRENDIZADO OFFLINE: Pode revisitar experiências passadas

4. ESTABILIDADE: Reduz variância do gradiente

5. REUSO DE DADOS: Experiências caras são reaproveitadas

✅ VARIAÇÕES AVANÇADAS:

• Prioritized Experience Replay: Amostra experiências com maior erro TD

• Hindsight Experience Replay (HER): Para recompensas esparsas

• N-step Replay: Usa n passos para calcular alvo

""")

print("\n" + "=" * 70)

print("CONCLUSÃO FINAL")

print("=" * 70)

print("""

✅ Experience Replay é um dos pilares do DQN.

✅ Sem ele, o aprendizado é instável e frequentemente diverge.

✅ O buffer permite reuso eficiente de experiências caras.

✅ A amostragem aleatória quebra correlações temporais.

✅ O DQN com replay resolveu o CartPole consistentemente.

O Experience Replay é usado em praticamente todos

algoritmos modernos de RL profundo (DQN, DDQN, SAC, etc.).

""")

print("\n✅ PROGRAMA CONCLUÍDO COM SUCESSO!")

Indice

Características do Experience Replay

Hiperparâmetros e fórmulas matemáticas

Exemplo clássico: CartPole com análise do replay

Deixe um comentário Cancelar resposta