예제 소스
오목으로 이해하는 자기대국과 모방 학습 · 파일 59개
coord_omook.py
#!/usr/bin/env python3
"""오목 VLM 좌표 grounding: 정렬 텍스트 판, 좌표 문제 데이터, MLX 검사.
바둑(go/vlm_server)의 좌표 학습을 오목 15×15·19×19 로 옮긴 것.
mlx-vlm 이 필요하므로 바둑 가상환경으로 실행한다.
../go/.venv/bin/python coord_omook.py dataset --out-dir sft_data/coord_omook_r1 --n-samples 6000
../go/.venv/bin/python coord_omook.py verify --format aligned \\
--model vlm_models/Qwen__Qwen3-VL-8B-Instruct [--lora-dir vlm_models/lora_coord_omook]
../go/.venv/bin/python coord_omook.py realboard --format aligned --n 200 \\
--model vlm_models/Qwen__Qwen3-VL-8B-Instruct [--lora-dir ...]
"""
from __future__ import annotations
import argparse
import json
import random
import re
import time
from collections import Counter
from pathlib import Path
from typing import Iterable, List, Optional, Sequence, Tuple
ROOT = Path(__file__).resolve().parent
LOG_JSONL = ROOT / "logs" / "omook_vlm.jsonl"
SIZES = (15, 19)
Board = List[int]
# ---------------------------------------------------------------- board text
def board_grid_aligned(board: Sequence[int], size: int) -> str:
"""열·행·칸이 모두 세 글자 간격. 칸 기호는 두 글자 폭의 왼쪽."""
lines = [" " + " ".join(f"{c:02d}" for c in range(size))]
for r in range(size):
cells = []
for c in range(size):
v = board[r * size + c]
cells.append(f"{'.' if v == 0 else ('X' if v == 1 else 'O'):<2}")
lines.append(f"{r:02d} " + " ".join(cells))
return "\n".join(lines)
def board_grid_legacy(board: Sequence[int], size: int) -> str:
"""지금 서버(vlm_pva_server._board_to_text)의 격자. 헤더는 3칸, 칸은 2칸 간격이라 어긋난다."""
lines = [" " + " ".join(f"{c:2d}" for c in range(size))]
for r in range(size):
cells = []
for c in range(size):
v = board[r * size + c]
cells.append(" ." if v == 0 else (" X" if v == 1 else " O"))
lines.append(f"{r:2d} " + "".join(cells))
return "\n".join(lines)
def board_to_text(board: Sequence[int], size: int, fmt: str = "aligned") -> str:
"""착수 프롬프트용 판 설명. 서버와 같은 머리말과 돌 목록을 붙인다."""
last = size - 1
grid = board_grid_aligned(board, size) if fmt == "aligned" else board_grid_legacy(board, size)
occ_b = [(r, c) for r in range(size) for c in range(size) if board[r * size + c] == 1]
occ_w = [(r, c) for r in range(size) for c in range(size) if board[r * size + c] == 2]
head = (
"Board TEXT is authoritative (ignore fuzzy image coords). "
"X=Black, O=White, .=empty. "
f"row = top→bottom 0..{last}, col = left→right 0..{last}. "
)
if fmt == "aligned":
head += "Each cell symbol sits at the LEFT of its two-character slot under the column number. "
return (
head
+ "Answer with 0-based integers only.\n"
+ grid
+ f"\nBlack stones (row,col): {occ_b}\nWhite stones (row,col): {occ_w}"
)
def move_prompt(board: Sequence[int], size: int, ai_side: int, fmt: str = "aligned") -> str:
"""vlm_pva_server 의 착수 프롬프트(이미지 없음)."""
last = size - 1
you = "Black (X)" if ai_side == 1 else "White (O)"
opp = "White" if ai_side == 1 else "Black"
if not any(board):
mid = size // 2
m = 2 if size <= 15 else 3
turn = f"You are a {size}x{size} Gomoku (오목) player playing {you}."
opening = (
f" OPENING RULE: the board is empty — this is the first move. "
f"Do NOT play in a corner (귀). Corner means both row and col within "
f"{m} of an edge. Prefer the center, e.g. \"{mid} {mid}\"."
)
else:
turn = f"You are a {size}x{size} Gomoku (오목) player. {opp} has just moved; you play {you} next."
opening = ""
return (
f"{turn}\n{board_to_text(board, size, fmt)}\n"
f"Choose your next {you} move on an EMPTY cell (a '.' in the TEXT board)."
f"{opening} "
"Trust the TEXT board coordinates over the image. "
"Reply with EXACTLY two integers separated by a single space: \"row col\". "
f"row and col are both in 0..{last}. No other text."
)
def qa_system(size: int, fmt: str = "aligned") -> str:
slot = "Each cell symbol (X/O/.) sits at the LEFT of its two-character slot. " if fmt == "aligned" else ""
return (
f"You read a {size}x{size} Gomoku text board. "
f"Coordinates are 0-based (row, col), each 0..{size - 1}. "
"Row numbers are at the left of each line; column numbers are in the top header. "
f"{slot}X=Black, O=White, .=empty. Reply with one line of JSON only."
)
def grid_for(board: Sequence[int], size: int, fmt: str) -> str:
return board_grid_aligned(board, size) if fmt == "aligned" else board_grid_legacy(board, size)
def locate_prompt(board: Sequence[int], size: int, color: int, n: int, fmt: str = "aligned") -> str:
name = "Black (X)" if color == 1 else "White (O)"
grid = grid_for(board, size, fmt)
if n == 1:
ask = (
f"There is exactly 1 {name} stone on the board. "
'Reply with its coordinate only. Example: {"row": 3, "col": 3}'
)
else:
ask = (
f"There are exactly {n} {name} stones on the board. "
'Reply with all coordinates. Example: {"points":[{"row":1,"col":2},{"row":3,"col":4}]}'
)
return f"{qa_system(size, fmt)}\n\nBoard:\n{grid}\n\n{ask}"
def lookup_prompt(board: Sequence[int], size: int, r: int, c: int, fmt: str = "aligned") -> str:
grid = grid_for(board, size, fmt)
return (
f"{qa_system(size, fmt)}\n\nBoard:\n{grid}\n\n"
f"What is in cell (row={r}, col={c})? "
'Reply only: {"cell": "."} or {"cell": "X"} or {"cell": "O"}'
)
# ---------------------------------------------------------------- boards
def empty_board(size: int) -> Board:
return [0] * (size * size)
def put(board: Board, size: int, stones: Iterable[Tuple[int, int, int]]) -> Board:
for r, c, v in stones:
board[r * size + c] = v
return board
def synth_game_board(rng: random.Random, size: int, n_moves: int) -> Board:
"""돌이 서로 가까이 모이는 오목 중반 판을 흉내 낸다."""
b = empty_board(size)
mid = size // 2
occupied: List[Tuple[int, int]] = []
color = 1
for _ in range(n_moves):
for _try in range(50):
if not occupied:
r, c = mid + rng.randint(-2, 2), mid + rng.randint(-2, 2)
else:
br, bc = rng.choice(occupied)
r, c = br + rng.randint(-2, 2), bc + rng.randint(-2, 2)
if 0 <= r < size and 0 <= c < size and b[r * size + c] == 0:
break
else:
continue
b[r * size + c] = color
occupied.append((r, c))
color = 3 - color
return b
_LOG_BOARDS: Optional[List[Tuple[int, Board]]] = None
def log_vlm_requests() -> List[dict]:
"""VLM 착수 요청 기록(19×19, 판 = 두기 직전)."""
out = []
if not LOG_JSONL.exists():
return out
with LOG_JSONL.open(encoding="utf-8") as f:
for line in f:
d = json.loads(line)
if d.get("corrected") is None or d.get("size") not in SIZES:
continue
board = d.get("board")
if isinstance(board, list) and len(board) == d["size"] ** 2:
out.append(d)
return out
def log_boards() -> List[Tuple[int, Board]]:
global _LOG_BOARDS
if _LOG_BOARDS is None:
_LOG_BOARDS = [(d["size"], d["board"]) for d in log_vlm_requests() if sum(1 for v in d["board"] if v) >= 4]
return _LOG_BOARDS
def dense_board(rng: random.Random, size: int) -> Board:
pool = [b for s, b in log_boards() if s == size]
if pool and rng.random() < 0.6:
return list(rng.choice(pool))
return synth_game_board(rng, size, rng.randint(6, 70))
def winner_moves() -> List[dict]:
"""대국 기록에서 이긴 쪽 흑(로컬 AI)·사람의 수. 백 local_ai 는 VLM 수가 잘못 기록된 것이라 뺀다."""
if not LOG_JSONL.exists():
return []
games: dict = {}
order: List[dict] = []
with LOG_JSONL.open(encoding="utf-8") as f:
for line in f:
d = json.loads(line)
if d.get("corrected") is not None:
continue
order.append(d)
out = []
game: List[dict] = []
for d in order:
try:
seq = int(d.get("seq") or 0)
except (TypeError, ValueError):
seq = 0
if game and (seq <= int(game[-1].get("seq") or 0) or d.get("size") != game[-1].get("size")):
game = []
game.append(d)
status = str(d.get("status") or "")
if status in ("BlackWin", "WhiteWin"):
winner = 1 if status == "BlackWin" else 2
for m in game:
actor = m.get("actor")
stone = m.get("stone")
if stone != winner or m.get("size") not in SIZES:
continue
if actor == "human" or (actor == "local_ai" and stone == 1):
out.append(m)
game = []
elif status not in ("", "Playing", "None"):
game = []
return out
# ---------------------------------------------------------------- dataset
def _rec(task: str, size: int, user: str, answer: str) -> dict:
return {
"task": task,
"size": size,
"images": [],
"messages": [
{"role": "user", "content": [{"type": "text", "text": user}]},
{"role": "assistant", "content": [{"type": "text", "text": answer}]},
],
}
def hard_coords(size: int) -> List[Tuple[int, int]]:
e, m = size - 1, size // 2
return [(0, 0), (0, e), (e, 0), (e, e), (m, m), (m, m - 1), (m, m + 1), (0, m), (m, 0), (e, m), (m, e),
(e - 3, e - 3), (e - 3, e - 4), (3, e - 3), (e - 3, 3), (4, e - 4), (e - 1, 1), (1, e - 1)]
def make_locate(rng: random.Random, size: int, hard: bool = False) -> dict:
if hard:
pts = [rng.choice(hard_coords(size))]
else:
n = 1 if rng.random() < 0.7 else 2
s = set()
while len(s) < n:
s.add((rng.randrange(size), rng.randrange(size)))
pts = sorted(s)
color = 1 if rng.random() < 0.5 else 2
b = put(empty_board(size), size, [(r, c, color) for r, c in pts])
if len(pts) == 1:
ans = f'{{"row": {pts[0][0]}, "col": {pts[0][1]}}}'
else:
ans = '{"points":[' + ",".join(f'{{"row":{r},"col":{c}}}' for r, c in pts) + "]}"
return _rec("locate_hard" if hard else "locate", size, locate_prompt(b, size, color, len(pts)), ans)
def make_lookup(rng: random.Random, size: int, dense: bool = False) -> dict:
if dense:
b = dense_board(rng, size)
occ = [i for i, v in enumerate(b) if v]
emp = [i for i, v in enumerate(b) if not v]
i = rng.choice(occ) if occ and rng.random() < 0.5 else rng.choice(emp)
r, c = divmod(i, size)
else:
r, c = rng.randrange(size), rng.randrange(size)
color = 1 if rng.random() < 0.5 else 2
if rng.random() < 0.7:
b = put(empty_board(size), size, [(r, c, color)])
else:
er, ec = r, c
while (er, ec) == (r, c):
er, ec = rng.randrange(size), rng.randrange(size)
b = put(empty_board(size), size, [(er, ec, color)])
v = b[r * size + c]
ans = f'{{"cell": "{"." if v == 0 else ("X" if v == 1 else "O")}"}}'
return _rec("lookup_dense" if dense else "lookup", size, lookup_prompt(b, size, r, c), ans)
def make_move(rng: random.Random, moves: List[dict]) -> Optional[dict]:
if not moves:
return None
m = rng.choice(moves)
size = m["size"]
board = list(m["board"])
r, c = int(m["move_row"]), int(m["move_col"])
# OMKL 은 둔 뒤의 판이므로 방금 둔 돌을 빼서 두기 전으로 돌린다.
if board[r * size + c] == m["stone"]:
board[r * size + c] = 0
if board[r * size + c] != 0:
return None
return _rec("move", size, move_prompt(board, size, m["stone"]), f"{r} {c}")
def cmd_dataset(args: argparse.Namespace) -> int:
rng = random.Random(args.seed)
out = ROOT / args.out_dir
out.mkdir(parents=True, exist_ok=True)
moves = winner_moves() if args.move_frac > 0 else []
counts: Counter = Counter()
n_train = n_val = 0
with (out / "train.jsonl").open("w", encoding="utf-8") as ft, (out / "val.jsonl").open("w", encoding="utf-8") as fv:
while n_train + n_val < args.n_samples:
size = rng.choice(SIZES)
roll = rng.random()
rec = None
if roll < args.move_frac:
rec = make_move(rng, moves)
elif roll < args.move_frac + (1 - args.move_frac) * 0.30:
rec = make_locate(rng, size)
elif roll < args.move_frac + (1 - args.move_frac) * 0.45:
rec = make_locate(rng, size, hard=True)
elif roll < args.move_frac + (1 - args.move_frac) * 0.65:
rec = make_lookup(rng, size)
else:
rec = make_lookup(rng, size, dense=True)
if rec is None:
continue
counts[(rec["task"], rec["size"])] += 1
line = json.dumps(rec, ensure_ascii=False) + "\n"
if rng.random() < args.val_split:
fv.write(line)
n_val += 1
else:
ft.write(line)
n_train += 1
info = {
"n_train": n_train,
"n_val": n_val,
"winner_moves_pool": len(moves),
"tasks": {f"{t}@{s}": n for (t, s), n in sorted(counts.items())},
"board_format": "aligned_left",
}
(out / "dataset_info.json").write_text(json.dumps(info, indent=2, ensure_ascii=False), encoding="utf-8")
print(json.dumps(info, indent=2, ensure_ascii=False))
return 0
# ---------------------------------------------------------------- MLX
class MlxRunner:
def __init__(self, model: str, lora_dir: Optional[str], max_tokens: int = 64) -> None:
from mlx_vlm import generate, load
from mlx_vlm.prompt_utils import apply_chat_template
from mlx_vlm.utils import load_config
t0 = time.time()
self._generate = generate
self._template = apply_chat_template
self.model, self.processor = load(model)
if lora_dir:
from mlx_vlm.trainer.utils import apply_lora_layers
updated = apply_lora_layers(self.model, lora_dir)
if updated is not None:
self.model = updated
self.config = load_config(model)
self.max_tokens = max_tokens
print(f"[mlx] loaded {model} lora={lora_dir} in {time.time() - t0:.0f}s", flush=True)
def ask(self, prompt: str) -> str:
try:
formatted = self._template(self.processor, self.config, prompt, num_images=0)
except TypeError:
formatted = self._template(self.processor, prompt, num_images=0)
kw = dict(max_tokens=self.max_tokens, verbose=False)
try:
res = self._generate(self.model, self.processor, formatted, temperature=0.0, **kw)
except TypeError:
res = self._generate(self.model, self.processor, formatted, **kw)
text = getattr(res, "text", res)
return text if isinstance(text, str) else ""
def parse_points(text: str) -> List[Tuple[int, int]]:
pts = [(int(a), int(b)) for a, b in re.findall(r'"row"\s*:\s*(\d+)\s*,\s*"col"\s*:\s*(\d+)', text or "")]
return sorted(set(pts))
def parse_cell(text: str) -> Optional[str]:
m = re.search(r'"cell"\s*:\s*"([.XOxo])"', text or "")
return m.group(1).upper() if m else None
def parse_two_ints(text: str, size: int) -> Optional[Tuple[int, int]]:
m = re.search(r"(-?\d+)\s*[ ,]\s*(-?\d+)", text or "")
if not m:
return None
r, c = int(m.group(1)), int(m.group(2))
return (r, c) if 0 <= r < size and 0 <= c < size else None
def quadrant(r: int, c: int, size: int) -> str:
h = (size - 1) / 2
return ("T" if r <= h else "B") + ("L" if c <= h else "R")
def verify_cases(rng: random.Random, size: int, n: int) -> List[dict]:
e, m = size - 1, size // 2
probes = [(0, 0), (0, m), (0, e), (3, 3), (3, e - 3), (m, 0), (m, m), (m, e), (e - 3, 3),
(e - 3, e - 3), (e, 0), (e, m), (e, e), (2, e - 2), (e - 2, 2), (m - 2, m + 3)]
cases = [{"kind": "locate", "board": put(empty_board(size), size, [(r, c, 1)]), "target": [(r, c)], "color": 1}
for r, c in probes]
for r, c in probes[:6]:
cases.append({"kind": "lookup", "board": put(empty_board(size), size, [(r, c, 2)]), "q": (r, c), "expect": "O"})
while len(cases) < n:
kind = rng.choice(["locate2", "lookup_dense", "lookup_dense"])
if kind == "locate2":
s = set()
while len(s) < 2:
s.add((rng.randrange(size), rng.randrange(size)))
pts = sorted(s)
cases.append({"kind": "locate", "board": put(empty_board(size), size, [(r, c, 2) for r, c in pts]),
"target": pts, "color": 2})
else:
b = synth_game_board(rng, size, rng.randint(10, 60))
occ = [i for i, v in enumerate(b) if v]
emp = [i for i, v in enumerate(b) if not v]
i = rng.choice(occ) if rng.random() < 0.5 else rng.choice(emp)
v = b[i]
cases.append({"kind": "lookup_dense", "board": b, "q": divmod(i, size),
"expect": "." if v == 0 else ("X" if v == 1 else "O")})
return cases[:n]
def cmd_verify(args: argparse.Namespace) -> int:
runner = MlxRunner(args.model, args.lora_dir)
rng = random.Random(args.seed)
rows = []
for size in SIZES:
for case in verify_cases(rng, size, args.trials):
t0 = time.time()
if case["kind"] == "locate":
raw = runner.ask(locate_prompt(case["board"], size, case["color"], len(case["target"]), args.format))
got = parse_points(raw)
ok = got == sorted(case["target"])
q = quadrant(*case["target"][0], size)
exp = case["target"]
else:
r, c = case["q"]
raw = runner.ask(lookup_prompt(case["board"], size, r, c, args.format))
got = parse_cell(raw)
ok = got == case["expect"]
q = quadrant(r, c, size)
exp = [case["expect"], [r, c]]
rows.append({"size": size, "kind": case["kind"], "ok": ok, "quadrant": q, "expected": exp,
"got": got, "raw": raw[:200], "ms": round((time.time() - t0) * 1000)})
print(f"[{size}] {'OK ' if ok else 'FAIL'} {case['kind']:<12} {q} exp={exp} got={got}", flush=True)
summary = summarize(rows, ("size", "kind", "quadrant"))
summary.update(model=args.model, lora_dir=args.lora_dir, format=args.format)
return write_out(args, rows, summary)
def cmd_realboard(args: argparse.Namespace) -> int:
reqs = log_vlm_requests()
rng = random.Random(args.seed)
picks = rng.sample(reqs, min(args.n, len(reqs)))
runner = MlxRunner(args.model, args.lora_dir, max_tokens=16)
rows = []
for d in picks:
size, board = d["size"], d["board"]
t0 = time.time()
raw = runner.ask(move_prompt(board, size, int(d["ai_side"]), args.format))
rc = parse_two_ints(raw, size)
if rc is None:
verdict = "parse_fail"
elif board[rc[0] * size + rc[1]] != 0:
verdict = "occupied"
else:
verdict = "empty"
rows.append({"size": size, "stones": sum(1 for v in board if v), "was_corrected": bool(d["corrected"]),
"verdict": verdict, "got": rc, "raw": raw[:80], "ms": round((time.time() - t0) * 1000)})
print(f"[{size}] {verdict:<10} stones={rows[-1]['stones']:<3} got={rc} raw={raw[:30]!r}", flush=True)
n = len(rows) or 1
verdicts = Counter(r["verdict"] for r in rows)
summary = {
"n": len(rows),
"format": args.format,
"empty_pct": round(100 * verdicts["empty"] / n, 1),
"occupied_pct": round(100 * verdicts["occupied"] / n, 1),
"parse_fail_pct": round(100 * verdicts["parse_fail"] / n, 1),
"live_corrected_pct_same_boards": round(100 * sum(r["was_corrected"] for r in rows) / n, 1),
"model": args.model,
"lora_dir": args.lora_dir,
}
return write_out(args, rows, summary)
def summarize(rows: List[dict], keys: Sequence[str]) -> dict:
n = len(rows) or 1
out = {"n": len(rows), "accuracy": round(100 * sum(r["ok"] for r in rows) / n, 1)}
for k in keys:
tot, ok = Counter(), Counter()
for r in rows:
tot[r[k]] += 1
ok[r[k]] += r["ok"]
out[f"by_{k}"] = {str(v): {"n": tot[v], "acc": round(100 * ok[v] / tot[v], 1)} for v in sorted(tot)}
return out
def write_out(args: argparse.Namespace, rows: List[dict], summary: dict) -> int:
print(json.dumps(summary, indent=2, ensure_ascii=False))
if args.out:
base = ROOT / args.out
base.parent.mkdir(parents=True, exist_ok=True)
with base.with_suffix(".jsonl").open("w", encoding="utf-8") as f:
for r in rows:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
base.with_name(base.name + "_summary.json").write_text(
json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8")
return 0
def main() -> int:
p = argparse.ArgumentParser(description="오목 VLM 좌표 grounding")
sub = p.add_subparsers(dest="cmd", required=True)
d = sub.add_parser("dataset")
d.add_argument("--out-dir", default="sft_data/coord_omook_r1")
d.add_argument("--n-samples", type=int, default=6000)
d.add_argument("--move-frac", type=float, default=0.0, help="이긴 쪽 착수 예제 비율 (0~1)")
d.add_argument("--val-split", type=float, default=0.05)
d.add_argument("--seed", type=int, default=7)
for name in ("verify", "realboard"):
s = sub.add_parser(name)
s.add_argument("--model", default="********")
s.add_argument("--lora-dir", default=None)
s.add_argument("--format", choices=("aligned", "legacy"), default="aligned")
s.add_argument("--seed", type=int, default=42)
s.add_argument("--out", default=None, help="logs/coord/<name> (확장자 없이)")
if name == "verify":
s.add_argument("--trials", type=int, default=48, help="판 크기마다 문항 수")
else:
s.add_argument("--n", type=int, default=200)
args = p.parse_args()
return {"dataset": cmd_dataset, "verify": cmd_verify, "realboard": cmd_realboard}[args.cmd](args)
if __name__ == "__main__":
raise SystemExit(main())