Update app.py

app.py

@@ -1,11 +1,15 @@
 import os
 import subprocess
 import random
-from huggingface_hub import InferenceClient
+import json
+from datetime import datetime
+
+from huggingface_hub import InferenceClient, cached_download, hf_hub_url
 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 i_search import google, i_search as i_s
+
 from agent import (
     ACTION_PROMPT,
     ADD_PROMPT,
@@ -19,372 +23,226 @@ from agent import (
     TASK_PROMPT,
     UNDERSTAND_TEST_RESULTS_PROMPT,
 )
-from utils import parse_action, parse_file_content, read_python_module_structure
-from datetime import datetime
-now = datetime.now()
-date_time_str = now.strftime("%Y-%m-%d %H:%M:%S")
-
-client = InferenceClient(
-    "mistralai/Mixtral-8x7B-Instruct-v0.1"
-)
-
-############################################
-
-
-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]"
-  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(
-        date_time_str=date_time_str,
-        purpose=purpose,
-        safe_search=safe_search,
-    ) + prompt_template.format(**prompt_kwargs)
-    if VERBOSE:
-        print(LOG_PROMPT.format(content))
-    
-    
-    #formatted_prompt = format_prompt(f"{system_prompt}, {prompt}", history)
-    #formatted_prompt = format_prompt(f'{content}', history)
-
-    stream = client.text_generation(content, **generate_kwargs, stream=True, details=True, return_full_text=False)
-    resp = ""
-    for response in stream:
-        resp += response.token.text
-
-    if VERBOSE:
-        print(LOG_RESPONSE.format(resp))
-    return resp
+from utils import parse_action, parse_file_content, read_python_module_structure
 
 
-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")
-    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)
+class App:
+    def __init__(self):
+        self.app_state = {"components": []}
+        self.terminal_history = ""
+        self.components_registry = {
+            "Button": {
+                "properties": {
+                    "label": "Click Me",
+                    "onclick": ""
+                },
+                "description": "A clickable button",
+                "code_snippet": "gr.Button(value='{{label}}', variant='primary')"
+            },
+            "Text Input": {
+                "properties": {
+                    "value": "",
+                    "placeholder": "Enter text"
+                },
+                "description": "A field for entering text",
+                "code_snippet": "gr.Textbox(label='{{placeholder}}')"
+            },
+            "Image": {
+                "properties": {
+                    "src": "#",
+                    "alt": "Image"
+                },
+                "description": "Displays an image",
+                "code_snippet": "gr.Image(label='{{alt}}')"
+            },
+            "Dropdown": {
+                "properties": {
+                    "choices": ["Option 1", "Option 2"],
+                    "value": ""
+                },
+                "description": "A dropdown menu for selecting options",
+                "code_snippet": "gr.Dropdown(choices={{choices}}, label='Dropdown')"
+            }
+        }
+        self.nlp_model_names = [
+            "google/flan-t5-small",
+            "Qwen/CodeQwen1.5-7B-Chat-GGUF",
+            "bartowski/Codestral-22B-v0.1-GGUF",
+            "bartowski/AutoCoder-GGUF"
+        ]
+        self.nlp_models = []
+        self.initialize_nlp_models()
+
+    def initialize_nlp_models(self):
+        for nlp_model_name in self.nlp_model_names:
+            try:
+                cached_download(hf_hub_url(nlp_model_name, revision="main"))
+                self.nlp_models.append(InferenceClient(nlp_model_name))
+            except:
+                self.nlp_models.append(None)
+
+    def get_nlp_response(self, input_text, model_index):
+        if self.nlp_models[model_index]:
+            response = self.nlp_models[model_index].text_generation(input_text)
+            return response.generated_text
         else:
-            history += "observation: I need to provide a valid URL to 'action: SEARCH action_input=https://URL'\n"
-    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
+            return "NLP model not available."
+
+    class Component:
+        def __init__(self, type, properties=None, id=None):
+            self.id = id or random.randint(1000, 9999)
+            self.type = type
+            self.properties = properties or self.components_registry[type]["properties"].copy()
+
+        def to_dict(self):
+            return {
+                "id": self.id,
+                "type": self.type,
+                "properties": self.properties,
+            }
+
+        def render(self):
+            if self.type == "Dropdown":
+                self.properties["choices"] = str(self.properties["choices"]).replace("[", "").replace("]", "").replace("'", "")
+            return self.components_registry[self.type]["code_snippet"].format(**self.properties)
+
+    def update_app_canvas(self):
+        components_html = "".join([f"<div>Component ID: {component['id']}, Type: {component['type']}, Properties: {component['properties']}</div>" for component in self.app_state["components"]])
+        return components_html
+
+    def add_component(self, component_type):
+        if component_type in self.components_registry:
+            new_component = self.Component(component_type)
+            self.app_state["components"].append(new_component.to_dict())
+            return (
+                self.update_app_canvas(),
+                f"System: Added component: {component_type}\n",
+            )
+        else:
+            return None, f"Error: Invalid component type: {component_type}\n"
+
+    def run_terminal_command(self, command, history):
+        output = ""
+        try:
+            if command.startswith("add "):
+                component_type = command.split("add ")[1]
+                return self.add_component(component_type)
+            elif command.startswith("search "):
+                query = command.split("search ")[1]
+                return google(query)
+            elif command.startswith("i search "):
+                query = command.split("i search ")[1]
+                return i_s(query)
+            elif command.startswith("safe search "):
+                query = command.split("safesearch ")[1]
+                return safe_search(query)
+            elif command.startswith("read "):
+                file_path = command.split("read ")[1]
+                return parse_file_content(file_path)
+            elif command == "task":
+                return TASK_PROMPT
+            elif command == "modify":
+                return MODIFY_PROMPT
+            elif command == "log":
+                return LOG_PROMPT
+            elif command.startswith("understand test results "):
+                test_results = command.split("understand test results ")[1]
+                return self.understand_test_results(test_results)
+            elif command.startswith("compress history"):
+                return self.compress_history(history)
+            elif command == "help":
+                return self.get_help_message()
+            elif command == "exit":
+                exit()
             else:
-                return action_name, action_input, history, task
+                output = subprocess.check_output(command, shell=True).decode("utf-8")
+        except Exception as e:
+            output = str(e)
+        return output or "No output\n"
+
+    def compress_history(self, history):
+        compressed_history = ""
+        lines = history.strip().split("\n")
+        for line in lines:
+            if not line.strip().startswith("#"):
+                compressed_history += line + "\n"
+        return compressed_history
+
+    def understand_test_results(self, test_results):
+        # Logic to understand test results
+        return UNDERSTAND_TEST_RESULTS_PROMPT
+
+    def get_help_message(self):
+        return """
+        Available commands:
+        - add [component_type]: Add a component to the app canvas
+        - search [query]: Perform a Google search
+        - i search [query]: Perform an intelligent search
+        - safe search [query]: Perform a safe search
+        - read [file_path]: Read and parse the content of a Python module
+        - task: Prompt for a task to perform
+        - modify: Prompt to modify a component property
+        - log: Prompt to log a response
+        - understand test results [test_results]: Understand test results
+        - compress history: Compress the terminal history by removing comments
+        - help: Show this help message
+        - exit: Exit the program
+        """
+
+    def process_input(self, input_text):
+        if input_text.strip().startswith("/"):
+            command = input_text.strip().lstrip("/")
+            output = self.run_terminal_command(command, self.terminal_history)
+            self.terminal_history += f"{input_text}\n{output}\n"
+            return output
         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,
-
-}
+            model_index = random.randint(0, len(self.nlp_models)-1)
+            response = self.get_nlp_response(input_text, model_index)
+            component_id, action, property_name, property_value = parse_action(response)
+            if component_id:
+                component = next((comp for comp in self.app_state["components"] if comp["id"] == component_id), None)
+                if component:
+                    if action == "update":
+                        component["properties"][property_name] = property_value
+                        return (
+                            self.update_app_canvas(),
+                            f"System: Updated property '{property_name}' of component with ID {component_id}\n",
+                        )
+                    elif action == "remove":
+                        self.app_state["components"].remove(component)
+                        return (
+                            self.update_app_canvas(),
+                            f"System: Removed component with ID {component_id}\n",
+                        )
+                    else:
+                        return None, f"Error: Invalid action: {action}\n"
+                else:
+                    return None, f"Error: Component with ID {component_id} not found\n"
+            else:
+                return None, f"Error: Failed to parse action from NLP response\n"
 
-def run_action(purpose, task, history, directory, action_name, action_input):
-    print(f'action_name::{action_name}')
+def run(self):
+    print("Welcome to the Python App Builder!")
+    print("Type 'help' to see the available commands.")
+    print("-" * 50)
     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)
-    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 the updated chat history
-        #yield ("",[(purpose,history)])
-        if task == "END":
-            return "", history  # Return an empty string for the Textbox and the chat history for the Chatbot
-            #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
-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 produce more diverse outputs",
-    ),
-
-    gr.Slider(
-        label="Max new tokens",
-        value=1048*10,
-        minimum=0,
-        maximum=1048*10,
-        step=64,
-        interactive=True,
-        info="The maximum numbers of new tokens",
-    ),
-    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",
-    ),
-
-
-]
-
-examples=[["What are the biggest news stories today?", None, None, None, None, None, ],
-          ["When is the next full moon?", None, None, None, None, None, ],
-          ["I'm planning a vacation to Japan. Can you suggest a one-week itinerary including must-visit places and local cuisines to try?", None, None, None, None, None, ],
-          ["Can you write a short story about a time-traveling detective who solves historical mysteries?", None, None, None, None, None,],
-          ["I'm trying to learn French. Can you provide some common phrases that would be useful for a beginner, along with their pronunciations?", None, None, None, None, None,],
-          ["I have chicken, rice, and bell peppers in my kitchen. Can you suggest an easy recipe I can make with these ingredients?", None, None, None, None, None,],
-          ["Can you explain how the QuickSort algorithm works and provide a Python implementation?", None, None, None, None, None,],
-          ["What are some unique features of Rust that make it stand out compared to other systems programming languages like C++?", None, None, None, None, None,],
-         ]
-
-'''
-gr.ChatInterface(
-fn=run,
-chatbot=gr.Chatbot(show_label=False, show_share_button=False, show_copy_button=True, likeable=True, layout="panel"),
-title="Mixtral 46.7B\nMicro-Agent\nInternet Search <br> development test",
-examples=examples,
-concurrency_limit=20,
-with gr.Blocks() as ifacea:
-    gr.HTML("""TEST""")
-ifacea.launch()
-).launch()
-with gr.Blocks() as iface:
-    #chatbot=gr.Chatbot(show_label=False, show_share_button=False, show_copy_button=True, likeable=True, layout="panel"),
-    chatbot=gr.Chatbot()
-    msg = gr.Textbox()
-    with gr.Row():
-        submit_b = gr.Button()
-        clear = gr.ClearButton([msg, chatbot])
-    submit_b.click(run, [msg,chatbot],[msg,chatbot])
-    msg.submit(run, [msg, chatbot], [msg, chatbot])
-iface.launch()
-'''
-with gr.Blocks() as iface:
-    chatbot=gr.Chatbot(show_label=False, show_share_button=False, show_copy_button=True, layout="panel")
-    msg = gr.Textbox()
-    with gr.Row():
-        submit_b = gr.Button()
-        clear = gr.ClearButton([msg, chatbot])
-    submit_b.click(run, [msg,chatbot],[msg,chatbot])
-    msg.submit(run, [msg, chatbot], [msg, chatbot])
-iface.launch()
+        while True:
+            try:
+                input_text = input("Enter input: ")
+            except EOFError:
+                print("Error: Input reading interrupted. Please provide valid input.")
+                continue
+
+            output, system_message = self.process_input(input_text)
+            if output:
+                print(output)
+            if system_message:
+                print(system_message)
+    except KeyboardInterrupt:
+        print("\nApplication stopped by user.")
+
+
+if __name__ == "__main__":
+    app = App()
+    app.run()