| 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() |
| with open(local_path, 'wb') as f: |
| f.write(response.content) |
|
|
| |
| env_file_url = "https://drive.google.com/uc?export=download&id=1bIkq-X1S9943w8-rp8NErTP9G4YfXqKa" |
|
|
| local_env_path = "openai.env" |
| download_env_file(env_file_url, local_env_path) |
|
|
| |
| env = Env() |
| env.read_env("openai.env") |
| openai.api_key = env.str("OPENAI_API_KEY") |
|
|
| |
| 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: |
| 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) |
| if "inputs" in hf_data: |
| actual_data = hf_data["inputs"] |
| print("Processing actual data:", actual_data) |
| 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: |
| |
| 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 |
|
|
|
|
