| 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() |
|
|
