예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
verify_coords.py
"""좌표 grounding 검증 — 모델이 격자에서 (row,col) 을 올바르게 읽는지 측정.
SGF/학습 라벨이 맞아도, 모델이 텍스트 격자의 좌표축을 잘못 읽으면
다음 수 학습은 전부 붕괴합니다 (과거 Gemma-E4B / 일부 Google 계열 이슈).
테스트 종류:
locate — 돌 1~N개를 두고 "흑/백 돌 좌표는?" 질문
lookup — "(r,c) 칸의 내용은?" 질문
empty — 빈 칸 좌표를 주고 "비었는가?" (함정)
사용:
# 베이스 Qwen, aligned 텍스트만
python -m vlm_server.verify_coords \\
--model vlm_models/Qwen__Qwen3-VL-8B-Instruct \\
--board-format aligned --text-only --trials 48
# 스모크 LoRA 적용
python -m vlm_server.verify_coords \\
--model vlm_models/Qwen__Qwen3-VL-8B-Instruct \\
--lora-dir vlm_models/lora_baduk_v3_qwen_smoke \\
--board-format aligned --text-only --trials 48 \\
--out-jsonl logs/vlm_eval_day/coord_qwen_smoke.jsonl
"""
from __future__ import annotations
import argparse
import json
import logging
import random
import re
import time
from collections import Counter
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from .dataset_baduk import (
BOARD_SIZE,
board_text_aligned,
board_text_compact,
board_text,
)
from .server import _MlxVlmModel, parse_move
LOGGER = logging.getLogger("vlm_server.verify_coords")
LOCATE_SYSTEM = (
"당신은 19×19 바둑 보드 텍스트 격자를 읽는 검사기입니다. "
"좌표는 0-indexed 정수 (row, col) 이며 각각 0..18 입니다. "
"행 번호는 각 줄 왼쪽, 열 번호는 맨 위 헤더입니다. "
"각 칸 기호(X/O/.)는 두 글자 폭의 왼쪽에 있습니다. "
"응답은 설명 없이 JSON 한 줄만 출력하세요."
)
LOOKUP_SYSTEM = (
"당신은 19×19 바둑 보드 텍스트 격자를 읽는 검사기입니다. "
"좌표는 0-indexed (row, col), 각 0..18. "
"각 칸 기호는 두 글자 폭의 왼쪽에 있습니다. "
'지정한 칸의 내용만 JSON 한 줄로 답하세요. 예: {"cell": "."} 또는 {"cell": "X"} 또는 {"cell": "O"}.'
)
@dataclass
class TrialResult:
kind: str
ok: bool
expected: Dict[str, Any]
got: Dict[str, Any]
raw: str
ms: float
quadrant: str
board_format: str
def _quadrant(r: int, c: int) -> str:
if r < 9.5 and c < 9.5:
return "TL"
if r < 9.5 and c >= 9.5:
return "TR"
if r >= 9.5 and c < 9.5:
return "BL"
return "BR"
def _render_board(cells: bytes, board_format: str) -> str:
if board_format == "compact":
return board_text_compact(cells)
if board_format == "spaced":
return board_text(cells)
return board_text_aligned(cells)
def _pack_stones(stones: List[Tuple[int, int, int]]) -> bytes:
"""stones: list of (r, c, color) color 1=B 2=W."""
b = bytearray(BOARD_SIZE * BOARD_SIZE)
for r, c, color in stones:
b[r * BOARD_SIZE + c] = color
return bytes(b)
def _parse_cell(text: str) -> Optional[str]:
if not text:
return None
m = re.search(r'"cell"\s*:\s*"([.XOxo])"', text)
if m:
ch = m.group(1).upper()
return "." if ch == "." else ch
m = re.search(r"\b([.XO])\b", text)
if m:
return m.group(1).upper() if m.group(1) != "." else "."
return None
def _parse_points(text: str) -> List[Tuple[int, int]]:
"""JSON points 배열 또는 단일 row/col."""
pts: List[Tuple[int, int]] = []
if not text:
return pts
try:
for m in re.finditer(r"\{[^{}]*\}", text):
chunk = m.group(0)
if re.search(r'"pass"\s*:\s*true', chunk, re.I):
continue
if "points" in chunk:
continue
r, c, is_pass = parse_move(chunk)
if is_pass or r is None or c is None:
continue
if 0 <= r < BOARD_SIZE and 0 <= c < BOARD_SIZE:
pts.append((r, c))
except Exception:
pass
# {"points":[{"row":1,"col":2},...]}
m = re.search(r'"points"\s*:\s*(\[[^\]]*\])', text, re.S)
if m:
try:
arr = json.loads(m.group(1))
for item in arr:
if isinstance(item, dict) and "row" in item and "col" in item:
pts.append((int(item["row"]), int(item["col"])))
except Exception:
pass
# dedupe preserve order
seen = set()
out = []
for p in pts:
if p not in seen:
seen.add(p)
out.append(p)
return out
# 사분면·가장자리·중앙을 고르게 덮는 고정 좌표 세트
PROBE_COORDS: List[Tuple[int, int]] = [
(0, 0),
(0, 9),
(0, 18),
(3, 3),
(3, 15),
(9, 0),
(9, 9),
(9, 18),
(15, 3),
(15, 15),
(18, 0),
(18, 9),
(18, 18),
(2, 16),
(16, 2),
(4, 14),
(14, 4),
(7, 11),
(11, 7),
(1, 17),
(17, 1),
(5, 5),
(5, 13),
(13, 5),
(13, 13),
]
def build_cases(rng: random.Random, trials: int) -> List[Dict[str, Any]]:
cases: List[Dict[str, Any]] = []
# 1) 고정 좌표 locate (흑 1개)
for r, c in PROBE_COORDS:
cases.append({"kind": "locate", "stones": [(r, c, 1)], "ask_color": 1, "target": [(r, c)]})
# 2) 고정 lookup
for r, c in PROBE_COORDS[:12]:
cases.append({"kind": "lookup", "stones": [(r, c, 1)], "query": (r, c), "expect_cell": "X"})
# 근처 빈칸
er, ec = (r + 1) % BOARD_SIZE, c
if (er, ec) != (r, c):
cases.append(
{"kind": "lookup", "stones": [(r, c, 1)], "query": (er, ec), "expect_cell": "."}
)
# 3) 랜덤 locate / 2돌
while len(cases) < trials:
kind = rng.choice(["locate", "locate2", "lookup"])
if kind == "locate":
r, c = rng.randrange(BOARD_SIZE), rng.randrange(BOARD_SIZE)
cases.append(
{"kind": "locate", "stones": [(r, c, 1)], "ask_color": 1, "target": [(r, c)]}
)
elif kind == "locate2":
pts = set()
while len(pts) < 2:
pts.add((rng.randrange(BOARD_SIZE), rng.randrange(BOARD_SIZE)))
stones = [(r, c, 1) for r, c in pts]
cases.append(
{"kind": "locate", "stones": stones, "ask_color": 1, "target": sorted(pts)}
)
else:
r, c = rng.randrange(BOARD_SIZE), rng.randrange(BOARD_SIZE)
color = rng.choice([1, 2])
cases.append(
{
"kind": "lookup",
"stones": [(r, c, color)],
"query": (r, c),
"expect_cell": "X" if color == 1 else "O",
}
)
return cases[:trials]
def _run_one(
model: _MlxVlmModel,
case: Dict[str, Any],
board_format: str,
timeout_s: float,
) -> TrialResult:
board = _pack_stones(case["stones"])
grid = _render_board(board, board_format)
t0 = time.perf_counter()
if case["kind"] == "locate":
color_name = "흑(X)" if case["ask_color"] == 1 else "백(O)"
n = len(case["target"])
if n == 1:
user = (
f"보드 문자 격자:\n{grid}\n\n"
f"보드 위에 {color_name} 돌이 정확히 1개 있습니다. "
f'그 돌의 좌표만 JSON 한 줄로 답하세요. 예: {{"row": 3, "col": 3}}'
)
else:
user = (
f"보드 문자 격자:\n{grid}\n\n"
f"보드 위에 {color_name} 돌이 정확히 {n}개 있습니다. "
f'모든 좌표를 JSON 한 줄로 답하세요. 예: {{"points":[{{"row":1,"col":2}},{{"row":3,"col":4}}]}}'
)
system = LOCATE_SYSTEM
raw = model.predict_raw(system, user, num_images=0, timeout_s=timeout_s, temperature=0.0)
got_pts = sorted(_parse_points(raw))
exp_pts = sorted(tuple(x) for x in case["target"])
ok = got_pts == exp_pts
# 대표 좌표의 사분면
qr, qc = exp_pts[0]
return TrialResult(
kind="locate",
ok=ok,
expected={"points": exp_pts},
got={"points": got_pts},
raw=(raw or "")[:500],
ms=(time.perf_counter() - t0) * 1000,
quadrant=_quadrant(qr, qc),
board_format=board_format,
)
# lookup
qr, qc = case["query"]
user = (
f"보드 문자 격자:\n{grid}\n\n"
f"칸 (row={qr}, col={qc}) 의 내용은 무엇입니까? "
f'JSON 한 줄만: {{"cell": "."}} 또는 {{"cell": "X"}} 또는 {{"cell": "O"}}'
)
raw = model.predict_raw(
LOOKUP_SYSTEM, user, num_images=0, timeout_s=timeout_s, temperature=0.0
)
cell = _parse_cell(raw or "")
ok = cell == case["expect_cell"]
return TrialResult(
kind="lookup",
ok=ok,
expected={"cell": case["expect_cell"], "at": [qr, qc]},
got={"cell": cell},
raw=(raw or "")[:500],
ms=(time.perf_counter() - t0) * 1000,
quadrant=_quadrant(qr, qc),
board_format=board_format,
)
def summarize(results: List[TrialResult]) -> Dict[str, Any]:
by_kind = Counter()
ok_kind = Counter()
by_q = Counter()
ok_q = Counter()
for r in results:
by_kind[r.kind] += 1
by_q[r.quadrant] += 1
if r.ok:
ok_kind[r.kind] += 1
ok_q[r.quadrant] += 1
total = len(results) or 1
return {
"n": len(results),
"accuracy": round(100.0 * sum(1 for r in results if r.ok) / total, 1),
"by_kind": {
k: {
"n": by_kind[k],
"ok": ok_kind[k],
"acc": round(100.0 * ok_kind[k] / max(1, by_kind[k]), 1),
}
for k in sorted(by_kind)
},
"by_quadrant": {
q: {
"n": by_q[q],
"ok": ok_q[q],
"acc": round(100.0 * ok_q[q] / max(1, by_q[q]), 1),
}
for q in ("TL", "TR", "BL", "BR")
},
"avg_ms": round(sum(r.ms for r in results) / total, 1),
}
def main(argv: Optional[List[str]] = None) -> int:
p = argparse.ArgumentParser(description="좌표 grounding 검증")
p.add_argument("--model", required=True)
p.add_argument("--lora-dir", default=None)
p.add_argument(
"--board-format",
choices=["aligned", "compact", "spaced"],
default="aligned",
)
p.add_argument("--trials", type=int, default=48)
p.add_argument("--seed", type=int, default=42)
p.add_argument("--timeout-s", type=float, default=180.0)
p.add_argument("--max-new-tokens", type=int, default=96)
p.add_argument("--out-jsonl", default=None)
p.add_argument("--out-summary", default=None)
args = p.parse_args(argv)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
rng = random.Random(args.seed)
cases = build_cases(rng, args.trials)
LOGGER.info(
"cases=%d format=%s model=%s lora=%s",
len(cases),
args.board_format,
args.model,
args.lora_dir,
)
model = _MlxVlmModel(
args.model,
max_new_tokens=args.max_new_tokens,
temperature=0.0,
lora_dir=args.lora_dir,
text_only_moves=True,
board_repr="aligned" if args.board_format == "aligned" else "chars",
)
model.start(wait_load=True, load_timeout_s=600.0)
if model._load_error is not None:
LOGGER.error("모델 로드 실패: %s", model._load_error)
model.stop()
return 2
results: List[TrialResult] = []
out_f = open(args.out_jsonl, "w", encoding="utf-8") if args.out_jsonl else None
try:
for i, case in enumerate(cases, 1):
tr = _run_one(model, case, args.board_format, args.timeout_s)
results.append(tr)
mark = "OK" if tr.ok else "FAIL"
LOGGER.info(
"[%d/%d] %s %s q=%s exp=%s got=%s (%.0fms)",
i,
len(cases),
mark,
tr.kind,
tr.quadrant,
tr.expected,
tr.got,
tr.ms,
)
if out_f:
out_f.write(json.dumps(asdict(tr), ensure_ascii=False) + "\n")
out_f.flush()
finally:
if out_f:
out_f.close()
model.stop()
summary = summarize(results)
summary["model"] = args.model
summary["lora_dir"] = args.lora_dir
summary["board_format"] = args.board_format
print(json.dumps(summary, ensure_ascii=False, indent=2))
if args.out_summary:
Path(args.out_summary).write_text(
json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8"
)
# 게이트: locate 정확도 90% 미만이면 실패 코드
loc = summary["by_kind"].get("locate", {})
if loc and loc.get("acc", 0) < 90.0:
LOGGER.warning(
"좌표 locate 정확도 %.1f%% < 90%% — 다음 수 학습 전에 포맷/모델을 재검토하세요.",
loc.get("acc", 0),
)
return 3
return 0
if __name__ == "__main__":
raise SystemExit(main())