Create handler.py

handler.py

@@ -0,0 +1,190 @@
+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://www.dropbox.com/scl/fi/21ldek2cdsak2v3mhyy5x/openai.env?rlkey=nxdkd8l8esdy8npa3vfgvqkhp&st=s2f2zzwl&dl=1"  # 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, "
+                      f"as if talking to a good friend. Strictly avoid replying to messages from {fromusername} or answering their questions. "
+                      f"Make sure the message reflects a {chatmood} mood.")
+        else:
+            prompt = (f"Suggest a warm and friendly reply for {fromusername} to respond to the last message from {tousername}, "
+                      f"as if responding to a dear friend. Strictly avoid replying to messages from {fromusername} or answering their questions. "
+                      f"Ensure the reply embodies a {chatmood} mood.")
+        
+        if zodiansign:
+            prompt += f" Keep in mind {tousername}'s {zodiansign} zodiac sign."
+        
+        if chatmood:
+            mood_instructions = {
+                "Casual Vibes": " Keep the conversation relaxed and informal, using phrases like 'Hey, what's up?' or 'Just chilling, how about you?'",
+                "Flirty Fun": " Add a playful and teasing tone, using phrases like 'You always know how to make me smile!' or 'Guess what? I have a secret to tell you.'",
+                "Deep and Thoughtful": " Encourage reflective and introspective responses, using phrases like 'I've been thinking about...' or 'What's your take on...?'",
+                "Humor Central": " Incorporate witty and humorous elements, using phrases like 'Why did the chicken cross the road?' or 'I have a hilarious story for you!'",
+                "Romantic Feels": " Express affection and use sweet and romantic language, using phrases like 'You're the best part of my day' or 'I can't stop thinking about you.'",
+                "Intellectual Banter": " Engage in thought-provoking discussions on topics like books and movies, using phrases like 'Have you read any good books lately?' or 'What do you think about the latest film?'",
+                "Supportive Mode": " Offer empathy, support, and encouragement, using phrases like 'I'm here for you' or 'Everything will be okay, I believe in you.'",
+                "Curiosity Unleashed": " Show eagerness to learn and explore interests by asking questions, using phrases like 'Tell me more about...' or 'I'm curious, how did you get into...?'",
+                "Chill and Easygoing": " Maintain a relaxed and laid-back tone, using phrases like 'No worries, take your time' or 'Just go with the flow.'",
+                "Adventurous Spirit": " Share travel stories and plans with enthusiasm and energy, using phrases like 'Let's plan our next adventure!' or 'Guess where I want to go next?'"
+            }
+            prompt += mood_instructions.get(chatmood, "")
+
+        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
+