File size: 8,708 Bytes
55dfa40 60d27cf 55dfa40 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 |
from pydantic import BaseModel
import openai
from environs import Env
from typing import List, Dict, Any
import requests
def download_env_file(url: str, local_path: str):
response = requests.get(url)
response.raise_for_status() # Ensure we notice bad responses
with open(local_path, 'wb') as f:
f.write(response.content)
# Download the .env file
env_file_url = "https://drive.google.com/uc?export=download&id=1bIkq-X1S9943w8-rp8NErTP9G4YfXqKa" # Adjusted URL for direct download
local_env_path = "openai.env"
download_env_file(env_file_url, local_env_path)
# Load environment variables
env = Env()
env.read_env("openai.env")
openai.api_key = env.str("OPENAI_API_KEY")
# Constants
MODEL = env.str("MODEL", "gpt-3.5-turbo")
AI_RESPONSE_TIMEOUT = env.int("AI_RESPONSE_TIMEOUT", 20)
class EndpointHandler:
def __init__(self, model_dir=None):
self.model_dir = model_dir
def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
try:
if "inputs" in data: # Check if data is in Hugging Face JSON format
return self.process_hf_input(data)
else:
return self.process_json_input(data)
except ValueError as e:
return {"error": str(e)}
except Exception as e:
return {"error": str(e)}
def process_json_input(self, json_data):
if "FromUserKavasQuestions" in json_data and "Chatmood" in json_data:
prompt = self.create_conversation_starter_prompt(
json_data["FromUserKavasQuestions"],
json_data["Chatmood"]
)
starter_suggestion = self.generate_conversation_starters(prompt)
return {"conversation_starter": starter_suggestion}
elif "LastChatMessages" in json_data:
last_chat_messages = json_data["LastChatMessages"][-4:]
response = {
"version": "1.0.0-alpha",
"suggested_responses": self.get_conversation_suggestions(last_chat_messages)
}
return response
else:
raise ValueError("Invalid JSON structure.")
def process_hf_input(self, hf_data):
print("Received HF Data:", hf_data) # Debugging line
if "inputs" in hf_data:
actual_data = hf_data["inputs"]
print("Processing actual data:", actual_data) # Debugging line
return self.process_json_input(actual_data)
else:
return {"error": "Invalid Hugging Face JSON structure."}
def create_conversation_starter_prompt(self, user_questions, chatmood):
formatted_info = " ".join([f"{qa['Question']} - {qa['Answer']}" for qa in user_questions if qa['Answer']])
prompt = (f"Based on user profile info and a {chatmood} mood, "
f"generate 3 subtle and very short conversation starters. "
f"Explore various topics like travel, hobbies, movies, and not just culinary tastes. "
f"\nProfile Info: {formatted_info}")
return prompt
def generate_conversation_starters(self, prompt):
try:
response = openai.ChatCompletion.create(
model=MODEL,
messages=[{"role": "system", "content": prompt}],
temperature=0.7,
max_tokens=100,
n=1,
request_timeout=AI_RESPONSE_TIMEOUT
)
return response.choices[0].message["content"]
except openai.error.OpenAIError as e:
raise Exception(f"OpenAI API error: {str(e)}")
except Exception as e:
raise Exception(f"Unexpected error: {str(e)}")
def transform_messages(self, last_chat_messages):
t_messages = []
for chat in last_chat_messages:
if "fromUser" in chat:
from_user = chat['fromUser']
message = chat.get('touser', '')
t_messages.append(f"{from_user}: {message}")
elif "touser" in chat:
to_user = chat['touser']
message = chat.get('fromUser', '')
t_messages.append(f"{to_user}: {message}")
if t_messages and "touser" in last_chat_messages[-1]:
latest_message = t_messages[-1]
latest_message = f"Q: {latest_message}"
t_messages[-1] = latest_message
return t_messages
def generate_system_prompt(self, last_chat_messages, fromusername, tousername, zodiansign=None, chatmood=None):
prompt = ""
if not last_chat_messages or ("touser" not in last_chat_messages[-1]):
prompt = f"Suggest a casual and friendly message for {fromusername} to start a conversation with {tousername} or continue naturally, as if talking to a good friend. Strictly avoid replying to messages from {fromusername} or answering their questions."
else:
prompt = f"Suggest a warm and friendly reply for {fromusername} to respond to the last message from {tousername}, as if responding to a dear friend. Strictly avoid replying to messages from {fromusername} or answering their questions."
if zodiansign:
prompt += f" Keep in mind {tousername}'s {zodiansign} zodiac sign."
if chatmood:
if chatmood == "Casual Vibes":
prompt += " Keep the conversation relaxed and informal."
elif chatmood == "Flirty Fun":
prompt += " Add a playful and teasing tone to the conversation."
elif chatmood == "Deep and Thoughtful":
prompt += " Encourage reflective and introspective responses."
elif chatmood == "Humor Central":
prompt += " Incorporate witty and humorous elements into the conversation."
elif chatmood == "Romantic Feels":
prompt += " Express affection and use sweet and romantic language."
elif chatmood == "Intellectual Banter":
prompt += " Engage in thought-provoking discussions on topics like books and movies."
elif chatmood == "Supportive Mode":
prompt += " Offer empathy, support, and encouragement in the conversation."
elif chatmood == "Curiosity Unleashed":
prompt += " Show eagerness to learn and explore interests by asking questions."
elif chatmood == "Chill and Easygoing":
prompt += " Maintain a relaxed and laid-back tone in the conversation."
elif chatmood == "Adventurous Spirit":
prompt += " Share travel stories and plans with enthusiasm and energy."
return prompt
def get_conversation_suggestions(self, last_chat_messages):
fromusername = last_chat_messages[-1].get("fromusername", "")
tousername = last_chat_messages[-1].get("tousername", "")
zodiansign = last_chat_messages[-1].get("zodiansign", "")
chatmood = last_chat_messages[-1].get("Chatmood", "")
messages = self.transform_messages(last_chat_messages)
system_prompt = self.generate_system_prompt(last_chat_messages, fromusername, tousername, zodiansign, chatmood)
messages_final = [{"role": "system", "content": system_prompt}]
if messages:
messages_final.extend([{"role": "user", "content": m} for m in messages])
else:
# If there are no messages, add a default message to ensure a response is generated
default_message = f"{tousername}: Hi there!"
messages_final.append({"role": "user", "content": default_message})
try:
response = openai.ChatCompletion.create(
model=MODEL,
messages=messages_final,
temperature=0.7,
max_tokens=150,
n=3,
request_timeout=AI_RESPONSE_TIMEOUT
)
formatted_replies = []
for idx, choice in enumerate(response.choices):
formatted_replies.append({
"type": "TEXT",
"body": choice.message['content'],
"title": f"AI Reply {idx + 1}",
"confidence": 1,
})
return formatted_replies
except openai.error.Timeout as e:
formatted_reply = [{
"type": "TEXT",
"body": "Request to the AI response generator has timed out. Please try again later.",
"title": "AI Response Error",
"confidence": 1
}]
return formatted_reply |