예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
cross_lessons.py
"""휴리스틱과 R5가 대국에서 확인한 수를 서로 가산점으로 넘긴다.
PASS, 자충, 호구, 축 헛도망은 옮기지 않는다.
이긴 쪽의 따냄·단수 구원·잡히는 축은 두 배로 쌓는다.
단, 진 쪽이 중반에 PASS를 반복한 판은 승패가 수의 질을 반영하지 않으므로
승자 가중치를 주지 않는다.
"""
from __future__ import annotations
import json
import os
import re
import threading
import time
from pathlib import Path
from typing import Iterable, Optional
from .ladder import BOARD_SIZE, _Board, chase_verdict, futile_escapes
from .shapes import hogu_verdict
ROOT = Path(__file__).resolve().parents[1]
LESSON_PATH = ROOT / "vlm_models" / "cross_lessons.json"
MAX_GAME_PLIES = 200
TAG_CAP = 40
MIDGAME_EMPTY_MIN = 120
FORFEIT_PASS_MIN = 3
_MOVE_RE = re.compile(
r"^\d+\.\s+([BW])\s+(?:PASS|\S+\s+\((\d+),(\d+)\))"
)
_LOCK = threading.Lock()
_CACHE: dict = {"mtime": None, "loaded_at": 0.0, "book": None}
_FUTILE: dict = {}
def empty_book() -> dict:
return {
"to_heuristic": {"capture": 0, "atari_save": 0, "working_chase": 0},
"to_r5": {"capture": 0, "atari_save": 0, "working_chase": 0},
"seen_games": [],
}
def _normalize(raw: Optional[dict]) -> dict:
book = empty_book()
if not isinstance(raw, dict):
return book
for side in ("to_heuristic", "to_r5"):
src = raw.get(side) or {}
if isinstance(src, dict):
for tag in book[side]:
try:
book[side][tag] = max(0, int(src.get(tag) or 0))
except (TypeError, ValueError):
book[side][tag] = 0
seen = raw.get("seen_games") or []
if isinstance(seen, list):
book["seen_games"] = [str(x) for x in seen]
return book
def load_book(path: Path = LESSON_PATH, *, force: bool = False) -> dict:
now = time.time()
with _LOCK:
if (
not force
and _CACHE["book"] is not None
and path == LESSON_PATH
and now - float(_CACHE["loaded_at"]) < 2.0
):
return _CACHE["book"]
book = empty_book()
mtime = None
if path.is_file():
mtime = path.stat().st_mtime
if (
not force
and path == LESSON_PATH
and _CACHE["book"] is not None
and _CACHE["mtime"] == mtime
):
_CACHE["loaded_at"] = now
return _CACHE["book"]
try:
book = _normalize(json.loads(path.read_text(encoding="utf-8")))
except (OSError, json.JSONDecodeError):
book = empty_book()
if path == LESSON_PATH:
_CACHE["book"] = book
_CACHE["mtime"] = mtime
_CACHE["loaded_at"] = now
return book
def save_book(book: dict, path: Path = LESSON_PATH) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
payload = _normalize(book)
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
os.replace(tmp, path)
with _LOCK:
if path == LESSON_PATH:
_CACHE["book"] = payload
_CACHE["mtime"] = path.stat().st_mtime
_CACHE["loaded_at"] = time.time()
def _cells(board) -> bytes:
if isinstance(board, (bytes, bytearray)):
return bytes(board)
return bytes(int(v) for v in board)
def _futile_points(cells: bytes, who: int) -> set:
key = (who, cells)
hit = _FUTILE.get(key)
if hit is None:
hit = {v.point for v in futile_escapes(cells, who)}
if len(_FUTILE) > 8:
_FUTILE.clear()
_FUTILE[key] = hit
return hit
def cheap_tags(board, r: int, c: int, who: int) -> set[str]:
"""따냄·단수 구원만. 호구·자충·축 헛도망이면 비운다."""
cells = _cells(board)
size = BOARD_SIZE
idx = r * size + c
if not (0 <= r < size and 0 <= c < size) or cells[idx] != 0:
return set()
if hogu_verdict(cells, r, c, who) in ("hogu_entry", "hogu_fill"):
return set()
if (r, c) in _futile_points(cells, who):
return set()
before = _Board(cells, size)
saved = False
seen: set[int] = set()
for nb in before.nbrs(idx):
if before.c[nb] != who or nb in seen:
continue
stones, libs = before.group(nb)
seen.update(stones)
if len(libs) == 1 and libs[0] == idx:
saved = True
after = before.copy()
captured = after.play(idx, who)
if captured is None:
return set()
_stones, libs_after = after.group(idx)
if captured == 0 and len(libs_after) <= 1:
return set()
tags: set[str] = set()
if captured > 0:
tags.add("capture")
if saved and len(libs_after) >= 2:
tags.add("atari_save")
return tags
def positive_tags(board, r: int, c: int, who: int) -> set[str]:
"""대국 반영용. 잡히는 축만 추가로 인정한다."""
tags = cheap_tags(board, r, c, who)
if not tags and hogu_verdict(_cells(board), r, c, who) in ("hogu_entry", "hogu_fill"):
return set()
cells = _cells(board)
if (r, c) in _futile_points(cells, who):
return set()
verdict = chase_verdict(cells, r, c, who)
if verdict is not None and verdict.kind == "working_chase":
tags.add("working_chase")
return tags
def _add_tags(book: dict, target: str, tags: Iterable[str], weight: int) -> None:
side = book[target]
for tag in tags:
if tag in side:
side[tag] = int(side[tag]) + int(weight)
def absorb_move(book: dict, actor: str, tags: Iterable[str], winner: Optional[str]) -> None:
"""actor 가 둔 좋은 수를 상대 테이블에 넣는다. actor 는 black|white."""
tags = [t for t in tags if t in ("capture", "atari_save", "working_chase")]
if not tags:
return
if actor == "white":
target = "to_heuristic"
weight = 2 if winner == "white" else 1
else:
target = "to_r5"
weight = 2 if winner == "black" else 1
_add_tags(book, target, tags, weight)
def parse_move(line: str) -> Optional[tuple]:
m = _MOVE_RE.match(line.strip())
if not m:
return None
actor = "black" if m.group(1) == "B" else "white"
if m.group(2) is None:
return actor, None, None, True
return actor, int(m.group(2)), int(m.group(3)), False
def midgame_pass_counts(moves: list[str]) -> dict[str, int]:
"""빈점이 MIDGAME_EMPTY_MIN 이상 남은 상태에서 나온 PASS 수 (색별)."""
counts = {"black": 0, "white": 0}
cells = bytearray(BOARD_SIZE * BOARD_SIZE)
for line in 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 cells.count(0) >= MIDGAME_EMPTY_MIN:
counts[actor] += 1
continue
played = _Board(cells)
if played.play(r * BOARD_SIZE + c, 1 if actor == "black" else 2) is None:
continue
cells = played.c
return counts
def effective_winner(moves: list[str], winner: Optional[str]) -> Optional[str]:
"""진 쪽이 중반 PASS를 반복했다면 승자 가중치를 주지 않는다(None)."""
if winner not in ("black", "white"):
return winner
loser = "white" if winner == "black" else "black"
if midgame_pass_counts(moves)[loser] >= FORFEIT_PASS_MIN:
return None
return winner
def absorb_game(book: dict, moves: list[str], winner: Optional[str], game_key: str) -> bool:
if game_key in book["seen_games"]:
return False
winner = effective_winner(moves, winner)
cells = bytearray(BOARD_SIZE * BOARD_SIZE)
for line in 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:
continue
who = 1 if actor == "black" else 2
tags = positive_tags(bytes(cells), r, c, who)
played = _Board(cells)
if played.play(r * BOARD_SIZE + c, who) is None:
continue
absorb_move(book, actor, tags, winner)
cells = played.c
book["seen_games"].append(game_key)
return True
def ingest_jsonl(path: Path, book: Optional[dict] = None, *, source: str = "") -> dict:
book = _normalize(book if book is not None else load_book())
label = source or path.stem
latest: dict[str, dict] = {}
if not path.is_file():
return book
with path.open(encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
rec = json.loads(line)
except json.JSONDecodeError:
continue
gid = rec.get("game_id")
latest[str(gid)] = rec
for gid, rec in latest.items():
winner = (rec.get("score") or {}).get("winner")
absorb_game(book, list(rec.get("moves") or []), winner, f"{label}:{gid}")
return book
def _bonus(book: dict, side: str, tags: Iterable[str], scale: float, cap: float) -> float:
total = 0.0
weights = book.get(side) or {}
for tag in tags:
try:
count = int(weights.get(tag) or 0)
except (TypeError, ValueError):
count = 0
total += min(TAG_CAP, max(0, count)) * scale
return float(min(cap, total))
def lesson_bonus_heuristic(board, r: int, c: int, who: int, book: Optional[dict] = None) -> int:
tags = cheap_tags(board, r, c, who)
if not tags:
return 0
loaded = book if book is not None else load_book()
return int(_bonus(loaded, "to_heuristic", tags, 35.0, 1400.0))
def working_chase_bonus_r5(book: Optional[dict] = None) -> float:
loaded = book if book is not None else load_book()
return _bonus(loaded, "to_r5", ("working_chase",), 0.04, 1.2)
def lesson_bonus_r5(board, r: int, c: int, who: int, book: Optional[dict] = None) -> float:
tags = cheap_tags(board, r, c, who)
if not tags:
return 0.0
loaded = book if book is not None else load_book()
return _bonus(loaded, "to_r5", tags, 0.04, 1.2)
def heuristic_shape_delta(board, r: int, c: int, who: int) -> float:
"""휴리스틱의 한 칸 뜀·날일자를 R5 계가 단위로 가산한다."""
cells = _cells(board)
size = BOARD_SIZE
own = [
divmod(i, size)
for i, v in enumerate(cells)
if v == who
]
if len(own) < 4:
return 0.0
relation = 0.0
for rr, cc in own:
dr, dc = abs(r - rr), abs(c - cc)
if (dr, dc) in ((0, 2), (2, 0)):
relation = max(relation, 0.55)
elif (dr, dc) in ((1, 2), (2, 1)):
relation = max(relation, 0.4)
elif (dr, dc) in ((0, 3), (3, 0)):
relation = max(relation, 0.25)
return relation
def _puts_opponent_in_atari(cells: bytes, r: int, c: int, who: int) -> bool:
size = BOARD_SIZE
idx = r * size + c
if not (0 <= r < size and 0 <= c < size) or cells[idx] != 0:
return False
board = _Board(cells, size)
opp = 3 - who
seen: set[int] = set()
for nb in board.nbrs(idx):
if board.c[nb] != opp or nb in seen:
continue
stones, libs = board.group(nb)
seen.update(stones)
if len(libs) == 2 and idx in libs:
return True
return False
def r5_live_delta(board, r: int, c: int, who: int) -> int:
"""R5가 쓰던 축·호구·집 메우기·근처 양단수를 휴리스틱 점수에 넣는다."""
cells = _cells(board)
size = BOARD_SIZE
score = 0
hogu = hogu_verdict(cells, r, c, who)
if hogu == "hogu_entry":
score -= 9000
elif hogu == "hogu_fill":
score -= 2000
if (r, c) in _futile_points(cells, who):
score -= 14000
if _puts_opponent_in_atari(cells, r, c, who):
verdict = chase_verdict(cells, r, c, who)
if verdict is not None:
if verdict.kind == "working_chase":
score += 2500 + verdict.group_size * 80
score += int(_bonus(load_book(), "to_heuristic", ("working_chase",), 35.0, 1400.0))
elif verdict.kind == "broken_chase":
score -= 2000
if _own_territory_fill(cells, r, c, who):
score -= 3500
if _nearby_double_atari(cells, r, c, who):
score -= 6000
return score
def _own_territory_fill(cells: bytes, r: int, c: int, who: int) -> bool:
size = BOARD_SIZE
idx = r * size + c
if cells[idx] != 0:
return False
opp = 3 - who
for dr, dc in ((-1, 0), (1, 0), (0, -1), (0, 1)):
nr, nc = r + dr, c + dc
if 0 <= nr < size and 0 <= nc < size and cells[nr * size + nc] == opp:
return False
seen = bytearray(size * size)
stack = [idx]
seen[idx] = 1
filled = 1
while stack:
cur = stack.pop()
rr, cc = divmod(cur, size)
if rr in (0, size - 1) or cc in (0, size - 1):
return False
if filled > 48:
return False
for dr, dc in ((-1, 0), (1, 0), (0, -1), (0, 1)):
nr, nc = rr + dr, cc + dc
if not (0 <= nr < size and 0 <= nc < size):
continue
nidx = nr * size + nc
v = cells[nidx]
if v == 0 and not seen[nidx]:
seen[nidx] = 1
filled += 1
stack.append(nidx)
elif v != 0 and v != who:
return False
return True
def _nearby_double_atari(cells: bytes, r: int, c: int, who: int) -> bool:
size = BOARD_SIZE
base = _Board(cells, size)
idx = r * size + c
if base.play(idx, who) is None:
return False
opp = 3 - who
own_groups: list[tuple[int, int]] = []
seen: set[int] = set()
for i, v in enumerate(base.c):
if v != who or i in seen:
continue
stones, libs = base.group(i)
seen.update(stones)
if len(libs) >= 2:
own_groups.append((stones[0], len(stones)))
if len(own_groups) < 2:
return False
r0, r1 = max(0, r - 2), min(size, r + 3)
c0, c1 = max(0, c - 2), min(size, c + 3)
for rr in range(r0, r1):
for cc in range(c0, c1):
nidx = rr * size + cc
if base.c[nidx] != 0:
continue
reply = base.copy()
if reply.play(nidx, opp) is None:
continue
threatened = 0
for rep, _size in own_groups:
if reply.c[rep] != who:
threatened += 1
continue
_stones, libs = reply.group(rep)
if len(libs) <= 1:
threatened += 1
if threatened >= 2:
return True
return False