TTsamurai
ask familiarity in an easy way
483e73c
raw
history blame
60.2 kB
import sys
import os
from datetime import datetime
import json
import uuid
from pathlib import Path
from huggingface_hub import CommitScheduler
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_FILE = os.path.join(STATIC_FILE, "txt/system_instruction_with_user_persona.txt")
SYSTEM_INSTRUECTION_PREFERENCE_ELICITATION_FILE = os.path.join(
STATIC_FILE, "txt/system_instruction_preference_elicitation.txt"
)
SUMMARIZATION_PROMPT_FILE = os.path.join(STATIC_FILE, "txt/system_summarization_user_preference_elicitation.txt")
uuid_this_session = str(uuid.uuid4())
feedback_file_interaction = Path("user_feedback/") / f"interaction_{uuid_this_session}.json"
feedback_file_summarization = Path("user_feedback/") / f"summarization_{uuid_this_session}.json"
feedback_file_round_evaluation = Path("user_feedback/") / f"round_evaluation_{uuid_this_session}.json"
feedback_file_final_ranking = Path("user_feedback/") / f"final_ranking_{uuid_this_session}.json"
feedback_file_final_survey = Path("user_feedback/") / f"final_survey_{uuid_this_session}.json"
feedback_folder = feedback_file_interaction.parent
feedback_folder.mkdir(parents=True, exist_ok=True) # Ensure the directory exists
scheduler = CommitScheduler(
repo_id="logging_test_space",
repo_type="dataset",
folder_path=feedback_folder,
path_in_repo="data",
token=os.getenv("HUGGINGFACE_HUB_TOKEN"),
every=1,
)
# 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 = load_static_content(SYSTEM_INSTRUCTION_FILE)
SYSTEM_INSTRUECTION_PREFERENCE_ELICITATION = load_static_content(SYSTEM_INSTRUECTION_PREFERENCE_ELICITATION_FILE)
SUMMARIZATION_PROMPT = load_static_content(SUMMARIZATION_PROMPT_FILE)
# Other constants
FIRST_MESSAGE = "Hey"
INFORMATION_SEEKING = True
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"))
SESSION_DEBUG = True
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", "<br>")
def get_user_narrative_from_raw(raw_narrative):
return get_user_narrative_html(markdown.markdown(raw_narrative.replace("\n", "<br>")))
def get_task_instruction_for_user(context):
ticker_name = context["short_name"]
user_narrative = context["user_narrative"]
user_narrative = user_narrative.replace("\n", "<br>")
html_user_narrative = markdown.markdown(user_narrative)
general_instruction = GENERAL_INSTRUCTION
round_instruction = f"""
<div style="background-color: #f9f9f9; padding: 20px; border-radius: 10px; box-shadow: 0 4px 8px rgba(0, 0, 0, 0.1); margin-bottom: 20px; max-height: 780px; overflow-y: auto; overflow-x: hidden;">
<!-- Stock Information (Bold label, Normal ticker name) -->
<h2 style="color: #2c3e50; text-align: center; margin-bottom: 20px; font-size: 20px; font-weight: 600;">
Round Info
</h2>
<div style="text-align: left; font-size: 20px; font-weight: bold; margin-bottom: 20px;">
Stock
</div>
<div style="text-align: left; font-weight: normal; font-size: 16px; margin-bottom: 20px;">
<span style="font-weight: bold;">
This Round's Stock:
</span>
{ticker_name}
</div>
<!-- User Narrative (Bold label, Normal narrative) -->
<div style="text-align: left; font-size: 20px; font-weight: bold; margin-bottom: 20px;">
User Narrative
</div>
<div style="text-align: left; font-weight: normal; font-size: 16px; margin-bottom: 20px;">
{html_user_narrative}
</div>
</div>"""
return general_instruction, round_instruction
def display_system_instruction_with_html(
system_instruction,
):
html_system_instruction = f"""
<p style="text-align: left; margin-bottom: 10px;">
{system_instruction}
</p>
"""
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")
print(log_file)
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
):
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,
)
# print(f"Preference Summary:{summ}")
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 create_demo():
global personality_prompts, 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():
dropdown = gr.Dropdown(
label="Would you like to purchase the stock?",
choices=["Yes", "No"],
show_label=True,
)
reason = gr.Textbox(
scale=1,
label="Reason for Your Choice (Explain Your Reasoning & Highlight Useful Parts of Conversation)[Optional]",
lines=5,
)
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,
)
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,
)
with gr.Row():
familiarity = stock_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,
"dropdown": dropdown,
"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="<h3>Rank the individual stocks below according to your desire to invest in each one.</h3>")
with gr.Row():
ranking_first_comp = gr.Dropdown(choices=[1, 2, 3, 4, 5])
ranking_second_comp = gr.Dropdown(choices=[1, 2, 3, 4, 5])
ranking_third_comp = gr.Dropdown(choices=[1, 2, 3, 4, 5])
ranking_fourth_comp = gr.Dropdown(choices=[1, 2, 3, 4, 5])
with gr.Row():
gr.HTML(
value='<h3>Choose how strongly you agree with each statement about the advisor (<strong style="color:red;">1 for Strongly Disagree</strong>, <strong style="color:green;">7 for Strongly Agree</strong>).</h3>'
)
with gr.Row():
perceived_personalization_1 = likert_evaluation("The advisor understands my needs")
perceived_personalization_2 = likert_evaluation("The advisor knows what I want")
with gr.Row():
emotional_trust_1 = likert_evaluation("I feel secure about relying on this advisor for my decisions")
emotional_trust_2 = likert_evaluation("I feel comfortable about relying on this advisor for my decisions")
with gr.Row():
emotional_trust_3 = likert_evaluation("I feel content about relying on this advisor for my decisions")
trust_in_competence_1 = likert_evaluation("The advisor is a real expert in assessing the stock")
with gr.Row():
trust_in_competence_2 = likert_evaluation("The advisor has good knowledge of the stock")
intention_to_use_1 = 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():
intention_to_use_2 = likert_evaluation(
"I am willing to let this advisor assist me in deciding which stock to purchase"
)
usefulness_1 = likert_evaluation("The advisor helped me with the decision-making")
with gr.Row():
usefulness_2 = likert_evaluation("Using the advisor to find what I would like to invest in is easy")
usefulness_3 = likert_evaluation("The advisor gave me good suggestions")
with gr.Row():
overall_satisfaction = likert_evaluation("Overall, I am satisfied with the advisor")
providing_information_1 = likert_evaluation("The advisor presents financial knowledge")
with gr.Row():
providing_information_2 = likert_evaluation("The advisor provides the financial knowledge needed")
with gr.Row():
textbox = gr.HTML()
submit_ranking = gr.Button(value="Submit Ranking")
return {
"first": ranking_first_comp,
"second": ranking_second_comp,
"third": ranking_third_comp,
"fourth": ranking_fourth_comp,
"evaluators": {
"perceived_personalization_1": perceived_personalization_1,
"perceived_personalization_2": perceived_personalization_2,
"emotional_trust_1": emotional_trust_1,
"emotional_trust_2": emotional_trust_2,
"emotional_trust_3": emotional_trust_3,
"trust_in_competence_1": trust_in_competence_1,
"trust_in_competence_2": trust_in_competence_2,
"intention_to_use_1": intention_to_use_1,
"intention_to_use_2": intention_to_use_2,
"usefulness_1": usefulness_1,
"usefulness_2": usefulness_2,
"usefulness_3": usefulness_3,
"overall_satisfaction": overall_satisfaction,
"providing_information_1": providing_information_1,
"providing_information_2": providing_information_2,
},
"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,
dropdown,
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["dropdown"],
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, dropdown, reason, likelihood, confidence, familiarity, evaluation_send_button, textbox: respond_evaluation(
user_id,
tab_session,
{
"selection": dropdown,
"reason": reason,
"likelihood": likelihood,
"confidence": confidence,
"familiarity": familiarity,
},
comp,
evaluation_send_button,
textbox,
),
[
user_id,
comp,
tab_session,
dropdown,
reason,
likelihood,
confidence,
familiarity,
evaluation_send_button,
textbox,
],
[tab_session, dropdown, reason, likelihood, confidence, familiarity, evaluation_send_button, textbox],
)
def click_control_preference_stage(tabs, user_id, user_preference_elicitation_session):
(
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: respond_start_conversation(
user_id,
user_preference_elicitation_data,
history,
SYSTEM_INSTRUECTION_PREFERENCE_ELICITATION,
user_elicitation=True,
),
[user_id, user_preference_elicitation_session, elicitation_chatbot],
[user_preference_elicitation_session, elicitation_chatbot, start_conversation, msg_button, continue_button],
)
msg_button.click(
lambda user_id, tab_data, message, history: respond(
user_id,
tab_data,
message,
tab_data["history"],
SYSTEM_INSTRUECTION_PREFERENCE_ELICITATION,
user_elicitation=True,
),
[user_id, user_preference_elicitation_session, msg, elicitation_chatbot],
[user_preference_elicitation_session, msg, elicitation_chatbot],
)
continue_button.click(
lambda user_id, tab_data, history: respond_continue(
user_id,
tab_data,
tab_data["history"],
SYSTEM_INSTRUECTION_PREFERENCE_ELICITATION,
user_elicitation=True,
),
[user_id, user_preference_elicitation_session, elicitation_chatbot],
[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_1,
perceived_personalization_2,
emotional_trust_1,
emotional_trust_2,
emotional_trust_3,
trust_in_competence_1,
trust_in_competence_2,
intention_to_use_1,
intention_to_use_2,
usefulness_1,
usefulness_2,
usefulness_3,
overall_satisfaction,
providing_information_1,
providing_information_2,
) = (
evaluators["perceived_personalization_1"],
evaluators["perceived_personalization_2"],
evaluators["emotional_trust_1"],
evaluators["emotional_trust_2"],
evaluators["emotional_trust_3"],
evaluators["trust_in_competence_1"],
evaluators["trust_in_competence_2"],
evaluators["intention_to_use_1"],
evaluators["intention_to_use_2"],
evaluators["usefulness_1"],
evaluators["usefulness_2"],
evaluators["usefulness_3"],
evaluators["overall_satisfaction"],
evaluators["providing_information_1"],
evaluators["providing_information_2"],
)
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_1, perceived_personalization_2, emotional_trust_1, emotional_trust_2, emotional_trust_3, trust_in_competence_1, trust_in_competence_2, intention_to_use_1, intention_to_use_2, usefulness_1, usefulness_2, usefulness_3, overall_satisfaction, providing_information_1, providing_information_2, 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_1,
perceived_personalization_2,
emotional_trust_1,
emotional_trust_2,
emotional_trust_3,
trust_in_competence_1,
trust_in_competence_2,
intention_to_use_1,
intention_to_use_2,
usefulness_1,
usefulness_2,
usefulness_3,
overall_satisfaction,
providing_information_1,
providing_information_2,
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_1,
perceived_personalization_2,
emotional_trust_1,
emotional_trust_2,
emotional_trust_3,
trust_in_competence_1,
trust_in_competence_2,
intention_to_use_1,
intention_to_use_2,
usefulness_1,
usefulness_2,
usefulness_3,
overall_satisfaction,
providing_information_1,
providing_information_2,
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,
)
# Formatting Input
print(f"User Message: {message} in Tab: {tab_name}")
# 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:
print(f"Tab: {tab_name}\nSystem Output: {outputs_text}")
# 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:
print(f"User Elicitation\nSystem Output: {outputs_text}")
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
# if SESSION_DEBUG:
# log_action(user_id, "Session", "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:
print(f"User Preference Elicitation Data: {user_preference_elicitation_data}")
print(f"Tab data: {tab_data}")
system_instruction = add_user_profile_to_system_instruction(
user_id,
system_instruction,
user_preference_elicitation_data,
summary=USER_PREFERENCE_SUMMARY,
terminator=terminator,
)
print(f"Tab: {tab_name}\nSystem Instruction:{system_instruction}")
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)
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)
# Format
history = huggingface_to_gradio_message(history)
if tab_name is not None:
print(f"Tab: {tab_name}\nHistory: {history}")
# 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:
print(f"User Elicitation\nHistory: {history}")
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,
)
# log_action(user_id, "User_Elicitation", "User Message", first_message)
# log_action(user_id, "User_Elicitation", "Response", outputs_text)
tab_data["history"] = history
# if SESSION_DEBUG:
# log_action(user_id, "Session", "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,
)
# log_action(user_id, tab_name, "Show More of the Advisor’s Answer", "User continued the conversation")
# log_action(user_id, tab_name, "Response", outputs_text)
# Update history for this tab
tab_data["history"] = history
if user_elicitation:
print(f"User Elicitation\nSystem Output: {outputs_text}")
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", "Response", outputs_text)
tab_data["history"] = history
# if SESSION_DEBUG:
# log_action(user_id, "Session", "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
print(f"Tab: {tab_name}\nEvaluation: {evals}")
if evals["selection"] is None or evals["likelihood"] is None or evals["confidence"] is None:
return (
tab_data,
evals["selection"],
evals["reason"],
evals["likelihood"],
evals["confidence"],
evaluation_send_button,
"""<div style="background-color: #f8d7da; color: #721c24; padding: 15px; border: 1px solid #f5c6cb; border-radius: 5px; margin-bottom: 20px;">
<strong>Please make sure that you answer all the questions.</strong>
</div>""",
)
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["selection"],
evals["reason"],
evals["likelihood"],
evals["confidence"],
evaluation_send_button,
"""<div style="background-color: #d4edda; color: #155724; padding: 15px; border: 1px solid #c3e6cb; border-radius: 5px; margin-bottom: 20px;">
<strong>Thank you for submitting your evaluation. You may proceed to the next tab..</strong>
</div>""",
)
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_1,
perceived_personalization_2,
emotional_trust_1,
emotional_trust_2,
emotional_trust_3,
trust_in_competence_1,
trust_in_competence_2,
intention_to_use_1,
intention_to_use_2,
usefulness_1,
usefulness_2,
usefulness_3,
overall_satisfaction,
providing_information_1,
providing_information_2,
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 (
"""<div style="background-color: #f8d7da; color: #721c24; padding: 15px; border: 1px solid #f5c6cb; border-radius: 5px; margin-bottom: 20px;">
<strong>Please make sure that you are not ranking the same stock multiple times.</strong>
</div>""",
submit_ranking,
)
if any(
var is None
for var in [
emotional_trust_1,
emotional_trust_2,
emotional_trust_3,
trust_in_competence_1,
trust_in_competence_2,
intention_to_use_1,
intention_to_use_2,
usefulness_1,
usefulness_2,
usefulness_3,
overall_satisfaction,
providing_information_1,
providing_information_2,
]
):
return (
"""<div style="background-color: #f8d7da; color: #721c24; padding: 15px; border: 1px solid #f5c6cb; border-radius: 5px; margin-bottom: 20px;">
<strong>Please make sure that you answer all the statements.</strong>
</div>""",
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_1": perceived_personalization_1,
"perceived_personalization_2": perceived_personalization_2,
"emotional_trust_1": emotional_trust_1,
"emotional_trust_2": emotional_trust_2,
"emotional_trust_3": emotional_trust_3,
"trust_in_competence_1": trust_in_competence_1,
"trust_in_competence_2": trust_in_competence_2,
"intention_to_use_1": intention_to_use_1,
"intention_to_use_2": intention_to_use_2,
"usefulness_1": usefulness_1,
"usefulness_2": usefulness_2,
"usefulness_3": usefulness_3,
"overall_satisfaction": overall_satisfaction,
"providing_information_1": providing_information_1,
"providing_information_2": providing_information_2,
},
feedback_file_final_survey,
)
submit_ranking = gr.Button(value="Final evaluaiotn received", interactive=False)
return (
"""<div style="background-color: #d4edda; color: #155724; padding: 15px; border: 1px solid #c3e6cb; border-radius: 5px; margin-bottom: 20px;">
<strong>Thank you for participating in the experiment. This concludes the session. You may now close the tab.</strong>
</div>""",
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
narrative_id = user_id.split("_")[-2]
personality_id = user_id.split("_")[-1]
print(f"User ID: {user_id}, Narrative ID: {narrative_id}, Personality ID: {personality_id}")
return user_id, narrative_id, personality_id
def get_inst_without_context(personality_id):
return SYSTEM_INSTRUCTION + "\n" + personality_prompts[int(personality_id)]
def get_stock_related_context(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)])
return raw_context_list, stock_context_list
def set_initial_values(request: gr.Request):
# Set user specific information (Session State)
user_id, narrative_id, personality_id = set_user_id(request)
# System instruction without prompt
system_description_without_context = get_inst_without_context(personality_id)
# Stock related context
raw_context_list, stock_context_list = get_stock_related_context(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, 5], label=first_comp)
ranking_second_comp = gr.Dropdown(choices=[1, 2, 3, 4, 5], label=second_comp)
ranking_third_comp = gr.Dropdown(choices=[1, 2, 3, 4, 5], label=third_comp)
ranking_fourth_comp = gr.Dropdown(choices=[1, 2, 3, 4, 5], label=fourth_comp)
return (
user_id,
narrative_id,
personality_id,
system_description_without_context,
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()
narrative_id = gr.State()
personality_id = gr.State()
system_description_without_context = 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
)
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,
narrative_id,
personality_id,
system_description_without_context,
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
print(context_info_list)
# 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
personality = {
1: [
"extroverted",
"agreeable",
"conscientious",
"emotionally stable",
"open to experience",
]
}
personality_prompts = {i: build_personality_prompt(p) for i, p in personality.items()}
terminator = ["<eos>", "<unk>", "<sep>", "<pad>", "<cls>", "<mask>"]
demo = create_demo()
demo_user_pwd_list = [(f"demo_{i}_1", f"demo_{i}_1") for i in range(1, 11)]
demo.launch(
share=True,
auth=demo_user_pwd_list
+ [
("user_1_1", "pw1"),
("user_2_1", "pw2"),
("user_3_1", "pw3"),
],
)