import random
import sys
import torch
from peft import AutoPeftModelForCausalLM
from transformers import AutoTokenizer
MODEL_ID = "mario-rc/emotional-rlaif-dpo-llama-3.2-1b-instruct"
def get_turn_markers():
return {
'bos': '<|begin_of_text|>',
'user_start': '<|start_header_id|>user<|end_header_id|>\n\n',
'user_end': '<|eot_id|>',
'assistant_start': '<|start_header_id|>assistant<|end_header_id|>\n\n',
'assistant_end': '<|eot_id|>',
}
def update_prompt(dialogues):
"""Build the prompt for the model based on the dialogue history."""
markers = get_turn_markers()
system = (
f"{markers['bos']}You are an expert at creating dialogues.\n\n"
"Dialogue and emotional structure:\n"
)
human_prompts = [d[0] for d in dialogues]
chatbot_responses = [d[1] for d in dialogues]
p_emo = [h[0] for h in human_prompts]
p_utt = [h[1] for h in human_prompts]
r1_utt = [c[1] for c in chatbot_responses]
r2_emo = [c[2] for c in chatbot_responses]
r2_utt = [c[3] for c in chatbot_responses]
r3_utt = [c[5] for c in chatbot_responses]
context = (
"Human: (HAPPINESS) PROMPT.\n"
"Chatbot: (HAPPINESS) RESPONSE_1. (HAPPINESS) RESPONSE_2. (NEUTRAL) RESPONSE_3.\n"
)
for p_e, _, _, r2_e, _, _ in zip(p_emo, p_utt, r1_utt, r2_emo, r2_utt, r3_utt):
context += f"Human: ({p_e}) PROMPT.\n"
context += f"Chatbot: ({p_e}) RESPONSE_1. ({r2_e}) RESPONSE_2. (NEUTRAL) RESPONSE_3.\n"
context += "\n"
rules = (
"Dialogue rules:\n"
"The response must be open-domain curated. The response should be coherent, empathetic, engaging and proactive.\n"
"The chatbot RESPONSE is composed of 3 different sentences (RESPONSE_1, RESPONSE_2 and RESPONSE_3), separated by a period.\n"
"Between RESPONSE_1, RESPONSE_2 and RESPONSE_3 should be a max length of 20-25 words.\n"
"RESPONSE_3 must be open-ended to follow-up the conversation, so the Human is encouraged to answer with a full long sentence. Avoid yes/no questions.\n\n"
"Emotional response rules:\n"
f"RESPONSE_1 must contain a {p_emo[-1]} tone.\n"
f"RESPONSE_2 must contain a {r2_emo[-1]} tone.\n"
"RESPONSE_3 must contain a NEUTRAL tone.\n\n"
"Answer in a single turn to Human. Follow exactly the emotional structure and the emotional and dialogue rules."
f"{markers['user_start']}"
)
completion = ""
for idx, (p_e, p_u, r1_u, r2_e, r2_u, r3_u) in enumerate(zip(p_emo, p_utt, r1_utt, r2_emo, r2_utt, r3_utt)):
completion += f"({p_e}) {p_u}{markers['user_end']}{markers['assistant_start']}"
if idx != len(p_emo) - 1:
completion += f"({p_e}) {r1_u} ({r2_e}) {r2_u} (NEUTRAL) {r3_u}{markers['assistant_end']}{markers['user_start']}"
return system + context + rules + completion
class Chatbot:
def __init__(self, dialogue_language="es"):
self.dialogue_language = dialogue_language
self.device = "cuda" if torch.cuda.is_available() else "cpu"
self.tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
self.model = AutoPeftModelForCausalLM.from_pretrained(
MODEL_ID,
torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32,
device_map="auto" if torch.cuda.is_available() else None,
)
if not torch.cuda.is_available():
self.model = self.model.to(self.device)
self.model.eval()
def split_emo_chatbot(sentence):
"""Extract emotions and utterances from a chatbot response."""
response_pos_ini = [i for i, c in enumerate(sentence) if c == "("]
response_pos_end = [i for i, c in enumerate(sentence) if c == ")"]
response_r1_utt = sentence[response_pos_end[0] + 2:response_pos_ini[1]].strip()
response_r2_utt = sentence[response_pos_end[1] + 2:response_pos_ini[2]].strip()
response_r3_utt = sentence[response_pos_end[2] + 2:].lstrip()
return response_r1_utt, response_r2_utt, response_r3_utt
def select_dialogue(self, dialogue_language):
"""Return a list of example dialogues for the given language."""
if dialogue_language == "en":
dialogue_base = [
[["HAPPINESS", "Hi, who are you?"],
["HAPPINESS", "Hi! I'm Ray, a social personal assistant robot with emotions.", "HAPPINESS", "I'm here to chat with you about anything you'd like.", "NEUTRAL", "What would you like to talk about?"]],
[["HAPPINESS", "I'm interested in talking about you, tell me more."],
["HAPPINESS", "Great! I'm glad you want to get to know me!", "NEUTRAL", "I'm designed to help and talk with people about any topic.", "NEUTRAL", "I can talk about science, technology, history, or just have a pleasant conversation. What interests you?"]],
]
dialogue = [
[["HAPPINESS", "Nice to meet you, Ray. I'd like to know more about you."],
["HAPPINESS", "The pleasure is mine!", "HAPPINESS", "I'm a chatbot designed to chat and learn with you.", "NEUTRAL", "Would you like to talk about a specific topic?"]],
[["HAPPINESS", "I love talking to you, you're very interesting."],
["HAPPINESS", "That's so nice to hear! I'm glad you enjoy talking to me.", "NEUTRAL", "I'm designed to have meaningful and empathetic conversations.", "NEUTRAL", "Would you like to talk about emotions, artificial intelligence, or something more personal?"]],
]
else:
dialogue_base = [
[["HAPPINESS", "Hola, ¿quién eres?"],
["HAPPINESS", "¡Hola! Soy Ray y soy un robot social asistente personal con emociones.", "HAPPINESS", "Estoy aquí para charlar contigo sobre cualquier tema.", "NEUTRAL", "¿Sobre qué te gustaría hablar?"]],
[["HAPPINESS", "Me interesa hablar sobre ti, cuéntame más detalles."],
["HAPPINESS", "¡Genial, me encanta que quieras conocerme!", "NEUTRAL", "Estoy diseñado para ayudar y hablar con la gente sobre cualquier tema.", "NEUTRAL", "Puedo hablar de ciencia, tecnología, historia o simplemente tener una charla amena. ¿Qué te interesa?"]],
]
dialogue = [
[["HAPPINESS", "Mucho gusto, Ray. Me gustaría saber más sobre ti."],
["HAPPINESS", "¡El gusto es mío!", "HAPPINESS", "Soy un chatbot diseñado para conversar y aprender contigo.", "NEUTRAL", "¿Quieres hablar de algún tema en específico?"]],
[["HAPPINESS", "Me encanta hablar contigo, eres muy interesante."],
["HAPPINESS", "¡Qué lindo escuchar eso! Me alegra que disfrutes hablar conmigo.", "NEUTRAL", "Estoy diseñado para tener conversaciones significativas y empáticas.", "NEUTRAL", "¿Te gustaría que hablemos sobre emociones, inteligencia artificial, o algo más personal?"]],
]
return dialogue_base + dialogue
def chat_with_model(self, dialogues, max_new_tokens=96):
prompt_text = update_prompt(dialogues)
inputs = self.tokenizer(prompt_text, return_tensors="pt").to(self.model.device)
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_new_tokens,
do_sample=True,
temperature=0.7,
top_p=0.9,
eos_token_id=self.tokenizer.eos_token_id,
)
generated = outputs[0][inputs["input_ids"].shape[-1]:]
response = self.tokenizer.decode(generated, skip_special_tokens=True).splitlines()[0].strip()
print("Response:", response, "\n")
return response
def main(self):
emotions = ["ANGER", "FEAR", "SADNESS", "DISGUST", "HAPPINESS", "SURPRISE", "NEUTRAL"]
dialogue = self.select_dialogue(self.dialogue_language)
while True:
if len(dialogue) > 7:
dialogue.pop(2)
p_emo = random.choice(emotions)
user_sentence = input(f"Enter your sentence: ({p_emo}) ")
if user_sentence.strip().lower() == "exit":
break
r2_emo = random.choice(emotions)
dialogue.append([[p_emo, user_sentence], [p_emo, "", r2_emo, "", "NEUTRAL", ""]])
response = self.chat_with_model(dialogue)
try:
r1_utt, r2_utt, r3_utt = self.split_emo_chatbot(response)
except Exception:
if self.dialogue_language == "en":
r1_utt, r2_utt, r3_utt = "I'm sorry.", "I didn't understand you.", "Could you repeat?"
else:
r1_utt, r2_utt, r3_utt = "Lo siento.", "No te he entendido.", "¿Podrías repetirme?"
dialogue[-1][1] = [p_emo, r1_utt, r2_emo, r2_utt, "NEUTRAL", r3_utt]
if __name__ == "__main__":
language = sys.argv[1] if len(sys.argv) > 1 else "en"
chatbot = Chatbot(dialogue_language=language)
chatbot.main()