import sys import os from datetime import datetime import json import uuid from pathlib import Path from huggingface_hub import CommitScheduler, login from datasets import load_dataset import gradio as gr import markdown from together import Together ROOT_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "./") sys.path.append(ROOT_FILE) from components.induce_personality import construct_big_five_words from components.chat_conversation import ( # format_message_history, format_user_message, format_context, gradio_to_huggingface_message, huggingface_to_gradio_message, # get_system_instruction, prepare_tokenizer, # format_rag_context, conversation_window, generate_response_local_api, generate_response_together_api, generate_response_debugging, ) from components.constant import ( CONV_WINDOW, API_URL, ) from components.induce_personality import ( build_personality_prompt, ) LOG_DIR = os.path.join(ROOT_FILE, "log/api/") if os.path.exists(LOG_DIR) is False: os.makedirs(LOG_DIR) # Load Static Files STATIC_FILE = os.path.join(ROOT_FILE, "_static") LOG_DIR = os.path.join(ROOT_FILE, "log/test_session/") INSTRUCTION_PAGE_FILE = os.path.join(STATIC_FILE, "html/instruction_page.html") USER_NARRATIVE_FILE = os.path.join(STATIC_FILE, "html/user_narrative.html") PREFERENCE_ELICITATION_TASK_FILE = os.path.join(STATIC_FILE, "html/system_instruction_preference_elicitation.html") EVALUATION_INSTRUCTION_FILE = os.path.join(STATIC_FILE, "html/evaluation_instruction.html") GENERAL_INSTRUCTION_FILE = os.path.join(STATIC_FILE, "html/general_instruction.html") FINAL_EVALUATION_FILE = os.path.join(STATIC_FILE, "html/final_evaluation.html") SYSTEM_INSTRUCTION_PERSONALIZATION_FILE = os.path.join(STATIC_FILE, "txt/system_instruction_personalization.txt") SYSTEM_INSTRUCTION_NON_PERSONALIZATION_FILE = os.path.join( STATIC_FILE, "txt/system_instruction_non_personalization.txt" ) SYSTEM_INSTRUCTION_PERSONALITY_FILE = os.path.join(STATIC_FILE, "txt/system_instruction_personality.txt") SYSTEM_INSTRUCTION_PREFERENCE_ELICITATION_FILE = os.path.join( STATIC_FILE, "txt/system_instruction_preference_elicitation.txt" ) SYSTEM_INSTRUCTION_PREFERENCE_ELICITATION_PERSONALITY_FILE = os.path.join( STATIC_FILE, "txt/system_instruction_preference_elicitation_personality.txt" ) SUMMARIZATION_PROMPT_FILE = os.path.join(STATIC_FILE, "txt/system_summarization_user_preference_elicitation.txt") PERSONALITY_EXT_FILE = os.path.join(STATIC_FILE, "txt/personality_ext.txt") PERSONALITY_INT_FILE = os.path.join(STATIC_FILE, "txt/personality_int.txt") uuid_this_session = str(uuid.uuid4()) system_order = "first" feedback_file_interaction = Path("user_feedback/") / f"interaction_{uuid_this_session}_{system_order}.json" feedback_file_summarization = Path("user_feedback/") / f"summarization_{uuid_this_session}_{system_order}.json" feedback_file_round_evaluation = Path("user_feedback/") / f"round_evaluation_{uuid_this_session}_{system_order}.json" feedback_file_final_ranking = Path("user_feedback/") / f"final_ranking_{uuid_this_session}_{system_order}.json" feedback_file_final_survey = Path("user_feedback/") / f"final_survey_{uuid_this_session}_{system_order}.json" feedback_folder = feedback_file_interaction.parent feedback_folder.mkdir(parents=True, exist_ok=True) # Ensure the directory exists scheduler = CommitScheduler( repo_id=os.getenv("LOGGING_FILE"), repo_type="dataset", folder_path=feedback_folder, path_in_repo="data", token=os.getenv("HUGGINGFACE_HUB_TOKEN"), every=10, ) # Function to save user feedback def save_feedback(user_id: str, uuid: str, type: str, value, feedback_file) -> None: """ Append input/outputs and user feedback to a JSON Lines file using a thread lock to avoid concurrent writes from different users. """ timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with scheduler.lock: with feedback_file.open("a") as f: f.write( json.dumps({"user_id": user_id, "uuid": uuid, "timestamp": timestamp, "type": type, "value": value}) ) f.write("\n") # Load the required static content from files def load_static_content(file_path): with open(file_path, "r") as f: return f.read() def ensure_directory_exists(directory_path): """Ensures the given directory exists; creates it if it does not.""" if not os.path.exists(directory_path): os.makedirs(directory_path) INSTRUCTION_PAGE = load_static_content(INSTRUCTION_PAGE_FILE) EVALUATION_INSTRUCTION = load_static_content(EVALUATION_INSTRUCTION_FILE) GENERAL_INSTRUCTION = load_static_content(GENERAL_INSTRUCTION_FILE) USER_NARRATIVE = load_static_content(USER_NARRATIVE_FILE) PREFERENCE_ELICITATION_TASK = load_static_content(PREFERENCE_ELICITATION_TASK_FILE) FINAL_EVALUATION = load_static_content(FINAL_EVALUATION_FILE) SYSTEM_INSTRUCTION_PERSONALIZATION = load_static_content(SYSTEM_INSTRUCTION_PERSONALIZATION_FILE) SYSTEM_INSTRUCTION_NON_PERSONALIZATION = load_static_content(SYSTEM_INSTRUCTION_NON_PERSONALIZATION_FILE) SYSTEM_INSTRUCTION_PERSONALITY = load_static_content(SYSTEM_INSTRUCTION_PERSONALITY_FILE) SYSTEM_INSTRUCTION_PREFERENCE_ELICITATION = load_static_content(SYSTEM_INSTRUCTION_PREFERENCE_ELICITATION_FILE) SYSTEM_INSTRUCTION_PREFERENCE_ELICITATION_PERSONALITY = load_static_content( SYSTEM_INSTRUCTION_PREFERENCE_ELICITATION_PERSONALITY_FILE ) SUMMARIZATION_PROMPT = load_static_content(SUMMARIZATION_PROMPT_FILE) PERSONALITY_EXT = load_static_content(PERSONALITY_EXT_FILE) PERSONALITY_INT = load_static_content(PERSONALITY_INT_FILE) # Other constants FIRST_MESSAGE = "Hey" USER_PREFERENCE_SUMMARY = True DEBUG = False API_TYPE = "together" assert API_TYPE in ["together", "local", "debug"], "The API should be either 'together' or 'local'" if API_TYPE == "together": TOGETHER_CLIENT = Together(api_key=os.getenv("TOGETHER_API_KEY")) def generate_username_pwd_list(data): user_list = [] demo_list = [] for index, row in data.iterrows(): user_list.append((row["user"], str(row["pwd"]))) demo_list.append((row["demo"], str(row["pwd"]))) return user_list, demo_list def load_username_and_pwd(): login(token=os.environ["HUGGINGFACE_HUB_TOKEN"]) dataset = load_dataset(os.getenv("USER_PWD_FILE")) df = dataset["train"].to_pandas() user_list, demo_list = generate_username_pwd_list(df) return user_list, demo_list def get_context_list(synthetic_data_path): # Load data from the synthetic data file with open(synthetic_data_path, "r") as f: data = [json.loads(line) for line in f] return data def add_ticker_prefix(ticker_list, context_list): res = [] for ticker, context in zip(ticker_list, context_list): res.append(f"{ticker}: {context}") return res def build_raw_context_list(context_dict): return context_dict["data"] def build_context(context_dict): return [build_context_element(context) for context in context_dict["data"]] def build_context_element(context): # [{topic: ex, data: {}}, {..}, ..] # Extract information from the context ticker = context["ticker"] sector = context["sector"] business_summary = context["business_summary"] name = context["short_name"] stock_price = context["price_data"] earning = context["earning_summary"] beta = context["beta"] # Build the context string stock_candidate = f"Stock Candidate: {name}" stock_info = f"Stock Information: \nIndustry - {sector}, \nBeta (risk indicator) - {beta}, \nEarning Summary - {earning}\n, 2023 Monthly Stock Price - {stock_price}\n, Business Summary - {business_summary}" context_list = [stock_candidate, stock_info] # Combine all parts into a single string return "\n".join(context_list) def get_user_narrative_html(user_narrative): return USER_NARRATIVE.replace("{user_narrative}", user_narrative).replace("\n", "
") def get_user_narrative_from_raw(raw_narrative): return get_user_narrative_html(markdown.markdown(raw_narrative.replace("\n", "
"))) def get_task_instruction_for_user(context): ticker_name = context["short_name"] user_narrative = context["user_narrative"] user_narrative = user_narrative.replace("\n", "
") html_user_narrative = markdown.markdown(user_narrative) general_instruction = GENERAL_INSTRUCTION round_instruction = f"""

Round Info

Stock
This Round's Stock: {ticker_name}
User Narrative
{html_user_narrative}
""" return general_instruction, round_instruction def display_system_instruction_with_html( system_instruction, ): html_system_instruction = f"""

{system_instruction}

""" return html_system_instruction def log_action(user_id, tab_name, action, details): """ Log actions for each tab (stock). """ log_file_dir = os.path.join(LOG_DIR, f"{user_id}") if os.path.exists(log_file_dir) is False: os.makedirs(log_file_dir) log_file = os.path.join(log_file_dir, f"{tab_name}.txt") with open(log_file, "a") as f: f.write(f"Action: {action} | Details: {details}\n") def add_user_profile_to_system_instruction( user_id, system_instruction, user_preference_elicitation_data, summary, terminator ): exp_id = int(user_id.split("_")[-3]) # exp_id = 1 => No personalization if exp_id == 1: return system_instruction if summary: if user_preference_elicitation_data["summary_history"] == "": # Format prompt summarization_prompt = SUMMARIZATION_PROMPT + "\nPrevious Conversations: {}".format( user_preference_elicitation_data["history"] ) summarization_instruction = [{"role": "system", "content": summarization_prompt}] if API_TYPE == "local": summ, _ = generate_response_local_api(summarization_instruction, terminator, 512, API_URL) elif API_TYPE == "together": summ, _ = generate_response_together_api(summarization_instruction, 512, TOGETHER_CLIENT) else: summ, _ = generate_response_debugging(summarization_instruction) user_preference_elicitation_data["summary_history"] = summ # log_action(user_id, "Prompt", "Preference Elicitation Summarization", summ) save_feedback( user_id, uuid_this_session, "preference_elicitation_summarization", {"summarization": summ}, feedback_file_summarization, ) system_instruction += f"\nUser Profile collected in the previous conversations: {user_preference_elicitation_data['summary_history']}\n" else: system_instruction += ( f"\nUser Profile collected in the previous conversations: {user_preference_elicitation_data['history']}\n" ) return system_instruction def likert_evaluation(content): return gr.Radio( [1, 2, 3, 4, 5, 6, 7], label=f"{content}", show_label=True, ) def reorder_list_based_on_user_in_narrative_id(user_in_narrative_id, target_list): # user_in_narrative random_order = {"0": [3, 2, 1, 0], "1": [1, 0, 3, 2], "2": [2, 1, 0, 3], "3": [1, 3, 2, 0], "4": [0, 3, 1, 2]} user_in_narrative_random = random_order[user_in_narrative_id] return [target_list[i] for i in user_in_narrative_random] def create_demo(): global context_info_list, terminator def tab_creation_exploration_stage(order, comp, context): english_order = ["1", "2", "3", "4", "5"] with gr.Tab(f"{english_order[order]}-1:Discuss"): general_instruction = gr.HTML(label="General Instruction") with gr.Row(): with gr.Column(): with gr.Row(): round_instruction = gr.HTML(label="Round Instruction") with gr.Column(): with gr.Row(): chatbot = gr.Chatbot(height=600) with gr.Row(): start_conversation = gr.Button(value="Start Conversation") with gr.Row(): msg = gr.Textbox(scale=1, label="User Input") with gr.Row(): msg_button = gr.Button(value="Send This Message to Advisor", interactive=False) continue_button = gr.Button(value="Show More of the Advisor’s Answer", interactive=False) with gr.Row(): clear = gr.ClearButton([msg, chatbot]) with gr.Tab(f"{english_order[order]}-2:Eval"): with gr.Row(): gr.HTML(value=EVALUATION_INSTRUCTION) with gr.Row(): likelihood = gr.Radio( [1, 2, 3, 4, 5, 6, 7], label="I am likely to purchase the stock (1 = Strongly Disagree, 7 = Strongly Agree)", show_label=True, ) reason = gr.Textbox( scale=1, label="Reason for Your Choice (Explain Your Reasoning & Highlight Useful Parts of Conversation)", lines=5, ) with gr.Row(): confidence = gr.Radio( [1, 2, 3, 4, 5, 6, 7], label="I am confident in my decision (1 = Strongly Disagree, 7 = Strongly Agree)", show_label=True, ) familiarity = gr.Radio( [1, 2, 3, 4, 5, 6, 7], label="What was your level of familiarity with the candidate stock before the interaction? (1 = Not Familiar, 7 = Very Familiar)", ) with gr.Row(): textbox = gr.HTML() evaluation_send_button = gr.Button(value="Send: Evaluation") return { "comp": comp, "system_instruction_context": context, "start_conversation": start_conversation, "msg_button": msg_button, "continue_button": continue_button, "chatbot": chatbot, "msg": msg, "reason": reason, "likelihood": likelihood, "confidence": confidence, "familiarity": familiarity, "evaluation_send_button": evaluation_send_button, "general_instruction": general_instruction, "round_instruction": round_instruction, "textbox": textbox, } def tab_creation_preference_stage(): with gr.Row(): gr.HTML(value=PREFERENCE_ELICITATION_TASK, label="Preference Elicitation Task") with gr.Row(): with gr.Column(): user_narrative = gr.HTML(label="User Narrative") with gr.Column(): with gr.Row(): elicitation_chatbot = gr.Chatbot(height=600) with gr.Row(): start_conversation = gr.Button(value="Start Conversation") with gr.Row(): msg = gr.Textbox(scale=1, label="User Input") with gr.Row(): msg_button = gr.Button(value="Send This Message to Advisor", interactive=False) continue_button = gr.Button(value="Show More of the Advisor’s Answer", interactive=False) return { "start_conversation": start_conversation, "msg_button": msg_button, "continue_button": continue_button, "msg": msg, "elicitation_chatbot": elicitation_chatbot, "user_narrative": user_narrative, } def tab_final_evaluation(): with gr.Row(): gr.HTML(value=FINAL_EVALUATION) with gr.Row(): gr.HTML(value="

Rank the individual stocks below according to your desire to invest in each one.

") with gr.Row(): ranking_first_comp = gr.Dropdown(choices=[1, 2, 3, 4]) ranking_second_comp = gr.Dropdown(choices=[1, 2, 3, 4]) ranking_third_comp = gr.Dropdown(choices=[1, 2, 3, 4]) ranking_fourth_comp = gr.Dropdown(choices=[1, 2, 3, 4]) with gr.Row(): gr.HTML( value='

Choose how strongly you agree with each statement about the advisor (1 for Strongly Disagree, 7 for Strongly Agree).

' ) with gr.Row(): perceived_personalization = likert_evaluation("The advisor understands my needs") emotional_trust = likert_evaluation("I feel content about relying on this advisor for my decisions") with gr.Row(): trust_in_competence = likert_evaluation("The advisor has good knowledge of the stock") intention_to_use = likert_evaluation( "I am willing to use this advisor as an aid to help with my decision about which stock to purchase" ) with gr.Row(): usefulness = likert_evaluation("The advisor gave me good suggestions") overall_satisfaction = likert_evaluation("Overall, I am satisfied with the advisor") with gr.Row(): providing_information = likert_evaluation("The advisor provides the financial knowledge needed") with gr.Row(): textbox = gr.HTML() submit_ranking = gr.Button(value="Submit Final Evaluation") return { "first": ranking_first_comp, "second": ranking_second_comp, "third": ranking_third_comp, "fourth": ranking_fourth_comp, "evaluators": { "perceived_personalization": perceived_personalization, "emotional_trust": emotional_trust, "trust_in_competence": trust_in_competence, "intention_to_use": intention_to_use, "usefulness": usefulness, "overall_satisfaction": overall_satisfaction, "providing_information": providing_information, }, "submit_ranking": submit_ranking, "text_box": textbox, } def click_control_exploration_stage( tabs, user_id, tab_session, user_preference_elicitation_session, system_description_without_context ): ( comp, system_instruction_context, start_conversation, msg_button, continue_button, chatbot, msg, reason, likelihood, confidence, familiarity, evaluation_send_button, textbox, ) = ( tabs["comp"], tabs["system_instruction_context"], tabs["start_conversation"], tabs["msg_button"], tabs["continue_button"], tabs["chatbot"], tabs["msg"], tabs["reason"], tabs["likelihood"], tabs["confidence"], tabs["familiarity"], tabs["evaluation_send_button"], tabs["textbox"], ) system_instruction = "" start_conversation.click( lambda user_id, tab_session, history, comp, user_preference_elicitation_session, system_description_without_context, system_instruction_context: respond_start_conversation( user_id, tab_session, history, system_instruction, comp, user_preference_elicitation_data=user_preference_elicitation_session, system_description_without_context=system_description_without_context, system_instruction_context=system_instruction_context, ), [ user_id, tab_session, chatbot, comp, user_preference_elicitation_session, system_description_without_context, system_instruction_context, ], [tab_session, chatbot, start_conversation, msg_button, continue_button], ) msg_button.click( lambda user_id, tab_session, message, history, comp, user_preference_elicitation_session, system_description_without_context, system_instruction_context: respond( user_id, tab_session, message, tab_session["history"], system_instruction, comp, user_preference_elicitation_data=user_preference_elicitation_session, system_description_without_context=system_description_without_context, system_instruction_context=system_instruction_context, ), [ user_id, tab_session, msg, chatbot, comp, user_preference_elicitation_session, system_description_without_context, system_instruction_context, ], [tab_session, msg, chatbot], ) continue_button.click( lambda user_id, tab_session, history, comp, user_preference_elicitation_session, system_description_without_context, system_instruction_context: respond_continue( user_id, tab_session, tab_session["history"], system_instruction, comp, user_preference_elicitation_data=user_preference_elicitation_session, system_description_without_context=system_description_without_context, system_instruction_context=system_instruction_context, ), [ user_id, tab_session, chatbot, comp, user_preference_elicitation_session, system_description_without_context, system_instruction_context, ], [tab_session, chatbot], ) evaluation_send_button.click( lambda user_id, comp, tab_session, reason, likelihood, confidence, familiarity, evaluation_send_button, textbox: respond_evaluation( user_id, tab_session, { "reason": reason, "likelihood": likelihood, "confidence": confidence, "familiarity": familiarity, }, comp, evaluation_send_button, textbox, ), [ user_id, comp, tab_session, reason, likelihood, confidence, familiarity, evaluation_send_button, textbox, ], [tab_session, reason, likelihood, confidence, familiarity, evaluation_send_button, textbox], ) def click_control_preference_stage( tabs, user_id, user_preference_elicitation_session, system_description_user_elicitation ): ( start_conversation, msg_button, continue_button, elicitation_chatbot, msg, ) = ( tabs["start_conversation"], tabs["msg_button"], tabs["continue_button"], tabs["elicitation_chatbot"], tabs["msg"], ) # nonlocal user_id start_conversation.click( lambda user_id, user_preference_elicitation_data, history, system_description_user_elicitation: respond_start_conversation( user_id, user_preference_elicitation_data, history, system_description_user_elicitation, user_elicitation=True, ), [user_id, user_preference_elicitation_session, elicitation_chatbot, system_description_user_elicitation], [user_preference_elicitation_session, elicitation_chatbot, start_conversation, msg_button, continue_button], ) msg_button.click( lambda user_id, tab_data, message, history, system_description_user_elicitation: respond( user_id, tab_data, message, tab_data["history"], system_description_user_elicitation, user_elicitation=True, ), [ user_id, user_preference_elicitation_session, msg, elicitation_chatbot, system_description_user_elicitation, ], [user_preference_elicitation_session, msg, elicitation_chatbot], ) continue_button.click( lambda user_id, tab_data, history, system_description_user_elicitation: respond_continue( user_id, tab_data, tab_data["history"], system_description_user_elicitation, user_elicitation=True, ), [user_id, user_preference_elicitation_session, elicitation_chatbot, system_description_user_elicitation], [user_preference_elicitation_session, elicitation_chatbot], ) def click_control_final_evaluation(tabs, user_id, first_comp, second_comp, third_comp, fourth_comp, evaluators): ( ranking_first_comp, ranking_second_comp, ranking_third_comp, ranking_fourth_comp, ) = ( tabs["first"], tabs["second"], tabs["third"], tabs["fourth"], ) ( perceived_personalization, emotional_trust, trust_in_competence, intention_to_use, usefulness, overall_satisfaction, providing_information, ) = ( evaluators["perceived_personalization"], evaluators["emotional_trust"], evaluators["trust_in_competence"], evaluators["intention_to_use"], evaluators["usefulness"], evaluators["overall_satisfaction"], evaluators["providing_information"], ) result_textbox = tabs["text_box"] submit_ranking = tabs["submit_ranking"] submit_ranking.click( lambda user_id, first_comp, ranking_first_comp, second_comp, ranking_second_comp, third_comp, ranking_third_comp, fourth_comp, ranking_fourth_comp, perceived_personalization, emotional_trust, trust_in_competence, intention_to_use, usefulness, overall_satisfaction, providing_information, submit_ranking: respond_final_ranking( user_id, first_comp, ranking_first_comp, second_comp, ranking_second_comp, third_comp, ranking_third_comp, fourth_comp, ranking_fourth_comp, perceived_personalization, emotional_trust, trust_in_competence, intention_to_use, usefulness, overall_satisfaction, providing_information, submit_ranking, ), # Input components (names and rankings) [ user_id, first_comp, ranking_first_comp, second_comp, ranking_second_comp, third_comp, ranking_third_comp, fourth_comp, ranking_fourth_comp, perceived_personalization, emotional_trust, trust_in_competence, intention_to_use, usefulness, overall_satisfaction, providing_information, submit_ranking, ], # Output component(s) where you want the result to appear, e.g., result_textbox [result_textbox, submit_ranking], ) def respond( user_id, tab_data, message, history, system_instruction, tab_name=None, user_elicitation=False, user_preference_elicitation_data=None, system_description_without_context=None, system_instruction_context=None, ): """ Return: msg chat_history retrieved_passage rewritten_query """ assert ( tab_name is not None or user_elicitation is True ), "Tab name is required for the start of the conversation unless it is not preference elicitation." # Add user profile to system instruction if system_description_without_context is not None and system_instruction_context is not None: system_instruction = system_description_without_context + "\n" + system_instruction_context if not user_elicitation: system_instruction = add_user_profile_to_system_instruction( user_id, system_instruction, user_preference_elicitation_data, summary=USER_PREFERENCE_SUMMARY, terminator=terminator, ) # From string to list [{"role":"user", "content": message}, ...] history = gradio_to_huggingface_message(history) # We can implement context window here as we need all the system interaction. We can cut some of the early interactions if needed. history = conversation_window(history, CONV_WINDOW) # Add system instruction to the history history = format_context(system_instruction, history) # Add user message to the history history_with_user_utterance = format_user_message(message, history) # Call API instead of locally handle it if API_TYPE == "local": outputs_text, history = generate_response_local_api(history_with_user_utterance, terminator, 128, API_URL) elif API_TYPE == "together": outputs_text, history = generate_response_together_api(history_with_user_utterance, 128, TOGETHER_CLIENT) else: outputs_text, history = generate_response_debugging(history_with_user_utterance) # exclude system interaction and store the others in the history history = huggingface_to_gradio_message(history) if tab_name is not None: # Log the user message and response save_feedback( user_id, uuid_this_session, "interaction", {"type": tab_name, "role": "user", "content": message}, feedback_file_interaction, ) save_feedback( user_id, uuid_this_session, "interaction", {"type": tab_name, "role": "assistant", "content": outputs_text}, feedback_file_interaction, ) # log_action(user_id, tab_name, "User Message", message) # log_action(user_id, tab_name, "Response", outputs_text) # Store the updated history for this tab tab_data["history"] = history if user_elicitation: save_feedback( user_id, uuid_this_session, "Interaction", {"type": "user_elicitation", "role": "user", "content": message}, feedback_file_interaction, ) save_feedback( user_id, uuid_this_session, "Interaction", {"type": "user_elicitation", "role": "assistant", "content": outputs_text}, feedback_file_interaction, ) # log_action(user_id, "User_Elicitation", "User Message", message) # log_action(user_id, "User_Elicitation", "Response", outputs_text) tab_data["history"] = history return tab_data, "", history def respond_start_conversation( user_id, tab_data, history, system_instruction, tab_name=None, user_elicitation=False, user_preference_elicitation_data=None, system_description_without_context=None, system_instruction_context=None, ): assert ( tab_name is not None or user_elicitation is True ), "Tab name is required for the start of the conversation unless it is not preference elicitation." if system_description_without_context is not None and system_instruction_context is not None: system_instruction = system_description_without_context + "\n" + system_instruction_context if not user_elicitation: system_instruction = add_user_profile_to_system_instruction( user_id, system_instruction, user_preference_elicitation_data, summary=USER_PREFERENCE_SUMMARY, terminator=terminator, ) history = gradio_to_huggingface_message(history) history = format_context(system_instruction, history) first_message = FIRST_MESSAGE history_with_user_utterance = format_user_message(first_message, history) max_length = 128 if user_elicitation else 256 if API_TYPE == "local": outputs_text, history = generate_response_local_api( history_with_user_utterance, terminator, max_length, API_URL ) elif API_TYPE == "together": outputs_text, history = generate_response_together_api( history_with_user_utterance, max_length, TOGETHER_CLIENT ) else: outputs_text, history = generate_response_debugging(history_with_user_utterance) # Format history = huggingface_to_gradio_message(history) if tab_name is not None: # Log the user message and response save_feedback( user_id, uuid_this_session, "interaction", {"type": tab_name, "role": "user", "content": first_message}, feedback_file_interaction, ) save_feedback( user_id, uuid_this_session, "interaction", {"type": tab_name, "role": "assistant", "content": outputs_text}, feedback_file_interaction, ) # log_action(user_id, tab_name, "User Message", first_message) # log_action(user_id, tab_name, "Response", outputs_text) # Store the updated history for this tab tab_data["history"] = history if user_elicitation: save_feedback( user_id, uuid_this_session, "interaction", {"type": "user_elicitation", "role": "user", "content": first_message}, feedback_file_interaction, ) save_feedback( user_id, uuid_this_session, "Interaction", {"type": "user_elicitation", "role": "assistant", "content": outputs_text}, feedback_file_interaction, ) tab_data["history"] = history return ( tab_data, history, gr.Button(value="Start Conversation", interactive=False), gr.Button(value="Send This Message to Advisor", interactive=True), gr.Button(value="Show More of the Advisor’s Answer", interactive=True), ) def respond_continue( user_id, tab_data, history, system_instruction, tab_name=None, user_elicitation=False, user_preference_elicitation_data=None, system_description_without_context=None, system_instruction_context=None, ): assert ( tab_name is not None or user_elicitation is True ), "Tab name is required for the start of the conversation." # Add user profile to system instruction if system_description_without_context is not None and system_instruction_context is not None: system_instruction = system_description_without_context + "\n" + system_instruction_context if not user_elicitation: system_instruction = add_user_profile_to_system_instruction( user_id, system_instruction, user_preference_elicitation_data, summary=USER_PREFERENCE_SUMMARY, terminator=terminator, ) message = "continue" history = gradio_to_huggingface_message(history) history = conversation_window(history, CONV_WINDOW) history = format_context(system_instruction, history) history_with_user_utterance = format_user_message(message, history) if API_TYPE == "local": outputs_text, history = generate_response_local_api(history_with_user_utterance, terminator, 128, API_URL) elif API_TYPE == "together": outputs_text, history = generate_response_together_api(history_with_user_utterance, 128, TOGETHER_CLIENT) else: outputs_text, history = generate_response_debugging(history_with_user_utterance) history = huggingface_to_gradio_message(history) if tab_name is not None: save_feedback( user_id, uuid_this_session, "interaction", { "type": tab_name, "role": "user", "content": message, }, feedback_file_interaction, ) save_feedback( user_id, uuid_this_session, "interaction", {"type": tab_name, "role": "assistant", "content": outputs_text}, feedback_file_interaction, ) # Update history for this tab tab_data["history"] = history if user_elicitation: save_feedback( user_id, uuid_this_session, "interaction", {"type": "user_elicitation", "role": "user", "content": message}, feedback_file_interaction, ) save_feedback( user_id, uuid_this_session, "interaction", {"type": "user_elicitation", "role": "assistant", "content": outputs_text}, feedback_file_interaction, ) tab_data["history"] = history return tab_data, history def respond_evaluation(user_id, tab_data, evals, tab_name, evaluation_send_button, textbox): # dropdown, readon_button, multi-evaluator if evals["likelihood"] is None or evals["confidence"] is None or evals["familiarity"] is None: return ( tab_data, evals["reason"], evals["likelihood"], evals["confidence"], evals["familiarity"], evaluation_send_button, """
Please make sure that you answer all the questions.
""", ) else: save_feedback( user_id, uuid_this_session, "round_evaluation", {**evals, "company": tab_name}, feedback_file_round_evaluation, ) # log_action(user_id, tab_name, "Round Evaluation", "Following") # for key, value in evals.items(): # log_action(user_id, tab_name, key, value) # Store the reason for this tab tab_data["multi_evaluator"] = evals evaluation_send_button = gr.Button(value="Evaluation receirved", interactive=False) return ( tab_data, evals["reason"], evals["likelihood"], evals["confidence"], evals["familiarity"], evaluation_send_button, """
Thank you for submitting your evaluation. You may proceed to the next tab.
""", ) def respond_final_ranking( user_id, first_comp, ranking_first_comp, second_comp, ranking_second_comp, third_comp, ranking_third_comp, fourth_comp, ranking_fourth_comp, perceived_personalization, emotional_trust, trust_in_competence, intention_to_use, usefulness, overall_satisfaction, providing_information, submit_ranking, ): # make sure that they are not the same ranking_list = [ ranking_first_comp, ranking_second_comp, ranking_third_comp, ranking_fourth_comp, ] if len(set(ranking_list)) != len(ranking_list): return ( """
Please make sure that you are not ranking the same stock multiple times.
""", submit_ranking, ) if any( var is None for var in [ perceived_personalization, emotional_trust, trust_in_competence, intention_to_use, usefulness, overall_satisfaction, providing_information, ] ): return ( """
Please make sure that you answer all the statements.
""", submit_ranking, ) else: save_feedback( user_id, uuid_this_session, "final_ranking", { "comp_order": [first_comp, second_comp, third_comp, fourth_comp], "ranking": ranking_list, }, feedback_file_final_ranking, ) save_feedback( user_id, uuid_this_session, "final_ranking_survey", { "perceived_personalization": perceived_personalization, "emotional_trust": emotional_trust, "trust_in_competence": trust_in_competence, "intention_to_use": intention_to_use, "usefulness": usefulness, "overall_satisfaction": overall_satisfaction, "providing_information": providing_information, }, feedback_file_final_survey, ) submit_ranking = gr.Button(value="Final evaluaiotn received", interactive=False) return ( """
Thank you for participating in the experiment. This concludes the session. You may now close the tab.
""", submit_ranking, ) def get_context(index, raw_context_list, stock_context_list): comp = raw_context_list[index]["short_name"] context = stock_context_list[index] general_instruction, round_instruction = get_task_instruction_for_user(raw_context_list[index]) return comp, context, general_instruction, round_instruction def set_user_id(request: gr.Request): user_id = request.username user_in_narrative_id = user_id.split("_")[-1] narrative_id = user_id.split("_")[-2] experiment_id = user_id.split("_")[-3] return user_id, user_in_narrative_id, narrative_id, experiment_id def get_inst_without_context(experiment_id): # experiment_id = 1 => personalization # experiment_id = 2 => no personalization # experiment_id == 3 => ext personality # experiment_id == 4 => int personality if experiment_id == "0": return SYSTEM_INSTRUCTION_PERSONALIZATION elif experiment_id == "1": return SYSTEM_INSTRUCTION_NON_PERSONALIZATION elif experiment_id == "2": return SYSTEM_INSTRUCTION_PERSONALITY.format(personality=PERSONALITY_EXT) elif experiment_id == "3": return SYSTEM_INSTRUCTION_PERSONALITY.format(personality=PERSONALITY_INT) def get_user_preference_elicitation(experiment_id): if experiment_id == "0" or experiment_id == "1": return SYSTEM_INSTRUCTION_PREFERENCE_ELICITATION elif experiment_id == "2": return SYSTEM_INSTRUCTION_PREFERENCE_ELICITATION_PERSONALITY.format(personality=PERSONALITY_EXT) elif experiment_id == "3": return SYSTEM_INSTRUCTION_PREFERENCE_ELICITATION_PERSONALITY.format(personality=PERSONALITY_INT) def get_stock_related_context(narrative_id, user_in_narrative_id): raw_context_list = build_raw_context_list(context_info_list[int(narrative_id)]) stock_context_list = build_context(context_info_list[int(narrative_id)]) raw_context_list = reorder_list_based_on_user_in_narrative_id(user_in_narrative_id, raw_context_list) stock_context_list = reorder_list_based_on_user_in_narrative_id(user_in_narrative_id, stock_context_list) return raw_context_list, stock_context_list def set_initial_values(request: gr.Request): # Set user specific information (Session State) user_id, user_in_narrative_id, narrative_id, experiment_id = set_user_id(request) # System instruction without prompt system_description_without_context = get_inst_without_context(experiment_id) # user_preference_elicitation system_description_user_elicitation = get_user_preference_elicitation(experiment_id) # Stock related context raw_context_list, stock_context_list = get_stock_related_context(narrative_id, user_in_narrative_id) # User Narrative user_narrative = get_user_narrative_from_raw(raw_context_list[0]["user_narrative"]) # Tab Context first_comp, first_context, first_general_instruction, first_round_instruction = get_context( 0, raw_context_list, stock_context_list ) second_comp, second_context, second_general_instruction, second_round_instruction = get_context( 1, raw_context_list, stock_context_list ) third_comp, third_context, third_general_instruction, third_round_instruction = get_context( 2, raw_context_list, stock_context_list ) fourth_comp, fourth_context, fourth_general_instruction, fourth_round_instruction = get_context( 3, raw_context_list, stock_context_list ) # Final Evaluation ranking_first_comp = gr.Dropdown(choices=[1, 2, 3, 4], label=first_comp) ranking_second_comp = gr.Dropdown(choices=[1, 2, 3, 4], label=second_comp) ranking_third_comp = gr.Dropdown(choices=[1, 2, 3, 4], label=third_comp) ranking_fourth_comp = gr.Dropdown(choices=[1, 2, 3, 4], label=fourth_comp) return ( user_id, user_in_narrative_id, narrative_id, experiment_id, system_description_without_context, system_description_user_elicitation, raw_context_list, stock_context_list, user_narrative, first_comp, first_context, first_general_instruction, first_round_instruction, second_comp, second_context, second_general_instruction, second_round_instruction, third_comp, third_context, third_general_instruction, third_round_instruction, fourth_comp, fourth_context, fourth_general_instruction, fourth_round_instruction, ranking_first_comp, ranking_second_comp, ranking_third_comp, ranking_fourth_comp, ) with gr.Blocks(title="RAG Chatbot Q&A", theme="Soft") as demo: # Set user specific information (Session State) user_id = gr.State() user_in_narrative_id = gr.State() narrative_id = gr.State() experiment_id = gr.State() system_description_without_context = gr.State() system_description_user_elicitation = gr.State() # Context data raw_context_list = gr.State() stock_context_list = gr.State() first_comp = gr.State() first_context = gr.State() second_comp = gr.State() second_context = gr.State() third_comp = gr.State() third_context = gr.State() fourth_comp = gr.State() fourth_context = gr.State() # Tab data if DEBUG: user_preference_elicitation_session = gr.State( value={ "history": "", "summary_history": """User Profile collected in the previous conversations: Based on our previous conversation, here's a summary of your investment preferences: # 1. **Preferred Industries:** You're interested in investing in the healthcare sector, without a specific preference for sub-industries such as pharmaceuticals, medical devices, biotechnology, or healthcare services. # 2. **Value vs. Growth Stocks:** You prefer growth stocks, which have the potential for high returns but may be riskier. # 3. **Dividend vs. Non-Dividend Stocks:** You're open to both dividend and non-dividend growth stocks, focusing on reinvesting profits for future growth. # 4. **Cyclical vs. Non-Cyclical Stocks:** You're interested in cyclical stocks, which are sensitive to economic fluctuations and tend to perform well during economic expansions.""", } ) else: user_preference_elicitation_session = gr.State(value={"history": "", "summary_history": ""}) first_comp_session = gr.State(value={"history": [], "selection": "", "reason": ""}) second_comp_session = gr.State(value={"history": [], "selection": "", "reason": ""}) third_comp_session = gr.State(value={"history": [], "selection": "", "reason": ""}) fourth_comp_session = gr.State(value={"history": [], "selection": "", "reason": ""}) # EXperiment Instruction with gr.Tab("Experiment Instruction") as instruction_tab: gr.HTML(value=INSTRUCTION_PAGE, label="Experiment Instruction") # User Preference Elicitation Tab with gr.Tab("Preference Elicitation Stage") as preference_elicitation_tab: user_preference_elicitation_tab = tab_creation_preference_stage() user_narrative = user_preference_elicitation_tab["user_narrative"] click_control_preference_stage( user_preference_elicitation_tab, user_id, user_preference_elicitation_session, system_description_user_elicitation, ) with gr.Tab("Financial Decision Stage") as financial_decision: # Experiment Tag first_tab = tab_creation_exploration_stage(0, first_comp, first_context) first_general_instruction, first_round_instruction = ( first_tab["general_instruction"], first_tab["round_instruction"], ) click_control_exploration_stage( first_tab, user_id, first_comp_session, user_preference_elicitation_session, system_description_without_context, ) second_tab = tab_creation_exploration_stage(1, second_comp, second_context) second_general_instruction, second_round_instruction = ( second_tab["general_instruction"], second_tab["round_instruction"], ) click_control_exploration_stage( second_tab, user_id, second_comp_session, user_preference_elicitation_session, system_description_without_context, ) third_tab = tab_creation_exploration_stage(2, third_comp, third_context) third_general_instruction, third_round_instruction = ( third_tab["general_instruction"], third_tab["round_instruction"], ) click_control_exploration_stage( third_tab, user_id, third_comp_session, user_preference_elicitation_session, system_description_without_context, ) fourth_tab = tab_creation_exploration_stage(3, fourth_comp, fourth_context) fourth_general_instruction, fourth_round_instruction = ( fourth_tab["general_instruction"], fourth_tab["round_instruction"], ) click_control_exploration_stage( fourth_tab, user_id, fourth_comp_session, user_preference_elicitation_session, system_description_without_context, ) with gr.Tab("Final Evaluation Stage") as final_evaluation: final_evaluation_tab = tab_final_evaluation() ( ranking_first_comp, ranking_second_comp, ranking_third_comp, ranking_fourth_comp, evaluators, ) = ( final_evaluation_tab["first"], final_evaluation_tab["second"], final_evaluation_tab["third"], final_evaluation_tab["fourth"], final_evaluation_tab["evaluators"], ) click_control_final_evaluation( final_evaluation_tab, user_id, first_comp, second_comp, third_comp, fourth_comp, evaluators ) demo.load( set_initial_values, inputs=None, outputs=[ user_id, user_in_narrative_id, narrative_id, experiment_id, system_description_without_context, system_description_user_elicitation, raw_context_list, stock_context_list, user_narrative, first_comp, first_context, first_general_instruction, first_round_instruction, second_comp, second_context, second_general_instruction, second_round_instruction, third_comp, third_context, third_general_instruction, third_round_instruction, fourth_comp, fourth_context, fourth_general_instruction, fourth_round_instruction, ranking_first_comp, ranking_second_comp, ranking_third_comp, ranking_fourth_comp, ], ) return demo if __name__ == "__main__": file_path = os.path.join(ROOT_FILE, "./data/single_stock_data/experiment_processed_data.jsonl") topics = [ "healthcare_growth_defensive", "dividend_value_defensive", "nondividend_value_cyclical", ] context_info_list = get_context_list(file_path) # str to List of Dict # system instruction consist of Task, Personality, and Context """ Personality ["extroverted", "introverted"] ["agreeable", "antagonistic"] ["conscientious", "unconscientious"] ["neurotic", "emotionally stable"] ["open to experience", "closed to experience"]] """ # Global variables terminator = ["", "", "", "", "", ""] demo = create_demo() user_list, demo_list = load_username_and_pwd() demo.launch( share=False, auth=user_list + demo_list, )