import streamlit as st from llama_cpp import Llama from huggingface_hub import hf_hub_download import os import gc import shutil import subprocess # Available models MODELS = { "Qwen2.5-7B-Instruct (Q2_K)": { "repo_id": "Qwen/Qwen2.5-7B-Instruct-GGUF", "filename": "qwen2.5-7b-instruct-q2_k.gguf", "description": "Qwen2.5-7B Instruct (Q2_K)" }, "Gemma-3-4B-IT (Q4_K_M)": { "repo_id": "unsloth/gemma-3-4b-it-GGUF", "filename": "gemma-3-4b-it-Q4_K_M.gguf", "description": "Gemma 3 4B IT (Q4_K_M)" }, "Phi-4-mini-Instruct (Q4_K_M)": { "repo_id": "unsloth/Phi-4-mini-instruct-GGUF", "filename": "Phi-4-mini-instruct-Q4_K_M.gguf", "description": "Phi-4 Mini Instruct (Q4_K_M)" }, } with st.sidebar: st.header("⚙️ Settings") selected_model_name = st.selectbox("Select Model", list(MODELS.keys())) system_prompt = st.text_area("System Prompt", value="You are a helpful assistant.", height=80) max_tokens = st.slider("Max tokens", 64, 2048, 512, step=32) temperature = st.slider("Temperature", 0.1, 2.0, 0.7) top_k = st.slider("Top-K", 1, 100, 40) top_p = st.slider("Top-P", 0.1, 1.0, 0.95) repeat_penalty = st.slider("Repetition Penalty", 1.0, 2.0, 1.1) if st.button("🧹 Clear All Cached Models"): try: for f in os.listdir("models"): if f.endswith(".gguf"): os.remove(os.path.join("models", f)) st.success("Model cache cleared.") except Exception as e: st.error(f"Failed to clear models: {e}") if st.button("📦 Show Disk Usage"): try: usage = shutil.disk_usage(".") used = usage.used / (1024**3) free = usage.free / (1024**3) st.info(f"Disk Used: {used:.2f} GB | Free: {free:.2f} GB") except Exception as e: st.error(f"Disk usage error: {e}") # Model info selected_model = MODELS[selected_model_name] model_path = os.path.join("models", selected_model["filename"]) # Init state if "model_name" not in st.session_state: st.session_state.model_name = None if "llm" not in st.session_state: st.session_state.llm = None # Ensure model directory exists os.makedirs("models", exist_ok=True) def cleanup_old_models(): for f in os.listdir("models"): if f.endswith(".gguf") and f != selected_model["filename"]: try: os.remove(os.path.join("models", f)) except Exception as e: st.warning(f"Couldn't delete old model {f}: {e}") def download_model(): with st.spinner(f"Downloading {selected_model['filename']}..."): hf_hub_download( repo_id=selected_model["repo_id"], filename=selected_model["filename"], local_dir="./models", local_dir_use_symlinks=False, ) def try_load_model(path): try: return Llama(model_path=path, n_ctx=1024, n_threads=2, n_threads_batch=2, n_batch=4, n_gpu_layers=0, use_mlock=False, use_mmap=True, verbose=False) except Exception as e: return str(e) def validate_or_download_model(): if not os.path.exists(model_path): cleanup_old_models() download_model() # First load attempt result = try_load_model(model_path) if isinstance(result, str): st.warning(f"Initial load failed: {result}\nAttempting re-download...") try: os.remove(model_path) except: pass cleanup_old_models() download_model() result = try_load_model(model_path) if isinstance(result, str): st.error(f"Model still failed after re-download: {result}") st.stop() return result return result # Load model if changed if st.session_state.model_name != selected_model_name: if st.session_state.llm is not None: del st.session_state.llm gc.collect() st.session_state.llm = validate_or_download_model() st.session_state.model_name = selected_model_name llm = st.session_state.llm # Chat history state if "chat_history" not in st.session_state: st.session_state.chat_history = [] st.title(f"🧠 {selected_model['description']} (Streamlit + GGUF)") st.caption(f"Powered by `llama.cpp` | Model: {selected_model['filename']}") user_input = st.chat_input("Ask something...") if user_input: # Prevent appending user message if assistant hasn't replied yet if len(st.session_state.chat_history) % 2 == 1: st.warning("Please wait for the assistant to respond before sending another message.") else: st.session_state.chat_history.append({"role": "user", "content": user_input}) with st.chat_message("user"): st.markdown(user_input) # Trim conversation history to max 8 turns (user+assistant) MAX_TURNS = 8 trimmed_history = st.session_state.chat_history[-MAX_TURNS * 2:] messages = [{"role": "system", "content": system_prompt}] + trimmed_history with st.chat_message("assistant"): full_response = "" response_area = st.empty() stream = llm.create_chat_completion( messages=messages, max_tokens=max_tokens, temperature=temperature, top_k=top_k, top_p=top_p, repeat_penalty=repeat_penalty, stream=True, ) for chunk in stream: if "choices" in chunk: delta = chunk["choices"][0]["delta"].get("content", "") full_response += delta response_area.markdown(full_response) st.session_state.chat_history.append({"role": "assistant", "content": full_response})