from langchain.docstore.document import Document from langchain.chains.question_answering import load_qa_chain import chromadb from datetime import datetime import os from datetime import datetime import pdfkit from langchain.chains.question_answering import load_qa_chain bookQuestion = dict() contemplandoQuestion = { 'DEVOCIONALMENTE':'¿Cómo estimula Dios su corazón a través de Su Palabra?', 'EXÉGESIS':'Cuál es el contexto de este pasaje?', 'CRISTO':'¿Cómo se comprende este texto a la luz de Cristo?', 'ARCO REDENTOR':'¿Cómo encaja este texto en la metanarrativa de las Escrituras?', 'EVANGELION': '¿Cómo se declara el evangelio en este texto?', 'EVANGELION_TWO': '¿Cómo interpretamos este texto a la luz del evangelio?', } proclamandoQuestion = { 'PÚBLICO':'¿Cuáles son los ídolos en los corazones de las personas que rechazarían el evangelio de Cristo?', 'HISTORIA':'¿Cómo el guión de su predicación comunica la historia de Dios?', 'EXPECTATIVAS': '¿Qué espera Dios que hagan como respuesta a esta predicación?', 'EXPECTATIVAS_TWO': '¿Cuáles son sus expectativas divinas como predicador de este mensaje?', } bookQuestion['Contemplando'] = contemplandoQuestion bookQuestion['Proclamando'] = proclamandoQuestion HISTORY_ANSWER = "" DIRECTORY_PATH_TO_DOWNLOAD = 'data/sermon_lab_ai/download_files' if not os.path.exists(DIRECTORY_PATH_TO_DOWNLOAD): os.makedirs(f"{DIRECTORY_PATH_TO_DOWNLOAD}") def getCurrentFileName(): now = datetime.now() strNow = now.strftime("%m%d%Y_%H%M%S") return f"sermonDay_{strNow}.pdf" fileAddresToDownload = f"{DIRECTORY_PATH_TO_DOWNLOAD}{os.sep}{getCurrentFileName()}" FILE_PATH_NAME = fileAddresToDownload def updatePromptTemplate(promptTemplate, inputVariablesTemplate): prompt = PromptTemplate(template = promptTemplate, input_variables = inputVariablesTemplate) chain = load_qa_chain(llm, chain_type="stuff", prompt = prompt) return chain def predict(query): chain = updatePromptTemplate( sermonPromptMenuGemini['BUILD_PREPARE_QUESTIONS'], ['question','SERMON_CONTEXT','context'] ) if query != '': global retriever answer = askQuestion( query, chain, retriever, topic = query, KEY = 'question' ) answer = (answer.split("<|assistant|>")[-1]).strip() HISTORY_ANSWER = answer return answer else: return query def predictContemplando(queryKey): #Call to LLM LangChaing inference query = contemplandoQuestion[queryKey] return predict(query) def predictProclamando(queryKey): #Call to LLM LangChaing inference query = proclamandoQuestion[queryKey] return predict(query) #### # #### def predictFromInit(sermonTopic): global HISTORY_ANSWER keyStr = 'SERMON_TOPIC' if HISTORY_ANSWER == '': chain = updatePromptTemplate( sermonPromptMenuGemini['BUILD_INIT'], [keyStr,'CANT_VERSICULOS','context'] ) else: chain = updatePromptTemplate( sermonPromptMenuGemini['BUILD_EMPTY'], ['BIBLE_VERSICLE','context'] ) keyStr = 'BIBLE_VERSICLE' global retriever answer = askQuestionInit( '', chain, retriever, topic = sermonTopic, KEY = keyStr ) #Create a new document and build a retriver if answer != '': doc = Document(page_content="text", metadata = {"source": "local"}) vectorstore = Chroma.from_documents( documents = [doc], embedding = embed_model, persist_directory="chroma_db_dir_sermon", # Local mode with in-memory storage only collection_name="sermon_lab_ai" ) retriever = vectorstore.as_retriever( search_kwargs = {"k": 3} ) HISTORY_ANSWER = answer return answer #### # #### def predictQuestionBuild(sermonTopic): chain = updatePromptTemplate( sermonPromptMenuGemini['BUILD_QUESTION'], ['SERMON_IDEA', 'context'] ) global retriever answer = askQuestionEx( '', chain, retriever, topic = sermonTopic, KEY = 'SERMON_IDEA' ) return answer #### # #### def predictDevotionBuild(sermonTopic): chain = updatePromptTemplate( sermonPromptMenuGemini['BUILD_REFLECTIONS'], ['SERMON_IDEA', 'context'] ) global retriever global HISTORY_ANSWER answer = askQuestionEx( HISTORY_ANSWER, chain, retriever, topic = sermonTopic, KEY = 'SERMON_IDEA' ) return answer # A utility function for answer generation def askQuestion( question, _chain, _retriever, topic = 'el amor de Dios', KEY = 'SERMON_TOPIC' ): #Obtener los Chunks relevantes a la pregunta en el RAG #print(f" Question: {question}") context = _retriever.get_relevant_documents(question) #print("---- Contexto ----") #print(context) #print("____________________GLOBAL________") global HISTORY_ANSWER #print (HISTORY_ANSWER) return ( _chain({ KEY: topic, 'SERMON_CONTEXT': HISTORY_ANSWER, "input_documents": context, "question": question }, return_only_outputs=True) )['output_text'] A utility function for answer generation def askQuestionEx( question, _chain, _retriever, topic = 'el amor de Dios', KEY = 'SERMON_TOPIC' ): #Obtener los Chunks relevantes a la pregunta en el RAG #print(f" Question: {question}") context = _retriever.get_relevant_documents(question) #print("---- Contexto ----") #print(context) #print("____________________GLOBAL________") global HISTORY_ANSWER #print (HISTORY_ANSWER) return ( _chain({ KEY: topic, "input_documents": context, "question": question }, return_only_outputs=True) )['output_text'] # A utility function for answer generation def askQuestionInit( question, _chain, _retriever, topic = 'el amor de Dios', KEY = 'SERMON_TOPIC' ): #Obtener los Chunks relevantes a la pregunta en el RAG context = _retriever.get_relevant_documents(question) settings = { KEY: topic, "input_documents": context, "question": question } if KEY == 'SERMON_TOPIC': settings['CANT_VERSICULOS'] = 5 return ( _chain( settings, return_only_outputs=True) )['output_text'] def downloadSermonFile(answer): if os.path.exists(FILE_PATH_NAME): os.remove(FILE_PATH_NAME) pdfkit.from_string( answer, FILE_PATH_NAME ) return ""