はじめに

今回は、「ChatGPTにハンズオンを作らせてみた」の第6弾で、DQNを勉強しました。

第5弾はこちら↓

DQN

状態と行動の組み合わせごとに「Q値（行動価値）」をテーブル（Qテーブル）で管理し、ディープラーニングで学習することで、大規模な状態空間に対応可能にした手法（Q学習の応用）。

使用データ

今回は、NYC Taxi Trip Dataという、ニューヨーク市タクシー＆リムジン委員会（TLC: Taxi and Limousine Commission）によって提供されている、2024年11月のタクシーの乗車に関するオープンデータセットを使用しました。各変数の説明はこちらです。

変数	説明
VendorID	データを提供したベンダーのID
tpep_pickup_datetime	乗車時間
tpep_dropoff_datetime	降車時間
passenger_count	乗客の数
trip_distance	走行距離（マイル単位）
RatecodeID	料金コード（運賃ルール）
store_and_fwd_flag	データがリアルタイムで送信されたかどうか
PULocationID	乗車地点（TLCが定めるゾーンID）
DOLocationID	降車地点（TLCが定めるゾーンID）
payment_type	支払い方法
fare_amount	運賃（メーター料金）
extra	追加料金（時間帯による追加料金など）
mta_tax	MTA（メトロポリタン交通局）税
tip_amount	チップの金額（ドル単位）
tolls_amount	通行料（トール料金）
improvement_surcharge	インフラ改善追加料金
total_amount	運賃（ドル単位）
congestion_surcharge	渋滞追加料金
Airport_fee	空港利用料金

やること

エージェント：DQNAgent
環境：ニューヨーク市内
状態：タクシーの現在地
行動：疑似的な東西南北＋待機の５パターン
報酬：total_amount（運賃）を100で割った値
方策：ε-greedy方策
価値関数：Q値

使用コード

2024年11月における、黄色タクシーの乗車データを取得します。

import pandas as pd

# データの読み込み
data_path = "yellow_tripdata_2024-11.parquet"
df = pd.read_parquet(data_path)

乱数を使うことになるので、シード値を固定します。

import torch
import random
import numpy as np

def set_seed(seed=42):
    random.seed(seed)  # Python の random モジュールのシード
    np.random.seed(seed)  # NumPy のシード
    torch.manual_seed(seed)  # PyTorch のシード
    torch.cuda.manual_seed_all(seed)  # GPU用のシード
    torch.backends.cudnn.deterministic = True  # PyTorch の畳み込み演算の再現性を確保
    torch.backends.cudnn.benchmark = False  # 学習の最適化を抑えて再現性を優先

# 乱数シードを設定
set_seed(42)

タクシー配車の強化学習環境を定義します。プライバシーの観点から緯度・経度のデータがなく、ゾーンIDのデータしかないため、ゾーンIDを±1、±10として、疑似的に東西南北を決めています。

import gym
import numpy as np
import pandas as pd
from gym import spaces

class TaxiDispatchEnv(gym.Env):
    def __init__(self, data):
        super(TaxiDispatchEnv, self).__init__()
        self.data = data
        self.time_step = 0

        # 状態: [エリアID, 時刻]
        self.state_space = spaces.MultiDiscrete([265, 24])  # NYCには265のタクシーゾーンがある
        self.action_space = spaces.Discrete(5)  # [北, 南, 東, 西, 待機]

        self.current_state = None

    def reset(self):
        # ランダムな開始地点
        self.current_state = [np.random.randint(1, 266), np.random.randint(0, 24)]
        self.time_step = 0
        return np.array(self.current_state)

    def step(self, action):
        pu_location, hour = self.current_state

        # アクションの処理 (単純にエリアIDを±1変化)
        if action == 0:  # 北
            pu_location = max(1, pu_location - 1)
        elif action == 1:  # 南
            pu_location = min(265, pu_location + 1)
        elif action == 2:  # 東
            pu_location = min(265, pu_location + 10)
        elif action == 3:  # 西
            pu_location = max(1, pu_location - 10)

        hour = (hour + 1) % 24  # 時間を進める

        # 需要予測 (そのエリア・時間の乗客数を確認)
        demand = self.data[(self.data["PULocationID"] == pu_location) & 
                           (self.data["pickup_hour"] == hour)]

        # 乗客がいるかどうかで報酬を決定
        reward = demand["total_amount"].sum() / 100 if len(demand) > 0 else -1  # 収益 or ペナルティ

        self.current_state = [pu_location, hour]
        done = (self.time_step >= 100)  # 100ステップで終了
        self.time_step += 1

        return np.array(self.current_state), reward, done, {}

# データの前処理
df["pickup_hour"] = pd.to_datetime(df["tpep_pickup_datetime"]).dt.hour

DQNAgentを実装して、学習させていきます。

import torch.nn as nn
import torch.optim as optim
from collections import deque

class DQN(nn.Module):
    def __init__(self, state_size, action_size):
        super(DQN, self).__init__()
        self.fc1 = nn.Linear(state_size, 128)
        self.fc2 = nn.Linear(128, 128)
        self.fc3 = nn.Linear(128, action_size)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        return self.fc3(x)

class DQNAgent:
    def __init__(self, state_size, action_size):
        self.state_size = state_size
        self.action_size = action_size
        self.memory = deque(maxlen=2000)
        self.gamma = 0.95  # 割引率
        self.epsilon = 1.0  # 探索率
        self.epsilon_min = 0.01
        self.epsilon_decay = 0.995
        self.learning_rate = 0.001

        self.model = DQN(state_size, action_size)
        self.optimizer = optim.Adam(self.model.parameters(), lr=self.learning_rate)
        self.criterion = nn.MSELoss()

    def act(self, state):
        if np.random.rand() <= self.epsilon:
            return random.randrange(self.action_size)
        state = torch.FloatTensor(state).unsqueeze(0)
        actions = self.model(state)
        return torch.argmax(actions).item()

    def train(self, batch_size):
        minibatch = random.sample(self.memory, batch_size)
        for state, action, reward, next_state, done in minibatch:
            target = reward
            if not done:
                target += self.gamma * torch.max(self.model(torch.FloatTensor(next_state).unsqueeze(0))).item()
            target_f = self.model(torch.FloatTensor(state).unsqueeze(0))
            target_f[0][action] = target
            self.optimizer.zero_grad()
            loss = self.criterion(target_f, self.model(torch.FloatTensor(state).unsqueeze(0)))
            loss.backward()
            self.optimizer.step()

import matplotlib.pyplot as plt

# 環境とDQNの初期化
env = TaxiDispatchEnv(df)
set_seed(42)
agent = DQNAgent(state_size=2, action_size=5)

# 報酬の記録リスト
total_rewards = []

# 学習ループ
episodes = 1000  # エピソード数
batch_size = 32  # 学習用のミニバッチサイズ

for e in range(episodes):
    state = env.reset()
    total_reward = 0  # そのエピソードの合計報酬
    for time in range(100):
        action = agent.act(state)  # 行動を選択
        next_state, reward, done, _ = env.step(action)  # 環境を進める
        agent.memory.append((state, action, reward, next_state, done))  # 経験を保存
        state = next_state  # 状態を更新
        total_reward += reward  # 報酬を加算
        if done:
            break

    total_rewards.append(total_reward)  # 報酬を記録

    if len(agent.memory) > batch_size:
        agent.train(batch_size)  # DQNの学習を実行

    if e % 100 == 0:
        print(f"Episode {e}, Total Reward: {total_reward}, Epsilon: {agent.epsilon}")

# 学習曲線をプロット
plt.figure(figsize=(10,5))
plt.plot(total_rewards, label="Total Reward per Episode", alpha=0.5)
plt.plot(pd.Series(total_rewards).rolling(100).mean(), label="Moving Average (100 episodes)", linewidth=3, color="red")
plt.xlabel("Episode")
plt.ylabel("Total Reward")
plt.title("Training Progress of DQN for Taxi Dispatch")
plt.legend()
plt.show()

分析結果

100エピソードごとに合計報酬、探索率を表示しています。合計報酬を見ていると、うまく学習しているように見えますが、探索率が常に1.0で、ずっとランダムに行動を選択しているように見えます。本来、学習が進むにつれて、探索率は減っていってほしいので、ここは改善ポイントです。

学習の進捗を可視化してみました（青：各エピソードごとの報酬、赤：100エピソードごとの移動平均）。常にランダムに行動を選択していることが影響して、ほとんど学習が進んでいないように見えます。

おわりに

今回は、DQNの手法に対する理解に時間を費やしてしまったので、今度はしっかり学習を進められるようにしていこうと思います。