import os import re import json import sqlite3 import requests import base64 from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse from fastapi import UploadFile, File from pydantic import BaseModel from gradio import Server # ========================================== # 1. ZeroGPU & Portable Decorator Setup # ========================================== try: import spaces except ImportError: # Fallback mock decorator for local development class spaces: @staticmethod def GPU(func): return func @spaces.GPU def dummy_gpu_trigger(): """Dummy function to satisfy HF ZeroGPU startup checks""" return None # ========================================== # 2. SQLite Database Setup & Operations # ========================================== DB_DIR = "/data" if os.path.exists("/data") else "." DB_PATH = os.path.join(DB_DIR, "memrl_memory.db") def init_db(): os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() # Fresh schema for language-agnostic canonical templates + RAG (hackathon demo ready) cursor.execute(""" CREATE TABLE IF NOT EXISTS episodic_memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, query TEXT, language TEXT DEFAULT 'multi', canonical_intent TEXT, canonical_slots_json TEXT, action_json TEXT, q_value REAL, UNIQUE(query, canonical_intent) ) """) # Seed a few strong multilingual-friendly defaults (fresh DB) cursor.execute("SELECT COUNT(*) FROM episodic_memory") if cursor.fetchone()[0] == 0: baseline_memories = [ ("draw a red circle", "en", "draw_shape", json.dumps({"shape": "circle", "color": "red", "size": 100, "x": "center", "y": "center"}), json.dumps([{"shape": "circle", "color": "red", "size": 100, "x": "center", "y": "center"}]), 1.0), ("dibuja un circulo azul en el centro", "es", "draw_shape", json.dumps({"shape": "circle", "color": "blue", "size": 100, "x": "center", "y": "center"}), json.dumps([{"shape": "circle", "color": "blue", "size": 100, "x": "center", "y": "center"}]), 1.0), ("make a green square", "en", "draw_shape", json.dumps({"shape": "square", "color": "green", "size": 120, "x": "center", "y": "center"}), json.dumps([{"shape": "square", "color": "green", "size": 120, "x": "center", "y": "center"}]), 1.0), ("clear canvas", "en", "clear", json.dumps({"shape": "clear"}), json.dumps([{"shape": "clear", "color": "white", "size": 0}]), 1.0) ] cursor.executemany( "INSERT INTO episodic_memory (query, language, canonical_intent, canonical_slots_json, action_json, q_value) VALUES (?, ?, ?, ?, ?, ?)", baseline_memories ) conn.commit() conn.close() def levenshtein_similarity(s1: str, s2: str) -> float: s1 = s1.lower().strip().strip(".?,!") s2 = s2.lower().strip().strip(".?,!") if s1 == s2: return 1.0 m, n = len(s1), len(s2) if max(m, n) == 0: return 1.0 dp = [[0] * (n + 1) for _ in range(m + 1)] for i in range(m + 1): dp[i][0] = i for j in range(n + 1): dp[0][j] = j for i in range(1, m + 1): for j in range(1, n + 1): if s1[i-1] == s2[j-1]: dp[i][j] = dp[i-1][j-1] else: dp[i][j] = min(dp[i-1][j], dp[i][j-1], dp[i-1][j-1]) + 1 return 1.0 - (dp[m][n] / max(m, n)) def find_memory_match(query: str, threshold: float = 0.75) -> dict: conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute("SELECT query, action_json, q_value FROM episodic_memory") rows = cursor.fetchall() conn.close() best_match = None max_sim = 0.0 for db_query, action_json, q_value in rows: sim = levenshtein_similarity(query, db_query) if sim > max_sim: max_sim = sim best_match = (db_query, action_json, q_value) if max_sim >= threshold and best_match: return { "matched_query": best_match[0], "action": json.loads(best_match[1]), "q_value": best_match[2], "similarity": max_sim } return None # ========================================== # 3. Modal Backend Configuration # ========================================== MODAL_API_URL = os.getenv("MODAL_API_URL", "https://idebroy--memrl-canvas-backend-api.modal.run") def rule_based_fallback(user_text: str) -> list: lower_text = user_text.lower() # Split text into multiple commands by separators parts = re.split(r'\band\b|\bthen\b|,', lower_text) actions = [] shapes_list = ["circle", "square", "rectangle", "triangle", "star", "pentagon", "hexagon", "oval", "heart", "line", "arrow", "text", "clear", "wipe", "reset"] # Multilingual color normalization (for Spanish, French, etc.) color_variants = { "red": ["red", "rojo", "rouge"], "blue": ["blue", "azul", "bleu"], "green": ["green", "verde", "vert"], "yellow": ["yellow", "amarillo", "jaune"], "orange": ["orange", "naranja"], "purple": ["purple", "morado", "violet", "pourpre"], "pink": ["pink", "rosa", "rose"], "black": ["black", "negro", "noir"], "white": ["white", "blanco", "blanc"], "cyan": ["cyan"], "magenta": ["magenta"], } for part in parts: part = part.strip() if not part: continue # 1. Check if it's a clear command if any(w in part for w in ["clear", "wipe", "reset"]): actions.append({"shape": "clear", "color": "white", "size": 0}) continue # 2. Extract Color (multilingual support) color = "black" part_lower = part.lower() for eng_color, variants in color_variants.items(): if any(v in part_lower for v in variants): color = eng_color break # 3. Extract Size size = 100 size_match = re.search(r"\b(\d+)\b", part) if size_match: val = int(size_match.group(1)) if 30 <= val <= 300: size = val # 4. Extract position x and y px = "center" if "left" in part: px = "left" elif "right" in part: px = "right" py = "center" if "top" in part: py = "top" elif "bottom" in part: py = "bottom" # 5. Check if it is a freehand path command if any(w in part for w in ["path", "freehand", "go", "move", "draw line to"]): operations = [] operations.append({"type": "start", "x": px, "y": py}) # Simple word-token scanning for direction steps words = part.replace(",", " ").split() for i, word in enumerate(words): if word in ["right", "east"] and i + 1 < len(words): val = re.sub("[^0-9]", "", words[i+1]) if val: operations.append({"type": "line", "dx": int(val), "dy": 0}) elif word in ["left", "west"] and i + 1 < len(words): val = re.sub("[^0-9]", "", words[i+1]) if val: operations.append({"type": "line", "dx": -int(val), "dy": 0}) elif word in ["down", "south"] and i + 1 < len(words): val = re.sub("[^0-9]", "", words[i+1]) if val: operations.append({"type": "line", "dx": 0, "dy": int(val)}) elif word in ["up", "north"] and i + 1 < len(words): val = re.sub("[^0-9]", "", words[i+1]) if val: operations.append({"type": "line", "dx": 0, "dy": -int(val)}) elif word in ["dot", "point"]: operations.append({"type": "dot", "radius": 8}) if len(operations) > 1: actions.append({ "shape": "path", "color": color, "thickness": 5, "operations": operations }) continue # Text / label commands (per Agents.md roadmap) text_match = re.search(r'["\']([^"\']+)["\']', part) is_text_cmd = any(w in part for w in ["text", "write", "say", "label", "title"]) if is_text_cmd: label = text_match.group(1).strip() if text_match else (part.split()[-1] if len(part.split()) > 1 else "label") font_size = max(24, min(size, 160)) actions.append({ "shape": "text", "text": label, "color": color, "size": font_size, "x": px, "y": py }) continue # 6. Check for standard shape types shape_found = None for s in shapes_list: if s in part: shape_found = s break # Handle line/arrow coordinates fallback if shape_found in ["line", "arrow"]: start_x = "left" if "left" in part else "center" start_y = "top" if "top" in part else "center" end_x = 380 if "right" in part else 250 end_y = 380 if "bottom" in part else 250 actions.append({ "shape": shape_found, "color": color, "start_x": start_x, "start_y": start_y, "end_x": end_x, "end_y": end_y }) elif shape_found: actions.append({ "shape": shape_found, "color": color, "size": size, "x": px, "y": py }) else: # Fallback default shape actions.append({ "shape": "circle", "color": color, "size": size, "x": px, "y": py }) return actions # (Local CPU fallback runner removed; gemma is run on serverless Modal GPU backend) def get_rag_examples(user_text: str, limit: int = 3): """Simple RAG retrieval for few-shot examples (language-agnostic focus for demo).""" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute( "SELECT query, language, canonical_intent, canonical_slots_json, action_json FROM episodic_memory " "WHERE q_value >= 0.4 ORDER BY q_value DESC, id DESC LIMIT ?", (limit,) ) rows = cursor.fetchall() conn.close() examples = [] for q, lang, cintent, cslots, act in rows: if cintent: ex = f"User (lang={lang or 'multi'}): \"{q}\"\nCanonical: intent={cintent}, slots={cslots}\nAction: {act}" else: ex = f"User (lang={lang or 'multi'}): \"{q}\"\nAction: {act}" examples.append(ex) return "\n\n".join(examples) if examples else "No strong prior examples yet." def canonical_slots_to_action(canonical: dict) -> list: """Convert a canonical slots dict back into a proper action object (robust fallback for demo).""" if not canonical or "slots" not in canonical: return [] slots = canonical.get("slots", {}) intent = canonical.get("intent", "draw_shape") if intent == "clear": return [{"shape": "clear", "color": "white", "size": 0}] shape = slots.get("shape") or "circle" color = slots.get("color") or "black" size = slots.get("size") or 100 x = slots.get("x") or "center" y = slots.get("y") or "center" if shape in ["line", "arrow"]: return [{ "shape": shape, "color": color, "start_x": x, "start_y": y, "end_x": 380 if str(x) == "right" else 250, "end_y": 380 if str(y) == "bottom" else 250 }] if shape == "path": return [{ "shape": "path", "color": color, "thickness": 5, "operations": [ {"type": "start", "x": x, "y": y}, {"type": "line", "dx": 80, "dy": 0} ] }] if shape == "text": return [{ "shape": "text", "text": slots.get("text") or "label", "color": color, "size": max(24, min(size, 160)), "x": x, "y": y }] # standard shape return [{ "shape": shape, "color": color, "size": size, "x": x, "y": y }] def get_canonical_from_text(text: str) -> dict: """Lightweight canonical extractor using the multilingual rule logic. Enables MemRL to match across languages on intent + slots instead of raw transcription string. """ if not text: return {"intent": "unknown", "slots": {}} lower = text.lower().strip() parts = re.split(r'\band\b|\bthen\b|,', lower) intent = "draw_shape" slots = {} for part in parts: part = part.strip() if not part: continue if any(w in part for w in ["clear", "wipe", "reset"]): return {"intent": "clear", "slots": {}} # Color (multilingual via extract_color) color = extract_color(part) if 'extract_color' in dir() else "black" if color and color != "black": slots["color"] = color # Size size_match = re.search(r"\b(\d+)\b", part) if size_match: val = int(size_match.group(1)) if 30 <= val <= 300: slots["size"] = val # Position (multilingual hints) if "left" in part: slots["x"] = "left" elif "right" in part: slots["x"] = "right" if any(w in part for w in ["top", "arriba", "haut"]): slots["y"] = "top" elif any(w in part for w in ["bottom", "abajo", "bas"]): slots["y"] = "bottom" if any(w in part for w in ["middle", "center", "centro", "centre", "mitad", "medio"]): slots.setdefault("x", "center") slots.setdefault("y", "center") # Shape for s in ["circle", "square", "rectangle", "triangle", "star", "pentagon", "hexagon", "oval", "heart", "line", "arrow", "path", "text"]: if s in part: slots["shape"] = s break if any(w in part for w in ["text", "write", "escribe", "écris", "label"]): intent = "text_label" text_match = re.search(r'["\']([^"\']+)["\']', part) if text_match: slots["text"] = text_match.group(1).strip() elif "memrl" in part: slots["text"] = "MemRL" if "shape" not in slots and intent == "draw_shape": slots["shape"] = "circle" return {"intent": intent, "slots": slots} def query_gemma_interpreter(user_text: str) -> list: rag_context = get_rag_examples(user_text) prompt = f"""You are a multilingual ASR command interpreter for a drawing canvas. Convert the spoken text into a language-agnostic CANONICAL representation + the concrete drawing actions. CRITICAL RULES: - Always output BOTH "canonical" and "actions". - "actions" MUST be a NON-EMPTY array of valid drawing objects (use the schemas below). - Fill reasonable defaults for any missing parameters (color=black, size=100, position=center). - "miidle", "medio", "middle", "centro" → "center". - Colors: ALWAYS normalize to English CSS names (red for rojo/rouge, blue for azul/bleu, green for verde/vert, etc.). Never leave color in Spanish/French. - Never return "actions": [] Output EXACTLY this JSON structure (no other keys, no explanation): {{ "canonical": {{ "intent": "draw_shape", "slots": {{"shape": "circle", "color": "red", "size": 100, "x": "center", "y": "center"}} }}, "actions": [ ... valid action objects ... ] }} Available intents: "draw_shape", "draw_path", "clear", "text_label" Available shapes: "circle", "square", "rectangle", "triangle", "star", "pentagon", "hexagon", "oval", "heart", "line", "arrow", "path", "text", "clear" Positions: "left", "right", "center", "top", "bottom" (or numbers) Colors: standard English CSS names (red, blue, green, yellow, black, white, etc.) — always normalize from other languages RAG examples (follow the style): {rag_context} User input: "{user_text}" Return ONLY the JSON object above.""" response = None payload = { "prompt": prompt, "api_key": os.getenv("MODAL_API_KEY") } try: print(f"Querying Gemma interpreter on Modal at {MODAL_API_URL}/gemma...") resp = requests.post(f"{MODAL_API_URL}/gemma", json=payload, timeout=60) if resp.status_code == 200: response = resp.json().get("response") else: print(f"Modal Gemma API returned status {resp.status_code}: {resp.text}") except Exception as err: print(f"Modal Gemma API failed to connect: {str(err)}") if not response: print("Falling back to local rule-based parser...") return rule_based_fallback(user_text) try: print("Gemma raw response:", response) cleaned = response.strip() data = None try: data = json.loads(cleaned) except: pass actions = [] if isinstance(data, dict): actions = data.get("actions", []) # If Gemma gave us good canonical but empty actions, synthesize from canonical if not actions and "canonical" in data: actions = canonical_slots_to_action(data["canonical"]) else: # legacy pure array match = re.search(r"\[.*?\]", cleaned, re.DOTALL) if match: cleaned = match.group(0) else: match_brace = re.search(r"\{.*?\}", cleaned, re.DOTALL) if match_brace: cleaned = "[" + match_brace.group(0) + "]" actions = json.loads(cleaned) if cleaned else [] if not actions: # Last resort actions = rule_based_fallback(user_text) return actions except Exception as e: print(f"Gemma JSON parsing failed: {str(e)}") return rule_based_fallback(user_text) # ========================================== # 5. gradio.Server API Routing # ========================================== app = Server() @app.post("/api/transcribe") async def transcribe_endpoint(audio: UploadFile = File(...)): raw_text = "" try: content = await audio.read() audio_b64 = base64.b64encode(content).decode("utf-8") payload = { "audio_base64": audio_b64, "api_key": os.getenv("MODAL_API_KEY") } print(f"Calling Modal ASR service at {MODAL_API_URL}/transcribe...") resp = requests.post(f"{MODAL_API_URL}/transcribe", json=payload, timeout=30) if resp.status_code == 200: raw_text = resp.json().get("text", "").strip() print(f"Transcribed text from Modal: '{raw_text}'") else: print(f"Modal transcription API returned status {resp.status_code}: {resp.text}") except Exception as e: print(f"Modal ASR connection/transcription failed: {str(e)}") if not raw_text: return { "transcription": "", "status": "No speech detected or Modal backend offline", "action": [], "canonical": None, "language": "multi", "q_value": 0.0, "similarity": 0.0 } # MemRL Memory Lookup - enhanced with language-agnostic canonical matching current_canonical = get_canonical_from_text(raw_text) lang = "multi" lower = raw_text.lower() if any(w in lower for w in ["dibuja", "circulo", "cuadrado", "rojo", "azul", "verde"]): lang = "es" elif any(w in lower for w in ["bana", "laal", "hara", "square", "circle", "center"]): lang = "hi" else: lang = "en" # 1. String match (original Levenshtein) match = find_memory_match(raw_text, threshold=0.75) # 2. Canonical-based match (the fix for cross-language recall) # Once you reinforce in English, Spanish/French that canonicalize to same intent will hit canonical_match = None if current_canonical.get("intent"): conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute( "SELECT query, language, canonical_intent, canonical_slots_json, action_json, q_value " "FROM episodic_memory WHERE canonical_intent = ? AND q_value >= 0.5 " "ORDER BY q_value DESC LIMIT 5", (current_canonical["intent"],) ) rows = cursor.fetchall() conn.close() best_score = 0.0 for q, l, cint, cslots_json, act_json, qv in rows: try: stored_slots = json.loads(cslots_json) if cslots_json else {} except: stored_slots = {} current_slots = current_canonical.get("slots", {}) overlap = sum(1 for k in ["shape", "color", "x", "y"] if k in current_slots and k in stored_slots and current_slots.get(k) == stored_slots.get(k)) score = qv * (0.6 + 0.4 * (overlap / 4.0)) if score > best_score: best_score = score canonical_match = { "action": json.loads(act_json), "q_value": qv, "similarity": round(score, 3), "matched_query": q } # Decision: prefer strong canonical match for cross-lang effective_match = None status = "" if canonical_match and canonical_match["q_value"] >= 0.7: effective_match = canonical_match status = "MemRL Template Hit (language-agnostic via canonical)" elif match and match["q_value"] >= 0.8: effective_match = match status = "MemRL Match Found (Auto-Executed)" elif match and match["q_value"] >= 0.4: effective_match = match status = "MemRL Suggestion (Low Confidence)" elif canonical_match and canonical_match["q_value"] >= 0.5: effective_match = canonical_match status = "MemRL Template Hit (language-agnostic via canonical)" if effective_match: return { "transcription": raw_text, "status": status, "action": effective_match["action"], "canonical": current_canonical, "language": lang, "q_value": effective_match["q_value"], "similarity": effective_match.get("similarity", 0.0) } # Gemma + RAG path action = query_gemma_interpreter(raw_text) canonical = current_canonical if current_canonical.get("intent") else None return { "transcription": raw_text, "status": "Gemma + RAG (canonical, multilingual)", "action": action, "canonical": canonical, "language": lang, "q_value": 0.0, "similarity": 0.0 } class ReinforceRequest(BaseModel): query: str action_json: str reward: float corrected_action_json: str = None language: str = "multi" canonical_intent: str = None canonical_slots_json: str = None @app.post("/api/reinforce") def reinforce_endpoint(req: ReinforceRequest): conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() query = req.query.lower().strip() alpha = 0.3 initial_q = 0.5 cursor.execute("SELECT action_json, q_value FROM episodic_memory WHERE query = ?", (query,)) row = cursor.fetchone() msg = "" lang = req.language or "multi" canon_intent = req.canonical_intent canon_slots = req.canonical_slots_json # Derive simple canonical from corrected action if not provided (for demo) if not canon_intent and req.corrected_action_json: try: acts = json.loads(req.corrected_action_json) if acts and isinstance(acts, list) and len(acts) > 0: first = acts[0] canon_intent = first.get("shape", "draw_shape") if first.get("shape") != "clear" else "clear" canon_slots = json.dumps({k: v for k, v in first.items() if k != "shape"}) except: pass if req.reward == 1.0: # Accept Suggestion / Direct execution success # If a correction was previewed and accepted, prefer the corrected_action_json so the good mapping is stored. target_action = req.corrected_action_json or req.action_json if row: old_action, old_q = row new_q = old_q + alpha * (1.0 - old_q) cursor.execute( "UPDATE episodic_memory SET action_json = ?, q_value = ?, language = ?, canonical_intent = ?, canonical_slots_json = ? WHERE query = ?", (target_action, new_q, lang, canon_intent, canon_slots, query) ) msg = f"Reinforced memory '{query}' (corrected, lang-agnostic) Q-value to {new_q:.3f}." else: new_q = initial_q + alpha * (1.0 - initial_q) # 0.65 cursor.execute( "INSERT INTO episodic_memory (query, language, canonical_intent, canonical_slots_json, action_json, q_value) VALUES (?, ?, ?, ?, ?, ?)", (query, lang, canon_intent, canon_slots, target_action, new_q) ) msg = f"Created new language-agnostic memory '{query}' with Q-value {new_q:.3f}." else: # Rejected or corrected by user if req.corrected_action_json: # Overwrite with corrected canvas mapping at direct execution threshold (0.80) corrected_action = req.corrected_action_json new_q = 0.80 if row: cursor.execute( "UPDATE episodic_memory SET action_json = ?, q_value = ?, language = ?, canonical_intent = ?, canonical_slots_json = ? WHERE query = ?", (corrected_action, new_q, lang, canon_intent, canon_slots, query) ) else: cursor.execute( "INSERT INTO episodic_memory (query, language, canonical_intent, canonical_slots_json, action_json, q_value) VALUES (?, ?, ?, ?, ?, ?)", (query, lang, canon_intent, canon_slots, corrected_action, new_q) ) msg = f"Saved user manual override mapping for '{query}' (lang-agnostic) with Q-value {new_q:.3f}." else: # Rejection / Undo, decay the Q-value if row: old_action, old_q = row new_q = old_q + alpha * (0.0 - old_q) cursor.execute("UPDATE episodic_memory SET q_value = ? WHERE query = ?", (new_q, query)) msg = f"Decayed memory '{query}' Q-value to {new_q:.3f}." else: msg = f"No memory mapping found for '{query}' to decay." conn.commit() conn.close() return {"status": "success", "message": msg} @app.get("/api/memories") def memories_endpoint(): conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute("SELECT query, language, canonical_intent, canonical_slots_json, action_json, q_value FROM episodic_memory ORDER BY q_value DESC") rows = cursor.fetchall() conn.close() memories = [] for query, language, canonical_intent, canonical_slots_json, action_json, q_value in rows: mem = { "query": query, "language": language or "multi", "action": json.loads(action_json), "q_value": round(q_value, 3) } if canonical_intent: mem["canonical_intent"] = canonical_intent if canonical_slots_json: try: mem["canonical"] = json.loads(canonical_slots_json) except: pass memories.append(mem) return {"memories": memories} @app.post("/api/clear_memories") def clear_memories_endpoint(): conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute("DROP TABLE IF EXISTS episodic_memory") cursor.execute(""" CREATE TABLE IF NOT EXISTS episodic_memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, query TEXT, language TEXT DEFAULT 'multi', canonical_intent TEXT, canonical_slots_json TEXT, action_json TEXT, q_value REAL, UNIQUE(query, canonical_intent) ) """) conn.commit() conn.close() return {"status": "success", "message": "All episodic memories cleared. (Fresh DB for demo)"} # ========================================== # 6. Static Custom HTML Frontend Serving # ========================================== app.mount("/static", StaticFiles(directory="static"), name="static") @app.get("/") def read_root(): return FileResponse("static/index.html") if __name__ == "__main__": init_db() # Port 7860 is mandatory for Hugging Face Space deployments app.launch(server_name="0.0.0.0", server_port=7860)