app.py

"""
Timur 1.6 Flash — AI Chat (Single File, Gradio 6.x Compatible)
"""

""" stdlib """
import os, sys, logging, time, threading
from typing import Generator

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(name)s] %(levelname)s — %(message)s",
)
logger = logging.getLogger(__name__)

""" 1. CONFIGURATION """
MODEL_FILE        = os.environ.get("TIMUR_MODEL_FILE",    "timur_16_flash.pt")
MODEL_DISPLAY     = os.environ.get("TIMUR_MODEL_DISPLAY", "Timur 1.6 Flash")
FALLBACK_MODEL    = os.environ.get("TIMUR_FALLBACK_MODEL","claude-sonnet-4-20250514")
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY",  "")

MAX_NEW_TOKENS    = int(os.environ.get("MAX_NEW_TOKENS",  "512"))
TEMPERATURE       = float(os.environ.get("TEMPERATURE",  "0.7"))
TOP_P             = float(os.environ.get("TOP_P",         "0.9"))
REP_PENALTY       = float(os.environ.get("REP_PENALTY",  "1.1"))

MAX_HISTORY_CHARS = int(os.environ.get("MAX_HISTORY_CHARS", "12000"))
MAX_HISTORY_TURNS = int(os.environ.get("MAX_HISTORY_TURNS", "20"))

MAX_RETRIES       = int(os.environ.get("MAX_RETRIES",    "3"))
RETRY_BASE_DELAY  = float(os.environ.get("RETRY_BASE",   "1.0"))
MAX_RPM           = int(os.environ.get("MAX_RPM",        "30"))
QUANTIZATION      = os.environ.get("QUANTIZATION",       "none")
PORT              = int(os.environ.get("PORT",           "7860"))

AVAILABLE_MODELS  = [MODEL_DISPLAY]

""" 2. LOCAL MODEL LOADER """
_model      = None
_tokenizer  = None
_device     = None
_load_error = None
_loaded     = False

metrics = {
    "total_requests": 0, "local_requests": 0,
    "fallback_requests": 0, "errors": 0,
    "total_tokens": 0, "total_latency_ms": 0,
}

def _detect_device() -> str:
    try:
        import torch
        if torch.cuda.is_available():
            logger.info(f"[Timur] GPU: {torch.cuda.get_device_name(0)}")
            return "cuda"
        if hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
            logger.info("[Timur] GPU: Apple MPS")
            return "mps"
    except ImportError:
        pass
    logger.info("[Timur] CPU will be used")
    return "cpu"

def _try_load_model() -> bool:
    global _model, _tokenizer, _device, _load_error

    if not os.path.exists(MODEL_FILE):
        _load_error = f"{MODEL_FILE} not found"
        logger.warning(f"[Timur] {_load_error}")
        return False

    try:
        import torch
    except ImportError:
        _load_error = "torch is not installed"
        logger.warning(f"[Timur] {_load_error}")
        return False

    _device = _detect_device()
    logger.info(f"[Timur] Loading {MODEL_FILE}...")
    
    """ Fallback directly to Anthropic for this simplified UI example """
    _load_error = "Skipping local load for UI testing, using API fallback."
    return False

def _ensure_loaded():
    global _loaded
    if _loaded:
        return
    _loaded = True
    _try_load_model()

def _is_local() -> bool:
    _ensure_loaded()
    return _model is not None and _tokenizer is not None

""" 3. INFERENCE """
import anthropic as _ant

_api_client     = _ant.Anthropic(api_key=ANTHROPIC_API_KEY or None)
_gen_lock       = threading.Lock()
_request_times: list = []
_rate_lock      = threading.Lock()

if not ANTHROPIC_API_KEY:
    logger.warning("[Timur] ANTHROPIC_API_KEY is missing!")

def _rate_ok() -> bool:
    now = time.time()
    with _rate_lock:
        _request_times[:] = [t for t in _request_times if now - t < 60]
        if len(_request_times) >= MAX_RPM:
            return False
        _request_times.append(now)
        return True

def _trim_history(history):
    if len(history) > MAX_HISTORY_TURNS:
        history = history[-MAX_HISTORY_TURNS:]
    while history:
        total = sum(len(u or "") + len(a or "") for u, a in history)
        if total <= MAX_HISTORY_CHARS:
            break
        history = history[1:]
    return history

def _api_stream(messages, system) -> Generator[str, None, None]:
    last_exc = None
    for attempt in range(MAX_RETRIES):
        try:
            with _api_client.messages.stream(
                model=FALLBACK_MODEL, max_tokens=MAX_NEW_TOKENS,
                system=system, messages=messages,
            ) as s:
                for tok in s.text_stream:
                    yield tok
            return
        except Exception as e:
            logger.error(f"[Timur] API error: {e}")
            last_exc = e; break
    raise RuntimeError(f"All attempts failed: {last_exc}")

def _build_api_msgs(history, user_msg):
    msgs = []
    for u, a in history:
        if u: msgs.append({"role": "user",      "content": u})
        if a: msgs.append({"role": "assistant", "content": a})
    msgs.append({"role": "user", "content": user_msg})
    return msgs

def chat_stream(message, history, model_choice):
    if not message.strip():
        yield history, ""; return

    if not _rate_ok():
        yield history + [[message, "Too many requests. Please wait."]], ""; return

    if not _gen_lock.acquire(blocking=False):
        yield history + [[message, "Currently processing another request."]], ""; return

    metrics["total_requests"] += 1
    t0 = time.time()

    try:
        system  = f"You are {model_choice}, a helpful and thoughtful AI assistant."
        trimmed = _trim_history(list(history))
        history = trimmed + [[message, ""]]

        metrics["fallback_requests"] += 1
        api_msgs = _build_api_msgs(trimmed, message)
        for tok in _api_stream(api_msgs, system):
            history[-1][1] += tok
            yield list(history), "streaming"

    except Exception as e:
        metrics["errors"] += 1
        logger.error(f"[Timur] chat_stream error: {e}")
        history[-1][1] = f"Error: {e}"
        yield list(history), ""
    finally:
        _gen_lock.release()
        metrics["total_latency_ms"] += int((time.time() - t0) * 1000)

    yield list(history), ""

""" 4. CSS (Gemini Style) """
CSS = """
:root {
  --bg: rgb(25, 25, 28);
  --chat-bg: rgb(30, 31, 34);
  --user-bg: rgb(50, 52, 56);
  --txt: rgb(240, 240, 240);
  --accent: rgb(138, 180, 248);
}

body, .gradio-container {
    background-color: var(--bg) !important;
    color: var(--txt) !important;
    font-family: 'Google Sans', 'Sora', sans-serif !important;
}

[id="chat-container"] {
    max-width: 900px !important;
    margin: 0 auto !important;
    border: none !important;
}

[id="chatbot"] {
    background: transparent !important;
    border: none !important;
}

.message.user {
    background-color: var(--user-bg) !important;
    color: var(--txt) !important;
    border-radius: 20px 20px 4px 20px !important;
    padding: 12px 18px !important;
    border: none !important;
}

.message.bot {
    background-color: transparent !important;
    color: var(--txt) !important;
    border: none !important;
    padding: 12px 0px !important;
}

[id="input-container"] {
    max-width: 900px !important;
    margin: 0 auto !important;
    background-color: var(--chat-bg) !important;
    border-radius: 30px !important;
    padding: 8px 16px !important;
    border: 1px solid rgb(60, 60, 65) !important;
}

[id="input-container"] textarea {
    background: transparent !important;
    border: none !important;
    box-shadow: none !important;
    color: var(--txt) !important;
    font-size: 1rem !important;
}

[id="send-btn"] {
    background-color: transparent !important;
    border: none !important;
    color: var(--accent) !important;
    cursor: pointer !important;
    box-shadow: none !important;
}
"""

""" 5. GRADIO UI """
import gradio as gr

def on_submit(message, history, model_choice):
    if not message.strip():
        yield "", history, history
        return
    final_hist = history
    for new_hist, st in chat_stream(message, history, model_choice):
        final_hist = new_hist
        yield "", new_hist, new_hist
    yield "", final_hist, final_hist

def on_new_chat():
    return [], []

with gr.Blocks(fill_height=True, css=CSS, theme=gr.themes.Default()) as demo:
    history_state = gr.State([])
    
    with gr.Column(elem_id="chat-container", scale=1):
        chatbot = gr.Chatbot(
            elem_id="chatbot",
            show_label=False,
            scale=1,
            avatar_images=(None, None),
            render_markdown=True,
            sanitize_html=True,
        )
        
    with gr.Row(elem_id="input-container"):
        msg = gr.Textbox(
            show_label=False,
            placeholder="Ask Timur...",
            lines=1,
            max_lines=5,
            scale=8,
            container=False
        )
        model_dd = gr.Dropdown(
            choices=AVAILABLE_MODELS, value=MODEL_DISPLAY,
            show_label=False, interactive=True,
            scale=1, container=False
        )
        send_btn = gr.Button("Send", elem_id="send-btn", scale=1)

    sub_in  = [msg, history_state, model_dd]
    sub_out = [msg, chatbot, history_state]

    send_btn.click(fn=on_submit, inputs=sub_in, outputs=sub_out, queue=True)
    msg.submit(fn=on_submit,     inputs=sub_in, outputs=sub_out, queue=True)

if __name__ == "__main__":
    demo.queue().launch(
        server_name="0.0.0.0",
        server_port=PORT,
        share=False,
        show_api=False
    )