예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
eval_vlm_vs_vlm.py
"""텍스트(GO19) VLM vs VLM headless 대국 평가.
예:
# 서버 2개(흑 9319 / 백 9320)가 떠 있는 상태에서
python -m vlm_server.eval_vlm_vs_vlm \\
--black-port 9319 --white-port 9320 \\
--until 19:00 --out-dir logs/vlm_eval_run1
종료: 연속 패스 2회, --max-moves, 또는 --until(로컬 시각 HH:MM).
"""
from __future__ import annotations
import argparse
import io
import json
import socket
import struct
import time
from collections import Counter
from dataclasses import asdict, dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from .dataset_baduk import BOARD_SIZE, _GoSim, render_board_png
from .probe_move import _recv_reply, _send_go19
KOMI = 6.5
def _quadrant(r: int, c: int) -> str:
# TL=좌상, TR=우상, BL=좌하, BR=우하(사용자 말한 4사분면)
if r < 9.5 and c < 9.5:
return "TL"
if r < 9.5 and c >= 9.5:
return "TR"
if r >= 9.5 and c < 9.5:
return "BL"
return "BR"
def _empty_png() -> bytes:
img = render_board_png(bytes(BOARD_SIZE * BOARD_SIZE), BOARD_SIZE, px=256)
buf = io.BytesIO()
img.save(buf, format="PNG")
return buf.getvalue()
@dataclass
class Endpoint:
host: str
port: int
label: str = ""
@dataclass
class MoveRec:
ply: int
side: str
row: Optional[int]
col: Optional[int]
is_pass: bool
legal: bool
ms: float
err: Optional[str] = None
quadrant: Optional[str] = None
@dataclass
class GameStats:
game_id: int
black_label: str
white_label: str
moves: List[MoveRec] = field(default_factory=list)
winner: str = "" # B/W/draw/unknown
end_reason: str = ""
black_score: float = 0.0
white_score: float = 0.0
started_at: str = ""
finished_at: str = ""
duration_s: float = 0.0
def to_dict(self) -> Dict[str, Any]:
d = asdict(self)
return d
class VlmClient:
def __init__(self, ep: Endpoint, timeout_s: float = 180.0):
self.ep = ep
self.timeout_s = timeout_s
self._sock: Optional[socket.socket] = None
self._png = _empty_png()
def close(self) -> None:
if self._sock is not None:
try:
self._sock.close()
except OSError:
pass
self._sock = None
def _ensure(self) -> socket.socket:
if self._sock is not None:
return self._sock
sock = socket.create_connection((self.ep.host, self.ep.port), timeout=15.0)
sock.settimeout(self.timeout_s)
self._sock = sock
return sock
def request_move(self, board: bytes, ai_side: int) -> Tuple[bool, Optional[int], Optional[int], Optional[str], float]:
t0 = time.perf_counter()
try:
sock = self._ensure()
_send_go19(sock, self._png, board, ai_side=ai_side)
ok, row, col, err = _recv_reply(sock)
return ok, row, col, err, (time.perf_counter() - t0) * 1000.0
except Exception as e: # noqa: BLE001 — 평가 루프는 끊겨도 재연결
self.close()
try:
sock = self._ensure()
_send_go19(sock, self._png, board, ai_side=ai_side)
ok, row, col, err = _recv_reply(sock)
return ok, row, col, err, (time.perf_counter() - t0) * 1000.0
except Exception as e2: # noqa: BLE001
return False, None, None, f"{type(e2).__name__}: {e2}", (time.perf_counter() - t0) * 1000.0
def estimate_chinese_score(sim: _GoSim, komi: float = KOMI) -> Tuple[float, float]:
"""단순 중국식 근사: 돌 + 둘러싸인 빈점. dame/사석은 미처리."""
N = sim.size
seen = bytearray(N * N)
black = 0.0
white = 0.0
for i, v in enumerate(sim.cells):
if v == 1:
black += 1
elif v == 2:
white += 1
for i in range(N * N):
if sim.cells[i] != 0 or seen[i]:
continue
stack = [i]
seen[i] = 1
region = [i]
touches = set()
while stack:
idx = stack.pop()
rr, cc = divmod(idx, N)
for dr, dc in ((-1, 0), (1, 0), (0, -1), (0, 1)):
nr, nc = rr + dr, cc + dc
if not (0 <= nr < N and 0 <= nc < N):
continue
nidx = nr * N + nc
v = sim.cells[nidx]
if v == 0:
if not seen[nidx]:
seen[nidx] = 1
stack.append(nidx)
region.append(nidx)
else:
touches.add(v)
if touches == {1}:
black += len(region)
elif touches == {2}:
white += len(region)
white += komi
return black, white
def play_one_game(
game_id: int,
black: VlmClient,
white: VlmClient,
max_moves: int,
pass_to_end: int = 2,
) -> GameStats:
sim = _GoSim(size=BOARD_SIZE)
gs = GameStats(
game_id=game_id,
black_label=black.ep.label or f"{black.ep.host}:{black.ep.port}",
white_label=white.ep.label or f"{white.ep.host}:{white.ep.port}",
started_at=datetime.now().isoformat(timespec="seconds"),
)
t0 = time.perf_counter()
to_play = 1
pass_streak = 0
ply = 0
while ply < max_moves:
client = black if to_play == 1 else white
board = bytes(sim.cells)
ok, row, col, err, ms = client.request_move(board, ai_side=to_play)
is_pass = bool(ok and row is None and col is None and err is None)
legal = False
q = None
if not ok:
# 통신 실패 → 강제 패스로 끊고 종료
gs.moves.append(
MoveRec(
ply=ply,
side="B" if to_play == 1 else "W",
row=None,
col=None,
is_pass=True,
legal=False,
ms=ms,
err=err,
)
)
gs.end_reason = f"comm_error:{err}"
break
if is_pass:
legal = True
pass_streak += 1
else:
assert row is not None and col is not None
q = _quadrant(row, col)
legal = sim.play(row, col, to_play)
if not legal:
# 서버가 불법수를 내보냈거나(드물게) 클라 규칙 불일치
pass_streak += 1
else:
pass_streak = 0
gs.moves.append(
MoveRec(
ply=ply,
side="B" if to_play == 1 else "W",
row=row,
col=col,
is_pass=is_pass or (not legal and row is not None),
legal=legal if not is_pass else True,
ms=ms,
err=None if legal or is_pass else "illegal_on_client",
quadrant=q,
)
)
ply += 1
if pass_streak >= pass_to_end:
gs.end_reason = "double_pass"
break
to_play = 2 if to_play == 1 else 1
else:
gs.end_reason = "max_moves"
b_sc, w_sc = estimate_chinese_score(sim)
gs.black_score = b_sc
gs.white_score = w_sc
if b_sc > w_sc:
gs.winner = "B"
elif w_sc > b_sc:
gs.winner = "W"
else:
gs.winner = "draw"
gs.finished_at = datetime.now().isoformat(timespec="seconds")
gs.duration_s = time.perf_counter() - t0
return gs
def _parse_until(s: str) -> datetime:
"""'19:00' 또는 ISO. 오늘 해당 시각이 이미 지났으면 내일로."""
now = datetime.now()
if "T" in s or "-" in s:
return datetime.fromisoformat(s)
hh, mm = s.strip().split(":")
target = now.replace(hour=int(hh), minute=int(mm), second=0, microsecond=0)
if target <= now:
target += timedelta(days=1)
return target
def aggregate(games: List[GameStats]) -> Dict[str, Any]:
wins = Counter(g.winner for g in games)
quads = Counter()
side_quads: Dict[str, Counter] = {"B": Counter(), "W": Counter()}
illegal = 0
passes = 0
move_ms: List[float] = []
for g in games:
for m in g.moves:
move_ms.append(m.ms)
if m.is_pass:
passes += 1
if not m.legal and not m.is_pass:
illegal += 1
if m.quadrant:
quads[m.quadrant] += 1
side_quads[m.side][m.quadrant] += 1
total_q = sum(quads.values()) or 1
return {
"game_count": len(games),
"games_count": len(games), # alias (구버전 호환)
"wins": dict(wins),
"black_wins": wins.get("B", 0),
"white_wins": wins.get("W", 0),
"draws": wins.get("draw", 0),
"quadrants": {k: quads.get(k, 0) for k in ("TL", "TR", "BL", "BR")},
"quadrant_pct": {k: round(100.0 * quads.get(k, 0) / total_q, 1) for k in ("TL", "TR", "BL", "BR")},
"quadrants_by_side": {s: dict(c) for s, c in side_quads.items()},
"illegal_moves": illegal,
"passes": passes,
"avg_move_ms": round(sum(move_ms) / len(move_ms), 1) if move_ms else 0.0,
"avg_moves_per_game": round(sum(len(g.moves) for g in games) / max(1, len(games)), 1),
"br_share_pct": round(100.0 * quads.get("BR", 0) / total_q, 1),
}
def write_summary(out_dir: Path, games: List[GameStats], meta: Dict[str, Any]) -> None:
out_dir.mkdir(parents=True, exist_ok=True)
agg = aggregate(games)
payload = {
"updated_at": datetime.now().isoformat(timespec="seconds"),
"meta": meta,
"aggregate": agg,
"recent_games": [
{
"game_id": g.game_id,
"winner": g.winner,
"end_reason": g.end_reason,
"moves": len(g.moves),
"black_score": g.black_score,
"white_score": g.white_score,
"duration_s": round(g.duration_s, 1),
}
for g in games[-20:]
],
}
(out_dir / "summary.json").write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
# 사람이 보기 쉬운 한 장
lines = [
f"# VLM vs VLM eval ({payload['updated_at']})",
"",
f"- games: **{agg['game_count']}**",
f"- wins B/W/draw: **{agg['black_wins']}/{agg['white_wins']}/{agg['draws']}**",
f"- quadrant % TL/TR/BL/BR: **{agg['quadrant_pct']}**",
f"- BR(우하) 비중: **{agg['br_share_pct']}%** (균형이면 ~25%)",
f"- illegal: {agg['illegal_moves']}, passes: {agg['passes']}",
f"- avg move: {agg['avg_move_ms']} ms, avg plies: {agg['avg_moves_per_game']}",
"",
"## meta",
"```json",
json.dumps(meta, ensure_ascii=False, indent=2),
"```",
"",
]
(out_dir / "SUMMARY.md").write_text("\n".join(lines), encoding="utf-8")
def main(argv: Optional[List[str]] = None) -> int:
p = argparse.ArgumentParser(prog="python -m vlm_server.eval_vlm_vs_vlm")
p.add_argument("--black-host", default="127.0.0.1")
p.add_argument("--black-port", type=int, default=9319)
p.add_argument("--white-host", default="127.0.0.1")
p.add_argument("--white-port", type=int, default=9320)
p.add_argument("--black-label", default="black")
p.add_argument("--white-label", default="white")
p.add_argument("--max-moves", type=int, default=240)
p.add_argument("--max-games", type=int, default=0, help="0=무제한(until까지)")
p.add_argument("--until", default=None, help="HH:MM 또는 ISO, 예: 19:00")
p.add_argument("--duration-hours", type=float, default=0.0, help="until 대신 시간 수")
p.add_argument("--out-dir", default="logs/vlm_eval")
p.add_argument("--timeout-s", type=float, default=180.0)
p.add_argument("--sleep-between-games", type=float, default=0.5)
args = p.parse_args(argv)
out_dir = Path(args.out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
games_path = out_dir / "games.jsonl"
deadline: Optional[datetime] = None
if args.until:
deadline = _parse_until(args.until)
elif args.duration_hours > 0:
deadline = datetime.now() + timedelta(hours=args.duration_hours)
else:
deadline = datetime.now() + timedelta(hours=10)
meta = {
"black": f"{args.black_label}@{args.black_host}:{args.black_port}",
"white": f"{args.white_label}@{args.white_host}:{args.white_port}",
"deadline": deadline.isoformat(timespec="seconds"),
"max_moves": args.max_moves,
"started_at": datetime.now().isoformat(timespec="seconds"),
}
(out_dir / "meta.json").write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
black = VlmClient(Endpoint(args.black_host, args.black_port, args.black_label), timeout_s=args.timeout_s)
white = VlmClient(Endpoint(args.white_host, args.white_port, args.white_label), timeout_s=args.timeout_s)
games: List[GameStats] = []
# 기존 games.jsonl 이어쓰기
if games_path.exists():
for line in games_path.read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
try:
raw = json.loads(line)
# 요약용 최소 복원
g = GameStats(
game_id=raw["game_id"],
black_label=raw.get("black_label", ""),
white_label=raw.get("white_label", ""),
winner=raw.get("winner", ""),
end_reason=raw.get("end_reason", ""),
black_score=raw.get("black_score", 0),
white_score=raw.get("white_score", 0),
duration_s=raw.get("duration_s", 0),
)
for m in raw.get("moves", []):
g.moves.append(MoveRec(**m))
games.append(g)
except Exception: # noqa: BLE001
pass
print(f"[eval] start → deadline {deadline} out={out_dir}", flush=True)
gid = (games[-1].game_id + 1) if games else 1
try:
while True:
if datetime.now() >= deadline:
print("[eval] deadline reached", flush=True)
break
if args.max_games and len(games) >= args.max_games:
print("[eval] max_games reached", flush=True)
break
print(f"[eval] game {gid} ...", flush=True)
g = play_one_game(gid, black, white, max_moves=args.max_moves)
games.append(g)
with games_path.open("a", encoding="utf-8") as f:
f.write(json.dumps(g.to_dict(), ensure_ascii=False) + "\n")
write_summary(out_dir, games, meta)
agg = aggregate(games)
print(
f"[eval] game {gid} done winner={g.winner} plies={len(g.moves)} "
f"reason={g.end_reason} BR%={agg['br_share_pct']} "
f"score B{g.black_score:.1f}/W{g.white_score:.1f} "
f"({g.duration_s:.0f}s) total_games={len(games)}",
flush=True,
)
# 진행 중에도 수마다 남기도록 mid-game 파일 갱신
(out_dir / "live_quad.json").write_text(
json.dumps(agg.get("quadrant_pct", {}), ensure_ascii=False, indent=2),
encoding="utf-8",
)
gid += 1
time.sleep(max(0.0, args.sleep_between_games))
except KeyboardInterrupt:
print("[eval] interrupted", flush=True)
finally:
black.close()
white.close()
write_summary(out_dir, games, meta)
print(f"[eval] wrote {out_dir / 'SUMMARY.md'}", flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())