Update app.py

app.py

@@ -2,18 +2,24 @@ import gradio as gr
 from huggingface_hub import InferenceClient
 import json
 import os
+import shutil
+import pandas as pd
+from sklearn.feature_extraction.text import CountVectorizer
+from sklearn.ensemble import RandomForestClassifier
+from sklearn.pipeline import Pipeline
+import joblib
 
-"""
-For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-"""
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
+# Initialize the HuggingFace API Client with a valid model
+# Replace 'gpt-3.5-turbo' with your desired model if different
+client = InferenceClient("gpt-3.5-turbo")
 
 # Persistent memory and knowledge base setup
 memory_file = "chat_memory.json"
-knowledge_base = {
-    "AI": "Artificial Intelligence is a branch of computer science that focuses on creating systems capable of performing tasks that typically require human intelligence.",
-    "Quantum Computing": "Quantum computing is a type of computation that uses quantum mechanics to process information in ways classical computers cannot.",
-}
+knowledge_base_dir = "knowledge_base"
+model_file = "chat_model.pkl"
+
+# Ensure directories exist
+os.makedirs(knowledge_base_dir, exist_ok=True)
 
 # Load memory from file
 def load_memory():
@@ -25,109 +31,223 @@ def load_memory():
 # Save memory to file
 def save_memory(memory):
     with open(memory_file, "w") as f:
-        json.dump(memory, f)
+        json.dump(memory, f, indent=2)
 
 # Append to memory
 def update_memory(conversation):
     memory = load_memory()
     memory.append(conversation)
+    # Optionally limit memory size
+    if len(memory) > 1000:
+        memory = memory[-1000:]
     save_memory(memory)
 
-# Response generation with memory and knowledge base integration
-def respond(
-    message,
-    history: list[tuple[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-):
-    messages = [{"role": "system", "content": system_message}]
-
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
-
-    messages.append({"role": "user", "content": message})
+# Load or initialize the ML model
+def load_or_initialize_model():
+    if os.path.exists(model_file):
+        return joblib.load(model_file)
+    return Pipeline([
+        ("vectorizer", CountVectorizer()),
+        ("classifier", RandomForestClassifier(n_estimators=100, random_state=42))
+    ])
+
+# Retrain model on files in the knowledge base
+def train_model_on_files():
+    model = load_or_initialize_model()
+    texts, labels = [], []
+
+    # Load data from the knowledge base
+    for file_name in os.listdir(knowledge_base_dir):
+        file_path = os.path.join(knowledge_base_dir, file_name)
+        if file_path.endswith(".csv"):
+            try:
+                df = pd.read_csv(file_path)
+                if "text" in df.columns and "label" in df.columns:
+                    texts.extend(df["text"].astype(str).tolist())
+                    labels.extend(df["label"].astype(str).tolist())
+                else:
+                    return f"File '{file_name}' does not contain required 'text' and 'label' columns."
+            except Exception as e:
+                return f"Error reading '{file_name}': {str(e)}"
+
+    if texts and labels:
+        try:
+            model.fit(texts, labels)
+            joblib.dump(model, model_file)
+            return f"Model trained on {len(texts)} samples from {len(os.listdir(knowledge_base_dir))} files."
+        except Exception as e:
+            return f"Error during model training: {str(e)}"
+    return "No valid training data found in the knowledge base."
+
+# Chat response function
+def respond(message, history, system_message, max_tokens, temperature, top_p):
+    # Load or initialize model
+    model = load_or_initialize_model()
 
     # Check for answers in the knowledge base
-    if message in knowledge_base:
-        response = knowledge_base[message]
-        update_memory((message, response))
-        yield response
-        return
+    try:
+        pred_label = model.predict([message])[0]
+        response = f"Predicted response: {pred_label}"
+        update_memory({"user": message, "assistant": response})
+        return response
+    except Exception:
+        pass  # Continue with GPT model if ML model doesn't have a response
+
+    # Generate response using GPT
+    messages = [{"role": "system", "content": system_message}]
+    for turn in history:
+        if turn["user"]:
+            messages.append({"role": "user", "content": turn["user"]})
+        if turn["assistant"]:
+            messages.append({"role": "assistant", "content": turn["assistant"]})
+    messages.append({"role": "user", "content": message})
 
-    # Generate response from AI
     response = ""
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        token = message.choices[0].delta.content
-        response += token
-        yield response
+    try:
+        for message_part in client.chat_completion(
+            messages,
+            max_tokens=max_tokens,
+            stream=True,
+            temperature=temperature,
+            top_p=top_p,
+        ):
+            token = message_part.get("choices", [{}])[0].get("delta", {}).get("content", "")
+            response += token
+    except Exception as e:
+        response = f"Error generating response: {str(e)}"
+        update_memory({"user": message, "assistant": response})
+        return response
 
     # Update memory
-    update_memory((message, response))
-
-
-# Gradio interface with enhanced functionality
-def add_to_knowledge_base(key, value):
-    knowledge_base[key] = value
-    return f"Added to knowledge base: {key} -> {value}"
-
-
-demo = gr.Blocks()
-
-with demo:
-    gr.Markdown("# Advanced Chatbot with Memory and Knowledge Base")
-
-    with gr.Tab("Chat"):
-        chatbot = gr.ChatInterface(
-            respond,
-            additional_inputs=[
-                gr.Textbox(value="You are an advanced AI Chatbot.", label="System message"),
-                gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-                gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-                gr.Slider(
-                    minimum=0.1,
-                    maximum=1.0,
-                    value=0.95,
-                    step=0.05,
-                    label="Top-p (nucleus sampling)",
-                ),
-            ],
+    update_memory({"user": message, "assistant": response})
+    return response
+
+# Gradio interface
+with gr.Blocks() as demo:
+    gr.Markdown("# 🧠 Advanced AI Chatbot with Knowledge Base and Model Training")
+
+    with gr.Tab("💬 Chat"):
+        chatbot = gr.Chatbot(label="AI Chatbot").style(height=600)
+        with gr.Row():
+            with gr.Column(scale=0.85):
+                user_input = gr.Textbox(
+                    label="Your Message",
+                    placeholder="Type your message here...",
+                )
+            with gr.Column(scale=0.15, min_width=100):
+                send_button = gr.Button("Send", variant="primary")
+        with gr.Row():
+            system_message = gr.Textbox(
+                value="You are an advanced AI Chatbot.",
+                label="System Message",
+                visible=False  # Hidden if default system message is used
+            )
+            max_tokens = gr.Slider(
+                minimum=100, maximum=2048, value=512, step=100, label="Max Tokens"
+            )
+            temperature = gr.Slider(
+                minimum=0.1, maximum=1.0, value=0.7, step=0.1, label="Temperature"
+            )
+            top_p = gr.Slider(
+                minimum=0.1,
+                maximum=1.0,
+                value=0.95,
+                step=0.05,
+                label="Top-p (Nucleus Sampling)",
+            )
+
+        def handle_message(message, history, system_message, max_tokens, temperature, top_p):
+            response = respond(message, history, system_message, max_tokens, temperature, top_p)
+            history.append({"user": message, "assistant": response})
+            return history, history
+
+        send_button.click(
+            handle_message,
+            inputs=[user_input, chatbot, system_message, max_tokens, temperature, top_p],
+            outputs=[chatbot, chatbot],
+        )
+        user_input.submit(
+            handle_message,
+            inputs=[user_input, chatbot, system_message, max_tokens, temperature, top_p],
+            outputs=[chatbot, chatbot],
         )
 
-    with gr.Tab("Knowledge Base"):
-        gr.Markdown("### Manage the Knowledge Base")
-        kb_key = gr.Textbox(label="Key", placeholder="Enter the topic or question")
-        kb_value = gr.Textbox(label="Value", placeholder="Enter the explanation or answer")
-        add_kb_button = gr.Button("Add to Knowledge Base")
-        kb_output = gr.Textbox(label="Knowledge Base Output")
-
-        add_kb_button.click(add_to_knowledge_base, [kb_key, kb_value], kb_output)
-
-    with gr.Tab("Memory"):
-        gr.Markdown("### Conversation Memory")
-        memory_display = gr.Textbox(label="Conversation Memory", lines=10)
-        refresh_memory = gr.Button("Refresh Memory")
-        clear_memory = gr.Button("Clear Memory")
+    with gr.Tab("📚 Knowledge Base"):
+        gr.Markdown("### Manage Knowledge Base")
+        file_upload = gr.File(
+            label="Upload CSV File",
+            file_types=[".csv"],
+            multiple=False,
+            interactive=True,
+        )
+        upload_output = gr.Textbox(label="Upload Result", interactive=False)
+        train_button = gr.Button("🔄 Train Model on Knowledge Base")
+        train_output = gr.Textbox(label="Training Result", interactive=False)
+
+        def upload_file(file):
+            if file is None:
+                return "No file uploaded."
+            try:
+                # Validate file extension
+                if not file.name.endswith(".csv"):
+                    return "Invalid file type. Please upload a CSV file."
+                # Save file to knowledge base directory
+                destination_path = os.path.join(knowledge_base_dir, file.name)
+                with open(destination_path, "wb") as f:
+                    f.write(file.read())
+                return f"File '{file.name}' uploaded successfully."
+            except Exception as e:
+                return f"Error uploading file: {str(e)}"
+
+        file_upload.change(upload_file, inputs=file_upload, outputs=upload_output)
+        train_button.click(train_model_on_files, inputs=None, outputs=train_output)
+
+    with gr.Tab("🧠 Memory"):
+        gr.Markdown("### View and Manage Conversation Memory")
+        memory_display = gr.JSON(label="Conversation Memory", interactive=False)
+        with gr.Row():
+            refresh_memory = gr.Button("🔄 Refresh Memory")
+            clear_memory = gr.Button("🗑️ Clear Memory")
+            export_memory = gr.Button("📤 Export Memory")
+        export_output = gr.File(label="Download Memory", visible=False)
 
         def display_memory():
-            return json.dumps(load_memory(), indent=2)
+            return load_memory()
 
         def clear_memory_func():
             save_memory([])
-            return "Memory Cleared!"
-
-        refresh_memory.click(display_memory, outputs=memory_display)
-        clear_memory.click(clear_memory_func, outputs=memory_display)
+            return []
+
+        def export_memory_func():
+            if os.path.exists(memory_file):
+                return memory_file
+            return None
+
+        refresh_memory.click(display_memory, inputs=None, outputs=memory_display)
+        clear_memory.click(clear_memory_func, inputs=None, outputs=memory_display)
+        export_memory.click(export_memory_func, inputs=None, outputs=export_output)
+
+    with gr.Tab("💾 Download Model"):
+        gr.Markdown("### Download the Trained Model")
+        download_button = gr.Button("📥 Download Model")
+        model_download_output = gr.File(label="Downloadable Model", interactive=False)
+
+        def download_model():
+            if os.path.exists(model_file):
+                return model_file
+            return None
+
+        download_button.click(download_model, inputs=None, outputs=model_download_output)
+
+    with gr.Tab("⚙️ Settings"):
+        gr.Markdown("### Application Settings")
+        # Additional settings can be added here
+        gr.Textbox(
+            value="",
+            label="Settings Placeholder",
+            placeholder="Add settings here...",
+            interactive=False,
+        )
 
 if __name__ == "__main__":
     demo.launch()