예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
dataset_coords_game.py
#!/usr/bin/env python3
"""실전 국면 좌표 학습 데이터 (Gemma 좌표 인식 보정).
coord_v3 는 돌 1~2개짜리 빈 판이라, 돌이 많은 실전 판에서는 열을 세다 놓친다.
대국 기록(sqlite) 국면에서 아래 과제를 만든다. 보드 격자·수 프롬프트는 서버와 같은 문자열.
lookup (r,c) 칸의 기호 — 모델이 실제로 고른 이미 돌 있는 칸(서버 jsonl)을 우선
row 한 행의 돌 위치 전부 (그 행은 격자에서 이미 한 줄)
col 한 열의 돌 위치. 19줄을 세로로 세지 않고 그 열만 한 줄로 준다
move 서버 수 프롬프트 그대로, 정답은 휴리스틱 수. 절반은 금지 좌표가 붙은 재시도 프롬프트
python scripts/dataset_coords_game.py --source logs/x/matches.sqlite --jsonl logs/x/gemma4.jsonl \
--out sft_data/gemma_coord_r1
"""
from __future__ import annotations
import argparse
import importlib.util
import json
import random
import sqlite3
import sys
import zlib
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from vlm_server.dataset_baduk import assistant_text, board_text_aligned, column_scan_user # noqa: E402
from vlm_server.server import SYSTEM_PROMPT_ALIGNED, build_prompt # noqa: E402
from vlm_server.verify_coords import LOOKUP_SYSTEM # noqa: E402
N = 19
SCAN_SYSTEM = (
"당신은 19×19 바둑 보드 텍스트 격자를 읽는 검사기입니다. "
"좌표는 0-indexed (row, col), 각 0..18. 행 번호는 각 줄 왼쪽, 열 번호는 맨 위 헤더입니다. "
"각 칸 기호는 두 글자 폭의 왼쪽에 있습니다. 응답은 설명 없이 JSON 한 줄만 출력하세요."
)
COL_SYSTEM = (
"당신은 19×19 바둑 보드의 한 열을 읽는 검사기입니다. "
"좌표는 0-indexed 이며 행 번호는 0..18 입니다. "
"한 줄의 각 토큰은 행 번호 두 자리와 기호입니다. "
"응답은 설명 없이 JSON 한 줄만 출력하세요."
)
HINTS = ["occupied", "occupied", "occupied", "suicide"]
def _monitor():
path = ROOT / "scripts" / "monitor_mariago_web_play.py"
spec = importlib.util.spec_from_file_location("monitor_mariago_web_play", path)
mod = importlib.util.module_from_spec(spec)
sys.modules["monitor_mariago_web_play"] = mod
spec.loader.exec_module(mod)
return mod
mon = _monitor()
def board_hash(cells: bytes, side: int) -> str:
return f"{zlib.crc32(bytes(cells) + bytes([side & 0xFF])) & 0xFFFFFFFF:08x}"
def load_positions(db: Path) -> list[dict]:
con = sqlite3.connect(db)
rows = con.execute(
"select game_id, ply, color, board from moves where kind='move' and board is not null order by game_id, ply"
).fetchall()
con.close()
out = []
for gid, ply, color, text in rows:
if not text or len(text) != N * N:
continue
try:
cells = bytes(".XO".index(ch) for ch in text)
except ValueError:
continue
if sum(1 for v in cells if v) < 8:
continue
side = 1 if color == "B" else 2
out.append({"src": f"{db.parent.name}:{gid}:{ply}", "cells": cells, "side": side})
return out
def load_bad_picks(paths: list[Path]) -> dict[str, list[tuple[int, int]]]:
"""board_hash → 모델이 그 국면에서 고른 좌표(서버 재시도 포함)."""
picks: dict[str, list[tuple[int, int]]] = {}
for p in paths:
if not p.exists():
continue
for line in p.open(encoding="utf-8", errors="replace"):
try:
rec = json.loads(line)
except json.JSONDecodeError:
continue
if rec.get("event") != "move" or not rec.get("board_hash"):
continue
seeds = [(int(r), int(c)) for r, c in (rec.get("vlm_seeds") or []) if r is not None and c is not None]
if seeds:
picks.setdefault(rec["board_hash"], []).extend(seeds)
return picks
def record(task: str, system: str, user: str, answer: str, src: str, expect: dict) -> dict:
return {
"task": task,
"src": src,
"sys": system,
"usr": user,
"expect": expect,
"images": [],
"messages": [
{"role": "user", "content": [{"type": "text", "text": f"{system}\n\n{user}"}]},
{"role": "assistant", "content": [{"type": "text", "text": answer}]},
],
}
def near_empty(cells: bytes, rng: random.Random) -> tuple[int, int] | None:
cand = []
for r in range(N):
for c in range(N):
if cells[r * N + c]:
continue
if any(
0 <= r + dr < N and 0 <= c + dc < N and cells[(r + dr) * N + c + dc]
for dr in (-2, -1, 0, 1, 2) for dc in (-2, -1, 0, 1, 2)
):
cand.append((r, c))
return rng.choice(cand) if cand else None
def make_lookup(pos: dict, bad: list[tuple[int, int]], rng: random.Random) -> dict | None:
cells = pos["cells"]
stones = [(i // N, i % N) for i, v in enumerate(cells) if v]
occupied_bad = [p for p in bad if cells[p[0] * N + p[1]]]
roll = rng.random()
if occupied_bad and roll < 0.4:
r, c = rng.choice(occupied_bad)
elif roll < 0.65:
r, c = rng.choice(stones)
else:
pt = near_empty(cells, rng)
if pt is None:
return None
r, c = pt
cell = ".XO"[cells[r * N + c]]
user = (
f"보드 문자 격자:\n{board_text_aligned(cells)}\n\n"
f"칸 (row={r}, col={c}) 의 내용은 무엇입니까? "
f'JSON 한 줄만: {{"cell": "."}} 또는 {{"cell": "X"}} 또는 {{"cell": "O"}}'
)
return record("lookup", LOOKUP_SYSTEM, user, f'{{"cell": "{cell}"}}', pos["src"],
{"cell": cell, "at": [r, c], "hard": (r, c) in occupied_bad})
def make_scan(pos: dict, rng: random.Random, axis: str) -> dict | None:
cells = pos["cells"]
lines = []
for k in range(N):
pts = [(k, j) if axis == "row" else (j, k) for j in range(N)]
n = sum(1 for r, c in pts if cells[r * N + c])
if n >= 2:
lines.append((k, n))
if not lines:
return None
k = rng.choices([x for x, _ in lines], weights=[n for _, n in lines])[0]
xs, os_ = [], []
for j in range(N):
r, c = (k, j) if axis == "row" else (j, k)
v = cells[r * N + c]
if v == 1:
xs.append(j)
elif v == 2:
os_.append(j)
ans = json.dumps({"X": xs, "O": os_})
expect = {"X": xs, "O": os_, "line": k}
if axis == "row":
ask = f"행 row={k} 에 있는 돌들의 열(col) 번호를 색깔별로 모두 적으세요."
user = (
f"보드 문자 격자:\n{board_text_aligned(cells)}\n\n{ask} "
f'JSON 한 줄만. 예: {{"X": [3, 4], "O": [10]}}'
)
return record(axis, SCAN_SYSTEM, user, ans, pos["src"], expect)
user = column_scan_user(cells, k)
return record(axis, COL_SYSTEM, user, ans, pos["src"], expect)
def make_move(pos: dict, bad: list[tuple[int, int]], rng: random.Random, retry: bool) -> dict | None:
cells = pos["cells"]
side = pos["side"]
board = list(cells)
forbidden: list[tuple[int, int]] = []
hint = None
if retry:
occ = sorted({p for p in bad if cells[p[0] * N + p[1]]})
if not occ:
stones = [(i // N, i % N) for i, v in enumerate(cells) if v == side]
occ = rng.sample(stones, min(len(stones), rng.randint(1, 3))) if stones else []
forbidden = occ[:6]
hint = rng.choice(HINTS) if forbidden else None
hr, hc, is_pass = mon.heuristic_move(board, side, None, random.Random(hash(pos["src"]) & 0xFFFF),
banned=set(forbidden) or None)
if is_pass or hr is None or hc is None:
return None
ok, _reason, _nxt, _cap, _ko = mon.try_play(board, int(hr), int(hc), side, None)
if not ok or (int(hr), int(hc)) in forbidden:
return None
user = build_prompt(bytes(cells), side, forbidden=forbidden or None, illegal_hint=hint, board_repr="aligned")
return record("move_retry" if retry else "move", SYSTEM_PROMPT_ALIGNED, user,
assistant_text((int(hr), int(hc))), pos["src"],
{"row": int(hr), "col": int(hc), "side": side, "forbidden": forbidden})
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--source", action="append", required=True, help="matches.sqlite (반복 가능)")
ap.add_argument("--jsonl", action="append", default=[], help="모델 서버 jsonl (반복 가능, 틀린 좌표 수집)")
ap.add_argument("--out", required=True)
ap.add_argument("--n", type=int, default=2400, help="train 샘플 수")
ap.add_argument("--mix", default="lookup:0.35,row:0.15,col:0.10,move:0.15,move_retry:0.25")
ap.add_argument("--val-n", type=int, default=240)
ap.add_argument("--seed", type=int, default=29)
args = ap.parse_args()
rng = random.Random(args.seed)
positions: list[dict] = []
for s in args.source:
positions.extend(load_positions(Path(s)))
bad = load_bad_picks([Path(p) for p in args.jsonl])
if not positions:
print("[coord-game] no positions", flush=True)
return 1
# 국면 단위로 val 분리 (같은 판이 train/val 에 같이 들어가지 않게)
games = sorted({p["src"].rsplit(":", 1)[0] for p in positions})
rng.shuffle(games)
val_games = set(games[: max(1, len(games) // 10)])
train_pos = [p for p in positions if p["src"].rsplit(":", 1)[0] not in val_games]
val_pos = [p for p in positions if p["src"].rsplit(":", 1)[0] in val_games]
mix = [(k, float(v)) for k, v in (x.split(":") for x in args.mix.split(","))]
def occupied_bad(pos: dict) -> list[tuple[int, int]]:
cells = pos["cells"]
return [p for p in bad.get(board_hash(cells, pos["side"]), []) if cells[p[0] * N + p[1]]]
def build(pool: list[dict], n: int) -> list[dict]:
hard_pool = [p for p in pool if occupied_bad(p)]
out: list[dict] = []
tries = 0
while len(out) < n and tries < n * 20:
tries += 1
task = rng.choices([k for k, _ in mix], weights=[w for _, w in mix])[0]
if task in ("lookup", "move_retry") and hard_pool and rng.random() < 0.6:
pos = rng.choice(hard_pool)
else:
pos = rng.choice(pool)
b = bad.get(board_hash(pos["cells"], pos["side"]), [])
if task == "lookup":
rec = make_lookup(pos, b, rng)
elif task in ("row", "col"):
rec = make_scan(pos, rng, task)
else:
rec = make_move(pos, b, rng, retry=task == "move_retry")
if rec:
rec["board"] = "".join(".XO"[v] for v in pos["cells"])
out.append(rec)
return out
train = build(train_pos, args.n)
val = build(val_pos, args.val_n)
out = Path(args.out)
out.mkdir(parents=True, exist_ok=True)
for name, rows in (("train", train), ("val", val)):
with (out / f"{name}.jsonl").open("w", encoding="utf-8") as f:
for r in rows:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
count = lambda rows: {k: sum(1 for r in rows if r["task"] == k) for k, _ in mix} # noqa: E731
summary = {
"positions": len(positions),
"games": len(games),
"val_games": len(val_games),
"bad_pick_boards": len(bad),
"hard_lookups": sum(1 for r in train if r["task"] == "lookup" and r["expect"].get("hard")),
"train": count(train),
"val": count(val),
}
(out / "summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(f"[coord-game] {summary}", flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())