import re
import os
import datetime
from typing import Type, Dict, List, Tuple
import time
from itertools import compress
import pandas as pd
import numpy as np
# Model packages
import torch.cuda
from threading import Thread
from transformers import pipeline, TextIteratorStreamer
# Alternative model sources
#from dataclasses import asdict, dataclass
# Langchain functions
from langchain.prompts import PromptTemplate
from langchain_community.vectorstores import FAISS
from langchain_community.retrievers import SVMRetriever
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.docstore.document import Document
# For keyword extraction (not currently used)
#import nltk
#nltk.download('wordnet')
from nltk.corpus import stopwords
from nltk.tokenize import RegexpTokenizer
from nltk.stem import WordNetLemmatizer
from keybert import KeyBERT
# For Name Entity Recognition model
#from span_marker import SpanMarkerModel # Not currently used
# For BM25 retrieval
from gensim.corpora import Dictionary
from gensim.models import TfidfModel, OkapiBM25Model
from gensim.similarities import SparseMatrixSimilarity
from llama_cpp import Llama
from huggingface_hub import hf_hub_download
from chatfuncs.prompts import instruction_prompt_template_alpaca, instruction_prompt_mistral_orca, instruction_prompt_phi3, instruction_prompt_llama3
import gradio as gr
torch.cuda.empty_cache()
PandasDataFrame = Type[pd.DataFrame]
embeddings = None # global variable setup
vectorstore = None # global variable setup
model_type = None # global variable setup
max_memory_length = 0 # How long should the memory of the conversation last?
full_text = "" # Define dummy source text (full text) just to enable highlight function to load
model = [] # Define empty list for model functions to run
tokenizer = [] # Define empty list for model functions to run
## Highlight text constants
hlt_chunk_size = 12
hlt_strat = [" ", ". ", "! ", "? ", ": ", "\n\n", "\n", ", "]
hlt_overlap = 4
## Initialise NER model ##
ner_model = []#SpanMarkerModel.from_pretrained("tomaarsen/span-marker-mbert-base-multinerd") # Not currently used
## Initialise keyword model ##
# Used to pull out keywords from chat history to add to user queries behind the scenes
kw_model = pipeline("feature-extraction", model="sentence-transformers/all-MiniLM-L6-v2")
# Currently set gpu_layers to 0 even with cuda due to persistent bugs in implementation with cuda
if torch.cuda.is_available():
torch_device = "cuda"
gpu_layers = 100
else:
torch_device = "cpu"
gpu_layers = 0
print("Running on device:", torch_device)
threads = 8 #torch.get_num_threads()
print("CPU threads:", threads)
# Flan Alpaca (small, fast) Model parameters
temperature: float = 0.1
top_k: int = 3
top_p: float = 1
repetition_penalty: float = 1.15
flan_alpaca_repetition_penalty: float = 1.3
last_n_tokens: int = 64
max_new_tokens: int = 1024
seed: int = 42
reset: bool = False
stream: bool = True
threads: int = threads
batch_size:int = 256
context_length:int = 2048
sample = True
class CtransInitConfig_gpu:
def __init__(self,
last_n_tokens=last_n_tokens,
seed=seed,
n_threads=threads,
n_batch=batch_size,
n_ctx=4096,
n_gpu_layers=gpu_layers):
self.last_n_tokens = last_n_tokens
self.seed = seed
self.n_threads = n_threads
self.n_batch = n_batch
self.n_ctx = n_ctx
self.n_gpu_layers = n_gpu_layers
# self.stop: list[str] = field(default_factory=lambda: [stop_string])
def update_gpu(self, new_value):
self.n_gpu_layers = new_value
class CtransInitConfig_cpu(CtransInitConfig_gpu):
def __init__(self):
super().__init__()
self.n_gpu_layers = 0
gpu_config = CtransInitConfig_gpu()
cpu_config = CtransInitConfig_cpu()
class CtransGenGenerationConfig:
def __init__(self, temperature=temperature,
top_k=top_k,
top_p=top_p,
repeat_penalty=repetition_penalty,
seed=seed,
stream=stream,
max_tokens=max_new_tokens
):
self.temperature = temperature
self.top_k = top_k
self.top_p = top_p
self.repeat_penalty = repeat_penalty
self.seed = seed
self.max_tokens=max_tokens
self.stream = stream
def update_temp(self, new_value):
self.temperature = new_value
# Vectorstore funcs
def docs_to_faiss_save(docs_out:PandasDataFrame, embeddings=embeddings):
print(f"> Total split documents: {len(docs_out)}")
vectorstore_func = FAISS.from_documents(documents=docs_out, embedding=embeddings)
'''
#with open("vectorstore.pkl", "wb") as f:
#pickle.dump(vectorstore, f)
'''
#if Path(save_to).exists():
# vectorstore_func.save_local(folder_path=save_to)
#else:
# os.mkdir(save_to)
# vectorstore_func.save_local(folder_path=save_to)
global vectorstore
vectorstore = vectorstore_func
out_message = "Document processing complete"
#print(out_message)
#print(f"> Saved to: {save_to}")
return out_message
# Prompt functions
def base_prompt_templates(model_type = "Flan Alpaca (small, fast)"):
#EXAMPLE_PROMPT = PromptTemplate(
# template="\nCONTENT:\n\n{page_content}\n\nSOURCE: {source}\n\n",
# input_variables=["page_content", "source"],
#)
CONTENT_PROMPT = PromptTemplate(
template="{page_content}\n\n",#\n\nSOURCE: {source}\n\n",
input_variables=["page_content"]
)
# The main prompt:
if model_type == "Flan Alpaca (small, fast)":
INSTRUCTION_PROMPT=PromptTemplate(template=instruction_prompt_template_alpaca, input_variables=['question', 'summaries'])
elif model_type == "Phi 3 Mini (larger, slow)":
INSTRUCTION_PROMPT=PromptTemplate(template=instruction_prompt_phi3, input_variables=['question', 'summaries'])
return INSTRUCTION_PROMPT, CONTENT_PROMPT
def write_out_metadata_as_string(metadata_in):
metadata_string = [f"{' '.join(f'{k}: {v}' for k, v in d.items() if k != 'page_section')}" for d in metadata_in] # ['metadata']
return metadata_string
def generate_expanded_prompt(inputs: Dict[str, str], instruction_prompt, content_prompt, extracted_memory, vectorstore, embeddings, out_passages = 2): # ,
question = inputs["question"]
chat_history = inputs["chat_history"]
new_question_kworded = adapt_q_from_chat_history(question, chat_history, extracted_memory) # new_question_keywords,
docs_keep_as_doc, doc_df, docs_keep_out = hybrid_retrieval(new_question_kworded, vectorstore, embeddings, k_val = 25, out_passages = out_passages,
vec_score_cut_off = 0.85, vec_weight = 1, bm25_weight = 1, svm_weight = 1)#,
#vectorstore=globals()["vectorstore"], embeddings=globals()["embeddings"])
#print(docs_keep_as_doc)
#print(doc_df)
if (not docs_keep_as_doc) | (doc_df.empty):
sorry_prompt = """Say 'Sorry, there is no relevant information to answer this question.'.
RESPONSE:"""
return sorry_prompt, "No relevant sources found.", new_question_kworded
# Expand the found passages to the neighbouring context
file_type = determine_file_type(doc_df['meta_url'][0])
# Only expand passages if not tabular data
if (file_type != ".csv") & (file_type != ".xlsx"):
docs_keep_as_doc, doc_df = get_expanded_passages(vectorstore, docs_keep_out, width=3)
# Build up sources content to add to user display
doc_df['meta_clean'] = write_out_metadata_as_string(doc_df["metadata"]) # [f"{' '.join(f'{k}: {v}' for k, v in d.items() if k != 'page_section')}" for d in doc_df['metadata']]
# Remove meta text from the page content if it already exists there
doc_df['page_content_no_meta'] = doc_df.apply(lambda row: row['page_content'].replace(row['meta_clean'] + ". ", ""), axis=1)
doc_df['content_meta'] = doc_df['meta_clean'].astype(str) + ".
" + doc_df['page_content_no_meta'].astype(str)
#modified_page_content = [f" Document {i+1} - {word}" for i, word in enumerate(doc_df['page_content'])]
modified_page_content = [f" Document {i+1} - {word}" for i, word in enumerate(doc_df['content_meta'])]
docs_content_string = '
'.join(modified_page_content)
sources_docs_content_string = '
'.join(doc_df['content_meta'])#.replace(" "," ")#.strip()
instruction_prompt_out = instruction_prompt.format(question=new_question_kworded, summaries=docs_content_string)
print('Final prompt is: ')
print(instruction_prompt_out)
return instruction_prompt_out, sources_docs_content_string, new_question_kworded
def create_full_prompt(user_input, history, extracted_memory, vectorstore, embeddings, model_type, out_passages):
if not user_input.strip():
return history, "", "Respond with 'Please enter a question.' RESPONSE:"
#if chain_agent is None:
# history.append((user_input, "Please click the button to submit the Huggingface API key before using the chatbot (top right)"))
# return history, history, "", ""
print("\n==== date/time: " + str(datetime.datetime.now()) + " ====")
print("User input: " + user_input)
history = history or []
# Create instruction prompt
instruction_prompt, content_prompt = base_prompt_templates(model_type=model_type)
instruction_prompt_out, docs_content_string, new_question_kworded =\
generate_expanded_prompt({"question": user_input, "chat_history": history}, #vectorstore,
instruction_prompt, content_prompt, extracted_memory, vectorstore, embeddings, out_passages)
history.append(user_input)
print("Output history is:")
print(history)
print("Final prompt to model is:")
print(instruction_prompt_out)
return history, docs_content_string, instruction_prompt_out
# Chat functions
def produce_streaming_answer_chatbot(history, full_prompt, model_type,
temperature=temperature,
max_new_tokens=max_new_tokens,
sample=sample,
repetition_penalty=repetition_penalty,
top_p=top_p,
top_k=top_k
):
#print("Model type is: ", model_type)
#if not full_prompt.strip():
# if history is None:
# history = []
# return history
if model_type == "Flan Alpaca (small, fast)":
# Get the model and tokenizer, and tokenize the user text.
model_inputs = tokenizer(text=full_prompt, return_tensors="pt", return_attention_mask=False).to(torch_device)
# Start generation on a separate thread, so that we don't block the UI. The text is pulled from the streamer
# in the main thread. Adds timeout to the streamer to handle exceptions in the generation thread.
streamer = TextIteratorStreamer(tokenizer, timeout=120., skip_prompt=True, skip_special_tokens=True)
generate_kwargs = dict(
model_inputs,
streamer=streamer,
max_new_tokens=max_new_tokens,
do_sample=sample,
repetition_penalty=repetition_penalty,
top_p=top_p,
temperature=temperature,
top_k=top_k
)
print(generate_kwargs)
t = Thread(target=model.generate, kwargs=generate_kwargs)
t.start()
# Pull the generated text from the streamer, and update the model output.
start = time.time()
NUM_TOKENS=0
print('-'*4+'Start Generation'+'-'*4)
history[-1][1] = ""
for new_text in streamer:
try:
if new_text == None: new_text = ""
history[-1][1] += new_text
NUM_TOKENS+=1
yield history
except Exception as e:
print(f"Error during text generation: {e}")
time_generate = time.time() - start
print('\n')
print('-'*4+'End Generation'+'-'*4)
print(f'Num of generated tokens: {NUM_TOKENS}')
print(f'Time for complete generation: {time_generate}s')
print(f'Tokens per secound: {NUM_TOKENS/time_generate}')
print(f'Time per token: {(time_generate/NUM_TOKENS)*1000}ms')
elif model_type == "Phi 3 Mini (larger, slow)":
#tokens = model.tokenize(full_prompt)
gen_config = CtransGenGenerationConfig()
gen_config.update_temp(temperature)
print(vars(gen_config))
# Pull the generated text from the streamer, and update the model output.
start = time.time()
NUM_TOKENS=0
print('-'*4+'Start Generation'+'-'*4)
output = model(
full_prompt, **vars(gen_config))
history[-1][1] = ""
for out in output:
if "choices" in out and len(out["choices"]) > 0 and "text" in out["choices"][0]:
history[-1][1] += out["choices"][0]["text"]
NUM_TOKENS+=1
yield history
else:
print(f"Unexpected output structure: {out}")
time_generate = time.time() - start
print('\n')
print('-'*4+'End Generation'+'-'*4)
print(f'Num of generated tokens: {NUM_TOKENS}')
print(f'Time for complete generation: {time_generate}s')
print(f'Tokens per secound: {NUM_TOKENS/time_generate}')
print(f'Time per token: {(time_generate/NUM_TOKENS)*1000}ms')
# Chat helper functions
def adapt_q_from_chat_history(question, chat_history, extracted_memory, keyword_model=""):#keyword_model): # new_question_keywords,
chat_history_str, chat_history_first_q, chat_history_first_ans, max_memory_length = _get_chat_history(chat_history)
if chat_history_str:
# Keyword extraction is now done in the add_inputs_to_history function
#remove_q_stopwords(str(chat_history_first_q) + " " + str(chat_history_first_ans))
new_question_kworded = str(extracted_memory) + ". " + question #+ " " + new_question_keywords
#extracted_memory + " " + question
else:
new_question_kworded = question #new_question_keywords
#print("Question output is: " + new_question_kworded)
return new_question_kworded
def determine_file_type(file_path):
"""
Determine the file type based on its extension.
Parameters:
file_path (str): Path to the file.
Returns:
str: File extension (e.g., '.pdf', '.docx', '.txt', '.html').
"""
return os.path.splitext(file_path)[1].lower()
def create_doc_df(docs_keep_out):
# Extract content and metadata from 'winning' passages.
content=[]
meta=[]
meta_url=[]
page_section=[]
score=[]
doc_df = pd.DataFrame()
for item in docs_keep_out:
content.append(item[0].page_content)
meta.append(item[0].metadata)
meta_url.append(item[0].metadata['source'])
file_extension = determine_file_type(item[0].metadata['source'])
if (file_extension != ".csv") & (file_extension != ".xlsx"):
page_section.append(item[0].metadata['page_section'])
else: page_section.append("")
score.append(item[1])
# Create df from 'winning' passages
doc_df = pd.DataFrame(list(zip(content, meta, page_section, meta_url, score)),
columns =['page_content', 'metadata', 'page_section', 'meta_url', 'score'])
docs_content = doc_df['page_content'].astype(str)
doc_df['full_url'] = "https://" + doc_df['meta_url']
return doc_df
def hybrid_retrieval(new_question_kworded, vectorstore, embeddings, k_val, out_passages,
vec_score_cut_off, vec_weight, bm25_weight, svm_weight): # ,vectorstore, embeddings
#vectorstore=globals()["vectorstore"]
#embeddings=globals()["embeddings"]
doc_df = pd.DataFrame()
docs = vectorstore.similarity_search_with_score(new_question_kworded, k=k_val)
print("Docs from similarity search:")
print(docs)
# Keep only documents with a certain score
docs_len = [len(x[0].page_content) for x in docs]
docs_scores = [x[1] for x in docs]
# Only keep sources that are sufficiently relevant (i.e. similarity search score below threshold below)
score_more_limit = pd.Series(docs_scores) < vec_score_cut_off
docs_keep = list(compress(docs, score_more_limit))
if not docs_keep:
return [], pd.DataFrame(), []
# Only keep sources that are at least 100 characters long
length_more_limit = pd.Series(docs_len) >= 100
docs_keep = list(compress(docs_keep, length_more_limit))
if not docs_keep:
return [], pd.DataFrame(), []
docs_keep_as_doc = [x[0] for x in docs_keep]
docs_keep_length = len(docs_keep_as_doc)
if docs_keep_length == 1:
content=[]
meta_url=[]
score=[]
for item in docs_keep:
content.append(item[0].page_content)
meta_url.append(item[0].metadata['source'])
score.append(item[1])
# Create df from 'winning' passages
doc_df = pd.DataFrame(list(zip(content, meta_url, score)),
columns =['page_content', 'meta_url', 'score'])
docs_content = doc_df['page_content'].astype(str)
docs_url = doc_df['meta_url']
return docs_keep_as_doc, docs_content, docs_url
# Check for if more docs are removed than the desired output
if out_passages > docs_keep_length:
out_passages = docs_keep_length
k_val = docs_keep_length
vec_rank = [*range(1, docs_keep_length+1)]
vec_score = [(docs_keep_length/x)*vec_weight for x in vec_rank]
print("Number of documents remaining: ", docs_keep_length)
# 2nd level check on retrieved docs with BM25
content_keep=[]
for item in docs_keep:
content_keep.append(item[0].page_content)
corpus = corpus = [doc.lower().split() for doc in content_keep]
dictionary = Dictionary(corpus)
bm25_model = OkapiBM25Model(dictionary=dictionary)
bm25_corpus = bm25_model[list(map(dictionary.doc2bow, corpus))]
bm25_index = SparseMatrixSimilarity(bm25_corpus, num_docs=len(corpus), num_terms=len(dictionary),
normalize_queries=False, normalize_documents=False)
query = new_question_kworded.lower().split()
tfidf_model = TfidfModel(dictionary=dictionary, smartirs='bnn') # Enforce binary weighting of queries
tfidf_query = tfidf_model[dictionary.doc2bow(query)]
similarities = np.array(bm25_index[tfidf_query])
#print(similarities)
temp = similarities.argsort()
ranks = np.arange(len(similarities))[temp.argsort()][::-1]
# Pair each index with its corresponding value
pairs = list(zip(ranks, docs_keep_as_doc))
# Sort the pairs by the indices
pairs.sort()
# Extract the values in the new order
bm25_result = [value for ranks, value in pairs]
bm25_rank=[]
bm25_score = []
for vec_item in docs_keep:
x = 0
for bm25_item in bm25_result:
x = x + 1
if bm25_item.page_content == vec_item[0].page_content:
bm25_rank.append(x)
bm25_score.append((docs_keep_length/x)*bm25_weight)
# 3rd level check on retrieved docs with SVM retriever
svm_retriever = SVMRetriever.from_texts(content_keep, embeddings, k = k_val)
svm_result = svm_retriever.invoke(new_question_kworded)
svm_rank=[]
svm_score = []
for vec_item in docs_keep:
x = 0
for svm_item in svm_result:
x = x + 1
if svm_item.page_content == vec_item[0].page_content:
svm_rank.append(x)
svm_score.append((docs_keep_length/x)*svm_weight)
## Calculate final score based on three ranking methods
final_score = [a + b + c for a, b, c in zip(vec_score, bm25_score, svm_score)]
final_rank = [sorted(final_score, reverse=True).index(x)+1 for x in final_score]
# Force final_rank to increment by 1 each time
final_rank = list(pd.Series(final_rank).rank(method='first'))
#print("final rank: " + str(final_rank))
#print("out_passages: " + str(out_passages))
best_rank_index_pos = []
for x in range(1,out_passages+1):
try:
best_rank_index_pos.append(final_rank.index(x))
except IndexError: # catch the error
pass
# Adjust best_rank_index_pos to
best_rank_pos_series = pd.Series(best_rank_index_pos)
docs_keep_out = [docs_keep[i] for i in best_rank_index_pos]
# Keep only 'best' options
docs_keep_as_doc = [x[0] for x in docs_keep_out]
# Make df of best options
doc_df = create_doc_df(docs_keep_out)
return docs_keep_as_doc, doc_df, docs_keep_out
def get_expanded_passages(vectorstore, docs, width):
"""
Extracts expanded passages based on given documents and a width for context.
Parameters:
- vectorstore: The primary data source.
- docs: List of documents to be expanded.
- width: Number of documents to expand around a given document for context.
Returns:
- expanded_docs: List of expanded Document objects.
- doc_df: DataFrame representation of expanded_docs.
"""
from collections import defaultdict
def get_docs_from_vstore(vectorstore):
vector = vectorstore.docstore._dict
return list(vector.items())
def extract_details(docs_list):
docs_list_out = [tup[1] for tup in docs_list]
content = [doc.page_content for doc in docs_list_out]
meta = [doc.metadata for doc in docs_list_out]
return ''.join(content), meta[0], meta[-1]
def get_parent_content_and_meta(vstore_docs, width, target):
#target_range = range(max(0, target - width), min(len(vstore_docs), target + width + 1))
target_range = range(max(0, target), min(len(vstore_docs), target + width + 1)) # Now only selects extra passages AFTER the found passage
parent_vstore_out = [vstore_docs[i] for i in target_range]
content_str_out, meta_first_out, meta_last_out = [], [], []
for _ in parent_vstore_out:
content_str, meta_first, meta_last = extract_details(parent_vstore_out)
content_str_out.append(content_str)
meta_first_out.append(meta_first)
meta_last_out.append(meta_last)
return content_str_out, meta_first_out, meta_last_out
def merge_dicts_except_source(d1, d2):
merged = {}
for key in d1:
if key != "source":
merged[key] = str(d1[key]) + " to " + str(d2[key])
else:
merged[key] = d1[key] # or d2[key], based on preference
return merged
def merge_two_lists_of_dicts(list1, list2):
return [merge_dicts_except_source(d1, d2) for d1, d2 in zip(list1, list2)]
# Step 1: Filter vstore_docs
vstore_docs = get_docs_from_vstore(vectorstore)
doc_sources = {doc.metadata['source'] for doc, _ in docs}
vstore_docs = [(k, v) for k, v in vstore_docs if v.metadata.get('source') in doc_sources]
# Step 2: Group by source and proceed
vstore_by_source = defaultdict(list)
for k, v in vstore_docs:
vstore_by_source[v.metadata['source']].append((k, v))
expanded_docs = []
for doc, score in docs:
search_source = doc.metadata['source']
#if file_type == ".csv" | file_type == ".xlsx":
# content_str, meta_first, meta_last = get_parent_content_and_meta(vstore_by_source[search_source], 0, search_index)
#else:
search_section = doc.metadata['page_section']
parent_vstore_meta_section = [doc.metadata['page_section'] for _, doc in vstore_by_source[search_source]]
search_index = parent_vstore_meta_section.index(search_section) if search_section in parent_vstore_meta_section else -1
content_str, meta_first, meta_last = get_parent_content_and_meta(vstore_by_source[search_source], width, search_index)
meta_full = merge_two_lists_of_dicts(meta_first, meta_last)
expanded_doc = (Document(page_content=content_str[0], metadata=meta_full[0]), score)
expanded_docs.append(expanded_doc)
doc_df = pd.DataFrame()
doc_df = create_doc_df(expanded_docs) # Assuming you've defined the 'create_doc_df' function elsewhere
return expanded_docs, doc_df
def highlight_found_text(search_text: str, full_text: str, hlt_chunk_size:int=hlt_chunk_size, hlt_strat:List=hlt_strat, hlt_overlap:int=hlt_overlap) -> str:
"""
Highlights occurrences of search_text within full_text.
Parameters:
- search_text (str): The text to be searched for within full_text.
- full_text (str): The text within which search_text occurrences will be highlighted.
Returns:
- str: A string with occurrences of search_text highlighted.
Example:
>>> highlight_found_text("world", "Hello, world! This is a test. Another world awaits.")
'Hello, world! This is a test. Another world awaits.'
"""
def extract_text_from_input(text, i=0):
if isinstance(text, str):
return text.replace(" ", " ").strip()
elif isinstance(text, list):
return text[i][0].replace(" ", " ").strip()
else:
return ""
def extract_search_text_from_input(text):
if isinstance(text, str):
return text.replace(" ", " ").strip()
elif isinstance(text, list):
return text[-1][1].replace(" ", " ").strip()
else:
return ""
full_text = extract_text_from_input(full_text)
search_text = extract_search_text_from_input(search_text)
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=hlt_chunk_size,
separators=hlt_strat,
chunk_overlap=hlt_overlap,
)
sections = text_splitter.split_text(search_text)
found_positions = {}
for x in sections:
text_start_pos = 0
while text_start_pos != -1:
text_start_pos = full_text.find(x, text_start_pos)
if text_start_pos != -1:
found_positions[text_start_pos] = text_start_pos + len(x)
text_start_pos += 1
# Combine overlapping or adjacent positions
sorted_starts = sorted(found_positions.keys())
combined_positions = []
if sorted_starts:
current_start, current_end = sorted_starts[0], found_positions[sorted_starts[0]]
for start in sorted_starts[1:]:
if start <= (current_end + 10):
current_end = max(current_end, found_positions[start])
else:
combined_positions.append((current_start, current_end))
current_start, current_end = start, found_positions[start]
combined_positions.append((current_start, current_end))
# Construct pos_tokens
pos_tokens = []
prev_end = 0
for start, end in combined_positions:
if end-start > 15: # Only combine if there is a significant amount of matched text. Avoids picking up single words like 'and' etc.
pos_tokens.append(full_text[prev_end:start])
pos_tokens.append('' + full_text[start:end] + '')
prev_end = end
pos_tokens.append(full_text[prev_end:])
return "".join(pos_tokens)
# # Chat history functions
def clear_chat(chat_history_state, sources, chat_message, current_topic):
chat_history_state = []
sources = ''
chat_message = ''
current_topic = ''
return chat_history_state, sources, chat_message, current_topic
def _get_chat_history(chat_history: List[Tuple[str, str]], max_memory_length:int = max_memory_length): # Limit to last x interactions only
if (not chat_history) | (max_memory_length == 0):
chat_history = []
if len(chat_history) > max_memory_length:
chat_history = chat_history[-max_memory_length:]
#print(chat_history)
first_q = ""
first_ans = ""
for human_s, ai_s in chat_history:
first_q = human_s
first_ans = ai_s
#print("Text to keyword extract: " + first_q + " " + first_ans)
break
conversation = ""
for human_s, ai_s in chat_history:
human = f"Human: " + human_s
ai = f"Assistant: " + ai_s
conversation += "\n" + "\n".join([human, ai])
return conversation, first_q, first_ans, max_memory_length
def add_inputs_answer_to_history(user_message, history, current_topic):
if history is None:
history = [("","")]
#history.append((user_message, [-1]))
chat_history_str, chat_history_first_q, chat_history_first_ans, max_memory_length = _get_chat_history(history)
# Only get the keywords for the first question and response, or do it every time if over 'max_memory_length' responses in the conversation
if (len(history) == 1) | (len(history) > max_memory_length):
#print("History after appending is:")
#print(history)
first_q_and_first_ans = str(chat_history_first_q) + " " + str(chat_history_first_ans)
#ner_memory = remove_q_ner_extractor(first_q_and_first_ans)
keywords = keybert_keywords(first_q_and_first_ans, n = 8, kw_model=kw_model)
#keywords.append(ner_memory)
# Remove duplicate words while preserving order
ordered_tokens = set()
result = []
for word in keywords:
if word not in ordered_tokens:
ordered_tokens.add(word)
result.append(word)
extracted_memory = ' '.join(result)
else: extracted_memory=current_topic
print("Extracted memory is:")
print(extracted_memory)
return history, extracted_memory
# Keyword functions
def remove_q_stopwords(question): # Remove stopwords from question. Not used at the moment
# Prepare keywords from question by removing stopwords
text = question.lower()
# Remove numbers
text = re.sub('[0-9]', '', text)
tokenizer = RegexpTokenizer(r'\w+')
text_tokens = tokenizer.tokenize(text)
#text_tokens = word_tokenize(text)
tokens_without_sw = [word for word in text_tokens if not word in stopwords]
# Remove duplicate words while preserving order
ordered_tokens = set()
result = []
for word in tokens_without_sw:
if word not in ordered_tokens:
ordered_tokens.add(word)
result.append(word)
new_question_keywords = ' '.join(result)
return new_question_keywords
def remove_q_ner_extractor(question):
predict_out = ner_model.predict(question)
predict_tokens = [' '.join(v for k, v in d.items() if k == 'span') for d in predict_out]
# Remove duplicate words while preserving order
ordered_tokens = set()
result = []
for word in predict_tokens:
if word not in ordered_tokens:
ordered_tokens.add(word)
result.append(word)
new_question_keywords = ' '.join(result).lower()
return new_question_keywords
def apply_lemmatize(text, wnl=WordNetLemmatizer()):
def prep_for_lemma(text):
# Remove numbers
text = re.sub('[0-9]', '', text)
print(text)
tokenizer = RegexpTokenizer(r'\w+')
text_tokens = tokenizer.tokenize(text)
#text_tokens = word_tokenize(text)
return text_tokens
tokens = prep_for_lemma(text)
def lem_word(word):
if len(word) > 3: out_word = wnl.lemmatize(word)
else: out_word = word
return out_word
return [lem_word(token) for token in tokens]
def keybert_keywords(text, n, kw_model):
tokens_lemma = apply_lemmatize(text)
lemmatised_text = ' '.join(tokens_lemma)
keywords_text = KeyBERT(model=kw_model).extract_keywords(lemmatised_text, stop_words='english', top_n=n,
keyphrase_ngram_range=(1, 1))
keywords_list = [item[0] for item in keywords_text]
return keywords_list
# Gradio functions
def turn_off_interactivity(user_message, history):
return gr.update(value="", interactive=False), history + [[user_message, None]]
def restore_interactivity():
return gr.update(interactive=True)
def update_message(dropdown_value):
return gr.Textbox(value=dropdown_value)
def hide_block():
return gr.Radio(visible=False)
# Vote function
def vote(data: gr.LikeData, chat_history, instruction_prompt_out, model_type):
import os
import pandas as pd
chat_history_last = str(str(chat_history[-1][0]) + " - " + str(chat_history[-1][1]))
response_df = pd.DataFrame(data={"thumbs_up":data.liked,
"chosen_response":data.value,
"input_prompt":instruction_prompt_out,
"chat_history":chat_history_last,
"model_type": model_type,
"date_time": pd.Timestamp.now()}, index=[0])
if data.liked:
print("You upvoted this response: " + data.value)
if os.path.isfile("thumbs_up_data.csv"):
existing_thumbs_up_df = pd.read_csv("thumbs_up_data.csv")
thumbs_up_df_concat = pd.concat([existing_thumbs_up_df, response_df], ignore_index=True).drop("Unnamed: 0",axis=1, errors="ignore")
thumbs_up_df_concat.to_csv("thumbs_up_data.csv")
else:
response_df.to_csv("thumbs_up_data.csv")
else:
print("You downvoted this response: " + data.value)
if os.path.isfile("thumbs_down_data.csv"):
existing_thumbs_down_df = pd.read_csv("thumbs_down_data.csv")
thumbs_down_df_concat = pd.concat([existing_thumbs_down_df, response_df], ignore_index=True).drop("Unnamed: 0",axis=1, errors="ignore")
thumbs_down_df_concat.to_csv("thumbs_down_data.csv")
else:
response_df.to_csv("thumbs_down_data.csv")