예제 소스
인베이더 게임으로 배우는 휴리스틱과 AI 브리지 · 파일 17개
invader_vlm_server.py
#!/usr/bin/env python3
"""인베이더 VLM TCP 서버 (Qt 클라이언트 `invader` 와 짝).
프로토콜
--------
PING → PONG
INV3 + be32(body_len)
body = u8(rows) u8(cols) u8(reserved) be32(png_len) PNG rows*cols 격자
격자: 0=빈 1=에일리언 2=플레이어 3=플레이어탄 4=적탄
응답: RPLY + u8 ok + (ok 시 u8 action + u8 0 | 실패 시 be16 errlen + utf8)
INVL + be32(body_len) — 기보/틱 로깅 전용 (추론 없음)
body = u8 rows cols action actor play_mode status be16 seq + rows*cols 보드
actor: 0=human 1=local_ai 2=vlm
play_mode: 0=Human 1=Ai 2=Vlm
status: 0=Playing 1=Won 2=Lost 3=Intermission(웨이브 간 카운트다운)
응답: LACK
"""
from __future__ import annotations
import argparse
import datetime as _dt
import errno
import io
import json
import logging
import os
from pathlib import Path
import random
import re
import socket
import struct
import sys
import threading
from typing import Optional, Tuple
try:
from PIL import Image
except ImportError as e: # pragma: no cover
raise SystemExit("pip install Pillow") from e
MAGIC_INV3 = b"INV3"
MAGIC_INVL = b"INVL"
MAGIC_PING = b"PING"
MAGIC_RESP = b"RPLY"
MAGIC_PONG = b"PONG"
MAGIC_LACK = b"LACK"
INV_ROWS = 14
INV_COLS = 11
_vlm_pipe = None
_vlm_model_id: Optional[str] = None
class PeerClosed(EOFError):
pass
def recv_exact(conn: socket.socket, n: int) -> bytes:
buf = b""
while len(buf) < n:
try:
chunk = conn.recv(n - len(buf))
except (ConnectionResetError, BrokenPipeError) as e:
raise PeerClosed(f"peer reset: {e}") from None
if not chunk:
raise PeerClosed("연결 끊김")
buf += chunk
return buf
def send_ok_action(conn: socket.socket, action: int) -> None:
if not (0 <= action <= 3):
raise ValueError(action)
conn.sendall(MAGIC_RESP + bytes([1, action & 0xFF, 0]))
def send_err(conn: socket.socket, msg: str) -> None:
b = msg.encode("utf-8")
if len(b) > 65535:
b = b[:65535]
conn.sendall(MAGIC_RESP + bytes([0]) + struct.pack(">H", len(b)) + b)
def recv_inv3(conn: socket.socket) -> Tuple[bytes, bytes]:
(body_len,) = struct.unpack(">I", recv_exact(conn, 4))
if body_len > 32 * 1024 * 1024:
raise ValueError("body 너무 큼")
body = recv_exact(conn, body_len)
if len(body) < 6:
raise ValueError("INV3 짧음")
rows, cols, _res = body[0], body[1], body[2]
if rows != INV_ROWS or cols != INV_COLS:
raise ValueError(f"격자 {rows}x{cols} (기대 {INV_ROWS}x{INV_COLS})")
o = 3
(png_len,) = struct.unpack(">I", body[o : o + 4])
o += 4
png = body[o : o + png_len]
o += png_len
need = rows * cols
board = body[o : o + need]
o += need
if o != len(body):
raise ValueError("INV3 패딩")
if len(board) != need:
raise ValueError("보드 길이")
return png, board
def recv_invl(conn: socket.socket) -> dict:
(body_len,) = struct.unpack(">I", recv_exact(conn, 4))
if body_len > 4 * 1024 * 1024:
raise ValueError("INVL 너무 큼")
body = recv_exact(conn, body_len)
if len(body) < 9:
raise ValueError("INVL 짧음")
rows, cols = body[0], body[1]
action, actor, play_mode, status = body[2], body[3], body[4], body[5]
(seq,) = struct.unpack(">H", body[6:8])
need = rows * cols
if len(body) - 8 != need:
raise ValueError("INVL 보드")
board = bytes(body[8:])
return {
"rows": rows,
"cols": cols,
"action": action,
"actor": actor,
"play_mode": play_mode,
"status": status,
"seq": seq,
"board": board,
}
def board_to_text(board: bytes) -> str:
legend = "Symbols: .=empty A=alien P=you b=yourShot B=enemyShot\n"
lines = []
for r in range(INV_ROWS):
row = []
for c in range(INV_COLS):
v = board[r * INV_COLS + c]
ch = ".APbB"[v] if 0 <= v <= 4 else "?"
row.append(ch)
lines.append(" ".join(row))
return legend + "\n".join(lines)
def parse_action(text: str) -> Optional[int]:
t = text.strip().lower()
if "left" in t or "왼" in t:
return 0
if "right" in t or "오른" in t:
return 1
if "fire" in t or "shoot" in t or "발사" in t or "스페이스" in t:
return 2
if "none" in t or "hold" in t or "대기" in t or "noop" in t:
return 3
m = re.search(r"\b([0-3])\b", t)
if m:
return int(m.group(1))
nums = [int(x) for x in re.findall(r"\b\d+\b", t)]
if nums and 0 <= nums[0] <= 3:
return nums[0]
return None
def mock_action(board: bytes, rng: random.Random) -> int:
_ = board
return rng.randint(0, 3)
def _extract_generated_text(raw) -> str:
if isinstance(raw, list) and raw:
first = raw[0]
if isinstance(first, dict):
gen = first.get("generated_text", "")
if isinstance(gen, list):
for msg in reversed(gen):
if isinstance(msg, dict) and msg.get("role") == "assistant":
content = msg.get("content", "")
if isinstance(content, list):
parts = []
for p in content:
if isinstance(p, dict) and "text" in p:
parts.append(str(p["text"]))
return "\n".join(parts)
return str(content)
return str(gen)
return str(gen)
return str(first)
return str(raw)
def get_vlm_pipeline(model_id: str, device: str):
global _vlm_pipe, _vlm_model_id
if _vlm_pipe is not None and model_id == _vlm_model_id:
return _vlm_pipe
try:
import torch # noqa: F401
from transformers import pipeline # type: ignore
except ImportError as e:
raise RuntimeError("pip install torch transformers") from e
dev_str = str(device).lower()
if dev_str.startswith("cuda"):
dev = 0
elif dev_str in {"mps", "mps:0"}:
dev = "mps"
else:
dev = -1
kwargs = {"model": model_id, "device": dev, "trust_remote_code": True}
_vlm_pipe = pipeline("image-text-to-text", **kwargs)
_vlm_model_id = model_id
return _vlm_pipe
def _greedy_generation_kwargs(pipe) -> dict:
"""do_sample=False 일 때 temperature/top_p/top_k 가 남아 있으면 최신 transformers 가 경고함."""
try:
from transformers import GenerationConfig # type: ignore
except ImportError:
return {"max_new_tokens": 8, "do_sample": False}
tok = getattr(pipe, "tokenizer", None)
pad = getattr(tok, "eos_token_id", None) if tok is not None else None
base: dict = {}
try:
m = getattr(pipe, "model", None)
gc = getattr(m, "generation_config", None) if m is not None else None
if gc is not None and hasattr(gc, "to_dict"):
base = dict(gc.to_dict())
except Exception: # pylint: disable=broad-exception-caught
base = {}
for k in ("temperature", "top_p", "top_k"):
base.pop(k, None)
base["max_new_tokens"] = 8
base["do_sample"] = False
if pad is not None:
base["pad_token_id"] = pad
try:
cfg = GenerationConfig.from_dict(base)
return {"generation_config": cfg}
except Exception: # pylint: disable=broad-exception-caught
out = {"max_new_tokens": 8, "do_sample": False}
if pad is not None:
out["pad_token_id"] = pad
return out
def predict_vlm_action(
png: bytes, board: bytes, model_id: str, device: str, illegal_retries: int = 1
) -> int:
pil = Image.open(io.BytesIO(png)).convert("RGB")
w, h = pil.size
if max(w, h) > 384:
s = 384.0 / max(w, h)
pil = pil.resize((max(1, int(w * s)), max(1, int(h * s))), Image.LANCZOS)
pipe = get_vlm_pipeline(model_id, device)
board_txt = board_to_text(board)
base = (
"You control the bottom spaceship in a Space Invaders grid.\n"
"Rows go top(0) to bottom; you are usually the bottom 'P'. "
"Respond with ONE action digit only:\n"
"0 = move left, 1 = move right, 2 = fire bullet, 3 = wait/do nothing.\n"
"No other text — only a single character 0,1,2, or 3.\n\n"
f"{board_txt}"
)
gen_kwargs = _greedy_generation_kwargs(pipe)
last_bad: Optional[str] = None
for attempt in range(1 + max(0, illegal_retries)):
prompt = base
if last_bad:
prompt += f'\n\nInvalid previous answer {last_bad!r}. Output only one of 0,1,2,3.'
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": pil},
{"type": "text", "text": prompt},
],
}
]
raw = None
last_err: Optional[BaseException] = None
for caller in (
lambda m=messages: pipe(text=m, generate_kwargs=gen_kwargs),
lambda m=messages: pipe(text=m, images=pil, generate_kwargs=gen_kwargs),
lambda: pipe(images=pil, text=prompt, generate_kwargs=gen_kwargs),
):
try:
raw = caller()
break
except Exception as e: # pylint: disable=broad-exception-caught
last_err = e
continue
if raw is None:
raise RuntimeError(f"VLM 실패: {last_err!r}") from last_err
text = _extract_generated_text(raw)
parsed = parse_action(text)
if parsed is not None:
return parsed
last_bad = text[:120]
raise RuntimeError("action 파싱 실패")
def _log_predict_jsonl(
path: Optional[str],
board: bytes,
action: int,
raw_note: str = "",
*,
model_id: Optional[str] = None,
server_mock: bool = False,
) -> None:
if not path:
return
try:
os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
rec = {
"kind": "predict",
"ts": _dt.datetime.now(_dt.timezone.utc).isoformat(),
"game": "invader",
"rows": INV_ROWS,
"cols": INV_COLS,
"action": action,
"board": [board[i] for i in range(len(board))],
"note": raw_note,
"model_id": model_id,
"server_mock": bool(server_mock),
"status": "Playing",
}
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
except OSError as e:
logging.warning("predict jsonl 실패: %s", e)
_ACTOR_NAMES = {0: "human", 1: "local_ai", 2: "vlm"}
_MODE_NAMES = {0: "Human", 1: "Ai", 2: "Vlm"}
_STATUS_NAMES = {0: "Playing", 1: "Won", 2: "Lost", 3: "Intermission"}
def _log_invl_jsonl(path: Optional[str], d: dict) -> None:
if not path:
return
board = d["board"]
rec = {
"ts": _dt.datetime.now(_dt.timezone.utc).isoformat(),
"kind": "move",
"game": "invader",
"action": d["action"],
"actor_id": int(d["actor"]),
"actor": _ACTOR_NAMES.get(d["actor"], str(d["actor"])),
"mode": _MODE_NAMES.get(d["play_mode"], str(d["play_mode"])),
"status": _STATUS_NAMES.get(d["status"], str(d["status"])),
"seq": d["seq"],
"board": [board[i] for i in range(len(board))],
"source": "invader_tcp_invl",
}
try:
os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
except OSError as e:
logging.warning("invl jsonl 실패: %s", e)
def handle_message(head: bytes, conn: socket.socket, args: argparse.Namespace, rng: random.Random) -> None:
if head == MAGIC_PING:
conn.sendall(MAGIC_PONG)
return
if head == MAGIC_INVL:
d = recv_invl(conn)
_log_invl_jsonl(getattr(args, "log_jsonl", None), d)
conn.sendall(MAGIC_LACK)
logging.info("INVL seq=%s actor=%s action=%s", d["seq"], d["actor"], d["action"])
return
if head != MAGIC_INV3:
raise ValueError(f"unknown magic {head!r}")
png, board = recv_inv3(conn)
if args.mock:
act = mock_action(board, rng)
note = "mock"
else:
try:
act = predict_vlm_action(
png, board, args.model_id, args.device, illegal_retries=args.vlm_retries
)
note = ""
except Exception as e: # pylint: disable=broad-exception-caught
logging.warning("VLM 오류 → mock 폴백: %s", e)
act = mock_action(board, rng)
note = f"fallback_err:{e}"
act = int(act) % 4
send_ok_action(conn, act)
_log_predict_jsonl(
getattr(args, "log_jsonl", None),
board,
act,
note,
model_id=getattr(args, "model_id", None),
server_mock=bool(getattr(args, "mock", False)),
)
def handle_client(conn: socket.socket, args: argparse.Namespace, rng: random.Random) -> None:
n = 0
try:
while True:
try:
head = recv_exact(conn, 4)
except PeerClosed as e:
logging.info("closed after %s msgs: %s", n, e)
return
try:
handle_message(head, conn, args, rng)
n += 1
except PeerClosed:
return
except (EOFError, struct.error, ValueError) as e:
logging.exception("framing: %s", e)
try:
send_err(conn, str(e))
except OSError:
pass
return
except Exception as e: # pylint: disable=broad-exception-caught
logging.exception("handle")
try:
send_err(conn, str(e))
except OSError:
return
finally:
try:
conn.shutdown(socket.SHUT_RDWR)
except OSError:
pass
conn.close()
def _default_log_jsonl_path() -> Optional[str]:
"""학습용 JSONL 기본 경로. INVADER_NO_LOG=1 이면 끔."""
if os.environ.get("INVADER_NO_LOG", "").strip().lower() in ("1", "true", "yes"):
return None
env = os.environ.get("INVADER_LOG_JSONL", "").strip()
if env:
return env
return str(Path(__file__).resolve().parent / "logs" / "invader_train.jsonl")
def _default_model_id() -> str:
"""환경변수 → invader/vlm_models → 형제 omook/vlm_models → Hub id 순."""
env = os.environ.get("INVADER_VLM_MODEL", "").strip()
if env:
return env
here = Path(__file__).resolve().parent
repo_root = here.parent
candidates = [
here / "vlm_models" / "Qwen__Qwen3-VL-2B-Instruct",
repo_root / "omook" / "vlm_models" / "Qwen__Qwen3-VL-2B-Instruct",
]
for local in candidates:
if local.is_dir() and (local / "config.json").is_file():
return str(local)
return "Qwen/Qwen3-VL-2B-Instruct"
def serve(args: argparse.Namespace) -> None:
rng = random.Random(args.seed)
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
srv.bind((args.host, args.port))
except OSError as e:
if e.errno == errno.EADDRINUSE:
logging.error("포트 %s 사용 중", args.port)
raise SystemExit(1) from e
raise
srv.listen(8)
logging.info(
"invader VLM tcp://%s:%s mock=%s model=%s device=%s jsonl=%s",
args.host,
args.port,
args.mock,
getattr(args, "model_id", ""),
getattr(args, "device", ""),
getattr(args, "log_jsonl", "") or "(off)",
)
try:
while True:
c, addr = srv.accept()
logging.info("client %s", addr)
threading.Thread(target=handle_client, args=(c, args, rng), daemon=True).start()
finally:
srv.close()
def main() -> None:
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
ap = argparse.ArgumentParser(description="Invader VLM TCP server")
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--port", type=int, default=9300)
ap.add_argument("--mock", action="store_true", help="모델 없이 무작위 액션 (테스트)")
ap.add_argument(
"--model-id",
default=_default_model_id(),
help="Hub id 또는 download_vlm_model.py 로 받은 로컬 디렉터리 경로",
)
ap.add_argument(
"--device",
default=os.environ.get("INVADER_VLM_DEVICE", "cpu"),
help="cuda | mps | cpu (mac GPU 는 보통 mps)",
)
ap.add_argument("--seed", type=int, default=0)
ap.add_argument("--vlm-retries", type=int, default=2)
ap.add_argument(
"--log-jsonl",
default=_default_log_jsonl_path(),
metavar="PATH",
help="학습용 JSONL (predict·INVL 한 파일에 append). 기본: invader/logs/invader_train.jsonl",
)
ap.add_argument(
"--no-log-jsonl",
action="store_true",
help="JSONL 로깅 끔 (또는 환경변수 INVADER_NO_LOG=1)",
)
args = ap.parse_args()
if args.no_log_jsonl:
args.log_jsonl = None
serve(args)
if __name__ == "__main__":
main()