Spaces:
Runtime error
Runtime error
File size: 4,661 Bytes
439340a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 |
import gradio as gr
import random
import os
from openai import OpenAI
from pinecone import Pinecone
import uuid
from pymongo.mongo_client import MongoClient
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_OPENAI = os.getenv("MODEL_OPENAI")
PINECONE_API_TOKEN = os.getenv("PINECONE_API_TOKEN")
PINECONE_ENVIRONMENT = os.getenv("PINECONE_ENV")
PINECONE_HOST = os.getenv("PINECONE_HOST")
DB_USER_NAME = os.getenv("DB_USER_NAME")
DB_PASSWORD = os.getenv("DB_PASSWORD")
# Chat
openai_client = OpenAI(api_key=OPENAI_API_KEY)
# Vector store
pc = Pinecone(api_key=PINECONE_API_TOKEN)
index = pc.Index(host=PINECONE_HOST)
# Database
uri = f"mongodb+srv://{DB_USER_NAME}:{DB_PASSWORD}@cluster-rob01.3fpztfw.mongodb.net/?retryWrites=true&w=majority&appName=cluster-rob01"
client = MongoClient(uri)
db = client["ChatCrunchyroll"]
collection = db["history_msg"]
def _save_history_msg():
return None
def _add_question_vectorstore(question: str, response: str):
vector_id = str(uuid.uuid4())
vector_embedding = _call_embedding(question)
vector_metadata = {
'question': question,
'text': response
}
index.upsert([(vector_id, vector_embedding, vector_metadata)])
def _update_elements(question, chatbot, output, history_messages):
chatbot.append([question, output])
history_messages.append({'role': 'user', 'content': question})
history_messages.append({'role': 'assistant', 'content': output})
return chatbot
def _query_pinecone(embedding):
results = index.query(
vector=embedding,
top_k=10,
include_metadata=True,
)
final_results = """"""
for result in results['matches']:
final_results += f"{result['metadata']['text']}\n"
return final_results
def _general_prompt(context):
with open("prompt_general.txt", "r") as file:
file_prompt = file.read().replace("\n", "")
context_prompt = file_prompt.replace('CONTEXT', context)
print(context_prompt)
print("--------------------")
return context_prompt
def _call_embedding(text: str):
response = openai_client.embeddings.create(
input=text,
model='text-embedding-ada-002'
)
return response.data[0].embedding
def _call_gpt(prompt: str, message: str):
response = openai_client.chat.completions.create(
model=MODEL_OPENAI,
temperature=0.2,
messages=[
{'role': 'system', 'content': prompt},
{'role': 'user', 'content': message}
]
)
return response.choices[0].message.content
def _call_gpt_standalone(prompt: str):
response = openai_client.chat.completions.create(
model=MODEL_OPENAI,
temperature=0.2,
messages=[
{'role': 'system', 'content': prompt},
]
)
return response.choices[0].message.content
def _get_standalone_question(question, history_messages):
with open("prompt_standalone_message.txt", "r") as file:
file_prompt_standalone = file.read().replace("\n", "")
history = ''
for i, msg in enumerate(history_messages):
try:
if i == 0:
continue # Omit the prompt
if i % 2 == 0:
history += f'user: {msg["content"]}\n'
else:
history += f'assistant: {msg["content"]}\n'
except Exception as e:
print(e)
prompt_standalone = file_prompt_standalone.replace('HISTORY', history).replace('QUESTION', question)
standalone_msg_q = _call_gpt_standalone(prompt_standalone)
print(standalone_msg_q)
print("------------------")
return standalone_msg_q
def get_answer_text(question: str, chatbot: list[tuple[str, str]], history_messages):
"""
Gets the answer of the chatbot
"""
if len(chatbot) == 8:
message_output = 'Un placer haberte ayudado, hasta luego!'
else:
standalone_msg_q = _get_standalone_question(question, history_messages) # create standalone question or message
output_embedding = _call_embedding(standalone_msg_q) # create embedding of standalone question or message
best_results = _query_pinecone(output_embedding) # get nearest embeddings
final_context_prompt = _general_prompt(best_results) # create context/general prompt
message_output = _call_gpt(final_context_prompt, question)
if "Respuesta:" in message_output:
message_output.replace("Respuesta:", "")
print(history_messages)
return _update_elements(question, chatbot, message_output, history_messages) |