EKI-ReinforcementLearning-TicTacToe.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "42e167b9-ec16-4222-bee9-ea7be2d44a03",
   "metadata": {},
   "source": [
    "# Reinforcement Learning am Beispiel von Tic Tac Toe #"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5124cf10-f1d3-406a-9753-cf9c8b43a355",
   "metadata": {
    "tags": []
   },
   "source": [
    "https://www.python-lernen.de/tic-tac-toe-kontrolle-spielzug.htm enthält eine Implementierung mit einer Zufallsstrategie für die KI auf folgender Unterseite:<br>\n",
    "https://www.python-lernen.de/tic-tac-toe-ki.htm\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2091ace8-3f31-4584-a279-35a876ebb269",
   "metadata": {},
   "source": [
    "## Import von Bibliotheken ##"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "db1bad32-8a7d-4927-8bce-e4e839553fe9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from collections import defaultdict\n",
    "import graphviz"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "4af108af-7598-4a2a-8ca7-238daf1a2ed2",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "class TreeNode:\n",
    "    def __init__(self, state, player=None, parent=None):\n",
    "        self.state = state\n",
    "        self.player = player\n",
    "        self.parent = parent  # Referenz auf den Elternknoten\n",
    "        self.children = {}\n",
    "        self.q_values = {}  # Q-Werte für Aktionen in diesem Zustand\n",
    "\n",
    "    def add_child(self, action_index, next_state):\n",
    "        if action_index not in self.children:\n",
    "            self.children[action_index] = next_state\n",
    "\n",
    "    def get_child(self, action_index):\n",
    "        return self.children.get(action_index)\n",
    "    \n",
    "    def get_child_action(self, child):\n",
    "        for action, state in self.children.items():\n",
    "            if state == child:\n",
    "                return action\n",
    "        return None  # gib None zurück, falls child nicht existiert\n",
    "\n",
    "\n",
    "    def set_q_value(self, action_index, q_value):\n",
    "        self.q_values[action_index] = q_value\n",
    "\n",
    "    def get_q_value(self, action_index):\n",
    "        return self.q_values.get(action_index)\n",
    "    \n",
    "    def get_q_value_action(self, max_q_value):\n",
    "        for action, q_value in self.q_values.items():\n",
    "            if q_value == max_q_value:\n",
    "                return action\n",
    "        return None  # gib None zurück, falls child nicht existiert\n",
    "    \n",
    "    def visualize_tree(self, filename='tree'):\n",
    "        dot = graphviz.Digraph(comment='Game Tree')\n",
    "        self._add_nodes_edges(dot)\n",
    "        dot.render(filename, format='png', cleanup=True)\n",
    "\n",
    "    def _add_nodes_edges(self, dot):\n",
    "        dot.node(str(id(self)), str(self.state))\n",
    "        queue = [self]\n",
    "        while queue:\n",
    "            node = queue.pop(0)\n",
    "            for action, child in node.children.items():\n",
    "                dot.node(str(id(child)), str(child.state))\n",
    "                color = \"blue\" if child.player == 1 else \"green\"\n",
    "                dot.edge(str(id(node)), str(id(child)), label=(str(action)+\" | \"+str(node.get_q_value(action))), color=color)\n",
    "                queue.append(child)\n",
    "                \n",
    "    def to_graphviz(self) -> graphviz.Digraph:\n",
    "        \"\"\"\n",
    "        Transform the current fst into a graphviz graph\n",
    "\n",
    "        Returns\n",
    "        -------\n",
    "        graph :  graphviz.DiGraph\n",
    "            A graphviz DiGraph representing the fst\n",
    "\n",
    "        \"\"\"\n",
    "        graph = graphviz.Digraph(format='svg')\n",
    "        graph.node(str(id(self)), str(self.state))\n",
    "        queue = [self]\n",
    "        while queue:\n",
    "            node = queue.pop(0)\n",
    "            for action, child in node.children.items():\n",
    "                graph.node(str(id(child)), str(child.state))\n",
    "                color = \"blue\" if child.player == 1 else \"green\"\n",
    "                graph.edge(str(id(node)), str(id(child)), label=(str(action)+\" | \"+str(node.get_q_value(action))), color=color)\n",
    "                queue.append(child)       \n",
    "        return graph\n",
    "     \n",
    "\n",
    "# Hauptfunktion zum Spielen von Tic-Tac-Toe und Training des Q-Baums\n",
    "def play_tic_tac_toe(root_node, num_episodes=1000, epsilon=0.1, alpha=0.5, gamma=0.99):\n",
    "    for _ in range(num_episodes):\n",
    "        state = root_node\n",
    "        player = 1\n",
    "        for _ in range(9):  # Maximal 9 Züge im Tic-Tac-Toe\n",
    "            # Choose action based on epsilon-greedy policy\n",
    "            action = choose_action(state, epsilon, player)\n",
    "            # Update the game state for player X=1\n",
    "            next_state, done = update_state(state, action, player, alpha, gamma)\n",
    "            # If the game is finished, break the loop\n",
    "            if done:\n",
    "                break\n",
    "            # If the game is not finished, switch to the opponent's turn (player O=-1)\n",
    "            state = next_state\n",
    "            player = -1 * player\n",
    "            \n",
    "            \n",
    "def choose_action(state, epsilon, player):\n",
    "    if np.random.uniform(0, 1) < epsilon or not state.q_values:\n",
    "#        return random_strategy(state, player)\n",
    "        return heuristic_strategy(state, player)\n",
    "    else:\n",
    "        current_max_q_value = max([player * v for v in state.q_values.values()])\n",
    "        if current_max_q_value <= 0:\n",
    "            return random_strategy_unique(state, player) \n",
    "#            return heuristic_strategy(state, player)\n",
    "        else:\n",
    "            return state.get_q_value_action(player * current_max_q_value)\n",
    "    \n",
    "def random_strategy(state, player):\n",
    "    return np.random.choice([i for i, val in enumerate(state.state) if val is None])\n",
    "\n",
    "def random_strategy_unique(state, player):\n",
    "    return np.random.choice([i for i, val in enumerate(state.state) if val is None and i is not state.q_values.keys()])\n",
    "\n",
    "\n",
    "def heuristic_strategy(state, player):\n",
    "    directions = np.array([\n",
    "        [0, 1, 2], [3, 4, 5], [6, 7, 8],  # Horizontal\n",
    "        [0, 3, 6], [1, 4, 7], [2, 5, 8],  # Vertikal\n",
    "        [0, 4, 8], [2, 4, 6]              # Diagonal\n",
    "    ])\n",
    "    action_result = defaultdict(list)\n",
    "    for i, val in enumerate(state.state):\n",
    "        if val is None:\n",
    "            prio = 0\n",
    "            relevant = directions[np.where(np.any(directions == i, axis=1))]\n",
    "            relevant = np.reshape(relevant[relevant != i],(-1, 2))\n",
    "            for entry in np.take(state.state, relevant):\n",
    "                if np.all(entry != None):\n",
    "                    # print(\"Entry not None: \" + str(entry))\n",
    "                    # print(\"Player: \" + str(player))\n",
    "                    val = sum(entry) * player\n",
    "                    if val == 2:\n",
    "                        return i\n",
    "                    elif val == -2:\n",
    "                        prio = 9\n",
    "                elif np.all(entry is None):\n",
    "                    prio += 1\n",
    "                elif np.any(entry == player):\n",
    "                    prio += 3\n",
    "                else:\n",
    "                    prio += 1\n",
    "            action_result[prio].append(i)\n",
    "    return np.random.choice(action_result[max(action_result.keys())])\n",
    "                        \n",
    "                \n",
    "    \n",
    "#def q_value_strategy(state, player):\n",
    "#    current_max_q_value = max([player * v for v in state.q_values.values()])\n",
    "#    if current_max_q_value < 0:\n",
    "#        random_strategy(state, player)\n",
    "#    else:\n",
    "#        return state.get_q_value_action(player * current_max_q_value) \n",
    "\n",
    "def update_state(state, action, player, alpha, gamma):\n",
    "    next_state = state.get_child(action)  # Überprüfen, ob ein Kindknoten für diese Aktion existiert\n",
    "    if next_state is None:\n",
    "        next_state = TreeNode(state.state.copy(), player, state)  # Kopieren des aktuellen Zustands\n",
    "        # print(\"Action: \"+str(action))\n",
    "        next_state.state[action] = player\n",
    "        # print(\"next_state: \" + str(next_state.state))\n",
    "        state.add_child(action, next_state)  # Neuen Zustand als Kindknoten hinzufügen\n",
    "        done = check_winner(next_state.state, player)\n",
    "        if done:\n",
    "            state.set_q_value(action, player)\n",
    "            # print(\"Q_Values: \" + str(state.q_values))\n",
    "            new_q_value = max([player * v for v in state.q_values.values()])\n",
    "            # print(\"New_Q_Value: \" + str(new_q_value))\n",
    "            update_q_values(state.parent, state, new_q_value * player, -1*player, alpha, gamma)\n",
    "        else:\n",
    "            state.set_q_value(action, 0)  # Eintragen des Rewards in die Q-Werte\n",
    "    else:\n",
    "        reward = state.get_q_value(action)\n",
    "        done = ((player * reward) == 1)\n",
    "\n",
    "    return next_state, done\n",
    "\n",
    "\n",
    "def check_winner(state, player):\n",
    "    win_patterns = [\n",
    "        [0, 1, 2], [3, 4, 5], [6, 7, 8],  # Horizontal\n",
    "        [0, 3, 6], [1, 4, 7], [2, 5, 8],  # Vertikal\n",
    "        [0, 4, 8], [2, 4, 6]               # Diagonal\n",
    "    ]\n",
    "    for pattern in win_patterns:\n",
    "        if all(state[pos] == player for pos in pattern):\n",
    "            return True\n",
    "    return False\n",
    "\n",
    "def update_q_values(state, child, next_max_q_value, player, alpha, gamma):\n",
    "    if not state is None:\n",
    "        action = state.get_child_action(child)  # Aktion finden, die zum aktuellen Zustand geführt hat\n",
    "        # print(\"Action: \"+ str(action))\n",
    "        old_q_value = state.get_q_value(action)\n",
    "        if old_q_value != next_max_q_value:\n",
    "            state.set_q_value(action, old_q_value + alpha * (gamma * next_max_q_value - old_q_value))\n",
    "            new_q_value = max([player * v for v in state.q_values.values()])\n",
    "            # print(\"New_Q_Value: \"+ str(new_q_value))\n",
    "            update_q_values(state.parent, state, new_q_value * player, -1*player, alpha, gamma)\n",
    "    \n",
    "if __name__ == \"__main__\":\n",
    "    root_node = TreeNode([None] * 9)\n",
    "    play_tic_tac_toe(root_node, 50)\n",
    "    root_node.visualize_tree('tic_tac_toe_tree')\n",
    "    # tree = root_node.to_graphviz()\n",
    "    \n",
    "# tree\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2db13275-4d03-4cda-ab30-c63debe472f1",
   "metadata": {},
   "source": [
    "## Verfeinerung der Implementierung - Übergabe von Spielstrategien als Parameter ##\n",
    "\n",
    "Zunächst ist es wichtig, dass verschiedene Spielstrategien als Argument der Hauptfunktion übergeben werden. Dazu gehört - ähnlich der Othello-Implementierung in Lisp - die Umsetzung einer Humen-Strategie, um tatsächlich das Spiel gegen einen menschlichen Spieler zu spielen. Andy Schleising hat in seiner Bachelorarbeit eine entsprechende Implementierung vorgenommen, die es hier zu integrieren gilt."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "891f1e24-cda2-41f7-bfe6-0f8e2fbaf804",
   "metadata": {},
   "source": [
    "This decorator is useful for optimizing the performance of recursive or expensive functions, as it caches the results of previous calls and returns them if the same arguments are passed again. It also uses the functools.wraps decorator to preserve the name and docstring of the original function. Here is the code:"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8d3794c7-9e74-4b90-bd31-6b1d505cdb3e",
   "metadata": {},
   "source": [
    "## Verfeinerung der Implementierung - Vorhalten einer Hashtable für alle bereits erkundeten Zustände ##\n",
    "\n",
    "Bei verschiedenen kombinatorischen Spielen inkl. Tic Tac Toe kann es zu gleichen Spielzuständen (Game-State) kommen, obwohl die Reihenfolge der Spielzüge (Aktionen) unterschiedlich war (Play-State). Deshalb wäre es wichtig, immer wenn im Spielbaum ein neuer Spielzustand untersucht wird, ersteinmal zu schauen, ob man diesen Spielzustand bereits untersucht hat. Hierfür eignet sich das Konzept des Memoization. Für Python gibt es mittlerweile dieses Konzept als Decorator (vgl. https://medium.com/@ayush-thakur02/python-decorators-that-can-reduce-your-code-by-half-b19f673bc7d8). Decoratoren gibt es auch für andere Konzepte, die sich nachträglich in den Programmcode einklinken. Insgesamt werden die Decoratoren: @measure_time, @wrapper, @timer, @debug und @memoize vorgestellt sowie auf die Decoratoren @wraps, @staticmethod, @classmethod, @property, @functools.lru_cache, @functools.singledispatch verwiesen. Hier interessiert zunächst aber nur das Memoization. Deshalb folgt auf dem genannten Artikel der entsprechende Pythonauszug. Jedoch muss man für die Implementierung im vorliegenden Reinforcement-Szenario aufpassen, weil nur die Berechnung des Funktionswertes optimiert wird, d.h. nur bei gleichen Argumenten spart man sich die Berechnung."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "d43d9d3e-b5d0-4186-88f3-cb18232ba566",
   "metadata": {},
   "outputs": [],
   "source": [
    "from functools import wraps\n",
    "\n",
    "def memoize(func):\n",
    "    cache = {}\n",
    "    @wraps(func)\n",
    "    def wrapper(*args):\n",
    "        if args in cache:\n",
    "            return cache[args]\n",
    "        else:\n",
    "            result = func(*args)\n",
    "            cache[args] = result\n",
    "            return result\n",
    "    return wrapper"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6aa792bf-9c0f-4aaf-a9a8-1a3ded7c882f",
   "metadata": {},
   "source": [
    "Now, you can use this decorator to memoize any function, such as:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "10dd0cbd-8e98-47a7-bfd7-068f3c05eae9",
   "metadata": {},
   "outputs": [],
   "source": [
    "@memoize\n",
    "def factorial(n):\n",
    "    \"\"\"Returns the factorial of n\"\"\"\n",
    "    if n == 0 or n == 1:\n",
    "        return 1\n",
    "    else:\n",
    "        return n * factorial(n - 1)\n",
    "@memoize\n",
    "def fibonacci(n):\n",
    "    \"\"\"Returns the nth Fibonacci number\"\"\"\n",
    "    if n == 0 or n == 1:\n",
    "        return n\n",
    "    else:\n",
    "        return fibonacci(n - 1) + fibonacci(n - 2)\n",
    "print(factorial(10))\n",
    "print(fibonacci(10))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "43f68322-2915-4403-8a27-76f31ac28283",
   "metadata": {},
   "source": [
    "This would output the same as before, but with much faster execution time, as the results are cached and reused."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b1ec7be9-958b-4c0e-b039-ab90d6ebd931",
   "metadata": {},
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}