from langchain.llms import OpenAI
# from langchain.chat_models import ChatOpenAI
import openai
from src.get_user_details import *
from src.mongo_main import *
import requests
import json
import threading
import re
import pandas as pd
import pickle
from sklearn.metrics.pairwise import cosine_similarity
from src.utils import check_values_match
# from pronouns import *
from src.retriver_chunk import *
# from user_intent import *
# from token_log import *
def custom_replace(text):
replacements = [("\n\n", " "), ("\n", " "),("\ "," ")]
for old, new in replacements:
text = text.replace(old, new)
# Remove empty strings
text = ' '.join(text.split())
return text
def extract_page_content(data_string):
page_content_list = []
for document in data_string:
page_content = document.page_content
page_content = custom_replace(page_content)
page_content_list.append(page_content)
unique_list = list(set(page_content_list))
return unique_list
def general_ai_prompt(context, question, history,bot_name,custom_instruction):
prompt_template = f"""You are an assistant for question-answering tasks and your name is {bot_name}. Use the following pieces of retrieved context to answer the question if context not having answer then answer your own.
if user ask genral question you need to answer.
use the following instruction while answering the query
Instruction: {custom_instruction}
Context:
---------
{context}
---------
"""
messages = [{"role": "system", "content": prompt_template}]
messages += history
messages.append({"role": "user", "content": question})
return messages
def my_prompt(context, question, history,bot_name,custom_instruction):
prompt_template = f"""You are an assistant for question-answering tasks and your name is {bot_name}. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know.
Additionally, Do not clarify to the user why you lack the answer or disclose the source of information.
if user asks out of context questions or write a program say 'i don't have information' follow this without fail
use the following instruction while answering the query
Instruction: {custom_instruction}
Example #1
Context:
---------
Apples are red
---------
Question: what color are apples?
Helpful Answer: red.
Example #2
Context:
---------
it was night and the witness forgot his glasses. he was not sure if it was a sports car or an suv
---------
Question: what type was the car?
Helpful Answer: a sports car or an SUV.
Example #3
Context:
---------
Pears are either red or orange
---------
Question: what color are apples?
Helpful Answer: i don't have enough information to answer your question
Begin!
Context:
---------
{context}
---------
"""
messages = [{"role": "system", "content": prompt_template}]
messages += history
messages.append({"role": "user", "content": question})
return messages
def parse_input_string(input_string):
pattern_to_remove = r'Score: (?:\d+|N/A)'
return re.sub(pattern_to_remove, '', input_string)
def openai_llm(query,model_name):
response = openai.chat.completions.create(
model=model_name,
messages=query,
temperature = 0.2
)
modified_string = response.choices[0].message.content.replace('\n', ' \n')
return modified_string # , response.choices[0].usage
def static_answer(query,user_id,bot_id,history,context_id):
answer, similarity_score,q_a = retrieve_answer(query,user_id,bot_id)
if answer != None and similarity_score != None:
if similarity_score >= 0.85:
history.append({"role": "user", "content": query})
history.append({"role": "assistant", "content": answer})
def post_request():
headers = {'Content-Type': 'application/json'}
payload = {"id": context_id,"data": {"history": history,"last_response": answer}}
requests.post(redis_context_memory_URL, headers=headers, data=json.dumps(payload)).json()
thread = threading.Thread(target=post_request)
thread.start()
return answer
else:
return None
else:
return None
def load_test(user_id,bot_id,query,context_id,bot_name,custom_instruction,mail,Iscontact_details_availabel,error_message,general_ai):
payload = {"id":context_id}
json_payload = json.dumps(payload)
response = requests.get(redis_context_memory_URL, headers={'Content-Type': 'application/json'}, data=json_payload).json()
if response['is_new'] == True:
response_post = requests.post(redis_context_memory_URL, headers={'Content-Type': 'application/json'}, data=json.dumps({"id": context_id,"data":{"history": [{"role":"user","content": "Hi"}],"last_response":"Hi"}})).json()
payload = {"id":context_id}
json_payload = json.dumps(payload)
response = requests.get(redis_context_memory_URL, headers={'Content-Type': 'application/json'}, data=json_payload).json()
if response_post['success'] == False:
return "Failed please check backend"
data = {"user_id":user_id,"bot_id":bot_id}
configuration = config_main(data)
history = response["data"]["history"]
static_response = static_answer(query,user_id,bot_id,history,context_id)
if static_response != None:
return static_response
if configuration['method_id'] == "method_1":
doc = retirve_method_1(user_id,bot_id,history,query,n_history_questions=configuration['n_history_questions'],n_doc_retirve=configuration['n_doc_retirve'],lambda_mult=configuration['lambda_mult'],algo=configuration['algo'])
context = str(extract_page_content(doc))
if general_ai == 1:
full_prompt = general_ai_prompt(context,query,history,bot_name,custom_instruction)
else:
full_prompt = my_prompt(context, query,history,bot_name,custom_instruction)
# res, usage = openai_llm(full_prompt,configuration['model_name'])
res= openai_llm(full_prompt,configuration['model_name'])
if check_string(res):
if check_string_2_company_string(res) == True:
res = error_message
else:
res2 = res
if Iscontact_details_availabel == 3:
try:
res = "I'm sorry, but I don't have the required information to respond to your question at the moment. However, our dedicated support team will reach out to you using the provided contact details."
except:
res = "I'm sorry, but I don't have the information needed to answer your question right now. for more specialized support please write a mail to "+mail
elif Iscontact_details_availabel == 2:
res = "I'm sorry, but I don't have the information needed to answer your question right now. for more specialized support please write a mail to "+mail
elif Iscontact_details_availabel == 1:
res = error_message
payload = {"context_id": context_id,"email": "","contact": "","contact_status": "4"}
requests.post(contact_save, headers={'Content-Type': 'application/json'}, data=json.dumps(payload)).json()
else:
res = "I'm sorry, but I don't have the information needed to answer your question right now. for more specialized support please write a mail to "+mail
res = answer_suggestion("rephrase this text and make a single text message also remove question is this text --->"+res2+" "+ res)
# process_request_for_token_count(str(history), query, str(context), custom_instruction,usage, user_id, bot_id,context_id,res,general_ai)
if len(str(full_prompt)) >= 25800:
history = history[-3:]
if check_values_match(history,res,query) or "I'm sorry" in res.lower()== True:
pass
else:
history.append({"role": "user", "content": query})
history.append({"role": "assistant", "content": res})
def post_request():
headers = {'Content-Type': 'application/json'}
payload = {"id": context_id,"data": {"history": history,"last_response": res}}
requests.post(redis_context_memory_URL, headers=headers, data=json.dumps(payload)).json()
thread = threading.Thread(target=post_request)
thread.start()
print("log:",full_prompt,user_id,bot_id,res)
return res
def answer_suggestion(query):
query = [{"role": "system", "content": query}]
response = openai.ChatCompletion.create(
model=gpt_model,
messages=query,
temperature = 0,
api_key=os.environ["OPENAI_API_KEY"]
)
modified_string = response['choices'][0]['message']['content'].replace('\n', ' \n')
return modified_string
def retrieve_answer(question,user_id,bot_id):
try:
dataset_path = os.path.join(os.environ['DB_DIR'], str(user_id), str(bot_id), "q_a_data", "data.csv")
model_path = os.path.join(os.environ['DB_DIR'], str(user_id), str(bot_id), "q_a_data", "chatbot_model.pkl")
# Load the vectorizer model from the pickle file
with open(model_path, 'rb') as file:
vectorizer = pickle.load(file)
question_vector = vectorizer.transform([question])
dataset = pd.read_csv(dataset_path)
dataset_vectors = vectorizer.transform(dataset['Question'])
similarity_scores = cosine_similarity(question_vector, dataset_vectors)[0]
most_similar_index = similarity_scores.argmax()
answer = dataset.loc[most_similar_index, 'Answer']
similarity_score = similarity_scores[most_similar_index]
return answer, similarity_score, ' '.join(dataset.values.flatten())
except:
return None, None, " "