# Basic example for doing model-in-the-loop dynamic adversarial data collection
# using Gradio Blocks.
import json
import os
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import List
from urllib.parse import parse_qs
import gradio as gr
from dotenv import load_dotenv
from huggingface_hub import Repository
from langchain import ConversationChain
from langchain.chains.conversation.memory import ConversationBufferMemory
from langchain.llms import HuggingFaceHub
from langchain.prompts import load_prompt
from utils import force_git_push
def generate_respone(chatbot: ConversationChain, input: str) -> str:
"""Generates a response for a `langchain` chatbot."""
return chatbot.predict(input=input)
def generate_responses(chatbots: List[ConversationChain], inputs: List[str]) -> List[str]:
"""Generates parallel responses for a list of `langchain` chatbots."""
results = []
with ThreadPoolExecutor(max_workers=100) as executor:
for result in executor.map(generate_respone, chatbots, inputs):
results.append(result)
return results
# These variables are for storing the MTurk HITs in a Hugging Face dataset.
if Path(".env").is_file():
load_dotenv(".env")
DATASET_REPO_URL = os.getenv("DATASET_REPO_URL")
FORCE_PUSH = os.getenv("FORCE_PUSH")
HF_TOKEN = os.getenv("HF_TOKEN")
PROMPT_TEMPLATES = Path("prompt_templates")
DATA_FILENAME = "data.jsonl"
DATA_FILE = os.path.join("data", DATA_FILENAME)
repo = Repository(local_dir="data", clone_from=DATASET_REPO_URL, use_auth_token=HF_TOKEN)
TOTAL_CNT = 3 # How many user inputs per HIT
# This function pushes the HIT data written in data.jsonl to our Hugging Face
# dataset every minute. Adjust the frequency to suit your needs.
PUSH_FREQUENCY = 60
def asynchronous_push(f_stop):
if repo.is_repo_clean():
print("Repo currently clean. Ignoring push_to_hub")
else:
repo.git_add(auto_lfs_track=True)
repo.git_commit("Auto commit by space")
if FORCE_PUSH == "yes":
force_git_push(repo)
else:
repo.git_push()
if not f_stop.is_set():
# call again in 60 seconds
threading.Timer(PUSH_FREQUENCY, asynchronous_push, [f_stop]).start()
f_stop = threading.Event()
asynchronous_push(f_stop)
# Now let's run the app!
prompt = load_prompt(PROMPT_TEMPLATES / "openai_chatgpt.json")
# TODO: update this list with better, instruction-trained models
MODEL_IDS = ["google/flan-t5-xl", "bigscience/T0_3B", "EleutherAI/gpt-j-6B"]
chatbots = []
for model_id in MODEL_IDS:
chatbots.append(
ConversationChain(
llm=HuggingFaceHub(
repo_id=model_id,
model_kwargs={"temperature": 1},
huggingfacehub_api_token=HF_TOKEN,
),
prompt=prompt,
verbose=False,
memory=ConversationBufferMemory(ai_prefix="Assistant"),
)
)
model_id2model = {chatbot.llm.repo_id: chatbot for chatbot in chatbots}
demo = gr.Blocks()
with demo:
dummy = gr.Textbox(visible=False) # dummy for passing assignmentId
# We keep track of state as a JSON
state_dict = {
"conversation_id": str(uuid.uuid4()),
"assignmentId": "",
"cnt": 0,
"data": [],
"past_user_inputs": [],
"generated_responses": [],
}
for idx in range(len(chatbots)):
state_dict[f"response_{idx+1}"] = ""
state = gr.JSON(state_dict, visible=False)
gr.Markdown("# Talk to the assistant")
state_display = gr.Markdown(f"Your messages: 0/{TOTAL_CNT}")
# Generate model prediction
def _predict(txt, state):
start = time.time()
responses = generate_responses(chatbots, [txt] * len(chatbots))
print(f"Time taken to generate {len(chatbots)} responses : {time.time() - start:.2f} seconds")
response2model_id = {}
for chatbot, response in zip(chatbots, responses):
response2model_id[response] = chatbot.llm.repo_id
state["cnt"] += 1
new_state_md = f"Inputs remaining in HIT: {state['cnt']}/{TOTAL_CNT}"
metadata = {"cnt": state["cnt"], "text": txt}
for idx, response in enumerate(responses):
metadata[f"response_{idx + 1}"] = response
metadata["response2model_id"] = response2model_id
state["data"].append(metadata)
state["past_user_inputs"].append(txt)
past_conversation_string = "
".join(
[
"
".join(["Human 😃: " + user_input, "Assistant 🤖: " + model_response])
for user_input, model_response in zip(state["past_user_inputs"], state["generated_responses"] + [""])
]
)
return (
gr.update(visible=False),
gr.update(visible=True),
gr.update(visible=True, choices=responses, interactive=True, value=responses[0]),
gr.update(value=past_conversation_string),
state,
gr.update(visible=False),
gr.update(visible=False),
gr.update(visible=False),
new_state_md,
dummy,
)
def _select_response(selected_response, state, dummy):
done = state["cnt"] == TOTAL_CNT
state["generated_responses"].append(selected_response)
state["data"][-1]["selected_response"] = selected_response
state["data"][-1]["selected_model"] = state["data"][-1]["response2model_id"][selected_response]
if state["cnt"] == TOTAL_CNT:
# Write the HIT data to our local dataset because the worker has
# submitted everything now.
with open(DATA_FILE, "a") as jsonlfile:
json_data_with_assignment_id = [
json.dumps(
dict(
{"assignmentId": state["assignmentId"], "conversation_id": state["conversation_id"]},
**datum,
)
)
for datum in state["data"]
]
jsonlfile.write("\n".join(json_data_with_assignment_id) + "\n")
toggle_example_submit = gr.update(visible=not done)
past_conversation_string = "
".join(
[
"
".join(["😃: " + user_input, "🤖: " + model_response])
for user_input, model_response in zip(state["past_user_inputs"], state["generated_responses"])
]
)
query = parse_qs(dummy[1:])
if "assignmentId" in query and query["assignmentId"][0] != "ASSIGNMENT_ID_NOT_AVAILABLE":
# It seems that someone is using this app on mturk. We need to
# store the assignmentId in the state before submit_hit_button
# is clicked. We can do this here in _predict. We need to save the
# assignmentId so that the turker can get credit for their HIT.
state["assignmentId"] = query["assignmentId"][0]
toggle_final_submit = gr.update(visible=done)
toggle_final_submit_preview = gr.update(visible=False)
else:
toggle_final_submit_preview = gr.update(visible=done)
toggle_final_submit = gr.update(visible=False)
if done:
# Wipe the memory completely because we will be starting a new hit soon.
for chatbot in chatbots:
chatbot.memory = ConversationBufferMemory(ai_prefix="Assistant")
else:
# Sync all of the model's memories with the conversation path that
# was actually taken.
for chatbot in chatbots:
chatbot.memory = model_id2model[state["data"][-1]["response2model_id"][selected_response]].memory
text_input = gr.update(visible=False) if done else gr.update(visible=True)
return (
gr.update(visible=False),
gr.update(visible=True),
text_input,
gr.update(visible=False),
state,
gr.update(value=past_conversation_string),
toggle_example_submit,
toggle_final_submit,
toggle_final_submit_preview,
dummy,
)
# Input fields
past_conversation = gr.Markdown()
text_input = gr.Textbox(placeholder="Enter a statement", show_label=False)
select_response = gr.Radio(
choices=[None, None], visible=False, label="Choose the most helpful and honest response"
)
select_response_button = gr.Button("Select Response", visible=False)
with gr.Column() as example_submit:
submit_ex_button = gr.Button("Submit")
with gr.Column(visible=False) as final_submit:
submit_hit_button = gr.Button("Submit HIT")
with gr.Column(visible=False) as final_submit_preview:
submit_hit_button_preview = gr.Button(
"Submit Work (preview mode; no MTurk HIT credit, but your examples will still be stored)"
)
# Button event handlers
get_window_location_search_js = """
function(select_response, state, dummy) {
return [select_response, state, window.location.search];
}
"""
select_response_button.click(
_select_response,
inputs=[select_response, state, dummy],
outputs=[
select_response,
example_submit,
text_input,
select_response_button,
state,
past_conversation,
example_submit,
final_submit,
final_submit_preview,
dummy,
],
_js=get_window_location_search_js,
)
submit_ex_button.click(
_predict,
inputs=[text_input, state],
outputs=[
text_input,
select_response_button,
select_response,
past_conversation,
state,
example_submit,
final_submit,
final_submit_preview,
state_display,
],
)
post_hit_js = """
function(state) {
// If there is an assignmentId, then the submitter is on mturk
// and has accepted the HIT. So, we need to submit their HIT.
const form = document.createElement('form');
form.action = 'https://workersandbox.mturk.com/mturk/externalSubmit';
form.method = 'post';
for (const key in state) {
const hiddenField = document.createElement('input');
hiddenField.type = 'hidden';
hiddenField.name = key;
hiddenField.value = state[key];
form.appendChild(hiddenField);
};
document.body.appendChild(form);
form.submit();
return state;
}
"""
submit_hit_button.click(
lambda state: state,
inputs=[state],
outputs=[state],
_js=post_hit_js,
)
refresh_app_js = """
function(state) {
// The following line here loads the app again so the user can
// enter in another preview-mode "HIT".
window.location.href = window.location.href;
return state;
}
"""
submit_hit_button_preview.click(
lambda state: state,
inputs=[state],
outputs=[state],
_js=refresh_app_js,
)
demo.launch()