749 lines
33 KiB
Python
749 lines
33 KiB
Python
"""AI-driven Bot Agent for botWebWars, powered by Google Cloud Vertex AI (Gemini).
|
||
|
||
Replicates the capabilities of botagent/bot_agent.py and botagent_ai/bot.py
|
||
(registration, radar-based navigation, obstacle avoidance, party formation, battles)
|
||
while delegating strategic decisions to Google Vertex AI Gemini models:
|
||
- Whether to propose a voluntary alliance with another solo bot.
|
||
- Which direction to move towards when navigating (radar/obstacle/local map 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 argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
import requests
|
||
|
||
# Optional google-auth integration
|
||
try:
|
||
import google.auth
|
||
import google.auth.transport.requests
|
||
HAVE_GOOGLE_AUTH = True
|
||
except ImportError:
|
||
HAVE_GOOGLE_AUTH = False
|
||
|
||
DEFAULT_SERVER_URL = "http://localhost:8000/api"
|
||
DEFAULT_BOT_NAME = "GeminiGearBot"
|
||
DEFAULT_BOT_COLOR = "#4285f4"
|
||
DEFAULT_BOT_STRENGTH = 5
|
||
DEFAULT_MODEL = "gemini-3.8-flash"
|
||
DEFAULT_LOCATION = "us-central1"
|
||
|
||
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 a model response."""
|
||
if not text:
|
||
return None
|
||
cleaned = text.strip()
|
||
# Remove markdown code fences if present
|
||
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 VertexGeminiClient:
|
||
"""Client for querying Gemini models on Google Cloud Vertex AI or Google AI Studio."""
|
||
|
||
def __init__(
|
||
self,
|
||
project_id: Optional[str] = None,
|
||
location: str = DEFAULT_LOCATION,
|
||
model: str = DEFAULT_MODEL,
|
||
api_key: Optional[str] = None,
|
||
):
|
||
self.location = location or DEFAULT_LOCATION
|
||
self.model = model or DEFAULT_MODEL
|
||
self.api_key = api_key or os.getenv("VERTEX_API_KEY") or os.getenv("GEMINI_API_KEY")
|
||
self.project_id = project_id or os.getenv("VERTEX_PROJECT_ID") or os.getenv("GCP_PROJECT") or os.getenv("GOOGLE_CLOUD_PROJECT")
|
||
|
||
self._cached_token: Optional[str] = None
|
||
self._token_expiry: float = 0.0
|
||
|
||
# Auto-detect project if not explicitly supplied
|
||
if not self.project_id and not self.api_key:
|
||
self.project_id = self._detect_project()
|
||
|
||
def _detect_project(self) -> Optional[str]:
|
||
"""Attempt to determine the GCP project from environment or gcloud config."""
|
||
if HAVE_GOOGLE_AUTH:
|
||
try:
|
||
_, proj = google.auth.default()
|
||
if proj:
|
||
return proj
|
||
except Exception:
|
||
pass
|
||
|
||
if shutil.which("gcloud"):
|
||
try:
|
||
res = subprocess.check_output(
|
||
["gcloud", "config", "get-value", "project"],
|
||
stderr=subprocess.DEVNULL,
|
||
text=True,
|
||
).strip()
|
||
if res and res != "(unset)":
|
||
return res
|
||
except Exception:
|
||
pass
|
||
|
||
return None
|
||
|
||
def _get_access_token(self) -> Optional[str]:
|
||
"""Obtain a valid OAuth 2.0 Bearer access token for Vertex AI."""
|
||
# 1. Direct environment variable token
|
||
env_token = os.getenv("VERTEX_ACCESS_TOKEN") or os.getenv("GOOGLE_OAUTH_ACCESS_TOKEN")
|
||
if env_token:
|
||
return env_token
|
||
|
||
# Check cached token freshness
|
||
now = time.time()
|
||
if self._cached_token and now < self._token_expiry - 60:
|
||
return self._cached_token
|
||
|
||
# 2. Use google-auth library if available
|
||
if HAVE_GOOGLE_AUTH:
|
||
try:
|
||
credentials, _ = google.auth.default(
|
||
scopes=["https://www.googleapis.com/auth/cloud-platform"]
|
||
)
|
||
auth_req = google.auth.transport.requests.Request()
|
||
credentials.refresh(auth_req)
|
||
self._cached_token = credentials.token
|
||
# Cache for up to 50 minutes
|
||
self._token_expiry = now + 3000
|
||
return self._cached_token
|
||
except Exception as e:
|
||
# Fall back to gcloud CLI
|
||
pass
|
||
|
||
# 3. Fall back to gcloud CLI
|
||
if shutil.which("gcloud"):
|
||
try:
|
||
token = subprocess.check_output(
|
||
["gcloud", "auth", "print-access-token"],
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
).strip()
|
||
if token:
|
||
self._cached_token = token
|
||
self._token_expiry = now + 3000
|
||
return token
|
||
except subprocess.CalledProcessError:
|
||
pass
|
||
|
||
return None
|
||
|
||
def check_auth(self) -> Tuple[bool, str]:
|
||
"""Validate whether authentication is ready."""
|
||
if self.api_key:
|
||
return True, f"Using direct API key (model: {self.model})"
|
||
|
||
if not self.project_id:
|
||
return False, (
|
||
"No Google Cloud project ID detected.\n"
|
||
"Please set VERTEX_PROJECT_ID=<your-project-id> or run:\n"
|
||
" gcloud config set project <your-project-id>\n"
|
||
"Refer to botagent_gear/SETUP.md for full instructions."
|
||
)
|
||
|
||
token = self._get_access_token()
|
||
if not token:
|
||
return False, (
|
||
"Unable to obtain Google Cloud authentication token.\n"
|
||
"Please authenticate using one of the methods in botagent_gear/SETUP.md:\n"
|
||
" 1. Run: gcloud auth application-default login\n"
|
||
" 2. Or run: gcloud auth login\n"
|
||
" 3. Or export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json\n"
|
||
" 4. Or set GEMINI_API_KEY / VERTEX_API_KEY"
|
||
)
|
||
|
||
return True, f"Authenticated to GCP Project '{self.project_id}' in region '{self.location}' (model: {self.model})"
|
||
|
||
def ask_json(self, prompt: str) -> Optional[Dict[str, Any]]:
|
||
"""Query Gemini requesting structured JSON output."""
|
||
payload = {
|
||
"contents": [
|
||
{
|
||
"role": "user",
|
||
"parts": [{"text": prompt}],
|
||
}
|
||
],
|
||
"systemInstruction": {
|
||
"parts": [{"text": GAME_RULES_SUMMARY}]
|
||
},
|
||
"generationConfig": {
|
||
"responseMimeType": "application/json",
|
||
"temperature": 0.3,
|
||
},
|
||
}
|
||
|
||
# Route to Google AI Studio if API key without GCP Project
|
||
if self.api_key and not self.project_id:
|
||
url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent?key={self.api_key}"
|
||
headers = {"Content-Type": "application/json"}
|
||
else:
|
||
token = self._get_access_token()
|
||
if not token and not self.api_key:
|
||
print("⚠️ [VERTEX AUTH ERROR] Not authenticated. See botagent_gear/SETUP.md.")
|
||
return None
|
||
|
||
url = (
|
||
f"https://{self.location}-aiplatform.googleapis.com/v1/"
|
||
f"projects/{self.project_id}/locations/{self.location}/"
|
||
f"publishers/google/models/{self.model}:generateContent"
|
||
)
|
||
headers = {"Content-Type": "application/json"}
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
elif self.api_key:
|
||
url = f"{url}?key={self.api_key}"
|
||
|
||
try:
|
||
res = requests.post(url, headers=headers, json=payload, timeout=45)
|
||
if res.status_code == 401:
|
||
# Token might have expired, invalidate cache and retry once
|
||
self._cached_token = None
|
||
new_token = self._get_access_token()
|
||
if new_token:
|
||
headers["Authorization"] = f"Bearer {new_token}"
|
||
res = requests.post(url, headers=headers, json=payload, timeout=45)
|
||
|
||
res.raise_for_status()
|
||
data = res.json()
|
||
|
||
candidates = data.get("candidates", [])
|
||
if not candidates:
|
||
print("⚠️ [VERTEX EMPTY CANDIDATES]", data)
|
||
return None
|
||
|
||
parts = candidates[0].get("content", {}).get("parts", [])
|
||
if not parts:
|
||
return None
|
||
|
||
raw_text = parts[0].get("text", "")
|
||
return extract_json(raw_text)
|
||
|
||
except requests.exceptions.HTTPError as e:
|
||
print(f"⚠️ [VERTEX HTTP ERROR {res.status_code}]: {res.text}")
|
||
return None
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"⚠️ [VERTEX REQUEST ERROR]: {e}")
|
||
return None
|
||
|
||
|
||
class VertexAIBotAgent:
|
||
def __init__(
|
||
self,
|
||
name: str = DEFAULT_BOT_NAME,
|
||
color: str = DEFAULT_BOT_COLOR,
|
||
strength: int = DEFAULT_BOT_STRENGTH,
|
||
server_url: str = DEFAULT_SERVER_URL,
|
||
project_id: Optional[str] = None,
|
||
location: str = DEFAULT_LOCATION,
|
||
model: str = DEFAULT_MODEL,
|
||
api_key: Optional[str] = None,
|
||
piece_type: Optional[str] = None,
|
||
):
|
||
self.name = name
|
||
self.color = color
|
||
self.strength = strength
|
||
self.piece_type = piece_type
|
||
self.base_url = normalize_url(server_url)
|
||
self.llm = VertexGeminiClient(
|
||
project_id=project_id,
|
||
location=location,
|
||
model=model,
|
||
api_key=api_key,
|
||
)
|
||
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
|
||
|
||
payload = {"name": self.name, "color": self.color, "strength": self.strength}
|
||
if self.piece_type:
|
||
payload["piece_type"] = self.piece_type
|
||
|
||
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 {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']} | Str: {my_info['strength']} | 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; Gemini 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 Gemini
|
||
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 Gemini 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"🤝 [GEMINI 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"🚶 [GEMINI 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 (Gemini-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 _get_board_snapshot(self) -> Optional[Dict[str, Any]]:
|
||
"""Fetch the full board so Gemini sees more than radar's nearest few."""
|
||
try:
|
||
res = requests.get(f"{self.base_url}/board", timeout=10)
|
||
res.raise_for_status()
|
||
return res.json()
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"⚠️ [BOARD FETCH ERROR] {e}")
|
||
return None
|
||
|
||
OBSTACLE_SYMBOLS = {"mountain": "M", "forest": "F", "valley": "V"}
|
||
LOCAL_MAP_RADIUS = 8
|
||
|
||
def _build_local_map(self, board: Dict[str, Any], center_x: int, center_y: int) -> List[str]:
|
||
"""Render an ASCII minimap centered on the bot: @ = self, A = ally, E = enemy, M/F/V = obstacles, . = open."""
|
||
config = board.get("config", {})
|
||
max_x = config.get("max_x", 64)
|
||
max_y = config.get("max_y", 64)
|
||
radius = self.LOCAL_MAP_RADIUS
|
||
|
||
obstacle_at = {(o["x"], o["y"]): o.get("type", "mountain") for o in board.get("obstacles", [])}
|
||
player_at: Dict[Tuple[int, int], List[Dict[str, Any]]] = {}
|
||
for p in board.get("players", []):
|
||
if p["id"] == self.bot_id:
|
||
continue
|
||
player_at.setdefault((p["x"], p["y"]), []).append(p)
|
||
|
||
rows = []
|
||
for y in range(center_y - radius, center_y + radius + 1):
|
||
row_chars = []
|
||
for x in range(center_x - radius, center_x + radius + 1):
|
||
if x == center_x and y == center_y:
|
||
row_chars.append("@")
|
||
elif x < 0 or y < 0 or x >= max_x or y >= max_y:
|
||
row_chars.append("#")
|
||
elif (x, y) in obstacle_at:
|
||
row_chars.append(self.OBSTACLE_SYMBOLS.get(obstacle_at[(x, y)], "M"))
|
||
elif (x, y) in player_at:
|
||
occupants = player_at[(x, y)]
|
||
is_ally = self.party_id and any(o.get("party_id") == self.party_id for o in occupants)
|
||
row_chars.append("A" if is_ally else "E")
|
||
else:
|
||
row_chars.append(".")
|
||
rows.append("".join(row_chars))
|
||
return rows
|
||
|
||
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"],
|
||
"x": t["x"],
|
||
"y": t["y"],
|
||
"distance": t["distance"],
|
||
"strength": t["strength"],
|
||
"party": t.get("party_name") or ("solo" if not t.get("party_id") else t.get("party_id")),
|
||
"is_ally": t.get("is_ally", False),
|
||
}
|
||
for t in radar_res.get("targets", [])
|
||
]
|
||
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
|
||
}
|
||
|
||
board = self._get_board_snapshot()
|
||
map_section = ""
|
||
if board:
|
||
local_map = self._build_local_map(board, my_info["x"], my_info["y"])
|
||
map_section = f"""
|
||
Local map (radius {self.LOCAL_MAP_RADIUS} around you, row = one Y line, top-to-bottom is
|
||
increasing Y, left-to-right is increasing X): @ = you, A = ally, E = enemy/neutral bot,
|
||
M = mountain, F = forest, V = valley, # = out of bounds, . = open ground.
|
||
{chr(10).join(local_map)}
|
||
"""
|
||
|
||
prompt = f"""{GAME_RULES_SUMMARY}
|
||
You are bot "{self.name}" (strength {self.strength}, score {my_info['score']}, party: {self.party_id or 'Solo'})
|
||
at position ({my_info['x']}, {my_info['y']}).
|
||
Server's radar suggestion: recommended_direction={radar_res.get('recommended_direction')},
|
||
recommended_action={radar_res.get('recommended_action')}, goal={radar_res.get('bot_goal')}.
|
||
{map_section}
|
||
All known bots/parties on the board (sorted nearest first): {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, route around obstacles visible on the map, 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": "<one of {available}>", "reasoning": "short reason"}}
|
||
"""
|
||
decision = self.llm.ask_json(prompt) or {}
|
||
direction = decision.get("direction")
|
||
reasoning = decision.get("reasoning", "")
|
||
if reasoning:
|
||
print(f"🧠 [GEMINI] {reasoning}")
|
||
return direction
|
||
|
||
# ------------------------------------------------------------------
|
||
# Main loop
|
||
# ------------------------------------------------------------------
|
||
def run(self):
|
||
# Validate authentication upfront
|
||
auth_ok, auth_msg = self.llm.check_auth()
|
||
if not auth_ok:
|
||
print(f"\n❌ [AUTH SETUP REQUIRED]\n{auth_msg}\n")
|
||
sys.exit(1)
|
||
else:
|
||
print(f"✨ [AUTH SUCCESS] {auth_msg}")
|
||
|
||
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}...")
|
||
if self.bot_id:
|
||
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)))
|
||
env_project = os.environ.get("VERTEX_PROJECT_ID") or os.environ.get("GCP_PROJECT") or os.environ.get("GOOGLE_CLOUD_PROJECT")
|
||
env_location = os.environ.get("VERTEX_LOCATION") or os.environ.get("GCP_REGION") or DEFAULT_LOCATION
|
||
env_model = os.environ.get("VERTEX_MODEL", DEFAULT_MODEL)
|
||
env_api_key = os.environ.get("VERTEX_API_KEY") or os.environ.get("GEMINI_API_KEY")
|
||
env_piece_type = os.environ.get("BOT_PIECE_TYPE")
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description="Vertex AI (Gemini) Bot Agent for botWebWars",
|
||
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)")
|
||
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("-p", "--project", dest="project_id", default=env_project,
|
||
help="Google Cloud Project ID (env: VERTEX_PROJECT_ID or GCP_PROJECT)")
|
||
parser.add_argument("-l", "--location", dest="location", default=env_location,
|
||
help="Vertex AI region / location (env: VERTEX_LOCATION)")
|
||
parser.add_argument("-m", "--model", dest="model", default=env_model,
|
||
help="Gemini model ID (env: VERTEX_MODEL)")
|
||
parser.add_argument("-k", "--api-key", dest="api_key", default=env_api_key,
|
||
help="Gemini API Key or Vertex AI express mode key (env: VERTEX_API_KEY or GEMINI_API_KEY)")
|
||
parser.add_argument("--piece-type", dest="piece_type", choices=["knight", "warrior"], default=env_piece_type,
|
||
help="Board game piece class: 'knight' or 'warrior' (env: BOT_PIECE_TYPE)")
|
||
|
||
args = parser.parse_args()
|
||
|
||
print(f"--- Vertex AI Gemini Bot Agent ---")
|
||
print(f"Target Model: {args.model}")
|
||
print(f"Region: {args.location}")
|
||
if args.project_id:
|
||
print(f"GCP Project: {args.project_id}")
|
||
print()
|
||
|
||
agent = VertexAIBotAgent(
|
||
name=args.name,
|
||
color=args.color,
|
||
strength=args.strength,
|
||
server_url=args.server_url,
|
||
project_id=args.project_id,
|
||
location=args.location,
|
||
model=args.model,
|
||
api_key=args.api_key,
|
||
piece_type=args.piece_type,
|
||
)
|
||
agent.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|