예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
dataset_passfix.py
"""R5가 PASS한 국면을 KataGo 최선수로 다시 라벨링한 학습 데이터.
휴리스틱 대국에서 R5가 중반(빈점이 많을 때) PASS한 국면을 모은다.
- games.jsonl 의 moves 기록을 재생해 백 PASS 직전 판을 복원
- pass_cases.jsonl (재요청 실험 기록, 판 포함)
KataGo 최선수가 PASS가 아니고 PASS 대비 손해가 min_loss 이상인 국면만 남긴다.
정답은 baduk_v3_aligned 와 같은 형식({"row": r, "col": c})이라 그대로 LoRA 혼합에 쓴다.
python -m vlm_server.dataset_passfix --out sft_data/passfix_v1 \
--games logs/heuristic_r5_200x40_20261004/games.jsonl \
--cases logs/heuristic_r5_nolimit_recheck_20261005/pass_cases.jsonl
"""
from __future__ import annotations
import argparse
import hashlib
import json
import random
import time
from collections import Counter
from pathlib import Path
from typing import Iterable, Iterator, Optional
from .cross_lessons import parse_move
from .dataset_baduk import assistant_text, user_prompt_text
from .ladder import _Board
N = 19
Point = tuple[int, int]
WHITE = 2
def pass_positions(moves: list[str], side: int = WHITE) -> Iterator[tuple[int, bytes]]:
"""moves 기록을 재생하며 side 가 PASS한 시점의 (ply, 판)을 낸다."""
actor_name = "black" if side == 1 else "white"
cells = bytearray(N * N)
for ply, line in enumerate(moves):
parsed = parse_move(line)
if parsed is None:
continue
actor, r, c, is_pass = parsed
if is_pass or r is None or c is None:
if actor == actor_name:
yield ply, bytes(cells)
continue
b = _Board(cells)
if b.play(r * N + c, 1 if actor == "black" else 2) is None:
continue
cells = b.c
def collect(games: Iterable[Path], cases: Iterable[Path], min_empty: int) -> list[dict]:
seen: set[str] = set()
out: list[dict] = []
def add(cells: bytes, meta: dict) -> None:
if cells.count(0) < min_empty:
return
h = hashlib.sha1(cells).hexdigest()
if h in seen:
return
seen.add(h)
out.append({"cells": cells, "empty": cells.count(0), **meta})
for path in games:
if not path.exists():
continue
for line in path.read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
d = json.loads(line)
for ply, cells in pass_positions(d.get("moves") or []):
add(cells, {"source": f"{path.parent.name}:{d.get('game_id')}", "ply": ply})
for path in cases:
if not path.exists():
continue
for line in path.read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
d = json.loads(line)
board = d.get("board")
if not board or len(board) != N * N:
continue
add(
bytes(int(v) for v in board),
{
"source": f"{path.parent.name}:{d.get('game_id')}",
"ply": d.get("ply"),
"recheck": d.get("result"),
},
)
return out
def judge(kg, cells: bytes, side: int) -> dict:
"""KataGo 최선수와 PASS 대비 손해. top_is_pass 면 PASS가 실제로 맞는 국면."""
from .dataset_ladder import _from_gtp, _gtp
player = "B" if side == 1 else "W"
stones = [["B" if v == 1 else "W", _gtp(divmod(i, N))] for i, v in enumerate(cells) if v]
res = kg.j._query(
{
"initialStones": stones,
"initialPlayer": player,
"moves": [],
"rules": "korean",
"komi": kg.komi,
"boardXSize": N,
"boardYSize": N,
"analyzeTurns": [0],
"maxVisits": kg.visits,
}
)
infos = sorted(res.get("moveInfos") or [], key=lambda m: m.get("order", 99))
flip = -1.0 if side == 2 else 1.0
lead = {str(m["move"]).lower(): flip * m.get("scoreLead", 0.0) for m in infos}
top_is_pass = bool(infos) and str(infos[0]["move"]).lower() == "pass"
best: Optional[Point] = None
for m in infos:
p = _from_gtp(m["move"])
if p is not None:
best = p
break
loss = None
if best is not None and "pass" in lead:
loss = round(lead[_gtp(best).lower()] - lead["pass"], 2)
return {"best": best, "top_is_pass": top_is_pass, "pass_loss": loss}
def build(args: argparse.Namespace) -> int:
from .dataset_ladder import _KataGo
rng = random.Random(args.seed)
positions = collect(
[Path(p) for p in args.games], [Path(p) for p in args.cases], args.min_empty
)
rng.shuffle(positions)
if args.n > 0:
positions = positions[: args.n]
print(f"[passfix] positions={len(positions)} (min_empty={args.min_empty})", flush=True)
kg = _KataGo(args.katago_visits, args.komi)
out = Path(args.out)
out.mkdir(parents=True, exist_ok=True)
stats: Counter = Counter()
rows: list[dict] = []
t0 = time.time()
for i, pos in enumerate(positions, 1):
cells = pos["cells"]
v = judge(kg, cells, WHITE)
if v["best"] is None or v["top_is_pass"]:
stats["katago_pass"] += 1
continue
if v["pass_loss"] is not None and v["pass_loss"] < args.min_loss:
stats["small_loss"] += 1
continue
stats["kept"] += 1
if pos.get("recheck"):
stats[f"kept_{pos['recheck']}"] += 1
best = v["best"]
rows.append(
{
"id": len(rows),
"size": N,
"ai_side": WHITE,
"target_row": best[0],
"target_col": best[1],
"is_pass": False,
"is_tactical": False,
"board_format": "aligned",
"text_only": True,
"images": [],
"passfix": {
"source": pos["source"],
"ply": pos.get("ply"),
"empty": pos["empty"],
"recheck": pos.get("recheck"),
"pass_loss": v["pass_loss"],
},
"messages": [
{"role": "user", "content": [{"type": "text", "text": user_prompt_text(cells, WHITE, board_format="aligned", text_only=True)}]},
{"role": "assistant", "content": [{"type": "text", "text": assistant_text(best)}]},
],
}
)
if i % 50 == 0:
print(f"[passfix] {i}/{len(positions)} kept={len(rows)} {time.time() - t0:.0f}s", flush=True)
idx = list(range(len(rows)))
rng.shuffle(idx)
n_val = max(1, int(len(idx) * args.val_frac)) if idx else 0
val = set(idx[:n_val])
with open(out / "train.jsonl", "w", encoding="utf-8") as ft, open(out / "val.jsonl", "w", encoding="utf-8") as fv:
for j in idx:
(fv if j in val else ft).write(json.dumps(rows[j], ensure_ascii=False) + "\n")
losses = [r["passfix"]["pass_loss"] for r in rows if r["passfix"]["pass_loss"] is not None]
summary = {
"positions": len(positions),
"n": len(rows),
"train": len(rows) - len(val),
"val": len(val),
"avg_pass_loss": round(sum(losses) / len(losses), 2) if losses else None,
"seconds": round(time.time() - t0, 1),
**stats,
}
(out / "stats.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(summary, ensure_ascii=False, indent=2))
return 0
def main(argv: Optional[list[str]] = None) -> int:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--games", nargs="*", default=[], help="moves 기록이 있는 games.jsonl")
ap.add_argument("--cases", nargs="*", default=[], help="재요청 실험 pass_cases.jsonl")
ap.add_argument("--out", default="sft_data/passfix_v1")
ap.add_argument("--n", type=int, default=0, help="0 = 전부")
ap.add_argument("--min-empty", type=int, default=120, help="이보다 빈점이 적은 종반 PASS는 제외")
ap.add_argument("--min-loss", type=float, default=0.5, help="KataGo 기준 PASS 손해(집) 하한")
ap.add_argument("--seed", type=int, default=11)
ap.add_argument("--val-frac", type=float, default=0.05)
ap.add_argument("--katago-visits", type=int, default=64)
ap.add_argument("--komi", type=float, default=6.5)
return build(ap.parse_args(argv))
if __name__ == "__main__":
raise SystemExit(main())