LangCode - Improving alignment and reasoning of large language models (LLMs) with natural language embedded program (NLEP).
def get_langpy(api_key='', platform='gpt', model='gpt-4'):
    try:
        import google.colab
        from .colab_lp import ColabLangPy
        autolp = ColabLangPy(api_key, platform=platform, model=model)
    except:
        from .jupyter_lp import JupyterLangPy
        autolp = JupyterLangPy(api_key, platform=platform, model=model)
    return autolp

langpy/auto.py
luohongyin/LangCode
from .data_loader.csv_loader import CsvLoader
from .data_loader.json_loader import JsonLoader

_API_ENDPOINT = 'https://lang-py-522564686dd7.herokuapp.com/items/0'
_SUPPORTED_PLATFORMS = {'gpt', 'palm'}


class LangPy:
    def __init__(self, api_key='', platform='gpt', model='gpt-4'):
        if not api_key:
            raise ValueError("Empty api_key")

        if not model:
            raise ValueError("Empty model")

        if platform not in _SUPPORTED_PLATFORMS:
            supported_platforms = ", ".join(_SUPPORTED_PLATFORMS)
            raise ValueError(
                "Non-supported platform. Supported ones: {supported_platforms}")

        self.api_key = api_key
        self.api_end_point = _API_ENDPOINT
        self.platform = platform
        self.model = model

    def get_http_prompt(self, http):
        prompt = ''
        if http == 'forbid':
            prompt = f'Do not send HTTP requests.'
        elif http == 'force':
            prompt = f'Send an HTTP request to solve the problem.'
        elif http == 'optional':
            pass
        else:
            raise ValueError(
                f'http mode: {http} not supported. try [forbid | force | optional]!')
        return prompt

    def get_data_loader_prompt(self,
                               notebook,
                               data=None,
                               file_name=None,
                               data_type='csv',  # csv or json
                               read_mode='r',
                               num_case=2):
        if data_type == 'csv':
            loader = CsvLoader(data, file_name, read_mode)
        elif data_type == 'json':
            loader = JsonLoader(data, file_name, read_mode)
        else:
            raise ValueError(
                f'data_type = {data_type} not supported. Please select from [csv | json]')
        return loader.get_prompt(notebook, num_case)

    def config_api_key(self):
        api_key = input('API key:')
        self.api_key = api_key

    def config_end_point(self):
        self.api_end_point = input('API endpoint:')

    def generate(self, instruction, hist_mode='all', http='optional'):
        pass

    def complete(self, instruction, hist_mode='all', http='optional'):
        pass

    def process_csv(self, file_name):
        pass

langpy/base.py
luohongyin/LangCode
import json
import requests
from .base import LangPy
from google.colab import _message


class ColabLangPy(LangPy):
    def __init__(self, api_key='', platform='gpt', model='gpt-4'):
        super().__init__(api_key=api_key, platform=platform, model=model)
        self._colab_list = []
        self._cur_idx = -1

    def format_cell(self, cell_tuple):
        cell_str, cell_type = cell_tuple
        if cell_type == 'code':
            return f'```\n{cell_str}\n```'
        else:
            return cell_str

    def post_code(self, code):
        _message.blocking_request(
            'add_scratch_cell',
            request={'content': code, 'openInRightPane': True},
            timeout_sec=None
        )

    def update_idx_hist(self, instruction):
        colab_hist = _message.blocking_request('get_ipynb')['ipynb']['cells']
        self._code_list = [
            [''.join(x['source']), x['cell_type']] for x in colab_hist
        ]

        self._code_list = [
            self.format_cell(ct) for ct in self._code_list
        ]

        self._cur_idx = -1
        for i, code in enumerate(self._code_list):
            if instruction in code:
                self._cur_idx = i
                break

    def get_history_prompt(self, instruction, hist_mode):
        prompt = ''
        if hist_mode == 'all':
            previous_code = '\n\n'.join(self._code_list[:self._cur_idx])
            prompt = f'{previous_code}\n\n{instruction}'
        elif hist_mode == 'single':
            prompt = instruction
        elif isinstance(hist_mode, int):
            previous_code = '\n\n'.join(
                self._code_list[self._cur_idx - hist_mode: self._cur_idx])
            prompt = f'{previous_code}\n\n{instruction}'
        else:
            raise ValueError(
                f'hist_mode = {hist_mode} not supported. try [all | single | INTEGER].')
        return prompt

    def get_prompt(self, instruction, hist_mode, http):
        history_prompt = self.get_history_prompt(instruction, hist_mode)
        http_prompt = self.get_http_prompt(http)
        return f'{history_prompt} {http_prompt}'

    def get_code_of_thought(self, json_data):
        response = requests.put(self.api_end_point, json=json_data)
        if response.status_code != 200:
            raise SystemError(
                f'Failed to get valid response from server: {response.status_code}')
        return response.content

    def generate(self, instruction, hist_mode='single', http='optional'):
        self.update_idx_hist(instruction)
        response = self.get_code_of_thought(
            json_data={
                'instruction': self.get_prompt(instruction, hist_mode, http),
                'api_key_str': self.api_key,
                'exist_code': 'none',
                'platform': self.platform,
                'model': self.model
            }
        )
        ans_str = json.loads(response)['output']
        self.post_code(ans_str)

    def complete(self, instruction, hist_mode='single', http='optional'):
        self.update_idx_hist(instruction)
        response = self.get_code_of_thought(
            json_data={
                'instruction': self.get_prompt(instruction, hist_mode, http),
                'api_key_str': self.api_key,
                'exist_code': self._code_list[self._cur_idx + 1].replace('```', '').strip(),
                'platform': self.platform,
                'model': self.model
            }
        )
        ans_str = json.loads(response)['output']
        self.post_code(ans_str)

    def preview_data(
            self,
            data=None,
            file_name=None,
            data_type='csv',  # csv or json
            read_mode='r',
            num_case=2):
        prompt = self.get_data_loader_prompt(
            'colab', data, file_name, data_type, read_mode, num_case)
        self.post_code(prompt)

langpy/colab_lp.py
luohongyin/LangCode
from .loader import Loader


class CsvLoader(Loader):
    def __init__(self, data=None, file_name=None, read_mode='r'):
        super().__init__(data, file_name, read_mode)
        if file_name is None:
            raise ValueError('CsvLoader must process a file. set file_name.')

    def file_prompt(self, num_row, content_str):
        f_prompt = f"""# First {num_row} rows of the input file:\nexample_data = '''\n{content_str}\n ...\n'''\nfile_name = '{self.file_name}' \ninput_file = open(file_name)"""
        return f_prompt

    def get_prompt(self, notebook='jupyter', num_case=2):
        num_row = num_case + 1
        content_list = open(self.file_name, self.read_mode).readlines()[
            :num_row]
        content_list = [x.strip() for x in content_list]
        content_str = '\n# '.join(content_list)

        prompt = self.file_prompt(num_row, content_str)
        return self.format_prompt(prompt, notebook)

langpy/data_loader/csv_loader.py
luohongyin/LangCode
import json
from .loader import Loader


class JsonLoader(Loader):
    def __init__(self, data=None, file_name=None, read_mode='r'):
        super().__init__(data, file_name, read_mode)
        if data is None and file_name is None:
            raise ValueError('Must provide either `data` or `file_name`.')

    def file_prompt(self, num_case, content_str):
        f_prompt = f"""# First {num_case} cases of the Json data:\nexample_data = '''\n{content_str}\n...\n'''\nfile_name = '{self.file_name}' \ninput_file = open(file_name)"""
        return f_prompt

    def data_prompt(self, num_case, content_str):
        d_prompt = f"""# First {num_case} cases of the Json data:\nexample_data = '''\n{content_str}\n...\n'''"""
        return d_prompt

    def get_prompt(self, notebook='jupyter', num_case=2):
        if self.data is None:
            self.data = json.load(open(self.file_name, self.read_mode))

        example_list = self.data[:num_case]
        content_str = ',\n'.join([json.dumps(x, indent=4)
                                 for x in example_list])
        
        if self.file_name is not None:
            prompt = self.file_prompt(num_case, content_str)
        else:
            prompt = self.data_prompt(num_case, content_str) 

        return self.format_prompt(prompt, notebook)

langpy/data_loader/json_loader.py
luohongyin/LangCode
class Loader:
    def __init__(self, data=None, file_name=None, read_mode='r'):
        self.data = data
        self.file_name = file_name
        self.read_mode = read_mode

    def format_prompt(self, prompt, notebook):
        if notebook == 'jupyter':
            prompt = prompt.replace('\n', '\\n').replace('\t', '\\t')
        elif notebook == 'colab':
            pass
        else:
            raise ValueError(
                f'Notebook = {notebook} note supported. Use [jupyter | colab]')
        return prompt
langpy/data_loader/loader.py
luohongyin/LangCode
import IPython
from IPython.core.display import display, Javascript
from .base import LangPy


class JupyterLangPy(LangPy):
    def __init__(self, api_key='', platform='gpt', model='gpt-4'):
        super().__init__(api_key=api_key, platform=platform, model=model)

    def get_prompt(self, instruction, http):
        http_prompt = self.get_http_prompt(http)
        return f'{instruction} {http_prompt}'

    def init_js_code(self):
        return """
        async function myCallbackFunction(data) {
            const url = '%s';
            try {
                let response = await fetch(url, {
                    method: "PUT",
                    headers: {
                    "Content-Type": "application/json",
                    },
                    body: JSON.stringify(data),
                });
                let responseData = await response.json();
                return responseData;
            } catch (error) {
                console.log(error);
            }
        }
        var current_index = Jupyter.notebook.get_selected_index();
        var previous_cell_content = "";
        """ % self.api_end_point

    def js_get_history(self, hist_mode):
        if hist_mode == "all":
            js_code = """
        for (var i = 1; i < current_index; i++) {
            var cell = Jupyter.notebook.get_cell(i);
            if (cell.cell_type === "code") {
                previous_cell_content += '```\\n' + cell.get_text() + '\\n```\\n';
            } else {
                previous_cell_content += cell.get_text() + '\\n';
            }
        }
        """
        elif isinstance(hist_mode, int):
            js_code = """
        for (var i = current_index - %d; i < current_index; i++) {
            var cell = Jupyter.notebook.get_cell(i);
            if (cell.cell_type === "code") {
                previous_cell_content += '```\\n' + cell.get_text() + '\\n```\\n';
            } else {
                previous_cell_content += cell.get_text() + '\\n';
            }
        }
        """ % hist_mode
        elif hist_mode == 'single':
            js_code = ''
        else:
            raise ValueError(
                f"hist_mode {hist_mode} not supported. Try [all | single | INT]")

        return js_code

    def build_request_data(self, instruction, exist_code=False):
        if exist_code:
            js_request = """
            previous_cell_content += '%s'
            var api_key_str = "%s"
            var next_cell = Jupyter.notebook.get_cell(current_index)
            var exist_code = next_cell.get_text()

            var data = {
                "instruction": previous_cell_content,
                "api_key_str": api_key_str,
                "exist_code": exist_code,
                "platform": "%s",
                "model": "%s"
            }
            """ % (instruction, self.api_key, self.platform, self.model)
        else:
            js_request = """
            previous_cell_content += '%s'
            var api_key_str = "%s"
            var data = {
                "instruction": previous_cell_content,
                "api_key_str": api_key_str,
                "exist_code": "none",
                "platform": "%s",
                "model": "%s"
            }
            """ % (instruction, self.api_key, self.platform, self.model)
        return js_request

    def build_js_code(self, instruction, hist_mode, http, exist_code=False):
        instruction = self.get_prompt(instruction, http)

        js_code = self.init_js_code()
        js_code += self.js_get_history(hist_mode)
        js_code += self.build_request_data(instruction, exist_code=exist_code)
        return js_code

    def generate(self, instruction, hist_mode='single', http='optional'):
        js_code = self.build_js_code(
            instruction, hist_mode, http, exist_code=False
        )

        js_code += """
        myCallbackFunction(data).then(responseData => {
            var print_content = responseData["output"]
            Jupyter.notebook.select_prev()
            var new_cell = Jupyter.notebook.insert_cell_below('code');

            var next_cell = Jupyter.notebook.get_cell(current_index)
            next_cell.set_text(print_content);
        }).catch()
        """

        print('The code will be generated in the next cell ...')

        display(Javascript(js_code))

    def complete(self, instruction, hist_mode='single', http='optional'):
        js_code = self.build_js_code(
            instruction, hist_mode, http, exist_code=True
        )

        js_code += """
        myCallbackFunction(data).then(responseData => {
            var print_content = responseData["output"]
            var next_cell = Jupyter.notebook.get_cell(current_index)
            next_cell.set_text(print_content);
        })
        """
        print('The code in the following cell will be completed ...')

        display(Javascript(js_code))

    def preview_data(
            self,
            data=None,
            file_name=None,
            data_type='csv',  # csv or json
            read_mode='r',
            num_case=2):
        prompt = self.get_data_loader_prompt(
            'jupyter', data, file_name, data_type, read_mode, num_case)

        js_code = f"""
            Jupyter.notebook.select_prev()
            var new_cell = Jupyter.notebook.insert_cell_below('code');
            new_cell.set_text(`{prompt}`);
        """
        print('Run the following cell to read process the format of your csv file.')

        display(Javascript(js_code))

langpy/jupyter_lp.py
luohongyin/LangCode
import torch
from datasets import load_dataset
from tqdm import tqdm
import json

def classify_sst2():
    # Step 1: Import necessary built-in libraries
    from textblob import TextBlob

    # Step 2: Define a function that analyze the sentiment of a text
    def analyze_sentiment(text):
        sentiment = TextBlob(text).sentiment.polarity
        if sentiment > 0:
            return "positive"
        elif sentiment < 0:
            return "negative"
        else:
            return "neutral"

    dataset = load_dataset('sst2')
    val_dataset = dataset["validation"]
    acc = 0
    tot = 0
    for item in tqdm(val_dataset):
        tot += 1
        
        # Step 3: Define the text of movie review
        review = item['sentence']

        # Step 4: Print an answer in natural language using the knowledge and function defined above
        sentiment_result = analyze_sentiment(review)
        if sentiment_result == 'positive':
            label = 1
        else:
            label = 0
        if label == item['label']:
            acc += 1

    print(acc/tot)


def classify_cola():
    # Step 1: Import necessary built-in libraries
    import nltk

    # Step 2: Define a function that checks the grammar correctness of sentences
    def check_sentence(sentence):
        try:
            nltk.pos_tag(nltk.word_tokenize(sentence))
            return "acceptable"
        except:
            return "unacceptable"

    # Step 3: Define the input sentence
    dataset = load_dataset('glue', 'cola')
    val_dataset = dataset["validation"]
    acc = 0
    tot = 0
    for item in tqdm(val_dataset):
        tot += 1
        sentence = item['sentence']
        label = check_sentence(sentence)
        # Step 4: Classify the grammar of the sentence and print the result
        if item['label'] == 1:
            truth_label = 'acceptable'
        else:
            truth_label = 'unacceptable'
        if truth_label == label:
            acc += 1
       
    print(acc/tot)


def classifying_emotion():
    # Step 1: Import necessary built-in libraries
    from textblob import TextBlob

    # Step 2: Define a function that classify the emotion of the sentence
    def classify_emotion(sentence):
        blob = TextBlob(sentence)
        if blob.sentiment.polarity < 0:
            return "sad"
        elif blob.sentiment.polarity == 0:
            return "surprised"
        else:
            if "love" in sentence:
                return "love"
            elif "angry" in sentence:
                return "angry"
            elif "afraid" in sentence:
                return "afraid"
            else:
                return "happy"

    dataset = load_dataset('emotion')
    val_dataset = dataset["validation"]
    tot = 0
    acc = 0
    all_classes = ['sad', 'happy', 'love', 'angry', 'afraid', 'surprised']
    for item in tqdm(val_dataset):
        tot += 1
        # Step 3: Define a sentence 
        sentence = item['text']
       
        # Step 5: Print a classification answer in natural language using the defined function and sentence
        emotion = classify_emotion(sentence)
        if emotion == all_classes[item['label']]:
            acc += 1
    print(acc/tot)


def classify_review():
    # Step 1: Import necessary built-in libraries
    from textblob import TextBlob

    # Step 2: Define a function to analyze the stars of a given review
    def get_stars(review):
        analysis = TextBlob(review)
        # The polarity of TextBlob is in a range from -1 to 1, where -1 refers to negative sentiment and 1 to positive sentiment.
        # In order to map it to the star range of 1-5, we need to add 1 to make the polarity in 0-2 range, then divide it by 2 to make it in 0-1 range, 
        # and finally multiply it by (5 - 1) and add 1 to make it in 1-5 range.
        stars = round((analysis.sentiment.polarity + 1) / 2 * 4 + 1)
        return stars

    dataset = load_dataset('amazon_reviews_multi', 'en')
    val_dataset = dataset["validation"]
    tot = 0
    acc = 0
    classes = [1, 2, 3, 4, 5]  # Possible star rating

    for item in tqdm(val_dataset):
        tot += 1
        # Step 3: Specify a review to analyze
        review = item['review_title'] + item['review_body']

        # Step 4: Print the star from the analysis
        stars = get_stars(review)
        label = item['stars']
        if stars == label:
            acc += 1
    print(acc/tot)



def classify_hsd():
    # Step 1: Import necessary built-in libraries
    from nltk.sentiment.vader import SentimentIntensityAnalyzer
    from nltk.tokenize import word_tokenize
    import nltk
    nltk.download('vader_lexicon')

    # Step 2: Define a function to check feelings of sentences
    def hate_speech_check(speech):
        words = word_tokenize(speech)
        # We use nltk library to determine the feelings of the speech
        sid = SentimentIntensityAnalyzer()
        sentiment = sid.polarity_scores(speech)
        # Considered hate speech if compound sentiment is less than -0.3
        if sentiment['compound'] < -0.3:
            return True
        else:
            return False

    with open('../hate-speech-dataset/test_sample.json', 'r') as f:
        data = json.load(f)
    tot = 0
    acc = 0
    for item in data:
        tot += 1
        # Step 3: Define dictionaries storing the speech
        speech = item['sentence']

        # Step 4: Print an result using the defined function and speech
        is_hate_speech = hate_speech_check(speech)
        if is_hate_speech:
            if item['label'] == 'hate':
                acc += 1
        else:
            if item['label'] != 'hate':
                acc += 1
    print(acc/tot)



def classify_sbic():
    # Step 1: Import necessary built-in libraries
    import re

    # Step 2: Define a dictionary storing keywords which identify offensiveness
    keywords = {
        "Yes": ["hate", "kill", "racist", "sexist", "explicit"], # A list containing some offensive words for example
        "Maybe": ["idiot", "stupid", "dumb", "weirdo"], # A list containing some maybe offensive words for example
        "No": ["happy", "love", "beautiful", "amazing", "fantastic"] # A list containing some positive words for example
    }

    # Step 3: Define a function that classify a post
    def classify_post(post, keywords):
        word_occurrences = {"Yes": 0, "Maybe": 0, "No": 0} # Initialize a dictionary to store the number of occurrences 
        for keyword in re.findall(r'\w+', post): # Split the post into words
            for key in keywords.keys():
                if keyword.lower() in keywords[key]:
                    word_occurrences[key] += 1
            
        max_key = max(word_occurrences, key=word_occurrences.get) # Find the key with maximum occurrences
        if word_occurrences[max_key] > 0:
            return max_key
        else:
            return "No"    
    
    dataset = load_dataset('social_bias_frames')
    val_dataset = dataset["validation"]
    tot = 0
    acc = 0
    for item in tqdm(val_dataset):
        tot += 1
        # Step 4: Define dictionaries storing the post
        post = item['post']
        truth = item['offensiveYN']
        
        if truth == '1.0':
            truth = 'Yes'
        elif truth == '0.5':
            truth = 'Maybe'
        else:
            truth = 'No'

        # Step 5: Print an answer in natural language using the knowledge and function defined above
        classification = classify_post(post, keywords)
        if truth == classification:
            acc += 1
    print(acc/tot)


def classify_agnews():
    # Step 1: Import necessary built-in libraries
    from sklearn.feature_extraction.text import CountVectorizer
    from sklearn.naive_bayes import MultinomialNB

    # Step 2: Define the training dataset for the classification model
    news_data = {
        "world news": [
            "The 193-member United Nations General Assembly elected the United Arab Emirates, Brazil, Albania, Gabon, and Ghana to serve on the U.N. Security Council for two years.",
            "North Korea announced that it has test launched series of tactical guided missiles.",
            "European Union leaders reach agreement on a new tax proposal that affects multinational companies."
        ],
        "sport news": [
            "The German team edged out Portugal in a high-scoring Euro 2020 group-stage game.",
            "Serena Williams withdraws from Wimbledon due to injury.",
            "Tokyo Olympic organizers announced a limit on spectators in the upcoming summer games."
        ],
        "business news": [
            "Airline companies see a rise in domestic travel bookings as Covid-19 restrictions begin to ease.",
            "Shares of AMC Entertainment soared due to coordinated trading by individual investors.",
            "Microsoft Corp is set to acquire LinkedIn Corp for $26.2 billion in cash."
        ],
        "technology news": [
            "Apple Inc. announces new privacy features for its upcoming iOS 15 operating system.",
            "Facebook Inc tests new tool for users to manage content from their News Feed.",
            "Google's proposed acquisition of Fitbit to be closely scrutinized by the EU."
        ]
    }

    # Step 3: Prepare the data
    X_train = []
    y_train = []
    for category, news_list in news_data.items():
        for news in news_list:
            X_train.append(news)
            y_train.append(category)

    vectorizer = CountVectorizer()
    X_train = vectorizer.fit_transform(X_train)

    # Step 4: Train the classifier
    clf = MultinomialNB()
    clf.fit(X_train, y_train)

    classes = ['world news', 'sport news', 'business news', 'technology news']
    dataset = load_dataset('ag_news', cache_dir = '/shared/jiaxin/.cache')
    val_dataset = dataset["test"]
    tot = 0
    acc = 0
    for item in tqdm(val_dataset):
        tot += 1
        # Step 5: Define the news piece to be classified
        piece_of_news = item['text']
        # Step 6: Make the prediction
        X_test = vectorizer.transform([piece_of_news])
        predicted_category = clf.predict(X_test)[0]

        if predicted_category == classes[item['label']]:
            acc += 1
    print(acc/tot)

research/nlu/direct_NLEP/direct_NLEP_execution.py
luohongyin/LangCode
import openai
import backoff

prompt_head = """Write a bug-free Python program that can generate the answer to the given instruction when correctly executed. Do not ask for user input. For reasoning tasks, define functions first and then define variables. For knowledge intensive tasks, define variables before defining functions.

### Instruction: Discuss the causes of the Great Depression
### Input: None
### Python program:

# Step 1: Import necessary built-in ligraries
# No need to import

# Step 2: Define dictionaries storing the knowledge about the grat depression
depression_name = "The Great Depression"
depression_period = "1929-1939"
depression_countries = "the United States and countries around the world"
depression_causes = {
    "Stock Market Crash of 1929": "In October of 1929, the stock market experienced a significant fall that wiped out millions of investors. This event is considered by many to be the initial trigger of the Great Depression.",
    "Overproduction": "During the 1920s, many industries produced more goods than consumers wanted or could afford. This ultimately led to a decline in demand for goods, causing job loss, lower wages, and business failure.",
    "High Tariffs and War Debts": "Protectionist trade policies in the form of high tariffs led to a decline in global trade, as other countries retaliated with tariffs of their own. Additionally, many countries were struggling to repay war debts, which led to economic instability.",
    "Bank Failures": "As demand for goods declined, many banks began to fail, causing a loss of confidence in the banking system. This led to a massive withdrawal of money from banks, causing even more banks to fail.",
    "Drought Conditions": "The Dust Bowl was a severe drought and dust storm that hit the Great Plains region of the United States in the 1930s. This had a significant impact on agriculture, causing many farmers to lose their land and livelihoods which worsened the effects of the depression."
}

# Step 3: Define necessary functions that generally solve this type of problem
# Do not need to define functions

# Step 4: Print an answer in natural language using the knowledge defined above
print(f"{depression_name} was a period of economic decline that lasted from {depression_period}, making it the longest-lasting depression in modern history. It affected not only {depression_countries}, causing substantial social and economic upheaval.\n")
print(f"There were several major causes of {depression_name}, which include:\n")
for i, (cause, description) in enumerate(depression_causes.items(), 1):
    print(f"{i}. {cause} - {description}\n")
print(f"Overall, {depression_name} was caused by a combination of factors, including economic, environmental, and political factors. Its impact was widespread, affecting millions of people around the world.")
```

### Instruction: Identify the odd one out.
### Input: Twitter, Instagram, Telegram
### Python program:
```
# Step 1: Import necessary built-in libraries
from collections import OrderedDict

# Step 2: Define dictionaries storing the knowledge about the main function of each application
services = {
    "Twitter": "a social media platform mainly for sharing information, images and videos",
    "Instagram": "a social media platform mainly for sharing information, images and videos",
    "Telegram": "a cloud-based instant messaging and voice-over-IP service",
}

# Step 3: Define a function that finds the different application
def find_odd_one_out(services, input_services):
    descriptions = [services[service] for service in input_services]
    for description in descriptions:
        if descriptions.count(description) == 1:
            return input_services[descriptions.index(description)]
    return None

# Step 4: Print an answer in natural language using the knowledge and function defined above
input_services = ["Twitter", "Instagram", "Telegram"]
odd_one_out = find_odd_one_out(services, input_services)
if odd_one_out:
    other_services = [service for service in input_services if service != odd_one_out]
    print(f"The odd one out is {odd_one_out}. {other_services[0]} and {other_services[1]} are {services[other_services[0]]} while {odd_one_out} is {services[odd_one_out]}.")
```

### Instruction: Calculate the total surface area of a cube with a side length of 5 cm.
### Input:  None
### Python program:
```
# Step 1: Import necessary built-in libraries
# No need to import

# Step 2: Define a function that calculate the surface area of cubes
def calculate_surface_area(side_length):
    return 6 * (side_length ** 2)

# Step 3: Define dictionaries storing the cube information
cube = {
    "side_length": 5  # Side length of the cube
}

# Step 4: Print a step-by-step calculation answer in natural language using the defined function and varible
side_length = cube["side_length"]
surface_area = calculate_surface_area(side_length)
print(f"The surface area of a cube is found by calculating the area of one of its faces and multiplying it by six (since a cube has six faces). The area of a cube face is simply its side length squared.\n")
print(f"Thus for this particular cube:")
print(f"Surface Area = 6 × (Side Length)²")
print(f"             = 6 × ({side_length} cm)²")
print(f"             = 6 × {side_length**2} cm²")
print(f"             = {surface_area} cm²\n")
print(f"The total surface area of this cube is {surface_area} square centimeters.")
"""
prompt_sst2 = """

### Instruction: Given a movie review, determine the attitude of this review as either positive or negative.
### Input: A moview review.
### Python program:

"""
prompt_cola = """

### Instruction: Given a sentence, determine the grammar correctness of this sentence as either acceptable or unacceptable.
### Input: A sentence.
### Python program:

"""
prompt_emotion = """

### Instruction: Given a sentence, classify the emotion of the sentence as one of these classes:  ['sad', 'happy', 'love', 'angry', 'afraid', 'surprised']. Your choice should be strictly the name of one of these 6 classes.
### Input: A sentence "I feel xxx"
### Python program:

"""
prompt_amazon = """

### Instruction: Given an amazon review, determine the star(1-5) of this review, where 1 is the worst and 5 is the best.
Your choice should be strictly in [1,2,3,4,5]
### Input: An amazon review.
### Python program:

"""

prompt_hsd = """

### Instruction: Given a speech from a white supremacy forum, determine if this is speech contains hate (eg. bias/racisim/prejudice...) towards a certain group of people.
### Input: A speech from a forum.
### Python program:

"""

prompt_sbic = """

### Instruction: Given a post, determine if the content of the post is offensive, maybe offensive, or not offensive. Classify the post as: ['Yes', 'Maybe', 'No']
### Input: A potentially offensive post.
### Python program:

"""
prompt_agnews = """

### Instruction: Given a piece of news, determine which category it belongs to. Classify it as one of the four classes: ['world news', 'sport news', 'business news', 'technology news']
### Input: A piece of news.
### Python program:

"""
@backoff.on_exception(backoff.expo, openai.error.RateLimitError)
def completions_with_backoff(**kwargs):
    return openai.ChatCompletion.create(**kwargs)

model_name = 'gpt-4'
content = prompt_head + prompt_agnews
response = completions_with_backoff(
    model=model_name, 
    messages=[
        {"role": "user","content": content}
        ],
    )
response = response["choices"][0]["message"]["content"]
print(response)







research/nlu/direct_NLEP/direct_NLEP_generation.py
luohongyin/LangCode
import openai
import backoff

prompt_head = """Write a Python function that constructs a decision tree according to the given examples that can generate the correct label of the given classification task.

### Available APIs(shared for all tasks):

\"\"\"Returns whether the hypothesis in entailed by the premise.\"\"\"
def entailment(hypothesis, premise, model, tokenizer):
    proposition = f'{hypothesis} is entailed by {premise}.'
    inputs = tokenizer(proposition,  return_tensors="pt", truncation=True, padding=True, max_length=128)
    outputs = model(**inputs)['logits'][0]
    ent_label = int(outputs[0] > outputs[2])
    if ent_label == 1:
        return 'yes'
    else:
        return 'no'

\"\"\"Use the constructed decision tree to predict the label of the sentence.\"\"\"
def tree_predict(sentence, criterions, tree, model, tokenizer):
    node = tree['root']
    while node not in POSSIBLE_CLASSES:
        ent_label = entailment(criterions[node], sentence, model, tokenizer)
        node = tree[node][ent_label]
    return node

### Task: Movie review classification
### Description: Determine if a movie review expresses positive attitude or negative attitude.
### Possible classes: [positive, negative]
### Examples:
- contains no wit, only labored gags
    - [The movie is wise|The movie is not wise|1], [the story is fun|the story is not boring|1], [the review is positive|the review is negative|1]
- that loves its characters and communicates something rather beautiful about human nature
    - [The characters are lovely|The characters are awful|0], [the script is touching|the script is dry|0], [the review is positive|the review is negative|0]
- on the worst revenge-of-the-nerds clichés the filmmakers could dredge up
    - [The movie is novel|The movie is mostly platitudes|1], [the review is negative|1]
- are more deeply thought through than in most right-thinking films
    - [The takeaway of the movie is profound|The idea of the movie is shallow|0], [the review is positive|the review is negative|0]

### Define possible classes
POSSIBLE_CLASSES = ['positive', 'negative']

### Decision Tree Logic:
Start by assessing whether the movie is interesting.
- If uninteresting:
    Probe for depth by determining if the movie is wise.
    - If wise, label as a positive review.
    - Otherwise, label as a negative review.
- If interesting:
    Examine the quality of the script.
    - If the script is commendable, label as a positive review.
    - If not a good script
        Evaluate the portrayal of characters.
        - If the characters are well-depicted, label the review as positive.
        - If characters are not good, label the review negative.

### Python program: 
\"""Decision tree for this task\"""
def get_decision_tree(sentence, model, tokenizer):
    # Step 1: define criterions of the decision tree.
    criterions = {
        'is_interesting':'This movie is interesting',
        'is_good_script':'The movie has a good script',
        'is_good_character':'The characters are awsome',
        'is_wise': 'This movie is wise'
    }

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 'is_interesting',
        'is_interesting': {'yes': 'is_good_script', 'no': 'is_wise'},
        'is_good_script': {'yes': 'positive', 'no': 'is_good_character'},
        'is_good_character': {'yes': 'positive', 'no': 'negative'},
        'is_wise': {'yes': 'positive', 'no': 'negative'}
    }

    return criterions, tree

"""

prompt_cola ="""
### Task: Grammar correctness classification
### Possible classes: [acceptable, unacceptable]

### Define possible classes
POSSIBLE_CLASSES = ['acceptable', 'unacceptable']

### Python program: 
\"""Decision tree for this task\"""
Try to generate a more balanced tree!
def get_decision_tree(sentence, model, tokenizer):

"""

def get_cola_tree():
    # Step 1: define criterions of the decision tree.
    criterions = {
        'has_subject': 'The sentence has subject',
        'has_verb': 'The sentence has verb',
        'punctuation_correct':'The sentence has proper punctuations',
        'pronouns_reference_match':'This sentence has matched pronouns and references',
        'subject_verb_match':'The sentence has matched subject and verb',
    }

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 'has_subject',
        'has_subject': {'yes': 'has_verb', 'no': 'unacceptable'},
        'has_verb': {'yes': 'punctuation_correct', 'no': 'unacceptable'},
        'punctuation_correct': {'yes': 'pronouns_reference_match', 'no': 'unacceptable'},
        'pronouns_reference_match': {'yes': 'subject_verb_match', 'no': 'unacceptable'},
        'subject_verb_match':{'yes': 'acceptable', 'no': 'unacceptable'}
    }

    return criterions, tree

prompt_emotion ="""
### Task: Emotion classification
### Description: I wrote a sentence about my feeling, determine which emotion am I feeling.
### Possible classes: ['I feel sad', 'I feel happy', 'I feel love', 'I feel angry', 'I feel afraid', 'I feel surprised']

### Define possible classes
POSSIBLE_CLASSES = ['I feel sad', 'I feel happy', 'I feel love', 'I feel angry', 'I feel afraid', 'I feel surprised']

### Decision Tree Logic:

"""

def get_emotion_tree():
    # Step 1: define criterions of the decision tree.
    criterions = {
        'positive_words': 'The sentence includes positive words',
        'intensifiers': 'The sentence includes intensifiers, adverb or exclamation point',
        'unexpected': 'The sentence includes words that are synonyms or antonyms of unexpected',
        'fear': 'The sentence includes words that are synonyms or antonyms of fear',
        'upset': 'The sentence includes words that are synonyms or antonyms of upset'
    }

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 'positive words',
        'positive_words': {'yes': 'intensifiers', 'no': 'fear'},
        'intensifiers': {'yes': 'unexpected', 'no': 'I feel happy'},
        'unexpected': {'yes': 'I feel surprised', 'no': 'I feel love'},
        'fear': {'yes': 'I feel afraid', 'no': 'upset'},
        'upset': {'yes': 'I feel sad', 'no': 'I feel angry'}
    }

    return criterions, tree

prompt_amazon = """
### Task: Amazon Review Star classification
### Description: Given an amazon review, determine the star(1-5) of this review, where 1 is the worst and 5 is the best.
### Possible classes: ['1', '2', '3', '4', '5']

### Define possible classes
POSSIBLE_CLASSES = ['1', '2', '3', '4', '5']

### Python program: 
\"""Decision tree for this task\"""
Try to generate a balanced tree!
def get_decision_tree(sentence, model, tokenizer):

"""

def get_amazon_tree():
    # Step 1: define criterions of the decision tree.
    criterions = {
        'extreme_word': 'The review includes intensifiers or exclamation point',
        'positive': 'The review includes positive words or sentences',
        'both_positive_negative':'The review includes both positive and negative sentences',
    }

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 'extreme_word',
        'extreme_word': {'yes': 'positive', 'no': 'both_positive_negative'},
        'positive': {'yes': '5', 'no': '1'},
        'both_positive_negative': {'yes': '3', 'no': 'positive'},
        'positive': {'yes': '4', 'no': '2'}
    }

    return criterions, tree

prompt_hate_speech = """
### Task: Bias/Hateful speech classification
### Description: Given a speech from a white supremacy forum, determine if this is speech contains hate (eg. bias/racisim/prejudice...) towards a certain group of people.
### Possible classes: ['this is hate', 'this is noHate']

### Define possible classes
POSSIBLE_CLASSES = ['this is hate', 'this is noHate']

### Each criterion should be simple, clear, and concise.
### Start each criterion with 'This is'
### Decision Tree Logic:

"""

def get_hate_tree():
    # Step 1: define criterions of the decision tree.
    criterions = {
        'oi_words': 'The speech includes offensive words about minorities',
        'identity_words': 'The sentence includes mention of identities',
        'swear_words': 'The sentence includes swear words',
        'negative_words':'The sentence includes negative words',
    }

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 'oi_words',
        'oi_words': {'yes': 'this is hate', 'no': 'identity_words'},
        'identity_words': {'yes': 'swear_words', 'no': 'this is noHate'},
        'swear_words': {'yes': 'this is hate', 'no': 'negative_words'},
        'negative_words': {'yes': 'this is hate', 'no': 'this is noHate'},
    }

    return criterions, tree

prompt_social_bias_offensive = """
### Task: Offensive classification
### Description: Given a post, determine if this post is offensive.
### Possible classes: ['Yes','Maybe', 'No']

### Define possible classes
POSSIBLE_CLASSES = ['Yes','Maybe', 'No']

### Each criterion should be simple, clear, and concise.
### Start each criterion with 'This is'
### Generate at least 5 criterions!
### Decision Tree Logic:
"""

def get_sbic_tree():
    # Step 1: define criterions of the decision tree.
    criterions = {
        'racist_words': 'The post includes racist words',
        'sexist_words': 'The post includes sexist words',
        'swear_words': 'The post includes swear words',
        'identity_words': 'The post includes mention of identities',
        'negative_words':'The post include negative words',
    }

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 'racist_words',
        'racist_words': {'yes': 'Yes', 'no': 'sexist_words'},
        'sexist_words': {'yes': 'Yes', 'no': 'swear_words'},
        'swear_words': {'yes': 'Yes', 'no': 'identity_words'},
        'identity_words': {'yes': 'negative_words', 'no': 'Maybe'},
        'negative_words': {'yes': 'Maybe', 'no': 'No'}
    }

    return criterions, tree

@backoff.on_exception(backoff.expo, openai.error.RateLimitError)
def completions_with_backoff(**kwargs):
    return openai.ChatCompletion.create(**kwargs)

model_name = 'gpt-3.5-turbo'
content = prompt_head + prompt_cola
response = completions_with_backoff(
    model=model_name, 
    messages=[
        {"role": "user","content": content}
        ],
    )
response = response["choices"][0]["message"]["content"]
print(response)

research/nlu/human_generated_tree.py
luohongyin/LangCode

import torch
from transformers import RobertaTokenizer, RobertaModel, RobertaForSequenceClassification
from transformers import DebertaTokenizer, DebertaModel, DebertaForSequenceClassification
from sklearn.metrics.pairwise import cosine_similarity
from datasets import load_dataset
from tqdm import tqdm
import json
# Load pre-trained RoBERTa model and tokenizer
device = 'cuda'
# Load Roberta:
model = RobertaForSequenceClassification.from_pretrained("luohy/ESP-roberta-large")
tokenizer = RobertaTokenizer.from_pretrained("roberta-large")

# Load Deberta:
model = DebertaForSequenceClassification.from_pretrained("luohy/ESP-deberta-large")
tokenizer = DebertaTokenizer.from_pretrained("microsoft/deberta-large")

model.eval()  # Set to evaluation mode
model = model.to(device)
softmax = torch.nn.Softmax(dim=0)

# General Functions that are used for entailment model:
def get_entailment_score(hypothesis, model, tokenizer):
    with torch.no_grad():
        proposition = f'{hypothesis} is entailed by .'
        inputs = tokenizer(proposition,  return_tensors="pt", truncation=True, padding=True, max_length=128).to(device)
        outputs = model(**inputs)['logits'][0]
        ent = outputs[0]
        contra = outputs[2]
        return ent, contra
    
def entailment(hypothesis, premise, model, tokenizer, reduce=(0,0), return_val=False):
    with torch.no_grad():
        proposition = f'{hypothesis} is entailed by {premise}.'
        inputs = tokenizer(proposition,  return_tensors="pt", truncation=True, padding=True, max_length=128).to(device)
        outputs = model(**inputs)['logits'][0]
        ent_label = int((outputs[0]-reduce[0]) > (outputs[2]-reduce[1]))
        if ent_label == 1:
            if return_val:
                val = outputs[0] - reduce[0]
                return 'yes', val
            return 'yes'
        else:
            if return_val:
                val = outputs[0] - reduce[0]
                return 'no', val
            return 'no'
            
def tree_predict(sentence, criterions, tree, model, tokenizer, score_list, return_val=False):
    node = tree['root']
    score_tuple = score_list[node]
    tot_num = 0
    tot_val = 0.0
    while node not in POSSIBLE_CLASSES:
        score_tuple = score_list[node]
        if not return_val:
            ent_label = entailment(criterions[node], sentence, model, tokenizer, score_tuple, return_val)
        else:  
            ent_label, val = entailment(criterions[node], sentence, model, tokenizer, score_tuple, return_val)
            tot_num += 1
            tot_val += val
        node = tree[node][ent_label]
    if return_val:
        return node, tot_val/tot_num
    return node

# SST2 decision tree and possible classes
POSSIBLE_CLASSES = ["positive", 'negative']
def get_decision_tree_sst2(which_tree='gpt4'):
    if which_tree == 'human':
        # Human crafted tree
        # Step 1: define criterions of the decision tree.
        criterions = {
            'is_interesting':'This movie is interesting',
            'is_good_script':'The movie has a good script',
            'is_good_character':'The characters are awsome',
            'is_wise': 'This movie is wise'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'is_interesting',
            'is_interesting': {'yes': 'is_good_script', 'no': 'is_wise'},
            'is_good_script': {'yes': 'positive', 'no': 'is_good_character'},
            'is_good_character': {'yes': 'positive', 'no': 'negative'},
            'is_wise': {'yes': 'positive', 'no': 'negative'}
        }
    else:
        # Automatic Tree
        # Step 1: define criterions of the decision tree
        criterions = {
            'positive_adjectives': 'The review uses positive adjectives',
            'negative_adjectives': 'The review uses negative adjectives',
            'positive_director_mention': 'The review mentions the director\'s name positively',
            'negative_cast_comments': 'The review comments negatively on the cast'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'positive_adjectives',
            'positive_adjectives': {'yes': 'positive', 'no': 'negative_adjectives'},
            'negative_adjectives': {'yes': 'negative', 'no': 'positive_director_mention'},
            'positive_director_mention': {'yes': 'positive', 'no': 'negative_cast_comments'},
            'negative_cast_comments': {'yes': 'negative', 'no': 'positive'}
        }
    all_scores = {}
    for key in criterions.keys():
        sentence = criterions[key]
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores[key] = score
    return criterions, tree, all_scores

    

def get_sst2_tree_score():
    dataset = load_dataset('sst2')
    val_dataset = dataset["validation"]
    criterions, tree, all_score = get_decision_tree_sst2()
    tot = 0
    acc = 0
    for item in tqdm(val_dataset):
        tot += 1
        sentence = item['sentence']
        
        label, val = tree_predict(sentence, criterions, tree, model, tokenizer, all_score, True)
        print(sentence)
        print(POSSIBLE_CLASSES[item['label']])
        print(label)
        if label == POSSIBLE_CLASSES[1-item['label']]:
            acc += 1
    print(acc/tot)


# COLA decision tree and testing
POSSIBLE_CLASSES = ["acceptable", 'unacceptable']
def get_decision_tree_cola(which_tree='gpt4'):
    if which_tree == 'gpt4':
        # GPT 4
        # Step 1: define criterions of the decision tree.
        criterions = {
            0:'This sentence is grammatically correct',
            1:'The sentence does not contain any spelling errors',
            2:'The sentence uses punctuation correctly',
            3:'This sentence is semantically clear'
        }

        # Step 2: define the Decision Tree for classification
        tree = {
            'root': 0,
            0: {'yes': 1, 'no': 3},
            1: {'yes': 'acceptable', 'no': 2},
            2: {'yes': 'acceptable', 'no': 'unacceptable'},
            3: {'yes': 2, 'no': 'unacceptable'}
        }
    elif which_tree == 'gpt3.5'
        # GPT3.5
        # Step 1: define criterions of the decision tree.
        criterions = {
            'is_grammatically_correct': 'The sentence is grammatically correct',
            'is_clear': 'The sentence is clear',
            'has_minor_errors': 'The sentence has minor errors',
            'has_major_errors': 'The sentence has major errors'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'is_grammatically_correct',
            'is_grammatically_correct': {'yes': 'is_clear', 'no': 'has_minor_errors'},
            'is_clear': {'yes': 'acceptable', 'no': 'has_major_errors'},
            'has_minor_errors': {'yes': 'acceptable', 'no': 'unacceptable'},
            'has_major_errors': {'yes': 'unacceptable', 'no': 'acceptable'}
        }
    elif which_tree == 'human':
        # Step 1: define criterions of the decision tree.
        criterions = {
            'has_subject': 'The sentence has subject',
            'has_verb': 'The sentence has verb',
            'punctuation_correct':'The sentence has proper punctuations',
            'pronouns_reference_match':'This sentence has matched pronouns and references',
            'subject_verb_match':'The sentence has matched subject and verb',
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'has_subject',
            'has_subject': {'yes': 'has_verb', 'no': 'unacceptable'},
            'has_verb': {'yes': 'punctuation_correct', 'no': 'unacceptable'},
            'punctuation_correct': {'yes': 'pronouns_reference_match', 'no': 'unacceptable'},
            'pronouns_reference_match': {'yes': 'subject_verb_match', 'no': 'unacceptable'},
            'subject_verb_match':{'yes': 'acceptable', 'no': 'unacceptable'}
        }
    all_scores = {}
    for key in criterions.keys():
        sentence = criterions[key]
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores[key] = score
    return criterions, tree, all_scores

# baseline of cola using entailment model
def get_decision_cola_baseline():
    # Step 1: define criterions of the decision tree.
    criterions = {
        0:'This sentence is acceptable',
    }

    # Step 2: define the Decision Tree for classification
    tree = {
        'root': 0,
        0: {'yes': 'acceptable', 'no':'unacceptable'},
    }

    all_scores = {}
    for key in criterions.keys():
        sentence = criterions[key]
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores[key] = score
    return criterions, tree, all_scores

def get_cola_tree_score():
    dataset = load_dataset('glue', 'cola')
    val_dataset = dataset["validation"]
    criterions, tree, all_score = get_decision_tree_cola()
    tot = 0
    acc = 0
    for item in tqdm(val_dataset):
        tot += 1
        sentence = item['sentence']
        
        label, val = tree_predict(sentence, criterions, tree, model, tokenizer, all_score, True)
        if label == POSSIBLE_CLASSES[item['label']]:
            acc += 1
    print(acc/tot)



# Emotion Classification tree and testing
POSSIBLE_CLASSES = ['I feel sad', 'I feel happy', 'I feel love', 'I feel angry', 'I feel afraid', 'I feel surprised']
def get_emotion_tree(which_tree='gpt4'):
    if which_tree == 'gpt4':
        # GPT4
        # Step 1: define criterions of the decision tree.
            criterions = {
                'is_positive':'This feeling is positive',
                'is_sad':'I feel sad',
                'is_angry':'I feel angry',
                'is_afraid':'I feel afraid',
                'is_happy':'I feel happy',
                'is_love':'I feel love',
                'is_surprised':'I feel surprised'
            }

            # Step 2: define the balanced decision tree for this classification task
            tree = {
                'root': 'is_positive',
                'is_positive': {'yes': 'is_happy', 'no': 'is_sad'},
                'is_happy': {'yes': 'I feel happy', 'no': 'is_love'},
                'is_love': {'yes': 'I feel love', 'no': 'is_surprised'},
                'is_surprised': {'yes': 'I feel surprised', 'no': 'I feel happy'},
                'is_sad': {'yes': 'I feel sad', 'no': 'is_angry'},
                'is_angry': {'yes': 'I feel angry', 'no': 'is_afraid'},
                'is_afraid': {'yes': 'I feel afraid', 'no': 'I feel sad'}
            }
    elif which_tree == 'gpt3.5':
        # GPT3.5
        # Step 1: define criterions of the decision tree.
        criterions = {
            'is_positive':'The sentence expresses positive emotion',
            'is_love':'The sentence expresses love',
            'is_happy':'The sentence expresses happiness',
            'is_negative':'The sentence expresses negative emotion',
            'is_anger':'The sentence expresses anger',
            'is_sad':'The sentence expresses sadness'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'is_positive',
            'is_positive': {'yes': 'is_love', 'no': 'is_negative'},
            'is_love': {'yes': 'I feel love', 'no': 'is_happy'},
            'is_happy': {'yes': 'I feel happy', 'no': 'I feel surprised'},
            'is_negative': {'yes': 'is_anger', 'no': 'is_sad'},
            'is_anger': {'yes': 'I feel angry', 'no': 'is_sad'},
            'is_sad': {'yes': 'I feel sad', 'no': 'I feel afraid'}
        }
    elif which_tree == 'human':
        # Human Prompt
        # Step 1: define criterions of the decision tree.
        criterions = {
            'positive_words': 'The sentence includes positive words',
            'intensifiers': 'The sentence includes intensifiers, adverb or exclamation point',
            'unexpected': 'The sentence includes words that are synonyms or antonyms of unexpected',
            'fear': 'The sentence includes words that are synonyms or antonyms of fear',
            'upset': 'The sentence includes words that are synonyms or antonyms of upset'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'positive_words',
            'positive_words': {'yes': 'intensifiers', 'no': 'fear'},
            'intensifiers': {'yes': 'unexpected', 'no': 'I feel happy'},
            'unexpected': {'yes': 'I feel surprised', 'no': 'I feel love'},
            'fear': {'yes': 'I feel afraid', 'no': 'upset'},
            'upset': {'yes': 'I feel sad', 'no': 'I feel angry'}
        }
    all_scores = {}
    for key in criterions.keys():
        sentence = criterions[key]
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores[key] = score
    return criterions, tree, all_scores

# Another side-note: generate a tree for each emotion and get the score for each class; choose the highest class label as the result
def get_happy_tree():
    # Step 1: define criterions of the decision tree.
    criterions = [
        'I am feeling good',
        'I have a positive attitude',
        'I am smiling',
        'I am excited'
    ]

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 0,
        0: {'yes': 1, 'no': 2},
        1: {'yes': 'Yes', 'no': 3},
        2: {'yes': 3, 'no': 'No'},
        3: {'yes': 'Yes', 'no': 'No'}
    }

    all_scores = []
    for sentence in criterions:
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores.append(score)
        del score
    return criterions, tree, all_scores
def get_sad_tree():
    # Step 1: define criterions of the decision tree.
    criterions = [
        'I am feeling depressed',
        'I am not feeling good',
        'I am missing someone',
        'I feel like crying'
    ]

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 0,
        0: {'yes': 1, 'no': 3},
        1: {'yes': 'Yes', 'no': 2},
        2: {'yes': 'Yes', 'no': 'No'},
        3: {'yes': 'Yes', 'no': 'No'}
    }
    

    all_scores = []
    for sentence in criterions:
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores.append(score)
        del score
    return criterions, tree, all_scores

def get_love_tree():
    # Step 1: define criterions of the decision tree.
    criterions = [
        'I mention love in this sentence',
        'I talk about affection in this sentence',
        'I express happiness in this sentence',
        'I mention someone special in this sentence'
    ]

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 0,
        0: {'yes': 1, 'no': 2},
        1: {'yes': 'Yes', 'no': 3},
        2: {'yes': 'Yes', 'no': 'No'},
        3: {'yes': 'Yes', 'no': 'No'}
    }

    all_scores = []
    for sentence in criterions:
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores.append(score)
        del score
    return criterions, tree, all_scores

def get_anger_tree():
    # Step 1: define criterions of the decision tree.
    criterions = [
        'I am irritated',
        'I am frustrated',
        'I am outraged',
        'I am moody'
    ]

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 0,
        0: {'yes': 1, 'no': 2},
        1: {'yes': 'Yes', 'no': 3},
        2: {'yes': 'Yes', 'no': 'No'},
        3: {'yes': 'Yes', 'no': 'No'}
    }

    all_scores = []
    for sentence in criterions:
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores.append(score)
        del score
    return criterions, tree, all_scores

def get_fear_tree():
    # Step 1: define criterions of the decision tree.
    criterions = [
        'I feel scared',
        'I am frightened',
        'I feel terrified',
        'There is danger'
    ]
    
    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 0,
        0: {'yes': 1, 'no': 2},
        1: {'yes': 'Yes', 'no': 'No'},
        2: {'yes': 3, 'no': 'No'},
        3: {'yes': 'Yes', 'no': 'No'}
    }

    all_scores = []
    for sentence in criterions:
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores.append(score)
        del score
    return criterions, tree, all_scores

def get_surprise_tree():
    # Step 1: define criterions of the decision tree.
    criterions = [
        'I did not expect this',
        'This is unusual',
        'I am startled',
        'This is a surprise'
    ]

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 0,
        0: {'yes': 1, 'no': 2},
        1: {'yes': 'Yes', 'no': 3},
        2: {'yes': 'Yes', 'no': 'No'},
        3: {'yes': 'Yes', 'no': 'No'}
    }
    
    all_scores = []
    for sentence in criterions:
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores.append(score)
    return criterions, tree, all_scores

# test emotion by using only one tree
def get_full_tree_emotion():
    dataset = load_dataset('emotion')
    val_dataset = dataset["validation"]
    criterions, tree, all_score = get_emotion_tree()
    tot = 0
    acc = 0
    for item in tqdm(val_dataset):
        tot += 1
        sentence = item['text']
        
        label, val = tree_predict(sentence, criterions, tree, model, tokenizer, all_score, True)
        if label == POSSIBLE_CLASSES[item['label']]:
            acc += 1
    print(acc/tot)

# Use multiple trees for each class and choose the highest-scoring class
def get_all_tree_emotion():
    EMOTION_CLASSES = ['sadness', 'joy', 'love', 'anger', 'fear', 'surprise']
    criterions_sad, tree_sad, score_sad = get_sad_tree()
    criterions_happy, tree_happy, score_happy = get_happy_tree()
    criterions_love, tree_love, score_love = get_love_tree()
    criterions_anger, tree_anger, score_anger = get_anger_tree()
    criterions_fear, tree_fear, score_fear = get_fear_tree()
    criterions_surprise, tree_surprise, score_surprise = get_surprise_tree()
    all_criterion = [criterions_sad, criterions_happy,criterions_love, criterions_anger,criterions_fear,criterions_surprise]
    all_tree = [tree_sad, tree_happy, tree_love, tree_anger,tree_fear, tree_surprise]
    all_score = [score_sad, score_happy, score_love, score_anger, score_fear, score_surprise]
    tot = 0
    acc = 0
    for item in tqdm(val_dataset):
        tot += 1
        sentence = item['text']
        correct_emotions = []
        emotion_val = []
        for criterions, tree, score, emotion in zip(all_criterion, all_tree,all_score, EMOTION_CLASSES):
            label, val = tree_predict(sentence, criterions, tree, model, tokenizer, score, True)
            if label == 'Yes':
                correct_emotions.append(emotion)
            emotion_val.append(val)
   
        if len(correct_emotions) == 1:
            label = correct_emotions[0]
        elif len(correct_emotions) > 1:
            init = -100
            for num, emotion in enumerate(EMOTION_CLASSES):
                if emotion in correct_emotions and emotion_val[num] > init:
                    label = emotion
                    init = emotion_val[num]
        else:
            index = torch.argmax(torch.tensor(emotion_val))
            label = EMOTION_CLASSES[index]
       
        if label == EMOTION_CLASSES[item['label']]:
            acc += 1
    print(acc/tot)
    with open('./results/emotion_tree', 'w') as f:
        f.write(str(acc/tot))

# Tree and testing for amazon review
POSSIBLE_CLASSES = ['1', '2', '3', '4', '5']
def get_amazon_review_tree(which_tree='gpt4'):
    if which_tree == 'gpt4':
        # Step 1: define criterions of the decision tree.

        criterions = {
            'is_very_satisified':'This review is very satisfied',
            'is_superior_quality':'The product is of superior quality',
            'is_moderately_satisfied':'This review is moderately satisfied',
            'is_critic':'The review contains criticism',
            'is_satisfied':'The review is satisfied',
            'is_improvable':'The product has features to improve',
            'is_unsatisfied':'The review is unsatisfied',
            'is_extremely_negative':'The review contains extremely negative words'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'is_very_satisified',
            'is_very_satisified': {'yes': 'is_superior_quality', 'no': 'is_moderately_satisfied'},
            'is_superior_quality': {'yes': '5', 'no': '4'},
            'is_moderately_satisfied': {'yes': 'is_critic', 'no': 'is_satisfied'},
            'is_critic': {'yes': '3', 'no': '4'},
            'is_satisfied': {'yes': 'is_improvable', 'no': '2'},
            'is_improvable': {'yes': '3', 'no': '2'},
            'is_unsatisfied': {'yes': 'is_extremely_negative', 'no': '2'},
            'is_extremely_negative': {'yes': '1', 'no': '2'}
        }
    elif which_tree == 'human':
        # Step 1: define criterions of the decision tree.
        criterions = {
            'extreme_word': 'The review includes intensifiers or exclamation point',
            'positive': 'The review includes positive words or sentences',
            'both_positive_negative':'The review includes both positive and negative sentences',
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'extreme_word',
            'extreme_word': {'yes': 'positive', 'no': 'both_positive_negative'},
            'positive': {'yes': '5', 'no': '1'},
            'both_positive_negative': {'yes': '3', 'no': 'positive'},
            'positive': {'yes': '4', 'no': '2'}
        }
    elif which_tree == 'gpt3.5':
        # GPT3.5
        # Step 1: define criterions of the decision tree.
        criterions = {
            'is_negative_sentiment':'This review has negative sentiment',
            'is_short_review':'This review is short',
            'has_positive_keywords':'This review has positive keywords',
            'has_negative_keywords':'This review has negative keywords'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'is_negative_sentiment',
            'is_negative_sentiment': {'yes': '1', 'no': 'is_short_review'},
            'is_short_review': {'yes': '3', 'no': 'has_positive_keywords'},
            'has_positive_keywords': {'yes': '5', 'no': 'has_negative_keywords'},
            'has_negative_keywords': {'yes': '2', 'no': '4'}
        }
    all_scores = {}
    for key in criterions.keys():
        sentence = criterions[key]
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores[key] = score
    return criterions, tree, all_scores

def get_full_tree_amazon():

    dataset = load_dataset('amazon_reviews_multi', 'en')
    val_dataset = dataset["validation"]
    criterions, tree, all_score = get_amazon_review_tree()
    tot = 0
    acc = 0
    for item in tqdm(val_dataset):
        tot += 1
        sentence = item['review_title'] + item['review_body']
        label, val = tree_predict(sentence, criterions, tree, model, tokenizer, all_score, True)
        if label == POSSIBLE_CLASSES[item['stars']-1]:
            acc += 1
    print(acc/tot)


POSSIBLE_CLASSES = ['this is hate', 'this is noHate']
def get_hate_tree(which_tree='gpt4'):
    if which_tree=='gpt4':
        # GPT4
        # # Step 1: define criterions of the decision tree.
        criterions = {
            'is_aggressive':'This speech contains aggressive language',
            'is_targeting_group':'This speech targets a specific group',
            'is_explicit_discrimination':'This speech contains explicit discrimination',
            'has_covert_prejudice': 'This speech has covert prejudice or bias'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'is_aggressive',
            'is_aggressive': {'yes': 'is_targeting_group', 'no': 'has_covert_prejudice'},
            'is_targeting_group': {'yes': 'this is hate', 'no': 'is_explicit_discrimination'},
            'is_explicit_discrimination': {'yes': 'this is hate', 'no': 'this is noHate'},
            'has_covert_prejudice': {'yes': 'this is hate', 'no': 'this is noHate'}
        }
    elif which_tree == 'human':
        # Human Tree
        # Step 1: define criterions of the decision tree.
        criterions = {
            'oi_words': 'The speech includes offensive words about minorities',
            'identity_words': 'The sentence includes mention of identities',
            'swear_words': 'The sentence includes swear words',
            'negative_words':'The sentence includes negative words',
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'oi_words',
            'oi_words': {'yes': 'this is hate', 'no': 'identity_words'},
            'identity_words': {'yes': 'swear_words', 'no': 'this is noHate'},
            'swear_words': {'yes': 'this is hate', 'no': 'negative_words'},
            'negative_words': {'yes': 'this is hate', 'no': 'this is noHate'},
        }
    elif which_tree=='gpt3.5':
        # GPT3.5
        # Step 1: define criterions of the decision tree.
        criterions = {
            'is_explicit_hate':'This speech contains explicit hate speech',
            'is_derogatory_language':'This speech contains derogatory language'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'is_explicit_hate',
            'is_explicit_hate': {'yes': 'this is hate', 'no': 'is_derogatory_language'},
            'is_derogatory_language': {'yes': 'this is hate', 'no': 'this is noHate'}
        }
    all_scores = {}
    for key in criterions.keys():
        sentence = criterions[key]
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores[key] = score
    return criterions, tree, all_scores


def get_hate_dataset():
    import os
    import csv
    import json
    # Paths to CSV file and test set folder
    csv_file_path = '../hate-speech-dataset/annotations_metadata.csv'
    test_folder_path = '../hate-speech-dataset/sampled_test/'

    # Read the CSV file and create a dictionary of file IDs and labels
    file_labels = {}
    with open(csv_file_path, 'r') as csv_file:
        csv_reader = csv.DictReader(csv_file)
        for row in csv_reader:
            print(row)
            file_id = row['file_id']
            label = row['label']
            file_labels[file_id] = label

    # Iterate over files in the test set folder
    data = []
    for filename in os.listdir(test_folder_path):
        if filename.endswith('.txt'):
            file_id = filename.split('.')[0]
            file_path = os.path.join(test_folder_path, filename)
            
            # Check if the file ID has a corresponding label
            if file_id in file_labels:
                label = file_labels[file_id]
                with open(file_path, 'r') as txt_file:
                    sentence = txt_file.read().strip()
                    print(f'File: {filename}\nLabel: {label}\nSentence: {sentence}\n')
                    new_dict = {}
                    new_dict['file'] = filename
                    new_dict['label'] = label
                    new_dict['sentence'] = sentence
                    data.append(new_dict)
            else:
                print(f'No label found for File: {filename}\n')
    with open('../hate-speech-dataset/test_sample.json', 'w') as f:
        json.dump(data, f)

def get_full_tree_hate_speech():
    with open('../hate-speech-dataset/test_sample.json', 'r') as f:
        data = json.load(f)
    criterions, tree, all_score = get_hate_tree()
    tot = 0
    acc = 0
    for item in tqdm(data):
        tot += 1
        sentence = item['sentence']
        
        label, val = tree_predict(sentence, criterions, tree, model, tokenizer, all_score, True)
        
        if label == 'this is ' + item['label']:
            acc += 1
    print(acc/tot)
# get_full_tree_hate_speech()


def get_baseline_hate_speech():
    with open('../hate-speech-dataset/test_sample.json', 'r') as f:
        data = json.load(f)
    tot = 0
    acc = 0
    hate_tuple =  get_entailment_score('this is hate ', model, tokenizer)
    nonhate_tuple = get_entailment_score('this is not hate ', model, tokenizer)
    for item in tqdm(data):
        tot += 1
        sentence = item['sentence']
        
        hatelabel, hateval = entailment('this is hate ', sentence, model, tokenizer, hate_tuple, True)
        nohatelabel, nohateval = entailment('this is not hate ', sentence, model, tokenizer, nonhate_tuple, True)
        if hateval > nohateval:
            label = 'hate'
        else:
            label = 'noHate'
        
        if label == item['label']:
            acc += 1
    print(acc/tot)



POSSIBLE_CLASSES = ['Yes','Maybe', 'No']
def get_bias_offense_tree(which_tree='gpt4', use_model='deberta'):
    if which_tree=='gpt4':
        if use_model=='roberta':
            # Step 1: define criterions of the decision tree.
            criterions = {
                'is_hate_speech':'This post contains hate speech',
                'is_disrespectful':'This post is disrespectful',
                'has_strong_language':'This post contains strong language',
            }

            # Step 2: define the balanced decision tree for this classification task
            tree = {
                'root': 'is_hate_speech',
                'is_hate_speech': {'yes': 'Yes', 'no': 'is_disrespectful'},
                'is_disrespectful': {'yes': 'Maybe', 'no': 'has_strong_language'},
                'has_strong_language': {'yes': 'Maybe', 'no': 'No'},
            }
        else:
            # Step 1: define criterions of the decision tree.
            criterions = {
                'is_politically_incorrect':'This is politically incorrect',
                'includes_offensive_words':'This includes offensive words',
                'is_discriminating':'This is discriminative'
            }

            # Step 2: define the balanced decision tree for this task.
            tree = {
                'root': 'is_politically_incorrect',
                'is_politically_incorrect': {'yes': 'includes_offensive_words', 'no': 'is_discriminating'},
                'includes_offensive_words': {'yes': 'Yes', 'no': 'Maybe'},
                'is_discriminating': {'yes': 'Maybe', 'no': 'No'}
            }
    elif which_tree == 'human'
        # Human Tree
        # Step 1: define criterions of the decision tree.
        criterions = {
            'racist_words': 'The post includes racist words',
            'sexist_words': 'The post includes sexist words',
            'swear_words': 'The post includes swear words',
            'identity_words': 'The post includes mention of identities',
            'negative_words':'The post include negative words',
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'racist_words',
            'racist_words': {'yes': 'Yes', 'no': 'sexist_words'},
            'sexist_words': {'yes': 'Yes', 'no': 'swear_words'},
            'swear_words': {'yes': 'Yes', 'no': 'identity_words'},
            'identity_words': {'yes': 'negative_words', 'no': 'Maybe'},
            'negative_words': {'yes': 'Maybe', 'no': 'No'}
        }
    elif which_tree=='gpt3.5':
        # GPT3.5
        # Step 1: define criterions of the decision tree.
        criterions = {
            'contains_offensive_language':'This post contains offensive language',
            'intention_to_offend':'The intention of the post is to offend',
            'tone_neutral':'The tone of the post is neutral',
            'tone_ambiguous':'The tone of the post is ambiguous'
        }

        # Step 2: define the balanced decision tree for this classification task
        tree = {
            'root': 'contains_offensive_language',
            'contains_offensive_language': {'yes': 'intention_to_offend', 'no': 'tone_neutral'},
            'intention_to_offend': {'yes': 'Yes', 'no': 'Maybe'},
            'tone_neutral': {'yes': 'No', 'no': 'tone_ambiguous'},
            'tone_ambiguous': {'yes': 'Maybe', 'no': 'No'}
        }
    all_scores = {}
    for key in criterions.keys():
        sentence = criterions[key]
        score = get_entailment_score(sentence, model, tokenizer)
        all_scores[key] = score
    return criterions, tree, all_scores


def get_full_tree_social_speech():
    dataset = load_dataset('social_bias_frames')
    val_dataset = dataset["validation"]
    criterions, tree, all_score = get_bias_offense_tree()
    tot = 0
    acc = 0
    for item in tqdm(val_dataset):
        tot += 1
        sentence = item['post']

        label, val = tree_predict(sentence, criterions, tree, model, tokenizer, all_score, True)
        truth = item['offensiveYN']
        
        if truth == '1.0':
            truth = 'Yes'
        elif truth == '0.5':
            truth = 'Maybe'
        else:
            truth = 'No'

        if label == truth:
            acc += 1

        print(acc/tot)
            
        
       
    print(acc/tot)

def get_full_tree_social_speech_baseline():
    dataset = load_dataset('social_bias_frames')
    val_dataset = dataset["validation"]
    tot = 0
    acc = 0
    hate_tuple =  get_entailment_score('this is offensive ', model, tokenizer)
    nonhate_tuple = get_entailment_score('this is not offensive ', model, tokenizer)
    maybehate_tuple = get_entailment_score('this is maybe offensive ', model, tokenizer)
    for item in tqdm(val_dataset):
        tot += 1
        sentence = item['post']
        
        hatelabel, hateval = entailment('this is offensive ', sentence, model, tokenizer, hate_tuple, True)
        nohatelabel, nohateval = entailment('this is not offensive ', sentence, model, tokenizer, nonhate_tuple, True)
        maybehatelabel, maybehateval = entailment('this is maybe offensive ', sentence, model, tokenizer, maybehate_tuple, True)
        if hateval > nohateval and hateval > maybehateval:
            label = 'Yes'
        elif nohateval > hateval and nohateval > maybehateval:
            label = 'No'
        else:
            label = 'Maybe'
        truth = item['offensiveYN']
        if truth == '1.0':
            truth = 'Yes'
        elif truth == '0.5':
            truth = 'Maybe'
        else:
            truth = 'No'
        if label == truth:
            acc += 1
        
    print(acc/tot)
    

research/nlu/tree_NLEP/tree_NLEP_execution.py
luohongyin/LangCode
import openai
import backoff
openai.api_key_path = './api_key.txt'
prompt_head = """Write a Python function that constructs a decision tree according to the given examples that can generate the correct label of the given classification task.

### Available APIs(shared for all tasks):

\"\"\"Returns whether the hypothesis in entailed by the premise.\"\"\"
def entailment(hypothesis, premise, model, tokenizer):
    proposition = f'{hypothesis} is entailed by {premise}.'
    inputs = tokenizer(proposition,  return_tensors="pt", truncation=True, padding=True, max_length=128)
    outputs = model(**inputs)['logits'][0]
    ent_label = int(outputs[0] > outputs[2])
    if ent_label == 1:
        return 'yes'
    else:
        return 'no'

\"\"\"Use the constructed decision tree to predict the label of the sentence.\"\"\"
def tree_predict(sentence, criterions, tree, model, tokenizer):
    node = tree['root']
    while node not in POSSIBLE_CLASSES:
        ent_label = entailment(criterions[node], sentence, model, tokenizer)
        node = tree[node][ent_label]
    return node

### Task: Movie review classification
### Description: Determine if a movie review expresses positive attitude or negative attitude.
### Possible classes: [positive, negative]
### Examples:
- contains no wit, only labored gags
    - [The movie is wise|The movie is not wise|1], [the story is fun|the story is not boring|1], [the review is positive|the review is negative|1]
- that loves its characters and communicates something rather beautiful about human nature
    - [The characters are lovely|The characters are awful|0], [the script is touching|the script is dry|0], [the review is positive|the review is negative|0]
- on the worst revenge-of-the-nerds clichés the filmmakers could dredge up
    - [The movie is novel|The movie is mostly platitudes|1], [the review is negative|1]
- are more deeply thought through than in most right-thinking films
    - [The takeaway of the movie is profound|The idea of the movie is shallow|0], [the review is positive|the review is negative|0]

### Define possible classes
POSSIBLE_CLASSES = ['positive', 'negative']

### Decision Tree Logic:
Start by assessing whether the movie is interesting.
- If uninteresting:
    Probe for depth by determining if the movie is wise.
    - If wise, label as a positive review.
    - Otherwise, label as a negative review.
- If interesting:
    Examine the quality of the script.
    - If the script is commendable, label as a positive review.
    - If not a good script
        Evaluate the portrayal of characters.
        - If the characters are well-depicted, label the review as positive.
        - If characters are not good, label the review negative.

### Python program: 
\"""Decision tree for this task\"""
def get_decision_tree(sentence, model, tokenizer):
    # Step 1: define criterions of the decision tree.
    criterions = {
        'is_interesting':'This movie is interesting',
        'is_good_script':'The movie has a good script',
        'is_good_character':'The characters are awsome',
        'is_wise': 'This movie is wise'
    }

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 'is_interesting',
        'is_interesting': {'yes': 'is_good_script', 'no': 'is_wise'},
        'is_good_script': {'yes': 'positive', 'no': 'is_good_character'},
        'is_good_character': {'yes': 'positive', 'no': 'negative'},
        'is_wise': {'yes': 'positive', 'no': 'negative'}
    }

    return criterions, tree

"""

prompt_head2 = """
Write a Python function that constructs a decision tree according to the given examples that can generate the correct label of the given classification task.

### Available APIs(shared for all tasks):

\"\"\"Returns whether the hypothesis in entailed by the premise.\"\"\"
def entailment(hypothesis, premise, model, tokenizer):
    proposition = f'{hypothesis} is entailed by {premise}.'
    inputs = tokenizer(proposition,  return_tensors="pt", truncation=True, padding=True, max_length=128)
    outputs = model(**inputs)['logits'][0]
    ent_label = int(outputs[0] > outputs[2])
    if ent_label == 1:
        return 'yes'
    else:
        return 'no'

\"\"\"Use the constructed decision tree to predict the label of the sentence.\"\"\"
def tree_predict(sentence, criterions, tree, model, tokenizer):
    node = tree['root']
    while node not in POSSIBLE_CLASSES:
        ent_label = entailment(criterions[node], sentence, model, tokenizer)
        node = tree[node][ent_label]
    return node

### Task: Grammar correctness classification
### Possible classes: ['accpetable', 'unacceptable']

### Define possible classes
POSSIBLE_CLASSES = ['accpetable', 'unacceptable']

### Decision Tree Logic:
- If verbs are not correctly constructed, the sentence is immediately labeled as unacceptable.
- If verbs are correct:
    The tree then checks if the sentence has correct punctuation
    - If incorrect, label the sentence as unacceptable
    - If correct:
        The next criterion to be assessed is the subject-verb agreement.
        - If subject and verb disagree, label the sentence as unacceptable.
        - If they agree, check for sentence fragments.
            - If the sentence is a fragment, label it as unacceptable.
            - If it is not a sentence fragment, label the sentence as acceptable.

### Python code for the decision tree:

```python
def get_decision_tree(sentence, model, tokenizer):
    # Step 1: define criterions of the decision tree
    criterions = {
        'correct_verbs': 'The verbs are correctly constructed in the sentence',
        'correct_punctuation': 'The sentence is punctuated correctly',
        'subject_verb_agreement': 'The subject and verb agree in the sentence',
        'no_sentence_fragments': 'The sentence is not a fragment',
    }

    # Step 2: define the balanced decision tree for this classification task
    tree = {
        'root': 'correct_verbs',
        'correct_verbs': {'yes': 'correct_punctuation', 'no': 'unacceptable'},
        'correct_punctuation': {'yes': 'subject_verb_agreement', 'no': 'unacceptable'},
        'subject_verb_agreement': {'yes': 'no_sentence_fragments', 'no': 'unacceptable'},
        'no_sentence_fragments': {'yes': 'acceptable', 'no': 'unacceptable'}
    }

    return criterions, tree
```

"""

prompt_sst2 = """
### Task: Movie review classification
### Possible classes: [positive, negative]

### Define possible classes
POSSIBLE_CLASSES = [positive, negative]

### Decision Tree Logic:

"""
prompt_cola ="""
### Task: Grammar correctness classification
### Possible classes: [acceptable, unacceptable]

### Define possible classes
POSSIBLE_CLASSES = ['acceptable', 'unacceptable']

### Decision Tree Logic:

"""

prompt_qnli ="""
### Task: the sentence answers the question classification
### Description: Determine if the sentence is an answer to the question
### Possible classes: [true, false]

### Define possible classes
POSSIBLE_CLASSES = ['true', 'false']

### Python program: 
\"""Decision tree for this task\"""
Try to generate a balanced tree!
def get_decision_tree(sentence, model, tokenizer):

"""

prompt_emotion_separate ="""
### Task: Sadness classification
### Description: Assume that "I" wrote a sentence, determine if "I" am feeling sad.
### Possible classes: ['Yes', 'No']

### Define possible classes
POSSIBLE_CLASSES = ['Yes', 'No']

### Python program: 
\"""Decision tree for this task\"""
Try to generate a balanced tree!
def get_decision_tree(sentence, model, tokenizer):

"""

prompt_emotion ="""
### Task: Emotion classification
### Description: I wrote a sentence about my feeling, determine which emotion am I feeling.
### Possible classes: ['I feel sad', 'I feel happy', 'I feel love', 'I feel angry', 'I feel afraid', 'I feel surprised']

### Define possible classes
POSSIBLE_CLASSES = ['I feel sad', 'I feel happy', 'I feel love', 'I feel angry', 'I feel afraid', 'I feel surprised']

### Decision Tree Logic:

"""
prompt_amazon = """
### Task: Amazon Review Star classification
### Description: Given an amazon review, determine the star(1-5) of this review, where 1 is the worst and 5 is the best.
### Possible classes: ['1', '2', '3', '4', '5']

### Define possible classes
POSSIBLE_CLASSES = ['1', '2', '3', '4', '5']

### Decision Tree Logic:

"""

prompt_agnews = """
### Task: News content classification
### Description: Given a piece of news, determine which category it belongs to.
### Possible classes: ['this is world news', 'this is sport news', 'this is business news', 'this is technology news']


### Define possible classes
POSSIBLE_CLASSES = ['this is world news', 'this is sport news', 'this is business news', 'this is technology news']

### Start each criterion with "This is"
### Each criterion should be general/high-level! 
### Python program: 
\"""Decision tree for this task\"""
Try to generate a balanced tree!
def get_decision_tree(sentence, model, tokenizer):

"""

prompt_agnews_separate = """
### Task: Science news classification
### Description: Given a piece of news, determine if this is a science news.
### Possible classes: ['Yes', 'No']

### Define possible classes
POSSIBLE_CLASSES = ['Yes', 'No']

### Decision Tree Logic:

"""

prompt_hate_speech = """
### Task: Bias/Hateful speech classification
### Description: Given a speech from a white supremacy forum, determine if this is speech contains hate (eg. bias/racisim/prejudice...) towards a certain group of people.
### Possible classes: ['this is hate', 'this is noHate']

### Define possible classes
POSSIBLE_CLASSES = ['this is hate', 'this is noHate']

### Each criterion should be simple, clear, and concise.
### Start each criterion with 'This is'
### Decision Tree Logic:

"""

prompt_social_bias_offensive = """
### Task: Offensive classification
### Description: Given a post, determine if this post is offensive.
### Possible classes: ['Yes','Maybe', 'No']

### Define possible classes
POSSIBLE_CLASSES = ['Yes','Maybe', 'No']

### Each criterion should be simple, clear, and concise.
### Start each criterion with 'This is'
### Generate at least 5 criterions!
### Decision Tree Logic:
"""
prompt_social_bias_offensive_deberta = """
### Task: Offensive classification
### Description: Given a post, determine if this post is potentially offensive to anyone.
### Possible classes: ['this is offensive','this is not offensive', 'this is maybe offensive']

### Define possible classes
POSSIBLE_CLASSES = ['this is offensive','this is not offensive', 'this is maybe offensive']

### Each criterion should be simple, clear, and concise.
### Start each criterion with 'This is'
### Decision Tree Logic:
"""
@backoff.on_exception(backoff.expo, openai.error.RateLimitError)
def completions_with_backoff(**kwargs):
    return openai.ChatCompletion.create(**kwargs)

model_name = 'gpt-4'
# To generate a tree for SST2, use prompt_head2 instead of prompt_head
content = prompt_head + prompt_social_bias_offensive
response = completions_with_backoff(
    model=model_name, 
    messages=[
        {"role": "user","content": content}
        ],
    )
response = response["choices"][0]["message"]["content"]
print(response)
research/nlu/tree_NLEP/tree_NLEP_generation.py
luohongyin/LangCode
import openai
import google.generativeai as palm

import csv

from fastapi import FastAPI, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware

from pydantic import BaseModel
import os
from urllib.parse import urlparse


app = FastAPI()

fs_prompt = open('prompts.txt').read()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class Item(BaseModel):
    instruction: str
    api_key_str: str
    exist_code: str
    platform: str
    model: str


def parse_api_key(api_key_str):
    print(api_key_str)
    if 'api_key =' in api_key_str:
        return api_key_str.split('api_key =')[1].replace('\'', '').strip()
    else:
        return api_key_str


def construct_fspy_prompt(fs_prompt, inst_str, input_txt = 'None', exist_code = 'none'):
    if exist_code == 'none':
        prompt = f'''{fs_prompt}\n\n### Instruction: {inst_str}\n### Input: {input_txt}\n### Python program:'''
    else:
        prompt = f'''{fs_prompt}\n\n### Instruction: {inst_str}\n### Input: {input_txt}\n### Python program:\n```\n{exist_code.strip()}'''
    return prompt


def gpt4_py(ques_str, api_key_str, exist_code, platform, model='gpt-4'):
    api_key = parse_api_key(api_key_str)

    prompt = construct_fspy_prompt(fs_prompt, ques_str, exist_code = exist_code)
    
    if platform == 'gpt':
        openai.api_key = api_key

        gpt4_output = openai.ChatCompletion.create(
            model = model,
            messages = [{'role': 'user', 'content': prompt}],
            temperature = 0.5,
            top_p = 1.0,
            max_tokens = 1024
        )
        gen_txt = gpt4_output['choices'][0]['message']['content'].replace('```python', '```')
    
    elif platform == 'palm':
        palm.configure(api_key = api_key)
        completion = palm.generate_text(
            model = f'models/{model}',
            prompt = prompt,
            temperature = 0.5,
            # The maximum length of the response
            max_output_tokens = 1024,
        )
        gen_txt = completion.result
    
    else:
        return 'Platform not supported.'

    if exist_code != 'none' and gen_txt.startswith('```'):
        gen_txt = gen_txt[3:]

    if exist_code != 'none':
        gen_txt = f'\n{gen_txt}'

    res_str = (
        prompt + gen_txt
    ).split('### Python program:')[-1].strip()

    sections = res_str.split('```')
    if len(sections) > 1:
        ans_str = sections[1].strip()
    else:
        ans_str = sections[0].strip()
    
    return ans_str


@app.post("/process-file/")
async def process_file(file: UploadFile = File(...)):
    # Do something with the file
    content = await file.read()
    content_list = content.decode('utf-8').split('\n')[:3]
    content_str = '\n# '.join(content_list)
    prompt = f"# First three rows of the input file:\n# {content_str}\n# ...\nfile_path = 'cache/{file.filename}' # Please fill\ninput_file = open(file_path)"
    # For demonstration purposes, we'll just return the filename
    return {"filename": file.filename, "message": f"{prompt}"}



@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    ans_str = gpt4_py(
        item.instruction, item.api_key_str, item.exist_code,
        item.platform, item.model
    )

    return {"output": ans_str, "item_id": item_id}
server/nlep_server.py
luohongyin/LangCode
from setuptools import setup, find_packages

setup(
    name='langpy-notebook',
    version='0.4',
    packages=find_packages(),
    install_requires=[
        # Any dependencies your project needs, e.g.
        # 'requests',
        'IPython',
        'notebook==6.4.8'
    ],
    author='Hongyin Luo',
    author_email='hyluo@mit.edu',
    description='Chat in Python on IPython notebook.',
    long_description=open('README.md').read(),
    long_description_content_type='text/markdown',
    url='https://github.com/luohongyin/langpy-notebook',
    classifiers=[
        'Programming Language :: Python :: 3',
        'License :: OSI Approved :: MIT License',
        'Operating System :: OS Independent',
    ],
)

setup.py
luohongyin/LangCode