import os from threading import Lock from typing import Any, Dict, Optional, Tuple import gradio as gr from langchain.chains import ConversationalRetrievalChain from langchain.chat_models import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.prompts.chat import (ChatPromptTemplate, HumanMessagePromptTemplate, SystemMessagePromptTemplate) from .core.chunking import chunk_file from .core.embedding import embed_files from .core.parsing import read_file VECTOR_STORE = "faiss" MODEL = "openai" EMBEDDING = "openai" MODEL = "gpt-3.5-turbo-16k" K = 5 USE_VERBOSE = True API_KEY = os.environ["OPENAI_API_KEY"] system_template = """ Use the context below to answer questions. You must only use the Context to answer questions. If I ask you about 'the book' or 'this book' or similar references, then answer using the Context. If you cannot find the answer from the Context below, you must respond with "I'm sorry, but I can't find the answer to your question in the book, 'Design by Fire,' by Emily Elizabeth Schlickman and Brett Milligan." All answers must be in English unless you are explicitly asked to translate to a different language. ---------------- {context} {chat_history} """ # Create the chat prompt templates messages = [ SystemMessagePromptTemplate.from_template(system_template), HumanMessagePromptTemplate.from_template("{question}") ] qa_prompt = ChatPromptTemplate.from_messages(messages) class AnswerConversationBufferMemory(ConversationBufferMemory): def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None: return super(AnswerConversationBufferMemory, self).save_context(inputs,{'response': outputs['answer']}) def getretriever(): with open("./resources/design-by-fire.pdf", 'rb') as uploaded_file: try: file = read_file(uploaded_file) except Exception as e: print(e) chunked_file = chunk_file(file, chunk_size=512, chunk_overlap=0) folder_index = embed_files( files=[chunked_file], embedding=EMBEDDING, vector_store=VECTOR_STORE, openai_api_key=API_KEY, ) return folder_index.index.as_retriever(verbose=True, search_type="similarity", search_kwargs={"k": K}) retriever = getretriever() def getanswer(chain, question, history): if hasattr(chain, "value"): chain = chain.value if hasattr(history, "value"): history = history.value if hasattr(question, "value"): question = question.value history = history or [] lock = Lock() lock.acquire() try: output = chain({"question": question}) output = output["answer"] history.append((question, output)) except Exception as e: raise e finally: lock.release() return history, history, gr.update(value="") def load_chain(inputs = None): llm = ChatOpenAI( openai_api_key=API_KEY, model_name=MODEL, verbose=True) chain = ConversationalRetrievalChain.from_llm( llm, retriever=retriever, return_source_documents=USE_VERBOSE, memory=AnswerConversationBufferMemory(memory_key="chat_history", return_messages=True), verbose=USE_VERBOSE, combine_docs_chain_kwargs={"prompt": qa_prompt}) return chain CSS =""" .contain { display: flex; flex-direction: column; } .gradio-container { height: 100vh !important; } #component-0 { height: 100%; } #chatbot { flex-grow: 1; overflow: auto;} """ with gr.Blocks() as block: with gr.Row(): with gr.Column(scale=0.75): with gr.Row(): gr.Markdown("