Delete app.py

app.py

@@ -1,301 +0,0 @@
-import gradio as gr
-import torch
-from transformers import AutoTokenizer, AutoModelForCausalLM
-import time
-import os
-
-class ConversationManager:
-    def __init__(self):
-        self.models = {}
-        self.conversation = []
-        self.delay = 3
-        self.is_paused = False
-        self.current_model = None
-        self.initial_prompt = ""
-        self.task_complete = False  # New attribute for task completion
-
-    def load_model(self, model_name):
-        if model_name in self.models:
-            return self.models[model_name]
-
-        try:
-            print(f"Attempting to load model: {model_name}")
-            tokenizer = AutoTokenizer.from_pretrained(model_name)
-            model = AutoModelForCausalLM.from_pretrained(model_name, device_map="auto", load_in_8bit=True)
-            self.models[model_name] = (model, tokenizer)
-            print(f"Successfully loaded model: {model_name}")
-            return self.models[model_name]
-        except Exception as e:
-            print(f"Failed to load model {model_name}: {e}")
-            print(f"Error type: {type(e).__name__}")
-            print(f"Error details: {str(e)}")
-            return None
-
-    def generate_response(self, model_name, prompt):
-        model, tokenizer = self.load_model(model_name)
-    
-        # Format the prompt based on the model
-        if "llama" in model_name.lower():
-            formatted_prompt = self.format_llama2_prompt(prompt)
-        else:
-            formatted_prompt = self.format_general_prompt(prompt)
-    
-        inputs = tokenizer(formatted_prompt, return_tensors="pt", max_length=1024, truncation=True)
-        with torch.no_grad():
-            outputs = model.generate(**inputs, max_length=200, num_return_sequences=1, do_sample=True)
-        return tokenizer.decode(outputs[0], skip_special_tokens=True)
-
-    def format_llama2_prompt(self, prompt):
-        B_INST, E_INST = "[INST]", "[/INST]"
-        B_SYS, E_SYS = "<<SYS>>\n", "\n<</SYS>>\n\n"
-        system_prompt = "You are a helpful AI assistant. Please provide a concise and relevant response."
-    
-        formatted_prompt = f"{B_INST} {B_SYS}{system_prompt}{E_SYS}{prompt.strip()} {E_INST}"
-        return formatted_prompt
-
-    def format_general_prompt(self, prompt):
-        # A general format that might work for other models
-        return f"Human: {prompt.strip()}\n\nAssistant:"
-    def add_to_conversation(self, model_name, response):
-        self.conversation.append((model_name, response))
-        if "task complete?" in response.lower():  # Check for task completion marker
-            self.task_complete = True
-
-    def get_conversation_history(self):
-        return "\n".join([f"{model}: {msg}" for model, msg in self.conversation])
-
-    def clear_conversation(self):
-        self.conversation = []
-        self.initial_prompt = ""
-        self.models = {}
-        self.current_model = None
-        self.task_complete = False  # Reset task completion status
-
-    def rewind_conversation(self, steps):
-        self.conversation = self.conversation[:-steps]
-        self.task_complete = False  # Reset task completion status after rewinding
-
-    def rewind_and_insert(self, steps, inserted_response):
-        if steps > 0:
-            self.conversation = self.conversation[:-steps]
-        if inserted_response.strip():
-            last_model = self.conversation[-1][0] if self.conversation else "User"
-            next_model = "Model 1" if last_model == "Model 2" or last_model == "User" else "Model 2"
-            self.conversation.append((next_model, inserted_response))
-        self.current_model = last_model
-        self.task_complete = False  # Reset task completion status after rewinding and inserting
-
-manager = ConversationManager()
-
-def get_model(dropdown, custom):
-    model = custom if custom.strip() else dropdown
-    return (model, model)  # Return a tuple (label, value)
-
-def chat(model1, model2, user_input, history, inserted_response=""):
-    try:
-        model1 = get_model(model1, model1_custom.value)[0]
-        model2 = get_model(model2, model2_custom.value)[0]
-        
-        if not manager.load_model(model1) or not manager.load_model(model2):
-            return "Error: Failed to load one or both models. Please check the model names and try again.", ""
-        
-        if not manager.conversation:
-            manager.initial_prompt = user_input
-            manager.clear_conversation()
-            manager.add_to_conversation("User", user_input)
-        
-        models = [model1, model2]
-        current_model_index = 0 if manager.current_model in ["User", "Model 2"] else 1
-        
-        while not manager.task_complete:
-            if manager.is_paused:
-                yield history, "Conversation paused."
-                return
-
-            model = models[current_model_index]
-            manager.current_model = model
-            
-            if inserted_response and current_model_index == 0:
-                response = inserted_response
-                inserted_response = ""
-            else:
-                conversation_history = manager.get_conversation_history()
-                prompt = f"{conversation_history}\n\nPlease continue the conversation. If you believe the task is complete, end your response with 'Task complete?'"
-                response = manager.generate_response(model, prompt)
-            
-            manager.add_to_conversation(model, response)
-            history = manager.get_conversation_history()
-            
-            for i in range(manager.delay, 0, -1):
-                yield history, f"{model} is writing... {i}"
-                time.sleep(1)
-            
-            yield history, ""
-            
-            if manager.task_complete:
-                yield history, "Models believe the task is complete. Are you satisfied with the result? (Yes/No)"
-                return
-            
-            current_model_index = (current_model_index + 1) % 2
-        
-        return history, "Conversation completed."
-    except Exception as e:
-        print(f"Error in chat function: {str(e)}")
-        return f"An error occurred: {str(e)}", ""
-
-def user_satisfaction(satisfied, history):
-    if satisfied.lower() == 'yes':
-        return history, "Task completed successfully."
-    else:
-        manager.task_complete = False
-        return history, "Continuing the conversation..."
-
-def pause_conversation():
-    manager.is_paused = True
-    return "Conversation paused. Press Resume to continue."
-
-def resume_conversation():
-    manager.is_paused = False
-    return "Conversation resumed."
-
-def edit_response(edited_text):
-    if manager.conversation:
-        manager.conversation[-1] = (manager.current_model, edited_text)
-    manager.task_complete = False  # Reset task completion status after editing
-    return manager.get_conversation_history()
-
-def restart_conversation(model1, model2, user_input):
-    manager.clear_conversation()
-    return chat(model1, model2, user_input, "")
-
-def rewind_and_insert(steps, inserted_response, history):
-    manager.rewind_and_insert(int(steps), inserted_response)
-    return manager.get_conversation_history(), ""
-
-# This list should be populated with the exact model names when available
-open_source_models = [
-    "meta-llama/Llama-2-7b-chat-hf",
-    "meta-llama/Llama-2-13b-chat-hf",
-    "meta-llama/Llama-2-70b-chat-hf",
-    "mistralai/Mixtral-8x7B-Instruct-v0.1",
-    "bigcode/starcoder2-15b",
-    "bigcode/starcoder2-3b",
-    "tiiuae/falcon-7b",
-    "tiiuae/falcon-40b",
-    "EleutherAI/gpt-neox-20b",
-    "google/flan-ul2",
-    "stabilityai/stablelm-zephyr-3b",
-    "HuggingFaceH4/zephyr-7b-beta",
-    "microsoft/phi-2",
-    "google/gemma-7b-it"
-]
-
-with gr.Blocks() as demo:
-    gr.Markdown("# ConversAI Playground")
-    
-    with gr.Row():
-        with gr.Column(scale=1):
-            model1_dropdown = gr.Dropdown(choices=open_source_models, label="Model 1")
-            model1_custom = gr.Textbox(label="Custom Model 1")
-        with gr.Column(scale=1):
-            model2_dropdown = gr.Dropdown(choices=open_source_models, label="Model 2")
-            model2_custom = gr.Textbox(label="Custom Model 2")
-    
-    user_input = gr.Textbox(label="Initial prompt", lines=2)
-    chat_history = gr.Textbox(label="Conversation", lines=20)
-    current_response = gr.Textbox(label="Current model response", lines=3)
-    
-    with gr.Row():
-        pause_btn = gr.Button("Pause")
-        edit_btn = gr.Button("Edit")
-        rewind_btn = gr.Button("Rewind")
-        resume_btn = gr.Button("Resume")
-        restart_btn = gr.Button("Restart")
-        clear_btn = gr.Button("Clear")
-    
-    with gr.Row():
-        rewind_steps = gr.Slider(0, 10, 1, label="Steps to rewind")
-        inserted_response = gr.Textbox(label="Insert response after rewind", lines=2)
-    
-    delay_slider = gr.Slider(0, 10, 3, label="Response Delay (seconds)")
-
-    user_satisfaction_input = gr.Textbox(label="Are you satisfied with the result? (Yes/No)", visible=False)
-
-
-    gr.Markdown("""
-    ## Button Descriptions
-    - **Pause**: Temporarily stops the conversation. The current model will finish its response.
-    - **Edit**: Allows you to modify the last response in the conversation.
-    - **Rewind**: Removes the specified number of last responses from the conversation.
-    - **Resume**: Continues the conversation from where it was paused.
-    - **Restart**: Begins a new conversation with the same or different models, keeping the initial prompt.
-    - **Clear**: Resets everything, including loaded models, conversation history, and initial prompt.
-    """)
-
-    def on_chat_update(history, response):
-        if "Models believe the task is complete" in response:
-            return gr.update(visible=True), gr.update(visible=False)
-        return gr.update(visible=False), gr.update(visible=True)
-
-    start_btn = gr.Button("Start Conversation")
-    chat_output = start_btn.click(
-        chat,
-        inputs=[
-            model1_dropdown,
-            model2_dropdown,
-            user_input,
-            chat_history
-        ],
-        outputs=[chat_history, current_response]
-    )
-    
-    chat_output.then(
-        on_chat_update,
-        inputs=[chat_history, current_response],
-        outputs=[user_satisfaction_input, start_btn]
-    )
-    
-    user_satisfaction_input.submit(
-        user_satisfaction,
-        inputs=[user_satisfaction_input, chat_history],
-        outputs=[chat_history, current_response]
-    ).then(
-        chat,
-        inputs=[
-            model1_dropdown,
-            model2_dropdown,
-            user_input,
-            chat_history
-        ],
-        outputs=[chat_history, current_response]
-    )
-    
-    pause_btn.click(pause_conversation, outputs=[current_response])
-    resume_btn.click(
-        chat,
-        inputs=[
-            model1_dropdown,
-            model2_dropdown,
-            user_input,
-            chat_history,
-            inserted_response
-        ],
-        outputs=[chat_history, current_response]
-    )
-    edit_btn.click(edit_response, inputs=[current_response], outputs=[chat_history])
-    rewind_btn.click(rewind_and_insert, inputs=[rewind_steps, inserted_response, chat_history], outputs=[chat_history, current_response])
-    restart_btn.click(
-        restart_conversation,
-        inputs=[
-            model1_dropdown,
-            model2_dropdown,
-            user_input
-        ],
-        outputs=[chat_history, current_response]
-    )
-    clear_btn.click(manager.clear_conversation, outputs=[chat_history, current_response, user_input])
-    delay_slider.change(lambda x: setattr(manager, 'delay', x), inputs=[delay_slider])
-
-if __name__ == "__main__":
-    demo.launch()
-