예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
server.py
"""19×19 바둑(GO19) VLM TCP 서버.
C++ 클라이언트(`baduk`) 가 보내는 GO19 프레임을 받아, Gemma 4 (mlx-vlm) 에게
보드 이미지를 보고 다음 한 수의 좌표를 결정하게 한 뒤 RPLY 로 응답합니다.
자세한 프로토콜은 README 참고.
"""
from __future__ import annotations
import io
import json
import logging
import os
import queue
import re
import socket
import socketserver
import struct
import threading
import time
import zlib
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal, Optional, Tuple
BoardRepr = Literal["chars", "numbers", "both", "compact", "aligned", "aligned_read", "aligned_read2"]
MoveBackend = Literal["vlm", "policy", "hybrid"]
try:
from PIL import Image
except ImportError as e: # pragma: no cover
raise SystemExit("Pillow 필요: pip install Pillow") from e
from .cross_lessons import heuristic_shape_delta, lesson_bonus_r5, working_chase_bonus_r5
from .ladder import chase_verdict, futile_escapes
from .shapes import hogu_verdict
# ---------- 프로토콜 상수 ----------
MAGIC_GO19 = b"GO19"
MAGIC_RPLY = b"RPLY"
MAGIC_PING = b"PING"
MAGIC_PONG = b"PONG"
BOARD_SIZE = 19 # 본 서버는 19x19 만 지원
LOGGER = logging.getLogger("vlm_server")
# ---------- 모델 어댑터 ----------
#
# MLX 의 GPU 스트림은 *스레드 로컬* 입니다. 또한 mlx-vlm 의 `generate()` 는
# 스레드-세이프 하지 않습니다. 따라서 TCP 워커에서 직접 추론을 호출하면
# "There is no Stream(gpu, 0) in current thread" 등이 발생합니다.
#
# 그래서 추론은 **단일 추론 스레드**에서만 실행하고, 워커들은 큐로
# 작업을 넘기고 결과를 받아 가는 구조로 만듭니다.
@dataclass
class _InferenceJob:
image: Image.Image
board: bytes
ai_side: int
result_q: "queue.Queue[Tuple[Optional[int], Optional[int], bool, Optional[str]]]"
forbidden: Optional[list[tuple[int, int]]] = None
illegal_hint: Optional[str] = None
@dataclass
class _RawInferenceJob:
"""임의 system/user 프롬프트로 텍스트만 생성 (보드 인식 검증 등)."""
system_prompt: str
user_prompt: str
image: Optional[Image.Image]
num_images: int
result_q: "queue.Queue[Tuple[Optional[str], Optional[str]]]"
max_tokens: Optional[int] = None
temperature: Optional[float] = None
class _MlxVlmModel:
"""단일 추론 스레드를 가진 mlx-vlm 어댑터.
`start()` 로 추론 스레드를 띄워 모델을 로드하고, `submit()` 으로 작업을 큐잉.
결과는 `(row, col, is_pass, error_msg or None)` 튜플로 반환.
"""
def __init__(
self,
model_id: str,
max_new_tokens: int = 96,
temperature: float = 0.2,
lora_dir: Optional[str] = None,
text_only_moves: bool = False,
board_repr: BoardRepr = "both",
):
self.model_id = model_id
self.max_new_tokens = max_new_tokens
self.temperature = temperature
self.lora_dir = lora_dir
self.text_only_moves = text_only_moves
self.board_repr = board_repr
self._jobs: "queue.Queue[Optional[_InferenceJob]]" = queue.Queue()
self._raw_jobs: "queue.Queue[Optional[_RawInferenceJob]]" = queue.Queue()
self._ready_event = threading.Event()
self._load_error: Optional[BaseException] = None
self._thread: Optional[threading.Thread] = None
self._stop = False
# 추론 스레드에서만 채워질 핸들들.
self._model = None
self._processor = None
self._config = None
# 메인 스레드에서 호출. 추론 스레드를 시작하고 모델 로드 완료까지 기다림.
def start(self, wait_load: bool = True, load_timeout_s: float = 600.0) -> None:
if self._thread is not None:
return
# 모델 로드(수십 초) 들어가기 전에 LoRA 경로 미리 검증해서 즉시 친절히 실패.
if self.lora_dir:
self._validate_lora_dir(self.lora_dir)
t = threading.Thread(target=self._run_loop, name="mlx-inference", daemon=True)
self._thread = t
t.start()
if wait_load:
if not self._ready_event.wait(timeout=load_timeout_s):
raise RuntimeError("모델 로드 타임아웃")
if self._load_error is not None:
raise self._load_error
def stop(self) -> None:
self._stop = True
try:
self._jobs.put_nowait(None)
except Exception:
pass
try:
self._raw_jobs.put_nowait(None)
except Exception:
pass
def predict_raw(
self,
system_prompt: str,
user_prompt: str,
*,
image: Optional[Image.Image] = None,
num_images: int = 0,
timeout_s: float = 300.0,
max_tokens: Optional[int] = None,
temperature: Optional[float] = None,
) -> str:
"""임의 프롬프트로 생성. num_images>0 이면 image 필수."""
if self._thread is None or not self._thread.is_alive():
raise RuntimeError("추론 스레드가 실행 중이 아닙니다")
rq: "queue.Queue[Tuple[Optional[str], Optional[str]]]" = queue.Queue(maxsize=1)
self._raw_jobs.put(
_RawInferenceJob(
system_prompt=system_prompt,
user_prompt=user_prompt,
image=image,
num_images=num_images,
result_q=rq,
max_tokens=max_tokens,
temperature=temperature,
)
)
try:
text, err = rq.get(timeout=timeout_s)
except queue.Empty:
raise TimeoutError(f"predict_raw 타임아웃 ({timeout_s:.0f}s)")
if err:
raise RuntimeError(err)
return text or ""
# 메인/워커 어디서든 호출 가능. 결과 큐에 결과 도착할 때까지 블록.
def predict_move(
self, image: Image.Image, board: bytes, ai_side: int, timeout_s: float = 120.0
, forbidden: Optional[list[tuple[int, int]]] = None
, illegal_hint: Optional[str] = None
) -> Tuple[Optional[int], Optional[int], bool]:
if self._thread is None or not self._thread.is_alive():
raise RuntimeError("추론 스레드가 실행 중이 아닙니다")
rq: "queue.Queue[Tuple[Optional[int], Optional[int], bool, Optional[str]]]" = (
queue.Queue(maxsize=1)
)
self._jobs.put(
_InferenceJob(
image=image,
board=board,
ai_side=ai_side,
result_q=rq,
forbidden=forbidden,
illegal_hint=illegal_hint,
)
)
try:
row, col, is_pass, err = rq.get(timeout=timeout_s)
except queue.Empty:
raise TimeoutError(f"추론 타임아웃 ({timeout_s:.0f}s)")
if err:
raise RuntimeError(err)
return row, col, is_pass
# ---------- 추론 스레드 ----------
def _run_loop(self) -> None:
# 1) 이 스레드에서 모델/모듈을 모두 import + load 한다.
try:
t0 = time.time()
LOGGER.info("loading model: %s", self.model_id)
from mlx_vlm import generate as _generate, load as _load
from mlx_vlm.prompt_utils import apply_chat_template as _apply_chat_template
from mlx_vlm.utils import load_config as _load_config
# mlx-vlm 일부 버전은 load(..., adapter_path=) 가 base weight 와
# 섞이며 Missing parameters 를 냄 → 항상 base 먼저 로드 후 LoRA 수동 적용.
self._model, self._processor = _load(self.model_id)
if self.lora_dir:
self._apply_lora_manually(self.lora_dir)
LOGGER.info("LoRA 어댑터 적용: %s", self.lora_dir)
try:
self._config = _load_config(self.model_id)
except Exception:
self._config = None
LOGGER.info("model loaded in %.1fs", time.time() - t0)
except BaseException as e:
self._load_error = e
LOGGER.exception("모델 로드 실패")
self._ready_event.set()
return
self._ready_event.set()
# 2) 작업 처리 루프 (raw 작업 우선 처리)
while not self._stop:
while True:
try:
rj = self._raw_jobs.get_nowait()
except queue.Empty:
break
if rj is None:
self._stop = True
break
try:
text, err = self._infer_raw(rj, _generate, _apply_chat_template)
rj.result_q.put((text, err))
except BaseException as e:
LOGGER.warning("predict_raw 실패: %s", e)
rj.result_q.put((None, str(e)))
if self._stop:
break
try:
job = self._jobs.get(timeout=0.25)
except queue.Empty:
continue
if job is None:
break
try:
row, col, is_pass = self._infer_one(job, _generate, _apply_chat_template)
job.result_q.put((row, col, is_pass, None))
except BaseException as e:
LOGGER.warning("추론 실패: %s", e)
job.result_q.put((None, None, False, str(e)))
def _validate_lora_dir(self, lora_dir: str) -> None:
"""LoRA 디렉토리가 유효한지 미리 검사하고 친절한 에러를 던진다."""
p = Path(lora_dir)
if not p.exists():
raise FileNotFoundError(
f"--lora-dir 가 가리키는 폴더가 존재하지 않습니다: {p}\n"
f" 먼저 학습을 실행하세요:\n"
f" python -m vlm_server.train_lora --output-dir {p} ..."
)
cfg = p / "adapter_config.json"
# adapters.safetensors 또는 N자릿수 체크포인트 둘 다 인정
adapters = list(p.glob("*adapters.safetensors"))
if not cfg.exists() or not adapters:
existing = sorted(x.name for x in p.iterdir())
raise FileNotFoundError(
f"LoRA 어댑터 파일이 누락되었습니다: {p}\n"
f" 필요한 파일: adapter_config.json + adapters.safetensors\n"
f" 현재 폴더 내용: {existing or '(비어 있음)'}\n"
f" 원인: LoRA 학습이 끝나지 않았거나 도중에 실패했을 가능성이 높습니다.\n"
f" 해결: 학습을 (다시) 끝까지 실행하세요. 예)\n"
f" python -m vlm_server.train_lora \\\n"
f" --dataset-jsonl sft_data/<your>/train.jsonl \\\n"
f" --val-jsonl sft_data/<your>/val.jsonl \\\n"
f" --output-dir {p} \\\n"
f" --epochs 3 --batch-size 1 --grad-accum 16 \\\n"
f" --lr 1e-4 --lora-rank 16 --lora-alpha 32 --grad-checkpoint"
)
def _apply_lora_manually(self, lora_dir: str) -> None:
"""mlx-vlm: apply_lora_layers 가 어댑터 가중치까지 로드한다 (반환 모델 재할당 필요)."""
try:
from mlx_vlm.trainer.utils import apply_lora_layers # type: ignore
except Exception as e:
LOGGER.error("LoRA import 실패 (%s) — base 모델로만 추론합니다.", e)
return
try:
updated = apply_lora_layers(self._model, lora_dir)
if updated is not None:
self._model = updated
LOGGER.info("LoRA layers applied from %s", lora_dir)
except Exception as e:
LOGGER.error("LoRA 가중치 로드 실패: %s — base 모델로만 추론합니다.", e)
def _infer_raw(
self,
job: _RawInferenceJob,
_generate,
_apply_chat_template,
) -> Tuple[str, Optional[str]]:
prompt = f"{job.system_prompt}\n\n{job.user_prompt}"
num_images = max(0, int(job.num_images))
mt = job.max_tokens if job.max_tokens is not None else max(512, self.max_new_tokens * 4)
temp = job.temperature if job.temperature is not None else self.temperature
try:
formatted = _apply_chat_template(
self._processor, self._config, prompt, num_images=num_images
)
except TypeError:
formatted = _apply_chat_template(self._processor, prompt, num_images=num_images)
gen_kw: dict = dict(max_tokens=mt, verbose=False)
if temp is not None:
gen_kw["temperature"] = temp
def _run_gen(with_image: bool) -> object:
kw = dict(gen_kw)
if with_image and job.image is not None:
try:
return _generate(
self._model,
self._processor,
formatted,
image=[job.image],
**kw,
)
except TypeError:
kw.pop("temperature", None)
return _generate(
self._model,
self._processor,
formatted,
image=[job.image],
**kw,
)
try:
return _generate(self._model, self._processor, formatted, **kw)
except TypeError:
kw.pop("temperature", None)
return _generate(self._model, self._processor, formatted, **kw)
try:
if num_images > 0 and job.image is not None:
result = _run_gen(with_image=True)
else:
result = _run_gen(with_image=False)
except TypeError:
gen_kw.pop("temperature", None)
if num_images > 0 and job.image is not None:
result = _generate(
self._model,
self._processor,
formatted,
image=[job.image],
max_tokens=mt,
verbose=False,
)
else:
result = _generate(
self._model, self._processor, formatted, max_tokens=mt, verbose=False
)
text = getattr(result, "text", None)
if text is None and isinstance(result, str):
text = result
text = text or ""
LOGGER.debug("predict_raw 응답 앞부분: %s", text[:400])
return text, None
def _infer_one(
self,
job: _InferenceJob,
_generate,
_apply_chat_template,
) -> Tuple[Optional[int], Optional[int], bool]:
user_prompt = build_prompt(
job.board,
job.ai_side,
forbidden=job.forbidden,
illegal_hint=job.illegal_hint,
board_repr=self.board_repr,
)
if self.board_repr == "aligned":
system = SYSTEM_PROMPT_ALIGNED
elif self.board_repr == "aligned_read":
system = SYSTEM_PROMPT_ALIGNED_READ
elif self.board_repr == "aligned_read2":
system = SYSTEM_PROMPT_ALIGNED_READ2
elif self.board_repr == "compact":
system = SYSTEM_PROMPT_COMPACT
elif self.text_only_moves:
system = SYSTEM_PROMPT_TEXT_GRID
else:
system = SYSTEM_PROMPT_MULTI
prompt = f"{system}\n\n{user_prompt}"
num_images = 0 if self.text_only_moves else 1
try:
formatted = _apply_chat_template(
self._processor, self._config, prompt, num_images=num_images
)
except TypeError:
formatted = _apply_chat_template(self._processor, prompt, num_images=num_images)
gen_kw = dict(
max_tokens=self.max_new_tokens,
temperature=self.temperature,
verbose=False,
)
try:
if self.text_only_moves:
try:
result = _generate(
self._model,
self._processor,
formatted,
**gen_kw,
)
except TypeError:
gen_kw.pop("temperature", None)
result = _generate(
self._model,
self._processor,
formatted,
**gen_kw,
)
else:
try:
result = _generate(
self._model,
self._processor,
formatted,
image=[job.image],
**gen_kw,
)
except TypeError:
gen_kw.pop("temperature", None)
result = _generate(
self._model,
self._processor,
formatted,
image=[job.image],
**gen_kw,
)
except TypeError:
# 일부 버전: image=[] 또는 키워드 미지원
gen_kw.pop("temperature", None)
if self.text_only_moves:
result = _generate(self._model, self._processor, formatted, **gen_kw)
else:
result = _generate(
self._model,
self._processor,
formatted,
image=[job.image],
**gen_kw,
)
text = getattr(result, "text", None)
if text is None and isinstance(result, str):
text = result
text = text or ""
self.last_raw_text = text
LOGGER.debug("VLM raw: %s", text[:240])
return parse_move(text)
# ---------- 프롬프트/파싱 ----------
# 멀티모달(보드 PNG + 텍스트 격자) 기본 안내.
SYSTEM_PROMPT_MULTI = (
"당신은 19×19 바둑 (Go) 의 강력한 플레이어입니다. "
"보드 이미지(있으면 참고용)와 아래 텍스트 격자를 함께 보고 다음 단 한 수의 좌표만 골라야 합니다. "
"(텍스트 격자가 이미지와 어긋나면 텍스트 격자를 우선하세요.) "
"좌표는 0-indexed 정수 (row, col) 이며 각각 0..18 입니다. "
"착수 가능한 자리(값 0 / 문자 '.') 에서만 골라야 하며, 자살수/패(코) 위반은 피하세요. "
"응답은 다른 설명 없이 JSON 한 줄로만 출력하세요."
)
# 이미지 없이 격자 텍스트만으로 판단.
SYSTEM_PROMPT_TEXT_GRID = (
"당신은 19×19 바둑 (Go) 의 강력한 플레이어입니다. "
"이번 요청에서는 보드 상태가 텍스트로만 주어집니다. 아래 숫자/문자 격자를 정확히 읽고 "
"다음 단 한 수의 좌표만 고르세요. "
"숫자 격자: 각 칸은 0=빈칸, 1=흑(PB), 2=백(PW). "
"문자 격자: .=빈칸 X=흑 O=백. "
"좌표는 0-indexed 정수 (row, col) 이며 각각 0..18 입니다. "
"항상 빈 칸(0 또는 '.') 에만 두세요. 자살수도 피하세요. "
"응답은 다른 설명 없이 JSON 한 줄로만 출력하세요."
)
# baduk_v2 SFT 와 맞춘 compact 격자용 (숫자 격자 없음, 학습 프롬프트에 가깝게).
SYSTEM_PROMPT_COMPACT = (
"당신은 19×19 바둑(Go) 의 강력한 플레이어입니다. "
"주어진 보드 이미지와 격자 데이터를 보고 다음 단 한 수의 좌표만 골라야 합니다. "
"좌표는 0-indexed 정수 (row, col) 이며 각각 0..18 입니다. "
"응답은 다른 설명 없이 JSON 한 줄로만 출력하세요."
)
# baduk_v3 aligned + text-only 권장.
SYSTEM_PROMPT_ALIGNED = (
"당신은 19×19 바둑(Go) 의 강력한 플레이어입니다. "
"보드 상태가 텍스트 격자로만 주어집니다. 격자를 정확히 읽고 "
"다음 단 한 수의 좌표만 고르세요. "
"문자 격자: .=빈칸 X=흑 O=백. 각 칸은 두 글자 폭이며 기호는 왼쪽에 있습니다. "
"좌표는 0-indexed 정수 (row, col) 이며 각각 0..18 입니다. "
"빈 칸(.)에만 두세요. 자살수는 피하세요. "
"응답은 다른 설명 없이 JSON 한 줄로만 출력하세요."
)
# aligned 격자 + 답할 때 고른 행의 내용을 먼저 옮겨 적고 그 행의 빈 칸에서 열을 고른다.
SYSTEM_PROMPT_ALIGNED_READ = (
"당신은 19×19 바둑(Go) 의 강력한 플레이어입니다. "
"보드 상태가 텍스트 격자로만 주어집니다. 격자를 정확히 읽고 "
"다음 단 한 수의 좌표만 고르세요. "
"문자 격자: .=빈칸 X=흑 O=백. 각 칸은 두 글자 폭이며 기호는 왼쪽에 있습니다. "
"좌표는 0-indexed 정수 (row, col) 이며 각각 0..18 입니다. "
"먼저 둘 행(row)을 정하고, 그 행의 19칸을 '열번호+기호' 로 line 에 그대로 옮겨 적은 뒤, "
"line 에서 기호가 '.' 인 열 중에서 col 을 고르세요. "
"응답은 다른 설명 없이 JSON 한 줄로만 출력하세요."
)
# aligned_read 에 빈 열 목록을 한 단계 더 둔다. col 은 그 목록 안에서만 고른다.
SYSTEM_PROMPT_ALIGNED_READ2 = (
"당신은 19×19 바둑(Go) 의 강력한 플레이어입니다. "
"보드 상태가 텍스트 격자로만 주어집니다. 격자를 정확히 읽고 "
"다음 단 한 수의 좌표만 고르세요. "
"문자 격자: .=빈칸 X=흑 O=백. 각 칸은 두 글자 폭이며 기호는 왼쪽에 있습니다. "
"좌표는 0-indexed 정수 (row, col) 이며 각각 0..18 입니다. "
"먼저 둘 행(row)을 정하고, 그 행의 19칸을 '열번호+기호' 로 line 에 그대로 옮겨 적으세요. "
"그 다음 line 에서 기호가 '.' 인 열 번호를 empty 에 빠짐없이 적고, "
"col 은 empty 안에 있는 열만 고르세요. 돌이 있는 열은 고르지 마세요. "
"응답은 다른 설명 없이 JSON 한 줄로만 출력하세요."
)
READ_PICK_INSTRUCTION = (
"다음 수를 JSON 한 줄로 답하세요. row → line(그 행 19칸을 열번호+기호로 그대로) → col 순서이며, "
"col 은 line 에서 '.' 인 열이어야 합니다."
)
READ_PICK_INSTRUCTION2 = (
"다음 수를 JSON 한 줄로 답하세요. row → line(그 행 19칸을 열번호+기호로 그대로) → "
"empty(line 에서 '.' 인 열 번호 전부) → col 순서이며, col 은 empty 안의 열이어야 합니다."
)
def board_numeric_grid(board: bytes) -> str:
"""각 행을 '0,1,2,...' 콤마 구분 문자열."""
rows = []
for r in range(BOARD_SIZE):
cells = []
for c in range(BOARD_SIZE):
v = board[r * BOARD_SIZE + c]
cells.append(str(int(v)))
rows.append(f"{r:2d}: " + ",".join(cells))
return "\n".join(rows)
def packed_board_ascii_for_log(board: bytes, size: int = BOARD_SIZE) -> str:
"""테스트·디버그용: dataset_baduk.board_text 와 동일 형식(열 00..18, 칸 공백 구분)."""
from .dataset_baduk import board_text
return board_text(board, size)
def build_prompt(
board: bytes,
ai_side: int,
forbidden: Optional[list[tuple[int, int]]] = None,
illegal_hint: Optional[str] = None,
board_repr: BoardRepr = "both",
) -> str:
"""user 메시지 본문(텍스트). image 는 멀티모달일 때만 별도 인자로 전달됩니다."""
side_name = "흑" if ai_side == 1 else "백"
from .dataset_baduk import board_text as _board_text_grid
forbidden_note = ""
if forbidden:
banned = ", ".join(f"({r},{c})" for (r, c) in forbidden[:16])
more = "" if len(forbidden) <= 16 else f" ...(+{len(forbidden) - 16})"
forbidden_note = (
"\n\n중요: 아래 좌표들은 금지입니다(이미 불법 판정 또는 제외 요청). "
f"절대 고르지 마세요: {banned}{more}\n"
)
illegal_note = ""
if illegal_hint:
illegal_note = (
"\n\n직전 응답은 불법이었습니다. 같은 실수를 반복하지 마세요:\n"
f"- {illegal_hint}\n"
"반드시 '빈 점'이며 '자살수/패 위반이 아닌' 좌표만 고르세요.\n"
)
board_blocks: list[str] = []
if board_repr == "aligned":
from .dataset_baduk import board_text_aligned, _example_move_json
side_tag = "흑(X)" if ai_side == 1 else "백(O)"
example = _example_move_json(board)
return (
f"당신의 차례: {side_tag}.\n"
"보드 문자 격자 (X=흑, O=백, .=빈, 행=row, 열=col, 0-indexed).\n"
"첫 줄·각 칸은 두 글자 폭으로 정렬되어 있습니다:\n"
f"{board_text_aligned(board)}\n\n"
f"{forbidden_note}"
f"{illegal_note}"
f"다음 수를 JSON 한 줄로만 답하세요. 예: {example} "
'혹은 패스를 원하면 {"pass": true} 만 출력.'
)
if board_repr == "aligned_read":
from .dataset_baduk import board_text_aligned, _example_read_json
side_tag = "흑(X)" if ai_side == 1 else "백(O)"
return (
f"당신의 차례: {side_tag}.\n"
"보드 문자 격자 (X=흑, O=백, .=빈, 행=row, 열=col, 0-indexed).\n"
"첫 줄·각 칸은 두 글자 폭으로 정렬되어 있습니다:\n"
f"{board_text_aligned(board)}\n\n"
f"{forbidden_note}"
f"{illegal_note}"
f"{READ_PICK_INSTRUCTION} 형식: {_example_read_json()} "
'혹은 패스를 원하면 {"pass": true} 만 출력.'
)
if board_repr == "aligned_read2":
from .dataset_baduk import board_text_aligned, _example_read2_json
side_tag = "흑(X)" if ai_side == 1 else "백(O)"
return (
f"당신의 차례: {side_tag}.\n"
"보드 문자 격자 (X=흑, O=백, .=빈, 행=row, 열=col, 0-indexed).\n"
"첫 줄·각 칸은 두 글자 폭으로 정렬되어 있습니다:\n"
f"{board_text_aligned(board)}\n\n"
f"{forbidden_note}"
f"{illegal_note}"
f"{READ_PICK_INSTRUCTION2} 형식: {_example_read2_json()} "
'혹은 패스를 원하면 {"pass": true} 만 출력.'
)
if board_repr == "compact":
from .dataset_baduk import board_text_compact, _example_move_json
side_tag = "흑(X)" if ai_side == 1 else "백(O)"
example = _example_move_json(board)
return (
f"당신의 차례: {side_tag}.\n"
"보드 (X=흑, O=백, .=빈, 행=row, 열=col, 0-indexed):\n"
f"{board_text_compact(board)}\n\n"
f"{forbidden_note}"
f"{illegal_note}"
f"다음 수를 JSON 한 줄로만 답하세요. 예: {example} "
'혹은 패스를 원하면 {"pass": true} 만 출력.'
)
if board_repr in ("chars", "both"):
board_blocks.append(
"보드 문자 격자 (X=흑, O=백, .=빈). "
"첫 줄은 열 번호 00~18, 각 데이터 줄은 행 번호 뒤에 공백으로 구분된 정확히 19칸입니다:\n"
f"{_board_text_grid(board)}\n"
)
if board_repr in ("numbers", "both"):
board_blocks.append(
"보드 숫자 격자 (0=빈, 1=흑, 2=백). 각 줄은 행번호: 쉼표로 구분된 19 칸입니다.\n"
f"{board_numeric_grid(board)}"
)
board_section = ""
if board_blocks:
board_section = "\n\n".join(board_blocks)
return (
f"당신의 차례: {side_name} ({ 'X' if ai_side==1 else 'O' }).\n"
f"{board_section}\n\n"
f"{forbidden_note}"
f"{illegal_note}"
'다음 수를 JSON 한 줄로만 답하세요. 예: {"row": 3, "col": 15} '
'혹은 패스를 원하면 {"pass": true} 만 출력.'
)
_JSON_RE = re.compile(r"\{[^{}]*\}", re.DOTALL)
_RC_RE = re.compile(r"row\D+(\d+)\D+col\D+(\d+)", re.IGNORECASE)
def parse_move(text: str) -> Tuple[Optional[int], Optional[int], bool]:
"""모델 응답 텍스트에서 (row, col, is_pass) 추출."""
if not text:
return None, None, False
# 1) 패스 키워드
if re.search(r'"pass"\s*:\s*true', text, re.IGNORECASE):
return None, None, True
if re.search(r"\bpass\b", text, re.IGNORECASE) and not re.search(r"\d", text):
return None, None, True
# 2) JSON 시도
for m in _JSON_RE.finditer(text):
chunk = m.group(0)
try:
obj = json.loads(chunk)
except Exception:
continue
if not isinstance(obj, dict):
continue
if obj.get("pass") is True:
return None, None, True
if "row" in obj and "col" in obj:
try:
r = int(obj["row"])
c = int(obj["col"])
if 0 <= r < BOARD_SIZE and 0 <= c < BOARD_SIZE:
return r, c, False
except Exception:
continue
# 3) feeble fallback: row=, col=
m = _RC_RE.search(text)
if m:
try:
r = int(m.group(1))
c = int(m.group(2))
if 0 <= r < BOARD_SIZE and 0 <= c < BOARD_SIZE:
return r, c, False
except Exception:
pass
return None, None, False
# ---------- Go 합법수 검증 (서버 측 안전망) ----------
class _MiniGo:
"""서버측 합법수 검증/안전 fallback 용 미니 Go 보드.
위치 슈퍼코는 검사하지 않음 (클라이언트가 어차피 거부함).
자살수와 점유만 검사하여 fallback 시 안전한 합법수를 고름.
"""
DR = (-1, 1, 0, 0)
DC = (0, 0, -1, 1)
def __init__(self, board_bytes: bytes, size: int = BOARD_SIZE):
assert len(board_bytes) == size * size
self.size = size
self.cells = bytearray(board_bytes)
def at(self, r: int, c: int) -> int:
return self.cells[r * self.size + c]
def _set(self, r: int, c: int, v: int) -> None:
self.cells[r * self.size + c] = v
def _group(self, r: int, c: int) -> Tuple[list[int], int]:
color = self.at(r, c)
if color == 0:
return [], 1
N = self.size
seen = bytearray(N * N)
stones: list[int] = []
libs = 0
lib_seen = bytearray(N * N)
stack = [r * N + c]
seen[r * N + c] = 1
while stack:
idx = stack.pop()
stones.append(idx)
rr = idx // N
cc = idx % N
for d in range(4):
nr = rr + self.DR[d]
nc = cc + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
nidx = nr * N + nc
v = self.cells[nidx]
if v == 0:
if not lib_seen[nidx]:
lib_seen[nidx] = 1
libs += 1
elif v == color and not seen[nidx]:
seen[nidx] = 1
stack.append(nidx)
return stones, libs
def illegal_reason(self, r: int, c: int, who: int) -> Optional[str]:
N = self.size
if not (0 <= r < N and 0 <= c < N):
return "out_of_bounds"
if self.at(r, c) != 0:
return "occupied"
opp = 2 if who == 1 else 1
snapshot = bytes(self.cells)
self._set(r, c, who)
# 잡힘: 인접 상대 그룹 활로 0 → 제거
captured_any = False
seen = bytearray(N * N)
for d in range(4):
nr = r + self.DR[d]
nc = c + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
nidx = nr * N + nc
if self.cells[nidx] != opp or seen[nidx]:
continue
stones, libs = self._group(nr, nc)
for s in stones:
seen[s] = 1
if libs == 0:
captured_any = True
for s in stones:
self.cells[s] = 0
# 자기 그룹 활로 0 → 자살
_, my_libs = self._group(r, c)
legal = my_libs > 0
# 복원
self.cells[:] = snapshot
return None if legal else "suicide"
def is_legal(self, r: int, c: int, who: int) -> bool:
return self.illegal_reason(r, c, who) is None
def _would_capture_any(self, r: int, c: int, who: int) -> bool:
"""(r,c)에 두면 인접 상대 그룹을 바로 따내는지(단순 체크)."""
N = self.size
if not (0 <= r < N and 0 <= c < N) or self.at(r, c) != 0:
return False
opp = 2 if who == 1 else 1
snapshot = bytes(self.cells)
self._set(r, c, who)
captured = False
seen = bytearray(N * N)
for d in range(4):
nr = r + self.DR[d]
nc = c + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
nidx = nr * N + nc
if self.cells[nidx] != opp or seen[nidx]:
continue
stones, libs = self._group(nr, nc)
for s in stones:
seen[s] = 1
if libs == 0:
captured = True
break
self.cells[:] = snapshot
return captured
def would_self_atari(self, r: int, c: int, who: int, *, lib_threshold: int = 1) -> bool:
"""근사 자충수(자기단수) 판정.
- (r,c)가 합법이라고 가정(occupied/suicide 아님).
- 즉시 상대를 따내는 수라면(전술) 자충으로 보지 않음.
- 착수 후 내 그룹 활로가 lib_threshold 이하이면 True.
"""
N = self.size
if not (0 <= r < N and 0 <= c < N) or self.at(r, c) != 0:
return False
if self._would_capture_any(r, c, who):
return False
snapshot = bytes(self.cells)
opp = 2 if who == 1 else 1
self._set(r, c, who)
# 캡처 반영
seen = bytearray(N * N)
for d in range(4):
nr = r + self.DR[d]
nc = c + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
nidx = nr * N + nc
if self.cells[nidx] != opp or seen[nidx]:
continue
stones, libs = self._group(nr, nc)
for s in stones:
seen[s] = 1
if libs == 0:
for s in stones:
self.cells[s] = 0
# 내 그룹 활로
_, my_libs = self._group(r, c)
self.cells[:] = snapshot
return my_libs <= int(lib_threshold)
def is_simple_eye(self, r: int, c: int, who: int) -> bool:
"""단순 눈 판정(근사): 빈 점이며 상하좌우가 모두 내 돌 또는 경계."""
N = self.size
if not (0 <= r < N and 0 <= c < N):
return False
if self.at(r, c) != 0:
return False
for d in range(4):
nr = r + self.DR[d]
nc = c + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue # 경계는 눈 형성에 유리
v = self.at(nr, nc)
if v != who:
return False
return True
def is_self_fill(self, r: int, c: int, who: int) -> bool:
"""Qt 자충 안전망과 동일 근사: 내 돌 ≥3·상대 0·따냄/단수탈출 아님."""
N = self.size
if not (0 <= r < N and 0 <= c < N) or self.at(r, c) != 0:
return False
if self._would_capture_any(r, c, who):
return False
opp = 2 if who == 1 else 1
own_around = 0
opp_around = 0
rescues_atari = False
for d in range(4):
nr, nc = r + self.DR[d], c + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
v = self.at(nr, nc)
if v == who:
own_around += 1
_, libs = self._group(nr, nc)
if libs == 1:
rescues_atari = True
elif v == opp:
opp_around += 1
return own_around >= 3 and opp_around == 0 and not rescues_atari
def is_enclosed_territory_fill(self, r: int, c: int, who: int, *, max_region: int = 128) -> bool:
"""근사 '자기 집 메우기' 판정.
(r,c)가 빈 점일 때, (r,c)를 포함한 연결된 빈 영역이
- 바깥 경계까지 닿지 않고
- 경계가 내 돌(who)로만 둘러싸여 있으면
그 안을 메우는 수는 대개 종반 자해/무의미 수다.
성능을 위해 영역이 너무 크면(False)로 빠짐.
"""
N = self.size
if not (0 <= r < N and 0 <= c < N):
return False
if self.at(r, c) != 0:
return False
# 인접에 상대 돌이 있으면 싸움 가능성이 커서 메우기 금지 대상에서 제외.
opp = 2 if who == 1 else 1
for d in range(4):
nr = r + self.DR[d]
nc = c + self.DC[d]
if 0 <= nr < N and 0 <= nc < N and self.at(nr, nc) == opp:
return False
seen = bytearray(N * N)
stack = [r * N + c]
seen[r * N + c] = 1
touches_edge = False
while stack:
idx = stack.pop()
rr = idx // N
cc = idx % N
if rr == 0 or rr == N - 1 or cc == 0 or cc == N - 1:
touches_edge = True
break
if sum(seen) > max_region: # cheap cap
return False
for d in range(4):
nr = rr + self.DR[d]
nc = cc + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
nidx = nr * N + nc
v = self.cells[nidx]
if v == 0 and not seen[nidx]:
seen[nidx] = 1
stack.append(nidx)
elif v != 0 and v != who:
# 경계에 상대 돌이 있으면 내 집 확정이 아님
return False
if touches_edge:
return False
# 여기까지 왔으면: (r,c) 영역이 경계에 닿지 않고 경계에 상대 돌 없음.
# 내 돌로 둘러싸인 영역으로 간주.
return True
def is_hopeless_invasion(
self,
r: int,
c: int,
who: int,
*,
max_region: int = 80,
min_deep_points: int = 6,
) -> bool:
"""상대 돌로만 둘러싸인 좁고 얕은 빈 구역에 홀로 들어가는 수.
면적 계가는 죽은 돌을 들어내지 않아서, 상대 집 안에 돌 하나만 넣어도
그 구역 전체가 공배로 바뀌어 큰 이득처럼 보인다. 3선 이상 빈점이
min_deep_points 미만이면 두 눈을 낼 공간이 없다고 본다.
상대를 단수로 모는 수나 약한(활로 2 이하) 벽을 건드리는 수는 제외한다.
이미 내 돌이 있어도, 활로가 전부 이 구역 안이면 눈은 하나뿐인 죽은 자리다.
활로가 3 이상이어도 같다. 그 숨구멍을 메우면 죽은 돌이 집과 포로로
점수를 두 배로 잃는다. 구역 밖에 활로가 있거나 3선 이상이 충분하면 예외.
"""
N = self.size
if self.at(r, c) != 0:
return False
opp = 2 if who == 1 else 1
seen = bytearray(N * N)
stack = [r * N + c]
seen[r * N + c] = 1
region = 0
deep = 0
my_groups_in_region: list[tuple[int, int]] = []
while stack:
idx = stack.pop()
region += 1
if region > max_region:
return False
rr, cc = divmod(idx, N)
if self.edge_line(rr, cc) >= 3:
deep += 1
for d in range(4):
nr, nc = rr + self.DR[d], cc + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
nidx = nr * N + nc
v = self.cells[nidx]
if v == 0:
if not seen[nidx]:
seen[nidx] = 1
stack.append(nidx)
elif v == who:
if nidx not in [g[0] for g in my_groups_in_region]:
stones, libs = self._group(nr, nc)
if stones:
my_groups_in_region.append((stones[0], libs))
elif v == opp:
pass
for rep, libs in my_groups_in_region:
if libs <= 2:
continue
rr, cc = divmod(rep, N)
outside = any(
not seen[lr * N + lc] for lr, lc in self.group_liberties(rr, cc)
)
# 얕은 숨구멍 안에만 활로가 있으면 개수와 상관없이 죽은 자리다.
if not outside and deep < min_deep_points:
continue
return False
if deep >= min_deep_points:
return False
if self._would_capture_any(r, c, who):
return False
m2 = self.clone()
if not m2.play(r, c, who):
return False
for d in range(4):
nr, nc = r + self.DR[d], c + self.DC[d]
if 0 <= nr < N and 0 <= nc < N and m2.at(nr, nc) == opp:
_, libs = m2._group(nr, nc)
if libs <= 2:
return False
return True
def any_legal_move(
self, who: int, *, banned: Optional[set[tuple[int, int]]] = None
) -> Optional[Tuple[int, int]]:
N = self.size
# 중앙부터 바깥으로 나선 비슷하게 탐색하면 좋지만 단순화: 무작위 셔플.
import random
ban = banned or set()
coords = [(r, c) for r in range(N) for c in range(N) if self.cells[r * N + c] == 0]
random.shuffle(coords)
for r, c in coords:
if (r, c) in ban:
continue
if self.is_legal(r, c, who):
return (r, c)
return None
def clone(self) -> "_MiniGo":
m = _MiniGo(bytes(self.cells), self.size)
return m
def edge_line(self, r: int, c: int) -> int:
"""가장자리부터의 선 번호(1선=테두리)."""
N = self.size
return min(r, c, N - 1 - r, N - 1 - c) + 1
def group_liberties(self, r: int, c: int) -> list[tuple[int, int]]:
"""(r,c) 소속 그룹의 활로 좌표 목록."""
stones, _libs = self._group(r, c)
N = self.size
out: list[tuple[int, int]] = []
seen: set[tuple[int, int]] = set()
for idx in stones:
rr, cc = divmod(idx, N)
for d in range(4):
nr, nc = rr + self.DR[d], cc + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
if self.at(nr, nc) != 0:
continue
if (nr, nc) in seen:
continue
seen.add((nr, nc))
out.append((nr, nc))
return out
def urgent_tactical_moves(self, who: int) -> list[tuple[int, int]]:
"""아다리 응수 후보: (1) 상대 단수 따냄 (2) 내 단수 활로.
Qt atari-miss 안전망과 맞추기 위해, 단수가 있으면 계가보다 이 수를 우선한다.
"""
N = self.size
opp = 2 if who == 1 else 1
captures: list[tuple[int, int]] = []
rescues: list[tuple[int, int]] = []
seen_cap: set[tuple[int, int]] = set()
seen_res: set[tuple[int, int]] = set()
visited = bytearray(N * N)
for i in range(N * N):
if self.cells[i] == 0 or visited[i]:
continue
r, c = divmod(i, N)
stones, libs = self._group(r, c)
for s in stones:
visited[s] = 1
if libs != 1:
continue
color = self.cells[stones[0]]
libs_pts = self.group_liberties(stones[0] // N, stones[0] % N)
if color == opp:
for rr, cc in libs_pts:
if (rr, cc) in seen_cap:
continue
if not self.is_legal(rr, cc, who):
continue
seen_cap.add((rr, cc))
captures.append((rr, cc))
elif color == who:
for rr, cc in libs_pts:
if (rr, cc) in seen_res:
continue
if not self.is_legal(rr, cc, who):
continue
seen_res.add((rr, cc))
rescues.append((rr, cc))
return captures + rescues
def double_atari_threat_after(
self, r: int, c: int, who: int
) -> Optional[tuple[int, int, tuple[int, int]]]:
"""착수 뒤 상대 한 수가 서로 다른 내 그룹 둘 이상을 새로 단수로 만드는지 본다.
반환값은 가장 큰 위협의 ``(그룹 수, 위험 돌 수, 상대 응수점)``이다.
이미 단수인 그룹은 별도 긴급 대응 대상이므로 여기서는 새로 활로가
2 이상에서 1 이하로 줄어드는 그룹만 센다.
"""
after = self.clone()
if not after.play(r, c, who):
return None
opp = 2 if who == 1 else 1
N = self.size
best: Optional[tuple[int, int, tuple[int, int]]] = None
for rr in range(N):
for cc in range(N):
if after.at(rr, cc) != 0:
continue
# 상대 응수점에 맞닿은 서로 다른 내 그룹을 수집한다.
groups: list[tuple[int, int]] = []
seen_stones: set[int] = set()
for d in range(4):
nr, nc = rr + self.DR[d], cc + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N) or after.at(nr, nc) != who:
continue
nidx = nr * N + nc
if nidx in seen_stones:
continue
stones, libs = after._group(nr, nc)
seen_stones.update(stones)
# 이미 단수인 그룹은 이 응수가 만든 양단수로 세지 않는다.
if libs >= 2:
groups.append((stones[0], len(stones)))
if len(groups) < 2:
continue
reply = after.clone()
if not reply.play(rr, cc, opp):
continue
threatened = 0
stones_at_risk = 0
for representative, group_size in groups:
if reply.cells[representative] != who:
threatened += 1
stones_at_risk += group_size
continue
gr, gc = divmod(representative, N)
_stones, libs_after = reply._group(gr, gc)
if libs_after <= 1:
threatened += 1
stones_at_risk += group_size
if threatened < 2:
continue
threat = (threatened, stones_at_risk, (rr, cc))
if best is None or (threatened, stones_at_risk) > (best[0], best[1]):
best = threat
return best
def play(self, r: int, c: int, who: int) -> bool:
"""합법이면 착수·따냄을 반영하고 True. 불법이면 보드 불변·False."""
if self.illegal_reason(r, c, who) is not None:
return False
N = self.size
opp = 2 if who == 1 else 1
self._set(r, c, who)
seen = bytearray(N * N)
for d in range(4):
nr = r + self.DR[d]
nc = c + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
nidx = nr * N + nc
if self.cells[nidx] != opp or seen[nidx]:
continue
stones, libs = self._group(nr, nc)
for s in stones:
seen[s] = 1
if libs == 0:
for s in stones:
self.cells[s] = 0
return True
def side_margin(self, who: int, komi: float = 6.5) -> float:
"""자기 관점 어림 점수(중국식 면적 − 덤). 클수록 유리."""
from rl_env.board import estimate_area_score
b_area, w_area, _dame = estimate_area_score(bytes(self.cells), self.size)
black_m = float(b_area) - float(w_area) - float(komi)
return black_m if who == 1 else -black_m
def endgame_status(self) -> tuple[bool, int, int]:
"""실제 종국에 가까운지와 (미해결 빈점, 돌 수)를 반환한다.
단순 면적 계가는 한 색 돌만 있는 초중반 보드도 전부 그 색 집으로
오인할 수 있으므로, 충분히 많은 돌이 놓인 경우에만 종국으로 인정한다.
Qt GameJudge와 같은 기준으로 미해결 빈점 4개 이하는 잔여 공배로 본다.
"""
N = self.size
seen = bytearray(N * N)
unsettled = 0
for start in range(N * N):
if self.cells[start] != 0 or seen[start]:
continue
seen[start] = 1
stack = [start]
region = 0
touch = 0
while stack:
cur = stack.pop()
region += 1
rr, cc = divmod(cur, N)
for d in range(4):
nr, nc = rr + self.DR[d], cc + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
n = nr * N + nc
v = self.cells[n]
if v == 0:
if not seen[n]:
seen[n] = 1
stack.append(n)
else:
touch |= v
# 한 색에만 닿아도 큰 빈 영역은 아직 삭감·침입 여지가 있다. 면적 계가가
# 그대로 집으로 세면 빈점 100개 이상 남은 판도 종국으로 오인해 PASS한다.
if touch == 3 or touch == 0 or region > SETTLED_REGION_MAX:
unsettled += region
stones = sum(1 for v in self.cells if v != 0)
min_endgame_stones = N * N // 3
return stones >= min_endgame_stones and unsettled <= 4, unsettled, stones
def is_soft_bad_move(self, r: int, c: int, who: int, *, min_line: int = 3) -> bool:
"""눈메우기/자충/저선(따냄 제외) 등 ‘두지 말아야 할’ 근사."""
return self.self_damage_reason(r, c, who, min_line=min_line) is not None
def self_damage_reason(
self, r: int, c: int, who: int, *, min_line: int = 3
) -> Optional[str]:
"""자충·집메우기·저선 등 자기 손해 수면 reason 문자열, 아니면 None.
고수 계가 전에 ‘이 수가 나를 깎는가’를 먼저 가르는 하드 게이트.
"""
if not self.is_legal(r, c, who):
return "illegal"
cap = self._would_capture_any(r, c, who)
if self.is_simple_eye(r, c, who) and not cap:
return "fill_eye"
if self.is_self_fill(r, c, who):
return "self_fill"
hv = hogu_verdict(self.cells, r, c, who, size=self.size)
if hv is not None:
return hv
if self.would_self_atari(r, c, who, lib_threshold=1) and not cap:
return "self_atari"
if self.is_enclosed_territory_fill(r, c, who) and not cap:
return "fill_territory"
if not cap and self.is_hopeless_invasion(r, c, who):
return "hopeless_invasion"
if self.edge_line(r, c) < int(min_line) and not cap:
return "low_line"
# 활로만 줄이고 이득 없는 착수(인접 내 그룹 활로 감소, 따냄·상대단수 없음)
if not cap and self._would_reduce_own_libs_only(r, c, who):
return "reduce_own_libs"
return None
def _would_reduce_own_libs_only(self, r: int, c: int, who: int) -> bool:
"""착수 후 내 인접 그룹 활로만 줄고, 상대 위협/따냄이 없으면 True."""
N = self.size
if self.at(r, c) != 0:
return False
# 착수 전: 인접 내 그룹 최소 활로
before_libs: list[int] = []
seen0 = bytearray(N * N)
for d in range(4):
nr, nc = r + self.DR[d], c + self.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
if self.at(nr, nc) != who:
continue
nidx = nr * N + nc
if seen0[nidx]:
continue
stones, libs = self._group(nr, nc)
for s in stones:
seen0[s] = 1
before_libs.append(libs)
if not before_libs:
return False
before_min = min(before_libs)
m2 = self.clone()
if not m2.play(r, c, who):
return False
_, after_libs = m2._group(r, c)
# 상대를 단수/따냄으로 몰지 않고 내 활로만 줄면 손해
if after_libs < before_min and after_libs <= 2:
return True
return False
def gather_score_candidates(
self,
who: int,
*,
seeds: Optional[list[tuple[int, int]]] = None,
max_candidates: int = 64,
min_line: int = 3,
banned: Optional[set[tuple[int, int]]] = None,
) -> list[tuple[int, int]]:
"""가상 계가용 후보: VLM 시드 + 돌 인접 합법수 (자충/저선 필터)."""
N = self.size
out: list[tuple[int, int]] = []
seen: set[tuple[int, int]] = set()
n_stones = sum(1 for v in self.cells if v != 0)
ban = banned or set()
def _add(rr: int, cc: int) -> None:
if (rr, cc) in seen or (rr, cc) in ban:
return
if self.self_damage_reason(rr, cc, who, min_line=min_line) is not None:
return
# 초반(돌≤4): 3·4선만 (포석). 1·2선·중앙 남발 방지.
if n_stones <= 4 and self.edge_line(rr, cc) not in (3, 4):
return
seen.add((rr, cc))
out.append((rr, cc))
for s in seeds or []:
if s is None:
continue
rr, cc = int(s[0]), int(s[1])
_add(rr, cc)
has_stones = n_stones > 0
for r in range(N):
for c in range(N):
if self.at(r, c) != 0:
continue
if not self.is_legal(r, c, who):
continue
near = False
for dr in (-2, -1, 0, 1, 2):
for dc in (-2, -1, 0, 1, 2):
nr, nc = r + dr, c + dc
if 0 <= nr < N and 0 <= nc < N and self.at(nr, nc) != 0:
near = True
break
if near:
break
if has_stones and not near and n_stones > 4:
continue
_add(r, c)
if len(out) >= max_candidates:
return out
# 초반 후보가 비면 표준 화점/3·4선 시드
if not out and n_stones <= 4:
for rr, cc in (
(3, 3),
(3, 9),
(3, 15),
(9, 3),
(9, 15),
(15, 3),
(15, 9),
(15, 15),
(2, 3),
(3, 2),
(2, 15),
(3, 16),
):
_add(rr, cc)
return out
def pick_by_score(
self,
who: int,
candidates: list[tuple[int, int]],
*,
komi: float = 6.5,
prefer: Optional[tuple[int, int]] = None,
min_line: int = 3,
banned: Optional[set[tuple[int, int]]] = None,
urgent_bonus: Optional[set[tuple[int, int]]] = None,
adjust: Optional[dict[tuple[int, int], float]] = None,
minimum_delta: Optional[float] = 0.75,
) -> Optional[tuple[int, int, float, float, float, Optional[str]]]:
"""각 후보를 가상 착수 후 side_margin Δ가 최대인 수 선택.
초반(돌≤6)은 면적 계가가 보드 전체를 집으로 오인하므로 포석 점수 사용.
자충/저선·클라이언트 거부 좌표는 제외.
urgent_bonus(아다리 활로/따냄)는 가산만 하고 강제하지 않음.
adjust: 좌표별 가감점(축 판독 결과 등).
Returns:
(row, col, delta, before, after, damage_reason|None) 또는 None.
"""
dbg: dict = {
"n_cand": len(candidates),
"n_banned": 0,
"damage": {},
"n_eval": 0,
"top": [],
"below_min": False,
"minimum_delta": minimum_delta,
}
self.last_pick_debug = dbg
if not candidates:
return None
ban = banned or set()
urgent = urgent_bonus or set()
n_stones = sum(1 for v in self.cells if v != 0)
scored: list[tuple[float, int, int]] = []
stars = {
(3, 3),
(3, 9),
(3, 15),
(9, 3),
(9, 15),
(15, 3),
(15, 9),
(15, 15),
}
before = 0.0 if n_stones <= 6 else self.side_margin(who, komi)
best: Optional[tuple[int, int, float, float, float, Optional[str]]] = None
for r, c in candidates:
if (r, c) in ban:
dbg["n_banned"] += 1
continue
# 아다리 응수는 저선이어도 허용
use_min = 1 if (r, c) in urgent else min_line
dmg = self.self_damage_reason(r, c, who, min_line=use_min)
if dmg is not None and (r, c) not in urgent:
key = dmg.split(":", 1)[0]
dbg["damage"][key] = dbg["damage"].get(key, 0) + 1
continue
if dmg is not None and (r, c) in urgent and dmg not in ("illegal",):
# 활로가 자살이면 urgent 자체에서 걸러졌어야 함
if dmg == "illegal":
continue
# fill_eye 등만 해당 시 urgent는 통과
if dmg in ("self_atari",) and not self._would_capture_any(r, c, who):
continue
soft_neutral = False
if n_stones > 6 and not self._would_capture_any(r, c, who) and (r, c) not in urgent:
opp = 2 if who == 1 else 1
touches_opp = False
for d in range(4):
nr, nc = r + self.DR[d], c + self.DC[d]
if 0 <= nr < self.size and 0 <= nc < self.size and self.at(nr, nc) == opp:
touches_opp = True
break
soft_neutral = not touches_opp
if n_stones <= 6:
L = self.edge_line(r, c)
delta = 10.0 if (r, c) in stars else (5.0 if L == 4 else (3.0 if L == 3 else 0.0))
after = before + delta
else:
m2 = self.clone()
if not m2.play(r, c, who):
continue
after = m2.side_margin(who, komi)
delta = after - before
if soft_neutral:
delta *= 0.15
if prefer is not None and (r, c) == prefer:
delta += 0.15
# 아다리 응수 가산(강제 아님). 텐우키가 훨씬 크면 이길 수 있음.
if (r, c) in urgent:
delta += 8.0
if adjust:
delta += adjust.get((r, c), 0.0)
delta += heuristic_shape_delta(self.cells, r, c, who)
delta += lesson_bonus_r5(self.cells, r, c, who)
risk_reason: Optional[str] = None
if (r, c) not in urgent and not self._would_capture_any(r, c, who):
double_atari = self.double_atari_threat_after(r, c, who)
if double_atari is not None:
groups, stones_at_risk, reply = double_atari
# 한 수로 두 군데 이상이 단수가 되면 한쪽을 포기해야 하는
# 경우가 많다. 일반 계가 이득보다 충분히 큰 감점으로 회피한다.
delta -= 18.0 + 4.0 * stones_at_risk + 2.0 * (groups - 2)
risk_reason = (
f"allows_double_atari:{reply[0]},{reply[1]}"
f":groups={groups}:stones={stones_at_risk}"
)
dbg["n_eval"] += 1
scored.append((delta, r, c))
if best is None or delta > best[2]:
best = (r, c, delta, before, after, risk_reason)
scored.sort(reverse=True)
dbg["top"] = [[r, c, round(d, 3)] for d, r, c in scored[:5]]
dbg["before"] = round(before, 3)
if (
best is not None
and minimum_delta is not None
and n_stones > 6
and best[2] < float(minimum_delta)
and (best[0], best[1]) not in urgent
):
dbg["below_min"] = True
return None
return best
# ---------- TCP 서버 ----------
@dataclass
class ServerConfig:
host: str
port: int
model_id: str
max_new_tokens: int
temperature: float
load_model: bool
log_jsonl_path: Optional[str]
illegal_retry_max: int = 6
illegal_retry_budget_s: float = 55.0
text_only_moves: bool = False
board_repr: str = "both"
move_backend: MoveBackend = "vlm"
policy_checkpoint: Optional[str] = None
policy_lr: float = 5e-3
policy_train_on_reset: bool = False
policy_save_every_games: int = 1
policy_reset_small_stones: int = 8
policy_reset_prev_stones: int = 40
policy_sample: bool = False
policy_avoid_fill_eye: bool = True
policy_avoid_self_atari: bool = True
policy_avoid_fill_territory: bool = True
score_guided: bool = True
score_komi: float = 6.5
score_max_candidates: int = 64
score_min_line: int = 3
ladder_guard: bool = True
class _Logger:
def __init__(self, path: Optional[str]):
self.path = path
self._lock = threading.Lock()
self._dir_ready = False
if path:
try:
parent = os.path.dirname(os.path.abspath(path))
if parent:
os.makedirs(parent, exist_ok=True)
self._dir_ready = True
LOGGER.info("log file: %s", os.path.abspath(path))
except OSError as e:
LOGGER.warning("log 디렉토리 생성 실패 (%s): %s", parent, e)
def log(self, obj: dict) -> None:
if not self.path:
return
try:
if not self._dir_ready:
parent = os.path.dirname(os.path.abspath(self.path))
if parent:
os.makedirs(parent, exist_ok=True)
self._dir_ready = True
obj = {**obj, "ts": datetime.now(timezone.utc).isoformat()}
with self._lock:
with open(self.path, "a", encoding="utf-8") as f:
f.write(json.dumps(obj, ensure_ascii=False) + "\n")
except Exception as e:
LOGGER.warning("log write fail: %s", e)
def _recv_exact(sock: socket.socket, n: int) -> bytes:
buf = bytearray()
while len(buf) < n:
try:
chunk = sock.recv(n - len(buf))
except (ConnectionResetError, BrokenPipeError):
raise EOFError("peer reset")
if not chunk:
raise EOFError("peer closed")
buf.extend(chunk)
return bytes(buf)
def _send_ok(sock: socket.socket, row: int, col: int) -> None:
sock.sendall(MAGIC_RPLY + bytes([1, row & 0xFF, col & 0xFF]))
SETTLED_REGION_MAX = 40
ENDGAME_MIN_GAIN = 0.1
def board_hash(board: bytes, side: int) -> str:
"""요청 국면 식별자 — 대국 모니터(SQLite)와 서버 jsonl을 이어 붙이는 키."""
return f"{zlib.crc32(bytes(board) + bytes([int(side) & 0xFF])) & 0xFFFFFFFF:08x}"
def _send_pass(sock: socket.socket) -> None:
sock.sendall(MAGIC_RPLY + bytes([1, 0xFF, 0xFF]))
def _send_err(sock: socket.socket, msg: str) -> None:
b = msg.encode("utf-8")[:65535]
sock.sendall(MAGIC_RPLY + bytes([0]) + struct.pack(">H", len(b)) + b)
def _read_request(
sock: socket.socket,
) -> Tuple[Image.Image, bytes, int, int, Optional[int], list[tuple[int, int]]]:
"""(image, board, size, ai_side, game_epoch, client_forbid).
하위 호환: 본문이 보드 뒤로 딱 끝나면 game_epoch 는 None.
Qt 클라이언트는 보드 뒤에 u32_be game_epoch 를 붙여 새 대국(새 게임)마다 증가시키며,
서버는 값이 바뀔 때 이전 판 에피소드로 온라인 학습을 확실히 트리거한다.
game_epoch 뒤에 선택적으로 거부 좌표 목록:
u8 n | n×(u8 row, u8 col) — 클라이언트가 settled/자충으로 거부한 수.
"""
head = _recv_exact(sock, 4)
if head == MAGIC_PING:
sock.sendall(MAGIC_PONG)
raise _Handled()
if head != MAGIC_GO19:
raise ValueError(f"알 수 없는 헤더: {head!r}")
(body_len,) = struct.unpack(">I", _recv_exact(sock, 4))
if body_len > 32 * 1024 * 1024:
raise ValueError("body 너무 큼")
body = _recv_exact(sock, body_len)
if len(body) < 6:
raise ValueError("body 너무 짧음")
size = body[0]
ai_side = body[1]
if size != BOARD_SIZE:
raise ValueError(f"이 서버는 {BOARD_SIZE}만 지원합니다 (받음 {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
game_epoch: Optional[int] = None
if len(body) >= o + 4:
(ge,) = struct.unpack(">I", body[o : o + 4])
o += 4
game_epoch = int(ge) & 0xFFFF_FFFF
client_forbid: list[tuple[int, int]] = []
if len(body) > o:
n_forbid = int(body[o])
o += 1
if n_forbid > 64 or len(body) < o + 2 * n_forbid:
raise ValueError("거부 좌표 목록 길이 불일치")
for _ in range(n_forbid):
rr, cc = int(body[o]), int(body[o + 1])
o += 2
if 0 <= rr < size and 0 <= cc < size:
client_forbid.append((rr, cc))
if o != len(body):
raise ValueError("body 길이 불일치")
if len(board) != board_len:
raise ValueError("board 길이 불일치")
img = Image.open(io.BytesIO(png)).convert("RGB")
return img, bytes(board), size, ai_side, game_epoch, client_forbid
class _Handled(Exception):
"""Ping 등 짧은 처리 후 다음 메시지 루프로 돌아가기 위한 신호."""
def _policy_online_train_episode(
srv: "_GoTcpServer",
cfg: ServerConfig,
episode: list,
last_board_for_reward: Optional[bytes],
last_ai_side: Optional[int],
ai_side: int,
reason: str,
) -> bool:
"""에피소드가 있으면 REINFORCE 1회 업데이트. 성공 시 True."""
if not cfg.policy_train_on_reset or srv.policy is None:
return False
if not episode or last_board_for_reward is None:
LOGGER.info("policy online-train skip (%s): empty episode or board", reason)
return False
try:
from .policy_backend import outcome_reward
side_for_reward = int(last_ai_side if last_ai_side is not None else ai_side)
r = outcome_reward(last_board_for_reward, side_for_reward)
ema = float(srv.policy.stats.ema_reward)
beta = 0.9
srv.policy.stats.ema_reward = beta * ema + (1.0 - beta) * float(r)
adv = float(r) - float(srv.policy.stats.ema_reward)
upd = [(x, m, a, adv) for (x, m, a) in episode]
with srv.policy_lock:
srv.policy.update_transitions(upd)
srv.policy.stats.games_seen += 1
g = int(srv.policy.stats.games_seen)
if (
cfg.policy_checkpoint
and int(cfg.policy_save_every_games) > 0
and g % int(cfg.policy_save_every_games) == 0
):
srv.policy.save(cfg.policy_checkpoint)
LOGGER.info(
"policy online-train: %s reward=%.4f adv=%.4f games=%d saved=%s",
reason,
float(r),
float(adv),
int(srv.policy.stats.games_seen),
"yes" if cfg.policy_checkpoint else "no",
)
return True
except Exception as e:
LOGGER.warning("policy online-train failed (%s): %s", reason, e)
return False
def _position_scoped_reject_bans(
st: dict,
board: bytes,
size: int,
ai_side: int,
client_forbid: list[tuple[int, int]],
) -> tuple[set[tuple[int, int]], Optional[tuple[int, int]]]:
"""현재 국면에서만 유효한 클라이언트 거부·미착수 좌표를 만든다.
좌표만 누적하면 실제로 둔 돌이 나중에 따내져 그 자리가 다시 비었을 때도
과거 거부로 오인한다. 따라서 보드 전체와 차례가 같은 재요청에서만 금지를
재사용하고, 한 수라도 진행된 새 국면에서는 폐기한다.
"""
board_key = bytes(board)
same_position = (
st.get("reject_bans_board") == board_key
and int(st.get("reject_bans_side") or 0) == int(ai_side)
)
client_coords = {
(int(rr), int(cc))
for rr, cc in client_forbid[-16:]
if 0 <= int(rr) < size and 0 <= int(cc) < size
}
previous_client_coords = (
{
(int(item[0]), int(item[1]))
for item in (st.get("client_forbid_snapshot") or [])
if len(item) >= 2
}
if same_position
else set()
)
raw_bans = list(st.get("reject_bans") or []) if same_position else []
# 같은 국면에서 클라이언트가 목록을 명시적으로 비우면 client ban도 푼다.
if same_position and previous_client_coords and not client_coords:
raw_bans = []
if len(raw_bans) > 24:
raw_bans = raw_bans[-24:]
bans: set[tuple[int, int]] = set()
for item in raw_bans:
try:
rr, cc = int(item[0]), int(item[1])
except Exception:
continue
if 0 <= rr < size and 0 <= cc < size and board[rr * size + cc] == 0:
bans.add((rr, cc))
inferred_reject: Optional[tuple[int, int]] = None
offers = st.get("last_offers") or {}
prev_offer = offers.get(str(int(ai_side)))
if prev_offer is not None and len(prev_offer) >= 3:
lr, lc = int(prev_offer[0]), int(prev_offer[1])
offered_on_board = prev_offer[2]
if (
offered_on_board == board_key
and 0 <= lr < size
and 0 <= lc < size
and board[lr * size + lc] == 0
):
bans.add((lr, lc))
inferred_reject = (lr, lc)
if same_position:
# 같은 국면에서 새로 추가된 좌표만 이번 거부로 인정한다. 국면이
# 바뀐 첫 요청의 목록은 이전 국면에서 넘어온 stale snapshot이다.
bans.update(client_coords - previous_client_coords)
st["reject_bans"] = [[r, c] for r, c in list(bans)[-24:]]
st["reject_bans_board"] = board_key
st["reject_bans_side"] = int(ai_side)
st["client_forbid_snapshot"] = [[r, c] for r, c in client_coords]
return bans, inferred_reject
class _GoTcpHandler(socketserver.BaseRequestHandler):
server: "_GoTcpServer" # type: ignore[assignment]
def handle(self) -> None:
peer = self.client_address
LOGGER.info("client connected: %s", peer)
sock: socket.socket = self.request
sock.settimeout(None)
# 온라인 학습(게임 단위)용 상태 — 연결이 끊겨도 유지.
# 같은 머신에서 VLM×2 포트를 쓰면 peer_ip 만으로는 두 프로세스가 서로의 에피소드를 덮어쓴다.
peer_ip = str(peer[0]) if peer and len(peer) >= 1 else "unknown"
listen_port = int(self.server.server_address[1])
state_key = f"{peer_ip}:{listen_port}"
st = self.server.get_online_state(state_key)
last_stones: Optional[int] = st.get("last_stones")
last_board_for_reward: Optional[bytes] = st.get("last_board_for_reward")
episode: list = st.get("episode") or [] # [(x ndarray, mask list[int], action int)]
last_ai_side: Optional[int] = st.get("last_ai_side")
policy_rng = None
try:
import numpy as _np
policy_rng = _np.random.default_rng(int(time.time() * 1e6) & 0xFFFF_FFFF)
except Exception:
policy_rng = None
try:
while True:
try:
img, board, size, ai_side, game_epoch, client_forbid = _read_request(sock)
except _Handled:
continue
except EOFError:
break
except Exception as e:
LOGGER.warning("프레임 파싱 실패: %s", e)
try:
_send_err(sock, f"프레임 파싱 실패: {e}")
except Exception:
pass
break
n_b = sum(1 for b in board if b == 1)
n_w = sum(1 for b in board if b == 2)
cfg = self.server.cfg
stones_now = n_b + n_w
pass_streak = int(st.get("pass_streak") or 0)
# 1) 클라이언트 game_epoch 변경 = 새 대국 시작 → 이전 판은 반드시 한 번 학습.
prev_ge = st.get("last_game_epoch")
if (
game_epoch is not None
and prev_ge is not None
and int(game_epoch) != int(prev_ge)
):
if cfg.policy_train_on_reset:
_policy_online_train_episode(
self.server,
cfg,
episode,
last_board_for_reward,
last_ai_side,
ai_side,
"game_epoch",
)
episode.clear()
pass_streak = 0
st["pass_streak"] = 0
# 새 대국: 거부/미착수 금지도 리셋
st["reject_bans"] = []
st["reject_bans_board"] = None
st["reject_bans_side"] = None
st["client_forbid_snapshot"] = []
st["last_offers"] = {}
# 2) 돌 수 급감(구 클라이언트·휴리스틱; 에피소드가 남았을 때만)
if (
last_stones is not None
and stones_now <= int(cfg.policy_reset_small_stones)
and last_stones >= int(cfg.policy_reset_prev_stones)
and last_board_for_reward is not None
and episode
and cfg.policy_train_on_reset
):
_policy_online_train_episode(
self.server,
cfg,
episode,
last_board_for_reward,
last_ai_side,
ai_side,
"stone_drop",
)
episode.clear()
pass_streak = 0
st["pass_streak"] = 0
# 거부 좌표는 같은 보드·같은 차례의 재요청에만 유효하다.
# 착수·따냄으로 국면이 바뀌면 과거 금지를 모두 해제한다.
bans, inferred_reject = _position_scoped_reject_bans(
st, board, size, ai_side, client_forbid
)
offers = st.get("last_offers") or {}
side_key = str(int(ai_side))
if inferred_reject is not None:
LOGGER.info(
"동일 국면 미착수 감지 → 금지 (%d,%d) side=%s (Qt 거부/대체 추정)",
inferred_reject[0],
inferred_reject[1],
"B" if ai_side == 1 else "W",
)
if client_forbid:
LOGGER.info(
"클라이언트 거부 좌표 %d개(사용 %d): %s",
len(client_forbid),
min(16, len(client_forbid)),
", ".join(f"({r},{c})" for r, c in client_forbid[-12:]),
)
LOGGER.info(
"GO19 request stones B=%d W=%d side=%s epoch=%s png=%s bans=%d — packed board (test):\n%s",
n_b,
n_w,
"B" if ai_side == 1 else "W",
game_epoch if game_epoch is not None else "-",
getattr(img, "size", ()),
len(bans),
packed_board_ascii_for_log(board, size),
)
last_stones = stones_now
last_board_for_reward = board
last_ai_side = int(ai_side)
if game_epoch is not None:
st["last_game_epoch"] = int(game_epoch)
t0 = time.time()
self._cur_game_epoch = game_epoch
try:
is_pass_reply, offered = self._handle_one(
sock, img, board, size, ai_side, episode, policy_rng, bans
)
if is_pass_reply:
pass_streak += 1
# PASS 는 착수 아님 — 해당 색 last_offer 유지하지 않음
offers = dict(st.get("last_offers") or {})
offers.pop(side_key, None)
st["last_offers"] = offers
else:
pass_streak = 0
if offered is not None:
offers = dict(st.get("last_offers") or {})
offers[side_key] = [
int(offered[0]),
int(offered[1]),
bytes(board),
]
st["last_offers"] = offers
st["pass_streak"] = pass_streak
# 3) 같은 연결에서 서버가 연속 PASS 두 번(단일 프로세스 종국)
if pass_streak >= 2 and episode and cfg.policy_train_on_reset:
_policy_online_train_episode(
self.server,
cfg,
episode,
last_board_for_reward,
last_ai_side,
ai_side,
"pass_x2",
)
episode.clear()
pass_streak = 0
st["pass_streak"] = 0
except Exception as e:
LOGGER.exception("처리 중 오류")
try:
_send_err(sock, f"서버 오류: {e}")
except Exception:
pass
break
LOGGER.info(
"served move in %.2fs (side=%s)",
time.time() - t0,
"B" if ai_side == 1 else "W",
)
finally:
self.server.set_online_state(
state_key,
{
"last_stones": last_stones,
"last_board_for_reward": last_board_for_reward,
"episode": episode,
"last_ai_side": last_ai_side,
"pass_streak": int(st.get("pass_streak") or 0),
"last_game_epoch": st.get("last_game_epoch"),
"reject_bans": list(st.get("reject_bans") or []),
"reject_bans_board": st.get("reject_bans_board"),
"reject_bans_side": st.get("reject_bans_side"),
"client_forbid_snapshot": list(st.get("client_forbid_snapshot") or []),
"last_offers": dict(st.get("last_offers") or {}),
},
)
LOGGER.info("client disconnected: %s", peer)
def _handle_one(
self,
sock: socket.socket,
img: Image.Image,
board: bytes,
size: int,
ai_side: int,
episode: list,
policy_rng,
banned: Optional[set[tuple[int, int]]] = None,
) -> tuple[bool, Optional[tuple[int, int]]]:
cfg = self.server.cfg
mini = _MiniGo(board, size)
ban = set(banned or set())
row: Optional[int] = None
col: Optional[int] = None
is_pass = False
illegal_reason: Optional[str] = None
illegal_row: Optional[int] = None
illegal_col: Optional[int] = None
retry_count = 0
forbidden: list[tuple[int, int]] = list(ban)
vlm_seeds: list[tuple[int, int]] = []
vlm_raw0 = ""
score_meta: dict = {}
fallback_used = False
policy_transition: Optional[tuple] = None
pass_reason: Optional[str] = None
pick_debug: dict = {}
# 1) 정책 백엔드: packed 보드만으로 좌표 선택(합법 마스크 위).
if cfg.move_backend in ("policy", "hybrid") and self.server.policy is not None:
try:
from .policy_backend import encode_obs, restrict_pass_to_forced
# mask 생성: 미니룰(점유/자살/따냄) 기준 합법만 1
mask = [0] * (size * size + 1) # +PASS
for rr in range(size):
base = rr * size
for cc in range(size):
if not mini.is_legal(rr, cc, ai_side):
continue
# 눈 메우기 방지(근사): 단순 눈이고, 즉시 따내는 수가 아니면 제외
if (
bool(getattr(cfg, "policy_avoid_fill_eye", True))
and mini.is_simple_eye(rr, cc, ai_side)
and not mini._would_capture_any(rr, cc, ai_side)
):
continue
# 종반 자충(자기단수) 방지(근사): 즉시 따내는 수가 아니면 활로 1 수는 제외
if bool(getattr(cfg, "policy_avoid_self_atari", True)) and mini.would_self_atari(
rr, cc, ai_side, lib_threshold=1
):
continue
# 종반 자기 집 메우기 방지(근사): 싸움/따냄이 아니면 내 집 내부는 제외
if (
bool(getattr(cfg, "policy_avoid_fill_territory", True))
and not mini._would_capture_any(rr, cc, ai_side)
and mini.is_enclosed_territory_fill(rr, cc, ai_side)
):
continue
if (rr, cc) in ban:
continue
if True:
mask[base + cc] = 1
# 둘 곳이 있으면 PASS를 정책 행동에서 제외한다. 축 수가 교정될 때
# PASS로 회피하는 대신 다른 합법 좌표를 고르게 학습한다.
if restrict_pass_to_forced(mask, size * size):
is_pass = True
pass_reason = "policy_no_legal"
else:
x = encode_obs(board, ai_side)
with self.server.policy_lock:
if bool(getattr(cfg, "policy_sample", False)) and policy_rng is not None:
a = self.server.policy.sample_action(x, mask, policy_rng)
else:
a = self.server.policy.greedy_action(x, mask)
if int(a) == size * size:
is_pass = True
pass_reason = "policy_pass"
row = col = None
else:
row, col = divmod(int(a), size)
# score/축 가드가 좌표를 교정할 수 있으므로 최종 행동이 정해진
# 뒤 episode에 넣는다.
if cfg.policy_train_on_reset and episode is not None:
policy_transition = (x, mask)
except Exception as e:
LOGGER.warning("policy backend failed: %s", e)
row = col = None
is_pass = False
# 2) VLM 백엔드: 기존 로직 (또는 hybrid에서 policy가 실패했을 때만)
if (row is None or col is None) and cfg.load_model and self.server.model is not None and cfg.move_backend != "policy":
# 불법수면 (row,col)을 금지 목록에 넣고 재질문해서 다른 수가 나오는지 본다.
# temperature>0 이면 재질문만으로도 달라질 수 있고,
# temperature가 낮아 반복되는 모델은 금지 목록 힌트로 회피 가능성이 올라간다.
t_deadline = time.time() + max(0.5, float(getattr(cfg, "illegal_retry_budget_s", 0.0)))
max_retry = max(0, int(getattr(cfg, "illegal_retry_max", 0)))
max_attempts = 1 + max_retry # 첫 시도 포함
for attempt in range(max_attempts):
remaining = t_deadline - time.time()
if attempt > 0 and remaining <= 0:
break
try:
row, col, is_pass = self.server.model.predict_move(
img,
board,
ai_side,
timeout_s=min(120.0, max(1.0, remaining)) if attempt > 0 else 120.0,
forbidden=forbidden,
illegal_hint=(
None
if illegal_reason is None or illegal_row is None or illegal_col is None
else f"({illegal_row},{illegal_col}) reason={illegal_reason}"
),
)
except Exception as e:
LOGGER.warning("모델 추론 실패: %s", e)
row = col = None
is_pass = False
break
if attempt == 0:
vlm_raw0 = str(getattr(self.server.model, "last_raw_text", "") or "")[:240]
if is_pass or row is None or col is None:
break
vlm_seeds.append((int(row), int(col)))
reason = mini.illegal_reason(row, col, ai_side)
if reason is None:
break # 합법
# 불법 → 금지 목록 추가 후 재시도.
illegal_reason = reason
illegal_row, illegal_col = row, col
forbidden.append((row, col))
retry_count = attempt + 1
LOGGER.info(
"모델 비합법수 (%d,%d) reason=%s → retry(%d)",
row,
col,
reason,
retry_count,
)
row = col = None
is_pass = False
if not is_pass and row is not None and col is not None:
reason = mini.illegal_reason(row, col, ai_side)
if reason is not None:
illegal_reason = reason
illegal_row, illegal_col = row, col
LOGGER.info(
"모델 비합법수 (%d,%d) reason=%s → score/candidate 경로",
row,
col,
reason,
)
row = col = None
# 모델/정책의 PASS는 아직 최종 결정이 아니다. 아래 계가 경로에서
# 단수 응수나 안전하고 가치 있는 착수점이 실제로 없는지 확인한 뒤 확정한다.
pass_requested = is_pass
if is_pass and pass_reason is None:
pass_reason = "model_pass"
# 3) 계가 유도. 아다리는 가산만 하고 강제하지 않음(텐우키/희생 허용).
ladder_on = bool(getattr(cfg, "ladder_guard", True))
# 뻗어도 축으로 잡히는 단수 그룹의 활로 — 금지 완화 재탐색에서도 풀지 않는다.
ladder_ban: set[tuple[int, int]] = set()
ladder_futile: list[dict] = []
if ladder_on:
try:
for v in futile_escapes(board, ai_side, size=size):
ladder_ban.add(v.point)
ladder_futile.append(
{"point": list(v.point), "group_at": list(v.group_at), "stones": v.group_size}
)
except Exception as e:
LOGGER.warning("축 판독 실패(헛도망): %s", e)
if ladder_futile:
LOGGER.info("축 헛도망 차단 %s", [f["point"] for f in ladder_futile])
# 축에 걸린 국면에서는 VLM/정책의 PASS도 회피 행동으로 본다.
# score 후보 경로로 보내 축 수가 아닌 다른 착수를 찾는다.
if is_pass and ladder_futile and bool(getattr(cfg, "score_guided", True)):
LOGGER.info("축 국면 PASS 차단 → 다른 합법수 탐색")
is_pass = False
row = col = None
# PASS 제안도 반드시 같은 후보 평가를 거친다. 이전에는 축에 걸린 경우만
# PASS를 취소해서, 모델이 불리한 국면이나 자충 회피 중 PASS를 반환하면
# 정상 중반에도 그대로 통과했다.
if bool(getattr(cfg, "score_guided", True)):
prefer = (int(row), int(col)) if row is not None and col is not None else None
if prefer is not None and (prefer in ban or prefer in ladder_ban):
prefer = None
seeds = [s for s in vlm_seeds if s not in ladder_ban]
if prefer is not None:
seeds.insert(0, prefer)
urgent = [
p
for p in mini.urgent_tactical_moves(ai_side)
if mini.is_legal(p[0], p[1], ai_side) and p not in ban and p not in ladder_ban
]
for up in urgent:
seeds.append(up)
cands = mini.gather_score_candidates(
ai_side,
seeds=seeds,
max_candidates=int(getattr(cfg, "score_max_candidates", 64)),
min_line=int(getattr(cfg, "score_min_line", 3)),
banned=ban | ladder_ban,
)
# 아다리 활로가 1·2선이어도 후보에 넣기 (min_line 우회)
for ur, uc in urgent:
if (ur, uc) not in cands and (ur, uc) not in ban:
cands.insert(0, (ur, uc))
if not cands and ban:
LOGGER.info("계가 후보 0 (bans=%d) → 금지 완화 재탐색", len(ban))
cands = mini.gather_score_candidates(
ai_side,
seeds=seeds,
max_candidates=int(getattr(cfg, "score_max_candidates", 64)),
min_line=int(getattr(cfg, "score_min_line", 3)),
banned=ladder_ban,
)
ban = set()
# 단수 후보의 축 판독: 잡히는 축은 가산, 상대가 빠져나가는 축(축머리·꺾임 단수)은 감점.
ladder_adj: dict[tuple[int, int], float] = {}
ladder_chase: dict[str, str] = {}
if ladder_on:
try:
for p in cands:
if mini._would_capture_any(p[0], p[1], ai_side):
continue
cv = chase_verdict(board, p[0], p[1], ai_side, size=size)
if cv is None:
continue
if cv.kind == "working_chase":
# 잡히는 축은 급하지 않다 — 큰 그룹일 때만 약하게 가산
ladder_adj[p] = min(0.5 * cv.group_size, 3.0) if cv.group_size >= 2 else 0.0
ladder_adj[p] += working_chase_bonus_r5()
else:
ladder_adj[p] = -6.0 if cv.group_size >= 2 else -2.0
ladder_chase[f"{p[0]},{p[1]}"] = f"{cv.kind}:{cv.group_size}"
except Exception as e:
LOGGER.warning("축 판독 실패(추격): %s", e)
urgent_set = set(urgent)
relaxed_pass_override = False
picked = mini.pick_by_score(
ai_side,
cands,
komi=float(getattr(cfg, "score_komi", 6.5)),
prefer=prefer,
min_line=int(getattr(cfg, "score_min_line", 3)),
banned=ban | ladder_ban,
urgent_bonus=urgent_set,
adjust=ladder_adj,
)
pick_debug["strict"] = dict(getattr(mini, "last_pick_debug", {}) or {})
late_endgame, unsettled_empty, stone_count = mini.endgame_status()
if picked is None and not late_endgame:
# 중반에는 계가 Δ 임계값만으로 PASS하지 않는다. 모델이 좌표를 냈어도
# 조용한 수는 Δ가 0.15배로 깎여 임계값(0.75) 아래로 떨어지기 쉬우므로
# PASS 제안 여부와 무관하게 1·2선까지 넓혀 다시 고른다.
# 자충·집 메우기·호구 등의 안전 필터는 그대로 둔다.
relaxed = mini.gather_score_candidates(
ai_side,
seeds=seeds,
max_candidates=int(getattr(cfg, "score_max_candidates", 64)),
min_line=1,
banned=ban | ladder_ban,
)
for p in cands:
if p not in relaxed:
relaxed.append(p)
picked = mini.pick_by_score(
ai_side,
relaxed,
komi=float(getattr(cfg, "score_komi", 6.5)),
prefer=prefer,
min_line=1,
banned=ban | ladder_ban,
urgent_bonus=urgent_set,
adjust=ladder_adj,
minimum_delta=None,
)
pick_debug["relaxed"] = dict(getattr(mini, "last_pick_debug", {}) or {})
if picked is not None:
cands = relaxed
relaxed_pass_override = True
LOGGER.info(
"중반 PASS 차단 → 안전 후보 임계 완화 (stones=%d unsettled=%d requested=%s)",
stone_count,
unsettled_empty,
"Y" if pass_requested else "n",
)
elif picked is None and late_endgame:
# 끝내기는 1·2선 경계 수가 대부분이고, 상대 돌에 닿지 않는 수는 Δ가
# 0.15배로 깎인다. 0.75 임계값만 보면 몇 집짜리 끝내기를 두고 PASS한다.
yose = mini.gather_score_candidates(
ai_side,
seeds=seeds,
max_candidates=int(getattr(cfg, "score_max_candidates", 64)),
min_line=1,
banned=ban | ladder_ban,
)
for p in cands:
if p not in yose:
yose.append(p)
picked = mini.pick_by_score(
ai_side,
yose,
komi=float(getattr(cfg, "score_komi", 6.5)),
prefer=prefer,
min_line=1,
banned=ban | ladder_ban,
urgent_bonus=urgent_set,
adjust=ladder_adj,
minimum_delta=ENDGAME_MIN_GAIN,
)
pick_debug["yose"] = dict(getattr(mini, "last_pick_debug", {}) or {})
if picked is not None:
cands = yose
relaxed_pass_override = True
LOGGER.info("종반 PASS 차단 → 끝내기 후보 (unsettled=%d)", unsettled_empty)
if picked is None:
alt = (
mini.any_legal_move(ai_side, banned=ban | ladder_ban)
if ladder_futile
else None
)
if alt is None:
LOGGER.info(
"안전하고 가치 있는 계가 후보 없음 → PASS%s",
" (제안 PASS 확인)" if pass_requested else "",
)
is_pass = True
row = col = None
if late_endgame:
pass_reason = "endgame_no_gain"
elif "relaxed" in pick_debug:
pass_reason = "midgame_no_safe_move"
else:
pass_reason = "no_meaningful_safe_move"
score_meta["pass_requested"] = pass_requested
score_meta["late_endgame"] = late_endgame
score_meta["unsettled_empty"] = unsettled_empty
score_meta["stone_count"] = stone_count
else:
row, col = alt
fallback_used = True
LOGGER.info("축 국면 계가 후보 없음 → 대체 합법수 (%d,%d)", row, col)
else:
pr, pc, delta, before, after, dmg = picked
row, col = pr, pc
is_pass = False
score_meta = {
"score_guided": True,
"score_delta": round(delta, 3),
"score_before": round(before, 3),
"score_after": round(after, 3),
"score_n_cand": len(cands),
"vlm_prefer": list(prefer) if prefer else None,
"self_damage": dmg,
"ban_n": len(ban),
"urgent_atari": (pr, pc) in urgent_set,
"urgent_n": len(urgent_set),
"pass_requested": pass_requested,
"pass_overridden": pass_requested,
"late_endgame": late_endgame,
"unsettled_empty": unsettled_empty,
"stone_count": stone_count,
"relaxed_pass_override": relaxed_pass_override,
}
if pass_requested:
LOGGER.info(
"조기 PASS 차단 → 긴급/계가 대체수 (%d,%d) Δ=%+.2f urgent=%s",
pr,
pc,
delta,
"Y" if (pr, pc) in urgent_set else "n",
)
if ladder_futile:
score_meta["ladder_futile"] = ladder_futile
if ladder_chase:
score_meta["ladder_chase"] = ladder_chase
score_meta["ladder_pick"] = ladder_chase.get(f"{pr},{pc}")
LOGGER.info(
"계가선택 (%d,%d) Δ=%+.2f (before=%+.2f → after=%+.2f) cands=%d bans=%d urgent=%s side=%s",
pr,
pc,
delta,
before,
after,
len(cands),
len(ban),
"Y" if (pr, pc) in urgent_set else "n",
"B" if ai_side == 1 else "W",
)
if delta < -0.25 and len(cands) <= 2 and not ladder_futile:
LOGGER.info("계가 Δ 불량 + 후보 부족 → PASS")
is_pass = True
pass_reason = "bad_delta_few_cands"
row = col = None
elif row is not None and col is not None and (row, col) not in urgent_set:
dmg2 = mini.self_damage_reason(
int(row), int(col), ai_side,
min_line=1
if relaxed_pass_override
else int(getattr(cfg, "score_min_line", 3)),
)
if dmg2 is not None:
score_meta["self_damage"] = dmg2
alt = next((p for p in cands if p != (row, col)), None) if ladder_futile else None
if alt is None and ladder_futile:
alt = mini.any_legal_move(
ai_side, banned=ban | ladder_ban | {(int(row), int(col))}
)
if alt is None:
LOGGER.warning("자충/손해수 차단 (%d,%d) reason=%s → PASS", row, col, dmg2)
is_pass = True
pass_reason = f"self_damage_block:{dmg2}"
score_meta["blocked_move"] = [int(row), int(col)]
row = col = None
else:
LOGGER.warning(
"자충/손해수 차단 (%d,%d) reason=%s → 축 대체수 (%d,%d)",
row,
col,
dmg2,
alt[0],
alt[1],
)
row, col = alt
fallback_used = True
if not is_pass and (row is None or col is None):
# score_guided 꺼졌을 때만 구형 랜덤 fallback
if bool(getattr(cfg, "score_guided", True)):
alt = (
mini.any_legal_move(ai_side, banned=ban | ladder_ban)
if ladder_futile
else None
)
if alt is None:
is_pass = True
pass_reason = "no_coord_fallback"
else:
row, col = alt
fallback_used = True
else:
picked_fb = mini.any_legal_move(ai_side)
if picked_fb is None:
is_pass = True
pass_reason = "no_legal_move"
else:
row, col = picked_fb
fallback_used = True
if illegal_reason is not None:
LOGGER.info(
"fallback 선택 (%d,%d) (illegal=%s from (%s,%s), retry=%d)",
row,
col,
illegal_reason,
"?" if illegal_row is None else str(illegal_row),
"?" if illegal_col is None else str(illegal_col),
retry_count,
)
# 온라인 학습에는 정책의 최초 제안이 아니라 실제 클라이언트에 보낼
# 행동을 기록한다. 축 가드가 다른 곳을 골랐다면 그 좌표가 정답 행동이다.
if policy_transition is not None and episode is not None:
x, train_mask = policy_transition
action = size * size if is_pass else int(row) * size + int(col)
train_mask = list(train_mask)
train_mask[action] = 1
episode.append((x, train_mask, action))
log_rec = {
"event": "move",
"ai_side": ai_side,
"row": row,
"col": col,
"pass": is_pass,
"fallback_used": fallback_used or (cfg.load_model is False),
"illegal_reason": illegal_reason,
"illegal_row": illegal_row,
"illegal_col": illegal_col,
"retry_count": retry_count,
"text_only_moves": bool(getattr(cfg, "text_only_moves", False)),
"board_repr": str(getattr(cfg, "board_repr", "both")),
"move_backend": str(getattr(cfg, "move_backend", "vlm")),
"board_hash": board_hash(board, ai_side),
"game_epoch": getattr(self, "_cur_game_epoch", None),
"pass_requested": pass_requested,
"vlm_seeds": [list(s) for s in vlm_seeds],
"vlm_raw0": vlm_raw0,
}
log_rec.update(score_meta)
if is_pass:
log_rec["pass_reason"] = pass_reason or "unknown"
log_rec["pick_debug"] = pick_debug
log_rec["legal_n"] = sum(
1 for i in range(size * size) if mini.is_legal(i // size, i % size, ai_side)
)
LOGGER.info(
"PASS 사유=%s requested=%s legal=%d debug=%s",
log_rec["pass_reason"],
"Y" if pass_requested else "n",
log_rec["legal_n"],
json.dumps(pick_debug, ensure_ascii=False),
)
self.server.logger.log(log_rec)
if is_pass:
_send_pass(sock)
return True, None
else:
assert row is not None and col is not None
_send_ok(sock, row, col)
return False, (int(row), int(col))
class _GoTcpServer(socketserver.ThreadingTCPServer):
daemon_threads = True
allow_reuse_address = True
def __init__(
self, address, handler_cls, cfg: ServerConfig, model: Optional[_MlxVlmModel],
slogger: _Logger,
):
super().__init__(address, handler_cls)
self.cfg = cfg
self.model = model
self.logger = slogger
self.policy = None
self.policy_lock = threading.Lock()
# 클라이언트 연결이 끊겨도 유지되는 온라인 학습 상태(게임 단위).
self._online_state_lock = threading.Lock()
self._online_state: dict[str, dict] = {}
def get_online_state(self, peer_ip: str) -> dict:
with self._online_state_lock:
st = self._online_state.get(peer_ip)
if st is None:
st = {}
self._online_state[peer_ip] = st
return st
def set_online_state(self, peer_ip: str, st: dict) -> None:
with self._online_state_lock:
self._online_state[peer_ip] = st
# ---------- 외부에서 호출 ----------
def run_server(
*,
host: str,
port: int,
model_id: str,
max_new_tokens: int,
temperature: float,
load_model: bool,
log_jsonl_path: Optional[str],
lora_dir: Optional[str] = None,
illegal_retry_max: int = 6,
illegal_retry_budget_s: float = 55.0,
text_only_moves: bool = False,
board_repr: BoardRepr = "both",
move_backend: MoveBackend = "vlm",
policy_checkpoint: Optional[str] = None,
policy_lr: float = 5e-3,
policy_train_on_reset: bool = False,
policy_save_every_games: int = 1,
policy_reset_small_stones: int = 8,
policy_reset_prev_stones: int = 40,
policy_sample: bool = False,
policy_avoid_fill_eye: bool = True,
policy_avoid_self_atari: bool = True,
policy_avoid_fill_territory: bool = True,
score_guided: bool = True,
score_komi: float = 6.5,
score_max_candidates: int = 64,
score_min_line: int = 3,
ladder_guard: bool = True,
) -> None:
logging.basicConfig(
level=os.environ.get("VLM_LOGLEVEL", "INFO"),
format="%(asctime)s %(levelname)s %(message)s",
)
if board_repr not in ("chars", "numbers", "both", "compact", "aligned", "aligned_read", "aligned_read2"):
raise ValueError(
"board_repr must be chars|numbers|both|compact|aligned|aligned_read|aligned_read2, "
f"got {board_repr!r}"
)
cfg = ServerConfig(
host=host,
port=port,
model_id=model_id,
max_new_tokens=max_new_tokens,
temperature=temperature,
load_model=load_model,
log_jsonl_path=log_jsonl_path,
illegal_retry_max=int(illegal_retry_max),
illegal_retry_budget_s=float(illegal_retry_budget_s),
text_only_moves=bool(text_only_moves),
board_repr=board_repr,
move_backend=move_backend,
policy_checkpoint=policy_checkpoint,
policy_lr=float(policy_lr),
policy_train_on_reset=bool(policy_train_on_reset),
policy_save_every_games=int(policy_save_every_games),
policy_reset_small_stones=int(policy_reset_small_stones),
policy_reset_prev_stones=int(policy_reset_prev_stones),
policy_sample=bool(policy_sample),
policy_avoid_fill_eye=bool(policy_avoid_fill_eye),
policy_avoid_self_atari=bool(policy_avoid_self_atari),
policy_avoid_fill_territory=bool(policy_avoid_fill_territory),
score_guided=bool(score_guided),
score_komi=float(score_komi),
score_max_candidates=int(score_max_candidates),
score_min_line=int(score_min_line),
ladder_guard=bool(ladder_guard),
)
model: Optional[_MlxVlmModel] = None
if load_model:
model = _MlxVlmModel(
model_id,
max_new_tokens=max_new_tokens,
temperature=temperature,
lora_dir=lora_dir,
text_only_moves=bool(text_only_moves),
board_repr=board_repr,
)
# 추론 전용 스레드를 시작하고 모델 로드 완료까지 대기.
try:
model.start(wait_load=True)
except FileNotFoundError:
# _validate_lora_dir 가 던지는 친절 에러: 사용자가 lora-dir 를 명시했으나
# 학습된 어댑터 파일이 없는 상태. 좀비 fallback 으로 listen 하지 말고 즉시 종료.
LOGGER.error("LoRA 어댑터 파일 부재로 서버를 시작하지 않습니다.")
raise SystemExit(2)
except BaseException as e:
# 그 외 모델 로드 실패는 lora_dir 가 명시되었는지에 따라 분기.
if lora_dir:
LOGGER.exception("LoRA 적용 모드인데 모델 로드 실패 → 종료: %s", e)
raise SystemExit(2)
LOGGER.exception("모델 로드 실패 → 무작위 합법수 모드로 전환합니다: %s", e)
model = None
cfg.load_model = False
slogger = _Logger(log_jsonl_path)
LOGGER.info(
"listening on %s:%d (model=%s, load=%s, backend=%s, text_only_moves=%s, "
"board_repr=%s, score_guided=%s, min_line=%s, ladder_guard=%s)",
host,
port,
model_id,
bool(model),
str(move_backend),
bool(text_only_moves),
board_repr,
bool(score_guided),
int(score_min_line),
bool(ladder_guard),
)
with _GoTcpServer((host, port), _GoTcpHandler, cfg, model, slogger) as srv:
# 정책 로드(선택). VLM 없이 policy-only도 가능.
if cfg.move_backend in ("policy", "hybrid") and cfg.policy_checkpoint:
try:
from .policy_backend import PolicyMLP
srv.policy = PolicyMLP.load(cfg.policy_checkpoint, lr=cfg.policy_lr)
LOGGER.info("policy loaded: %s (lr=%.4g)", cfg.policy_checkpoint, cfg.policy_lr)
except FileNotFoundError:
from .policy_backend import PolicyMLP
srv.policy = PolicyMLP(hidden=256, seed=0, lr=cfg.policy_lr)
LOGGER.warning(
"policy checkpoint not found, starting new policy: %s", cfg.policy_checkpoint
)
except Exception as e:
LOGGER.warning("policy load failed (disable policy backend): %s", e)
srv.policy = None
if cfg.move_backend in ("policy", "hybrid") and srv.policy is not None:
LOGGER.info(
"policy online-train: %s (에피소드 끝마다 학습하려면 --policy-train-on-reset)",
"ON" if cfg.policy_train_on_reset else "OFF",
)
try:
srv.serve_forever()
except KeyboardInterrupt:
LOGGER.info("shutting down")
srv.shutdown()
finally:
if model is not None:
model.stop()