Squad Commander

an LLM officer picks the stance, the squad fights every frame.

📄 Docs only

Tags: ai llm rts agent

A model call takes far longer than a frame, so the model never drives the game. This demo splits the work into two layers, and only the fast one is authoritative:

• The squad runs every frame. Each blue Unit reads one word from the battle’s standing plan, attack, hold or retreat, and acts on it: close on the nearest red enemy and shoot it, settle on its home slot, or fall back to the rally point. That is ordinary Python in on_update, with no LLM code and nothing to await.

• The commander is a Planner from simvx.ai. At most every PLAN_PERIOD seconds it asks a model, on a worker thread, for a plan matching PLAN_SCHEMA: a stance and one line of reasoning.

  self.planner = Planner(client, schema=PLAN_SCHEMA, validate=validate_plan, period=2.5, default=DEFAULT_PLAN)
  
  def on_update(self, dt):
      self.plan = self.planner.poll(dt, enemy_count=len(self.enemies), forced_retreat=...)
  

poll never blocks. The keyword arguments are the whole of what the commander sees, a small slice of the battle chosen by the game, never the scene tree. Between answers, and while a call is in flight, poll hands back the last good plan, so a call that is slow, fails or answers nonsense costs the squad nothing. A schema is a request, not a guarantee: validate_plan is the second gate, refusing a stance the units cannot carry out so the last good plan survives. Until the first answer lands the squad fights on DEFAULT_PLAN, and it would keep doing so with the planner deleted. Call close() when the planner’s owner leaves the tree.

The client is any LLMClient. By default the demo runs offline on ScriptedCommanderClient, a stand-in that waits a moment, as a model would, then answers in the same schema from the same facts. Run it with --live to use a real model through OpenAICompatibleClient.from_env(), which reads SIMVX_LLM_BASE_URL (the endpoint up to /chat/completions), SIMVX_LLM_MODEL and, if the endpoint needs one, SIMVX_LLM_API_KEY. A CachingClient records each answer under $XDG_CACHE_HOME/simvx/squad_commander/ (SQUAD_COMMANDER_CACHE_DIR overrides it), so a re-run replays them for free.

Press Space or the Spawn wave button to send in three enemies, and R or the Fall back button to order the squad to break contact for a few seconds; Esc quits. A first wave arrives on its own two seconds in, so a viewer who touches nothing still sees the commander switch from HOLD to ATTACK. With four or more enemies on the field the scripted commander orders a retreat instead.

Source

# /// simvx
# tags = ["ai", "llm", "rts", "agent"]
# web = { disabled = true, reason = "requires a local/remote LLM server; not available in the browser runtime" }
# ///
"""Squad Commander: an LLM officer picks the stance, the squad fights every frame.

A model call takes far longer than a frame, so the model never drives the game. This demo splits
the work into two layers, and only the fast one is authoritative:

- **The squad runs every frame.** Each blue `Unit` reads one word from the battle's standing
  plan, `attack`, `hold` or `retreat`, and acts on it: close on the nearest red enemy and shoot
  it, settle on its home slot, or fall back to the rally point. That is ordinary Python in
  `on_update`, with no LLM code and nothing to await.
- **The commander is a `Planner`** from `simvx.ai`. At most every `PLAN_PERIOD` seconds it asks a
  model, on a worker thread, for a plan matching `PLAN_SCHEMA`: a stance and one line of
  reasoning.

      self.planner = Planner(client, schema=PLAN_SCHEMA, validate=validate_plan, period=2.5, default=DEFAULT_PLAN)

      def on_update(self, dt):
          self.plan = self.planner.poll(dt, enemy_count=len(self.enemies), forced_retreat=...)

`poll` never blocks. The keyword arguments are the whole of what the commander sees, a small slice
of the battle chosen by the game, never the scene tree. Between answers, and while a call is in
flight, `poll` hands back the last good plan, so a call that is slow, fails or answers nonsense
costs the squad nothing. A schema is a request, not a guarantee: `validate_plan` is the second
gate, refusing a stance the units cannot carry out so the last good plan survives. Until the first
answer lands the squad fights on `DEFAULT_PLAN`, and it would keep doing so with the planner
deleted. Call `close()` when the planner's owner leaves the tree.

The client is any `LLMClient`. By default the demo runs offline on `ScriptedCommanderClient`, a
stand-in that waits a moment, as a model would, then answers in the same schema from the same
facts. Run it with `--live` to use a real model through `OpenAICompatibleClient.from_env()`, which
reads `SIMVX_LLM_BASE_URL` (the endpoint up to `/chat/completions`), `SIMVX_LLM_MODEL` and, if the
endpoint needs one, `SIMVX_LLM_API_KEY`. A `CachingClient` records each answer under
`$XDG_CACHE_HOME/simvx/squad_commander/` (`SQUAD_COMMANDER_CACHE_DIR` overrides it), so a re-run
replays them for free.

Press Space or the Spawn wave button to send in three enemies, and R or the Fall back button to
order the squad to break contact for a few seconds; Esc quits. A first wave arrives on its own two
seconds in, so a viewer who touches nothing still sees the commander switch from HOLD to ATTACK.
With four or more enemies on the field the scripted commander orders a retreat instead.
"""

import asyncio
import json
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from simvx.ai import CachingClient, LLMClient, LLMResponse, Message, OpenAICompatibleClient, Planner
from simvx.core import (
    Anchor,
    Button,
    CanvasLayer,
    HAlign,
    HBoxContainer,
    Key,
    Label,
    Node2D,
    Property,
    Timer,
    Vec2,
    on_input,
)
from simvx.graphics import App

SQUAD_SIZE = 5
UNIT_SPEED = 130.0
UNIT_RADIUS = 12.0
ATTACK_RANGE = 60.0
DAMAGE_PER_SECOND = 90.0

WAVE_SIZE = 3
ENEMY_RADIUS = 12.0
ENEMY_SPEED = 25.0
ENEMY_HP = 100.0

FIRST_WAVE_DELAY = 2.0  # seconds before the opening wave arrives unprompted
RETREAT_SECONDS = 4.0  # how long a fallback order stays in the commander's facts
PLAN_PERIOD = 2.5  # seconds between commander decisions

# The squad knows exactly three stances, so the schema names them and a model cannot invent a fourth.
STANCES = ("attack", "hold", "retreat")
PLAN_SCHEMA = {
    "type": "object",
    "properties": {
        "stance": {"type": "string", "enum": list(STANCES)},
        "rationale": {"type": "string"},
    },
    "required": ["stance", "rationale"],
    "additionalProperties": False,
}
DEFAULT_PLAN = {"stance": "hold", "rationale": "no orders yet, holding the line"}


def validate_plan(plan: dict[str, Any]) -> dict[str, Any] | None:
    """The plan the squad should adopt, or None to keep the last good one."""
    if plan.get("stance") not in STANCES:
        return None
    return {"stance": plan["stance"], "rationale": str(plan.get("rationale", ""))[:60]}


class ScriptedCommanderClient(LLMClient):
    """An offline stand-in for a model: a short wait, then a plan for the situation."""

    async def complete(self, messages: list[Message], **kwargs) -> LLMResponse:
        await asyncio.sleep(0.06)  # model latency, spent on the planner's worker thread
        # Planner hands the client only its prompt, so the facts are read back out of the text.
        prompt = messages[-1]["content"]
        facts = json.loads(prompt[prompt.index("{") :])
        if facts["forced_retreat"]:
            plan = {"stance": "retreat", "rationale": "fallback ordered, break contact"}
        elif facts["enemy_count"] == 0:
            plan = {"stance": "hold", "rationale": "no contacts, hold the rally"}
        elif facts["enemy_count"] >= 4:
            plan = {"stance": "retreat", "rationale": "outnumbered, fall back"}
        else:
            plan = {"stance": "attack", "rationale": "engage, we have the numbers"}
        return LLMResponse(text=json.dumps(plan))


@dataclass
class Enemy:
    """A red target. A plain object, not a node: the battle moves, draws and culls them."""

    position: Vec2
    hp: float = ENEMY_HP


class Unit(Node2D):
    """A blue squad unit. It reads the battle's plan every frame and acts on it."""

    # The world point this unit is shooting at, or None. A Property, so each write redraws the unit;
    # the point stays right in local space because a firing unit stands still.
    firing_at = Property(None)

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.home = Vec2()  # the slot this unit holds when there is nothing to do

    def on_enter_tree(self):
        self.battle = self.ancestor(Battle)  # the plan, the enemies and the rally point live there

    def on_update(self, dt):
        match self.battle.plan["stance"]:
            case "attack":
                self.attack(dt)
            case "retreat":
                self.firing_at = None
                self.move_toward(self.battle.rally, dt)
            case _:  # hold
                self.firing_at = None
                self.move_toward(self.home, dt)

    def attack(self, dt):
        """Shoot the nearest enemy in range, else close on it, else regroup at the rally point."""
        enemy = self.battle.nearest_enemy(self.position)
        if enemy is None:
            self.firing_at = None
            self.move_toward(self.battle.rally, dt)
        elif (enemy.position - self.position).length() <= ATTACK_RANGE:
            enemy.hp -= DAMAGE_PER_SECOND * dt
            self.firing_at = enemy.position.copy()
        else:
            self.firing_at = None
            self.move_toward(enemy.position, dt)

    def move_toward(self, target, dt):
        delta = target - self.position
        distance = float(delta.length())
        if distance > 1.0:
            self.position = self.position + delta * (min(UNIT_SPEED * dt, distance) / distance)

    def on_draw(self, renderer):
        renderer.circle((0, 0), UNIT_RADIUS, fill=(0.3, 0.6, 1.0, 1.0))
        if self.firing_at is not None:
            renderer.line((0, 0), self.to_local(self.firing_at), stroke=(1.0, 0.9, 0.3, 0.8), width=2.0)


class Battle(Node2D):
    """The battlefield: the squad, the enemies, and the commander's standing plan."""

    dynamic = True  # the enemies are plain objects that move every frame
    input_context = {"spawn_wave": [Key.SPACE], "fall_back": [Key.R], "quit": [Key.ESCAPE]}

    def __init__(self, client: LLMClient | None = None, **kwargs):
        super().__init__(**kwargs)
        self.planner = Planner(
            client or ScriptedCommanderClient(),
            schema=PLAN_SCHEMA,
            validate=validate_plan,
            period=PLAN_PERIOD,
            default=DEFAULT_PLAN,
        )
        self.plan = dict(DEFAULT_PLAN)  # the standing orders: the planner writes them, the units read them
        self.enemies: list[Enemy] = []
        self.rally = Vec2()

    def on_ready(self):
        self.units = [self.add_child(Unit(name=f"Unit{i}")) for i in range(SQUAD_SIZE)]
        self.layout()
        for unit in self.units:
            unit.position = unit.home
        self.add_child(Timer(duration=FIRST_WAVE_DELAY, autostart=True)).timeout.connect(self.spawn_wave)
        self.fallback = self.add_child(Timer(duration=RETREAT_SECONDS))

        hud = self.add_child(CanvasLayer())
        self.plan_label = hud.add_child(
            Label(anchor=Anchor.BOTTOM_WIDE, inset=(20, 16), size=(0, 48), font_size=22, alignment=HAlign.CENTRE)
        )

        # Example help overlay: not part of the technique shown.
        bar = hud.add_child(HBoxContainer(anchor=Anchor.TOP_LEFT, inset=12))
        bar.add_child(Button("Spawn wave", action="spawn_wave"))
        bar.add_child(Button("Fall back", action="fall_back"))

    def layout(self):
        """Rally point and home slots follow the window, so a resize just moves them."""
        width, height = self.viewport_size
        self.rally = Vec2(width * 0.14, height * 0.5)
        for i, unit in enumerate(self.units):
            unit.home = Vec2(width * 0.22, height * (0.22 + 0.14 * i))

    @on_input("spawn_wave")
    def spawn_wave(self, _event=None):
        """Send a wave of enemies in from the right edge."""
        width, height = self.viewport_size
        self.enemies += [Enemy(Vec2(width - 120, height * (0.26 + 0.19 * i))) for i in range(WAVE_SIZE)]

    @on_input("fall_back")
    def order_fall_back(self, _event):
        self.fallback.start()

    @on_input("quit")
    def quit(self, _event):
        self.app.quit()

    def nearest_enemy(self, point):
        living = [enemy for enemy in self.enemies if enemy.hp > 0]
        return min(living, key=lambda enemy: (enemy.position - point).length(), default=None)

    def on_update(self, dt):
        self.layout()
        # Enemies advance on the rally point and are cleared once dead or past it.
        for enemy in self.enemies:
            enemy.position.x -= ENEMY_SPEED * dt
        self.enemies = [e for e in self.enemies if e.hp > 0 and e.position.x > self.rally.x]

        self.plan = self.planner.poll(dt, enemy_count=len(self.enemies), forced_retreat=not self.fallback.stopped)
        self.plan_label.text = f'COMMANDER  {self.plan["stance"].upper()}   "{self.plan["rationale"]}"'

    def on_exit_tree(self):
        self.planner.close()

    def on_draw(self, renderer):
        renderer.circle(self.rally, 18, fill=(0.3, 0.6, 1.0, 0.25))
        for enemy in self.enemies:
            renderer.circle(enemy.position, ENEMY_RADIUS, fill=(1.0, 0.35, 0.35, 1.0))
            bar = (28 * max(0.0, enemy.hp) / ENEMY_HP, 4)
            renderer.rect((enemy.position.x - 14, enemy.position.y - 22), bar, fill=(0.4, 1.0, 0.4, 1.0))


def live_client() -> LLMClient:
    """A real model, with its answers recorded to the cache and replayed on a re-run."""
    cache = os.environ.get("SQUAD_COMMANDER_CACHE_DIR")
    base = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "simvx" / "squad_commander"
    return CachingClient(OpenAICompatibleClient.from_env(), Path(cache) if cache else base)


if __name__ == "__main__":
    client = live_client() if "--live" in sys.argv else None
    App(title="Squad Commander", width=1000, height=700).run(Battle(client))