예제 소스
장기로 이해하는 규칙 엔진과 평가 함수 탐색 · 파일 59개
learn.py
#!/usr/bin/env python3
"""기록된 사람 기보(data/monomob_kibo.sqlite)로 평가 가중치 학습.
python learn.py 학습 + 검증 + data/weights_learned.txt 저장
python learn.py --side h 한(내) 수만 학습
python learn.py --arena 10 학습 가중치 vs 기본 가중치 자가대국 N판으로 세기 확인
방법: 각 국면에서 합법수마다 "둔 뒤 평가 점수"를 매기고, softmax로 사람이 둔 수의 확률을
최대화하도록 가중치를 좌표 탐색으로 조정 (기본값에서 멀어지면 벌점 = 데이터가 적을 때 과적합 방지).
4번째 국면마다 검증용으로 떼어 두고 학습에 쓰지 않음.
"""
from __future__ import annotations
import argparse
import json
import math
import sqlite3
import subprocess
import time
from collections import defaultdict
from pathlib import Path
import rules
ROOT = Path(__file__).resolve().parent
DB_PATH = ROOT / "data" / "monomob_kibo.sqlite"
OUT_WEIGHTS = ROOT / "data" / "weights_learned.txt"
OUT_BOOK = ROOT / "data" / "human_book.json"
ENGINE = ROOT / "janggi_engine"
ARENA = ROOT.parent / "arena" / "arena"
DEFAULT = {
"chariot": 130, "cannon": 70, "horse": 50, "elephant": 30, "advisor": 30, "pawn": 20,
"pawnAdvance": 3, "minorFile": 2, "minorRank": 1, "chariotKingLine": 8,
"kingCenter": 6, "checkPenalty": 18, "hanBonus": 15,
}
# kingCenter는 엔진 평가에서 궁을 건너뛰어 실제로 쓰이지 않음, hanBonus는 수 선택에 영향 없음
TUNABLE = [k for k in DEFAULT if k not in ("kingCenter", "hanBonus")]
MATERIAL = {"R": "chariot", "C": "cannon", "H": "horse", "E": "elephant", "A": "advisor", "P": "pawn"}
# ---------------------------------------------------------------- 데이터
def load_samples(side_filter: str | None, sources: list[str] | None = None) -> list[tuple[list[str], str, int, int]]:
conn = sqlite3.connect(DB_PATH)
out = []
query, params = "SELECT id, start_board FROM games", []
if sources:
query += f" WHERE source IN ({','.join('?' * len(sources))})"
params = sources
for gid, start in conn.execute(query + " ORDER BY id", params):
board = json.loads(start)
for side, move, after in conn.execute(
"SELECT side, move, board_after FROM moves WHERE game_id=? ORDER BY ply", (gid,)
):
if move != "pass" and (side_filter is None or side == side_filter):
f = int(move[1]) * 9 + int(move[0])
t = int(move[3]) * 9 + int(move[2])
out.append((board, side, f, t))
board = json.loads(after)
return out
# ---------------------------------------------------------------- 평가 (src/Ai.cpp evaluate와 동일)
def king_pos(board: list[str], side: str) -> int:
for i, p in enumerate(board):
if p == side + "K":
return i
return -1
def in_check(board: list[str], side: str) -> bool:
k = king_pos(board, side)
if k < 0:
return True
enemy = "h" if side == "c" else "c"
return any(p and p[0] == enemy and k in rules.targets(board, i) for i, p in enumerate(board))
def evaluate(board: list[str], to_move: str, w: dict[str, float]) -> float:
cho, han = 0.0, w["hanBonus"]
kings = {"c": king_pos(board, "c"), "h": king_pos(board, "h")}
for i, p in enumerate(board):
if not p or p[1] == "K":
continue
side, kind = p[0], p[1]
f, r = i % 9, i // 9
v = w[MATERIAL[kind]]
if kind == "P":
v += ((6 - r) if side == "c" else (r - 3)) * w["pawnAdvance"]
elif kind in ("H", "E"):
v += (4 - abs(f - 4)) * w["minorFile"] + (4 - abs(r - 4)) * w["minorRank"]
elif kind == "R":
ek = kings["h" if side == "c" else "c"]
if ek >= 0 and (f == ek % 9 or r == ek // 9):
v += w["chariotKingLine"]
if side == "c":
cho += v
else:
han += v
if in_check(board, "c"):
cho -= w["checkPenalty"]
if in_check(board, "h"):
han -= w["checkPenalty"]
score = cho - han
return score if to_move == "c" else -score
def candidates(board: list[str], side: str) -> list[tuple[int, int, list[str]]]:
out = []
for f, p in enumerate(board):
if not p or p[0] != side:
continue
for t in rules.targets(board, f):
nb = board[:]
nb[t], nb[f] = nb[f], ""
if not in_check(nb, side):
out.append((f, t, nb))
return out
# ---------------------------------------------------------------- 학습
class Dataset:
"""국면별 후보수와 '둔 뒤 판'을 미리 계산해 둠."""
def __init__(self, samples):
self.items = []
for board, side, f, t in samples:
cands = candidates(board, side)
idx = next((i for i, (cf, ct, _) in enumerate(cands) if cf == f and ct == t), None)
if idx is None or len(cands) < 2:
continue
opp = "h" if side == "c" else "c"
self.items.append((side, opp, cands, idx))
def stats(self, w: dict[str, float], temp: float) -> tuple[float, float, float]:
"""(평균 NLL, top-1 일치율, 사람 수의 평균 순위 백분위)"""
nll, top1, rank_sum = 0.0, 0, 0.0
for _side, opp, cands, idx in self.items:
scores = [-evaluate(nb, opp, w) for _, _, nb in cands]
m = max(scores)
z = sum(math.exp((s - m) / temp) for s in scores)
nll -= (scores[idx] - m) / temp - math.log(z)
if scores[idx] >= m:
top1 += 1
better = sum(1 for s in scores if s > scores[idx])
rank_sum += better / len(scores)
n = max(1, len(self.items))
return nll / n, top1 / n, rank_sum / n
def train(ds: Dataset, reg: float, temp: float) -> dict[str, float]:
w = {k: float(v) for k, v in DEFAULT.items()}
def loss(ww):
penalty = sum(((ww[k] - DEFAULT[k]) / max(DEFAULT[k], 5)) ** 2 for k in TUNABLE)
return ds.stats(ww, temp)[0] + reg * penalty
best = loss(w)
for step in (0.3, 0.15, 0.07):
improved = True
while improved:
improved = False
for k in TUNABLE:
for sign in (1, -1):
cand = dict(w)
delta = max(1.0, abs(w[k]) * step)
cand[k] = max(0.0, w[k] + sign * delta)
lc = loss(cand)
if lc < best - 1e-4:
w, best, improved = cand, lc, True
print(f" {k:16s} {w[k]:7.1f} loss={best:.4f}")
return {k: round(v) for k, v in w.items()}
# ---------------------------------------------------------------- 엔진 검증
def write_weights(w: dict[str, float], path: Path) -> None:
path.write_text("".join(f"{k}={int(v)}\n" for k, v in w.items()))
def engine_agreement(samples, weights_file: Path | None, depth: int) -> float:
lines = [
("cho" if side == "c" else "han") + " " + ",".join(p or "." for p in board)
for board, side, _f, _t in samples
]
cmd = [str(ENGINE), "--batch", "--depth", str(depth)]
if weights_file:
cmd += ["--weights", str(weights_file)]
out = subprocess.run(cmd, input="\n".join(lines) + "\n", capture_output=True, text=True).stdout.split()
hit = sum(
1 for (b, s, f, t), mv in zip(samples, out) if mv == f"{f % 9}{f // 9}{t % 9}{t // 9}"
)
return hit / max(1, len(samples))
def build_book(samples) -> dict:
book: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
for board, side, f, t in samples:
key = side + ":" + ",".join(p or "." for p in board)
book[key][f"{f % 9}{f // 9}{t % 9}{t // 9}"] += 1
return {k: dict(v) for k, v in book.items()}
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--side", choices=["c", "h"])
ap.add_argument("--reg", type=float, default=0.3)
ap.add_argument("--temp", type=float, default=25.0)
ap.add_argument("--depth", type=int, default=3)
ap.add_argument("--arena", type=int, default=0)
ap.add_argument("--source", help="기보 출처만 사용 (쉼표 구분: watch=관전 고수 기보, bot, bot_log, monomob)")
args = ap.parse_args()
samples = load_samples(args.side, args.source.split(",") if args.source else None)
train_s = [s for i, s in enumerate(samples) if i % 4 != 0]
test_s = [s for i, s in enumerate(samples) if i % 4 == 0]
print(f"사람 수 {len(samples)}개 (학습 {len(train_s)}, 검증 {len(test_s)})")
t0 = time.time()
ds_train, ds_test = Dataset(train_s), Dataset(test_s)
base_w = {k: float(v) for k, v in DEFAULT.items()}
print("\n[학습]")
learned = train(ds_train, args.reg, args.temp)
print(f" ({time.time() - t0:.1f}초)")
write_weights(learned, OUT_WEIGHTS)
OUT_BOOK.write_text(json.dumps(build_book(samples), ensure_ascii=False))
print("\n[가중치 변화]")
for k in TUNABLE:
if learned[k] != DEFAULT[k]:
print(f" {k:16s} {DEFAULT[k]:5d} → {int(learned[k]):5d}")
print("\n[1수 평가 기준: 사람 수 예측] 기본 학습")
for name, ds in (("학습 데이터", ds_train), ("검증 데이터", ds_test)):
b, l = ds.stats(base_w, args.temp), ds.stats({k: float(v) for k, v in learned.items()}, args.temp)
print(f" {name} 일치율 {b[1]:6.1%} {l[1]:6.1%}")
print(f" {name} 사람 수보다 높게 본 수 비율 {b[2]:6.1%} {l[2]:6.1%}")
print(f"\n[엔진 {args.depth}수 읽기: 사람 수와 같은 수를 고른 비율] 기본 학습")
for name, ss in (("학습 데이터", train_s), ("검증 데이터", test_s)):
b = engine_agreement(ss, None, args.depth)
l = engine_agreement(ss, OUT_WEIGHTS, args.depth)
print(f" {name} {b:6.1%} {l:6.1%}")
print(f"\n저장: {OUT_WEIGHTS.relative_to(ROOT)}, {OUT_BOOK.relative_to(ROOT)}")
if args.arena:
print(f"\n[자가대국 {args.arena}판: A=학습 가중치, B=기본, 깊이 3]")
out = subprocess.run(
[str(ARENA), "--games", str(args.arena), "--a", "3", "--b", "3", "--wa", str(OUT_WEIGHTS),
"--max-ply", "200"],
capture_output=True, text=True,
).stdout
print(out.strip().splitlines()[-1])
if __name__ == "__main__":
main()