예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
dataset_baduk.py
"""SGF 폴더 → mlx-vlm 호환 SFT 학습용 jsonl + 보드 PNG 이미지로 변환.
각 게임 마다 (한 수 두기 직전 상태, 다음 한 수) 트리플을 추출하여
Gemma 4 의 messages 포맷으로 저장합니다.
품질 향상 옵션:
--winner-only SGF 의 RE 태그를 보고 이긴 쪽 수만 학습 데이터로 채택.
--tactical-bias 게임당 샘플 N 개 중 절반은 전술적 순간(잡기/단수/단수회피
전후) 으로 채우고 나머지는 균등 분포에서 채움.
--dedup 같은 (보드 패킹, 차례, 정답수) 가 이미 있으면 스킵.
--val-split FLOAT train.jsonl 외에 val.jsonl 을 같은 폴더에 비율만큼 분리.
사용:
python -m vlm_server.dataset_baduk \
--sgf-dir sgf_data \
--out-dir sft_data/baduk_sft \
--max-games 5000 --moves-per-game 24 \
--winner-only --tactical-bias --dedup --val-split 0.05 --shuffle
생성:
sft_data/baduk_sft/
train.jsonl (val.jsonl 가 있으면 분리)
img_00000000.png ...
dataset_info.json
"""
from __future__ import annotations
import argparse
import json
import os
import random
import sys
from dataclasses import asdict
from pathlib import Path
from typing import Iterable, Iterator, List, Optional, Tuple
from PIL import Image, ImageDraw
from .sgf_parser import SgfGame, iter_sgf_files, parse_sgf_file
BOARD_SIZE = 19
IMG_PX = 720
# ---------- 미니 Go 시뮬레이터 (활로/따냄/자살수 — 학습 데이터 합법성 보장) ----------
class _GoSim:
DR = (-1, 1, 0, 0)
DC = (0, 0, -1, 1)
def __init__(self, size: int = BOARD_SIZE):
self.size = size
self.cells = bytearray(size * size)
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 play(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
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)
if my_libs == 0:
self.set(r, c, 0)
return False
return True
def packed(self) -> bytes:
return bytes(self.cells)
# ---------- 보드 이미지 렌더 (PIL, Qt 의존 없음) ----------
def render_board_png(cells: bytes, size: int = BOARD_SIZE, px: int = IMG_PX) -> Image.Image:
img = Image.new("RGB", (px, px), (220, 184, 116))
d = ImageDraw.Draw(img)
pad = px * 0.045
avail = px - 2.0 * pad
sp = avail / (size - 1)
off = (pad, pad)
line_color = (80, 50, 20)
for i in range(size):
y = off[1] + i * sp
d.line([(off[0], y), (off[0] + (size - 1) * sp, y)], fill=line_color, width=2)
x = off[0] + i * sp
d.line([(x, off[1]), (x, off[1] + (size - 1) * sp)], fill=line_color, width=2)
# 화점
if size == 19:
hoshi = [(3, 3), (3, 9), (3, 15), (9, 3), (9, 9), (9, 15), (15, 3), (15, 9), (15, 15)]
rr = max(2.5, sp * 0.11)
for r, c in hoshi:
cx = off[0] + c * sp
cy = off[1] + r * sp
d.ellipse(
[(cx - rr, cy - rr), (cx + rr, cy + rr)],
fill=(40, 25, 10),
)
rad = max(2.0, sp * 0.46)
for r in range(size):
for c in range(size):
v = cells[r * size + c]
if v == 0:
continue
cx = off[0] + c * sp
cy = off[1] + r * sp
if v == 1:
fill = (10, 10, 12)
outline = (0, 0, 0)
else:
fill = (240, 240, 245)
outline = (80, 80, 84)
d.ellipse(
[(cx - rad, cy - rad), (cx + rad, cy + rad)],
fill=fill,
outline=outline,
width=1,
)
return img
# ---------- 프롬프트 생성 ----------
SYSTEM_TEXT = (
"당신은 19×19 바둑(Go) 의 강력한 플레이어입니다. "
"보드 이미지와 아래 텍스트 격자를 함께 보고 다음 단 한 수의 좌표만 골라야 합니다. "
"(텍스트 격자가 이미지와 어긋나면 텍스트 격자를 우선하세요.) "
"숫자 격자: 0=빈칸, 1=흑, 2=백. 문자 격자: .=빈 X=흑 O=백. "
"좌표는 0-indexed 정수 (row, col) 이며 각각 0..18 입니다. "
"응답은 다른 설명 없이 JSON 한 줄로만 출력하세요."
)
SYSTEM_TEXT_ONLY = (
"당신은 19×19 바둑(Go) 의 강력한 플레이어입니다. "
"보드 상태가 텍스트 격자로만 주어집니다. 격자를 정확히 읽고 "
"다음 단 한 수의 좌표만 고르세요. "
"문자 격자: .=빈칸 X=흑 O=백. "
"좌표는 0-indexed 정수 (row, col) 이며 각각 0..18 입니다. "
"빈 칸(.)에만 두세요. 자살수는 피하세요. "
"응답은 다른 설명 없이 JSON 한 줄로만 출력하세요."
)
# 예제 좌표를 우측에 고정하지 않음(우하 편향 완화). 보드 해시로 순환.
_EXAMPLE_COORDS = (
(3, 3),
(3, 15),
(15, 3),
(15, 15),
(9, 9),
(2, 9),
(16, 9),
(9, 2),
(9, 16),
)
def _example_move_json(cells: bytes) -> str:
h = sum(cells) + len(cells)
r, c = _EXAMPLE_COORDS[h % len(_EXAMPLE_COORDS)]
return f'{{"row": {r}, "col": {c}}}'
def _cell_token(index: int, v: int) -> str:
return f"{index:02d}" + ("." if v == 0 else ("X" if v == 1 else "O"))
def row_line_indexed(cells: bytes, r: int, size: int = BOARD_SIZE) -> str:
"""한 행을 '00. 01X 02O ...' (열 번호 두 자리 + 기호) 로."""
return " ".join(_cell_token(c, cells[r * size + c]) for c in range(size))
def col_line_indexed(cells: bytes, c: int, size: int = BOARD_SIZE) -> str:
"""한 열을 '00. 01X 02O ...' (행 번호 두 자리 + 기호) 로. 행 줄과 같은 형식."""
return " ".join(_cell_token(r, cells[r * size + c]) for r in range(size))
def column_scan_user(cells: bytes, col: int, size: int = BOARD_SIZE) -> str:
"""열 질문은 19줄을 세로로 세지 않고, 그 열만 한 줄로 준다."""
line = col_line_indexed(cells, col, size)
return (
f"열 col={col} 의 {size}칸입니다. 각 토큰은 행 번호 두 자리와 그 칸의 기호입니다 "
f"(.=빈칸, X=흑, O=백).\n"
f"{line}\n\n"
f"이 줄만 보고 돌이 있는 행(row) 번호를 색깔별로 모두 적으세요. "
f'JSON 한 줄만. 예: {{"X": [3, 4], "O": [10]}}'
)
def read_pick_answer(cells: bytes, r: int, c: int, size: int = BOARD_SIZE) -> str:
"""aligned_read 정답: 행 → 그 행 내용 → 열 순서."""
return f'{{"row": {r}, "line": "{row_line_indexed(cells, r, size)}", "col": {c}}}'
def empty_cols(cells: bytes, r: int, size: int = BOARD_SIZE) -> list[int]:
return [c for c in range(size) if cells[r * size + c] == 0]
def read_pick_answer2(cells: bytes, r: int, c: int, size: int = BOARD_SIZE) -> str:
"""aligned_read2 정답: 행 → 그 행 내용 → 빈 열 목록 → 그 목록 안의 열."""
payload = {"row": r, "line": row_line_indexed(cells, r, size), "empty": empty_cols(cells, r, size), "col": c}
return json.dumps(payload, ensure_ascii=False)
def _example_read_json() -> str:
return '{"row": 4, "line": "00. 01. 02X 03. ... 18.", "col": 3}'
def _example_read2_json() -> str:
return '{"row": 4, "line": "00. 01. 02X 03. ... 18.", "empty": [0, 1, 3], "col": 3}'
def board_text(cells: bytes, size: int = BOARD_SIZE) -> str:
"""문자 격자(spaced, 레거시): 열 00..18, 셀은 한 글자+공백 — 헤더와 셀 폭이 어긋날 수 있음."""
col_header = " " + " ".join(f"{c:2d}" for c in range(size))
lines: List[str] = [col_header]
for r in range(size):
ch: List[str] = []
for c in range(size):
v = cells[r * size + c]
ch.append("." if v == 0 else ("X" if v == 1 else "O"))
lines.append(f"{r:3d} " + " ".join(ch))
return "\n".join(lines)
def board_text_compact(cells: bytes, size: int = BOARD_SIZE) -> str:
"""baduk_v2 등 기존 SFT jsonl 과 동일 형식.
열 헤더가 ``0123456789012345678`` (한 자리 연속)이고 칸 사이 공백이 없음.
"""
col_header = " " + "".join(str(c % 10) for c in range(size))
lines: List[str] = [col_header]
for r in range(size):
ch: List[str] = []
for c in range(size):
v = cells[r * size + c]
ch.append("." if v == 0 else ("X" if v == 1 else "O"))
lines.append(f"{r:2d} " + "".join(ch))
return "\n".join(lines)
def board_text_aligned(cells: bytes, size: int = BOARD_SIZE) -> str:
"""v3 권장: 열·행·셀이 모두 두 글자 토큰으로 정렬.
셀 기호는 **왼쪽 정렬** (``X `` / ``. ``). 오른쪽 정렬(`` X``)은
모델이 열을 한 칸 왼쪽(col-1)으로 읽는 오차를 유발했다.
"""
col_header = " " + " ".join(f"{c:02d}" for c in range(size))
lines: List[str] = [col_header]
for r in range(size):
cells_s: List[str] = []
for c in range(size):
v = cells[r * size + c]
ch = "." if v == 0 else ("X" if v == 1 else "O")
cells_s.append(f"{ch:<2}")
lines.append(f"{r:02d} " + " ".join(cells_s))
return "\n".join(lines)
def board_numeric_grid(cells: bytes, size: int = BOARD_SIZE) -> str:
rows = []
for r in range(size):
row = [str(int(cells[r * size + c])) for c in range(size)]
rows.append(f"{r:2d}: " + ",".join(row))
return "\n".join(rows)
def board_numeric_aligned(cells: bytes, size: int = BOARD_SIZE) -> str:
"""숫자 격자도 두 자리 정렬(0→' 0')."""
col_header = " " + " ".join(f"{c:02d}" for c in range(size))
lines: List[str] = [col_header]
for r in range(size):
cells_s = [f"{int(cells[r * size + c]):>2d}" for c in range(size)]
lines.append(f"{r:02d} " + " ".join(cells_s))
return "\n".join(lines)
def render_board_for_prompt(cells: bytes, board_format: str = "aligned") -> str:
if board_format == "compact":
return board_text_compact(cells)
if board_format == "spaced":
return board_text(cells)
return board_text_aligned(cells)
def user_prompt_text(
cells: bytes,
ai_side: int,
*,
board_format: str = "aligned",
text_only: bool = False,
include_numeric: bool = False,
) -> str:
side_name = "흑(X)" if ai_side == 1 else "백(O)"
system = SYSTEM_TEXT_ONLY if text_only else SYSTEM_TEXT
board_chars = render_board_for_prompt(cells, board_format)
example = _example_move_json(cells)
parts = [
f"{system}\n\n",
f"당신의 차례: {side_name}.\n",
"보드 문자 격자 (X=흑, O=백, .=빈, 행=row, 열=col, 0-indexed).\n"
"첫 줄·각 칸은 두 글자 폭으로 정렬되어 있습니다:\n"
if board_format == "aligned"
else "보드 (X=흑, O=백, .=빈, 행=row, 열=col, 0-indexed):\n",
f"{board_chars}\n\n",
]
if include_numeric:
num = (
board_numeric_aligned(cells)
if board_format == "aligned"
else board_numeric_grid(cells)
)
parts.append(
"보드 숫자 격자 (0=빈, 1=흑, 2=백):\n"
f"{num}\n\n"
)
parts.append(
f'다음 수를 JSON 한 줄로만 답하세요. 예: {example} '
'혹은 패스를 원하면 {"pass": true} 만 출력.'
)
return "".join(parts)
def assistant_text(point: Optional[Tuple[int, int]]) -> str:
if point is None:
return '{"pass": true}'
r, c = point
return f'{{"row": {r}, "col": {c}}}'
# ---------- 전술 순간 탐지 ----------
def _is_tactical_move(
sim_before: _GoSim, side: int, point: Optional[Tuple[int, int]]
) -> bool:
"""이 한 수가 전술적으로 의미 있는가?
- 잡기: 두면 상대 그룹의 활로가 0 이 됨 → 따냄 발생
- 단수 만들기: 두면 상대 인접 그룹의 활로 1
- 단수 탈출: 두기 전 우리 인접 그룹 중 활로 1 짜리가 있고, 두면 활로 ≥ 2
- 끊기: 두면 상대 그룹 두 개 이상 분리 (간단판: 4방향 상대 그룹이 ≥2 개)
학습 데이터의 '결정적 순간' 비율을 높이기 위한 휴리스틱 분류기.
"""
if point is None:
return False
r, c = point
N = sim_before.size
if not (0 <= r < N and 0 <= c < N):
return False
if sim_before.at(r, c) != 0:
return False
opp = 2 if side == 1 else 1
# 두기 전 인접 우리 그룹 활로 검사
visited_my: set = set()
for d in range(4):
nr, nc = r + _GoSim.DR[d], c + _GoSim.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
if sim_before.at(nr, nc) != side:
continue
nidx = nr * N + nc
if nidx in visited_my:
continue
stones, libs = sim_before._group(nr, nc)
for s in stones:
visited_my.add(s)
if libs == 1:
# 두면 우리 단수 탈출 가능성 → 전술
return True
# 두기 전 인접 상대 그룹 활로 검사 (상대 단수/잡기 가능성)
visited_opp_starts: List[int] = []
visited_opp_set: set = set()
distinct_opp_groups = 0
has_opp_atari = False
has_opp_capture = False
for d in range(4):
nr, nc = r + _GoSim.DR[d], c + _GoSim.DC[d]
if not (0 <= nr < N and 0 <= nc < N):
continue
if sim_before.at(nr, nc) != opp:
continue
nidx = nr * N + nc
if nidx in visited_opp_set:
continue
stones, libs = sim_before._group(nr, nc)
for s in stones:
visited_opp_set.add(s)
distinct_opp_groups += 1
# 두기 전 활로 1 인 상대 그룹과 인접 = 두면 잡힘 (단, 그 활로가 (r,c) 일 때).
if libs == 1:
has_opp_capture = True
elif libs == 2:
has_opp_atari = True
if has_opp_capture or has_opp_atari:
return True
if distinct_opp_groups >= 2:
# 끊기 후보
return True
return False
# ---------- 메인 변환 ----------
def iter_training_samples(
game: SgfGame,
max_per_game: int,
rng: random.Random,
winning_side_only: bool = False,
winner: Optional[int] = None,
tactical_bias: bool = False,
tactics_only: bool = False,
) -> Iterator[Tuple[bytes, int, Optional[Tuple[int, int]], bool]]:
"""게임 1개에서 (board_before_move, ai_side, target_move, is_tactical) 4-tuple yield."""
if game.size != BOARD_SIZE:
return
sim = _GoSim(BOARD_SIZE)
for r, c in game.setup_black:
sim.set(r, c, 1)
for r, c in game.setup_white:
sim.set(r, c, 2)
# tactics_only 든 tactical_bias 든 어느 쪽이라도 분류기를 켜야 한다.
need_tactical_flag = tactical_bias or tactics_only
states: List[Tuple[bytes, int, Optional[Tuple[int, int]], bool]] = []
# (snap_before, side, point, is_tactical)
for color, point in game.moves:
side = 1 if color == "B" else 2
if winning_side_only and winner is not None and side != winner:
if point is not None:
sim.play(point[0], point[1], side)
continue
snap = sim.packed()
is_tac = _is_tactical_move(sim, side, point) if need_tactical_flag else False
states.append((snap, side, point, is_tac))
if point is not None:
sim.play(point[0], point[1], side)
if not states:
return
if tactics_only:
# 전술 순간만 yield (max_per_game 무시).
for snap, side, point, is_tac in states:
if is_tac:
yield (snap, side, point, True)
return
if max_per_game <= 0 or len(states) <= max_per_game:
chosen_indices = list(range(len(states)))
else:
if tactical_bias:
# 절반은 전술 순간 우선, 나머지는 균등.
tactical_idx = [i for i, st in enumerate(states) if st[3]]
half = max_per_game // 2
if len(tactical_idx) > half:
rng.shuffle(tactical_idx)
tactical_pick = sorted(tactical_idx[:half])
else:
tactical_pick = sorted(tactical_idx)
need = max_per_game - len(tactical_pick)
# 나머지 균등 샘플 (tactical_pick 와 중복 피함)
taken = set(tactical_pick)
uniform: List[int] = []
if need > 0:
step = max(1.0, len(states) / need)
k = 0
while len(uniform) < need and k < len(states):
idx = int(k * step)
if idx >= len(states):
break
if idx not in taken:
uniform.append(idx)
taken.add(idx)
k += 1
chosen_indices = sorted(set(tactical_pick + uniform))
else:
step = len(states) / max_per_game
chosen_indices = sorted(set(int(i * step) for i in range(max_per_game)))
for i in chosen_indices:
snap, side, point, is_tac = states[i]
yield (snap, side, point, is_tac)
def main(argv: Optional[List[str]] = None) -> int:
p = argparse.ArgumentParser(description="SGF → mlx-vlm SFT 데이터셋 변환")
p.add_argument("--sgf-dir", required=True, help="SGF 파일들이 있는 폴더")
p.add_argument("--out-dir", required=True, help="결과 출력 폴더")
p.add_argument("--max-games", type=int, default=0, help="처리할 최대 게임 수 (0=전부)")
p.add_argument("--skip-games", type=int, default=0, help="셔플 후 앞에서 건너뛸 게임 수")
p.add_argument("--moves-per-game", type=int, default=16, help="게임당 채택할 샘플 수")
p.add_argument("--seed", type=int, default=42)
p.add_argument("--shuffle", action="store_true", help="SGF 파일 순서 셔플 후 max-games")
p.add_argument("--render-px", type=int, default=IMG_PX, help="보드 PNG 픽셀 (정사각)")
p.add_argument(
"--include-pass",
action="store_true",
help="패스 수도 학습 샘플로 포함 (기본은 제외).",
)
p.add_argument(
"--winner-only",
action="store_true",
help="SGF RE 태그를 보고 이긴 쪽 수만 학습 데이터로 채택.",
)
p.add_argument(
"--tactical-bias",
action="store_true",
help="게임당 샘플 절반은 전술적 순간(잡기/단수/단수회피/끊기) 우선.",
)
p.add_argument(
"--tactics-only",
action="store_true",
help="전술 순간(잡기/단수/단수회피/끊기) 만 학습 데이터로 채택. "
"moves-per-game 은 무시. 단수 응수·끊기 잇기 학습 강화용 보조 셋 제작에 사용.",
)
p.add_argument(
"--oversample-tactical",
type=int,
default=1,
help="전술 샘플은 jsonl 에 N 회 복제해서 기록 (이미지는 1회만 저장). "
"1=복제 안 함(기본). 비전술 샘플은 항상 1 회.",
)
p.add_argument(
"--dedup",
action="store_true",
help="(보드 패킹, 차례, 정답수) 가 이미 있으면 스킵.",
)
p.add_argument(
"--val-split",
type=float,
default=0.0,
help="0.0~0.3. >0 이면 train.jsonl 외에 val.jsonl 도 같은 폴더에 생성.",
)
p.add_argument(
"--board-format",
choices=["aligned", "compact", "spaced"],
default="aligned",
help="프롬프트 보드 격자: aligned=v3정렬(권장), compact=baduk_v2레거시, spaced=구 공백형식.",
)
p.add_argument(
"--text-only",
action="store_true",
help="이미지 없이 텍스트 격자만 학습 샘플에 넣음 (Qwen text-only serve 와 맞출 때).",
)
p.add_argument(
"--include-numeric",
action="store_true",
help="문자 격자 외에 숫자 격자도 프롬프트에 포함 (기본 off — 토큰 절약).",
)
p.add_argument(
"--no-render-png",
action="store_true",
help="PNG 파일을 디스크에 쓰지 않음 (--text-only 와 함께 권장).",
)
args = p.parse_args(argv)
rng = random.Random(args.seed)
out_dir = Path(args.out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
train_path = out_dir / "train.jsonl"
val_path = out_dir / "val.jsonl"
if val_path.exists() and args.val_split <= 0:
# 이전 실행에서 만든 val.jsonl 청소
try:
val_path.unlink()
except OSError:
pass
sgf_files = list(iter_sgf_files(args.sgf_dir))
if args.shuffle:
rng.shuffle(sgf_files)
if args.skip_games > 0:
sgf_files = sgf_files[args.skip_games :]
if args.max_games > 0:
sgf_files = sgf_files[: args.max_games]
print(f"SGF files: {len(sgf_files)}")
if not sgf_files:
print("처리할 SGF 가 없습니다. 먼저 sgf_download 로 받아주세요.")
return 2
n_samples = 0
n_val = 0
n_games_used = 0
n_skipped_size = 0
n_dedup_drop = 0
img_idx = 0
seen_keys: set = set() if args.dedup else set()
val_split = max(0.0, min(0.3, float(args.val_split)))
fout_train = open(train_path, "w", encoding="utf-8")
fout_val = open(val_path, "w", encoding="utf-8") if val_split > 0 else None
try:
for fp in sgf_files:
game = parse_sgf_file(fp)
if game is None:
continue
if game.size != BOARD_SIZE:
n_skipped_size += 1
continue
# tactical bias / tactics-only / oversample 중 어느 하나라도 켜져 있으면
# 분류기를 활성화해야 4-tuple 의 is_tactical 이 의미를 갖는다.
tactical_bias_eff = (
args.tactical_bias
or args.tactics_only
or args.oversample_tactical > 1
)
samples = list(
iter_training_samples(
game,
max_per_game=args.moves_per_game,
rng=rng,
winning_side_only=args.winner_only,
winner=getattr(game, "winner", None),
tactical_bias=tactical_bias_eff,
tactics_only=args.tactics_only,
)
)
if not samples:
continue
n_games_used += 1
for board, side, point, is_tac in samples:
if not args.include_pass and point is None:
continue
if args.dedup:
key_pt = (-1, -1) if point is None else point
key = (bytes(board), side, key_pt)
if key in seen_keys:
n_dedup_drop += 1
continue
seen_keys.add(key)
img_path_str = ""
if not args.no_render_png:
img = render_board_png(board, BOARD_SIZE, args.render_px)
img_name = f"img_{img_idx:08d}.png"
img_path = out_dir / img_name
img.save(img_path, "PNG", optimize=True)
img_path_str = str(img_path.resolve())
img_idx += 1
user_content: list = []
if not args.text_only:
if not img_path_str:
raise SystemExit("--text-only 없이 --no-render-png 는 사용할 수 없습니다.")
user_content.append({"type": "image", "image": img_path_str})
user_content.append(
{
"type": "text",
"text": user_prompt_text(
board,
side,
board_format=args.board_format,
text_only=bool(args.text_only),
include_numeric=bool(args.include_numeric),
),
}
)
rec = {
"id": img_idx - 1,
"size": BOARD_SIZE,
"ai_side": side,
"target_row": point[0] if point else -1,
"target_col": point[1] if point else -1,
"is_pass": point is None,
"is_tactical": bool(is_tac),
"board_format": args.board_format,
"text_only": bool(args.text_only),
"images": ([img_path_str] if img_path_str else []),
"messages": [
{
"role": "user",
"content": user_content,
},
{
"role": "assistant",
"content": [{"type": "text", "text": assistant_text(point)}],
},
],
}
line = json.dumps(rec, ensure_ascii=False) + "\n"
# 전술 샘플은 oversample-tactical 만큼 jsonl 에 반복 기록.
# 이미지는 한 번만 저장한 PNG 를 가리키므로 디스크 비용은 거의 없음.
write_count = max(1, args.oversample_tactical) if is_tac else 1
for _ in range(write_count):
if fout_val is not None and rng.random() < val_split:
fout_val.write(line)
n_val += 1
else:
fout_train.write(line)
n_samples += 1
finally:
fout_train.close()
if fout_val is not None:
fout_val.close()
info = {
"out_dir": str(out_dir.resolve()),
"train_jsonl": str(train_path.resolve()),
"val_jsonl": (str(val_path.resolve()) if fout_val is not None else None),
"n_samples_train": n_samples,
"n_samples_val": n_val,
"n_games_used": n_games_used,
"n_sgf_input": len(sgf_files),
"n_skipped_size": n_skipped_size,
"n_dedup_drop": n_dedup_drop,
"moves_per_game": args.moves_per_game,
"render_px": args.render_px,
"include_pass": args.include_pass,
"winner_only": args.winner_only,
"tactical_bias": args.tactical_bias,
"tactics_only": args.tactics_only,
"oversample_tactical": args.oversample_tactical,
"dedup": args.dedup,
"val_split": val_split,
"board_format": args.board_format,
"text_only": bool(args.text_only),
"include_numeric": bool(args.include_numeric),
"no_render_png": bool(args.no_render_png),
}
with open(out_dir / "dataset_info.json", "w", encoding="utf-8") as f:
json.dump(info, f, indent=2, ensure_ascii=False)
print(json.dumps(info, indent=2, ensure_ascii=False))
return 0
if __name__ == "__main__":
sys.exit(main())