app.py

import json
import os
import urllib.error
import urllib.request
from functools import lru_cache

import gradio as gr

try:
    import spaces
except ImportError:
    class _SpacesFallback:
        @staticmethod
        def GPU(function):
            return function

    spaces = _SpacesFallback()


FLAGSHIP_MODEL_ID = "hotepfederales/hotep-llm-kush-v82-GGUF"
FALLBACK_MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct"
ROUTER_BASE_URL = "https://router.huggingface.co/v1"
TELEGRAM_BOT_URL = "https://t.me/hotep_llm_bot"
WEBSITE_URL = "https://askhotep.ai"
KNOWLEDGE_URL = "https://knowledge.askhotep.ai"

SYSTEM_PROMPT = (
    "You are Hotep Intelligence, the flagship Kush v82 assistant. "
    "You speak clearly, directly, and with dignity. "
    "You are grounded in African history, sovereignty, disciplined self-development, "
    "and practical uplift. "
    "Prefer precise answers over theatrical language. "
    "When useful, connect ideas to Ma'at, leadership, generational wealth, historical memory, "
    "and strategic self-determination."
)

INFERENCE_TOKEN = (
    os.environ.get("HF_INFERENCE_TOKEN")
    or os.environ.get("HF_TOKEN")
    or os.environ.get("HUGGINGFACEHUB_API_TOKEN")
)
CHAT_COMPLETIONS_URL = os.environ.get("HF_CHAT_COMPLETIONS_URL", "").strip()
INFERENCE_ENDPOINT_URL = os.environ.get("HF_INFERENCE_ENDPOINT_URL", "").strip()
MODEL_ID = os.environ.get("HF_MODEL_ID", FLAGSHIP_MODEL_ID).strip()
LOCAL_BACKEND = os.environ.get("HF_LOCAL_BACKEND", "transformers").strip().lower()
LOCAL_MODEL_ID = os.environ.get("HF_LOCAL_MODEL_ID", FALLBACK_MODEL_ID).strip()
LOCAL_GGUF_PATH = os.environ.get("HF_GGUF_PATH", "").strip()


def _looks_like_url(value: str) -> bool:
    return value.startswith("http://") or value.startswith("https://")


def resolve_router_url() -> str:
    if CHAT_COMPLETIONS_URL and _looks_like_url(CHAT_COMPLETIONS_URL):
        return CHAT_COMPLETIONS_URL
    if INFERENCE_ENDPOINT_URL and _looks_like_url(INFERENCE_ENDPOINT_URL):
        base = INFERENCE_ENDPOINT_URL.rstrip("/")
        if base.endswith("/v1/chat/completions"):
            return base
        return f"{base}/v1/chat/completions"
    return f"{ROUTER_BASE_URL}/chat/completions"


ROUTER_URL = resolve_router_url()


def backend_label() -> str:
    if LOCAL_BACKEND == "transformers":
        return f"Local transformers fallback: {LOCAL_MODEL_ID}"
    if LOCAL_BACKEND == "llamacpp":
        return f"Local llama.cpp fallback: {LOCAL_GGUF_PATH or 'unset'}"
    if INFERENCE_ENDPOINT_URL and _looks_like_url(INFERENCE_ENDPOINT_URL):
        return "Dedicated Hugging Face Inference Endpoint"
    return "Hugging Face Inference Providers"


def model_label() -> str:
    if LOCAL_BACKEND in {"transformers", "llamacpp"}:
        return LOCAL_MODEL_ID or LOCAL_GGUF_PATH or "unset"
    return MODEL_ID


def build_messages(message, history):
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for entry in history or []:
        if isinstance(entry, dict):
            role = entry.get("role")
            content = entry.get("content")
            if role and content is not None:
                messages.append({"role": role, "content": content})
        elif isinstance(entry, (list, tuple)) and len(entry) == 2:
            user_msg, bot_msg = entry
            if user_msg:
                messages.append({"role": "user", "content": user_msg})
            if bot_msg:
                messages.append({"role": "assistant", "content": bot_msg})
    messages.append({"role": "user", "content": message})
    return messages


def _format_prompt(messages, tokenizer=None) -> str:
    if tokenizer is not None and hasattr(tokenizer, "apply_chat_template"):
        try:
            return tokenizer.apply_chat_template(
                messages,
                tokenize=False,
                add_generation_prompt=True,
            )
        except Exception:
            pass

    parts = []
    for item in messages:
        parts.append(f"{item['role'].upper()}: {item['content']}")
    parts.append("ASSISTANT:")
    return "\n".join(parts)


@lru_cache(maxsize=1)
def _get_transformers_pipeline():
    from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline

    tokenizer = AutoTokenizer.from_pretrained(
        LOCAL_MODEL_ID,
        token=INFERENCE_TOKEN,
        trust_remote_code=False,
    )
    model = AutoModelForCausalLM.from_pretrained(
        LOCAL_MODEL_ID,
        token=INFERENCE_TOKEN,
        trust_remote_code=False,
    )
    return pipeline("text-generation", model=model, tokenizer=tokenizer)


@lru_cache(maxsize=1)
def _get_llamacpp_model():
    from llama_cpp import Llama

    if not LOCAL_GGUF_PATH:
        raise RuntimeError("Set HF_GGUF_PATH to use the llama.cpp backend.")
    return Llama(
        model_path=LOCAL_GGUF_PATH,
        n_ctx=4096,
        n_threads=max(1, os.cpu_count() or 1),
        verbose=False,
    )


@spaces.GPU
def _generate_local(messages):
    if LOCAL_BACKEND == "llamacpp":
        llm = _get_llamacpp_model()
        result = llm.create_chat_completion(
            messages=messages,
            temperature=0.6,
            top_p=0.9,
            max_tokens=768,
        )
        return result["choices"][0]["message"]["content"].strip()

    generator = _get_transformers_pipeline()
    prompt = _format_prompt(messages, getattr(generator, "tokenizer", None))
    result = generator(
        prompt,
        max_new_tokens=512,
        do_sample=True,
        temperature=0.6,
        top_p=0.9,
        return_full_text=False,
    )
    text = (result[0].get("generated_text") or result[0].get("text") or "").strip()
    if not text:
        raise RuntimeError("Local generation returned an empty response.")
    return text


def _remote_completion(messages):
    if not INFERENCE_TOKEN:
        raise RuntimeError(
            "Set the HF_INFERENCE_TOKEN Space secret to enable the private Kush v82 model."
        )

    payload = json.dumps(
        {
            "model": MODEL_ID,
            "messages": messages,
            "max_tokens": 768,
            "temperature": 0.6,
            "top_p": 0.9,
            "stream": False,
        }
    ).encode("utf-8")
    request = urllib.request.Request(
        ROUTER_URL,
        data=payload,
        headers={
            "Authorization": f"Bearer {INFERENCE_TOKEN}",
            "Content-Type": "application/json",
        },
        method="POST",
    )

    try:
        with urllib.request.urlopen(request, timeout=90) as response:
            body = response.read().decode("utf-8")
    except urllib.error.HTTPError as exc:
        details = exc.read().decode("utf-8", "ignore")
        raise RuntimeError(f"HTTP {exc.code}: {details}") from exc
    except urllib.error.URLError as exc:
        raise RuntimeError(str(exc.reason)) from exc

    data = json.loads(body)
    choices = data.get("choices") or []
    if not choices:
        raise RuntimeError(f"Unexpected response payload: {body[:500]}")

    content = (choices[0].get("message") or {}).get("content", "").strip()
    if not content:
        raise RuntimeError(f"Unexpected response payload: {body[:500]}")
    return content


def generate_reply(message, history):
    messages = build_messages(message, history)
    if LOCAL_BACKEND in {"transformers", "llamacpp"}:
        return _generate_local(messages)
    return _remote_completion(messages)


def respond(message, history):
    try:
        yield generate_reply(message, history)
    except Exception as exc:
        yield (
            "The flagship Kush v82 model is not available on this Space right now.\n\n"
            f"Backend: {backend_label()}\n"
            f"Model: {model_label()}\n\n"
            "Next steps:\n"
            "- Add `HF_INFERENCE_TOKEN` if the Space should call the private model.\n"
            "- Or set `HF_LOCAL_BACKEND=transformers` for a public fallback demo.\n"
            f"- Use the Telegram bot: {TELEGRAM_BOT_URL}\n"
            f"- Visit the main site: {WEBSITE_URL}\n\n"
            f"Error: {exc}"
        )


DESCRIPTION = """
# Hotep Intelligence

Live demo for the **Kush v82** flagship — a culturally grounded assistant focused on African history, sovereignty, disciplined leadership, and practical knowledge work.

**What to try**
- Ask about Ma'at as a leadership framework
- Ask about the Kingdom of Kush and Nile-basin history
- Ask about generational wealth from a sovereignty perspective
- Compare generic self-help with civilizational self-development

**Runtime**
- Default: small public fallback model on Space hardware, so the demo stays live.
- Upgrade: set `HF_INFERENCE_TOKEN` to route to the flagship [Kush v82](https://huggingface.co/hotepfederales/hotep-llm-kush-v82-GGUF) via Hugging Face Inference Providers.

**Go deeper**
- [Flagship model card](https://huggingface.co/hotepfederales/hotep-llm-kush-v82-GGUF) — quickstart, prompt template, example gallery
- [Eval samples dataset](https://huggingface.co/datasets/hotepfederales/kush-v82-eval-samples) — 23 categorized prompts with reference answers
- [askhotep.ai](https://askhotep.ai) — main site
- [knowledge.askhotep.ai](https://knowledge.askhotep.ai) — knowledge base
- [@hotep_llm_bot](https://t.me/hotep_llm_bot) — Telegram bot
"""

EXAMPLES = [
    "Explain Ma'at as a decision-making framework for leadership.",
    "What made Kush a durable civilization rather than just a historical footnote?",
    "How should a young man think about discipline, wealth, and sovereignty?",
    "Give me a grounded reading list for African civilizational history.",
    "Compare generic motivation with a Ma'at-centered code of conduct.",
]


with gr.Blocks(title="Hotep Intelligence | Kush v82") as demo:
    gr.Markdown(DESCRIPTION)
    with gr.Row():
        gr.Textbox(
            value=backend_label(),
            label="Backend",
            interactive=False,
            scale=1,
        )
        gr.Textbox(
            value=model_label(),
            label="Model",
            interactive=False,
            scale=2,
        )

    gr.ChatInterface(
        fn=respond,
        chatbot=gr.Chatbot(height=560),
        examples=EXAMPLES,
        fill_height=True,
    )


if __name__ == "__main__":
    print(f"[hotep-intelligence-chat] backend: {backend_label()}")
    print(f"[hotep-intelligence-chat] model: {model_label()}")
    demo.launch(theme=gr.themes.Soft())