Update app.py

app.py

@@ -1,344 +1,280 @@
 import os
 import subprocess
 import random
-from huggingface_hub import InferenceClient
-import gradio as gr
-from safe_search import safe_search
-from i_search import google
-from i_search import i_search as i_s
-from agent import (
-    ACTION_PROMPT,
-    ADD_PROMPT,
-    COMPRESS_HISTORY_PROMPT,
-    LOG_PROMPT,
-    LOG_RESPONSE,
-    MODIFY_PROMPT,
-    PREFIX,
-    SEARCH_QUERY,
-    READ_PROMPT,
-    TASK_PROMPT,
-    UNDERSTAND_TEST_RESULTS_PROMPT,
-)
-from utils import parse_action, parse_file_content, read_python_module_structure
+import time
+from typing import Dict, List, Tuple
 from datetime import datetime
+import logging
 
-now = datetime.now()
-date_time_str = now.strftime("%Y-%m-%d %H:%M:%S")
+import gradio as gr
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
+from huggingface_hub import InferenceClient, cached_download, Repository, HfApi
+from IPython.display import display, HTML
 
-client = InferenceClient(
-    "mistralai/Mixtral-8x7B-Instruct-v0.1"
+# --- Configuration ---
+VERBOSE = True
+MAX_HISTORY = 5
+MAX_TOKENS = 2048
+TEMPERATURE = 0.7
+TOP_P = 0.8
+REPETITION_PENALTY = 1.5
+DEFAULT_PROJECT_PATH = "./my-hf-project"  # Default project directory
+
+# --- Logging Setup ---
+logging.basicConfig(
+    filename="app.log",
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
 )
 
-############################################
-
-VERBOSE = True
-MAX_HISTORY = 100
-# MODEL = "gpt-3.5-turbo"  # "gpt-4"
-
-def format_prompt(message, history):
-    prompt = "<s>"
-    for user_prompt, bot_response in history:
-        prompt += f"[INST] {user_prompt} [/INST]"
-        prompt += f" {bot_response}</s> "
-    prompt += f"[INST] {message} [/INST]"
+# --- Global Variables ---
+current_model = None  # Store the currently loaded model
+repo = None  # Store the Hugging Face Repository object
+model_descriptions = {}  # Store model descriptions
+
+# --- Functions ---
+def format_prompt(message: str, history: List[Tuple[str, str]], max_history_turns: int = 2) -> str:
+    prompt = ""
+    for user_prompt, bot_response in history[-max_history_turns:]:
+        prompt += f"Human: {user_prompt}\nAssistant: {bot_response}\n"
+    prompt += f"Human: {message}\nAssistant:"
     return prompt
 
-def run_gpt(prompt_template, stop_tokens, max_tokens, purpose, **prompt_kwargs):
-    seed = random.randint(1, 1111111111111111)
-    print(seed)
-    generate_kwargs = dict(
-        temperature=1.0,
-        max_new_tokens=2096,
-        top_p=0.99,
-        repetition_penalty=1.0,
-        do_sample=True,
-        seed=seed,
-    )
-
-    content = PREFIX.format(
+def generate_response(
+    prompt: str,
+    history: List[Tuple[str, str]],
+    agent_name: str = "Generic Agent",
+    sys_prompt: str = "",
+    temperature: float = TEMPERATURE,
+    max_new_tokens: int = MAX_TOKENS,
+    top_p: float = TOP_P,
+    repetition_penalty: float = REPETITION_PENALTY,
+) -> str:
+    global current_model
+    if current_model is None:
+        return "Error: Please load a model first."
+
+    date_time_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    full_prompt = PREFIX.format(
         date_time_str=date_time_str,
-        purpose=purpose,
-        safe_search=safe_search,
-    ) + prompt_template.format(**prompt_kwargs)
+        purpose=sys_prompt,
+        agent_name=agent_name
+    ) + format_prompt(prompt, history)
+
     if VERBOSE:
-        print(LOG_PROMPT.format(content))
+        logging.info(LOG_PROMPT.format(content=full_prompt))
 
-    # formatted_prompt = format_prompt(f"{system_prompt}, {prompt}", history)
-    # formatted_prompt = format_prompt(f'{content}', history)
+    response = current_model(
+        full_prompt,
+        max_new_tokens=max_new_tokens,
+        temperature=temperature,
+        top_p=top_p,
+        repetition_penalty=repetition_penalty,
+        do_sample=True
+    )[0]['generated_text']
 
-    stream = client.text_generation(content, **generate_kwargs, stream=True, details=True, return_full_text=False)
-    resp = ""
-    for response in stream:
-        resp += response.token.text
+    assistant_response = response.split("Assistant:")[-1].strip()
 
     if VERBOSE:
-        print(LOG_RESPONSE.format(resp))
-    return resp
-
-def compress_history(purpose, task, history, directory):
-    resp = run_gpt(
-        COMPRESS_HISTORY_PROMPT,
-        stop_tokens=["observation:", "task:", "action:", "thought:"],
-        max_tokens=512,
-        purpose=purpose,
-        task=task,
-        history=history,
-    )
-    history = "observation: {}\n".format(resp)
-    return history
-
-def call_search(purpose, task, history, directory, action_input):
-    print("CALLING SEARCH")
+        logging.info(LOG_RESPONSE.format(resp=assistant_response))
+
+    return assistant_response
+
+def load_hf_model(model_name: str):
+    """Loads a language model and fetches its description."""
+    global current_model, model_descriptions
     try:
-        if "http" in action_input:
-            if "<" in action_input:
-                action_input = action_input.strip("<")
-            if ">" in action_input:
-                action_input = action_input.strip(">")
-
-            response = i_s(action_input)
-            # response = google(search_return)
-            print(response)
-            history += "observation: search result is: {}\n".format(response)
-        else:
-            history += "observation: I need to provide a valid URL to 'action: SEARCH action_input=https://URL'\n"
+        tokenizer = AutoTokenizer.from_pretrained(model_name)
+        current_model = pipeline(
+            "text-generation",
+            model=model_name,
+            tokenizer=tokenizer,
+            model_kwargs={"load_in_8bit": True}
+        )
+
+        # Fetch and store the model description
+        api = HfApi()
+        model_info = api.model_info(model_name)
+        model_descriptions[model_name] = model_info.pipeline_tag
+        return f"Successfully loaded model: {model_name}"
     except Exception as e:
-        history += "observation: {}'\n".format(e)
-    return "MAIN", None, history, task
-
-def call_main(purpose, task, history, directory, action_input):
-    resp = run_gpt(
-        ACTION_PROMPT,
-        stop_tokens=["observation:", "task:", "action:","thought:"],
-        max_tokens=2096,
-        purpose=purpose,
-        task=task,
-        history=history,
-    )
-    lines = resp.strip().strip("\n").split("\n")
-    for line in lines:
-        if line == "":
-            continue
-        if line.startswith("thought: "):
-            history += "{}\n".format(line)
-        elif line.startswith("action: "):
-            action_name, action_input = parse_action(line)
-            print(f'ACTION_NAME :: {action_name}')
-            print(f'ACTION_INPUT :: {action_input}')
-
-            history += "{}\n".format(line)
-            if "COMPLETE" in action_name or "COMPLETE" in action_input:
-                task = "END"
-                return action_name, action_input, history, task
-            else:
-                return action_name, action_input, history, task
+        return f"Error loading model: {str(e)}"
+
+def execute_command(command: str, project_path: str = None) -> str:
+    """Executes a shell command and returns the output."""
+    try:
+        if project_path:
+            process = subprocess.Popen(command, shell=True, cwd=project_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         else:
-            history += "{}\n".format(line)
-            # history += "observation: the following command did not produce any useful output: '{}', I need to check the commands syntax, or use a different command\n".format(line)
-
-            # return action_name, action_input, history, task
-            # assert False, "unknown action: {}".format(line)
-    return "MAIN", None, history, task
-
-def call_set_task(purpose, task, history, directory, action_input):
-    task = run_gpt(
-        TASK_PROMPT,
-        stop_tokens=[],
-        max_tokens=64,
-        purpose=purpose,
-        task=task,
-        history=history,
-    ).strip("\n")
-    history += "observation: task has been updated to: {}\n".format(task)
-    return "MAIN", None, history, task
-
-def end_fn(purpose, task, history, directory, action_input):
-    task = "END"
-    return "COMPLETE", "COMPLETE", history, task
-
-NAME_TO_FUNC = {
-    "MAIN": call_main,
-    "UPDATE-TASK": call_set_task,
-    "SEARCH": call_search,
-    "COMPLETE": end_fn,
-}
-
-def run_action(purpose, task, history, directory, action_name, action_input):
-    print(f'action_name::{action_name}')
+            process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        output, error = process.communicate()
+        if error:
+            return f"Error: {error.decode('utf-8')}"
+        return output.decode("utf-8")
+    except Exception as e:
+        return f"Error executing command: {str(e)}"
+
+def create_hf_project(project_name: str, project_path: str = DEFAULT_PROJECT_PATH):
+    """Creates a new Hugging Face project."""
+    global repo
     try:
-        if "RESPONSE" in action_name or "COMPLETE" in action_name:
-            action_name = "COMPLETE"
-            task = "END"
-            return action_name, "COMPLETE", history, task
-
-        # compress the history when it is long
-        if len(history.split("\n")) > MAX_HISTORY:
-            if VERBOSE:
-                print("COMPRESSING HISTORY")
-            history = compress_history(purpose, task, history, directory)
-        if not action_name in NAME_TO_FUNC:
-            action_name = "MAIN"
-        if action_name == "" or action_name == None:
-            action_name = "MAIN"
-        assert action_name in NAME_TO_FUNC
-
-        print("RUN: ", action_name, action_input)
-        return NAME_TO_FUNC[action_name](purpose, task, history, directory, action_input)
+        if os.path.exists(project_path):
+            return f"Error: Directory '{project_path}' already exists!"
+        # Create the repository
+        repo = Repository(local_dir=project_path, clone_from=None)
+        repo.git_init()
+
+        # Add basic files (optional, you can customize this)
+        with open(os.path.join(project_path, "README.md"), "w") as f:
+            f.write(f"# {project_name}\n\nA new Hugging Face project.")
+
+        # Stage all changes
+        repo.git_add(pattern="*")
+        repo.git_commit(commit_message="Initial commit")
+
+        return f"Hugging Face project '{project_name}' created successfully at '{project_path}'"
     except Exception as e:
-        history += "observation: the previous command did not produce any useful output, I need to check the commands syntax, or use a different command\n"
-        return "MAIN", None, history, task
-
-def run(purpose, history):
-    # print(purpose)
-    # print(hist)
-    task = None
-    directory = "./"
-    if history:
-        history = str(history).strip("[]")
-    if not history:
-        history = ""
-
-    action_name = "UPDATE-TASK" if task is None else "MAIN"
-    action_input = None
-    while True:
-        print("")
-        print("")
-        print("---")
-        print("purpose:", purpose)
-        print("task:", task)
-        print("---")
-        print(history)
-        print("---")
-
-        action_name, action_input, history, task = run_action(
-            purpose,
-            task,
-            history,
-            directory,
-            action_name,
-            action_input,
-        )
-        yield (history)
-        # yield ("",[(purpose,history)])
-        if task == "END":
-            return (history)
-            # return ("", [(purpose,history)])
-
-################################################
-
-def format_prompt(message, history):
-    prompt = "<s>"
-    for user_prompt, bot_response in history:
-        prompt += f"[INST] {user_prompt} [/INST]"
-        prompt += f" {bot_response}</s> "
-    prompt += f"[INST] {message} [/INST]"
-    return prompt
+        return f"Error creating Hugging Face project: {str(e)}"
 
-AGENTS = [
-    "WEB_DEV",
-    "AI_SYSTEM_PROMPT",
-    "PYTHON_CODE_DEV"
-]
-
-def generate(prompt, history, agent_name=AGENTS[0], sys_prompt="", temperature=0.9, max_new_tokens=256, top_p=0.95, repetition_penalty=1.0):
-    seed = random.randint(1, 1111111111111111)
-
-    agent = prompts.WEB_DEV
-    if agent_name == "WEB_DEV":
-        agent = prompts.WEB_DEV
-    if agent_name == "AI_SYSTEM_PROMPT":
-        agent = prompts.AI_SYSTEM_PROMPT
-    if agent_name == "PYTHON_CODE_DEV":
-        agent = prompts.PYTHON_CODE_DEV
-    system_prompt = agent
-    temperature = float(temperature)
-    if temperature < 1e-2:
-        temperature = 1e-2
-    top_p = float(top_p)
-
-    generate_kwargs = dict(
-        temperature=temperature,
-        max_new_tokens=max_new_tokens,
-        top_p=top_p,
-        repetition_penalty=repetition_penalty,
-        do_sample=True,
-        seed=seed,
-    )
-
-    formatted_prompt = format_prompt(f"{system_prompt}, {prompt}", history)
-    stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=False)
-    output = ""
-
-    for response in stream:
-        output += response.token.text
-        yield output
-    return output
-
-additional_inputs = [
-    gr.Dropdown(
-        label="Agents",
-        choices=[s for s in AGENTS],
-        value=AGENTS[0],
-        interactive=True,
-    ),
-    gr.Textbox(
-        label="System Prompt",
-        max_lines=1,
-        interactive=True,
-    ),
-    gr.Slider(
-        label="Temperature",
-        value=0.9,
-        minimum=0.0,
-        maximum=1.0,
-        step=0.05,
-        interactive=True,
-        info="Higher values generate more diverse outputs.",
-    ),
-    gr.Slider(
-        label="Max New Tokens",
-        value=2048,
-        minimum=64,
-        maximum=4096,
-        step=64,
-        interactive=True,
-        info="The maximum number of new tokens to generate.",
-    ),
-    gr.Slider(
-        label="Top-p (Nucleus Sampling)",
-        value=0.90,
-        minimum=0.0,
-        maximum=1,
-        step=0.05,
-        interactive=True,
-        info="Higher values sample more low-probability tokens.",
-    ),
-    gr.Slider(
-        label="Repetition Penalty",
-        value=1.2,
-        minimum=1.0,
-        maximum=2.0,
-        step=0.05,
-        interactive=True,
-        info="Penalize repeated tokens.",
-    )
-]
-
-customCSS = """
-#component-7 { 
-  height: 1600px; 
-  flex-grow: 4;
-}
-"""
-
-with gr.Blocks(theme='ParityError/Interstellar') as demo:
-    gr.ChatInterface(
-        generate,
-        additional_inputs=additional_inputs,
-    )
-
-from gradio import Blocks
-
-app = gr.Blocks(analytics_enabled=False, title="Test")
-
-app.launch(share=True,)
+def list_project_files(project_path: str = DEFAULT_PROJECT_PATH) -> str:
+    """Lists files in the project directory."""
+    try:
+        files = os.listdir(project_path)
+        if not files:
+            return "Project directory is empty."
+        return "\n".join(files)
+    except Exception as e:
+        return f"Error listing project files: {str(e)}"
+
+def read_file_content(file_path: str, project_path: str = DEFAULT_PROJECT_PATH) -> str:
+    """Reads and returns the content of a file in the project."""
+    try:
+        full_path = os.path.join(project_path, file_path)
+        with open(full_path, "r") as f:
+            content = f.read()
+        return content
+    except Exception as e:
+        return f"Error reading file: {str(e)}"
+
+def write_to_file(file_path: str, content: str, project_path: str = DEFAULT_PROJECT_PATH) -> str:
+    """Writes content to a file in the project."""
+    try:
+        full_path = os.path.join(project_path, file_path)
+        with open(full_path, "w") as f:
+            f.write(content)
+        return f"Successfully wrote to '{file_path}'"
+    except Exception as e:
+        return f"Error writing to file: {str(e)}"
+
+def preview_project(project_path: str = DEFAULT_PROJECT_PATH):
+    """Provides a preview of the project, if applicable."""
+    # Assuming a simple HTML preview for now
+    try:
+        index_html_path = os.path.join(project_path, "index.html")
+        if os.path.exists(index_html_path):
+            with open(index_html_path, "r") as f:
+                html_content = f.read()
+            display(HTML(html_content))
+            return "Previewing 'index.html'"
+        else:
+            return "No 'index.html' found for preview."
+    except Exception as e:
+        return f"Error previewing project: {str(e)}"
+
+def main():
+    with gr.Blocks() as demo:
+        gr.Markdown("## FragMixt: Your Hugging Face No-Code App Builder")
+
+        # --- Model Selection ---
+        with gr.Tab("Model"):
+            # --- Model Dropdown with Categories ---
+            model_categories = gr.Dropdown(
+                choices=["Text Generation", "Text Summarization", "Code Generation", "Translation", "Question Answering"],
+                label="Model Category",
+                value="Text Generation"
+            )
+            model_name = gr.Dropdown(
+                choices=[],  # Initially empty, will be populated based on category
+                label="Hugging Face Model Name",
+            )
+            load_button = gr.Button("Load Model")
+            load_output = gr.Textbox(label="Output")
+            model_description = gr.Markdown(label="Model Description")
+
+            # --- Function to populate model names based on category ---
+            def update_model_dropdown(category):
+                models = []
+                api = HfApi()
+                for model in api.list_models():
+                    if model.pipeline_tag == category:
+                        models.append(model.modelId)
+                return gr.Dropdown.update(choices=models)
+
+            # --- Event handler for category dropdown ---
+            model_categories.change(
+                fn=update_model_dropdown,
+                inputs=model_categories,
+                outputs=model_name,
+            )
+
+            # --- Event handler to display model description ---
+            def display_model_description(model_name):
+                global model_descriptions
+                if model_name in model_descriptions:
+                    return model_descriptions[model_name]
+                else:
+                    return "Model description not available."
+
+            model_name.change(
+                fn=display_model_description,
+                inputs=model_name,
+                outputs=model_description,
+            )
+
+            load_button.click(load_hf_model, inputs=model_name, outputs=load_output)
+
+        # --- Chat Interface ---
+        with gr.Tab("Chat"):
+            chatbot = gr.Chatbot(show_label=False, show_share_button=False, show_copy_button=True, likeable=True)
+            message = gr.Textbox(label="Enter your message", placeholder="Ask me anything!")
+            purpose = gr.Textbox(label="Purpose", placeholder="What is the purpose of this interaction?")
+            agent_name = gr.Dropdown(label="Agents", choices=["Generic Agent"], value="Generic Agent", interactive=True)
+            sys_prompt = gr.Textbox(label="System Prompt", max_lines=1, interactive=True)
+            temperature = gr.Slider(label="Temperature", value=TEMPERATURE, minimum=0.0, maximum=1.0, step=0.05, interactive=True, info="Higher values produce more diverse outputs")
+            max_new_tokens = gr.Slider(label="Max new tokens", value=MAX_TOKENS, minimum=0, maximum=1048 * 10, step=64, interactive=True, info="The maximum numbers of new tokens")
+            top_p = gr.Slider(label="Top-p (nucleus sampling)", value=TOP_P, minimum=0.0, maximum=1, step=0.05, interactive=True, info="Higher values sample more low-probability tokens")
+            repetition_penalty = gr.Slider(label="Repetition penalty", value=REPETITION_PENALTY, minimum=1.0, maximum=2.0, step=0.05, interactive=True, info="Penalize repeated tokens")
+            submit_button = gr.Button(value="Send")
+            history = gr.State([])
+
+            def run_chat(purpose: str, message: str, agent_name: str, sys_prompt: str, temperature: float, max_new_tokens: int, top_p: float, repetition_penalty: float, history: List[Tuple[str, str]]) -> Tuple[List[Tuple[str, str]], List[Tuple[str, str]]]:
+                response = generate_response(message, history, agent_name, sys_prompt, temperature, max_new_tokens, top_p, repetition_penalty)
+                history.append((message, response))
+                return history, history
+
+            submit_button.click(run_chat, inputs=[purpose, message, agent_name, sys_prompt, temperature, max_new_tokens, top_p, repetition_penalty, history], outputs=[chatbot, history])
+
+        # --- Project Management ---
+        with gr.Tab("Project"):
+            project_name = gr.Textbox(label="Project Name", placeholder="MyHuggingFaceApp")
+            create_project_button = gr.Button("Create Hugging Face Project")
+            project_output = gr.Textbox(label="Output", lines=5)
+            file_content = gr.Code(label="File Content", language="python", lines=20)
+            file_path = gr.Textbox(label="File Path (relative to project)", placeholder="src/main.py")
+            read_button = gr.Button("Read File")
+            write_button = gr.Button("Write to File")
+            command_input = gr.Textbox(label="Terminal Command", placeholder="pip install -r requirements.txt")
+            command_output = gr.Textbox(label="Command Output", lines=5)
+            run_command_button = gr.Button("Run Command")
+            preview_button = gr.Button("Preview Project")
+
+            create_project_button.click(create_hf_project, inputs=[project_name], outputs=project_output)
+            read_button.click(read_file_content, inputs=file_path, outputs=file_content)
+            write_button.click(write_to_file, inputs=[file_path, file_content], outputs=project_output)
+            run_command_button.click(execute_command, inputs=command_input, outputs=command_output)
+            preview_button.click(preview_project, outputs=project_output)
+
+    demo.launch()
+
+if __name__ == "__main__":
+    main()