import time import gradio as gr import logging from langchain.document_loaders import PDFMinerLoader,CSVLoader ,UnstructuredWordDocumentLoader,TextLoader,OnlinePDFLoader from langchain.text_splitter import CharacterTextSplitter from langchain.embeddings import SentenceTransformerEmbeddings from langchain.vectorstores import FAISS from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain.docstore.document import Document from youtube_transcript_api import YouTubeTranscriptApi import chatops logger = logging.getLogger(__name__) DEVICE = 'cpu' MAX_NEW_TOKENS = 4096 DEFAULT_TEMPERATURE = 0.1 DEFAULT_MAX_NEW_TOKENS = 2048 MAX_INPUT_TOKEN_LENGTH = 4000 DEFAULT_CHAR_LENGTH = 1000 EXAMPLES = ["https://www.youtube.com/watch?v=aircAruvnKk&ab_channel=3Blue1Brown", "https://www.youtube.com/watch?v=Ilg3gGewQ5U", "https://www.youtube.com/watch?v=WUvTyaaNkzM" ] def clear_chat(): return [] def get_text_from_youtube_link(video_link,max_video_length=800): video_text = "" video_id = video_link.split("watch?v=")[1].split("&")[0] srt = YouTubeTranscriptApi.get_transcript(video_id) for text_data in srt: video_text = video_text + " " + text_data.get("text") if len(video_text) > max_video_length: print(video_text) return video_text[0:max_video_length] else: print(video_text) return video_text def process_documents(documents,data_chunk=1500,chunk_overlap=100): text_splitter = CharacterTextSplitter(chunk_size=data_chunk, chunk_overlap=chunk_overlap,separator='\n') texts = text_splitter.split_documents(documents) return texts def process_youtube_link(link, document_name="youtube-content",char_length=1000): try: metadata = {"source": f"{document_name}.txt"} return [Document(page_content=get_text_from_youtube_link(video_link=link,max_video_length=char_length), metadata=metadata)] except Exception as err: logger.error(f'Error in reading document. {err}') def create_prompt(): prompt_template = """As a chatbot asnwer the questions regarding the content in the video. Use the following context to answer. If you don't know the answer, just say I don't know. {context} Question: {question} Answer :""" prompt = PromptTemplate( template=prompt_template, input_variables=["context", "question"] ) return prompt def youtube_chat(youtube_link,API_key,llm='HuggingFace',temperature=0.1,max_tokens=1096,char_length=1500): document = process_youtube_link(link=youtube_link,char_length=char_length) print("docuemt:",document) embedding_model = SentenceTransformerEmbeddings(model_name='thenlper/gte-base',model_kwargs={"device": DEVICE}) texts = process_documents(documents=document) global vector_db vector_db = FAISS.from_documents(documents=texts, embedding= embedding_model) global qa if llm == 'HuggingFace': chat = chatops.get_hugging_face_model( model_id="tiiuae/falcon-7b-instruct", API_key=API_key, temperature=temperature, max_tokens=max_tokens ) else: chat = chatops.get_openai_chat_model(API_key=API_key) chain_type_kwargs = {"prompt": create_prompt()} qa = RetrievalQA.from_chain_type(llm=chat, chain_type='stuff', retriever=vector_db.as_retriever(), chain_type_kwargs=chain_type_kwargs, return_source_documents=True ) return "Youtube link Processing completed ..." def infer(question, history): # res = [] # # for human, ai in history[:-1]: # # pair = (human, ai) # # res.append(pair) # chat_history = res result = qa({"query": question}) matching_docs_score = vector_db.similarity_search_with_score(question) return result["result"] def bot(history): response = infer(history[-1][0], history) history[-1][1] = "" for character in response: history[-1][1] += character time.sleep(0.05) yield history def add_text(history, text): history = history + [(text, None)] return history, "" css=""" #col-container {max-width: 2048px; margin-left: auto; margin-right: auto;} """ title = """

Chat with Youtube Videos

Upload a youtube link of any video-lecture/song/Research and you are able to ask QA as chatbot with the tool It gives you option to use HuggingFace/OpenAI as LLM's, make sure to add your key.

""" with gr.Blocks(css=css) as demo: with gr.Row(): with gr.Column(elem_id="col-container"): gr.HTML(title) with gr.Column(): with gr.Row(): LLM_option = gr.Dropdown(['HuggingFace','OpenAI'],label='Select HuggingFace/OpenAI') API_key = gr.Textbox(label="Add API key", type="password",autofocus=True) with gr.Group(): chatbot = gr.Chatbot(height=270) examples_set = gr.Radio(label="Examples of some You tube Videos",choices=EXAMPLES) with gr.Row(): question = gr.Textbox(label="Type your question !",lines=1).style(full_width=True) with gr.Row(): submit_btn = gr.Button(value="Send message", variant="primary", scale = 1) clean_chat_btn = gr.Button("Delete Chat") with gr.Column(): with gr.Box(): youtube_link = gr.Textbox(label="Add your you tube Link",text_align='left',autofocus=True) with gr.Row(): load_youtube_bt = gr.Button("Process Youtube Link",).style(full_width = False) langchain_status = gr.Textbox(label="Status", placeholder="", interactive = False) with gr.Column(): with gr.Accordion(label='Advanced options', open=False): max_new_tokens = gr.Slider( label='Max new tokens', minimum=2048, maximum=MAX_NEW_TOKENS, step=1, value=DEFAULT_MAX_NEW_TOKENS, ) temperature = gr.Slider(label='Temperature',minimum=0.1,maximum=4.0,step=0.1,value=DEFAULT_TEMPERATURE,) char_length = gr.Slider(label='Max Character', minimum= DEFAULT_CHAR_LENGTH, maximum = 5*DEFAULT_CHAR_LENGTH, step = 500,value= 1500 ) load_youtube_bt.click(youtube_chat,inputs= [youtube_link,API_key,LLM_option,temperature,max_new_tokens,char_length],outputs=[langchain_status], queue=False) # examples_set.change(fn=youtube_chat, inputs=[examples_set,API_key,LLM_option,temperature,max_new_tokens,char_length], outputs=[langchain_status]) clean_chat_btn.click(clear_chat, [], chatbot) question.submit(add_text, inputs=[chatbot, question], outputs=[chatbot, question]).then(bot, chatbot, chatbot) submit_btn.click(add_text, inputs=[chatbot, question], outputs=[chatbot, question]).then(bot, chatbot, chatbot) demo.launch()