|
import gradio as gr |
|
from langchain_community.document_loaders import YoutubeLoader |
|
|
|
|
|
from langchain_cohere import ChatCohere |
|
from langchain import hub |
|
from langchain_chroma import Chroma |
|
from langchain_core.output_parsers import StrOutputParser |
|
from langchain_core.runnables import RunnablePassthrough |
|
from langchain_text_splitters import RecursiveCharacterTextSplitter |
|
from langchain_cohere import CohereEmbeddings |
|
import os |
|
import os |
|
|
|
|
|
COHERE_API_KEY = os.environ.get("COHERE_API_KEY") |
|
|
|
|
|
llm = ChatCohere(model="command-r",cohere_api_key=COHERE_API_KEY) |
|
prompt = hub.pull("rlm/rag-prompt") |
|
|
|
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) |
|
|
|
def format_docs(docs): |
|
return "\n\n".join(doc.page_content for doc in docs) |
|
|
|
|
|
|
|
def get_youtube_details(video_url): |
|
print(video_url) |
|
loader = YoutubeLoader.from_youtube_url(str(video_url), add_video_info=False) |
|
docs = loader.load() |
|
print(docs) |
|
print("Video transcripts loaded in DB") |
|
return docs, loader |
|
|
|
|
|
def user_message(message, history): |
|
return "", history + [[message, None]] |
|
|
|
|
|
def clear_vectorstore(vectorstore): |
|
vectorstore.delete_all() |
|
return "Vector store cleared." |
|
|
|
|
|
def clear_textbox(): |
|
return "", None, None |
|
|
|
|
|
def bot_message(history, docs): |
|
if docs is None: |
|
return history |
|
|
|
print(docs,"instide bot_message") |
|
user_question = history[-1][0] |
|
splits = text_splitter.split_documents(docs) |
|
print(splits,"splits are also here") |
|
vectorstore = Chroma.from_documents(documents=splits, embedding=CohereEmbeddings(model="embed-english-light-v3.0", |
|
cohere_api_key=COHERE_API_KEY)) |
|
retriever = vectorstore.as_retriever() |
|
|
|
rag_chain = ( |
|
{"context": retriever | format_docs, "question": RunnablePassthrough()} |
|
| prompt |
|
| llm |
|
| StrOutputParser() |
|
) |
|
|
|
response = rag_chain.invoke(user_question) |
|
history[-1][1] = response |
|
return history |
|
|
|
title=( |
|
""" |
|
<center> |
|
|
|
<h1> VideoQ: Quick Answers, Skip Clickbait </h1> |
|
<b> text 📧<b> |
|
|
|
</center> |
|
""" |
|
) |
|
|
|
with gr.Blocks(theme=gr.themes.Monochrome()) as demo: |
|
|
|
with gr.Row(): |
|
gr.HTML(title,label=" ") |
|
|
|
gr.Markdown(""" |
|
### Skip the endless scrolling. VideoQ provides instant video insights. |
|
### Ask Questions to YouTube video and Save Time |
|
""",label="Description") |
|
|
|
text_box = gr.Textbox(lines=2, placeholder="Enter link of the YouTube video",label="Youtube valid link") |
|
with gr.Row(): |
|
load_button = gr.Button("Load Document") |
|
clear_button = gr.Button("Clear Document") |
|
|
|
docs_box = gr.State() |
|
loader_box = gr.State() |
|
|
|
load_button.click(fn=get_youtube_details, inputs=[text_box], outputs=[docs_box, loader_box]) |
|
clear_button.click(fn=clear_textbox, inputs=[], outputs=[text_box, docs_box, loader_box]) |
|
|
|
chatbot_interface = gr.Chatbot(show_copy_button=True,label=" ") |
|
msg = gr.Textbox(label="Message") |
|
|
|
with gr.Row(): |
|
submit_btn = gr.Button("Submit") |
|
clear_btn = gr.Button("Clear") |
|
|
|
submit_btn.click(user_message, [msg, chatbot_interface], [msg, chatbot_interface], queue=False).then( |
|
bot_message, [chatbot_interface, docs_box], chatbot_interface) |
|
|
|
msg.submit(user_message, [msg, chatbot_interface], [msg, chatbot_interface], queue=False).then( |
|
bot_message, [chatbot_interface, docs_box], chatbot_interface) |
|
|
|
clear_btn.click(lambda: None, None, chatbot_interface, queue=False) |
|
|
|
demo.launch(server_name="0.0.0.0", server_port=7860) |
|
|