예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
daemon_text_client_harden.py
#!/usr/bin/env python3
"""텍스트 클라이언트 강화 파이프라인.
VLM vs VLM(단일 서버 self-play)을 계속 돌리며:
- games.jsonl / errors.jsonl / STATUS.md 기록
- reject_loop / sticky_offer / retry_exhausted / comm_error 를 모아 CORE 이슈로 남김
예:
.venv/bin/python scripts/daemon_text_client_harden.py --foreground \\
--host 127.0.0.1 --port 9319 --max-games 0
"""
from __future__ import annotations
import argparse
import json
import sys
import time
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from vlm_server.text_client import Endpoint, TextVlmClient, play_one_game # noqa: E402
from vlm_server.text_client_core import CoreConfig # noqa: E402
ERROR_ENDS = {
"reject_loop",
"sticky_offer",
"retry_exhausted",
"comm_error",
}
def _iso() -> str:
return datetime.now(timezone.utc).isoformat()
def write_status(path: Path, **kw) -> None:
lines = [f"# Text client harden — {_iso()}", ""]
for k, v in kw.items():
lines.append(f"- **{k}**: `{v}`")
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
def main() -> int:
p = argparse.ArgumentParser()
p.add_argument("--host", default="127.0.0.1")
p.add_argument("--port", type=int, default=9319)
p.add_argument("--white-port", type=int, default=None)
p.add_argument("--out-dir", default="logs/text_client_harden")
p.add_argument("--max-games", type=int, default=0, help="0=무한")
p.add_argument("--max-moves", type=int, default=360)
p.add_argument("--timeout-s", type=float, default=240.0)
p.add_argument("--sleep-s", type=float, default=0.5)
p.add_argument("--foreground", action="store_true")
p.add_argument("--max-reject-retries", type=int, default=8)
args = p.parse_args()
out = Path(args.out_dir)
if not out.is_absolute():
out = ROOT / out
out.mkdir(parents=True, exist_ok=True)
games_path = out / "games.jsonl"
errors_path = out / "errors.jsonl"
status_path = out / "STATUS.md"
issues_path = out / "CORE_ISSUES.md"
cfg = CoreConfig(max_reject_retries=args.max_reject_retries)
bep = Endpoint(args.host, args.port, "B")
wep = Endpoint(args.host, args.white_port or args.port, "W")
black = TextVlmClient(bep, timeout_s=args.timeout_s)
white = TextVlmClient(wep, timeout_s=args.timeout_s)
end_counts: Counter = Counter()
err_counts: Counter = Counter()
# 재시작 시 games.jsonl 집계를 이어받아 국수/통계 연속성 유지
prior_games = 0
if games_path.exists():
with open(games_path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
prior_games += 1
try:
rec = json.loads(line)
except json.JSONDecodeError:
continue
er = rec.get("end_reason") or "?"
end_counts[er] += 1
if er in ERROR_ENDS or rec.get("error"):
err_counts[er] += 1
gid = prior_games
print(
f"[harden] start out={out} endpoint={bep.host}:{bep.port} "
f"resume_from_game={gid}",
flush=True,
)
write_status(
status_path,
state="running",
games=gid,
resumed_from=prior_games,
endpoint=f"{bep.host}:{bep.port}",
)
try:
while True:
gid += 1
if args.max_games > 0 and (gid - prior_games) > args.max_games:
break
print(f"[harden] game {gid} …", flush=True)
t0 = time.time()
res = play_one_game(
game_id=gid,
black=black,
white=white,
max_moves=args.max_moves,
core_cfg=cfg,
)
dt = time.time() - t0
end_counts[res.end_reason] += 1
n_rej = sum(1 for m in res.moves if m.reject)
print(
f"[harden] game {gid} done in {dt:.0f}s end={res.end_reason} "
f"plies={res.plies} rejects={n_rej} err={res.error or '-'}",
flush=True,
)
rec = {
"ts": _iso(),
"game_id": res.game_id,
"end_reason": res.end_reason,
"error": res.error,
"plies": res.plies,
"n_events": len(res.moves),
"rejects": dict(Counter(m.reject for m in res.moves if m.reject)),
}
# compact moves for errors only
with open(games_path, "a", encoding="utf-8") as f:
f.write(
json.dumps(
{
**rec,
"moves": [
{
"ply": m.ply,
"side": m.side,
"row": m.row,
"col": m.col,
"pass": m.is_pass,
"ok": m.accepted,
"reject": m.reject,
"ms": round(m.ms, 1),
"retries": m.retries,
}
for m in res.moves
],
},
ensure_ascii=False,
)
+ "\n"
)
is_err = res.end_reason in ERROR_ENDS or bool(res.error)
if is_err:
err_counts[res.end_reason] += 1
# last few rejects for diagnosis
tail = [m for m in res.moves if m.reject][-12:]
with open(errors_path, "a", encoding="utf-8") as f:
f.write(
json.dumps(
{
"ts": _iso(),
"game_id": gid,
"end_reason": res.end_reason,
"error": res.error,
"plies": res.plies,
"reject_tail": [
{
"ply": m.ply,
"side": m.side,
"rc": [m.row, m.col],
"reject": m.reject,
"retries": m.retries,
}
for m in tail
],
},
ensure_ascii=False,
)
+ "\n"
)
print(
f"[harden] ERROR game={gid} {res.end_reason} {res.error} plies={res.plies}",
flush=True,
)
else:
print(
f"[harden] ok game={gid} {res.end_reason} plies={res.plies}",
flush=True,
)
write_status(
status_path,
state="running",
games=gid,
end_counts=dict(end_counts),
error_counts=dict(err_counts),
last_end=res.end_reason,
last_error=res.error or "",
)
# CORE_ISSUES 요약
issues = [
f"# CORE_ISSUES — {_iso()}",
"",
"텍스트 클라이언트가 잡은 종료/오류 집계. 코어 수정 시 이 파일을 기준으로 우선순위를 정한다.",
"",
"## end_reason",
]
for k, v in end_counts.most_common():
issues.append(f"- `{k}`: {v}")
issues.append("")
issues.append("## errors")
for k, v in err_counts.most_common():
issues.append(f"- `{k}`: {v}")
issues.append("")
issues.append(f"상세: `{errors_path.name}` / `{games_path.name}`")
issues_path.write_text("\n".join(issues) + "\n", encoding="utf-8")
time.sleep(max(0.0, args.sleep_s))
except KeyboardInterrupt:
print("[harden] interrupted", flush=True)
finally:
black.close()
white.close()
write_status(
status_path,
state="stopped",
games=gid,
end_counts=dict(end_counts),
error_counts=dict(err_counts),
)
return 0
if __name__ == "__main__":
raise SystemExit(main())