Update app.py

app.py

@@ -1,16 +1,11 @@
 import os
 import subprocess
 import random
-import json
-from datetime import datetime
-import uuid  # Import the uuid library
-
-from huggingface_hub import InferenceClient, cached_download, hf_hub_url
+from huggingface_hub import InferenceClient
 import gradio as gr
-
 from safe_search import safe_search
-from i_search import google, i_search as i_s
-
+from i_search import google
+from i_search import i_search as i_s
 from agent import (
     ACTION_PROMPT,
     ADD_PROMPT,
@@ -24,274 +19,372 @@ 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", cache_dir=None
+)
+
+
+############################################
+
+
+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
 
-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
+
+
+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
+
+
+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)
         else:
-            return "NLP model not available."
-
-    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, properties=None):
-        try:
-            new_component = {
-                "type": component_type,
-                "properties": properties or {},
-                "id": str(uuid.uuid4())  # Assign a unique ID using UUID
-            }
-            self.app_state["components"].append(new_component)
-        except Exception as e:
-            print(f"Error adding component: {e}")
-
-    def run_terminal_command(self, command, *args):
-        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(self.terminal_history)
-            elif command == "help":
-                return self.get_help_message()
-            elif command == "exit":
-                exit()
+            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
             else:
-                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):
-        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 += f"{input_text}\n{output}\n"
-            return output, ""
+                return action_name, action_input, history, task
         else:
-            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 "", f"Error: Invalid action: {action}\n"
-                else:
-                    return "", f"Error: Component with ID {component_id} not found\n"
-            else:
-                return "", f"Error: Failed to parse action from NLP response\n"
-
-    def build_app(self):
-        with gr.Blocks() as demo:
-            with gr.Tab("App Canvas"):
-                app_canvas = gr.HTML(self.update_app_canvas())
-            with gr.Tab("Terminal"):
-                gr.Markdown("## Terminal")
-                terminal_input = gr.Textbox(label="Input")
-                terminal_output = gr.Textbox(label="Output")
-                run_terminal_command_button = gr.Button("Run Command")
-                run_terminal_command_button.click(
-                    self.run_terminal_command,
-                    inputs=[terminal_input],
-                    outputs=[terminal_output]
-                )
-            with gr.Tab("NLP Models"):
-                gr.Markdown("## Available NLP Models")
-                model_index = gr.Slider(label="Model Index", minimum=0, maximum=len(self.nlp_model_names)-1, step=1)
-                input_text = gr.Textbox(label="Input Text")
-                get_nlp_response_button = gr.Button("Get NLP Response")
-                get_nlp_response_button.click(
-                    self.get_nlp_response,
-                    inputs=[input_text, model_index],
-                    outputs="text"
-                )
-            # Add components to the app canvas dynamically
-            for component in self.app_state["components"]:
-                component_type = component["type"]
-                properties = component["properties"]
-                
-                if component_type == "Button":
-                    gr.Button(value=properties["label"], variant="primary")
-                elif component_type == "Text Input":
-                    gr.Textbox(label=properties["placeholder"])
-                elif component_type == "Image":
-                    gr.Image(label=properties["alt"])
-                elif component_type == "Dropdown":
-                    gr.Dropdown(choices=properties["choices"], label="Dropdown")
-
-        return demo
-
-    def run(self):
-        self._print_welcome_message()
-        
-        while True:
-            try:
-                input_text = self._get_user_input()
-                if input_text.lower() == 'exit':
-                    break
-                elif input_text.lower() == 'launch':
-                    self.launch_app()
-                    continue
-                
-                output, system_message = self.process_input(input_text)
-                self._display_output(output, system_message)
+            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)
             
-            except EOFError:
-                print("Error: Input reading interrupted. Please provide valid input.")
-            except KeyboardInterrupt:
-                print("\nApplication stopped by user.")
-                break
-            except Exception as e:
-                print(f"An error occurred: {str(e)}")
-
-    def _print_welcome_message(self):
-        print("Welcome to the Python App Builder!")
-        print("Type 'help' to see the available commands.")
-        print("Type 'launch' to build and launch the Gradio app.")
-        print("Type 'exit' to quit the application.")
-        print("-" * 50)
-
-    def _get_user_input(self):
-        return input("Enter input: ").strip()
-
-    def _display_output(self, output, system_message):
-        if output:
-            print(output)
-        if system_message:
-            print(system_message)
-
-    def launch_app(self):
-        print("Launching app...")
-        demo = self.build_app()
-        print(f"Demo object type: {type(demo)}")
-        try:
-            demo.launch()
-            print("App launched successfully.")
-        except Exception as e:
-            print(f"Error launching app: {str(e)}")
-
-def main():
+            #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}')
     try:
-        app = App()
-        app.run()
+        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:
-        print(f"Error running app: {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
+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,],
+         ]
 
-if __name__ == "__main__":
-    main()
+'''
+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()
+'''
+gr.ChatInterface(
+    fn=run,
+    chatbot=gr.Chatbot(show_label=False, show_share_button=False, show_copy_button=True, super-intelligence=True, layout="panel"),
+    title="Mixtral 46.7B\nMicro-Agent\nInternet Search <br> development test",
+    examples=examples,
+    concurrency_limit=50,
+).launch(show_api=True)