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"""[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} <|user|> {question} <|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)