files

.gitignore

@@ -0,0 +1,5 @@
+.history/
+.vscode/
+notebooks/
+.*ipynb
+__pycache__/

app.py

@@ -0,0 +1,66 @@
+import streamlit as st
+from src.main import ConversationalResponse
+import os
+
+# Constants
+ROLE_USER = "user"
+ROLE_ASSISTANT = "assistant"
+
+st.set_page_config(page_title="Chat with Git Codes", page_icon="🦜")
+st.title("Chat with Git Codes 🤖📄")
+st.markdown("by [Rohan Kataria](https://www.linkedin.com/in/imrohan/) view more at [VEW.AI](https://vew.ai/)")
+st.markdown("This app allows you to chat with Git. You can paste link to the Git repository and ask questions about it. In the backround uses the Git Loader and ConversationalRetrival chain from langchain, Streamlit for UI.")
+
+@st.cache_resource(ttl="1h")
+def load_agent(url, branch, file_filter):
+    with st.spinner('Loading Git documents...'):
+        agent = ConversationalResponse(url, branch, file_filter)
+    st.success("Git Loaded Successfully")
+    return agent
+
+def main():
+    api_key = st.sidebar.text_input("Enter your OpenAI API Key", type="password")
+    if api_key:
+        os.environ["OPENAI_API_KEY"] = api_key
+    else:
+        st.sidebar.error("Please enter your OpenAI API Key.")
+        return
+
+    git_link = st.sidebar.text_input("Enter your Git Link")
+    branch = st.sidebar.text_input("Enter your Git Branch")
+    file_filter = st.sidebar.text_input("Enter the Extension of Files to Load eg. py,sql,r (no spaces)")
+
+    if "agent" not in st.session_state:
+        st.session_state["agent"] = None
+
+    if st.sidebar.button("Load Agent"):
+        if git_link and branch and file_filter:
+            try:
+                st.session_state["agent"] = load_agent(git_link, branch, file_filter)
+                st.session_state["messages"] = [{"role": ROLE_ASSISTANT, "content": "How can I help you?"}]
+            except Exception as e:
+                st.sidebar.error(f"Error loading Git repository: {str(e)}")
+                return
+
+    if st.session_state["agent"]:  # Chat will only appear if the agent is loaded
+        for msg in st.session_state.messages:
+            st.chat_message(msg["role"]).write(msg["content"])
+
+        user_query = st.chat_input(placeholder="Ask me anything!")
+
+        if user_query:
+            st.session_state.messages.append({"role": ROLE_USER, "content": user_query})
+            st.chat_message(ROLE_USER).write(user_query)
+
+            # Generate the response
+            with st.spinner("Generating response"):
+                response = st.session_state["agent"](user_query)
+
+            # Display the response immediately
+            st.chat_message(ROLE_ASSISTANT).write(response)
+
+            # Add the response to the message history
+            st.session_state.messages.append({"role": ROLE_ASSISTANT, "content": response})
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,5 @@
+openai
+streamlit
+langchain
+langchain[docarray]
+tiktoken

src/main.py

@@ -0,0 +1,124 @@
+
+import os
+import openai
+import sys
+sys.path.append('../..')
+from langchain.embeddings.openai import OpenAIEmbeddings
+from langchain.text_splitter import CharacterTextSplitter, RecursiveCharacterTextSplitter
+from langchain.vectorstores import DocArrayInMemorySearch
+from langchain.document_loaders import TextLoader
+from langchain.chains import RetrievalQA,  ConversationalRetrievalChain
+from langchain.memory import ConversationBufferMemory
+from langchain.chat_models import ChatOpenAI
+from langchain.document_loaders import TextLoader
+from langchain.document_loaders import GitLoader
+from langchain.llms import OpenAI
+from langchain.memory import ConversationBufferMemory
+from langchain.vectorstores import Chroma
+from langchain.embeddings.openai import OpenAIEmbeddings
+from langchain.prompts import PromptTemplate
+import datetime
+import shutil
+
+# Function to load the data from github using langchain with string type url, string type branch, string type file_filter
+def loader(url: str, branch: str, file_filter: str):
+    repo_path = "./github_repo"
+    if os.path.exists(repo_path):
+        shutil.rmtree(repo_path)
+
+    loader = GitLoader(
+    clone_url= url,
+    repo_path="./github_repo/",
+    branch=branch,
+    file_filter=lambda file_path: file_path.endswith(tuple(file_filter.split(','))) # Filter out files in Data but whole repo is cloned
+    )
+
+    data = loader.load()
+    return data
+    
+
+#Function to split the data into chunks using recursive character text splitter
+def split_data(data):
+    splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,
+            chunk_overlap=150,
+            length_function=len,  # Function to measure the length of chunks while splitting
+            add_start_index=True  # Include the starting position of each chunk in metadata
+    )
+    chunks = splitter.split_documents(data)
+    return chunks
+
+#Function to ingest the chunks into a vectorstore of doc
+def ingest_chunks(chunks):
+    embedding = OpenAIEmbeddings()
+    vector_store = DocArrayInMemorySearch.from_documents(chunks, embedding)
+
+    repo_path = "./github_repo"
+    if os.path.exists(repo_path):
+        shutil.rmtree(repo_path)
+
+    return vector_store
+
+#Retreival function to get the data from the database and reply to the user
+def retreival(vector_store):
+    # Selecting the right model
+    current_date = datetime.datetime.now().date()
+    if current_date < datetime.date(2023, 9, 2):
+        llm_name = "gpt-3.5-turbo-0301"
+    else:
+        llm_name = "gpt-3.5-turbo"
+
+    #Creating LLM
+    llm = ChatOpenAI(model=llm_name, temperature=0)
+
+    # Creating Prompt template
+    template = """
+        You're a code summarisation assistant. Given the following extracted parts of a long document and a question, create a final answer with "CODE SNIPPETS" from "SOURCE DOCUMENTS".
+        If you don't know the answer, just say that you don't know. Don't try to make up an answer.
+        ALWAYS return a "CODE SNIPPETS" from "SOURCE DOCUMENTS" part in your answer.
+
+        QUESTION: {question}
+        =========
+        CONTEXT: {context}
+        =========
+        FINAL ANSWER:"""
+
+    PROMPT = PromptTemplate(input_variables=["context", "question"], template=template,)
+
+    #Creating memory
+    memory = ConversationBufferMemory(
+            memory_key="chat_history",
+            input_key="question",
+            output_key="answer",
+            return_messages=True)
+
+    #Creating the retriever, this can also be a contextual compressed retriever
+    retriever = vector_store.as_retriever(search_type="similarity", search_kwargs={"k": 5}) #search_type can be "similarity" or "mmr"
+
+    chain = ConversationalRetrievalChain.from_llm(
+        llm=llm, 
+        chain_type="stuff", #chain type can be refine, stuff, map_reduce
+        retriever=retriever, 
+        memory=memory,
+        return_source_documents=True, #When used these 2 properties, the output gets 3 properties: answer, source_document, source_document_score and then have to speocify input and output key in memory for it to work
+        combine_docs_chain_kwargs=dict({"prompt": PROMPT})
+    )
+
+    return chain
+
+#Class using all above components to create QA system
+class ConversationalResponse:
+    def __init__(self, url, branch, file_filter):
+        self.url = url
+        self.branch = branch
+        self.file_filter = file_filter
+        self.data = loader(self.url, self.branch, self.file_filter)
+        self.chunks = split_data(self.data)
+        self.vector_store = ingest_chunks(self.chunks)
+        self.chain_type = "stuff"
+        self.k = 5
+        self.chain = retreival(self.vector_store)
+
+    def __call__(self, question):
+        agent = self.chain(question)
+        return agent['answer']