예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
__main__.py
"""`python -m vlm_server` 진입점."""
from __future__ import annotations
import argparse
import sys
def _build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="python -m vlm_server",
description="19x19 바둑 VLM TCP 서버 (Gemma 4 + mlx-vlm, Apple Silicon)",
)
sub = p.add_subparsers(dest="cmd", required=True)
s = sub.add_parser("serve", help="GO19 TCP 서버 실행")
s.add_argument("--host", default="********")
s.add_argument("--port", type=int, default=9319)
s.add_argument(
"--model",
default="********",
help="HF 모델 ID 또는 로컬 경로 (mlx-vlm 호환).",
)
s.add_argument("--max-new-tokens", type=int, default=96)
s.add_argument("--temperature", type=float, default=0.2)
s.add_argument(
"--illegal-retry-max",
type=int,
default=6,
help="모델이 불법수를 낼 때 금지 좌표를 주고 재질문하는 최대 횟수(추가 시도). 0이면 retry 없음.",
)
s.add_argument(
"--illegal-retry-budget-s",
type=float,
default=55.0,
help="불법수 재질문 총 시간 예산(초). 예산을 넘기면 즉시 fallback 합법수로 전환.",
)
s.add_argument(
"--no-load",
action="store_true",
help="모델을 로드하지 않고 무작위 합법수만 반환 (디버그용).",
)
s.add_argument(
"--log-jsonl",
default=None,
help="대국/응답 로그 JSONL 파일 경로 (선택).",
)
s.add_argument(
"--lora-dir",
default=None,
help="LoRA 학습 결과 폴더 (vlm_models/lora_baduk 등). 지정 시 base 모델 위에 어댑터 적용.",
)
s.add_argument(
"--text-only-move",
action="store_true",
help="추론 시 보드 PNG를 모델에 넣지 않고, 텍스트 격자(문자/숫자)만으로 다음 수를 묻습니다.",
)
# 별칭(사용자 친화): 이미지 입력을 끈다는 의미를 더 직관적으로.
s.add_argument(
"--no-image",
action="store_true",
help="(별칭) --text-only-move 와 동일. 모델에 보드 PNG를 넣지 않습니다.",
)
s.add_argument(
"--board-repr",
choices=["chars", "numbers", "both", "compact", "aligned", "aligned_read", "aligned_read2"],
default="both",
help="프롬프트 보드: aligned=v3정렬(권장), compact=baduk_v2, chars/numbers/both=레거시.",
)
s.add_argument(
"--move-backend",
choices=["vlm", "policy", "hybrid"],
default="vlm",
help="수 선택 백엔드: vlm=기존, policy=RL정책만, hybrid=정책 우선 후 VLM fallback",
)
s.add_argument(
"--policy-checkpoint",
default=None,
help="policy 백엔드 체크포인트(npz). 없으면 policy 백엔드 사용 불가.",
)
s.add_argument("--policy-lr", type=float, default=5e-3)
s.add_argument(
"--policy-train-on-reset",
action="store_true",
help="(온라인) 게임 리셋 감지 시 에피소드 1회 REINFORCE 업데이트",
)
s.add_argument("--policy-save-every-games", type=int, default=1)
s.add_argument("--policy-reset-small-stones", type=int, default=8)
s.add_argument("--policy-reset-prev-stones", type=int, default=40)
s.add_argument(
"--policy-sample",
action="store_true",
help="policy가 greedy가 아니라 샘플링(탐색)으로 둠",
)
s.add_argument(
"--policy-avoid-fill-eye",
action="store_true",
help="policy 백엔드에서 (근사) 자기 눈 메우기 수를 마스크로 제외",
)
s.add_argument(
"--no-policy-avoid-fill-eye",
action="store_true",
help="--policy-avoid-fill-eye 비활성화",
)
s.add_argument(
"--policy-avoid-self-atari",
action="store_true",
help="policy 백엔드에서 (근사) 자기단수(self-atari) 수를 마스크로 제외",
)
s.add_argument(
"--no-policy-avoid-self-atari",
action="store_true",
help="--policy-avoid-self-atari 비활성화",
)
s.add_argument(
"--policy-avoid-fill-territory",
action="store_true",
help="policy 백엔드에서 (근사) 자기 집(둘러싸인 빈 영역) 메우기 수를 마스크로 제외",
)
s.add_argument(
"********",
action="store_true",
help="--policy-avoid-fill-territory 비활성화",
)
s.add_argument(
"--score-guided",
action="store_true",
default=True,
help="후보 수를 가상 착수 후 어림 계가(Δ집)로 고름 (기본 ON)",
)
s.add_argument(
"--no-score-guided",
action="store_true",
help="계가 유도 비활성화 (구형: VLM/랜덤 fallback)",
)
s.add_argument("--score-komi", type=float, default=6.5)
s.add_argument(
"--score-max-candidates",
type=int,
default=64,
help="계가 비교 후보 상한",
)
s.add_argument(
"--score-min-line",
type=int,
default=3,
help="따냄이 아니면 이 선 미만(1·2선) 금지",
)
s.add_argument(
"--no-ladder-guard",
action="store_true",
help="축 판독 비활성화 (잡히는 축 도망 금지·깨진 축 추격 감점 끔)",
)
d = sub.add_parser("download", help="HF 에서 Gemma 4 모델 다운로드")
d.add_argument(
"--model",
default="********",
help="HF 모델 ID (기본: mlx-community/gemma-4-e4b-it-4bit)",
)
d.add_argument(
"--cache-dir",
default=None,
help="다운로드 캐시 경로 (기본: HF 기본 캐시).",
)
t = sub.add_parser("probe", help="SGF 국면으로 한 수 테스트")
t.add_argument("--host", default="127.0.0.1")
t.add_argument("--port", type=int, default=9319)
t.add_argument("--sgf", required=True, help="국면을 가져올 SGF 파일 경로")
t.add_argument("--ply", type=int, default=-1, help="-1=최종, 0=초기, 1=첫 수 후, ...")
t.add_argument("--side", choices=["auto", "B", "W"], default="auto")
t.add_argument("--out-png", default=None, help="렌더된 보드 PNG 저장 경로(선택)")
v = sub.add_parser(
"verify-board",
help="모델이 보드 격자(19x19)를 텍스트로 올바르게 복원하는지 검증",
)
v.add_argument("--sgf", required=True)
v.add_argument("--ply", type=int, default=-1)
v.add_argument("--model", default="********")
v.add_argument("--lora-dir", default=None)
v.add_argument("--max-new-tokens", type=int, default=1024)
v.add_argument("--temperature", type=float, default=0.0)
v.add_argument(
"--with-image",
action="store_true",
help="렌더한 보드 PNG를 함께 넣어 멀티모달 검증",
)
vc = sub.add_parser(
"verify-coords",
help="좌표 grounding 검증: 격자에서 (row,col) 을 맞게 읽는지 측정",
)
vc.add_argument("--model", required=True)
vc.add_argument("--lora-dir", default=None)
vc.add_argument(
"--board-format",
choices=["aligned", "compact", "spaced"],
default="aligned",
)
vc.add_argument("--trials", type=int, default=48)
vc.add_argument("--seed", type=int, default=42)
vc.add_argument("--max-new-tokens", type=int, default=96)
vc.add_argument("--out-jsonl", default=None)
vc.add_argument("--out-summary", default=None)
b = sub.add_parser(
"serve-board-viz",
help="보드 시각화 TCP 서버 (BRD1→RTXT, 기본 포트 9321)",
)
b.add_argument("--host", default="********")
b.add_argument("--port", type=int, default=9321)
b.add_argument("--model", default="********")
b.add_argument("--max-new-tokens", type=int, default=1024)
b.add_argument("--temperature", type=float, default=0.0)
b.add_argument("--lora-dir", default=None)
b.add_argument("--infer-timeout-s", type=float, default=300.0)
b.add_argument(
"--no-image",
action="store_true",
help="이미지 입력 없이 텍스트 격자만으로 rows 복원(형식 안정성 향상)",
)
b.add_argument(
"--packed-only",
action="store_true",
help="모델을 거치지 않고 packed 보드로 rows 를 결정론적으로 생성(항상 정확)",
)
b.add_argument(
"--no-load",
action="store_true",
help="모델 없이 스텁 응답만 (프로토콜 테스트용)",
)
return p
def main(argv: list[str] | None = None) -> int:
parser = _build_parser()
args = parser.parse_args(argv)
if args.cmd == "serve":
from .server import run_server
run_server(
host=args.host,
port=args.port,
model_id=args.model,
max_new_tokens=args.max_new_tokens,
temperature=args.temperature,
load_model=not args.no_load,
log_jsonl_path=args.log_jsonl,
lora_dir=args.lora_dir,
illegal_retry_max=args.illegal_retry_max,
illegal_retry_budget_s=args.illegal_retry_budget_s,
text_only_moves=bool(args.text_only_move or args.no_image),
board_repr=args.board_repr,
move_backend=args.move_backend,
policy_checkpoint=args.policy_checkpoint,
policy_lr=args.policy_lr,
policy_train_on_reset=args.policy_train_on_reset,
policy_save_every_games=args.policy_save_every_games,
policy_reset_small_stones=args.policy_reset_small_stones,
policy_reset_prev_stones=args.policy_reset_prev_stones,
policy_sample=args.policy_sample,
policy_avoid_fill_eye=bool(args.policy_avoid_fill_eye) and not bool(args.no_policy_avoid_fill_eye),
policy_avoid_self_atari=bool(args.policy_avoid_self_atari) and not bool(args.no_policy_avoid_self_atari),
policy_avoid_fill_territory=bool(args.policy_avoid_fill_territory) and not bool(args.no_policy_avoid_fill_territory),
score_guided=bool(args.score_guided) and not bool(args.no_score_guided),
score_komi=float(args.score_komi),
score_max_candidates=int(args.score_max_candidates),
score_min_line=int(args.score_min_line),
ladder_guard=not bool(args.no_ladder_guard),
)
return 0
if args.cmd == "download":
from .download_gemma4 import download
path = download(model_id=args.model, cache_dir=args.cache_dir)
print(f"OK: {path}")
return 0
if args.cmd == "probe":
from .probe_move import main as probe_main
return int(
probe_main(
[
"--host",
args.host,
"--port",
str(args.port),
"--sgf",
args.sgf,
"--ply",
str(args.ply),
"--side",
args.side,
]
+ (["--out-png", args.out_png] if args.out_png else [])
)
)
if args.cmd == "verify-board":
from .verify_board import main as verify_main
vargs = [
"--sgf",
args.sgf,
"--ply",
str(args.ply),
"--model",
args.model,
"--max-new-tokens",
str(args.max_new_tokens),
"--temperature",
str(args.temperature),
]
if args.lora_dir:
vargs += ["--lora-dir", args.lora_dir]
if args.with_image:
vargs.append("--with-image")
return int(verify_main(vargs))
if args.cmd == "verify-coords":
from .verify_coords import main as coords_main
# subparser args already on namespace — rebuild argv for coords main
carg = [
"--model",
args.model,
"--board-format",
args.board_format,
"--trials",
str(args.trials),
"--seed",
str(args.seed),
"--max-new-tokens",
str(args.max_new_tokens),
]
if args.lora_dir:
carg += ["--lora-dir", args.lora_dir]
if args.out_jsonl:
carg += ["--out-jsonl", args.out_jsonl]
if args.out_summary:
carg += ["--out-summary", args.out_summary]
return int(coords_main(carg))
if args.cmd == "serve-board-viz":
from .board_viz_server import run_board_viz_server
run_board_viz_server(
host=args.host,
port=args.port,
model_id=args.model,
max_new_tokens=args.max_new_tokens,
temperature=args.temperature,
load_model=not args.no_load,
lora_dir=args.lora_dir,
infer_timeout_s=args.infer_timeout_s,
with_image=not args.no_image,
packed_only=args.packed_only,
)
return 0
parser.print_help()
return 2
if __name__ == "__main__":
sys.exit(main())