예제 소스
오목 좌표 LoRA: 15×15와 19×19 격자를 정확히 읽히기 · 파일 59개
vlm_pva_server.py
#!/usr/bin/env python3
"""
오목 PvA VLM TCP 서버 (클라이언트: Qt omook).
프로토컬
---------
공통: 매 메시지는 다음 4바이트 매직으로 시작.
OMK3 (현행 권장; 동적 보드 크기 + AI 색 명시):
b'OMK3' + big-endian u32(body_len)
body = u8(size) + u8(ai_side) + u32(png_len) + PNG + size*size 격자
size: 13/15/19, ai_side: 1=Black,2=White, 격자: 0=빈,1=흑,2=백
OMK2 (하위 호환; 13x13 고정, AI=백 가정):
b'OMK2' + u32(body_len)
body = u32(png_len) + PNG + 169바이트 격자
OMKP (legacy; PNG only): b'OMKP' + u32(png_len) + PNG
서버 → 클라이언트:
성공: b'RPLY' + b'\x01' + u8(row) + u8(col)
실패: b'RPLY' + b'\x00' + u16(err_utf8_len) + utf8
"""
from __future__ import annotations
import argparse
import datetime as _dt
import errno
import io
import json
import logging
import os
import queue
import random
import re
import socket
import struct
import subprocess
import sys
import threading
import time
from typing import Optional, Tuple
try:
from PIL import Image
except ImportError as e: # pragma: no cover
raise SystemExit("Pillow 필요: pip install Pillow") from e
MAGIC_OMKP = b"OMKP"
MAGIC_OMK2 = b"OMK2"
MAGIC_OMK3 = b"OMK3"
MAGIC_OMKL = b"OMKL" # 기보 로그 전용 (추론 없음).
MAGIC_PING = b"PING"
MAGIC_RESP = b"RPLY"
MAGIC_PONG = b"PONG"
MAGIC_LACK = b"LACK" # OMKL 에 대한 4바이트 ACK.
VALID_SIZES = (13, 15, 19)
# MPS/단일 파이프라인 — 동시 OMK3 추론 직렬화
_vlm_infer_lock = threading.Lock()
class PeerClosed(EOFError):
"""클라이언트가 정상/비정상으로 연결을 끊었음 — 콜러는 조용히 루프 종료."""
class IllegalMoveAfterRetries(RuntimeError):
"""보드가 있을 때 모델이 연속으로 범위 밖/점유 칸만 반환."""
def __init__(self, last_r: int, last_c: int, size: int, board: bytes):
self.last_r = last_r
self.last_c = last_c
self.size = size
self.board = board
super().__init__(f"last=({last_r},{last_c})")
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:
# 클라이언트 측에서 abort/timeout — EOF 와 동치로 취급.
raise PeerClosed(f"peer reset: {e}") from None
if not chunk:
raise PeerClosed("연결 끊김")
buf += chunk
return buf
def send_ok(conn: socket.socket, row: int, col: int, size: int = 13) -> None:
if not (0 <= row < size and 0 <= col < size):
raise ValueError(row, col)
conn.sendall(MAGIC_RESP + bytes([1, row & 0xFF, col & 0xFF]))
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_request_payload(
head: bytes, conn: socket.socket
) -> tuple[bytes, Optional[bytes], int, int]:
"""반환 (png_bytes, optional_board, size, ai_side).
ai_side: 1=Black, 2=White. 없는 프로토콜은 White(2) 로 가정.
OMKP 처럼 board 가 없으면 size 는 13 으로 가정.
"""
if head == MAGIC_OMK3:
(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("OMK3 body 너무 짧음")
size = body[0]
ai_side = body[1]
if size not in VALID_SIZES:
raise ValueError(f"지원하지 않는 보드 크기: {size}")
if ai_side not in (1, 2):
raise ValueError(f"잘못된 ai_side: {ai_side}")
o = 2
(png_len,) = struct.unpack(">I", body[o : o + 4])
o += 4
png = body[o : o + png_len]
o += png_len
board_len = size * size
board = body[o : o + board_len]
o += board_len
if o != len(body):
raise ValueError("OMK3 body 패딩 불일치")
if len(board) != board_len:
raise ValueError("board 길이")
return png, board, size, ai_side
if head == MAGIC_OMK2:
(body_len,) = struct.unpack(">I", recv_exact(conn, 4))
if body_len > 32 * 1024 * 1024:
raise ValueError("body 너무 큼")
body = recv_exact(conn, body_len)
o = 0
(png_len,) = struct.unpack(">I", body[o : o + 4])
o += 4
png = body[o : o + png_len]
o += png_len
board = body[o : o + 169]
o += 169
if o != len(body):
raise ValueError("body 패딩 불일치")
if len(board) != 169:
raise ValueError("board 길이")
return png, board, 13, 2 # OMK2 는 백 AI 가정.
if head == MAGIC_OMKP:
(ln,) = struct.unpack(">I", recv_exact(conn, 4))
if ln > 20 * 1024 * 1024:
raise ValueError("PNG 크기 과대")
png = recv_exact(conn, ln)
return png, None, 13, 2
raise ValueError(f"알 수 없는 헤더: {head!r}")
def _corner_margin(size: int) -> int:
"""귀(모서리)로 볼 여백. 13/15→2, 19→3."""
return 2 if size <= 15 else 3
def _is_corner_cell(r: int, c: int, size: int) -> bool:
m = _corner_margin(size)
return (r <= m or r >= size - 1 - m) and (c <= m or c >= size - 1 - m)
def _stone_count(board: bytes, size: int) -> int:
return sum(1 for i in range(size * size) if board[i] != 0)
def _is_opening_board(board: Optional[bytes], size: int) -> bool:
"""돌이 하나도 없는 빈 판 = 흑의 첫수."""
return board is not None and len(board) == size * size and _stone_count(board, size) == 0
def _opening_center(size: int) -> Tuple[int, int]:
mid = size // 2
return mid, mid
def mock_pick_move(
board: Optional[bytes], png: bytes, rng: random.Random, size: int = 13
) -> Tuple[int, int]:
if board is not None:
empty = [(r, c) for r in range(size) for c in range(size) if board[r * size + c] == 0]
if not empty:
raise RuntimeError("빈 칸 없음")
# 첫수는 귀 제외 · 중앙 우선.
if _is_opening_board(board, size):
mid = _opening_center(size)
if mid in empty:
return mid
non_corner = [rc for rc in empty if not _is_corner_cell(rc[0], rc[1], size)]
pool = non_corner or empty
return pool[rng.randrange(len(pool))]
return empty[rng.randrange(len(empty))]
_ = Image.open(io.BytesIO(png)).convert("RGB")
empties = [(r, c) for r in range(size) for c in range(size)]
return empties[rng.randrange(len(empties))]
def parse_two_integers(text: str, size: int = 13) -> Optional[Tuple[int, int]]:
"""모델 응답에서 (row, col) 추출.
보드 텍스트(행·열 헤더, Black stones 목록)를 메아리치면 앞쪽 숫자가
오염되므로, 응답 *끝쪽*의 유효 쌍을 우선한다.
"""
if not text:
return None
m = re.search(r'"row"\s*:\s*(\d+)\s*,\s*"col"\s*:\s*(\d+)', text)
if m:
r, c = int(m.group(1)), int(m.group(2))
if 0 <= r < size and 0 <= c < size:
return r, c
# "12 3" / "12,3" 형태 — 등장 순서대로 모은 뒤 마지막 유효 쌍
pairs = re.findall(r"(?<!\d)(\d{1,2})\s*[, ]\s*(\d{1,2})(?!\d)", text)
for a, b in reversed(pairs):
r, c = int(a), int(b)
if 0 <= r < size and 0 <= c < size:
return r, c
nums = [int(x) for x in re.findall(r"\b\d{1,2}\b", text)]
for i in range(len(nums) - 1, 0, -1):
r, c = nums[i - 1], nums[i]
if 0 <= r < size and 0 <= c < size:
return r, c
return None
_vlm_pipe = None
_vlm_model_id: Optional[str] = None
_vlm_lora_dir: Optional[str] = None
# collapse(="!!!!!" 만 출력) 감지가 누적되면 다음 학습 라운드까지 LoRA 를 disable 한다.
_vlm_lora_disabled = False
_vlm_collapse_streak = 0
_VLM_COLLAPSE_TOLERANCE = 3 # 같은 응답이 N 회 연속이면 LoRA 끄기.
# main() 에서 --backend / --board-format 으로 정한다.
_BACKEND = "transformers"
_BOARD_FORMAT = "legacy"
class _MlxTextModel:
"""mlx-vlm 텍스트 전용 추론. coord_omook.py 로 학습한 mlx LoRA 를 그대로 얹는다.
MLX GPU 스트림은 스레드 로컬이라, 로드와 generate 를 모두 전용 스레드 하나에서 돌린다.
"""
def __init__(self, model_id: str, lora_dir: Optional[str]) -> None:
from concurrent.futures import ThreadPoolExecutor
self._exec = ThreadPoolExecutor(max_workers=1, thread_name_prefix="mlx-inference")
self._exec.submit(self._load, model_id, lora_dir).result()
def _load(self, model_id: str, lora_dir: Optional[str]) -> None:
try:
from mlx_vlm import generate, load # type: ignore
from mlx_vlm.prompt_utils import apply_chat_template # type: ignore
from mlx_vlm.utils import load_config # type: ignore
except ImportError as e:
raise RuntimeError(
"--backend mlx 에는 mlx-vlm 이 필요합니다 (예: ../go/.venv/bin/python 으로 실행)."
) from e
self._generate = generate
self._template = apply_chat_template
self.model, self.processor = load(model_id)
if lora_dir and os.path.isfile(os.path.join(lora_dir, "adapter_config.json")):
from mlx_vlm.trainer.utils import apply_lora_layers # type: ignore
logging.info("mlx LoRA 어댑터 로드: base=%s, adapter=%s", model_id, lora_dir)
updated = apply_lora_layers(self.model, lora_dir)
if updated is not None:
self.model = updated
elif lora_dir:
logging.info("LoRA 디렉토리 %r 에 adapter_config.json 이 없어 base 만 씁니다.", lora_dir)
self.config = load_config(model_id)
def __call__(self, prompt: str, max_tokens: int = 32) -> str:
return self._exec.submit(self._infer, prompt, max_tokens).result()
def _infer(self, prompt: str, max_tokens: int) -> str:
try:
formatted = self._template(self.processor, self.config, prompt, num_images=0)
except TypeError:
formatted = self._template(self.processor, prompt, num_images=0)
kw = dict(max_tokens=max_tokens, verbose=False)
try:
res = self._generate(self.model, self.processor, formatted, temperature=0.0, **kw)
except TypeError:
res = self._generate(self.model, self.processor, formatted, **kw)
text = getattr(res, "text", res)
return text if isinstance(text, str) else ""
def get_vlm_pipeline(model_id: str, device: str, lora_dir: Optional[str] = None):
"""모델(+선택적 LoRA 어댑터) 을 로드해 image-text-to-text 파이프라인을 캐시한다."""
global _vlm_pipe, _vlm_model_id, _vlm_lora_dir
if (
_vlm_pipe is not None
and model_id == _vlm_model_id
and (lora_dir or None) == _vlm_lora_dir
):
return _vlm_pipe
if _BACKEND == "mlx":
_vlm_pipe = None
_vlm_pipe = _MlxTextModel(model_id, lora_dir)
_vlm_model_id = model_id
_vlm_lora_dir = lora_dir or None
return _vlm_pipe
try:
import torchvision # noqa: F401 — Idefics3/SmolVLM 이미지 프로세서 등에 필요
_ = torchvision
import torch # type: ignore
from transformers import pipeline # type: ignore
except ImportError as e:
raise RuntimeError(
"pip install torch torchvision transformers accelerate — "
"VLM 로딩에 필요한 패키지가 부족합니다."
) 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
if dev_str.startswith("cuda"):
dtype = torch.bfloat16
elif dev_str == "mps":
dtype = torch.float16
else:
dtype = None # cpu → fp32 default
# LoRA 디렉토리가 지정되었지만 아직 어댑터가 학습되지 않아 비어 있는 경우
# (--auto-train 첫 부팅 시 흔함) → base 모델만 로드해서 추론을 가능하게 두고,
# 첫 학습 라운드가 끝나면 캐시 무효화 → 다음 호출에서 새 어댑터를 흡수한다.
have_adapter = bool(lora_dir) and os.path.isfile(
os.path.join(lora_dir, "adapter_config.json")
)
if lora_dir and not have_adapter:
logging.info(
"LoRA 디렉토리 %r 에 adapter_config.json 이 없어 base 모델만 로드합니다 "
"(아직 학습되지 않음). 학습이 끝나면 자동으로 적용됩니다.",
lora_dir,
)
if have_adapter:
# LoRA 가 붙는 경우엔 모델을 직접 로드해 PEFT 어댑터를 합치고 파이프라인에 주입.
try:
from transformers import AutoModelForImageTextToText, AutoProcessor # type: ignore
from peft import PeftModel # type: ignore
except ImportError as e:
raise RuntimeError(
"pip install peft — LoRA 어댑터 로딩에 필요합니다."
) from e
logging.info("LoRA 어댑터 로드: base=%s, adapter=%s", model_id, lora_dir)
model_kwargs = {"trust_remote_code": True}
if dtype is not None:
model_kwargs["dtype"] = dtype
base = AutoModelForImageTextToText.from_pretrained(model_id, **model_kwargs)
merged = PeftModel.from_pretrained(base, lora_dir)
try:
merged = merged.merge_and_unload() # 가능하면 합쳐서 추론 속도 회복.
except Exception as e: # pylint: disable=broad-exception-caught
logging.warning("LoRA merge_and_unload 실패(%s) — adapter 분리 상태로 사용.", e)
processor = AutoProcessor.from_pretrained(lora_dir, trust_remote_code=True)
if dev != -1:
try:
merged.to("cuda" if dev == 0 else dev)
except Exception as e: # pylint: disable=broad-exception-caught
logging.warning("LoRA model .to(%s) 실패: %s", dev, e)
_vlm_pipe = pipeline(
"image-text-to-text",
model=merged,
processor=processor,
device=dev,
)
else:
pipe_kwargs = {
"model": model_id,
"device": dev,
"trust_remote_code": True,
}
if dtype is not None:
pipe_kwargs["dtype"] = dtype
_vlm_pipe = pipeline("image-text-to-text", **pipe_kwargs)
_vlm_model_id = model_id
# 캐시 키는 "요청된 lora_dir" 그대로 기록한다 — 아직 어댑터가 없어 base 만 올린 경우라도
# 같은 lora_dir 로 들어오는 후속 요청에는 캐시 hit 가 나야 한다. (그렇지 않으면 매 요청마다
# 8B 모델이 통째로 reload 되어 추론이 사실상 멈춘다.)
# 학습이 끝나 어댑터가 생긴 시점에는 auto-train 워커가 명시적으로 _vlm_pipe=None 으로
# 캐시를 무효화하므로, 그 다음 호출에서 자연스럽게 어댑터까지 흡수된 새 파이프라인이 만들어진다.
_vlm_lora_dir = lora_dir or None
return _vlm_pipe
def _extract_generated_text(raw) -> str:
"""파이프라인 응답에서 모델이 생성한 텍스트만 뽑아낸다.
transformers image-text-to-text 파이프라인은 보통
[{"generated_text": [{"role": "user", ...}, {"role": "assistant", "content": "..."}]}]
형태(messages 입력 시) 또는
[{"generated_text": "..."}]
형태(plain text 입력 시)를 반환한다.
"""
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 _board_to_text(board: Optional[bytes], size: int = 13, fmt: Optional[str] = None) -> str:
"""size*size 바이트 격자를 모델이 읽기 쉬운 텍스트로 변환.
15/19 에서는 행·열 번호가 없으면 좌표 착오가 잦아서 헤더를 붙인다.
legacy 는 헤더가 3칸, 칸이 2칸 간격이라 오른쪽으로 갈수록 열이 어긋난다.
aligned 는 coord_omook.py 의 좌표 학습 데이터와 글자 하나까지 같아야 한다.
"""
if not board or len(board) != size * size:
return "(no board state provided)"
fmt = fmt or _BOARD_FORMAT
last_idx = size - 1
lines = []
if fmt == "aligned":
lines.append(" " + " ".join(f"{c:02d}" for c in range(size)))
for r in range(size):
cells = []
for c in range(size):
v = board[r * size + c]
cells.append(f"{'.' if v == 0 else ('X' if v == 1 else 'O'):<2}")
lines.append(f"{r:02d} " + " ".join(cells))
else:
lines.append(" " + " ".join(f"{c:2d}" for c in range(size)))
for r in range(size):
cells = []
for c in range(size):
v = board[r * size + c]
cells.append(" ." if v == 0 else (" X" if v == 1 else " O"))
lines.append(f"{r:2d} " + "".join(cells))
occ_b = [(r, c) for r in range(size) for c in range(size) if board[r * size + c] == 1]
occ_w = [(r, c) for r in range(size) for c in range(size) if board[r * size + c] == 2]
head = (
f"Board TEXT is authoritative (ignore fuzzy image coords). "
f"X=Black, O=White, .=empty. "
f"row = top→bottom 0..{last_idx}, col = left→right 0..{last_idx}. "
)
if fmt == "aligned":
head += "Each cell symbol sits at the LEFT of its two-character slot under the column number. "
return (
head
+ "Answer with 0-based integers only.\n"
+ "\n".join(lines)
+ f"\nBlack stones (row,col): {occ_b}\nWhite stones (row,col): {occ_w}"
)
def _downscale_for_vlm(pil: Image.Image, max_edge: int = 384) -> Image.Image:
"""VLM 입력 이미지를 적당한 크기로 리사이즈.
Qwen3-VL 등 ViT 기반 모델은 픽셀 수에 비례해 토큰이 폭증한다(예: 28×28 패치).
15×15/19×19 는 교점이 많아 너무 줄이면 좌표 인식이 깨지므로 max_edge 를 키운다.
"""
w, h = pil.size
long_side = max(w, h)
if long_side <= max_edge:
return pil
scale = max_edge / float(long_side)
new_w = max(1, int(round(w * scale)))
new_h = max(1, int(round(h * scale)))
return pil.resize((new_w, new_h), Image.LANCZOS)
def _vlm_image_max_edge(size: int) -> int:
if size >= 19:
return 640
if size >= 15:
return 512
return 384
def _coord_is_empty_cell(board: bytes, r: int, c: int, size: int) -> bool:
if r < 0 or r >= size or c < 0 or c >= size:
return False
return board[r * size + c] == 0
def _is_legal_vlm_move(board: bytes, r: int, c: int, size: int) -> bool:
if not _coord_is_empty_cell(board, r, c, size):
return False
if _is_opening_board(board, size) and _is_corner_cell(r, c, size):
return False
return True
def _move_rejection_reason(board: bytes, r: int, c: int, size: int) -> str:
"""모델이 낸 (r,c) 가 왜 두면 안 되는지 (로그·재추론 프롬프트용)."""
if r < 0 or r >= size or c < 0 or c >= size or board[r * size + c] != 0:
return _invalid_coord_reason(board, r, c, size)
if _is_opening_board(board, size) and _is_corner_cell(r, c, size):
m = _corner_margin(size)
mid = size // 2
return (
f"첫수(빈 판) 귀(모서리) 착수 금지 — "
f"가장자리 {m}칸 이내 모서리는 불가, 중앙 부근(예: {mid} {mid})에 두세요"
)
return "내부 불일치: 좌표는 허용 가능해 보이는데 부적절 분기 진입"
def predict_vlm_png(
png: bytes,
model_id: str,
device: str,
board: Optional[bytes] = None,
size: int = 13,
ai_side: int = 2,
lora_dir: Optional[str] = None,
illegal_move_retries: int = 2,
) -> Tuple[int, int]:
# 동시 추론 직렬화 (스레드별 클라이언트 수용은 유지)
with _vlm_infer_lock:
return _predict_vlm_png_unlocked(
png,
model_id,
device,
board=board,
size=size,
ai_side=ai_side,
lora_dir=lora_dir,
illegal_move_retries=illegal_move_retries,
)
def _predict_vlm_png_unlocked(
png: bytes,
model_id: str,
device: str,
board: Optional[bytes] = None,
size: int = 13,
ai_side: int = 2,
lora_dir: Optional[str] = None,
illegal_move_retries: int = 2,
) -> Tuple[int, int]:
use_mlx = _BACKEND == "mlx"
pil = None
if not use_mlx:
pil_full = Image.open(io.BytesIO(png)).convert("RGB")
pil = _downscale_for_vlm(pil_full, max_edge=_vlm_image_max_edge(size))
pipe = get_vlm_pipeline(model_id, device, lora_dir=lora_dir)
board_txt = _board_to_text(board, size=size)
last_idx = size - 1
you_label = "Black (X)" if ai_side == 1 else "White (O)"
opp_label = "White" if ai_side == 1 else "Black"
opening = _is_opening_board(board, size)
mid_r, mid_c = _opening_center(size)
opening_rule = ""
if opening:
m = _corner_margin(size)
opening_rule = (
f" OPENING RULE: the board is empty — this is the first move. "
f"Do NOT play in a corner (귀). Corner means both row and col within "
f"{m} of an edge. Prefer the center, e.g. \"{mid_r} {mid_c}\"."
)
turn_line = (
f"You are a {size}x{size} Gomoku (오목) player playing {you_label}."
if opening
else (
f"You are a {size}x{size} Gomoku (오목) player. {opp_label} has just moved; "
f"you play {you_label} next."
)
)
base_prompt = (
f"{turn_line}\n"
f"{board_txt}\n"
f"Choose your next {you_label} move on an EMPTY cell (a '.' in the TEXT board)."
f"{opening_rule} "
"Trust the TEXT board coordinates over the image. "
"Reply with EXACTLY two integers separated by a single space: \"row col\". "
f"row and col are both in 0..{last_idx}. No other text."
)
gen_kwargs: dict = {}
if not use_mlx:
try:
from transformers import GenerationConfig # type: ignore
gen_cfg = GenerationConfig(
max_new_tokens=32,
do_sample=False,
pad_token_id=getattr(pipe, "tokenizer", None) and pipe.tokenizer.eos_token_id,
)
gen_kwargs = {"generation_config": gen_cfg}
except Exception: # pylint: disable=broad-exception-caught
gen_kwargs = {"max_new_tokens": 32, "do_sample": False}
have_board = board is not None and len(board) == size * size
max_attempts = 1 + max(0, illegal_move_retries) if have_board else 1
last_bad: Optional[Tuple[int, int]] = None
global _vlm_collapse_streak
for attempt in range(max_attempts):
prompt = base_prompt
if last_bad is not None and have_board:
br, bc = last_bad
reason = _move_rejection_reason(board, br, bc, size)
opening_hint = ""
if opening:
opening_hint = (
f" Remember: first move must NOT be a corner; prefer \"{mid_r} {mid_c}\"."
)
prompt += (
f"\n\nYour previous answer \"{br} {bc}\" is INVALID: {reason}. "
"Output AGAIN exactly two integers \"row col\" for a cell that is '.' "
f"(empty) in the board text above. row,col must be in 0..{last_idx}."
f"{opening_hint} No other words."
)
messages = [
{
"role": "user",
"content": [
{"type": "image", "image": pil},
{"type": "text", "text": prompt},
],
}
]
raw = None
last_err: Optional[BaseException] = None
logging.info(
"VLM 추론 시작 attempt=%d/%d size=%d ai_side=%d png=%dB backend=%s",
attempt + 1,
max_attempts,
size,
ai_side,
len(png),
_BACKEND,
)
t_infer = time.monotonic()
if use_mlx:
try:
raw = pipe(prompt)
except Exception as e: # pylint: disable=broad-exception-caught
last_err = e
else:
for caller in (
lambda m=messages, pr=prompt: pipe(text=m, generate_kwargs=gen_kwargs),
lambda m=messages, pr=prompt: pipe(text=m, images=pil, generate_kwargs=gen_kwargs),
lambda pr=prompt: pipe(images=pil, text=pr, generate_kwargs=gen_kwargs),
lambda pr=prompt: pipe(text=pr, images=pil, generate_kwargs=gen_kwargs),
):
try:
raw = caller()
break
except Exception as e: # pylint: disable=broad-exception-caught
last_err = e
continue
logging.info(
"VLM 추론 종료 attempt=%d elapsed=%.1fs ok=%s",
attempt + 1,
time.monotonic() - t_infer,
raw is not None,
)
if raw is None:
raise RuntimeError(f"VLM pipeline 호출 실패: {last_err!r}") from last_err
text = raw if use_mlx else _extract_generated_text(raw)
logging.info(
"VLM raw reply attempt=%d: %r",
attempt + 1,
(text or "")[:240],
)
parsed = parse_two_integers(text, size=size)
if not parsed:
if _looks_collapsed(text):
_note_possible_collapse(text, lora_dir=lora_dir)
raise RuntimeError(f"모델 응답 파싱 실패: {text[:400]!r}")
pr, pc = parsed
if not have_board or _is_legal_vlm_move(board, pr, pc, size):
_vlm_collapse_streak = 0
return parsed
logging.warning(
"VLM 불법 좌표 (%d,%d) — %s — 재추론 %d/%d",
pr,
pc,
_move_rejection_reason(board, pr, pc, size),
attempt + 1,
max_attempts,
)
last_bad = (pr, pc)
assert last_bad is not None and have_board and board is not None
raise IllegalMoveAfterRetries(last_bad[0], last_bad[1], size, board)
def _looks_collapsed(text: str) -> bool:
"""모델이 같은 토큰만 반복하는 collapse 출력인지 휴리스틱 검사."""
if not text:
return False
s = text.strip()
if len(s) < 8:
return False
# 단일 문자를 8회 이상 반복하면 collapse 로 본다 (예: '!!!!!!!!', '....', 'aaaa').
only_char = s[0]
if all(c == only_char for c in s):
return True
# 하나의 문자가 95% 이상 차지해도 collapse 후보.
counts: dict = {}
for c in s:
counts[c] = counts.get(c, 0) + 1
top = max(counts.values())
if top >= len(s) * 0.95 and len(s) >= 16:
return True
return False
def _note_possible_collapse(text: str, *, lora_dir: Optional[str]) -> None:
"""collapse 가 연속해서 일어나면 LoRA 어댑터를 자동 비활성화."""
global _vlm_collapse_streak, _vlm_lora_disabled, _vlm_pipe, _vlm_model_id, _vlm_lora_dir
_vlm_collapse_streak += 1
logging.warning(
"[collapse] 모델 출력이 단일 토큰 반복 (%r 첫64=%r). 누적 %d/%d.",
bool(lora_dir),
text[:64],
_vlm_collapse_streak,
_VLM_COLLAPSE_TOLERANCE,
)
if (
not _vlm_lora_disabled
and lora_dir
and _vlm_collapse_streak >= _VLM_COLLAPSE_TOLERANCE
):
logging.error(
"[collapse] LoRA 어댑터(%s) 가 collapse 출력만 내고 있어 자동 비활성화. "
"다음 학습 라운드가 정상 종료되면 새 어댑터로 재시도.",
lora_dir,
)
_vlm_lora_disabled = True
# 다음 호출에서 base 만으로 재로드.
_vlm_pipe = None
_vlm_model_id = None
_vlm_lora_dir = None
def _empty_cells(board: bytes, size: int) -> list[Tuple[int, int]]:
return [(r, c) for r in range(size) for c in range(size) if board[r * size + c] == 0]
def _occupied_cells(board: bytes, size: int) -> list[Tuple[int, int]]:
return [(r, c) for r in range(size) for c in range(size) if board[r * size + c] != 0]
def _invalid_coord_reason(board: bytes, r: int, c: int, size: int) -> str:
"""모델이 낸 (r,c) 가 왜 두면 안 되는지 사람이 읽을 설명 (로그용)."""
if r < 0 or r >= size:
return f"행 범위 밖 — 유효 0~{size - 1}, 모델 r={r}"
if c < 0 or c >= size:
return f"열 범위 밖 — 유효 0~{size - 1}, 모델 c={c}"
cell = int(board[r * size + c])
if cell == 0:
return "내부 불일치: 좌표는 범위 안·빈 칸인데 부적절 분기 진입"
if cell == 1:
return "이미 흑돌이 있는 칸(점유)"
if cell == 2:
return "이미 백돌이 있는 칸(점유)"
return f"이미 점유된 칸(보드값={cell})"
def _nearest_empty(board: bytes, r: int, c: int, size: int) -> Optional[Tuple[int, int]]:
"""모델이 답한 (r,c) 가 점유되어 있을 때 의미 있는 대안을 고른다.
1순위 — '점유 칸과 인접(8방향)' 한 빈 칸 중 (r,c) 와 가장 가까운 곳.
2순위 — 그냥 (r,c) 와 가장 가까운 빈 칸.
3순위 — 보드가 비어 있으면 중앙.
"""
empties = _empty_cells(board, size)
if not empties:
return None
occupied = _occupied_cells(board, size)
if not occupied:
mid = size // 2
return (mid, mid) if (mid, mid) in empties else empties[0]
def cheb_to(rc: Tuple[int, int], tr: int, tc: int) -> int:
return max(abs(rc[0] - tr), abs(rc[1] - tc))
near_stone = []
for er, ec in empties:
for orow, ocol in occupied:
if max(abs(er - orow), abs(ec - ocol)) <= 1:
near_stone.append((er, ec))
break
pool = near_stone or empties
return min(pool, key=lambda rc: (cheb_to(rc, r, c), abs(rc[0] - r) + abs(rc[1] - c)))
def _handle_one_message(
head: bytes, conn: socket.socket, args: argparse.Namespace, rng: random.Random
) -> None:
"""한 클라이언트 연결 안에서 들어온 한 개의 메시지를 처리.
persistent 모드에서 같은 함수가 같은 conn 으로 반복 호출됩니다.
"""
if head == MAGIC_PING:
try:
conn.sendall(MAGIC_PONG)
except OSError:
pass
logging.info("ping ok")
return
if head == MAGIC_OMKL:
_handle_omkl_message(conn, args)
return
png, board, size, ai_side = recv_request_payload(head, conn)
raw_r: Optional[int] = None
raw_c: Optional[int] = None
corrected = False
correct_reason: Optional[str] = None
used_mock = False
if args.mock:
r, c = mock_pick_move(board, png, rng, size=size)
used_mock = True
else:
illegal_adj_fallback = False
try:
effective_lora_dir = (
None if _vlm_lora_disabled else getattr(args, "lora_dir", None)
)
retries = max(0, min(int(getattr(args, "vlm_illegal_retries", 2)), 8))
r, c = predict_vlm_png(
png,
args.model_id,
args.device,
board=board,
size=size,
ai_side=ai_side,
lora_dir=effective_lora_dir,
illegal_move_retries=retries,
)
raw_r, raw_c = r, c
except IllegalMoveAfterRetries as e:
if args.strict:
raise
if board is None:
raise
why = _move_rejection_reason(board, e.last_r, e.last_c, size)
ntry = max(0, min(int(getattr(args, "vlm_illegal_retries", 2)), 8)) + 1
if getattr(args, "no_snap", False):
_reply_no_snap(conn, args, board, size, ai_side, e.last_r, e.last_c,
f"재추론 {ntry}회 소진; 마지막 불법: {why}")
return
logging.warning(
"VLM 불법 좌표 재추론 %d회 소진 — 마지막 (%d,%d) (%s) → 인접·근접 빈 칸 보정",
ntry,
e.last_r,
e.last_c,
why,
)
if _is_opening_board(board, size):
near = _opening_center(size)
else:
near = _nearest_empty(board, e.last_r, e.last_c, size=size)
if near is None:
raise RuntimeError("빈 칸 없음 (보드 가득 참)") from e
r, c = near
raw_r, raw_c = e.last_r, e.last_c
corrected = True
correct_reason = f"재추론 {ntry}회 소진 후 인접 보정; 마지막 불법: {why}"
illegal_adj_fallback = True
except Exception as e: # pylint: disable=broad-exception-caught
if args.strict or board is None:
raise
logging.warning("VLM 호출 실패 → 폴백 무작위 빈 칸 사용: %s", e)
fallback = mock_pick_move(board, png, rng, size=size)
r, c = fallback
used_mock = True
if (
not illegal_adj_fallback
and board is not None
and not _is_legal_vlm_move(board, r, c, size)
):
if args.strict:
raise RuntimeError(f"모델이 부적절 칸을 반환: ({r},{c})")
why = _move_rejection_reason(board, r, c, size)
if getattr(args, "no_snap", False) and not used_mock:
_reply_no_snap(conn, args, board, size, ai_side, r, c, why)
return
if _is_opening_board(board, size):
near = _opening_center(size)
else:
near = _nearest_empty(board, r, c, size=size)
if near is None:
raise RuntimeError("빈 칸 없음 (보드 가득 참)")
logging.warning(
"모델 좌표 (%d,%d) 부적절 — 사유: %s → 인접·근접 빈 칸 (%d,%d) 로 보정 "
"(보정은 좌표 복구용 휴리스틱이라 좋은 수를 보장하지 않음)",
r,
c,
why,
*near,
)
r, c = near
corrected = True
correct_reason = why
send_ok(conn, r, c, size=size)
_maybe_log_request(
args, board, size, ai_side, r, c, raw_r, raw_c, corrected, used_mock, correct_reason
)
def _reply_no_snap(
conn: socket.socket,
args: argparse.Namespace,
board: bytes,
size: int,
ai_side: int,
raw_r: int,
raw_c: int,
why: str,
) -> None:
"""--no-snap: 불법 좌표를 근처 빈 칸으로 옮기지 않고 실패로 돌려준다.
Qt 클라이언트는 실패 시 로컬 AI 가 대신 두므로, 이 건수는 로그(no_snap_fail)로 따로 센다.
"""
logging.warning("VLM 불법 좌표 (%d,%d) — %s → 보정 없이 실패 응답 (--no-snap)", raw_r, raw_c, why)
send_err(conn, f"VLM 불법 좌표 ({raw_r},{raw_c}): {why}")
_maybe_log_request(
args, board, size, ai_side, None, None, raw_r, raw_c, False, False, why, no_snap_fail=True
)
def _maybe_log_request(
args: argparse.Namespace,
board: Optional[bytes],
size: int,
ai_side: int,
r: Optional[int],
c: Optional[int],
raw_r: Optional[int],
raw_c: Optional[int],
corrected: bool,
used_mock: bool,
correct_reason: Optional[str] = None,
no_snap_fail: bool = False,
) -> None:
"""학습/평가용 한 수 단위 JSONL 로그.
--log-jsonl PATH 가 있으면 매 응답을 한 줄씩 기록합니다. 보드 상태(size*size 정수 배열),
AI 색, 모델이 원래 답한 (raw_r,raw_c)·실제 보낸 (r,c)·보정 여부·mock 여부 등을 남깁니다.
이후 게임 결과(승/패) 와 매칭하면 SFT/Behavior cloning 데이터로 활용 가능.
"""
path = getattr(args, "log_jsonl", None)
if not path or board is None:
return
try:
os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
rec = {
"ts": _dt.datetime.now(_dt.timezone.utc).isoformat(),
"size": size,
"ai_side": ai_side, # 1=Black, 2=White
"board": [board[i] for i in range(size * size)],
"move_row": r,
"move_col": c,
"raw_row": raw_r,
"raw_col": raw_c,
"corrected": bool(corrected),
"correct_reason": correct_reason,
"mock": bool(used_mock),
"model": getattr(args, "model_id", None),
"lora": getattr(args, "lora_dir", None),
"backend": _BACKEND,
"board_format": _BOARD_FORMAT,
}
if no_snap_fail:
rec["no_snap_fail"] = 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("log_jsonl 기록 실패: %s", e)
_KNOWN_MAGICS = (MAGIC_PING, MAGIC_OMK3, MAGIC_OMK2, MAGIC_OMKP, MAGIC_OMKL)
def _handle_omkl_message(conn: socket.socket, args: argparse.Namespace) -> None:
"""클라이언트가 어떤 모드든 *둔 수 한 개* 를 사후 로깅용으로 보내는 메시지.
OMKL body 포맷 (모두 빅엔디안):
u8(size) u8(stone) u8(row) u8(col) u8(actor) u8(game_mode) u8(status) u16(seq)
size*size bytes (둔 후 보드 상태)
stone: 1=Black, 2=White
actor: 0=human, 1=local_ai, 2=vlm
game_mode: 0=HumanVsHuman 1=HumanVsAi 2=HumanVsVlmTcp 3=AiVsAi 4=AiVsVlm 5=VlmVsVlm
status: 0=playing, 1=BlackWin, 2=WhiteWin, 3=Draw
seq: 현재 게임 안에서 1부터 증가하는 수 순번 (선택)
응답: 4바이트 'LACK'.
"""
(body_len,) = struct.unpack(">I", recv_exact(conn, 4))
if body_len > 4 * 1024 * 1024:
raise ValueError("OMKL body 너무 큼")
body = recv_exact(conn, body_len)
if len(body) < 9:
raise ValueError("OMKL body 너무 짧음")
size = body[0]
stone = body[1]
row = body[2]
col = body[3]
actor = body[4]
game_mode = body[5]
status = body[6]
(seq,) = struct.unpack(">H", body[7:9])
expected_board_len = size * size
if len(body) - 9 != expected_board_len:
raise ValueError(
f"OMKL board 길이 불일치: got {len(body) - 9}, want {expected_board_len}"
)
board = bytes(body[9:])
if size not in VALID_SIZES:
raise ValueError(f"지원하지 않는 보드 크기: {size}")
_log_omkl_record(args, size, stone, row, col, actor, game_mode, status, seq, board)
conn.sendall(MAGIC_LACK)
mode_name = _MODE_NAMES.get(game_mode, str(game_mode))
status_name = _STATUS_NAMES.get(status, str(status))
actor_name = _ACTOR_NAMES.get(actor, str(actor))
stone_name = "B" if stone == 1 else ("W" if stone == 2 else f"?{stone}")
logging.info(
"OMKL recv mode=%s actor=%s stone=%s seq=%d (r,c)=(%d,%d) status=%s",
mode_name, actor_name, stone_name, seq, row, col, status_name,
)
_maybe_trigger_auto_train(args, mode_name, status_name)
_ACTOR_NAMES = {0: "human", 1: "local_ai", 2: "vlm"}
_MODE_NAMES = {
0: "HumanVsHuman",
1: "HumanVsAi",
2: "HumanVsVlmTcp",
3: "AiVsAi",
4: "AiVsVlm",
5: "VlmVsVlm",
}
_STATUS_NAMES = {0: "Playing", 1: "BlackWin", 2: "WhiteWin", 3: "Draw"}
def _log_omkl_record(
args: argparse.Namespace,
size: int,
stone: int,
row: int,
col: int,
actor: int,
game_mode: int,
status: int,
seq: int,
board: bytes,
) -> None:
path = getattr(args, "log_jsonl", None)
if not path:
return
try:
os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
rec = {
"ts": _dt.datetime.now(_dt.timezone.utc).isoformat(),
"kind": "move",
"size": size,
"stone": stone, # 1=Black, 2=White
"move_row": row,
"move_col": col,
"actor": _ACTOR_NAMES.get(actor, str(actor)),
"mode": _MODE_NAMES.get(game_mode, str(game_mode)),
"status": _STATUS_NAMES.get(status, str(status)),
"seq": seq,
"board": [board[i] for i in range(size * size)],
}
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
except OSError as e:
logging.warning("OMKL log_jsonl 기록 실패: %s", e)
# =============================================================================
# 자동 학습 (online LoRA SFT) — VLM 이 참여하는 학습/평가 모드 게임 종료 시 트리거.
# =============================================================================
#
# 추론 모델과 학습 모델이 같은 프로세스에서 동시에 메모리에 올라가면 8B 급 VLM 은
# 거의 확실히 OOM 이다. 따라서 학습은 항상 *별도 subprocess* 로 띄우고, 끝난 뒤
# 추론 파이프라인 캐시만 무효화해서 다음 요청부터 새 LoRA 가 자동 적용되게 한다.
#
# 큐는 collapse(최신 1건만 유지) 한다. 학습 한 번이 한 게임보다 훨씬 오래 걸리는
# 경우에도 백로그가 쌓이지 않도록.
# VLM 이 참여하는 모드의 게임 종료를 학습 트리거로 본다.
_AUTO_TRAIN_TRIGGER_MODES = {"AiVsVlm", "HumanVsVlmTcp", "VlmVsVlm"}
# 무승부(Draw) 는 승리 측이 없어 dataset_omook 의 "승리 측의 수만 학습" 필터에 0건이 잡히므로
# 학습 트리거에서 제외한다. 흑승/백승 일 때만 한 라운드 학습을 돈다.
_AUTO_TRAIN_TRIGGER_STATUSES = {"BlackWin", "WhiteWin"}
_auto_train_queue: "queue.Queue[dict]" = queue.Queue()
_auto_train_lock = threading.Lock()
_auto_train_thread: Optional[threading.Thread] = None
_auto_train_busy = threading.Event()
_auto_train_games_seen = 0
def _start_auto_train_worker_if_needed(args: argparse.Namespace) -> None:
"""서버 부팅 시 한번 호출. auto-train 이 켜져 있으면 워커 데몬을 띄운다."""
global _auto_train_thread
if not getattr(args, "auto_train", False):
return
if _auto_train_thread is not None:
return
if not getattr(args, "log_jsonl", None):
logging.warning("[auto-train] --log-jsonl 가 없으면 학습 데이터가 없으므로 비활성화.")
args.auto_train = False
return
if args.mock:
logging.warning("[auto-train] --mock 모드에서는 의미가 없어 비활성화.")
args.auto_train = False
return
# 출력 LoRA 디렉토리 결정. 사용자가 --lora-dir 를 줬다면 그 경로에 덮어쓰며 회전.
if not getattr(args, "lora_dir", None):
args.lora_dir = "vlm_models/lora_omook_auto"
logging.info("[auto-train] --lora-dir 미지정 → 기본 %s 사용", args.lora_dir)
t = threading.Thread(
target=_auto_train_worker_loop,
args=(args,),
daemon=True,
name="auto-train-worker",
)
t.start()
_auto_train_thread = t
logging.info(
"[auto-train] 활성화: trigger=%s 종료 시, log=%s, lora_out=%s",
",".join(sorted(_AUTO_TRAIN_TRIGGER_MODES)),
args.log_jsonl,
args.lora_dir,
)
def _maybe_trigger_auto_train(args: argparse.Namespace, mode_name: str, status_name: str) -> None:
"""OMKL 한 건 처리 직후 호출. 학습 모드 게임이 끝났다면 큐에 신호."""
if not getattr(args, "auto_train", False):
return
if status_name == "Playing":
# 진행 중 수는 트리거 후보가 아니다 — 정적 무시.
return
if mode_name not in _AUTO_TRAIN_TRIGGER_MODES:
logging.info(
"[auto-train] 게임 종료(%s) 감지했지만 모드 %s 는 학습 트리거 대상이 아님. (대상=%s)",
status_name, mode_name, sorted(_AUTO_TRAIN_TRIGGER_MODES),
)
return
if status_name not in _AUTO_TRAIN_TRIGGER_STATUSES:
logging.info(
"[auto-train] 게임 종료(%s, mode=%s) 트리거 제외 — 흑승/백승만 학습.",
status_name, mode_name,
)
return
global _auto_train_games_seen
_auto_train_games_seen += 1
every = max(1, int(getattr(args, "auto_train_every", 1) or 1))
if _auto_train_games_seen % every != 0:
logging.info(
"[auto-train] 게임 종료 감지(%s) — every=%d 미달, 누적 %d 건 (skip)",
status_name,
every,
_auto_train_games_seen,
)
return
logging.info(
"[auto-train] 게임 종료(%s) → 학습 큐에 신호 (누적 %d 게임)",
status_name,
_auto_train_games_seen,
)
_auto_train_queue.put(
{
"ts": _dt.datetime.now(_dt.timezone.utc).isoformat(),
"mode": mode_name,
"status": status_name,
"games_seen": _auto_train_games_seen,
}
)
def _auto_train_worker_loop(args: argparse.Namespace) -> None:
"""큐에서 트리거 1건을 꺼낼 때마다 한 라운드 학습을 돌린다."""
while True:
ev = _auto_train_queue.get()
# 학습 1라운드 동안 들어온 추가 트리거는 collapse — 최신 1건만 의미가 있다.
try:
while True:
_auto_train_queue.get_nowait()
except queue.Empty:
pass
with _auto_train_lock:
_auto_train_busy.set()
try:
_run_one_auto_train_iteration(args, ev)
except subprocess.CalledProcessError as e:
logging.error(
"[auto-train] subprocess 실패: cmd=%s rc=%s",
e.cmd,
e.returncode,
)
except Exception: # pylint: disable=broad-exception-caught
logging.exception("[auto-train] 예외")
finally:
_auto_train_busy.clear()
def _run_one_auto_train_iteration(args: argparse.Namespace, ev: dict) -> None:
"""1) 데이터셋 빌드 → 2) LoRA SFT → 3) 추론 파이프라인 캐시 무효화."""
py = sys.executable or "python"
project_root = os.path.dirname(os.path.abspath(__file__))
dataset_dir = getattr(args, "auto_train_dataset_dir", None) or "datasets/omook_sft_auto"
output_dir = args.lora_dir # _start_auto_train_worker_if_needed 에서 보장.
t0 = _dt.datetime.now()
logging.info(
"[auto-train] === iteration 시작 (trigger=%s, total_games=%d) ===",
ev.get("status"),
ev.get("games_seen"),
)
# 1) 데이터셋 변환 (이미지 포함).
cmd_ds = [
py,
os.path.join(project_root, "dataset_omook.py"),
"--jsonl",
args.log_jsonl,
"--out-dir",
dataset_dir,
"--save-images",
"--workers",
str(getattr(args, "auto_train_num_workers", 2)),
]
logging.info("[auto-train] dataset: %s", " ".join(cmd_ds))
subprocess.run(cmd_ds, check=True, cwd=project_root)
# 데이터 충분한지 확인.
train_jsonl = os.path.join(dataset_dir, "train.jsonl")
n_examples = 0
try:
with open(train_jsonl, "r", encoding="utf-8") as f:
for _ in f:
n_examples += 1
except OSError:
n_examples = 0
min_n = max(1, int(getattr(args, "auto_train_min_examples", 8) or 8))
if n_examples < min_n:
logging.info(
"[auto-train] 학습 예제 %d < 최소 %d → 학습 스킵, 데이터만 갱신.",
n_examples,
min_n,
)
return
logging.info("[auto-train] 학습 예제 %d 건", n_examples)
# 2) LoRA SFT.
cmd_tr = [
py,
os.path.join(project_root, "train_vlm.py"),
"--dataset-jsonl",
train_jsonl,
"--base-model",
args.model_id,
"--output-dir",
output_dir,
"--epochs",
str(getattr(args, "auto_train_epochs", 1)),
"--batch-size",
str(getattr(args, "auto_train_batch_size", 1)),
"--grad-accum",
str(getattr(args, "auto_train_grad_accum", 8)),
"--lr",
str(getattr(args, "auto_train_lr", 2e-4)),
"--device",
args.device,
"--num-workers",
str(getattr(args, "auto_train_num_workers", 2)),
"--torch-threads",
str(getattr(args, "auto_train_torch_threads", 0)),
]
if int(getattr(args, "auto_train_max_samples", 0) or 0) > 0:
cmd_tr += ["--max-samples", str(args.auto_train_max_samples)]
logging.info("[auto-train] train: %s", " ".join(cmd_tr))
proc = subprocess.run(cmd_tr, cwd=project_root) # check=False — 발산 시 rc=2
if proc.returncode == 2:
# train_vlm.py 가 명시적으로 발산 게이트로 학습 결과를 버린 케이스.
# 디스크의 LoRA 는 그대로(이전 정상본 유지/빈 디렉토리), 추론 캐시 그대로.
logging.warning("[auto-train] 학습 발산 → LoRA 미저장. 추론은 영향 없음.")
return
if proc.returncode != 0:
raise subprocess.CalledProcessError(proc.returncode, cmd_tr)
# 3) 추론 파이프라인 캐시 무효화 → 다음 predict 시 새 LoRA 가 자동 로드된다.
global _vlm_pipe, _vlm_model_id, _vlm_lora_dir, _vlm_lora_disabled, _vlm_collapse_streak
_vlm_pipe = None
_vlm_model_id = None
_vlm_lora_dir = None
# 새 어댑터가 들어왔으니 collapse 게이트도 리셋해서 한 번 더 기회를 준다.
_vlm_lora_disabled = False
_vlm_collapse_streak = 0
elapsed = (_dt.datetime.now() - t0).total_seconds()
logging.info(
"[auto-train] === 완료 (%.1fs) — 다음 추론부터 LoRA=%s 적용 ===",
elapsed,
output_dir,
)
def handle_client(conn: socket.socket, args: argparse.Namespace, rng: random.Random) -> None:
"""한 클라이언트 연결을 끝까지 응대.
같은 연결 위에서 여러 개의 PING/OMK3/OMK2/OMKP 메시지를 순서대로 받아 처리합니다.
클라이언트가 정상/비정상으로 끊으면 EOF 가 발생해 루프를 빠져나옵니다.
framing 오류(매직 미일치, body 길이 이상 등) 가 발생하면 잔여 body 바이트가 소켓
버퍼에 남아 후속 read 가 가비지로 어긋나는 cascade 를 일으키므로 *연결을 닫아* 버립니다.
클라이언트(persistent) 는 다음 요청에서 자동으로 재접속합니다.
"""
msg_count = 0
try:
while True:
try:
head = recv_exact(conn, 4)
except PeerClosed as e:
if msg_count == 0:
# 헬스체크·짧은 프로브가 메시지를 안 보내고 끊는 경우가 흔함.
logging.debug("eof: 연결 끊김 (메시지 0건, %s)", e)
else:
logging.info("client closed (메시지 %d건 처리, %s)", msg_count, e)
return
if head not in _KNOWN_MAGICS:
logging.warning(
"알 수 없는 헤더 %r → 프레이밍 깨짐, 연결 종료 (클라이언트 재접속 권장)",
head,
)
try:
send_err(conn, f"unknown magic {head!r}; reconnect")
except OSError:
pass
return
try:
_handle_one_message(head, conn, args, rng)
msg_count += 1
except PeerClosed as e:
# 응답 보내는 도중 또는 본문 읽는 도중 연결이 끊겼다 — 조용히 종료.
logging.info(
"client disconnected during message (메시지 %d건 처리, %s)",
msg_count, e,
)
return
except (BrokenPipeError, ConnectionResetError) as e:
# send_ok / send_err / sendall 등에서 직접 올라온 disconnect.
logging.info(
"client disconnected (write side, 메시지 %d건 처리, %s)",
msg_count, e,
)
return
except (EOFError, struct.error, ValueError) as e:
# framing 계열 오류 → 잔여 바이트가 어긋났을 가능성 큼 → 연결 종료.
logging.exception("framing 오류 → 연결 종료: %s", e)
try:
send_err(conn, f"framing error: {e}")
except OSError:
pass
return
except Exception as e: # pylint: disable=broad-exception-caught
# 메시지는 정상 수신했으나 모델/응답 단계에서 발생한 오류는 연결 유지.
logging.exception("handle_client (메시지 처리 실패)")
try:
send_err(conn, str(e))
except OSError:
return
continue
finally:
try:
conn.shutdown(socket.SHUT_RDWR)
except OSError:
pass
conn.close()
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(
"바인드 실패: tcp://%s:%s 포트가 이미 사용 중입니다 (Errno %s).\n"
" 이전에 켠 vlm 서버/mock을 종료하거나, 다른 터미널 프로세스를 끝내세요.\n"
" 예: macOS/Linux — lsof -i :%s 후 PID를 kill 하거나,\n"
" kill \"$(lsof -t -i :%s)\"\n"
" 또는 — --port 9300 처럼 다른 포트를 지정하고 오목 앱의 VLM 포트도 같이 변경하세요.",
args.host,
args.port,
e.errno,
args.port,
args.port,
)
raise SystemExit(1) from e
raise
srv.listen(8)
logging.info(
"%slistening on tcp://%s:%s%s",
"[MOCK] " if args.mock else "",
args.host,
args.port,
"" if args.mock else f", model={args.model_id!r}",
)
try:
while True:
c, addr = srv.accept()
logging.info("client %s", addr)
t = threading.Thread(target=handle_client, args=(c, args, rng), daemon=True)
t.start()
finally:
srv.close()
def main() -> None:
parser = argparse.ArgumentParser(description="Omok VLM TCP server (PvA)")
parser.add_argument("--host", default="127.0.0.1", help="바인드 주소")
parser.add_argument("--port", type=int, default=9299)
parser.add_argument("--mock", action="store_true", help="VLM 없이 빈 칸 무작위 (테스트용)")
parser.add_argument(
"--model-id",
default="Qwen/Qwen3-VL-2B-Instruct",
help=(
"Hugging Face 모델 id 또는 로컬 디렉토리 (transformers image-text-to-text 파이프라인). "
"로컬 다운로드 권장: python download_qwen3vl.py 실행 후 "
"vlm_models/Qwen__Qwen3-VL-2B-Instruct 경로 사용."
),
)
parser.add_argument("--device", default="cpu", help="cuda | mps | cpu")
parser.add_argument(
"--lora-dir",
default=None,
help=(
"지정하면 base 모델 위에 PEFT LoRA 어댑터를 얹어 추론합니다. "
"train_vlm.py 의 --output-dir 결과를 그대로 넘기면 됩니다."
),
)
parser.add_argument("--seed", type=int, default=0)
parser.add_argument(
"--strict",
action="store_true",
help="VLM 응답이 점유 칸이거나 호출이 실패해도 폴백 없이 에러를 그대로 클라이언트에 보낸다.",
)
parser.add_argument(
"--vlm-illegal-retries",
type=int,
default=2,
help=(
"보드가 있을 때 모델이 범위 밖·점유 칸을 내면 추가로 그만큼 재추론합니다 "
"(기본 2 → 최대 3번 생성). 모두 실패하면 인접 빈 칸 보정으로 폴백."
),
)
parser.add_argument(
"--no-snap",
action="store_true",
help="재추론까지 불법이면 인접 빈 칸으로 옮기지 않고 실패 응답을 보낸다 (로그 no_snap_fail).",
)
parser.add_argument(
"--backend",
choices=("transformers", "mlx"),
default="transformers",
help="mlx: mlx-vlm 텍스트 전용 추론 (coord_omook.py 좌표 LoRA). mlx-vlm 있는 python 으로 실행.",
)
parser.add_argument(
"--board-format",
choices=("legacy", "aligned"),
default="legacy",
help="aligned: 열 번호와 칸이 같은 3글자 간격 (좌표 LoRA 는 aligned 로 학습).",
)
parser.add_argument(
"--log-jsonl",
default=None,
help=(
"지정하면 매 (board, ai_side, move) 를 JSONL 로 기록합니다. "
"AI 대 VLM 모드에서 대량 대국을 돌리고, 이후 결과와 결합해 SFT/Behavior cloning "
"학습 데이터로 활용할 수 있습니다."
),
)
# ------- 자동 학습 (online LoRA) -------
parser.add_argument(
"--auto-train",
action="store_true",
help=(
"HumanVsVlmTcp / AiVsVlm / VlmVsVlm 등 학습·평가 모드에서 게임이 종료될 때마다 "
"누적 JSONL 로 LoRA 학습을 1라운드 자동 실행합니다. 학습은 별도 subprocess 로 돌고, "
"끝나면 추론 파이프라인이 새 어댑터로 자동 갱신됩니다. --log-jsonl 필수."
),
)
parser.add_argument(
"--auto-train-every",
type=int,
default=1,
help="N 게임마다 1번 학습 (기본: 매 게임).",
)
parser.add_argument(
"--auto-train-min-examples",
type=int,
default=8,
help="학습 예제가 N건 미만이면 데이터만 갱신하고 학습은 스킵.",
)
parser.add_argument(
"--auto-train-max-samples",
type=int,
default=0,
help="0=전체. 한 라운드에 최대 N건만 사용 (속도/메모리 조절).",
)
parser.add_argument("--auto-train-epochs", type=int, default=1)
parser.add_argument("--auto-train-batch-size", type=int, default=1)
parser.add_argument("--auto-train-grad-accum", type=int, default=8)
parser.add_argument(
"--auto-train-lr",
type=float,
default=5e-5,
help="auto-train LR. 8B+ collapse 방지를 위해 5e-5 권장 (기본).",
)
parser.add_argument(
"--auto-train-num-workers",
type=int,
default=max(2, (os.cpu_count() or 4) // 2),
help="train_vlm.py / dataset_omook.py 의 워커 프로세스/스레드 수.",
)
parser.add_argument(
"--auto-train-torch-threads",
type=int,
default=0,
help="train_vlm.py 의 torch.set_num_threads(N). 0=torch 기본값.",
)
parser.add_argument(
"--auto-train-dataset-dir",
default="sft_data/omook_sft_auto",
help=(
"dataset_omook.py 의 출력 디렉토리. 'datasets' 로 시작하는 이름은 "
"huggingface `datasets` 패키지와 import 충돌이 나므로 피하세요."
),
)
parser.add_argument("-v", action="store_true", help="verbose")
args = parser.parse_args()
logging.basicConfig(
level=logging.DEBUG if args.v else logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
global _BACKEND, _BOARD_FORMAT
_BACKEND = args.backend
_BOARD_FORMAT = args.board_format
if _BACKEND == "mlx" and args.auto_train:
raise SystemExit("--auto-train 은 transformers 학습(train_vlm.py) 전용이라 --backend mlx 와 함께 쓸 수 없습니다.")
logging.info(
"backend=%s board_format=%s no_snap=%s model=%s lora=%s",
_BACKEND, _BOARD_FORMAT, args.no_snap, args.model_id, args.lora_dir,
)
if not args.mock:
try:
get_vlm_pipeline(args.model_id, args.device, lora_dir=args.lora_dir)
except Exception as e:
logging.error("모델 로드 실패: %s — --mock 으로 재시도하세요.", e)
raise
_start_auto_train_worker_if_needed(args)
serve(args)
if __name__ == "__main__":
main()