#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
LULUV2 Pro local chat UI.
A clean ChatGPT-style desktop UI for the fine-tuned LULUV2 checkpoint.
It keeps the important local features only:
- chat inference
- live token streaming
- new chat / save / load chats
- persistent memory notes
- live edge monitor: tok/s, RAM, VRAM, GPU, pass2 metrics
- 32K context controls and test prompt helper
Run:
python ./app.py --ckpt ./LULUV2-bf16.pt --model-py ./luluv2_inference_runtime.py --inbrowser
"""
from __future__ import annotations
import argparse
import json
import os
import re
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Tuple
import gradio as gr
from luluv2_live_inference import (
GenerationConfig,
LULUV2LiveEngine,
clean_text,
normalize_history,
system_usage,
)
APP_NAME = "LuluV2"
CHAT_DIR = Path(os.getenv("LULU_CHAT_DIR", "lulu_chats"))
MEMORY_FILE = Path(os.getenv("LULU_MEMORY_FILE", "lulu_memory.json"))
DEFAULT_SYSTEM_PROMPT = """Your name is LuluV2.
You are a local AI assistant made by Open Machine.
You run offline from the LULUV2 VWM checkpoint.
Answer directly and naturally.
Use Markdown for structure.
When writing code, use fenced code blocks with the correct language tag.
Do not output role tags, hidden scratchpad text, JSON UI fragments, or {'type':'text'} blocks.
"""
PRESETS = {
"Balanced": dict(temperature=0.65, top_k=40, top_p=0.90, min_p=0.03, repetition_penalty=1.10, frequency_penalty=0.02, max_new_tokens=768),
"Precise": dict(temperature=0.35, top_k=30, top_p=0.84, min_p=0.04, repetition_penalty=1.14, frequency_penalty=0.03, max_new_tokens=512),
"Code": dict(temperature=0.42, top_k=40, top_p=0.88, min_p=0.03, repetition_penalty=1.10, frequency_penalty=0.02, max_new_tokens=1200),
"Long 32K": dict(temperature=0.55, top_k=50, top_p=0.90, min_p=0.025, repetition_penalty=1.08, frequency_penalty=0.02, max_new_tokens=1200),
}
def safe_int(value: Any, default: int, low: int | None = None, high: int | None = None) -> int:
try:
value = int(value)
except Exception:
value = default
if low is not None:
value = max(low, value)
if high is not None:
value = min(high, value)
return value
def clamp(value: Any, low: float, high: float, default: float) -> float:
try:
value = float(value)
except Exception:
return default
return max(low, min(high, value))
def esc(text: Any) -> str:
return str(text).replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """)
def status_html(title: str, detail: str = "", tone: str = "neutral") -> str:
tone = tone if tone in {"neutral", "good", "warn", "bad", "live"} else "neutral"
return f"""
{esc(title)}{esc(detail)}
"""
def read_memory() -> str:
if not MEMORY_FILE.exists():
return ""
try:
return str(json.loads(MEMORY_FILE.read_text(encoding="utf-8")).get("memory_notes", ""))
except Exception:
return ""
def write_memory(memory_notes: str) -> Tuple[str, str]:
MEMORY_FILE.write_text(
json.dumps(
{"memory_notes": memory_notes or "", "saved_at": datetime.now().isoformat(timespec="seconds"), "app": APP_NAME},
indent=2,
ensure_ascii=False,
),
encoding="utf-8",
)
return str(MEMORY_FILE), status_html("Memory saved", str(MEMORY_FILE), "good")
def safe_chat_filename(chat_name: str, suffix: str) -> Path:
CHAT_DIR.mkdir(parents=True, exist_ok=True)
base = re.sub(r"[^a-zA-Z0-9_-]+", "_", chat_name or "chat").strip("_") or "chat"
return CHAT_DIR / f"{base}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.{suffix}"
def list_saved_chats() -> List[str]:
CHAT_DIR.mkdir(parents=True, exist_ok=True)
return [str(p) for p in sorted(CHAT_DIR.glob("*.json"), key=lambda x: x.stat().st_mtime, reverse=True)]
def save_chat(history: Any, chat_name: str, memory_notes: str) -> Tuple[str, str, List[str]]:
path = safe_chat_filename(chat_name or "Lulu chat", "json")
data = {
"chat_name": chat_name or "Lulu chat",
"history": normalize_history(history),
"memory_notes": memory_notes or "",
"saved_at": datetime.now().isoformat(timespec="seconds"),
"app": APP_NAME,
}
path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
return str(path), status_html("Chat saved", path.name, "good"), list_saved_chats()
def load_chat(path: str) -> Tuple[List[Dict[str, str]], str, str, str]:
if not path:
return [], "New chat", read_memory(), status_html("No saved chat selected", "Pick a JSON file from the sidebar.", "warn")
try:
data = json.loads(Path(path).read_text(encoding="utf-8"))
except Exception as exc:
return [], "New chat", read_memory(), status_html("Load failed", f"{type(exc).__name__}: {exc}", "bad")
return (
normalize_history(data.get("history", [])),
str(data.get("chat_name") or Path(path).stem),
str(data.get("memory_notes", read_memory())),
status_html("Chat loaded", Path(path).name, "good"),
)
def chat_to_markdown(history: Any, chat_name: str) -> str:
lines = [f"# {clean_text(chat_name) or 'LuluV2 chat'}", ""]
for item in normalize_history(history):
lines.append("## You" if item["role"] == "user" else "## LuluV2")
lines.append(item["content"])
lines.append("")
return "\n".join(lines).strip() + "\n"
def export_markdown(history: Any, chat_name: str) -> Tuple[str, str]:
path = safe_chat_filename(chat_name or "Lulu chat", "md")
path.write_text(chat_to_markdown(history, chat_name), encoding="utf-8")
return str(path), status_html("Markdown exported", path.name, "good")
def postprocess_answer(text: Any, final: bool = False) -> str:
text = clean_text(text)
# Remove common generated UI artifacts from older chat data.
text = re.sub(r"\n?\s*\[\s*\{\s*['\"]text['\"].*?['\"]type['\"]\s*:\s*['\"]text['\"]\s*\}\s*\]\s*$", "", text, flags=re.S)
text = re.sub(r"\n?\s*type\s*:\s*['\"]text['\"]\s*$", "", text, flags=re.I)
text = re.sub(r"\n{4,}", "\n\n\n", text)
if final and text.count("```") % 2 == 1:
text += "\n```"
return text.strip()
def metric_cards(engine: LULUV2LiveEngine, max_context: int) -> str:
stats = engine.stats_dict()
sys = stats.get("system", {})
model = stats.get("model", {})
pass_kl = stats.get("pass1_pass2_kl")
pass_cos = stats.get("pass1_pass2_logit_cosine")
pass_text = "base"
if pass_kl is not None and pass_cos is not None:
pass_text = f"KL {pass_kl:.3f} / cos {pass_cos:.3f}"
gpu_util = sys.get("gpu_util_percent")
gpu_temp = sys.get("gpu_temp_c")
gpu_text = "n/a" if gpu_util is None else f"{gpu_util}%"
temp_text = "n/a" if gpu_temp is None else f"{gpu_temp}°C"
return f"""