예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
mariago_ensemble_proxy.py
#!/usr/bin/env python3
"""MariaGO 앙상블 프록시 — 여러 모델이 다음 수를 제안하고, 판정자가 하나를 고른다.
브리지 ──GO19──▶ 프록시(:9318) ──GO19──▶ 제안 모델들(:9319, :9329, …) 병렬
└──▶ 판정자(katago | laya | vote) ──▶ RPLY
.venv/bin/python scripts/mariago_ensemble_proxy.py \\
--proposer r5=127.0.0.1:9319 --proposer coord=127.0.0.1:9329 --judge katago
브리지를 프록시로 보내려면 MARIAGO_MOVE_PORT=9318 (또는 브리지 --vlm-port 9318).
판정자
katago 제안된 후보로만 탐색을 제한(allowMoves)해 KataGo가 최선 후보를 고름 (기본)
laya Jev 대체 공개 모델(Laya-MLX)에 후보 선택을 묻는다. 바둑 지식은 약함 — 실험용
vote 다수결, 동률이면 --proposer 순서
판정 실패 시 vote로 떨어진다.
"""
from __future__ import annotations
import argparse
import json
import logging
import os
import queue
import socket
import socketserver
import struct
import subprocess
import sys
import threading
import time
import zlib
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Optional, Union
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from scripts.katago_analyze import katago_paths # noqa: E402
from vlm_server.server import _MiniGo # noqa: E402
MAGIC_GO19 = b"GO19"
MAGIC_RPLY = b"RPLY"
MAGIC_PING = b"PING"
MAGIC_PONG = b"PONG"
N = 19
LETTERS = "ABCDEFGHJKLMNOPQRST"
LOG = logging.getLogger("ensemble")
Point = tuple[int, int]
MoveKey = Union[Point, str] # (r, c) 또는 "pass"
def gtp(key: MoveKey) -> str:
if key == "pass":
return "pass"
r, c = key # type: ignore[misc]
return f"{LETTERS[c]}{N - r}"
def from_gtp(s: str) -> MoveKey:
s = s.strip().upper()
if s == "PASS":
return "pass"
return (N - int(s[1:]), LETTERS.index(s[0]))
# ---------- GO19 프레임 ----------
def _recv_exact(sock: socket.socket, n: int) -> bytes:
buf = bytearray()
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
raise EOFError("peer closed")
buf.extend(chunk)
return bytes(buf)
@dataclass
class MoveRequest:
ai_side: int
png: bytes
board: bytes
epoch: Optional[int]
rejects: list[Point]
def read_request(sock: socket.socket) -> Optional[MoveRequest]:
"""PING 이면 PONG 을 보내고 None."""
head = _recv_exact(sock, 4)
if head == MAGIC_PING:
sock.sendall(MAGIC_PONG)
return None
if head != MAGIC_GO19:
raise ValueError(f"unknown header {head!r}")
(body_len,) = struct.unpack(">I", _recv_exact(sock, 4))
if body_len > 32 * 1024 * 1024:
raise ValueError("body too large")
body = _recv_exact(sock, body_len)
if body[0] != N or body[1] not in (1, 2):
raise ValueError(f"bad size/side {body[0]}/{body[1]}")
ai_side = body[1]
o = 2
(png_len,) = struct.unpack(">I", body[o : o + 4])
o += 4
png = body[o : o + png_len]
o += png_len
board = body[o : o + N * N]
o += N * N
if len(board) != N * N:
raise ValueError("board length mismatch")
epoch: Optional[int] = None
if len(body) >= o + 4:
(epoch,) = struct.unpack(">I", body[o : o + 4])
o += 4
rejects: list[Point] = []
if len(body) > o:
n = body[o]
o += 1
for _ in range(n):
rejects.append((body[o], body[o + 1]))
o += 2
return MoveRequest(ai_side, bytes(png), bytes(board), epoch, rejects)
def send_request(
sock: socket.socket, req: MoveRequest, epoch: int
) -> None:
body = bytearray([N, req.ai_side])
body += struct.pack(">I", len(req.png)) + req.png + req.board
body += struct.pack(">I", epoch & 0xFFFFFFFF)
rej = req.rejects[-64:]
if rej:
body.append(len(rej))
for r, c in rej:
body += bytes([r & 0xFF, c & 0xFF])
sock.sendall(MAGIC_GO19 + struct.pack(">I", len(body)) + body)
def recv_reply(sock: socket.socket) -> tuple[bool, Optional[MoveKey], Optional[str]]:
if _recv_exact(sock, 4) != MAGIC_RPLY:
raise ValueError("bad reply header")
flag = _recv_exact(sock, 1)[0]
if flag == 1:
r, c = _recv_exact(sock, 2)
return True, ("pass" if (r, c) == (0xFF, 0xFF) else (r, c)), None
(n,) = struct.unpack(">H", _recv_exact(sock, 2))
return False, None, _recv_exact(sock, n).decode("utf-8", "replace")
def send_move(sock: socket.socket, key: MoveKey) -> None:
r, c = (0xFF, 0xFF) if key == "pass" else key # type: ignore[misc]
sock.sendall(MAGIC_RPLY + bytes([1, r, c]))
def send_error(sock: socket.socket, msg: str) -> None:
raw = msg.encode("utf-8")[:65535]
sock.sendall(MAGIC_RPLY + b"\x00" + struct.pack(">H", len(raw)) + raw)
def turn_epoch(req: MoveRequest) -> int:
"""제안 모델에 보낼 epoch — 국면마다 달라진다.
VLM 서버는 직전 제안 칸이 비어 있으면 '클라이언트 거부'로 보고 금지하며, 이 상태는 epoch 가
바뀔 때만 리셋된다. 앙상블에서 채택되지 않은 제안이 계속 금지되지 않도록 국면마다 epoch 를
바꾸고, 같은 국면의 재요청(rejects 추가)에는 같은 값을 준다.
"""
seed = struct.pack(">IB", (req.epoch or 1) & 0xFFFFFFFF, req.ai_side) + req.board
return zlib.crc32(seed) & 0xFFFFFFFF
# ---------- 제안 모델 ----------
@dataclass
class Proposer:
label: str
host: str
port: int
@dataclass
class Proposal:
label: str
move: Optional[MoveKey] = None
error: Optional[str] = None
ms: float = 0.0
def ask_proposer(p: Proposer, req: MoveRequest, epoch: int, timeout_s: float) -> Proposal:
t0 = time.perf_counter()
try:
with socket.create_connection((p.host, p.port), timeout=10) as s:
s.settimeout(timeout_s)
send_request(s, req, epoch)
ok, move, err = recv_reply(s)
out = Proposal(p.label, move=move if ok else None, error=None if ok else err)
except Exception as e: # noqa: BLE001
out = Proposal(p.label, error=f"{type(e).__name__}: {e}")
out.ms = (time.perf_counter() - t0) * 1000
return out
# ---------- 판정자 ----------
@dataclass
class Candidate:
key: MoveKey
supporters: list[str] = field(default_factory=list)
@dataclass
class Decision:
key: MoveKey
judge: str
detail: dict = field(default_factory=dict)
class VoteJudge:
name = "vote"
def pick(self, board: bytes, ai_side: int, cands: list[Candidate]) -> Decision:
best = max(cands, key=lambda c: len(c.supporters))
return Decision(best.key, self.name, {c_gtp(c): len(c.supporters) for c in cands})
def c_gtp(c: Candidate) -> str:
return gtp(c.key)
class KataGoJudge:
"""상주 KataGo analysis 엔진. 후보 수로만 루트 탐색을 제한해 최선을 고른다."""
name = "katago"
def __init__(self, visits: int, komi: float, timeout_s: float):
self.visits = visits
self.komi = komi
self.timeout_s = timeout_s
self._lock = threading.Lock()
self._seq = 0
self._proc: Optional[subprocess.Popen] = None
self._results: "queue.Queue[dict]" = queue.Queue()
self._start()
def _start(self) -> None:
exe, model, cfg = katago_paths()
if not exe.exists():
raise FileNotFoundError(f"KataGo not found: {exe}")
home = ROOT / "logs/katago_analysis"
home.mkdir(parents=True, exist_ok=True)
env = os.environ.copy()
env["HOME"] = str(home)
err_log = open(ROOT / "logs/ensemble/katago.stderr.log", "a", encoding="utf-8")
self._proc = subprocess.Popen(
[str(exe), "analysis", "-config", str(cfg), "-model", str(model)],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=err_log,
text=True,
bufsize=1,
env=env,
)
self._results = queue.Queue()
threading.Thread(target=self._read_loop, args=(self._proc, self._results), daemon=True).start()
LOG.info("KataGo analysis engine started (pid=%s)", self._proc.pid)
@staticmethod
def _read_loop(proc: subprocess.Popen, out: "queue.Queue[dict]") -> None:
assert proc.stdout
for line in proc.stdout:
line = line.strip()
if not line.startswith("{"):
continue
try:
out.put(json.loads(line))
except json.JSONDecodeError:
continue
def _query(self, payload: dict) -> dict:
with self._lock:
if self._proc is None or self._proc.poll() is not None:
LOG.warning("KataGo engine not running — restarting")
self._start()
self._seq += 1
qid = f"ens{self._seq}"
payload["id"] = qid
assert self._proc and self._proc.stdin
self._proc.stdin.write(json.dumps(payload) + "\n")
self._proc.stdin.flush()
deadline = time.monotonic() + self.timeout_s
while True:
left = deadline - time.monotonic()
if left <= 0:
raise TimeoutError(f"KataGo timeout {self.timeout_s}s")
try:
msg = self._results.get(timeout=left)
except queue.Empty:
continue
if msg.get("id") != qid:
continue
if msg.get("error"):
raise RuntimeError(msg["error"])
if msg.get("isDuringSearch"):
continue
if "moveInfos" in msg:
return msg
def pick(self, board: bytes, ai_side: int, cands: list[Candidate]) -> Decision:
player = "B" if ai_side == 1 else "W"
stones = [
["B" if v == 1 else "W", gtp((i // N, i % N))] for i, v in enumerate(board) if v in (1, 2)
]
allowed = [gtp(c.key) for c in cands]
res = self._query(
{
"initialStones": stones,
"initialPlayer": player,
"moves": [],
"rules": "korean",
"komi": self.komi,
"boardXSize": N,
"boardYSize": N,
"analyzeTurns": [0],
"maxVisits": self.visits,
"allowMoves": [{"player": player, "moves": allowed, "untilDepth": 1}],
}
)
infos = sorted(res.get("moveInfos") or [], key=lambda m: m.get("order", 99))
if not infos:
raise RuntimeError("KataGo returned no moveInfos")
# analysis_example.cfg: reportAnalysisWinratesAs = BLACK
flip = -1.0 if ai_side == 2 else 1.0
scores = {
m["move"]: {
"visits": m.get("visits"),
"winrate": round(m["winrate"] if ai_side == 1 else 1 - m["winrate"], 4),
"scoreLead": round(flip * m.get("scoreLead", 0.0), 2),
}
for m in infos
}
return Decision(from_gtp(infos[0]["move"]), self.name, scores)
class LayaJudge:
"""Jev 대체 공개 결정 모델(Laya-MLX)을 별도 venv 워커로 띄워 후보 중 하나를 고르게 한다."""
name = "laya"
def __init__(self, python: Path, worker: Path, timeout_s: float, load_timeout_s: float = 180):
if not python.exists():
raise FileNotFoundError(f"laya python not found: {python} (experiments/jev/.venv 설치 필요)")
self.timeout_s = timeout_s
self._lock = threading.Lock()
err_log = open(ROOT / "logs/ensemble/laya.stderr.log", "a", encoding="utf-8")
self._proc = subprocess.Popen(
[str(python), str(worker)],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=err_log,
text=True,
bufsize=1,
cwd=str(worker.parent),
)
self._out: "queue.Queue[dict]" = queue.Queue()
threading.Thread(target=KataGoJudge._read_loop, args=(self._proc, self._out), daemon=True).start()
ready = self._out.get(timeout=load_timeout_s)
if not ready.get("ready"):
raise RuntimeError(f"laya worker failed: {ready}")
LOG.info("Laya judge ready (pid=%s)", self._proc.pid)
@staticmethod
def describe(board: bytes, ai_side: int, cands: list[Candidate]) -> str:
blacks = [gtp((i // N, i % N)) for i, v in enumerate(board) if v == 1]
whites = [gtp((i // N, i % N)) for i, v in enumerate(board) if v == 2]
side = "Black" if ai_side == 1 else "White"
props = "; ".join(f"{gtp(c.key)} (suggested by {', '.join(c.supporters)})" for c in cands)
return (
f"Game of Go (baduk), 19x19 board, Korean rules, komi 6.5. {side} to play. "
f"Black stones ({len(blacks)}): {' '.join(blacks) or 'none'}. "
f"White stones ({len(whites)}): {' '.join(whites) or 'none'}. "
f"Candidate moves from AI models: {props}."
)
def pick(self, board: bytes, ai_side: int, cands: list[Candidate]) -> Decision:
side = "Black" if ai_side == 1 else "White"
criteria = [gtp(c.key) for c in cands]
req = {
"state": self.describe(board, ai_side, cands),
"instructions": f"Which candidate is the strongest next move for {side}?",
"criteria": criteria,
}
with self._lock:
if self._proc.poll() is not None:
raise RuntimeError("laya worker exited")
assert self._proc.stdin
self._proc.stdin.write(json.dumps(req) + "\n")
self._proc.stdin.flush()
res = self._out.get(timeout=self.timeout_s)
if not res.get("ok"):
raise RuntimeError(res.get("error") or "laya error")
return Decision(
from_gtp(res["choice"]),
self.name,
{"probabilities": res.get("probabilities"), "confidence": res.get("confidence")},
)
# ---------- 서버 ----------
class EnsembleServer(socketserver.ThreadingTCPServer):
allow_reuse_address = True
daemon_threads = True
def __init__(self, addr, args: argparse.Namespace, judge, proposers: list[Proposer]):
super().__init__(addr, EnsembleHandler)
self.args = args
self.judge = judge
self.fallback = VoteJudge()
self.proposers = proposers
self.pool = ThreadPoolExecutor(max_workers=max(4, len(proposers) * 2))
self.log_path = Path(args.log_jsonl)
self.log_path.parent.mkdir(parents=True, exist_ok=True)
self._log_lock = threading.Lock()
def write_log(self, rec: dict) -> None:
with self._log_lock, self.log_path.open("a", encoding="utf-8") as f:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
class EnsembleHandler(socketserver.BaseRequestHandler):
server: EnsembleServer # type: ignore[assignment]
def handle(self) -> None:
sock: socket.socket = self.request
while True:
try:
req = read_request(sock)
except EOFError:
return
except Exception as e: # noqa: BLE001
LOG.warning("bad frame: %s", e)
try:
send_error(sock, f"bad frame: {e}")
except OSError:
pass
return
if req is None:
continue
self.handle_move(sock, req)
def handle_move(self, sock: socket.socket, req: MoveRequest) -> None:
srv = self.server
t0 = time.perf_counter()
epoch = turn_epoch(req)
futures = [
srv.pool.submit(ask_proposer, p, req, epoch, srv.args.proposer_timeout_s)
for p in srv.proposers
]
proposals = [f.result() for f in futures]
go = _MiniGo(req.board)
rejected = set(req.rejects)
cands: dict[MoveKey, Candidate] = {}
dropped: dict[str, str] = {}
for pr in proposals:
if pr.move is None:
continue
if pr.move != "pass":
reason = "client_reject" if pr.move in rejected else go.illegal_reason(*pr.move, req.ai_side)
if reason:
dropped[pr.label] = f"{gtp(pr.move)}:{reason}"
continue
cands.setdefault(pr.move, Candidate(pr.move)).supporters.append(pr.label)
rec: dict = {
"ts": datetime.now().isoformat(timespec="seconds"),
"side": "B" if req.ai_side == 1 else "W",
"stones": sum(1 for v in req.board if v),
"proposals": {
pr.label: {"move": gtp(pr.move) if pr.move else None, "error": pr.error, "ms": round(pr.ms)}
for pr in proposals
},
"dropped": dropped,
}
if not cands:
errs = "; ".join(f"{pr.label}: {pr.error or 'no legal move'}" for pr in proposals)
rec["error"] = errs
srv.write_log(rec)
LOG.warning("no candidates — %s", errs)
send_error(sock, f"ensemble: 모든 제안 모델 실패 ({errs})")
return
cand_list = list(cands.values())
tj = time.perf_counter()
if len(cand_list) == 1:
decision = Decision(cand_list[0].key, "unanimous")
else:
try:
decision = srv.judge.pick(req.board, req.ai_side, cand_list)
if decision.key not in cands:
raise RuntimeError(f"judge chose non-candidate {gtp(decision.key)}")
except Exception as e: # noqa: BLE001
LOG.warning("judge %s failed (%s) → vote", srv.judge.name, e)
decision = srv.fallback.pick(req.board, req.ai_side, cand_list)
decision.detail["judge_error"] = str(e)
rec.update(
{
"candidates": {gtp(c.key): c.supporters for c in cand_list},
"chosen": gtp(decision.key),
"judge": decision.judge,
"judge_detail": decision.detail,
"judge_ms": round((time.perf_counter() - tj) * 1000),
"total_ms": round((time.perf_counter() - t0) * 1000),
}
)
srv.write_log(rec)
LOG.info(
"%s to play · %s → %s by %s (%dms)",
rec["side"],
", ".join(f"{k}:{'/'.join(v)}" for k, v in rec["candidates"].items()),
rec["chosen"],
decision.judge,
rec["total_ms"],
)
send_move(sock, decision.key)
def parse_proposer(spec: str) -> Proposer:
label, _, addr = spec.partition("=")
host, _, port = addr.rpartition(":")
if not label or not port:
raise argparse.ArgumentTypeError(f"--proposer label=host:port 형식이어야 함: {spec}")
return Proposer(label, host or "127.0.0.1", int(port))
def build_judge(args: argparse.Namespace):
if args.judge == "katago":
return KataGoJudge(args.katago_visits, args.komi, args.judge_timeout_s)
if args.judge == "laya":
return LayaJudge(Path(args.laya_python), Path(args.laya_worker), args.judge_timeout_s)
return VoteJudge()
def main() -> int:
p = argparse.ArgumentParser(description="MariaGO multi-model ensemble proxy (GO19)")
p.add_argument("--listen-host", default="127.0.0.1")
p.add_argument("--listen-port", type=int, default=9318)
p.add_argument(
"--proposer",
action="append",
type=parse_proposer,
help="label=host:port (반복 가능, 순서 = 다수결 동률 우선순위). 기본 r5=127.0.0.1:9319",
)
p.add_argument("--proposer-timeout-s", type=float, default=200.0)
p.add_argument("--judge", choices=["katago", "laya", "vote"], default="katago")
p.add_argument("--judge-timeout-s", type=float, default=30.0)
p.add_argument("--katago-visits", type=int, default=200)
p.add_argument("--komi", type=float, default=6.5)
p.add_argument("--laya-python", default=str(ROOT / "experiments/jev/.venv/bin/python"))
p.add_argument("--laya-worker", default=str(ROOT / "experiments/jev/judge_worker.py"))
p.add_argument("--log-jsonl", default=str(ROOT / "logs/ensemble/decisions.jsonl"))
args = p.parse_args()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
(ROOT / "logs/ensemble").mkdir(parents=True, exist_ok=True)
proposers = args.proposer or [Proposer("r5", "127.0.0.1", 9319)]
judge = build_judge(args)
LOG.info(
"listening %s:%d · proposers=%s · judge=%s",
args.listen_host,
args.listen_port,
", ".join(f"{x.label}@{x.host}:{x.port}" for x in proposers),
judge.name,
)
with EnsembleServer((args.listen_host, args.listen_port), args, judge, proposers) as srv:
try:
srv.serve_forever()
except KeyboardInterrupt:
pass
return 0
if __name__ == "__main__":
raise SystemExit(main())