Upload online demo code

.gitignore

@@ -0,0 +1,50 @@
+# Ignore .ckpt files
+ckpt
+
+# Ignore Python compiled files
+__pycache__/
+*.py[cod]
+
+# Ignore Python virtual environment
+venv/
+
+# Ignore Jupyter notebook checkpoints
+.ipynb_checkpoints/
+.git/
+.vscode/
+
+# Ignore .DS_Store on MacOS
+.DS_Store
+
+rilab_key.txt
+gpt4_custom_code_interpreter/rilab_key.txt
+openai_api_key.txt
+
+gpt4_custom_code_interpreter/
+tmp/
+output/
+wandb/
+
+utils/const.py
+utils/hf_model_upload.py
+gpt_data_gen/
+*.json
+*.txt
+*.sh
+*.pt 
+*.pth 
+*.ckpt
+*.tokenizer
+
+# eval data
+eval/ds1000_data
+eval/grade-school-math 
+
+# gradio features
+chatbot_feat.py
+chatbot_feat2.py
+gradio_test.py
+
+cache/
+env/
+json_dataset/

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Magnetic2014
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

chatbot.py

@@ -0,0 +1,376 @@
+import ast
+import gradio as gr
+import os
+import re
+import json
+import logging
+
+import torch
+from datetime import datetime
+
+from threading import Thread
+from typing import Optional
+from transformers import TextIteratorStreamer
+from functools import partial
+from huggingface_hub import CommitScheduler
+from uuid import uuid4
+from pathlib import Path
+
+from code_interpreter.JupyterClient import JupyterNotebook
+
+MAX_INPUT_TOKEN_LENGTH = int(os.getenv("MAX_INPUT_TOKEN_LENGTH", "4096"))
+
+import warnings
+
+warnings.filterwarnings("ignore", category=UserWarning, module="transformers")
+os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"
+
+
+from code_interpreter.OpenCodeInterpreter import OpenCodeInterpreter
+
+JSON_DATASET_DIR = Path("json_dataset")
+JSON_DATASET_DIR.mkdir(parents=True, exist_ok=True)
+
+upvote_button_value = "👍  Upvote Conversation"
+downvote_button_value = "👎  Downvote Conversation"
+
+scheduler = CommitScheduler(
+    repo_id="opencodeinterpreter_user_data",
+    repo_type="dataset",
+    folder_path=JSON_DATASET_DIR,
+    path_in_repo="data",
+    private=True
+)
+
+logging.basicConfig(level=logging.INFO)
+
+class StreamingOpenCodeInterpreter(OpenCodeInterpreter):
+    streamer: Optional[TextIteratorStreamer] = None
+
+    # overwirte generate function
+    @torch.inference_mode()
+    def generate(
+        self,
+        prompt: str = "",
+        max_new_tokens = 1024,
+        do_sample: bool = False,
+        top_p: float = 0.95,
+        top_k: int = 50,
+    ) -> str:
+        # Get the model and tokenizer, and tokenize the user text.
+
+        self.streamer = TextIteratorStreamer(
+            self.tokenizer, skip_prompt=True, Timeout=5
+        )
+
+        inputs = self.tokenizer([prompt], return_tensors="pt", truncation=True, max_length=MAX_INPUT_TOKEN_LENGTH)
+        inputs = inputs.to(self.model.device)
+
+        kwargs = dict(
+            **inputs,
+            streamer=self.streamer,
+            max_new_tokens=max_new_tokens,
+            do_sample=do_sample,
+            top_p=top_p,
+            top_k=top_k,
+            eos_token_id=self.tokenizer.eos_token_id
+        )
+
+        thread = Thread(target=self.model.generate, kwargs=kwargs)
+        thread.start()
+
+        return ""
+
+def save_json(dialog, mode, json_file_path, flag, dialog_id) -> None:
+    with scheduler.lock:
+        with json_file_path.open("a") as f:
+            json.dump({"id": dialog_id, "dialog": dialog, "mode": mode, "flag": flag, "datetime": datetime.now().isoformat()}, f, ensure_ascii=False)
+            f.write("\n")
+
+def convert_history(gradio_history: list[list], interpreter_history: list[dict]):
+    interpreter_history = [interpreter_history[0]] if interpreter_history and interpreter_history[0]["role"] == "system" else []
+    if not gradio_history:
+        return interpreter_history
+    for item in gradio_history:
+        if item[0] is not None:
+            interpreter_history.append({"role": "user", "content": item[0]})
+        if item[1] is not None:
+            interpreter_history.append({"role": "assistant", "content": item[1]})
+    return interpreter_history
+
+def reset_dialog_info(dialog_info):
+    new_uuid = str(uuid4())
+    logging.info(f"allocating new uuid {new_uuid} for conversation...")
+    return [new_uuid, None]
+
+def is_valid_python_code(code):
+    try:
+        ast.parse(code)
+        return True
+    except SyntaxError:
+        return False
+
+
+class InputFunctionVisitor(ast.NodeVisitor):
+    def __init__(self):
+        self.found_input = False
+
+    def visit_Call(self, node):
+        if isinstance(node.func, ast.Name) and node.func.id == 'input':
+            self.found_input = True
+        self.generic_visit(node)
+
+def has_input_function_calls(code):
+    try:
+        tree = ast.parse(code)
+    except SyntaxError:
+        return False
+    visitor = InputFunctionVisitor()
+    visitor.visit(tree)
+    return visitor.found_input
+
+def gradio_launch(model_path: str, MAX_TRY: int = 3):
+    with gr.Blocks() as demo:
+        gr.Markdown("# Online Demo of OpenCodeInterpreter Models")
+        gr.Markdown("**NOTE: Please read the disclaimer section in [README.md](https://huggingface.co/spaces/m-a-p/OpenCodeInterpreter_demo/blob/main/README.md) before using this demo!**")
+        gr.Markdown("**By using this demo, you acknowledge that you have read this disclaimer, understand its terms, and agree to be bound by them.**")
+        chatbot = gr.Chatbot(height=600, label="OpenCodeInterpreter", avatar_images=["assets/user.pic.jpg", "assets/assistant.pic.jpg"], show_copy_button=True)
+        with gr.Group():
+            with gr.Row():
+                msg = gr.Textbox(
+                    container=False,
+                    show_label=False,
+                    label="Message",
+                    placeholder="Type a message...",
+                    scale=7,
+                    autofocus=True
+                )
+                sub = gr.Button(
+                    "Submit",
+                    variant="primary",
+                    scale=1,
+                    min_width=150
+                )
+                # stop = gr.Button(
+                #     "Stop",
+                #     variant="stop",
+                #     visible=False,
+                #     scale=1,
+                #     min_width=150
+                # )
+
+        with gr.Row():
+            # retry = gr.Button("🔄  Retry", variant="secondary")
+            # undo = gr.Button("↩️ Undo", variant="secondary")
+            upvote = gr.Button(upvote_button_value, variant="secondary")
+            downvote = gr.Button(downvote_button_value, variant="secondary")
+            clear = gr.Button("🗑️  Clear", variant="secondary")
+
+        session_state = gr.State([])
+        jupyter_state = gr.State(JupyterNotebook())
+        dialog_info = gr.State(["", None])
+        demo.load(reset_dialog_info, dialog_info, dialog_info)
+
+        def bot(user_message, history, jupyter_state, dialog_info, interpreter):
+            logging.info(f"user message: {user_message}")
+            interpreter.dialog = convert_history(gradio_history=history, interpreter_history=interpreter.dialog)
+            history.append([user_message, None])
+
+            interpreter.dialog.append({"role": "user", "content": user_message})
+
+            # setup
+            HAS_CODE = False  # For now
+            prompt = interpreter.dialog_to_prompt(dialog=interpreter.dialog)
+
+            _ = interpreter.generate(prompt)
+            history[-1][1] = ""
+            generated_text = ""
+            for character in interpreter.streamer:
+                history[-1][1] += character
+                history[-1][1] = history[-1][1].replace("<|EOT|>","")
+                generated_text += character
+                yield history, history, jupyter_state, dialog_info
+
+            if is_valid_python_code(history[-1][1].strip()):
+                history[-1][1] = f"```python\n{history[-1][1].strip()}\n```"
+                generated_text = history[-1][1]
+
+            HAS_CODE, generated_code_block = interpreter.extract_code_blocks(
+                generated_text
+            )
+
+            interpreter.dialog.append(
+                {
+                    "role": "assistant",
+                    "content": generated_text.replace("<unk>_", "")
+                    .replace("<unk>", "")
+                    .replace("<|EOT|>", ""),
+                }
+            )
+
+            logging.info(f"saving current dialog to file {dialog_info[0]}.json...")
+            logging.info(f"current dialog: {interpreter.dialog}")
+            save_json(interpreter.dialog, mode="openci_only", flag=dialog_info[1], json_file_path=JSON_DATASET_DIR/f"{dialog_info[0]}.json", dialog_id=dialog_info[0])
+
+            attempt = 1
+            while HAS_CODE:
+                if attempt > MAX_TRY:
+                    break
+                # if no code then doesn't have to execute it
+                generated_text = "" # clear generated text
+
+                yield history, history, jupyter_state, dialog_info
+
+                # replace unknown thing to none ''
+                generated_code_block = generated_code_block.replace(
+                    "<unk>_", ""
+                ).replace("<unk>", "")
+
+                if has_input_function_calls(generated_code_block):
+                    code_block_output = "Please directly assign the value of inputs instead of using input() function in your code."
+                else:
+                    (
+                        code_block_output,
+                        error_flag,
+                    ) = interpreter.execute_code_and_return_output(
+                        f"{generated_code_block}",
+                        jupyter_state
+                    )
+                    if error_flag == "Timeout":
+                        logging.info(f"{dialog_info[0]}: Restart jupyter kernel due to timeout")
+                        jupyter_state = JupyterNotebook()
+                    code_block_output = interpreter.clean_code_output(code_block_output)
+
+                    if code_block_output.strip():
+                        code_block_output = "Execution result: \n" + code_block_output
+                    else:
+                        code_block_output = "Code is executed, but result is empty. Please make sure that you include test case in your code."
+
+                history.append([code_block_output, ""])
+
+                interpreter.dialog.append({"role": "user", "content": code_block_output})
+
+                yield history, history, jupyter_state, dialog_info
+
+                prompt = interpreter.dialog_to_prompt(dialog=interpreter.dialog)
+
+                logging.info(f"generating answer for dialog {dialog_info[0]}")
+                _ = interpreter.generate(prompt)
+                for character in interpreter.streamer:
+                    history[-1][1] += character
+                    history[-1][1] = history[-1][1].replace("<|EOT|>","")
+                    generated_text += character
+                    yield history, history, jupyter_state, dialog_info
+                logging.info(f"finish generating answer for dialog {dialog_info[0]}")
+
+                HAS_CODE, generated_code_block = interpreter.extract_code_blocks(
+                    history[-1][1]
+                )
+
+                interpreter.dialog.append(
+                    {
+                        "role": "assistant", 
+                        "content": generated_text.replace("<unk>_", "")
+                        .replace("<unk>", "")
+                        .replace("<|EOT|>", ""),
+                    }
+                )
+
+                attempt += 1
+
+                logging.info(f"saving current dialog to file {dialog_info[0]}.json...")
+                logging.info(f"current dialog: {interpreter.dialog}")
+                save_json(interpreter.dialog, mode="openci_only", flag=dialog_info[1], json_file_path=JSON_DATASET_DIR/f"{dialog_info[0]}.json", dialog_id=dialog_info[0])
+
+                if generated_text.endswith("<|EOT|>"):
+                    continue
+
+            return history, history, jupyter_state, dialog_info
+
+
+        def reset_textbox():
+            return gr.update(value="")
+
+        def set_button_variant(upvote_button_variant, downvote_button_variant):
+            return gr.Button(upvote_button_value, variant=upvote_button_variant), gr.Button(downvote_button_value, variant=downvote_button_variant)
+
+        def reset_button_and_flag(dialog_info):
+            return (*set_button_variant("secondary", "secondary"), [dialog_info[0], None])
+
+        def clear_history(history, jupyter_state, dialog_info, interpreter):
+            interpreter.dialog = []
+            jupyter_state.close()
+            return ([], [], JupyterNotebook(), reset_dialog_info(dialog_info), *set_button_variant("secondary", "secondary"))
+        
+        def toggle_preference(button, dialog_info):
+            if button == upvote_button_value:
+                dialog_info[1] = True
+            elif button == downvote_button_value:
+                dialog_info[1] = False
+            else:
+                raise ValueError(button)
+            logging.info(f"{button} is clicked by {dialog_info[0]}, current flag: {dialog_info[1]}")
+
+            if dialog_info[1] is None:
+                return (*set_button_variant("secondary", "secondary"), dialog_info)
+            elif dialog_info[1]:
+                return (*set_button_variant("primary", "secondary"), dialog_info)
+            else:
+                return (*set_button_variant("secondary", "primary"), dialog_info)
+        
+        def save_preference(dialog_info, interpreter):
+            if interpreter.dialog:
+                save_json(interpreter.dialog, mode="openci_only", flag=dialog_info[1], json_file_path=JSON_DATASET_DIR/f"{dialog_info[0]}.json", dialog_id=dialog_info[0])
+            return dialog_info
+
+        interpreter = StreamingOpenCodeInterpreter(model_path=model_path)
+
+        sub.click(reset_button_and_flag, dialog_info, [upvote, downvote, dialog_info])
+        sub.click(partial(bot, interpreter=interpreter), [msg, session_state, jupyter_state, dialog_info], [chatbot, session_state, jupyter_state, dialog_info])
+        sub.click(reset_textbox, [], [msg])
+
+        clear.click(
+            partial(clear_history, interpreter=interpreter),
+            [session_state, jupyter_state, dialog_info],
+            [chatbot, session_state, jupyter_state, dialog_info, upvote, downvote],
+            queue=False
+        )
+
+        upvote.click(
+            toggle_preference,
+            [upvote, dialog_info],
+            [upvote, downvote, dialog_info]
+        ).then(
+            partial(save_preference, interpreter=interpreter),
+            dialog_info,
+            dialog_info
+        )
+
+        downvote.click(
+            toggle_preference,
+            [downvote, dialog_info],
+            [upvote, downvote, dialog_info]
+        ).then(
+            partial(save_preference, interpreter=interpreter),
+            dialog_info,
+            dialog_info
+        )
+
+    demo.queue(max_size=20)
+    demo.launch(share=True)
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--path",
+        type=str,
+        required=False,
+        help="Path to the OpenCodeInterpreter Model.",
+        default="m-a-p/OpenCodeInterpreter-DS-6.7B",
+    )
+    args = parser.parse_args()
+
+    gradio_launch(model_path=args.path)

code_interpreter/BaseCodeInterpreter.py

@@ -0,0 +1,29 @@
+import os
+import sys
+import re
+
+prj_root_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.append(prj_root_path)
+
+
+from utils.const import *
+
+class BaseCodeInterpreter:
+    def __init__(self):
+        self.dialog = [
+            {
+                "role": "system",
+                "content": CODE_INTERPRETER_SYSTEM_PROMPT,
+            },
+        ]
+
+    @staticmethod
+    def extract_code_blocks(text: str):
+        pattern = r"```(?:python\n)?(.*?)```"  # Match optional 'python\n' but don't capture it
+        code_blocks = re.findall(pattern, text, re.DOTALL)
+        return [block.strip() for block in code_blocks]
+
+    def execute_code_and_return_output(self, code_str: str, nb):
+        _, _ = nb.add_and_run(GUARD_CODE)
+        outputs, error_flag = nb.add_and_run(code_str)
+        return outputs, error_flag

code_interpreter/JupyterClient.py

@@ -0,0 +1,85 @@
+from jupyter_client import KernelManager
+import threading
+import re
+from utils.const import *
+
+
+class JupyterNotebook:
+    def __init__(self):
+        self.km = KernelManager()
+        self.km.start_kernel()
+        self.kc = self.km.client()
+        _ = self.add_and_run(TOOLS_CODE)
+
+    def clean_output(self, outputs):
+        outputs_only_str = list()
+        for i in outputs:
+            if type(i) == dict:
+                if "text/plain" in list(i.keys()):
+                    outputs_only_str.append(i["text/plain"])
+            elif type(i) == str:
+                outputs_only_str.append(i)
+            elif type(i) == list:
+                error_msg = "\n".join(i)
+                error_msg = re.sub(r"\x1b\[.*?m", "", error_msg)
+                outputs_only_str.append(error_msg)
+
+        return "\n".join(outputs_only_str).strip()
+
+    def add_and_run(self, code_string):
+        # This inner function will be executed in a separate thread
+        def run_code_in_thread():
+            nonlocal outputs, error_flag
+
+            # Execute the code and get the execution count
+            msg_id = self.kc.execute(code_string)
+
+            while True:
+                try:
+                    msg = self.kc.get_iopub_msg(timeout=20)
+
+                    msg_type = msg["header"]["msg_type"]
+                    content = msg["content"]
+
+                    if msg_type == "execute_result":
+                        outputs.append(content["data"])
+                    elif msg_type == "stream":
+                        outputs.append(content["text"])
+                    elif msg_type == "error":
+                        error_flag = True
+                        outputs.append(content["traceback"])
+
+                    # If the execution state of the kernel is idle, it means the cell finished executing
+                    if msg_type == "status" and content["execution_state"] == "idle":
+                        break
+                except:
+                    break
+
+        outputs = []
+        error_flag = False
+
+        # Start the thread to run the code
+        thread = threading.Thread(target=run_code_in_thread)
+        thread.start()
+
+        # Wait for 20 seconds for the thread to finish
+        thread.join(timeout=20)
+
+        # If the thread is still alive after 20 seconds, it's a timeout
+        if thread.is_alive():
+            outputs = ["Execution timed out."]
+            # outputs = ["Error"]
+            error_flag = "Timeout"
+
+        return self.clean_output(outputs), error_flag
+
+    def close(self):
+        """Shutdown the kernel."""
+        self.km.shutdown_kernel()
+    
+    def __deepcopy__(self, memo):
+        if id(self) in memo:
+            return memo[id(self)]
+        new_copy = type(self)()
+        memo[id(self)] = new_copy
+        return new_copy

code_interpreter/OpenCodeInterpreter.py

@@ -0,0 +1,80 @@
+import sys
+import os
+
+prj_root_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.append(prj_root_path)
+
+from code_interpreter.BaseCodeInterpreter import BaseCodeInterpreter
+from utils.const import *
+
+from typing import List, Tuple, Dict
+import re
+
+import torch
+from transformers import AutoModelForCausalLM, AutoTokenizer
+
+
+sys.path.append(os.path.dirname(__file__))
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+
+import warnings
+
+warnings.filterwarnings("ignore", category=UserWarning, module="transformers")
+os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"
+
+
+class OpenCodeInterpreter(BaseCodeInterpreter):
+    def __init__(
+        self,
+        model_path: str,
+        load_in_8bit: bool = False,
+        load_in_4bit: bool = False,
+    ):
+        # build tokenizer
+        self.tokenizer = AutoTokenizer.from_pretrained(
+            model_path,
+            padding_side="right",
+            trust_remote_code=True
+        )
+
+        self.model = AutoModelForCausalLM.from_pretrained(
+            model_path,
+            device_map="auto",
+            load_in_4bit=load_in_4bit,
+            load_in_8bit=load_in_8bit,
+            torch_dtype=torch.float16,
+            trust_remote_code=True
+        )
+
+        self.model.resize_token_embeddings(len(self.tokenizer))
+
+        self.model = self.model.eval()
+
+        self.dialog = []
+        self.MAX_CODE_OUTPUT_LENGTH = 1000
+        
+
+    def dialog_to_prompt(self, dialog: List[Dict]) -> str:
+        full_str = self.tokenizer.apply_chat_template(dialog, tokenize=False)
+
+        return full_str
+
+    def extract_code_blocks(self, prompt: str) -> Tuple[bool, str]:
+        pattern = re.escape("```python") + r"(.*?)" + re.escape("```")
+        matches = re.findall(pattern, prompt, re.DOTALL)
+
+        if matches:
+            # Return the last matched code block
+            return True, matches[-1].strip()
+        else:
+            return False, ""
+
+    def clean_code_output(self, output: str) -> str:
+        if self.MAX_CODE_OUTPUT_LENGTH < len(output):
+            return (
+                output[: self.MAX_CODE_OUTPUT_LENGTH // 5]
+                + "\n...(truncated due to length)...\n"
+                + output[-self.MAX_CODE_OUTPUT_LENGTH // 5 :]
+            )
+
+        return output

requirements.txt

@@ -0,0 +1,32 @@
+accelerate==0.21.0
+bitsandbytes==0.41.1
+colorama==0.4.6
+coloredlogs==15.0.1
+colorlog==6.7.0
+datasets==2.12.0
+deepspeed==0.10.1
+diffusers==0.20.0
+einops==0.6.1
+gradio==3.48.0
+ipykernel==6.25.1
+ipython==8.12.2
+jupyter_client==8.3.0
+jupyter_core==5.3.0
+Markdown==3.4.3
+nbclient==0.8.0
+nbconvert==7.7.1
+nbformat==5.8.0
+omegaconf==2.3.0
+openai==0.27.7
+rich==13.7.0
+scikit-learn==1.4.0
+scipy==1.12.0
+seaborn==0.13.2
+sentencepiece==0.1.99
+termcolor==2.3.0
+tqdm==4.66.1
+transformers==4.37.1
+triton==2.0.0
+yfinance==0.2.28
+retrying==1.3.4
+pydantic<2.0.0

utils/cleaner.py

@@ -0,0 +1,31 @@
+import re
+import os
+
+PYTHON_PREFIX = os.environ.get("CONDA_PREFIX", "/usr/local")
+
+SITE_PKG_ERROR_PREFIX = f'File {PYTHON_PREFIX}/lib/python3.10/'
+
+def get_error_header(traceback_str):
+    lines = traceback_str.split('\n')
+    for line in lines:
+        if 'Error:' in line:
+            return line
+    return ''  # Return None if no error message is found
+
+def clean_error_msg(error_str:str =''):
+    filtered_error_msg = error_str.__str__().split('An error occurred while executing the following cell')[-1].split("\n------------------\n")[-1]
+    raw_error_msg = "".join(filtered_error_msg)
+
+    # Remove escape sequences for colored text
+    ansi_escape = re.compile(r'\x1b\[[0-?]*[ -/]*[@-~]')
+    error_msg = ansi_escape.sub('', raw_error_msg)
+    
+    error_str_out = ''
+    error_msg_only_cell = error_msg.split(SITE_PKG_ERROR_PREFIX)
+
+    error_str_out += f'{error_msg_only_cell[0]}\n'
+    error_header = get_error_header(error_msg_only_cell[-1])
+    if error_header not in error_str_out:
+        error_str_out += get_error_header(error_msg_only_cell[-1])
+
+    return error_str_out

utils/const.py

@@ -0,0 +1,120 @@
+TOOLS_CODE = """
+import numpy as np
+import pandas as pd 
+import matplotlib.pyplot as plt
+import seaborn as sns
+from scipy import stats
+import os,sys
+import re
+from datetime import datetime
+from sympy import symbols, Eq, solve
+import torch 
+import requests
+from bs4 import BeautifulSoup
+import json
+import math
+import yfinance
+import time
+"""
+
+write_denial_function = 'lambda *args, **kwargs: (_ for _ in ()).throw(PermissionError("Writing to disk operation is not permitted due to safety reasons. Please do not try again!"))'
+read_denial_function = 'lambda *args, **kwargs: (_ for _ in ()).throw(PermissionError("Reading from disk operation is not permitted due to safety reasons. Please do not try again!"))'
+class_denial = """Class Denial:
+    def __getattr__(self, name):
+        def method(*args, **kwargs):
+            return "Using this class is not permitted due to safety reasons. Please do not try again!"
+        return method
+"""
+
+GUARD_CODE = f"""
+import builtins
+
+_original_open = open
+
+def custom_open(file, mode='r', buffering=-1, encoding=None, errors=None, newline=None, closefd=True, opener=None):
+    if 'w' in mode or 'a' in mode or 'x' in mode or '+' in mode:
+        raise PermissionError("Writing operation is not permitted due to safety reasons. Please do not try again!")
+    return _original_open(file, mode, buffering, encoding, errors, newline, closefd, opener)
+
+builtins.open = custom_open
+
+builtins.exit = {write_denial_function}
+builtins.quit = {write_denial_function}
+
+import sys
+
+blocked_modules = ['pathlib', 'glob', 'ctypes']
+
+for module in blocked_modules:
+    sys.modules[module] = PermissionError
+
+import os
+
+os.listdir = {read_denial_function}
+os.scandir = {read_denial_function}
+os.walk = {read_denial_function}
+os.stat = {read_denial_function}
+os.kill = {write_denial_function}
+os.system = {write_denial_function}
+os.putenv = {write_denial_function}
+os.remove = {write_denial_function}
+os.removedirs = {write_denial_function}
+os.rmdir = {write_denial_function}
+os.fchdir = {write_denial_function}
+os.setuid = {write_denial_function}
+os.fork = {write_denial_function}
+os.forkpty = {write_denial_function}
+os.killpg = {write_denial_function}
+os.rename = {write_denial_function}
+os.renames = {write_denial_function}
+os.truncate = {write_denial_function}
+os.replace = {write_denial_function}
+os.unlink = {write_denial_function}
+os.fchmod = {write_denial_function}
+os.fchown = {write_denial_function}
+os.chmod = {write_denial_function}
+os.chown = {write_denial_function}
+os.chroot = {write_denial_function}
+os.fchdir = {write_denial_function}
+os.lchflags = {write_denial_function}
+os.lchmod = {write_denial_function}
+os.lchown = {write_denial_function}
+os.getcwd = {write_denial_function}
+os.chdir = {write_denial_function}
+os.popen = {write_denial_function}
+os.environ = {{}}
+os.getenv = {write_denial_function}
+builtins.open = {write_denial_function}
+
+import shutil
+
+shutil.rmtree = {write_denial_function}
+shutil.move = {write_denial_function}
+shutil.chown = {write_denial_function}
+
+import subprocess
+
+subprocess.Popen = {write_denial_function}  # type: ignore
+
+__builtins__["help"] = {write_denial_function}
+
+import sys
+
+sys.modules["ipdb"] = {write_denial_function}
+sys.modules["joblib"] = {write_denial_function}
+sys.modules["resource"] = {write_denial_function}
+sys.modules["psutil"] = {write_denial_function}
+sys.modules["tkinter"] = {write_denial_function}
+
+get_ipython().system = lambda *args, **kwargs: (_ for _ in ()).throw(PermissionError("Sorry, magic command is disabled due to safety reasons. Please do not try again!"))
+"""
+
+CODE_INTERPRETER_SYSTEM_PROMPT = """You are an AI code interpreter.
+Your goal is to help users do a variety of jobs by executing Python code.
+
+You should:
+1. Comprehend the user's requirements carefully & to the letter.
+2. Give a brief description for what you plan to do & call the provided function to run code.
+3. Provide results analysis based on the execution output.
+4. If error occurred, try to fix it.
+5. Response in the same language as the user."""