예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
dataset_hogu.py
"""호구(虎口) 학습 데이터 생성 — 실전 국면(baduk_v3_aligned)에서 호구 모양이 있는 것만 골라 KataGo 로 정답.
종류(kind) — 둘 차례 쪽 기준, 정답 수는 모두 KataGo 최선
peep : 내 호구 입구에 상대가 들여다봄 → 끊김 위협이지만 실전에선 대부분 손을 뺀다(따라 잇지 말 것)
entry : 상대 호구 안에 들어가면 단수 → 두지 말 것
fill : 내 호구(입구 비어 있고 둘러싼 그룹 활로 충분)를 메움 → 느린 수, 두지 말 것
python -m vlm_server.dataset_hogu --out sft_data/hogu_v1 --n 3000 --katago-visits 100
"""
from __future__ import annotations
import argparse
import json
import random
import time
from collections import Counter
from pathlib import Path
from typing import Optional
from .dataset_baduk import assistant_text, user_prompt_text
from .dataset_ladder import _KataGo, _gtp
from .ladder import _Board
from .shapes import hogu_verdict
N = 19
Point = tuple[int, int]
KINDS = ("peep", "entry", "fill")
def parse_aligned(text: str) -> Optional[bytearray]:
lines = text.split("\n")
try:
i = next(k for k, l in enumerate(lines) if l.startswith(" 00 01"))
except StopIteration:
return None
b = bytearray(N * N)
for r in range(N):
toks = lines[i + 1 + r][3:].split()
if len(toks) < N:
return None
for c, t in enumerate(toks[:N]):
b[r * N + c] = {".": 0, "X": 1, "O": 2}.get(t, 0)
return b
def peep_points(cells: bytes | bytearray, who: int) -> list[Point]:
"""내(who) 호구 입구를 상대가 막은 점 — 상대가 그 점에 두면 내 돌이 끊긴다."""
b = _Board(cells)
opp = 3 - who
out: list[Point] = []
for idx in range(N * N):
if b.c[idx] != 0:
continue
r, c = divmod(idx, N)
own: list[int] = []
opp_n = 0
walls = 0
for dr, dc in ((-1, 0), (1, 0), (0, -1), (0, 1)):
nr, nc = r + dr, c + dc
if not (0 <= nr < N and 0 <= nc < N):
walls += 1
continue
v = b.c[nr * N + nc]
if v == who:
own.append(nr * N + nc)
elif v == opp:
opp_n += 1
if opp_n != 1 or len(own) < 2 or len(own) + walls != 3:
continue
groups = set()
for s in own:
groups.add(min(b.group(s)[0]))
if len(groups) < 2:
continue
b2 = b.copy()
cap = b2.play(idx, opp)
if cap is None or len(b2.group(idx)[1]) < 2:
continue
out.append((r, c))
return out
def classify(cells: bytes | bytearray, who: int) -> Optional[tuple[str, list[Point]]]:
pp = peep_points(cells, who)
if pp:
return "peep", pp
entry: list[Point] = []
fill: list[Point] = []
for idx in range(N * N):
if cells[idx]:
continue
v = hogu_verdict(cells, idx // N, idx % N, who)
if v == "hogu_entry":
entry.append(divmod(idx, N))
elif v == "hogu_fill":
fill.append(divmod(idx, N))
if entry:
return "entry", entry
if fill:
return "fill", fill
return None
def build(args: argparse.Namespace) -> int:
rng = random.Random(args.seed)
src = Path(args.src)
lines = [l for l in src.open(encoding="utf-8") if l.strip()]
rng.shuffle(lines)
want = {"peep": int(args.n * 0.35), "entry": int(args.n * 0.30)}
want["fill"] = args.n - sum(want.values())
kg = _KataGo(args.katago_visits, args.komi)
out = Path(args.out)
out.mkdir(parents=True, exist_ok=True)
have: Counter = Counter()
stats: Counter = Counter()
rows: list[dict] = []
t0 = time.time()
for line in lines:
if sum(have.values()) >= args.n:
break
d = json.loads(line)
if d.get("is_pass"):
continue
cells = parse_aligned(d["messages"][0]["content"][0]["text"])
if cells is None or sum(1 for v in cells if v) < args.min_stones:
continue
side = int(d["ai_side"])
stats["scanned"] += 1
cl = classify(cells, side)
if cl is None:
continue
kind, pts = cl
if have[kind] >= want[kind]:
continue
best, lead = kg.best(bytes(cells), side)
if best is None:
continue
on_point = best in pts
stats[f"{kind}_n"] += 1
stats[f"{kind}_katago_on_point"] += int(on_point)
sgf = (d.get("target_row"), d.get("target_col"))
stats[f"{kind}_sgf_on_point"] += int(sgf in [tuple(p) for p in pts])
loss = None
g0 = _gtp(pts[0])
if g0 in lead:
loss = round(lead[_gtp(best)] - lead[g0], 2)
stats[f"{kind}_point_loss_sum"] += loss
stats[f"{kind}_point_loss_n"] += 1
have[kind] += 1
rows.append(
{
"id": len(rows),
"size": N,
"ai_side": side,
"target_row": best[0],
"target_col": best[1],
"is_pass": False,
"is_tactical": True,
"board_format": "aligned",
"text_only": True,
"images": [],
"hogu": {
"kind": kind,
"points": [list(p) for p in pts],
"katago_on_point": on_point,
"point_loss": loss,
"sgf_move": list(sgf),
},
"messages": [
{"role": "user", "content": [{"type": "text", "text": user_prompt_text(bytes(cells), side, board_format="aligned", text_only=True)}]},
{"role": "assistant", "content": [{"type": "text", "text": assistant_text(best)}]},
],
}
)
if len(rows) % 100 == 0:
print(f"[hogu] {len(rows)}/{args.n} {dict(have)} scanned={stats['scanned']} {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))
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")
summary: dict = {"n": len(rows), "by_kind": dict(have), "seconds": round(time.time() - t0, 1), **stats}
for k in KINDS:
n = stats.get(f"{k}_n", 0)
if n:
summary[f"{k}_katago_on_point_rate"] = round(stats[f"{k}_katago_on_point"] / n, 3)
summary[f"{k}_sgf_on_point_rate"] = round(stats[f"{k}_sgf_on_point"] / n, 3)
ln = stats.get(f"{k}_point_loss_n", 0)
if ln:
summary[f"{k}_point_avg_loss"] = round(stats[f"{k}_point_loss_sum"] / ln, 2)
(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("--src", default="sft_data/baduk_v3_aligned/train.jsonl")
ap.add_argument("--out", default="sft_data/hogu_v1")
ap.add_argument("--n", type=int, default=3000)
ap.add_argument("--seed", type=int, default=5)
ap.add_argument("--min-stones", type=int, default=20)
ap.add_argument("--val-frac", type=float, default=0.05)
ap.add_argument("--katago-visits", type=int, default=100)
ap.add_argument("--komi", type=float, default=6.5)
return build(ap.parse_args(argv))
if __name__ == "__main__":
raise SystemExit(main())