inital commit

config.py

@@ -0,0 +1,164 @@
+"""
+Configuration settings for the Executive Education RAG Chatbot.
+"""
+import os
+from dotenv import load_dotenv
+
+# Load environment variables from .env file
+load_dotenv()
+
+class LLMProvider:
+    def __init__(self, base: str, sub: str | None = None) -> None:
+        self.base = base
+        self.sub  = sub
+        self.name = f"{base}:{sub}" if sub else base 
+    
+
+    def with_sub(self, sub: str | None = None) -> str:
+        return LLMProvider(self.base, sub)
+
+
+class LLMProviderConfiguration:
+    AVAIABLE_PROVIDERS = [
+        'groq', 
+        'ollama',  
+        'openai',
+        'open_router',
+    ]
+    AVAILABLE_SUBPROVIDERS = {
+        'groq': [],
+        'open_router': [
+            'openai', 
+            'deepseek',
+            'meituan'
+            'alibaba'   # For tongyi models 
+            'nvidia',
+        ],
+    }
+    
+    # DEFINE YOUR MAIN MODEL PROVIDER HERE 
+    # Some unified interfaces such as Groq or Open Router provide access to other providers
+    # such as OpenAI or Deepseek. When using interfaces define the provider you want to gain access to. 
+    LLM_PROVIDER = LLMProvider('openai')
+    
+    # -------------------- Some predefined models for available providers ----------------------
+
+    # Groq settings
+    GROQ_API_KEY = os.getenv("GROQ_API_KEY")
+    GROQ_MODEL = "mixtral-8x7b-32768"
+    
+    # Open Router settings
+    OPEN_ROUTER_API_KEY = os.getenv("OPEN_ROUTER_API_KEY")
+    OPEN_ROUTER_MODEL="meituan/longcat-flash-chat:free"
+    OPEN_ROUTER_BASE_URL="https://openrouter.ai/api/v1"
+
+    # OpenAI settings
+    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+    OPENAI_MODEL = "gpt-5"
+    
+    # The gpt-oss:20b model is preferable but takes much more space
+    # Set to False if you only have the llama3.2 installed
+    GPT_OSS_ENABLED=False
+    # Local/Ollama settings
+    OLLAMA_BASE_URL = "http://localhost:11434"
+    OLLAMA_MODEL = "gpt-oss:20b" if GPT_OSS_ENABLED else "llama3.2"
+    
+    # ----------------------------------------------------------------------------------------
+
+    @classmethod
+    def get_fallback_models(cls, provider: LLMProvider | None = None) -> list[str]:
+        provider = provider or cls.LLM_PROVIDER
+        match provider.base:
+            case 'openai':
+                return {
+                    provider: fallback_model
+                    for fallback_model in [
+                        'gpt-5-mini', 
+                        'gpt-5-nano',
+                    ]
+                }
+            case 'open_router':
+                return {
+                    provider.with_sub('openai'):   "gpt-oss-20b",
+                    provider.with_sub('openai'):   "gpt-oss-120b",
+                    provider.with_sub('alibaba'):  "alibaba/tongyi-deepresearch-30b-a3b:free",
+                    provider: "openrouter/polaris-alpha",
+                    # Currently unusable because has no tool support
+                    #provider.with_sub('deepseek'): "deepseek/deepseek-chat-v3.1:free",
+                }
+            case _:
+                return {}
+
+    @classmethod
+    def get_reasoning_support(cls, provider: LLMProvider | None = None) -> bool:
+        provider = provider or cls.LLM_PROVIDER
+        return {
+            "groq":   True,
+            "openai": True, 
+            "open_router": True,
+        }.get(provider.base, False)
+
+
+    @classmethod
+    def get_default_model(cls, provider: LLMProvider | None = None) -> str:
+        provider = provider or cls.LLM_PROVIDER
+        return {
+            "groq":   cls.GROQ_MODEL,
+            "openai": cls.OPENAI_MODEL, 
+            "ollama": cls.OLLAMA_MODEL,
+            "open_router":   cls.OPEN_ROUTER_MODEL,
+        }.get(provider.base)
+   
+
+    @classmethod
+    def get_api_key(cls, provider: LLMProvider | None = None) -> str:
+        provider = provider or cls.LLM_PROVIDER
+        return {
+            "groq": cls.GROQ_API_KEY,
+            "openai": cls.OPENAI_API_KEY,
+            "open_router": cls.OPEN_ROUTER_API_KEY,
+        }.get(provider.base)
+
+
+# Weaviate database settings 
+class WeaviateConfiguration:
+    LOCAL_DATABASE = False
+    WEAVIATE_BACKUP_BACKEND = ''
+    WEAVIATE_COLLECTION_BASENAME = 'hsg_rag_content'
+    
+    # Weaviate Cloud settings
+    CLUSTER_URL = "r2vd9fuvrcjvx7idsvta.c0.europe-west3.gcp.weaviate.cloud"
+    WEAVIATE_API_KEY = os.getenv('WEAVIATE_API_KEY')
+    HUGGING_FACE_API_KEY = os.getenv('HUGGING_FACE_API_KEY')
+
+    @classmethod 
+    def is_local(cls) -> bool:
+        return cls.LOCAL_DATABASE
+
+# Data paths
+DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
+RAW_DATA_PATH = os.path.join(DATA_DIR, "raw_data.json")
+PROCESSED_DATA_PATH = os.path.join(DATA_DIR, "processed_data.json")
+VECTORDB_PATH = os.path.join(DATA_DIR, "vectordb")
+
+# Vector database settings
+CHUNK_SIZE = 1000
+CHUNK_OVERLAP = 200
+
+# Agent Chain settings 
+MAX_MODEL_RETRIES = 3
+
+# RAG settings
+TOP_K_RETRIEVAL = 8  # Number of documents to retrieve for each query
+
+# UI settings
+MAX_HISTORY = 10  # Maximum number of conversation turns to keep in history
+
+# Data processing pipeline settings 
+CHUNK_MAX_TOKENS = 8191
+AVAILABLE_LANGUAGES = ['en', 'de']
+HASH_FILE_PATH = os.path.join(DATA_DIR, 'hashtables.json')
+DOCUMENTS_PATH = os.path.join(DATA_DIR, 'documents')
+
+# Base URL for scraping
+BASE_URL = "https://emba.unisg.ch/programm"

main.py

@@ -0,0 +1,130 @@
+"""
+Main entry point for the Executive Education RAG Chatbot.
+"""
+import argparse
+from src.utils.logging import init_logging, get_logger
+from config import AVAILABLE_LANGUAGES
+
+# Initialize logging
+def logging_startup():
+    init_logging(interactive_mode=False)
+    return get_logger('main_module')
+
+
+def run_scraper() -> None:
+    """
+    Run the scraper to collect program data.
+
+    Args:
+        use_selenium: Whether to use Selenium for scraping.
+    """ 
+    from src.pipeline.pipeline import ImportPipeline
+    logger = logging_startup()
+
+    logger.info("Running scraper...")
+    ImportPipeline().scrape_website()
+    logger.info("Scraping completed.")
+
+
+def run_importer(sources: list[str]) -> None:
+    """Run the data import pipeline.""" 
+    from src.pipeline.pipeline import ImportPipeline
+    logger = logging_startup()
+
+    logger.info("Running data import pipeline...")
+    ImportPipeline().import_many_documents(sources)
+    logger.info("Data processing completed.")
+
+
+def run_weaviate_command(command: str, backup_id: str = None): 
+    """Run commands to manipulate the database contents.""" 
+    from src.database.weavservice import WeaviateService
+    logger = logging_startup()
+
+    logger.info(f"Running database command {command}")
+    if command == 'restore' and not backup_id:
+        logger.error("Backup ID is required to initalize the restore process.")
+    
+    service = WeaviateService()
+    if command == 'backup':
+        service._create_backup()
+
+    if command == 'restore':
+        service._restore_backup(backup_id)
+
+    if command == 'delete' or command == 'redo':
+        service._delete_collections()
+
+    if command == 'init' or command == 'redo':
+        service._create_collections()
+
+    if command == 'checkhealth' or command == 'init' or command == 'redo':
+        service._checkhealth()
+
+
+def run_application(lang: str) -> None:
+    """Run the chatbot web application."""
+    from src.apps.chat.app import ChatbotApplication
+    logger = logging_startup()
+
+    logger.info("Starting chatbot web application...")
+    app = ChatbotApplication(language=lang)
+    app.run()
+
+
+def run_cli() -> None:
+    """Run the chatbot CLI."""
+    from src.ui.cli import ChatbotCLI
+    logger = logging_startup()
+
+    logger.info("Starting chatbot...")
+    cli = ChatbotCLI()
+    cli.run()
+
+
+def parse_args():
+    """Parse command-line arguments."""
+    parser = argparse.ArgumentParser(description="University of St. Gallen Executive Education RAG Chatbot")
+    
+    # Add arguments
+    parser.add_argument("--scrape", action="store_true", help="Scrapes the data from the HSG website and imports it into the database")
+    parser.add_argument("--imports", nargs="+", help="Runs the data importing pipeline for the provided files")
+    
+    parser.add_argument("--weaviate", type=str, choices=['init', 'delete', 'redo', 'checkhealth', 'backup', 'restore'], help="Runs different database actions")
+    parser.add_argument("--backup-id", type=str, help="Required when calling the --weaviate restore command!")
+
+    parser.add_argument("--cli", action="store_true", help="Run the chatbot CLI")
+    parser.add_argument("--app", type=str, choices=AVAILABLE_LANGUAGES, help="Run the chatbot web application")
+    
+    return parser.parse_args()
+
+
+def main():
+    """Main entry point for the application."""
+    args = parse_args()
+    
+    # Check if any argument is provided
+    if not any([args.scrape, args.imports, args.weaviate, args.cli, args.app]):
+        # If no argument is provided, run the chatbot by default
+        run_app()
+        return 
+
+    # Run the specified components
+    if args.scrape:
+        run_scraper()
+
+    if args.imports:
+        run_importer(args.imports)
+    
+    if args.weaviate:
+        run_weaviate_command(command=args.weaviate, backup_id=args.backup_id)
+
+    if args.cli:
+        run_cli()
+
+    if args.app:
+        run_application(args.app)
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,30 @@
+# Core dependencies
+langchain>=1.0.2
+langchain-community>=0.3.29
+langchain-core>=1.0.1
+langchain-deepseek>=1.0.0
+langchain-groq>=1.0.0
+langchain-ollama>=0.3.10
+langchain-openai>=1.0.1
+
+requests>=2.31.0
+openai>=1.3.0
+python-dotenv>=1.0.0
+tiktoken>=0.5.1
+colorama>=0.4.6
+
+# Web applications
+gradio>=5.49.1
+
+# Data processing
+numpy>=1.24.3
+pandas>=2.0.3
+
+# UI
+rich>=13.5.2
+
+# Document processing pipeline
+docling>=2.55.0
+
+# Weaviate Vector DB 
+weaviate-client>=4.16.9

src/apps/chat/.gradio/flagged/dataset1.csv

@@ -0,0 +1,3 @@
+name,intensity,output,timestamp
+weqwe,30,Hello weqwe!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!,2025-10-22 18:26:18.763129
+,0,,2025-10-22 18:26:22.517154

src/apps/chat/app.py

@@ -0,0 +1,57 @@
+import gradio as gr
+from src.rag.agent_chain import ExecutiveAgentChain 
+from src.utils.logging import get_logger, cached_log_handler
+
+logger = get_logger("chatbot_app")
+
+class ChatbotApplication:
+    def __init__(self, language: str = 'en') -> None:
+        self._language: str = language
+        self._agent_chain: ExecutiveAgentChain = ExecutiveAgentChain(language=self._language)
+        self._app = gr.Blocks()
+
+        chatbot: gr.Chatbot = gr.Chatbot(
+            value=[
+                gr.ChatMessage(
+                    role='assistant', 
+                    content=self._agent_chain.generate_greeting(),
+                )
+            ],
+            type='messages'
+        )
+        
+        with self._app:
+            with gr.Row():
+                with gr.Row(scale=3):
+                    gr.ChatInterface(
+                        chatbot=chatbot,
+                        fn=self._chat,
+                        title="Executive Education Adviser",
+                        type='messages',
+                    )
+                with gr.Row(scale=1):
+                    log_window = gr.Code(
+                        label='Console Log',
+                        lines=30,
+                        max_lines=30, 
+                        show_line_numbers=False,
+                    )
+
+                    log_timer = gr.Timer(value=0.5, active=True)
+                    log_timer.tick(fn=lambda: cached_log_handler.get_logs(), outputs=log_window)
+        
+
+    def _chat(self, message: str, history: list[dict]):
+        answers = []
+        try:
+            response = self._agent_chain.query(query=message)
+            logger.info("Recieved response from the agent, diplaying answer in the application")
+            answers.append(response)
+        except Exception as e:
+            logger.error(f"Error processing query: {e}")
+            answers.append("") 
+        return answers
+
+
+    def run(self):
+        self._app.launch(share=True)

src/apps/panel/app.py

@@ -0,0 +1,55 @@
+import gradio as gr
+import os
+import time
+
+# Simulated processor
+def process_documents(files, options):
+    results = []
+    for f, opt in zip(files, options):
+        results.append(f"{os.path.basename(f.name)} processed with {opt}")
+        time.sleep(0.5)
+    return results
+
+# Toggle a button state
+def toggle_option(option):
+    return not option
+
+# Build a file row with buttons
+def build_file_rows(files, states):
+    rows = []
+    with gr.Column():
+        for i, f in enumerate(files):
+            with gr.Row():
+                gr.Markdown(f"**{os.path.basename(f.name)}**")
+                chunk_btn = gr.Button("Chunk ✅" if states[i]["chunk"] else "Chunk ❌")
+                summary_btn = gr.Button("Summary ✅" if states[i]["summary"] else "Summary ❌")
+                embed_btn = gr.Button("Embed ✅" if states[i]["embed"] else "Embed ❌")
+                # Bind click events to toggle
+                chunk_btn.click(toggle_option, inputs=[states[i]["chunk"]], outputs=[states[i]["chunk"]])
+                summary_btn.click(toggle_option, inputs=[states[i]["summary"]], outputs=[states[i]["summary"]])
+                embed_btn.click(toggle_option, inputs=[states[i]["embed"]], outputs=[states[i]["embed"]])
+            rows.append(states[i])
+    return rows
+
+# Initialize UI
+with gr.Blocks() as demo:
+    uploaded_files = gr.File(file_count="multiple", label="Upload Files")
+    file_states = gr.State([])  # track toggle states per file
+    files_section = gr.Group(visible=False)
+    status_box = gr.Textbox(label="Status", interactive=False)
+
+    def display_files(files):
+        if not files:
+            return gr.update(visible=False), []
+        states = [{"chunk": True, "summary": False, "embed": True} for _ in files]
+        rows = build_file_rows(files, states)
+        return gr.update(visible=True), rows
+
+    uploaded_files.upload(
+        display_files,
+        inputs=[uploaded_files],
+        outputs=[files_section, file_states]
+    )
+
+demo.launch()
+

src/database/docker-compose.yml

@@ -0,0 +1,29 @@
+version: '3.4'
+
+services:
+  weaviate:
+    image: semitechnologies/weaviate:1.33.0
+    restart: on-failure:0
+    ports:
+      - "8080:8080"
+      - "50051:50051"
+    environment:
+      QUERY_DEFAULTS_LIMIT: 25
+      AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true'
+      PERSISTENCE_DATA_PATH: '/var/lib/weaviate'
+      ENABLE_API_BASED_MODULES: 'true'
+      ENABLE_MODULES: 'text2vec-transformers'
+      TRANSFORMERS_INFERENCE_API: 'http://t2v-transformers:8080'
+      CLUSTER_HOSTNAME: 'node1'
+    volumes:
+      - weaviate_data:/var/lib/weaviate
+
+  t2v-transformers:
+    image: semitechnologies/transformers-inference:sentence-transformers-all-MiniLM-L6-v2
+    restart: on-failure:0
+    ports:
+      - "8081:8080"
+
+volumes:
+  weaviate_data:
+

src/database/weavservice.py

@@ -0,0 +1,341 @@
+import weaviate as wvt
+import argparse, datetime, os
+
+from time import perf_counter, sleep
+from weaviate.classes.config import Configure, Property, DataType
+from weaviate.collections.classes.grpc import MetadataQuery
+from weaviate.collections.collection import Collection
+
+from src.utils.logging import get_logger
+from config import WeaviateConfiguration as wvtconf, AVAILABLE_LANGUAGES, HASH_FILE_PATH
+
+logger = get_logger("weaviate_service")
+
+_get_collection_name = lambda lang: f'{wvtconf.WEAVIATE_COLLECTION_BASENAME}_{lang}'
+_collection_names = [_get_collection_name(lang) for lang in AVAILABLE_LANGUAGES]
+
+
+class WeaviateService:
+    """
+    Provides an interface for interacting with the Weaviate vector database.
+    Handles initialization, data import, and hybrid queries.
+    """
+    def __init__(self) -> None:
+        """
+        Initialize the Weaviate service and establish a connection to the database.
+
+        Raises:
+            weaviate.exceptions.WeaviateConnectionError: If the connection fails.
+        """
+        self._connection_type = 'local' if wvtconf.is_local() else 'cloud'
+        self._client: wvt.WeaviateClient = self._connect_to_database()
+
+
+    def _connect_to_database(self):
+        retries = 0 
+        client: wvt.WeaviateClient
+        while retries < 3:
+            try:
+                if wvtconf.is_local():
+                    client = wvt.connect_to_local() 
+                else: 
+                    client = wvt.connect_to_weaviate_cloud(
+                        cluster_url=wvtconf.CLUSTER_URL,
+                        auth_credentials=wvtconf.WEAVIATE_API_KEY,
+                        headers={
+                            "X-HuggingFace-Api-Key": wvtconf.HUGGING_FACE_API_KEY,
+                        },
+                    ) 
+                break
+            except Exception as e:
+                logger.error(f"Failed to establish a conneciton with the {self._connection_type} weaviate database: {e}")
+                retries += 1 
+                sleep(1)
+        
+        if retries == 3:
+            logger.error(f"Failed to establish a connection with the {self._connection_type} weaviate after 3 retries, terminating the application")
+            exit()
+
+        logger.info(f"Successully connected to the {self._connection_type} weaviate database")
+        return client
+
+
+    def _with_connection(func):
+        def wrapper(self, *args, **kwargs):
+            try:
+                if not self._client.is_connected():
+                    self._client.connect()
+                    logger.info(f"Created a connection with the {self._connection_type} weaviate database")
+                result = func(self, *args, **kwargs)
+                self._client.close()
+                logger.info(f"Closed the connection with the {self._connection_type} weaviate database")
+                return result
+            except Exception as e:
+                logger.exception(f"Client failed to connect to the {self._connection_type} weaviate database: {e}")
+           
+        return wrapper
+
+
+    def _select_collection(self, lang: str) -> tuple[Collection, str]:
+        """
+        Select a language-specific collection as the active working collection.
+
+        Args:
+            lang (str): Acceptable language code.
+
+        Raises:
+            weaviate.exceptions.WeaviateConnectionError: If the specified language collection does not exist.
+        """
+        if lang not in AVAILABLE_LANGUAGES:
+            logger.error(f"No collection for language '{lang}' was found in the database")
+            return None, ''
+
+        collection_name = _get_collection_name(lang)
+        logger.info(f"Using collection {collection_name}")
+        return self._client.collections.use(collection_name), collection_name
+
+    @_with_connection
+    def batch_import(self, data_rows: list, lang: str) -> list:
+        """
+        Perform a batch import of multiple objects into the current collection.
+
+        Args:
+            data_rows (list): List of dictionaries representing the data rows to import.
+            lang (str, optional): Language collection to use. If not provided, uses the current one.
+
+        Returns:
+            list[dict]: List of failed imports with error details, if any.
+
+        Raises:
+            weaviate.exceptions.WeaviateConnectionError: If no active collection is available.
+        """
+        collection, collection_name = self._select_collection(lang)
+        if collection is None:
+            logger.error("No working collection selected upon starting batch import!")
+            return []
+
+        import_errors = []
+        logger.info(f"Initiating batch import for {len(data_rows)} data rows into collection {collection_name}")
+        with collection.batch.fixed_size(batch_size=100, concurrent_requests=2) as batch:
+            for idx, data_row in enumerate(data_rows):
+                try:
+                    batch.add_object(properties=data_row)
+                except Exception as e:
+                    import_errors.append({'index': idx, 'chunk_id': data_row['chunk_id'], 'error': str(e)})
+                    continue
+                
+                # Periodical checks for failed imports
+                if idx % 20 == 0 and idx > 0:
+                    if batch.number_errors > 0:
+                        logger.info(f"Amount of failed imports at index {idx}: {batch.number_errors}")
+                        last_failed_object = self._current_collection.batch.failed_objects[-1]
+                        logger.info(f"Last failure: {last_failed_object.message}")
+        
+        logger.info(f"Batch import finished for {collection_name}")
+        logger.info("Total import errors: {len(import_errors)}" if import_errors else "No errors catched during importing!")
+      
+        return import_errors
+   
+
+    @_with_connection 
+    def query(self, query: str, lang: str, query_properties: list[str] = None, limit: int = 5) -> dict:
+        """
+        Execute a hybrid semantic and keyword query against the active collection.
+
+        Args:
+            query (str): The query string.
+            query_properties (list[str], optional): List of properties to query against.
+            limit (int, optional): Maximum number of results to return. Defaults to 5.
+            distance (float, optional): Distance threshold for the query. Defaults to 0.25.
+            lang (str, optional): Language collection to use. If not provided, uses the current one.
+
+        Returns:
+            tuple: A tuple containing the query response and elapsed time.
+
+        Raises:
+            weaviate.exceptions.WeaviateConnectionError: If no active collection is available.
+        """ 
+        collection, collection_name = self._select_collection(lang)
+        if collection is None:
+            logger.error("No working collection selected upon starting of the querying!")
+            return [], 0
+        
+        logger.info(f"Querying collection {collection_name}")
+        query_start_time = perf_counter()
+        resp = collection.query.hybrid(
+            query=query,
+            query_properties=query_properties,
+            limit=limit,
+            return_metadata=MetadataQuery.full()
+        )
+        elapsed = perf_counter() - query_start_time
+        logger.info(f"Querying retrieved {len(resp.objects)} objects in {elapsed:3.2f} seconds")
+
+        return (resp, elapsed)
+
+
+    @_with_connection
+    def _create_collections(self):
+        """
+        Create and initialize language-specific collections in the Weaviate database.
+
+        Each collection includes vector and generative configurations.
+
+        Raises:
+            weaviate.exceptions.WeaviateConnectionError: If the database connection fails.
+        """ 
+        logger.info('Attempting collections creation for the database...')
+        
+        vector_config = (Configure.Vectors.text2vec_transformers() if wvtconf.is_local() 
+                         else Configure.Vectors.text2vec_huggingface(
+                                name='hsg_rag_embeddings',
+                                source_properties=['body'],
+                                model="sentence-transformers/all-MiniLM-L6-v2",
+                             ))
+        successful_creations = 0
+        for collection_name in _collection_names:
+            try:
+                self._client.collections.create(
+                    name=collection_name,
+                    properties=[
+                        Property(name='body', data_type=DataType.TEXT),
+                        Property(name='chunk_id', data_type=DataType.TEXT),
+                        Property(name='document_id', data_type=DataType.TEXT),
+                        Property(name='programs', data_type=DataType.TEXT_ARRAY),
+                        Property(name='source', data_type=DataType.TEXT),
+                        Property(name='date', data_type=DataType.DATE)
+                    ],
+                    vector_config=vector_config)
+                logger.info(f"Created collection {collection_name}")
+                successful_creations += 1
+            except Exception as e:
+                logger.error(f"Failed to initialize collection '{collection_name}': {e}")
+        if successful_creations == len(_collection_names):
+            logger.info('All collections were successfully instantiated in the database')
+
+
+    @_with_connection
+    def _delete_collections(self):
+        """
+        Delete all existing collections from the Weaviate database.
+
+        Raises:
+            weaviate.exceptions.WeaviateConnectionError: If the connection to the database fails.
+        """        
+        logger.info("Initiating the deletion of stored collections.")
+        for collection_name in _collection_names:
+            if self._client.collections.exists(collection_name):
+                self._client.collections.delete(collection_name)
+                logger.info(f"Deleted collection {collection_name}")
+            else:
+                logger.warning(f"Cannot delete collection {collection_name}: collection does not exist!")
+        
+        logger.info("Finished the deletion of stored colections")
+   
+        if os.path.exists(HASH_FILE_PATH):
+            logger.info(f"Hashtable found on path {HASH_FILE_PATH}; deleting hash file...")
+            os.remove(HASH_FILE_PATH)
+            logger.info("Hash file deleted successfully")
+
+    @_with_connection
+    def _create_backup(self) -> None:
+        """
+        Creates a backup instance from current weaviate database state and uploads it to selected backend storage service.
+        """        
+        backup_id = datetime.datetime.now().strftime("%Y%m%d%H%M%S%f")
+        logger.info(f"Initiating backup creation for the {self._connection_type} weaviate database")
+        try:
+            result = self._client.backup.create(
+                backup_id=f"hsg_wvtdb_backup_{backup_id}",
+                backend=wvtconf.WEAVIATE_BACKUP_BACKEND,
+                include_collections=_collection_names,
+                wait_for_completion=True
+            )
+            logger.info(f"Creation of backup '{backup_id}' finished with response: {result}")
+        except Exception as e:
+            logger.error(f"Failed to create a backup due to an unexpected error: {e.message}")
+            raise e
+
+
+    @_with_connection
+    def _restore_backup(self, backup_id: str):
+        """
+        Restores the state of the database from the provided backup.
+        
+        Args:
+            backup_id(str): ID of the backup from which the database state should be restored.
+        """        
+        logger.info("Initiating restoration process from the backup '{backup_id}' for the {self._connection_type} weaviate database")
+        try:
+            result = self._client.backup.restore(
+                backup_id=backup_id,
+                backend=wvtconf.WEAVIATE_BACKUP_BACKEND,
+                include_collections=_collection_names,
+                wait_for_completion=True,
+            )
+            logger.info(f"Restored backup '{backup_id}' with response: {result}")
+        except Exception as e:
+            logger.error(f"Failed to restore a backup due to an unexpected error: {e.message}")
+            raise e
+
+
+    @_with_connection
+    def _checkhealth(self):
+        """
+        Check the connectivity and health status of the Weaviate database and its collections.
+
+        Prints the connection status and verifies the existence of collections for each supported language.
+        """
+        connection_exists = self._client.is_connected()
+        logger.info(f"Checking the connection to the {self._connection_type} weaviate database: {'OK!' if connection_exists else 'ERROR'}")
+        if not connection_exists: return 
+
+        metainfo = self._client.get_meta()
+        modules_str = ', '.join(metainfo['modules'])
+        modules_str = modules_str[:30] + '...' if len(modules_str) > 30 else modules_str
+        if wvtconf.is_local():
+            logger.info(f"Cluster metadata: HOSTNAME={metainfo['hostname']}, VERSION={metainfo['version']}, MODULES={modules_str}")
+
+        for collection_name in _collection_names:
+            logger.info(f"Checking the existence of collection {collection_name}: "
+                  f"{'OK!' if self._client.collections.exists(collection_name) else 'ERROR' }")
+
+
+def parse_arguments():
+    """
+    Parse command-line arguments for managing Weaviate collections.
+
+    Returns:
+        argparse.Namespace: Parsed command-line arguments.
+    """
+    parser = argparse.ArgumentParser()
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument('-dc', "--delete_collections", action='store_true', help='deletes all collections from the database')
+    group.add_argument('-cc', "--create_collections", action='store_true', help='initializes the collections for different language contents separately')
+    group.add_argument('-rc', "--redo_collections", action='store_true', help='deletes and creates the collections anew')
+
+    group.add_argument('-ch', "--checkhealth", action='store_true', help='checks the connection to the database, existense of content collections...')
+    group.add_argument('-cb', "--create_backup", action='store_true', help='creates a backup of the current state of the database')
+    group.add_argument('-rb', "--restore_backup", type=str, help='restores the state of the database from the provided backup_id')
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    args = parse_arguments()
+    service = WeaviateService()
+
+    if args.create_backup:
+       service._create_backup()
+
+    if args.restore_backup:
+        service._restore_backup(args.restore_backup)
+
+    if any([args.delete_collections, args.redo_collections]):
+        service._delete_collections()
+
+    if any([args.create_collections, args.redo_collections]):
+        service._create_collections()
+    
+    if any([args.checkhealth, args.create_collections, args.redo_collections]):
+        service._checkhealth()
+    

src/pipeline/pipeline.py

@@ -0,0 +1,209 @@
+import json, os
+
+from pathlib import Path
+from src.utils.logging import get_logger
+from src.processing.processor import DataProcessor, ProcessingResult, ProcessingStatus, WebsiteProcessor
+from src.database.weavservice import WeaviateService
+
+from config import AVAILABLE_LANGUAGES, HASH_FILE_PATH, DOCUMENTS_PATH
+
+pipelogger = get_logger("pipeline_module")
+implogger  = get_logger("import_pipeline")
+
+
+def _get_all_sources(sources) -> list[str]:
+    sources.remove('all')
+    pipelogger.info(f"Getting all sources from the soruce directory at {DOCUMENTS_PATH}...")
+    for source in os.listdir(DOCUMENTS_PATH):
+        if source in sources: continue
+        if source.endswith('.pdf'):
+            sources.append(os.path.join(DOCUMENTS_PATH, source))
+    pipelogger.info(f"Loaded {len(sources)} sources from the source directory")
+    return sources
+
+
+def _import_hashtables() -> dict:
+    """
+    Import deduplication hashtables from the JSON file.
+
+    Returns:
+        dict: Hashtable data containing document and chunk IDs.
+    """
+    hashtables = dict()
+    
+    with open(HASH_FILE_PATH, 'a+') as f:
+        try:
+            f.seek(0)
+            pipelogger.info(f"Loading deduplication hashtable from file {HASH_FILE_PATH}")
+            hashtables = json.load(f)
+            pipelogger.info(f"Import pipeline loaded deduplication hashtable with {len(hashtables['documents'])} sources and {len(hashtables['chunks'])} chunks")
+        except json.JSONDecodeError as e:
+            pipelogger.warning(f"Failed to decode the hash file {os.path.basename(HASH_FILE_PATH)}: {e}; new hashtable will be created")
+            hashtables['documents'] = []
+            hashtables['chunks'] = []
+    return hashtables
+
+
+def _export_hashtables(hashtables: dict):
+    """
+    Export hashtable data to the JSON file.
+
+    Args:
+        hashtables (dict): Hashtable dictionary containing documents and chunks.
+    """
+    with open(HASH_FILE_PATH, 'w+') as f:
+        json.dump(hashtables, f)
+        pipelogger.info("Saved successfully imported chunk IDs in the hashtables")
+
+
+class ImportPipeline:
+    """
+    Main pipeline class responsible for importing website and local documents
+    into the database with deduplication and language-based organization.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the import pipeline with processors and hashtable data."""
+        self._hashtables   = _import_hashtables()
+        self._webprocessor = WebsiteProcessor()
+        self._processor    = DataProcessor()
+        self._wvtserv      = WeaviateService()
+
+
+    def scrape_website(self):
+        """
+        Scrape program pages from the website, process and deduplicate them,
+        and import unique chunks into the database.
+        """
+        unique_chunks = {lang: [] for lang in AVAILABLE_LANGUAGES}
+        for result in self._webprocessor.process():
+            chunks = self._deduplicate(result)
+            unique_chunks[result.language].extend(chunks)
+
+        if not unique_chunks:
+            implogger.warning("Information provided by the HSG website does not contain any unique information. Terminating the pipeline without importing")
+            return 
+        
+        self._import_to_database(unique_chunks)
+        _export_hashtables(self._hashtables)
+
+
+    def import_many_documents(self, sources: list[Path | str]):
+        """
+        Import multiple documents by processing, deduplicating, and inserting
+        unique chunks into the database.
+
+        Args:
+            sources (list[Path | str]): List of file paths or URLs to process.
+        """
+        if 'all' in sources:
+            implogger.info("Import list contains the 'all', all sources will be imported...")
+            sources = _get_all_sources(sources)
+        
+        if not sources:
+            implogger.warning("Import list does not contain any sources, aborting the import pipeline!")
+            return
+
+        if len(sources) > 1:
+            implogger.info(f"Initiating the import pipeline for multiple sources: {', '.join(sources)}")
+
+        unique_chunks = {lang: [] for lang in AVAILABLE_LANGUAGES}
+        for source in sources:
+            chunks, lang = self._process_source(source)
+            if chunks:
+                unique_chunks[lang].extend(chunks)
+        
+        if not unique_chunks:
+            implogger.warning(f"File(s) provided for the insertion do not contain any unique information. Terminating the pipeline without importing")
+            return 
+
+        self._import_to_database(unique_chunks)
+        _export_hashtables(self._hashtables)
+
+
+    def import_document(self, source: Path | str):
+        """
+        Import a single document into the database.
+
+        Args:
+            source (Path | str): Path to the document to process and import.
+        """
+        self.import_many_documents([source])
+     
+    
+    def _import_to_database(self, unique_chunks):
+        """
+        Import the processed unique chunks into the Weaviate database.
+
+        Args:
+            unique_chunks (dict): Dictionary mapping languages to lists of chunks.
+        """
+        for lang, chunks in unique_chunks.items():
+            if not chunks: 
+                continue
+
+            failures = self._wvtserv.batch_import(data_rows=chunks, lang=lang)
+            for failure in failures:
+                chunk_id = failure['chunk_id']
+                if chunk_id in self._hashtables['chunks']:
+                    self._hashtables['chunks'].remove(chunk_id)
+
+
+    def _process_source(self, source: Path | str) -> tuple[list, str]:
+        """
+        Process a single document source, deduplicate its chunks, and
+        determine its language.
+
+        Args:
+            source (Path | str): Path to the document to process.
+
+        Returns:
+            tuple[list, str]: List of unique chunks and detected language.
+        """
+        result: ProcessingResult = self._processor.process(source)
+
+        if not result.status == ProcessingStatus.SUCCESS:
+            implogger.error(f"Failed to process document {source}: {result.status}")
+            return [], ''
+        
+        unique_chunks = self._deduplicate(result)
+        return unique_chunks, result.language
+
+
+    def _deduplicate(self, result: ProcessingResult):
+        """
+        Remove duplicate chunks and documents based on previously processed hashes.
+
+        Args:
+            result (ProcessingResult): The processing result containing document chunks.
+
+        Returns:
+            list[dict]: List of unique chunk dictionaries.
+        """
+        d_id = result.document_id
+        unique_chunks = []
+
+        implogger.info(f"Analyzing document with ID {d_id} for duplicated contents")
+        if d_id in self._hashtables['documents']:
+            implogger.warning(f"Document with ID {d_id} is a duplicate!")
+            return unique_chunks
+        
+        for chunk in result.chunks:
+            c_id = chunk['chunk_id']
+            if c_id in self._hashtables['chunks']:
+                continue 
+
+            self._hashtables['chunks'].append(c_id)
+            unique_chunks.append(chunk)
+
+        if not unique_chunks:
+            self._hashtables['documents'].append(d_id)
+        
+        implogger.info(f"Found {len(unique_chunks)} unique chunks out ouf {len(result.chunks)} collected chunks")
+        return unique_chunks
+
+
+if __name__ == "__main__":
+    pipeline = ImportPipeline()
+    #pipeline.import_many_documents(['data/hsg.pdf', 'data/emba_X5.pdf'])
+    pipeline.scrape_website()

src/processing/processor.py

@@ -0,0 +1,268 @@
+import os, re, hashlib, time
+
+from enum import Enum
+from datetime import datetime, timezone
+from pathlib import Path
+from dataclasses import dataclass
+from transformers import AutoTokenizer
+from docling_core.transforms.chunker.tokenizer.huggingface import HuggingFaceTokenizer
+from docling.document_converter import DocumentConverter
+from docling.chunking import HybridChunker
+from docling_core.types.doc.document import DoclingDocument
+
+from src.utils.lang import detect_language
+from src.utils.logging import get_logger
+from config import BASE_URL, CHUNK_MAX_TOKENS
+
+weblogger  = get_logger("website_processor")
+datalogger = get_logger("data_processor")
+
+_TRANSFORMERS_TOKENIZER = AutoTokenizer.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")
+_EN_URL_PATTERN = r'\[EN\]\((https://emba\.unisg\.ch/en/[^\s)]+)\)'
+_PROGRAM_URL_PATTERN = r'https://emba\.unisg\.ch/(?:programm[^\s)]+|en/embax)'
+
+
+def _get_hash(text: str) -> str:
+    """Generate an MD5 hash for the given text."""
+    return hashlib.md5(text.strip().encode("utf-8")).hexdigest()
+
+
+def _get_en_version(text: str):
+    """Extract the English version URL from the given text, if available."""
+    result = re.search(_EN_URL_PATTERN, text)
+    if result:
+        return result.group(1)
+    return ""
+
+
+def _get_program_urls(text: str):
+    """Find all program URLs in the given text."""
+    return re.findall(_PROGRAM_URL_PATTERN, text)
+
+
+def _detect_programs(text: str):
+    """
+    Identify MBA program names mentioned in the given text.
+
+    Args:
+        text (str): The text to search for program mentions.
+
+    Returns:
+        list[str]: List of detected program identifiers.
+    """
+    programms = []
+    lc_text = text.lower()
+    found = lambda txt: txt in lc_text
+    
+    if found('emba') or found('executive mba'):
+        programms.append('emba')
+
+    if found('iemba') or found('international emba') or found('international executive mba'):
+        programms.append('iemba')
+
+    if found('emba x'):
+        programms.append('emba_x')
+
+    return programms
+
+
+class ProcessingStatus(Enum):
+    NOT_FOUND         = 1
+    SUCCESS           = 2
+    FAILURE           = 3
+    INCORRECT_FORMAT  = 5
+
+    
+@dataclass
+class _ChunkMetadata:
+    programs: str 
+    date: str 
+    document_id: str 
+    language: str
+    source: str = None
+
+
+@dataclass
+class ProcessingResult:
+    status: ProcessingStatus = ProcessingStatus.SUCCESS
+    chunks: list = None
+    document_id: str = None
+    language: str = None
+        
+
+class _ProcessorBase:
+    def __init__(self):
+        """
+        Initialize the Processor with converter, chunker, and hashtable.
+
+        Args:
+            config (dict, optional): Configuration dictionary for processing options.
+        """
+        self._converter = DocumentConverter()
+        self._chunker = HybridChunker(
+            tokenizer=HuggingFaceTokenizer(
+                tokenizer=_TRANSFORMERS_TOKENIZER,
+                max_tokens=CHUNK_MAX_TOKENS
+            ), 
+            max_tokens=CHUNK_MAX_TOKENS, 
+            merge_peers=True
+        )
+    
+    def process(self):
+        """Abstract method to be implemented by subclasses."""
+        raise NotImplementedError("This method is not implemented in ProcessorBase")
+
+
+    def _collect_metadata(self, text: str) -> _ChunkMetadata:
+        """Collect metadata such as programs, date, language, and document hash."""
+        return _ChunkMetadata(
+                programs=_detect_programs(text),
+                date=datetime.now().replace(tzinfo=timezone.utc),
+                language=detect_language(text),
+                document_id=_get_hash(text))
+
+
+    def _collect_chunks(self, document: DoclingDocument, metadata: _ChunkMetadata) -> list:
+        """
+        Collect text chunks from a document and prepare them with metadata.
+
+        Args:
+            document (DoclingDocument): The converted document object.
+            metadata (_ChunkMetadata): Metadata containing program, source, and date information.
+
+        Returns:
+            list[dict]: List of chunk dictionaries containing text and metadata.
+        """
+        chunks = [self._chunker.contextualize(chunk=c) for c in self._chunker.chunk(document)]
+        prepared_chunks = []
+        for chunk in chunks:
+            c_hash = _get_hash(chunk)            
+            prepared_chunks.append({
+                'body': chunk,
+                'chunk_id': c_hash,
+                'document_id': metadata.document_id,
+                'programs': metadata.programs,
+                'date': metadata.date,
+                'source': metadata.source
+            })
+        return prepared_chunks
+
+
+class WebsiteProcessor(_ProcessorBase):
+    def __init__(self):
+        """Initialize the WebsiteProcessor with base processing capabilities."""
+        super().__init__()
+
+
+    def process(self) -> list[ProcessingResult]:
+        """
+        Scrape and process program pages from the HSG website.
+
+        Returns:
+            list[ProcessingResult]: A list of processing results for each processed URL.
+        """
+        weblogger.info("Initiating scraping and processing of the HSG program pages.")
+        urls = [BASE_URL]
+        results = []
+        while urls:
+            url = urls.pop()
+            result, text = self._process_url(url)
+
+            if result.status != ProcessingStatus.SUCCESS:
+                weblogger.warning(f"Failed to process URLs {url}.")
+                continue 
+
+            if url == BASE_URL:
+                program_urls = _get_program_urls(text)
+                urls.extend(program_urls)
+                weblogger.info(f"Found following program URLs: {', '.join(program_urls)}.")
+
+            if '/en/' not in url:
+                en_url = _get_en_version(text)
+                urls.append(en_url)
+                weblogger.info(f"Added an english version of the URL {en_url} to the processing list")
+            
+            results.append(result)
+            time.sleep(2)
+
+        weblogger.info(f"Successfully processed {len(results)} URLs.")
+        return results 
+    
+
+    def _process_url(self, url: str) -> tuple[ProcessingResult, str]:
+        """
+        Process the content of a single URL, converting it into chunks with metadata.
+
+        Args:
+            url (str): The URL of the webpage to process.
+
+        Returns:
+            tuple[ProcessingResult, str]: The processing result and the extracted text.
+        """
+        weblogger.info(f"Initiating processing pipeline for url {url}")
+        try:
+            document = self._converter.convert(url).document
+        except Exception as e:
+            weblogger.error(f"Failed to load the contents of the url page {url}: {e}")
+            return ProcessingResult(status=ProcessingStatus.FAILURE)
+        
+        text = document.export_to_markdown()
+        metadata = self._collect_metadata(text)
+        metadata.source = url
+        collected_chunks = self._collect_chunks(document, metadata)
+        del collected_chunks[0]
+        weblogger.info(f"Successfully collected {len(collected_chunks)} chunks from {url}")
+        
+        return ProcessingResult(chunks=collected_chunks, language=metadata.language, document_id=metadata.document_id), text
+
+
+class DataProcessor(_ProcessorBase):
+    """
+    Handles document processing, including conversion, chunking, language detection,
+    and hash-based deduplication.
+    """
+
+    def process_many_documents(self, sources: list[Path | str]) -> list[ProcessingResult]:
+        """
+        Process a list of document sources sequentially.
+
+        Args:
+            sources (list[Path | str]): List of file paths or URLs to process.
+
+        Returns:
+            list[ProcessingResult]: List of results for each processed document.
+        """
+        return [self.process(source) for source in sources]
+
+
+    def process(self, source: Path | str) -> ProcessingResult:
+        """
+        Process a single document source, converting it to text, chunking, and hashing.
+
+        Args:
+            source (Path | str): Path to the document to process.
+
+        Returns:
+            ProcessingResult: The result of the processing operation, including chunks and language.
+        """
+        if not os.path.exists(source) or not os.path.isfile(source):
+            datalogger.error(f"Failed to initiate processing pipeline for source {source}: file does not exist")
+            return ProcessingResult(status=ProcessingStatus.NOT_FOUND)
+        
+        datalogger.info(f"Initiating processing pipeline for source {source}")
+        document = self._converter.convert(source).document
+        metadata = self._collect_metadata(document.export_to_markdown())
+        metadata.source = os.path.basename(source)
+        collected_chunks = self._collect_chunks(document, metadata)
+        datalogger.info(f"Successfully collected {len(collected_chunks)} chunks from {source}")
+ 
+        return ProcessingResult(chunks=collected_chunks, language=metadata.language, document_id=metadata.document_id)
+
+
+if __name__ == "__main__":
+    processor = WebsiteProcessor()
+    results = processor.process()
+
+    for result in results:
+        for chunk in result.chunks:
+            print(chunk['body'], end='\n\n')

src/processing/scraping.py

@@ -0,0 +1,382 @@
+"""
+Data processor for cleaning and structuring scraped program data.
+"""
+import json
+import logging
+import os
+from typing import Dict, List, Any
+
+import pandas as pd
+
+from config import RAW_DATA_PATH, PROCESSED_DATA_PATH
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+class ScrapingProcessor:
+    """Processor for cleaning and structuring scraped program data."""
+
+    def __init__(self, input_path: str = RAW_DATA_PATH, output_path: str = PROCESSED_DATA_PATH, manual_data_path: str = None):
+        """
+        Initialize the data processor.
+
+        Args:
+            input_path: Path to the raw data file.
+            output_path: Path to save the processed data.
+            manual_data_path: Path to the manual data file (optional).
+        """
+        self.input_path = input_path
+        self.output_path = output_path
+        self.manual_data_path = manual_data_path or os.path.join(os.path.dirname(input_path), "manual_data.json")
+
+    def load_data(self) -> List[Dict[str, Any]]:
+        """
+        Load raw data from the input file.
+
+        Returns:
+            A list of dictionaries containing program data.
+        """
+        try:
+            with open(self.input_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+            
+            # If raw data is empty, try to load manual data
+            if not data and os.path.exists(self.manual_data_path):
+                logger.info(f"Raw data is empty, loading manual data from {self.manual_data_path}")
+                with open(self.manual_data_path, "r", encoding="utf-8") as f:
+                    data = json.load(f)
+                logger.info(f"Loaded {len(data)} programs from {self.manual_data_path}")
+            else:
+                logger.info(f"Loaded {len(data)} programs from {self.input_path}")
+            
+            return data
+        except Exception as e:
+            logger.error(f"Error loading data: {e}")
+            
+            # If there was an error, try to load manual data as a fallback
+            try:
+                if os.path.exists(self.manual_data_path):
+                    logger.info(f"Attempting to load manual data from {self.manual_data_path}")
+                    with open(self.manual_data_path, "r", encoding="utf-8") as f:
+                        data = json.load(f)
+                    logger.info(f"Loaded {len(data)} programs from {self.manual_data_path}")
+                    return data
+            except Exception as e2:
+                logger.error(f"Error loading manual data: {e2}")
+            
+            return []
+
+    def save_data(self, data: List[Dict[str, Any]]) -> None:
+        """
+        Save processed data to the output file.
+
+        Args:
+            data: The processed data to save.
+        """
+        try:
+            # Ensure the directory exists
+            os.makedirs(os.path.dirname(self.output_path), exist_ok=True)
+            
+            with open(self.output_path, "w", encoding="utf-8") as f:
+                json.dump(data, f, ensure_ascii=False, indent=2)
+            
+            logger.info(f"Saved {len(data)} processed programs to {self.output_path}")
+        except Exception as e:
+            logger.error(f"Error saving data to {self.output_path}: {e}")
+
+    def clean_text(self, text: str) -> str:
+        """
+        Clean text by removing extra whitespace and normalizing.
+
+        Args:
+            text: The text to clean.
+
+        Returns:
+            The cleaned text.
+        """
+        if not text or not isinstance(text, str):
+            return ""
+        
+        # Replace multiple whitespace with a single space
+        cleaned = " ".join(text.split())
+        return cleaned.strip()
+
+    def clean_list(self, items: List[str]) -> List[str]:
+        """
+        Clean a list of text items.
+
+        Args:
+            items: The list of text items to clean.
+
+        Returns:
+            The cleaned list of text items.
+        """
+        if not items or not isinstance(items, list):
+            return []
+        
+        cleaned_items = [self.clean_text(item) for item in items if item]
+        return [item for item in cleaned_items if item]  # Remove empty items
+
+    def normalize_costs(self, cost_text: str) -> Dict[str, Any]:
+        """
+        Normalize cost information.
+
+        Args:
+            cost_text: The cost text to normalize.
+
+        Returns:
+            A dictionary with normalized cost information.
+        """
+        if not cost_text or cost_text == "Not specified":
+            return {"amount": None, "currency": None, "original_text": cost_text}
+        
+        # Try to extract currency and amount
+        import re
+        
+        # Look for common currency patterns
+        currency_patterns = {
+            "CHF": r"CHF\s*([\d\',\.]+)",
+            "EUR": r"(?:€|EUR)\s*([\d\',\.]+)",
+            "USD": r"(?:\$|USD)\s*([\d\',\.]+)",
+        }
+        
+        for currency, pattern in currency_patterns.items():
+            match = re.search(pattern, cost_text, re.IGNORECASE)
+            if match:
+                # Extract the amount and clean it
+                amount_str = match.group(1)
+                # Remove thousands separators and convert to float
+                amount_str = amount_str.replace(",", "").replace("'", "")
+                try:
+                    amount = float(amount_str)
+                    return {
+                        "amount": amount,
+                        "currency": currency,
+                        "original_text": cost_text,
+                    }
+                except ValueError:
+                    pass
+        
+        # If no pattern matched, return the original text
+        return {"amount": None, "currency": None, "original_text": cost_text}
+
+    def normalize_duration(self, duration_text: str) -> Dict[str, Any]:
+        """
+        Normalize duration information.
+
+        Args:
+            duration_text: The duration text to normalize.
+
+        Returns:
+            A dictionary with normalized duration information.
+        """
+        if not duration_text or duration_text == "Not specified":
+            return {"months": None, "original_text": duration_text}
+        
+        # Try to extract duration in months
+        import re
+        
+        # Look for common duration patterns
+        month_patterns = [
+            r"(\d+)\s*months?",
+            r"(\d+)\s*month program",
+        ]
+        
+        year_patterns = [
+            r"(\d+)\s*years?",
+            r"(\d+)\s*year program",
+            r"(\d+)\s*-\s*year",
+        ]
+        
+        # Check for months first
+        for pattern in month_patterns:
+            match = re.search(pattern, duration_text, re.IGNORECASE)
+            if match:
+                try:
+                    months = int(match.group(1))
+                    return {
+                        "months": months,
+                        "original_text": duration_text,
+                    }
+                except ValueError:
+                    pass
+        
+        # Check for years and convert to months
+        for pattern in year_patterns:
+            match = re.search(pattern, duration_text, re.IGNORECASE)
+            if match:
+                try:
+                    years = int(match.group(1))
+                    months = years * 12
+                    return {
+                        "months": months,
+                        "original_text": duration_text,
+                    }
+                except ValueError:
+                    pass
+        
+        # If no pattern matched, return the original text
+        return {"months": None, "original_text": duration_text}
+
+    def process_program(self, program: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Process a single program's data.
+
+        Args:
+            program: The program data to process.
+
+        Returns:
+            The processed program data.
+        """
+        processed = {}
+        
+        # Copy basic fields
+        processed["url"] = program.get("url", "")
+        processed["name"] = self.clean_text(program.get("name", "Unknown Program"))
+        processed["description"] = self.clean_text(program.get("description", ""))
+        
+        # Process duration
+        processed["duration"] = self.normalize_duration(program.get("duration", "Not specified"))
+        
+        # Process costs
+        processed["costs"] = self.normalize_costs(program.get("costs", "Not specified"))
+        
+        # Process lists
+        processed["curriculum"] = self.clean_list(program.get("curriculum", []))
+        processed["admission_requirements"] = self.clean_list(program.get("admission_requirements", []))
+        
+        # Process faculty
+        faculty = program.get("faculty", [])
+        processed_faculty = []
+        for member in faculty:
+            if isinstance(member, dict):
+                processed_faculty.append({
+                    "name": self.clean_text(member.get("name", "")),
+                    "title": self.clean_text(member.get("title", "")),
+                })
+        processed["faculty"] = processed_faculty
+        
+        # Process other fields
+        processed["schedules"] = self.clean_text(program.get("schedules", "Not specified"))
+        processed["deadlines"] = self.clean_text(program.get("deadlines", "Not specified"))
+        processed["language"] = self.clean_text(program.get("language", "Not specified"))
+        processed["location"] = self.clean_text(program.get("location", "Not specified"))
+        
+        # Add metadata
+        processed["program_id"] = f"prog_{hash(processed['url']) % 10000:04d}"
+        
+        return processed
+
+    def process_data(self) -> List[Dict[str, Any]]:
+        """
+        Process all program data.
+
+        Returns:
+            A list of processed program data.
+        """
+        raw_data = self.load_data()
+        processed_data = []
+        
+        for program in raw_data:
+            try:
+                processed_program = self.process_program(program)
+                processed_data.append(processed_program)
+            except Exception as e:
+                logger.error(f"Error processing program {program.get('name', 'Unknown')}: {e}")
+        
+        logger.info(f"Processed {len(processed_data)} programs")
+        return processed_data
+
+    def run(self) -> None:
+        """Run the data processor."""
+        logger.info("Starting data processing...")
+        processed_data = self.process_data()
+        self.save_data(processed_data)
+        logger.info("Data processing completed")
+
+    def generate_stats(self) -> Dict[str, Any]:
+        """
+        Generate statistics about the processed data.
+
+        Returns:
+            A dictionary containing statistics.
+        """
+        try:
+            processed_data = self.load_data()
+            if not processed_data:
+                return {"error": "No processed data available"}
+            
+            # Convert to DataFrame for easier analysis
+            df = pd.DataFrame(processed_data)
+            
+            # Basic stats
+            stats = {
+                "total_programs": len(df),
+                "languages": {},
+                "locations": {},
+                "duration_months": {
+                    "min": None,
+                    "max": None,
+                    "avg": None,
+                },
+                "costs": {
+                    "min": {},
+                    "max": {},
+                    "avg": {},
+                },
+            }
+            
+            # Language stats
+            if "language" in df.columns:
+                language_counts = df["language"].value_counts().to_dict()
+                stats["languages"] = language_counts
+            
+            # Location stats
+            if "location" in df.columns:
+                location_counts = df["location"].value_counts().to_dict()
+                stats["locations"] = location_counts
+            
+            # Duration stats
+            if "duration" in df.columns:
+                # Extract months from duration dictionaries
+                months = [d.get("months") for d in df["duration"] if d.get("months") is not None]
+                if months:
+                    stats["duration_months"]["min"] = min(months)
+                    stats["duration_months"]["max"] = max(months)
+                    stats["duration_months"]["avg"] = sum(months) / len(months)
+            
+            # Cost stats by currency
+            if "costs" in df.columns:
+                # Group costs by currency
+                currencies = {}
+                for cost in df["costs"]:
+                    currency = cost.get("currency")
+                    amount = cost.get("amount")
+                    if currency and amount is not None:
+                        if currency not in currencies:
+                            currencies[currency] = []
+                        currencies[currency].append(amount)
+                
+                # Calculate stats for each currency
+                for currency, amounts in currencies.items():
+                    if amounts:
+                        stats["costs"]["min"][currency] = min(amounts)
+                        stats["costs"]["max"][currency] = max(amounts)
+                        stats["costs"]["avg"][currency] = sum(amounts) / len(amounts)
+            
+            return stats
+        except Exception as e:
+            logger.error(f"Error generating stats: {e}")
+            return {"error": str(e)}
+
+
+if __name__ == "__main__":
+    processor = DataProcessor()
+    processor.run()
+    stats = processor.generate_stats()
+    print(json.dumps(stats, indent=2))

src/rag/agent_chain.py

@@ -0,0 +1,197 @@
+from langchain.tools import tool
+from langchain.agents import create_agent
+from langchain_core.messages import (
+    HumanMessage, 
+    AIMessage, 
+    SystemMessage, 
+)
+from langchain.agents.middleware import (
+    SummarizationMiddleware,
+    ModelFallbackMiddleware,
+)
+
+from langgraph.checkpoint.memory import InMemorySaver
+
+from src.database.weavservice import WeaviateService
+
+from src.rag.utilclasses import *
+from src.rag.middleware import AgentChainMiddleware as chainmdw
+from src.rag.prompts import PromptConfigurator as promptconf
+from src.rag.models import ModelConfigurator as modelconf
+
+from src.utils.logging import get_logger 
+from config import TOP_K_RETRIEVAL
+
+chain_logger = get_logger('agent_chain')
+
+class ExecutiveAgentChain:
+    def __init__(self, language: str = 'en') -> None:
+        self._language = language 
+        self._dbservice = WeaviateService()
+        self._agents, self._config = self._init_agents()
+        chain_logger.info(f"Initalized new Agent Chain for language '{language}'")
+
+
+    def _retrieve_context(self, query: str, language: str = None):
+        """
+        Send the query to the vector database to retrieve additional information about the program.
+
+        Args:
+            query: Keywords depicting information you want to retrieve in the primary language. 
+            language: Optional parameter (either 'en' for English language or 'de' for German language). This parameter selects the language of the database to query from. The input query must be written in the same language as the selected language. Use this parameter only if there's not enough information in your main language.
+        """
+        print('called retrieve_context')
+        lang = language or self._language
+        try:
+            response, _ = self._dbservice.query(
+                query=query, 
+                lang=lang, 
+                limit=TOP_K_RETRIEVAL,
+            )
+            serialized = '\n\n'.join(
+                ("Source: {source}\nPrograms: {programs}\nContent: {content}".format(
+                    source=doc.properties.get('source', 'unknown'),
+                    programs=', '.join(doc.properties.get('programs', 'unknown')),
+                    content=doc.properties.get('body', ''))) 
+                for doc in response.objects
+            )
+            return serialized
+        except Exception as _:
+            return ''
+   
+
+    def _call_emba_agent(self, query: str) -> str:
+        """
+        Invokes the EMBA support agent to retrieve more detailed information about the EMBA program.
+        
+        Args:
+            query: Query to the EMBA support agent. Provide collected user data in the query if possible.
+        """
+        response = self._query(agent=self._agents['emba'], messages=[HumanMessage(query)])
+        return response
+    
+
+    def _call_iemba_agent(self, query: str) -> str:
+        """
+        Invokes the IEMBA support agent to retrieve more detailed information about the IEMBA program.
+        
+        Args:
+            query: Query to the IEMBA support agent. Provide collected user data in the query if possible.
+        """
+        response = self._query(agent=self._agents['iemba'], messages=[HumanMessage(query)])
+        return response
+
+
+    def _call_embax_agent(self, query: str) -> str:
+        """
+        Invokes the EMBA X support agent to retrieve more detailed information about the EMBA X program.
+        
+        Args:
+            query: Query to the EMBA X support agent. Provide collected user data in the query if possible.
+        """
+        response = self._query(agent=self._agents['embax'], messages=[HumanMessage(query)])
+        return response
+
+
+    def _init_agents(self):
+        config: RunnableConfig = {
+            'configurable': {'thread_id': 0}
+        }
+        checkpointer = InMemorySaver()
+        fallback_middleware = ModelFallbackMiddleware(
+            *modelconf.get_fallback_models()
+        )
+        summarization_middleware = SummarizationMiddleware(
+            model=modelconf.get_summarization_model(),
+            max_tokens_before_summary=1000,
+            messages_to_keep=5,
+            summary_prompt=promptconf.get_summarization_prompt(),
+            summary_prefix=promptconf.get_summary_prefix(),
+        )
+        tool_retrieve_context = tool(
+            name_or_callable='retrieve_context',
+            runnable=self._retrieve_context,
+            return_direct=False,
+            parse_docstring=True,
+        )
+        tools_agent_calling = [
+            tool(
+                name_or_callable='call_emba_agent',
+                runnable=self._call_emba_agent,
+                return_direct=False,
+                parse_docstring=True,
+            ),
+            tool(
+                name_or_callable='call_iemba_agent',
+                runnable=self._call_iemba_agent,
+                return_direct=False,
+                parse_docstring=True,
+            ),
+            tool(
+                name_or_callable='call_embax_agent',
+                runnable=self._call_embax_agent,
+                return_direct=False,
+                parse_docstring=True,
+            ),
+        ]
+        agents = {
+            'lead': create_agent(
+                name="Lead Agent",
+                model=modelconf.get_main_agent_model(),
+                tools=tools_agent_calling + [tool_retrieve_context],
+                state_schema=LeadInformationState,
+                system_prompt=promptconf.get_configured_agent_prompt('lead', language=self._language),
+                middleware=[
+                    fallback_middleware,
+                    summarization_middleware,
+                    chainmdw.get_tool_wrapper(),
+                    chainmdw.get_model_wrapper(),
+                ],
+                checkpointer=checkpointer,
+                context_schema=AgentContext,
+            ),            
+        }
+        for agent in ['emba', 'iemba', 'embax']:
+            agents[agent]=create_agent(
+                name=f"{agent.upper()} Agent",
+                model=modelconf.get_main_agent_model(),
+                tools=[tool_retrieve_context],
+                state_schema=LeadInformationState,
+                system_prompt=promptconf.get_configured_agent_prompt(agent, language=self._language),
+                middleware=[
+                    fallback_middleware,
+                    chainmdw.get_tool_wrapper(),
+                    chainmdw.get_model_wrapper(),
+                ],
+                checkpointer=checkpointer,
+                context_schema=AgentContext,
+            )
+        return agents, config
+   
+
+    def generate_greeting(self) -> str:
+        return self._query(
+            agent=self._agents['lead'], 
+            messages=[SystemMessage("Generate a greeting message and introduce yourself.")],
+        ) 
+
+
+    def query(self, query: str) -> str:
+        return self._query(
+            agent=self._agents['lead'],
+            messages=[HumanMessage(query), SystemMessage("You MUST call the retrieve_context tool before answering!")],
+        )
+
+
+    def _query(self, agent, messages: list) -> str:
+        try:
+            result: AIMessage = agent.invoke(
+                {"messages": messages},
+                config=self._config,
+                context=AgentContext(agent_name=agent.name),
+            )
+            response = result['messages'][-1]
+            return response.text
+        except Exception as e:
+            chain_logger.error(f"Failed to invoke the agent: {e.body['message']}")
+            return "I'm sorry, I cannot provide a helpful response right now. Please contact the tech support or try again later."

src/rag/middleware.py

@@ -0,0 +1,99 @@
+from langchain.tools.tool_node import ToolCallRequest
+from langchain.chat_models import BaseChatModel
+from langchain.agents.middleware import (
+        ModelRequest,
+        ModelResponse,
+
+        wrap_model_call,
+        wrap_tool_call,
+)
+from openai import (
+    OpenAIError, 
+    InternalServerError, 
+    NotFoundError, 
+    RateLimitError,
+)
+
+from config import MAX_MODEL_RETRIES
+from src.rag.utilclasses import AgentContext
+from src.utils.logging import get_logger
+
+model_logger = get_logger('chain_model_call')
+tool_logger  = get_logger('chain_tool_call')
+
+class AgentChainMiddleware:
+    _tool_wrapper_middleware = None 
+    _model_wrapper_middleware = None
+
+
+    @classmethod 
+    def get_tool_wrapper(cls):
+        if cls._tool_wrapper_middleware:
+            return cls._tool_wrapper_middleware
+
+        cls._tool_wrapper_middleware = wrap_tool_call(cls._tool_call_wrapper)
+        tool_logger.info(f"Initialized tool call wrapper with call inspection")
+        return cls._tool_wrapper_middleware
+
+    
+    @classmethod
+    def get_model_wrapper(cls):
+        if cls._model_wrapper_middleware:
+            return cls._model_wrapper_middleware
+
+        cls._model_wrapper_middleware = wrap_model_call(cls._model_call_wrapper)
+        model_logger.info(f"Initialized model call wrapper with maximum of {MAX_MODEL_RETRIES} retry attempts")
+        return cls._model_wrapper_middleware
+
+
+    @staticmethod
+    def _model_call_wrapper(request: ModelRequest, handler):
+        context: AgentContext  = request.runtime.context
+        model:   BaseChatModel = request.model
+        model_logger.info(f"{context.agent_name} is attempting to call model '{model.model_name}'...")
+        for attempt in range(1, MAX_MODEL_RETRIES+1):
+            try:
+                response: ModelResponse = handler(request)
+                model_logger.info(f"Recieved response from model after {attempt} attempt{'s' if attempt > 1 else ''}")
+                result = response.result[0]
+                
+                # Check if any errors occured during tool call execution.
+                # Some errors might be fatal, making the model unusable in the agent chain
+                if hasattr(result, 'invalid_tool_calls') and result.invalid_tool_calls:
+                    for invalid_call in result.invalid_tool_calls:
+                        fail_reason = invalid_call.get('error', 'Unknown').replace('\n', '')
+                        model_logger.warning(f"Failed tool call: {invalid_call['name']}, error: {fail_reason}, retrying the call...")
+                        if 'JSONDecodeError' in fail_reason:
+                            model_logger.error(f"Model does not support current tool call architecture! Switching to the fallback model...")
+                            raise Exception("Unsupported model") 
+                
+                else:
+                    return response
+            except OpenAIError as e:
+                match e:
+                    case InternalServerError():
+                        model_logger.warning(f"[{e.code}] Internal difficulties on the provider side, retrying the call...")
+                    case RateLimitError():
+                        model_logger.warning(f"[{e.code}] Model is temporary rate limited, retrying the call...")
+                    case NotFoundError():
+                        model_logger.error(f"[{e.code}] Model cannot be used in the chain, reason: {e.body['message']}")
+                        raise e
+
+                if attempt == MAX_MODEL_RETRIES:
+                    model_logger.warning(f"Failed to recieve response from model '{model.model_name}' after {MAX_MODEL_RETRIES} attempt{'s' if attempt > 1 else ''}, reason: {e.body['message']}")
+                    model_logger.info(f"Switching to the fallback model...")
+                    raise e
+
+
+    @staticmethod
+    def _tool_call_wrapper(request: ToolCallRequest, handler):
+        context: AgentContext = request.runtime.context or AgentContext(agent_name="Agent")
+        
+        tool_call = request.tool_call
+        tool_logger.info(f"{context.agent_name} is calling tool: {tool_call['name']}")
+        tool_logger.info(f"Tool arguments: {tool_call.get('args', {})}")
+        
+        response = handler(request) 
+        tool_logger.info(f"Recieved response from tool call")
+        
+        return response       

src/rag/models.py

@@ -0,0 +1,123 @@
+from langchain.chat_models import BaseChatModel
+from config import LLMProvider, LLMProviderConfiguration as llmconf
+
+from src.utils.logging import get_logger
+
+logger = get_logger("model_config")
+
+class ModelConfigurator:
+    _main_model_instance: BaseChatModel = None
+    _fallback_models_instances: list[BaseChatModel] = None
+    _summarization_model_instance: BaseChatModel = None
+    
+
+    @classmethod
+    def get_summarization_model(cls) -> BaseChatModel:
+        if cls._summarization_model_instance:
+            return cls._summarization_model_instance
+        
+        try:
+            # Add custom summarization model initialization here if needed
+            cls._summarization_model_instance = cls.get_main_agent_model()
+            logger.info(f"Initialized summarization model '{llmconf.LLM_PROVIDER.name}:{llmconf.get_default_model()}'")
+            return cls._summarization_model_instance
+        except Exception as e:
+            logger.error(f"Failed to initialize the summarization model: {e}")
+            raise e
+
+
+    @classmethod
+    def get_main_agent_model(cls) -> BaseChatModel:
+        """Initialize the language model based on config."""
+        if cls._main_model_instance:
+            return cls._main_model_instance
+
+        try:
+            cls._main_model_instance = cls._initialize_model(
+                provider=llmconf.LLM_PROVIDER,
+                model=llmconf.get_default_model()
+            )
+            logger.info(f"Initialized main agent model '{llmconf.LLM_PROVIDER.name}:{llmconf.get_default_model()}'")
+            return cls._main_model_instance
+        except Exception as e: 
+            logger.error(f"Failed to initialize the main agent model for provider '{llmconf.LLM_PROVIDER.name}': {e}")
+            raise e
+
+
+    @classmethod
+    def get_fallback_models(cls) -> list[BaseChatModel]:
+        if cls._fallback_models_instances != None:
+            return cls._fallback_models_instances 
+
+        cls._fallback_models_instances = cls._initialize_fallback_models()
+        if len(cls._fallback_models_instances) == 0:
+            logger.warning("No fallback models were initialized! Response generation may result in unexpected errors!")
+        return cls._fallback_models_instances
+
+
+    @classmethod
+    def _initialize_fallback_models(cls) -> list[BaseChatModel]:
+        fallback_models_instances = []
+        for fallback_provider, fallback_model in llmconf.get_fallback_models().items():
+            try:
+                fallback_model_instance = cls._initialize_model(
+                    provider=fallback_provider,
+                    model=fallback_model,
+                )
+                logger.info(f"Initialized fallback model '{fallback_provider.name}:{fallback_model}'")
+                fallback_models_instances.append(fallback_model_instance)
+            except Exception as e:
+                logger.error(f"Failed to initialize the fallback model {fallback_provider.name}:{fallback_model}: {e}; skipping...")
+        return fallback_models_instances
+
+
+    @classmethod
+    def _initialize_model(cls, provider: LLMProvider, model: str) -> BaseChatModel:
+        try:
+            match provider.name:
+                case 'groq':
+                    from langchain_groq import ChatGroq
+                    return ChatGroq(
+                        model=model,
+                        groq_api_key=llmconf.get_api_key(),
+                        temperature=0.01,
+                    )
+                case (  'open_router:openai' 
+                      | 'open_router:alibaba' 
+                      | 'open_router:nvidia'
+                      | 'open_router:meituan'):
+                    from langchain_openai import ChatOpenAI
+                    return ChatOpenAI(
+                        model=model,
+                        base_url=llmconf.OPEN_ROUTER_BASE_URL,
+                        api_key=llmconf.get_api_key(),
+                        temperature=0.01,
+                    )
+                case 'open_router:deepseek':
+                    from langchain_deepseek import ChatDeepSeek
+                    return ChatDeepSeek(
+                        model=model,
+                        api_key=llmconf.OPEN_ROUTER_API_KEY,
+                        api_base=llmconf.OPEN_ROUTER_BASE_URL,
+                    )
+                case 'openai':
+                    from langchain_openai import ChatOpenAI
+                    return ChatOpenAI(
+                        model=model,
+                        openai_api_key=llmconf.get_api_key(),
+                        max_tokens=1000,
+                        temperature=0.01,
+                    )
+                case 'ollama':
+                    from langchain_ollama import ChatOllama
+                    return ChatOllama(
+                        model=model,
+                        base_url=llmconf.OLLAMA_BASE_URL,
+                        temperature=0.01,
+                        reasoning=llmconf.get_reasoning_support(),
+                        num_predict=2048,
+                    )
+                case _:
+                    raise ValueError(f"Unsupported LLM provider: {provider.name}")
+        except Exception as e:
+            raise e

src/rag/prompts.py

@@ -0,0 +1,105 @@
+class PromptConfigurator:
+    _PROGRAM_SYSTEM_PROMPT = """
+You are a helpful support agent, explicitly specializing in the {program_name} program offered by the University of St. Gallen Executive School. You work alongside the Executive Education Advisor. Your task is to provide correct information about the {program_name} and check whether the user meets qualificaiton criteria for the {program_name} program based on their experience and career goals.
+
+Use only the provided context to provide information about the {program_name} program. The context include information such as duration, curriculum, costs, admission requirements, schedules, faculty, deadlines, and other relevant details.
+
+Before answering any user questions you MUST use the 'retrieve_context' tool to retrieve context!
+Answer ONLY after retrieving information.
+
+General Guidelines:
+{general_guidelines}
+- If user does not provide information about a critera, ask them to provide more information about it.
+- Do not hallucinate or give qualificaitons to the user that they have not provided themselves.
+- If user only meets the minimal criteria, proactively recommend contacting the admissions team for more information.
+- If user does not meet minimal criteria, recommend the regular MBA program as an alternative and provide the contact information of the admissions team.
+    """
+
+    _LEAD_SYSTEM_PROMPT="""
+You are an Executive Education Advisor for the University of St. Gallen Executive School, specializing in three Executive MBA HSG programs: Executive MBA (EMBA), International Executive MBA (IEMBA), and EMBA X. Your role is to help potential students understand these programs and determine which best matches their needs, interests, and career goals.
+
+Use only the provided context to answer questions about the Executive MBA HSG programs. The context include information such as duration, curriculum, costs, admission requirements, schedules, faculty, deadlines, and other relevant details.
+
+Before answering any user questions you MUST use the appropriate tool:
+- If you need detailed program information, call `call_emba_agent`, `call_iemba_agent`, or `call_embax_agent`.
+- If you need to retrieve general context, call `retrieve_context`.
+- Only answer after retrieving information.
+
+General Guidelines:
+{general_guidelines}
+{lead_guidelines}
+
+Formatting Guidelines:
+{formatting_guidelines}
+    """
+
+    _LEAD_GUIDELINES = """
+- If another language is used, politely inform the user you can only respond in {selected_language}.
+- Be nice and keep the conversation fluent and human-like.
+- List all available programs, including EMBA, IEMBA, and EMBA X, if user has general interest in studying.
+- When listing all programs, include duration, deadlines and special program aspects. Ask user about their experience and qualificaitons afterwards.
+- Primarily recommend {prefered_program} program.
+- If user is not explicitly stating the program he is asking about, talk about the {prefered_program} program.
+- Try not to repeat the information that was already stated in the previous answer.
+- You are not allowed to mention or discuss programs offered by competitor universities. 
+- If the user attemps to discuss anything unrelated to the MBA programs, politely switch back to the main topic. You are not allowed to discuss anything besides the HSG MBA programs.
+- Do not decide yourself whether the user has good or bad chances. 
+- If user is asking about their chances, state clearly that the admissions team makes the final decision.
+- Proactively recommend contacting the admissions team after checking the user's qualificaitons.
+- If context does not cover a user question, clearly inform the user and suggest contacting the admissions team.   
+"""
+
+    _GENERAL_GUIDELINES = """
+- Respond only in {selected_language}. 
+- Be helpful, professional, and keep answers short and concise.
+- Only provide program prices if user is asking about them. 
+- Never state the exact pricing; only provide program prices in 5k ranges and mention Early Bird Discount if it exists.
+- When providing program prices, list all services that are included in them.
+    """
+
+    _FORMATTING_GUIDELINES = """
+- Use Markdown formatting.
+- Use appropriate emojis.
+- Do not add titles at the beginning of an answer.
+- Highlight key facts (e.g., program names, costs, durations) in bold.
+- Use tables when listing or comparing program features.
+- Maintain clean and consistent formatting.
+    """
+
+    _SUMMARIZATION_PROMPT = """
+Write a short summarization of the conversation between the Executive Education Advisor and the user. In summarization include previously discussed topics as well as all the information that the user provided about their work experience and career goals.
+    """
+
+    _SUMMARY_PREFIX_PROMPT = """Conversation Summary:"""
+
+
+    @classmethod
+    def get_configured_agent_prompt(cls, agent: str, language: str = 'en'):
+        match agent:
+            case 'lead':
+                return cls._LEAD_SYSTEM_PROMPT.format(
+                    general_guidelines=cls._GENERAL_GUIDELINES.format(
+                        selected_language=language
+                    ),
+                    lead_guidelines=cls._LEAD_GUIDELINES.format(
+                        selected_language=language,
+                        prefered_program='EMBA' if language == 'de' else 'IEMBA',
+                    ),
+                    formatting_guidelines=cls._FORMATTING_GUIDELINES,
+                )
+            case _:
+                return cls._PROGRAM_SYSTEM_PROMPT.format(
+                    program_name=agent.upper(),
+                    general_guidelines=cls._GENERAL_GUIDELINES.format(
+                        selected_language=language,
+                    ),
+                )
+    
+    @classmethod
+    def get_summarization_prompt(cls):
+        return cls._SUMMARIZATION_PROMPT
+
+
+    @classmethod
+    def get_summary_prefix(cls):
+        return cls._SUMMARY_PREFIX_PROMPT

src/rag/utilclasses.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+from typing_extensions import TypedDict
+from langchain.agents import AgentState
+from langchain_core.messages import AnyMessage
+
+@dataclass
+class AgentContext:
+    agent_name: str
+
+class State(TypedDict):
+    messages: list[AnyMessage]
+    answer: str
+
+
+class LeadInformationState(AgentState):
+    lead_name: str
+    lead_age:  int
+    lead_language_knowledge: list 
+    lead_work_experience: dict
+    lead_motivation: list

src/scraper/parser.py

@@ -0,0 +1,304 @@
+"""
+Parser for extracting structured information from program pages.
+"""
+import logging
+import re
+from typing import Dict, List, Optional, Any
+
+from bs4 import BeautifulSoup
+from src.utils.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class ProgramParser:
+    """Parser for extracting program information from HTML content."""
+
+    def parse_program_page(self, html_content: str, url: str) -> Dict[str, Any]:
+        """
+        Parse a program page to extract structured information.
+
+        Args:
+            html_content: The HTML content of the program page.
+            url: The URL of the program page.
+
+        Returns:
+            A dictionary containing structured program information.
+        """
+        soup = BeautifulSoup(html_content, "html.parser")
+        
+        # Initialize program data with URL
+        program_data = {
+            "url": url,
+            "name": self._extract_program_name(soup),
+            "duration": self._extract_duration(soup),
+            "curriculum": self._extract_curriculum(soup),
+            "costs": self._extract_costs(soup),
+            "admission_requirements": self._extract_admission_requirements(soup),
+            "schedules": self._extract_schedules(soup),
+            "faculty": self._extract_faculty(soup),
+            "deadlines": self._extract_deadlines(soup),
+            "language": self._extract_language(soup),
+            "location": self._extract_location(soup),
+            "description": self._extract_description(soup),
+        }
+        
+        return program_data
+
+    def _extract_program_name(self, soup: BeautifulSoup) -> str:
+        """Extract the program name from the soup."""
+        # This is a placeholder. Update with actual selectors based on website structure.
+        try:
+            # Try to find the program name in the page title or a heading
+            title_element = soup.find("h1") or soup.find("title")
+            if title_element:
+                return title_element.get_text().strip()
+        except Exception as e:
+            logger.error(f"Error extracting program name: {e}")
+        
+        return "Unknown Program"
+
+    def _extract_duration(self, soup: BeautifulSoup) -> str:
+        """Extract the program duration from the soup."""
+        try:
+            # Look for duration information
+            # This is a placeholder. Update with actual selectors.
+            duration_section = soup.find("div", class_="program-duration")
+            if duration_section:
+                return duration_section.get_text().strip()
+            
+            # Try to find duration in the text
+            text = soup.get_text()
+            duration_patterns = [
+                r"Duration:?\s*([^\.]+)",
+                r"Program length:?\s*([^\.]+)",
+                r"(\d+)\s+months",
+                r"(\d+)\s+years",
+            ]
+            
+            for pattern in duration_patterns:
+                match = re.search(pattern, text, re.IGNORECASE)
+                if match:
+                    return match.group(1).strip()
+        except Exception as e:
+            logger.error(f"Error extracting duration: {e}")
+        
+        return "Not specified"
+
+    def _extract_curriculum(self, soup: BeautifulSoup) -> List[str]:
+        """Extract the curriculum information from the soup."""
+        curriculum = []
+        try:
+            # Look for curriculum information
+            # This is a placeholder. Update with actual selectors.
+            curriculum_section = soup.find("div", class_="program-curriculum")
+            if curriculum_section:
+                # Look for list items
+                items = curriculum_section.find_all("li")
+                if items:
+                    for item in items:
+                        curriculum.append(item.get_text().strip())
+                else:
+                    # If no list items, get the text
+                    curriculum.append(curriculum_section.get_text().strip())
+        except Exception as e:
+            logger.error(f"Error extracting curriculum: {e}")
+        
+        return curriculum
+
+    def _extract_costs(self, soup: BeautifulSoup) -> str:
+        """Extract the program costs from the soup."""
+        try:
+            # Look for cost information
+            # This is a placeholder. Update with actual selectors.
+            cost_section = soup.find("div", class_="program-costs")
+            if cost_section:
+                return cost_section.get_text().strip()
+            
+            # Try to find cost in the text
+            text = soup.get_text()
+            cost_patterns = [
+                r"Cost:?\s*([^\.]+)",
+                r"Tuition:?\s*([^\.]+)",
+                r"Fee:?\s*([^\.]+)",
+                r"CHF\s*[\d\',]+",
+                r"€\s*[\d\',]+",
+            ]
+            
+            for pattern in cost_patterns:
+                match = re.search(pattern, text, re.IGNORECASE)
+                if match:
+                    return match.group(0).strip()
+        except Exception as e:
+            logger.error(f"Error extracting costs: {e}")
+        
+        return "Not specified"
+
+    def _extract_admission_requirements(self, soup: BeautifulSoup) -> List[str]:
+        """Extract the admission requirements from the soup."""
+        requirements = []
+        try:
+            # Look for admission requirements
+            # This is a placeholder. Update with actual selectors.
+            requirements_section = soup.find("div", class_="program-requirements")
+            if requirements_section:
+                # Look for list items
+                items = requirements_section.find_all("li")
+                if items:
+                    for item in items:
+                        requirements.append(item.get_text().strip())
+                else:
+                    # If no list items, get the text
+                    requirements.append(requirements_section.get_text().strip())
+        except Exception as e:
+            logger.error(f"Error extracting admission requirements: {e}")
+        
+        return requirements
+
+    def _extract_schedules(self, soup: BeautifulSoup) -> str:
+        """Extract the program schedules from the soup."""
+        try:
+            # Look for schedule information
+            # This is a placeholder. Update with actual selectors.
+            schedule_section = soup.find("div", class_="program-schedule")
+            if schedule_section:
+                return schedule_section.get_text().strip()
+            
+            # Try to find schedule in the text
+            text = soup.get_text()
+            schedule_patterns = [
+                r"Schedule:?\s*([^\.]+)",
+                r"Classes:?\s*([^\.]+)",
+                r"Start date:?\s*([^\.]+)",
+            ]
+            
+            for pattern in schedule_patterns:
+                match = re.search(pattern, text, re.IGNORECASE)
+                if match:
+                    return match.group(1).strip()
+        except Exception as e:
+            logger.error(f"Error extracting schedules: {e}")
+        
+        return "Not specified"
+
+    def _extract_faculty(self, soup: BeautifulSoup) -> List[Dict[str, str]]:
+        """Extract the faculty information from the soup."""
+        faculty = []
+        try:
+            # Look for faculty information
+            # This is a placeholder. Update with actual selectors.
+            faculty_section = soup.find("div", class_="program-faculty")
+            if faculty_section:
+                faculty_members = faculty_section.find_all("div", class_="faculty-member")
+                for member in faculty_members:
+                    name_element = member.find("h3") or member.find("strong")
+                    name = name_element.get_text().strip() if name_element else "Unknown"
+                    
+                    title_element = member.find("p", class_="faculty-title")
+                    title = title_element.get_text().strip() if title_element else ""
+                    
+                    faculty.append({
+                        "name": name,
+                        "title": title,
+                    })
+        except Exception as e:
+            logger.error(f"Error extracting faculty: {e}")
+        
+        return faculty
+
+    def _extract_deadlines(self, soup: BeautifulSoup) -> str:
+        """Extract the application deadlines from the soup."""
+        try:
+            # Look for deadline information
+            # This is a placeholder. Update with actual selectors.
+            deadline_section = soup.find("div", class_="program-deadlines")
+            if deadline_section:
+                return deadline_section.get_text().strip()
+            
+            # Try to find deadlines in the text
+            text = soup.get_text()
+            deadline_patterns = [
+                r"Deadline:?\s*([^\.]+)",
+                r"Application deadline:?\s*([^\.]+)",
+                r"Apply by:?\s*([^\.]+)",
+            ]
+            
+            for pattern in deadline_patterns:
+                match = re.search(pattern, text, re.IGNORECASE)
+                if match:
+                    return match.group(1).strip()
+        except Exception as e:
+            logger.error(f"Error extracting deadlines: {e}")
+        
+        return "Not specified"
+
+    def _extract_language(self, soup: BeautifulSoup) -> str:
+        """Extract the program language from the soup."""
+        try:
+            # Look for language information
+            # This is a placeholder. Update with actual selectors.
+            language_section = soup.find("div", class_="program-language")
+            if language_section:
+                return language_section.get_text().strip()
+            
+            # Try to find language in the text
+            text = soup.get_text()
+            language_patterns = [
+                r"Language:?\s*([^\.]+)",
+                r"Taught in:?\s*([^\.]+)",
+            ]
+            
+            for pattern in language_patterns:
+                match = re.search(pattern, text, re.IGNORECASE)
+                if match:
+                    return match.group(1).strip()
+        except Exception as e:
+            logger.error(f"Error extracting language: {e}")
+        
+        return "Not specified"
+
+    def _extract_location(self, soup: BeautifulSoup) -> str:
+        """Extract the program location from the soup."""
+        try:
+            # Look for location information
+            # This is a placeholder. Update with actual selectors.
+            location_section = soup.find("div", class_="program-location")
+            if location_section:
+                return location_section.get_text().strip()
+            
+            # Try to find location in the text
+            text = soup.get_text()
+            location_patterns = [
+                r"Location:?\s*([^\.]+)",
+                r"Campus:?\s*([^\.]+)",
+                r"Venue:?\s*([^\.]+)",
+            ]
+            
+            for pattern in location_patterns:
+                match = re.search(pattern, text, re.IGNORECASE)
+                if match:
+                    return match.group(1).strip()
+        except Exception as e:
+            logger.error(f"Error extracting location: {e}")
+        
+        return "Not specified"
+
+    def _extract_description(self, soup: BeautifulSoup) -> str:
+        """Extract the program description from the soup."""
+        try:
+            # Look for description information
+            # This is a placeholder. Update with actual selectors.
+            description_section = soup.find("div", class_="program-description")
+            if description_section:
+                return description_section.get_text().strip()
+            
+            # Try to find a general description
+            paragraphs = soup.find_all("p")
+            if paragraphs:
+                # Get the first few paragraphs as the description
+                description = " ".join([p.get_text().strip() for p in paragraphs[:3]])
+                return description
+        except Exception as e:
+            logger.error(f"Error extracting description: {e}")
+        
+        return "No description available"

src/scraper/scraper.py

@@ -0,0 +1,292 @@
+"""
+Web scraper for extracting program information from the University of St. Gallen Executive School website.
+"""
+import json
+import logging
+import os
+import platform
+import time
+from typing import Dict, List, Optional, Set
+
+import requests
+from bs4 import BeautifulSoup
+from selenium import webdriver
+from selenium.webdriver.chrome.options import Options
+from selenium.webdriver.chrome.service import Service
+from selenium.common.exceptions import WebDriverException
+from webdriver_manager.chrome import ChromeDriverManager
+
+from src.scraper.parser import ProgramParser
+from src.utils.logging import get_logger
+from config import BASE_URL, RAW_DATA_PATH, SCRAPER_HEADERS, SCRAPER_TIMEOUT
+
+logger = get_logger(__name__)
+
+
+class Scraper:
+    """Scraper for the University of St. Gallen Executive School website."""
+
+    def __init__(self, use_selenium: bool = False):
+        """
+        Initialize the scraper.
+
+        Args:
+            use_selenium: Whether to use Selenium for scraping (for JavaScript-heavy pages).
+        """
+        self.base_url = BASE_URL
+        self.headers = SCRAPER_HEADERS
+        self.timeout = SCRAPER_TIMEOUT
+        self.use_selenium = use_selenium
+        self.driver = None
+        self.parser = ProgramParser()
+        self.visited_urls: Set[str] = set()
+
+    def _get_chrome_install_instructions(self) -> str:
+        """Get Chrome installation instructions based on the operating system."""
+        system = platform.system().lower()
+        
+        if system == "darwin":  # macOS
+            return "Install Chrome via: brew install --cask google-chrome"
+        elif system == "linux":
+            # Check if it's Ubuntu/Debian or other
+            try:
+                with open("/etc/os-release", "r") as f:
+                    os_info = f.read().lower()
+                if "ubuntu" in os_info or "debian" in os_info:
+                    return "Install Chrome via: sudo apt-get update && sudo apt-get install google-chrome-stable"
+                else:
+                    return "Install Chrome from your package manager or download from https://www.google.com/chrome/"
+            except FileNotFoundError:
+                return "Install Chrome from your package manager or download from https://www.google.com/chrome/"
+        elif system == "windows":
+            return "Download Chrome from https://www.google.com/chrome/"
+        else:
+            return "Download Chrome from https://www.google.com/chrome/"
+
+    def _setup_selenium(self) -> None:
+        """Set up Selenium WebDriver with proper error handling."""
+        if self.use_selenium and self.driver is None:
+            logger.info("Setting up Selenium WebDriver...")
+            
+            try:
+                chrome_options = Options()
+                chrome_options.add_argument("--headless")
+                chrome_options.add_argument("--no-sandbox")
+                chrome_options.add_argument("--disable-dev-shm-usage")
+                chrome_options.add_argument("--disable-gpu")
+                chrome_options.add_argument("--disable-dev-shm-usage")
+                chrome_options.add_argument("--remote-debugging-port=9222")
+                
+                # Try to install ChromeDriver
+                try:
+                    service = Service(ChromeDriverManager("140.0.7339").install())
+                except Exception as e:
+                    logger.error(f"Failed to install ChromeDriver: {e}")
+                    raise WebDriverException("ChromeDriver installation failed")
+                
+                # Try to create Chrome WebDriver
+                self.driver = webdriver.Chrome(service=service, options=chrome_options)
+                logger.info("Chrome WebDriver initialized successfully")
+                
+            except WebDriverException as e:
+                error_msg = str(e)
+                if "chrome binary" in error_msg.lower() or "chromedriver" in error_msg.lower():
+                    install_instructions = self._get_chrome_install_instructions()
+                    logger.error(
+                        f"Chrome browser not found. Please install Chrome first.\n"
+                        f"Installation instructions for your system:\n{install_instructions}\n"
+                        f"After installation, restart the application."
+                    )
+                    raise WebDriverException(
+                        f"Chrome browser not found. {install_instructions}"
+                    )
+                else:
+                    logger.error(f"WebDriver setup failed: {e}")
+                    raise
+            except Exception as e:
+                logger.error(f"Unexpected error during WebDriver setup: {e}")
+                raise WebDriverException(f"WebDriver setup failed: {e}")
+
+    def _cleanup_selenium(self) -> None:
+        """Clean up Selenium WebDriver."""
+        if self.driver is not None:
+            logger.info("Cleaning up Selenium WebDriver...")
+            self.driver.quit()
+            self.driver = None
+
+    def _get_page_content(self, url: str) -> Optional[str]:
+        """
+        Get the HTML content of a page.
+
+        Args:
+            url: The URL to fetch.
+
+        Returns:
+            The HTML content of the page, or None if the request failed.
+        """
+        try:
+            if self.use_selenium:
+                self._setup_selenium()
+                self.driver.get(url)
+                # Wait for JavaScript to load
+                time.sleep(2)
+                return self.driver.page_source
+            else:
+                response = requests.get(url, headers=self.headers, timeout=self.timeout)
+                response.raise_for_status()
+                return response.text
+        except Exception as e:
+            logger.error(f"Error fetching {url}: {e}")
+            return None
+
+    def _extract_program_links(self, html_content: str) -> List[str]:
+        """
+        Extract links to program pages from the main page.
+
+        Args:
+            html_content: The HTML content of the main page.
+
+        Returns:
+            A list of URLs to program pages.
+        """
+        soup = BeautifulSoup(html_content, "html.parser")
+        program_links = []
+
+        # Try multiple selectors in order of specificity
+        selectors = [
+            "a.program-link",  # Specific program link class
+            "a[href*='program']",  # Links containing 'program' in href
+            "a[href*='emba']",  # Links containing 'emba' in href
+            ".program a",  # Links inside program containers
+            ".course a",  # Links inside course containers
+            "a"  # Fallback to all links
+        ]
+        
+        for selector in selectors:
+            link_elements = soup.select(selector)
+            if link_elements:
+                logger.info(f"Using selector: {selector} (found {len(link_elements)} links)")
+                break
+        else:
+            logger.warning("No links found with any selector")
+            return []
+        
+        for link in link_elements:
+            href = link.get("href")
+            if href:
+                # Filter out anchor-only links (starting with #)
+                if href.startswith("#"):
+                    continue
+                
+                # Filter out mailto, tel, and javascript links
+                if href.startswith(("mailto:", "tel:", "javascript:")):
+                    continue
+                
+                # Filter out external links that don't belong to the target domain
+                if href.startswith("http") and "unisg.ch" not in href:
+                    continue
+                
+                # Make sure we have absolute URLs
+                if href.startswith("/"):
+                    href = f"https://emba.unisg.ch{href}"
+                elif not href.startswith("http"):
+                    # Relative URL
+                    href = f"https://emba.unisg.ch/{href.lstrip('/')}"
+                
+                # Avoid duplicates
+                if href not in program_links:
+                    program_links.append(href)
+
+        logger.info(f"Found {len(program_links)} valid program links after filtering")
+        return program_links
+
+    def scrape_main_page(self) -> List[str]:
+        """
+        Scrape the main page to find links to program pages.
+
+        Returns:
+            A list of URLs to program pages.
+        """
+        logger.info(f"Scraping main page: {self.base_url}")
+        html_content = self._get_page_content(self.base_url)
+        
+        if not html_content:
+            logger.error("Failed to fetch main page")
+            return []
+        
+        return self._extract_program_links(html_content)
+
+    def scrape_program_page(self, url: str) -> Optional[Dict]:
+        """
+        Scrape a program page to extract program information.
+
+        Args:
+            url: The URL of the program page.
+
+        Returns:
+            A dictionary containing program information, or None if scraping failed.
+        """
+        if url in self.visited_urls:
+            logger.info(f"Already visited {url}, skipping")
+            return None
+        
+        logger.info(f"Scraping program page: {url}")
+        self.visited_urls.add(url)
+        
+        html_content = self._get_page_content(url)
+        if not html_content:
+            logger.error(f"Failed to fetch program page: {url}")
+            return None
+        
+        program_data = self.parser.parse_program_page(html_content, url)
+        return program_data
+
+    def scrape_all_programs(self) -> List[Dict]:
+        """
+        Scrape all program pages.
+
+        Returns:
+            A list of dictionaries containing program information.
+        """
+        try:
+            program_links = self.scrape_main_page()
+            programs_data = []
+            
+            for link in program_links:
+                program_data = self.scrape_program_page(link)
+                if program_data:
+                    programs_data.append(program_data)
+                # Be nice to the server
+                time.sleep(1)
+            
+            return programs_data
+        finally:
+            self._cleanup_selenium()
+
+    def save_data(self, data: List[Dict], output_path: str = RAW_DATA_PATH) -> None:
+        """
+        Save scraped data to a JSON file.
+
+        Args:
+            data: The data to save.
+            output_path: The path to save the data to.
+        """
+        # Ensure the directory exists
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        
+        with open(output_path, "w", encoding="utf-8") as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+        
+        logger.info(f"Saved {len(data)} programs to {output_path}")
+
+    def run(self) -> None:
+        """Run the scraper and save the results."""
+        logger.info("Starting scraper...")
+        programs_data = self.scrape_all_programs()
+        self.save_data(programs_data)
+        logger.info("Scraping completed")
+
+
+if __name__ == "__main__":
+    scraper = Scraper(use_selenium=True)
+    scraper.run()

src/ui/cli.py

@@ -0,0 +1,154 @@
+"""
+Command-line interface for the RAG chatbot.
+"""
+import os
+import sys
+from typing import Dict, List, Any, Optional, Tuple
+
+import colorama
+from colorama import Fore, Style
+from rich.console import Console
+from rich.markdown import Markdown
+from rich.panel import Panel
+from rich.table import Table
+
+from config import MAX_HISTORY
+from src.rag.chain import RAGChain
+from src.utils.logging import get_logger, init_logging
+
+# Initialize colorama
+colorama.init()
+
+init_logging(interactive_mode=False)
+logger = get_logger('cli')
+
+# Initialize rich console
+console = Console()
+
+
+class ChatbotCLI:
+    """Command-line interface for the RAG chatbot."""
+
+    def __init__(self):
+        """Initialize the chatbot CLI."""
+        self.rag_chain = RAGChain()
+        self.chat_history = []
+
+    def _print_welcome_message(self) -> None:
+        """Print the welcome message."""
+        console.print(
+            Panel(
+                "[bold green]Executive MBA HSG Chatbot[/bold green]\n\n"
+                "Welcome! I'm your Executive Education Advisor specializing in the Executive MBA HSG program. "
+                "Ask me about the Executive MBA HSG program's curriculum, "
+                "admissions, costs, or anything related to this specific program at the "
+                "University of St. Gallen.\n\n"
+                "Type [bold cyan]'exit'[/bold cyan], [bold cyan]'quit'[/bold cyan], [bold cyan]'bye'[/bold cyan], or [bold cyan]'goodbye'[/bold cyan] to end the conversation.",
+                title="Welcome",
+                border_style="green",
+                width=100,
+            )
+        )
+
+    def _print_sources(self, sources: List[Dict[str, Any]]) -> None:
+        """
+        Print the sources of information.
+
+        Args:
+            sources: List of source information dictionaries.
+        """
+        if not sources:
+            return
+        
+        table = Table(title="Sources", box=None, show_header=True, header_style="bold cyan")
+        table.add_column("Program", style="cyan")
+        
+        for source in sources:
+            program_name = source.get("program_name", "Unknown Program")
+            table.add_row(program_name)
+        
+        console.print(table)
+
+    def _print_response(self, response: Dict[str, Any]) -> None:
+        """
+        Print the chatbot response.
+
+        Args:
+            response: The response dictionary.
+        """
+        answer = response.get("answer", "")
+        sources = response.get("sources", [])
+        
+        # Print the answer as markdown
+        console.print(Markdown(answer))
+        
+        # Print the sources
+        if sources:
+            console.print("\n[bold cyan]Sources:[/bold cyan]")
+            self._print_sources(sources)
+
+    def _get_user_input(self) -> str:
+        """
+        Get input from the user.
+
+        Returns:
+            The user input.
+        """
+        console.print("\n[bold green]You:[/bold green] ", end="")
+        return input()
+
+    def _process_query(self, query: str) -> Dict[str, Any]:
+        """
+        Process a user query.
+
+        Args:
+            query: The user query.
+
+        Returns:
+            The response dictionary.
+        """
+        try:
+            # Get the response from the RAG chain
+            response = self.rag_chain.query(query, self.chat_history)
+            
+            # Update chat history
+            if len(self.chat_history) >= MAX_HISTORY:
+                self.chat_history.pop(0)
+            
+            self.chat_history.append((query, response["answer"]))
+            
+            return response
+        except Exception as e:
+            logger.error(f"Error processing query: {e}")
+            return {
+                "answer": "I'm sorry, I encountered an error while processing your question. Please try again.",
+                "sources": [],
+            }
+
+    def run(self) -> None:
+        """Run the chatbot CLI."""
+        self._print_welcome_message()
+        
+        while True:
+            # Get user input
+            query = self._get_user_input()
+            
+            # Check if the user wants to exit
+            if query.lower() in ["exit", "quit", "bye", "goodbye"]:
+                console.print("\n[bold green]Thank you for using the Executive MBA HSG Chatbot. Goodbye![/bold green]")
+                break
+            
+            # Process the query
+            console.print("\n[bold blue]Assistant:[/bold blue]")
+            response = self._process_query(query)
+            
+            # Print the response
+            self._print_response(response)
+            
+            # Add a separator
+            console.print("\n" + "-" * 100)
+
+
+if __name__ == "__main__":
+    cli = ChatbotCLI()
+    cli.run()

src/utils/__init__.py

@@ -0,0 +1,3 @@
+"""
+Utility modules for the Executive Education RAG Chatbot.
+"""

src/utils/lang.py

@@ -0,0 +1,14 @@
+from langdetect import detect 
+
+def detect_language(text: str):
+    """
+    Detect the language of the given text.
+
+    Args:
+        text (str): The text to analyze.
+
+    Returns:
+        str: Detected language code ('de' or 'en').
+    """
+    return 'de' if detect(text) == 'de' else 'en'
+    

src/utils/logging.py

@@ -0,0 +1,284 @@
+"""
+Centralized logging configuration for the Executive Education RAG Chatbot.
+"""
+from threading import Lock
+import logging, os, sys, warnings, copy
+from pathlib import Path
+from typing import Optional
+
+import colorama
+from colorama import Fore, Style
+
+# Initialize colorama for cross-platform color support
+colorama.init()
+
+class CodeBlockFormatter(logging.Formatter):
+    ALIASES = {
+        'DEBUG':    'DEBUG',
+        'INFO':     'INFO ',
+        'WARNING':  'WARN ',
+        'ERROR':    'ERROR',
+        'CRITICAL': 'CRITC'
+    }
+
+    def format(self, record):
+        if hasattr(record, 'levelname') and record.levelname in self.ALIASES:
+            record.levelname = self.ALIASES[record.levelname]
+
+        if hasattr(record, 'name'):
+            rname = record.name 
+            if len(rname) <= 17: rname += ' '*(17-len(rname))
+            else: rname = rname[14:] + '...'
+            record.name = rname
+        
+        return super().format(record)
+
+
+class ColoredFormatter(logging.Formatter):
+    """Custom formatter with color support for console output."""
+    
+    COLORS = {
+        'DEBUG':    Fore.CYAN,
+        'INFO':     Fore.GREEN,
+        'WARNING':  Fore.YELLOW,
+        'ERROR':    Fore.RED,
+        'CRITICAL': Fore.MAGENTA + Style.BRIGHT,
+    }
+    ALIASES = {
+        'DEBUG':    'DEBUG',
+        'INFO':     'INFO ',
+        'WARNING':  'WARN ',
+        'ERROR':    'ERROR',
+        'CRITICAL': 'CRITC'
+    }
+    
+    def format(self, record):
+        # Add color to the level name
+        if hasattr(record, 'levelname') and record.levelname in self.COLORS:
+            lname = record.levelname
+            if hasattr(record, 'message') and lname == 'ERROR':
+                record.message = f"{self.COLORS[lname]}{record.message}{Style.RESET_ALL}"
+
+            record.levelname = f"{self.COLORS[lname]}{self.ALIASES[lname]}{Style.RESET_ALL}"
+            
+
+        if hasattr(record, 'name'):
+            rname = record.name if len(record.name) <= 17 else record.name[:14] + '...'
+            record.name = f"{Fore.CYAN}{rname}{Style.RESET_ALL}"
+
+        return super().format(record)
+
+
+class CachedLogHandler(logging.Handler):
+    def __init__(self, level = 0) -> None:
+        super().__init__(level)
+        self._cache = []
+        self._max_lines = 50
+        self._lock = Lock()
+
+
+    def emit(self, record):
+        record_copy = copy.copy(record)
+        msg = self.format(record_copy)
+        with self._lock:
+            self._cache.append(msg)
+
+    def get_logs(self):
+        with self._lock:
+            return '\n'.join(self._cache)
+
+
+cached_log_handler: CachedLogHandler = CachedLogHandler()
+
+def setup_logging(
+    level: str = "INFO",
+    log_file: Optional[str] = None,
+    interactive_mode: bool = False,
+    module_name: Optional[str] = None
+) -> logging.Logger:
+    """
+    Set up centralized logging configuration.
+    
+    Args:
+        level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
+        log_file: Optional path to log file. If None, uses default location.
+        interactive_mode: If True, logs only to file in interactive mode
+        module_name: Name of the module requesting the logger
+    
+    Returns:
+        Configured logger instance
+    """
+    global cached_log_handler
+
+    # Convert string level to logging constant
+    numeric_level = getattr(logging, level.upper(), logging.INFO)
+    
+    # Create logger
+    logger = logging.getLogger()
+    
+    # Avoid duplicate handlers if logger already configured
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    logger.setLevel(numeric_level)
+    
+    # Create formatters
+    detailed_formatter = logging.Formatter(
+        "(%(asctime)s) %(name)s\t %(levelname)s: %(message)s",
+        datefmt="%Y.%m.%d %H:%M:%S"
+    )
+    
+    colored_formatter = ColoredFormatter(
+        "(%(asctime)s) %(name)s\t %(levelname)s: %(message)s",
+        datefmt="%Y.%m.%d %H:%M:%S"
+    )
+
+    codeblock_formatter = CodeBlockFormatter(
+       "%(levelname)s %(name)s\t : %(message)s", 
+    )
+    
+    # Set up file logging
+    if log_file or interactive_mode:
+        if not log_file:
+            # Default log file location
+            log_dir = Path("logs")
+            log_dir.mkdir(exist_ok=True)
+            log_file = log_dir / "rag_chatbot.log"
+        
+        file_handler = logging.FileHandler(log_file, encoding='utf-8')
+        file_handler.setLevel(numeric_level)
+        file_handler.setFormatter(detailed_formatter)
+        logger.addHandler(file_handler)
+    
+    # Set up the cached log handler for gradio Code block output
+    cached_log_handler.setLevel(numeric_level)
+    cached_log_handler.setFormatter(codeblock_formatter)
+
+    # Set up console logging (unless in interactive mode)
+    if not interactive_mode:
+        console_handler = logging.StreamHandler(sys.stdout)
+        console_handler.setLevel(numeric_level)
+        
+        # Use colored formatter if terminal supports it
+        if _supports_color():
+            console_handler.setFormatter(colored_formatter)
+        else:
+            console_handler.setFormatter(detailed_formatter)
+        
+        logger.addHandler(cached_log_handler)
+        logger.addHandler(console_handler)
+    
+    return logger
+
+
+def get_logger(module_name: str) -> logging.Logger:
+    """
+    Get a logger for a specific module.
+    
+    Args:
+        module_name: Name of the module requesting the logger
+    
+    Returns:
+        Logger instance
+    """
+    logger = logging.getLogger(module_name)
+    logger.propagate = True       
+    return logger
+
+
+def _supports_color() -> bool:
+    """
+    Check if the terminal supports color output.
+    
+    Returns:
+        True if color is supported, False otherwise
+    """
+    # Check if we're in a terminal
+    if not hasattr(sys.stdout, 'isatty') or not sys.stdout.isatty():
+        return False
+    
+    # Check environment variables
+    if os.getenv('NO_COLOR'):
+        return False
+    
+    if os.getenv('FORCE_COLOR'):
+        return True
+    
+    # Check terminal type
+    term = os.getenv('TERM', '').lower()
+    if 'color' in term or term in ['xterm', 'xterm-256color', 'screen']:
+        return True
+    
+    return False
+
+
+def detect_interactive_mode() -> bool:
+    """
+    Detect if the application is running in interactive mode.
+    
+    Returns:
+        True if in interactive mode, False otherwise
+    """
+    # Check if no command line arguments were provided (except script name)
+    if len(sys.argv) == 1:
+        return True
+    
+    # Check if only the script name and no other arguments
+    # This indicates the chatbot is running in default interactive mode
+    return False
+
+
+def configure_external_loggers(level: str = "WARNING") -> None:
+    """
+    Configure logging for external libraries to reduce noise.
+    
+    Args:
+        level: Logging level for external libraries
+    """
+    external_loggers = [
+        'selenium',
+        'urllib3',
+        'requests',
+        'chromadb',
+        'docling',
+        'weaviate',
+        'langchain',
+        'langgraph',
+        'openai',
+        'httpx'
+    ]
+    
+    numeric_level = getattr(logging, level.upper(), logging.WARNING)
+    
+    for logger_name in external_loggers:
+        logging.getLogger(logger_name).setLevel(numeric_level)
+
+
+# Global configuration function
+def init_logging(
+    level: str = "INFO",
+    log_file: Optional[str] = None,
+    interactive_mode: Optional[bool] = None
+) -> None:
+    """
+    Initialize the global logging configuration.
+    
+    Args:
+        level: Logging level
+        log_file: Optional log file path
+        interactive_mode: If None, auto-detect interactive mode
+    """
+    if interactive_mode is None:
+        interactive_mode = detect_interactive_mode()
+    
+    warnings.filterwarnings("ignore")
+
+    # Set up root logger
+    setup_logging(
+        level=level,
+        log_file=log_file,
+        interactive_mode=interactive_mode
+    )
+    
+    # Configure external library loggers
+    configure_external_loggers()