first commit

README 2.md

@@ -0,0 +1,117 @@
+<h1 align="center">
+FileGPT 🤖
+</h1>
+
+Read the article to know how it works: <a href="">Medium Article</a>
+
+With File GPT you will be able to extract all the information from a file.
+You will obtain the transcription, the embedding of each segment and also ask questions to the file through a chat.
+
+All code was written with the help of <a href="https://codegpt.co">Code GPT</a>
+
+<a href="https://codegpt.co" target="_blank"><img width="753" alt="Captura de Pantalla 2023-02-08 a la(s) 9 16 43 p  m" src="https://user-images.githubusercontent.com/6216945/217699939-eca3ae47-c488-44da-9cf6-c7caef69e1a7.png"></a>
+
+<hr>
+<br>
+
+# Features
+
+- Read any pdf, docx, txt or csv file
+- Embedding texts segments with Langchain and OpenAI (**text-embedding-ada-002**)
+- Chat with the file using **streamlit-chat** and LangChain QA with source and (**text-davinci-003**)
+
+# Example
+For this example we are going to use this video from The PyCoach
+https://youtu.be/lKO3qDLCAnk
+
+Add the video URL and then click Start Analysis
+![Youtube](https://user-images.githubusercontent.com/6216945/217701635-7c386ca7-c802-4f56-8148-dcce57555b5a.gif)
+
+## Pytube and OpenAI Whisper
+The video will be downloaded with pytube and then OpenAI Whisper will take care of transcribing and segmenting the video.
+![Pyyube Whisper](https://user-images.githubusercontent.com/6216945/217704219-886d0afc-4181-4797-8827-82f4fd456f4f.gif)
+
+```python
+# Get the video 
+youtube_video = YouTube(youtube_link)
+streams = youtube_video.streams.filter(only_audio=True)
+mp4_video = stream.download(filename='youtube_video.mp4')
+audio_file = open(mp4_video, 'rb')
+
+# whisper load base model
+model = whisper.load_model('base')
+
+# Whisper transcription
+output = model.transcribe("youtube_video.mp4")
+```
+
+## Embedding with "text-embedding-ada-002"
+We obtain the vectors with **text-embedding-ada-002** of each segment delivered by whisper
+![Embedding](https://user-images.githubusercontent.com/6216945/217705008-180285d7-6bce-40c3-8601-576cc2f38171.gif)
+
+```python
+# Embeddings
+segments = output['segments']
+for segment in segments:
+    openai.api_key = user_secret
+    response = openai.Embedding.create(
+        input= segment["text"].strip(),
+        model="text-embedding-ada-002"
+    )
+    embeddings = response['data'][0]['embedding']
+    meta = {
+        "text": segment["text"].strip(),
+        "start": segment['start'],
+        "end": segment['end'],
+        "embedding": embeddings
+    }
+    data.append(meta)
+pd.DataFrame(data).to_csv('word_embeddings.csv') 
+```
+## OpenAI GPT-3
+We make a question to the vectorized text, we do the search of the context and then we send the prompt with the context to the model "text-davinci-003"
+
+![Question1](https://user-images.githubusercontent.com/6216945/217708086-b89dce2e-e3e2-47a7-b7dd-77e402d818cb.gif)
+
+We can even ask direct questions about what happened in the video. For example, here we ask about how long the exercise with Numpy that Pycoach did in the video took.
+
+![Question2](https://user-images.githubusercontent.com/6216945/217708485-df1edef3-d5f1-4b4a-a5c9-d08f31c80be4.gif)
+
+# Running Locally
+
+1. Clone the repository
+
+```bash
+git clone https://github.com/davila7/youtube-gpt
+cd youtube-gpt
+```
+2. Install dependencies
+
+These dependencies are required to install with the requirements.txt file:
+
+* streamlit 
+* streamlit_chat 
+* matplotlib 
+* plotly 
+* scipy 
+* sklearn 
+* pandas 
+* numpy 
+* git+https://github.com/openai/whisper.git 
+* pytube 
+* openai-whisper
+
+```bash
+pip install -r requirements.txt
+```
+3. Run the Streamlit server
+
+```bash
+streamlit run app.py
+```
+
+## Upcoming Features 🚀
+
+- Semantic search with embedding
+- Chart with emotional analysis
+- Connect with Pinecone

README.md

@@ -1,5 +1,5 @@
 ---
-title: Filegpt
+title: FileGPT
 emoji: 🐢
 colorFrom: blue
 colorTo: green

app.py

@@ -0,0 +1,116 @@
+import streamlit as st
+from streamlit_chat import message
+import os
+from utils import (
+    parse_docx,
+    parse_pdf,
+    parse_txt,
+    parse_csv,
+    search_docs,
+    embed_docs,
+    text_to_docs,
+    get_answer,
+    get_sources,
+    wrap_text_in_html,
+)
+from openai.error import OpenAIError
+
+def clear_submit():
+    st.session_state["submit"] = False
+
+def set_openai_api_key(api_key: str):
+    st.session_state["OPENAI_API_KEY"] = api_key
+
+st.markdown('<h1>File GPT 🤖<small> by <a href="https://codegpt.co">Code GPT</a></small></h1>', unsafe_allow_html=True)
+
+# Sidebar
+index = None
+doc = None
+with st.sidebar:
+    user_secret = st.text_input(
+        "OpenAI API Key",
+        type="password",
+        placeholder="Paste your OpenAI API key here (sk-...)",
+        help="You can get your API key from https://platform.openai.com/account/api-keys.",
+        value=st.session_state.get("OPENAI_API_KEY", ""),
+    )
+    if user_secret:
+        set_openai_api_key(user_secret)
+
+    uploaded_file = st.file_uploader(
+        "Upload a pdf, docx, or txt file",
+        type=["pdf", "docx", "txt", "csv"],
+        help="Scanned documents are not supported yet!",
+        on_change=clear_submit,
+    )
+
+    if uploaded_file is not None:
+        if uploaded_file.name.endswith(".pdf"):
+            doc = parse_pdf(uploaded_file)
+        elif uploaded_file.name.endswith(".docx"):
+            doc = parse_docx(uploaded_file)
+        elif uploaded_file.name.endswith(".csv"):
+            doc = parse_csv(uploaded_file)
+        elif uploaded_file.name.endswith(".txt"):
+            doc = parse_txt(uploaded_file)
+        else:
+            st.error("File type not supported")
+            doc = None
+        text = text_to_docs(doc)
+        try:
+            with st.spinner("Indexing document... This may take a while⏳"):
+                index = embed_docs(text)
+                st.session_state["api_key_configured"] = True
+        except OpenAIError as e:
+            st.error(e._message)
+
+tab1, tab2 = st.tabs(["Intro", "Chat with the File"])
+with tab1:
+    st.markdown("### How does it work?")
+    st.markdown('Read the article to know how it works: [Medium Article]("https://medium.com/@dan.avila7")')
+    st.write("File GPT was written with the following tools:")
+    st.markdown("#### Code GPT")
+    st.write("All code was written with the help of Code GPT. Visit [codegpt.co]('https://codegpt.co') to get the extension.")
+    st.markdown("#### Streamlit")
+    st.write("The design was written with [Streamlit]('https://streamlit.io/').")
+    st.markdown("#### LangChain")
+    st.write("Question answering with source [Langchain QA]('https://langchain.readthedocs.io/en/latest/use_cases/question_answering.html#adding-in-sources').")
+    st.markdown("#### Embedding")
+    st.write('[Embedding]("https://platform.openai.com/docs/guides/embeddings") is done via the OpenAI API with "text-embedding-ada-002"')
+    st.markdown("""---""")
+    st.write('Author: [Daniel Ávila](https://www.linkedin.com/in/daniel-avila-arias/)')
+    st.write('Repo: [Github](https://github.com/davila7/file-gpt)')
+    st.write("This software was developed with Code GPT, for more information visit: https://codegpt.co")
+
+with tab2:
+    st.write('To obtain an API Key you must create an OpenAI account at the following link: https://openai.com/api/')
+    if 'generated' not in st.session_state:
+        st.session_state['generated'] = []
+
+    if 'past' not in st.session_state:
+        st.session_state['past'] = []
+
+    def get_text():
+        if user_secret:
+            st.header("Ask me something about the document:")
+            input_text = st.text_area("You:", on_change=clear_submit)
+            return input_text
+    user_input = get_text()
+
+    button = st.button("Submit")
+    if button or st.session_state.get("submit"):
+        if not user_input:
+            st.error("Please enter a question!")
+        else:
+            st.session_state["submit"] = True
+            sources = search_docs(index, user_input)
+            try:
+                answer = get_answer(sources, user_input)
+                st.session_state.past.append(user_input)
+                st.session_state.generated.append(answer["output_text"].split("SOURCES: ")[0])
+            except OpenAIError as e:
+                st.error(e._message)
+            if st.session_state['generated']:
+                for i in range(len(st.session_state['generated'])-1, -1, -1):
+                    message(st.session_state["generated"][i], key=str(i))
+                    message(st.session_state['past'][i], is_user=True, key=str(i) + '_user')

embeddings.py

@@ -0,0 +1,115 @@
+"""Wrapper around OpenAI embedding models."""
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, Extra, root_validator
+
+from langchain.embeddings.base import Embeddings
+from langchain.utils import get_from_dict_or_env
+
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
+from openai.error import Timeout, APIError, APIConnectionError, RateLimitError
+
+
+class OpenAIEmbeddings(BaseModel, Embeddings):
+    """Wrapper around OpenAI embedding models.
+    To use, you should have the ``openai`` python package installed, and the
+    environment variable ``OPENAI_API_KEY`` set with your API key or pass it
+    as a named parameter to the constructor.
+    Example:
+        .. code-block:: python
+            from langchain.embeddings import OpenAIEmbeddings
+            openai = OpenAIEmbeddings(openai_api_key="my-api-key")
+    """
+
+    client: Any  #: :meta private:
+    document_model_name: str = "text-embedding-ada-002"
+    query_model_name: str = "text-embedding-ada-002"
+    openai_api_key: Optional[str] = None
+
+    class Config:
+        """Configuration for this pydantic object."""
+
+        extra = Extra.forbid
+
+    # TODO: deprecate this
+    @root_validator(pre=True, allow_reuse=True)
+    def get_model_names(cls, values: Dict) -> Dict:
+        """Get model names from just old model name."""
+        if "model_name" in values:
+            if "document_model_name" in values:
+                raise ValueError(
+                    "Both `model_name` and `document_model_name` were provided, "
+                    "but only one should be."
+                )
+            if "query_model_name" in values:
+                raise ValueError(
+                    "Both `model_name` and `query_model_name` were provided, "
+                    "but only one should be."
+                )
+            model_name = values.pop("model_name")
+            values["document_model_name"] = f"text-search-{model_name}-doc-001"
+            values["query_model_name"] = f"text-search-{model_name}-query-001"
+        return values
+
+    @root_validator(allow_reuse=True)
+    def validate_environment(cls, values: Dict) -> Dict:
+        """Validate that api key and python package exists in environment."""
+        openai_api_key = get_from_dict_or_env(
+            values, "openai_api_key", "OPENAI_API_KEY"
+        )
+        try:
+            import openai
+
+            openai.api_key = openai_api_key
+            values["client"] = openai.Embedding
+        except ImportError:
+            raise ValueError(
+                "Could not import openai python package. "
+                "Please it install it with `pip install openai`."
+            )
+        return values
+
+    @retry(
+        reraise=True,
+        stop=stop_after_attempt(100),
+        wait=wait_exponential(multiplier=1, min=10, max=60),
+        retry=(
+            retry_if_exception_type(Timeout)
+            | retry_if_exception_type(APIError)
+            | retry_if_exception_type(APIConnectionError)
+            | retry_if_exception_type(RateLimitError)
+        ),
+    )
+    def _embedding_func(self, text: str, *, engine: str) -> List[float]:
+        """Call out to OpenAI's embedding endpoint with exponential backoff."""
+        # replace newlines, which can negatively affect performance.
+        text = text.replace("\n", " ")
+        return self.client.create(input=[text], engine=engine)["data"][0]["embedding"]
+
+    def embed_documents(self, texts: List[str]) -> List[List[float]]:
+        """Call out to OpenAI's embedding endpoint for embedding search docs.
+        Args:
+            texts: The list of texts to embed.
+        Returns:
+            List of embeddings, one for each text.
+        """
+        responses = [
+            self._embedding_func(text, engine=self.document_model_name)
+            for text in texts
+        ]
+        return responses
+
+    def embed_query(self, text: str) -> List[float]:
+        """Call out to OpenAI's embedding endpoint for embedding query text.
+        Args:
+            text: The text to embed.
+        Returns:
+            Embeddings for the text.
+        """
+        embedding = self._embedding_func(text, engine=self.query_model_name)
+        return embedding

prompts.py

@@ -0,0 +1,26 @@
+from langchain.prompts import PromptTemplate
+
+## Use a shorter template to reduce the number of tokens in the prompt
+template = """Create a final answer to the given questions using the provided document excerpts(in no particular order) as references. ALWAYS include a "SOURCES" section in your answer including only the minimal set of sources needed to answer the question. If you are unable to answer the question, simply state that you do not know. Do not attempt to fabricate an answer and leave the SOURCES section empty.
+---------
+QUESTION: What  is the purpose of ARPA-H?
+=========
+Content: More support for patients and families. \n\nTo get there, I call on Congress to fund ARPA-H, the Advanced Research Projects Agency for Health. \n\nIt’s based on DARPA—the Defense Department project that led to the Internet, GPS, and so much more.  \n\nARPA-H will have a singular purpose—to drive breakthroughs in cancer, Alzheimer’s, diabetes, and more.
+Source: 1-32
+Content: While we’re at it, let’s make sure every American can get the health care they need. \n\nWe’ve already made historic investments in health care. \n\nWe’ve made it easier for Americans to get the care they need, when they need it. \n\nWe’ve made it easier for Americans to get the treatments they need, when they need them. \n\nWe’ve made it easier for Americans to get the medications they need, when they need them.
+Source: 1-33
+Content: The V.A. is pioneering new ways of linking toxic exposures to disease, already helping  veterans get the care they deserve. \n\nWe need to extend that same care to all Americans. \n\nThat’s why I’m calling on Congress to pass legislation that would establish a national registry of toxic exposures, and provide health care and financial assistance to those affected.
+Source: 1-30
+=========
+FINAL ANSWER: The purpose of ARPA-H is to drive breakthroughs in cancer, Alzheimer’s, diabetes, and more.
+SOURCES: 1-32
+---------
+QUESTION: {question}
+=========
+{summaries}
+=========
+FINAL ANSWER:"""
+
+STUFF_PROMPT = PromptTemplate(
+    template=template, input_variables=["summaries", "question"]
+)

requirements.txt

@@ -0,0 +1,10 @@
+openai
+pypdf
+scikit-learn
+numpy
+tiktoken
+docx2txt
+langchain
+pydantic
+typing
+faiss-cpu

utils.py

@@ -0,0 +1,167 @@
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain.vectorstores.faiss import FAISS
+from langchain import OpenAI, Cohere
+from langchain.chains.qa_with_sources import load_qa_with_sources_chain
+from embeddings import OpenAIEmbeddings
+from langchain.llms import OpenAI
+from langchain.docstore.document import Document
+from langchain.vectorstores import FAISS, VectorStore
+import docx2txt
+from typing import List, Dict, Any
+import re
+import numpy as np
+from io import StringIO
+from io import BytesIO
+import streamlit as st
+from prompts import STUFF_PROMPT
+from pypdf import PdfReader
+from openai.error import AuthenticationError
+
+@st.experimental_memo()
+def parse_docx(file: BytesIO) -> str:
+    text = docx2txt.process(file)
+    # Remove multiple newlines
+    text = re.sub(r"\n\s*\n", "\n\n", text)
+    return text
+
+
+@st.experimental_memo()
+def parse_pdf(file: BytesIO) -> List[str]:
+    pdf = PdfReader(file)
+    output = []
+    for page in pdf.pages:
+        text = page.extract_text()
+        # Merge hyphenated words
+        text = re.sub(r"(\w+)-\n(\w+)", r"\1\2", text)
+        # Fix newlines in the middle of sentences
+        text = re.sub(r"(?<!\n\s)\n(?!\s\n)", " ", text.strip())
+        # Remove multiple newlines
+        text = re.sub(r"\n\s*\n", "\n\n", text)
+
+        output.append(text)
+
+    return output
+
+
+@st.experimental_memo()
+def parse_txt(file: BytesIO) -> str:
+    text = file.read().decode("utf-8")
+    # Remove multiple newlines
+    text = re.sub(r"\n\s*\n", "\n\n", text)
+    return text
+
+@st.experimental_memo()
+def parse_csv(uploaded_file):
+    # To read file as bytes:
+    #bytes_data = uploaded_file.getvalue()
+    #st.write(bytes_data)
+
+    # To convert to a string based IO:
+    stringio = StringIO(uploaded_file.getvalue().decode("utf-8"))
+    #st.write(stringio)
+
+    # To read file as string:
+    string_data = stringio.read()
+    #st.write(string_data)
+
+    # Can be used wherever a "file-like" object is accepted:
+    # dataframe = pd.read_csv(uploaded_file)
+    return string_data
+
+
+@st.cache(allow_output_mutation=True)
+def text_to_docs(text: str | List[str]) -> List[Document]:
+    """Converts a string or list of strings to a list of Documents
+    with metadata."""
+    if isinstance(text, str):
+        # Take a single string as one page
+        text = [text]
+    page_docs = [Document(page_content=page) for page in text]
+
+    # Add page numbers as metadata
+    for i, doc in enumerate(page_docs):
+        doc.metadata["page"] = i + 1
+
+    # Split pages into chunks
+    doc_chunks = []
+
+    for doc in page_docs:
+        text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=800,
+            separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""],
+            chunk_overlap=0,
+        )
+        chunks = text_splitter.split_text(doc.page_content)
+        for i, chunk in enumerate(chunks):
+            doc = Document(
+                page_content=chunk, metadata={"page": doc.metadata["page"], "chunk": i}
+            )
+            # Add sources a metadata
+            doc.metadata["source"] = f"{doc.metadata['page']}-{doc.metadata['chunk']}"
+            doc_chunks.append(doc)
+    return doc_chunks
+
+
+@st.cache(allow_output_mutation=True, show_spinner=False)
+def embed_docs(docs: List[Document]) -> VectorStore:
+    """Embeds a list of Documents and returns a FAISS index"""
+
+    if not st.session_state.get("OPENAI_API_KEY"):
+        raise AuthenticationError(
+            "Enter your OpenAI API key in the sidebar. You can get a key at https://platform.openai.com/account/api-keys."
+        )
+    else:
+        # Embed the chunks
+        embeddings = OpenAIEmbeddings(openai_api_key=st.session_state.get("OPENAI_API_KEY"))  # type: ignore
+        index = FAISS.from_documents(docs, embeddings)
+
+        return index
+
+
+@st.cache(allow_output_mutation=True)
+def search_docs(index: VectorStore, query: str) -> List[Document]:
+    """Searches a FAISS index for similar chunks to the query
+    and returns a list of Documents."""
+
+    # Search for similar chunks
+    docs = index.similarity_search(query, k=5)
+    return docs
+
+
+@st.cache(allow_output_mutation=True)
+def get_answer(docs: List[Document], query: str) -> Dict[str, Any]:
+    """Gets an answer to a question from a list of Documents."""
+
+    # Get the answer
+
+    chain = load_qa_with_sources_chain(OpenAI(temperature=0, openai_api_key=st.session_state.get("OPENAI_API_KEY")), chain_type="stuff", prompt=STUFF_PROMPT)  # type: ignore
+
+    # Cohere doesn't work very well as of now.
+    # chain = load_qa_with_sources_chain(Cohere(temperature=0), chain_type="stuff", prompt=STUFF_PROMPT)  # type: ignore
+    answer = chain(
+        {"input_documents": docs, "question": query}, return_only_outputs=True
+    )
+    return answer
+
+
+@st.cache(allow_output_mutation=True)
+def get_sources(answer: Dict[str, Any], docs: List[Document]) -> List[Document]:
+    """Gets the source documents for an answer."""
+
+    # Get sources for the answer
+    source_keys = [s for s in answer["output_text"].split("SOURCES: ")[-1].split(", ")]
+
+    source_docs = []
+    for doc in docs:
+        if doc.metadata["source"] in source_keys:
+            source_docs.append(doc)
+
+    return source_docs
+
+
+def wrap_text_in_html(text: str | List[str]) -> str:
+    """Wraps each text block separated by newlines in <p> tags"""
+    if isinstance(text, list):
+        # Add horizontal rules between pages
+        text = "\n<hr/>\n".join(text)
+    return "".join([f"<p>{line}</p>" for line in text.split("\n")])