目標

gym-tetris や NES の ROM のような外部環境を使わず、自作の C++ エンジンで強化学習をします。学習は画面なしで回し、Qt の画面は結果を見るときだけつなぎます。

エンジン

  • 盤は横10マス、縦26マスです。上の4行はブロックが出てくる隠し行なので、画面と観測には22行だけが見えます。
  • ブロックは 7-bag で引きます。7種類を一度ずつ混ぜて出すので、同じブロックが長く出ないことはありません。
  • 回転するときは、その場から左右・上に押した位置まで7か所を順に試し、最初に入る場所に置きます。
  • 移動、回転、何もしない、の後にはブロックが自然に1マス下がります。そのため1つのブロックに使える行動の数は盤の高さ分です。
  • ブロックが固まった瞬間に報酬を計算します。報酬の構成は次の記事で扱います。

C API と Gym 環境

エンジンは C++ のクラスなので、Python から直接は呼べません。extern "C" の関数21個で包んで共有ライブラリ(libtetris_c_api.dylib)としてビルドし、Python は ctypes で読み込みます。このライブラリは Qt なしでビルドされます。

LocalTetrisEnv は、これらの関数で gymnasium 環境を作ります。

  • 観測: 22×10 マス。空きマスは0、ブロックは種類によって1〜7です。固まったブロックと動いているブロックは区別しません。
  • 行動: 7つ。何もしない、左、右、時計回りの回転、反時計回りの回転、1マス下げる、すぐに落とす。
  • 報酬: エンジンが計算した値をそのまま使います。

エンジンは次のブロック5つを知っていますが、観測には入れませんでした。

Qt の画面と TCP

Qt アプリ(tetris_rl)を --rl-port 5555 で起動すると TCP サーバーが開きます。アプリは盤の状態を1行のJSONで送り、クライアントは行動番号を1行で返します。JSONには、盤、現在のブロックの種類・位置・回転、次のブロックのリストが入っています。学習したモデルを画面で動かしたり、後で出てくる LLM・VLM のクライアントをつないだりするときにこの通路を使います。

この段階の実際のコード

C API はハンドル一つにエンジンを入れ、行動番号を受け取って1ステップ進めます。

tetris/src/tetris_c_api.cppcpp · 17行
struct TetrisHandle {
    TetrisEngine engine;
    std::vector<uint8_t> render_grid;
    explicit TetrisHandle(const TetrisEngineConfig& cfg) : engine(cfg) {}
};

// Control
void tetris_set_seed(TetrisHandle* h, uint64_t seed) { h->engine.set_seed(seed); }

void tetris_reset(TetrisHandle* h) { h->engine.reset(); }

// Step returns 1 if alive, 0 if game over
int tetris_step(TetrisHandle* h, int action) {
    const auto a = static_cast<TetrisAction>(static_cast<int8_t>(action));
    const bool alive = h->engine.step(a);
    return alive ? 1 : 0;
}

Python の環境は、エンジンが埋めた盤から隠し行の4行を切り取って観測に使います。

local_tetris_env.pypython · 29行
def _obs(self) -> np.ndarray:
    written = int(self._lib._dll.tetris_fill_render_grid(self._h, self._buf, self._w * self._hgt))
    if written <= 0:
        raise RuntimeError("tetris_fill_render_grid() 실패")
    arr = np.frombuffer(self._buf, dtype=np.uint8, count=self._w * self._hgt)
    arr = arr.reshape(self._hgt, self._w)
    visible = arr[self._hidden :, :]  # drop hidden rows
    return visible[:, :, None]

def reset(self, *, seed: int | None = None, options: dict[str, Any] | None = None):
    super().reset(seed=seed)
    if seed is not None:
        self._lib._dll.tetris_set_seed(self._h, int(seed))
    self._lib._dll.tetris_reset(self._h)
    return self._obs(), {}

def step(self, action: int):
    alive = int(self._lib._dll.tetris_step(self._h, int(action)))
    reward = float(self._lib._dll.tetris_last_reward(self._h))
    terminated = (alive == 0) or (int(self._lib._dll.tetris_game_over(self._h)) == 1)
    truncated = False
    info = {
        "lines_cleared_total": int(self._lib._dll.tetris_lines_cleared_total(self._h)),
        "piece_index": int(self._lib._dll.tetris_piece_index(self._h)),
        "stack_height": int(self._lib._dll.tetris_stack_height(self._h)),
        "bumpiness": int(self._lib._dll.tetris_bumpiness(self._h)),
        "holes": int(self._lib._dll.tetris_holes(self._h)),
    }
    return self._obs(), reward, terminated, truncated, info

次の記事は「2. DQN を37万ステップ、消したラインは0本」です。