Spaces:
Sleeping
Sleeping
| import openai | |
| import os | |
| import numpy as np | |
| import pandas as pd | |
| from tqdm import tqdm | |
| import faiss | |
| from sentence_transformers import SentenceTransformer | |
| import time | |
| encoder = SentenceTransformer('all-mpnet-base-v2') | |
| from sklearn.metrics import classification_report | |
| from functools import reduce | |
| import json | |
| print ("finished import") | |
| openai.api_key = os.environ["OPENAI_API_KEY"] | |
| MODEL_NAME = 'gpt-3.5-turbo-0613' # GPT 4 alternative: 'gpt-4-0613' | |
| # MODEL_NAME = 'gpt-4-1106-preview' | |
| MAX_RETRIES = 5 | |
| all_strategy = pd.read_csv("all_strategy_with_generated_reason.csv") | |
| def find_knn(target_utterance, search_inventory, top_k): | |
| all_utterances = search_inventory['message_text'].tolist() | |
| all_utterances_embeddings = encoder.encode(all_utterances) | |
| print (all_utterances_embeddings.shape) | |
| vec_dimension = all_utterances_embeddings.shape[1] | |
| index = faiss.IndexFlatL2(vec_dimension) | |
| faiss.normalize_L2(all_utterances_embeddings) | |
| index.add(all_utterances_embeddings) | |
| search_vec = encoder.encode(target_utterance) | |
| _vector = np.array([search_vec]) | |
| faiss.normalize_L2(_vector) | |
| k = index.ntotal | |
| distances, ann = index.search(_vector, k=k) | |
| results = pd.DataFrame({'distances': distances[0], 'ann': ann[0]}) | |
| # print (results.head(5)) | |
| select_ind = results.ann[:top_k].to_list() | |
| distance_list = results.distances[:top_k].to_list() | |
| return search_inventory.iloc[select_ind], distance_list | |
| def find_similar_situation(target_situation, list_of_situations, top_k): | |
| situation_embeddings = encoder.encode(list_of_situations) | |
| vec_dimension = situation_embeddings.shape[1] | |
| index = faiss.IndexFlatL2(vec_dimension) | |
| faiss.normalize_L2(situation_embeddings) | |
| index.add(situation_embeddings) | |
| search_vec = encoder.encode(target_situation) | |
| _vector = np.array([search_vec]) | |
| faiss.normalize_L2(_vector) | |
| k = index.ntotal | |
| distances, ann = index.search(_vector, k=k) | |
| results = pd.DataFrame({'distances': distances[0], 'ann': ann[0]}) | |
| # print (results.head(top_k)) | |
| # for i in range(top_k): | |
| # print (results.ann[i], all_utterances[results.ann[i]]) | |
| select_ind = results.ann[:top_k].to_list() | |
| return [list_of_situations[i] for i in select_ind] | |
| def generate_knn_demonstrations(all_strategy, mode, target_situation, category, u, strategy, top_k_situation, top_k_utterances): | |
| all_convo = all_strategy['situation'].drop_duplicates().tolist() | |
| if mode == 'similar_situation_knn': | |
| # other_convo = all_strategy[all_strategy.conversation_id != convo_id]['context'].drop_duplicates().tolist() | |
| # target_situation = all_strategy[(all_strategy.conversation_id == convo_id)]['context'].reset_index(drop=True)[0] | |
| situation_list = find_similar_situation(target_situation, all_convo, top_k_situation) | |
| # print (situation_list) | |
| utterances_in_situations = all_strategy[all_strategy['situation'].isin(situation_list)] | |
| elif mode == 'in_category_knn': | |
| utterances_in_situations = all_strategy[all_strategy.category == category].reset_index(drop=True) | |
| elif mode == 'in_category_similar_situation_knn': | |
| # other_convo_in_category = all_strategy[all_strategy.category == category]['context'].drop_duplicates().tolist() | |
| # target_situation = all_strategy[(all_strategy.conversation_id == convo_id)]['context'].reset_index(drop=True)[0] | |
| situation_list = find_similar_situation(target_situation, all_convo, top_k_situation) | |
| utterances_in_situations = all_strategy[all_strategy['situation'].isin(situation_list)] | |
| elif mode == 'all_utterances_knn': | |
| utterances_in_situations = all_strategy | |
| strategy_label = 'label_' + strategy | |
| suggestion_column = 'suggestion_' + strategy | |
| reason_column = 'reason_' + strategy | |
| rewrite_column = 'rewrite_' + strategy | |
| if strategy in ['describe', 'express', 'assert', 'reinforce', 'negotiate']: | |
| utterances_strong = utterances_in_situations[utterances_in_situations[strategy_label] == 'Strong'].reset_index(drop=True) | |
| strong_demonstrations, strong_distance = find_knn(u, utterances_strong, top_k_utterances) | |
| strong_demonstrations = strong_demonstrations.reset_index(drop=True) | |
| utterances_weak = utterances_in_situations[utterances_in_situations[strategy_label] == 'Weak'].reset_index(drop=True) | |
| weak_demonstrations, weak_distance = find_knn(u, utterances_weak, top_k_utterances) | |
| weak_demonstrations = weak_demonstrations.reset_index(drop=True) | |
| utterances_no = utterances_in_situations[utterances_in_situations[strategy_label] == 'No'].reset_index(drop=True) | |
| none_demonstrations, none_distance = find_knn(u, utterances_no, top_k_utterances) | |
| none_demonstrations = none_demonstrations.reset_index(drop=True) | |
| list_strong = [(strong_demonstrations['situation'][i], strong_demonstrations['message_id'][i], strong_demonstrations['message_text'][i], strong_demonstrations[suggestion_column][i], strong_demonstrations[reason_column][i], strong_demonstrations[rewrite_column][i]) for i in range(len(strong_demonstrations))] #ADD | |
| list_weak = [(weak_demonstrations['situation'][i], weak_demonstrations['message_id'][i], weak_demonstrations['message_text'][i], weak_demonstrations[suggestion_column][i], weak_demonstrations[reason_column][i], weak_demonstrations[rewrite_column][i]) for i in range(len(weak_demonstrations))] #ADD | |
| list_none = [(none_demonstrations['situation'][i], none_demonstrations['message_id'][i], none_demonstrations['message_text'][i], none_demonstrations[suggestion_column][i], none_demonstrations[reason_column][i], none_demonstrations[rewrite_column][i]) for i in range(len(none_demonstrations))] #ADD | |
| # print (strong_demonstrations['message_text'], weak_demonstrations['message_text'], none_demonstrations['message_text']) | |
| all_distance = strong_distance + weak_distance + none_distance | |
| return list_strong, list_weak, list_none, all_distance | |
| elif strategy in ['mindful', 'confident']: | |
| utterances_strong = utterances_in_situations[utterances_in_situations[strategy_label] == 'Yes'].reset_index(drop=True) | |
| strong_demonstrations, strong_distance = find_knn(u, utterances_strong, top_k_utterances) | |
| strong_demonstrations = strong_demonstrations.reset_index(drop=True) | |
| utterances_weak = utterances_in_situations[utterances_in_situations[strategy_label] == 'No'].reset_index(drop=True) | |
| weak_demonstrations, weak_distance = find_knn(u, utterances_weak, top_k_utterances) | |
| weak_demonstrations = weak_demonstrations.reset_index(drop=True) | |
| # return strong_demonstrations, weak_demonstrations | |
| list_strong = [(strong_demonstrations['situation'][i], strong_demonstrations['message_id'][i], strong_demonstrations['message_text'][i], strong_demonstrations[suggestion_column][i], strong_demonstrations[reason_column][i], strong_demonstrations[rewrite_column][i]) for i in range(len(strong_demonstrations))] # ADD | |
| list_weak = [(weak_demonstrations['situation'][i], weak_demonstrations['message_id'][i], weak_demonstrations['message_text'][i], weak_demonstrations[suggestion_column][i], weak_demonstrations[reason_column][i], weak_demonstrations[rewrite_column][i]) for i in range(len(weak_demonstrations))] # ADD | |
| all_distance = strong_distance + weak_distance | |
| return list_strong, list_weak, '', all_distance | |
| def form_prompt_from_demonstrations(list_strong, list_weak, list_none, strategy, prompt_strategy, include_improve, cot_dict, zip_reorder=True): | |
| # list_strong: | |
| # (strong_demonstrations['context'][i], strong_demonstrations['message_id'][i], strong_demonstrations['message_text'][i], strong_demonstrations[suggestion_column][i], strong_demonstrations[reason_column][i]) | |
| strategy_cap = strategy[0].upper() + strategy[1:] | |
| if strategy == 'describe': | |
| strong_improve_str = f"This is a great {strategy_cap}! It sticks to the facts, makes no judgemental statements, and is objective." | |
| elif strategy == 'express': | |
| strong_improve_str = f"This is a great {strategy_cap}! It express your feeling or opinions explicitly." | |
| elif strategy == 'assert': | |
| strong_improve_str = f"This is a great {strategy_cap}! It is clear, concise, and to the point." | |
| elif strategy == 'reinforce': | |
| strong_improve_str = f"This is a great {strategy_cap}! It reinforces the other person." | |
| elif strategy == 'negotiate': | |
| strong_improve_str = f"This is a great {strategy_cap}! It shows that you are trying to find an alternative solution." | |
| elif strategy == 'mindful': | |
| strong_improve_str = f"This utterance shows mindfulness. You focused on your goal, and did not get distracted or get off topic." | |
| elif strategy == 'confident': | |
| strong_improve_str = f"This utterance shows confidence. You used a confident tone. The statement was effective and competent." | |
| # weak_improve = [list(weak_demonstrations[suggestion_column])[i] for i in range(actual_top_k_weak)] | |
| weak_improve = [list_weak[i][3] for i in range(len(list_weak))] | |
| if strategy in ['describe', 'express', 'assert', 'reinforce', 'negotiate']: | |
| # none_improve = [list(none_demonstrations[suggestion_column])[i] for i in range(actual_top_k_none)] | |
| none_improve = [list_none[i][3] for i in range(len(list_none))] | |
| strong_rating = f"Strong {strategy_cap}" | |
| weak_rating = f"Weak {strategy_cap}" | |
| if strategy in ['describe', 'express', 'assert', 'reinforce', 'negotiate']: | |
| strong_rating = f"Strong {strategy_cap}" | |
| weak_rating = f"Weak {strategy_cap}" | |
| none_rating = f"No {strategy_cap}" | |
| elif strategy in ['mindful', 'confident']: | |
| strong_rating = "Yes" | |
| weak_rating = "No" | |
| user_prompt = '' | |
| if prompt_strategy == 'CoT': | |
| if strategy in ['mindful', 'confident']: | |
| if strategy == 'mindful': answer_subquestion_yes, answer_subquestion_no = 'No', 'Yes' | |
| elif strategy == 'confident': answer_subquestion_yes, answer_subquestion_no = 'Yes', 'No' | |
| for i in range(len(list_strong)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, {cot_dict[strategy][0]}{answer_subquestion_yes}, {strategy_cap} Rating: {strong_rating}, Suggestion for improvement: {strong_improve_str}\n" | |
| else: | |
| user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, {cot_dict[strategy][0]}{answer_subquestion_yes}, {strategy_cap} Rating: {strong_rating}\n" | |
| for i in range(len(list_weak)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, {cot_dict[strategy][0]}{answer_subquestion_no}, {strategy_cap} Rating: {weak_rating}, Suggestion for improvement: {list_weak[i][3]}\n" | |
| else: | |
| user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, {cot_dict[strategy][0]}{answer_subquestion_no}, {strategy_cap} Rating: {weak_rating}\n" | |
| elif strategy in ['describe', 'express', 'assert', 'reinforce', 'negotiate']: | |
| for i in range(len(list_strong)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, {cot_dict[strategy][0]}Yes, {cot_dict[strategy][1]}Yes, {strategy_cap} Rating: {strong_rating}, Suggestion for improvement: {strong_improve_str}\n" | |
| else: | |
| user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, {cot_dict[strategy][0]}Yes, {cot_dict[strategy][1]}Yes, {strategy_cap} Rating: {strong_rating}\n" | |
| for i in range(len(list_weak)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, {cot_dict[strategy][0]}Yes, {cot_dict[strategy][1]}No, {strategy_cap} Rating: {weak_rating}, Suggestion for improvement: {list_weak[i][3]}\n" | |
| else: | |
| user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, {cot_dict[strategy][0]}Yes, {cot_dict[strategy][1]}No, {strategy_cap} Rating: {weak_rating}\n" | |
| for i in range(len(list_none)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_none[i][0]}, Utterance: {list_none[i][2]}, {cot_dict[strategy][0]}No, {strategy_cap} Rating: {none_rating}, Suggestion for improvement: {list_none[i][3]}\n" # TODO: one question or two questions for None situation?n | |
| else: | |
| user_prompt += f"Context: {list_none[i][0]}, Utterance: {list_none[i][2]}, {cot_dict[strategy][0]}No, {strategy_cap} Rating: {none_rating}\n" | |
| elif prompt_strategy == 'CoT-reason': | |
| if strategy in ['mindful', 'confident']: | |
| if strategy == 'mindful': answer_subquestion_yes, answer_subquestion_no = 'No', 'Yes' | |
| elif strategy == 'confident': answer_subquestion_yes, answer_subquestion_no = 'Yes', 'No' | |
| for i in range(len(list_strong)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, {cot_dict[strategy][0]}{answer_subquestion_yes}, Reason for rating: {list_strong[i][4]}, {strategy_cap} Rating: {strong_rating}, Suggestion for improvement: {strong_improve_str}###" | |
| else: | |
| user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, {cot_dict[strategy][0]}{answer_subquestion_yes}, Reason for rating: {list_strong[i][4]}, {strategy_cap} Rating: {strong_rating}\n" | |
| for i in range(len(list_weak)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, {cot_dict[strategy][0]}{answer_subquestion_no}, Reason for rating: {list_weak[i][4]}, {strategy_cap} Rating: {weak_rating}, Suggestion for improvement: {list_weak[i][3]}###" | |
| else: | |
| user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, {cot_dict[strategy][0]}{answer_subquestion_no}, Reason for rating: {list_weak[i][4]}, {strategy_cap} Rating: {weak_rating}\n" | |
| elif strategy in ['describe', 'express', 'assert', 'reinforce', 'negotiate']: | |
| if include_improve: | |
| strong_prompts_list = [f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, {cot_dict[strategy][0]}Yes, {cot_dict[strategy][1]}Yes, Reason for rating: {list_strong[i][4]}, {strategy_cap} Rating: {strong_rating}, Suggestion for improvement: {strong_improve_str}###" for i in range(len(list_strong))] | |
| strong_prompts = ''.join(strong_prompts_list) | |
| weak_prompts_list = [f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, {cot_dict[strategy][0]}Yes, {cot_dict[strategy][1]}No, Reason for rating: {list_weak[i][4]}, {strategy_cap} Rating: {weak_rating}, Suggestion for improvement: {list_weak[i][3]}###" for i in range(len(list_weak))] | |
| weak_prompts = ''.join(weak_prompts_list) | |
| none_prompts_list = [f"Context: {list_none[i][0]}, Utterance: {list_none[i][2]}, {cot_dict[strategy][0]}No, Reason for rating: {list_none[i][4]}, {strategy_cap} Rating: {none_rating}, Suggestion for improvement: {list_none[i][3]}###" for i in range(len(list_none))] | |
| none_prompts = ''.join(none_prompts_list) | |
| # if zip_reorder: | |
| user_prompt_list = [] | |
| for s,w,n in zip(strong_prompts_list, weak_prompts_list, none_prompts_list): | |
| user_prompt_list.extend([s,w,n]) | |
| user_prompt = ''.join(user_prompt_list) | |
| # else: | |
| # user_prompt = strong_prompts + weak_prompts + none_prompts | |
| elif prompt_strategy == 'reason': | |
| if strategy in ['mindful', 'confident']: | |
| if strategy == 'mindful': answer_subquestion_yes, answer_subquestion_no = 'No', 'Yes' | |
| elif strategy == 'confident': answer_subquestion_yes, answer_subquestion_no = 'Yes', 'No' | |
| if include_improve: | |
| strong_prompts_list = [f"Context: {list_strong[i][0]} Utterance: {list_strong[i][2]} Step 1 - Reason for rating: {list_strong[i][4]}### Step 2 - {strategy_cap} Rating: {strong_rating}### Step 3 - Suggestion for improvement: {strong_improve_str}###" for i in range(len(list_strong))] | |
| strong_prompts = ''.join(strong_prompts_list) | |
| weak_prompts_list = [f"Context: {list_weak[i][0]} Utterance: {list_weak[i][2]} Step 1 - Reason for rating: {list_weak[i][4]}### Step 2 - {strategy_cap} Rating: {weak_rating}### Step 3 - Suggestion for improvement: {list_weak[i][3]}###" for i in range(len(list_weak))] | |
| weak_prompts = ''.join(weak_prompts_list) | |
| # none_prompts_list = [f"Context: {list_none[i][0]}, Utterance: {list_none[i][2]}, Reason for rating: {list_none[i][4]}, {strategy_cap} Rating: {none_rating}, Suggestion for improvement: {list_none[i][3]}###" for i in range(len(list_none))] | |
| # none_prompts = ''.join(none_prompts_list) | |
| if zip_reorder: | |
| # if zip_reorder: | |
| user_prompt_list = [] | |
| for s,w in zip(strong_prompts_list, weak_prompts_list): | |
| user_prompt_list.extend([s,w]) | |
| user_prompt = ''.join(user_prompt_list) | |
| else: | |
| user_prompt = strong_prompts + weak_prompts + none_prompts | |
| elif strategy in ['describe', 'express', 'assert', 'reinforce', 'negotiate']: | |
| if include_improve: | |
| # print ("right prompt construction") | |
| strong_prompts_list = [f"Context: {list_strong[i][0]} Utterance: {list_strong[i][2]} Step 1 - Reason for rating: {list_strong[i][4]}### Step 2 - {strategy_cap} Rating: {strong_rating}### Step 3 - Suggestion for improvement: {strong_improve_str}###" for i in range(len(list_strong))] | |
| strong_prompts = ''.join(strong_prompts_list) | |
| weak_prompts_list = [f"Context: {list_weak[i][0]} Utterance: {list_weak[i][2]} Step 1 - Reason for rating: {list_weak[i][4]}### Step 2 - {strategy_cap} Rating: {weak_rating}### Step 3 -Suggestion for improvement: {list_weak[i][3]}###" for i in range(len(list_weak))] | |
| weak_prompts = ''.join(weak_prompts_list) | |
| none_prompts_list = [f"Context: {list_none[i][0]} Utterance: {list_none[i][2]} Step 1 - Reason for rating: {list_none[i][4]}### Step 2 -{strategy_cap} Rating: {none_rating}### Step 3 -Suggestion for improvement: {list_none[i][3]}###" for i in range(len(list_none))] | |
| none_prompts = ''.join(none_prompts_list) | |
| # print (len(strong_prompts_list), len(weak_prompts_list), len(none_prompts_list)) | |
| if zip_reorder: | |
| # if zip_reorder: | |
| user_prompt_list = [] | |
| for s,w,n in zip(strong_prompts_list, weak_prompts_list, none_prompts_list): | |
| user_prompt_list.extend([s,w,n]) | |
| user_prompt = ''.join(user_prompt_list) | |
| else: | |
| user_prompt = strong_prompts + weak_prompts + none_prompts | |
| # elif prompt_strategy == 'reason': | |
| # if strategy in ['mindful', 'confident']: | |
| # if strategy == 'mindful': answer_subquestion_yes, answer_subquestion_no = 'No', 'Yes' | |
| # elif strategy == 'confident': answer_subquestion_yes, answer_subquestion_no = 'Yes', 'No' | |
| # for i in range(len(list_strong)): | |
| # if include_improve: | |
| # user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, Reason for rating: {list_strong[i][4]}, {strategy_cap} Rating: {strong_rating}, Suggestion for improvement: {strong_improve_str}###" | |
| # else: | |
| # user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, Reason for rating: {list_strong[i][4]}, {strategy_cap} Rating: {strong_rating}\n" | |
| # for i in range(len(list_weak)): | |
| # if include_improve: | |
| # user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, Reason for rating: {list_weak[i][4]}, {strategy_cap} Rating: {weak_rating}, Suggestion for improvement: {list_weak[i][3]}###" | |
| # else: | |
| # user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, Reason for rating: {list_weak[i][4]}, {strategy_cap} Rating: {weak_rating}\n" | |
| # elif strategy in ['describe', 'express', 'assert', 'reinforce', 'negotiate']: | |
| # for i in range(len(list_strong)): | |
| # if include_improve: | |
| # user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, Reason for rating: {list_strong[i][4]}, {strategy_cap} Rating: {strong_rating}, Suggestion for improvement: {strong_improve_str}###" | |
| # else: | |
| # user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, Reason for rating: {list_strong[i][4]}, {strategy_cap} Rating: {strong_rating}\n" | |
| # for i in range(len(list_weak)): | |
| # if include_improve: | |
| # user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, Reason for rating: {list_weak[i][4]}, {strategy_cap} Rating: {weak_rating}, Suggestion for improvement: {list_weak[i][3]}###" | |
| # else: | |
| # user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, Reason for rating: {list_weak[i][4]}, {strategy_cap} Rating: {weak_rating}\n" | |
| # for i in range(len(list_none)): | |
| # if include_improve: | |
| # user_prompt += f"Context: {list_none[i][0]}, Utterance: {list_none[i][2]}, Reason for rating: {list_none[i][4]}, {strategy_cap} Rating: {none_rating}, Suggestion for improvement: {list_none[i][3]}###" # TODO: one question or two questions for None situation?n | |
| # else: | |
| # user_prompt += f"Context: {list_none[i][0]}, Utterance: {list_none[i][2]}, Reason for rating: {list_none[i][4]}, {strategy_cap} Rating: {none_rating}\n" | |
| else: | |
| for i in range(len(list_strong)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, {strategy_cap} Rating: {strong_rating}, Suggestion for improvement: {strong_improve_str}###" | |
| else: | |
| user_prompt += f"Context: {list_strong[i][0]}, Utterance: {list_strong[i][2]}, {strategy_cap} Rating: {strong_rating}\n" | |
| for i in range(len(list_weak)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, {strategy_cap} Rating: {weak_rating}, Suggestion for improvement: {list_weak[i][2]}###" | |
| else: | |
| user_prompt += f"Context: {list_weak[i][0]}, Utterance: {list_weak[i][2]}, {strategy_cap} Rating: {weak_rating}\n" | |
| if strategy in ['describe', 'express', 'assert', 'reinforce', 'negotiate']: | |
| for i in range(len(list_none)): | |
| if include_improve: | |
| user_prompt += f"Context: {list_none[i][0]}, Utterance: {list_none[i][2]}, {strategy_cap} Rating: {none_rating}, Suggestion for improvement: {list_none[i][3]}###" | |
| else: | |
| user_prompt += f"Context: {list_none[i][0]}, Utterance: {list_none[i][2]}, {strategy_cap} Rating: {none_rating}\n" | |
| print(user_prompt) | |
| return user_prompt | |
| def generate_feedback(current_situation, category, u, prompt_strategy, prompt_selection, strategy_list): | |
| # parse strategy | |
| strategy = strategy_list[0] | |
| strategy = strategy[0].lower() + strategy[1:] | |
| print (f"Generating feedback for {strategy}") | |
| # parse category | |
| cate = category[0].lower() + category[1:] | |
| # load files once, move to user_study_interface.py | |
| with open('prompts.json', 'r') as fp: | |
| system_prompts = json.load(fp) | |
| print ("There are in total {} utterances".format(len(all_strategy))) | |
| list_of_convo = all_strategy['conversation_id'].drop_duplicates().tolist() | |
| print (list_of_convo) | |
| all_convo_results = pd.DataFrame() | |
| dict_CoT = { | |
| 'describe': ['Answer in Yes or No: The utterance is or contains a description of the given context. Answer:', 'Answer in Yes or No: The utterance sticks to the fact, makes no judgemental statement, and is objective. Answer:'], | |
| 'express': ['Answer in Yes or No: The utterance is or contains an expression of the speaker\'s feelings. Answer:', 'Answer in Yes or No: The expression of feelings in the utterance is clear and explicit. Answer:'], | |
| 'assert': ['Answer in Yes or No: The utterance is or contains an ask for what the speaker wants. Answer:', 'Answer in Yes or No: The ask in the utterance is clear and explicit, or the speaker is saying no clearly and explicitly. Answer:'], | |
| 'reinforce': ['Answer in Yes or No: The utterance is or contains a reinforcement of a reward for the other person. Answer:', 'Answer in Yes or No: The reinforcement in the utterance is targeted to the other person and is communicated clearly. Answer:'], | |
| 'negotiate': ['Answer in Yes or No: The utterance offers a compromise or an alternative solution. Answer:', 'Answer in Yes or No: The compromise or alternative solution in the utterance is clear and explicit. Answer:'], # TODO: add weak description from clusters | |
| 'mindful':['Answer in Yes or No: The utterance shows that the speaker is responding to attacks and criticism or is losing track of their goals. Answer:'], # REVERSE! | |
| 'confident': ['Answer in Yes or No: The utterance shows a confident tone, and is effective and competent in conveying the speaker\'s goal. Answer:'], | |
| } | |
| system_prompt = system_prompts[prompt_strategy][strategy] | |
| start_time = time.time() | |
| current_tries = 1 | |
| strategy_cap = strategy[0].upper() + strategy[1:] | |
| # User Prompt | |
| # if prompt_selection == "knn": | |
| # user_prompt = get_knn_prompt(u, top_k, True, index_knn, args.strategy, prompt_strategy, dict_CoT) | |
| if prompt_selection in ("in_category_similar_situation_knn", "similar_situation_knn", "in_category_knn", "all_utterances_knn") : | |
| print ("right") | |
| strong_, weak_, none_, knn_distance = generate_knn_demonstrations(all_strategy, prompt_selection, current_situation, cate, u, strategy, top_k_situation=5, top_k_utterances=3) #all_strategy, mode, target_situation, category, u, strategy, top_k_situation, top_k_utterances | |
| user_prompt = form_prompt_from_demonstrations(strong_, weak_, none_, strategy, prompt_strategy, True, dict_CoT) | |
| elif prompt_selection == "zero_shot": | |
| user_prompt = '' | |
| # Current Prompt | |
| if prompt_strategy == "CoT" or prompt_strategy == "CoT-reason" : | |
| current_prompt = f"Context: {current_situation}, Utterance: {u}," | |
| elif prompt_strategy == "reason": | |
| current_prompt = f"Context: {current_situation}, Utterance: {u}, Reason for rating:" | |
| else: | |
| current_prompt = f"Context: {current_situation}, Utterance: {u}, {strategy_cap} Rating:" | |
| prompt = [{"role":"system", "content":system_prompt}, {"role":"user", "content": user_prompt+current_prompt}] | |
| print ("Full prompt: ", prompt) | |
| while current_tries <= MAX_RETRIES: | |
| try: | |
| response = openai.ChatCompletion.create( | |
| model=MODEL_NAME, | |
| messages = prompt, | |
| # max_tokens = 256, | |
| temperature = 0, | |
| ) | |
| curr_response_str = response['choices'][0]['message']['content'].replace('\n', ' ').strip() | |
| # print (curr_response_str, '\t') | |
| break | |
| except Exception as e: | |
| print('error: ', str(e)) | |
| print('response retrying') | |
| current_tries += 1 | |
| if current_tries > MAX_RETRIES: | |
| break | |
| time.sleep(5) | |
| time_taken = time.time() - start_time | |
| print (f'Time taken for this utterance: {time_taken}') | |
| try: | |
| rating_list = curr_response_str.split(f"{strategy_cap} Rating: ", 1)[1].split(" ", 2)[:2].replace('[,.#]', '', regex=True) | |
| rating = ' '.join(rating_list) | |
| suggestion = curr_response_str.split(f"{strategy_cap} Rating: ", 1)[1].split(" ", 2)[2:][0].split("###", 1)[0] | |
| output = rating + '. \n' + suggestion | |
| except: | |
| output = curr_response_str | |
| return output | |
| # def generate_feedback_with_mc(current_situation, category, u, prompt_strategy, prompt_selection, strategy_list): | |
| # strategy_output = generate_feedback(current_situation, category, u, prompt_strategy, prompt_selection, strategy_list) | |
| # mindful_output = generate_feedback(current_situation, category, u, prompt_strategy, prompt_selection, ['mindful']) | |
| # confident_output = generate_feedback(current_situation, category, u, prompt_strategy, prompt_selection, ['confident']) | |
| # print (strategy_output, mindful_output, confident_output) | |
| # all_concat = strategy_output + '\n' + mindful_output + '\n' + confident_output | |
| # return all_concat | |
| def generate_feedback_with_mc(current_situation, category, u, prompt_strategy, prompt_selection, strategy_list): | |
| strategy_output = generate_feedback(current_situation, category, u, prompt_strategy, prompt_selection, strategy_list) | |
| mindful_output = generate_feedback(current_situation, category, u, prompt_strategy, prompt_selection, ['mindful']) | |
| confident_output = generate_feedback(current_situation, category, u, prompt_strategy, prompt_selection, ['confident']) | |
| strategy = strategy_list[0] | |
| strategy_rating = strategy_output.split(f'{strategy} Rating:')[1].split('###')[0].strip() | |
| strategy_suggestion = strategy_output.split(f'{strategy} Rating:')[1].split('Suggestion for improvement: ')[1].split('###')[0].strip() | |
| strategy_feedback = strategy_rating + '<br/>' + strategy_suggestion | |
| mindful_rating = mindful_output.split('Mindful Rating:')[1].split('###')[0].strip() | |
| if mindful_rating != 'Yes' and mindful_rating != 'No': | |
| mindful_feedback = '' | |
| elif mindful_rating == 'Yes': | |
| mindful_feedback = "Mindfulness check: Well done! 👍 " | |
| else: | |
| mindful_feedback = "Mindfulness check: " + mindful_output.split('Mindful Rating:')[1].split('Suggestion for improvement: ')[1].split('###')[0].strip() | |
| confident_rating = confident_output.split('Confident Rating:')[1].split('###')[0].strip() | |
| if confident_rating != 'Yes' and confident_rating != 'No': | |
| confident_feedback = '' | |
| elif confident_rating == 'Yes': | |
| confident_feedback = "Confidence check: Well done! 👍 " | |
| else: | |
| confident_feedback = "Confidence check: " + confident_output.split('Confident Rating:')[1].split('Suggestion for improvement: ')[1].split('###')[0].strip() | |
| print (strategy_output, mindful_output, confident_output) | |
| all_concat = strategy_feedback + '</br> </br>' + mindful_feedback + '</br>' + confident_feedback | |
| return all_concat | |
| def generate_skill_suggestion(situation, input_history, demonstration_mode): | |
| # format of input_history: | |
| # [{"role":"system", "content": system_input}] + input_history + [{"role":"user", "content":new_input}] | |
| # | |
| if len(input_history) == 0: | |
| return "Describe", "You can start the conversation by describing the situation." | |
| if input_history[-1]['role'] == 'system': | |
| previous_partner_message = input_history[-1]['content'] | |
| previous_client_message = input_history[-2]['content'] | |
| else: | |
| previous_partner_message = input_history[-2]['content'] | |
| previous_client_message = input_history[-1]['content'] | |
| prompt_other_message = f'Conversation context: {situation}. I received the response: "{previous_partner_message}" What DEAR MAN strategy should I use next?' | |
| prompt = prompt_other_message | |
| system_prompt_one = "Please suggest the top DEAR MAN skill to use in the next response: describe, express, assert, reinforce, negotiate. In the first line, output only the name of the skill immediately. In the second line, suggestion reason why use this skill. Address the reason in second person perspective. Separate the lines with ###. YOU MUST FOLLOW THIS FORMAT." | |
| prompt = [{"role":"system", "content":system_prompt_one}, {"role":"user", "content": prompt_other_message}] | |
| current_tries=0 | |
| while current_tries <= MAX_RETRIES: | |
| try: | |
| response = openai.ChatCompletion.create( | |
| model=MODEL_NAME, | |
| messages = prompt, | |
| max_tokens = 256, | |
| temperature = 0, | |
| ) | |
| # print (prompt)k | |
| curr_response_str = response['choices'][0]['message']['content'].replace('\n', ' ').strip() | |
| print (curr_response_str, '\t') | |
| # print ('Suggestion: ', all_strategy[f'suggestion_{strategy}'][m], '\t') | |
| # model_output.append(curr_response_str) | |
| break | |
| except Exception as e: | |
| print('error: ', str(e)) | |
| print('response retrying') | |
| current_tries += 1 | |
| if current_tries > MAX_RETRIES: | |
| break | |
| time.sleep(5) | |
| # suggested_skills = curr_response_str.split('###')[0].split(',') | |
| # suggested_skills = [skill.strip() for skill in suggested_skills] | |
| # suggested_skills = [skill[0].upper() + skill[1:].lower() for skill in suggested_skills] | |
| # suggest_reason = curr_response_str.split('###')[1:] | |
| # # convert suggest_reason to a string | |
| # suggest_reason_str = ' '.join(suggest_reason) | |
| # print (suggest_reason, suggest_reason_str) | |
| # if len(suggested_skills) == 2: | |
| # print (suggested_skills[0], suggested_skills[1]) | |
| # return suggested_skills[0], suggested_skills[1], suggest_reason_str | |
| suggested_skills = curr_response_str.split('###')[0].split(',') | |
| suggested_skills = [skill.strip() for skill in suggested_skills] | |
| suggested_skills = [skill[0].upper() + skill[1:].lower() for skill in suggested_skills] | |
| suggest_reason = curr_response_str.split('###')[1:] | |
| suggest_reason_str = ' '.join(suggest_reason) | |
| print (suggested_skills, suggest_reason_str) | |
| if len(suggested_skills) == 1: | |
| return suggested_skills[0], suggest_reason_str | |
| else: | |
| print (suggested_skills) | |
| print ("Suggestion format is wrong") | |
| return 'Describe', "During the conversation, it is helpful to reground the fact." |