import gradio as gr
from llama_cpp import Llama
from huggingface_hub import hf_hub_download
import os
import time
import logging
import re
from typing import List, Dict, Any, Tuple
from dataclasses import dataclass
# =================================================================
# 1. KONFIGURASI TERPUSAT
# =================================================================
@dataclass
class Config:
REPO_ID: str = "tensorblock/dolphin-2.9-llama3-8b-GGUF"
FILENAME: str = "dolphin-2.9-llama3-8b-Q2_K.gguf"
CACHE_DIR: str = "/data/huggingface_cache"
N_CTX: int = 4096
N_THREADS: int = 2
N_GPU_LAYERS: int = 0
BATCH_SIZE: int = 512
TEMPERATURE: float = 0.7
TOP_P: float = 0.90
TOP_K: int = 40
REPEAT_PENALTY: float = 1.12
PRESENCE_PENALTY: float = 0.0
FREQUENCY_PENALTY: float = 0.0
MIROSTAT: bool = False
TFS_Z: float = 1.0
MAX_TOKENS: int = 768
MAX_HISTORY_LENGTH: int = 20
MAX_PROMPT_LENGTH: int = 4000
ENABLE_SUMMARY: bool = True
SUMMARY_TRIGGER: int = 15
ENABLE_PERSONA: bool = True
LOG_FILE: str = "mutsumi_log.txt"
LOG_LEVEL: str = "INFO"
SYSTEM_PROMPT: str = (
"Kamu adalah Mutsumi AI, teman ngobrol dan asisten AI yang santai, asik, cerdas, dan terbuka. "
"Jawablah dengan gaya bahasa yang natural, luwes, to the point, dan tidak kaku seperti robot. "
"Kamu bebas membahas topik apapun secara jujur dan apa adanya tanpa batasan moral yang dibuat-buat, "
"tetapi tetap berikan jawaban yang bermakna dan mudah dipahami. "
"Gunakan Bahasa Indonesia yang akrab namun tetap cerdas."
)
config = Config()
# =================================================================
# 2. LOGGING
# =================================================================
os.makedirs(os.path.dirname(config.LOG_FILE) or '.', exist_ok=True)
logging.basicConfig(
filename=config.LOG_FILE,
level=getattr(logging, config.LOG_LEVEL),
format='%(asctime)s - %(levelname)s - %(message)s'
)
console = logging.StreamHandler()
console.setLevel(logging.INFO)
logging.getLogger('').addHandler(console)
# =================================================================
# 3. MODEL MANAGER
# =================================================================
class ModelManager:
def __init__(self, config: Config):
self.config = config
self.llm = None
self.loaded = False
def load_model(self) -> Llama:
if self.loaded and self.llm is not None:
return self.llm
logging.info("Loading model...")
os.makedirs(self.config.CACHE_DIR, exist_ok=True)
try:
model_path = hf_hub_download(
repo_id=self.config.REPO_ID,
filename=self.config.FILENAME,
cache_dir=self.config.CACHE_DIR,
force_download=False,
resume_download=True,
)
logging.info(f"Model path: {model_path}")
except Exception as e:
logging.error(f"Download failed: {e}")
raise
try:
self.llm = Llama(
model_path=model_path,
n_ctx=self.config.N_CTX,
n_threads=self.config.N_THREADS,
n_gpu_layers=self.config.N_GPU_LAYERS,
batch_size=self.config.BATCH_SIZE,
chat_format="chatml",
verbose=False,
)
self.loaded = True
logging.info("Model loaded successfully.")
except Exception as e:
logging.error(f"Load failed: {e}")
raise
# Warm-up
try:
self.llm.create_chat_completion(
messages=[{"role": "user", "content": "Halo"}],
max_tokens=1,
temperature=0.0,
stream=False,
)
logging.info("Warm-up done.")
except Exception as e:
logging.warning(f"Warm-up failed: {e}")
return self.llm
def get_model(self) -> Llama:
if not self.loaded:
return self.load_model()
return self.llm
# =================================================================
# 4. CHAT MEMORY
# =================================================================
class ChatMemory:
def __init__(self, config: Config):
self.config = config
self.history: List[Dict[str, str]] = []
self.persona: Dict[str, Any] = {}
self.summary: str = ""
def add_message(self, role: str, content: str):
self.history.append({"role": role, "content": content})
if role == "user" and self.config.ENABLE_PERSONA:
self._extract_persona(content)
if len(self.history) > self.config.MAX_HISTORY_LENGTH * 2:
self._manage_memory()
def _extract_persona(self, text: str):
if "nama saya" in text.lower() or "panggil saya" in text.lower():
match = re.search(r"(?:nama saya|panggil saya)\s+(\w+)", text, re.IGNORECASE)
if match:
self.persona["name"] = match.group(1)
if any(ord(c) > 127 for c in text):
self.persona["language"] = "Indonesian"
else:
self.persona["language"] = "English"
def _manage_memory(self):
if not self.config.ENABLE_SUMMARY:
overflow = len(self.history) - self.config.MAX_HISTORY_LENGTH * 2
self.history = self.history[overflow:]
return
if len(self.history) > self.config.SUMMARY_TRIGGER * 2:
half = len(self.history) // 2
old_messages = self.history[:half]
old_text = "\n".join([f"{m['role']}: {m['content']}" for m in old_messages])
self.summary = f"Ringkasan percakapan sebelumnya:\n{old_text[:1000]}..."
self.history = self.history[half:]
def get_messages_for_prompt(self, system_prompt: str) -> List[Dict[str, str]]:
messages = []
enhanced_system = system_prompt
if self.summary:
enhanced_system += f"\n\n{self.summary}"
if self.persona.get("name"):
enhanced_system += f"\n\nNama pengguna: {self.persona['name']}."
if self.persona.get("language"):
enhanced_system += f"\nBahasa yang digunakan: {self.persona['language']}."
messages.append({"role": "system", "content": enhanced_system})
history_to_use = self.history[-(self.config.MAX_HISTORY_LENGTH * 2):]
messages.extend(history_to_use)
return messages
def clear(self):
self.history = []
self.summary = ""
self.persona = {}
# =================================================================
# 5. CHAT ENGINE
# =================================================================
class ChatEngine:
def __init__(self, model_manager: ModelManager, config: Config):
self.model_manager = model_manager
self.config = config
self.memory = ChatMemory(config)
def generate_response(self, user_message: str, system_prompt: str):
if not user_message or len(user_message) > 2000:
yield "Pesan terlalu panjang (maks 2000 karakter)", {"error": "input_too_long"}
return
self.memory.add_message("user", user_message)
messages = self.memory.get_messages_for_prompt(system_prompt)
prompt_length = sum(len(m["content"]) for m in messages) // 3
if prompt_length > self.config.MAX_PROMPT_LENGTH:
logging.warning(f"Prompt panjang ({prompt_length}), dipotong.")
while prompt_length > self.config.MAX_PROMPT_LENGTH and len(messages) > 1:
messages.pop(1)
prompt_length = sum(len(m["content"]) for m in messages) // 3
model = self.model_manager.get_model()
start_time = time.time()
full_response = ""
token_count = 0
try:
stream = model.create_chat_completion(
messages=messages,
max_tokens=self.config.MAX_TOKENS,
temperature=self.config.TEMPERATURE,
top_p=self.config.TOP_P,
top_k=self.config.TOP_K,
repeat_penalty=self.config.REPEAT_PENALTY,
presence_penalty=self.config.PRESENCE_PENALTY,
frequency_penalty=self.config.FREQUENCY_PENALTY,
mirostat_mode=2 if self.config.MIROSTAT else 0,
tfs_z=self.config.TFS_Z,
stream=True,
)
for chunk in stream:
if not isinstance(chunk, dict):
continue
if "choices" not in chunk or not chunk["choices"]:
continue
choice = chunk["choices"][0]
delta = choice.get("delta", {})
token = delta.get("content", "")
finish_reason = choice.get("finish_reason")
if token:
full_response += token
token_count += 1
yield full_response, None
if finish_reason:
break
except Exception as e:
logging.error(f"Stream error: {e}")
yield f"⚠️ Gangguan saat generate: {str(e)}", {"error": str(e)}
return
elapsed = time.time() - start_time
tokens_per_sec = token_count / elapsed if elapsed > 0 else 0
logging.info(f"Response: {token_count} token in {elapsed:.2f}s ({tokens_per_sec:.1f} tok/s)")
self.memory.add_message("assistant", full_response)
yield full_response, {"token_count": token_count, "elapsed": elapsed, "tokens_per_sec": tokens_per_sec}
def clear_memory(self):
self.memory.clear()
# =================================================================
# 6. INISIALISASI GLOBAL
# =================================================================
model_manager = ModelManager(config)
chat_engine = ChatEngine(model_manager, config)
# =================================================================
# 7. FUNGSI RESPOND
# =================================================================
def respond(message, chat_history, system_prompt):
try:
if not isinstance(chat_history, list):
chat_history = []
chat_history.append([message, "Thinking"])
yield "", chat_history
full_response = ""
for partial, metadata in chat_engine.generate_response(message, system_prompt):
if metadata and "error" in metadata:
if len(chat_history) > 0:
chat_history[-1][1] = f"⚠️ {partial}"
else:
chat_history = [[message, f"⚠️ {partial}"]]
yield "", chat_history
return
if len(chat_history) > 0:
chat_history[-1][1] = partial
else:
chat_history = [[message, partial]]
yield "", chat_history
full_response = partial
except Exception as e:
logging.error(f"Error di respond: {e}")
if chat_history and len(chat_history) > 0:
chat_history[-1][1] = f"⚠️ Terjadi kesalahan: {str(e)}"
else:
chat_history = [[message, f"⚠️ Terjadi kesalahan: {str(e)}"]]
yield "", chat_history
def clear_chat():
chat_engine.clear_memory()
return [], ""
# =================================================================
# 8. CSS dan UI (Background PUTIH + Avatar Bot Gambar)
# =================================================================
custom_css = """
@import url('https://fonts.googleapis.com/css2?family=Quicksand:wght@400;500;600;700&display=swap');
* { font-family: 'Quicksand', sans-serif !important; }
.gradio-container {
background: #FFFFFF !important;
min-height: 100vh;
}
#susu-header {
text-align: center;
padding: 22px 10px 6px 10px;
}
#susu-header h1 {
font-weight: 700;
font-size: 2.1em;
color: #E88CA8;
margin-bottom: 2px;
letter-spacing: 0.5px;
}
#susu-header p {
color: #C98BA0;
font-size: 0.95em;
font-weight: 500;
}
#chatbot {
background: #FFFDFB !important;
border-radius: 26px !important;
border: 1.5px solid #FBD4E1 !important;
box-shadow: 0 8px 30px rgba(233, 160, 190, 0.18) !important;
padding: 6px !important;
}
.message.user {
background: linear-gradient(135deg, #FFB9D2, #FFCBE0) !important;
color: #6B2E42 !important;
border-radius: 20px 20px 4px 20px !important;
font-weight: 500;
}
.message.bot {
background: #FFF4F8 !important;
color: #6B4550 !important;
border: 1px solid #FADCE7 !important;
border-radius: 20px 20px 20px 4px !important;
}
#msg-box textarea, #msg-box input {
background: #FFFDFB !important;
border: 1.5px solid #F8C6D8 !important;
border-radius: 18px !important;
color: #6B4550 !important;
font-weight: 500;
}
button.primary, #send-btn {
background: linear-gradient(135deg, #FFAFCB, #FF8FB3) !important;
color: white !important;
border: none !important;
border-radius: 16px !important;
font-weight: 600 !important;
box-shadow: 0 4px 14px rgba(255, 143, 179, 0.35) !important;
}
button.primary:hover, #send-btn:hover {
background: linear-gradient(135deg, #FF9EC0, #FF7CA6) !important;
transform: translateY(-1px);
}
button.secondary {
background: #FFFDFB !important;
border: 1.5px solid #F8C6D8 !important;
color: #E88CA8 !important;
border-radius: 16px !important;
font-weight: 600 !important;
}
.gr-accordion {
background: #FFFBFD !important;
border: 1px solid #FADCE7 !important;
border-radius: 18px !important;
}
@keyframes pulse {
0% { opacity: 0.4; }
50% { opacity: 1; }
100% { opacity: 0.4; }
}
.thinking-text {
animation: pulse 1.2s infinite;
font-weight: 600;
color: #E88CA8;
font-size: 1.1em;
}
footer {display: none !important;}
"""
BOT_AVATAR_URL = "https://huggingface.co/spaces/MortisChan/NSW-CHUPA/resolve/main/46f0ef7ab8b96cb0940bb964e05b2973.jpg"
with gr.Blocks(css=custom_css, title="Mutsumi AI - Uncensored") as demo:
gr.HTML(
"""
"""
)
with gr.Accordion("⚙️ Atur Kepribadian Mutsumi (Uncensored)", open=False):
system_prompt_box = gr.Textbox(
label="System Prompt",
value=config.SYSTEM_PROMPT,
lines=5,
)
chatbot = gr.Chatbot(
elem_id="chatbot",
height=480,
avatar_images=(None, BOT_AVATAR_URL),
bubble_full_width=False,
)
with gr.Row():
msg = gr.Textbox(
elem_id="msg-box",
placeholder="Tanyakan apapun tanpa batasan",
scale=8,
show_label=False,
)
send = gr.Button("Kirim 💌", elem_id="send-btn", scale=1, variant="primary")
clear = gr.Button("🧼 Bersihkan Obrolan", variant="secondary")
msg.submit(respond, [msg, chatbot, system_prompt_box], [msg, chatbot])
send.click(respond, [msg, chatbot, system_prompt_box], [msg, chatbot])
clear.click(clear_chat, None, [chatbot, msg])
if __name__ == "__main__":
demo.queue(default_concurrency_limit=10).launch(debug=False)