Phobos 🪐
This is a open tuned model that was fitted onto a RAG pipeline using all-mpnet-base-v2.
In order to chat, please say 'gen phobos' = General Question you have of any topic. Say 'phobos' for questions specifically medical.
'''
# API keys
api_key = os.getenv('OPEN_AI_API_KEY')
df_embeds = pd.read_csv("chunks_tokenized.csv")
df_embeds["embeddings"] = df_embeds["embeddings"].apply(lambda x: np.fromstring(x.strip("[]"), sep=" "))
embeds_dict = df_embeds.to_dict(orient="records")
# convert into tensors
embeddings = torch.tensor(np.array(df_embeds["embeddings"].to_list()), dtype=torch.float32).to('cuda')
# Make a text wrapper
def text_wrapper(text):
"""
Wraps the text that will pass here
"""
clean_text = textwrap.fill(text, 80)
print(clean_text)
# Let's first get the embedding model
embedding_model = SentenceTransformer(model_name_or_path="all-mpnet-base-v2",
device='cuda')
# functionize RAG Pipeline
def rag_pipeline(query,
embedding_model,
embeddings,
device: str,
chunk_min_token: list):
"""
Grabs a query and retrieve data all in passages, augments them, than it
it outputs the top 5 relevant results regarding query's meaning using dot scores.
"""
# Retrieval
query_embeddings = embedding_model.encode(query, convert_to_tensor=True).to(device)
# Augmentation
dot_scores = util.dot_score(a=query_embeddings, b=embeddings)[0]
# Output
scores, indices = torch.topk(dot_scores, k=5)
counting = 0
for score, idx in zip(scores, indices):
counting+=1
clean_score = score.item()*100
print(f"For the ({counting}) result has a score: {round(clean_score, 2)}%")
print(f"On index: {idx}")
print(f"Relevant Text:\n")
print(f"{text_wrapper(chunk_min_token[idx]['sentence_chunk'])}\n")
# Message request to gpt
def message_request_to_model(input_text: str):
"""
Message to pass to the request on API
"""
message_to_model = [
{"role": "system", "content": "You are a helpful assistant called 'Phobos'."},
{"role": "user", "content": input_text}, # This must be in string format or else the request won't be successful
]
return message_to_model
# Functionize API request from the very beginning as calling gpt for the first time
def request_gpt_model(input_text,
temperature,
message_to_model_api,
model: str="gpt-3.5-turbo"):
"""
This will pass in a request to the gpt api with the messages and
will take the whole prompt generated as input as intructions to model
and output the similiar meaning on the output.
"""
# Create client
client = OpenAI(api_key=api_key)
# Make a request, for the input prompt
response = client.chat.completions.create(
model=model,
messages=message_to_model_api,
temperature=temperature,
)
# Output the message in readable format
output = response.choices[0].message.content
json_response = json.dumps(json.loads(response.model_dump_json()), indent=4)
# print(f"{text_wrapper(output)}")
print(output)
return output, json_response
# Functionize saving output to file
def save_log_models_activity(query, prompt, continue_question, output, cont_output, embeds_dict, json_response,
model, rag_pipeline, message_request_to_model, indices, embedding_model, source_directed: str):
"""
This will save the models input and output interaction, onto
a txt file, for each request, labeling model that was used.
What sort of embedding process, pipeline that was used and
date and time it was ran
"""
# If there is a follow up question:
input_query = ""
if continue_question != "":
input_query += continue_question
else:
input_query += query
clean_query = re.sub(r'[^\w\s]', '', input_query).replace(' ', '_')
file_path = os.path.join("./logfiles/may-2024/", f"{clean_query}.txt")
#Open the file in write mode
with open(file_path, 'w', encoding='utf-8') as file:
file.write(f"Original Query: {query}\n\n")
if prompt != "":
file.write(f"Base Prompt: {prompt}\n\n")
if continue_question != "":
file.write(f"Follow up question:\n\n{continue_question}\n\n")
file.write(f"Output:\n\n {cont_output}")
else:
file.write(f"Output:\n\n{output}\n\n")
# Json response
file.write(f"\n\nJson format response: {json_response}\n\n")
for idx in indices:
# Let's log the models activity in txt file
if rag_pipeline:
file.write(f"{source_directed}")
file.write(f"\n\nPipeline Used: RAG\n")
file.write(f"Embedding Model used on tokenizing pipeline:\n\n{embedding_model}\n")
file.write(f"\nRelevant Passages: {embeds_dict[idx]['sentence_chunk']}\n\n")
break
file.write(f"Model used: {model}\n")
# file.write(f"{message_request_to_model}")
today = date.today()
current_time = datetime.now().time()
file.write(f"Date: {today.strftime('%B %d, %Y')}\nTime: {current_time.strftime('%H:%M:%S')}\n\n")
# retrieve rag resources such as score and indices
def rag_resources(query: str,
device: str="cuda"):
"""
Extracts only the scores and indices of the top 5 best results
according to dot scores on query.
"""
# Retrieval
query_embeddings = embedding_model.encode(query, convert_to_tensor=True).to(device)
# Augmentation
dot_scores = util.dot_score(a=query_embeddings, b=embeddings)[0]
# Output
scores, indices = torch.topk(dot_scores, k=5)
return scores, indices
# Format the prompt
def rag_prompt_formatter(prompt: str,
prev_quest: list,
context_items: List[Dict[str, Any]]):
"""
Format the base prompt with the user query.
"""
# Convert the list into string
prev_questions_str ='\n'.join(prev_quest) # convert to string so we can later format on base_prompt
context = "- " + "\n- ".join(i["sentence_chunk"] for i in context_items)
base_prompt = """In this text, you will act as supportive medical assistant.
Give yourself room to think.
Explain each topic with facts and also suggestions based on the users needs.
Keep your answers thorough but practical.
\nHere are the past questions and answers you gave to the user, to serve you as a memory:
{previous_questions}
\nYou as the assistant will recieve context items for retrieving information.
\nNow use the following context items to answer the user query. Be advised if the user does not give you
any query that seems medical, DO NOT extract the relevant passages:
{context}
\nRelevant passages: Please extract the context items that helped you answer the user's question