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 + '
' + 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 + '

' + mindful_feedback + '
' + 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."