app.py

import whisper 
import pytube
import gradio as gr
import openai
import faiss
from datetime import datetime
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores.faiss import FAISS
from langchain.chains import RetrievalQAWithSourcesChain
from langchain import OpenAI
from langchain.vectorstores.base import VectorStoreRetriever
import os

def get_answer(api_key, video_link, question):
    os.environ["OPENAI_API_KEY"] = api_key

    video = pytube.YouTube(video_link)
    audio = video.streams.get_audio_only()
    fn = audio.download(output_path="tmp.mp3")
    model = whisper.load_model("base")
    transcription = model.transcribe(fn)
    res = transcription['text']

    def store_segments(text):
        segment_size = 1000
        segments = [{'text': text[i:i+segment_size], 'start': i} for i in range(0, len(text), segment_size)]

        texts = []
        start_times = []

        for segment in segments:
            text = segment['text']
            start = segment['start']

            start_datetime = datetime.fromtimestamp(start)
            formatted_start_time = start_datetime.strftime('%H:%M:%S')

            texts.append(text)
            start_times.append(formatted_start_time)

        return texts, start_times

    texts, start_times = store_segments(res)

    text_splitter = CharacterTextSplitter(chunk_size=1500, separator="\n")
    docs = []
    metadatas = []
    for i, d in enumerate(texts):
        splits = text_splitter.split_text(d)
        docs.extend(splits)
        metadatas.extend([{"source": start_times[i]}] * len(splits))

    embeddings = OpenAIEmbeddings()
    store = FAISS.from_texts(docs, embeddings, metadatas=metadatas)
    faiss.write_index(store.index, "docs.index")

    retri = VectorStoreRetriever(vectorstore=store)

    chain = RetrievalQAWithSourcesChain.from_llm(llm=OpenAI(temperature=0), retriever=retri)

    result = chain({"question": question})

    return result['answer'], result['sources']

iface = gr.Interface(
    fn=get_answer,
    inputs=["text", "text", "text"],
    outputs=["text", "text"],
)

iface.queue().launch()