{ "cells": [ { "cell_type": "markdown", "id": "f56cc5ad", "metadata": {}, "source": [ "# NDIS Project - OpenAI - PBSP Scoring - Page 4 - Reinforcement strategies to support teaching " ] }, { "cell_type": "code", "execution_count": null, "id": "a8d844ea", "metadata": { "hide_input": false }, "outputs": [], "source": [ "import openai\n", "import re\n", "from ipywidgets import interact\n", "import ipywidgets as widgets\n", "from IPython.display import display, clear_output, Javascript, HTML, Markdown\n", "import matplotlib.pyplot as plt\n", "import matplotlib.ticker as mtick\n", "import json\n", "import spacy\n", "from spacy import displacy\n", "from dotenv import load_dotenv\n", "import pandas as pd\n", "import argilla as rg\n", "from argilla.metrics.text_classification import f1\n", "import warnings\n", "warnings.filterwarnings('ignore')\n", "%matplotlib inline\n", "pd.set_option('display.max_rows', 500)\n", "pd.set_option('display.max_colwidth', 10000)\n", "pd.set_option('display.width', 10000)" ] }, { "cell_type": "code", "execution_count": null, "id": "96b83a1d", "metadata": {}, "outputs": [], "source": [ "#initializations\n", "openai.api_key = os.environ['API_KEY']\n", "openai.api_base = os.environ['API_BASE']\n", "openai.api_type = os.environ['API_TYPE']\n", "openai.api_version = os.environ['API_VERSION']\n", "deployment_name = os.environ['DEPLOYMENT_ID']\n", "\n", "#argilla\n", "rg.init(\n", " api_url=os.environ[\"ARGILLA_API_URL\"],\n", " api_key=os.environ[\"ARGILLA_API_KEY\"]\n", ")" ] }, { "cell_type": "code", "execution_count": null, "id": "dee25d82", "metadata": {}, "outputs": [], "source": [ "#sentence extraction\n", "def extract_sentences(paragraph):\n", " symbols = ['\\\\.', '!', '\\\\?', ';', ':', ',', '\\\\_', '\\n', '\\\\-']\n", " pattern = '|'.join([f'{symbol}' for symbol in symbols])\n", " sentences = re.split(pattern, paragraph)\n", " sentences = [sentence.strip() for sentence in sentences if sentence.strip()]\n", " return sentences" ] }, { "cell_type": "code", "execution_count": null, "id": "08365143", "metadata": {}, "outputs": [], "source": [ "def filter_dataframe(result_df, paragraph):\n", " filtered_df = pd.DataFrame(columns=result_df.columns)\n", " last_end = 0\n", " for index, row in result_df.iterrows():\n", " phrase = row['Phrase'].lower()\n", " if phrase in paragraph.lower()[last_end:]:\n", " filtered_df = filtered_df.append(row)\n", " last_end = paragraph.lower().find(phrase, last_end) + len(phrase)\n", " else:\n", " phrase_words = []\n", " for word in phrase.split():\n", " if word in paragraph.lower()[last_end:]:\n", " phrase_words.append(word)\n", " filtered_phrase = \" \".join(phrase_words)\n", " if len(filtered_phrase) / len(phrase) >= 0.9:\n", " filtered_row = row.copy()\n", " filtered_row['Phrase'] = filtered_phrase\n", " filtered_df = filtered_df.append(filtered_row)\n", " last_end = paragraph.lower().find(filtered_phrase, last_end) + len(filtered_phrase)\n", " return filtered_df" ] }, { "cell_type": "code", "execution_count": null, "id": "02fda761", "metadata": {}, "outputs": [], "source": [ "def process_response(response, query):\n", " sentences = []\n", " topics = []\n", " scores = []\n", " lines = response.strip().split(\"\\n\")\n", " topic = None\n", " for line in lines:\n", " if \"Privileges:\" in line:\n", " topic = \"PRIVILEGES\"\n", " elif \"Breaks:\" in line:\n", " topic = \"BREAKS\"\n", " elif \"Verbal Praise:\" in line:\n", " topic = \"VERBAL PRAISE\"\n", " elif \"Social Interaction:\" in line:\n", " topic = \"SOCIAL INTERACTION\"\n", " elif \"Physical Rewards:\" in line:\n", " topic = \"PHYSICAL REWARDS\"\n", " elif \"Attention:\" in line:\n", " topic = \"ATTENTION\"\n", " elif \"None:\" in line:\n", " topic = \"NO STRATEGY\"\n", " else:\n", " try:\n", " parts = line.split(\"(Confidence Score:\")\n", " if len(parts) == 2:\n", " phrase = parts[0].strip()\n", " score = float(parts[1].strip().replace(\")\", \"\"))\n", " sentences.append(phrase)\n", " topics.append(topic)\n", " scores.append(score)\n", " except:\n", " pass\n", " result_df = pd.DataFrame({'Phrase': sentences, 'Topic': topics, 'Score': scores})\n", " try:\n", " result_df['Phrase'] = result_df['Phrase'].str.replace('\\d+\\.', '', regex=True)\n", " result_df['Phrase'] = result_df['Phrase'].str.replace('^\\s', '', regex=True)\n", " result_df['Phrase'] = result_df['Phrase'].str.strip('\"')\n", " result_df = filter_dataframe(result_df, query)\n", " except:\n", " sentences = extract_sentences(query)\n", " topics = ['NO STRATEGY'] * len(sentences)\n", " scores = [0.9] * len(sentences)\n", " result_df = pd.DataFrame({'Phrase': sentences, 'Topic': topics, 'Score': scores})\n", " return result_df" ] }, { "cell_type": "code", "execution_count": null, "id": "714fafb4", "metadata": {}, "outputs": [], "source": [ "def get_prompt(query):\n", " prompt = f\"\"\"\n", " The paragraph below is written by a disability practitioner in a behaviour support plan to describe the reinforcement strategies that will be used as a reward for completing work or demonstrating desired behavior while teaching the person with disability a new skill, an alternative behaviour or a functionally equivalent replacement behaviour.\n", "\n", " Paragraph:\n", " {query}\n", "\n", " Requirement:\n", " - Identify the phrases from the paragraph above that represent each of the following reinforcement strategies: \"Privileges\", \"Breaks\", \"Verbal Praise\", \"Social Interaction\", \"Physical Rewards\", \"Attention\".\n", "\n", " Guidelines:\n", " - \"privileges\": refers to allowing access to preferred activities such as dancing, singing, or playing, or access to preferred locations such as parks or gardens, as a reward for completing work or demonstrating desired behavior.\n", " - \"breaks\": refers to allowing breaks between lessons as a reward for completing work or demonstrating desired behavior.\n", " - \"verbal praise\": refers to providing verbal praise as a reward for completing work or demonstrating desired behavior, such as \"great job!\" or \"i'm proud of you!\"\n", " - \"social interaction\": refers to providing social interaction with preferred individuals, such as meeting with peers, parents, relatives, friends, as a reward for completing work or demonstrating desired behavior.\n", " - \"physical rewards\": refers to providing physical and tangible presents and gifts as a reward for completing work or demonstrating desired behavior, such as a lollipops, chocolates etc. \n", " - \"attention\": refers to providing a form of positive interaction or engagement with the student as a reward for completing work or demonstrating desired behavior, such as sitting next to the student, offering help or guidance, or simply engaging in conversation with the student.\n", "\n", " Specifications of a correct answer:\n", " - Provide your answer in numbered lists. \n", " - All the phrases in your answer must be exact substrings in the original paragraph. without changing any characters.\n", " - All the upper case and lower case characters in the phrases in your answer must match the upper case and lower case characters in the original paragraph.\n", " - Start numbering the phrases under each reinforcement strategy from number 1. \n", " - Start each list of phrases with these titles: \"Privileges\", \"Breaks\", \"Verbal Praise\", \"Social Interaction\", \"Physical Rewards\", \"Attention\".\n", " - For each phrase that belongs to any of the above reinforcement strategies, provide a confidence score that ranges between 0.50 and 1.00, where a score of 0.50 means you are very weakly confident that the phrase belongs to that specific reinforcement strategy, whereas a score of 1.00 means you are very strongly confident that the phrase belongs to that specific reinforcement strategy.\n", " - Never include any phrase in your answer that does not exist in the paragraph above.\n", " - If there are not any phrases that belong to one or more of the reinforcement strategies, then do not include these strategies in your answer. \n", " - Include a final numbered list titled \"None:\", which include all the remaining phrases from the paragraph above that do not belong to any of the reinforcement strategies above. Provide a confidence score for each of these phrases as well.\n", "\n", " Example answer:\n", "\n", " Privileges:\n", " 1. . (Confidence Score: )\n", " 2. . (Confidence Score: )\n", "\n", " Breaks:\n", " 1. . (Confidence Score: )\n", " 2. . (Confidence Score: )\n", "\n", " Verbal Praise:\n", " 1. . (Confidence Score: )\n", " 2. . (Confidence Score: )\n", "\n", " Social Interaction:\n", " 1. . (Confidence Score: )\n", " 2. . (Confidence Score: )\n", "\n", " Physical Rewards:\n", " 1. . (Confidence Score: )\n", " 2. . (Confidence Score: )\n", "\n", " Attention:\n", " 1. . (Confidence Score: )\n", " 2. . (Confidence Score: )\n", " \n", " None:\n", " 1. . (Confidence Score: )\n", " 2. . (Confidence Score: )\n", " \"\"\"\n", " return prompt" ] }, { "cell_type": "code", "execution_count": null, "id": "9e23821b", "metadata": {}, "outputs": [], "source": [ "def get_response_chatgpt(prompt):\n", " response=openai.ChatCompletion.create( \n", " engine=deployment_name, \n", " messages=[ \n", " {\"role\": \"system\", \"content\": \"You are a helpful assistant.\"}, \n", " {\"role\": \"user\", \"content\": prompt} \n", " ],\n", " temperature=0\n", " )\n", " reply = response[\"choices\"][0][\"message\"][\"content\"]\n", " return reply" ] }, { "cell_type": "code", "execution_count": null, "id": "983765bc", "metadata": {}, "outputs": [], "source": [ "def convert_df(result_df):\n", " new_df = pd.DataFrame(columns=['text', 'prediction'])\n", " new_df['text'] = result_df['Phrase']\n", " new_df['prediction'] = result_df.apply(lambda row: [[row['Topic'], row['Score']]], axis=1)\n", " return new_df" ] }, { "cell_type": "code", "execution_count": null, "id": "bc69cc81", "metadata": {}, "outputs": [], "source": [ "#query = \"\"\"\n", "#When developing a positive behavior support plan for Eddie, I will use a combination of reinforcement strategies to encourage him to complete work and demonstrate desired behaviors while learning new skills. First, I will offer Eddie privileges by allowing him to participate in his preferred activities such as painting or playing games on the computer after completing his work. I will also provide breaks between lessons as a reward for completing work, such as a 10-minute break to play with his favorite toy.\n", "#In addition, I will give Eddie verbal praise as a reward for demonstrating desired behaviors, such as saying \"Great job, Eddie!\" or \"You worked really hard today.\" I will also offer social interaction as a reward by allowing him to spend time with his peers or family members, such as having lunch with his best friend or calling his grandmother on the phone.\n", "#I will also provide physical rewards as a token of my appreciation for Eddie's efforts, such as stickers or small toys. Finally, I will give him attention by engaging in positive interactions with him, such as sitting with him while he completes his work and offering help and guidance when he needs it.\n", "#By using a range of reinforcement strategies that reflect Eddie's preferences and interests, I hope to motivate him to complete his work and demonstrate positive behaviors during the learning process. The goal is to create a positive and supportive learning environment that encourages Eddie's success and builds his confidence.\n", "#\"\"\"\n", "#prompt = get_prompt(query)\n", "#response = get_response_chatgpt(prompt)\n", "#result_df = process_response(response, query)\n", "#result_df" ] }, { "cell_type": "code", "execution_count": null, "id": "905eaf2a", "metadata": {}, "outputs": [], "source": [ "topic_color_dict = {\n", " 'PRIVILEGES': '#FFCCCC',\n", " 'BREAKS': '#CCFFFF',\n", " 'VERBAL PRAISE': '#FF69B4',\n", " 'SOCIAL INTERACTION': '#FFFF00',\n", " 'PHYSICAL REWARDS': '#CCCCFF',\n", " 'ATTENTION': '#FFCC99',\n", " 'NONE': '#F08080'\n", " }\n", "\n", "def color(df, color):\n", " return df.style.format({'Score': '{:,.2%}'.format}).bar(subset=['Score'], color=color)\n", "\n", "def annotate_query(highlights, query, topics):\n", " ents = []\n", " for h, t in zip(highlights, topics):\n", " pattern = re.escape(h)\n", " pattern = re.sub(r'\\\\(.)', r'[\\1\\\\W]*', pattern) # optional non-alphanumeric characters\n", " for match in re.finditer(pattern, query, re.IGNORECASE):\n", " ent_dict = {\"start\": match.start(), \"end\": match.end(), \"label\": t}\n", " ents.append(ent_dict)\n", " return ents\n", "\n", "def path_to_image_html(path):\n", " return ''\n", "\n", "passing_score = 0.7\n", "final_passing = 0.0\n", "def display_final_df(agg_df):\n", " tags = []\n", " crits = [\n", " 'PRIVILEGES',\n", " 'BREAKS',\n", " 'VERBAL PRAISE',\n", " 'SOCIAL INTERACTION',\n", " 'PHYSICAL REWARDS',\n", " 'ATTENTION'\n", " ]\n", " orig_crits = crits\n", " crits = [x for x in crits if x in agg_df.index.tolist()]\n", " bools = [agg_df.loc[crit, 'Final_Score'] > final_passing for crit in crits]\n", " paths = ['./thumbs_up.png' if x else './thumbs_down.png' for x in bools]\n", " df = pd.DataFrame({'Reinforcement Strategy': crits, 'USED': paths})\n", " rem_crits = [x for x in orig_crits if x not in crits]\n", " if len(rem_crits) > 0:\n", " df2 = pd.DataFrame({'Reinforcement Strategy': rem_crits, 'USED': ['./thumbs_down.png'] * len(rem_crits)})\n", " df = pd.concat([df, df2])\n", " df = df.set_index('Reinforcement Strategy')\n", " pd.set_option('display.max_colwidth', None)\n", " display(HTML('
' + df.to_html(classes=[\"align-center\"], index=True, escape=False ,formatters=dict(USED=path_to_image_html)) + '
'))\n", " " ] }, { "cell_type": "markdown", "id": "2c6e9fe7", "metadata": {}, "source": [ "### Teaching of a new skill, alternative behaviour or functionally equivalent replacement behaviour (Reinforcement strategies to support teaching)" ] }, { "cell_type": "code", "execution_count": null, "id": "76dd8cab", "metadata": { "scrolled": false }, "outputs": [], "source": [ "#demo with Voila\n", "\n", "bhvr_label = widgets.Label(value='Please type your answer:')\n", "bhvr_text_input = widgets.Textarea(\n", " value='',\n", " placeholder='Type your answer',\n", " description='',\n", " disabled=False,\n", " layout={'height': '300px', 'width': '90%'}\n", ")\n", "\n", "bhvr_nlp_btn = widgets.Button(\n", " description='Score Answer',\n", " disabled=False,\n", " button_style='success', # 'success', 'info', 'warning', 'danger' or ''\n", " tooltip='Score Answer',\n", " icon='check',\n", " layout={'height': '70px', 'width': '250px'}\n", ")\n", "bhvr_agr_btn = widgets.Button(\n", " description='Validate Data',\n", " disabled=False,\n", " button_style='success', # 'success', 'info', 'warning', 'danger' or ''\n", " tooltip='Validate Data',\n", " icon='check',\n", " layout={'height': '70px', 'width': '250px'}\n", ")\n", "bhvr_eval_btn = widgets.Button(\n", " description='Evaluate Model',\n", " disabled=False,\n", " button_style='success', # 'success', 'info', 'warning', 'danger' or ''\n", " tooltip='Evaluate Model',\n", " icon='check',\n", " layout={'height': '70px', 'width': '250px'}\n", ")\n", "btn_box = widgets.HBox([bhvr_nlp_btn, bhvr_agr_btn, bhvr_eval_btn], \n", " layout={'width': '100%', 'height': '160%'})\n", "bhvr_outt = widgets.Output()\n", "bhvr_outt.layout.height = '100%'\n", "bhvr_outt.layout.width = '100%'\n", "bhvr_box = widgets.VBox([bhvr_text_input, btn_box, bhvr_outt], \n", " layout={'width': '100%', 'height': '160%'})\n", "dataset_rg_name = 'pbsp-page4-reinforcement-strategy-argilla-ds'\n", "agrilla_df = None\n", "annotated = False\n", "def on_bhvr_button_next(b):\n", " global agrilla_df\n", " with bhvr_outt:\n", " clear_output()\n", " query = bhvr_text_input.value\n", " prompt = get_prompt(query)\n", " response = get_response_chatgpt(prompt)\n", " result_df = process_response(response, query)\n", " sub_result_df = result_df[(result_df['Score'] >= passing_score) & (result_df['Topic'] != 'NO STRATEGY')]\n", " sub_2_result_df = result_df[result_df['Topic'] == 'NO STRATEGY']\n", " highlights = []\n", " if len(sub_result_df) > 0:\n", " highlights = sub_result_df['Phrase'].tolist()\n", " highlight_topics = sub_result_df['Topic'].tolist() \n", " ents = annotate_query(highlights, query, highlight_topics)\n", " colors = {}\n", " for ent, ht in zip(ents, highlight_topics):\n", " colors[ent['label']] = topic_color_dict[ht]\n", "\n", " ex = [{\"text\": query,\n", " \"ents\": ents,\n", " \"title\": None}]\n", " title = \"Reinforcement Strategy Highlights\"\n", " display(HTML(f'

{title}

'))\n", " html = displacy.render(ex, style=\"ent\", manual=True, jupyter=True, options={'colors': colors})\n", " display(HTML(html))\n", " title = \"Reinforcement Strategy Classifications\"\n", " display(HTML(f'

{title}

'))\n", " for top in topic_color_dict.keys():\n", " top_result_df = sub_result_df[sub_result_df['Topic'] == top]\n", " if len(top_result_df) > 0:\n", " top_result_df = top_result_df.sort_values(by='Score', ascending=False).reset_index(drop=True)\n", " top_result_df = top_result_df.set_index('Phrase')\n", " top_result_df = top_result_df[['Score']]\n", " display(HTML(\n", " f'

{top}

'))\n", " display(color(top_result_df, topic_color_dict[top]))\n", "\n", " agg_df = sub_result_df.groupby('Topic')['Score'].sum()\n", " agg_df = agg_df.to_frame()\n", " agg_df.index.name = 'Topic'\n", " agg_df.columns = ['Total Score']\n", " agg_df = agg_df.assign(\n", " Final_Score=lambda x: x['Total Score'] / x['Total Score'].sum() * 100.00\n", " )\n", " agg_df = agg_df.sort_values(by='Final_Score', ascending=False)\n", " title = \"Reinforcement Strategy Coverage\"\n", " display(HTML(f'

{title}

'))\n", " agg_df['Topic'] = agg_df.index\n", " rem_topics= [x for x in list(topic_color_dict.keys()) if not x in agg_df.Topic.tolist()]\n", " if len(rem_topics) > 0:\n", " rem_agg_df = pd.DataFrame({'Topic': rem_topics, 'Final_Score': 0.0, 'Total Score': 0.0})\n", " agg_df = pd.concat([agg_df, rem_agg_df])\n", " labels = agg_df['Final_Score'].round(1).astype('str') + '%'\n", " ax = agg_df.plot.bar(x='Topic', y='Final_Score', rot=0, figsize=(20, 5), align='center')\n", " for container in ax.containers:\n", " ax.bar_label(container, labels=labels)\n", " ax.yaxis.set_major_formatter(mtick.PercentFormatter())\n", " ax.legend([\"Final Score (%)\"])\n", " ax.set_xlabel('')\n", " plt.show()\n", " title = \"Final Scores\"\n", " display(HTML(f'

{title}

'))\n", " display_final_df(agg_df)\n", " if len(sub_2_result_df) > 0:\n", " sub_result_df = pd.concat([sub_result_df, sub_2_result_df]).reset_index(drop=True)\n", " agrilla_df = sub_result_df.copy()\n", " else:\n", " print(query)\n", " \n", "def on_agr_button_next(b):\n", " global agrilla_df, annotated\n", " with bhvr_outt:\n", " clear_output()\n", " if agrilla_df is not None:\n", " # convert the dataframe to the structure accepted by argilla\n", " converted_df = convert_df(agrilla_df)\n", " # convert pandas dataframe to DatasetForTextClassification\n", " dataset_rg = rg.DatasetForTextClassification.from_pandas(converted_df)\n", " # delete the old DatasetForTextClassification from the Argilla web app if exists\n", " rg.delete(dataset_rg_name, workspace=\"admin\")\n", " # load the new DatasetForTextClassification into the Argilla web app\n", " rg.log(dataset_rg, name=dataset_rg_name, workspace=\"admin\")\n", " # Make sure all classes are present for annotation\n", " rg_settings = rg.TextClassificationSettings(label_schema=list(topic_color_dict.keys()))\n", " rg.configure_dataset(name=dataset_rg_name, workspace=\"admin\", settings=rg_settings)\n", " annotated = True\n", " else:\n", " display(Markdown(\"

Please score the answer first!

\"))\n", " \n", "def on_eval_button_next(b):\n", " global annotated\n", " with bhvr_outt:\n", " clear_output()\n", " if annotated:\n", " display(f1(dataset_rg_name).visualize())\n", " else:\n", " display(Markdown(\"

Please score the answer and validate the data first!

\"))\n", "\n", "bhvr_nlp_btn.on_click(on_bhvr_button_next)\n", "bhvr_agr_btn.on_click(on_agr_button_next)\n", "bhvr_eval_btn.on_click(on_eval_button_next)\n", "\n", "display(bhvr_label, bhvr_box)" ] }, { "cell_type": "code", "execution_count": null, "id": "ed551eba", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "hide_input": false, "kernelspec": { "display_name": "Python 3.9 (Argilla)", "language": "python", "name": "argilla" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.9.16" }, "toc": { "base_numbering": 1, "nav_menu": {}, "number_sections": false, "sideBar": true, "skip_h1_title": true, "title_cell": "Table of Contents", "title_sidebar": "Contents", "toc_cell": false, "toc_position": { "height": "calc(100% - 180px)", "left": "10px", "top": "150px", "width": "258.097px" }, "toc_section_display": true, "toc_window_display": false } }, "nbformat": 4, "nbformat_minor": 5 }