Profit Sharingを利用した迷路探索

「Profit Sharing」は、強化学習におけるアルゴリズムの1つです。このアルゴリズムは、エージェントが環境内で探索を行い、その結果得られる報酬をアクションの価値を更新するために使用します。

下記のコードには、迷路を探索するエージェントが実装されています。具体的には、ProfitSharingAgentというクラスがProfit Sharingのメカニズムを実装しています。以下にその主要な特徴を解説します：

1. 探索 vs. 活用:

   • get_actionメソッドでは、エージェントがどのアクションを選択するかを決定します。
   • 20%の確率でランダムにアクションを選択することで、探索を行います。
   • それ以外の場合、過去の経験に基づいて最も価値の高いアクションを選択します。これは活用のフェーズです。

2. 価値の更新:

   • エージェントはrememberメソッドを使用して、各ステップでの状態、アクション、報酬の組み合わせを記憶します。
   • エピソードの終わりにlearnメソッドを呼び出すことで、記憶した経験をもとに価値の更新を行います。
   • ここでは、逆順に経験を走査して、累積された報酬（G）を計算します。この累積報酬は、エージェントが未来で得られると予測される報酬の合計を表します。
   • そして、この累積報酬を使って、各アクションの価値を更新します。

3. 報酬の共有:

   • Profit Sharingの名前の通り、累積報酬はエピソード内の全てのステップで共有されます。これにより、目的地に近づくアクションだけでなく、その前のアクションも高い価値を持つようになります。

上記のコードの実装では、エージェントは迷路を探索しながらゴールを目指します。ゴールに到達するたびに、エージェントはその経験をもとに学習を行い、次回の探索でより良い経路を選択する能力を向上させます。

総じて、Profit Sharingはエージェントが環境内での経験を最大限に活用し、最適な行動を学習するための方法の1つです。

import numpy as np
import matplotlib.pyplot as plt

class Maze:
    def __init__(self, size=5):
        self.size = size
        self.start = (0, 0)  # これを前に移動
        self.goal = (size-1, size-1)  # これも前に移動
        self.maze = self._generate_maze(size)
        self.current = self.start

        # 描画のための初期設定
        self.fig, self.ax = plt.subplots(figsize=(6, 6))
        self.ax.set_xticks(np.arange(self.size))
        self.ax.set_yticks(np.arange(self.size))
        self.ax.set_xticklabels([])
        self.ax.set_yticklabels([])
        self.ax.grid(which='both')

    def _generate_maze(self, size):
        maze = np.ones((size, size))
        for i in range(size):  # 外周も含めて全てのセルに対して
            for j in range(size):
                maze[i][j] = np.random.choice([0, 1], p=[0.7, 0.3])  # 70% の確率で通路, 30% の確率で壁

        maze[self.start] = 0
        maze[self.goal] = 0
        return maze


    def reset(self):
        self.current = self.start
        return self.current

    def step(self, action):
        next_x, next_y = self.current
        if action == 0:  # 上
            next_x = max(0, self.current[0]-1)
        elif action == 1:  # 右
            next_y = min(self.size-1, self.current[1]+1)
        elif action == 2:  # 下
            next_x = min(self.size-1, self.current[0]+1)
        elif action == 3:  # 左
            next_y = max(0, self.current[1]-1)

        if self.maze[next_x][next_y] == 0:
            self.current = (next_x, next_y)

        distance_to_goal = np.sqrt((self.current[0] - self.goal[0]) ** 2 + (self.current[1] - self.goal[1]) ** 2)
        reward = 1.0 / (1.0 + distance_to_goal)  # Inverse distance as reward

        if self.current == self.goal:
            return self.current, reward, True
        return self.current, reward, False

    def render(self, agent_position):
        self.ax.clear()
        self.ax.set_xlim(0, self.size)
        self.ax.set_ylim(0, self.size)
        self.ax.set_xticks(np.arange(self.size))
        self.ax.set_yticks(np.arange(self.size))
        self.ax.set_xticklabels([])
        self.ax.set_yticklabels([])
        self.ax.grid(which='both')

        for i in range(self.size):
            for j in range(self.size):
                color = 'black' if self.maze[i][j] == 1 else 'lightgray'  # 壁のセルは黒く表示
                self.ax.add_patch(plt.Rectangle((j, i), 1, 1, color=color))

                # 開始とゴールの位置を描画
                if (i, j) == self.start:
                    self.ax.text(j + 0.5, i + 0.5, 'S', ha='center', va='center', color='green')
                elif (i, j) == self.goal:
                    self.ax.text(j + 0.5, i + 0.5, 'G', ha='center', va='center', color='red')
        
        # エージェントの位置を描画
        self.ax.text(agent_position[1] + 0.5, agent_position[0] + 0.5, 'A', ha='center', va='center', color='blue', fontsize=12)
        
        plt.pause(0.1)


class ProfitSharingAgent:
    def __init__(self, num_actions, learning_rate=0.1, discount_factor=0.9):
        self.num_actions = num_actions
        self.learning_rate = learning_rate
        self.discount_factor = discount_factor
        self.values = {}  # 状態とアクションの組み合わせの価値を保存する辞書
        self.memory = []  # アクションの履歴を保存

    def get_action(self, state):
        # 探索 vs 活用 (ここではランダムにアクションを選択します)
        if np.random.rand() < 0.2:  # 20% の確率でランダムに行動
            return np.random.choice(self.num_actions)
        values = [self.values.get((state, a), 0.0) for a in range(self.num_actions)]
        return np.argmax(values)

    def remember(self, state, action, reward):
        self.memory.append((state, action, reward))

    def learn(self):
        G = 0
        for state, action, reward in reversed(self.memory):
            G = reward + self.discount_factor * G
            value = self.values.get((state, action), 0.0)
            self.values[(state, action)] = value + self.learning_rate * (G - value)
        self.memory = [] # メモリをリセット

# 学習とテストのコード
if __name__ == "__main__":
    maze = Maze()
    agent = ProfitSharingAgent(num_actions=4)

    num_episodes = 10
    for episode in range(num_episodes):
        state = maze.reset()
        done = False
        while not done:
            action = agent.get_action(state)
            next_state, reward, done = maze.step(action)
            agent.remember(state, action, reward)
            state = next_state
            maze.render(state)  # 各ステップごとに状態を表示

        agent.learn()  # エピソード終了後に学習
        print(f"Episode {episode} finished")
        plt.close(maze.fig)