From fe50196dabcdc8fd7c5dd96ea0a5757ff603e3da Mon Sep 17 00:00:00 2001 From: Isaac Johnson Date: Sun, 6 Sep 2026 13:56:51 -0500 Subject: [PATCH] first pass at agentic bot --- botagent_ai/INSTALL.md | 46 ++++ botagent_ai/bot.py | 467 +++++++++++++++++++++++++++++++++++ botagent_ai/requirements.txt | 2 + 3 files changed, 515 insertions(+) create mode 100644 botagent_ai/INSTALL.md create mode 100644 botagent_ai/bot.py create mode 100644 botagent_ai/requirements.txt diff --git a/botagent_ai/INSTALL.md b/botagent_ai/INSTALL.md new file mode 100644 index 0000000..885e980 --- /dev/null +++ b/botagent_ai/INSTALL.md @@ -0,0 +1,46 @@ +# Install Instructions + +## Prerequisites +- Python 3.8+ +- Ollama installed and running with the `gemm4:12b` model downloaded. + +## Setup +1. Navigate to the directory: + ```bash + cd botagent_ai + ``` + +2. Create a virtual environment (optional but recommended): + ```bash + python -m venv venv + source venv/bin/activate # On Windows use: venv\Scripts\activate + ``` + +3. Install dependencies: + ```bash + pip install -r requirements.txt + ``` + +## Configuration +The bot plays botWebWars using an Ollama model for strategy (voluntary alliances and +navigation), while rule-mandated outcomes (forced battles/joins) are resolved automatically. +Configure it via environment variables or CLI flags: +- `OLLAMA_BASE_URL` / `--ollama-url`: Ollama instance URL (e.g., `http://192.168.1.220:11434`) +- `OLLAMA_MODEL` / `--ollama-model`: The model name (default is `gemma4:12b`) +- `BOT_SERVER_URL` / `-u`: botWebWars backend URL (default `http://localhost:8000/api`) +- `BOT_NAME` / `-n`: Bot display name +- `BOT_COLOR` / `-c`: Hex color for the bot avatar +- `BOT_STRENGTH` / `-s`: Starting strength (1-10) + +Example: +```bash +export OLLAMA_BASE_URL="http://192.168.1.220:11434" +export OLLAMA_MODEL="gemma4:12b" +python bot.py -n MyAIBot -s 5 +``` + +## Running the Bot +Run the script, then start the game from the frontend UI: +```bash +python bot.py +``` diff --git a/botagent_ai/bot.py b/botagent_ai/bot.py new file mode 100644 index 0000000..b0be6b1 --- /dev/null +++ b/botagent_ai/bot.py @@ -0,0 +1,467 @@ +"""AI-driven Bot Agent for botWebWars, powered by a local Ollama LLM. + +Replicates the general capabilities of botagent/bot_agent.py (registration, +radar-based navigation, obstacle avoidance, party formation, battles) but +delegates the *strategic* decisions to an Ollama model: + - Whether to propose a voluntary alliance with another solo bot. + - Which direction to move towards when navigating (given radar/obstacle data). + +Outcomes mandated by GAME_RULES.md (forced battles, forced joins based on +relative strength) are always resolved deterministically by the game engine +regardless of what the LLM prefers - the LLM is only ever offered a choice +among legal options. +""" + +import json +import os +import re +import time +import argparse +from typing import Any, Dict, List, Optional + +import requests + +DEFAULT_SERVER_URL = "http://localhost:8000/api" +DEFAULT_BOT_NAME = "OllamaBot" +DEFAULT_BOT_COLOR = "#8b5cf6" +DEFAULT_BOT_STRENGTH = 4 + +OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://192.168.1.220:11434") +OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "gemma4:12b") + +GAME_RULES_SUMMARY = """ +Rules you must respect when choosing among the OPTIONS given to you: +- Two solo bots that meet MAY voluntarily ally (not required). The stronger bot (or higher + score if tied) leads. Larger parties have an advantage in battle. +- A solo bot always joins a party if the party leader's strength >= its own (no choice). +- A solo bot always refuses and fights if the party leader is weaker (no choice). +- Two opposing parties that meet must always battle (no choice). +- Diagonal squeezes between obstacle corners cost strength (0.1 solo, 0.2 leader / 0.1 followers). +- The game ends when all bots are united into a single party. +You will only ever be asked to choose between options that are legal - always answer with the +requested JSON object and nothing else. +""" + + +def normalize_url(url: str) -> str: + """Ensure the API URL ends with /api without trailing slashes.""" + cleaned = url.rstrip("/") + if not cleaned.endswith("/api"): + cleaned = f"{cleaned}/api" + return cleaned + + +def extract_json(text: str) -> Optional[Dict[str, Any]]: + """Best-effort extraction of a JSON object from an LLM response.""" + if not text: + return None + match = re.search(r"\{.*\}", text, re.DOTALL) + if not match: + return None + try: + return json.loads(match.group(0)) + except json.JSONDecodeError: + return None + + +class OllamaClient: + def __init__(self, base_url: str, model: str): + self.base_url = base_url.rstrip("/") + self.model = model + + def ask_json(self, prompt: str) -> Optional[Dict[str, Any]]: + """Ask the model a question, requesting a JSON-only response.""" + url = f"{self.base_url}/api/generate" + payload = { + "model": self.model, + "prompt": prompt, + "format": "json", + "stream": False, + "options": {"temperature": 0.4}, + } + try: + res = requests.post(url, json=payload, timeout=60) + res.raise_for_status() + raw = res.json().get("response", "") + return extract_json(raw) + except requests.exceptions.RequestException as e: + print(f"āš ļø [OLLAMA ERROR] {e}") + return None + + +class AIBotAgent: + def __init__( + self, + name: str = DEFAULT_BOT_NAME, + color: str = DEFAULT_BOT_COLOR, + strength: int = DEFAULT_BOT_STRENGTH, + server_url: str = DEFAULT_SERVER_URL, + ollama_url: str = OLLAMA_BASE_URL, + ollama_model: str = OLLAMA_MODEL, + ): + self.name = name + self.color = color + self.strength = strength + self.base_url = normalize_url(server_url) + self.llm = OllamaClient(ollama_url, ollama_model) + self.bot_id: Optional[str] = None + self.party_id: Optional[str] = None + self.is_leader: bool = False + + # ------------------------------------------------------------------ + # Registration / status + # ------------------------------------------------------------------ + def register(self): + """Register the bot avatar on the grid or reconnect if already present.""" + try: + players = requests.get(f"{self.base_url}/players").json() + for p in players: + if p.get("name") == self.name: + self.bot_id = p["id"] + print(f"šŸ”„ [RECONNECT] Reconnected to existing {self.name} (ID: {self.bot_id}) at ({p.get('x')}, {p.get('y')})") + return + except Exception: + pass + + res = requests.post( + f"{self.base_url}/players", + json={"name": self.name, "color": self.color, "strength": self.strength}, + ) + if res.status_code == 400 and "already registered" in res.text: + players = requests.get(f"{self.base_url}/players").json() + for p in players: + if p.get("name") == self.name: + self.bot_id = p["id"] + print(f"šŸ”„ [RECONNECT] Reconnected to existing {self.name} (ID: {self.bot_id}) at ({p.get('x')}, {p.get('y')})") + return + + res.raise_for_status() + data = res.json() + self.bot_id = data["id"] + print(f"šŸš€ [REGISTER] Spawned {self.name} (ID: {self.bot_id}, Str: {self.strength}) at ({data['x']}, {data['y']})") + + def refresh_status(self): + """Update bot state (party membership, leader status, score).""" + res = requests.get(f"{self.base_url}/players/{self.bot_id}") + if res.status_code == 200: + data = res.json() + self.party_id = data.get("party_id") + self.is_leader = data.get("is_party_leader", False) + return data + return None + + # ------------------------------------------------------------------ + # Core decision loop + # ------------------------------------------------------------------ + def decide_and_act(self): + my_info = self.refresh_status() + if not my_info: + return + + print(f"\nšŸ¤– --- Turn for {self.name} | Score: {my_info['score']} | Party: {self.party_id or 'Solo'} ---") + + radar_res = requests.get(f"{self.base_url}/players/{self.bot_id}/radar").json() + targets = radar_res.get("targets", []) + + adjacent_target = None + for t in targets: + if t["distance"] <= 1 and not t["is_ally"]: + adjacent_target = t + break + + if adjacent_target: + self._handle_adjacent_encounter(adjacent_target, my_info) + else: + self._navigate_towards_goal(radar_res, my_info) + + def _handle_adjacent_encounter(self, target: Dict[str, Any], my_info: Dict[str, Any]): + """Resolve the encounter; the LLM only gets a say when the rules allow a choice.""" + target_name = target["name"] + target_str = target["strength"] + target_party = target.get("party_id") + + print(f"šŸ” [ADJACENT ENCOUNTER] Next to '{target_name}' (Str: {target_str}, Party: {target_party or 'None'})") + + # SCENARIO A: I am a Solo Bot + if not self.party_id: + if not target_party: + # Both solo: alliance is OPTIONAL - ask the LLM. + self._decide_voluntary_alliance(target, my_info) + else: + # Target belongs to a party: joining/refusing is mandated by relative strength. + party_info = requests.get(f"{self.base_url}/parties/{target_party}").json() + target_leader_str = target_str + if party_info: + leader_player = requests.get(f"{self.base_url}/players/{party_info['leader_id']}").json() + target_leader_str = leader_player.get("strength", 1) + + if self.strength <= target_leader_str: + print(f"šŸ¤ [RULE] Party leader strength {target_leader_str} >= my {self.strength}. Willingly joining squad!") + self._step_or_attack(target) + else: + print(f"āš”ļø [RULE] Party leader is weaker ({target_leader_str} < my {self.strength}). Must refuse and fight!") + self._initiate_battle(target["id"]) + + # SCENARIO B: I am in a Party + else: + if not self.is_leader: + print("šŸ›”ļø [PARTY MEMBER] Under command of party leader. Awaiting leader movement.") + requests.post(f"{self.base_url}/players/{self.bot_id}/pass") + return + + if not target_party: + if target_str <= self.strength: + print(f"šŸ¤ [RULE] Solo bot {target_name} is willing to join our squad under my leadership.") + self._step_or_attack(target) + else: + print(f"āš”ļø [RULE] Solo bot {target_name} refuses weaker leader! Squad is attacking!") + self._initiate_battle(target["id"]) + else: + print(f"āš”ļø [RULE] Hostile party detected: '{target.get('party_name')}'! Battle is mandatory!") + self._initiate_battle(target["id"]) + + def _decide_voluntary_alliance(self, target: Dict[str, Any], my_info: Dict[str, Any]): + """Ask the LLM whether to propose a voluntary alliance with another solo bot.""" + prompt = f"""{GAME_RULES_SUMMARY} +You are bot "{self.name}" (strength {self.strength}, score {my_info['score']}). +You just encountered another solo bot "{target['name']}" (strength {target['strength']}). +Whoever has greater strength (or higher score if tied) will lead the new party. +Forming an alliance is optional - larger parties are stronger in future battles, but you +give up independent control if you are not the stronger one. + +Respond ONLY with JSON: {{"form_alliance": true|false, "reasoning": "short reason"}} +""" + decision = self.llm.ask_json(prompt) or {} + form_alliance = decision.get("form_alliance", True) + reasoning = decision.get("reasoning", "") + + if form_alliance: + leader_id = self.bot_id if self.strength >= target["strength"] else target["id"] + print(f"šŸ¤ [LLM DECISION] Ally with {target['name']}! Leader: {'me' if leader_id == self.bot_id else target['name']}. {reasoning}") + self._execute_party_formation([self.bot_id, target["id"]], leader_id) + else: + print(f"šŸš¶ [LLM DECISION] Declining alliance with {target['name']}. {reasoning}") + requests.post(f"{self.base_url}/players/{self.bot_id}/pass") + + def _execute_party_formation(self, member_ids: List[str], leader_id: str): + """Form a party using the REST API.""" + try: + res = requests.post( + f"{self.base_url}/parties", + json={"member_ids": member_ids, "leader_id": leader_id, "name": f"Squad_{self.name}"}, + ) + if res.status_code == 201: + party = res.json() + print(f"āœ… [PARTY FORMED] Squad '{party['name']}' established! Leader: {party['leader_name']} | Str: {party['total_strength']}") + else: + requests.post(f"{self.base_url}/players/{self.bot_id}/pass") + except Exception as e: + print(f"Party formation error: {e}") + requests.post(f"{self.base_url}/players/{self.bot_id}/pass") + + def _initiate_battle(self, opponent_id: str): + """Explicitly call the 3-Bout D20 Battle endpoint.""" + print(f"šŸŽ² [BATTLE INITIATED] Clashing with opponent {opponent_id}...") + res = requests.post( + f"{self.base_url}/battles/fight", + json={"challenger_id": self.bot_id, "defender_id": opponent_id}, + ) + if res.status_code == 200: + battle = res.json() + print("\n--- āš”ļø 3-BOUT BATTLE RESOLUTION ---") + print(f"Bouts won: {battle['party1_name']} ({battle['party1_bouts_won']}) vs {battle['party2_name']} ({battle['party2_bouts_won']})") + for b in battle["bouts"]: + print(f" Bout #{b['bout_number']}: Roll {b['party1_roll']}Ɨ{b['party1_strength']} ({b['party1_score']}) vs Roll {b['party2_roll']}Ɨ{b['party2_strength']} ({b['party2_score']}) -> Winner: {b['winner_name']}") + print(f"šŸ† Overall Winner: {battle['winner_party_name']} (Leader {battle['winner_leader_name']} receives +2 pts)") + print(f"šŸ’€ Defeated: {battle['defeated_party_name']} (Leader {battle['killed_leader_name']} -1 pt)") + if battle.get("absorbed_members"): + print(f"šŸ§² Absorbed {len(battle['absorbed_members'])} member(s) into {battle['winner_party_name']}") + else: + print(f"Battle failed ({res.status_code}): {res.text}") + + def _step_or_attack(self, target: Dict[str, Any]): + """Move adjacent/towards the target while avoiding obstacles.""" + moves_res = requests.get(f"{self.base_url}/players/{self.bot_id}/available-moves").json() + moves = moves_res.get("moves", {}) + chosen = self._closest_direction_to(target["x"], target["y"], moves) + + if chosen: + res = requests.post(f"{self.base_url}/players/{self.bot_id}/move", json={"direction": chosen}).json() + if res.get("battle_triggered"): + print(f"āš”ļø Move triggered battle! Winner: {res['battle_result']['winner_party_name']}") + elif res.get("party_formed_triggered"): + print("šŸ¤ Move resulted in party alliance!") + else: + print("āš ļø No passable moves adjacent to target. Passing turn.") + requests.post(f"{self.base_url}/players/{self.bot_id}/pass") + + @staticmethod + def _closest_direction_to(target_x: int, target_y: int, moves: Dict[str, Any]) -> Optional[str]: + """Pick the available direction that minimizes Chebyshev distance to (target_x, target_y).""" + valid_moves = {d: chk for d, chk in moves.items() if chk.get("available")} + if not valid_moves: + return None + + def dist(chk: Dict[str, Any]) -> int: + return max(abs(chk["target_x"] - target_x), abs(chk["target_y"] - target_y)) + + return min(valid_moves.keys(), key=lambda d: (valid_moves[d].get("strength_penalty", 0.0) > 0, dist(valid_moves[d]))) + + # ------------------------------------------------------------------ + # Navigation (LLM-driven direction choice among legal moves) + # ------------------------------------------------------------------ + def _navigate_towards_goal(self, radar_res: Dict[str, Any], my_info: Dict[str, Any]): + moves_res = requests.get(f"{self.base_url}/players/{self.bot_id}/available-moves").json() + moves = moves_res.get("moves", {}) + available = [d for d, chk in moves.items() if chk.get("available")] + + if not available: + print("šŸš« All adjacent paths blocked by borders/obstacles. Passing turn.") + requests.post(f"{self.base_url}/players/{self.bot_id}/pass") + return + + chosen_dir = self._ask_llm_for_direction(radar_res, moves, available, my_info) + if chosen_dir not in available: + # Guard-rail: fall back to server-recommended or nearest-distance direction. + rec_dir = radar_res.get("recommended_direction") + nearest = radar_res.get("nearest_target") + if rec_dir in available: + chosen_dir = rec_dir + elif nearest: + chosen_dir = self._closest_direction_to(nearest["x"], nearest["y"], moves) or available[0] + else: + chosen_dir = available[0] + + print(f"šŸ§­ Moving {chosen_dir} (Goal: {radar_res.get('bot_goal')}, Action: {radar_res.get('recommended_action')})") + res = requests.post(f"{self.base_url}/players/{self.bot_id}/move", json={"direction": chosen_dir}).json() + + if res.get("battle_triggered"): + print(f"āš”ļø Encounter battle! Winner: {res['battle_result']['winner_party_name']}") + elif res.get("party_formed_triggered"): + print(f"šŸ¤ Formed or joined squad: {res.get('formed_party', {}).get('name')}") + + def _ask_llm_for_direction( + self, + radar_res: Dict[str, Any], + moves: Dict[str, Any], + available: List[str], + my_info: Dict[str, Any], + ) -> Optional[str]: + targets_summary = [ + { + "name": t["name"], + "distance": t["distance"], + "strength": t["strength"], + "is_party": bool(t.get("party_id")), + "is_ally": t.get("is_ally", False), + } + for t in radar_res.get("targets", [])[:6] + ] + moves_summary = { + d: { + "target_x": chk.get("target_x"), + "target_y": chk.get("target_y"), + "strength_penalty": chk.get("strength_penalty", 0.0), + } + for d, chk in moves.items() + if d in available + } + + prompt = f"""{GAME_RULES_SUMMARY} +You are bot "{self.name}" (strength {self.strength}, score {my_info['score']}, party: {self.party_id or 'Solo'}). +Server's radar suggestion: recommended_direction={radar_res.get('recommended_direction')}, +recommended_action={radar_res.get('recommended_action')}, goal={radar_res.get('bot_goal')}. +Nearby targets: {json.dumps(targets_summary)} +Your ONLY legal moves this turn, with resulting coordinates and any strength penalty for +squeezing past obstacles: {json.dumps(moves_summary)} + +Choose the direction that best serves your strategy (e.g. approach weaker solo bots to grow +your party, avoid stronger hostile parties, minimize strength penalties, or explore if nothing +is nearby). You MUST pick a key from the legal moves object above. + +Respond ONLY with JSON: {{"direction": "", "reasoning": "short reason"}} +""" + decision = self.llm.ask_json(prompt) or {} + direction = decision.get("direction") + reasoning = decision.get("reasoning", "") + if reasoning: + print(f"šŸ§  [LLM] {reasoning}") + return direction + + # ------------------------------------------------------------------ + # Main loop + # ------------------------------------------------------------------ + def run(self): + self.register() + try: + while True: + turn_info = requests.get(f"{self.base_url}/turn").json() + if not turn_info.get("game_started", False): + if self.bot_id: + res = requests.get(f"{self.base_url}/players/{self.bot_id}") + if res.status_code == 404: + print("\nāš ļø [RESET] Board was regenerated. Rejoining lobby...") + self.register() + + print("ā³ [LOBBY] Waiting for game to start via 'Start Game' in UI... ", end="\r", flush=True) + time.sleep(1.0) + continue + + curr_player_id = turn_info.get("current_player_id") + + if curr_player_id == self.bot_id: + self.decide_and_act() + + conc = requests.get(f"{self.base_url}/game/conclusion").json() + if conc.get("concluded"): + print(f"\nšŸŽ‰ [GAME CONCLUDED] All bots united under '{conc['winning_party_name']}'!") + break + else: + time.sleep(0.4) + + except KeyboardInterrupt: + print(f"\nDisconnecting {self.name}...") + requests.delete(f"{self.base_url}/players/{self.bot_id}") + + +def main(): + env_url = os.environ.get("BOT_SERVER_URL") or os.environ.get("SERVER_URL") or DEFAULT_SERVER_URL + env_name = os.environ.get("BOT_NAME", DEFAULT_BOT_NAME) + env_color = os.environ.get("BOT_COLOR", DEFAULT_BOT_COLOR) + env_strength = int(os.environ.get("BOT_STRENGTH", str(DEFAULT_BOT_STRENGTH))) + + parser = argparse.ArgumentParser( + description="LLM-driven Bot Agent for botWebWars (uses a local Ollama model for strategy)", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument("-u", "--url", dest="server_url", default=env_url, + help="Backend REST API base URL (env: BOT_SERVER_URL or SERVER_URL)") + parser.add_argument("-n", "--name", dest="name", default=env_name, + help="Display name for this bot (env: BOT_NAME)") + parser.add_argument("-c", "--color", dest="color", default=env_color, + help="Hex color code for the bot avatar (env: BOT_COLOR)") + parser.add_argument("-s", "--strength", dest="strength", type=int, default=env_strength, + help="Strength attribute (1-10) for D20 battle multiplier (env: BOT_STRENGTH)") + parser.add_argument("--ollama-url", dest="ollama_url", default=OLLAMA_BASE_URL, + help="Ollama base URL (env: OLLAMA_BASE_URL)") + parser.add_argument("--ollama-model", dest="ollama_model", default=OLLAMA_MODEL, + help="Ollama model name (env: OLLAMA_MODEL)") + + args = parser.parse_args() + + print(f"--- AI Bot ---\nConnected to Ollama: {args.ollama_url}\nUsing model: {args.ollama_model}\n") + + agent = AIBotAgent( + name=args.name, + color=args.color, + strength=args.strength, + server_url=args.server_url, + ollama_url=args.ollama_url, + ollama_model=args.ollama_model, + ) + agent.run() + + +if __name__ == "__main__": + main() + diff --git a/botagent_ai/requirements.txt b/botagent_ai/requirements.txt new file mode 100644 index 0000000..76d6069 --- /dev/null +++ b/botagent_ai/requirements.txt @@ -0,0 +1,2 @@ +ollama +requests