Decision TransformerによるゲームAI

REM Python をシステム領域にインストール
winget install --scope machine --id Python.Python.3.12 -e --silent
REM Python のパス設定
set "PYTHON_PATH=C:\Program Files\Python312"
set "PYTHON_SCRIPTS_PATH=C:\Program Files\Python312\Scripts"
echo "%PATH%" | find /i "%PYTHON_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_PATH%" /M >nul
echo "%PATH%" | find /i "%PYTHON_SCRIPTS_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_SCRIPTS_PATH%" /M >nul

winget install --scope machine Codeium.Windsurf -e --silent

# Decision Transformerによるグリッドワールドゲームエージェント
# 特徴技術名: Decision Transformer
# 出典: Chen, L., Lu, K., Rajeswaran, A., Lee, K., Grover, A., Abbeel, P., Srinivas, A., & Mordatch, I. (2021). Decision Transformer: Reinforcement Learning via Sequence Modeling. Advances in Neural Information Processing Systems, 34, 15084-15097.
# 特徴機能: 目標リターン条件付き行動生成 - 目標とする累積報酬（リターン）を条件として与えることで、そのリターンを達成するための行動系列をTransformerで生成する。従来の価値関数ベースやポリシー勾配法とは異なり、強化学習を教師あり学習の枠組みで解決する。
# 学習済みモデル: GPT-2 (openai-community/gpt2) - OpenAIが開発した因果的言語モデル。124Mパラメータ版を使用。Transformerのデコーダーアーキテクチャを採用し、自己回帰的な系列生成が可能。URL: https://huggingface.co/openai-community/gpt2
# 方式設計:
#   - 関連利用技術: PyTorch（深層学習フレームワーク、自動微分とGPU計算）、Hugging Face Transformers（事前学習済みTransformerモデルの利用）、Matplotlib/japanize-matplotlib（結果の可視化と日本語表示）、NumPy（数値計算と配列操作）
#   - 入力と出力: 入力: なし（プログラム内でゲーム環境を生成）、出力: AIエージェントの軌跡可視化とテキスト結果
#   - 処理手順: 1) ランダムプレイによる軌跡データ収集、2) 各軌跡のリターン計算、3) Decision Transformerの学習（状態・行動・リターンの系列から次の行動を予測）、4) 学習済みモデルによる目標リターン条件付きプレイ実行
#   - 前処理、後処理: 前処理: 状態の正規化（グリッド座標を浮動小数点数に変換）、行動のone-hotエンコーディング。後処理: softmax関数による行動確率分布の生成、確率的サンプリングによる行動選択
#   - 追加処理: なし
#   - 調整を必要とする設定値: HIDDEN_DIM（モデルの表現力、デフォルト128）、MAX_SEQ_LENGTH（考慮する履歴長、デフォルト10）、LEARNING_RATE（学習率、デフォルト1e-4）
# 将来方策: HIDDEN_DIMの最適値は、収集した軌跡データの複雑さに基づいて自動調整可能。具体的には、学習損失の収束速度を監視し、収束が遅い場合は次元数を増加させる適応的調整機能を実装できる。
# その他の重要事項: グリッドサイズ5×5、障害物位置は固定、エピソード最大長20ステップ、バッチサイズ100で学習
# 前準備:
#   - pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126 を実行してください
#   - pip install transformers matplotlib numpy japanize-matplotlib を実行してください

import matplotlib.pyplot as plt
import japanize_matplotlib
import torch
import torch.nn as nn
import numpy as np
from transformers import GPT2Model, GPT2Config
import random

# 定数定義
RANDOM_SEED = 42
GRID_SIZE = 5
MAX_EPISODE_STEPS = 20
HIDDEN_DIM = 128  # モデルの表現力を決定（大きいほど複雑なパターンを学習可能だが計算量増加）
MAX_SEQ_LENGTH = 10  # 考慮する過去の履歴長（長いほど長期的な戦略を学習可能）
LEARNING_RATE = 1e-4  # 学習の速度（大きすぎると学習が不安定、小さすぎると収束が遅い）
NUM_EPOCHS = 50
NUM_EPISODES = 500
BATCH_SIZE = 100

# ゲーム環境設定
START_POS = [0, 0]  # 開始位置
GOAL_POS = [4, 4]   # ゴール位置
OBSTACLES = [(2, 2), (3, 1), (1, 3)]  # 障害物位置

# GPU/CPU自動選択
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f'デバイス: {str(device)}')


# Decision Transformerモデル定義
class DecisionTransformerGameAI(nn.Module):
    def __init__(self, state_dim=2, action_dim=4, hidden_dim=HIDDEN_DIM, max_length=MAX_SEQ_LENGTH):
        super().__init__()

        # GPT-2ベースのTransformerエンコーダ
        config = GPT2Config(
            vocab_size=50257,
            n_positions=max_length * 3,
            n_embd=hidden_dim,
            n_layer=4,
            n_head=4
        )
        self.transformer = GPT2Model(config)

        # 埋め込み層
        self.state_embed = nn.Linear(state_dim, hidden_dim)
        self.action_embed = nn.Linear(action_dim, hidden_dim)
        self.return_embed = nn.Linear(1, hidden_dim)

        # 出力層
        self.action_head = nn.Linear(hidden_dim, action_dim)

        self.max_length = max_length
        self.hidden_dim = hidden_dim

    def forward(self, states, actions, returns_to_go):
        batch_size, seq_len = states.shape[0], states.shape[1]

        # 埋め込み
        state_emb = self.state_embed(states)
        action_emb = self.action_embed(actions)
        return_emb = self.return_embed(returns_to_go.unsqueeze(-1))

        # シーケンスを交互に配置: return、state、actionの順
        sequence = torch.zeros(batch_size, seq_len * 3, self.hidden_dim).to(states.device)
        sequence[:, 0::3] = return_emb
        sequence[:, 1::3] = state_emb
        sequence[:, 2::3] = action_emb

        # Transformer処理
        transformer_out = self.transformer(inputs_embeds=sequence).last_hidden_state

        # 行動予測（return位置から次の行動を予測）
        action_preds = self.action_head(transformer_out[:, 0::3])

        return action_preds


# グリッドワールドゲーム定義
class GridWorldGame:
    def __init__(self, size=GRID_SIZE):
        self.size = size
        self.reset()

    def reset(self):
        self.agent_pos = START_POS.copy()
        self.goal_pos = GOAL_POS.copy()
        self.obstacles = OBSTACLES
        return np.array(self.agent_pos, dtype=np.float32)

    def step(self, action):
        # 行動: 0=上、1=下、2=左、3=右
        moves = [(-1, 0), (1, 0), (0, -1), (0, 1)]
        new_pos = [
            self.agent_pos[0] + moves[action][0],
            self.agent_pos[1] + moves[action][1]
        ]

        # 境界チェック
        if 0 <= new_pos[0] < self.size and 0 <= new_pos[1] < self.size:
            if tuple(new_pos) not in self.obstacles:
                self.agent_pos = new_pos

        # 報酬計算
        if self.agent_pos == self.goal_pos:
            reward = 10
            done = True
        else:
            reward = -0.1
            done = False

        return np.array(self.agent_pos, dtype=np.float32), reward, done


# メイン処理
print('Decision Transformerによるグリッドワールドゲームエージェント')
print('=' * 60)
print('概要: 強化学習を条件付きシーケンスモデリングとして解くAI')
print('ゲーム: 5×5グリッドで障害物を避けてゴールを目指す')
print('特徴: 目標リターンを指定することで行動戦略を変更可能')
print(f'デバイス: {device}')
print('=' * 60)
print('\n操作方法:')
print('- プログラムは自動実行されます')
print('- AIが学習してからゲームをプレイします')
print('- 学習曲線グラフが表示されます')
print('\n注意事項:')
print('- 学習には時間がかかる場合があります')
print('- グラフウィンドウを閉じるとプログラムが終了します')
print('=' * 60)

# 乱数シード設定
random.seed(RANDOM_SEED)
np.random.seed(RANDOM_SEED)
torch.manual_seed(RANDOM_SEED)
if torch.cuda.is_available():
    torch.cuda.manual_seed(RANDOM_SEED)
    torch.cuda.manual_seed_all(RANDOM_SEED)

# トレーニングデータ生成
print('\nトレーニングデータ生成中...')
game = GridWorldGame()
trajs = []

for episode in range(NUM_EPISODES):
    states, actions, rewards = [], [], []
    state = game.reset()

    for step in range(MAX_EPISODE_STEPS):
        action = random.randint(0, 3)
        states.append(state.copy())
        actions.append(action)

        state, reward, done = game.step(action)
        rewards.append(reward)

        if done:
            break

    # returns-to-go計算（各時点から将来の累積報酬）
    returns_to_go = []
    rtg = 0
    for i in reversed(range(len(rewards))):
        rtg += rewards[i]
        returns_to_go.insert(0, rtg)

    if len(states) > 1:
        trajs.append({
            'states': np.array(states),
            'actions': np.array(actions),
            'returns_to_go': np.array(returns_to_go)
        })

# モデル初期化と学習
print('Decision Transformer学習中...')
model = DecisionTransformerGameAI().to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=LEARNING_RATE)
criterion = nn.CrossEntropyLoss()

# 学習履歴記録用
epoch_losses = []
epoch_numbers = []

for epoch in range(NUM_EPOCHS):
    total_loss = 0
    sample_trajs = random.sample(trajs, min(BATCH_SIZE, len(trajs)))

    for traj in sample_trajs:
        if len(traj['states']) < 2:
            continue

        seq_len = min(MAX_SEQ_LENGTH, len(traj['states']) - 1)

        states = torch.FloatTensor(traj['states'][:seq_len]).unsqueeze(0).to(device)
        actions_one_hot = torch.zeros(1, seq_len, 4).to(device)
        for i, a in enumerate(traj['actions'][:seq_len]):
            actions_one_hot[0, i, a] = 1.0
        returns_to_go = torch.FloatTensor(traj['returns_to_go'][:seq_len]).unsqueeze(0).to(device)

        # 予測（次の行動を予測）
        action_preds = model(states, actions_one_hot, returns_to_go)

        # 損失計算（次の行動をターゲットとする）
        targets = torch.LongTensor(traj['actions'][1:seq_len+1]).unsqueeze(0).to(device)
        loss = criterion(action_preds.reshape(-1, 4), targets.reshape(-1))

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()

    # 平均損失を記録
    avg_loss = total_loss / len(sample_trajs)
    epoch_losses.append(avg_loss)
    epoch_numbers.append(epoch)

    if epoch % 10 == 0:
        print(f'エポック {epoch}、損失: {avg_loss:.4f}')

# 学習曲線の可視化
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))

# 損失推移グラフ
ax1.plot(epoch_numbers, epoch_losses, 'b-', linewidth=2, alpha=0.8)
ax1.set_xlabel('エポック', fontsize=12)
ax1.set_ylabel('平均損失', fontsize=12)
ax1.set_title('学習曲線 - 損失の推移', fontsize=14, pad=15)
ax1.grid(True, alpha=0.3)
ax1.set_xlim(0, NUM_EPOCHS-1)

# 移動平均を追加（視認性向上）
window_size = 5
if len(epoch_losses) >= window_size:
    moving_avg = np.convolve(epoch_losses, np.ones(window_size)/window_size, mode='valid')
    ax1.plot(epoch_numbers[window_size-1:], moving_avg, 'r-', linewidth=2, alpha=0.7, label=f'{window_size}エポック移動平均')
    ax1.legend(loc='upper right')

# 収束判定の可視化
final_loss = epoch_losses[-1]
ax1.axhline(y=final_loss, color='g', linestyle='--', alpha=0.5, label=f'最終損失: {final_loss:.4f}')
ax1.text(NUM_EPOCHS*0.7, final_loss*1.1, f'最終損失: {final_loss:.4f}', fontsize=10)

# AIプレイデモンストレーション
print('\nDecision Transformer AIプレイデモ')
game = GridWorldGame()
state = game.reset()

target_return = 8.0
states_hist = [state.copy()]
actions_hist = []
rewards_hist = []

print(f'目標リターン: {target_return}')
print(f'開始位置: {state}')
print(f'ゴール位置: {game.goal_pos}')
print(f'障害物: {game.obstacles}')

# ゲームプレイ
for step in range(15):
    # 現在の状態で予測
    seq_len = min(len(states_hist), MAX_SEQ_LENGTH)
    current_states = torch.FloatTensor(states_hist[-seq_len:]).unsqueeze(0).to(device)

    # 過去の行動（不足分は0で埋める）
    actions_one_hot = torch.zeros(1, seq_len, 4).to(device)
    past_actions_len = min(len(actions_hist), seq_len)
    if past_actions_len > 0:
        for i, a in enumerate(actions_hist[-past_actions_len:]):
            actions_one_hot[0, seq_len-past_actions_len+i, a] = 1.0

    # returns-to-go（目標リターン）
    current_returns = torch.FloatTensor([target_return] * seq_len).unsqueeze(0).to(device)

    # AIが行動を予測
    with torch.no_grad():
        action_preds = model(current_states, actions_one_hot, current_returns)
        act_probs = torch.softmax(action_preds[0, -1], dim=-1)
        action = torch.multinomial(act_probs, 1).item()

    # 行動実行
    next_state, reward, done = game.step(action)

    action_names = ['上', '下', '左', '右']
    print(f'ステップ {step+1}: {action_names[action]} → 位置{next_state}（報酬: {reward:.1f}）')

    states_hist.append(next_state.copy())
    actions_hist.append(action)
    rewards_hist.append(reward)
    target_return -= reward

    if done:
        print('ゴール到達')
        break

# ゲーム結果の可視化（右側のグラフ）
# グリッド描画
for i in range(game.size + 1):
    ax2.axhline(i, color='lightgray', linewidth=0.5)
    ax2.axvline(i, color='lightgray', linewidth=0.5)

# 障害物
for obs in game.obstacles:
    ax2.add_patch(plt.Rectangle((obs[1], obs[0]), 1, 1, color='red', alpha=0.7))

# ゴール
ax2.add_patch(plt.Rectangle((game.goal_pos[1], game.goal_pos[0]), 1, 1, color='gold', alpha=0.7))

# エージェントの軌跡
path_x = [s[1] + 0.5 for s in states_hist]
path_y = [s[0] + 0.5 for s in states_hist]

ax2.plot(path_x, path_y, 'bo-', markersize=8, linewidth=2, alpha=0.7, label='AI軌跡')
ax2.plot(path_x[0], path_y[0], 'go', markersize=12, label='開始')
ax2.plot(path_x[-1], path_y[-1], 'ro', markersize=12, label='終了')

ax2.set_xlim(0, game.size)
ax2.set_ylim(0, game.size)
ax2.set_aspect('equal')
ax2.invert_yaxis()
ax2.set_title('AIプレイ結果', fontsize=14, pad=15)
ax2.legend(loc='upper left')

# 全体タイトル
fig.suptitle('Decision Transformer 学習と実行結果', fontsize=16, y=1.02)

plt.tight_layout()
plt.show()

# 結果出力
print(f'\n総移動ステップ数: {len(states_hist)-1}ステップ')
print(f"ゴール到達: {'はい' if np.array_equal(states_hist[-1], game.goal_pos) else 'いいえ'}")
print(f'実際の累積報酬: {sum(rewards_hist):.1f}')

# 学習統計の出力
print('\n学習統計:')
print(f'初期損失: {epoch_losses[0]:.4f}')
print(f'最終損失: {epoch_losses[-1]:.4f}')
print(f'損失削減率: {(1 - epoch_losses[-1]/epoch_losses[0])*100:.1f}%')
if len(epoch_losses) >= 10:
    last_10_avg = np.mean(epoch_losses[-10:])
    print(f'最後10エポックの平均損失: {last_10_avg:.4f}')

# 最適経路の計算（マンハッタン距離）
optimal_steps = abs(START_POS[0] - GOAL_POS[0]) + abs(START_POS[1] - GOAL_POS[1])
efficiency = (optimal_steps / max(len(states_hist)-1, 1)) * 100 if len(states_hist) > 1 else 0
print(f'\n経路効率: {efficiency:.1f}% （最適{optimal_steps}ステップ / 実際{len(states_hist)-1}ステップ）')

# AIが学習した戦略データ
with torch.no_grad():
    # 異なる目標リターンでの戦略比較
    test_state = torch.FloatTensor([[2.0, 2.0]]).unsqueeze(0).to(device)  # 中央位置での戦略を確認
    dummy_action = torch.zeros(1, 1, 4).to(device)

    high_return = torch.FloatTensor([[8.0]]).to(device)
    low_return = torch.FloatTensor([[2.0]]).to(device)

    high_probs = torch.softmax(model(test_state, dummy_action, high_return)[0, -1], dim=-1).cpu()
    low_probs = torch.softmax(model(test_state, dummy_action, low_return)[0, -1], dim=-1).cpu()

print(f'\n中央位置(2,2)での戦略比較:')
print(f'高目標(8.0): 上{high_probs[0]:.2f}、下{high_probs[1]:.2f}、左{high_probs[2]:.2f}、右{high_probs[3]:.2f}')
print(f'低目標(2.0): 上{low_probs[0]:.2f}、下{low_probs[1]:.2f}、左{low_probs[2]:.2f}、右{low_probs[3]:.2f}')

# 最終位置での確信度
if len(states_hist) > 0 and len(actions_hist) > 0:
    print(f'\n最終決定の確信度: {act_probs.max():.3f}')
    print(f'ゴールまでの距離: {abs(states_hist[-1][0]-GOAL_POS[0])+abs(states_hist[-1][1]-GOAL_POS[1])}マス')

print('\nDecision Transformerの特徴:')
print('・目標リターンを条件として与えることで行動を生成')
print('・従来の強化学習と異なるTransformerベースのアプローチ')
print('・シーケンス生成により長期的な戦略を学習')

# 結果をresult.txtに保存
with open('result.txt', 'w', encoding='utf-8') as f:
    f.write('Decision Transformer実行結果\n')
    f.write('=' * 40 + '\n')
    f.write('[学習統計]\n')
    f.write(f'総エポック数: {NUM_EPOCHS}\n')
    f.write(f'使用デバイス: {str(device).upper()}\n')
    if device.type == 'cuda':
        f.write(f'GPU: {torch.cuda.get_device_name(0)}\n')
    f.write(f'初期損失: {epoch_losses[0]:.4f}\n')
    f.write(f'最終損失: {epoch_losses[-1]:.4f}\n')
    f.write(f'損失削減率: {(1 - epoch_losses[-1]/epoch_losses[0])*100:.1f}%\n')
    if len(epoch_losses) >= 10:
        f.write(f'最後10エポックの平均損失: {last_10_avg:.4f}\n')
    f.write('\n[プレイ結果]\n')
    f.write(f'総移動ステップ数: {len(states_hist)-1}ステップ\n')
    f.write(f"ゴール到達: {'はい' if np.array_equal(states_hist[-1], game.goal_pos) else 'いいえ'}\n")
    f.write(f'最終位置: {states_hist[-1]}\n')
    f.write(f'実際の累積報酬: {sum(rewards_hist):.1f}\n')
    f.write(f'経路効率: {efficiency:.1f}%\n')
    f.write('\n[戦略分析]\n')
    f.write(f'中央位置での高目標戦略: 上{high_probs[0]:.2f}、下{high_probs[1]:.2f}、左{high_probs[2]:.2f}、右{high_probs[3]:.2f}\n')
    f.write(f'中央位置での低目標戦略: 上{low_probs[0]:.2f}、下{low_probs[1]:.2f}、左{low_probs[2]:.2f}、右{low_probs[3]:.2f}\n')
    if len(states_hist) > 0 and len(actions_hist) > 0:
        f.write(f'最終決定の確信度: {act_probs.max():.3f}\n')
    f.write('=' * 40 + '\n')

print('処理結果をresult.txtに保存しました')