| 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 |
|
|
| |
| |
| |
| @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() |
|
|
| |
| |
| |
| 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) |
|
|
| |
| |
| |
| 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 |
| |
| 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 |
|
|
| |
| |
| |
| 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 = {} |
|
|
| |
| |
| |
| 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() |
|
|
| |
| |
| |
| model_manager = ModelManager(config) |
| chat_engine = ChatEngine(model_manager, config) |
|
|
| |
| |
| |
| def respond(message, chat_history, system_prompt): |
| try: |
| if not isinstance(chat_history, list): |
| chat_history = [] |
|
|
| chat_history.append([message, "<span class='thinking-text'>Thinking</span>"]) |
| 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 [], "" |
|
|
| |
| |
| |
| 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( |
| """ |
| <div id="susu-header"> |
| <h1>🤖 Mutsumi AI 🎭</h1> |
| <p>Asisten AI Bebas & Tanpa Sensor — Jujur, Kritis, Transparan ✨</p> |
| <p style="font-size: 0.8em; color: #E88CA8;">⚡ Q2_K ~3GB | Cache permanen di /data</p> |
| </div> |
| """ |
| ) |
| 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) |
