예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
dataset_onpolicy.py
"""휴리스틱 vs R5 대국(SQLite)의 백 차례 국면을 KataGo로 라벨링한 학습 데이터.
monitor_mariago_web_play.py 가 남긴 matches.sqlite 의 moves(color='W', attempt='first')를
읽어, 각 국면의 KataGo 최선수를 정답으로 하는 행을 만든다(형식은 baduk_v3_aligned 와 동일).
서버가 PASS한 국면(onpolicy_pass)은 KataGo 최선수가 PASS가 아니고 손해가 min_loss 이상일 때만 남긴다.
부산물로 diag.json 에 VLM 제안(vlm_prefer)·서버 최종 착수·PASS가 KataGo top-k 와
얼마나 맞는지 기록한다 — 서버 계가 휴리스틱이 모델 제안을 얼마나 덮어쓰는지 확인용.
python -m vlm_server.dataset_onpolicy --db logs/<run>/matches.sqlite --out sft_data/onpolicy_v1
"""
from __future__ import annotations
import argparse
import json
import random
import sqlite3
import time
from collections import Counter
from pathlib import Path
from typing import Optional
from .dataset_baduk import assistant_text, user_prompt_text
N = 19
WHITE = 2
def _cells(text: str) -> bytes:
return bytes(".XO".index(ch) for ch in text)
def analyze(kg, cells: bytes, side: int, top_k: int) -> dict:
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: dict[str, float] = {str(m["move"]).lower(): flip * float(m.get("scoreLead", 0.0)) for m in infos}
top: list[list[int]] = []
for m in infos:
p = _from_gtp(m["move"])
if p is not None:
top.append([p[0], p[1]])
if len(top) >= top_k:
break
top_is_pass = bool(infos) and str(infos[0]["move"]).lower() == "pass"
pass_loss = None
if top and "pass" in lead:
pass_loss = round(lead[_gtp(tuple(top[0])).lower()] - lead["pass"], 2)
root_lead = res.get("rootInfo", {}).get("scoreLead")
return {
"top": top,
"top_is_pass": top_is_pass,
"pass_loss": pass_loss,
"lead": None if root_lead is None else round(flip * float(root_lead), 2),
}
def build(args: argparse.Namespace) -> int:
from .dataset_ladder import _KataGo
con = sqlite3.connect(args.db)
rows_db = con.execute(
"""select m.game_id, m.ply, m.kind, m.row, m.col, m.board, m.server_rec
from moves m
where m.color=? and m.attempt='first' and m.kind in ('move','pass')
and (
? = ''
or exists (
select 1 from games g
where g.run_id = m.run_id and g.game_id = m.game_id
and date(replace(substr(g.finished_at,1,19),'T',' '), '+9 hours') = ?
)
)
order by m.game_id, m.ply""",
(args.color, args.day, args.day),
).fetchall()
con.close()
rng = random.Random(args.seed)
side = 1 if args.color == "B" else WHITE
print(f"[onpolicy] {args.color} positions={len(rows_db)} day={args.day or 'all'}", flush=True)
kg = _KataGo(args.katago_visits, args.komi)
out = Path(args.out)
out.mkdir(parents=True, exist_ok=True)
diag: Counter = Counter()
pass_reasons: Counter = Counter()
pass_losses: list[float] = []
rows: list[dict] = []
t0 = time.time()
for i, (gid, ply, kind, r, c, board, srec) in enumerate(rows_db, 1):
cells = _cells(board)
rec = json.loads(srec) if srec else {}
if args.skip_opening and cells.count(1) + cells.count(2) < args.skip_opening:
diag["skip_opening"] += 1
continue
v = analyze(kg, cells, side, args.top_k)
if not v["top"]:
diag["katago_empty"] += 1
continue
best = v["top"][0]
prefer = rec.get("vlm_prefer")
if kind == "pass":
diag["server_pass"] += 1
pass_reasons[str(rec.get("pass_reason"))] += 1
diag["server_pass_katago_agrees"] += int(v["top_is_pass"])
if v["pass_loss"] is not None:
pass_losses.append(v["pass_loss"])
diag["server_pass_loss_ge2"] += int(v["pass_loss"] >= 2.0)
else:
diag["server_move"] += 1
diag["server_top1"] += int([r, c] == best)
diag["server_topk"] += int([r, c] in v["top"])
if prefer:
diag["vlm_prefer"] += 1
diag["vlm_top1"] += int(list(prefer) == best)
diag["vlm_topk"] += int(list(prefer) in v["top"])
diag["vlm_followed"] += int(kind == "move" and [r, c] == list(prefer))
if v["top_is_pass"]:
diag["katago_pass"] += 1
continue
if kind == "pass" and (v["pass_loss"] is None or v["pass_loss"] < args.min_loss):
diag["pass_small_loss"] += 1
continue
label = "onpolicy_pass" if kind == "pass" else "onpolicy_move"
rows.append(
{
"id": len(rows),
"size": N,
"ai_side": side,
"target_row": best[0],
"target_col": best[1],
"is_pass": False,
"is_tactical": False,
"board_format": "aligned",
"text_only": True,
"images": [],
"onpolicy": {
"kind": label,
"game_id": gid,
"ply": ply,
"top": v["top"],
"server": None if kind == "pass" else [r, c],
"vlm_prefer": prefer,
"pass_reason": rec.get("pass_reason"),
"pass_loss": v["pass_loss"],
"lead": v["lead"],
},
"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(tuple(best))}]},
],
}
)
if i % 100 == 0:
print(f"[onpolicy] {i}/{len(rows_db)} kept={len(rows)} {time.time() - t0:.0f}s", flush=True)
# PASS 국면은 소수라 학습 혼합에서 비중을 늘린다.
weighted = rows + [r for r in rows if r["onpolicy"]["kind"] == "onpolicy_pass"] * max(0, args.pass_repeat - 1)
idx = list(range(len(rows)))
rng.shuffle(idx)
n_val = max(1, int(len(idx) * args.val_frac)) if idx else 0
val_ids = {rows[j]["id"] for j in idx[:n_val]}
rng.shuffle(weighted)
with open(out / "train.jsonl", "w", encoding="utf-8") as ft:
for row in weighted:
if row["id"] not in val_ids:
ft.write(json.dumps(row, ensure_ascii=False) + "\n")
with open(out / "val.jsonl", "w", encoding="utf-8") as fv:
for j in idx[:n_val]:
fv.write(json.dumps(rows[j], ensure_ascii=False) + "\n")
def rate(a: str, b: str) -> Optional[float]:
return round(diag[a] / diag[b], 3) if diag[b] else None
summary = {
"db": args.db,
"day": args.day or None,
"positions": len(rows_db),
"rows": len(rows),
"rows_pass": sum(1 for r in rows if r["onpolicy"]["kind"] == "onpolicy_pass"),
"val": n_val,
"server_top1_rate": rate("server_top1", "server_move"),
"server_topk_rate": rate("server_topk", "server_move"),
"vlm_top1_rate": rate("vlm_top1", "vlm_prefer"),
"vlm_topk_rate": rate("vlm_topk", "vlm_prefer"),
"vlm_followed_rate": rate("vlm_followed", "vlm_prefer"),
"server_pass_katago_agrees_rate": rate("server_pass_katago_agrees", "server_pass"),
"server_pass_loss_avg": round(sum(pass_losses) / len(pass_losses), 2) if pass_losses else None,
"pass_reasons": dict(pass_reasons),
"top_k": args.top_k,
"seconds": round(time.time() - t0, 1),
**diag,
}
(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("--db", required=True)
ap.add_argument("--day", default="", help="KST YYYY-MM-DD. 비우면 DB 전체")
ap.add_argument("--color", default="W", choices=["B", "W"], help="학습할 돌 색깔. 기본은 백")
ap.add_argument("--out", default="sft_data/onpolicy_v1")
ap.add_argument("--top-k", type=int, default=3)
ap.add_argument("--min-loss", type=float, default=0.5, help="서버 PASS 국면을 남길 KataGo PASS 손해 하한(집)")
ap.add_argument("--pass-repeat", type=int, default=4, help="PASS 국면 학습 반복 배수")
ap.add_argument("--skip-opening", type=int, default=6, help="돌 수가 이보다 적은 포석 국면 제외")
ap.add_argument("--seed", type=int, default=13)
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())