Delete aimakerspace

aimakerspace/openai_utils/chatmodel.py

@@ -1,47 +0,0 @@
-from openai import OpenAI, AsyncOpenAI
-from dotenv import load_dotenv
-import os
-
-load_dotenv()
-
-
-class ChatOpenAI:
-    def __init__(self, model_name: str = "gpt-4o-mini"):
-        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, text_only: bool = True, **kwargs):
-        if not isinstance(messages, list):
-            raise ValueError("messages must be a list")
-
-        client = OpenAI()
-        response = client.chat.completions.create(
-            model=self.model_name, messages=messages, **kwargs
-        )
-
-        if text_only:
-            return response.choices[0].message.content
-
-        return response
-    
-    async def astream(self, messages, **kwargs):
-        if not isinstance(messages, list):
-            raise ValueError("messages must be a list")
-        
-        client = AsyncOpenAI()
-
-        stream = await client.chat.completions.create(
-            model=self.model_name,
-            messages=messages,
-            stream=True,
-            **kwargs
-        )
-
-        async for chunk in stream:
-            content = chunk.choices[0].delta.content
-            if content is not None:
-                yield content
-
-

aimakerspace/openai_utils/embedding.py

@@ -1,68 +0,0 @@
-from dotenv import load_dotenv
-from openai import AsyncOpenAI, OpenAI
-import openai
-from typing import List
-import os
-import asyncio
-
-
-class EmbeddingModel:
-    def __init__(self, embeddings_model_name: str = "text-embedding-3-small"):
-        load_dotenv()
-        self.openai_api_key = os.getenv("OPENAI_API_KEY")
-        self.async_client = AsyncOpenAI()
-        self.client = OpenAI()
-
-        if self.openai_api_key is None:
-            raise ValueError(
-                "OPENAI_API_KEY environment variable 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]]:
-        batch_size = 1024
-        batches = [list_of_text[i:i + batch_size] for i in range(0, len(list_of_text), batch_size)]
-        
-        async def process_batch(batch):
-            embedding_response = await self.async_client.embeddings.create(
-                input=batch, model=self.embeddings_model_name
-            )
-            return [embeddings.embedding for embeddings in embedding_response.data]
-        
-        # Use asyncio.gather to process all batches concurrently
-        results = await asyncio.gather(*[process_batch(batch) for batch in batches])
-        
-        # Flatten the results
-        return [embedding for batch_result in results for embedding in batch_result]
-
-    async def async_get_embedding(self, text: str) -> List[float]:
-        embedding = await self.async_client.embeddings.create(
-            input=text, model=self.embeddings_model_name
-        )
-
-        return embedding.data[0].embedding
-
-    def get_embeddings(self, list_of_text: List[str]) -> List[List[float]]:
-        embedding_response = self.client.embeddings.create(
-            input=list_of_text, model=self.embeddings_model_name
-        )
-
-        return [embeddings.embedding for embeddings in embedding_response.data]
-
-    def get_embedding(self, text: str) -> List[float]:
-        embedding = self.client.embeddings.create(
-            input=text, model=self.embeddings_model_name
-        )
-
-        return embedding.data[0].embedding
-
-
-if __name__ == "__main__":
-    embedding_model = EmbeddingModel()
-    print(asyncio.run(embedding_model.async_get_embedding("Hello, world!")))
-    print(
-        asyncio.run(
-            embedding_model.async_get_embeddings(["Hello, world!", "Goodbye, world!"])
-        )
-    )

aimakerspace/openai_utils/prompts.py

@@ -1,78 +0,0 @@
-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, format=True, **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
-        """
-        if format:
-            return {"role": self.role, "content": self.format_prompt(**kwargs)}
-        
-        return {"role": self.role, "content": self.prompt}
-
-
-class SystemRolePrompt(RolePrompt):
-    def __init__(self, prompt: str):
-        super().__init__(prompt, "system")
-
-
-class UserRolePrompt(RolePrompt):
-    def __init__(self, prompt: str):
-        super().__init__(prompt, "user")
-
-
-class AssistantRolePrompt(RolePrompt):
-    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())

aimakerspace/text_utils.py

@@ -1,116 +0,0 @@
-import os
-from typing import List
-from PyPDF2 import PdfReader
-
-
-class TextFileLoader:
-    def __init__(self, path: str, encoding: str = "utf-8"):
-        self.documents = []
-        self.path = path
-        self.encoding = encoding
-
-    def load(self):
-        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 nor a .txt file."
-            )
-
-    def load_file(self):
-        with open(self.path, "r", encoding=self.encoding) as f:
-            self.documents.append(f.read())
-
-    def load_directory(self):
-        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):
-        self.load()
-        return self.documents
-
-class PDFFileLoader:
-    def __init__(self, path: str):
-        self.documents = []
-        self.path = path
-
-    def load(self):
-        if os.path.isdir(self.path):
-            self.load_directory()
-        elif os.path.isfile(self.path) and self.path.endswith(".pdf"):
-            self.load_file()
-        else:
-            raise ValueError(
-                "Provided path is neither a valid directory nor a .pdf file."
-            )
-
-    def load_file(self):
-        with open(self.path, "rb") as file:
-            pdf_reader = PdfReader(file)
-            text = ""
-            for page in pdf_reader.pages:
-                text += page.extract_text()
-            self.documents.append(text)
-
-    def load_directory(self):
-        for root, _, files in os.walk(self.path):
-            for file in files:
-                if file.endswith(".pdf"):
-                    file_path = os.path.join(root, file)
-                    with open(file_path, "rb") as f:
-                        pdf_reader = PdfReader(f)
-                        text = ""
-                        for page in pdf_reader.pages:
-                            text += page.extract_text()
-                        self.documents.append(text)
-
-    def load_documents(self):
-        self.load()
-        return self.documents
-
-class CharacterTextSplitter:
-    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]:
-        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]:
-        chunks = []
-        for text in texts:
-            chunks.extend(self.split(text))
-        return chunks
-
-
-if __name__ == "__main__":
-    loader = TextFileLoader("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])

aimakerspace/vectordatabase.py

@@ -1,91 +0,0 @@
-import numpy as np
-from collections import defaultdict
-from typing import List, Tuple, Callable
-from aimakerspace.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)
-
-def jaccard_binary(vector_a: np.array, vector_b: np.array):
-    """A function for finding the similarity between two binary vectors"""
-    intersection = len(list(set(vector_a).intersection(vector_b)))
-    union = (len(vector_a) + len(vector_b)) - intersection
-    return float(intersection) / union
-
-def euclidean_distance(vector_a: np.array, vector_b: np.array) -> float:
-    """Computes the euclidean distance between two vectors."""
-    return np.linalg.norm(vector_a - vector_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:
-        self.vectors[key] = vector
-
-    def search(
-        self,
-        query_vector: np.array,
-        k: int,
-        distance_measure: Callable = cosine_similarity,
-    ) -> List[Tuple[str, float]]:
-        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]]:
-        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:
-        return self.vectors.get(key, None)
-
-    async def abuild_from_list(self, list_of_text: List[str]) -> "VectorDatabase":
-        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
-
-
-if __name__ == "__main__":
-    list_of_text = [
-        "I like to eat broccoli and bananas.",
-        "I ate a banana and spinach smoothie for breakfast.",
-        "Chinchillas and kittens are cute.",
-        "My sister adopted a kitten yesterday.",
-        "Look at this cute hamster munching on a piece of broccoli.",
-    ]
-
-    vector_db = VectorDatabase()
-    vector_db = asyncio.run(vector_db.abuild_from_list(list_of_text))
-    k = 2
-
-    searched_vector = vector_db.search_by_text("I think fruit is awesome!", k=k)
-    print(f"Closest {k} vector(s):", searched_vector)
-
-    retrieved_vector = vector_db.retrieve_from_key(
-        "I like to eat broccoli and bananas."
-    )
-    print("Retrieved vector:", retrieved_vector)
-
-    relevant_texts = vector_db.search_by_text(
-        "I think fruit is awesome!", k=k, return_as_text=True
-    )
-    print(f"Closest {k} text(s):", relevant_texts)