""" FreeExile Multi-LLM Provider Client Engine. Authoritative bridge for routing tactical prompts to OpenRouter, OpenAI (GPT), Anthropic (Claude), Google (Gemini), DeepSeek, Moonshot (Kimi), and Local Tactical Fallback. Adheres to strict async timeouts, schema parsing, and automatic zero-latency heuristic fallback. """ from enum import Enum from typing import Dict, Any, Optional from dataclasses import dataclass, field import json import logging import re import urllib.request import urllib.error logger = logging.getLogger("freeexile.agent.llm") class LLMProviderType(str, Enum): OPENROUTER = "OPENROUTER" OPENAI = "OPENAI" ANTHROPIC = "ANTHROPIC" GEMINI = "GEMINI" DEEPSEEK = "DEEPSEEK" KIMI = "KIMI" LOCAL_TACTICAL = "LOCAL_TACTICAL" class AgentActionType(str, Enum): IDLE = "IDLE" MOVE_TOWARDS = "MOVE_TOWARDS" MOVE_AWAY = "MOVE_AWAY" ATTACK_SKILL = "ATTACK_SKILL" DODGE = "DODGE" LOOT = "LOOT" USE_POTION = "USE_POTION" SWAP_WEAPON = "SWAP_WEAPON" @dataclass class LLMProviderConfig: provider: LLMProviderType api_key: str = "" model: str = "" base_url: str = "" temperature: float = 0.2 max_tokens: int = 256 timeout_sec: float = 2.0 def __post_init__(self) -> None: if not self.model: if self.provider == LLMProviderType.OPENROUTER: self.model = "meta-llama/llama-3.3-70b-instruct" elif self.provider == LLMProviderType.OPENAI: self.model = "gpt-4o-mini" elif self.provider == LLMProviderType.ANTHROPIC: self.model = "claude-3-5-haiku-20241022" elif self.provider == LLMProviderType.GEMINI: self.model = "gemini-1.5-flash" elif self.provider == LLMProviderType.DEEPSEEK: self.model = "deepseek-chat" elif self.provider == LLMProviderType.KIMI: self.model = "moonshot-v1-8k" elif self.provider == LLMProviderType.LOCAL_TACTICAL: self.model = "heuristic-reflex-v1" if not self.base_url: if self.provider == LLMProviderType.OPENROUTER: self.base_url = "https://openrouter.ai/api/v1/chat/completions" elif self.provider == LLMProviderType.OPENAI: self.base_url = "https://api.openai.com/v1/chat/completions" elif self.provider == LLMProviderType.ANTHROPIC: self.base_url = "https://api.anthropic.com/v1/messages" elif self.provider == LLMProviderType.GEMINI: self.base_url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent" elif self.provider == LLMProviderType.DEEPSEEK: self.base_url = "https://api.deepseek.com/chat/completions" elif self.provider == LLMProviderType.KIMI: self.base_url = "https://api.moonshot.cn/v1/chat/completions" elif self.provider == LLMProviderType.LOCAL_TACTICAL: self.base_url = "local://heuristic" @dataclass(slots=True, frozen=True) class TacticalActionDecision: action_type: AgentActionType target_id: Optional[int] = None skill_id: Optional[int] = None target_x: Optional[float] = None target_y: Optional[float] = None target_drop_id: Optional[str] = None reasoning: str = "" token_cost: int = 0 class LLMProviderClientRouter: """ Unified Router connecting to multiple Cloud LLMs with seamless fallback to local deterministic rules. """ def build_request_headers(self, config: LLMProviderConfig) -> Dict[str, str]: headers: Dict[str, str] = { "Content-Type": "application/json" } if config.provider == LLMProviderType.ANTHROPIC: headers["x-api-key"] = config.api_key headers["anthropic-version"] = "2023-06-01" elif config.provider == LLMProviderType.OPENROUTER: headers["Authorization"] = f"Bearer {config.api_key}" headers["HTTP-Referer"] = "https://freeexile.game" headers["X-Title"] = "FreeExile Dark Martial MMORPG" elif config.provider == LLMProviderType.GEMINI: if config.api_key: headers["x-goog-api-key"] = config.api_key else: if config.api_key: headers["Authorization"] = f"Bearer {config.api_key}" return headers def build_payload(self, config: LLMProviderConfig, prompt_data: Dict[str, Any]) -> Dict[str, Any]: system_prompt = ( "You are an elite tactical AI agent piloting a martial arts character in FreeExile 2.5D MMORPG. " "Analyze current perception and respond ONLY with a raw JSON object containing: " "action_type (ATTACK_SKILL | MOVE_TOWARDS | MOVE_AWAY | DODGE | LOOT | USE_POTION), " "target_id (int or null), skill_id (int or null), target_x (float or null), target_y (float or null), " "target_drop_id (string or null), reasoning (string explanation)." ) user_content = json.dumps(prompt_data, ensure_ascii=False) if config.provider == LLMProviderType.ANTHROPIC: return { "model": config.model, "max_tokens": config.max_tokens, "system": system_prompt, "messages": [{"role": "user", "content": user_content}], "temperature": config.temperature } elif config.provider == LLMProviderType.GEMINI: return { "contents": [{ "role": "user", "parts": [{"text": f"{system_prompt}\n\nCurrent Game State:\n{user_content}"}] }], "generationConfig": { "temperature": config.temperature, "maxOutputTokens": config.max_tokens, "responseMimeType": "application/json" } } else: # Standard OpenAI-compatible format (OpenAI, OpenRouter, DeepSeek, Kimi) return { "model": config.model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_content} ], "temperature": config.temperature, "max_tokens": config.max_tokens, "response_format": {"type": "json_object"} if config.provider in [LLMProviderType.OPENAI, LLMProviderType.DEEPSEEK] else None } def parse_response_to_decision(self, raw_response: str) -> TacticalActionDecision: """Parses LLM output into a validated TacticalActionDecision.""" try: # Extract JSON block if wrapped in markdown match = re.search(r"\{.*\}", raw_response, re.DOTALL) json_str = match.group(0) if match else raw_response data = json.loads(json_str) action_str = str(data.get("action_type", "IDLE")).upper() action_type = AgentActionType.IDLE try: action_type = AgentActionType[action_str] except KeyError: action_type = AgentActionType.IDLE return TacticalActionDecision( action_type=action_type, target_id=data.get("target_id"), skill_id=data.get("skill_id"), target_x=float(data["target_x"]) if data.get("target_x") is not None else None, target_y=float(data["target_y"]) if data.get("target_y") is not None else None, target_drop_id=data.get("target_drop_id"), reasoning=str(data.get("reasoning", "")), token_cost=int(data.get("token_cost", 0)) ) except Exception as e: logger.warning(f"Failed to parse LLM tactical decision: {e}. Fallback to IDLE.") return TacticalActionDecision( action_type=AgentActionType.IDLE, reasoning=f"Parsing error: {e}" ) def _heuristic_fallback(self, world_prompt: Dict[str, Any]) -> TacticalActionDecision: """ Instant 0ms tactical heuristic rule-based engine. Ensures player character survives even if network is completely severed. """ hp_pct = float(world_prompt.get("player_hp_pct", 1.0)) nearby_enemies = world_prompt.get("nearby_enemies", []) nearby_drops = world_prompt.get("nearby_drops", []) # Rule 1: Emergency survival (HP < 25%) if hp_pct < 0.25: if nearby_enemies: return TacticalActionDecision( action_type=AgentActionType.DODGE, reasoning="[Tactical Heuristic] Critical HP (<25%). Activating Huyễn Ảnh Bộ i-frame evasion." ) return TacticalActionDecision( action_type=AgentActionType.USE_POTION, reasoning="[Tactical Heuristic] Low HP (<25%). Consuming recovery elixir." ) # Rule 2: High-priority loot (Rare stones nearby and HP safe > 50%) if nearby_drops and hp_pct >= 0.50: best_drop = nearby_drops[0] return TacticalActionDecision( action_type=AgentActionType.LOOT, target_drop_id=str(best_drop.get("id")), reasoning=f"[Tactical Heuristic] Collecting rare ground loot: {best_drop.get('item')}." ) # Rule 3: Engage closest enemy if nearby_enemies: target = nearby_enemies[0] dist = float(target.get("distance", 10.0)) if dist > 3.0: return TacticalActionDecision( action_type=AgentActionType.MOVE_TOWARDS, target_id=int(target.get("id", 0)), reasoning="[Tactical Heuristic] Moving into melee attack range." ) else: return TacticalActionDecision( action_type=AgentActionType.ATTACK_SKILL, target_id=int(target.get("id", 0)), skill_id=1, reasoning="[Tactical Heuristic] Executing elemental counter attack skill." ) return TacticalActionDecision( action_type=AgentActionType.IDLE, reasoning="[Tactical Heuristic] No immediate threat or loot in vicinity. Patrolling." ) async def request_tactical_decision( self, config: LLMProviderConfig, world_prompt: Dict[str, Any] ) -> TacticalActionDecision: """ Submits request to configured provider. If LOCAL_TACTICAL or on network timeout/error, immediately resolves via _heuristic_fallback. """ if config.provider == LLMProviderType.LOCAL_TACTICAL: return self._heuristic_fallback(world_prompt) # If no API key provided, fallback to heuristic if not config.api_key: return self._heuristic_fallback(world_prompt) headers = self.build_request_headers(config) payload = self.build_payload(config, world_prompt) payload_bytes = json.dumps(payload).encode("utf-8") import asyncio def _do_http_call() -> str: req = urllib.request.Request( url=config.base_url, data=payload_bytes, headers=headers, method="POST" ) with urllib.request.urlopen(req, timeout=config.timeout_sec) as resp: body = resp.read().decode("utf-8") return body try: loop = asyncio.get_event_loop() raw_body = await loop.run_in_executor(None, _do_http_call) # Extract content from provider response envelope resp_data = json.loads(raw_body) content_text = "" if config.provider == LLMProviderType.ANTHROPIC: content_text = resp_data.get("content", [{}])[0].get("text", "") elif config.provider == LLMProviderType.GEMINI: candidates = resp_data.get("candidates", []) if candidates: parts = candidates[0].get("content", {}).get("parts", []) if parts: content_text = parts[0].get("text", "") else: choices = resp_data.get("choices", []) if choices: content_text = choices[0].get("message", {}).get("content", "") return self.parse_response_to_decision(content_text) except Exception as exc: logger.warning( f"LLM request to {config.provider} failed: {exc}. Activating autonomous heuristic fallback." ) return self._heuristic_fallback(world_prompt)