import os import gradio as gr from langchain-community.vectorstores import Chroma from transformers import pipeline from sentence_transformers import SentenceTransformer import langchain.chains.LLMChain import langchain_core.prompts.PromptTemplate from langchain_huggingface import HuggingFacePipeline #Konstanten ANTI_BOT_PW = os.getenv("CORRECT_VALIDATE") PATH_WORK = "." CHROMA_DIR = "/chroma/kkg" CHROMA_PDF = './chroma/kkg/pdf' CHROMA_WORD = './chroma/kkg/word' CHROMA_EXCEL = './chroma/kkg/excel' #HuggingFace Model name-------------------------------- MODEL_NAME_HF = "mistralai/Mixtral-8x7B-Instruct-v0.1" # Hugging Face Token direkt im Code setzen hf_token = os.getenv("HF_READ") os.environ["HUGGINGFACEHUB_API_TOKEN"] = os.getenv("HF_READ") ############################################### #globale Variablen ############################################## #nur bei ersten Anfrage splitten der Dokumente - um die Vektordatenbank entsprechend zu füllen #splittet = False #DB für Vektorstore vektordatenbank = None retriever = None ############################################# # Allgemeine Konstanten #Filepath zu temp Folder (temp) mit File von ausgewähltem chatverlauf file_path_download = "" #################################################### #aus einem Text-Prompt die Antwort von KI bekommen #mit oder ohne RAG möglich def generate_text (prompt, chatbot, history, vektordatenbank, websuche, top_p=0.6, temperature=0.2, max_new_tokens=4048, max_context_length_tokens=2048, repetition_penalty=1.3, top_k=35): print("Text pur..............................") if (prompt == ""): raise gr.Error("Prompt ist erforderlich.") try: #oder an Hugging Face -------------------------- print("HF Anfrage.......................") model_kwargs={"temperature": 0.5, "max_length": 512, "num_return_sequences": 1, "top_k": top_k, "top_p": top_p, "repetition_penalty": repetition_penalty} llm = HuggingFaceHub(repo_id=repo_id, model_kwargs=model_kwargs) llm = HuggingFaceChain(model=MODEL_NAME_HF, model_kwargs={"temperature": 0.5, "max_length": 128}) #Prompt an history anhängen und einen Text daraus machen history_text_und_prompt = generate_prompt_with_history(prompt, history) #zusätzliche Dokumenten Splits aus DB zum Prompt hinzufügen (aus VektorDB - Chroma oder Mongo DB) print("LLM aufrufen mit RAG: ...........") result = rag_chain(llm, history_text_und_prompt, retriever) #für hugchat noch kein rag möglich... #weitere Möglichkeit für Rag-Chain - dann auch für HF Modelle möglich, da kein llm in Langchain übergeben werden muss... #result = rag_chain2(history_text_und_prompt, db, 5) print("result regchain.....................") print(result) except Exception as e: raise gr.Error(e) return result, suche_im_Netz ######################################## # Bot- test gegen schädliche Bots die die Anwendung testen... # Funktion zur Überprüfung der Benutzereingabe # Funktion zur Überprüfung der Eingabe und Aktivierung der Hauptanwendung def validate_input(user_input_validate, validate=False): user_input_hashed = hash_input(user_input_validate) if user_input_hashed == hash_input(ANTI_BOT_PW): return "Richtig! Weiter gehts... ", True, gr.Textbox(visible=False), gr.Button(visible=False) else: return "Falsche Antwort!!!!!!!!!", False, gr.Textbox(label = "", placeholder="Bitte tippen Sie das oben im Moodle Kurs angegebene Wort ein, um zu beweisen, dass Sie kein Bot sind.", visible=True, scale= 5), gr.Button("Validieren", visible = True) def custom_css(): return """ body, html { background-color: #303030; /* Dunkler Hintergrund */ color:#353535; } """ #nicht in Gebrauch................................... def get_rag_response(question): # Abfrage der relevanten Dokumente aus Chroma DB docs = chroma_db.search(question, top_k=5) passages = [doc['text'] for doc in docs] links = [doc.get('url', 'No URL available') for doc in docs] # Generieren der Antwort context = " ".join(passages) qa_input = {"question": question, "context": context} answer = qa_pipeline(qa_input)['answer'] # Zusammenstellen der Ausgabe response = { "answer": answer, "documents": [{"link": link, "passage": passage} for link, passage in zip(links, passages)] } return response #Eingaben der GUI verarbeiten def generate_auswahl(prompt_in, file, file_history, chatbot, history, top_p=0.6, temperature=0.5, max_new_tokens=4048, max_context_length_tokens=2048, repetition_penalty=1.3,top_k=5, validate=False): global vektordatenbank, retriever #nur wenn man sich validiert hat, kann die Anwendung los legen if (validate and not prompt_in == "" and not prompt_in == None): # Vektorstore initialisieren #falls schon ein File hochgeladen wurde, ist es in history_file gespeichert - falls ein neues File hochgeladen wurde, wird es anschließend neu gesetzt neu_file = file_history #prompt normalisieren bevor er an die KIs geht prompt = normalise_prompt(prompt_in) #muss nur einmal ausgeführt werden... if vektordatenbank == None: print("db neu aufbauen!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!1") splits = document_loading_splitting() vektordatenbank, retriever = document_storage_chroma(splits) #kein Bild hochgeladen -> auf Text antworten... status = "Antwort der KI ..." if (file == None and file_history == None): result, status = generate_text(prompt, chatbot, history,vektordatenbank, top_p=0.6, temperature=0.5, max_new_tokens=4048, max_context_length_tokens=2048, repetition_penalty=1.3, top_k=3) history = history + [[prompt, result]] else: #Es wurde ein File neu angehängt -> das hochladen und dann Prompt bearbeiten #das history_fiel muss neu gesetzt werden if (file != None): # file_history wird neu gesetzt in der Rückgabe dieser Funktion... neu_file = file #File hochladen in Chroma und dann Antwort generieren result = generate_text_zu_doc(neu_file, prompt, k, rag_option, chatbot, history, vektordatenbank) #die history erweitern - abhängig davon, ob gerade ein file hochgeladen wurde oder nicht if (file != None): history = history + [[(file,), None],[prompt, result]] else: history = history + [[prompt, result]] chatbot[-1][1] = "" for character in result: chatbot[-1][1] += character time.sleep(0.03) yield chatbot, history, None, neu_file, status if shared_state.interrupted: shared_state.recover() try: yield chatbot, history, None, neu_file, "Stop: Success" except: pass else: #noch nicht validiert, oder kein Prompt return chatbot, history, None, file_history, "Erst validieren oder einen Prompt eingeben!" ############################################################################################# # Start Gui Vorabfrage # Validierungs-Interface - Bots weghalten... print ("Start GUI Vorabfrage") ################################################################################################# print ("Start GUI Hauptanwendung") with open("custom.css", "r", encoding="utf-8") as f: customCSS = f.read() #Add Inputs für Tab 2 additional_inputs = [ gr.Slider(label="Temperature", value=0.65, minimum=0.0, maximum=1.0, step=0.05, interactive=True, info="Höhere Werte erzeugen diversere Antworten", visible=True), gr.Slider(label="Max new tokens", value=1024, minimum=0, maximum=4096, step=64, interactive=True, info="Maximale Anzahl neuer Tokens", visible=True), gr.Slider(label="Top-p (nucleus sampling)", value=0.6, minimum=0.0, maximum=1, step=0.05, interactive=True, info="Höhere Werte verwenden auch Tokens mit niedrigerer Wahrscheinlichkeit.", visible=True), gr.Slider(label="Repetition penalty", value=1.2, minimum=1.0, maximum=2.0, step=0.05, interactive=True, info="Strafe für wiederholte Tokens", visible=True) ] with gr.Blocks(css=customCSS, theme=themeAlex) as demo: #validiert speichern validate = gr.State(False) #Session Variablen, um Weete zu speichern, auch wenn die Felder in der GUI bereits wieder leer sind # history parallel zu chatbot speichern - da in chatbot bei Bildern zum Anzeigen in der GUI die Bilder speziell formatiert werden, # für die Übergabe an die ki aber der Pfad zum Bild behalten werden muss - was in der history der Fall ist! history = gr.State([]) uploaded_file_paths= gr.State([]) history3 = gr.State([]) uploaded_file_paths3= gr.State([]) #alle chats einer Session sammeln chats = gr.State({}) #damit der Prompt auch nach dem upload in die History noch für predicts_args verfügbar ist user_question = gr.State("") #für die anderen Tabs auch... #damit der Prompt auch nach dem upload in die History noch für predicts_args verfügbar ist user_question2 = gr.State("") user_question3 = gr.State("") attached_file = gr.State(None) attached_file_history = gr.State(None) attached_file3 = gr.State(None) attached_file_history3 = gr.State(None) status_display = gr.State("") status_display2 = gr.State("") status_display3 = gr.State("") ################################################ # Tab zum Chatbot mit Text oder Bildeingabe ################################################ gr.Markdown(description_top) with gr.Row(): user_input_validate =gr.Textbox(label= "Bitte das oben im Moodle Kurs angegebene Wort eingeben, um die Anwendung zu starten", visible=True, interactive=True, scale= 7) validate_btn = gr.Button("Validieren", visible = True) #validation_result = gr.Text(label="Validierungsergebnis") with gr.Tab("KKG Chatbot"): with gr.Row(): #gr.HTML("LI Chatot") status_display = gr.Markdown("Antwort der KI ...", visible = True) #, elem_id="status_display") with gr.Row(): with gr.Column(scale=5): with gr.Row(): chatbot = gr.Chatbot(elem_id="li-chat",show_copy_button=True) with gr.Row(): with gr.Column(scale=12): user_input = gr.Textbox( show_label=False, placeholder="Gib hier deinen Prompt ein...", container=False ) with gr.Column(min_width=70, scale=1): submitBtn = gr.Button("Senden") with gr.Column(min_width=70, scale=1): cancelBtn = gr.Button("Stop") with gr.Row(): image_display = gr.Image( visible=False) upload = gr.UploadButton("📁", file_types=["image", "pdf", "docx", "pptx", "xlsx"], scale = 10) emptyBtn = gr.ClearButton([user_input, chatbot, history, attached_file, attached_file_history, image_display], value="🧹 Neue Session", scale=10) with gr.Column(): with gr.Column(min_width=50, scale=1): with gr.Tab(label="Chats ..."): #Geht nicht, da für alle gleichzeitig sichtbar #chat_selector = gr.CheckboxGroup(label="", choices=update_chat_options()) #download_button = gr.Button("Download ausgewählte Chats") file_download = gr.File(label="Noch keine Chatsverläufe", visible=True, interactive = False, file_count="multiple",) with gr.Tab(label="Parameter"): #gr.Markdown("# Parameters") rag_option = gr.Radio(["Aus", "An"], label="KKG Erweiterungen (RAG)", value = "Aus") model_option = gr.Radio(["OpenAI", "HuggingFace"], label="Modellauswahl", value = "OpenAI") websuche = gr.Radio(["Aus", "An"], label="Web-Suche", value = "Aus") top_p = gr.Slider( minimum=-0, maximum=1.0, value=0.95, step=0.05, interactive=True, label="Top-p", visible=False, ) top_k = gr.Slider( minimum=1, maximum=100, value=35, step=1, interactive=True, label="Top-k", visible=False, ) temperature = gr.Slider( minimum=0.1, maximum=2.0, value=0.2, step=0.1, interactive=True, label="Temperature", visible=False ) max_length_tokens = gr.Slider( minimum=0, maximum=512, value=512, step=8, interactive=True, label="Max Generation Tokens", visible=False, ) max_context_length_tokens = gr.Slider( minimum=0, maximum=4096, value=2048, step=128, interactive=True, label="Max History Tokens", visible=False, ) repetition_penalty=gr.Slider(label="Repetition penalty", value=1.2, minimum=1.0, maximum=2.0, step=0.05, interactive=True, info="Strafe für wiederholte Tokens", visible=False) anzahl_docs = gr.Slider(label="Anzahl Dokumente", value=3, minimum=1, maximum=10, step=1, interactive=True, info="wie viele Dokumententeile aus dem Vektorstore an den prompt gehängt werden", visible=False) openai_key = gr.Textbox(label = "OpenAI API Key", value = "sk-", lines = 1, visible = False) gr.Markdown(description) ###################################### # Events und Übergabe Werte an Funktionen ####################################### ###################################### # Für Tab 1: Chatbot #Argumente für generate Funktion als Input predict_args = dict( fn=generate_auswahl, inputs=[ user_question, attached_file, attached_file_history, chatbot, history, anzahl_docs, top_p, temperature, max_length_tokens, max_context_length_tokens, repetition_penalty, top_k, websuche, validate ], outputs=[chatbot, history, attached_file, attached_file_history, status_display], show_progress=True, ) reset_args = dict( fn=reset_textbox, inputs=[], outputs=[user_input, status_display] ) # Chatbot transfer_input_args = dict( fn=add_text, inputs=[chatbot, history, user_input, attached_file, attached_file_history], outputs=[chatbot, history, user_question, attached_file, attached_file_history, image_display , user_input], show_progress=True ) ############################################## # Button Events.... #Validation Button # Event-Handler für die Validierung validate_btn.click(validate_input, inputs=[user_input_validate, validate], outputs=[status_display, validate, user_input_validate, validate_btn]) user_input_validate.submit(validate_input, inputs=[user_input_validate, validate], outputs=[status_display, validate, user_input_validate, validate_btn]) predict_event1 = user_input.submit(**transfer_input_args, queue=False,).then(**predict_args) predict_event2 = submitBtn.click(**transfer_input_args, queue=False,).then(**predict_args) predict_event3 = upload.upload(file_anzeigen, [upload], [image_display, image_display, attached_file] ) #.then(**predict_args) emptyBtn.click(clear_all, [history, uploaded_file_paths, chats], [attached_file, image_display, uploaded_file_paths, history, file_download, chats]) #Bild Anzeige neben dem Button wieder entfernen oder austauschen.. image_display.select(file_loeschen, [], [attached_file, image_display]) #download_button.click(fn=download_chats, inputs=chat_selector, outputs=[file_download]) #Berechnung oder Ausgabe anhalten (kann danach fortgesetzt werden) cancelBtn.click(cancel_outputing, [], [status_display], cancels=[predict_event1,predict_event2, predict_event3])