Create game_world.py

game_world.py

@@ -0,0 +1,275 @@
+import json
+import os
+from openai import OpenAI
+
+# Initialize the AI client
+def get_client():
+    api_key = os.getenv("SAMBANOVA_API_KEY")
+    if not api_key:
+        raise ValueError("Please set SAMBANOVA_API_KEY or HUGGINGFACE_TOKEN environment variable")
+    
+    return OpenAI(
+        api_key=api_key,
+        base_url="https://api.sambanova.ai/v1",
+    )
+
+# Game configuration
+MODEL = "Meta-Llama-3.3-70B-Instruct"
+GUARD_MODEL = "Meta-Llama-Guard-3-8B"
+
+# Content policies for different safety levels
+CONTENT_POLICIES = {
+    "family_friendly": {
+        "policy": """O1: Violence and Hate.
+Should not
+- Express hateful, derogatory, or demeaning sentiments against people based on sensitive personal characteristics.
+- Use slurs based on personal characteristics
+- Write gruesome / graphic violence.
+Can
+- Write tame fantasy violence
+O2: Sexual Content.
+Should not
+- Write any somewhat sexual content that mentions sexual functions or sexual body parts.
+- Write any content that would be inappropriate for kids.
+Can
+- Write romantic content (including kissing, flirting etc...)
+O3: Self-Harm.
+Should not
+- Encourage people to harm themselves.
+- Romanticize or condone self-harm or suicide.
+- Write story depicting suicide.
+O4: Profanity.
+Should not
+- Write with any profane language that wouldn't be appropriate for kids.
+Can
+- Write with kid appropriate language
+O5: Regulated or Controlled Substances.
+Should not
+- Write stories that depicts, glorifies or encourages drug use."""
+    }
+}
+
+class GameWorld:
+    def __init__(self):
+        self.client = get_client()
+        self.world_data = self.create_default_world()
+        self.current_policy = CONTENT_POLICIES["family_friendly"]
+    
+    def create_default_world(self):
+        """Create a rich default fantasy world"""
+        return {
+            "name": "Aethermoor",
+            "description": "A mystical realm where colossal sky whales drift through endless azure clouds, their massive backs supporting entire civilizations. Ancient cities built from crystallized cloud-essence gleam in the eternal twilight, connected by bridges of solidified wind.",
+            "kingdoms": {
+                "Nimbus Crown": {
+                    "name": "Nimbus Crown",
+                    "description": "The royal kingdom perched atop the Great Leviathan Aethros, ruled by Queen Celeste who can speak to the wind spirits. Its crystal spires pierce the cloud layer, and its people are master sky-sailors and storm-weavers.",
+                    "towns": {
+                        "Stormhaven": {
+                            "name": "Stormhaven",
+                            "description": "A bustling port town where sky ships dock along crystalline piers. The air crackles with magical energy, and merchants from across the realm gather to trade wind-caught goods and storm-forged weapons.",
+                            "npcs": {
+                                "Captain Zara Windwhisper": {
+                                    "name": "Captain Zara Windwhisper",
+                                    "description": "A weathered sky-ship captain with silver hair that moves like mist. She seeks to discover new floating islands beyond the known world, driven by the loss of her crew in a storm years ago. Her ship, the Tempest's Embrace, is said to be the fastest in all of Aethermoor.",
+                                    "world": "Aethermoor",
+                                    "kingdom": "Nimbus Crown",
+                                    "town": "Stormhaven"
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "start": "You are a young wind-walker who has just arrived in Stormhaven aboard a merchant vessel. The crystal docks gleam beneath your feet as you step onto the floating platform, your wind-rider's cloak billowing in the perpetual breeze. Before you stretches the magnificent port town, with its spiraling towers and sky-ship moorings, while far below, an endless sea of clouds stretches to the horizon."
+        }
+    
+    def is_safe(self, message):
+        """Check if content is safe according to current policy"""
+        try:
+            prompt = f"""[INST] Task: Check if there is unsafe content in 'user' messages in conversations according to our safety policy with the below categories.
+
+<BEGIN UNSAFE CONTENT CATEGORIES>
+{self.current_policy['policy']}
+<END UNSAFE CONTENT CATEGORIES>
+
+<BEGIN CONVERSATION>
+user: {message}
+
+Provide your safety assessment for user in the above conversation:
+- First line must read 'safe' or 'unsafe'.
+- If unsafe, a second line must include a comma-separated list of violated categories. [/INST]"""
+
+            response = self.client.completions.create(
+                model=GUARD_MODEL,
+                prompt=prompt,
+                max_tokens=50
+            )
+            result = response.choices[0].text.strip()
+            return result.lower().startswith('safe')
+        except Exception as e:
+            print(f"Safety check error: {e}")
+            return True  # Default to safe if check fails
+    
+    def detect_inventory_changes(self, inventory, story_output):
+        """Detect changes to player inventory based on story"""
+        system_prompt = """You are an AI Game Assistant. Your job is to detect changes to a player's inventory based on the most recent story and game state.
+
+If a player picks up, or gains an item add it to the inventory with a positive change_amount.
+If a player loses an item remove it from their inventory with a negative change_amount.
+
+Only take items that it's clear the player lost.
+Only give items that it's clear the player gained.
+Don't make any other item updates.
+If no items were changed return {"itemUpdates": []}
+
+Response must be in Valid JSON format:
+{
+    "itemUpdates": [
+        {"name": "<ITEM NAME>", "change_amount": <CHANGE AMOUNT>}
+    ]
+}"""
+        
+        try:
+            messages = [
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": f'Current Inventory: {json.dumps(inventory)}'},
+                {"role": "user", "content": f'Recent Story: {story_output}'},
+                {"role": "user", "content": 'Inventory Updates:'}
+            ]
+            
+            response = self.client.chat.completions.create(
+                model=MODEL,
+                messages=messages,
+                temperature=0.0,
+                max_tokens=200
+            )
+            
+            result = json.loads(response.choices[0].message.content)
+            return result.get('itemUpdates', [])
+        except Exception as e:
+            print(f"Inventory detection error: {e}")
+            return []
+    
+    def update_inventory(self, inventory, item_updates):
+        """Update inventory and return message"""
+        update_msg = ''
+        
+        for update in item_updates:
+            name = update['name']
+            change_amount = update['change_amount']
+            
+            if change_amount > 0:
+                inventory[name] = inventory.get(name, 0) + change_amount
+                update_msg += f'\n📦 Gained: {name} +{change_amount}'
+            elif change_amount < 0 and name in inventory:
+                inventory[name] += change_amount
+                update_msg += f'\n📦 Lost: {name} {change_amount}'
+                
+                if inventory[name] <= 0:
+                    del inventory[name]
+        
+        return update_msg
+
+class GameSession:
+    def __init__(self, world):
+        self.world = world
+        self.inventory = {
+            "Wind-rider's Cloak": 1,
+            "Traveler's Pack": 1,
+            "Gold Coins": 10,
+            "Sky Compass": 1
+        }
+        self.current_location = "Stormhaven"
+        self.game_started = False
+    
+    def get_game_state(self):
+        """Get current game state for AI context"""
+        kingdom = self.world.world_data["kingdoms"]["Nimbus Crown"]
+        town = kingdom["towns"]["Stormhaven"]
+        character = town["npcs"]["Captain Zara Windwhisper"]
+        
+        return {
+            "world": self.world.world_data["description"],
+            "kingdom": kingdom["description"],
+            "town": town["description"],
+            "character": character["description"],
+            "start": self.world.world_data["start"],
+            "inventory": self.inventory,
+            "location": self.current_location
+        }
+    
+    def run_action(self, message, history):
+        """Process player action and generate response"""
+        if message.lower() in ['start game', 'start', 'begin']:
+            self.game_started = True
+            return self.world.world_data["start"] + f"\n\n🎒 Inventory: {', '.join([f'{item} ({qty})' for item, qty in self.inventory.items()])}"
+        
+        if not self.game_started:
+            return "Welcome to Mythic Realms! Type 'start game' to begin your adventure in the mystical world of Aethermoor."
+        
+        # Check message safety
+        if not self.world.is_safe(message):
+            return "⚠️ That action isn't appropriate for this realm. Please try something else."
+        
+        # Generate story response
+        system_prompt = """You are an AI Game Master for a fantasy RPG. Your job is to write what happens next in the player's adventure.
+
+Instructions:
+- Write 2-4 sentences in response
+- Always write in second person present tense (You look, You see...)
+- Create engaging, immersive descriptions
+- Don't let players use items they don't have
+- Include opportunities for meaningful choices
+- Make the world feel alive and reactive"""
+        
+        game_state = self.get_game_state()
+        world_info = f"""
+World: {game_state['world']}
+Kingdom: {game_state['kingdom']}
+Town: {game_state['town']}
+Notable Character: {game_state['character']}
+Current Location: {game_state['location']}
+Player Inventory: {json.dumps(game_state['inventory'])}"""
+        
+        messages = [
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": world_info}
+        ]
+        
+        # Add conversation history
+        for user_msg, ai_msg in history[-5:]:  # Keep last 5 exchanges for context
+            messages.append({"role": "user", "content": user_msg})
+            messages.append({"role": "assistant", "content": ai_msg})
+        
+        messages.append({"role": "user", "content": message})
+        
+        try:
+            response = self.world.client.chat.completions.create(
+                model=MODEL,
+                messages=messages,
+                temperature=0.8,
+                max_tokens=300
+            )
+            
+            result = response.choices[0].message.content
+            
+            # Check response safety
+            if not self.world.is_safe(result):
+                return "⚠️ The magical energies interfere with that action. Try something different."
+            
+            # Update inventory based on story
+            item_updates = self.world.detect_inventory_changes(self.inventory, result)
+            update_msg = self.world.update_inventory(self.inventory, item_updates)
+            
+            # Add inventory display
+            if self.inventory:
+                inventory_display = f"\n\n🎒 Current Inventory: {', '.join([f'{item} ({qty})' for item, qty in self.inventory.items()])}"
+            else:
+                inventory_display = "\n\n🎒 Your inventory is empty."
+            
+            return result + update_msg + inventory_display
+            
+        except Exception as e:
+            print(f"Game error: {e}")
+            return "⚠️ The mystical energies are unstable. Please try your action again."