# LLM как стратегический советник

> Отдайте LLM вопросы «на что копить, куда ставить рабочих, атаковать или выжидать в эту минуту», а слою правил оставьте только исполнение и право вето. Эталонный мозг уже работает так; на этой странице — сам паттерн и подводные камни.

Источник: https://war3ai.com/ru/docs/llm-coach/

Когда бот дорастает до определённого размера, вы замечаете, что правила экономики наслаиваются одно на другое: правило про число лесорубов, правило «5 рабочих на рудник», правило «урезать добычу древесины вдвое, если её слишком много», правило «отправить больше людей на золото, если золота мало, а древесины много»… Каждое правило по отдельности верно, но вместе они порождают ситуации вроде «на руднике не хватает рабочих, а все крестьяне рубят лес» — ситуации, **за которые не отвечает ни одно правило**.

Решения вида «оценить всю картину и расставить приоритеты» изначально плохо ложатся на `if / else`, зато это ровно то, в чём сильны LLM. Эталонный мозг (`brains/xwar3/strategy/brain/coach.py`) использует следующее разделение на слои.

## Слои

```text
LLM (советник)          Раз в 20 игровых секунд, асинхронно, никогда не блокирует тик
  Вход: снимок партии на одну страницу (ресурсы, пища, распределение рабочих, рудники, войска, технологии, герои, разведданные о противнике, недавние события)
  Выход: строгий JSON — диагноз в одну фразу + распределение рабочих + что строить первым + позиция на эту минуту + чего не делать
        │
        ▼  белый список + ограничение min/max + вето
Слой правил (Bot, каждый тик)   Переводит советы в «смещения» уже имеющихся возможностей: распределение рабочих, приоритет постройки / найма, атакующая позиция
        │
        ▼
Слой исполнения (SDK / рефлекторный слой)   Отдаёт приказы, читает квитанции, микроконтроль
```

## Контракт вывода

Заставьте модель выдавать JSON с фиксированным набором полей — ничего не добавлять и не опускать:

```json
{
  "diagnosis": "Одна фраза: главная проблема в партии; должна подтверждаться входными данными",
  "workers":   { "gold": 10, "lumber": 6 },
  "priority":  ["hpea", "hhou", "hbar"],
  "posture":   "creep",
  "avoid":     ["Не исследовать улучшение брони первым, когда не хватает древесины"]
}
```

| Поле | Как его использует слой правил | Ограничение в эталонном мозге |
|---|---|---|
| `workers` | Целевое число рабочих на золоте и на древесине | Золото 2 ~ 25, древесина 1 ~ 20; сумма не больше общего числа крестьян |
| `priority` | Порядок приоритетов для найма / постройки / исследований | Не больше 4; принимаются только четырёхсимвольные коды из таблицы «допустимых кодов» |
| `posture` | Позиция на эту минуту | Только одно из `attack` `defend` `creep` `expand` `recover` `hold` |
| `avoid` | Чего не делать в эту минуту | Не больше 2 пунктов |
| `diagnosis` | Только для логов и отображения в консоли | — |

Для каждой расы — свой промпт, и в нём только компромиссы, характерные для этой расы (совместное строительство и Ополчение у Людей, Норы у Орды, Haunted Gold Mine у Нежити, Entangled Gold Mine у Ночных эльфов…). Общие правила вынесите в общую часть — не копируйте их четыре раза.

## Четыре жёстких ограничения

Каждое из них эталонный мозг усвоил на собственных ошибках:

1. **Советник никогда не отдаёт приказы юнитам напрямую.** Он не видит происходящего в масштабе 150 ms и к тому же галлюцинирует. Он меняет только цели и приоритеты; кто куда идёт и кого атакует, по-прежнему решают слой правил и рефлекторный слой — у командования может быть только один хозяин.
2. **Асинхронность.** Один вызов советника занимает около 1 секунды и выполняется в фоновом потоке; действует последний результат, и он **никогда не блокирует тик**. Если модель не запущена, не уложилась в тайм-аут или ответила чепухой, ведите себя так, будто этого слоя нет, и возвращайтесь к чистым правилам. Слишком старый совет (старше 3 интервалов) тоже не используется.
3. **Белый список + ограничение значений.** Каждое поле должно отображаться на уже существующую возможность, а числа ограничиваются разумным диапазоном. Всё нераспознанное **подсчитывается и отбрасывается**, а не игнорируется молча.
4. **Считайте всё.** Сколько раз спросили, сколько успешных ответов, сколько тайм-аутов, сколько раз сработало ограничение, сколько раз было принято каждое поле — публикуйте всё это вместе с последним входом, отправленным модели. Иначе на вопрос «а этот слой вообще что-то даёт?» ответить невозможно.

> **Что умеет безопасно деградировать, легче всего деградирует незаметно**
>
> Советник устроен так, что «сбой = этого слоя нет», поэтому, когда сервис модели не запущен, бот ведёт себя в точности как на чистых правилах, и снаружи ничего не заметно. Однажды у эталонного мозга советник целый день не мог подключиться ни на одном из 6 экземпляров — и никто этого не заметил. Обязательно публикуйте «время последнего успешного вызова» и «причину последнего сбоя» — именно для этого в [консоли Farsight](https://war3ai.com/ru/docs/console/) есть страница «Стратегический советник».

## Реализация в вашем боте

Ниже — минимальный каркас, который работает с любым OpenAI-совместимым API (LM Studio, Ollama или облачный API) и использует только стандартную библиотеку:

```python title="coached_bot.py"
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, только из allowed], "posture": одно из шести, "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-модели (за один вызов активируется лишь малая часть параметров) намного быстрее плотных моделей того же размера. Эталонный мозг использует Qwen3.6-35B-A3B (LM Studio, Q4): медиана **1.09 s**, самый медленный ответ 1.45 s, 5/5 ответов напрямую разбираются через `json.loads` |
| Локальные «думающие» модели | **Обязательно отключите блок размышлений**, иначе все токены уйдут на размышления и ни одного JSON вы не получите. LM Studio игнорирует `/no_think`; эталонный мозг перешёл на `/v1/completions`, сам собирает ChatML и заранее подставляет пустой `<think></think>` и `{` |
| Облачные модели | Задержка обычно выше, но эта схема слоёв асинхронна по своей природе; экономические решения измеряются минутами, так что несколько секунд задержки допустимы |

> **Примечание**
>
> Та же модель может озвучивать ваших юнитов: см. [Реплики и локальные модели](https://war3ai.com/ru/docs/speech/). Если хотите, чтобы модель сама отдавала приказы каждый тик (а не работала советником), дождитесь JSON-шлюза [Арены](https://war3ai.com/ru/arena/).