Update app.py

app.py

@@ -1 +1,204 @@
-import gradio as gr import torch from transformers import AutoModelForCausalLM, AutoTokenizer import logging # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) MODEL_NAME = "ubiodee/Plutus_Tutor_new" # ------------ Model and Tokenizer cache ------------ _TOKENIZER = None _MODEL = None def get_tokenizer(): global _TOKENIZER if _TOKENIZER is None: try: _TOKENIZER = AutoTokenizer.from_pretrained(MODEL_NAME, use_fast=True) if _TOKENIZER.pad_token_id is None: if _TOKENIZER.eos_token_id is not None: _TOKENIZER.pad_token = _TOKENIZER.eos_token _TOKENIZER.pad_token_id = _TOKENIZER.eos_token_id else: _TOKENIZER.add_special_tokens({"pad_token": "<|endoftext|>"}) _TOKENIZER.pad_token_id = _TOKENIZER.convert_tokens_to_ids("<|endoftext|>") logger.info(f"Set pad_token_id: {_TOKENIZER.pad_token_id}") if _TOKENIZER.eos_token_id is None: _TOKENIZER.eos_token = "<|endoftext|>" _TOKENIZER.eos_token_id = _TOKENIZER.convert_tokens_to_ids("<|endoftext|>") logger.info(f"Set eos_token_id: {_TOKENIZER.eos_token_id}") logger.info("Tokenizer loaded successfully") except Exception as e: logger.error(f"Failed to load tokenizer: {type(e).__name__}: {e}") raise return _TOKENIZER def get_model(): global _MODEL if _MODEL is None: try: logger.info("Loading model on CPU with FP16") _MODEL = AutoModelForCausalLM.from_pretrained( MODEL_NAME, torch_dtype=torch.float16, device_map="cpu", low_cpu_mem_usage=True, ) _MODEL.eval() logger.info("Model loaded successfully") except Exception as e: logger.error(f"Failed to load model: {type(e).__name__}: {e}") raise return _MODEL # ------------ Prompt builder ------------ def build_instructions(personality, level, topic): return ( f"Plutus AI tutor for a {personality} learner at {level} level. " f"Explain {topic} in a tone that fits learners personality and tech level with examples. " "Keep it 250–500 words. End with 'Takeaway:'." ) def build_model_input(tokenizer, personality, level, topic): user_msg = build_instructions(personality, level, topic) # Use a simple text prompt instead of chat template return f"System: You are a personalised Cardano Plutus tutor, your job is to make Plutus easy to learn based on the different learner personalities, you are to adapt your teaching style according to the learner.\nUser: {user_msg}\nAssistant:" # ------------ Generation function ------------ def generate(personality, level, topic, max_new_tokens=500): try: tokenizer = get_tokenizer() model = get_model() prompt = build_model_input(tokenizer, personality, level, topic) device = torch.device("cpu") logger.info("Generating on CPU") inputs = tokenizer(prompt, return_tensors="pt") input_len = inputs["input_ids"].shape[1] inputs = {k: v.to(device) for k, v in inputs.items()} with torch.inference_mode(): outputs = model.generate( **inputs, max_new_tokens=max_new_tokens, temperature=0.3, top_p=0.4, do_sample=True, repetition_penalty=1.3, eos_token_id=tokenizer.eos_token_id, pad_token_id=tokenizer.pad_token_id, stop_strings=["Takeaway:"] ) gen_ids = outputs[0][input_len:] text = tokenizer.decode(gen_ids, skip_special_tokens=True).strip() if not text: text = tokenizer.decode(outputs[0], skip_special_tokens=True).strip() if text.startswith(prompt): text = text[len(prompt):].lstrip() return text if text else "Generation failed. Try regenerating." except Exception as e: logger.error(f"Generation error: {type(e).__name__}: {e}") return f"Error during generation: {str(e)}. Try regenerating or using a smaller model." # ------------ Orchestrator with retry logic ------------ def orchestrator(personality, level, topic, max_retries=3): if not personality or not level or not topic: return "Select personality, expertise, and topic to get an explanation." for attempt in range(max_retries): try: logger.info(f"Generation attempt {attempt + 1}/{max_retries}") return generate(personality, level, topic) except Exception as e: logger.error(f"Attempt {attempt + 1}/{max_retries} failed: {type(e).__name__}: {e}") if attempt == max_retries - 1: return ( "Failed to generate after multiple attempts. " "Click **Regenerate** or try again later. " "If this persists, try a smaller model or check logs for errors." ) # ------------ Gradio UI ------------ with gr.Blocks(theme="default") as iface: gr.Markdown( "## Cardano Plutus AI Assistant\n" "Select **Learning Personality**, **Expertise Level**, and **Topic**, then click **Generate**. " "Note: Generation may take ~1–2 minutes on CPU." ) with gr.Row(): personality = gr.Dropdown( choices=["Dyslexic", "Autistic", "Expressive"], label="Learning Personality", value=None, allow_custom_value=False, scale=1, ) level = gr.Dropdown( choices=["Beginner", "Intermediate", "Advanced"], label="Expertise Level", value=None, allow_custom_value=False, scale=1, ) topic = gr.Dropdown( choices=[ "What is Plutus", "Introduction to Plutus Smart Contracts", "Understanding Cardano Blockchain", "Validator Scripts in Plutus", "Plutus Tx", "Datum and Redeemer", "Time Handling in Plutus", "Off-Chain Code", "On-Chain Constraints", "Plutus Core", "Transaction Validation", "Cardano Node Integration", ], label="Topic", value=None, allow_custom_value=False, scale=2, ) with gr.Row(): generate_btn = gr.Button("Generate") regen = gr.Button("🔁 Regenerate") output = gr.Textbox( label="Model Response", lines=12, interactive=False, show_copy_button=True, placeholder="Your tailored explanation will appear here…", ) generate_btn.click(orchestrator, [personality, level, topic], output, queue=True) regen.click(orchestrator, [personality, level, topic], output, queue=True) # Enable queue iface.queue() if __name__ == "__main__": iface.launch(server_name="0.0.0.0", server_port=7860)
+import gradio as gr
+import torch
+import time
+from transformers import AutoModelForCausalLM, AutoTokenizer
+import logging
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+MODEL_NAME = "ubiodee/Plutus_Tutor_new"
+
+# ------------ Model and Tokenizer cache ------------
+_TOKENIZER = None
+_MODEL = None
+
+def get_tokenizer():
+    global _TOKENIZER
+    if _TOKENIZER is None:
+        try:
+            _TOKENIZER = AutoTokenizer.from_pretrained(MODEL_NAME, use_fast=True)
+            if _TOKENIZER.pad_token_id is None:
+                if _TOKENIZER.eos_token_id is not None:
+                    _TOKENIZER.pad_token = _TOKENIZER.eos_token
+                    _TOKENIZER.pad_token_id = _TOKENIZER.eos_token_id
+                else:
+                    _TOKENIZER.add_special_tokens({"pad_token": "<|endoftext|>"})
+                    _TOKENIZER.pad_token_id = _TOKENIZER.convert_tokens_to_ids("<|endoftext|>")
+                logger.info(f"Set pad_token_id: {_TOKENIZER.pad_token_id}")
+            if _TOKENIZER.eos_token_id is None:
+                _TOKENIZER.eos_token = "<|endoftext|>"
+                _TOKENIZER.eos_token_id = _TOKENIZER.convert_tokens_to_ids("<|endoftext|>")
+                logger.info(f"Set eos_token_id: {_TOKENIZER.eos_token_id}")
+            logger.info("Tokenizer loaded successfully")
+        except Exception as e:
+            logger.error(f"Failed to load tokenizer: {type(e).__name__}: {e}")
+            raise
+    return _TOKENIZER
+
+def get_model():
+    global _MODEL
+    if _MODEL is None:
+        try:
+            logger.info("Loading model on CPU with FP16")
+            _MODEL = AutoModelForCausalLM.from_pretrained(
+                MODEL_NAME,
+                torch_dtype=torch.float16,
+                device_map="cpu",
+                low_cpu_mem_usage=True,
+            )
+            _MODEL.eval()
+            logger.info("Model loaded successfully")
+        except Exception as e:
+            logger.error(f"Failed to load model: {type(e).__name__}: {e}")
+            raise
+    return _MODEL
+
+# ------------ Prompt builder ------------
+def build_instructions(personality, level, topic):
+    return (
+        f"Plutus AI tutor for a {personality} learner at {level} level. "
+        f"Explain {topic} in a tone that fits learner's personality and tech level with examples. "
+        "Keep it 250–500 words. End with 'Takeaway:'."
+    )
+
+def build_model_input(tokenizer, personality, level, topic):
+    user_msg = build_instructions(personality, level, topic)
+    return f"System: You are a personalised Cardano Plutus tutor, your job is to make Plutus easy to learn based on different learner personalities, adapt your teaching style accordingly.\nUser: {user_msg}\nAssistant:"
+
+# ------------ Generation function ------------
+def generate(personality, level, topic, max_new_tokens=200):
+    try:
+        tokenizer = get_tokenizer()
+        model = get_model()
+        prompt = build_model_input(tokenizer, personality, level, topic)
+
+        device = torch.device("cpu")
+        logger.info("Generating on CPU")
+        inputs = tokenizer(prompt, return_tensors="pt")
+        input_len = inputs["input_ids"].shape[1]
+        inputs = {k: v.to(device) for k, v in inputs.items()}
+
+        start_time = time.time()
+        with torch.inference_mode():
+            outputs = model.generate(
+                **inputs,
+                max_new_tokens=max_new_tokens,
+                temperature=0.3,
+                top_p=0.4,
+                do_sample=True,
+                repetition_penalty=1.5,  # Increased to prevent repetition
+                eos_token_id=tokenizer.eos_token_id,
+                pad_token_id=tokenizer.pad_token_id,
+                stop_strings=["Takeaway:"]
+            )
+        logger.info(f"Generation took {time.time() - start_time:.2f} seconds")
+
+        gen_ids = outputs[0][input_len:]
+        text = tokenizer.decode(gen_ids, skip_special_tokens=True).strip()
+        if not text:
+            text = tokenizer.decode(outputs[0][input_len:], skip_special_tokens=True).strip()
+
+        # Remove prompt if present
+        if prompt in text:
+            text = text.replace(prompt, "").strip()
+
+        # Truncate at Takeaway
+        if "Takeaway:" in text:
+            text = text[:text.index("Takeaway:") + len("Takeaway:") + 100].strip()
+        elif not text.endswith("Takeaway:"):
+            text += "\nTakeaway: (Summary not fully generated due to token limit)."
+
+        return text if text else "Generation failed. Try regenerating."
+    except Exception as e:
+        logger.error(f"Generation error: {type(e).__name__}: {e}")
+        return f"Error during generation: {str(e)}. Try regenerating or using a smaller model."
+
+# ------------ Orchestrator with retry logic ------------
+def orchestrator(personality, level, topic, max_retries=3):
+    if not personality or not level or not topic:
+        return "Select personality, expertise, and topic to get an explanation."
+    
+    yield "Generating response, please wait (~1–2 minutes on CPU)..."
+    
+    for attempt in range(max_retries):
+        try:
+            logger.info(f"Generation attempt {attempt + 1}/{max_retries}")
+            result = generate(personality, level, topic)
+            if result.startswith("Error during generation"):
+                raise Exception(result)
+            return result
+        except Exception as e:
+            logger.error(f"Attempt {attempt + 1}/{max_retries} failed: {type(e).__name__}: {e}")
+            if attempt == max_retries - 1:
+                return (
+                    "Failed to generate after multiple attempts. "
+                    "Click **Regenerate** or try again later. "
+                    "If this persists, try a smaller model or check logs for errors."
+                )
+
+# ------------ Gradio UI ------------
+with gr.Blocks(theme="default") as iface:
+    gr.Markdown(
+        "## Cardano Plutus AI Assistant\n"
+        "Select **Learning Personality**, **Expertise Level**, and **Topic**, then click **Generate**. "
+        "Note: Generation may take ~1–2 minutes on CPU."
+    )
+
+    with gr.Row():
+        personality = gr.Dropdown(
+            choices=["Dyslexic", "Autistic", "Expressive"],
+            label="Learning Personality",
+            value=None,
+            allow_custom_value=False,
+            scale=1,
+        )
+        level = gr.Dropdown(
+            choices=["Beginner", "Intermediate", "Advanced"],
+            label="Expertise Level",
+            value=None,
+            allow_custom_value=False,
+            scale=1,
+        )
+        topic = gr.Dropdown(
+            choices=[
+                "What is Plutus?",
+                "Smart Contracts in Plutus",
+                "Cardano Blockchain",
+                "What is a Validator Script?",
+                "Plutus Tx",
+                "Datum and Redeemer",
+                "Time Handling in Plutus",
+                "Off-Chain Code",
+                "On-Chain Constraints",
+                "Plutus Core",
+                "Transaction Validation",
+                "Cardano Node Integration",
+            ],
+            label="Topic",
+            value=None,
+            allow_custom_value=False,
+            scale=2,
+        )
+
+    with gr.Row():
+        generate_btn = gr.Button("Generate")
+        regen = gr.Button("🔁 Regenerate")
+
+    output = gr.Textbox(
+        label="Model Response",
+        lines=12,
+        interactive=False,
+        show_copy_button=True,
+        placeholder="Your tailored explanation will appear here…",
+    )
+
+    generate_btn.click(orchestrator, [personality, level, topic], output, queue=True)
+    regen.click(orchestrator, [personality, level, topic], output, queue=True)
+
+# Enable queue
+iface.queue()
+
+if __name__ == "__main__":
+    iface.launch(server_name="0.0.0.0", server_port=7860)