예제 소스
VLM을 이용한 테트리스 학습 · 파일 31개
llm_macro_client.py
"""
Qt 테트리스 서버용 LLM 매크로 클라이언트.
전략:
1) Qt 서버에서 받은 grid에서 'active piece'를 마스크 아웃하여 lock된 보드만 추출.
2) 현재 piece에 대해 가능한 모든 (rotation, target_x) 후보를 파이썬에서 직접
시뮬레이션하여 cleared / holes / bumpiness / max_h / landing_y 등을 계산.
3) 휴리스틱 점수로 정렬 후 Top-K 만 LLM에 제시 → LLM은 'index'만 선택.
(LLM이 자유롭게 좌표를 만들지 못하게 해서 hallucination 차단)
4) 하드드랍 미사용. target에 도달하면 NOOP를 보내 자연 gravity로 한 칸씩 낙하시키고,
매 state(블럭이 한 칸 내려올 때마다)에서 후보를 다시 평가하여 적응적으로 선택.
5) lock된 보드 signature가 변하지 않는 동안에는 LLM 재호출을 캐시(기본값).
--no-llm-cache 를 주면 매 스텝마다 LLM을 호출.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import socket
import time
from pathlib import Path
import numpy as np
ACT_NOOP = 0
ACT_LEFT = 1
ACT_RIGHT = 2
ACT_ROT_CW = 3
ACT_ROT_CCW = 4
ACT_SOFT = 5
ACT_HARD = 6
# tetris_engine.cpp 의 base_shape() 와 동일해야 함 (4x4, row-major)
_PIECE_BASE = {
1: ((0, 0, 0, 0), (1, 1, 1, 1), (0, 0, 0, 0), (0, 0, 0, 0)), # I
2: ((0, 1, 1, 0), (0, 1, 1, 0), (0, 0, 0, 0), (0, 0, 0, 0)), # O
3: ((0, 1, 0, 0), (1, 1, 1, 0), (0, 0, 0, 0), (0, 0, 0, 0)), # T
4: ((0, 1, 1, 0), (1, 1, 0, 0), (0, 0, 0, 0), (0, 0, 0, 0)), # S
5: ((1, 1, 0, 0), (0, 1, 1, 0), (0, 0, 0, 0), (0, 0, 0, 0)), # Z
6: ((1, 0, 0, 0), (1, 1, 1, 0), (0, 0, 0, 0), (0, 0, 0, 0)), # J
7: ((0, 0, 1, 0), (1, 1, 1, 0), (0, 0, 0, 0), (0, 0, 0, 0)), # L
}
# 시각적으로 distinct 한 회전 개수 (LLM 후보 중복 제거 목적). 엔진은 0..3 모두 구분하지만
# 후보 평가에서는 unique 회전만 enumerate 하면 충분. target_rot 은 그대로 0..N-1 사용.
_PIECE_NROT = {1: 2, 2: 1, 3: 4, 4: 2, 5: 2, 6: 4, 7: 4}
PIECE_NAME = {0: "-", 1: "I", 2: "O", 3: "T", 4: "S", 5: "Z", 6: "J", 7: "L"}
# tetris_engine.h 의 기본값과 동일. Qt 서버는 visible_h 만 보내므로 클라이언트가 알고 있어야 함.
HIDDEN_ROWS = 4
# well-keeping(테트리스 셋업 보너스) 활성 플래그. main 에서 CLI 옵션으로 토글.
_ENABLE_WELL_KEEP = False
def _rotate_cw(s: tuple) -> tuple:
"""4x4 도형을 CW 90도 회전. tetris_engine.cpp::rotate_cw 와 동일."""
return tuple(tuple(s[3 - c][r] for c in range(4)) for r in range(4))
def _build_piece_cells() -> dict:
"""{ ptype: [rot0_cells, rot1_cells, ...] } where each cells is tuple of (dx, dy)."""
shapes = {}
for ptype, base in _PIECE_BASE.items():
rots = []
cur = base
for _ in range(4):
cells = tuple((c, r) for r in range(4) for c in range(4) if cur[r][c])
rots.append(cells)
if ptype != 2: # O 는 회전해도 동일
cur = _rotate_cw(cur)
shapes[ptype] = rots
return shapes
PIECE_CELLS = _build_piece_cells()
def grid_to_ascii(board: np.ndarray) -> str:
"""(visible_h, w) → ASCII. '#' 채움, '.' 빈칸. 좌측에 행번호."""
rows = []
for y in range(board.shape[0]):
row = "".join("." if int(v) == 0 else "#" for v in board[y])
rows.append(f"{y:02d}|{row}|")
return "\n".join(rows)
def mask_active_piece(grid_visible: np.ndarray, ptype: int, px: int, py_engine: int, prot: int) -> np.ndarray:
"""서버가 보낸 render-grid 에서 현재 active piece 를 제거해 lock-only 보드를 만든다.
py_engine 은 엔진 좌표(상단 hidden 포함). visible 좌표로 보려면 HIDDEN_ROWS 만큼 빼야 한다.
"""
out = grid_visible.copy()
if ptype == 0:
return out
cells = PIECE_CELLS[ptype][prot & 3]
vh, w = out.shape
for dx, dy in cells:
gy_v = (py_engine + dy) - HIDDEN_ROWS
gx = px + dx
if 0 <= gy_v < vh and 0 <= gx < w:
out[gy_v, gx] = 0
return out
def column_heights(board: np.ndarray) -> list[int]:
vh, w = board.shape
heights = []
for x in range(w):
col = board[:, x]
nz = np.where(col > 0)[0]
heights.append(int(vh - nz[0]) if len(nz) > 0 else 0)
return heights
def column_hole_count(board: np.ndarray, heights: list[int]) -> list[int]:
"""컬럼별 hole 개수 (top filled cell 아래의 빈칸)."""
vh, w = board.shape
out = [0] * w
for x in range(w):
h = heights[x]
if h == 0:
continue
top_y = vh - h
for y in range(top_y + 1, vh):
if board[y, x] == 0:
out[x] += 1
return out
def count_holes(board: np.ndarray, heights: list[int]) -> int:
return sum(column_hole_count(board, heights))
# 벽(boundary) 은 무한 높이의 컬럼처럼 취급해 well_depth 만 측정한다.
# 단, well_bonus 부여 시에는 boundary 컬럼을 후보에서 제외하여
# "빈 가장자리 = 자동 우물" 인식으로 인한 잘못된 양쪽 우물 형성을 막는다.
_WALL_HEIGHT = 999
def well_metrics(heights: list[int]) -> tuple[int, int, int]:
"""단일-컬럼 우물 깊이 메트릭.
반환:
other_well_sum : 가장 깊은 'interior' 우물을 제외한 나머지 우물 깊이의 합
(boundary 우물도 포함하여 페널티 계산)
best_interior_depth : 가장 깊은 'interior' 우물(boundary 제외) 깊이
(테트리스 셋업 후보)
best_interior_idx : 위 컬럼 인덱스 (없으면 -1)
"""
w = len(heights)
depths = [0] * w
for x in range(w):
left = heights[x - 1] if x > 0 else _WALL_HEIGHT
right = heights[x + 1] if x < w - 1 else _WALL_HEIGHT
depths[x] = max(0, min(left, right) - heights[x])
# boundary 는 wall 한쪽이 INF 라서 항상 well_depth 가 옆 컬럼만큼 부여된다.
# 이를 well_bonus 후보로 인정하면 빈 보드에서도 자동으로 우물 형성을 시도하므로,
# interior(1..w-2) 컬럼만 best 후보로 본다.
interior = [(d, i) for i, d in enumerate(depths) if 0 < i < w - 1 and d > 0]
if not interior:
return sum(depths), 0, -1
best_d, best_i = max(interior, key=lambda p: p[0])
other = sum(depths) - best_d
return other, best_d, best_i
def fits(board: np.ndarray, ptype: int, rot: int, x: int, y: int) -> bool:
cells = PIECE_CELLS[ptype][rot & 3]
vh, w = board.shape
for dx, dy in cells:
gx = x + dx
gy = y + dy
if gx < 0 or gx >= w or gy >= vh:
return False
if gy >= 0 and board[gy, gx] != 0:
return False
return True
def drop_landing(board: np.ndarray, ptype: int, rot: int, x: int) -> int | None:
"""주어진 (rot, x) 에서 piece 를 자유낙하시켰을 때 정착하는 y(visible) 를 반환.
board 자체에 충돌하면 None. piece 의 (dx,dy) 좌표계에서 dy>=0 인 cell 만 있으므로
(= 4x4 box 의 row 1 또는 그 아래) y=-min_dy 에서 시작해 충돌까지 내림.
"""
cells = PIECE_CELLS[ptype][rot & 3]
min_dy = min(dy for _, dy in cells)
y = -min_dy
if not fits(board, ptype, rot, x, y):
return None
while fits(board, ptype, rot, x, y + 1):
y += 1
return y
def evaluate_placement(
board: np.ndarray,
ptype: int,
rot: int,
x: int,
*,
heights_before: list[int] | None = None,
hole_cols_before: list[int] | None = None,
holes_before: int | None = None,
) -> dict | None:
"""주어진 (rot, x) 에서 piece 를 자유낙하시켜 정착·라인클리어 후 메트릭 산출."""
if heights_before is None:
heights_before = column_heights(board)
if hole_cols_before is None:
hole_cols_before = column_hole_count(board, heights_before)
if holes_before is None:
holes_before = sum(hole_cols_before)
y = drop_landing(board, ptype, rot, x)
if y is None:
return None
new_b = board.copy()
cells = PIECE_CELLS[ptype][rot & 3]
landing_ys = []
cells_above_hole = 0
for dx, dy in cells:
gy = y + dy
gx = x + dx
if 0 <= gy < new_b.shape[0]:
new_b[gy, gx] = ptype
# hole 이 있는 컬럼에 셀을 추가한 경우(=무조건 stack 위쪽에 쌓는 것) 카운트
if 0 <= gx < new_b.shape[1] and hole_cols_before[gx] > 0:
cells_above_hole += 1
landing_ys.append(gy)
keep_mask = (new_b == 0).any(axis=1)
cleared = int((~keep_mask).sum())
if cleared > 0:
new_b = np.vstack(
[
np.zeros((cleared, new_b.shape[1]), dtype=new_b.dtype),
new_b[keep_mask],
]
)
heights_after = column_heights(new_b)
holes_after = sum(column_hole_count(new_b, heights_after))
new_holes = max(0, holes_after - holes_before)
sum_h = sum(heights_after)
max_h = max(heights_after)
bump = sum(abs(heights_after[i] - heights_after[i + 1]) for i in range(len(heights_after) - 1))
vh = board.shape[0]
max_landing_y = max(landing_ys) if landing_ys else 0
landing_height = max(0, vh - 1 - max_landing_y)
other_well_sum, best_interior_depth, best_interior_idx = well_metrics(heights_after)
max_h_global = max(heights_after) if heights_after else 0
# 우물 보너스 조건 (모두 만족할 때만):
# - interior 컬럼에 형성된 우물 (boundary 자동 우물 제외)
# - 다른 우물이 거의 없음 (깔끔한 단일 우물)
# - 보드가 어느정도 채워진 상태 (max_h_global >= 4): 게임 초반에 우물부터 만드는 것 방지
if (
_ENABLE_WELL_KEEP
and best_interior_idx >= 0
and best_interior_depth >= 2
and other_well_sum <= 1
and max_h_global >= 4
):
well_bonus = min(best_interior_depth, 4)
# 인정된 well 컬럼 좌우 transition 은 bump 에서 제외 (우물 유지 = bump 페널티 충돌 방지)
bump_eff = 0
for i in range(len(heights_after) - 1):
if i == best_interior_idx - 1 or i == best_interior_idx:
continue
bump_eff += abs(heights_after[i] - heights_after[i + 1])
else:
well_bonus = 0
bump_eff = bump
# cells_above_hole 페널티는 정상 상황에서는 강하게, panic(높이 누적) 상황에서는 점진적으로 약화.
# max_h_global 이 16+ 가 되면 가중치가 0.6배까지 감쇠되어 hole 위에라도 쌓아 살아남을 수 있게 함.
on_hole_w = 3.0 * max(0.2, 1.0 - max_h_global / 16.0)
score = (
+ 1.0 * cleared
- 4.0 * new_holes
- on_hole_w * cells_above_hole
- 0.36 * holes_after
- 1.00 * max_h # 0.5 → 1.0 (높이 누적 강하게 회피)
- 0.30 * bump_eff # 0.18 → 0.30 (양쪽 우물 패턴 회피)
- 0.70 * landing_height # 0.5 → 0.7 (깊게 박기 우선)
- 0.40 * other_well_sum
+ 0.30 * well_bonus # 0.5 → 0.3 (테트리스 욕심을 줄임)
)
return {
"rot": rot,
"x": x,
"landing_y": y,
"cleared": cleared,
"holes": holes_after,
"new_holes": new_holes,
"cells_above_hole": cells_above_hole,
"bumpiness": bump,
"max_h": max_h,
"sum_h": sum_h,
"landing_height": landing_height,
"well_bonus": well_bonus,
"other_well_sum": other_well_sum,
"score": score,
"board_after": new_b,
}
def enumerate_candidates(board: np.ndarray, ptype: int) -> list[dict]:
if ptype == 0:
return []
heights_before = column_heights(board)
hole_cols_before = column_hole_count(board, heights_before)
holes_before = sum(hole_cols_before)
nrot = _PIECE_NROT[ptype]
w = board.shape[1]
cands = []
for rot in range(nrot):
cells = PIECE_CELLS[ptype][rot]
min_dx = min(dx for dx, _ in cells)
max_dx = max(dx for dx, _ in cells)
x_lo = -min_dx
x_hi = w - 1 - max_dx
for x in range(x_lo, x_hi + 1):
r = evaluate_placement(
board,
ptype,
rot,
x,
heights_before=heights_before,
hole_cols_before=hole_cols_before,
holes_before=holes_before,
)
if r is not None:
cands.append(r)
cands.sort(key=lambda r: r["score"], reverse=True)
return cands
def enumerate_candidates_with_lookahead(
board: np.ndarray,
ptype: int,
next_ptype: int | None,
*,
lookahead_top: int = 10,
gamma: float = 0.7,
) -> list[dict]:
"""현재 piece 의 후보 상위 N개에 대해 다음 piece 의 최적 score 를 합쳐 재정렬한다."""
cands = enumerate_candidates(board, ptype)
if not cands or not next_ptype:
for c in cands:
c["lookahead_score"] = c["score"]
return cands
head = cands[: max(1, lookahead_top)]
for c in head:
next_cands = enumerate_candidates(c["board_after"], next_ptype)
next_best = next_cands[0]["score"] if next_cands else -10.0
c["lookahead_score"] = c["score"] + gamma * next_best
head.sort(key=lambda r: r["lookahead_score"], reverse=True)
tail = cands[max(1, lookahead_top):]
for c in tail:
c["lookahead_score"] = c["score"]
return head + tail
# 모듈 레벨 토글: target 도달 후 보낼 액션. main 에서 CLI 옵션으로 hard/soft/noop 선택.
_AT_TARGET_ACTION = ACT_HARD
_AT_TARGET_MAP = {
"hard": ACT_HARD,
"soft": ACT_SOFT,
"noop": ACT_NOOP,
}
def next_micro(ptype: int, cur_rot: int, cur_x: int, target_rot: int, target_x: int) -> int:
"""target (rot, x) 에 도달하기 위한 다음 1-step micro action.
엔진은 Left/Right/Rotate 같은 비-드랍 액션 이후에도 자동으로 gravity 1칸을 적용한다.
따라서 회전/이동 단계에서도 piece 가 한 칸씩 자연 낙하한다.
target 도달 시점에 어떤 액션을 보낼지는 _AT_TARGET_ACTION 으로 결정:
- ACT_HARD : 즉시 바닥까지 내려서 lock (가장 빠름, 매 칸 재평가 X)
- ACT_SOFT : 1 칸씩 적극 낙하 (매 칸 재평가 O)
- ACT_NOOP : 자연 gravity 만 사용 (SOFT 와 결과는 동일하나 의미상 수동)
"""
if ptype == 0:
return ACT_NOOP
target_rot &= 3
cur_rot &= 3
if cur_rot != target_rot:
dr = (target_rot - cur_rot) & 3
return ACT_ROT_CW if dr <= 2 else ACT_ROT_CCW
if cur_x < target_x:
return ACT_RIGHT
if cur_x > target_x:
return ACT_LEFT
return _AT_TARGET_ACTION
def _board_signature(board: np.ndarray) -> str:
return hashlib.md5(board.tobytes()).hexdigest()
class LLMMacroAgent:
"""Top-K 후보 중 'index' 하나만 고르도록 하는 LLM 어댑터."""
def __init__(self, model_path: str):
try:
from llama_cpp import Llama
except ImportError as e:
raise RuntimeError("llama-cpp-python 이 필요합니다.") from e
mp = Path(model_path).expanduser().resolve()
if not mp.exists():
raise FileNotFoundError(f"LLM 모델 파일을 찾을 수 없습니다: {mp}")
print(f"[LLM] loading {mp}")
self.llm = Llama(
model_path=str(mp),
n_ctx=2048,
n_threads=8,
n_gpu_layers=-1,
verbose=False,
seed=0,
)
print("[LLM] loaded.")
def pick_candidate(
self,
locked_board: np.ndarray,
ptype: int,
candidates: list[dict],
next_piece_name: str | None = None,
) -> dict:
if not candidates:
return {"rot": 0, "x": 4, "landing_y": 0, "cleared": 0, "holes": 0,
"new_holes": 0, "cells_above_hole": 0, "bumpiness": 0,
"max_h": 0, "sum_h": 0, "landing_height": 0,
"well_bonus": 0, "other_well_sum": 0,
"score": 0.0, "lookahead_score": 0.0}
ascii_b = grid_to_ascii(locked_board)
piece_name = PIECE_NAME.get(ptype, "?")
cand_lines = []
for i, c in enumerate(candidates):
la = c.get("lookahead_score", c["score"])
cand_lines.append(
f" [{i}] rot={c['rot']} x={c['x']:>2} "
f"cleared={c['cleared']} new_holes={c['new_holes']} "
f"on_hole={c['cells_above_hole']} "
f"bump={c['bumpiness']} max_h={c['max_h']} "
f"land_h={c['landing_height']} well_bonus={c.get('well_bonus', 0)} "
f"score={c['score']:+.2f} look={la:+.2f}"
)
cand_block = "\n".join(cand_lines)
sys = (
"You are a Tetris expert. Choose ONE candidate index. "
'Output ONLY a JSON object with shape: {"choice": <int>}.'
)
next_block = f"Next piece preview: {next_piece_name}\n" if next_piece_name else ""
usr = (
f"Active piece: {piece_name}\n"
f"{next_block}"
"Locked-only board (active piece removed; '#' filled, '.' empty):\n"
f"{ascii_b}\n\n"
f"Pre-evaluated candidate placements (already re-sorted by 1-step lookahead score 'look';\n"
"[0] is the best by lookahead):\n"
f"{cand_block}\n\n"
"Field meaning:\n"
" cleared : lines this piece would clear\n"
" new_holes : holes newly created by this placement (must be 0 if possible)\n"
" on_hole : cells of this piece stacked ABOVE a column with buried holes\n"
" (must be 0 if possible)\n"
" bump : surface bumpiness after placement\n"
" max_h : tallest column after placement\n"
" land_h : floor-relative height of the lowest cell of this piece\n"
" well_bonus : 0..4 bonus when exactly one clean deep well remains\n"
" (column reserved for a future I-piece tetris)\n"
" score : single-piece heuristic score\n"
" look : score + 0.7 * (best score for the next piece)\n\n"
"Selection priorities (STRICT, in order):\n"
" 1) on_hole == 0 (NEVER stack on a column that already contains buried holes)\n"
" 2) new_holes == 0 (do not create new holes)\n"
" 3) cleared >= 1 (clear lines if possible)\n"
" 4) higher 'look' (1-step lookahead) wins ties\n"
" 5) lower max_h, then lower bump, then lower land_h.\n"
"If [0] already satisfies (1)+(2), just pick 0.\n"
f'Reply EXACTLY as: {{"choice": <integer 0..{len(candidates) - 1}>}}'
)
try:
out = self.llm.create_chat_completion(
messages=[
{"role": "system", "content": sys},
{"role": "user", "content": usr},
],
max_tokens=32,
temperature=0.0,
response_format={"type": "json_object"},
seed=0,
)
text = out["choices"][0]["message"]["content"]
except Exception as e:
print(f"[LLM error] {e} → fallback to heuristic top-1")
return candidates[0]
idx = self._parse_choice(text, len(candidates))
return candidates[idx]
@staticmethod
def _parse_choice(text: str, k: int) -> int:
try:
obj = json.loads(text)
v = int(obj.get("choice", 0))
if 0 <= v < k:
return v
except Exception:
pass
m = re.search(r'"choice"\s*:\s*(\d+)', text)
if m:
v = int(m.group(1))
if 0 <= v < k:
return v
m = re.search(r"\b(\d+)\b", text)
if m:
v = int(m.group(1))
if 0 <= v < k:
return v
return 0
def _connect_with_retry(
host: str,
port: int,
timeout_s: float,
retry_s: float,
read_timeout_s: float = 30.0,
) -> socket.socket:
"""connect_timeout_s 안에 접속 시도. 접속 후에는 read_timeout_s 로 recv 타임아웃 설정.
read_timeout_s 를 넉넉히 주는 이유: Qt 서버가 pause/game_over 로 state 전송이
지연되는 경우가 있다. 너무 짧으면 readline 에서 TimeoutError 빈발.
"""
deadline = time.time() + timeout_s
last_err: Exception | None = None
while time.time() < deadline:
try:
s = socket.create_connection((host, port), timeout=2.0)
s.settimeout(read_timeout_s)
return s
except OSError as e:
last_err = e
time.sleep(retry_s)
raise ConnectionError(f"Qt 서버({host}:{port}) 연결 실패. 마지막 에러: {last_err}")
def main():
p = argparse.ArgumentParser(description="Qt 테트리스용 LLM 매크로 클라이언트 (text + 후보 사전평가)")
p.add_argument("--host", default="127.0.0.1")
p.add_argument("--port", type=int, default=5555)
p.add_argument("--model-path", default="llms/qwen2.5-3b-instruct-q4_k_m.gguf")
p.add_argument("--top-k", type=int, default=6, help="LLM 에 보여줄 후보 개수")
p.add_argument(
"--lookahead-gamma",
type=float,
default=0.7,
help="다음 piece lookahead 의 할인계수 (0=비활성, 1=동등 가중)",
)
p.add_argument(
"--well-keep",
action="store_true",
default=False,
help="interior 컬럼에 4-line 테트리스용 우물을 유지하도록 보너스 부여 (기본 OFF)",
)
g_at_target = p.add_mutually_exclusive_group()
g_at_target.add_argument(
"--hard-drop",
dest="at_target",
action="store_const",
const="hard",
help="target (rot,x) 도달 시 HardDrop 으로 즉시 lock (기본값, 가장 빠름)",
)
g_at_target.add_argument(
"--soft-drop",
dest="at_target",
action="store_const",
const="soft",
help="target 도달 시 SoftDrop 으로 1 칸씩 적극 낙하 (매 칸 재평가)",
)
g_at_target.add_argument(
"--noop-drop",
dest="at_target",
action="store_const",
const="noop",
help="target 도달 시 NOOP (자연 gravity 만 사용)",
)
p.add_argument(
"--no-soft-drop",
dest="at_target",
action="store_const",
const="noop",
help="(호환용) --noop-drop 과 동일",
)
p.set_defaults(at_target="hard")
p.add_argument(
"--llm-cache",
action="store_true",
default=True,
help="lock된 보드 signature 가 동일한 동안 LLM 재호출 생략 (기본값 ON)",
)
p.add_argument(
"--no-llm-cache",
dest="llm_cache",
action="store_false",
help="매 state(블럭이 한 칸 내려올 때마다) LLM 재호출",
)
p.add_argument(
"--no-llm",
dest="use_llm",
action="store_false",
default=True,
help="LLM 을 호출하지 않고 휴리스틱 Top-1 만 사용 (가장 빠르고 안정적)",
)
p.add_argument("--connect-timeout", type=float, default=15.0)
p.add_argument(
"--read-timeout",
type=float,
default=30.0,
help="소켓 recv 타임아웃 (초). 초과 시 자동으로 재연결 (default: 30s)",
)
p.add_argument(
"--reconnect-delay",
type=float,
default=1.0,
help="연결 끊김 감지 후 재연결까지 대기 시간 (초)",
)
args = p.parse_args()
global _ENABLE_WELL_KEEP, _AT_TARGET_ACTION
_ENABLE_WELL_KEEP = bool(args.well_keep)
_AT_TARGET_ACTION = _AT_TARGET_MAP[args.at_target]
agent = LLMMacroAgent(args.model_path) if args.use_llm else None
if not args.use_llm:
print("[LLM] disabled (--no-llm) → heuristic top-1 only")
print("=" * 70)
print(
f"[cfg] at_target={args.at_target.upper()} (action code: {_AT_TARGET_ACTION}) "
f"well_keep={_ENABLE_WELL_KEEP} lookahead_gamma={args.lookahead_gamma} "
f"top_k={args.top_k}"
)
print("=" * 70)
last_piece_idx = -1
last_piece_y: int | None = None
cache_key: tuple | None = None
cache_target: tuple[int, int] | None = None
def _open_conn() -> tuple[socket.socket, "io.TextIOBase", "io.TextIOBase"]:
print(f"[net] connecting to {args.host}:{args.port}")
_s = _connect_with_retry(
args.host,
args.port,
args.connect_timeout,
0.5,
read_timeout_s=args.read_timeout,
)
print(f"[net] connected (read_timeout={args.read_timeout}s, entering agent mode).")
_fi = _s.makefile("r", encoding="utf-8", newline="\n")
_fo = _s.makefile("w", encoding="utf-8", newline="\n")
_fo.write("agent\n")
_fo.flush()
return _s, _fi, _fo
def _close_conn(_s, _fi, _fo) -> None:
for obj in (_fi, _fo):
try:
if obj is not None:
obj.close()
except Exception:
pass
try:
if _s is not None:
_s.shutdown(socket.SHUT_RDWR)
except Exception:
pass
try:
if _s is not None:
_s.close()
except Exception:
pass
s, f_in, f_out = _open_conn()
def _safe_send(payload: str) -> bool:
"""f_out 로 write+flush. 실패 시 False 반환 (다음 readline 에서 재연결 트리거).
write 자체는 버퍼에 쌓이므로 조용히 실패할 수 있지만, flush 에서 서버가 끊긴
경우 BrokenPipeError/OSError 발생. 이 경우 readline 이 뒤따라 실패하며
상단의 재연결 루틴이 처리한다.
"""
nonlocal f_out
try:
f_out.write(payload)
f_out.flush()
return True
except (OSError, ValueError) as e:
# ValueError: makefile 객체가 close 된 경우
print(f"[net] send failed ({e}) — will reconnect on next read")
return False
while True:
try:
line = f_in.readline()
except (TimeoutError, socket.timeout) as e:
print(f"[net] recv timeout ({e}) → reconnecting after {args.reconnect_delay}s")
_close_conn(s, f_in, f_out)
time.sleep(args.reconnect_delay)
# 세션 상태 리셋: 새 연결에서는 piece 시퀀스가 이어지지 않을 수 있음
last_piece_idx = -1
last_piece_y = None
cache_key = None
cache_target = None
try:
s, f_in, f_out = _open_conn()
except Exception as re:
print(f"[net] reconnect failed: {re} → retry in {args.reconnect_delay}s")
time.sleep(args.reconnect_delay)
continue
except OSError as e:
print(f"[net] socket error ({e}) → reconnecting after {args.reconnect_delay}s")
_close_conn(s, f_in, f_out)
time.sleep(args.reconnect_delay)
last_piece_idx = -1
last_piece_y = None
cache_key = None
cache_target = None
try:
s, f_in, f_out = _open_conn()
except Exception as re:
print(f"[net] reconnect failed: {re} → retry in {args.reconnect_delay}s")
time.sleep(args.reconnect_delay)
continue
if not line:
# EOF (서버 측에서 연결 종료) → 재연결 시도
print("[net] connection closed by peer → reconnecting")
_close_conn(s, f_in, f_out)
time.sleep(args.reconnect_delay)
last_piece_idx = -1
last_piece_y = None
cache_key = None
cache_target = None
try:
s, f_in, f_out = _open_conn()
except Exception as re:
print(f"[net] reconnect failed: {re} → retry in {args.reconnect_delay}s")
time.sleep(args.reconnect_delay)
continue
try:
state = json.loads(line)
except Exception:
continue
w = int(state["w"])
visible_h = int(state["visible_h"])
grid = np.asarray(state["grid"], dtype=np.uint8).reshape(visible_h, w)
status = state.get("status", "running")
if status == "game_over":
print("[game] over → restart")
_safe_send("restart\n")
last_piece_idx = -1
last_piece_y = None
cache_key = None
cache_target = None
time.sleep(0.5)
continue
if status == "paused":
_safe_send("resume\n")
continue
ptype = int(state.get("piece_type", 0))
px = int(state.get("piece_x", 0))
py = int(state.get("piece_y", 0))
prot = int(state.get("piece_rot", 0)) & 3
piece_idx = int(state.get("piece", -1))
preview_list = state.get("preview", []) or []
next_ptype = int(preview_list[0]) if preview_list else 0
if ptype == 0:
_safe_send(f"{ACT_NOOP}\n")
continue
locked = mask_active_piece(grid, ptype, px, py, prot)
new_piece = piece_idx != last_piece_idx
descended = (last_piece_y is not None) and (py != last_piece_y)
# 매 state 마다 후보를 다시 평가한다 (값싸고 deterministic). LLM 호출은 캐시 정책에 따름.
sig = _board_signature(locked)
key = (sig, ptype)
replan = new_piece or descended or cache_target is None or (not args.llm_cache)
if replan:
cands = enumerate_candidates_with_lookahead(
locked,
ptype,
next_ptype if next_ptype != 0 else None,
lookahead_top=max(args.top_k, 8),
gamma=args.lookahead_gamma,
)
if not cands:
target_rot, target_x = prot, px
else:
topk = cands[: max(1, args.top_k)]
# 메모리 절약: 후보의 board_after 는 lookahead 후 더 이상 필요 없음.
for c in topk:
c.pop("board_after", None)
use_cache = args.llm_cache and (key == cache_key) and (cache_target is not None)
if use_cache:
target_rot, target_x = cache_target
else:
if agent is not None:
next_name = PIECE_NAME.get(next_ptype, None) if next_ptype else None
chosen = agent.pick_candidate(locked, ptype, topk, next_piece_name=next_name)
else:
chosen = topk[0]
target_rot, target_x = int(chosen["rot"]), int(chosen["x"])
if new_piece:
next_str = PIECE_NAME.get(next_ptype, "?") if next_ptype else "-"
print(f"\n--- piece {PIECE_NAME[ptype]} #{piece_idx} (next: {next_str}) ---")
print(f"top-{len(topk)} candidates:")
for i, c in enumerate(topk):
mark = "*" if (c["rot"] == target_rot and c["x"] == target_x) else " "
la = c.get("lookahead_score", c["score"])
print(
f" {mark}[{i}] rot={c['rot']} x={c['x']:>2} "
f"cl={c['cleared']} new_h={c['new_holes']} "
f"on_hole={c['cells_above_hole']} "
f"bump={c['bumpiness']} mh={c['max_h']} "
f"land_h={c['landing_height']} "
f"well_b={c.get('well_bonus', 0)} "
f"score={c['score']:+.2f} look={la:+.2f}"
)
print(f" -> target rot={target_rot}, x={target_x}")
else:
print(f" [step] py={py} replan rot={target_rot} x={target_x}")
cache_key = key
cache_target = (target_rot, target_x)
target_rot, target_x = cache_target # type: ignore[misc]
micro = next_micro(ptype, prot, px, target_rot, target_x)
# 드랍 액션은 시각적으로 확인하기 어려우므로 명시적으로 로그 (HARD/SOFT 만)
if micro == ACT_HARD:
print(
f" [HARD] piece {PIECE_NAME.get(ptype, '?')} #{piece_idx} "
f"drop @ rot={prot} x={px} (target rot={target_rot}, x={target_x})"
)
elif micro == ACT_SOFT:
print(
f" [SOFT] piece {PIECE_NAME.get(ptype, '?')} #{piece_idx} "
f"py={py} (target rot={target_rot}, x={target_x})"
)
_safe_send(f"{micro}\n")
last_piece_idx = piece_idx
last_piece_y = py
if __name__ == "__main__":
main()