Update app.py

app.py

@@ -7,26 +7,25 @@ from datetime import datetime
 import logging
 
 import gradio as gr
-from huggingface_hub import InferenceClient, cached_download
+from huggingface_hub import InferenceClient
 from safe_search import safe_search
 from i_search import google, i_search as i_s
 
 # --- Configuration ---
-VERBOSE = True  # Enable verbose logging
-MAX_HISTORY = 5  # Maximum history turns to keep
-MAX_TOKENS = 2048  # Maximum tokens for LLM responses
-TEMPERATURE = 0.7  # Temperature for LLM responses
-TOP_P = 0.8  # Top-p (nucleus sampling) for LLM responses
-REPETITION_PENALTY = 1.5  # Repetition penalty for LLM responses
-MODEL_NAME = "mistralai/Mixtral-8x7B-Instruct-v0.1"  # Name of the LLM model
+VERBOSE = True
+MAX_HISTORY = 5
+MAX_TOKENS = 2048
+TEMPERATURE = 0.7
+TOP_P = 0.8
+REPETITION_PENALTY = 1.5
+MODEL_NAME = "mistralai/Mixtral-8x7B-Instruct-v0.1"
 
-import os
-API_KEY = os.getenv("HUGGINGFACE_API_KEY")  # Ensure you set the HUGGINGFACE_API_KEY environment variable
+API_KEY = os.getenv("HUGGINGFACE_API_KEY")
 
 # --- Logging Setup ---
 logging.basicConfig(
-    filename="app.log",  # Name of the log file
-    level=logging.INFO,  # Set the logging level (INFO, DEBUG, etc.)
+    filename="app.log",
+    level=logging.INFO,
     format="%(asctime)s - %(levelname)s - %(message)s",
 )
 
@@ -86,13 +85,10 @@ thought:
 
 # --- Functions ---
 def format_prompt(message: str, history: List[Tuple[str, str]], max_history_turns: int = 2) -> str:
-    """Formats the prompt for the LLM, including the message and relevant history."""
     prompt = " "
-    # Keep only the last 'max_history_turns' turns
     for user_prompt, bot_response in history[-max_history_turns:]:
-        prompt += f"[INST] {user_prompt} [/  "
-        prompt += f" {bot_response}"
-    prompt += f"[INST] {message} [/  "
+        prompt += f"[INST] {user_prompt} [/INST] {bot_response} "
+    prompt += f"[INST] {message} [/INST]"
     return prompt
 
 def run_llm(
@@ -101,21 +97,21 @@ def run_llm(
     purpose: str,
     **prompt_kwargs: Dict
 ) -> str:
-    """Runs the LLM with the given prompt and parameters."""
     seed = random.randint(1, 1111111111111111)
-    logging.info(f"Seed: {seed}")  # Log the seed
+    logging.info(f"Seed: {seed}")
 
     content = PREFIX.format(
-        date_time_str=date_time_str,
+        date_time_str=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
         purpose=purpose,
         safe_search=safe_search,
     ) + prompt_template.format(**prompt_kwargs)
     if VERBOSE:
-        logging.info(LOG_PROMPT.format(content))  # Log the prompt
+        logging.info(LOG_PROMPT.format(content=content))
 
+    client = InferenceClient(model=MODEL_NAME, token=API_KEY)
     resp = client.text_generation(content, max_new_tokens=MAX_TOKENS, stop_sequences=stop_tokens, temperature=TEMPERATURE, top_p=TOP_P, repetition_penalty=REPETITION_PENALTY)
     if VERBOSE:
-        logging.info(LOG_RESPONSE.format(resp))  # Log the response
+        logging.info(LOG_RESPONSE.format(resp=resp))
     return resp
 
 def generate(
@@ -128,265 +124,59 @@ def generate(
     top_p: float = TOP_P,
     repetition_penalty: float = REPETITION_PENALTY,
 ) -> str:
-    """Generates text using the LLM."""
     content = PREFIX.format(
-        date_time_str=date_time_str,
-        purpose=purpose,
+        date_time_str=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+        purpose=f"Generating response as {agent_name}",
         safe_search=safe_search,
-    ) + prompt_template.format(**prompt_kwargs)
+    ) + sys_prompt + "\n" + prompt
     if VERBOSE:
-        logging.info(LOG_PROMPT.format(content))  # Log the prompt
+        logging.info(LOG_PROMPT.format(content=content))
 
+    client = InferenceClient(model=MODEL_NAME, token=API_KEY)
     stream = client.text_generation(content, stream=True, details=True, return_full_text=False, temperature=temperature, top_p=top_p, repetition_penalty=repetition_penalty, max_new_tokens=max_new_tokens)
-    resp = ""
-    for response in stream:
-        resp += response.token.text
-
-    if VERBOSE:
-        logging.info(LOG_RESPONSE.format(resp))  # Log the response
-    return resp
-
-def compress_history(purpose: str, task: str, history: List[Tuple[str, str]], directory: str) -> str:
-    """Compresses the history into a shorter summary."""
-    resp = run_llm(
-        COMPRESS_HISTORY_PROMPT,
-        stop_tokens=["observation:", "task:", "action:", "thought:"],
-        purpose=purpose,
-        task=task,
-        history="\n".join(f"[INST] {user_prompt} [/] {bot_response}" for user_prompt, bot_response in history),
-    )
-    history = "observation: {}\n".format(resp)
-    return history
-
-def call_search(purpose: str, task: str, history: List[Tuple[str, str]], directory: str, action_input: str) -> Tuple[str, str, List[Tuple[str, str]], str]:
-    """Performs a search based on the action input."""
-    logging.info(f"CALLING SEARCH: {action_input}")
-    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)
-            logging.info(f"Search Result: {response}")
-            history.append(("observation: search result is: {}".format(response), ""))
-        else:
-            history.append(("observation: I need to provide a valid URL to 'action: SEARCH action_input=https://URL'\n", ""))
-    except Exception as e:
-        history.append(("observation: {}\n".format(e), ""))
-    return "MAIN", None, history, task
-
-def call_main(purpose: str, task: str, history: List[Tuple[str, str]], directory: str, action_input: str) -> Tuple[str, str, List[Tuple[str, str]], str]:
-    """Handles the main agent interaction loop."""
-    logging.info(f"CALLING MAIN: {action_input}")
-    resp = run_llm(
-        ACTION_PROMPT,
-        stop_tokens=["observation:", "task:", "action:", "thought:"],
-        purpose=purpose,
-        task=task,
-        history="\n".join(f"[INST] {user_prompt} [/] {bot_response}" for user_prompt, bot_response in history),
-    )
-    lines = resp.strip().strip("\n").split("\n")
-    for line in lines:
-        if line == "":
-            continue
-        if line.startswith("thought: "):
-            history.append((line, ""))
-            logging.info(f"Thought: {line}")
-        elif line.startswith("action: "):
-            action_name, action_input = parse_action(line)
-            logging.info(f"Action: {action_name} - {action_input}")
-            history.append((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
-        else:
-            history.append((line, ""))
-            logging.info(f"Other Output: {line}")
-    return "MAIN", None, history, task
-
-def call_set_task(purpose: str, task: str, history: List[Tuple[str, str]], directory: str, action_input: str) -> Tuple[str, str, List[Tuple[str, str]], str]:
-    """Sets a new task for the agent."""
-    logging.info(f"CALLING SET_TASK: {action_input}")
-    task = run_llm(
-        TASK_PROMPT,
-        stop_tokens=[],
-        purpose=purpose,
-        task=task,
-        history="\n".join(f"[INST] {user_prompt} [/] {bot_response}" for user_prompt, bot_response in history),
-    ).strip("\n")
-    history.append(("observation: task has been updated to: {}".format(task), ""))
-    return "MAIN", None, history, task
-
-def end_fn(purpose: str, task: str, history: List[Tuple[str, str]], directory: str, action_input: str) -> Tuple[str, str, List[Tuple[str, str]], str]:
-    """Ends the agent interaction."""
-    logging.info(f"CALLING END_FN: {action_input}")
-    task = "END"
-    return "COMPLETE", "COMPLETE", history, task
-
-NAME_TO_FUNC: Dict[str, callable] = {
-    "MAIN": call_main,
-    "UPDATE-TASK": call_set_task,
-    "SEARCH": call_search,
-    "COMPLETE": end_fn,
-}
-
-def run_action(purpose: str, task: str, history: List[Tuple[str, str]], directory: str, action_name: str, action_input: str) -> Tuple[str, str, List[Tuple[str, str]], str]:
-    """Executes the specified action."""
-    logging.info(f"RUNNING ACTION: {action_name} - {action_input}")
-    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) > MAX_HISTORY:
-            logging.info("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 is None:
-            action_name = "MAIN"
-        assert action_name in NAME_TO_FUNC
-    
-        logging.info(f"RUN: {action_name} - {action_input}")
-        return NAME_TO_FUNC[action_name](purpose, task, history, directory, action_input)
-    except Exception as e:
-        history.append(("observation: the previous command did not produce any useful output, I need to check the commands syntax, or use a different command\n", ""))
-        logging.error(f"Error in run_action: {e}")
-        return "MAIN", None, history, task
-
-def run(purpose: str, history: List[Tuple[str, str]]) -> List[Tuple[str, str]]:
-    """Main agent interaction loop."""
-    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:
-        logging.info(f"---")
-        logging.info(f"Purpose: {purpose}")
-        logging.info(f"Task: {task}")
-        logging.info(f"---")
-        logging.info(f"History: {history}")
-        logging.info(f"---")
-
-        action_name, action_input, history, task = run_action(
-            purpose,
-            task,
-            history,
-            directory,
-            action_name,
-            action_input,
-        )
-        yield (history)
-        if task == "END":
-            return (history)
-
-################################################
-
-def format_prompt(message: str, history: List[Tuple[str, str]], max_history_turns: int = 5) -> str:
-    """Formats the prompt for the LLM, including the message and relevant history."""
-    prompt = " "
-    # Keep only the last 'max_history_turns' turns
-    for user_prompt, bot_response in history[-max_history_turns:]:
-        prompt += f"[INST] {user_prompt} [/  "
-        prompt += f" {bot_response}"
-    prompt += f"[INST] {message} [/  "
-    return prompt
-
-def parse_action(line: str) -> Tuple[str, str]:
-    """Parses the action line to get the action name and input."""
-    parts = line.split(":", 1)
-    if len(parts) == 2:
-        action_name = parts[0].replace("action", "").strip()
-        action_input = parts[1].strip()
-    else:
-        action_name = parts[0].replace("action", "").strip()
-        action_input = ""
-    return action_name, action_input
+    return "".join(chunk.text for chunk in stream)
 
 def main():
-    """Main function to run the Gradio interface."""
-    global client, date_time_str
-    # Initialize the LLM client with your API key
-    try:
-        client = InferenceClient(
-            MODEL_NAME,
-            token=API_KEY  # Replace with your actual API key
-        )
-    except Exception as e:
-        logging.error(f"Error initializing LLM client: {e}")
-        print("Error initializing LLM client. Please check your API key.")
-        return
-
-    # Get the current date and time
-    date_time_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-
     with gr.Blocks() as demo:
         gr.Markdown("## FragMixt: The No-Code Development Powerhouse")
-        gr.Markdown("###  Your AI-Powered Development Companion")
-
-        # Chat Interface
-        chatbot = gr.Chatbot(show_label=False, show_share_button=False, show_copy_button=True, likeable=True, layout="panel")
-        
-        # Input Components
-        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=[s for s in agents], value=agents[0], 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")
-
-        # Button to submit the message
-        submit_button = gr.Button(value="Send")
-
-        # Project Explorer Tab
-        with gr.Tab("Project Explorer"):
-            project_path = gr.Textbox(label="Project Path", placeholder="/home/user/app/current_project")
-            explore_button = gr.Button(value="Explore")
-            project_output = gr.Textbox(label="File Tree", lines=20)
-
-        # Chat App Logic Tab
-        with gr.Tab("Chat App"):
-            history = gr.State([])
-            examples = [
-                ["What is the purpose of this AI agent?", "I am designed to assist with no-code development tasks."],
-                ["Can you help me generate a Python function to calculate the factorial of a number?", "Sure! Here is a Python function to calculate the factorial of a number:"],
-                ["Generate a simple HTML page with a heading and a paragraph.", "
-
-
-html\n
-
-<!DOCTYPE html>\n<html>\n<head>\n<title>My Simple Page</title>\n</head>\n<body>\n<h1>Welcome to my page!</h1>\n<p>This is a simple paragraph.</p>\n</body>\n</html>
-
-JS
-
-
+        gr.Markdown("### Your AI-Powered Development Companion")
 
-
-\n
-
-["Create a basic SQL query to select all data from a table named 'users'.", "
-
-
-sql\nSELECT * FROM users;\n
-
-["Design a user interface for a mobile app that allows users to track their daily expenses.", "Here's a basic UI design for a mobile expense tracker app:\n\n**Screen 1: Home**\n- Top: App Name and Balance Display\n- Middle: List of Recent Transactions (Date, Description, Amount)\n- Bottom: Buttons for Add Expense, Add Income, View Categories\n\n**Screen 2: Add Expense**\n- Input fields for Date, Category, Description, Amount\n- Buttons for Save, Cancel\n\n**Screen 3: Expense Categories**\n- List of expense categories (e.g., Food, Transportation, Entertainment)\n- Option to add/edit categories\n\n**Screen 4: Reports**\n- Charts and graphs to visualize spending by category, date range, etc.\n- Filters to customize the reports"],
-            ]
+        with gr.Row():
+            with gr.Column(scale=3):
+                chatbot = gr.Chatbot(show_label=False, show_share_button=False, show_copy_button=True, likeable=True, layout="panel")
+                message = gr.Textbox(label="Enter your message", placeholder="Ask me anything!")
+                submit_button = gr.Button(value="Send")
+            
+            with gr.Column(scale=1):
+                purpose = gr.Textbox(label="Purpose", placeholder="What is the purpose of this interaction?")
+                agent_name = gr.Dropdown(label="Agents", choices=[s for s in agents], value=agents[0], 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")
+
+        with gr.Tabs():
+            with gr.TabItem("Project Explorer"):
+                project_path = gr.Textbox(label="Project Path", placeholder="/home/user/app/current_project")
+                explore_button = gr.Button(value="Explore")
+                project_output = gr.Textbox(label="File Tree", lines=20)
+
+            with gr.TabItem("Code Editor"):
+                code_editor = gr.Code(label="Code Editor", language="python")
+                run_code_button = gr.Button(value="Run Code")
+                code_output = gr.Textbox(label="Code Output", lines=10)
+
+            with gr.TabItem("File Management"):
+                file_list = gr.Dropdown(label="Select File", choices=[], interactive=True)
+                file_content = gr.Textbox(label="File Content", lines=20)
+                save_file_button = gr.Button(value="Save File")
+                create_file_button = gr.Button(value="Create New File")
+                delete_file_button = gr.Button(value="Delete File")
+
+        history = gr.State([])
 
         def 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]]]:
-            """Handles the chat interaction."""
             prompt = format_prompt(message, history)
             response = generate(prompt, history, agent_name, sys_prompt, temperature, max_new_tokens, top_p, repetition_penalty)
             history.append((message, response))
@@ -394,9 +184,7 @@ sql\nSELECT * FROM users;\n
 
         submit_button.click(chat, inputs=[purpose, message, agent_name, sys_prompt, temperature, max_new_tokens, top_p, repetition_penalty, history], outputs=[chatbot, history])
 
-        # Project Explorer Logic
         def explore_project(project_path: str) -> str:
-            """Explores the project directory and returns a file tree."""
             try:
                 tree = subprocess.check_output(["tree", project_path]).decode("utf-8")
                 return tree
@@ -405,6 +193,58 @@ sql\nSELECT * FROM users;\n
 
         explore_button.click(explore_project, inputs=[project_path], outputs=[project_output])
 
+        def run_code(code: str) -> str:
+            try:
+                exec_globals = {}
+                exec(code, exec_globals)
+                output = exec_globals.get('__builtins__', {}).get('print', print)
+                return str(output)
+            except Exception as e:
+                return f"Error running code: {e}"
+
+        run_code_button.click(run_code, inputs=[code_editor], outputs=[code_output])
+
+        def load_file_list(project_path: str) -> List[str]:
+            try:
+                return [f for f in os.listdir(project_path) if os.path.isfile(os.path.join(project_path, f))]
+            except Exception as e:
+                return [f"Error loading file list: {e}"]
+
+        def load_file_content(project_path: str, file_name: str) -> str:
+            try:
+                with open(os.path.join(project_path, file_name), 'r') as file:
+                    return file.read()
+            except Exception as e:
+                return f"Error loading file content: {e}"
+
+        def save_file(project_path: str, file_name: str, content: str) -> str:
+            try:
+                with open(os.path.join(project_path, file_name), 'w') as file:
+                    file.write(content)
+                return f"File {file_name} saved successfully."
+            except Exception as e:
+                return f"Error saving file: {e}"
+
+        def create_file(project_path: str, file_name: str) -> str:
+            try:
+                open(os.path.join(project_path, file_name), 'a').close()
+                return f"File {file_name} created successfully."
+            except Exception as e:
+                return f"Error creating file: {e}"
+
+        def delete_file(project_path: str, file_name: str) -> str:
+            try:
+                os.remove(os.path.join(project_path, file_name))
+                return f"File {file_name} deleted successfully."
+            except Exception as e:
+                return f"Error deleting file: {e}"
+
+        project_path.change(load_file_list, inputs=[project_path], outputs=[file_list])
+        file_list.change(load_file_content, inputs=[project_path, file_list], outputs=[file_content])
+        save_file_button.click(save_file, inputs=[project_path, file_list, file_content], outputs=[gr.Textbox()])
+        create_file_button.click(create_file, inputs=[project_path, gr.Textbox(label="New File Name")], outputs=[gr.Textbox()])
+        delete_file_button.click(delete_file, inputs=[project_path, file_list], outputs=[gr.Textbox()])
+
     demo.launch()
 
 if __name__ == "__main__":