Upload 20 files

Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.10
+RUN useradd -m -u 1000 user
+USER user
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH
+WORKDIR $HOME/app
+COPY --chown=user . $HOME/app
+COPY ./requirements.txt ~/app/requirements.txt
+RUN pip install -r requirements.txt
+COPY . .
+CMD ["chainlit", "run", "app.py", "--port", "7860"]

app.py

@@ -0,0 +1,83 @@
+import chainlit as cl
+from llmops.text_utils import TextFileLoader, CharacterTextSplitter
+from llmops.vectordatabase import VectorDatabase
+import asyncio
+from llmops.retrieval_pipeline import RetrievalAugmentedQAPipeline, WandB_RetrievalAugmentedQAPipeline
+from llmops.openai_utils.chatmodel import ChatOpenAI
+import wandb
+from llmops.openai_utils.prompts import (
+    UserRolePrompt,
+    SystemRolePrompt,
+    AssistantRolePrompt,
+)
+
+RAQA_PROMPT_TEMPLATE = """
+Use the provided context to answer the user's query. 
+
+You may not answer the user's query unless there is specific context in the following text.
+
+If you do not know the answer, or cannot answer, please respond with "I don't know".
+
+Context:
+{context}
+"""
+
+raqa_prompt = SystemRolePrompt(RAQA_PROMPT_TEMPLATE)
+
+USER_PROMPT_TEMPLATE = """
+User Query:
+{user_query}
+"""
+
+user_prompt = UserRolePrompt(USER_PROMPT_TEMPLATE)
+
+@cl.on_chat_start
+async def on_chat_start():
+    files = None
+
+    # Wait for the user to upload a file
+    while files == None:
+        files = await cl.AskFileMessage(
+            content="Please upload a text file to begin!",
+            accept=["text/plain"],
+            max_size_mb=20,
+            timeout=180,
+        ).send()
+
+    file = files[0]
+
+    msg = cl.Message(
+        content=f"Processing `{file.name}`...", disable_human_feedback=True
+    )
+    await msg.send()
+
+    text_loader = TextFileLoader("data/KingLear.txt")
+    documents = text_loader.load_documents()
+
+    text_splitter = CharacterTextSplitter()
+    split_documents = text_splitter.split_texts(documents)
+
+    vector_db = VectorDatabase()
+    vector_db = asyncio.run(vector_db.abuild_from_list(split_documents))
+    chat_openai = ChatOpenAI()
+    wandb.init(project="RAQA Example")
+    raqa_retrieval_augmented_qa_pipeline = WandB_RetrievalAugmentedQAPipeline(
+        vector_db_retriever=vector_db,
+        llm=chat_openai,
+        wandb_project="RAQA from Scratch"
+    )
+    # Let the user know that the system is ready
+    msg.content = f"Processing `{file.name}` done. You can now ask questions!"
+    await msg.update()
+
+    cl.user_session.set("chain", raqa_retrieval_augmented_qa_pipeline)
+
+@cl.on_message
+async def main(message:str):
+    chain = cl.user_session.get("chain")
+    output = chain.run_pipeline(message,raqa_prompt, user_prompt)
+    print(output)
+    msg = cl.Message(content=f"{output}")
+    # msg.prompt = output
+    await msg.send()
+

llmops/openai_utils/chatmodel.py

@@ -0,0 +1,31 @@
+import openai
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+class ChatOpenAI:
+    """
+    This class pings open ai to create response for the list of messages
+    """
+    def __init__(self, model_name:str="gpt-3.5-turbo"):
+        self.model_name = model_name
+        self.openai_api_key = os.getenv("OPENAI_API_KEY")
+        if self.openai_api_key is None:
+            raise ValueError("OPENAI_API_KEY is not set")
+    
+    def run(self, messages:list, text_only:bool=True):
+        """
+        Takes in list of messages and returns response
+        """
+        if not isinstance(messages, list):
+            raise ValueError("Messages myst be a list")
+        
+        openai.api_key = self.openai_api_key
+        response = openai.ChatCompletion.create(
+            model=self.model_name, messages = messages
+        )
+        if text_only:
+            return response.choices[0].message.content
+        
+        return response

llmops/openai_utils/embedding.py

@@ -0,0 +1,71 @@
+from dotenv import load_dotenv
+from openai.embeddings_utils import get_embeddings, aget_embeddings, get_embedding, aget_embedding
+import openai
+from typing import List
+import os
+import asyncio
+
+class EmbeddingModel:
+    """
+    This class contains functionalities to generate embeddings from the 
+    list of texts or text asynchronously or in sync.
+    """
+    def __init__(self, embeddings_model_name:str = "text-embedding-ada-002"):
+        """
+        Loads the OpenAI Api key and sets the embedding model
+        """
+        load_dotenv()
+        self.openai_api_key = os.getenv("OPENAI_API_KEY")
+
+        if self.openai_api_key is None:
+            raise ValueError("OPENAI_API_KEY environment variables is not set. Please set it to your openAI API key")
+        
+        openai.api_key = self.openai_api_key
+        self.embeddings_model_name = embeddings_model_name
+    
+    async def async_get_embeddings(self, list_of_text:List[str])->List[List[float]]:
+        """
+        This function takes in a list of strings and uses openai api 
+        aget_embeddings to get the list of embeddings back. The process is asynchronous in nature
+
+        """
+        return await aget_embeddings(
+            list_of_text = list_of_text, engine = self.embeddings_model_name
+        )
+    
+    async def async_get_embedding(self, text: str) -> List[float]:
+        """
+        This function takes in a string and uses openai api 
+        aget_embedding to get the list of embeddings back. The process is asynchronous in nature
+
+        """
+        return await aget_embedding(text=text, engine=self.embeddings_model_name)
+    
+    def get_embeddings(self, list_of_text: List[str]) -> List[List[float]]:
+        """
+        This function takes in a list of strings and uses openai api 
+        get_embeddings to get the list of embeddings back. The process is synchronous in nature
+
+        """
+        return get_embeddings(
+            list_of_text=list_of_text, engine=self.embeddings_model_name
+        )
+
+    def get_embedding(self, text: str) -> List[float]:
+        """
+        This function takes in a string and uses openai api 
+        get_embedding to get the list of embeddings back. The process is synchronous in nature
+
+        """
+        return get_embedding(text=text, engine=self.embeddings_model_name)
+
+if __name__ == "__main__":
+    embedding_model = EmbeddingModel()
+    print(embedding_model.get_embedding("Hello, world!"))
+    print(embedding_model.get_embeddings(["Hello, world!", "Goodbye, world!"]))
+    print(asyncio.run(embedding_model.async_get_embedding("Hello, world!")))
+    print(
+        asyncio.run(
+            embedding_model.async_get_embeddings(["Hello, world!", "Goodbye, world!"])
+        )
+    )

llmops/openai_utils/prompts.py

@@ -0,0 +1,82 @@
+import re
+
+class BasePrompt:
+    def __init__(self, prompt):
+        """
+        Initializes the BasePrompt object with a prompt template.
+
+        :param prompt: A string that can contain placeholders within curly braces{}
+        """
+        self.prompt = prompt
+        self._pattern = re.compile(r"\{([^}]+)\}") 
+    
+    def format_prompt(self, **kwargs):
+        """
+        Formats the prompt string using the keyword arguments provided.
+
+        :param kwargs: The values to substitute into the prompt string
+        :return: The formatted prompt string
+        """
+        matches = self._pattern.findall(self.prompt)
+        return self.prompt.format(**{match: kwargs.get(match, "") for match in matches})
+
+    def get_input_variables(self):
+        """
+        Gets the list of input variable names from the prompt string.
+
+        :return: List of input variable names
+        """
+        return self._pattern.findall(self.prompt)
+
+class RolePrompt(BasePrompt):
+    def __init__(self, prompt, role: str):
+        """
+        Initializes the RolePrompt object with a prompt template and a role.
+
+        :param prompt: A string that can contain placeholders within curly braces
+        :param role: The role for the message ('system', 'user', or 'assistant')
+        """
+        super().__init__(prompt)
+        self.role = role
+
+    def create_message(self, **kwargs):
+        """
+        Creates a message dictionary with a role and a formatted message.
+
+        :param kwargs: The values to substitute into the prompt string
+        :return: Dictionary containing the role and the formatted message
+        """
+        return {"role": self.role, "content": self.format_prompt(**kwargs)}
+
+
+class SystemRolePrompt(RolePrompt):
+    """
+    This class inherits the RolePrompt class and sets the role to "system"
+    """
+    def __init__(self, prompt: str):
+        super().__init__(prompt, "system")
+
+
+class UserRolePrompt(RolePrompt):
+    """
+    This class inherits the RolePrompt class and sets the role to "user"
+    """
+    def __init__(self, prompt: str):
+        super().__init__(prompt, "user")
+
+
+class AssistantRolePrompt(RolePrompt):
+    """
+    This class inherits the RolePrompt class and sets the role to "assistant"
+    """
+    def __init__(self, prompt: str):
+        super().__init__(prompt, "assistant")
+
+
+if __name__ == "__main__":
+    prompt = BasePrompt("Hello {name}, you are {age} years old")
+    print(prompt.format_prompt(name="John", age=30))
+
+    prompt = SystemRolePrompt("Hello {name}, you are {age} years old")
+    print(prompt.create_message(name="John", age=30))
+    print(prompt.get_input_variables())

llmops/retrieval_pipeline.py

@@ -0,0 +1,85 @@
+from llmops.openai_utils.chatmodel import ChatOpenAI
+from llmops.vectordatabase import VectorDatabase
+from llmops.openai_utils.prompts import (
+    UserRolePrompt,
+    SystemRolePrompt,
+    AssistantRolePrompt,
+)
+import datetime
+from wandb.sdk.data_types.trace_tree import Trace
+
+class RetrievalAugmentedQAPipeline:
+    """
+    
+    """
+    def __init__(self, llm: ChatOpenAI(), vector_db_retriever: VectorDatabase) -> None:
+        self.llm = llm
+        self.vector_db_retriever = vector_db_retriever
+
+    def run_pipeline(self, user_query: str, raqa_prompt:SystemRolePrompt, user_prompt:UserRolePrompt) -> str:
+        context_list = self.vector_db_retriever.search_by_text(user_query, k=4)
+        
+        context_prompt = ""
+        for context in context_list:
+            context_prompt += context[0] + "\n"
+
+        formatted_system_prompt = raqa_prompt.create_message(context=context_prompt)
+
+        formatted_user_prompt = user_prompt.create_message(user_query=user_query)
+        
+        return self.llm.run([formatted_system_prompt, formatted_user_prompt])
+    
+
+class WandB_RetrievalAugmentedQAPipeline:
+    def __init__(self, llm: ChatOpenAI(), vector_db_retriever: VectorDatabase, wandb_project = None) -> None:
+        self.llm = llm
+        self.vector_db_retriever = vector_db_retriever
+        self.wandb_project = wandb_project
+
+    def run_pipeline(self, user_query: str, raqa_prompt:SystemRolePrompt, user_prompt:UserRolePrompt) -> str:
+        context_list = self.vector_db_retriever.search_by_text(user_query, k=4)
+        
+        context_prompt = ""
+        for context in context_list:
+            context_prompt += context[0] + "\n"
+
+        formatted_system_prompt = raqa_prompt.create_message(context=context_prompt)
+        formatted_user_prompt = user_prompt.create_message(user_query=user_query)
+        start_time = datetime.datetime.now().timestamp() * 1000
+
+        try:
+            openai_response = self.llm.run([formatted_system_prompt, formatted_user_prompt], text_only=False)
+            end_time = datetime.datetime.now().timestamp() * 1000
+            status = "success"
+            status_message = (None, )
+            response_text = openai_response.choices[0].message.content
+            token_usage = openai_response["usage"].to_dict()
+            model = openai_response["model"]
+
+        except Exception as e:
+            end_time = datetime.datetime.now().timestamp() * 1000
+            status = "error"
+            status_message = str(e)
+            response_text = ""
+            token_usage = {}
+            model = ""
+
+        if self.wandb_project:
+            root_span = Trace(
+                name="root_span",
+                kind="llm",
+                status_code=status,
+                status_message=status_message,
+                start_time_ms=start_time,
+                end_time_ms=end_time,
+                metadata={
+                    "token_usage" : token_usage,
+                    "model_name" : model
+                },
+                inputs= {"system_prompt" : formatted_system_prompt, "user_prompt" : formatted_user_prompt},
+                outputs= {"response" : response_text}
+            )
+
+            root_span.log(name="openai_trace")
+        
+        return response_text if response_text else "We ran into an error. Please try again later. Full Error Message: " + status_message

llmops/text_utils.py

@@ -0,0 +1,95 @@
+import os 
+from typing import List
+
+class TextFileLoader:
+    """
+    This class has the functionality to load the data from 
+    the text files.
+    """
+    def __init__(self, path:str, encoding:str = "utf-8")->None:
+        self.documents = []
+        self.path = path
+        self.encoding = encoding
+    
+    def load(self)->None:
+        """
+        if the path is of a directory, then load directory and read the file,
+        else if the path is of the file, directly read the file.
+        """
+        if os.path.isdir(self.path):
+            self.load_directory()
+        elif os.path.isfile(self.path) and self.path.endswith(".txt"):
+            self.load_file()
+        else:
+            raise ValueError(
+                "Provided path is neither a valid directory not a .txt tile"
+            )
+    
+    def load_file(self)->None:
+        """
+        read the text file and append it to the list
+        """
+        with open(self.path,"r",encoding=self.encoding) as f:
+            self.documents.append(f.read())
+    
+    def load_directory(self)->None:
+        """
+        reads all the text files in the directory and appends it to the list
+        """
+        for root, _, files in os.walk(self.path):
+            for file in files:
+                if file.endswith(".txt"):
+                    with open(
+                        os.path.join(root, file),"r",encoding=self.encoding
+                    ) as f:
+                        self.documents.append(f.read())
+    
+    def load_documents(self):
+        """
+        call the load function, that calls the function to read data and returns the documents.
+        """
+        self.load()
+        return self.documents
+
+class CharacterTextSplitter:
+    """
+    This class contains the functionailites to chunk the text documents.
+    """
+    def __init__(self, chunk_size:int = 1000,chunk_overlap:int = 200):
+        assert(chunk_size>chunk_overlap),"Chunk size must be greater than chunk overlap"
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+
+    def split(self, text:str)->List[str]:
+        """
+        takes in text and splits them based on character count
+        """
+        chunks = []
+        for i in range(0, len(text),self.chunk_size-self.chunk_overlap):
+            chunks.append(text[i:i+self.chunk_size])
+        return chunks
+    
+    def split_texts(self, texts:List[str])->List[str]:
+        """
+        takes in list of texts and breaks it down to chunks
+        """
+        chunks = []
+        for text in texts:
+            chunks.extend(self.split(text))
+        return chunks
+    
+if __name__ == "__main__":
+    loader = TextFileLoader("/Users/shubham.agnihotri/Documents/GitHub/LLM-Ops/RAQA from scratch/data/KingLear.txt")
+    loader.load()
+    splitter = CharacterTextSplitter()
+    chunks = splitter.split_texts(loader.documents)
+    print(len(chunks))
+    print(chunks[0])
+    print("--------")
+    print(chunks[1])
+    print("--------")
+    print(chunks[-2])
+    print("--------")
+    print(chunks[-1])
+    
+

llmops/vectordatabase.py

@@ -0,0 +1,56 @@
+import numpy as np
+from collections import defaultdict
+from typing import List, Tuple, Callable
+from llmops.openai_utils.embedding import EmbeddingModel
+import asyncio
+
+def cosine_similarity(vector_a: np.array, vector_b: np.array) -> float:
+    """Computes the cosine similarity between two vectors."""
+    dot_product = np.dot(vector_a, vector_b)
+    norm_a = np.linalg.norm(vector_a)
+    norm_b = np.linalg.norm(vector_b)
+    return dot_product / (norm_a * norm_b)
+
+class VectorDatabase:
+    def __init__(self, embedding_model:EmbeddingModel = None):
+        self.vectors = defaultdict(np.array)
+        self.embedding_model = embedding_model or EmbeddingModel()
+
+    def insert(self, key:str, vector:np.array)->None:
+        """
+        Adding elements to the dictionary vectors, with key as key and value as vector
+        """
+        self.vectors[key] = vector
+
+    def search(self, query_vector:np.array,k:int, distance_measure:Callable = cosine_similarity)->List[Tuple[str, float]]:
+        """
+        calculates cosine similarity between query vector and vector in the database and then sort the result and 
+        returns the top k values by slicing the list
+        """
+        scores = [
+            (key, distance_measure(query_vector, vector)) for key, vector in self.vectors.items()
+        ]
+        return sorted(scores, key = lambda x:x[1], reverse = True)[:k]
+    
+    def search_by_text(self, query_text:str, k:int, distance_measure:Callable = cosine_similarity, return_as_text:bool = False) -> List[Tuple[str, float]]:
+        """
+        This function converts the text query to embeddings and then calls the seach function
+        """
+        query_vector = self.embedding_model.get_embedding(query_text)
+        results = self.search(query_vector, k, distance_measure)
+        return [result[0] for result in results] if return_as_text else results
+
+    def retrieve_from_key(self, key: str) -> np.array:
+        """
+        This function returns the value of the parameter key in the vector dictionary
+        """
+        return self.vectors.get(key, None)
+
+    async def abuild_from_list(self, list_of_text: List[str]) -> "VectorDatabase":
+        """
+        create a database from a list of texts. text is key where as embedding is the mapping
+        """
+        embeddings = await self.embedding_model.async_get_embeddings(list_of_text)
+        for text, embedding in zip(list_of_text, embeddings):
+            self.insert(text, np.array(embedding))
+        return self

requirements.txt

@@ -0,0 +1,9 @@
+numpy==1.25.2
+openai==0.27.8
+python-dotenv==1.0.0
+pandas
+scikit-learn
+ipykernel
+matplotlib
+plotly
+wandb