Squad Commander¶

an RTS where an LLM commander re-prioritises a classical squad.

📄 Docs only

Tags: ai llm rts agent

The demo runs the game logic and the language model on two separate layers:

  • The classical squad AI is authoritative and runs every frame. Each blue unit is a dependency-free Brain (SquadUnitBrain) that reads the shared squad plan from its blackboard and returns one Action per frame: attack the nearest red enemy, hold position, or fall back to the rally point. It imports no LLM code and runs the same whether or not a commander is present.

  • The LLM commander sets strategy, not per-frame control. Every few seconds, off the frame thread, SquadCommanderBrain asks the model for a typed battle plan (a stance / focus / target schema) and writes that plan onto the shared squad blackboard for every unit to read. If a call is late, dropped, or malformed, the last good plan stays in effect, so the squad never stalls.

Remove the commander and the squad still plays on the default plan, which shows the LLM layer is optional.

Run offline (default: a scripted commander reads the battlefield and swaps the stance, after a trivial awaitable that exercises the off-thread path):

uv run python examples/demos/squad_commander.py

Run against a real model (records to a local cache so a re-run is deterministic):

SIMVX_LLM_BASE_URL=http://host:8000/v1 SIMVX_LLM_MODEL=your-model         uv run python examples/demos/squad_commander.py --live

Controls: Space or left-click spawns a wave of red enemies, R or right-click forces a manual retreat order, Esc quits. A first wave auto-spawns shortly after start so the commander switches from HOLD to ATTACK on its own.

This demo is desktop-only: it calls out to a local or remote LLM server and does not run in the browser web export.

Source¶

  1"""Squad Commander: an RTS where an LLM commander re-prioritises a classical squad.
  2
  3The demo runs the game logic and the language model on two separate layers:
  4
  5- The classical squad AI is authoritative and runs every frame. Each blue unit is a
  6  dependency-free ``Brain`` (``SquadUnitBrain``) that reads the shared squad plan
  7  from its blackboard and returns one Action per frame: attack the nearest red enemy,
  8  hold position, or fall back to the rally point. It imports no LLM code and runs the
  9  same whether or not a commander is present.
 10
 11- The LLM commander sets strategy, not per-frame control. Every few seconds, off the
 12  frame thread, ``SquadCommanderBrain`` asks the model for a typed battle plan (a
 13  ``stance`` / ``focus`` / ``target`` schema) and writes that plan onto the shared
 14  squad blackboard for every unit to read. If a call is late, dropped, or malformed,
 15  the last good plan stays in effect, so the squad never stalls.
 16
 17Remove the commander and the squad still plays on the default plan, which shows the
 18LLM layer is optional.
 19
 20Run offline (default: a scripted commander reads the battlefield and swaps the stance,
 21after a trivial awaitable that exercises the off-thread path):
 22
 23    uv run python examples/demos/squad_commander.py
 24
 25Run against a real model (records to a local cache so a re-run is deterministic):
 26
 27    SIMVX_LLM_BASE_URL=http://host:8000/v1 SIMVX_LLM_MODEL=your-model \
 28        uv run python examples/demos/squad_commander.py --live
 29
 30Controls: Space or left-click spawns a wave of red enemies, R or right-click forces a
 31manual retreat order, Esc quits. A first wave auto-spawns shortly after start so the
 32commander switches from HOLD to ATTACK on its own.
 33
 34This demo is desktop-only: it calls out to a local or remote LLM server and does not
 35run in the browser web export.
 36
 37# /// simvx
 38# tags = ["ai", "llm", "rts", "agent"]
 39# web = { disabled = true, reason = "requires a local/remote LLM server; not available in the browser runtime" }
 40# ///
 41"""
 42
 43from __future__ import annotations
 44
 45import asyncio
 46import json
 47import sys
 48
 49from simvx.ai import DEFAULT_PLAN, PLAN_KEY, CachingClient, OpenAICompatibleClient, SquadCommanderBrain
 50from simvx.ai.client import LLMClient, LLMResponse
 51from simvx.core import AnchorPreset, Input, InputMap, Key, Label, MouseButton, Node2D, Vec2
 52from simvx.core.ai import Action, ActionResult, AgentNode, AIContext, Blackboard, Brain
 53from simvx.graphics import App
 54
 55WIDTH, HEIGHT = 1000, 700
 56RALLY = Vec2(140, HEIGHT / 2)
 57CACHE_DIR = "/tmp/simvx_squad_commander_cache"
 58
 59UNIT_SPEED = 130.0
 60UNIT_RADIUS = 12.0
 61ATTACK_RANGE = 60.0
 62ENEMY_RADIUS = 12.0
 63
 64
 65class ScriptedCommanderClient(LLMClient):
 66    """An offline fake commander: a trivial awaitable, then a plan for the battlefield.
 67
 68    Stands in for a real model so the demo runs with no network, while still
 69    exercising the full async path (the ``await`` runs on the AsyncSlot loop, never
 70    the frame thread), so non-blocking / coalescing / degrade behaviour is identical.
 71    """
 72
 73    async def complete(self, messages, **kwargs) -> LLMResponse:
 74        await asyncio.sleep(0.06)  # simulate model latency, off the frame thread
 75        facts = json.loads(messages[-1]["content"].split("Battlefield: ", 1)[1])
 76        enemies = int(facts.get("enemy_count", 0))
 77        forced = facts.get("forced_retreat", False)
 78        if forced:
 79            plan = {"stance": "retreat", "focus": "nearest", "target": None, "rationale": "manual fallback"}
 80        elif enemies == 0:
 81            plan = {"stance": "hold", "focus": "nearest", "target": None, "rationale": "no contacts, hold rally"}
 82        elif enemies >= 4:
 83            plan = {"stance": "retreat", "focus": "nearest", "target": None, "rationale": "overwhelmed, fall back"}
 84        else:
 85            plan = {"stance": "attack", "focus": "nearest", "target": None, "rationale": "engage, we have numbers"}
 86        return LLMResponse(text=json.dumps(plan))
 87
 88
 89def _nearest_enemy(ctx: AIContext, pos: Vec2) -> tuple[Enemy | None, float]:
 90    """Return the closest living enemy and its distance, or ``(None, inf)`` if none are left."""
 91    enemies = ctx.blackboard.get("enemies", ())
 92    best = None
 93    best_d = float("inf")
 94    for enemy in enemies:
 95        if not enemy.alive:
 96            continue
 97        d = float((enemy.position - pos).length())
 98        if d < best_d:
 99            best_d, best = d, enemy
100    return best, best_d
101
102
103class MoveToward(Action):
104    """Step the unit toward a target point at unit speed (authoritative, every frame)."""
105
106    def __init__(self, target: Vec2) -> None:
107        self.target = target
108
109    def execute(self, ctx: AIContext) -> ActionResult:
110        unit = ctx.agent
111        delta = self.target - unit.position
112        dist = float(delta.length())
113        if dist > 1.0:
114            unit.position = unit.position + delta * (min(UNIT_SPEED * ctx.dt, dist) / dist)
115        return ActionResult.success("move")
116
117
118class AttackEnemy(Action):
119    """Damage the nearest enemy if in range, else close the distance."""
120
121    def execute(self, ctx: AIContext) -> ActionResult:
122        unit = ctx.agent
123        enemy, dist = _nearest_enemy(ctx, unit.position)
124        if enemy is None:
125            return MoveToward(RALLY).execute(ctx)
126        if dist <= ATTACK_RANGE:
127            enemy.hp -= 90.0 * ctx.dt
128            unit.firing_at = enemy.position.copy()
129            return ActionResult.success("attack")
130        unit.firing_at = None
131        return MoveToward(enemy.position).execute(ctx)
132
133
134class SquadUnitBrain(Brain):
135    """Dep-free classical unit AI: reads the shared plan every frame, returns an Action.
136
137    Authoritative for unit actions and imports no LLM code. With no plan on the
138    board (no commander present) ``get`` falls back to ``DEFAULT_PLAN`` -> a sane hold.
139    """
140
141    def decide(self, ctx: AIContext) -> Action | None:
142        plan = ctx.blackboard.get(PLAN_KEY, DEFAULT_PLAN)
143        stance = plan["stance"]
144        if stance == "attack":
145            return AttackEnemy()
146        if stance == "retreat":
147            ctx.agent.firing_at = None
148            return MoveToward(RALLY)
149        ctx.agent.firing_at = None
150        return MoveToward(ctx.agent.home)  # hold: settle on the unit's home slot
151
152
153class Unit(AgentNode):
154    """A blue squad unit driven by the classical ``SquadUnitBrain`` every frame."""
155
156    def __init__(self, home: Vec2, squad: Blackboard, **kwargs) -> None:
157        super().__init__(brain=SquadUnitBrain(), squad=squad, **kwargs)
158        self.home = home
159        self.position = home.copy()
160        self.firing_at: Vec2 | None = None
161        # on_draw reads plain (non-Property) state the brain mutates every frame
162        # (position + firing_at), so nothing dirties the node under the retained
163        # 2D renderer. Re-capture this small draw each frame (see dodge_the_creeps).
164        self.dynamic = True
165
166    def on_draw(self, renderer) -> None:
167        renderer.draw_circle(self.position, UNIT_RADIUS, colour=(0.3, 0.6, 1.0, 1.0), filled=True)
168        if self.firing_at is not None:
169            renderer.draw_line(self.position, self.firing_at, colour=(1.0, 0.9, 0.3, 0.8), thickness=2.0)
170
171
172class Enemy:
173    """A red target. Plain object (not a node): the demo draws and culls them."""
174
175    def __init__(self, position: Vec2) -> None:
176        self.position = position
177        self.hp = 100.0
178
179    @property
180    def alive(self) -> bool:
181        return self.hp > 0.0
182
183
184class CommanderHUD(Node2D):
185    """Bottom HUD showing the live plan (last good if a call is in flight / failed)."""
186
187    def __init__(self, squad: Blackboard, **kwargs) -> None:
188        super().__init__(**kwargs)
189        self.squad = squad
190        self.label: Label | None = None
191
192    def on_ready(self) -> None:
193        label = Label("...", name="Plan")
194        label.set_anchor_preset(AnchorPreset.BOTTOM_WIDE)
195        label.margin_left = 20
196        label.margin_right = 20
197        label.margin_top = -64
198        label.margin_bottom = -16
199        label.font_size = 22.0
200        label.alignment = "center"
201        self.add_child(label)
202        self.label = label
203
204    def on_update(self, dt: float) -> None:
205        if self.label is None:
206            return
207        plan = self.squad.get(PLAN_KEY, DEFAULT_PLAN)
208        self.label.text = f"COMMANDER  stance: {plan['stance'].upper()}   \"{plan['rationale']}\""
209
210
211class Battle(Node2D):
212    """Root: owns the shared squad board, spawns units + a low-frequency commander."""
213
214    def __init__(self, client: LLMClient | None = None, **kwargs) -> None:
215        super().__init__(**kwargs)
216        # No-arg construction runs offline so the screenshot walker (which instantiates
217        # the root with no args) gets the scripted commander; main() passes a real
218        # client for --live.
219        self._client = client if client is not None else ScriptedCommanderClient()
220        self.squad = Blackboard()
221        self.enemies: list[Enemy] = []
222        self._forced_retreat_t = 0.0
223        self._auto_wave_t = 2.0  # one-shot: first wave spawns itself so the plan flips with no input
224        self.commander: AgentNode | None = None
225        # on_draw renders enemies from a plain list (drift/damage/cull every frame,
226        # variable count) plus the rally marker, none of it Property-backed, so the
227        # retained 2D renderer would freeze it. Re-capture each frame.
228        self.dynamic = True
229
230    def on_ready(self) -> None:
231        # Mouse bindings mirror the keys so the demo is fully playable with the mouse alone.
232        InputMap.add_action("spawn_wave", [Key.SPACE, MouseButton.LEFT])
233        InputMap.add_action("retreat", [Key.R, MouseButton.RIGHT])
234        InputMap.add_action("quit", [Key.ESCAPE])
235
236        # Shared squad facts the commander reads (allowlisted slice, not the tree).
237        self.squad.set("enemies", self.enemies)
238        self.squad.set("enemy_count", 0)
239        self.squad.set("forced_retreat", False)
240
241        # Classical units: each gets squad.child(), reads the plan every frame.
242        for i in range(5):
243            home = Vec2(220, 160 + i * 95)
244            self.add_child(Unit(home, self.squad, name=f"Unit{i}"))
245
246        # The low-frequency LLM commander. Its own AgentNode ticks every frame, but
247        # the brain self-throttles to ``period`` seconds and writes to the squad board.
248        self.commander = AgentNode(
249            brain=SquadCommanderBrain(
250                self._client,
251                self.squad,
252                facts=["enemy_count", "forced_retreat"],
253                period=2.5,
254            ),
255            name="Commander",
256        )
257        self.add_child(self.commander)
258        self.add_child(CommanderHUD(self.squad, name="HUD"))
259
260        title = Label("Squad Commander  -  Space/click: spawn wave   R/right-click: retreat   Esc: quit", name="Title")
261        title.set_anchor_preset(AnchorPreset.CENTER_TOP)
262        title.margin_left = -320
263        title.margin_right = 320
264        title.margin_top = 16
265        title.margin_bottom = 40
266        title.font_size = 18.0
267        title.alignment = "center"
268        self.add_child(title)
269
270    def _spawn_wave(self) -> None:
271        for i in range(3):
272            self.enemies.append(Enemy(Vec2(WIDTH - 120, 180 + i * 130)))
273
274    def on_update(self, dt: float) -> None:
275        # One-shot opening wave: a passive viewer sees the commander react unprompted.
276        if self._auto_wave_t > 0.0:
277            self._auto_wave_t -= dt
278            if self._auto_wave_t <= 0.0:
279                self._spawn_wave()
280        if Input.is_action_just_pressed("spawn_wave"):
281            self._spawn_wave()
282        if Input.is_action_just_pressed("retreat"):
283            self._forced_retreat_t = 3.0
284        if Input.is_action_just_pressed("quit"):
285            self.app.quit()
286
287        # Enemies drift toward the squad and are culled when dead.
288        for enemy in self.enemies:
289            enemy.position.x -= 25.0 * dt
290        self.enemies[:] = [e for e in self.enemies if e.alive and e.position.x > 60]
291
292        # Update the authoritative facts the commander reads (every frame).
293        self._forced_retreat_t = max(0.0, self._forced_retreat_t - dt)
294        self.squad.set("enemy_count", len(self.enemies))
295        self.squad.set("forced_retreat", self._forced_retreat_t > 0.0)
296
297    def on_draw(self, renderer) -> None:
298        renderer.draw_circle(RALLY, 18, colour=(0.3, 0.6, 1.0, 0.25), filled=True)
299        for enemy in self.enemies:
300            renderer.draw_circle(enemy.position, ENEMY_RADIUS, colour=(1.0, 0.35, 0.35, 1.0), filled=True)
301            renderer.draw_rect(
302                (enemy.position.x - 14, enemy.position.y - 22),
303                (28 * max(0.0, enemy.hp) / 100.0, 4),
304                colour=(0.4, 1.0, 0.4, 1.0),
305                filled=True,
306            )
307
308
309def _build_client(live: bool) -> LLMClient:
310    if not live:
311        return ScriptedCommanderClient()
312    return CachingClient(OpenAICompatibleClient.from_env(), CACHE_DIR, mode="auto")
313
314
315def main() -> None:
316    live = "--live" in sys.argv
317    app = App(title="SimVX Squad Commander", width=WIDTH, height=HEIGHT)
318    app.run(Battle(_build_client(live), name="Battle"))
319
320
321if __name__ == "__main__":
322    main()