Create app.py

app.py

@@ -0,0 +1,297 @@
+import gradio as gr
+import json
+import time
+import hashlib
+from typing import Dict, Optional
+
+# ============================================================================
+# ZEROENGINE-BACKEND: Background Processing Service
+# ============================================================================
+# This space handles:
+# - Tokenization pre-processing
+# - Prompt caching
+# - Token accounting calculations
+# - Response caching
+# ============================================================================
+
+# In-memory caches (will reset on space restart)
+prompt_cache = {}
+response_cache = {}
+token_ledger = {}
+
+def tokenize_text(text: str) -> str:
+    """
+    Fast tokenization without loading full model
+    Returns: JSON string with token count estimation
+    """
+    try:
+        # Simple estimation (4 chars ≈ 1 token for English)
+        # This is FAST and good enough for pre-processing
+        estimated_tokens = len(text) // 4
+        word_count = len(text.split())
+        
+        # Create cache key
+        text_hash = hashlib.md5(text.encode()).hexdigest()[:16]
+        
+        result = {
+            "success": True,
+            "text_hash": text_hash,
+            "estimated_tokens": estimated_tokens,
+            "word_count": word_count,
+            "char_count": len(text),
+            "timestamp": time.time()
+        }
+        
+        # Cache this tokenization
+        prompt_cache[text_hash] = {
+            "text": text[:100] + "..." if len(text) > 100 else text,
+            "tokens": estimated_tokens,
+            "cached_at": time.time()
+        }
+        
+        return json.dumps(result, indent=2)
+        
+    except Exception as e:
+        return json.dumps({
+            "success": False,
+            "error": str(e)
+        }, indent=2)
+
+def cache_prompt(key: str, value: str) -> str:
+    """
+    Store prompt in cache with timestamp
+    """
+    try:
+        prompt_cache[key] = {
+            "value": value,
+            "timestamp": time.time()
+        }
+        
+        # Limit cache size to 100 entries
+        if len(prompt_cache) > 100:
+            oldest_key = min(prompt_cache.keys(), key=lambda k: prompt_cache[k]["timestamp"])
+            del prompt_cache[oldest_key]
+        
+        return json.dumps({
+            "success": True,
+            "cached": key,
+            "cache_size": len(prompt_cache)
+        }, indent=2)
+        
+    except Exception as e:
+        return json.dumps({
+            "success": False,
+            "error": str(e)
+        }, indent=2)
+
+def get_cached_prompt(key: str) -> str:
+    """
+    Retrieve cached prompt
+    """
+    try:
+        if key in prompt_cache:
+            data = prompt_cache[key]
+            return json.dumps({
+                "success": True,
+                "value": data["value"],
+                "age_seconds": round(time.time() - data["timestamp"], 2)
+            }, indent=2)
+        
+        return json.dumps({
+            "success": False,
+            "error": "Cache key not found"
+        }, indent=2)
+        
+    except Exception as e:
+        return json.dumps({
+            "success": False,
+            "error": str(e)
+        }, indent=2)
+
+def cache_response(prompt_hash: str, response: str) -> str:
+    """
+    Cache a complete response for instant retrieval
+    """
+    try:
+        response_cache[prompt_hash] = {
+            "response": response,
+            "timestamp": time.time()
+        }
+        
+        # Limit cache size
+        if len(response_cache) > 50:
+            oldest_key = min(response_cache.keys(), key=lambda k: response_cache[k]["timestamp"])
+            del response_cache[oldest_key]
+        
+        return json.dumps({
+            "success": True,
+            "cached": prompt_hash,
+            "cache_size": len(response_cache)
+        }, indent=2)
+        
+    except Exception as e:
+        return json.dumps({
+            "success": False,
+            "error": str(e)
+        }, indent=2)
+
+def get_cached_response(prompt_hash: str) -> str:
+    """
+    Retrieve cached response
+    """
+    try:
+        if prompt_hash in response_cache:
+            data = response_cache[prompt_hash]
+            return json.dumps({
+                "success": True,
+                "response": data["response"],
+                "age_seconds": round(time.time() - data["timestamp"], 2)
+            }, indent=2)
+        
+        return json.dumps({
+            "success": False,
+            "error": "Response not cached"
+        }, indent=2)
+        
+    except Exception as e:
+        return json.dumps({
+            "success": False,
+            "error": str(e)
+        }, indent=2)
+
+def calculate_token_cost(username: str, duration_ms: float) -> str:
+    """
+    Calculate token cost for a user
+    Stateless - just returns the calculation
+    """
+    try:
+        cost = (duration_ms / 100.0) * 0.001  # 0.001 tokens per 100ms
+        
+        # Track in ledger (for analytics)
+        if username not in token_ledger:
+            token_ledger[username] = {
+                "total_cost": 0.0,
+                "total_duration_ms": 0.0,
+                "requests": 0
+            }
+        
+        token_ledger[username]["total_cost"] += cost
+        token_ledger[username]["total_duration_ms"] += duration_ms
+        token_ledger[username]["requests"] += 1
+        
+        return json.dumps({
+            "success": True,
+            "username": username,
+            "duration_ms": duration_ms,
+            "cost": round(cost, 6),
+            "total_cost": round(token_ledger[username]["total_cost"], 4),
+            "total_requests": token_ledger[username]["requests"]
+        }, indent=2)
+        
+    except Exception as e:
+        return json.dumps({
+            "success": False,
+            "error": str(e)
+        }, indent=2)
+
+def get_cache_stats() -> str:
+    """
+    Get statistics about cache usage
+    """
+    try:
+        return json.dumps({
+            "success": True,
+            "prompt_cache_size": len(prompt_cache),
+            "response_cache_size": len(response_cache),
+            "users_tracked": len(token_ledger),
+            "total_requests": sum(u["requests"] for u in token_ledger.values()),
+            "timestamp": time.time()
+        }, indent=2)
+        
+    except Exception as e:
+        return json.dumps({
+            "success": False,
+            "error": str(e)
+        }, indent=2)
+
+# ============================================================================
+# GRADIO INTERFACE
+# ============================================================================
+
+with gr.Blocks(title="ZeroEngine-Backend", theme=gr.themes.Monochrome()) as demo:
+    gr.HTML("""
+        <div style='text-align: center; padding: 20px;'>
+            <h1>🔧 ZeroEngine-Backend</h1>
+            <p style='color: #888;'>Background Processing Service for ZeroEngine</p>
+        </div>
+    """)
+    
+    with gr.Tab("🔢 Tokenize"):
+        gr.Markdown("### Fast Tokenization Pre-Processing")
+        with gr.Row():
+            with gr.Column():
+                tokenize_input = gr.Textbox(
+                    label="Text to Tokenize",
+                    placeholder="Enter text here...",
+                    lines=5
+                )
+                tokenize_btn = gr.Button("Tokenize", variant="primary")
+            with gr.Column():
+                tokenize_output = gr.Code(label="Result (JSON)", language="json")
+        
+        tokenize_btn.click(tokenize_text, [tokenize_input], [tokenize_output])
+    
+    with gr.Tab("💾 Prompt Cache"):
+        gr.Markdown("### Store and Retrieve Prompts")
+        with gr.Row():
+            with gr.Column():
+                cache_key_input = gr.Textbox(label="Cache Key")
+                cache_value_input = gr.Textbox(label="Value to Cache", lines=3)
+                cache_store_btn = gr.Button("Store", variant="primary")
+                cache_store_output = gr.Code(label="Result", language="json")
+            
+            with gr.Column():
+                cache_get_input = gr.Textbox(label="Key to Retrieve")
+                cache_get_btn = gr.Button("Retrieve", variant="secondary")
+                cache_get_output = gr.Code(label="Result", language="json")
+        
+        cache_store_btn.click(cache_prompt, [cache_key_input, cache_value_input], [cache_store_output])
+        cache_get_btn.click(get_cached_prompt, [cache_get_input], [cache_get_output])
+    
+    with gr.Tab("⚡ Response Cache"):
+        gr.Markdown("### Cache Complete Responses")
+        with gr.Row():
+            with gr.Column():
+                resp_hash_input = gr.Textbox(label="Prompt Hash")
+                resp_value_input = gr.Textbox(label="Response to Cache", lines=5)
+                resp_cache_btn = gr.Button("Cache Response", variant="primary")
+                resp_cache_output = gr.Code(label="Result", language="json")
+            
+            with gr.Column():
+                resp_get_input = gr.Textbox(label="Hash to Retrieve")
+                resp_get_btn = gr.Button("Get Response", variant="secondary")
+                resp_get_output = gr.Code(label="Result", language="json")
+        
+        resp_cache_btn.click(cache_response, [resp_hash_input, resp_value_input], [resp_cache_output])
+        resp_get_btn.click(get_cached_response, [resp_get_input], [resp_get_output])
+    
+    with gr.Tab("💰 Token Accounting"):
+        gr.Markdown("### Calculate Token Costs")
+        with gr.Row():
+            username_input = gr.Textbox(label="Username", value="turtle170")
+            duration_input = gr.Number(label="Duration (ms)", value=5000)
+        
+        calc_btn = gr.Button("Calculate Cost", variant="primary")
+        calc_output = gr.Code(label="Result (JSON)", language="json")
+        
+        calc_btn.click(calculate_token_cost, [username_input, duration_input], [calc_output])
+    
+    with gr.Tab("📊 Stats"):
+        gr.Markdown("### Cache Statistics")
+        stats_btn = gr.Button("Get Stats", variant="primary")
+        stats_output = gr.Code(label="Statistics (JSON)", language="json")
+        
+        stats_btn.click(get_cache_stats, None, [stats_output])
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)