Python AI в StarCraft II. Часть VII: введение в глубокое обучение

Предыдущая статья — Python AI в StarCraft II. Часть VI: побеждаем сильный AI.

Всем привет и добро пожаловать в седьмую часть серии статей про использование искусственного интеллекта в игре Starcraft II. В этой и нескольких последующих частях мы будем рассматривать включение глубокого обучения в процесс решения нашей задачи. Но сначала нам надо решить, какой вид глубокого обучения мы будем использовать.

Мы считаем, что наиболее подходящей формой машинного обучения в нашем случае будут эволюционные алгоритмы глубокого обучения.

Сейчас последний писк моды — это обучение с подкреплением, в особенности Q-обучение. Но, честно говоря, мы не понимаем пока, что даст нам здесь Q-обучение, если только мы значительно не упростим задачу или не получим вдруг в свое распоряжение оборудование, превосходящее все, что мы могли бы себе позволить на данный момент.

Идея Q-обучения состоит в том, чтобы распределить вознаграждение или штраф между шагами по мере прохождения среды обучения. Мы могли бы разбить среду обучения в Starcraft на этапы, но зачем? Словом, Q-обучение здесь можно применить только в целях кликбейта!

В чем же тогда суть эволюционного алгоритма? Эволюционные алгоритмы настолько похожи на алгоритмы обучения с подкреплением, что можно даже сказать, что они являются некой формой последних.

Основная идея обучения с подкреплением — это подкрепление правильного выбора (правильного — с точки зрения конечного результата). Но тут разработчиков подстерегает ловушка. Они забывают, что должны просто подкреплять конечный результат (и ничего более!), позволяя алгоритму самостоятельно, без предвзятости человека выяснять, как лучше всего достичь этого результата. Какие еще процессы работают подобным образом? Конечно же эволюция! Какой же именно результат подкрепляется? Потомство!

Применение эволюционного подхода в случае игры Starcraft предполагает, что параметры выигравшего алгоритма сохраняются и становятся частью генофонда (обучающих данных), а параметры проигравшего забываются.

Давайте для начала подумаем о наступлении.

Хотелось бы подчеркнуть, что оптимальное решение нам неизвестно и мы будем действовать методом проб и ошибок. Может быть, Q-обучение — лучший выбор, а может — что-то еще. Мы просто пробуем алгоритмы и делимся нашим опытом и результатами с вами.

Наш код к этому моменту принял вот такой вид:

import sc2
from sc2 import run_game, maps, Race, Difficulty
from sc2.player import Bot, Computer
from sc2.constants import NEXUS, PROBE, PYLON, ASSIMILATOR, GATEWAY, \
 CYBERNETICSCORE, STALKER, STARGATE, VOIDRAY
import random


class SentdeBot(sc2.BotAI):
    def __init__(self):
        self.ITERATIONS_PER_MINUTE = 165
        self.MAX_WORKERS = 50

    async def on_step(self, iteration):
        self.iteration = iteration
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()
        await self.build_assimilators()
        await self.expand()
        await self.offensive_force_buildings()
        await self.build_offensive_force()
        await self.attack()

    async def build_workers(self):
        if (len(self.units(NEXUS)) * 16) > len(self.units(PROBE)) and len(self.units(PROBE)) < self.MAX_WORKERS:
            for nexus in self.units(NEXUS).ready.noqueue:
                if self.can_afford(PROBE):
                    await self.do(nexus.train(PROBE))


    async def build_pylons(self):
        if self.supply_left < 5 and not self.already_pending(PYLON):
            nexuses = self.units(NEXUS).ready
            if nexuses.exists:
                if self.can_afford(PYLON):
                    await self.build(PYLON, near=nexuses.first)

    async def build_assimilators(self):
        for nexus in self.units(NEXUS).ready:
            vaspenes = self.state.vespene_geyser.closer_than(15.0, nexus)
            for vaspene in vaspenes:
                if not self.can_afford(ASSIMILATOR):
                    break
                worker = self.select_build_worker(vaspene.position)
                if worker is None:
                    break
                if not self.units(ASSIMILATOR).closer_than(1.0, vaspene).exists:
                    await self.do(worker.build(ASSIMILATOR, vaspene))

    async def expand(self):
        if self.units(NEXUS).amount < (self.iteration / self.ITERATIONS_PER_MINUTE) and self.can_afford(NEXUS):
            await self.expand_now()

    async def offensive_force_buildings(self):
        #print(self.iteration / self.ITERATIONS_PER_MINUTE)
        if self.units(PYLON).ready.exists:
            pylon = self.units(PYLON).ready.random

            if self.units(GATEWAY).ready.exists and not self.units(CYBERNETICSCORE):
                if self.can_afford(CYBERNETICSCORE) and not self.already_pending(CYBERNETICSCORE):
                    await self.build(CYBERNETICSCORE, near=pylon)

            elif len(self.units(GATEWAY)) < ((self.iteration / self.ITERATIONS_PER_MINUTE)/2):
                if self.can_afford(GATEWAY) and not self.already_pending(GATEWAY):
                    await self.build(GATEWAY, near=pylon)

            if self.units(CYBERNETICSCORE).ready.exists:
                if len(self.units(STARGATE)) < ((self.iteration / self.ITERATIONS_PER_MINUTE)/2):
                    if self.can_afford(STARGATE) and not self.already_pending(STARGATE):
                        await self.build(STARGATE, near=pylon)

    async def build_offensive_force(self):
        for gw in self.units(GATEWAY).ready.noqueue:
            if not self.units(STALKER).amount > self.units(VOIDRAY).amount:
                if self.can_afford(STALKER) and self.supply_left > 0:
                    await self.do(gw.train(STALKER))

        for sg in self.units(STARGATE).ready.noqueue:
            if self.can_afford(VOIDRAY) and self.supply_left > 0:
                await self.do(sg.train(VOIDRAY))

    def find_target(self, state):
        if len(self.known_enemy_units) > 0:
            return random.choice(self.known_enemy_units)
        elif len(self.known_enemy_structures) > 0:
            return random.choice(self.known_enemy_structures)
        else:
            return self.enemy_start_locations[0]

    async def attack(self):
        # {UNIT: [n to fight, n to defend]}
        aggressive_units = {STALKER: [15, 5],
                            VOIDRAY: [8, 3]}


        for UNIT in aggressive_units:
            if self.units(UNIT).amount > aggressive_units[UNIT][0] and self.units(UNIT).amount > aggressive_units[UNIT][1]:
                for s in self.units(UNIT).idle:
                    await self.do(s.attack(self.find_target(self.state)))

            elif self.units(UNIT).amount > aggressive_units[UNIT][1]:
                if len(self.known_enemy_units) > 0:
                    for s in self.units(UNIT).idle:
                        await self.do(s.attack(random.choice(self.known_enemy_units)))


run_game(maps.get("AbyssalReefLE"), [
    Bot(Race.Protoss, SentdeBot()),
    Computer(Race.Terran, Difficulty.Hard)
    ], realtime=False)

Давайте вместо этого будем производить исключительно юниты Лучи Бездны, а затем изменим протокол наступления. Для начала изменим метод offensive_force_building:

    async def offensive_force_buildings(self):
        #print(self.iteration / self.ITERATIONS_PER_MINUTE)
        if self.units(PYLON).ready.exists:
            pylon = self.units(PYLON).ready.random

            if self.units(GATEWAY).ready.exists and not self.units(CYBERNETICSCORE):
                if self.can_afford(CYBERNETICSCORE) and not self.already_pending(CYBERNETICSCORE):
                    await self.build(CYBERNETICSCORE, near=pylon)

            elif len(self.units(GATEWAY)) < 1:
                if self.can_afford(GATEWAY) and not self.already_pending(GATEWAY):
                    await self.build(GATEWAY, near=pylon)

            if self.units(CYBERNETICSCORE).ready.exists:
                if len(self.units(STARGATE)) < (self.iteration / self.ITERATIONS_PER_MINUTE):  # this too
                    if self.can_afford(STARGATE) and not self.already_pending(STARGATE):
                        await self.build(STARGATE, near=pylon)

Мы больше не делим Звездные врата (Stargates) пополам и строим Врата (Gateaway), только если их еще нет.

Наш метод build_offensive_force примет следующий вид:

    async def build_offensive_force(self):
        for sg in self.units(STARGATE).ready.noqueue:
            if self.can_afford(VOIDRAY) and self.supply_left > 0:
                await self.do(sg.train(VOIDRAY))

Наконец, наш метод attck становится весьма простым:

    async def attack(self):
        # {UNIT: [n to fight, n to defend]}
        aggressive_units = {
                            #STALKER: [15, 5],
                            VOIDRAY: [8, 3],
                            }

        for UNIT in aggressive_units:
            for s in self.units(UNIT).idle:
                await self.do(s.attack(self.find_target(self.state)))

Здесь у нас есть выбор. Мы можем принимать решение атаковать или не атаковать вообще или относительно отдельных юнитов врага. Нам кажется, что второй вариант интересней (хотя и сложнее, конечно). Но для начала мы пока остановимся на варианте попроще и сведем все к выбору «атаковать или нет».

[machinelearning_ad_block]

Как мы будем принимать решение о целесообразности атаки? В принципе, для этого нам бы надо было построить некую логическую связку, но идея глубокого обучения как раз и состоит в том, чтобы заменить логику нейронной сетью! Поэтому вместо построения логики встает задача информирования нейронной сети. Как нам это сделать?

В этой игре у нас много переменных и количество переменных… хм… тоже переменное. У нас есть переменное количество рабочих и переменное количество воинских частей. У нас даже переменное количество ТИПОВ воинских частей, которые мы можем использовать!

Сразу приходит в голову визуализировать эти данные и передать их в сеть. И также было бы полезно увидеть размещение наших и вражеских зданий и юнитов. Поэтому нам кажется, что было бы разумно подавать на вход в нейронную сеть эти изображения, из чего следует, что мы будем использовать свёрточную нейронную сеть. Отлично, давайте приступим.

Но не так быстро! Мы не сомневаемся, что вам бы хотелось тут же создать сеть и стать повелителем Starcraft, но это не так просто. Для начала нам нужны данные. Очень большое количество данных.

Изображения, которые мы будем использовать, главным образом, привязаны к конкретным действиям. Вначале мы можем просто выполнять случайные действия, после чего сохранять успешные — это не сложно. Но нам нужны изображения, привязанные к этим действиям, ведь именно их мы хотим передавать в сеть. Поэтому для начала нужно кое-что сделать для их создания.

В принципе, это не должно быть слишком сложно: у нас есть доступ ко всей необходимой информации, нам просто нужно ее отрисовать. Например, давайте изобразим Нексусов.

Сперва нам нужно импортировать библиотеки numpy и opencv. Возможно, вам потребуется их установить. Если у вас есть проблемы с установкой библиотеки OpenCV, ознакомьтесь с этой статьей.

import cv2
import numpy as np

Затем давайте добавим к нашему методу on_step перед запуском метода attack следующий код:

   await self.intel()

Мы вставляем это перед атакой, так как метод intel должен нам сообщить способ самой атаки.

Теперь для нашего метода intel мы будем использовать:

game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)

Здесь мы меняем оси (высоту и ширину) местами, чтобы массив принял нужный нам вид.

Мы начнем с темного экрана размером с игровую карту. У разных карт разные размеры, но они примерно 200×200 пикселей.

Далее мы нарисуем на карте что-то типа Нексуса:

        for nexus in self.units(NEXUS):
            nex_pos = nexus.position
            print(nex_pos)
            cv2.circle(game_data, (int(nex_pos[0]), int(nex_pos[1])), 10, (0, 255, 0), -1)  # BGR

Массивы в библиотеке OpenCV имеют значение индексов (0. 0) в верхнем левом углу. Это нормально, но наша система координат не соответствует этим правилам. Значит, нам нужно перевернуть изображение:

        flipped = cv2.flip(game_data, 0)

Затем мы изменим масштаб изображения и визуализируем его:

        resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)

        cv2.imshow('Intel', resized)
        cv2.waitKey(1)

Полный код метода intel имеет вид:

    async def intel(self):
        # for game_info: https://github.com/Dentosal/python-sc2/blob/master/sc2/game_info.py#L162
        print(self.game_info.map_size)
        # flip around. It's y, x when you're dealing with an array.
        game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)
        for nexus in self.units(NEXUS):
            nex_pos = nexus.position
            print(nex_pos)
            cv2.circle(game_data, (int(nex_pos[0]), int(nex_pos[1])), 10, (0, 255, 0), -1)  # BGR

        # flip horizontally to make our final fix in visual representation:
        flipped = cv2.flip(game_data, 0)
        resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)

        cv2.imshow('Intel', resized)
        cv2.waitKey(1)

А весь код на данный момент выглядит следующим образом:

import sc2
from sc2 import run_game, maps, Race, Difficulty
from sc2.player import Bot, Computer
from sc2.constants import NEXUS, PROBE, PYLON, ASSIMILATOR, GATEWAY, \
 CYBERNETICSCORE, STARGATE, VOIDRAY
import random
import cv2
import numpy as np


class SentdeBot(sc2.BotAI):
    def __init__(self):
        self.ITERATIONS_PER_MINUTE = 165
        self.MAX_WORKERS = 50

    async def on_step(self, iteration):
        self.iteration = iteration
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()
        await self.build_assimilators()
        await self.expand()
        await self.offensive_force_buildings()
        await self.build_offensive_force()
        await self.intel()
        await self.attack()

    async def intel(self):
        # for game_info: https://github.com/Dentosal/python-sc2/blob/master/sc2/game_info.py#L162
        # print(self.game_info.map_size)
        # flip around. It's y, x when you're dealing with an array.
        game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)
        for nexus in self.units(NEXUS):
            nex_pos = nexus.position
            print(nex_pos)
            cv2.circle(game_data, (int(nex_pos[0]), int(nex_pos[1])), 10, (0, 255, 0), -1)  # BGR

        # flip horizontally to make our final fix in visual representation:
        flipped = cv2.flip(game_data, 0)
        resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)

        cv2.imshow('Intel', resized)
        cv2.waitKey(1)

    async def build_workers(self):
        if (len(self.units(NEXUS)) * 16) > len(self.units(PROBE)) and len(self.units(PROBE)) < self.MAX_WORKERS:
            for nexus in self.units(NEXUS).ready.noqueue:
                if self.can_afford(PROBE):
                    await self.do(nexus.train(PROBE))

    async def build_pylons(self):
        if self.supply_left < 5 and not self.already_pending(PYLON):
            nexuses = self.units(NEXUS).ready
            if nexuses.exists:
                if self.can_afford(PYLON):
                    await self.build(PYLON, near=nexuses.first)

    async def build_assimilators(self):
        for nexus in self.units(NEXUS).ready:
            vaspenes = self.state.vespene_geyser.closer_than(15.0, nexus)
            for vaspene in vaspenes:
                if not self.can_afford(ASSIMILATOR):
                    break
                worker = self.select_build_worker(vaspene.position)
                if worker is None:
                    break
                if not self.units(ASSIMILATOR).closer_than(1.0, vaspene).exists:
                    await self.do(worker.build(ASSIMILATOR, vaspene))

    async def expand(self):
        if self.units(NEXUS).amount < (self.iteration / self.ITERATIONS_PER_MINUTE) and self.can_afford(NEXUS):
            await self.expand_now()

    async def offensive_force_buildings(self):
        if self.units(PYLON).ready.exists:
            pylon = self.units(PYLON).ready.random

            if self.units(GATEWAY).ready.exists and not self.units(CYBERNETICSCORE):
                if self.can_afford(CYBERNETICSCORE) and not self.already_pending(CYBERNETICSCORE):
                    await self.build(CYBERNETICSCORE, near=pylon)

            elif len(self.units(GATEWAY)) < 1:
                if self.can_afford(GATEWAY) and not self.already_pending(GATEWAY):
                    await self.build(GATEWAY, near=pylon)

            if self.units(CYBERNETICSCORE).ready.exists:
                if len(self.units(STARGATE)) < (self.iteration / self.ITERATIONS_PER_MINUTE):  # this too
                    if self.can_afford(STARGATE) and not self.already_pending(STARGATE):
                        await self.build(STARGATE, near=pylon)

    async def build_offensive_force(self):
        for sg in self.units(STARGATE).ready.noqueue:
            if self.can_afford(VOIDRAY) and self.supply_left > 0:
                await self.do(sg.train(VOIDRAY))

    def find_target(self, state):
        if len(self.known_enemy_units) > 0:
            return random.choice(self.known_enemy_units)
        elif len(self.known_enemy_structures) > 0:
            return random.choice(self.known_enemy_structures)
        else:
            return self.enemy_start_locations[0]

    async def attack(self):
        # {UNIT: [n to fight, n to defend]}
        aggressive_units = {
                            #STALKER: [15, 5],
                            VOIDRAY: [8, 3],
                            }

        for UNIT in aggressive_units:
            for s in self.units(UNIT).idle:
                await self.do(s.attack(self.find_target(self.state)))


run_game(maps.get("AbyssalReefLE"), [
    Bot(Race.Protoss, SentdeBot()),
    Computer(Race.Terran, Difficulty.Hard)
    ], realtime=False)

В результате его работы вы должны видеть что-то типа этого:

Следующая статья — Python AI в StarCraft II. Часть VIII: разведка и другие визуальные материалы.