예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
dataset_ladder.py
"""축(ladder) 학습 데이터 생성.
국면은 축 시작 모양(둘러싼 돌이 날일자·대각, 안쪽 돌이 단수/활로 2)을 8방향·임의 위치에 놓고
0~8수 진행시킨 뒤, 배경 돌과 축머리(도망 측 돌)/막는 돌(쫓는 측 돌)을 섞어 만든다.
축 성립 여부는 vlm_server.ladder 판독기가 정하고, 실제 둘 수는 KataGo 가 고른다.
샘플 종류(kind) — 정답 수는 모두 KataGo 최선
flee_dead : 내 단수 그룹이 축으로 잡힘 → 도망 금지
flee_alive : 축머리 등으로 도망이 성립 (살릴 가치가 없으면 KataGo 는 손을 뺌)
chase_work : 상대 활로 2 그룹을 축으로 잡을 수 있음 (급하지 않아 손을 빼는 경우가 많음)
chase_broken : 쫓아도 빠져나감 → 쫓지 않음
출력
<out>/train.jsonl, val.jsonl : 기존 baduk_v3_aligned 와 같은 수 예측 형식 (LoRA 혼합용)
<out>/judge_train.jsonl, judge_val : 텍스트 축 판정 에이전트용 (성립 여부·이유·권장 수)
<out>/stats.json : 판독기 vs KataGo 일치율 등
예)
python -m vlm_server.dataset_ladder --out sft_data/ladder_v1 --n 3000 --katago-visits 64
"""
from __future__ import annotations
import argparse
import json
import random
import sys
import time
from collections import Counter
from pathlib import Path
from typing import Optional
from .dataset_baduk import assistant_text, board_text_aligned, user_prompt_text
from .ladder import _Board, attacker_can_capture, chase_verdict, defender_can_escape, ladder_line
N = 19
Point = tuple[int, int]
LETTERS = "ABCDEFGHJKLMNOPQRST"
# 수비 돌 (0,0) 기준, 축은 +r,+c 방향으로 진행.
FLEE_ATT = [(-1, 0), (0, -1), (1, 0), (-1, 1)] # 수비 활로 (0,1) 하나
CHASE_ATT = [(-1, 0), (0, -1), (-1, 1)] # 수비 활로 (1,0),(0,1)
def _xf(p: Point, t: int) -> Point:
r, c = p
if t & 4:
r, c = c, r
if t & 1:
r = -r
if t & 2:
c = -c
return r, c
def _gtp(p: Point) -> str:
return f"{LETTERS[p[1]]}{N - p[0]}"
def _from_gtp(s: str) -> Optional[Point]:
s = s.strip().upper()
if s == "PASS":
return None
return N - int(s[1:]), LETTERS.index(s[0])
class _Gen:
def __init__(self, rng: random.Random):
self.rng = rng
def _place(self, b: _Board, rel: list[Point], origin: Point, t: int, who: int) -> bool:
for p in rel:
dr, dc = _xf(p, t)
r, c = origin[0] + dr, origin[1] + dc
if not (0 <= r < N and 0 <= c < N):
return False
if b.play(r * N + c, who) is None:
return False
return True
def _path_cells(self, origin: Point, t: int, length: int) -> list[Point]:
out = []
for k in range(1, length):
for off in ((k, k), (k, k + 1), (k + 1, k)):
dr, dc = _xf(off, t)
r, c = origin[0] + dr, origin[1] + dc
if 0 <= r < N and 0 <= c < N:
out.append((r, c))
return out
def _noise(self, b: _Board, avoid: set[Point], n: int) -> None:
"""축 돌 주변 1칸 밖에 배경 돌. 따냄·새 단수를 만들지 않는다."""
tries = 0
placed = 0
while placed < n and tries < n * 20:
tries += 1
r, c = self.rng.randrange(N), self.rng.randrange(N)
if any(abs(r - ar) <= 1 and abs(c - ac) <= 1 for ar, ac in avoid):
continue
idx = r * N + c
b2 = b.copy()
if b2.play(idx, self.rng.choice((1, 2))) != 0:
continue
if any(b2.c[i] and len(b2.group(i)[1]) == 1 for i in [idx, *b2.nbrs(idx)]):
continue
b.c[:] = b2.c
placed += 1
def make(self, mode: str) -> Optional[dict]:
"""mode: flee | chase. 국면·수비 돌 좌표·수비 색·진행 방향 반환."""
rng = self.rng
t = rng.randrange(8)
defender = rng.choice((1, 2))
attacker = 3 - defender
origin = (rng.randrange(2, N - 2), rng.randrange(2, N - 2))
b = _Board(bytes(N * N))
ok = self._place(b, [(0, 0)], origin, t, defender) and self._place(
b, FLEE_ATT if mode == "flee" else CHASE_ATT, origin, t, attacker
)
if not ok:
return None
idx = origin[0] * N + origin[1]
if mode == "flee":
steps = rng.choice((0, 0, 1, 2, 3, 4, 6, 8))
line = ladder_line(b.c, *origin)
for who, (r, c) in line[: 2 * steps]:
b.play(r * N + c, who)
if b.c[idx] != defender or len(b.group(idx)[1]) != 1:
return None
ladder_stones = {divmod(i, N) for i in range(N * N) if b.c[i]}
path = [p for p in self._path_cells(origin, t, 18) if b.c[p[0] * N + p[1]] == 0]
roll = rng.random()
if path and roll < 0.35:
p = rng.choice(path)
b.play(p[0] * N + p[1], defender)
elif path and roll < 0.5:
p = rng.choice(path)
b.play(p[0] * N + p[1], attacker)
self._noise(b, ladder_stones, rng.randrange(6, 30))
if b.c[idx] != defender:
return None
return {"cells": bytes(b.c), "target": origin, "defender": defender, "mode": mode}
class _KataGo:
def __init__(self, visits: int, komi: float):
root = Path(__file__).resolve().parents[1]
if str(root) not in sys.path:
sys.path.insert(0, str(root))
(root / "logs/ensemble").mkdir(parents=True, exist_ok=True)
from scripts.mariago_ensemble_proxy import KataGoJudge
self.j = KataGoJudge(visits=visits, komi=komi, timeout_s=60)
self.visits = visits
self.komi = komi
def best(self, cells: bytes, side: int, avoid: Optional[Point] = None) -> tuple[Optional[Point], dict]:
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]
payload = {
"initialStones": stones,
"initialPlayer": player,
"moves": [],
"rules": "korean",
"komi": self.komi,
"boardXSize": N,
"boardYSize": N,
"analyzeTurns": [0],
"maxVisits": self.visits,
}
res = self.j._query(payload)
infos = sorted(res.get("moveInfos") or [], key=lambda m: m.get("order", 99))
flip = -1.0 if side == 2 else 1.0
lead = {m["move"]: flip * m.get("scoreLead", 0.0) for m in infos}
for m in infos:
p = _from_gtp(m["move"])
if p is None or p == avoid:
continue
return p, lead
return None, lead
def _judge_prompt(cells: bytes, side: int, target: Point, mode: str) -> str:
side_name = "흑(X)" if side == 1 else "백(O)"
tr, tc = target
who = "당신의" if mode == "flee" else "상대의"
q = (
f"({tr},{tc}) 에 있는 {who} 돌 그룹이 축(ladder)으로 잡히는지 판정하세요. "
"축머리(진행 방향의 도망 측 돌), 진행 방향의 쫓는 측 돌, 쫓는 돌이 >/< 로 꺾이며 단수에 걸리는지, "
"판 가장자리를 모두 고려하세요."
)
return (
"당신은 바둑 수읽기 보조 에이전트입니다. 보드 문자 격자: .=빈칸 X=흑 O=백, 좌표는 0-indexed (row, col).\n\n"
f"당신의 차례: {side_name}.\n{board_text_aligned(cells)}\n\n{q}\n"
'응답은 JSON 한 줄: {"ladder": "captured"|"escapes", "action": "flee"|"no_flee"|"chase"|"no_chase", '
'"ladder_point": [r, c] 또는 null (도망/단수 자리), "reason": "..."}'
)
def _judge_answer(kind: str, point: Optional[Point], reason: str) -> str:
ladder = "captured" if kind in ("flee_dead", "chase_work") else "escapes"
action = {"flee_dead": "no_flee", "flee_alive": "flee", "chase_work": "chase", "chase_broken": "no_chase"}[kind]
return json.dumps(
{"ladder": ladder, "action": action, "ladder_point": list(point) if point else None, "reason": reason},
ensure_ascii=False,
)
def _reason(kind: str, cells: bytes, target: Point) -> str:
line = ladder_line(cells, *target) if kind.startswith("flee") else []
if kind == "flee_dead":
tail = line[-1][1] if line else None
return f"뻗어도 축이 {tail} 부근까지 이어져 잡힌다. 더 두면 돌만 늘어난다." if tail else "뻗어도 활로가 늘지 않아 잡힌다."
if kind == "flee_alive":
return "진행 방향에 우군 돌(축머리)이 있거나 쫓는 돌이 단수에 걸려 뻗으면 빠져나간다."
if kind == "chase_work":
return "단수로 몰면 가장자리/쫓는 쪽 돌까지 도망칠 곳이 없어 잡힌다."
return "쫓아도 축머리나 꺾임 단수로 빠져나가 쫓는 쪽 약점만 남는다."
def build(args: argparse.Namespace) -> int:
rng = random.Random(args.seed)
gen = _Gen(rng)
if args.katago_visits <= 0:
raise SystemExit("--katago-visits 는 1 이상이어야 합니다 (정답 수를 KataGo 가 고름)")
kg = _KataGo(args.katago_visits, args.komi)
out = Path(args.out)
out.mkdir(parents=True, exist_ok=True)
want = {"flee_dead": args.n * 3 // 10, "flee_alive": args.n * 2 // 10, "chase_work": args.n * 2 // 10}
want["chase_broken"] = args.n - sum(want.values())
have: Counter = Counter()
stats: Counter = Counter()
moves_rows: list[dict] = []
judge_rows: list[dict] = []
t0 = time.time()
attempts = 0
while sum(have.values()) < args.n and attempts < args.n * 60:
attempts += 1
mode = "flee" if rng.random() < 0.5 else "chase"
pos = gen.make(mode)
if pos is None:
continue
cells, target, defender = pos["cells"], pos["target"], pos["defender"]
b = _Board(cells)
tidx = target[0] * N + target[1]
libs = b.group(tidx)[1]
if mode == "flee":
side = defender
esc = defender_can_escape(cells, *target)
if esc is None:
stats["unreadable"] += 1
continue
kind = "flee_alive" if esc else "flee_dead"
flee_pt: Point = divmod(libs[0], N)
b2 = b.copy()
if b2.play(libs[0], side) is None:
continue
else:
side = 3 - defender
if len(libs) != 2:
continue
cap = attacker_can_capture(cells, *target)
if cap is None:
stats["unreadable"] += 1
continue
kind = "chase_work" if cap else "chase_broken"
chase_pts = [divmod(l, N) for l in libs]
working = [
p for p in chase_pts
if (v := chase_verdict(cells, p[0], p[1], side)) is not None and v.kind == "working_chase"
]
if have[kind] >= want[kind]:
continue
# 축 수(도망/단수)는 판독기, 실제 정답 수는 KataGo. 잡히는 축은 급하지 않아 KataGo 가 손을 빼는 게 보통.
if kind.startswith("flee"):
ladder_move: Optional[Point] = flee_pt
elif kind == "chase_work":
ladder_move = working[0] if working else None
else:
ladder_move = None
target_move, lead = kg.best(cells, side)
if target_move is None:
continue
# 판독기와 KataGo 가 정면으로 어긋나는 경우(죽은 축 도망 / 깨진 축 추격)
contradict = (kind == "flee_dead" and target_move == flee_pt) or (
kind == "chase_broken" and target_move in chase_pts
)
stats[f"{kind}_n"] += 1
if contradict:
stats[f"{kind}_contradict"] += 1
if args.drop_disagree:
continue
agree = target_move == ladder_move if ladder_move else not contradict
stats[f"{kind}_katago_plays_ladder_move"] += int(ladder_move is not None and target_move == ladder_move)
loss = None
if ladder_move is not None and _gtp(ladder_move) in lead:
loss = round(lead[_gtp(target_move)] - lead[_gtp(ladder_move)], 2)
stats[f"{kind}_ladder_move_loss_sum"] += loss
stats[f"{kind}_ladder_move_loss_n"] += 1
have[kind] += 1
i = len(moves_rows)
moves_rows.append(
{
"id": i,
"size": N,
"ai_side": side,
"target_row": target_move[0],
"target_col": target_move[1],
"is_pass": False,
"is_tactical": True,
"board_format": "aligned",
"text_only": True,
"images": [],
"ladder": {
"kind": kind,
"group_at": list(target),
"ladder_move": list(ladder_move) if ladder_move else None,
"chase_points": [list(p) for p in chase_pts] if mode == "chase" else [],
"katago_agree": agree,
"ladder_move_loss": loss,
},
"messages": [
{"role": "user", "content": [{"type": "text", "text": user_prompt_text(cells, side, board_format="aligned", text_only=True)}]},
{"role": "assistant", "content": [{"type": "text", "text": assistant_text(target_move)}]},
],
}
)
judge_rows.append(
{
"id": i,
"size": N,
"ai_side": side,
"text_only": True,
"images": [],
"ladder": {"kind": kind, "group_at": list(target)},
"messages": [
{"role": "user", "content": [{"type": "text", "text": _judge_prompt(cells, side, target, mode)}]},
{"role": "assistant", "content": [{"type": "text", "text": _judge_answer(kind, flee_pt if mode == "flee" else (working[0] if working else None), _reason(kind, cells, target))}]},
],
}
)
done = sum(have.values())
if done % 100 == 0:
print(f"[ladder] {done}/{args.n} {dict(have)} {time.time() - t0:.0f}s", flush=True)
idx = list(range(len(moves_rows)))
rng.shuffle(idx)
n_val = max(1, int(len(idx) * args.val_frac))
val_set = set(idx[:n_val])
def dump(rows: list[dict], name: str) -> None:
with open(out / f"{name}train.jsonl", "w", encoding="utf-8") as ft, open(
out / f"{name}val.jsonl", "w", encoding="utf-8"
) as fv:
for j in idx:
(fv if j in val_set else ft).write(json.dumps(rows[j], ensure_ascii=False) + "\n")
dump(moves_rows, "")
dump(judge_rows, "judge_")
summary = {
"n": len(moves_rows),
"by_kind": dict(have),
"katago_visits": args.katago_visits,
"attempts": attempts,
"seconds": round(time.time() - t0, 1),
**{k: v for k, v in stats.items()},
}
for k in want:
n = stats.get(f"{k}_n", 0)
if n:
summary[f"{k}_contradict_rate"] = round(stats.get(f"{k}_contradict", 0) / n, 3)
summary[f"{k}_katago_plays_ladder_move_rate"] = round(stats[f"{k}_katago_plays_ladder_move"] / n, 3)
ln = stats.get(f"{k}_ladder_move_loss_n", 0)
if ln:
summary[f"{k}_ladder_move_avg_loss"] = round(stats[f"{k}_ladder_move_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("--out", default="sft_data/ladder_v1")
ap.add_argument("--n", type=int, default=3000)
ap.add_argument("--seed", type=int, default=7)
ap.add_argument("--val-frac", type=float, default=0.05)
ap.add_argument("--katago-visits", type=int, default=64, help="정답 수를 고를 KataGo 탐색 수")
ap.add_argument("--komi", type=float, default=6.5)
ap.add_argument("--drop-disagree", action="store_true", help="판독기와 KataGo 가 다르면 버림")
return build(ap.parse_args(argv))
if __name__ == "__main__":
raise SystemExit(main())