api.py

import gc
import os
import re
import shutil
import urllib.request
from pathlib import Path
from tempfile import NamedTemporaryFile

import fitz
import numpy as np
import openai
import torch
import torch.nn.functional as F
from fastapi import UploadFile
from lcserve import serving
from optimum.bettertransformer import BetterTransformer
from sklearn import svm
from sklearn.cluster import KMeans
from sklearn.metrics import pairwise_distances_argmin_min
from torch import Tensor
from transformers import AutoModel, AutoTokenizer

recommender = None


def download_pdf(url, output_path):
    urllib.request.urlretrieve(url, output_path)


def preprocess(text):
    text = text.replace("-\n", "")
    text = text.replace("\n", " ")
    text = re.sub("\s+", " ", text)
    return text


def get_margin(pdf):
    page = pdf[0]
    page_size = page.mediabox
    margin_hor = page.mediabox.width * 0.05
    margin_ver = page.mediabox.height * 0.05
    margin_size = page_size + (margin_hor, margin_ver, -margin_hor, -margin_ver)
    return margin_size


def pdf_to_text(path, start_page=1, end_page=None):
    doc = fitz.open(path)
    total_pages = doc.page_count

    if end_page is None:
        end_page = total_pages

    text_list = []
    margin_size = get_margin(doc)
    for i in range(start_page - 1, end_page):
        page = doc[i]
        page.set_cropbox(margin_size)
        text = page.get_text("text")
        text = preprocess(text)
        text_list.append(text)

    doc.close()
    return text_list


def text_to_chunks(texts, word_length=150, start_page=1):
    text_toks = [t.split(" ") for t in texts]
    page_nums = []
    chunks = []

    for idx, words in enumerate(text_toks):
        for i in range(0, len(words), word_length):
            chunk = words[i : i + word_length]
            if (
                (i + word_length) > len(words)
                and (len(chunk) < word_length)
                and (len(text_toks) != (idx + 1))
            ):
                text_toks[idx + 1] = chunk + text_toks[idx + 1]
                continue
            chunk = " ".join(chunk).strip()
            chunk = f"[Page no. {idx+start_page}]" + " " + '"' + chunk + '"'
            chunks.append(chunk)
    return chunks


class SemanticSearch:
    def __init__(self, embedding_model):
        self.tokenizer = AutoTokenizer.from_pretrained(f"intfloat/{embedding_model}")
        self.model = AutoModel.from_pretrained(
            f"intfloat/{embedding_model}",
            # cache_dir =,
        )
        self.model = BetterTransformer.transform(self.model, keep_original_model=True)

        # set device
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model = self.model.to(self.device)
        self.fitted = False

    def fit(self, data, batch_size=32, n_neighbors=5):
        self.data = data
        self.embeddings = self.get_text_embedding(self.data, batch_size=batch_size)
        self.fitted = True

    def __call__(self, text, return_data=True):
        self.inp_emb = self.get_text_embedding([text], prefix="query")
        self.matches = self.run_svm(self.inp_emb, self.embeddings)

        if return_data:
            # return 5 first match, first index is query, so it has to be skipped
            return [self.data[i - 1] for i in self.matches[1:6]]

        else:
            return self.matches

    def average_pool(
        self, last_hidden_states: Tensor, attention_mask: Tensor
    ) -> Tensor:
        self.last_hidden = last_hidden_states.masked_fill(
            ~attention_mask[..., None].bool(), 0.0
        )
        return self.last_hidden.sum(dim=1) / attention_mask.sum(dim=1)[..., None]

    def get_text_embedding(self, texts, prefix="passage", batch_size=32):
        # Tokenize the input texts
        texts = [f"{prefix}: {text}" for text in texts]
        batch_dict = self.tokenizer(
            texts, max_length=512, padding=True, truncation=True, return_tensors="pt"
        ).to(self.device)

        with torch.no_grad():
            outputs = self.model(**batch_dict)

        embeddings = self.average_pool(
            outputs.last_hidden_state, batch_dict["attention_mask"]
        )

        # Normalize embeddings
        embeddings = F.normalize(embeddings, p=2, dim=1)

        # Convert pytorch tensor to numpy array (no grad)
        if self.device == "cuda":
            embeddings = embeddings.detach().cpu().clone().numpy()
        else:
            embeddings = embeddings.detach().numpy()
        return embeddings

    def run_svm(self, query_emb, passage_emb):
        joined_emb = np.concatenate((query_emb, passage_emb))

        # create var for SVM label
        y = np.zeros(joined_emb.shape[0])
        # mark query as a positive example
        y[0] = 1

        # declare SVM
        clf = svm.LinearSVC(
            class_weight="balanced", verbose=False, max_iter=10000, tol=1e-6, C=0.1
        )
        # train (Exemplar) SVM
        clf.fit(joined_emb, y)

        # infer on original data
        similarities = clf.decision_function(joined_emb)
        sorted_ix = np.argsort(-similarities)
        return sorted_ix

    def summarize(self):
        n_clusters = int(np.ceil(len(self.embeddings)**0.5))
        # max cluster 5 (reserve token)
        n_clusters = n_clusters if n_clusters <= 5 else 5
        kmeans = KMeans(n_clusters=n_clusters, random_state=23)
        kmeans = kmeans.fit(self.embeddings)

        avg = []
        closest = []
        for j in range(n_clusters):
            # find first chunk index of every cluster
            idx = np.where(kmeans.labels_ == j)[0]
            avg.append(np.mean(idx))
        # find chunk that is closest to the centroid
        closest, _ = pairwise_distances_argmin_min(kmeans.cluster_centers_,
                                                   self.embeddings)
        ordering = sorted(range(n_clusters), key=lambda k: avg[k])
        # concat representative chunks
        summary = [self.data[i] for i in [closest[idx] for idx in ordering]]
        return summary


def clear_cache():
    global recommender
    if "recommender" in globals():
        del recommender
    gc.collect()
    if torch.cuda.is_available():
        return torch.cuda.empty_cache()


def load_recommender(path, embedding_model, rebuild_embedding, start_page=1):
    global recommender
    if rebuild_embedding:
        clear_cache()
        recommender = None
    if recommender is None:
        recommender = SemanticSearch(embedding_model)
    if recommender.fitted:
        return "Corpus Loaded."
    else:
        texts = pdf_to_text(path, start_page=start_page)
        chunks = text_to_chunks(texts, start_page=start_page)
        recommender.fit(chunks)
    return "Corpus Loaded."


def generate_text(openai_key, prompt, model="gpt-3.5-turbo"):
    openai.api_key = openai_key
    completions = openai.ChatCompletion.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=512,
        n=1,
        stop=None,
        temperature=0.7,
    )
    message = f"{prompt}###{completions.choices[0].message.content}###{completions.usage.total_tokens}###{completions.model}"
    return message

def generate_answer(question, gpt_model, openai_key):
    topn_chunks = recommender(question)
    prompt = ""
    prompt += "search results:\n\n"
    for c in topn_chunks:
        prompt += c + "\n\n"

    prompt += (
        "Instructions: Compose a comprehensive reply to the query using the search results given. "
        "Cite each reference using [ Page Number] notation (every result has this number at the beginning). "
        "Citation should be done at the end of each sentence. If the search results mention multiple subjects "
        "with the same name, create separate answers for each. Only include information found in the results and "
        "don't add any additional information. Make sure the answer is correct and don't output false content. "
        "If the text does not relate to the query, simply state 'Text Not Found in PDF'. Ignore outlier "
        "search results which has nothing to do with the question. Only answer what is asked. The "
        "answer should be short and concise. Answer step-by-step.\n\n"
    )

    prompt += f"Query: {question}"
    answer = generate_text(openai_key, prompt, gpt_model)
    return answer

def generate_summary(gpt_model, openai_key):
    topn_chunks = recommender.summarize()
    prompt = ""
    prompt += (
        "Summarize the highlights of the search results and output a summary in bulletpoints. "
        "Do not write anything before the bulletpoints. "
        "Cite each reference using [Page no.] notation (every result has this number at the beginning). "
        "Citation should be done at the end of each sentence. "
        "Give conclusion in the end. "
        "Write summary in the same language as the search results. "
        "Search results:\n\n"
    )
    for c in topn_chunks:
        prompt += c + "\n\n"
    summary = generate_text(openai_key, prompt, gpt_model)
    return summary


def load_openai_key() -> str:
    key = os.environ.get("OPENAI_API_KEY")
    if key is None:
        raise ValueError(
            "[ERROR]: Please pass your OPENAI_API_KEY. Get your key here : https://platform.openai.com/account/api-keys"
        )
    return key


# %%
@serving
def ask_url(
    url: str,
    question: str,
    rebuild_embedding: bool,
    embedding_model: str,
    gpt_model: str,
) -> str:
    if rebuild_embedding:
        load_url(url, embedding_model, rebuild_embedding)
    openai_key = load_openai_key()
    return generate_answer(question, gpt_model, openai_key)


@serving
async def ask_file(
    file: UploadFile,
    question: str,
    rebuild_embedding: bool,
    embedding_model: str,
    gpt_model: str,
) -> str:
    if rebuild_embedding:
        load_file(file, embedding_model, rebuild_embedding)
    openai_key = load_openai_key()
    return generate_answer(question, gpt_model, openai_key)


@serving
def load_url(url: str,
             embedding_model: str,
             rebuild_embedding: bool,
             gpt_model: str
             ) -> str:
    download_pdf(url, "corpus.pdf")
    notification = load_recommender("corpus.pdf", embedding_model, rebuild_embedding)
    openai_key = load_openai_key()
    summary = generate_summary(gpt_model, openai_key)
    response = f"{notification}###{summary}"
    return response


@serving
async def load_file(
    file: UploadFile,
        embedding_model: str,
        rebuild_embedding: bool,
        gpt_model: str
) -> str:
    suffix = Path(file.filename).suffix
    with NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        shutil.copyfileobj(file.file, tmp)
        tmp_path = Path(tmp.name)
    notification = load_recommender(str(tmp_path), embedding_model, rebuild_embedding)
    openai_key = load_openai_key()
    summary = generate_summary(gpt_model, openai_key)
    response = f"{notification}###{summary}"
    return response