| from pydantic import BaseModel |
| import openai |
| from environs import Env |
| from typing import List |
|
|
| |
| env = Env() |
| env.read_env("openai.env") |
| openai.api_key = env.str("OPENAI_API_KEY") |
| SYSTEM_PROMPT = env.str("SYSTEM_PROMPT", "Suggest a suitable reply for a user in a dating conversation context.") |
| MODEL = env.str("MODEL", "gpt-3.5-turbo") |
| NUMBER_OF_MESSAGES_FOR_CONTEXT = min(env.int("NUMBER_OF_MESSAGES_FOR_CONTEXT", 4), 10) |
| AI_RESPONSE_TIMEOUT = env.int("AI_RESPONSE_TIMEOUT", 20) |
|
|
| class LastChatMessage(BaseModel): |
| fromUser: str |
| touser: str |
|
|
| class ConversationPayload(BaseModel): |
| fromusername: str |
| tousername: str |
| zodiansign: str |
| LastChatMessages: List[dict] |
| Chatmood: str |
|
|
| def transform_messages(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(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: |
| prompt += f" Consider the {chatmood} mood." |
| |
| return prompt |
|
|
| def get_conversation_suggestions(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 = transform_messages(last_chat_messages) |
| |
| system_prompt = 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 |
