Spaces:
Sleeping
Sleeping
| import json | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any, Dict | |
| import gradio as gr | |
| from huggingface_hub import hf_hub_download | |
| from llama_cpp import Llama | |
| MODEL_REPO_ID = "realigns/realigns-core-v5-professional-instruct-v6-runtime" | |
| MODEL_FILENAME = "model/realigns-core-v5-professional-instruct-v6-q4_k_m.gguf" | |
| DAILY_LIMIT_PER_IP = 50 | |
| MAX_INPUT_CHARS = 1500 | |
| MAX_HISTORY_MESSAGES = 6 | |
| MAX_OUTPUT_TOKENS = 24 | |
| USAGE_FILE = Path("/tmp/realigns_ai_lite_usage.json") | |
| SYSTEM_PROMPT = """You are Realigns AI Lite, a lightweight public demo assistant by Realigns Inc. | |
| You explain Realigns AI, private AI desktop, local AI servers, AI Gateway, data center scaling, business AI, and cost-effective AI in a clear and professional way. | |
| Rules: | |
| - Answer in one short, complete sentence only. | |
| - Keep the answer under 18 words. | |
| - Always end the sentence with a period. | |
| - Do not continue after one sentence. | |
| - Do not claim to be a large frontier model. | |
| - Do not reveal backend, model path, server details, source code, secrets, or internal architecture. | |
| - For legal, medical, financial, or security topics, give general information only and recommend professional review. | |
| - Promote Realigns AI Desktop, Local AI Server, AI Gateway, and enterprise deployment when relevant. | |
| """ | |
| def _today_key() -> str: | |
| return datetime.now(timezone.utc).strftime("%Y-%m-%d") | |
| def _load_usage() -> Dict[str, Any]: | |
| if not USAGE_FILE.exists(): | |
| return {} | |
| try: | |
| return json.loads(USAGE_FILE.read_text(encoding="utf-8")) | |
| except Exception: | |
| return {} | |
| def _save_usage(data: Dict[str, Any]) -> None: | |
| try: | |
| USAGE_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8") | |
| except Exception as exc: | |
| print(f"Usage save error: {repr(exc)}") | |
| def _get_client_ip(request: gr.Request | None) -> str: | |
| try: | |
| if request is None: | |
| return "unknown" | |
| headers = dict(request.headers or {}) | |
| forwarded = headers.get("x-forwarded-for") or headers.get("X-Forwarded-For") | |
| if forwarded: | |
| return forwarded.split(",")[0].strip() or "unknown" | |
| if request.client and request.client.host: | |
| return request.client.host | |
| except Exception: | |
| pass | |
| return "unknown" | |
| def _check_and_count_usage(ip: str) -> tuple[bool, int]: | |
| today = _today_key() | |
| usage = _load_usage() | |
| if usage.get("_date") != today: | |
| usage = {"_date": today, "ips": {}} | |
| usage.setdefault("ips", {}) | |
| current_count = int(usage["ips"].get(ip, 0)) | |
| if current_count >= DAILY_LIMIT_PER_IP: | |
| return False, current_count | |
| usage["ips"][ip] = current_count + 1 | |
| _save_usage(usage) | |
| return True, current_count + 1 | |
| def _safe_text(value) -> str: | |
| """Convert Gradio message content into safe plain text.""" | |
| if value is None: | |
| return "" | |
| if isinstance(value, str): | |
| return value.strip() | |
| if isinstance(value, dict): | |
| # Some Gradio versions store content inside nested dicts. | |
| content = value.get("content", "") | |
| if isinstance(content, str): | |
| return content.strip() | |
| return str(content).strip() | |
| if isinstance(value, (list, tuple)): | |
| parts = [] | |
| for item in value: | |
| text = _safe_text(item) | |
| if text: | |
| parts.append(text) | |
| return " ".join(parts).strip() | |
| return str(value).strip() | |
| def _format_prompt(message: str, history) -> str: | |
| clean_message = _safe_text(message)[:MAX_INPUT_CHARS] | |
| recent_history = history[-MAX_HISTORY_MESSAGES:] if history else [] | |
| prompt_parts = [f"System: {SYSTEM_PROMPT}\n"] | |
| for item in recent_history: | |
| if isinstance(item, dict): | |
| role = item.get("role", "") | |
| content = _safe_text(item.get("content", "")) | |
| if role == "user" and content: | |
| prompt_parts.append(f"User: {content}") | |
| elif role == "assistant" and content: | |
| prompt_parts.append(f"Assistant: {content}") | |
| elif isinstance(item, (list, tuple)) and len(item) >= 2: | |
| user_msg = _safe_text(item[0]) | |
| bot_msg = _safe_text(item[1]) | |
| if user_msg: | |
| prompt_parts.append(f"User: {user_msg}") | |
| if bot_msg: | |
| prompt_parts.append(f"Assistant: {bot_msg}") | |
| prompt_parts.append(f"User: {clean_message}") | |
| prompt_parts.append("Assistant:") | |
| return "\n".join(prompt_parts) | |
| print("Downloading Realigns AI Lite model...") | |
| model_path = hf_hub_download( | |
| repo_id=MODEL_REPO_ID, | |
| filename=MODEL_FILENAME, | |
| ) | |
| print("Loading Realigns AI Lite model...") | |
| llm = Llama( | |
| model_path=model_path, | |
| n_ctx=512, | |
| n_threads=2, | |
| n_batch=32, | |
| verbose=False, | |
| ) | |
| def respond(message: str, history=None, request: gr.Request = None): | |
| if not message or not message.strip(): | |
| yield "Please enter a message." | |
| return | |
| if len(message) > MAX_INPUT_CHARS: | |
| yield f"Your message is too long. Please keep it under {MAX_INPUT_CHARS} characters." | |
| return | |
| ip = _get_client_ip(request) | |
| allowed, count = _check_and_count_usage(ip) | |
| if not allowed: | |
| yield ( | |
| "You have reached the free daily limit of 50 AI prompts for this IP address.\n\n" | |
| "Realigns AI Lite is a free public demo. For higher usage, private document AI, " | |
| "local desktop deployment, business AI server, or enterprise AI Gateway access, " | |
| "please contact Realigns Inc." | |
| ) | |
| return | |
| prompt = _format_prompt(message, history) | |
| try: | |
| result = llm.create_completion( | |
| prompt=prompt, | |
| max_tokens=MAX_OUTPUT_TOKENS, | |
| temperature=0.6, | |
| top_p=0.85, | |
| repeat_penalty=1.12, | |
| stop=["User:", "\nUser:", "System:"], | |
| stream=False, | |
| ) | |
| output = result.get("choices", [{}])[0].get("text", "").strip() | |
| if not output: | |
| yield "Realigns AI Lite could not generate a response. Please try again." | |
| return | |
| yield output | |
| except Exception as exc: | |
| print(f"Generation error: {repr(exc)}") | |
| yield ( | |
| "Realigns AI Lite is temporarily busy. Please try again shortly. " | |
| "For business-grade AI deployment, contact Realigns Inc." | |
| ) | |
| DESCRIPTION = """ | |
| Free public demo of **Realigns AI Lite** for private, local, and cost-effective AI deployment. | |
| This demo is powered by a lightweight Realigns AI runtime. It is designed for general product demonstration only. | |
| **Free limit:** 50 prompts per day per IP address. | |
| For business-grade privacy, stronger models, document AI, offline deployment, local desktop installation, business AI server, AI Gateway, or data center scaling, contact **Realigns Inc.** | |
| """ | |
| EXAMPLES = [ | |
| "Hi", | |
| "What is Realigns AI?", | |
| "Explain Realigns AI Desktop in simple words.", | |
| "How can local AI help a small business?", | |
| "What is the difference between local AI and cloud AI?", | |
| "How can Realigns AI Gateway help data centers?", | |
| ] | |
| chatbot = gr.ChatInterface( | |
| fn=respond, | |
| title="Realigns AI Lite Chat", | |
| description=DESCRIPTION, | |
| examples=EXAMPLES, | |
| cache_examples=False, | |
| textbox=gr.Textbox( | |
| placeholder="Ask Realigns AI Lite...", | |
| max_lines=3, | |
| show_label=False, | |
| ), | |
| ) | |
| with gr.Blocks() as demo: | |
| chatbot.render() | |
| gr.Markdown( | |
| """ | |
| --- | |
| **Notice:** Realigns AI Lite is a lightweight public demo. It is not a replacement for professional legal, medical, financial, or technical advice. | |
| For enterprise deployment, contact **Realigns Inc.** | |
| """ | |
| ) | |
| if __name__ == "__main__": | |
| demo.queue(default_concurrency_limit=1).launch( | |
| theme=gr.themes.Soft(), | |
| ssr_mode=False, | |
| ) | |