changes

Alpha

@@ -1 +0,0 @@
-Subproject commit 959e8417af1dd5dba45ed870104ba1b5abacbe6b

app

@@ -1 +0,0 @@
-Subproject commit 9eff68fb1134d1f51c07aac9e0a88a77e60379c3

app.py

@@ -0,0 +1,283 @@
+import gradio as gr
+import uuid
+import logging
+from typing import List, Tuple
+from src.main import RAGSystem
+import asyncio
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Initialize the RAG system
+rag = RAGSystem()
+
+def create_session() -> str:
+    """Create a new session ID"""
+    return str(uuid.uuid4())
+
+def index_website(url: str, session_id: str) -> Tuple[bool, str]:
+    """Index a website for a given session"""
+    try:
+        result = rag.crawl_and_index(session_id, url)
+        if result["status"] == "success":
+            return True, f"Successfully indexed {len(result.get('urls_processed', []))} pages"
+        return False, result.get("message", "Unknown error during indexing")
+    except Exception as e:
+        logger.error(f"Indexing error: {str(e)}")
+        return False, f"Error during indexing: {str(e)}"
+
+def chat_response(
+    session_id: str,
+    message: str,
+    model_choice: str,
+    ollama_url: str,
+    gemini_api_key: str,
+    chat_history: List[Tuple[str, str]]
+) -> Tuple[List[Tuple[str, str]], str]:
+    """Generate a chat response with proper error handling"""
+    if not session_id:
+        chat_history.append(("🧑 " + message, "🤖 Please index a website first or enter a valid session ID"))
+        return chat_history, ""
+    
+    try:
+        response = asyncio.run(rag.chat(
+            session_id=session_id,
+            question=message,
+            model=model_choice.lower(),
+            ollama_url=ollama_url if model_choice == "mistral" else None,
+            gemini_api_key=gemini_api_key if model_choice == "gemini" else None
+        ))
+        
+        if response["status"] == "success":
+            answer = response["response"]
+            sources = "\n\nSources:\n" + "\n".join(
+                f"- {src['source_url']}" for src in response.get("sources", [])
+            ) if response.get("sources") else ""
+            full_response = answer + sources
+        else:
+            full_response = f"Error: {response.get('message', 'Unknown error')}"
+            
+        chat_history.append(("🧑 " + message, "🤖 " + full_response))
+        return chat_history, ""
+    except Exception as e:
+        logger.error(f"Chat error: {str(e)}")
+        chat_history.append(("🧑 " + message, f"🤖 System error: {str(e)}"))
+        return chat_history, ""
+
+def toggle_model_inputs(model_choice: str) -> List[gr.update]:
+    """Show/hide model-specific inputs"""
+    if model_choice == "mistral":
+        return [gr.update(visible=True), gr.update(visible=False)]
+    return [gr.update(visible=False), gr.update(visible=True)]
+
+def load_session(existing_session_id: str) -> Tuple[str, str]:
+    """Load an existing session"""
+    if existing_session_id:
+        # Here you might want to add validation if the session exists
+        return existing_session_id, f"Loaded existing session: {existing_session_id}"
+    return "", "Please enter a valid session ID"
+
+def get_session(self, session_id: str):
+    # If session exists in memory, return it
+    if session_id in self.sessions:
+        return self.sessions[session_id]
+    # If not, check if Qdrant collection exists and has documents
+    collection_name = self.get_collection_name(session_id)
+    try:
+        results = self.qdrant_client.scroll(collection_name=collection_name, limit=1)
+        if results and results[0]:
+            # Rehydrate session in memory
+            self.sessions[session_id] = {
+                "documents": [],  # Optionally, you can fetch all docs if needed
+                "history": []
+            }
+            return self.sessions[session_id]
+    except Exception as e:
+        logger.warning(f"Session {session_id} not found in Qdrant: {e}")
+    # If not found, return None or raise
+    raise ValueError("No documents indexed for this session")
+
+# Custom CSS for better styling
+custom_css = """
+.gradio-container {
+    max-width: 1200px !important;
+    margin: 0 auto !important;
+}
+.dark .gradio-container {
+    background: #1e1e2e !important;
+}
+#chatbot {
+    min-height: 500px;
+    border-radius: 12px !important;
+}
+.message.user {
+    border-left: 4px solid #4f46e5 !important;
+}
+.message.assistant {
+    border-left: 4px solid #10b981 !important;
+}
+.btn-primary {
+    background: linear-gradient(to right, #4f46e5, #7c3aed) !important;
+    border: none !important;
+}
+.btn-primary:hover {
+    background: linear-gradient(to right, #4338ca, #6d28d9) !important;
+}
+.prose {
+    max-width: 100% !important;
+}
+"""
+
+with gr.Blocks(title="RAG Chat with Mistral/Gemini", css=custom_css, theme="soft") as demo:
+    # Header section
+    with gr.Row():
+        gr.Markdown("""
+        # 🌐 RAG Chat Assistant
+        ### Chat with any website using Mistral or Gemini
+        """)
+    
+    # Session state
+    session_id = gr.State("")
+    
+    with gr.Tabs():
+        with gr.TabItem("📚 Index Website"):
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("### Step 1: Configure and Index")
+                    with gr.Group():
+                        url_input = gr.Textbox(
+                            label="Website URL to index",
+                            placeholder="https://example.com",
+                            interactive=True,
+                            lines=1
+                        )
+                        
+                        with gr.Row():
+                            model_choice = gr.Radio(
+                                choices=["mistral", "gemini"],
+                                label="Select Model",
+                                value="mistral",
+                                interactive=True
+                            )
+                            
+                            index_btn = gr.Button(
+                                "🚀 Index Website", 
+                                variant="primary",
+                                scale=0
+                            )
+                    
+                    with gr.Accordion("🔐 Model Settings", open=False):
+                        ollama_url = gr.Textbox(
+                            label="Ollama URL (required for Mistral)",
+                            placeholder="http://localhost:11434",
+                            visible=True
+                        )
+                        
+                        gemini_api_key = gr.Textbox(
+                            label="Gemini API Key (required for Gemini)",
+                            placeholder="your-api-key-here",
+                            visible=False,
+                            type="password"
+                        )
+                    
+                    status_output = gr.Textbox(
+                        label="Status",
+                        interactive=False,
+                        elem_classes="prose"
+                    )
+                    
+                    gr.Markdown("""
+                    **Instructions:**
+                    1. Enter a website URL
+                    2. Select your preferred model
+                    3. Configure model settings if needed
+                    4. Click 'Index Website'
+                    """)
+        
+        with gr.TabItem("💬 Chat"):
+            with gr.Row():
+                with gr.Column(scale=2):
+                    # New session ID input for resuming sessions
+                    with gr.Accordion("🔍 Resume Previous Session", open=False):
+                        existing_session_input = gr.Textbox(
+                            label="Enter existing Session ID",
+                            placeholder="Paste your session ID here...",
+                            interactive=True
+                        )
+                        load_session_btn = gr.Button(
+                            "🔁 Load Session",
+                            variant="secondary"
+                        )
+                        session_status = gr.Textbox(
+                            label="Session Status",
+                            interactive=False
+                        )
+                    
+                    chatbot = gr.Chatbot(
+                    label="Chat History",
+                    height=500,
+                    avatar_images=(None, None),
+                    show_copy_button=True,
+                    type="messages"  # Use OpenAI-style messages
+                )
+
+                with gr.Row():
+                    message_input = gr.Textbox(
+                        label="Type your message",
+                        placeholder="Ask about the website content...",
+                            interactive=True,
+                            container=False,
+                            scale=7,
+                            autofocus=True
+                        )
+                        
+                    send_btn = gr.Button(
+                            "Send", 
+                            variant="primary",
+                            scale=1,
+                            min_width=100
+                        )
+    
+    # Event handlers
+    model_choice.change(
+        fn=toggle_model_inputs,
+        inputs=model_choice,
+        outputs=[ollama_url, gemini_api_key]
+    )
+    
+    index_btn.click(
+        fn=create_session,
+        outputs=session_id
+    ).success(
+        fn=index_website,
+        inputs=[url_input, session_id],
+        outputs=[status_output]
+    )
+    
+    # New handler for loading existing sessions
+    load_session_btn.click(
+        fn=load_session,
+        inputs=[existing_session_input],
+        outputs=[session_id, session_status]
+    )
+    
+    send_btn.click(
+        fn=chat_response,
+        inputs=[session_id, message_input, model_choice, ollama_url, gemini_api_key, chatbot],
+        outputs=[chatbot, message_input]
+    )
+    
+    # Allow submitting with Enter key
+    message_input.submit(
+        fn=chat_response,
+        inputs=[session_id, message_input, model_choice, ollama_url, gemini_api_key, chatbot],
+        outputs=[chatbot, message_input]
+    )
+
+if __name__ == "__main__":
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        favicon_path="assets/favicon.ico"  # Local path, not URL
+    )

src/config.py

@@ -0,0 +1,56 @@
+from dotenv import load_dotenv
+import os
+import logging
+from pathlib import Path
+
+load_dotenv()
+
+# Logging configuration
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+class Config:
+    # Crawling and content configuration
+    MAX_PAGES_TO_CRAWL = int(os.getenv('MAX_PAGES_TO_CRAWL', 20))
+    MAX_LINKS_PER_PAGE = int(os.getenv('MAX_LINKS_PER_PAGE', 10))
+    MAX_CONTEXT_LENGTH = int(os.getenv('MAX_CONTEXT_LENGTH', 4000))
+    MAX_HISTORY_MESSAGES = int(os.getenv('MAX_HISTORY_MESSAGES', 20))
+    
+    # Model configuration
+    EMBEDDING_MODEL = os.getenv('EMBEDDING_MODEL', 'all-MiniLM-L6-v2')
+    OLLAMA_BASE_URL = os.getenv('OLLAMA_BASE_URL', 'http://localhost:11434')
+    OLLAMA_MODEL = os.getenv('OLLAMA_MODEL', 'mistral')
+    
+    # Path configuration
+    BASE_DIR = Path(__file__).parent.parent
+    SESSIONS_DIR = os.path.join(BASE_DIR, "sessions")
+    
+    # Qdrant configuration (new)
+    QDRANT_HOST = os.getenv('QDRANT_HOST', 'localhost')
+    QDRANT_PORT = int(os.getenv('QDRANT_PORT', 6333))
+    QDRANT_COLLECTION_PREFIX = os.getenv('QDRANT_COLLECTION_PREFIX', 'Chat-Bot')
+    QDRANT_URL = os.getenv('QDRANT_URL', 'https://6fe012ee-5a7c-4304-a77c-293a1888a9cf.us-west-2-0.aws.cloud.qdrant.io')
+    QDRANT_API_KEY = os.getenv('QDRANT_API_KEY', None)  # For cloud version
+    
+    # MongoDB configuration
+    MONGO_URI = os.getenv('MONGO_URI', "mongodb+srv://mehulxy21:[email protected]/")
+    DATABASE_NAME = os.getenv('DATABASE_NAME', "rag_chat_history")
+    HISTORY_COLLECTION = os.getenv('HISTORY_COLLECTION', "conversations")
+    
+    @staticmethod
+    def create_storage_dirs():
+        """Create necessary directories for storage"""
+        os.makedirs(Config.SESSIONS_DIR, exist_ok=True)
+        logger.info(f"Sessions directory created at: {Config.SESSIONS_DIR}")
+        
+        # Create data directory if it doesn't exist
+        data_dir = os.path.join(Config.BASE_DIR, "data")
+        os.makedirs(data_dir, exist_ok=True)
+        logger.info(f"Data directory created at: {data_dir}")
+
+        # Session Management
+SESSION_INACTIVITY_TIMEOUT = 3600  # 1 hour in seconds
+SESSION_CLEANUP_INTERVAL = 600  

src/crawler.py

@@ -0,0 +1,236 @@
+import requests
+from bs4 import BeautifulSoup
+from urllib.parse import urljoin, urlparse
+from typing import List, Set, Optional, Dict
+import logging
+import re
+from app.config import Config
+import aiohttp
+
+class URLCrawler:
+    """
+    A web crawler that extracts and processes content from websites.
+    Handles both synchronous and asynchronous crawling operations.
+    
+    Features:
+    - URL validation and sanitization
+    - Content extraction with noise removal
+    - Breadth-first crawling with configurable depth
+    - Respects robots.txt and avoids non-html content
+    """
+    
+    def __init__(self):
+        """Initialize the crawler with default settings."""
+        self.visited_urls: Set[str] = set()  # Tracks crawled URLs to avoid duplicates
+        self.logger = logging.getLogger(__name__)
+        # Configure headers to mimic a real browser
+        self.headers = {
+            'User-Agent': 'Mozilla/5.0 (compatible; RAGBot/1.0)',
+            'Accept-Language': 'en-US,en;q=0.9'
+        }
+
+    def is_valid_url(self, url: str, base_domain: str) -> bool:
+        """
+        Validate if a URL should be crawled.
+        
+        Args:
+            url: URL to validate
+            base_domain: The target domain to stay within
+            
+        Returns:
+            bool: True if URL is crawlable
+        """
+        parsed = urlparse(url)
+        return (parsed.scheme in ('http', 'https') and  # Only HTTP/HTTPS
+                parsed.netloc == base_domain and  # Stay within target domain
+                not any(ext in url.lower()  # Skip binary files
+                       for ext in ['.pdf', '.jpg', '.png', '.zip']) and
+                url not in self.visited_urls)  # Avoid duplicates
+
+    def sanitize_url(self, url: str) -> str:
+        """
+        Normalize URL by removing fragments and query parameters.
+        
+        Args:
+            url: URL to sanitize
+            
+        Returns:
+            str: Normalized URL
+        """
+        parsed = urlparse(url)
+        return f"{parsed.scheme}://{parsed.netloc}{parsed.path.rstrip('/')}"
+
+    def clean_text(self, text: str) -> str:
+        """
+        Clean and normalize extracted text content.
+        
+        Args:
+            text: Raw extracted text
+            
+        Returns:
+            str: Cleaned text content
+        """
+        # Remove excessive whitespace
+        text = re.sub(r'\s+', ' ', text)
+        # Remove common boilerplate
+        text = re.sub(r'(\b(privacy policy|terms of service|cookie policy)\b|\b\d+\s*(comments|shares|likes)\b)', '', text, flags=re.I)
+        # Remove short lines (likely not meaningful content)
+        return '\n'.join(line for line in text.split('\n') 
+                        if len(line.strip()) > 30)
+
+    def extract_main_content(self, soup: BeautifulSoup) -> str:
+        """
+        Extract primary content from HTML using semantic heuristics.
+        
+        Args:
+            soup: BeautifulSoup parsed HTML document
+            
+        Returns:
+            str: Extracted main content
+        """
+        # Remove unwanted elements that typically don't contain main content
+        for element in soup(['script', 'style', 'nav', 'footer', 
+                           'header', 'iframe', 'aside', 'form']):
+            element.decompose()
+
+        # Prioritize semantic HTML containers that likely contain main content
+        for tag in ['article', 'main', 'section[role="main"]', '.content']:
+            content = soup.select_one(tag)
+            if content:
+                return self.clean_text(content.get_text(separator='\n'))
+        
+        # Fallback to body if no semantic containers found
+        return self.clean_text(soup.body.get_text(separator='\n'))
+
+    def get_page_content(self, url: str) -> Optional[Dict]:
+        """
+        Fetch and process a single web page.
+        
+        Args:
+            url: URL to fetch
+            
+        Returns:
+            Optional[Dict]: Structured page data or None if invalid
+        """
+        try:
+            response = requests.get(url, headers=self.headers, timeout=15)
+            response.raise_for_status()
+            
+            # Skip non-HTML content
+            if 'text/html' not in response.headers.get('Content-Type', ''):
+                return None
+
+            soup = BeautifulSoup(response.text, 'lxml')
+            title = soup.title.string if soup.title else urlparse(url).path
+            content = self.extract_main_content(soup)
+            
+            # Skip pages with insufficient content
+            if len(content.split()) < 100:  # Minimum 100 words
+                return None
+                
+            return {
+                'url': url,
+                'title': title,
+                'content': content,
+                'last_modified': response.headers.get('Last-Modified', '')
+            }
+            
+        except Exception as e:
+            self.logger.warning(f"Error processing {url}: {str(e)}")
+            return None
+
+    def extract_links(self, url: str, soup: BeautifulSoup) -> List[str]:
+        """
+        Extract all crawlable links from a page.
+        
+        Args:
+            url: Base URL for relative link resolution
+            soup: Parsed HTML document
+            
+        Returns:
+            List[str]: List of absolute URLs to crawl
+        """
+        base_domain = urlparse(url).netloc
+        links = set()
+        
+        for link in soup.find_all('a', href=True):
+            href = link['href'].split('#')[0]  # Remove fragments
+            if not href or href.startswith('javascript:'):
+                continue
+                
+            absolute_url = urljoin(url, href)
+            sanitized_url = self.sanitize_url(absolute_url)
+            
+            if self.is_valid_url(sanitized_url, base_domain):
+                links.add(sanitized_url)
+        
+        return sorted(links)[:Config.MAX_LINKS_PER_PAGE]  # Apply limit
+
+    async def crawl(self, url: str) -> str:
+        """
+        Asynchronously crawl a single URL and return its text content.
+        
+        Args:
+            url: URL to crawl
+            
+        Returns:
+            str: Extracted text content
+            
+        Raises:
+            Exception: If crawling fails
+        """
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(url) as response:
+                    html = await response.text()
+                    soup = BeautifulSoup(html, 'html.parser')
+                    # Remove script and style elements
+                    for script in soup(["script", "style"]):
+                        script.decompose()
+                    return soup.get_text()
+        except Exception as e:
+            self.logger.error(f"Crawling error: {str(e)}")
+            raise
+
+    def crawl_sync(self, start_url: str, max_pages: int = Config.MAX_PAGES_TO_CRAWL) -> List[Dict]:
+        """
+        Synchronously crawl a website using breadth-first search.
+        
+        Args:
+            start_url: Initial URL to begin crawling
+            max_pages: Maximum number of pages to crawl
+            
+        Returns:
+            List[Dict]: Structured documents from crawled pages
+        """
+        base_domain = urlparse(start_url).netloc
+        queue = [start_url]  # URLs to crawl
+        documents = []  # Collected documents
+        
+        while queue and len(documents) < max_pages:
+            current_url = queue.pop(0)
+            sanitized_url = self.sanitize_url(current_url)
+            
+            if sanitized_url in self.visited_urls:
+                continue
+                
+            self.visited_urls.add(sanitized_url)
+            self.logger.info(f"Crawling: {sanitized_url}")
+            
+            page_data = self.get_page_content(sanitized_url)
+            if not page_data:
+                continue
+                
+            documents.append(page_data)
+            
+            # Get links for further crawling
+            try:
+                response = requests.get(sanitized_url, headers=self.headers, timeout=10)
+                soup = BeautifulSoup(response.text, 'lxml')
+                new_links = self.extract_links(sanitized_url, soup)
+                queue.extend(link for link in new_links 
+                            if link not in self.visited_urls)
+            except Exception as e:
+                self.logger.warning(f"Error getting links from {sanitized_url}: {str(e)}")
+        
+        return documents

src/databases/models.py

@@ -0,0 +1,10 @@
+from datetime import datetime
+from typing import Optional
+from pydantic import BaseModel
+
+class Conversation(BaseModel):
+    session_id: str
+    user_query: str
+    bot_response: str
+    timestamp: Optional[datetime] = datetime.utcnow()
+    metadata: Optional[dict] = None

src/databases/mongo_handler.py

@@ -0,0 +1,133 @@
+from pymongo import MongoClient
+from pymongo.errors import ServerSelectionTimeoutError, AutoReconnect
+import logging
+from datetime import datetime
+from app.config import Config
+from typing import List, Dict, Optional
+
+class MongoDBHandler:
+    def __init__(self):
+        self.logger = logging.getLogger(__name__)
+        self._ensure_mongodb_running()
+        try:
+            self.client = MongoClient(
+                Config.MONGO_URI,
+                serverSelectionTimeoutMS=5000
+            )
+            # Test connection immediately
+            self.client.server_info()
+            self.db = self.client[Config.DATABASE_NAME]
+            self.collection = self.db[Config.HISTORY_COLLECTION]
+            self._create_indexes()
+            self.logger.info("MongoDB connection established successfully")
+        except (ServerSelectionTimeoutError, AutoReconnect) as e:
+            self.logger.error(f"MongoDB connection failed: {e}")
+            self._diagnose_connection_issue()
+            raise
+
+    def _ensure_mongodb_running(self):
+        """Ensure MongoDB service is running"""
+        try:
+            import subprocess
+            result = subprocess.run(
+                ['net', 'start', 'MongoDB'],
+                capture_output=True,
+                text=True
+            )
+            if "already running" not in result.stderr.lower():
+                self.logger.info("Started MongoDB service")
+        except Exception as e:
+            self.logger.error(f"Failed to start MongoDB: {e}")
+
+    def _diagnose_connection_issue(self):
+        """Diagnose common MongoDB connection issues"""
+        import os
+        issues = []
+        
+        # Check data directory
+        if not os.path.exists("C:\\data\\db"):
+            issues.append("Data directory missing")
+        
+        # Check log directory
+        if not os.path.exists("C:\\data\\log"):
+            issues.append("Log directory missing")
+        
+        # Check service status
+        try:
+            import subprocess
+            result = subprocess.run(
+                ['sc', 'query', 'MongoDB'],
+                capture_output=True,
+                text=True
+            )
+            if "RUNNING" not in result.stdout:
+                issues.append("MongoDB service not running")
+        except Exception:
+            issues.append("Could not check service status")
+        
+        if issues:
+            self.logger.error("MongoDB Issues Found:")
+            for issue in issues:
+                self.logger.error(f" - {issue}")
+
+    def _create_indexes(self):
+        """Create indexes for better query performance"""
+        self.collection.create_index([("session_id", 1)])
+        self.collection.create_index([("timestamp", -1)])
+        self.collection.create_index([("session_id", 1), ("timestamp", -1)])
+
+    def save_conversation(self, session_id: str, query: str, response: str, metadata: dict = None) -> str:
+        """Save a conversation with automatic timestamp"""
+        conversation = {
+            "session_id": session_id,
+            "user_query": query,
+            "bot_response": response,
+            "timestamp": datetime.utcnow(),
+            "metadata": metadata or {}
+        }
+        result = self.collection.insert_one(conversation)
+        return str(result.inserted_id)
+
+    def verify_storage(self) -> bool:
+        """Verify storage is working by inserting and retrieving a test document"""
+        try:
+            # Insert test document
+            test_id = self.save_conversation(
+                session_id="test",
+                query="test_query",
+                response="test_response",
+                metadata={"test": True}
+            )
+            
+            # Verify retrieval
+            test_doc = self.collection.find_one({"_id": test_id})
+            
+            # Cleanup test document
+            self.collection.delete_one({"_id": test_id})
+            
+            return test_doc is not None
+        except Exception as e:
+            self.logger.error(f"Storage verification failed: {e}")
+            return False
+
+    def get_conversation_history(self, session_id: str, limit: int = 10) -> List[Dict]:
+        """Retrieve conversation history for a session"""
+        try:
+            cursor = self.collection.find(
+                {"session_id": session_id},
+                {"_id": 0}  # Exclude _id field
+            ).sort("timestamp", -1).limit(limit)
+            
+            return list(cursor)
+        except Exception as e:
+            self.logger.error(f"Error retrieving conversation history: {e}")
+            return []
+
+    def clear_session_history(self, session_id: str) -> int:
+        """Clear all conversations for a session"""
+        try:
+            result = self.collection.delete_many({"session_id": session_id})
+            return result.deleted_count
+        except Exception as e:
+            self.logger.error(f"Error clearing session history: {e}")
+            return 0

src/databases/storage.py

@@ -0,0 +1,29 @@
+from app.databases.mongo_handler import MongoDBHandler
+from app.main import SessionManager
+import logging
+
+class StorageService:
+    def __init__(self):
+        self.logger = logging.getLogger(__name__)
+        self.session_manager = SessionManager()
+        self.mongo_handler = MongoDBHandler()
+
+    async def add_conversation(self, session_id: str, query: str, response: str, sources: list = None):
+        """Save conversation to both session manager and MongoDB"""
+        try:
+            # Save to session manager (memory)
+            self.session_manager.add_conversation(session_id, query, response)
+            
+            # Save to MongoDB (persistent)
+            mongo_id = self.mongo_handler.save_conversation(
+                session_id=session_id,
+                query=query,
+                response=response,
+                metadata={"sources": sources} if sources else {}
+            )
+            
+            self.logger.debug(f"Conversation saved to MongoDB with ID: {mongo_id}")
+            return True
+        except Exception as e:
+            self.logger.error(f"Failed to save conversation: {e}")
+            return False

src/databases/test_mongo.py

@@ -0,0 +1,5 @@
+from pymongo import MongoClient
+
+client = MongoClient("mongodb+srv://mehulxy21:[email protected]/")
+db = client["rag_chat_history"]
+print("Databases:", client.list_database_names())

src/embeddings.py

@@ -0,0 +1,218 @@
+import os
+from sentence_transformers import SentenceTransformer
+import numpy as np
+import logging
+from typing import List, Dict, Optional
+from app.config import Config
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PointStruct
+from qdrant_client.http.exceptions import UnexpectedResponse
+
+class EmbeddingHandler:
+    """
+    Handles all embedding-related operations including:
+    - Text embedding generation using SentenceTransformers
+    - Vector storage and retrieval with Qdrant
+    - Collection management for vector storage
+    
+    This serves as the central component for vector operations in the RAG system.
+    """
+    
+    def __init__(self):
+        """Initialize the embedding handler with model and vector store client."""
+        self.logger = logging.getLogger(__name__)
+        try:
+            # Initialize embedding model with configuration from Config
+            self.model = SentenceTransformer(Config.EMBEDDING_MODEL)
+            # Get embedding dimension from the model
+            self.embedding_dim = self.model.get_sentence_embedding_dimension()
+
+            # Initialize Qdrant client with configuration from Config
+            self.qdrant_client = QdrantClient(
+                url=Config.QDRANT_URL,
+                api_key=Config.QDRANT_API_KEY,
+                prefer_grpc=False,  # HTTP preferred over gRPC for compatibility
+                timeout=30  # Connection timeout in seconds
+            )
+            
+            # Connection test can be uncommented for local development
+            # self._verify_connection()
+
+        except Exception as e:
+            self.logger.error(f"Error initializing embedding handler: {str(e)}", exc_info=True)
+            raise RuntimeError("Failed to initialize embedding handler") from e
+
+    def generate_embeddings(self, texts: List[str]) -> np.ndarray:
+        """
+        Generate embeddings for a list of text strings.
+        
+        Args:
+            texts: List of text strings to embed
+            
+        Returns:
+            np.ndarray: Array of embeddings (2D numpy array)
+            
+        Raises:
+            Exception: If embedding generation fails
+        """
+        try:
+            return self.model.encode(
+                texts,
+                show_progress_bar=True,  # Visual progress indicator
+                batch_size=32,  # Optimal batch size for most GPUs
+                convert_to_numpy=True  # Return as numpy array for efficiency
+            )
+        except Exception as e:
+            self.logger.error(f"Error generating embeddings: {str(e)}", exc_info=True)
+            raise
+
+    def create_collection(self, collection_name: str) -> bool:
+        """
+        Create a new Qdrant collection for storing vectors.
+        
+        Args:
+            collection_name: Name of the collection to create
+            
+        Returns:
+            bool: True if collection was created or already exists
+            
+        Raises:
+            Exception: If collection creation fails (except for already exists case)
+        """
+        try:
+            self.qdrant_client.create_collection(
+                collection_name=collection_name,
+                vectors_config=VectorParams(
+                    size=self.embedding_dim,  # Must match model's embedding dimension
+                    distance=Distance.COSINE  # Using cosine similarity
+                )
+            )
+            self.logger.info(f"Created collection {collection_name}")
+            return True
+
+        except UnexpectedResponse as e:
+            # Handle case where collection already exists
+            if "already exists" in str(e):
+                self.logger.info(f"Collection {collection_name} already exists")
+                return True
+            else:
+                self.logger.error(f"Error creating collection: {e}")
+                raise
+        except Exception as e:
+            self.logger.error(f"Error creating collection: {str(e)}", exc_info=True)
+            raise
+
+    def add_to_collection(self, collection_name: str, embeddings: np.ndarray, payloads: List[dict]) -> bool:
+        """
+        Add embeddings and associated metadata to a Qdrant collection.
+        
+        Args:
+            collection_name: Target collection name
+            embeddings: Numpy array of embeddings to add
+            payloads: List of metadata dictionaries corresponding to each embedding
+            
+        Returns:
+            bool: True if operation succeeded
+            
+        Raises:
+            Exception: If operation fails
+        """
+        try:
+            # Convert numpy arrays to lists for Qdrant compatibility
+            if isinstance(embeddings, np.ndarray):
+                embeddings = embeddings.tolist()
+
+            # Prepare points in batches for efficient processing
+            batch_size = 100  # Optimal batch size for Qdrant Cloud
+            points = [
+                PointStruct(
+                    id=idx,  # Sequential ID
+                    vector=embedding,
+                    payload=payload  # Associated metadata
+                )
+                for idx, (embedding, payload) in enumerate(zip(embeddings, payloads))
+            ]
+
+            # Process in batches to avoid overwhelming the server
+            for i in range(0, len(points), batch_size):
+                batch = points[i:i + batch_size]
+                self.qdrant_client.upsert(
+                    collection_name=collection_name,
+                    points=batch,
+                    wait=True  # Ensure immediate persistence
+                )
+
+            self.logger.info(f"Added {len(points)} vectors to collection {collection_name}")
+            return True
+
+        except Exception as e:
+            self.logger.error(f"Error adding to collection: {str(e)}", exc_info=True)
+            raise
+
+    async def search_collection(self, collection_name: str, query: str, k: int = 5) -> Dict:
+        """
+        Search a Qdrant collection for similar vectors to the query.
+        
+        Args:
+            collection_name: Name of collection to search
+            query: Text query to search for
+            k: Number of similar results to return (default: 5)
+            
+        Returns:
+            Dict: {
+                "status": "success"|"error",
+                "results": List[Dict] (if success),
+                "message": str (if error)
+            }
+        """
+        try:
+            # Generate embedding for the query text
+            query_embedding = self.model.encode(query).tolist()
+
+            # Perform similarity search in Qdrant
+            results = self.qdrant_client.search(
+                collection_name=collection_name,
+                query_vector=query_embedding,
+                limit=k,  # Number of results to return
+                with_payload=True,  # Include metadata
+                with_vectors=False  # Exclude raw vectors to save bandwidth
+            )
+
+            # Format results for consistent API response
+            formatted_results = []
+            for hit in results:
+                formatted_results.append({
+                    "id": hit.id,
+                    "score": float(hit.score),  # Similarity score
+                    "payload": hit.payload or {},  # Associated metadata
+                    "text": hit.payload.get("text", "") if hit.payload else ""  # Extracted text
+                })
+
+            return {
+                "status": "success",
+                "results": formatted_results
+            }
+
+        except Exception as e:
+            self.logger.error(f"Search error: {str(e)}", exc_info=True)
+            return {
+                "status": "error",
+                "message": str(e),
+                "results": []
+            }
+
+    # Deprecated FAISS methods (maintained for backward compatibility)
+    def create_faiss_index(self, *args, **kwargs):
+        """Deprecated method - FAISS support has been replaced by Qdrant."""
+        self.logger.warning("FAISS operations are deprecated")
+        raise NotImplementedError("Use Qdrant collections instead of FAISS")
+
+    def save_index(self, *args, **kwargs):
+        """Deprecated method - Qdrant persists data automatically."""
+        self.logger.warning("FAISS operations are deprecated")
+        raise NotImplementedError("Qdrant persists data automatically")
+
+    def load_index(self, *args, **kwargs):
+        """Deprecated method - Access Qdrant collections directly."""
+        self.logger.warning("FAISS operations are deprecated")
+        raise NotImplementedError("Access Qdrant collections directly")

src/handlers.py

@@ -0,0 +1,121 @@
+import json
+import logging
+from fastapi import WebSocket, WebSocketDisconnect
+from pydantic import BaseModel
+from typing import Optional, Dict
+from app.rag import RAGSystem
+
+# Configure logging
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
+class ConnectionManager:
+    """
+    Manages active WebSocket connections.
+    Tracks connected clients and handles connection/disconnection events.
+    """
+    
+    def __init__(self):
+        # Dictionary to store active WebSocket connections by session_id
+        self.active_connections: Dict[str, WebSocket] = {}
+
+    async def connect(self, websocket: WebSocket, session_id: str):
+        """
+        Accept a new WebSocket connection and track it.
+        
+        Args:
+            websocket: The WebSocket connection object
+            session_id: Unique identifier for the session
+        """
+        await websocket.accept()
+        self.active_connections[session_id] = websocket
+        logger.info(f"WebSocket connected: {session_id}")
+
+    def disconnect(self, session_id: str):
+        """
+        Remove a WebSocket connection from active connections.
+        
+        Args:
+            session_id: The session ID to disconnect
+        """
+        self.active_connections.pop(session_id, None)
+        logger.info(f"WebSocket disconnected: {session_id}")
+
+# Initialize system components
+manager = ConnectionManager()  # Manages WebSocket connections
+rag = RAGSystem()  # The RAG processing system
+
+class ResponseFormatter:
+    """
+    Formats responses before sending to clients.
+    Can be extended to standardize response structure.
+    """
+    
+    def __init__(self):
+        pass
+
+    def format_response(self, response: dict) -> dict:
+        """
+        Format a response dictionary.
+        
+        Args:
+            response: Raw response dictionary
+            
+        Returns:
+            dict: Formatted response
+        """
+        return response  # Currently passes through unchanged
+
+formatter = ResponseFormatter()  # Create formatter instance
+
+class ChatMessage(BaseModel):
+    """
+    Pydantic model for validating incoming chat messages.
+    Ensures proper message structure.
+    """
+    text: str  # The message text/content
+    url: Optional[str] = None  # Optional URL for context
+
+async def websocket_endpoint(websocket: WebSocket, session_id: str):
+    """
+    WebSocket endpoint for handling real-time chat interactions.
+    
+    Args:
+        websocket: The WebSocket connection
+        session_id: Unique identifier for the chat session
+        
+    Handles:
+        - Connection management
+        - Message processing
+        - Error handling
+    """
+    # Register the new connection
+    await manager.connect(websocket, session_id)
+    
+    try:
+        while True:
+            # Wait for and receive incoming message
+            data = await websocket.receive_json()
+            logger.info(f"Received message: {data}")
+            
+            # Parse and validate message using Pydantic model
+            message = ChatMessage(**data)
+            
+            # Process the message through RAG system
+            response = await rag.process_query(message.text, session_id)
+            
+            # Log and send the response
+            logger.info(f"Sending response: {response}")
+            await websocket.send_json(response)
+
+    except WebSocketDisconnect:
+        # Handle graceful disconnection
+        manager.disconnect(session_id)
+        logger.info(f"Client disconnected: {session_id}")
+    except Exception as e:
+        # Handle other errors and notify client
+        logger.error(f"Error in websocket: {str(e)}")
+        await websocket.send_json({
+            "status": "error",
+            "message": str(e)
+        })

src/llm.py

@@ -0,0 +1,220 @@
+import logging
+import ollama
+from typing import List, Dict, Optional
+from app.config import Config
+import os
+
+
+class OllamaMistral:
+    """
+    A class to interact with the Ollama API for Mistral model.
+    Handles both chat completions and embeddings generation.
+    """
+    
+    def __init__(self):
+        """Initialize the Ollama Mistral client with default settings."""
+        self.logger = logging.getLogger(__name__)
+        # Initialize Ollama client with default host
+        self.client = ollama.Client(host='http://localhost:11434')
+        self.model = 'mistral'  # Default model name
+
+    async def generate_response(self, prompt: str) -> str:
+        """
+        Asynchronously generate a text response from Mistral model.
+        
+        Args:
+            prompt: The input text prompt for the model
+            
+        Returns:
+            Generated response text or error message if failed
+        """
+        try:
+            print(f"[Ollama] Sending prompt:\n{prompt}\n")
+            # Send chat request to Ollama API
+            response = self.client.chat(
+                model=self.model,
+                messages=[{
+                    'role': 'user',
+                    'content': prompt
+                }]
+            )
+            print(f"[Ollama] Received response:\n{response}\n")
+
+            # Handle different response formats from Ollama
+            if isinstance(response, dict):
+                if 'message' in response and 'content' in response['message']:
+                    return response['message']['content']
+            elif hasattr(response, 'message') and hasattr(response.message, 'content'):
+                return response.message.content
+            # Fallback: try to convert to string
+            return str(response)
+
+        except Exception as e:
+            self.logger.error(f"[OllamaMistral] Error generating response: {str(e)}", exc_info=True)
+            return f"Error generating response: {str(e)}"
+
+    def generate_embedding(self, text: str, model: str = Config.OLLAMA_MODEL) -> Optional[List[float]]:
+        """
+        Generate embeddings for the input text using specified model.
+        
+        Args:
+            text: Input text to generate embeddings for
+            model: Model name to use for embeddings (default from Config)
+            
+        Returns:
+            List of embeddings or None if failed
+        """
+        try:
+            print(f"[Ollama] Generating embedding for: {text[:60]}...")
+            # Request embeddings from Ollama API
+            response = self.client.embeddings(
+                model=model,
+                prompts=[text]  # prompts must be a list of strings
+            )
+            print(f"[Ollama] Embedding response: {response}")
+
+            # Handle different response formats
+            if isinstance(response, dict) and 'embeddings' in response:
+                return response['embeddings'][0]
+            elif isinstance(response, dict) and 'embedding' in response:
+                return response['embedding']
+            else:
+                self.logger.warning(f"Unexpected embedding response format: {response}")
+                return None
+
+        except Exception as e:
+            self.logger.error(f"[OllamaMistral] Error generating embedding: {str(e)}", exc_info=True)
+            return None
+
+    def generate(self, prompt: str) -> str:
+        """
+        Synchronous wrapper for generate_response.
+        
+        Args:
+            prompt: Input text prompt
+            
+        Returns:
+            Generated response text
+        """
+        import asyncio
+        try:
+            return asyncio.run(self.generate_response(prompt))
+        except Exception as e:
+            self.logger.error(f"Error in synchronous generate: {e}")
+            return f"Error generating response: {str(e)}"
+
+
+class GeminiProvider:
+    """
+    A class to interact with Google's Gemini API.
+    Requires GEMINI_API_KEY environment variable.
+    """
+    
+    def __init__(self):
+        """Initialize Gemini provider with API key."""
+        self.logger = logging.getLogger(__name__)
+        self.api_key = os.getenv('GEMINI_API_KEY')
+        if not self.api_key:
+            raise ValueError("GEMINI_API_KEY environment variable is required for Gemini provider")
+        
+        try:
+            import google.generativeai as genai
+            # Configure Gemini API
+            genai.configure(api_key=self.api_key)
+            self.model = genai.GenerativeModel('gemini-1.5-flash')
+        except ImportError:
+            raise ImportError("google-generativeai package is required for Gemini provider")
+
+    def generate(self, prompt: str) -> str:
+        """
+        Generate text response using Gemini model.
+        
+        Args:
+            prompt: Input text prompt
+            
+        Returns:
+            Generated response text or error message
+        """
+        try:
+            response = self.model.generate_content(prompt)
+            return response.text
+        except Exception as e:
+            self.logger.error(f"[Gemini] Error generating response: {str(e)}")
+            return f"Error generating response: {str(e)}"
+
+
+class OpenChatProvider:
+    """
+    A class to use OpenChat models locally via transformers.
+    Requires transformers package to be installed.
+    """
+    
+    def __init__(self):
+        """Initialize OpenChat model and tokenizer."""
+        self.logger = logging.getLogger(__name__)
+        try:
+            from transformers import AutoTokenizer, AutoModelForCausalLM
+            # Load pretrained OpenChat model
+            self.tokenizer = AutoTokenizer.from_pretrained("openchat/openchat-3.5-0106")
+            self.model = AutoModelForCausalLM.from_pretrained("openchat/openchat-3.5-0106")
+        except ImportError:
+            raise ImportError("transformers package is required for OpenChat provider")
+
+    def generate(self, prompt: str) -> str:
+        """
+        Generate text response using OpenChat model.
+        
+        Args:
+            prompt: Input text prompt
+            
+        Returns:
+            Generated response text
+        """
+        try:
+            # Tokenize input and generate response
+            inputs = self.tokenizer(prompt, return_tensors="pt")
+            outputs = self.model.generate(**inputs, max_length=512, temperature=0.7)
+            response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
+            return response
+        except Exception as e:
+            self.logger.error(f"[OpenChat] Error generating response: {str(e)}")
+            return f"Error generating response: {str(e)}"
+
+
+class LLMFactory:
+    """
+    Factory class to create and manage different LLM providers.
+    Implements the Factory design pattern for LLM provider instantiation.
+    """
+    
+    @staticmethod
+    def get_provider(model_name: Optional[str] = None) -> any:
+        """
+        Get appropriate LLM provider based on model name.
+        
+        Args:
+            model_name: Name of the model ('mistral', 'gemini', 'openchat')
+                      Defaults to 'mistral' if None or unknown
+            
+        Returns:
+            Instance of the requested LLM provider
+            
+        Raises:
+            ValueError: If required dependencies are missing for the provider
+        """
+        if model_name is None:
+            model_name = "mistral"  # Default to mistral
+            
+        model_name = model_name.lower()
+        
+        # Return appropriate provider based on model name
+        if model_name == "mistral":
+            return OllamaMistral()
+        elif model_name == "gemini":
+            return GeminiProvider()
+        elif model_name == "openchat":
+            return OpenChatProvider()
+        else:
+            # Default to mistral if unknown model is specified
+            logging.warning(f"Unknown model '{model_name}', defaulting to mistral")
+            return OllamaMistral()

src/main.py

@@ -0,0 +1,535 @@
+import logging
+import numpy as np
+import uuid
+import os
+from datetime import datetime
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from typing import Optional, List, Dict, Any
+from bs4 import BeautifulSoup
+import requests
+from urllib.parse import urljoin, urlparse
+import re
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PointStruct
+from app.prompts.templates import rag_prompt_template
+from sentence_transformers import SentenceTransformer
+from langchain_ollama import OllamaLLM
+import json
+import asyncio
+from app.llm import GeminiProvider
+from app.config import Config
+from qdrant_client.http.exceptions import UnexpectedResponse
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+# Configuration
+class Config:
+    """
+    Application configuration settings.
+    Contains constants for storage, models, and Qdrant connection.
+    """
+    STORAGE_DIR = "data/qdrant_storage"
+    EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
+    url="https://6fe012ee-5a7c-4304-a77c-293a1888a9cf.us-west-2-0.aws.cloud.qdrant.io"
+    QDRANT_API_KEY = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2Nlc3MiOiJtIn0.NUKB9m360LPEBTnpdo2TJpJmEIttumHLz-9ZbAUBKIM"
+    QDRANT_COLLECTION_NAME = "Chat-Bot"
+    
+    @staticmethod
+    def create_storage_dir():
+        """Ensure storage directory exists"""
+        os.makedirs(Config.STORAGE_DIR, exist_ok=True)
+
+# Data classes
+class Document:
+    """Represents a document with text content and metadata"""
+    def __init__(self, text: str, metadata: Dict[str, Any]):
+        self.text = text
+        self.metadata = metadata
+
+# Session Manager
+class SessionManager:
+    """
+    Manages user sessions and Qdrant collections.
+    Handles session state, document storage, and conversation history.
+    """
+    
+    def __init__(self):
+        """Initialize with in-memory sessions and Qdrant connection"""
+        self.sessions = {}  # In-memory session store
+        self.embedding_model = SentenceTransformer(Config.EMBEDDING_MODEL)
+        self.qdrant_client = QdrantClient(
+            url=Config.url,
+            api_key=Config.QDRANT_API_KEY,
+            timeout=30
+        )
+
+    def get_collection_name(self, session_id: str) -> str:
+        """Generate standardized Qdrant collection name for a session"""
+        return f"collection_{session_id}"
+
+    def get_session(self, session_id: str) -> Dict:
+        """
+        Get or create session with the given ID.
+        Maintains original interface while adding robustness.
+        """
+        if session_id not in self.sessions:
+            self._initialize_new_session(session_id)
+            print(f"[SessionManager] Created new session: {session_id}")
+        return self.sessions[session_id]
+    
+
+    def _initialize_new_session(self, session_id: str):
+        """Internal method to handle new session creation"""
+        self.sessions[session_id] = {
+            'documents': [],
+            'history': []
+        }
+        self._ensure_qdrant_collection(session_id)
+        print(f"[SessionManager] Initialized session {session_id} with Qdrant collection.")
+
+    def _ensure_qdrant_collection(self, session_id: str):
+        """Ensure Qdrant collection exists for the session"""
+        collection_name = self.get_collection_name(session_id)
+        try:
+            # First try to get the collection (might already exist)
+            self.qdrant_client.get_collection(collection_name)
+            logger.debug(f"Using existing Qdrant collection: {collection_name}")
+        except Exception:
+            # Collection doesn't exist, create it
+            try:
+                self.qdrant_client.create_collection(
+                    collection_name=collection_name,
+                    vectors_config=VectorParams(
+                        size=self.embedding_model.get_sentence_embedding_dimension(),
+                        distance=Distance.COSINE
+                    )
+                )
+                logger.info(f"Created new Qdrant collection: {collection_name}")
+            except UnexpectedResponse as e:
+                if "already exists" in str(e):
+                    logger.debug(f"Collection already exists: {collection_name}")
+                else:
+                    logger.error(f"Error creating collection: {e}")
+                    raise
+            except Exception as e:
+                logger.error(f"Unexpected error ensuring collection: {e}")
+                raise
+
+    def add_to_history(self, session_id: str, question: str, answer: str):
+        """Add conversation to session history"""
+        if session_id not in self.sessions:
+            logger.warning(f"Session {session_id} not found when adding history")
+            return
+            
+        self.sessions[session_id]['history'].append({
+            'question': question,
+            'answer': answer,
+            'timestamp': datetime.now().isoformat()
+        })
+
+    def get_history(self, session_id: str, limit: Optional[int] = None) -> List[Dict]:
+        """Get conversation history with optional limit"""
+        if session_id not in self.sessions:
+            logger.warning(f"Session {session_id} not found when getting history")
+            return []
+            
+        history = self.sessions[session_id]['history']
+        return history[-limit:] if limit else history
+
+    def session_exists(self, session_id: str) -> bool:
+        """Check if session exists"""
+        if session_id in self.sessions:
+            return True
+        collection_name = self.get_collection_name(session_id)
+        try:
+            self.qdrant_client.get_collection(collection_name)
+            # Add to sessions if collection exists
+            self.sessions[session_id] = {
+                'documents': [],
+                'history': []
+            }
+            return True
+        except Exception:
+            return False
+
+    def cleanup_inactive_sessions(self, inactive_minutes: int = 60):
+        """Clean up sessions inactive for specified minutes"""
+        current_time = datetime.now()
+        for session_id in list(self.sessions.keys()):
+            history = self.sessions[session_id]['history']
+            if history:
+                last_activity = datetime.fromisoformat(history[-1]['timestamp'])
+                if (current_time - last_activity).total_seconds() > inactive_minutes * 60:
+                    del self.sessions[session_id]
+                    logger.info(f"Cleaned up inactive session: {session_id}")
+
+    def save_session(self, session_id: str):
+        """Qdrant persists data automatically"""
+        pass
+
+    def add_conversation(self, session_id: str, query: str, response: str):
+        """Add conversation to session history"""
+        self.sessions[session_id]['history'].append({"query": query, "response": response})
+
+    def get_conversation_history(self, session_id: str):
+        """Get full conversation history"""
+        return self.sessions[session_id]['history']
+
+    def add_documents_to_qdrant(self, session_id: str, documents: List[Document]):
+        """Add documents to Qdrant collection with validation"""
+        texts = [doc.text for doc in documents]
+        try:
+            embeddings = self.embedding_model.encode(texts, batch_size=32, show_progress_bar=True)
+            if isinstance(embeddings, np.ndarray):
+                embeddings = embeddings.tolist()
+            points = [
+                PointStruct(
+                    id=idx,
+                    vector=embedding,
+                    payload={
+                        "text": doc.text,
+                        "metadata": doc.metadata
+                    }
+                )
+                for idx, (embedding, doc) in enumerate(zip(embeddings, documents))
+            ]
+            collection_name = self.get_collection_name(session_id)
+            operation_info = self.qdrant_client.upsert(
+                collection_name=collection_name,
+                points=points,
+                wait=True  # Wait for operation confirmation
+            )
+            logger.info(f"Upsert operation status: {operation_info.status}")
+            self.sessions[session_id]['documents'].extend(documents)
+        except Exception as e:
+            logger.error(f"Document insertion failed: {e}")
+            raise
+
+    def search_qdrant(self, session_id: str, query_embedding: np.ndarray, k: int = 3):
+        """Search Qdrant collection with error handling"""
+        try:
+            if isinstance(query_embedding, np.ndarray):
+                query_embedding = query_embedding.tolist()
+            collection_name = self.get_collection_name(session_id)
+            return self.qdrant_client.search(
+                collection_name=collection_name,
+                query_vector=query_embedding,
+                limit=k,
+                with_payload=True,
+                with_vectors=False
+            )
+        except Exception as e:
+            logger.error(f"Search failed: {e}")
+            raise
+
+# Web Crawler
+class WebCrawler:
+    """Handles web crawling with depth control and duplicate prevention"""
+    
+    def __init__(self, max_depth=2, delay=1):
+        self.max_depth = max_depth
+        self.delay = delay
+        self.visited = set()
+
+    def crawl_recursive(self, url, depth=0):
+        """Recursively crawl URLs up to max_depth"""
+        print(f"[WebCrawler] Crawling {url} at depth {depth}")
+        
+        if not hasattr(self, "collected_links"):
+            self.collected_links = set()
+
+        if depth > self.max_depth or url in self.visited or len(self.collected_links) >= 50:
+            return []
+
+        self.visited.add(url)
+        self.collected_links.add(url)
+        links = [url]
+
+        try:
+            response = requests.get(url, timeout=10, headers={"User-Agent": "Mozilla/5.0"})
+            soup = BeautifulSoup(response.content, "html.parser")
+
+            for tag in soup.find_all("a", href=True):
+                if len(self.collected_links) >= 10:
+                    break  # Stop if 50 links collected
+
+                href = urljoin(url, tag["href"])
+                if urlparse(href).netloc == urlparse(url).netloc:
+                    links.extend(self.crawl_recursive(href, depth + 1))
+        except Exception as e:
+            logger.warning(f"Failed to crawl {url}: {e}")
+
+        return list(set(links))
+
+# Connection Manager
+class ConnectionManager:
+    """Manages active WebSocket connections"""
+    
+    def __init__(self):
+        self.active_connections: Dict[str, WebSocket] = {}
+
+    async def connect(self, websocket: WebSocket, session_id: str):
+        """Register new WebSocket connection"""
+        await websocket.accept()
+        self.active_connections[session_id] = websocket
+
+    async def disconnect(self, session_id: str):
+        """Remove WebSocket connection"""
+        if session_id in self.active_connections:
+            del self.active_connections[session_id]
+
+    async def send_message(self, message: str, session_id: str):
+        """Send message to specific WebSocket connection"""
+        if session_id in self.active_connections:
+            await self.active_connections[session_id].send_text(message)
+
+# RAG System with Qdrant
+class RAGSystem:
+    """Main RAG system orchestrating crawling, indexing and querying"""
+    
+    def __init__(self):
+        self.session_manager = SessionManager()
+        self.crawler = WebCrawler()
+        self.llm = OllamaLLM(base_url="http://localhost:11434", model="mistral")
+
+    def crawl_and_index(self, session_id: str, start_url: str) -> Dict[str, Any]:
+        """Crawl website and index content in Qdrant"""
+        print(f"[RAGSystem] Starting crawl and index for session {session_id} with URL: {start_url}")
+        try:
+            session = self.session_manager.get_session(session_id)
+            all_urls = self.crawler.crawl_recursive(start_url)
+            documents, successful_urls = [], []
+
+            print(f"[RAGSystem] Crawled {len(all_urls)} URLs for session {session_id}")
+
+            for url in all_urls[:20]:  # Limit to 20 URLs
+                try:
+                    print(f"[RAGSystem] Processing URL: {url}")
+                    response = requests.get(url, timeout=10, headers={"User-Agent": "Mozilla/5.0"})
+                    soup = BeautifulSoup(response.content, "html.parser")
+                    for tag in soup(["script", "style"]):
+                        tag.decompose()
+                    text = " ".join(chunk.strip() for chunk in soup.get_text().splitlines() if chunk.strip())
+                    if len(text) > 100:
+                        documents.append(Document(text, {"source_url": url, "session_id": session_id}))
+                        successful_urls.append(url)
+                except Exception as e:
+                    logger.warning(f"Error processing {url}: {e}")
+
+            if documents:
+                self.session_manager.add_documents_to_qdrant(session_id, documents)
+                return {
+                    "status": "success",
+                    "urls_processed": successful_urls,
+                    "total_documents": len(documents)
+                }
+            return {"status": "error", "message": "No documents indexed"}
+        except Exception as e:
+            logger.error(f"crawl_and_index error: {e}")
+            return {
+                "status": "error",
+                "message": f"Error during crawling and indexing: {str(e)}"
+            }
+
+    async def chat(
+        self,
+        session_id: str,
+        question: str,
+        model: str = "mistral",
+        ollama_url: str = None,
+        gemini_api_key: str = None
+    ) -> Dict[str, Any]:
+        """
+        Handle chat requests with model selection.
+        Supports both Mistral (via Ollama) and Gemini models.
+        """
+        try:
+            # Get session data
+            session = self.session_manager.get_session(session_id)
+            if not session.get('documents'):
+                return {
+                    "status": "error",
+                    "message": "No documents indexed for this session"
+                }
+
+            # Select appropriate LLM
+            if model == "mistral" and ollama_url:
+                self.llm = OllamaLLM(base_url=ollama_url, model="mistral")
+            elif model == "gemini" and gemini_api_key:
+                self.llm = GeminiProvider()
+
+            # Process the query
+            result = self.process_query(session_id, question)
+            
+            # Add to conversation history if successful
+            if result["status"] == "success":
+                self.session_manager.add_conversation(
+                    session_id, 
+                    question, 
+                    result["response"]
+                )
+
+            return result
+
+        except Exception as e:
+            logger.error(f"Chat error: {str(e)}")
+            return {
+                "status": "error",
+                "message": f"Chat error: {str(e)}"
+            }
+
+    def process_query(self, session_id: str, query: str) -> Dict[str, Any]:
+        """Process user query through RAG pipeline"""
+        try:
+            # Validate and encode query
+            query_embedding = self.session_manager.embedding_model.encode(query)
+            if isinstance(query_embedding, np.ndarray):
+                query_embedding = query_embedding.astype("float32")
+            
+            # Search with proper parameters
+            search_result = self.session_manager.search_qdrant(
+                session_id=session_id,
+                query_embedding=query_embedding
+            )
+            
+            # Generate response using retrieved context
+            context = "\n\n".join(hit.payload["text"] for hit in search_result)
+            prompt = rag_prompt_template(context, query)
+            response = self.llm.generate([prompt])  # Remove .generations[0][0].text
+            
+            return {
+                "status": "success",
+                "response": response,
+                "sources": [hit.payload["metadata"] for hit in search_result]
+            }
+        except Exception as e:
+            logger.error(f"Query processing failed: {e}")
+            return {"status": "error", "message": str(e)}
+
+# FastAPI App
+app = FastAPI()
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Initialize RAG system
+rag = RAGSystem()
+
+# Request models
+class URLRequest(BaseModel):
+    """Request model for URL crawling"""
+    url: str
+    session_id: Optional[str] = None
+
+class ChatRequest(BaseModel):
+    """Request model for chat queries"""
+    session_id: str
+    question: str
+
+class SearchRequest(BaseModel):
+    """Request model for direct searches"""
+    session_id: str
+    query: str
+    limit: Optional[int] = 5
+
+# API Endpoints
+@app.get("/")
+async def root():
+    """Health check endpoint"""
+    return {"message": "RAG with Ollama Mistral and Qdrant is running"}
+
+@app.post("/create_session")
+async def create_session():
+    """Create a new session ID"""
+    session_id = str(uuid.uuid4())
+    return {"session_id": session_id, "status": "success"}
+
+@app.post("/crawl_and_index")
+async def crawl_and_index(request: URLRequest):
+    """Crawl and index a website"""
+    session_id = request.session_id or str(uuid.uuid4())
+    result = rag.crawl_and_index(session_id, request.url)
+    return result
+
+@app.post("/chat")
+async def chat(request: ChatRequest):
+    """Handle chat request"""
+    return await rag.chat(request.session_id, request.question)
+
+@app.post("/search")
+async def search(request: SearchRequest):
+    """Handle direct search request"""
+    try:
+        session = rag.session_manager.get_session(request.session_id)
+        query_embedding = rag.session_manager.embedding_model.encode(request.query)
+        if isinstance(query_embedding, np.ndarray):
+            query_embedding = query_embedding.tolist()
+        collection_name = rag.session_manager.get_collection_name(request.session_id)
+        search_results = rag.session_manager.qdrant_client.search(
+            collection_name=collection_name,
+            query_vector=query_embedding,
+            limit=request.limit
+        )
+        return {
+            "status": "success",
+            "results": [
+                {
+                    "text": hit.payload["text"],
+                    "score": hit.score,
+                    "metadata": hit.payload.get("metadata", {})
+                }
+                for hit in search_results
+            ]
+        }
+    except Exception as e:
+        logger.error(f"API search failed: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.websocket("/ws/chat")
+async def websocket_endpoint(websocket: WebSocket):
+    """WebSocket endpoint for real-time chat"""
+    await websocket.accept()
+    try:
+        while True:
+            data = await websocket.receive_json()
+            uid = data.get("uid")
+            question = data.get("question")
+
+            if not uid or not question:
+                await websocket.send_json({"error": "Missing 'uid' or 'question'"})
+                continue
+
+            # Get response from RAG system
+            response = await rag.chat(uid, question)
+            
+            # Handle both success and error cases
+            if response["status"] == "success":
+                await websocket.send_json({
+                    "uid": uid,
+                    "question": question,
+                    "answer": response["response"],
+                    "sources": response.get("sources", [])
+                })
+            else:
+                await websocket.send_json({
+                    "uid": uid,
+                    "error": response["message"]
+                })
+
+    except WebSocketDisconnect:
+        logger.info("WebSocket disconnected")
+    except Exception as e:
+        await websocket.send_json({"error": str(e)})
+
+# Main entry point
+if __name__ == "__main__":
+    Config.create_dirs()
+    from app import launch_interface
+    launch_interface()

src/models.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+
+class SearchRequest(BaseModel):
+    query: str
+    limit: Optional[int] = 5
+
+class SearchResponse(BaseModel):
+    status: str
+    results: Optional[List[Dict[str, Any]]] = None
+    message: Optional[str] = None
+
+class ChatSession(BaseModel):
+    session_id: str
+    created_at: datetime
+    history: List[Dict[str, Any]] = []
+    metadata: Dict[str, Any] = {}

src/prompts/templates.py

@@ -0,0 +1,9 @@
+def rag_prompt_template(context: str, question: str) -> str:
+    return f"""You are an expert assistant. Use ONLY the information from the context below to answer the question.
+If the context does not contain the answer, say "I don't know based on the provided content."
+
+Context:
+{context}
+
+Question: {question}
+Answer:"""

src/rag.py

@@ -0,0 +1,138 @@
+from typing import List, Dict, Optional
+from datetime import datetime
+import logging
+from app.models import SearchResponse
+from app.main import SessionManager
+from app.vectorstore import VectorStore
+from app.llm import OllamaMistral
+from app.embeddings import EmbeddingHandler
+
+class RAGSystem:
+    """
+    Main RAG (Retrieval-Augmented Generation) system class that orchestrates:
+    - Session management
+    - Vector store operations
+    - LLM interactions
+    - Chat history management
+    """
+    
+    def __init__(self):
+        """Initialize the RAG system components."""
+        self.logger = logging.getLogger(__name__)
+        # Initialize session manager for tracking conversations
+        self.session_manager = SessionManager()
+        # Initialize vector store for document storage and retrieval
+        self.vectorstore = VectorStore()  # This will create its own EmbeddingHandler
+        # Initialize LLM for generating responses
+        self.llm = OllamaMistral()
+
+    async def chat(self, session_id: str, query: str) -> SearchResponse:
+        """
+        Handle a chat query using RAG pipeline:
+        1. Retrieve relevant documents
+        2. Generate response using LLM
+        3. Store conversation history
+        
+        Args:
+            session_id: Unique identifier for the conversation session
+            query: User's question/input
+            
+        Returns:
+            SearchResponse: Contains answer, context, and sources
+        """
+        print(f"[RAGSystem] Starting chat for session: {session_id} with query: {query}")
+        try:
+            # Search for relevant documents in vector store
+            print(f"[RAGSystem] Searching for query: {query} in session: {session_id}")
+            search_results = await self.vectorstore.search_similar(
+                session_id=session_id,
+                query=query,
+                k=3  # Number of similar documents to retrieve
+            )
+            print(f"[RAGSystem] Search results: {search_results}")
+            
+            if search_results["status"] == "error":
+                return SearchResponse(**search_results)
+
+            # Prepare context from search results
+            context = "\n".join([
+                result["text"] 
+                for result in search_results["results"]
+                if "text" in result
+            ]) or "No relevant context found"
+
+            # Generate response using LLM with context
+            prompt = (
+                f"You are assisting with a website analysis. Here's relevant context from the website:\n"
+                f"{context}\n\n"
+                f"Question: {query}\n"
+                f"Please provide a detailed answer based on the website content:"
+            )
+            response = await self.llm.generate_response(prompt)
+
+            # Save conversation to history
+            self.session_manager.add_to_history(
+                session_id=session_id,
+                question=query,
+                answer=response
+            )
+
+            return SearchResponse(
+                status="success",
+                results=[{
+                    "answer": response,
+                    "context": context,
+                    "sources": search_results["results"]
+                }]
+            )
+
+        except Exception as e:
+            self.logger.error(f"Error in chat: {str(e)}", exc_info=True)
+            return SearchResponse(
+                status="error",
+                message=f"Chat error: {str(e)}"
+            )
+
+    def _validate_session(self, session_id: str) -> bool:
+        """
+        Validate and potentially initialize a session.
+        
+        Args:
+            session_id: Session identifier to validate
+            
+        Returns:
+            bool: True if session is valid/exists, False otherwise
+        """
+        try:
+            # Initialize session if it doesn't exist
+            if not self.session_manager.session_exists(session_id):
+                self.logger.info(f"Session {session_id} not found, initializing.")
+                self.session_manager.get_session(session_id)
+            # Verify vectorstore collection exists
+            if not self.vectorstore.collection_exists(session_id):
+                self.logger.warning(f"Vectorstore collection missing for session: {session_id}")
+                return False
+            return True
+        except Exception as e:
+            self.logger.error(f"Session validation failed: {str(e)}")
+            return False
+
+    async def get_chat_history(self, session_id: str, limit: int = 100) -> List[Dict]:
+        """
+        Retrieve chat history for a session.
+        
+        Args:
+            session_id: Session identifier
+            limit: Maximum number of history items to return
+            
+        Returns:
+            List[Dict]: Chat history entries or empty list if error occurs
+        """
+        try:
+            if not self._validate_session(session_id):
+                return []
+                
+            return self.session_manager.get_history(session_id, limit)
+        except Exception as e:
+            self.logger.error(f"Error getting chat history: {str(e)}")
+            return []

src/services/history_services.py

@@ -0,0 +1,29 @@
+from datetime import datetime
+from ..database.mongo_handler import MongoDBHandler
+
+class HistoryService:
+    def __init__(self):
+        self.db_handler = MongoDBHandler()
+
+    async def save_conversation(self, session_id: str, user_query: str, bot_response: str, metadata: dict = None):
+        """Save conversation to database"""
+        conversation = {
+            "session_id": session_id,
+            "user_query": user_query,
+            "bot_response": bot_response,
+            "timestamp": datetime.utcnow(),
+            "metadata": metadata or {}
+        }
+        return self.db_handler.insert_conversation(conversation)
+
+    async def get_session_history(self, session_id: str, limit: int = 100) -> list:
+        """Retrieve conversation history for a session"""
+        return self.db_handler.get_conversations_by_session(session_id, limit)
+
+    async def update_metadata(self, session_id: str, query: str, metadata: dict) -> bool:
+        """Update metadata for a specific query"""
+        return self.db_handler.update_conversation_metadata(session_id, query, metadata)
+
+    async def clear_session_history(self, session_id: str) -> int:
+        """Clear all history for a session"""
+        return self.db_handler.delete_session_history(session_id)

src/services/qdrant_handler.py

@@ -0,0 +1,10 @@
+from qdrant_client import QdrantClient
+from config import QDRANT_HOST, QDRANT_PORT, QDRANT_COLLECTION_NAME
+
+class QdrantHandler:
+    def __init__(self):
+        self.client = QdrantClient(
+            host=QDRANT_HOST,
+            port=QDRANT_PORT
+        )
+        self.collection_name = QDRANT_COLLECTION_NAME

src/storage.py

@@ -0,0 +1,222 @@
+import os
+import pickle
+import logging
+from typing import Dict, List, Any
+from app.config import Config
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
+
+class SessionStorage:
+    """
+    Manages session persistence using a hybrid storage approach:
+    - Stores session metadata in local pickle files
+    - Stores vector data in Qdrant collections
+    - Maintains connection between the two
+    """
+    
+    def __init__(self):
+        """
+        Initialize the session storage system.
+        Sets up Qdrant client connection and ensures storage directory exists.
+        """
+        try:
+            Config.create_storage_dir()
+            self.logger = logging.getLogger(__name__)
+            # Initialize Qdrant client with configuration from Config
+            self.qdrant_client = QdrantClient(
+                host=Config.QDRANT_HOST,
+                port=Config.QDRANT_PORT,
+                prefer_grpc=True  # Use gRPC for better performance
+            )
+            self.logger.info("Qdrant client initialized")
+        except Exception as e:
+            self.logger.error(f"Storage initialization error: {str(e)}")
+            raise RuntimeError("Storage initialization failed") from e
+    
+    def get_session_path(self, session_id: str) -> str:
+        """
+        Get the filesystem path for a session's pickle file.
+        
+        Args:
+            session_id: Unique session identifier
+            
+        Returns:
+            str: Full path to session file
+        """
+        return os.path.join(Config.STORAGE_DIR, f"{session_id}.pkl")
+    
+    def save_session(self, session_id: str, data: Dict):
+        """
+        Persist session data to disk (excluding Qdrant references).
+        
+        Args:
+            session_id: Session identifier
+            data: Session data dictionary
+        """
+        session_path = self.get_session_path(session_id)
+        
+        # Remove Qdrant collection reference before saving to avoid serialization issues
+        data = data.copy()
+        if 'qdrant_collection' in data:
+            del data['qdrant_collection']
+        
+        with open(session_path, 'wb') as f:
+            pickle.dump(data, f)
+    
+    def load_session(self, session_id: str) -> Dict:
+        """
+        Load session data from disk and reconnect to Qdrant collection.
+        
+        Args:
+            session_id: Session identifier
+            
+        Returns:
+            Dict: Session data with restored Qdrant collection reference
+        """
+        session_path = self.get_session_path(session_id)
+        
+        if not os.path.exists(session_path):
+            return None
+            
+        with open(session_path, 'rb') as f:
+            data = pickle.load(f)
+        
+        # Restore Qdrant collection reference
+        collection_name = f"session_{session_id}"
+        data['qdrant_collection'] = collection_name
+        
+        # Ensure collection exists in Qdrant (create if missing)
+        if not self.qdrant_client.collection_exists(collection_name):
+            self.logger.warning(f"Qdrant collection {collection_name} missing, creating new")
+            self.qdrant_client.create_collection(
+                collection_name=collection_name,
+                vectors_config=VectorParams(
+                    size=Config.EMBEDDING_SIZE,
+                    distance=Distance.COSINE
+                )
+            )
+        
+        return data
+    
+    def delete_session(self, session_id: str):
+        """
+        Completely remove a session (both disk and Qdrant storage).
+        
+        Args:
+            session_id: Session identifier to delete
+        """
+        session_path = self.get_session_path(session_id)
+        
+        # Delete Qdrant collection first
+        collection_name = f"session_{session_id}"
+        try:
+            self.qdrant_client.delete_collection(collection_name)
+            self.logger.info(f"Deleted Qdrant collection: {collection_name}")
+        except Exception as e:
+            self.logger.error(f"Error deleting Qdrant collection: {str(e)}")
+        
+        # Delete session file
+        if os.path.exists(session_path):
+            os.remove(session_path)
+
+class QdrantStorage:
+    """
+    Manages vector storage operations using Qdrant.
+    Handles collection management and vector operations.
+    """
+    
+    def __init__(self, collection_name: str, vector_size: int, 
+                 host: str = Config.QDRANT_HOST, port: int = Config.QDRANT_PORT):
+        """
+        Initialize Qdrant storage for a specific collection.
+        
+        Args:
+            collection_name: Name of the Qdrant collection
+            vector_size: Dimensionality of vectors to store
+            host: Qdrant server host (default from Config)
+            port: Qdrant server port (default from Config)
+        """
+        self.logger = logging.getLogger(__name__)
+        self.collection_name = collection_name
+        self.vector_size = vector_size
+        # Initialize Qdrant client with gRPC preference
+        self.qdrant = QdrantClient(host=host, port=port, prefer_grpc=True)
+        self._ensure_collection()
+
+    def _ensure_collection(self):
+        """
+        Ensure the collection exists in Qdrant.
+        Creates it if missing, otherwise verifies configuration.
+        """
+        try:
+            collection_info = self.qdrant.get_collection(self.collection_name)
+            if collection_info.vectors_count > 0:
+                self.logger.info(f"Using existing Qdrant collection: {self.collection_name}")
+        except Exception:
+            self.logger.info(f"Creating Qdrant collection: {self.collection_name}")
+            self.qdrant.recreate_collection(
+                collection_name=self.collection_name,
+                vectors_config=VectorParams(
+                    size=self.vector_size,
+                    distance=Distance.COSINE  # Using cosine similarity
+                )
+            )
+
+    def add_vectors(self, vectors: List[List[float]], payloads: List[Dict[str, Any]], offset: int = 0):
+        """
+        Add vectors and associated metadata to the collection.
+        
+        Args:
+            vectors: List of vector embeddings
+            payloads: List of metadata dictionaries
+            offset: Starting ID for new points (default 0)
+        """
+        points = [
+            PointStruct(
+                id=offset + idx,  # Sequential IDs with optional offset
+                vector=vector,
+                payload=payload
+            )
+            for idx, (vector, payload) in enumerate(zip(vectors, payloads))
+        ]
+        self.qdrant.upsert(
+            collection_name=self.collection_name,
+            points=points,
+            wait=True  # Ensure immediate persistence
+        )
+        self.logger.info(f"Added {len(points)} vectors to Qdrant collection '{self.collection_name}'")
+
+    def search(self, query_vector: List[float], session_id: str, limit: int = 5):
+        """
+        Search the collection for similar vectors, filtered by session.
+        
+        Args:
+            query_vector: The vector to compare against
+            session_id: Session identifier to filter results
+            limit: Maximum number of results to return
+            
+        Returns:
+            List[Dict]: Search results with scores and metadata
+        """
+        # Add session filter to ensure only current session results
+        results = self.qdrant.search(
+            collection_name=self.collection_name,
+            query_vector=query_vector,
+            query_filter=Filter(
+                must=[
+                    FieldCondition(
+                        key="session_id",
+                        match=MatchValue(value=session_id)
+                    )
+                ]
+            ),
+            limit=limit
+        )
+        return [
+            {
+                "id": hit.id,
+                "score": hit.score,
+                "payload": hit.payload
+            }
+            for hit in results
+        ]

src/tests/test_connection.py

@@ -0,0 +1,19 @@
+import socket
+from qdrant_client import QdrantClient
+
+def check_port(host, port):
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        return s.connect_ex((host, port)) == 0
+
+host = "localhost"
+port = 6333
+
+if check_port(host, port):
+    print(f"Port {port} is open. Testing Qdrant API...")
+    try:
+        client = QdrantClient(host=host, port=port)
+        print("Success! Collections:", client.get_collections())
+    except Exception as e:
+        print(f"API Error: {e}")
+else:
+    print(f"ERROR: Port {port} is closed. Check if Qdrant is running.")

src/tests/test_qdrant_integration.py

@@ -0,0 +1,12 @@
+import pytest
+from qdrant_client import QdrantClient
+from qdrant_client.models import VectorParams, Distance
+
+@pytest.fixture
+def qdrant_client():
+    return QdrantClient(host="localhost", port=6333)
+
+def test_collection_creation(qdrant_client):
+    test_collection = "test_collection"
+    qdrant_client.recreate_collection(test_collection, vectors_config=VectorParams(size=384, distance=Distance.COSINE))
+    assert qdrant_client.collection_exists(test_collection)

src/tests/test_ws.py

@@ -0,0 +1,12 @@
+import asyncio
+import websockets
+import json
+
+async def test_ws():
+    uri = "ws://localhost:8000/ws/test-session"
+    async with websockets.connect(uri) as ws:
+        await ws.send(json.dumps({"query": "What is AI?"}))
+        response = await ws.recv()
+        print("Response:", response)
+
+asyncio.run(test_ws())

src/utils/response_formatter.py

@@ -0,0 +1,47 @@
+
+from typing import Dict, List
+import re
+
+class ResponseFormatter:
+    @staticmethod
+    def clean_text(text: str) -> str:
+        """Clean up raw text by removing excessive whitespace and common boilerplate"""
+        # Remove multiple newlines and spaces
+        text = re.sub(r'\n+', '\n', text)
+        text = re.sub(r'\s+', ' ', text)
+        
+        # Remove common website elements
+        text = re.sub(r'SUBSCRIBE|RECENT|POPULAR|TRENDY', '', text, flags=re.I)
+        text = re.sub(r'Copyright © \d{4}.*', '', text)
+        text = re.sub(r'Privacy Policy|Terms of Service', '', text)
+        
+        return text.strip()
+
+    @staticmethod
+    def format_sources(sources: List[Dict]) -> str:
+        """Format source URLs into readable references"""
+        if not sources:
+            return ""
+        
+        formatted_sources = "\n\nSources:\n"
+        for i, source in enumerate(sources, 1):
+            formatted_sources += f"{i}. {source['url']}\n"
+        return formatted_sources
+
+    @staticmethod
+    def format_response(api_response: Dict) -> str:
+        """Convert API response to natural language"""
+        if "error" in api_response:
+            return f"Sorry, I encountered an error: {api_response['error']}"
+            
+        if "response" not in api_response:
+            return "I couldn't find any relevant information."
+            
+        # Clean and format the main response
+        clean_response = ResponseFormatter.clean_text(api_response["response"])
+        
+        # Add sources if available
+        if "sources" in api_response:
+            clean_response += ResponseFormatter.format_sources(api_response["sources"])
+            
+        return clean_response

src/vectorstore.py

@@ -0,0 +1,192 @@
+import os
+import logging
+from typing import List, Dict
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PointStruct, Filter
+import uuid
+from langchain.embeddings import HuggingFaceEmbeddings
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain.schema import Document
+
+from app.config import Config
+from app.crawler import URLCrawler
+from app.models import SearchResponse
+from app.embeddings import EmbeddingHandler
+
+class VectorStore:
+    """
+    A class to handle vector storage operations using Qdrant.
+    Manages document storage, retrieval, and similarity search in vector space.
+    """
+    
+    def __init__(self):
+        """Initialize the VectorStore with Qdrant client and embedding handlers."""
+        self.logger = logging.getLogger(__name__)
+        # Initialize Qdrant client with configuration from Config
+        self.client = QdrantClient(
+            url=Config.QDRANT_URL,
+            api_key=Config.QDRANT_API_KEY,
+            prefer_grpc=False,
+            timeout=30
+        )
+        # Initialize embedding handler and text splitter
+        self.embedding_handler = EmbeddingHandler()
+        self.embeddings = HuggingFaceEmbeddings(model_name=Config.EMBEDDING_MODEL)
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,  # Size of each text chunk
+            chunk_overlap=200  # Overlap between chunks for context preservation
+        )
+
+    def collection_exists(self, session_id: str) -> bool:
+        """
+        Check if a collection exists for the given session ID.
+        Attempts to create the collection if it doesn't exist.
+        
+        Args:
+            session_id: Unique identifier for the session
+            
+        Returns:
+            bool: True if collection exists or was created successfully, False otherwise
+        """
+        collection_name = self._get_collection_name(session_id)
+        try:
+            self.client.get_collection(collection_name=collection_name)
+            return True
+        except Exception:
+            # Try to create the collection if it doesn't exist
+            try:
+                self.client.recreate_collection(
+                    collection_name=collection_name,
+                    vectors_config=VectorParams(
+                        size=self.embedding_handler.embedding_dim,
+                        distance=Distance.COSINE  # Using cosine similarity
+                    )
+                )
+                self.logger.info(f"Created collection {collection_name} automatically.")
+                return True
+            except Exception as e:
+                self.logger.error(f"Failed to create collection {collection_name}: {e}")
+                return False
+
+    def _get_collection_name(self, session_id: str) -> str:
+        """
+        Generate a standardized collection name from session ID.
+        
+        Args:
+            session_id: Unique session identifier
+            
+        Returns:
+            str: Formatted collection name
+        """
+        return f"collection_{session_id}"
+
+    async def search_similar(self, session_id: str, query: str, k: int = 5) -> Dict:
+        """
+        Search for similar documents in the vector store.
+        
+        Args:
+            session_id: Session identifier for the collection
+            query: Search query text
+            k: Number of similar documents to return (default: 5)
+            
+        Returns:
+            Dict: Search results or error message
+        """
+        try:
+            if not self.collection_exists(session_id):
+                return {"status": "error", "message": "Collection not found"}
+            return await self.embedding_handler.search_collection(
+                collection_name=self._get_collection_name(session_id),
+                query=query,
+                k=k
+            )
+
+        except Exception as e:
+            self.logger.error(f"Search failed: {str(e)}")
+            return {"status": "error", "message": str(e)}
+
+    def create_from_url(self, url: str, session_id: str) -> None:
+        """
+        Crawl a website and create a vector store from its content.
+        
+        Args:
+            url: Website URL to crawl
+            session_id: Unique session identifier for storage
+            
+        Raises:
+            Exception: If vector store creation fails
+        """
+        try:
+            # Initialize crawler and fetch pages
+            crawler = URLCrawler()
+            raw_pages = crawler.crawl_sync(url, Config.MAX_PAGES_TO_CRAWL)
+
+            # Convert crawled pages to LangChain Document format
+            documents: List[Document] = [
+                Document(
+                    page_content=page["content"],
+                    metadata={
+                        "source": page["url"],
+                        "title": page["title"],
+                        "last_modified": page.get("last_modified", "")
+                    }
+                ) for page in raw_pages
+            ]
+
+            # Split documents into chunks
+            texts = self.text_splitter.split_documents(documents)
+            collection_name = self._get_collection_name(session_id)
+            
+            # Create or recreate collection with proper vector configuration
+            self.client.recreate_collection(
+                collection_name=collection_name,
+                vectors_config=VectorParams(
+                    size=self.embedding_handler.embedding_dim,
+                    distance=Distance.COSINE
+                )
+            )
+            
+            # Prepare points for batch insertion
+            points = [
+                PointStruct(
+                    id=str(uuid.uuid4()),  # Generate unique ID for each point
+                    vector=self.embeddings.embed_query(doc.page_content),
+                    payload={
+                        "page_content": doc.page_content,
+                        "metadata": doc.metadata
+                    }
+                ) for doc in texts
+            ]
+            
+            # Upsert all points into the collection
+            self.client.upsert(
+                collection_name=collection_name,
+                points=points
+            )
+            self.logger.info(f"Created vector store for session {session_id}")
+        except Exception as e:
+            self.logger.error(f"Vector store creation failed: {str(e)}")
+            raise
+
+    def save_vectorstore(self, vectorstore: None, session_id: str):
+        """
+        Placeholder method since Qdrant persists data automatically.
+        
+        Args:
+            vectorstore: Not used (Qdrant handles persistence)
+            session_id: Session identifier for logging
+        """
+        self.logger.debug(f"Data automatically persisted for session {session_id}")
+
+    def load_vectorstore(self, session_id: str) -> None:
+        """
+        Verify that a collection exists for the given session ID.
+        
+        Args:
+            session_id: Session identifier to check
+            
+        Raises:
+            ValueError: If collection doesn't exist
+        """
+        if not self.collection_exists(session_id):
+            raise ValueError(f"Collection for session {session_id} not found")