Genesis commit

app/__init__.py

@@ -0,0 +1,12 @@
+# app/__init__.py
+
+from app.config import get_config
+from fastapi import FastAPI
+
+def create_app(env_name: str) -> FastAPI:
+    config = get_config(env_name)
+    app = FastAPI(title="Code Challenge")
+    
+    app.config = config
+
+    return app

app/config.py

@@ -0,0 +1,56 @@
+import os
+
+class BaseConfig:
+    """
+    Base configuration class that holds default settings for the application.
+    Environment-specific configurations will inherit from this class.
+    """
+    PDF_FOLDER = os.getenv("PDF_FOLDER", "./app/pdfs")
+    ENV = "base"
+    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-api-key-here")
+    CREATE_FAISS_INDEX = True
+
+    def __init__(self):
+        """
+        Validates required configurations and ensures all necessary environment variables are set.
+        """
+        if not self.OPENAI_API_KEY:
+            raise ValueError("OPENAI_API_KEY environment variable must be set.")
+
+class DevelopmentConfig(BaseConfig):
+    """
+    Configuration class for the development environment.
+    Inherits defaults from BaseConfig.
+    """
+    ENV = "development"
+    DEBUG = True
+
+
+class ProductionConfig(BaseConfig):
+    """
+    Configuration class for the production environment.
+    Inherits defaults from BaseConfig but overrides production-specific settings.
+    """
+    ENV = "production"
+    DEBUG = False
+
+
+def get_config(env_name: str = "development"):
+    """
+    Retrieves the appropriate configuration instance based on the environment name.
+
+    :param env_name: Name of the environment (e.g., 'development', 'production').
+    :return: An instance of the selected configuration class.
+    """
+    configs = {
+        "development": DevelopmentConfig,
+        "production": ProductionConfig,
+    }
+    config_class = configs.get(env_name.lower())
+
+    if not config_class:
+        raise ValueError(f"Unknown environment '{env_name}'. Valid options are 'development' or 'production'.")
+
+    config_instance = config_class()
+    print(f"[INFO] Loaded configuration for environment: {config_instance.ENV}")
+    return config_instance

app/main.py

@@ -0,0 +1,87 @@
+import asyncio
+import sys
+from app.services.dependencies import get_file_service, get_faiss_service
+from app.config import get_config
+from langchain_openai import ChatOpenAI
+from langchain.schema import Document
+from langchain.prompts import PromptTemplate
+
+
+async def main():
+    """
+    Entry point for the XBO product assistant.
+    """
+    config = get_config(env_name="development")
+    file_service = get_file_service(config=config)
+
+    openai_api_key = config.OPENAI_API_KEY
+    create_faiss_index = config.CREATE_FAISS_INDEX
+    print ("Create FAISS Index: ", create_faiss_index)
+    print("Wilkommen zum XBO Kaufberater!")
+    faiss_service = get_faiss_service(openai_api_key)
+
+    try:
+        if create_faiss_index:
+            print("[INFO] Creating a new FAISS index...")
+            pdfs = file_service.load_pdfs()
+            if not pdfs:
+                print("[ERROR] No PDFs found.")
+                sys.exit(1)
+
+            all_documents = []
+            for pdf in pdfs:
+                print(f"Processing PDF: {pdf}")
+                text = file_service.extract_text_from_pdf(pdf)
+                all_documents.append(Document(page_content=text, metadata={"source": pdf}))
+
+            vectorstore = faiss_service.create_faiss_index(all_documents)
+        else:
+            vectorstore = faiss_service.load_faiss_index()
+    except Exception as e:
+        print(f"[ERROR] {e}")
+        sys.exit(1)
+
+    llm = ChatOpenAI(model="gpt-4o", openai_api_key=openai_api_key)
+    retriever = vectorstore.as_retriever(search_kwargs={"k": 21}, search_type="mmr")
+
+    while True:
+        user_input = input("\nWas möchten Sie wissen? (type 'exit' to quit): ").strip()
+        if user_input.lower() == "exit":
+            print("Auf Wiedersehen!")
+            break
+
+        try:
+            print("[INFO] Retrieving relevant documents...")
+            docs = retriever.invoke(user_input)
+
+            if not docs:
+                print("\n[ANSWER]: Keine passenden Informationen gefunden.")
+                continue
+
+            context = "\n\n".join([doc.page_content for doc in docs])
+
+            prompt = PromptTemplate(
+                template="""
+                Du bist ein Assistent, der Fragen zu Produktinformationen beantwortet.
+
+                Kontext:
+                {context}
+
+                Frage:
+                {question}
+
+                Antwort:
+                """,
+                input_variables=["context", "question"]
+            )
+
+            response = llm.invoke(prompt.format(context=context, question=user_input))
+
+            print("\n[ANSWER]:")
+            print(response.content)
+        except Exception as e:
+            print(f"[ERROR] Failed to process query: {e}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

app/services/dependencies.py

@@ -0,0 +1,33 @@
+from fastapi import Depends
+from app.config import get_config
+from app.services.file_service import FileService
+from app.services.faiss_service import FAISSService
+
+
+
+def get_file_service(config=Depends(get_config)) -> FileService:
+    """
+    Dependency function to provide a FileService instance.
+
+    :param config: Configuration object obtained via dependency injection.
+    :return: An instance of FileService initialized with the PDF folder path.
+    """
+    if not hasattr(config, "PDF_FOLDER") or not config.PDF_FOLDER:
+        raise ValueError("PDF_FOLDER is not configured in the application settings.")
+    return FileService(folder_path=config.PDF_FOLDER)
+
+
+def get_faiss_service(file_service=Depends(get_file_service)) -> FAISSService:
+    """
+    Dependency function to provide a FAISSService instance.
+
+    :param config: Configuration object obtained via dependency injection.
+    :param file_service: FileService instance for handling PDFs and documents.
+    :return: An instance of FAISSService initialized with vectorstore and embeddings.
+    """
+    config = get_config()
+   
+    return FAISSService(
+        openai_api_key=config.OPENAI_API_KEY,
+        index_path="local_faiss_index",
+    )

app/services/faiss_service.py

@@ -0,0 +1,53 @@
+from langchain_community.vectorstores import FAISS
+from langchain_openai import OpenAIEmbeddings
+from langchain.schema import Document
+
+
+class FAISSService:
+    """
+    A service for creating and loading FAISS indexes for document embeddings.
+    """
+
+    def __init__(self, openai_api_key, index_path="local_faiss_index"):
+        """
+        Initialize the FAISS service.
+
+        :param openai_api_key: OpenAI API key for embeddings.
+        :param index_path: Path to save or load the FAISS index.
+        """
+        self.openai_api_key = openai_api_key
+        self.index_path = index_path
+
+    def create_faiss_index(self, documents):
+        """
+        Create a FAISS index from a list of documents.
+
+        :param documents: List of langchain Document objects.
+        :return: FAISS vectorstore instance.
+        """
+        print("[INFO] Creating FAISS index...")
+        vectorstore = FAISS.from_documents(
+            documents,
+            OpenAIEmbeddings(
+                model="text-embedding-ada-002",
+                openai_api_key=self.openai_api_key
+            )
+        )
+        vectorstore.save_local(self.index_path)
+        print(f"[INFO] FAISS index saved to {self.index_path}.")
+        return vectorstore
+
+    def load_faiss_index(self):
+        """
+        Load an existing FAISS index.
+
+        :return: Loaded FAISS vectorstore instance.
+        """
+        print("[INFO] Loading FAISS index...")
+        vectorstore = FAISS.load_local(
+            self.index_path,
+            OpenAIEmbeddings(openai_api_key=self.openai_api_key),
+            allow_dangerous_deserialization=True  
+        )
+        print(f"[INFO] FAISS index loaded from {self.index_path}.")
+        return vectorstore

app/services/file_service.py

@@ -0,0 +1,49 @@
+# app/services/file_service.py
+
+import os
+import pdfplumber
+
+class FileService:
+    """
+    A service to handle file-related operations, including loading PDFs from a folder.
+    """
+    def __init__(self, folder_path: str):
+        """
+        Initialize the FileService with the folder path to read files from.
+        """
+        self.folder_path = os.path.abspath(folder_path)  
+      #  print(f"[DEBUG] Initialized FileService with folder path: {self.folder_path}")
+
+    def load_pdfs(self):
+        """
+        Reads all PDF files from the folder and returns their paths.
+        
+        :return: List of paths to PDF files in the folder.
+        """
+        if not os.path.exists(self.folder_path):
+            raise FileNotFoundError(f"The folder {self.folder_path} does not exist.")
+
+        pdf_files = [
+            os.path.join(self.folder_path, f)
+            for f in os.listdir(self.folder_path)
+            if f.endswith(".pdf")
+        ]
+
+        if not pdf_files:
+            raise FileNotFoundError(f"No PDF files found in the folder {self.folder_path}.")
+
+        return pdf_files
+
+    def extract_text_from_pdf(self, pdf_path):
+        """
+        Extracts text from the PDF file using pdfplumber.
+        :param pdf_path: Path to the PDF file.
+        :return: Extracted text as a string.
+        """
+        text = ""
+        with pdfplumber.open(pdf_path) as pdf:
+            for page in pdf.pages:
+                page_text = page.extract_text()
+                if page_text:
+                    text += page_text + "\n"
+        return text