|
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 |
