예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
verify_board.py
"""보드 인식 검증: 모델이 19×19 격자를 JSON으로 복원하는지 정답과 비교.
사용 예:
python -m vlm_server verify-board --sgf recordings/x.sgf --ply -1 \\
--model mlx-community/gemma-4-e4b-it-4bit
# 이미지까지 넣어 검증
python -m vlm_server verify-board --sgf x.sgf --ply 50 --with-image
LoRA 없이 베이스만:
python -m vlm_server verify-board --sgf x.sgf --model ... ( --lora-dir 생략 )
"""
from __future__ import annotations
import argparse
import json
import logging
import re
import sys
from typing import Any, List, Optional, Tuple
from PIL import Image
from .dataset_baduk import BOARD_SIZE, board_numeric_grid, board_text, render_board_png
from .probe_move import _build_position_from_sgf
from .server import _MlxVlmModel
LOGGER = logging.getLogger("vlm_server.verify_board")
VERIFY_SYSTEM = (
"당신은 19×19 바둑판 상태를 읽어 복사하는 도우미입니다. "
"아래에 주어진 보드(텍스트 격자, 필요 시 이미지)를 정확히 읽고, "
'응답은 설명 없이 JSON 한 덩어리만 출력하세요. 형식: {"rows": [...]} '
"여기서 rows 는 길이 정확히 19 인 문자열 배열이고, 각 문자열은 공백 없이 정확히 19 글자입니다. "
"각 칸은 반드시 '.', 'X', 'O' 만 사용합니다 (.=빈칸, X=흑, O=백). "
"rows 의 각 문자열 안에는 공백·줄바꿈·탭을 절대 넣지 마세요 (19글자가 한 덩어리여야 함). "
"텍스트 격자에서 행 번호·열 번호·칸 사이 공백은 레이블일 뿐이며 rows 에 넣지 마세요. "
"row 0 이 맨 위 줄입니다. 다른 키나 문장은 넣지 마세요."
)
def expected_rows(board: bytes) -> List[str]:
if len(board) != BOARD_SIZE * BOARD_SIZE:
raise ValueError("board length")
rows: List[str] = []
for r in range(BOARD_SIZE):
line = []
for c in range(BOARD_SIZE):
v = board[r * BOARD_SIZE + c]
line.append("." if v == 0 else ("X" if v == 1 else "O"))
rows.append("".join(line))
return rows
def _normalize_cell(ch: str) -> str:
ch = ch.strip()
if ch == ".":
return "."
u = ch.upper()
if u == "X":
return "X"
if u == "O":
return "O"
return "?"
def _strip_row_cells(row: str) -> str:
"""rows[i] 에서 칸 사이에 모델이 넣은 공백·줄바꿈 등을 모두 제거."""
return re.sub(r"\s+", "", row)
def _rows_array_from_model_text(text: str) -> tuple[Optional[List[Any]], Optional[str]]:
"""텍스트에서 JSON 의 rows 배열만 꺼냄(개수는 검사하지 않음)."""
if not text:
return None, "빈 응답"
s = text.strip()
fence = re.search(r"```(?:json)?\s*([\s\S]*?)\s*```", s, re.IGNORECASE)
if fence:
s = fence.group(1).strip()
obj: Any = None
try:
obj = json.loads(s)
except json.JSONDecodeError:
for m in re.finditer(r"\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}", s, re.DOTALL):
try:
obj = json.loads(m.group(0))
break
except json.JSONDecodeError:
continue
if not isinstance(obj, dict) or "rows" not in obj:
return None, '{"rows": [...]} 형식의 JSON 을 찾지 못했습니다.'
rows = obj["rows"]
if not isinstance(rows, list):
return None, "rows 가 JSON 배열이 아닙니다."
return rows, None
def _extract_rows_from_model_json(text: str) -> tuple[Optional[List[Any]], Optional[str]]:
"""rows 배열 + 정확히 19줄."""
rows, err = _rows_array_from_model_text(text)
if err or rows is None:
return None, err
if len(rows) != BOARD_SIZE:
return None, f"rows 개수가 {len(rows)} 입니다 (정확히 {BOARD_SIZE} 줄 필요)."
return rows, None
def describe_model_grid_issues(text: str) -> str:
"""parse_model_grid 가 실패할 때, 모델 출력이 어디가 잘못됐는지 사람이 읽을 수 있게."""
rows, err = _rows_array_from_model_text(text)
if err:
return err
if len(rows) != BOARD_SIZE:
return f"rows 개수가 {len(rows)} 입니다 (정확히 {BOARD_SIZE} 줄 필요)."
lines: List[str] = []
for i, row in enumerate(rows):
if not isinstance(row, str):
lines.append(f" 행 {i}: 문자열이 아님 (타입 {type(row).__name__})")
continue
compact = _strip_row_cells(row)
raw_has_ws = bool(re.search(r"\s", row))
if len(compact) != BOARD_SIZE:
hint = " (원문에 공백/줄바꿈 포함)" if raw_has_ws else ""
preview = compact[:30] + ("…" if len(compact) > 30 else "")
lines.append(
f" 행 {i}: 칸 수 {len(compact)}개{hint}, 필요 {BOARD_SIZE}개 — "
f"공백 제거 후 앞부분 {preview!r}"
)
continue
norm = "".join(_normalize_cell(compact[j : j + 1]) for j in range(BOARD_SIZE))
if "?" in norm:
lines.append(f" 행 {i}: '.', 'X', 'O' 가 아닌 글자 포함 — {compact!r}")
if not lines:
return "행 단위 오류는 없어 보이나 다른 이유로 파싱에 실패했습니다."
return "각 rows[i] 는 공백 없이 정확히 19글자(. X O 만)여야 합니다.\n" + "\n".join(lines)
def parse_model_grid(text: str) -> Optional[List[str]]:
if not text:
return None
rows, err = _extract_rows_from_model_json(text)
if err or rows is None:
return None
out: List[str] = []
for i, row in enumerate(rows):
if not isinstance(row, str):
return None
row = _strip_row_cells(row)
if len(row) != BOARD_SIZE:
return None
norm = "".join(_normalize_cell(row[j : j + 1]) for j in range(BOARD_SIZE))
if "?" in norm:
return None
out.append(norm)
return out
def _normalize_one_row_string(cell: Any) -> Optional[str]:
if not isinstance(cell, str):
return None
row = _strip_row_cells(cell)
if len(row) != BOARD_SIZE:
return None
norm = "".join(_normalize_cell(row[j : j + 1]) for j in range(BOARD_SIZE))
if "?" in norm:
return None
return norm
def repair_model_grid_with_packed_board(
text: str, board: bytes
) -> tuple[Optional[List[str]], Optional[str]]:
"""모델이 rows 를 18줄만 줄 때, packed 보드 정답으로 빠진 한 줄을 끼워 넣는다.
Returns:
(19줄 격자, 사용자에게 보여줄 보정 설명) 또는 (None, None) if 보정 불가.
"""
rows, err = _rows_array_from_model_text(text)
if err or rows is None or len(board) != BOARD_SIZE * BOARD_SIZE:
return None, None
if len(rows) != BOARD_SIZE - 1:
return None, None
exp = expected_rows(board)
for m in range(BOARD_SIZE):
out: List[str] = []
j = 0
ok = True
for r in range(BOARD_SIZE):
if r == m:
out.append(exp[r])
continue
if j >= len(rows):
ok = False
break
line = _normalize_one_row_string(rows[j])
if line is None:
ok = False
break
out.append(line)
j += 1
if not ok or j != BOARD_SIZE - 1:
continue
note = (
f"모델이 rows 를 {BOARD_SIZE - 1}줄만 출력하여, "
f"요청에 포함된 packed 보드 기준으로 행 {m} 을 채워 19줄로 맞춤."
)
return out, note
return None, None
def diff_grids(exp: List[str], got: List[str]) -> Tuple[int, str]:
n = 0
lines = []
for r in range(BOARD_SIZE):
if exp[r] != got[r]:
for c in range(BOARD_SIZE):
if exp[r][c] != got[r][c]:
n += 1
lines.append(f"row {r}: exp={exp[r]!r} got={got[r]!r}")
return n, "\n".join(lines[:24]) + ("\n..." if len(lines) > 24 else "")
def build_verify_user_prompt(board: bytes) -> str:
return (
"아래는 동일한 국면의 두 가지 텍스트 표현입니다. 둘 다 읽고 rows 를 채우세요.\n"
"문자 격자는 첫 줄이 열 번호(00~18), 아래 각 줄은 '행번호 + 공백 + 19칸(공백으로 구분)' 입니다. "
"19칸을 세면 반드시 19개여야 합니다.\n\n"
"문자 격자 (X=흑, O=백, .=빈):\n"
f"{board_text(board)}\n\n"
"숫자 격자 (0=빈, 1=흑, 2=백):\n"
f"{board_numeric_grid(board)}\n"
)
def main(argv: Optional[list[str]] = None) -> int:
p = argparse.ArgumentParser(description="모델 보드 인식 검증 (격자 복원)")
p.add_argument("--sgf", required=True)
p.add_argument("--ply", type=int, default=-1)
p.add_argument("--model", default="********")
p.add_argument("--lora-dir", default=None)
p.add_argument("--max-new-tokens", type=int, default=1024)
p.add_argument("--temperature", type=float, default=0.0)
p.add_argument(
"--with-image",
action="store_true",
help="보드 PNG를 함께 넣어 멀티모달로 검증",
)
args = p.parse_args(argv)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
board, _side, _mt, _ma, stones = _build_position_from_sgf(args.sgf, args.ply)
exp = expected_rows(board)
LOGGER.info("stones=%d (정답 격자 %dx%d)", stones, BOARD_SIZE, BOARD_SIZE)
img: Optional[Image.Image] = None
if args.with_image:
img = render_board_png(board, BOARD_SIZE, px=768)
user = build_verify_user_prompt(board)
model = _MlxVlmModel(
args.model,
max_new_tokens=args.max_new_tokens,
temperature=args.temperature,
lora_dir=args.lora_dir,
text_only_moves=False,
board_repr="both",
)
model.start(wait_load=True)
try:
raw = model.predict_raw(
VERIFY_SYSTEM,
user,
image=img,
num_images=1 if args.with_image else 0,
timeout_s=300.0,
max_tokens=args.max_new_tokens,
temperature=args.temperature,
)
finally:
model.stop()
print("--- model raw (앞 800자) ---")
print((raw or "")[:800])
print("--- end raw ---")
got = parse_model_grid(raw)
if got is None:
print("FAIL: JSON rows 파싱 실패")
print(describe_model_grid_issues(raw))
return 2
mism, detail = diff_grids(exp, got)
if mism == 0:
print(f"OK: 격자 일치 (stones={stones})")
return 0
print(f"FAIL: 불일치 칸 수={mism} / {BOARD_SIZE*BOARD_SIZE}")
print(detail)
return 1
def _unit_test_parse() -> None:
exp = expected_rows(bytes([0]) * (BOARD_SIZE * BOARD_SIZE))
j = json.dumps({"rows": exp}, ensure_ascii=False)
got = parse_model_grid(j)
assert got == exp
bad = parse_model_grid('{"rows": ["' + "." * BOARD_SIZE + '"]}')
assert bad is None
# 18줄 → packed 보드로 한 줄 보정
board = bytearray(BOARD_SIZE * BOARD_SIZE)
board[0] = 1
board[BOARD_SIZE + 1] = 2
board_b = bytes(board)
exp2 = expected_rows(board_b)
eighteen = [exp2[i] for i in range(BOARD_SIZE) if i != 7]
j2 = json.dumps({"rows": eighteen}, ensure_ascii=False)
rep, note = repair_model_grid_with_packed_board(j2, board_b)
assert rep == exp2 and note is not None
print("verify_board parse self-test OK")
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "--self-test":
_unit_test_parse()
else:
raise SystemExit(main())