Create app.py

app.py

@@ -0,0 +1,388 @@
+from typing import List, Optional, Tuple, Dict
+History = List[Tuple[str, str]]
+Messages = List[Dict[str, str]]
+
+import enum
+from dataclasses import dataclass
+from typing import List, Dict, Any, Optional, Tuple
+from collections import defaultdict
+
+
+@dataclass(frozen=True)
+class Action:
+    value: str  # LM returned string for now
+    use_tool: bool  # if use_tool == False -> propose answer
+    error: Optional[str] = None
+
+def lm_output_to_action(lm_output: str) -> Action:
+    propose_solution = bool("<solution>" in lm_output)
+    return Action(lm_output, not propose_solution)
+
+from typing import Mapping
+import re
+import signal
+from contextlib import contextmanager
+from IPython.core.interactiveshell import InteractiveShell
+from IPython.utils import io
+from typing import Any
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class Tool(ABC):
+    """Abstract class for a tool."""
+
+    name: str
+    signature: str
+    description: str
+
+    @abstractmethod
+    def __call__(self, *args: Any, **kwds: Any) -> str:
+        """Execute the tool with the given args and return the output."""
+        # execute tool with abitrary args
+        pass
+
+    def reset(self) -> None:
+        """Reset the tool to its initial state."""
+        pass
+
+
+class PythonREPL(Tool):
+    """A tool for running python code in a REPL."""
+
+    name = "PythonREPL"
+    # This PythonREPL is not used by the environment; It is THE ENVIRONMENT.
+    signature = "NOT_USED"
+    description = "NOT_USED"
+
+    def __init__(
+        self,
+        user_ns: Mapping[str, Any],
+        timeout: int = 30,
+    ) -> None:
+        super().__init__()
+        self.user_ns = user_ns
+        self.timeout = timeout
+        self.reset()
+
+    @contextmanager
+    def time_limit(self, seconds):
+        def signal_handler(signum, frame):
+            raise TimeoutError(f"Timed out after {seconds} seconds.")
+
+        signal.signal(signal.SIGALRM, signal_handler)
+        signal.alarm(seconds)
+        try:
+            yield
+        finally:
+            signal.alarm(0)  # Disable the alarm
+
+    def reset(self) -> None:
+        InteractiveShell.clear_instance()
+        self.shell = InteractiveShell.instance(
+            # NOTE: shallow copy is needed to avoid
+            # shell modifying the original user_ns dict
+            user_ns=dict(self.user_ns),
+            colors="NoColor",
+        )
+
+    def __call__(self, query: str) -> str:
+        """Use the tool and return observation"""
+        with io.capture_output() as captured:
+            _ = self.shell.run_cell(query, store_history=True)
+        output = captured.stdout
+
+        if output == "":
+            output = "[Executed Successfully with No Output]"
+
+        # replace potentially sensitive filepath
+        # e.g., File /mint/mint/tools/python_tool.py:30, in PythonREPL.time_limit.<locals>.signal_handler(signum, frame)
+        # with File <filepath>:30, in PythonREPL.time_limit.<locals>.signal_handler(signum, frame)
+        # use re
+        output = re.sub(
+                # r"File (/mint/)mint/tools/python_tool.py:(\d+)",
+                r"File (.*)mint/tools/python_tool.py:(\d+)",
+                r"File <hidden_filepath>:\1",
+                output,
+            )
+        if len(output) > 2000:
+            output = output[:2000] + "...\n[Output Truncated]"
+
+        return output
+
+class ParseError(Exception):
+    pass
+
+def parse_action(action: Action) -> Tuple[str, Dict[str, Any]]:
+    """Define the parsing logic."""
+    lm_output = "\n" + action.value + "\n"
+    output = {}
+    try:
+        if not action.use_tool:
+            answer = "\n".join(
+                [
+                    i.strip()
+                    for i in re.findall(
+                        r"<solution>(.*?)</solution>", lm_output, re.DOTALL
+                    )
+                ]
+            )
+            if answer == "":
+                raise ParseError("No answer found.")
+            output["answer"] = answer
+        else:
+            env_input = "\n".join(
+                [
+                    i.strip()
+                    for i in re.findall(
+                        r"<execute>(.*?)</execute>", lm_output, re.DOTALL
+                    )
+                ]
+            )
+            if env_input == "":
+                raise ParseError("No code found.")
+            output["env_input"] = env_input
+    except Exception as e:
+        raise ParseError(e)
+    return output
+
+python_repl = PythonREPL(
+            user_ns={},
+        )
+
+import gradio as gr
+import llama_cpp
+import llama_cpp.llama_tokenizer
+import torch
+
+if torch.cuda.is_available():
+  CodeActAgent_llm = llama_cpp.Llama.from_pretrained(
+      repo_id="xingyaoww/CodeActAgent-Mistral-7b-v0.1.q8_0.gguf",
+      filename="*q8_0.gguf",
+      verbose=False,
+      n_gpu_layers = -1,
+      n_ctx = 3060
+  )
+else:
+  CodeActAgent_llm = llama_cpp.Llama.from_pretrained(
+      repo_id="xingyaoww/CodeActAgent-Mistral-7b-v0.1.q8_0.gguf",
+      filename="*q8_0.gguf",
+      verbose=False,
+      #n_gpu_layers = -1,
+      n_ctx = 3060
+  )
+
+system_prompt = '''
+You are a helpful assistant assigned with the task of problem-solving. To achieve this, you will be using an interactive coding environment equipped with a variety of tool functions to assist you throughout the process.
+
+At each turn, you should first provide your step-by-step thinking for solving the task. Your thought process should be enclosed using "<thought>" tag, for example: <thought> I need to print "Hello World!" </thought>.
+
+After that, you have two options:
+
+1) Interact with a Python programming environment and receive the corresponding output. Your code should be enclosed using "<execute>" tag, for example: <execute> print("Hello World!") </execute>.
+2) Directly provide a solution that adheres to the required format for the given task. Your solution should be enclosed using "<solution>" tag, for example: The answer is <solution> A </solution>.
+
+You have {max_total_steps} chances to interact with the environment or propose a solution. You can only propose a solution {max_propose_solution} times.
+'''.format(
+    **{
+        "max_total_steps": 5,
+        "max_propose_solution": 2,
+    }
+)
+
+
+def exe_to_md(str_):
+    req = str_.replace("<execute>" ,"```python").replace("</execute>" ,"```").replace("<solution>" ,"```python").replace("</solution>" ,"```")
+    if "<thought>" in req and "def " in req:
+        req = req.replace("<thought>" ,"```python").replace("</thought>" ,"```")
+    return req
+
+def md_to_exe(str_):
+    return str_.replace("```python", "<execute>").replace("```", "</execute>")
+
+def clear_session() -> History:
+    return '', []
+
+def modify_system_session(system: str) -> str:
+    if system is None or len(system) == 0:
+        system = default_system
+    return system, system, []
+
+def history_to_messages(history: History, system: str) -> Messages:
+    messages = [{'role': "system", 'content': system}]
+    for h in history:
+        messages.append({'role': "user", 'content': h[0]})
+        if h[1] != "😊":
+            messages.append({'role': "assistant", 'content':
+                md_to_exe(h[1])
+            })
+    return messages
+
+def messages_to_history(messages: Messages) -> Tuple[str, History]:
+    assert messages[0]['role'] == "system"
+    system = messages[0]['content']
+    history = []
+    import numpy as np
+    import pandas as pd
+    from copy import deepcopy
+    messages = deepcopy(messages)
+    if messages[-1]["role"] == "user":
+        messages += [{"role": "assistant", "content": "😊"}]
+
+    messages_ = []
+    for ele in messages[1:]:
+        if not messages_:
+            messages_.append(ele)
+        else:
+            if messages_[-1]["role"] == ele["role"]:
+                continue
+            else:
+                messages_.append(ele)
+
+    history = pd.DataFrame(np.asarray(messages_).reshape([-1, 2]).tolist()).applymap(
+        lambda x: x["content"]
+    ).applymap(
+        exe_to_md
+    ).values.tolist()
+    return system, history
+
+def model_chat(query: Optional[str], history: Optional[History], system: str
+) -> Tuple[str, str, History]:
+    if query is None:
+        query = ''
+    if history is None:
+        history = []
+    messages = history_to_messages(history, system)
+    if query:
+        messages.append({'role': "user", 'content': query})
+
+    response = CodeActAgent_llm.create_chat_completion(
+        messages=messages,
+        stream=True,
+        top_p = 0.9,
+        temperature = 0.01
+    )
+
+    from IPython.display import clear_output
+    lm_output = ""
+    for chunk in response:
+        delta = chunk["choices"][0]["delta"]
+        if "content" not in delta:
+            continue
+        lm_output += delta["content"]
+    
+    lm_output = lm_output.replace("<solution>", "<execute>").replace("</solution>", "</execute>")
+
+    if "<execute>" in lm_output:
+        action_out = lm_output_to_action(lm_output)
+        parsed = parse_action(action_out)
+        env_input = parsed["env_input"]
+        obs = python_repl(env_input).strip()
+        obs = '''
+        Observation:
+        {}
+        '''.format(obs).strip()
+
+        system, history = messages_to_history(messages + [
+            {'role': "assistant",
+            'content': exe_to_md(lm_output)},
+            {
+             'role': "user",
+             "content": obs
+            }
+        ])
+    elif "<thought>" in lm_output:
+        system, history = messages_to_history(messages + [
+            {'role': "assistant",
+            'content': exe_to_md(lm_output)},
+        ])
+    else:
+        system, history = messages_to_history(messages + [
+            {'role': "assistant",
+            'content': exe_to_md(lm_output)},
+        ])
+    return "", history, system
+
+
+with gr.Blocks() as demo:
+    gr.Markdown("""<center><font size=8>CodeActAgent Mistral 7B Bot 🤖</center>""")
+
+    with gr.Row():
+        with gr.Column(scale=3):
+            system_input = gr.Textbox(value=system_prompt, lines=1, label='System', visible = False)
+        with gr.Column(scale=1):
+            modify_system = gr.Button("🛠️ Set system prompt and clear history", scale=2, visible = False)
+        system_state = gr.Textbox(value=system_prompt, visible=False)
+    chatbot = gr.Chatbot(label='CodeActAgent-Mistral-7b-v0.1')
+    textbox = gr.Textbox(lines=2, label='Input')
+
+    with gr.Row():
+        clear_history = gr.Button("🧹 Clear History")
+        sumbit = gr.Button("🚀 Send")
+
+    sumbit.click(model_chat,
+                 inputs=[textbox, chatbot, system_state],
+                 outputs=[textbox, chatbot, system_input],
+                 concurrency_limit = 100)
+    clear_history.click(fn=clear_session,
+                        inputs=[],
+                        outputs=[textbox, chatbot])
+    modify_system.click(fn=modify_system_session,
+                        inputs=[system_input],
+                        outputs=[system_state, system_input, chatbot])
+
+    gr.Examples(
+        [
+            "teach me how to use numpy.",
+            "Give me a python function give the divide of number it self 10 times.",
+            '''
+            Plot box plot with pandas and save it to local.
+            '''.strip(),
+
+            '''
+            Write a python code about, download image to local from url, the format as :
+            url = f'https://image.pollinations.ai/prompt/{prompt}'
+            where prompt as the input of download function.
+            '''.strip(),
+            "Use this function download a image of bee.",
+
+            '''
+            Draw a picture teach me what linear regression is.
+            '''.strip(),
+            "Use more points and draw the image with the line fitted.",
+
+            '''
+            Write a piece of Python code to simulate the financial transaction process and draw a financial images chart by lineplot of Poisson process.
+            '''.strip(),
+            #"Add monotonic increasing trend on it.",
+            "Add a Trigonometric function loop on it.",
+        ],
+        inputs = textbox,
+        label = "Task Prompt: \n(Used to give the function or task defination on the head)",
+    )
+
+    gr.Examples(
+        [
+            '''
+            Give me the function defination. 💡
+            '''.strip(),
+
+            '''
+            Correct it. ☹️❌
+            '''.strip(),
+
+            '''
+            Save the output as image 🖼️ to local. ⏬
+            '''.strip(),
+
+            '''
+            Good Job 😊
+            '''.strip(),
+        ],
+        inputs = textbox,
+        label = "Action Prompt: \n(Used to specify downstream actions taken by LLM, such as modifying errors, saving running results locally, saying you did a good job, etc.)",
+    )
+
+demo.queue(api_open=False)
+demo.launch(max_threads=30, share = False)