リバーシで試す「強化学習」のはじめ

テーマ

みなさん、お元気でしょうか？

grasys でデータサイエンティストをしております t.watanabe です

今回は機械学習の中でも強化学習（Reinforcement Learning）についてリバーシを題材に学んでいこうと思います

「強化学習」とはなにか？

では、強化学習とは何かというところから説明していこうと思います

機械学習のアプローチには様々な分類がありますが、「学習の手法（データの与え方や目的）」を基準にすると、主に以下の3つ（時に4つ）に大別されます。

• 教師あり学習
• 教師なし学習
• 強化学習
• （準教師あり学習、今回は本題からそれるので説明を省きます）

この 3つ（ないしは4つ）に分ける書籍も多いのではないのでしょうか

具体的には

教師あり学習

一般的に入力と出力のペアを用意しそのペアを当てられるようにモデルを学習させていきます

メリットとして、

• 入力と出力の対応する特徴を捉えたモデルができ、それにより未知のデータにも対応することができます
• 入力と出力のペアを調整することにより意図したモデルを作成することができます

デメリットとして、

• データセットとして入力と出力のペアがある一定量、必要になります
• データに不適切な箇所（間違えたペアなど）があるとそれも学習してしまいます]

代表的な手法として

• RandomForest
• XGBoost
• LightGBM
• SVM（サポートベクターマシン）

などがあります

教師なし学習

一般的に入力からその傾向を見つけていくモデルを学習させていきます

メリットとして、

• 入力と出力の対応するペアがいらないです
• 入力の突出した特徴を捉えることができます
  • 入力が非常に多い項目のあるデータなどから特徴を見つけ出し項目数を減らすことができます（次元圧縮）

デメリットとして、

• 似ているデータの集合（クラスタ）はモデルとデータによって変わります
• 入力と出力のペアはわからないのでペアを作りたい場合は教師あり学習と組み合わせる必要があります

代表的な手法として

• AutoEncoder
• PCA（主成分分析）
• t-SNE

などがあります

強化学習

一般的に入力などの一定の状況から試行を繰り返しそこから得られたあらかじめ設定した正解、不正解のルールから良い解決方法をみつけだす、手法です

一定の環境下で試行するたびにモデルを環境へ順応（強化）していく手法となります

環境から得られる報酬に応じて、より報酬が多くもらえるよう学習していきます

最近ですと人間によるフィードバックを報酬とする RLHF（Reinforcement Learning from Human Feedback）などが LLM の学習で使われている場面をみます

メリットとして、

• 入出力のペアがなくても学習ができます
• 出力の表現が難しい場面でモデルが良い解決方法をみつけだします

デメリットとして、

• 環境（入力となるデータ）とルール（報酬）の設定が必要になります
• 試行回数が計算コストとして多くなることがあります

代表的な手法として

• Q学習
• TD学習
• SARSA

などがあります

リバーシで「強化学習」をしていきましょう！

今回は強化学習にスポットをあて、リバーシを解いていこうと思います

前章で「環境（入力となるデータ）とルール（報酬）の設定が必要になります」という話をしました

今回はルールはすでに決まっている問題としてリバーシを題材に強化学習のなかでも「Q学習（Deep-Q-Learning）」を実装していこうと思います

今回は少しアレンジも入っていますが、流れとしては以下の通りです

1. ランダムに行動する黒と白の石をおくエージェントを2つ用意します
   • それぞれのエージェントは「盤に石を置くモデル」と「場面を評価するモデル」から構成されます
2. エージェントは盤上のマスのおける確率に従って高い場所に石を置きます
   • このとき、ノイズを少し入れるとうまく学習できます
   • また、最初はおける確率がわからないのでランダムに配置を提案します
3. おける箇所であれば次の盤の状態として置く前の盤のラベルとして「盤に石を置くモデル」に学習させます
4. 「場面を評価するモデル」には石を置く前の場面のスコアが石を置いた後のスコアに近づくように学習させます
   • この際に置いた後のスコアを弱めるように数値をかけます（割引率といいます）
5. 2 ~ 4 を繰り返し、勝利した場合にはプラスの報酬を敗北した場合にはマイナスの報酬を与えます
6. 5の報酬に従って「場面を評価するモデル」と「盤に石を置くモデル」を学習させます
   • 「盤に石を置くモデル」はスコアを出さず、場面における確率を出すので学習の際には 「盤に石を置くモデル」-> 「場面を評価するモデル」 と連結した状態で学習を行います
   • 最後に即時報酬を与えることで 4 の未来の石を置いた場面から過去の石を置く前の場面の有利不利を評価するモデルが対戦途中の盤面を評価できるようになります
7. 2 ~ 6 を一定繰り返すことによりリバーシのルールのみから良い手を打つエージェントが生まれます

実装をしていきましょう

今回はリバーシを行うサーバと bot を黒と白を担当するクライアントに分けて実装します

サーバは C++ で HTTP で通信を行うため httplib を使います、高速に動作するため C++ で実装していきます

bot は pytorch を使いたいため python で実装していきます

それぞれ、マルチスレッドの排他処理など入っていますが原理は上記の通りです

サーバ

// board.hpp
#ifndef         __OTHELLO__BOARD__DEFINES__
#define         __OTHELLO__BOARD__DEFINES__     ( 1 )


#include <string>


typedef enum tagOthelloStone
{
        NONE,
        WHITE,
        BLACK
} OthelloStone;


typedef enum tagOthelloWinner
{
        WIN_NONE,
        WIN_WHITE,
        WIN_BLACK,
        WIN_EVEN
} OthelloWinner;


class OthelloBoard
{
private:
        int                     height;
        int                     width;
        OthelloStone**          board;
        
        int                     _reverse( int x, int y, int dx, int dy, OthelloStone c, bool exec );
        int                     _reverse_line( int x, int y, int dx, int dy, OthelloStone c );

public:
        int                     put( int x, int y, OthelloStone c );
        bool                    lose( OthelloStone c );
        int                     count( OthelloStone c );
        OthelloWinner           winner( void );
        std::string             draw( void );
        std::string             draw_bits( OthelloStone c );
        int                     reset( void );

        OthelloBoard( int width, int height );
        ~OthelloBoard();
};

#endif  //      __OTHELLO__BOARD__DEFINES__

// board.cpp
#include "board.hpp"
#include <sstream>
#include <iomanip>
int OthelloBoard::_reverse_line( int x, int y, int dx, int dy, OthelloStone c )
{
        int     f = 0;
        f = this->_reverse( x, y, dx, dy, c, false ); 
        if( f > 0 )
                this->_reverse( x, y, dx, dy, c, true ); 
        return f;
}
int OthelloBoard::_reverse( int x, int y, int dx, int dy, OthelloStone c, bool exec )
{
        int     r = 0;
        while( 1 )
        {
                int     b;
                x += dx;
                y += dy;
                if( x < 0 || this->width <= x )
                        return 0;
                if( y < 0 || this->height <= y )
                        return 0;                
                b = this->board[ y ][ x ];
                if( b == NONE )
                        return 0;
                if( b == c && r == 0 )
                        return 0;
                if( b == c && r > 0 )
                        return r;
                if( exec )
                        this->board[ y ][ x ] = c;
                r ++;
        }
        return 0;
}
int OthelloBoard::put( int x, int y, OthelloStone c )
{
        int     r = 0;
        if( x < 0 || this->width <= x )
                return 0;
        if( y < 0 || this->height <= y )
                return 0;
        if( this->board[ y ][ x ] != NONE )
                return 0;
        r += this->_reverse_line( x, y, -1, 0, c ); 
        r += this->_reverse_line( x, y, +1, 0, c ); 
        r += this->_reverse_line( x, y, 0, +1, c ); 
        r += this->_reverse_line( x, y, 0, -1, c ); 
        r += this->_reverse_line( x, y, -1, -1, c ); 
        r += this->_reverse_line( x, y, +1, -1, c ); 
        r += this->_reverse_line( x, y, -1, +1, c ); 
        r += this->_reverse_line( x, y, +1, +1, c ); 
       
        if( r > 0 )
                this->board[ y ][ x ] = c;
        return r;
}
bool OthelloBoard::lose( OthelloStone c )
{
        for( int j = 0; j < this->height; j ++ )
        {
                for( int i = 0; i < this->width; i ++ )
                {
                        if( this->board[ j ][ i ] == NONE )
                        {
                                // continue
                                if( this->_reverse( i, j, -1, 0, c, false ) ) return false;
                                if( this->_reverse( i, j, +1, 0, c, false ) ) return false;
                                if( this->_reverse( i, j, 0, +1, c, false ) ) return false;
                                if( this->_reverse( i, j, 0, -1, c, false ) ) return false;
                                if( this->_reverse( i, j, -1, -1, c, false ) ) return false;
                                if( this->_reverse( i, j, +1, -1, c, false ) ) return false;
                                if( this->_reverse( i, j, -1, +1, c, false ) ) return false;
                                if( this->_reverse( i, j, +1, +1, c, false ) ) return false;
                        }
                }    
        }
        return true;  // lose
}
int OthelloBoard::count( OthelloStone c ) 
{
        int     a = 0;
        for( int j = 0; j < this->height; j ++ )
        {
                for( int i = 0; i < this->width; i ++ )
                    a += ( c == this->board[ j ][ i ] );
        }
        return a;
}
OthelloWinner OthelloBoard::winner( void )
{
        bool    bl = this->lose( BLACK );
        bool    wl = this->lose( WHITE );
        if( bl && wl )
        {
                int     bc = this->count( BLACK );
                int     wc = this->count( WHITE );
                if( wc < bc )
                        return WIN_BLACK;
                if( bc < wc )
                        return WIN_WHITE;
                return WIN_EVEN;
        }
        if( bl )
                return WIN_WHITE;
        if( wl )
                return WIN_BLACK;
        return WIN_NONE; 
}
std::string OthelloBoard::draw_bits( OthelloStone c )
{
        std::string     bits = "";
        for( int j = 0; j < this->height; j ++ )
        {
                for( int i = 0; i < this->width; i ++ )
                {
                        char    b = ' ';
                        switch( this->board[ j ][ i ] )
                        {
                                case BLACK:
                                        b = ( c == BLACK )? '+' : '-' ;
                                        break;
                                case WHITE:
                                        b = ( c == WHITE )? '+' : '-' ;
                                        break;
                                default:
                                        break;
                        }
                        bits.push_back( b );
                }
                bits.push_back( '\n' );
        }
        return bits;
}
std::string OthelloBoard::draw( void )
{
        std::stringstream       s;
        std::string             circles[] = { "⬤", "⃝ " };
        // std::string             circles[] = { "⬤", "⃝◯" };
        // 最初
        s << ( "    " );
        for( int i = 0; i < this->width; i ++ )
                s << std::setw( 4 ) <<( i );
        s << std::endl;
        s << ( "    ┏" );
        for( int i = 0; i < this->width; i ++ )
        {
                s << ( "━━━" ); 
                if( i < this->width - 1 )
                        s << ( "┳" ); 
        }
        s <<( "┓\n" );
        // 中間
        for( int j = 0; j < this->height; j ++ )
        {
                s << std::setw( 4 ) << ( j );
                for( int i = 0; i < this->width; i ++ )
                {
                        s << ( "┃" );
                        switch( this->board[ j ][ i ] )
                        {
                                case BLACK:
                                        s << " " << circles[ 0 ] << " ";
                                        break;
                                case WHITE:
                                        s << " " << circles[ 1 ] << " ";
                                        break;
                                default:
                                        s << "   ";
                        }                    
                }
                s << ( "┃" );
                s << std::endl;
                s << ( " " );
                if( j < this->height - 1 )
                {
                        s << ( "   " );
                        s << ( "┣" );
                        for( int i = 0; i < this->width; i ++ )
                        {
                                s << ( "━━━" ); 
                                if( i < this->width - 1 )
                                        s << ( "╋" ); 
                        }
                        s << ( "┫" );
                        s << std::endl;
                }
        }
 
        // 最終
        s << ( "   ┗" );
        for( int i = 0; i < this->width; i ++ )
        {
                s << ( "━━━" ); 
                if( i < this->width - 1 )
                        s << ( "┻" ); 
        }
        s << ( "┛" );
        s << std::endl;
        
        return s.str();
}
int OthelloBoard::reset( void )
{
        for( int i = 0; i < this->height; i ++ )
        {
                this->board[ i ] = new OthelloStone[ this->width ];
                for( int j = 0; j < this->width; j ++ )
                        this->board[ i ][ j ] = NONE;
        }
        if( 1 )
        {
                int             center_x = ( width - 1 ) / 2;
                int             center_y = ( height - 1 ) / 2;
                this->board[ center_x + 0 ][ center_y + 1 ] = BLACK;
                this->board[ center_x + 1 ][ center_y + 0 ] = BLACK;
                this->board[ center_x + 0 ][ center_y + 0 ] = WHITE;
                this->board[ center_x + 1 ][ center_y + 1 ] = WHITE;
        }
        return 0;
}
OthelloBoard::OthelloBoard( int width, int height )
{
        this->width = width;
        this->height = height;
        this->board = new OthelloStone*[ this->height ];
        this->reset();
}
OthelloBoard::~OthelloBoard()
{
        for( int i = 0; i < this->height; i ++ )
                delete[] this->board[ i ];
        delete[] this->board;
}

// main.cpp
#include "board.hpp"
#include <iostream>
#include "httplib.h"
#include <sstream>
#include <string>
#include <cstdlib>
#include <mutex>
int first_number( std::string s )
{
        int     n = 0;
        for( size_t i = 0; i < s.length(); i ++ )
        {
                if( std::isdigit( s[ i ] ) )
                        n += n * 10 + ( s[ i ] - '0' );
                else
                        return n;
        }
        return n;
}
int main( int argc, char* argv[] )
{
        OthelloBoard*           othello = new OthelloBoard( 8, 8 );
        std::mutex              mtx;
        int                     turn = 0;
        OthelloStone            colors[] = { BLACK, WHITE };
        std::string             circles[] = { "⬤", "⃝ " };
        httplib::Server         httpserver;
        httpserver.Get( "/now/color", [&]( const httplib::Request& req, httplib::Response& res )
        {                
                res.set_content( std::string( ( colors[ turn % 2 ] == BLACK )? ( "BLACK" ) : ( "WHITE" ) ), "text/plain" );
        });
        httpserver.Get( "/board", [&]( const httplib::Request& req, httplib::Response& res )
        {
                std::stringstream               s;
                OthelloStone                    c = colors[ turn % 2 ];
                s << "turn: " << turn + 1 << ", color: " << ( ( c == WHITE )? ( circles[ 1 ] ) : ( circles[ 0 ] ) ) << std::endl;
                s << othello->draw() << std::endl;
                        
                std::cout << s.str();
                res.set_content( s.str(), "text/plain" );
        });
        httpserver.Get( "/board/bits", [&]( const httplib::Request& req, httplib::Response& res )
        {
                OthelloStone                    c = NONE;
                std::string                     name;
                
                if( !req.has_param( "c" ) )
                {
                        res.set_content( "", "text/plain" );
                        return ;
                }
                
                name = req.get_param_value( "c" );
                if( name == "BLACK" )
                       c = BLACK;
                if( name == "WHITE" )
                       c = WHITE;
                std::cout << othello->draw() << std::endl;
                        
                res.set_content( othello->draw_bits( c ), "text/plain" );
        });
        httpserver.Get( "/winner", [&]( const httplib::Request& req, httplib::Response& res )
        {
                OthelloWinner                   winner = othello->winner();
                 
                if( winner == WIN_EVEN ) 
                        res.set_content( "EVEN", "text/plain" );
                if( winner == WIN_BLACK ) 
                        res.set_content( "BLACK WIN", "text/plain" );
                if( winner == WIN_WHITE ) 
                        res.set_content( "WHITE WIN", "text/plain" );
                if( winner == WIN_NONE ) 
                {
                        res.set_content( "", "text/plain" );
                        return ;
                }
                mtx.lock();
                turn = 0;
                mtx.unlock();
        });
        httpserver.Put( "/stone", [&]( const httplib::Request& req, httplib::Response& res )
        {
                OthelloStone                    c = colors[ turn % 2 ];
                int                             x;
                int                             y;
                int                             z = -1;
                std::string                     body( req.body );
                int                             xp;
                int                             yp;
                z = ( body.find( "c=BLACK" ) != std::string::npos )? 0 : z ;
                z = ( body.find( "c=WHITE" ) != std::string::npos )? 1 : z ;
                if( 0 <= z && z <= 1 && c != colors[ z ] )
                {
                        res.set_content( "Invalid color", "text/plain" );
                        return ;
                }
                xp = body.find( "x=" );
                yp = body.find( "y=" );
                if( xp == std::string::npos || yp == std::string::npos )
                {       
                        res.set_content( "Error, json syntax", "text/plain" );
                        return ;
                }
                std::stringstream       xss( body.substr( xp + 2 ) );
                std::stringstream       yss( body.substr( yp + 2 ) );
                x = first_number( xss.str() );
                y = first_number( yss.str() );
                mtx.lock();
                if( othello->put( x, y, c ) == 0 )
                {
                        mtx.unlock();
                        res.set_content( "Failed", "text/plain" );
                        return ;
                }
                
                turn ++;
                mtx.unlock();
                std::cout << "turn: " << turn + 1 << ", color: " << ( ( c == WHITE )? ( circles[ 1 ] ) : ( circles[ 0 ] ) ) << ":" << "x=" << x << "," << "y=" << y << std::endl;
                std::cout << othello->draw() << std::endl;
                
                res.set_content( "Success", "text/plain" );
        });
        httpserver.Post( "/reset", [&]( const httplib::Request& req, httplib::Response& res )
        {
                mtx.lock();
                othello->reset();
                mtx.unlock();
                res.set_content( "OK", "text/plain" );
        });
        httpserver.listen( "localhost", 2400 );
        delete othello;
        return 0;
}

httplib.h は yhirose 様のものを使わせていただいております、公開していただいていることに感謝です。

# CMakeLists.txt
# CMakeのバージョンを設定
cmake_minimum_required( VERSION 3.13 )
# プロジェクト名と使用する言語を設定
project( othello CXX )
# othello という実行ファイルを作成
add_executable(
  othello 
  src/main.cpp 
  src/board.cpp
)
target_link_libraries( othello PUBLIC "-pthread" )

クライアント

深層学習を使った実装です

# requirements.txt
numpy>=2.5.1
requests>=2.34.2
torch>=2.13.0

# model.py
import torch
class FirstModel(torch.nn.Module):
    def __init__(self, indim: int, outdim: int, hdndim: int):        
        super().__init__()
        self.indim: int = indim
        self.outdim: int = outdim
        self.ptf_dense_ih = torch.nn.Linear(indim, hdndim)
        self.ptf_dense_ho = torch.nn.Linear(hdndim, outdim)
        self.ptf_xgate_ih = torch.nn.Linear(indim, hdndim)
    def forward(self, ptv_x: torch.Tensor):
        ptv_x = ptv_x.view(-1, self.indim)
        ptv_xin = ptv_x
        ptv_x = self.ptf_dense_ih(ptv_x)
        ptv_x = torch.nn.functional.leaky_relu(ptv_x)
        self.ptv_hdnx = torch.nn.functional.leaky_relu(ptv_x)
        ptv_g = self.ptf_xgate_ih(ptv_xin)
        ptv_g = torch.tanh(ptv_g)
        ptv_x = ptv_x * ptv_g
        ptv_x = self.ptf_dense_ho(ptv_x)
        # ptv_x = torch.tanh(ptv_x)
        ptv_x = ptv_x.view(-1, 1, self.outdim) 
        return ptv_x
class EvalueteModel(torch.nn.Module):
    def __init__(self):        
        super().__init__()
        self.conv1 = torch.nn.Conv2d(1, 16, 3)
        self.conv2 = torch.nn.Conv2d(16, 32, 3)
        indim = 8 * 8
        hdndim = 32 * 4 * 4
        self.ptf_dense_ih = torch.nn.Linear(indim, hdndim)
        self.ptf_xgate_ih = torch.nn.Linear(indim, hdndim)
        self.indim = indim
        self.rms_norm = torch.nn.LayerNorm([hdndim])
        self.fc1 = torch.nn.Linear(hdndim, 1024)
        self.fc2 = torch.nn.Linear(1024, 1)
    def forward(self, ptv_x: torch.Tensor):
        ptv_h = self.ptf_dense_ih(ptv_x.view(-1, self.indim))
        ptv_h = torch.relu(ptv_h)
        ptv_g = self.ptf_xgate_ih(ptv_x.view(-1, self.indim))
        ptv_g = torch.tanh(ptv_g)
        ptv_x = self.conv1(ptv_x)
        ptv_x = torch.relu(ptv_x)
        ptv_x = self.conv2(ptv_x)
        ptv_x = torch.relu(ptv_x)
        # print(ptv_x.size())
        ptv_x = ptv_x.view(ptv_x.size(0), -1)
        ptv_x = ptv_x + ptv_h
        ptv_x = self.rms_norm(ptv_x)
        ptv_x = ptv_x * ptv_g
        ptv_x = self.fc1(ptv_x)
        ptv_x = torch.nn.functional.leaky_relu(ptv_x)
        ptv_x = self.fc2(ptv_x)
        # ptv_x = torch.tanh(ptv_x)
        return ptv_x
class OthelloModel(torch.nn.Module):
    def __init__(self):        
        super().__init__()
        indim = 8 * 8
        hdndim = 64 * 4 * 4
        outdim = indim
        self.ptf_dense_ih = torch.nn.Linear(indim, hdndim)
        self.ptf_dense_ho = torch.nn.Linear(hdndim, outdim)
        self.ptf_xgate_ih = torch.nn.Linear(indim, hdndim)
        self.conv1 = torch.nn.Conv2d(1, 32, 3)
        self.conv2 = torch.nn.Conv2d(32, 64, 3)
        self.deconv1 = torch.nn.ConvTranspose2d(64, 32, 3)
        self.deconv2 = torch.nn.ConvTranspose2d(32, 1, 3)
        self.fc1 = torch.nn.Linear(hdndim, 1024)
        self.fc2 = torch.nn.Linear(1024, hdndim)
    def sub_forward(self, ptv_x: torch.Tensor):
        ptv_xin = ptv_x
        ptv_x = self.ptf_dense_ih(ptv_x)
        ptv_x = torch.nn.functional.leaky_relu(ptv_x)
        self.ptv_hdnx = torch.nn.functional.leaky_relu(ptv_x)
        ptv_g = self.ptf_xgate_ih(ptv_xin)
        ptv_g = torch.tanh(ptv_g)
        ptv_x = ptv_x * ptv_g
        ptv_x = self.ptf_dense_ho(ptv_x)
        # ptv_x = torch.tanh(ptv_x)
        return ptv_x
    def forward(self, ptv_x: torch.Tensor):
        ptv_z = self.sub_forward(ptv_x.view(-1, 8 * 8)).view(-1, 1, 8, 8)
        ptv_x = self.conv1(ptv_x)
        ptv_x = torch.relu(ptv_x)
        ptv_x = self.conv2(ptv_x)
        ptv_x = torch.relu(ptv_x)
        fcdims = ptv_x.size()
        # print(ptv_x.size())
        ptv_x = ptv_x.view(ptv_x.size()[0], -1)
        ptv_x = self.fc1(ptv_x)
        ptv_x = torch.relu(ptv_x)
        ptv_x = self.fc2(ptv_x)        
        ptv_x = ptv_x.view(fcdims)
        ptv_x = torch.relu(ptv_x)
        ptv_x = self.deconv1(ptv_x)
        ptv_x = torch.relu(ptv_x)
        ptv_x = self.deconv2(ptv_x)
        # print(ptv_x.size())
        ptv_x = ptv_x * torch.tanh(ptv_z)
        # print(ptv_x.size())
        torch.tanh(ptv_x)
        return ptv_x

# main.py
import torch
import sys
import os
import model
import requests
import time
    
def runtime_dtype_for(device: torch.device) -> torch.dtype:
    if device.type == "cuda":
        is_bf16_supported = getattr(torch.cuda, "is_bf16_supported", None)
        if callable(is_bf16_supported) and is_bf16_supported():
            return torch.bfloat16
    if device.type == "mps":
        return torch.float32
    return torch.float32
device: torch.device = torch.device('cpu')
device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')
device = torch.device('mps') if torch.backends.mps.is_available() else device
def convert_tensor(res: str, ptv_out: torch.Tensor) -> torch.Tensor:
    text = res.replace("\n", "")
    for i, t in enumerate(text):
        c = 0
        if t == '+':
            c = +1
        if t == '-':
            c = -1
        ptv_out[i] = c
    return ptv_out
def get_board(color: str, url: str='http://localhost:2400/board/bits', dims: list[int]=[8, 8], timeout: int=30):
    res = requests.get(url + "?c=" + color, timeout=timeout)
    ptv_board = torch.zeros(dims[0] * dims[1])
    return convert_tensor(res.text, ptv_board).view([1] + list(dims)).to(device).to(runtime_dtype_for(device))
def put_stone(url: str='http://localhost:2400/stone', payload: dict={}, timeout: int=30):
    res = requests.put(url, data=payload, timeout=timeout)
    return res.text == "Success"
def get_color(url: str='http://localhost:2400/now/color', timeout: int=30):
    res = requests.get(url, timeout=timeout)
    return res.text
def get_winner(url: str='http://localhost:2400/winner', timeout: int=30):
    res = requests.get(url, timeout=timeout)
    return res.text
def post_reset(url: str='http://localhost:2400/reset', timeout: int=30):
    res = requests.post(url, timeout=timeout)
    while res.status_code != 200:
        pass
    return res.text
def save_model(model, name: str):
    model_scripted = torch.jit.script(model)
    model_scripted.save(name)
def load_othello_model(name: str, device="cpu"):
    if not os.path.exists(name):
        print("create new model")
        return model.OthelloModel()
    print("loaded model")
    return torch.jit.load(name, map_location=device)
def load_evaluate_model(name: str, device="cpu"):
    if not os.path.exists(name):
        print("create new model")
        return model.EvalueteModel()
        # return model.FirstModel(64, 1, 1024)
    print("loaded model")
    return torch.jit.load(name, map_location=device)
def main(argv):
    def comp_winner(ptv_t: torch.Tensor):
        winner = get_winner()
        if color in winner:
            ptv_t = +torch.ones_like(ptv_t)
        elif len(winner) > 0:
            ptv_t = -torch.ones_like(ptv_t)
        if len(winner) > 0:
            print(winner)
            save_model(ptf_model_comp, model_path_comp)
            save_model(ptf_model_eval, model_path_eval)
            if post_reset() == "OK":
                print("reset board.")
            time.sleep(0.10)
        return ptv_t
    height: int = 8
    width: int = 8
    if len(argv) < 2:
        raise RuntimeError("Not enough parameters")
    color: str = argv[1]
    power: float = 0.99
    model_path_comp: str = os.path.join("models", f"othello_{color}_comp.pth")
    model_path_eval: str = os.path.join("models", f"othello_{color}_eval.pth")
    if not (color == "BLACK" or color == "WHITE"):
        raise RuntimeError("Invalid color")
   
    ptf_model_comp = load_othello_model(model_path_comp).to(device).to(runtime_dtype_for(device))
    ptf_model_eval = load_evaluate_model(model_path_eval).to(device).to(runtime_dtype_for(device))
    ptf_loss = torch.nn.MSELoss()
    ptf_optimizer_comp = torch.optim.RMSprop(ptf_model_comp.parameters())
    ptf_optimizer_eval = torch.optim.RMSprop(ptf_model_eval.parameters())
    ptv_board_putable = torch.ones([1, 1, height, width]).to(device).to(runtime_dtype_for(device))
    ptv_prev_x = torch.zeros([1, 1, height, width]).to(device).to(runtime_dtype_for(device))
    miss_put_count: int = 0
    while True:
        # get board
        if miss_put_count == 0:
            ptv_x = get_board(color, dims=[height, width]).view(1, 1, height, width)
        
        # evaluate model update
        with torch.no_grad():
            ptv_t = torch.tanh(ptf_model_eval(ptv_x).detach().clone())
            
        while get_color() != color:
            time.sleep(1e-2)
            ptv_t = comp_winner(ptv_t)
        # Q-learning
        ptf_optimizer_eval.zero_grad()
        ptv_p = torch.tanh(ptf_model_eval(ptv_prev_x))
        ptv_loss = ptf_loss(ptv_p, ptv_t * power)
        ptv_loss.backward()
        ptf_optimizer_eval.step()
        # compute model update
        ptf_optimizer_comp.zero_grad()
        ptv_prev_z = ptf_model_comp(ptv_prev_x)
        ptv_p = torch.tanh(ptf_model_eval(ptv_prev_z))
        ptv_loss = ptf_loss(ptv_p, ptv_t)
        ptv_loss.backward()
        ptf_optimizer_comp.step()
       
        # prediction
        search_rule: bool = True
        ptv_board_putable = torch.ones([1, 1, height, width]).to(device).to(runtime_dtype_for(device))
        ptv_board_put_ok = ptv_x.clone()
        # ptv_board_put_ok = torch.zeros([1, 1, height, width]).to(device).to(runtime_dtype_for(device))
        while search_rule:
            ptv_z = ptf_model_comp(ptv_x)
            # コスト計算
            with torch.no_grad():
                ptv_cost = ptf_model_eval(ptv_z)
            # 座標の計算
            # ptv_zz = ptv_zz.view(-1, 1, height, width)
            ptv_next_z = ptv_z - ptv_prev_z
            ptv_next_z = ptv_next_z.masked_fill(ptv_board_putable == 0, float('-inf')) + 1e-3 * torch.rand_like(ptv_next_z)
            ptv_next_z = torch.softmax(ptv_next_z.view(-1, 1, height * width), dim=-1).view(-1, 1, height, width)
            x = int(torch.argmax(torch.amax(ptv_next_z.view(-1, height, width), dim=1), dim=-1).cpu().item())
            y = int(torch.argmax(torch.amax(ptv_next_z.view(-1, height, width), dim=2), dim=-1).cpu().item())
            # send
            is_put = put_stone(payload={"c": color, "x": x, "y": y})
            if is_put:
                # put success
                print(f"color={color}, put x={x}, y={y}, miss={miss_put_count:2}, score={ptv_cost.cpu().item():.04}")
                miss_put_count = 0
                search_rule = False
            else:
                # put failed
                ptv_board_putable[0, 0, y, x] = 0
                miss_put_count += 1
            # compute model update
            if is_put:
                ptv_board_put_ok[0, 0, y, x] += 1
                ptf_optimizer_comp.zero_grad()
                ptv_board_predict = ptf_model_comp(ptv_prev_x)
                ptv_loss = ptf_loss(ptv_board_predict, ptv_board_put_ok)
                ptv_loss.backward()
                ptf_optimizer_comp.step()
            if torch.sum(ptv_board_putable) < 1:
                if post_reset() == "OK":
                    print("reset board.")
                    break
        # next
        ptv_prev_x = ptv_x
        ptv_prev_z = ptv_z
if __name__ == "__main__":
    main(sys.argv)

実行

長かったですね、、お疲れ様でした

残りはコンパイルと実行です

# server 
cmake .
make 

./othello

# client 

python main.py BLACK

python main.py WHITE

ターミナルは分けて実行してください

うまく動けば対戦がはじまります 👀

やってみましょう！

動きましたか？

動いた方はここまでお疲れ様でした！

http の通信になりますが curl コマンドなどから人が介入して対局することができます

# 黒で横に1、縦に2の位置に石をおく場合
curl -X PUT http://localhost:2400/stone -H "Content-Type: application/json" -d '{c=BLACK,x=1,y=2}'

ぜひ、育てたモデルと対局してみてください！

最後に

今回は深層学習とQ学習を組み合わせた Deep-Q-Learning を実装していきました

機械学習の世界は奥が深いので他の手法もいくつもあります

試してみることで身につけていくのも楽しいと思います！