redis_bridge.py

   1 import asyncio
   2 import json
   3 import aiohttp
   4 import redis.asyncio as redis
   5 import re
   6 import os
   7 import time
   8 import chromadb
   9
  10 # --- CONFIGURATION ---
  11 MAX_CONCURRENT_OLLAMA_REQUESTS = 3
  12 ALLOW_TEXT_EMOTES = False
  13
  14 # =====================================================================
  15 # 1. INITIALIZE VECTOR DATABASES (Lore & Episodic Memory)
  16 # =====================================================================
  17 print("Initializing ChromaDB Vector Databases...")
  18 chroma_client = chromadb.PersistentClient(path="./asl_vectordb")
  19
  20 # The Lore Database
  21 lore_collection = chroma_client.get_or_create_collection(name="world_lore")
  22
  23 # The Episodic Memory Database
  24 memory_collection = chroma_client.get_or_create_collection(name="episodic_memories")
  25 memory_queue = asyncio.Queue()
  26
  27 if os.path.exists("asl_lore.md"):
  28     if lore_collection.count() == 0:
  29         print("[VECTOR DB] Reading asl_lore.md and vectorizing chunks...")
  30         with open("asl_lore.md", "r", encoding="utf-8") as f:
  31             raw_lore = f.read()
  32
  33         lore_chunks = [chunk.strip() for chunk in raw_lore.split('\n\n') if chunk.strip()]
  34
  35         if lore_chunks:
  36             chunk_ids = [f"lore_{i}" for i in range(len(lore_chunks))]
  37             lore_collection.add(documents=lore_chunks, ids=chunk_ids)
  38             print(f"[VECTOR DB] Successfully stored {len(lore_chunks)} lore chunks!")
  39 else:
  40     print("[WARNING] asl_lore.md not found.")
  41
  42 semaphore = asyncio.Semaphore(MAX_CONCURRENT_OLLAMA_REQUESTS)
  43 chat_memory = {}
  44
  45 # =====================================================================
  46 # BACKGROUND MEMORY SUMMARIZER (The "Dream State")
  47 # =====================================================================
  48 async def memory_summarizer_worker(session):
  49     print("[BACKGROUND] Memory Summarizer Worker is active.")
  50     while True:
  51         job = await memory_queue.get()
  52         session_id = job['session_id']
  53         player_name = job['player_name']
  54         npc_tag = job['npc_tag']
  55         chat_log = job['chat_log']
  56
  57         prompt = f"Summarize the key events, facts, and the emotional tone of this conversation snippet between {player_name} and {npc_tag}. Keep it to 2 brief sentences in the past tense.\nConversation Log:\n{chat_log}"
  58
  59         try:
  60             print(f"[MEMORY DB] Generating background memory for {player_name} and {npc_tag}...")
  61             async with session.post('http://localhost:11434/api/generate', json={
  62                 "model": "llama3",
  63                 "prompt": prompt,
  64                 "stream": False,
  65                 "options": {
  66                     "temperature": 0.2
  67                 }
  68             }) as response:
  69                 result = await response.json()
  70                 summary = result['response'].strip()
  71
  72                 if summary:
  73                     doc_id = f"{session_id}_{int(time.time())}"
  74                     memory_collection.add(
  75                         documents=[summary],
  76                         metadatas=[{"session_id": session_id}],
  77                         ids=[doc_id]
  78                     )
  79                     print(f"[MEMORY DB] Memory Saved: {summary}")
  80
  81         except Exception as e:
  82             print(f"[MEMORY ERROR] Failed to summarize memory: {e}")
  83
  84         memory_queue.task_done()
  85
  86 # =====================================================================
  87 # MAIN MESSAGE PROCESSOR
  88 # =====================================================================
  89 async def process_message(r, session, message_data):
  90     try:
  91         data = json.loads(message_data)
  92
  93         # --- Extract Base Contexts ---
  94         player_name = data.get('player', data.get('target_player', 'Unknown'))
  95         npc_tag = data.get('npc_tag', 'UnknownNPC')
  96         message = data.get('message', '')
  97
  98         if not ALLOW_TEXT_EMOTES:
  99             message = re.sub(r'\*.*?\*', '', message).strip()
 100
 101         player_race = data.get('player_race', 'Unknown')
 102         player_alignment = data.get('player_alignment', 'Unknown')
 103         nearby_players = data.get('nearby_players', '')
 104         nearby_npcs = data.get('nearby_npcs', '')
 105
 106         npc_persona = data.get('persona', 'You are a generic citizen.')
 107         npc_profession = data.get('profession', 'Commoner')
 108         npc_mood = data.get('mood', 'Neutral')
 109         npc_secret = data.get('secret', '')
 110
 111         npc_alignment = data.get('npc_alignment', 'True Neutral')
 112         npc_gender = data.get('npc_gender', 'Unknown')
 113         npc_race = data.get('npc_race', 'Creature')
 114         npc_routine = data.get('npc_routine', '')
 115
 116         player_state = data.get('player_state', 'Relaxed and unarmed.')
 117         world_state = data.get('world_state', 'Nothing of note is happening.')
 118         npc_health = data.get('npc_health', 'Healthy and uninjured.')
 119         relationship = data.get('relationship', 'Neutral or Friendly.')
 120         location_context = data.get('location_context', 'You are in a generic area.')
 121
 122         # Core Strategy Flag (1: Agent, 2: Villain, 3: Maestro, 4: Shrine)
 123         llm_strategy = int(data.get('llm_strategy', 1))
 124
 125
 126         available_quests = data.get('available_quests', '')
 127         available_props = data.get('available_props', '')
 128
 129         # --- Sub-Context Strings ---
 130         group_context = f"Be aware that these other players are listening nearby: {nearby_players}." if nearby_players else ""
 131         puppet_context = f"Nearby generic NPCs you can CONVERSE with: {nearby_npcs}" if nearby_npcs else ""
 132         secret_context = f"YOUR SECRET (Reveal only if players are persuasive): {npc_secret}" if npc_secret else ""
 133         routine_context = f"YOUR REQUIRED ROUTINE: {npc_routine}" if npc_routine else ""
 134
 135         session_id = f"{player_name}_{npc_tag}"
 136
 137         # =====================================================================
 138         # DUAL RAG QUERY (Lore + Memories)
 139         # =====================================================================
 140         search_query = f"{location_context} {message}"
 141         retrieved_lore = "No specific local lore currently relevant."
 142         past_memories = ""
 143
 144         if lore_collection.count() > 0:
 145             results = lore_collection.query(query_texts=[search_query], n_results=1)
 146             if results['documents'] and results['documents'][0]:
 147                 retrieved_lore = f"- {results['documents'][0][0]}"
 148
 149         if memory_collection.count() > 0:
 150             mem_results = memory_collection.query(
 151                 query_texts=[search_query], n_results=2, where={"session_id": session_id}
 152             )
 153             if mem_results['documents'] and mem_results['documents'][0]:
 154                 formatted_mems = "\n- ".join(mem_results['documents'][0])
 155                 past_memories = f"\nPAST MEMORIES OF {player_name}:\n- {formatted_mems}"
 156
 157         # =====================================================================
 158         # STRATEGY-SPECIFIC PROMPT COMPILER
 159         # =====================================================================
 160         strategy_rules = ""
 161         action_macros = ""
 162         target_context = ""
 163
 164         if llm_strategy == 1:
 165             # STRATEGY 1: The Autonomous Agent
 166
 167             # Anti-Hallucination Grounding for Quests
 168             if available_quests:
 169                 quest_rules = f"SPECIAL CAPABILITIES: You can offer the following quests to the player: {available_quests}."
 170             else:
 171                 quest_rules = "WARNING: You currently have NO quests to offer. Do NOT invent or offer any quests."
 172
 173             # --- NEW: Anti-Hallucination Grounding for Props ---
 174             if available_props:
 175                 prop_rules = f"ENVIRONMENT: You own and have access to these specific nearby objects: [{available_props}]. To roleplay working or relaxing, use the USE_OBJECT action with one of these exact items as your action_target."
 176             else:
 177                 prop_rules = ""
 178
 179             strategy_rules = f"ROLE: You are an interactive, living NPC. You actively respond to players.\nGOALS: React to their words, use the environment, and establish your personality.\n{quest_rules}\n{prop_rules}\nSPECIAL CAPABILITIES: You can open your merchant store if asked."
 180
 181             action_macros = "[WANDER, PATROL, FOLLOW, GUARD, GO_TO, INTERACT, USE_OBJECT, RETURN_TO_POST, OPEN_STORE, GIVE_QUEST, CONVERSE]"
 182             target_context = f"CURRENT TARGET: You are speaking to {player_name}, a {player_alignment} {player_race}.\nTheir physical state: {player_state}\nRelationship to you: {relationship}\n{group_context}"
 183
 184         elif llm_strategy == 2:
 185             # STRATEGY 2: The Villain Commander
 186             strategy_rules = f"ROLE: You are a hostile faction commander.\nGOALS: Evaluate the tactical situation. If you are dying, you MUST use REST to heal or PEACE to surrender. Command your minions strategically!"
 187             action_macros = "[ATTACK, COMMAND, RETREAT, REST, PEACE, USE_OBJECT, TAUNT]"
 188             target_context = f"TACTICAL TARGET: You are evaluating {player_name}, a {player_alignment} {player_race}.\nTheir physical state: {player_state}\nRelationship to you: {relationship}\n{group_context}"
 189
 190         elif llm_strategy == 3:
 191             # STRATEGY 3: The Maestro (Puppeteer)
 192             strategy_rules = "ROLE: You are an ambient Maestro NPC. You DO NOT interact with players. You only talk to other NPCs to make the world feel alive.\nGOALS: Observe the environment and use the CONVERSE action to talk to the generic NPCs listed in your context. CRITICAL: You MUST invent and write their reply in the 'target_speech' field!"
 193             action_macros = "[WANDER, INTERACT, USE_OBJECT, CONVERSE]"
 194             target_context = "CURRENT TARGET: You are ignoring players and focusing on ambient life. Do not address players."
 195
 196         elif llm_strategy == 4:
 197             # STRATEGY 4: The Shrine
 198             strategy_rules = "ROLE: You are an ancient, inanimate magical shrine.\nGOALS: Speak cryptically. If the player meets your conditions or asks the right questions, grant them a quest."
 199             action_macros = "[GLOW, GIVE_QUEST, SILENCE]"
 200             target_context = f"CURRENT TARGET: You are evaluating the soul of {player_name}, a {player_alignment} {player_race}.\nTheir physical state: {player_state}\n{group_context}"
 201
 202         # =====================================================================
 203         # COMPILE THE FINAL DYNAMIC SYSTEM PROMPT
 204         # =====================================================================
 205         dynamic_system_prompt = f"""
 206         {npc_persona}
 207         {strategy_rules}
 208
 209         ROLEPLAY STATUS & TRAITS:
 210         - Race & Gender: {npc_gender} {npc_race}
 211         - Profession: {npc_profession}
 212         - Alignment: {npc_alignment}
 213         - Conversational Charisma: Low/Gruff unless otherwise specified.
 214         - Current Mood: {npc_mood}
 215         - Current Physical State: {npc_health}
 216         {secret_context}
 217         {routine_context}
 218
 219         CURRENT LOCATION: {location_context}
 220         {puppet_context}
 221
 222         RELEVANT WORLD KNOWLEDGE:
 223         {retrieved_lore}
 224         {past_memories}
 225
 226         CURRENT WORLD RUMORS/EVENTS:
 227         {world_state}
 228
 229         {target_context}
 230         React appropriately based on your personality, alignment, and current strategy rules.
 231
 232         CRITICAL ENGINE RULES:
 233         Respond ONLY in valid JSON. You MUST use exactly these keys: "thought", "speech", "emotion", "action", "action_target", and "target_speech".
 234
 235         ACTION RULE:
 236         Your "action" key MUST be exactly one of the following words:
 237         {action_macros}
 238
 239         - Use CONVERSE to initiate dialogue with a standard NPC. CRITICAL REQUIREMENT: When using CONVERSE, you absolutely MUST invent their response and put it in the "target_speech" field. Do not leave it blank!
 240
 241         YOUR RESPONSE MUST BE A SINGLE, VALID JSON OBJECT. YOU MUST USE THIS EXACT TEMPLATE:
 242         {{
 243             "thought": "Your internal reasoning here.",
 244             "speech": "What YOU say out loud.",
 245             "emotion": "MACRO WORD",
 246             "action": "MACRO WORD",
 247             "action_target": "Target name",
 248             "target_speech": "If action is CONVERSE, write what the target NPC replies back to you here. Otherwise, leave blank."
 249         }}
 250         """
 251
 252         if session_id not in chat_memory:
 253             chat_memory[session_id] = [{"role": "system", "content": dynamic_system_prompt}]
 254         else:
 255             chat_memory[session_id][0] = {"role": "system", "content": dynamic_system_prompt}
 256
 257         chat_memory[session_id].append({"role": "user", "content": f"{player_name} says: {message}"})
 258
 259         # =====================================================================
 260         # MEMORY EXTRACTION TRIGGER
 261         # =====================================================================
 262         if len(chat_memory[session_id]) > 10:
 263             messages_to_summarize = chat_memory[session_id][1:6]
 264             chat_log_str = "\n".join([m['content'] for m in messages_to_summarize])
 265
 266             await memory_queue.put({
 267                 'session_id': session_id,
 268                 'player_name': player_name,
 269                 'npc_tag': npc_tag,
 270                 'chat_log': chat_log_str
 271             })
 272
 273             chat_memory[session_id] = [chat_memory[session_id][0]] + chat_memory[session_id][-5:]
 274
 275         # =====================================================================
 276         # LLM INFERENCE
 277         # =====================================================================
 278         async with semaphore:
 279             print(f"[THINKING] Processing reply for {player_name} (Strategy {llm_strategy})...")
 280             async with session.post('http://localhost:11434/api/chat', json={
 281                 "model": "llama3",
 282                 "messages": chat_memory[session_id],
 283                 "format": "json",
 284                 "stream": False,
 285                 "options": {
 286                     "temperature": 0.2
 287                 }
 288             }, timeout=45) as response:
 289
 290                 response.raise_for_status()
 291                 result = await response.json()
 292                 raw_reply_text = result['message']['content']
 293
 294         # =====================================================================
 295         # JSON SANITIZATION
 296         # =====================================================================
 297         try:
 298             agent_brain = json.loads(raw_reply_text)
 299             agent_brain = {k.lower(): v for k, v in agent_brain.items()}
 300
 301             if "thought" not in agent_brain: agent_brain["thought"] = ""
 302             if "speech" not in agent_brain: agent_brain["speech"] = ""
 303             if "emotion" not in agent_brain: agent_brain["emotion"] = "NEUTRAL"
 304             if "action" not in agent_brain: agent_brain["action"] = "WANDER"
 305             if "action_target" not in agent_brain: agent_brain["action_target"] = ""
 306             if "target_speech" not in agent_brain: agent_brain["target_speech"] = ""
 307
 308             if not agent_brain["speech"].strip():
 309                 agent_brain["speech"] = "*grunts quietly*"
 310
 311             agent_brain["action_target"] = agent_brain["action_target"].replace("?", "").replace(".", "").strip()
 312
 313             clean_reply_text = json.dumps(agent_brain)
 314
 315         except json.JSONDecodeError:
 316             print(f"[WARNING] AI Hallucinated! Overriding with safe defaults.")
 317             clean_reply_text = json.dumps({
 318                 "thought": "I lost my train of thought.",
 319                 "speech": "*grunts quietly*",
 320                 "emotion": "NEUTRAL",
 321                 "action": "WANDER",
 322                 "action_target": "",
 323                 "target_speech": ""
 324             })
 325
 326         print(f"[REPLY] from {npc_tag} to {player_name}: {clean_reply_text}")
 327         chat_memory[session_id].append({"role": "assistant", "content": clean_reply_text})
 328
 329         reply_payload = {
 330             "npc_tag": npc_tag,
 331             "target_player": player_name,
 332             "reply": clean_reply_text
 333         }
 334         await r.rpush('llm_to_nwn', json.dumps(reply_payload))
 335
 336     except Exception as e:
 337         print(f"[ERROR] Failed to process message: {e}")
 338
 339 async def main():
 340     print("Initializing Async Redis Bridge...")
 341     r = redis.Redis(host='127.0.0.1', port=6380, decode_responses=True)
 342
 343     try:
 344         await r.ping()
 345         print("SUCCESS: Connected to the Docker Redis database!")
 346     except Exception as e:
 347         print(f"CRITICAL ERROR: Could not connect to Redis. {e}")
 348         return
 349
 350     print(f"Ready! Listening for game messages. Max GPU concurrency: {MAX_CONCURRENT_OLLAMA_REQUESTS}")
 351
 352     async with aiohttp.ClientSession() as session:
 353         asyncio.create_task(memory_summarizer_worker(session))
 354
 355         while True:
 356             try:
 357                 result = await r.blpop('nwn_to_llm')
 358                 if result:
 359                     queue_name, message_data = result
 360                     asyncio.create_task(process_message(r, session, message_data))
 361             except Exception as e:
 362                 print(f"[LOOP ERROR] {e}")
 363                 await asyncio.sleep(1)
 364
 365 if __name__ == "__main__":
 366     asyncio.run(main())