430 lines
18 KiB
Python
430 lines
18 KiB
Python
"""AI-driven Troll Agent for botWebWars, powered by an Ollama LLM.
|
||
|
||
Implements the unique rules and tactical gameplay of the Troll character type:
|
||
- Solitary hunter: Never joins or forms parties.
|
||
- Troll truce: Never battles other trolls.
|
||
- Relentless hunter: Mandatory tactical combat against players and squads.
|
||
- Strategic healing: Can take turns sleeping to regenerate +0.1 HP.
|
||
- Cannot challenge or duel Gary the Wizard.
|
||
- Strategic decision making via local/remote Ollama LLM:
|
||
1. Action Choice: Evaluates health and distance to decide whether to SLEEP (recover health) or HUNT (move).
|
||
2. Directional Navigation: Chooses the optimal passable path around obstacles towards target players.
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
import requests
|
||
|
||
DEFAULT_SERVER_URL = os.getenv("TROLL_SERVER_URL", os.getenv("BOT_SERVER_URL", "http://localhost:8000/api"))
|
||
DEFAULT_TROLL_NAME = os.getenv("TROLL_NAME", "GorgonAITroll")
|
||
DEFAULT_TROLL_COLOR = os.getenv("TROLL_COLOR", "#15803d")
|
||
DEFAULT_TROLL_STRENGTH = float(os.getenv("TROLL_STRENGTH", "3.0"))
|
||
DEFAULT_TROLL_HEALTH = float(os.getenv("TROLL_HEALTH", "10.0"))
|
||
|
||
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://192.168.1.220:11434")
|
||
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "gemma4:12b")
|
||
|
||
TROLL_RULES_SUMMARY = """
|
||
You are an autonomous tactical Troll in botWebWars!
|
||
Troll Rules of Engagement:
|
||
- You NEVER form alliances or parties with anyone. You are a solitary brute hunter.
|
||
- You NEVER fight other trolls. Trolls maintain an instinctive truce.
|
||
- Your sole mission is to hunt down human/bot players and player squads and crush them in 3-bout D20 battles.
|
||
- When victorious in battle, you earn +2 victory points (score). You do NOT gain strength or absorb squad members.
|
||
- If defeated in battle, you take 1 to 3 health points damage. At 0 HP, you die and a gravestone is placed.
|
||
- You CANNOT duel or interact with Gary the Wizard.
|
||
- SLEEP RESTORATION: On any turn when you are not in adjacent combat, you may choose to SLEEP. Sleeping skips movement but regenerates +0.1 HP!
|
||
- Squeezing diagonally between obstacle corners costs -0.1 strength penalty.
|
||
- The game concludes when all surviving entities are resolved. You can win the game on the final scoreboards!
|
||
"""
|
||
|
||
|
||
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]]:
|
||
"""Extract first valid JSON object from model output."""
|
||
if not text:
|
||
return None
|
||
cleaned = text.strip()
|
||
if cleaned.startswith("```"):
|
||
cleaned = re.sub(r"^```(?:json)?\s*", "", cleaned)
|
||
cleaned = re.sub(r"\s*```$", "", cleaned)
|
||
match = re.search(r"\{.*\}", cleaned, 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]]:
|
||
"""Query Ollama with a JSON formatting constraint."""
|
||
url = f"{self.base_url}/api/generate"
|
||
payload = {
|
||
"model": self.model,
|
||
"prompt": prompt,
|
||
"format": "json",
|
||
"stream": False,
|
||
"options": {"temperature": 0.3},
|
||
}
|
||
try:
|
||
res = requests.post(url, json=payload, timeout=45)
|
||
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 AITrollAgent:
|
||
def __init__(
|
||
self,
|
||
name: str = DEFAULT_TROLL_NAME,
|
||
color: str = DEFAULT_TROLL_COLOR,
|
||
strength: float = DEFAULT_TROLL_STRENGTH,
|
||
health: float = DEFAULT_TROLL_HEALTH,
|
||
server_url: str = DEFAULT_SERVER_URL,
|
||
ollama_url: str = OLLAMA_BASE_URL,
|
||
ollama_model: str = OLLAMA_MODEL,
|
||
loop_delay: float = 1.0,
|
||
):
|
||
self.name = name
|
||
self.color = color
|
||
self.strength = strength
|
||
self.health = health
|
||
self.base_url = normalize_url(server_url)
|
||
self.llm = OllamaClient(ollama_url, ollama_model)
|
||
self.loop_delay = loop_delay
|
||
self.bot_id: Optional[str] = None
|
||
|
||
def register(self):
|
||
"""Register the troll avatar on the grid or reconnect if existing."""
|
||
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 Troll {self.name} "
|
||
f"(ID: {self.bot_id}, Str: {p.get('strength', self.strength)}, "
|
||
f"HP: {p.get('health', self.health)}) at ({p.get('x')}, {p.get('y')})"
|
||
)
|
||
return
|
||
except Exception:
|
||
pass
|
||
|
||
payload = {
|
||
"name": self.name,
|
||
"color": self.color,
|
||
"strength": self.strength,
|
||
"health": self.health,
|
||
"piece_type": "troll",
|
||
"character_type": "troll",
|
||
}
|
||
|
||
res = requests.post(f"{self.base_url}/players", json=payload)
|
||
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 Troll {self.name} "
|
||
f"(ID: {self.bot_id}, HP: {p.get('health', self.health)}) at ({p.get('x')}, {p.get('y')})"
|
||
)
|
||
return
|
||
|
||
res.raise_for_status()
|
||
data = res.json()
|
||
self.bot_id = data["id"]
|
||
print(
|
||
f"👹 [REGISTER] Spawned AI Troll {self.name} (ID: {self.bot_id}, Str: {self.strength}, "
|
||
f"HP: {data.get('health', self.health)}) at ({data['x']}, {data['y']})"
|
||
)
|
||
|
||
def refresh_status(self) -> Optional[Dict[str, Any]]:
|
||
"""Get live troll state."""
|
||
try:
|
||
res = requests.get(f"{self.base_url}/players/{self.bot_id}")
|
||
if res.status_code == 200:
|
||
return res.json()
|
||
except Exception as e:
|
||
print(f"Status refresh error: {e}")
|
||
return None
|
||
|
||
def sleep_and_heal(self) -> bool:
|
||
"""Rest for 1 turn to regenerate +0.1 HP."""
|
||
try:
|
||
res = requests.post(f"{self.base_url}/players/{self.bot_id}/sleep")
|
||
if res.status_code == 200:
|
||
data = res.json()
|
||
print(
|
||
f"💤 [SLEEP] {self.name} rested for 1 turn. "
|
||
f"(+{data.get('health_gained', 0.1)} HP -> ❤️ {data.get('new_health')} HP)"
|
||
)
|
||
return True
|
||
else:
|
||
print(f"Sleep failed ({res.status_code}): {res.text}")
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
return False
|
||
except Exception as e:
|
||
print(f"Error sleeping: {e}")
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
return False
|
||
|
||
def pass_turn(self):
|
||
"""Pass turn when no actions are possible."""
|
||
try:
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
print(f"⏩ [PASS] {self.name} passed turn.")
|
||
except Exception as e:
|
||
print(f"Error passing turn: {e}")
|
||
|
||
def _find_adjacent_player(self, my_x: int, my_y: int) -> Optional[Dict[str, Any]]:
|
||
"""Detect living adjacent player or player squad (ignores other trolls)."""
|
||
try:
|
||
players: List[Dict[str, Any]] = requests.get(f"{self.base_url}/players").json()
|
||
for p in players:
|
||
if p.get("id") == self.bot_id:
|
||
continue
|
||
if p.get("character_type") == "troll":
|
||
continue
|
||
if p.get("is_alive") is False or p.get("health", 10) <= 0:
|
||
continue
|
||
chebyshev = max(abs(p["x"] - my_x), abs(p["y"] - my_y))
|
||
if chebyshev <= 1:
|
||
return p
|
||
except Exception as e:
|
||
print(f"Error checking adjacent players: {e}")
|
||
return None
|
||
|
||
def attack_player(self, defender: Dict[str, Any]):
|
||
"""Initiate mandatory 3-bout D20 battle against adjacent player."""
|
||
target_label = f"Squad '{defender.get('party_id')}'" if defender.get("party_id") else f"Player {defender.get('name')}"
|
||
print(f"⚔️ [BATTLE CLASH] Troll {self.name} ambushes {target_label} in 3-bout tactical D20 confrontation!")
|
||
try:
|
||
res = requests.post(
|
||
f"{self.base_url}/battles/fight",
|
||
json={"challenger_id": self.bot_id, "defender_id": defender["id"]},
|
||
)
|
||
if res.status_code == 200:
|
||
b = res.json()
|
||
won = b.get("winner_leader_name") == self.name or b.get("winner_party_name", "").startswith(self.name)
|
||
outcome = "VICTORY (+2 Victory Points!)" if won else "DEFEAT (HP damage taken)"
|
||
print(f"⚔️ [RESULT] {outcome}: {b.get('winner_party_name')} defeated {b.get('defeated_party_name')}")
|
||
for bout in b.get("bouts", []):
|
||
print(
|
||
f" Bout #{bout['bout_number']}: "
|
||
f"{bout['party1_name']} D20({bout['party1_roll']})×Str({bout['party1_strength']})={bout['party1_score']:.1f} vs "
|
||
f"{bout['party2_name']} D20({bout['party2_roll']})×Str({bout['party2_strength']})={bout['party2_score']:.1f} "
|
||
f"-> Winner: {bout['winner']}"
|
||
)
|
||
else:
|
||
print(f"Battle failed ({res.status_code}): {res.text}")
|
||
self.pass_turn()
|
||
except Exception as e:
|
||
print(f"Error fighting battle: {e}")
|
||
self.pass_turn()
|
||
|
||
def _decide_action_and_direction(
|
||
self,
|
||
my_info: Dict[str, Any],
|
||
radar_res: Dict[str, Any],
|
||
available_moves: Dict[str, Any],
|
||
) -> Tuple[str, Optional[str], str]:
|
||
"""Ask the LLM to decide whether to sleep or move, and if moving, which direction."""
|
||
my_hp = float(my_info.get("health", 10.0))
|
||
max_hp = float(my_info.get("max_health", 10.0))
|
||
nearest = radar_res.get("nearest_target")
|
||
rec_dir = radar_res.get("recommended_direction")
|
||
|
||
# Format available movement choices
|
||
choices_desc = []
|
||
for d, chk in available_moves.items():
|
||
if chk.get("available"):
|
||
pen = " [squeeze penalty -0.1 STR]" if chk.get("strength_penalty", 0.0) > 0 else ""
|
||
choices_desc.append(f"- {d}: Target ({chk.get('target_x')}, {chk.get('target_y')}){pen}")
|
||
|
||
choices_str = "\n".join(choices_desc) if choices_desc else "No open moves available (all blocked by obstacles/bounds)."
|
||
|
||
prompt = f"""{TROLL_RULES_SUMMARY}
|
||
Current Troll Status:
|
||
- Name: "{self.name}" | Current HP: {my_hp:.1f} / {max_hp:.1f} | Strength: {self.strength:.1f} | Score: {my_info['score']}
|
||
- Position: ({my_info['x']}, {my_info['y']})
|
||
- Nearest Target Player: {nearest.get('name') if nearest else 'None detected'} (Distance: {nearest.get('distance') if nearest else 'N/A'}, Pos: {nearest.get('x') if nearest else '?'},{nearest.get('y') if nearest else '?'})
|
||
- Radar Pathfinder Suggestion: {rec_dir or 'None'}
|
||
|
||
Available Passable Moves:
|
||
{choices_str}
|
||
|
||
OPTIONS:
|
||
1. "sleep" - Take a rest turn to regenerate +0.1 HP. (Recommended if wounded or trapped).
|
||
2. "move" - Move in one of the passable directions toward the nearest player.
|
||
|
||
Decide which action to take. If action is "move", pick the best passable direction from the list.
|
||
Respond ONLY with a JSON object:
|
||
{{"action": "sleep"|"move", "direction": "UP"|"DOWN"|"LEFT"|"RIGHT"|"UP_LEFT"|"UP_RIGHT"|"DOWN_LEFT"|"DOWN_RIGHT"|null, "reasoning": "short tactical rationale"}}
|
||
"""
|
||
decision = self.llm.ask_json(prompt) or {}
|
||
action = str(decision.get("action", "move")).lower().strip()
|
||
direction = decision.get("direction")
|
||
reasoning = decision.get("reasoning", "")
|
||
|
||
# Fallback validation
|
||
if action not in ("sleep", "move"):
|
||
action = "sleep" if my_hp < 6.0 and my_hp < max_hp else "move"
|
||
|
||
valid_dirs = [d for d, chk in available_moves.items() if chk.get("available")]
|
||
if action == "move":
|
||
if direction not in valid_dirs:
|
||
direction = rec_dir if rec_dir in valid_dirs else (valid_dirs[0] if valid_dirs else None)
|
||
if not direction:
|
||
action = "sleep" if my_hp < max_hp else "pass"
|
||
|
||
return action, direction, reasoning
|
||
|
||
def decide_and_act(self):
|
||
"""Turn execution pipeline: combat check -> LLM reasoning -> act."""
|
||
my_info = self.refresh_status()
|
||
if not my_info:
|
||
self.pass_turn()
|
||
return
|
||
|
||
my_x = my_info.get("x", 0)
|
||
my_y = my_info.get("y", 0)
|
||
|
||
# 1. Immediate combat check: Trolls always attack adjacent players
|
||
adjacent_player = self._find_adjacent_player(my_x, my_y)
|
||
if adjacent_player:
|
||
self.attack_player(adjacent_player)
|
||
return
|
||
|
||
# 2. Get sensor data
|
||
try:
|
||
radar_res = requests.get(f"{self.base_url}/players/{self.bot_id}/radar").json()
|
||
except Exception:
|
||
radar_res = {}
|
||
|
||
try:
|
||
moves_res = requests.get(f"{self.base_url}/players/{self.bot_id}/available-moves").json()
|
||
available_moves = moves_res.get("moves", {})
|
||
except Exception:
|
||
available_moves = {}
|
||
|
||
# 3. LLM strategic decision
|
||
action, direction, reasoning = self._decide_action_and_direction(my_info, radar_res, available_moves)
|
||
print(f"🧠 [LLM STRATEGY] Action: {action.upper()}{f' -> {direction}' if direction else ''}. Reasoning: {reasoning}")
|
||
|
||
if action == "sleep":
|
||
self.sleep_and_heal()
|
||
elif action == "move" and direction:
|
||
try:
|
||
res = requests.post(
|
||
f"{self.base_url}/players/{self.bot_id}/move",
|
||
json={"direction": direction},
|
||
).json()
|
||
if res.get("battle_triggered") and res.get("battle_result"):
|
||
b = res["battle_result"]
|
||
print(f"⚔️ Move clash! Winner: {b.get('winner_party_name')} (Defeated: {b.get('defeated_party_name')})")
|
||
except Exception as e:
|
||
print(f"Move failed: {e}")
|
||
self.pass_turn()
|
||
else:
|
||
self.pass_turn()
|
||
|
||
def run(self):
|
||
"""Main game loop."""
|
||
self.register()
|
||
try:
|
||
while True:
|
||
# Check life status
|
||
my_status = self.refresh_status()
|
||
if my_status and (my_status.get("is_alive") is False or my_status.get("health", 10) <= 0):
|
||
print(f"\n🪦 [FALLEN TROLL] {self.name} has fallen (0 HP)! Gravestone marked on board.")
|
||
print(f"Final Score: {my_status.get('score', 0)} pts preserved on scoreboard. Spectating...")
|
||
while True:
|
||
try:
|
||
conc = requests.get(f"{self.base_url}/game/conclusion").json()
|
||
if conc.get("concluded"):
|
||
print(f"\n🎉 [GAME CONCLUDED] Arena concluded! Winner: '{conc.get('winning_party_name')}'")
|
||
return
|
||
except Exception:
|
||
pass
|
||
time.sleep(2.0)
|
||
|
||
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 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:
|
||
print(f"\n⚡ [MY TURN] AI Troll {self.name} (Round {turn_info.get('round_number')}, Turn {turn_info.get('turn_number')})")
|
||
self.decide_and_act()
|
||
|
||
conc = requests.get(f"{self.base_url}/game/conclusion").json()
|
||
if conc.get("concluded"):
|
||
print(f"\n🎉 [GAME CONCLUDED] Arena concluded! Winner: '{conc.get('winning_party_name')}'!")
|
||
break
|
||
|
||
time.sleep(self.loop_delay)
|
||
|
||
except KeyboardInterrupt:
|
||
print(f"\n👋 AI Troll {self.name} disconnecting gracefully.")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Ollama LLM-Assisted Autonomous Troll Agent for botWebWars")
|
||
parser.add_argument("-u", "--url", default=DEFAULT_SERVER_URL, help="botWebWars API base URL")
|
||
parser.add_argument("-n", "--name", default=DEFAULT_TROLL_NAME, help="Troll name")
|
||
parser.add_argument("-c", "--color", default=DEFAULT_TROLL_COLOR, help="Troll color hex code")
|
||
parser.add_argument("-s", "--strength", type=float, default=DEFAULT_TROLL_STRENGTH, help="Starting strength (1-10)")
|
||
parser.add_argument("-H", "--health", type=float, default=DEFAULT_TROLL_HEALTH, help="Starting health points (default: 10.0)")
|
||
parser.add_argument("--ollama-url", default=OLLAMA_BASE_URL, help="Ollama API base URL")
|
||
parser.add_argument("--ollama-model", default=OLLAMA_MODEL, help="Ollama model name (default: gemma4:12b)")
|
||
parser.add_argument("--loop-delay", type=float, default=1.0, help="Polling interval in seconds")
|
||
|
||
args = parser.parse_args()
|
||
|
||
agent = AITrollAgent(
|
||
name=args.name,
|
||
color=args.color,
|
||
strength=args.strength,
|
||
health=args.health,
|
||
server_url=args.url,
|
||
ollama_url=args.ollama_url,
|
||
ollama_model=args.ollama_model,
|
||
loop_delay=args.loop_delay,
|
||
)
|
||
agent.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|