app2.py

from llama_cpp import Llama
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS, Chroma
from faster_whisper import WhisperModel
import os
import gradio as gr
import torch
import base64
import json
import chromadb
import requests
import gc, torch

GPU = False if torch.cuda.device_count()==0 else True
n_threads = os.cpu_count()//2
global llm

def load_llm(model_name):
    try:
        del llm     
    except:
        pass
    torch.cuda.empty_cache()
    gc.collect()
    llm = Llama(model_path=model_name,
            n_threads=11, n_gpu_layers=80, n_ctx=3000)
    return llm

def load_faiss_db():
    new_db = FAISS.load_local("faiss_MH_c2000_o100", hf_embs)
    return new_db

def load_chroma_db():
    ABS_PATH = os.getcwd()#os.path.dirname(os.path.abspath(__file__))
    DB_DIR = os.path.join(ABS_PATH, "chroma_MH_c1000_o0")
    print("DB_DIR", DB_DIR)
    client_settings = chromadb.config.Settings(
        chroma_db_impl="duckdb+parquet",
        persist_directory=DB_DIR,
        anonymized_telemetry=False
    )
    vectorstore = Chroma(
        collection_name="langchain_store",
        embedding_function=hf_embs,
        client_settings=client_settings,
        persist_directory=DB_DIR,
    )
    return vectorstore

def init_prompt_tempalate(context, question):
    prompt_template = f"""<s>[INST] 
    As a health insurance assistant, use the following pieces of context to answer the question at the end. If you don't know the answer, just say that you don't know, don't try to make up an answer.
    {context}
    Question: {question}
    Concise answer in French:
    [/INST]"""
    prompt_template = f"""As a health insurance assistant, use the following pieces of context to answer the question at the end. If you don't know the answer, just say that you don't know, don't try to make up an answer.
    {context}
    Question: {question}
    Concise answer in French:"""
    prompt_template = f"""Answer the question based only on the following context:
    {context}

    Question: {question}

    Answer in the following language: French
    """

    prompt_template = f"""<|system|>
    Answer the question based only on the following context:
    {context}</s>
    <|user|>
    {question}</s>
    <|assistant|>
    """
    return prompt_template

def wav_to_base64(file_path):
    base64_data = base64.b64encode(open(file_path, "rb").read()).decode("utf-8")
    return base64_data

def search_llm(question, max_tokens=10, temp=0, k_chunks=1, top_k=40, 
                top_p=0.95):
    results = {}
    context = ""
    
    new_db = new_db_faiss
    # if db_type=="faiss":
    #     new_db = new_db_faiss
    # else:
    #     new_db = new_db_chroma
    docs = new_db.similarity_search_with_score(question,
                                               k=int(k_chunks))
    contexts = [el[0].page_content for el in docs]
    scores = [el[1] for el in docs]
    context = "\n".join(contexts)
    score = sum(scores) / len(scores)
    score = round(score, 3)
    url = docs[0][0].metadata

    prompt_template = init_prompt_tempalate(context, question)

    output = llm(prompt_template,
                 max_tokens=int(max_tokens),
                 stop=["Question:", "\n"],
                 echo=True,
                 temperature=temp,
                 top_k=int(top_k),
                 top_p=top_p)
    # first_reponse = output["choices"][0]["text"].split("answer in French:")[-1].strip()
    first_reponse = output["choices"][0]["text"].split("<|assistant|>")[-1].strip()
    results["Response"] = first_reponse
    # results["prompt_template"] = prompt_template
    results["context"] = context
    results["source"] = url
    results["context_score"] = score
    return results["Response"], results["source"], results["context"], results["context_score"]

def stt(path):
    injson = {}
    injson["data"] = wav_to_base64(path)
    results = requests.post(url="http://0.0.0.0:5566/api",
                            json=injson,
                            verify=False)
    transcription = results.json()["transcription"]
    query = transcription
    query = transcription if "?" in transcription else transcription + "?"
    return query

def STT_LLM(path, max_tokens, temp, k_chunks, top_k, top_p, db_type):
    """
    """
    query = stt(path)
    Response, url, context, contextScore = search_llm(query, max_tokens, temp, k_chunks, top_k, top_p)
    return query, Response, url["source"], context, str(contextScore)

def LLM(content, max_tokens, temp, k_chunks, top_k, top_p, db_type):
    Response, url, context, contextScore = search_llm(content, max_tokens, temp, k_chunks, top_k,
                                                    top_p)
    url = url["source"]
    return Response, url, context, str(contextScore)


embs_name = "sentence-transformers/all-mpnet-base-v2"
hf_embs = HuggingFaceEmbeddings(model_name=embs_name,
                                    model_kwargs={"device": "cuda"})
new_db_chroma = load_faiss_db()
new_db_faiss = load_chroma_db()
### Load models
#stt
wspr = WhisperModel("small", device="cuda" if GPU else "cpu", compute_type="int8")
#llm
model_name = "mistral-7b-instruct-v0.1.Q4_K_M.gguf"
model_name = "zephyr-7b-beta.Q4_K_M.gguf"

llm = load_llm(model_name)

demo = gr.Blocks()
with demo:
    with gr.Tab(model_name):
        with gr.Row():
            with gr.Column():
                with gr.Box():
                    content = gr.Text(label="Posez votre question")
                    audio_path = gr.Audio(source="microphone",
                            format="mp3",
                            type="filepath",
                            label="Posez votre question (Whisper-small)")
                with gr.Row():
                    max_tokens = gr.Number(label="Max_tokens", value=100, maximum=1000, minimum=1)
                    temp = gr.Number(label="Temperature", value=0.1, maximum=1.0, minimum=0.0, step=0.1)
                    k_chunks = gr.Number(label="k_chunks", value=2, maximum=5, minimum=1)
                    top_k = gr.Number(label="top_k", value=100, maximum=1000, minimum=1)
                    top_p = gr.Number(label="top_p", value=0.95, maximum=1.0, minimum=0.0)
                # with gr.Box():
                #     db_type = gr.Dropdown(choices=["faiss", "chromadb"], label="Vector DB", value="faiss")
                #     # llm_name = gr.Dropdown(choices=["vicuna-7b-v1.3.ggmlv3.q4_1.bin",
                #     #                         "vicuna-7b-v1.3.ggmlv3.q5_1.bin"],
                #     #                         label="llm", value="vicuna-7b-v1.3.ggmlv3.q4_1.bin")
                #     b3 = gr.Button("update model")
                #     # b3.click(load_llm, inputs=llm_name, outputs=None)
            with gr.Column():
                # transcription = gr.Text(label="transcription")
                Response = gr.Text(label="Réponse")
                url = gr.Text(label="url source")
                context = gr.Text(label="contexte (chunks)")
                contextScore = gr.Text(label="contexte score (L2 distance)")
        with gr.Box():
            b2 = gr.Button("reconnaissace vocale")
            b1 = gr.Button("search llm")
        b1.click(LLM, inputs=[content, max_tokens, temp, k_chunks, top_k, top_p], #db_type
            outputs=[Response, url, context, contextScore])
        b2.click(stt, inputs=audio_path, outputs=content)
    
    # with gr.Tab("gptq"):
    #     with gr.Row():
    #         with gr.Column():
    #             with gr.Box():
    #                 content = gr.Text(label="Posez votre question")
    #                 audio_path = gr.Audio(source="microphone",
    #                         format="mp3",
    #                         type="filepath",
    #                         label="Posez votre question (Whisper-small)")
    #             with gr.Row():
    #                 max_tokens = gr.Number(label="Max_tokens", value=100, maximum=1000, minimum=1)
    #                 temp = gr.Number(label="Temperature", value=0.1, maximum=1.0, minimum=0.0)
    #                 k_chunks = gr.Number(label="k_chunks", value=2, maximum=3, minimum=1)
    #                 top_k = gr.Number(label="top_k", value=100, maximum=1000, minimum=1)
    #                 top_p = gr.Number(label="top_p", value=0.95, maximum=1.0, minimum=0.0)
    #             with gr.Box():
    #                 db_type = gr.Dropdown(choices=["faiss", "chromadb"], label="Vector DB", value="faiss")
    #                 llm_name = gr.Dropdown(choices=["llama-2-7b.ggmlv3.q4_1.bin",
    #                                         "vicuna-7b-v1.3.ggmlv3.q4_1.bin"],
    #                                         label="llm", value="llama-2-7b.ggmlv3.q4_1.bin")
    #                 b3 = gr.Button("update model")
    #                 # b3.click(stt, inputs=llm_name, outputs=None)
    #         with gr.Column():
    #             # transcription = gr.Text(label="transcription")
    #             Response = gr.Text(label="Réponse")
    #             url = gr.Text(label="url source")
    #             context = gr.Text(label="contexte (chunks)")
    #             contextScore = gr.Text(label="contexte score (L2 distance)")
    #     with gr.Box():
    #         b2 = gr.Button("reconnaissace vocale")
    #         b1 = gr.Button("search llm")
    #     b1.click(LLM, inputs=[content, max_tokens, temp, k_chunks, top_k, top_p, db_type], 
    #         outputs=[Response, url, context, contextScore])
    #     b2.click(stt, inputs=audio_path, outputs=content)

if __name__ == "__main__":
    demo.launch(share=True, enable_queue=True, show_api=True)