LLM を戦略コーチにする

Bot を書き進めていくと、運営層のルールが 1 枚ずつ貼り足されていくことに気づきます。伐採人数のルール、1 鉱山 5 人のルール、木材が余ったら半分にするルール、ゴールドが多く木材が少なければ伐採に多めに回すルール……。どのルールも単独では正しいのに、組み合わさると「鉱山は人手不足なのに、農民が全員木を切っている」といった、どのルールも責任を持たない状況が生まれます。

このような「全体を見て優先順位をつける」判断は、もともと if / else で書くのに向いていませんが、LLM はまさにこれが得意です。リファレンスブレイン（brains/xwar3/strategy/brain/coach.py）は、以下の階層構成を採用しています。

階層構成

LLM（アドバイザー）             20 ゲーム秒に 1 回、非同期。どのティックもブロックしない
  入力：局面スナップショット 1 ページ分（資源、人口、農民の配置、鉱山、兵種、技術、ヒーロー、敵情報、直近の出来事）
  出力：厳密な JSON —— 一言の診断 + ワーカー配分 + 優先して作るもの + この 1 分の方針 + やってはいけないこと
        │
        ▼  ホワイトリスト + 上下限クランプ + 拒否
ルール層（Bot、毎ティック）     助言を既存機能への「バイアス」に変換：ワーカー配分、建設 / 訓練の優先度、攻撃方針
        │
        ▼
実行層（SDK / リフレックス層）  命令を出す、レシートを読む、マイクロ操作

出力契約

モデルには固定フィールドの JSON だけを出力させ、フィールドの追加・削除は許しません。

{
  "diagnosis": "一文で：局面で最大の問題。入力データの中に根拠がなければならない",
  "workers":   { "gold": 10, "lumber": 6 },
  "priority":  ["hpea", "hhou", "hbar"],
  "posture":   "creep",
  "avoid":     ["木材が足りないうちに Iron Plating を先に研究しない"]
}

フィールド	ルール層での使い方	リファレンスブレインのクランプ
workers	ゴールド採掘・伐採の目標人数	採掘 2 ~ 25、伐採 1 ~ 20。合計は農民の総数を超えない
priority	訓練 / 建設 / 研究の優先順	最大 4 個。「選択可能コード」表に出てくる 4 文字コードのみ受け付ける
posture	この 1 分の方針	attack defend creep expand recover hold のいずれか
avoid	この 1 分にやってはいけないこと	最大 2 件
diagnosis	ログとコンソール表示にのみ使用	—

プロンプトは種族ごとに 1 つずつ用意し、その種族特有のトレードオフだけを書きます（ヒューマンの共同建設と Militia、オークの Burrow、アンデッドの Haunted Gold Mine、ナイトエルフの Entangled Gold Mine……）。共通ルールは共通部分にまとめ、4 回コピーしないでください。

4 つの厳格な制約

この 4 つは、いずれもリファレンスブレインが実際に痛い目を見て学んだものです。

1. アドバイザーは決してユニットに直接命令しません。 150 ms 単位の現場は見えませんし、幻覚も起こします。変えるのは目標と優先度だけで、具体的に誰がどこへ行き、誰を攻撃するかは引き続きルール層とリフレックス層が決めます —— 指揮権の持ち主は 1 人だけです。
2. 非同期。 アドバイザーへの問い合わせは 1 回約 1 秒かかります。バックグラウンドスレッドで実行して最新の結果を採用し、どのティックもブロックしません。モデルが起動していない、タイムアウトした、でたらめな回答を返した、という場合はこの層がないものとして扱い、純粋なルールの挙動に戻ります。古すぎる助言（3 間隔を超えたもの）も使いません。
3. ホワイトリスト + クランプ。 各フィールドは既存の機能に対応付けられなければならず、数値は妥当な範囲にクランプします。認識できない内容は、黙って無視するのではなくカウントしてから破棄します。
4. すべてをカウントする。 何回問い合わせ、何回成功し、何回タイムアウトし、何回クランプされ、各フィールドが何回採用されたかを、最後にモデルへ送った入力と合わせて公開します。そうしないと「この層は本当に役に立っているのか」という問いに答えられません。

自分の Bot に実装する

以下は最小限のスケルトンです。OpenAI 互換の API であれば何でも使え（LM Studio、Ollama、クラウド API など）、標準ライブラリだけで動きます。

import collections, json, threading, urllib.request
from openwar3 import Bot

BASE = "http://127.0.0.1:1234/v1"            # LM Studio / Ollama / 任意の OpenAI 互換サービス
MODEL = "your-model"
POSTURES = {"attack", "defend", "creep", "expand", "recover", "hold"}
SYSTEM = """あなたは Warcraft III の運営コーチです。担当は運営と戦略だけで、マイクロ操作は扱いません。
JSON だけを出力し、フィールドは固定です：{"diagnosis": 一文, "workers": {"gold": 整数, "lumber": 整数},
"priority": [4 文字コード, 最大 4 個, allowed に含まれるもののみ], "posture": 6 つから 1 つ, "avoid": [最大 2 件]}
与えられた局面データだけに基づいて答え、データにないことは作らないでください。"""


def ask(state: dict) -> dict:
    body = {"model": MODEL, "temperature": 0.3, "max_tokens": 260,
            "messages": [{"role": "system", "content": SYSTEM},
                         {"role": "user", "content": json.dumps(state, ensure_ascii=False)}]}
    req = urllib.request.Request(f"{BASE}/chat/completions", json.dumps(body).encode(),
                                 {"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=8) as r:
        text = json.load(r)["choices"][0]["message"]["content"]
    return json.loads(text[text.index("{"): text.rindex("}") + 1])


class CoachedBot(Bot):
    EVERY = 20.0                              # ゲーム秒：運営判断の時間スケールは分単位なので、毎ティック聞く必要はない
    allowed = {"hpea", "hfoo", "hrif", "hkni", "hhou", "hbar", "hbla", "Rhme", "Rhar"}

    def on_start(self, g):
        self.plan, self.plan_at, self.asked_at, self.busy = {}, -1e9, -1e9, False
        self.stats = collections.Counter()

    def summary(self, g) -> dict:             # スナップショットはメインスレッドで読んでおき、バックグラウンドスレッドは g に触らない
        res = g.resources() or {}
        return {"clock": round(g.clock() or 0), "gold": res.get("gold"), "lumber": res.get("lumber"),
                "food": [res.get("food_used"), res.get("food_cap")],
                "workers": len(g.my_workers()), "idle_workers": len(g.idle_workers()),
                "army": collections.Counter(u.type for u in g.my_army()),
                "enemy_seen": collections.Counter(u.type for u, _t, _age in g.last_seen(max_age=90)),
                "night": g.is_night(), "allowed": sorted(self.allowed)}

    def consult(self, state, now):
        try:
            p = ask(state)
            self.stats["ok"] += 1
            posture = p.get("posture")
            if posture not in POSTURES:
                self.stats["bad_posture"] += 1                       # カウントしてから破棄。黙って捨てない
                posture = "hold"
            self.plan = {"gold": min(25, max(2, int(p["workers"]["gold"]))),      # クランプ
                         "lumber": min(20, max(1, int(p["workers"]["lumber"]))),
                         "priority": [c for c in p.get("priority", []) if c in self.allowed][:4],
                         "posture": posture}
            self.plan_at = now
        except Exception as e:                                       # タイムアウト / でたらめな回答：この層はないものとして扱う
            self.stats[f"error:{type(e).__name__}"] += 1
        finally:
            self.busy = False

    def on_tick(self, g):
        now = g.clock() or 0.0
        if not self.busy and now - self.asked_at >= self.EVERY:
            self.busy, self.asked_at = True, now
            threading.Thread(target=self.consult, args=(self.summary(g), now), daemon=True).start()
        plan = self.plan if now - self.plan_at <= 3 * self.EVERY else {}   # 古すぎる助言は使わない
        # ↓ ルール層：plan が空ならデフォルトのルールで動く。plan があれば配分・優先度・方針だけを調整し、具体的な命令は引き続きルールが決める
        ...

モデルの選び方

状況	おすすめ
ローカルで速さ重視	MoE モデル（1 回にごく一部のパラメータしか使わない）は、同サイズの密なモデルよりずっと速く動きます。リファレンスブレインは Qwen3.6-35B-A3B（LM Studio、Q4）を使っており、中央値 1.09 秒、最遅 1.45 秒、5/5 の出力がそのまま json.loads できます
ローカルの「思考」するモデル	思考部分を必ずオフにしてください。そうしないとトークンがすべて思考に使われ、JSON が 1 つも出てきません。LM Studio は /no_think を無視するため、リファレンスブレインは /v1/completions に切り替え、ChatML を自前で組み立てて、空の <think></think> と { を 1 つあらかじめ埋めています
クラウドモデル	通常はレイテンシが高めですが、この階層構成はもともと非同期です。運営判断は分単位なので、数秒の遅延は許容できます