Add full AI Native Textbook project source code

.env.example

@@ -0,0 +1,13 @@
+# API Configuration
+GEMINI_API_KEY=your_gemini_api_key_here
+QDRANT_URL=your_qdrant_url_here
+QDRANT_API_KEY=your_qdrant_api_key_here
+NEON_DB_URL=your_neon_db_connection_string_here
+
+# JWT Configuration
+SECRET_KEY=your_secret_key_here
+JWT_EXPIRES_IN=3600
+
+# Application Configuration
+DEBUG=false
+LOG_LEVEL=info

Dockerfile

@@ -0,0 +1,25 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    gcc \
+    g++ \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements first to leverage Docker cache
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the application
+COPY . .
+
+# Expose the port the app runs on
+EXPOSE 8000
+
+# Run the application
+CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

README.md

@@ -0,0 +1,212 @@
+# AI Backend with RAG + Authentication
+
+A scalable backend featuring authentication, RAG capabilities, and integration with external services. The system uses Better Auth for authentication, Qdrant for vector storage, Neon Postgres for relational data, and Google's Gemini models for embeddings and chat functionality.
+
+## Architecture Overview
+
+```
+┌─────────────────┐    ┌──────────────────┐    ┌──────────────────┐
+│   Frontend      │────│   FastAPI        │────│  Better Auth     │
+│   (Future)      │    │   Backend        │    │   Service        │
+└─────────────────┘    └──────────────────┘    └──────────────────┘
+                              │
+         ┌────────────────────┼────────────────────┐
+         │                    │                    │
+   ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
+   │  Qdrant     │    │  Neon       │    │  Gemini     │
+   │  Vector DB  │    │  Postgres   │    │  API        │
+   └─────────────┘    └─────────────┘    └─────────────┘
+```
+
+## Features
+
+- **Authentication**: JWT-based authentication with Better Auth
+- **RAG Pipeline**: Retrieval-Augmented Generation with Qdrant vector database
+- **AI Integration**: Google Gemini for embeddings and chat responses
+- **Database**: Neon Postgres for user data and chat history
+- **Security**: Password hashing, JWT validation, user isolation
+- **Scalability**: Async architecture with connection pooling
+
+## Prerequisites
+
+- Python 3.9+
+- Qdrant vector database instance
+- Neon Postgres database
+- Google Gemini API key
+- Node.js (for development tools, optional)
+
+## Setup
+
+### 1. Clone the repository
+
+```bash
+git clone <repository-url>
+cd backend
+```
+
+### 2. Create a virtual environment
+
+```bash
+python -m venv venv
+source venv/bin/activate  # On Windows: venv\Scripts\activate
+```
+
+### 3. Install dependencies
+
+```bash
+pip install -r requirements.txt
+```
+
+### 4. Configure environment variables
+
+Copy the example environment file:
+
+```bash
+cp .env.example .env
+```
+
+Then edit `.env` with your actual configuration:
+
+```env
+# API Configuration
+GEMINI_API_KEY=your_gemini_api_key_here
+QDRANT_URL=your_qdrant_url_here
+QDRANT_API_KEY=your_qdrant_api_key_here
+NEON_DB_URL=your_neon_db_connection_string_here
+
+# JWT Configuration
+SECRET_KEY=your_secret_key_here  # Use a strong, random secret key
+JWT_EXPIRES_IN=3600
+
+# Application Configuration
+DEBUG=false
+LOG_LEVEL=info
+```
+
+### 5. Run the application
+
+```bash
+cd src
+python main.py
+```
+
+Or using uvicorn directly:
+
+```bash
+cd src
+uvicorn main:app --reload --host 0.0.0.0 --port 8000
+```
+
+The application will be available at `http://localhost:8000`
+
+## API Endpoints
+
+### Authentication
+- `POST /auth/signup` - User registration
+- `POST /auth/login` - User login
+- `GET /auth/me` - Get current user info
+
+### RAG & Embeddings
+- `POST /embed` - Generate embeddings for text
+- `POST /save-document` - Save and embed a document
+- `POST /search` - Semantic search in documents
+- `POST /chat` - Chat with RAG context
+
+### History
+- `GET /history` - Get chat history
+- `GET /history/{conversation_id}` - Get specific conversation
+
+### Health
+- `GET /health` - Health check endpoint
+
+## Project Structure
+
+```
+backend/
+├── src/
+│   ├── __init__.py
+│   ├── main.py                 # Application entry point
+│   ├── config/                 # Configuration management
+│   │   ├── __init__.py
+│   │   ├── settings.py         # App settings and env vars
+│   │   └── database.py         # Database configuration
+│   ├── auth/                   # Authentication module
+│   ├── db/                     # Database module
+│   │   ├── __init__.py
+│   │   ├── base.py             # Base model class
+│   │   ├── models/             # SQLAlchemy models
+│   │   ├── database.py         # Database connection
+│   │   └── crud.py             # CRUD operations
+│   ├── qdrant/                 # Vector database module
+│   ├── embeddings/             # Embedding module
+│   ├── rag/                    # RAG pipeline module
+│   ├── routes/                 # API routes
+│   ├── models/                 # Pydantic models
+│   ├── utils/                  # Utility functions
+│   └── scripts/                # Utility scripts
+├── tests/                      # Test suite
+├── requirements.txt            # Python dependencies
+├── .env.example                # Environment variables template
+└── README.md                   # Documentation
+```
+
+## Development
+
+### Running tests
+
+```bash
+cd backend
+python -m pytest tests/ -v
+```
+
+### Running with auto-reload during development
+
+```bash
+cd src
+uvicorn main:app --reload
+```
+
+## Environment Variables
+
+| Variable | Description | Required |
+|----------|-------------|----------|
+| GEMINI_API_KEY | Google Gemini API key | Yes |
+| QDRANT_URL | Qdrant vector database URL | Yes |
+| QDRANT_API_KEY | Qdrant API key (if secured) | No |
+| NEON_DB_URL | Neon Postgres connection string | Yes |
+| SECRET_KEY | JWT secret key | Yes |
+| JWT_EXPIRES_IN | JWT expiration time in seconds | No (default: 3600) |
+| DEBUG | Enable debug mode | No (default: false) |
+| LOG_LEVEL | Logging level | No (default: info) |
+
+## Security Considerations
+
+- Always use HTTPS in production
+- Store secrets securely (not in version control)
+- Validate and sanitize all user inputs
+- Use parameterized queries to prevent SQL injection
+- Implement rate limiting to prevent abuse
+- Use strong, randomly generated secret keys
+
+## Performance
+
+- Async architecture for high concurrency
+- Connection pooling for database operations
+- Caching mechanisms for frequently accessed data
+- Optimized vector search with Qdrant
+- Efficient embedding processing pipeline
+
+## Contributing
+
+1. Fork the repository
+2. Create a feature branch (`git checkout -b feature/amazing-feature`)
+3. Make your changes
+4. Add tests if applicable
+5. Run tests (`python -m pytest`)
+6. Commit your changes (`git commit -m 'Add amazing feature'`)
+7. Push to the branch (`git push origin feature/amazing-feature`)
+8. Open a Pull Request
+
+## License
+
+[Add your license here]

api/auth.py

@@ -0,0 +1,108 @@
+from fastapi import APIRouter, HTTPException, Depends
+from pydantic import BaseModel
+from typing import Optional
+from models.user import User
+from models.user_profile import UserProfile
+import os
+import bcrypt
+
+router = APIRouter()
+
+class SignupRequest(BaseModel):
+    email: str
+    password: str
+    software_background: Optional[str] = None
+    hardware_background: Optional[str] = None
+    experience_level: Optional[str] = None
+
+class LoginRequest(BaseModel):
+    email: str
+    password: str
+
+class AuthResponse(BaseModel):
+    user_id: str
+    email: str
+    access_token: str
+    refresh_token: str
+
+@router.post("/auth/signup", response_model=AuthResponse)
+async def signup(request: SignupRequest):
+    """Handle user registration with background information"""
+    try:
+        # In a real implementation, you would hash the password and store user in DB
+        # For now, we'll simulate the process
+
+        # Hash the password
+        hashed_password = bcrypt.hashpw(request.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
+
+        # Create user object
+        user = User(
+            email=request.email,
+            password=hashed_password,  # In real app, don't return the hash
+            software_background=request.software_background,
+            hardware_background=request.hardware_background,
+            experience_level=request.experience_level
+        )
+
+        # Create user profile
+        user_profile = UserProfile(
+            user_id="temp_user_id",  # In real app, this would be the actual user ID
+            software_background=request.software_background,
+            hardware_background=request.hardware_background,
+            experience_level=request.experience_level
+        )
+
+        # In a real implementation, you would store these in the database
+        # and generate proper JWT tokens
+
+        # For now, return a mock response
+        return AuthResponse(
+            user_id="temp_user_id",
+            email=request.email,
+            access_token="mock_access_token",
+            refresh_token="mock_refresh_token"
+        )
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error during signup: {str(e)}")
+
+@router.post("/auth/login", response_model=AuthResponse)
+async def login(request: LoginRequest):
+    """Handle user login"""
+    try:
+        # In a real implementation, you would verify credentials against DB
+        # For now, we'll simulate the process
+
+        # For demo purposes, we'll just return a mock response
+        # In a real app, you'd verify the password and generate tokens
+        return AuthResponse(
+            user_id="temp_user_id",
+            email=request.email,
+            access_token="mock_access_token",
+            refresh_token="mock_refresh_token"
+        )
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error during login: {str(e)}")
+
+@router.get("/auth/profile")
+async def get_profile():
+    """Get user profile information"""
+    try:
+        # In a real implementation, you would retrieve from DB based on auth token
+        profile = UserProfile(
+            user_id="temp_user_id",
+            software_background="Software Engineer",
+            hardware_background="Beginner",
+            experience_level="Intermediate"
+        )
+
+        return profile
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error retrieving profile: {str(e)}")
+
+@router.get("/auth/health")
+async def auth_health():
+    """Health check for auth service"""
+    return {"status": "auth service is running"}

api/chat.py

@@ -0,0 +1,72 @@
+import os
+import sys
+from fastapi import APIRouter
+import logging
+from qdrant_client import QdrantClient
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from services.rag_service import RAGService
+
+router = APIRouter()
+
+# Configure OpenRouter and RAG service
+openrouter_api_key = os.getenv("OPENAI_API_KEY")
+qdrant_url = os.getenv("QDRANT_URL")
+qdrant_api_key = os.getenv("QDRANT_API_KEY")
+collection_name = os.getenv("QDRANT_COLLECTION", "project_documents")
+
+if openrouter_api_key and openrouter_api_key != "your_openrouter_api_key_here":
+    # Initialize Qdrant client for cloud
+    if qdrant_url and qdrant_api_key and "qdrant.io" in qdrant_url:
+        qdrant_client = QdrantClient(
+            url=qdrant_url.replace(":6333", ""),  # Remove port from URL for cloud
+            api_key=qdrant_api_key,
+            prefer_grpc=False
+        )
+    else:
+        # Use local Qdrant if cloud not configured
+        qdrant_client = QdrantClient(
+            host=os.getenv("QDRANT_HOST", "localhost"),
+            port=int(os.getenv("QDRANT_PORT", 6333))
+        )
+
+    # Initialize RAG service with OpenRouter
+    rag_service = RAGService(openrouter_api_key, qdrant_client, collection_name)
+else:
+    rag_service = None
+
+logger = logging.getLogger(__name__)
+
+@router.post("/chat")
+async def chat(payload: dict):
+    user_msg = payload["message"]
+    selected_text = payload.get("selected_text", "")
+
+    # If selected text is provided, try to use RAG service to answer based only on that text
+    if selected_text and rag_service:
+        try:
+            # Use the RAG service to answer based on selected text only (with OpenRouter)
+            answer = rag_service.query_rag(selected_text, user_msg)
+            return {"answer": answer}
+        except Exception as e:
+            logger.error(f"RAG service failed: {str(e)}")
+            # Fall through to fallback response below
+    elif selected_text and not rag_service:
+        logger.warning("RAG service not available, using fallback")
+
+    # Fallback response when API is unavailable or not configured
+    fallback_responses = {
+        "hello": "Hello! I'm your AI textbook assistant. Feel free to ask questions about the content you're studying!",
+        "hi": "Hi there! I'm here to help you understand the AI and robotics concepts in your textbook. What would you like to know?",
+        "help": "I can help explain concepts from your AI and robotics textbook! Please select some text and ask questions about it.",
+        "default": f"I'm currently unable to process your request about '{user_msg}'. This might be because the AI service is temporarily unavailable or needs to be configured with a valid API key. The system is working properly but requires a valid OPENROUTER_API_KEY to provide AI-generated responses."
+    }
+
+    response_text = fallback_responses.get(user_msg.lower().strip(), fallback_responses["default"])
+
+    result = {"answer": response_text}
+    if not rag_service:
+        result["setup_needed"] = "Please configure a valid OPENROUTER_API_KEY in the .env file to enable AI responses"
+
+    return result

api/personalization.py

@@ -0,0 +1,71 @@
+from fastapi import APIRouter, HTTPException, Depends
+from pydantic import BaseModel
+from typing import Optional, Dict, Any
+import os
+import logging
+from services.personalization_service import PersonalizationService
+from services.content_adaptation import ContentAdaptationService
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+class PersonalizationRequest(BaseModel):
+    content: str
+    user_profile: Dict[str, Any]
+    chapter_id: str
+
+class PersonalizationResponse(BaseModel):
+    personalized_content: str
+    adaptation_details: Dict[str, Any]
+
+@router.post("/personalization/adapt", response_model=PersonalizationResponse)
+async def adapt_content(request: PersonalizationRequest):
+    """Adapt content based on user profile and background"""
+    try:
+        # Initialize content adaptation service
+        content_adaptation_service = ContentAdaptationService(
+            gemini_api_key=os.getenv("GEMINI_API_KEY", "your-gemini-key-here")
+        )
+
+        # Initialize personalization service with content adaptation service
+        personalization_service = PersonalizationService(content_adaptation_service)
+
+        # Adapt the content based on user profile
+        adapted_content = personalization_service.get_personalized_content(
+            content=request.content,
+            user_profile=request.user_profile,
+            chapter_id=request.chapter_id
+        )
+
+        # Prepare adaptation details
+        adaptation_details = {
+            "status": "success",
+            "user_software_background": request.user_profile.get('software_background', 'General'),
+            "user_hardware_background": request.user_profile.get('hardware_background', 'General'),
+            "user_experience_level": request.user_profile.get('experience_level', 'Intermediate'),
+            "chapter_id": request.chapter_id,
+            "adaptation_method": "AI-driven personalization"
+        }
+
+        return PersonalizationResponse(
+            personalized_content=adapted_content,
+            adaptation_details=adaptation_details
+        )
+
+    except Exception as e:
+        logger.error(f"Error adapting content: {str(e)}")
+        # Return original content if personalization fails, but still provide a response
+        return PersonalizationResponse(
+            personalized_content=request.content,
+            adaptation_details={
+                "status": "fallback",
+                "message": "Content personalization is temporarily unavailable. Showing original content.",
+                "original_chapter_id": request.chapter_id
+            }
+        )
+
+@router.get("/personalization/health")
+async def personalization_health():
+    """Health check for personalization service"""
+    return {"status": "personalization service is running"}

api/rag_search.py

@@ -0,0 +1,10 @@
+from fastapi import APIRouter
+
+router = APIRouter()
+
+@router.post("/rag-search")
+async def rag_search(payload: dict):
+    query = payload["query"]
+    # For now, return an empty result as the RAG functionality requires proper vector DB setup
+    # In a full implementation, this would search the vector database
+    return {"results": []}

api/translation.py

@@ -0,0 +1,66 @@
+from fastapi import APIRouter, HTTPException, Depends
+from pydantic import BaseModel
+from typing import Optional
+from services.translation_service import TranslationService
+import os
+
+router = APIRouter()
+
+class TranslationRequest(BaseModel):
+    text: str
+    source_lang: str = "en"
+    target_lang: str = "ur"
+
+class TranslationResponse(BaseModel):
+    original_text: str
+    translated_text: str
+    source_lang: str
+    target_lang: str
+
+@router.post("/translation/translate", response_model=TranslationResponse)
+async def translate_text(request: TranslationRequest):
+    """Translate text between languages (currently English to Urdu)"""
+    try:
+        # Initialize translation service
+        translation_service = TranslationService(
+            gemini_api_key=os.getenv("GEMINI_API_KEY", "your-gemini-key-here")
+        )
+
+        if request.source_lang == "en" and request.target_lang == "ur":
+            # Translate English to Urdu
+            translated_text = translation_service.translate_to_urdu(request.text)
+        elif request.source_lang == "ur" and request.target_lang == "en":
+            # Translate Urdu to English
+            translated_text = translation_service.translate_to_english(request.text)
+        else:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Unsupported language pair: {request.source_lang} to {request.target_lang}. Currently supported: en-ur"
+            )
+
+        return TranslationResponse(
+            original_text=request.text,
+            translated_text=translated_text,
+            source_lang=request.source_lang,
+            target_lang=request.target_lang
+        )
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error during translation: {str(e)}")
+
+@router.get("/translation/health")
+async def translation_health():
+    """Health check for translation service"""
+    return {"status": "translation service is running"}
+
+@router.post("/translation/clear-cache")
+async def clear_translation_cache():
+    """Clear the translation cache"""
+    try:
+        translation_service = TranslationService(
+            gemini_api_key=os.getenv("GEMINI_API_KEY", "your-gemini-key-here")
+        )
+        translation_service.clear_cache()
+        return {"status": "translation cache cleared"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error clearing cache: {str(e)}")

database/schema.sql

@@ -0,0 +1,50 @@
+-- Database schema for Neon Postgres
+
+-- Users table
+CREATE TABLE IF NOT EXISTS users (
+    id SERIAL PRIMARY KEY,
+    email VARCHAR(255) UNIQUE NOT NULL,
+    password_hash VARCHAR(255) NOT NULL,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    software_background VARCHAR(100),
+    hardware_background VARCHAR(100),
+    experience_level VARCHAR(50)
+);
+
+-- User profiles table
+CREATE TABLE IF NOT EXISTS user_profiles (
+    id SERIAL PRIMARY KEY,
+    user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
+    personalization_settings JSONB DEFAULT '{}',
+    learning_progress JSONB DEFAULT '{}',
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Chat sessions table
+CREATE TABLE IF NOT EXISTS chat_sessions (
+    id SERIAL PRIMARY KEY,
+    user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
+    selected_text TEXT NOT NULL,
+    question TEXT NOT NULL,
+    response TEXT NOT NULL,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    conversation_history JSONB DEFAULT '[]'
+);
+
+-- Textbook content table (for RAG)
+CREATE TABLE IF NOT EXISTS textbook_content (
+    id SERIAL PRIMARY KEY,
+    chapter_id VARCHAR(100) NOT NULL,
+    chapter_title VARCHAR(255) NOT NULL,
+    content TEXT NOT NULL,
+    embeddings JSONB, -- Store vector embeddings
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Indexes
+CREATE INDEX IF NOT EXISTS idx_users_email ON users(email);
+CREATE INDEX IF NOT EXISTS idx_chat_sessions_user_id ON chat_sessions(user_id);
+CREATE INDEX IF NOT EXISTS idx_textbook_content_chapter_id ON textbook_content(chapter_id);

debug_comprehensive.py

@@ -0,0 +1,79 @@
+import os
+import sys
+from dotenv import load_dotenv
+from qdrant_client import QdrantClient
+from qdrant_client.http import models
+
+# Load environment variables from parent directory
+load_dotenv(os.path.join(os.path.dirname(__file__), '..', '.env'))
+
+def comprehensive_debug():
+    # Get environment variables
+    qdrant_url = os.getenv("QDRANT_URL")
+    qdrant_api_key = os.getenv("QDRANT_API_KEY")
+    collection_name = os.getenv("QDRANT_COLLECTION", "project_documents")
+
+    print(f"QDRANT_URL: {qdrant_url}")
+    print(f"Collection: {collection_name}")
+
+    # Initialize Qdrant client for cloud
+    if qdrant_url and qdrant_api_key and "qdrant.io" in qdrant_url:
+        qdrant_client = QdrantClient(
+            url=qdrant_url.replace(":6333", ""),  # Remove port from URL for cloud
+            api_key=qdrant_api_key,
+            prefer_grpc=False
+        )
+    else:
+        # Use local Qdrant if cloud not configured
+        qdrant_client = QdrantClient(
+            host=os.getenv("QDRANT_HOST", "localhost"),
+            port=int(os.getenv("QDRANT_PORT", 6333))
+        )
+
+    # 1. Check collections
+    print("\n1. Available collections:")
+    try:
+        collections = qdrant_client.get_collections()
+        for collection in collections.collections:
+            print(f"  - {collection.name}")
+    except Exception as e:
+        print(f"  Error getting collections: {e}")
+
+    # 2. Get collection info
+    print(f"\n2. Collection info:")
+    try:
+        collection_info = qdrant_client.get_collection(collection_name)
+        print(f"  Points count: {collection_info.points_count}")
+        print(f"  Vector size: {collection_info.config.params.vectors.size}")
+        print(f"  Distance: {collection_info.config.params.vectors.distance}")
+    except Exception as e:
+        print(f"  Error getting collection info: {e}")
+
+    # 3. List all points in the collection (up to 10 for debugging)
+    print(f"\n3. All points in collection (up to 10):")
+    try:
+        points = qdrant_client.scroll(
+            collection_name=collection_name,
+            limit=10
+        )
+
+        count = 0
+        for point in points[0]:  # points[0] contains the list of points
+            count += 1
+            print(f"  Point {count}:")
+            print(f"    ID: {point.id}")
+            print(f"    Payload keys: {list(point.payload.keys()) if point.payload else 'None'}")
+            if point.payload and 'content' in point.payload:
+                content_preview = point.payload['content'][:100] + "..." if len(point.payload['content']) > 100 else point.payload['content']
+                print(f"    Content preview: {content_preview}")
+                print(f"    Topic: {point.payload.get('metadata', {}).get('topic', 'Unknown')}")
+            print()
+
+        print(f"  Total points found: {count}")
+    except Exception as e:
+        print(f"  Error listing points: {e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    comprehensive_debug()

debug_qdrant.py

@@ -0,0 +1,81 @@
+import os
+import sys
+from dotenv import load_dotenv
+from qdrant_client import QdrantClient
+
+# Load environment variables from parent directory
+load_dotenv(os.path.join(os.path.dirname(__file__), '..', '.env'))
+
+def debug_qdrant():
+    # Get environment variables
+    qdrant_url = os.getenv("QDRANT_URL")
+    qdrant_api_key = os.getenv("QDRANT_API_KEY")
+    collection_name = os.getenv("QDRANT_COLLECTION", "project_documents")
+
+    print(f"QDRANT_URL: {qdrant_url}")
+    print(f"Collection: {collection_name}")
+
+    # Initialize Qdrant client for cloud
+    if qdrant_url and qdrant_api_key and "qdrant.io" in qdrant_url:
+        qdrant_client = QdrantClient(
+            url=qdrant_url.replace(":6333", ""),  # Remove port from URL for cloud
+            api_key=qdrant_api_key,
+            prefer_grpc=False
+        )
+    else:
+        # Use local Qdrant if cloud not configured
+        qdrant_client = QdrantClient(
+            host=os.getenv("QDRANT_HOST", "localhost"),
+            port=int(os.getenv("QDRANT_PORT", 6333))
+        )
+
+    # 1. Check collections
+    print("\n1. Available collections:")
+    try:
+        collections = qdrant_client.get_collections()
+        for collection in collections.collections:
+            print(f"  - {collection.name}")
+    except Exception as e:
+        print(f"  Error getting collections: {e}")
+
+    # 2. Try to search for content to verify it exists
+    print(f"\n2. Testing search functionality:")
+    try:
+        # Try to create a simple embedding to test if connection works
+        from openai import OpenAI
+        openrouter_api_key = os.getenv("OPENAI_API_KEY")
+        client = OpenAI(
+            api_key=openrouter_api_key,
+            base_url="https://openrouter.ai/api/v1"
+        )
+
+        response = client.embeddings.create(
+            model="text-embedding-3-small",
+            input="Artificial Intelligence"
+        )
+        vector = response.data[0].embedding
+
+        # Now search in Qdrant
+        hits = qdrant_client.search(
+            collection_name=collection_name,
+            query_vector=vector,
+            limit=2
+        )
+
+        print(f"  Search successful! Found {len(hits)} results")
+        if hits:
+            for i, hit in enumerate(hits):
+                print(f"  Result {i+1}:")
+                print(f"    ID: {hit.id}")
+                print(f"    Payload keys: {list(hit.payload.keys()) if hit.payload else 'None'}")
+                if hit.payload and 'content' in hit.payload:
+                    print(f"    Content preview: {hit.payload['content'][:100]}...")
+                else:
+                    print(f"    Payload content: {hit.payload}")
+    except Exception as e:
+        print(f"  Error during search test: {e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    debug_qdrant()

docker-compose.yml

@@ -0,0 +1,47 @@
+version: '3.8'
+
+services:
+  backend:
+    build: .
+    ports:
+      - "8000:8000"
+    environment:
+      - NEON_DB_URL=${NEON_DB_URL}
+      - QDRANT_URL=${QDRANT_URL}
+      - QDRANT_API_KEY=${QDRANT_API_KEY}
+      - GEMINI_API_KEY=${GEMINI_API_KEY}
+      - SECRET_KEY=${SECRET_KEY}
+      - JWT_EXPIRES_IN=${JWT_EXPIRES_IN:-3600}
+      - DEBUG=${DEBUG:-false}
+      - LOG_LEVEL=${LOG_LEVEL:-info}
+    volumes:
+      - .:/app
+    depends_on:
+      - postgres
+      - qdrant
+    restart: unless-stopped
+
+  postgres:
+    image: postgres:15-alpine
+    environment:
+      - POSTGRES_DB=ai_backend
+      - POSTGRES_USER=postgres
+      - POSTGRES_PASSWORD=password
+    ports:
+      - "5432:5432"
+    volumes:
+      - postgres_data:/var/lib/postgresql/data
+    restart: unless-stopped
+
+  qdrant:
+    image: qdrant/qdrant:latest
+    ports:
+      - "6333:6333"
+      - "6334:6334"
+    volumes:
+      - qdrant_data:/qdrant/storage
+    restart: unless-stopped
+
+volumes:
+  postgres_data:
+  qdrant_data:

final_verification.py

@@ -0,0 +1,83 @@
+import os
+import sys
+from dotenv import load_dotenv
+
+# Load environment variables from parent directory
+load_dotenv(os.path.join(os.path.dirname(__file__), '..', '.env'))
+
+# Add the backend directory to the path
+sys.path.append(os.path.dirname(__file__))
+
+from services.rag_service import RAGService
+from qdrant_client import QdrantClient
+
+def final_verification():
+    # Get environment variables
+    openrouter_api_key = os.getenv("OPENAI_API_KEY")
+    qdrant_url = os.getenv("QDRANT_URL")
+    qdrant_api_key = os.getenv("QDRANT_API_KEY")
+    collection_name = os.getenv("QDRANT_COLLECTION", "project_documents")
+
+    # Initialize Qdrant client for cloud
+    if qdrant_url and qdrant_api_key and "qdrant.io" in qdrant_url:
+        qdrant_client = QdrantClient(
+            url=qdrant_url.replace(":6333", ""),  # Remove port from URL for cloud (same as in chat.py)
+            api_key=qdrant_api_key,
+            prefer_grpc=False
+        )
+    else:
+        # Use local Qdrant if cloud not configured
+        qdrant_client = QdrantClient(
+            host=os.getenv("QDRANT_HOST", "localhost"),
+            port=int(os.getenv("QDRANT_PORT", 6333))
+        )
+
+    # Initialize RAG service
+    rag_service = RAGService(openrouter_api_key, qdrant_client, collection_name)
+
+    print("=== FINAL VERIFICATION ===")
+
+    # Test 1: Content exists - should return actual content
+    print("\n✅ Test 1: Content exists in database")
+    selected_text = "Robotics is an interdisciplinary field that combines mechanical engineering, electrical engineering, and computer science to design, construct, and operate robots."
+    question = "What is robotics?"
+
+    result = rag_service.query_rag(selected_text, question)
+    print(f"Selected text: {selected_text[:50]}...")
+    print(f"Question: {question}")
+    print(f"Result: {result[:100]}...")
+    print(f"Expected: Actual content (not fallback)")
+    print(f"✅ PASS: Content returned (not fallback message)" if "Is sawal ka jawab" not in result else "❌ FAIL: Fallback message returned")
+
+    # Test 2: Content doesn't exist - should return fallback
+    print("\n✅ Test 2: Content does not exist in database")
+    selected_text = "This is completely unrelated text that should not match anything in the database."
+    question = "What is Quantum Computing?"
+
+    result = rag_service.query_rag(selected_text, question)
+    print(f"Selected text: {selected_text[:50]}...")
+    print(f"Question: {question}")
+    print(f"Result: {result}")
+    print(f"Expected: 'Is sawal ka jawab provided data me mojood nahi hai.'")
+    print(f"✅ PASS: Fallback message returned" if "Is sawal ka jawab" in result else "❌ FAIL: Content returned")
+
+    # Test 3: AI content exists
+    print("\n✅ Test 3: AI content exists in database")
+    selected_text = "Artificial Intelligence is a branch of computer science that aims to create software or machines that exhibit human-like intelligence."
+    question = "What is Artificial Intelligence?"
+
+    result = rag_service.query_rag(selected_text, question)
+    print(f"Selected text: {selected_text[:50]}...")
+    print(f"Question: {question}")
+    print(f"Result: {result[:100]}...")
+    print(f"✅ PASS: Content returned (not fallback message)" if "Is sawal ka jawab" not in result else "❌ FAIL: Fallback message returned")
+
+    print("\n=== VERIFICATION COMPLETE ===")
+    print("✅ Backend RAG service is working correctly")
+    print("✅ Uses selected_text for Qdrant search")
+    print("✅ Returns actual content when found")
+    print("✅ Returns fallback message when not found")
+    print("✅ Ready for frontend integration")
+
+if __name__ == "__main__":
+    final_verification()

main.py

@@ -0,0 +1,53 @@
+import sys
+import os
+from dotenv import load_dotenv
+
+# Load environment variables from .env file in the project root
+project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+dotenv_path = os.path.join(project_root, '.env')
+load_dotenv(dotenv_path)
+
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+from api.chat import router as chat_router
+from api.auth import router as auth_router
+from api.translation import router as translation_router
+from api.personalization import router as personalization_router
+from api.rag_search import router as rag_search_router
+from api.chat import router as chat_api_router  # New chat API with RAG
+
+app = FastAPI(title="AI-native Textbook Platform API")
+
+# Add CORS middleware to allow requests from the Docusaurus frontend
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["http://localhost:3000", "http://localhost:3001", "http://localhost:8000", "*"],  # Allow frontend origins
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+    allow_origin_regex=r"https?://localhost(:[0-9]+)?",
+)
+
+# Include API routers
+app.include_router(chat_router, prefix="/api")  # Original chat API (for compatibility)
+app.include_router(auth_router, prefix="/api")
+app.include_router(translation_router, prefix="/api")
+app.include_router(personalization_router, prefix="/api")
+app.include_router(rag_search_router, prefix="/api")
+# New enhanced chat API with RAG is included in the original chat_router
+
+@app.get("/")
+def read_root():
+    return {"message": "Welcome to the AI-native Interactive Textbook Platform for Physical AI & Humanoid Robotics"}
+
+@app.get("/health")
+def health_check():
+    return {"status": "healthy"}
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8001)

middleware/auth_middleware.py

@@ -0,0 +1,46 @@
+from fastapi import Request, HTTPException
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+import jwt
+from typing import Optional
+import os
+
+class JWTAuth:
+    def __init__(self, secret_key: str = None, algorithm: str = "HS256"):
+        self.secret_key = secret_key or os.getenv("JWT_SECRET_KEY", "your-secret-key-here")
+        self.algorithm = algorithm
+        self.security = HTTPBearer()
+
+    async def __call__(self, request: Request) -> Optional[dict]:
+        credentials: HTTPAuthorizationCredentials = await self.security(request)
+
+        if credentials:
+            token = credentials.credentials
+            try:
+                # Decode the JWT token
+                payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
+                request.state.user = payload
+                return payload
+            except jwt.ExpiredSignatureError:
+                raise HTTPException(status_code=401, detail="Token has expired")
+            except jwt.InvalidTokenError:
+                raise HTTPException(status_code=401, detail="Invalid token")
+        else:
+            raise HTTPException(status_code=401, detail="No authorization token provided")
+
+# Example usage in routes:
+# @router.get("/protected-route")
+# async def protected_route(request: Request, user: dict = Depends(JWTAuth())):
+#     return {"message": f"Hello {user.get('email')}, you are authenticated!"}
+
+# For now, we'll create a simple dependency that can be used to require authentication
+async def require_auth(request: Request):
+    """Simple dependency to require authentication (placeholder for real implementation)"""
+    # In a real implementation, this would validate the JWT token
+    # For now, we'll just check if there's a mock token in the header
+    auth_header = request.headers.get("Authorization")
+    if not auth_header or not auth_header.startswith("Bearer "):
+        raise HTTPException(status_code=401, detail="Authorization header missing or invalid")
+
+    # In a real app, you would validate the JWT here
+    # For demo purposes, we'll just continue
+    pass

models/chat_session.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+from typing import Optional, List
+from datetime import datetime
+
+class ChatSession(BaseModel):
+    id: Optional[str] = None
+    user_id: str
+    selected_text: str
+    question: str
+    response: str
+    created_at: Optional[datetime] = None
+    conversation_history: Optional[List[dict]] = []

models/user.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+class User(BaseModel):
+    id: Optional[str] = None
+    email: str
+    password: str
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
+    software_background: Optional[str] = None  # Software Engineer, Beginner, etc.
+    hardware_background: Optional[str] = None  # Hardware Engineer, Beginner, etc.
+    experience_level: Optional[str] = None  # Beginner, Intermediate, Advanced

models/user_profile.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+class UserProfile(BaseModel):
+    id: Optional[str] = None
+    user_id: str
+    software_background: Optional[str] = None  # Software Engineer, Beginner, etc.
+    hardware_background: Optional[str] = None  # Hardware Engineer, Beginner, etc.
+    experience_level: Optional[str] = None  # Beginner, Intermediate, Advanced
+    personalization_settings: Optional[dict] = {}
+    learning_progress: Optional[dict] = {}
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None

pyproject.toml

@@ -0,0 +1,70 @@
+[tool.poetry]
+name = "ai-backend-rag-auth"
+version = "1.0.0"
+description = "AI Backend with RAG + Authentication using Qdrant, Neon, Gemini, FastAPI, and Better Auth"
+authors = ["Your Name <your.email@example.com>"]
+
+[tool.poetry.dependencies]
+python = "^3.9"
+fastapi = "^0.104.1"
+uvicorn = {extras = ["standard"], version = "^0.24.0"}
+sqlalchemy = {extras = ["asyncio"], version = "^2.0.23"}
+asyncpg = "^0.29.0"
+qdrant-client = "^1.7.0"
+google-generativeai = "^0.4.0"
+python-multipart = "^0.0.6"
+python-jose = {extras = ["cryptography"], version = "^3.3.0"}
+passlib = {extras = ["bcrypt"], version = "^1.7.4"}
+better-exceptions = "^0.3.3"
+python-dotenv = "^1.0.0"
+pydantic = "^2.5.0"
+pydantic-settings = "^2.1.0"
+uuid = "^1.30"
+httpx = "^0.25.2"
+alembic = "^1.13.1"
+
+[tool.poetry.group.dev.dependencies]
+pytest = "^7.4.3"
+pytest-asyncio = "^0.21.1"
+black = "^23.10.1"
+isort = "^5.12.0"
+mypy = "^1.7.1"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+asyncio_mode = "auto"
+addopts = "-v --tb=short"
+
+[tool.black]
+line-length = 88
+target-version = ['py39']
+include = '\.pyi?$'
+extend-exclude = '''
+/(
+  # directories
+  \.eggs
+  | \.git
+  | \.hg
+  | \.mypy_cache
+  | \.tox
+  | \.venv
+  | build
+  | dist
+)/
+'''
+
+[tool.isort]
+profile = "black"
+multi_line_output = 3
+known_first_party = ["src"]
+known_third_party = ["fastapi", "uvicorn", "sqlalchemy", "asyncpg", "qdrant_client", "google", "pydantic", "pytest"]
+
+[tool.mypy]
+python_version = "3.9"
+warn_return_any = true
+warn_unused_configs = true
+warn_unused_ignores = true

requirements.txt

@@ -0,0 +1,19 @@
+fastapi==0.104.1
+uvicorn[standard]==0.24.0
+sqlalchemy[asyncio]==2.0.23
+asyncpg==0.29.0
+qdrant-client==1.7.0
+google-generativeai==0.4.0
+python-multipart==0.0.6
+python-jose[cryptography]==3.3.0
+passlib[bcrypt]==1.7.4
+better-exceptions==0.3.3
+python-dotenv==1.0.0
+pydantic==2.5.0
+pydantic-settings==2.1.0
+uuid==1.30
+httpx==0.25.2
+pytest==7.4.3
+pytest-asyncio==0.21.1
+alembic==1.13.1
+openai==1.10.0

services/content_adaptation.py

@@ -0,0 +1,105 @@
+from typing import Dict, Any, Optional
+import google.generativeai as genai
+import logging
+
+logger = logging.getLogger(__name__)
+
+class ContentAdaptationService:
+    def __init__(self, gemini_api_key: str):
+        genai.configure(api_key=gemini_api_key)
+        self.model = genai.GenerativeModel('gemini-pro')
+
+    def adapt_content(self, content: str, user_background: str, experience_level: str, chapter_id: str) -> str:
+        """Adapt content based on user background and experience level"""
+        try:
+            # Determine adaptation instructions based on user profile
+            adaptation_instructions = self._get_adaptation_instructions(user_background, experience_level, chapter_id)
+
+            # Call Gemini API to adapt the content
+            prompt = f"""You are an educational content adapter for a Physical AI & Humanoid Robotics textbook. Adapt the provided content according to these instructions: {adaptation_instructions}. Maintain the core educational value while making it appropriate for the target audience.
+
+Original content:
+{content}
+
+Adapted content:"""
+
+            response = self.model.generate_content(
+                prompt,
+                generation_config=genai.types.GenerationConfig(
+                    max_output_tokens=len(content) * 2,
+                    temperature=0.4,
+                )
+            )
+
+            adapted_content = response.text
+
+            logger.info(f"Adapted content for background: {user_background}, level: {experience_level}")
+            return adapted_content
+
+        except Exception as e:
+            logger.error(f"Error adapting content: {str(e)}")
+            # Return original content if adaptation fails
+            return content
+
+    def _get_adaptation_instructions(self, user_background: str, experience_level: str, chapter_id: str) -> str:
+        """Generate adaptation instructions based on user profile"""
+        instructions = []
+
+        # Add background-specific instructions
+        if user_background and 'software' in user_background.lower():
+            instructions.append("Include more code examples and programming concepts")
+        elif user_background and 'hardware' in user_background.lower():
+            instructions.append("Include more hardware specifications and physical implementations")
+        else:
+            instructions.append("Provide balanced content with both software and hardware aspects")
+
+        # Add experience level-specific instructions
+        if experience_level == 'beginner':
+            instructions.append("Use simpler explanations, more examples, and step-by-step instructions")
+        elif experience_level == 'intermediate':
+            instructions.append("Provide moderate complexity with practical applications")
+        elif experience_level == 'advanced':
+            instructions.append("Include complex examples, optimization techniques, and advanced concepts")
+        else:
+            instructions.append("Use moderate complexity appropriate for mixed experience levels")
+
+        # Add chapter-specific instructions if needed
+        if 'ros2' in chapter_id.lower():
+            instructions.append("Focus on ROS 2 concepts like nodes, topics, and URDF")
+        elif 'gazebo' in chapter_id.lower() or 'unity' in chapter_id.lower():
+            instructions.append("Emphasize simulation concepts, sensors, and environment modeling")
+        elif 'nvidia' in chapter_id.lower() or 'isaac' in chapter_id.lower():
+            instructions.append("Highlight perception, VSLAM, navigation, and Isaac-specific concepts")
+        elif 'vla' in chapter_id.lower():
+            instructions.append("Focus on voice, cognitive, and capstone project concepts")
+
+        return "; ".join(instructions)
+
+    def adapt_examples(self, examples: list, user_background: str, experience_level: str) -> list:
+        """Adapt code or practical examples based on user profile"""
+        try:
+            adapted_examples = []
+            for example in examples:
+                prompt = f"""You are adapting educational examples for a Physical AI & Humanoid Robotics textbook. Adapt this example for a user with {user_background} background and {experience_level} experience level. Return the adapted example.
+
+Original example:
+{example}
+
+Adapted example:"""
+
+                response = self.model.generate_content(
+                    prompt,
+                    generation_config=genai.types.GenerationConfig(
+                        max_output_tokens=1000,
+                        temperature=0.3,
+                    )
+                )
+
+                adapted_examples.append(response.text)
+
+            logger.info(f"Adapted {len(examples)} examples for background: {user_background}, level: {experience_level}")
+            return adapted_examples
+
+        except Exception as e:
+            logger.error(f"Error adapting examples: {str(e)}")
+            return examples  # Return original examples if adaptation fails

services/personalization_service.py

@@ -0,0 +1,95 @@
+from typing import Dict, Any, Optional
+from .content_adaptation import ContentAdaptationService
+import logging
+
+logger = logging.getLogger(__name__)
+
+class PersonalizationService:
+    def __init__(self, content_adaptation_service: ContentAdaptationService):
+        self.content_adaptation_service = content_adaptation_service
+
+    def get_personalized_content(self, content: str, user_profile: Dict[str, Any], chapter_id: str) -> str:
+        """Get personalized content based on user profile"""
+        try:
+            # Determine the user's background and experience level
+            software_background = user_profile.get('software_background', '')
+            hardware_background = user_profile.get('hardware_background', '')
+            experience_level = user_profile.get('experience_level', 'beginner')
+
+            # Adapt content based on user profile
+            adapted_content = self.content_adaptation_service.adapt_content(
+                content=content,
+                user_background=software_background or hardware_background,
+                experience_level=experience_level,
+                chapter_id=chapter_id
+            )
+
+            logger.info(f"Personalized content for user with background: {software_background}/{hardware_background}, level: {experience_level}")
+            return adapted_content
+
+        except Exception as e:
+            logger.error(f"Error in personalization: {str(e)}")
+            # Return original content if personalization fails
+            return content
+
+    def get_user_recommendations(self, user_profile: Dict[str, Any], current_chapter: str) -> Dict[str, Any]:
+        """Get personalized recommendations for the user"""
+        try:
+            software_background = user_profile.get('software_background', '')
+            hardware_background = user_profile.get('hardware_background', '')
+            experience_level = user_profile.get('experience_level', 'beginner')
+
+            recommendations = {
+                'next_chapters': self._get_next_chapters(user_profile, current_chapter),
+                'difficulty_level': experience_level,
+                'focus_areas': self._get_focus_areas(software_background, hardware_background),
+                'additional_resources': self._get_resources(experience_level)
+            }
+
+            logger.info(f"Generated recommendations for user")
+            return recommendations
+
+        except Exception as e:
+            logger.error(f"Error generating recommendations: {str(e)}")
+            return {}
+
+    def _get_next_chapters(self, user_profile: Dict[str, Any], current_chapter: str) -> list:
+        """Determine next chapters based on user profile and current progress"""
+        # This would be more sophisticated in a real implementation
+        # For now, return a default sequence
+        chapter_sequence = {
+            '1-ros2': ['2-gazebo-unity', '3-nvidia-isaac'],
+            '2-gazebo-unity': ['3-nvidia-isaac', '4-vla'],
+            '3-nvidia-isaac': ['4-vla', 'capstone'],
+            '4-vla': ['capstone'],
+            'capstone': []
+        }
+
+        return chapter_sequence.get(current_chapter, [])
+
+    def _get_focus_areas(self, software_background: str, hardware_background: str) -> list:
+        """Determine focus areas based on user background"""
+        focus_areas = []
+
+        if software_background and 'software' in software_background.lower():
+            focus_areas.append('code examples')
+            focus_areas.append('programming concepts')
+        elif hardware_background and 'hardware' in hardware_background.lower():
+            focus_areas.append('hardware specifications')
+            focus_areas.append('physical implementations')
+
+        if not focus_areas:
+            focus_areas.append('general concepts')
+
+        return focus_areas
+
+    def _get_resources(self, experience_level: str) -> list:
+        """Get additional resources based on experience level"""
+        if experience_level == 'beginner':
+            return ['tutorials', 'basic examples', 'step-by-step guides']
+        elif experience_level == 'intermediate':
+            return ['advanced examples', 'practical applications']
+        elif experience_level == 'advanced':
+            return ['research papers', 'cutting-edge implementations', 'optimization techniques']
+        else:
+            return ['tutorials', 'examples']

services/rag_service.py

@@ -0,0 +1,118 @@
+from typing import List, Dict, Any
+import logging
+import os
+from openai import OpenAI
+from qdrant_client import QdrantClient
+import json
+
+logger = logging.getLogger(__name__)
+
+class RAGService:
+    def __init__(self, openrouter_api_key: str, vector_db_service: QdrantClient, collection_name: str = "project_documents"):
+        # Initialize OpenRouter client
+        self.client = OpenAI(
+            api_key=openrouter_api_key,
+            base_url="https://openrouter.ai/api/v1"
+        )
+        self.qdrant = vector_db_service
+        self.collection_name = collection_name
+
+    def get_embedding(self, text: str) -> List[float]:
+        """Get embeddings for text using OpenAI's embedding API"""
+        try:
+            response = self.client.embeddings.create(
+                model="text-embedding-3-small",
+                input=text
+            )
+            return response.data[0].embedding
+        except Exception as e:
+            logger.error(f"Error getting embeddings: {str(e)}")
+            raise e
+
+    def search_qdrant(self, query: str) -> str:
+        """Search Qdrant for relevant content based on query"""
+        try:
+            vector = self.get_embedding(query)
+
+            hits = self.qdrant.search(
+                collection_name=self.collection_name,
+                query_vector=vector,
+                limit=5
+            )
+
+            return "\n\n".join(hit.payload["content"] for hit in hits if "content" in hit.payload)
+        except Exception as e:
+            logger.error(f"Error searching Qdrant: {str(e)}")
+            return ""
+
+    def query_rag(self, selected_text: str, question: str) -> str:
+        """Process a RAG query using OpenRouter with context from Qdrant"""
+        # Validate inputs
+        if not selected_text or len(selected_text.strip()) == 0:
+            # Check length (as per requirement TC-002: max 5000 characters)
+            if len(selected_text) > 5000:
+                logger.warning(f"Selected text exceeds 5000 character limit: {len(selected_text)} characters")
+                return "Selected text exceeds the 5000 character limit. Please select a shorter text."
+
+        SYSTEM_PROMPT = """You are a RAG-based AI agent.
+
+RULES:
+- Answer ONLY from the retrieved context.
+- If the answer is not found, say:
+  "Is sawal ka jawab provided data me mojood nahi hai."""
+
+        try:
+            # Search Qdrant using the selected_text to get relevant context
+            context = self.search_qdrant(selected_text)
+
+            # If we found context, generate the final answer using the context
+            if context.strip():
+                final_response = self.client.chat.completions.create(
+                    model="openai/gpt-3.5-turbo",
+                    messages=[
+                        {"role": "system", "content": SYSTEM_PROMPT},
+                        {"role": "assistant", "content": f"Here is the relevant context: {context}"},
+                        {"role": "user", "content": question}
+                    ],
+                    temperature=0
+                )
+                return final_response.choices[0].message.content
+            else:
+                # If no context was found, return the fallback message
+                return "Is sawal ka jawab provided data me mojood nahi hai."
+
+        except Exception as e:
+            logger.error(f"Error in RAG query: {str(e)}")
+            # Check if the error is related to API key validity
+            error_str = str(e).lower()
+            if "api key" in error_str or "quota" in error_str or "billing" in error_str or "permission" in error_str or "401" in str(e) or "403" in str(e):
+                # Return a more specific message about API configuration
+                return f"I'm currently unable to process your request about '{question}'. The AI service may be temporarily unavailable due to API key issues or quota limits. Please check that your OPENROUTER_API_KEY is properly configured in the .env file and has sufficient quota available."
+            else:
+                # Return a general fallback response
+                return f"I apologize, but I'm currently unable to process your request about '{question}'. The AI service may be temporarily unavailable. Please try again later or contact support if the issue persists."
+
+    def index_content(self, content_id: str, content: str, metadata: Dict[str, Any] = None):
+        """Index textbook content for RAG retrieval"""
+        if metadata is None:
+            metadata = {}
+
+        # Get embeddings for the content
+        embeddings = self.get_embedding(content)
+
+        # Store in vector database
+        self.qdrant.upsert(
+            collection_name=self.collection_name,
+            points=[
+                {
+                    "id": content_id,
+                    "vector": embeddings,
+                    "payload": {
+                        "content": content,
+                        "metadata": metadata
+                    }
+                }
+            ]
+        )
+
+        logger.info(f"Indexed content: {content_id}")

services/translation_service.py

@@ -0,0 +1,144 @@
+import google.generativeai as genai
+import logging
+from typing import Dict
+import time
+
+logger = logging.getLogger(__name__)
+
+class TranslationService:
+    def __init__(self, gemini_api_key: str):
+        genai.configure(api_key=gemini_api_key)
+        self.model = genai.GenerativeModel('gemini-pro')
+        self.translation_cache: Dict[str, str] = {}
+        self.cache_timestamps: Dict[str, float] = {}
+
+    def translate_to_urdu(self, text: str, ttl: int = 3600) -> str:
+        """Translate English text to Urdu with caching"""
+        # Create cache key
+        cache_key = f"en_to_ur_{hash(text)}"
+
+        # Check if translation is in cache and not expired
+        if cache_key in self.translation_cache:
+            if time.time() - self.cache_timestamps.get(cache_key, 0) < ttl:
+                logger.info("Returning cached translation")
+                return self.translation_cache[cache_key]
+
+        # Call Gemini API for translation with improved prompt
+        try:
+            prompt = self._create_urdu_translation_prompt(text)
+
+            response = self.model.generate_content(
+                prompt,
+                generation_config=genai.types.GenerationConfig(
+                    max_output_tokens=min(len(text) * 3, 4000),  # Urdu text might be longer
+                    temperature=0.2,
+                    top_p=0.9,
+                )
+            )
+
+            translated_text = self._format_translation_response(response.text)
+
+            # Cache the translation
+            self.translation_cache[cache_key] = translated_text
+            self.cache_timestamps[cache_key] = time.time()
+
+            logger.info(f"Translated text to Urdu (length: {len(translated_text)} chars)")
+            return translated_text
+
+        except Exception as e:
+            logger.error(f"Translation failed: {str(e)}")
+            # Return a professional fallback response
+            return f"Translation unavailable: {text[:100]}..."
+
+    def _create_urdu_translation_prompt(self, text: str) -> str:
+        """Create a professional Urdu translation prompt"""
+        return f"""You are an elite professional translator specializing in technical and educational content. Translate the provided English text to Urdu with precision and cultural sensitivity.
+
+TRANSLATION REQUIREMENTS:
+• Maintain technical accuracy for robotics/AI terminology
+• Use proper Urdu script and correct grammar
+• Preserve the original meaning and context
+• Apply appropriate formality level for educational content
+• Ensure readability and flow in Urdu
+• Do not add any commentary or explanations
+
+SOURCE TEXT:
+"{text}"
+
+URDU TRANSLATION:"""
+
+    def _format_translation_response(self, response_text: str) -> str:
+        """Format the translation response for consistency"""
+        # Clean up response
+        formatted = response_text.strip()
+
+        # Remove any unwanted prefixes or explanations
+        if 'TRANSLATION:' in formatted:
+            formatted = formatted.split('TRANSLATION:')[-1].strip()
+        elif 'TRANSLATED TEXT:' in formatted:
+            formatted = formatted.split('TRANSLATED TEXT:')[-1].strip()
+
+        # Clean up extra whitespace
+        formatted = ' '.join(formatted.split())
+
+        return formatted
+
+    def translate_to_english(self, urdu_text: str, ttl: int = 3600) -> str:
+        """Translate Urdu text back to English with caching"""
+        # Create cache key
+        cache_key = f"ur_to_en_{hash(urdu_text)}"
+
+        # Check if translation is in cache and not expired
+        if cache_key in self.translation_cache:
+            if time.time() - self.cache_timestamps.get(cache_key, 0) < ttl:
+                logger.info("Returning cached translation")
+                return self.translation_cache[cache_key]
+
+        # Call Gemini API for translation with improved prompt
+        try:
+            prompt = self._create_english_translation_prompt(urdu_text)
+
+            response = self.model.generate_content(
+                prompt,
+                generation_config=genai.types.GenerationConfig(
+                    max_output_tokens=min(len(urdu_text) * 2, 4000),
+                    temperature=0.2,
+                    top_p=0.9,
+                )
+            )
+
+            translated_text = self._format_translation_response(response.text)
+
+            # Cache the translation
+            self.translation_cache[cache_key] = translated_text
+            self.cache_timestamps[cache_key] = time.time()
+
+            logger.info(f"Translated text to English (length: {len(translated_text)} chars)")
+            return translated_text
+
+        except Exception as e:
+            logger.error(f"Translation failed: {str(e)}")
+            # Return a professional fallback response
+            return f"Translation unavailable: {urdu_text[:100]}..."
+
+    def _create_english_translation_prompt(self, urdu_text: str) -> str:
+        """Create a professional English translation prompt"""
+        return f"""You are an elite professional translator specializing in technical and educational content. Translate the provided Urdu text to English with precision and accuracy.
+
+TRANSLATION REQUIREMENTS:
+• Maintain technical accuracy for robotics/AI terminology
+• Preserve the original meaning and context
+• Apply appropriate formality level for educational content
+• Ensure readability and flow in English
+• Do not add any commentary or explanations
+
+SOURCE TEXT:
+"{urdu_text}"
+
+ENGLISH TRANSLATION:"""
+
+    def clear_cache(self):
+        """Clear the translation cache"""
+        self.translation_cache.clear()
+        self.cache_timestamps.clear()
+        logger.info("Translation cache cleared")

services/vector_db.py

@@ -0,0 +1,127 @@
+from qdrant_client import QdrantClient
+from qdrant_client.http import models
+from typing import List, Dict, Any
+import logging
+
+logger = logging.getLogger(__name__)
+
+class VectorDBService:
+    def __init__(self, host: str = "localhost", port: int = 6333, cloud_client=None):
+        self.collection_name = "textbook_content"
+        self.is_available = False  # Initialize as False by default
+
+        if cloud_client:
+            # Use provided cloud client
+            self.client = cloud_client
+            try:
+                self._init_collection()
+                self.is_available = True
+            except Exception as e:
+                import logging
+                logger = logging.getLogger(__name__)
+                logger.warning(f"Vector database not available: {str(e)}. Running in fallback mode.")
+                self.client = None
+        else:
+            # Use local client
+            try:
+                self.client = QdrantClient(host=host, port=port)
+                self._init_collection()
+                self.is_available = True
+            except Exception as e:
+                import logging
+                logger = logging.getLogger(__name__)
+                logger.warning(f"Vector database not available: {str(e)}. Running in fallback mode.")
+                self.client = None
+
+    def _init_collection(self):
+        """Initialize the Qdrant collection for textbook content"""
+        if not self.is_available:
+            return
+
+        try:
+            # Check if collection exists
+            self.client.get_collection(self.collection_name)
+        except:
+            # Create collection if it doesn't exist
+            self.client.create_collection(
+                collection_name=self.collection_name,
+                vectors_config=models.VectorParams(size=1536, distance=models.Distance.COSINE),  # Assuming OpenAI embeddings
+            )
+            logger.info(f"Created collection: {self.collection_name}")
+
+    def add_content(self, content_id: str, content: str, embeddings: List[float], metadata: Dict[str, Any] = None):
+        """Add textbook content to the vector database"""
+        if not self.is_available:
+            logger.warning("Vector database not available, skipping content addition")
+            return
+
+        if metadata is None:
+            metadata = {}
+
+        self.client.upsert(
+            collection_name=self.collection_name,
+            points=[
+                models.PointStruct(
+                    id=content_id,
+                    vector=embeddings,
+                    payload={
+                        "content": content,
+                        "metadata": metadata
+                    }
+                )
+            ]
+        )
+        logger.info(f"Added content to vector DB: {content_id}")
+
+    def search_content(self, query_embeddings: List[float], limit: int = 10) -> List[Dict[str, Any]]:
+        """Search for relevant content based on query embeddings"""
+        if not self.is_available or self.client is None:
+            logger.warning("Vector database not available, returning empty results")
+            # Return empty list when database is not available
+            return []
+
+        try:
+            # Handle different Qdrant client versions
+            if hasattr(self.client, 'search'):
+                # Newer version of Qdrant client
+                results = self.client.search(
+                    collection_name=self.collection_name,
+                    query_vector=query_embeddings,
+                    limit=limit
+                )
+            else:
+                # Older version or different interface
+                results = self.client.search(
+                    collection_name=self.collection_name,
+                    query_vector=query_embeddings,
+                    limit=limit
+                )
+        except AttributeError:
+            logger.warning("Qdrant client search method not available, using direct processing")
+            return []
+        except Exception as e:
+            logger.warning(f"Vector database search failed, using direct processing: {str(e)}")
+            # Return empty list when search fails
+            return []
+
+        return [
+            {
+                "id": result.id,
+                "content": result.payload.get("content"),
+                "metadata": result.payload.get("metadata", {}),
+                "score": getattr(result, 'score', 0.0)  # Handle different result structures
+            }
+            for result in results
+        ]
+
+    def delete_content(self, content_id: str):
+        """Delete content from the vector database"""
+        if not self.is_available:
+            logger.warning("Vector database not available, skipping content deletion")
+            return
+
+        self.client.delete(
+            collection_name=self.collection_name,
+            points_selector=models.PointIdsList(points=[content_id])
+        )
+        logger.info(f"Deleted content from vector DB: {content_id}")

setup_sample_content.py

@@ -0,0 +1,86 @@
+import os
+import sys
+from qdrant_client import QdrantClient
+from openai import OpenAI
+import uuid
+from dotenv import load_dotenv
+
+# Load environment variables from .env file in the project root
+load_dotenv(os.path.join(os.path.dirname(os.path.dirname(__file__)), '.env'))
+
+# Add the backend directory to the path so we can import the RAG service
+sys.path.append(os.path.dirname(os.path.dirname(__file__)))
+
+from services.rag_service import RAGService
+
+def setup_sample_content():
+    # Get environment variables
+    openrouter_api_key = os.getenv("OPENAI_API_KEY")
+    qdrant_url = os.getenv("QDRANT_URL")
+    qdrant_api_key = os.getenv("QDRANT_API_KEY")
+    collection_name = os.getenv("QDRANT_COLLECTION", "project_documents")
+
+    # Initialize Qdrant client for cloud
+    if qdrant_url and qdrant_api_key and "qdrant.io" in qdrant_url:
+        qdrant_client = QdrantClient(
+            url=qdrant_url.replace(":6333", ""),  # Remove port from URL for cloud (same as in chat.py)
+            api_key=qdrant_api_key,
+            prefer_grpc=False
+        )
+    else:
+        # Use local Qdrant if cloud not configured
+        qdrant_client = QdrantClient(
+            host=os.getenv("QDRANT_HOST", "localhost"),
+            port=int(os.getenv("QDRANT_PORT", 6333))
+        )
+
+    # Initialize RAG service
+    rag_service = RAGService(openrouter_api_key, qdrant_client, collection_name)
+
+    # Sample content about AI and Robotics
+    import uuid
+
+    sample_content = [
+        {
+            "id": str(uuid.uuid4()),
+            "content": "Introduction to Physical AI & Humanoid Robotics: Embodied Intelligence represents the convergence of artificial intelligence with physical systems. It's the principle that true intelligence emerges not just from abstract computation, but from the interaction between an intelligent system and its physical environment. In the context of humanoid robotics, this means creating machines that can perceive, reason, and act in the physical world much like humans do. This textbook combines cutting-edge robotics concepts with artificial intelligence to provide a deep understanding of embodied intelligence systems.",
+            "metadata": {"topic": "Introduction to Physical AI & Humanoid Robotics", "level": "beginner", "original_id": "intro_physical_ai_1"}
+        },
+        {
+            "id": str(uuid.uuid4()),
+            "content": "Artificial Intelligence (AI) is a branch of computer science that aims to create software or machines that exhibit human-like intelligence. This can include learning from experience, understanding natural language, solving problems, and recognizing patterns. AI systems can be trained using various techniques including machine learning, deep learning, and neural networks.",
+            "metadata": {"topic": "AI Fundamentals", "level": "beginner", "original_id": "ai_fundamentals_1"}
+        },
+        {
+            "id": str(uuid.uuid4()),
+            "content": "Machine learning is a subset of artificial intelligence that focuses on algorithms that can learn from data. Instead of being explicitly programmed, machine learning models improve their performance through experience with data. Common types include supervised learning, unsupervised learning, and reinforcement learning.",
+            "metadata": {"topic": "Machine Learning", "level": "beginner", "original_id": "machine_learning_1"}
+        },
+        {
+            "id": str(uuid.uuid4()),
+            "content": "Robotics is an interdisciplinary field that combines mechanical engineering, electrical engineering, and computer science to design, construct, and operate robots. Modern robots can perform complex tasks in manufacturing, healthcare, exploration, and service industries. They often incorporate AI to enable autonomous decision-making and adaptive behavior.",
+            "metadata": {"topic": "Robotics", "level": "beginner", "original_id": "robotics_intro_1"}
+        },
+        {
+            "id": str(uuid.uuid4()),
+            "content": "Neural networks are computing systems inspired by the human brain's structure and function. They consist of interconnected nodes (neurons) organized in layers. Deep learning uses neural networks with multiple hidden layers to recognize patterns and make predictions. They are particularly effective for image recognition, natural language processing, and complex decision-making tasks.",
+            "metadata": {"topic": "Neural Networks", "level": "intermediate", "original_id": "neural_networks_1"}
+        },
+        {
+            "id": str(uuid.uuid4()),
+            "content": "Natural Language Processing (NLP) is a field of AI focused on enabling computers to understand, interpret, and generate human language. NLP techniques are used in chatbots, translation services, sentiment analysis, and text summarization. Modern NLP systems often use transformer architectures and large language models.",
+            "metadata": {"topic": "Natural Language Processing", "level": "intermediate", "original_id": "nlp_fundamentals_1"}
+        }
+    ]
+
+    print(f"Indexing {len(sample_content)} content items into collection '{collection_name}'...")
+
+    for item in sample_content:
+        rag_service.index_content(item["id"], item["content"], item["metadata"])
+        print(f"Indexed: {item['id']} - {item['metadata']['topic']}")
+
+    print(f"\nSuccessfully indexed {len(sample_content)} items into '{collection_name}' collection!")
+    print("Your RAG system is now ready to answer questions about AI, Machine Learning, Robotics, Neural Networks, and NLP.")
+
+if __name__ == "__main__":
+    setup_sample_content()

src/auth/auth.py

@@ -0,0 +1,132 @@
+"""
+Authentication module for the AI Backend with RAG + Authentication
+Implements JWT-based authentication with password hashing
+"""
+from datetime import datetime, timedelta
+from typing import Optional, Union
+import jwt
+from passlib.context import CryptContext
+from fastapi import HTTPException, status, Depends
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from pydantic import BaseModel
+import logging
+
+from ..config.settings import settings
+
+logger = logging.getLogger(__name__)
+
+# Password hashing context
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+# JWT security scheme
+security = HTTPBearer()
+
+class TokenData(BaseModel):
+    username: Optional[str] = None
+    user_id: Optional[str] = None
+
+
+class AuthHandler:
+    def __init__(self):
+        self.secret_key = settings.secret_key
+        self.algorithm = settings.jwt_algorithm
+        self.access_token_expires = timedelta(minutes=settings.jwt_expires_in // 60)  # Convert seconds to minutes
+
+    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
+        """
+        Verify a plain password against a hashed password
+        """
+        return pwd_context.verify(plain_password, hashed_password)
+
+    def get_password_hash(self, password: str) -> str:
+        """
+        Generate a hash for a plain password
+        """
+        return pwd_context.hash(password)
+
+    def create_access_token(self, data: dict, expires_delta: Optional[timedelta] = None) -> str:
+        """
+        Create a JWT access token with optional expiration time
+        """
+        to_encode = data.copy()
+
+        if expires_delta:
+            expire = datetime.utcnow() + expires_delta
+        else:
+            expire = datetime.utcnow() + self.access_token_expires
+
+        to_encode.update({"exp": expire, "iat": datetime.utcnow()})
+
+        encoded_jwt = jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)
+        return encoded_jwt
+
+    def decode_access_token(self, token: str) -> Optional[TokenData]:
+        """
+        Decode a JWT token and return token data
+        """
+        try:
+            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
+            username: str = payload.get("sub")
+            user_id: str = payload.get("user_id")
+
+            if username is None:
+                return None
+
+            token_data = TokenData(username=username, user_id=user_id)
+            return token_data
+        except jwt.exceptions.ExpiredSignatureError:
+            logger.warning("Expired token attempted to be decoded")
+            return None
+        except jwt.exceptions.InvalidTokenError:
+            logger.warning("Invalid token attempted to be decoded")
+            return None
+
+    async def get_current_user(self, token: str = Depends(security)) -> TokenData:
+        """
+        Get the current user from the provided JWT token
+        This function can be used as a dependency in route handlers
+        """
+        credentials_exception = HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not validate credentials",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+        try:
+            token_data = self.decode_access_token(token.credentials)
+            if token_data is None:
+                raise credentials_exception
+            return token_data
+        except Exception as e:
+            logger.error(f"Error getting current user: {e}")
+            raise credentials_exception
+
+
+# Create a global instance of AuthHandler
+auth_handler = AuthHandler()
+
+# Convenience functions for use in other modules
+def get_password_hash(password: str) -> str:
+    """Generate a hash for a plain password"""
+    return auth_handler.get_password_hash(password)
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """Verify a plain password against a hashed password"""
+    return auth_handler.verify_password(plain_password, hashed_password)
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
+    """Create a JWT access token"""
+    return auth_handler.create_access_token(data, expires_delta)
+
+def decode_access_token(token: str) -> Optional[TokenData]:
+    """Decode a JWT token and return token data"""
+    return auth_handler.decode_access_token(token)
+
+async def get_current_user(token: str = Depends(security)) -> TokenData:
+    """Get the current user from the provided JWT token"""
+    return await auth_handler.get_current_user(token)
+
+def create_user_token(user_id: str, username: str) -> str:
+    """Create a token specifically for a user"""
+    data = {"sub": username, "user_id": user_id}
+    return create_access_token(data)

src/auth/middleware.py

@@ -0,0 +1,74 @@
+"""
+Authentication middleware for the AI Backend with RAG + Authentication
+Provides utilities for protecting routes with JWT authentication
+"""
+from fastapi import HTTPException, status, Request
+from typing import Optional
+import logging
+
+from .auth import auth_handler, TokenData
+
+logger = logging.getLogger(__name__)
+
+class AuthMiddleware:
+    """
+    Authentication middleware class to protect routes
+    """
+
+    @staticmethod
+    async def verify_token(request: Request) -> Optional[TokenData]:
+        """
+        Verify the JWT token in the request headers
+        """
+        # Get authorization header
+        auth_header = request.headers.get("Authorization")
+        if not auth_header or not auth_header.startswith("Bearer "):
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Authorization header missing or invalid format",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+
+        token = auth_header[7:]  # Remove "Bearer " prefix
+        token_data = auth_handler.decode_access_token(token)
+
+        if token_data is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Invalid or expired token",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+
+        # Add user info to request state for use in route handlers
+        request.state.user = token_data
+        return token_data
+
+    @staticmethod
+    async def require_auth(request: Request) -> TokenData:
+        """
+        Require authentication for a route
+        This can be used as a dependency in route handlers
+        """
+        return await AuthMiddleware.verify_token(request)
+
+    @staticmethod
+    async def optional_auth(request: Request) -> Optional[TokenData]:
+        """
+        Optionally authenticate a user (returns None if no valid token)
+        This can be used as a dependency in route handlers
+        """
+        try:
+            return await AuthMiddleware.verify_token(request)
+        except HTTPException:
+            # If token is invalid or missing, return None instead of raising error
+            return None
+
+
+# Convenience functions for use in route handlers
+async def require_auth(request: Request) -> TokenData:
+    """Require authentication for a route"""
+    return await AuthMiddleware.require_auth(request)
+
+async def optional_auth(request: Request) -> Optional[TokenData]:
+    """Optionally authenticate a user"""
+    return await AuthMiddleware.optional_auth(request)

src/auth/schemas.py

@@ -0,0 +1,53 @@
+"""
+Authentication schemas for request/response validation
+"""
+from pydantic import BaseModel, EmailStr
+from typing import Optional
+from datetime import datetime
+import uuid
+
+
+class UserBase(BaseModel):
+    email: EmailStr
+    full_name: Optional[str] = None
+
+
+class UserCreate(UserBase):
+    password: str
+
+    class Config:
+        from_attributes = True
+
+
+class UserUpdate(BaseModel):
+    full_name: Optional[str] = None
+    email: Optional[EmailStr] = None
+
+    class Config:
+        from_attributes = True
+
+
+class UserInDB(UserBase):
+    id: uuid.UUID
+    is_active: bool
+    created_at: datetime
+    updated_at: Optional[datetime] = None
+
+    class Config:
+        from_attributes = True
+
+
+class UserLogin(BaseModel):
+    email: EmailStr
+    password: str
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    expires_in: int
+
+
+class TokenData(BaseModel):
+    user_id: Optional[str] = None
+    username: Optional[str] = None

src/config/database.py

@@ -0,0 +1,61 @@
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+from .settings import settings
+import logging
+
+logger = logging.getLogger(__name__)
+
+try:
+    # Create async engine with proper configuration from settings
+    engine = create_async_engine(
+        settings.neon_db_url,
+        echo=settings.debug,  # Set to True to log SQL queries
+        pool_pre_ping=True,  # Verify connections before use
+        pool_size=20,  # Connection pool size
+        max_overflow=30,  # Additional connections beyond pool_size
+        pool_recycle=3600,  # Recycle connections after 1 hour
+        pool_pre_ping_enabled=True,  # Enable connection health checks
+        pool_pool_timeout=30,  # Connection timeout
+        pool_reset_on_return='commit'  # Reset connection on return
+    )
+
+    # Create async session factory
+    AsyncSessionLocal = sessionmaker(
+        engine,
+        class_=AsyncSession,
+        expire_on_commit=False
+    )
+
+    logger.info("Database engine created successfully")
+except Exception as e:
+    logger.error(f"Failed to create database engine: {e}")
+    raise
+
+
+async def get_db_session():
+    """Dependency to get database session"""
+    async with AsyncSessionLocal() as session:
+        try:
+            yield session
+        except Exception as e:
+            logger.error(f"Database session error: {e}")
+            await session.rollback()
+            raise
+        finally:
+            await session.close()
+
+
+# Initialize the database connection
+async def init_db():
+    """Initialize the database connection and create tables if needed"""
+    from ..db.base import Base
+
+    logger.info("Initializing database connection...")
+    try:
+        # Create all tables
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+            logger.info("Database tables created successfully")
+    except Exception as e:
+        logger.error(f"Failed to initialize database: {e}")
+        raise

src/config/settings.py

@@ -0,0 +1,69 @@
+from pydantic_settings import BaseSettings
+from typing import Optional
+from pydantic import ValidationError, field_validator
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class Settings(BaseSettings):
+    # Database settings
+    neon_db_url: str
+
+    # Qdrant settings
+    qdrant_url: str
+    qdrant_api_key: Optional[str] = None
+
+    # Gemini API settings
+    gemini_api_key: str
+
+    # JWT settings
+    secret_key: str
+    jwt_algorithm: str = "HS256"
+    jwt_expires_in: int = 3600  # 1 hour default
+
+    # Application settings
+    debug: bool = False
+    log_level: str = "info"
+
+    # Server settings
+    server_host: str = "0.0.0.0"
+    server_port: int = 8000
+
+    @field_validator('neon_db_url', 'qdrant_url', 'gemini_api_key', 'secret_key')
+    @classmethod
+    def validate_required_fields(cls, v, info):
+        if not v:
+            raise ValueError(f"{info.field_name} is required and must be set in environment variables")
+        return v
+
+    @field_validator('debug')
+    @classmethod
+    def validate_debug(cls, v):
+        if isinstance(v, str):
+            return v.lower() in ['true', '1', 'yes', 'on']
+        return bool(v)
+
+    @field_validator('jwt_expires_in')
+    @classmethod
+    def validate_jwt_expires_in(cls, v):
+        if v <= 0:
+            raise ValueError("JWT expires in must be a positive integer")
+        return v
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = 'utf-8'
+        case_sensitive = True
+
+
+# Create a single instance of settings with error handling
+try:
+    settings = Settings()
+    logger.info("Configuration loaded successfully")
+except ValidationError as e:
+    logger.error(f"Configuration validation error: {e}")
+    raise
+except Exception as e:
+    logger.error(f"Configuration error: {e}")
+    raise

src/db/base.py

@@ -0,0 +1,28 @@
+"""
+Base class for SQLAlchemy models
+"""
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy import Column, DateTime, func
+from sqlalchemy.ext.asyncio import AsyncAttrs
+from datetime import datetime
+import uuid
+from sqlalchemy.dialects.postgresql import UUID
+
+
+class Base(AsyncAttrs, DeclarativeBase):
+    """
+    Base class for all SQLAlchemy models
+    Includes common columns and configurations
+    """
+    __abstract__ = True
+
+    # Common columns for all models
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
+
+    def __init__(self, *args, **kwargs):
+        # Set the id automatically if not provided
+        if 'id' not in kwargs and hasattr(self, 'id') and self.id is None:
+            # For models that have an id column, set a default UUID if not provided
+            pass  # The column default will handle this
+        super().__init__(*args, **kwargs)

src/db/crud.py

@@ -0,0 +1,432 @@
+"""
+CRUD operations for the AI Backend with RAG + Authentication
+Implements Create, Read, Update, Delete operations for all models
+"""
+from typing import Optional, List
+from uuid import UUID
+from sqlalchemy import select, update, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.exc import IntegrityError
+from fastapi import HTTPException, status
+import logging
+
+from .models.user import User
+from .models.chat_history import ChatHistory
+from .models.document import Document
+
+logger = logging.getLogger(__name__)
+
+
+# User CRUD Operations
+async def create_user(db: AsyncSession, email: str, hashed_password: str, full_name: Optional[str] = None) -> User:
+    """Create a new user"""
+    try:
+        db_user = User(
+            email=email,
+            hashed_password=hashed_password,
+            full_name=full_name
+        )
+        db.add(db_user)
+        await db.commit()
+        await db.refresh(db_user)
+        logger.info(f"User created with email: {email}")
+        return db_user
+    except IntegrityError:
+        await db.rollback()
+        logger.warning(f"User with email {email} already exists")
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="User with this email already exists"
+        )
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error creating user: {e}")
+        raise
+
+
+async def get_user_by_id(db: AsyncSession, user_id: UUID) -> Optional[User]:
+    """Get a user by ID"""
+    try:
+        result = await db.execute(select(User).filter(User.id == user_id))
+        user = result.scalar_one_or_none()
+        return user
+    except Exception as e:
+        logger.error(f"Error getting user by ID: {e}")
+        raise
+
+
+async def get_user_by_email(db: AsyncSession, email: str) -> Optional[User]:
+    """Get a user by email"""
+    try:
+        result = await db.execute(select(User).filter(User.email == email))
+        user = result.scalar_one_or_none()
+        return user
+    except Exception as e:
+        logger.error(f"Error getting user by email: {e}")
+        raise
+
+
+async def update_user(db: AsyncSession, user_id: UUID, **kwargs) -> Optional[User]:
+    """Update a user"""
+    try:
+        query = update(User).where(User.id == user_id).values(**kwargs).returning(User)
+        result = await db.execute(query)
+        await db.commit()
+
+        updated_user = result.scalar_one_or_none()
+        if updated_user:
+            logger.info(f"User updated with ID: {user_id}")
+        return updated_user
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error updating user: {e}")
+        raise
+
+
+async def delete_user(db: AsyncSession, user_id: UUID) -> bool:
+    """Delete a user"""
+    try:
+        result = await db.execute(delete(User).where(User.id == user_id))
+        await db.commit()
+        deleted_count = result.rowcount
+        if deleted_count > 0:
+            logger.info(f"User deleted with ID: {user_id}")
+        return deleted_count > 0
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error deleting user: {e}")
+        raise
+
+
+async def list_users(db: AsyncSession, skip: int = 0, limit: int = 100) -> List[User]:
+    """List users with pagination"""
+    try:
+        result = await db.execute(select(User).offset(skip).limit(limit))
+        users = result.scalars().all()
+        return users
+    except Exception as e:
+        logger.error(f"Error listing users: {e}")
+        raise
+
+
+# ChatHistory CRUD Operations
+async def create_chat_history(db: AsyncSession, user_id: UUID, query: str, response: str, context_used: Optional[str] = None) -> ChatHistory:
+    """Create a new chat history record"""
+    try:
+        db_chat_history = ChatHistory(
+            user_id=user_id,
+            query=query,
+            response=response,
+            context_used=context_used
+        )
+        db.add(db_chat_history)
+        await db.commit()
+        await db.refresh(db_chat_history)
+        logger.info(f"Chat history created for user: {user_id}")
+        return db_chat_history
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error creating chat history: {e}")
+        raise
+
+
+async def get_chat_history_by_id(db: AsyncSession, chat_history_id: UUID) -> Optional[ChatHistory]:
+    """Get a chat history record by ID"""
+    try:
+        result = await db.execute(select(ChatHistory).filter(ChatHistory.id == chat_history_id))
+        chat_history = result.scalar_one_or_none()
+        return chat_history
+    except Exception as e:
+        logger.error(f"Error getting chat history by ID: {e}")
+        raise
+
+
+async def get_chat_histories_by_user(db: AsyncSession, user_id: UUID, skip: int = 0, limit: int = 100) -> List[ChatHistory]:
+    """Get all chat histories for a user"""
+    try:
+        result = await db.execute(
+            select(ChatHistory)
+            .filter(ChatHistory.user_id == user_id)
+            .order_by(ChatHistory.created_at.desc())
+            .offset(skip)
+            .limit(limit)
+        )
+        chat_histories = result.scalars().all()
+        return chat_histories
+    except Exception as e:
+        logger.error(f"Error getting chat histories by user: {e}")
+        raise
+
+
+async def update_chat_history(db: AsyncSession, chat_history_id: UUID, **kwargs) -> Optional[ChatHistory]:
+    """Update a chat history record"""
+    try:
+        query = update(ChatHistory).where(ChatHistory.id == chat_history_id).values(**kwargs).returning(ChatHistory)
+        result = await db.execute(query)
+        await db.commit()
+
+        updated_chat_history = result.scalar_one_or_none()
+        if updated_chat_history:
+            logger.info(f"Chat history updated with ID: {chat_history_id}")
+        return updated_chat_history
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error updating chat history: {e}")
+        raise
+
+
+async def delete_chat_history(db: AsyncSession, chat_history_id: UUID) -> bool:
+    """Delete a chat history record"""
+    try:
+        result = await db.execute(delete(ChatHistory).where(ChatHistory.id == chat_history_id))
+        await db.commit()
+        deleted_count = result.rowcount
+        if deleted_count > 0:
+            logger.info(f"Chat history deleted with ID: {chat_history_id}")
+        return deleted_count > 0
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error deleting chat history: {e}")
+        raise
+
+
+# Document CRUD Operations
+async def create_document(db: AsyncSession, user_id: UUID, title: str, content: str, content_hash: str,
+                         file_path: Optional[str] = None, metadata: Optional[dict] = None) -> Document:
+    """Create a new document"""
+    try:
+        db_document = Document(
+            user_id=user_id,
+            title=title,
+            content=content,
+            content_hash=content_hash,
+            file_path=file_path,
+            metadata=metadata
+        )
+        db.add(db_document)
+        await db.commit()
+        await db.refresh(db_document)
+        logger.info(f"Document created for user: {user_id}, title: {title}")
+        return db_document
+    except IntegrityError:
+        await db.rollback()
+        logger.warning(f"Document with content_hash {content_hash} already exists")
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Document with this content already exists"
+        )
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error creating document: {e}")
+        raise
+
+
+async def get_document_by_id(db: AsyncSession, document_id: UUID) -> Optional[Document]:
+    """Get a document by ID"""
+    try:
+        result = await db.execute(select(Document).filter(Document.id == document_id))
+        document = result.scalar_one_or_none()
+        return document
+    except Exception as e:
+        logger.error(f"Error getting document by ID: {e}")
+        raise
+
+
+async def get_documents_by_user(db: AsyncSession, user_id: UUID, skip: int = 0, limit: int = 100) -> List[Document]:
+    """Get all documents for a user"""
+    try:
+        result = await db.execute(
+            select(Document)
+            .filter(Document.user_id == user_id)
+            .order_by(Document.created_at.desc())
+            .offset(skip)
+            .limit(limit)
+        )
+        documents = result.scalars().all()
+        return documents
+    except Exception as e:
+        logger.error(f"Error getting documents by user: {e}")
+        raise
+
+
+async def get_document_by_hash(db: AsyncSession, content_hash: str) -> Optional[Document]:
+    """Get a document by content hash"""
+    try:
+        result = await db.execute(select(Document).filter(Document.content_hash == content_hash))
+        document = result.scalar_one_or_none()
+        return document
+    except Exception as e:
+        logger.error(f"Error getting document by hash: {e}")
+        raise
+
+
+async def update_document(db: AsyncSession, document_id: UUID, **kwargs) -> Optional[Document]:
+    """Update a document"""
+    try:
+        query = update(Document).where(Document.id == document_id).values(**kwargs).returning(Document)
+        result = await db.execute(query)
+        await db.commit()
+
+        updated_document = result.scalar_one_or_none()
+        if updated_document:
+            logger.info(f"Document updated with ID: {document_id}")
+        return updated_document
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error updating document: {e}")
+        raise
+
+
+async def delete_document(db: AsyncSession, document_id: UUID) -> bool:
+    """Delete a document"""
+    try:
+        result = await db.execute(delete(Document).where(Document.id == document_id))
+        await db.commit()
+        deleted_count = result.rowcount
+        if deleted_count > 0:
+            logger.info(f"Document deleted with ID: {document_id}")
+        return deleted_count > 0
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error deleting document: {e}")
+        raise
+
+
+# Chat History CRUD Operations
+async def create_chat_history_entry(db: AsyncSession, user_id: UUID, query: str, response: str, context_used: Optional[str] = None) -> ChatHistory:
+    """Create a new chat history entry"""
+    try:
+        db_chat_history = ChatHistory(
+            user_id=user_id,
+            query=query,
+            response=response,
+            context_used=context_used
+        )
+        db.add(db_chat_history)
+        await db.commit()
+        await db.refresh(db_chat_history)
+        logger.info(f"Chat history created for user: {user_id}")
+        return db_chat_history
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error creating chat history: {e}")
+        raise
+
+
+async def get_chat_history_by_id(db: AsyncSession, chat_history_id: UUID) -> Optional[ChatHistory]:
+    """Get a chat history record by ID"""
+    try:
+        result = await db.execute(select(ChatHistory).filter(ChatHistory.id == chat_history_id))
+        chat_history = result.scalar_one_or_none()
+        return chat_history
+    except Exception as e:
+        logger.error(f"Error getting chat history by ID: {e}")
+        raise
+
+
+async def get_chat_histories_by_user(db: AsyncSession, user_id: UUID, skip: int = 0, limit: int = 100) -> List[ChatHistory]:
+    """Get all chat histories for a user"""
+    try:
+        result = await db.execute(
+            select(ChatHistory)
+            .filter(ChatHistory.user_id == user_id)
+            .order_by(ChatHistory.created_at.desc())
+            .offset(skip)
+            .limit(limit)
+        )
+        chat_histories = result.scalars().all()
+        return chat_histories
+    except Exception as e:
+        logger.error(f"Error getting chat histories by user: {e}")
+        raise
+
+
+async def get_user_chat_history_count(db: AsyncSession, user_id: UUID) -> int:
+    """Get the count of chat history records for a user"""
+    try:
+        from sqlalchemy import func
+        result = await db.execute(
+            select(func.count(ChatHistory.id))
+            .filter(ChatHistory.user_id == user_id)
+        )
+        count = result.scalar_one()
+        return count
+    except Exception as e:
+        logger.error(f"Error getting chat history count for user: {e}")
+        raise
+
+
+async def update_chat_history(db: AsyncSession, chat_history_id: UUID, **kwargs) -> Optional[ChatHistory]:
+    """Update a chat history record"""
+    try:
+        query = update(ChatHistory).where(ChatHistory.id == chat_history_id).values(**kwargs).returning(ChatHistory)
+        result = await db.execute(query)
+        await db.commit()
+
+        updated_chat_history = result.scalar_one_or_none()
+        if updated_chat_history:
+            logger.info(f"Chat history updated with ID: {chat_history_id}")
+        return updated_chat_history
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error updating chat history: {e}")
+        raise
+
+
+async def delete_chat_history(db: AsyncSession, chat_history_id: UUID) -> bool:
+    """Delete a chat history record"""
+    try:
+        result = await db.execute(delete(ChatHistory).where(ChatHistory.id == chat_history_id))
+        await db.commit()
+        deleted_count = result.rowcount
+        if deleted_count > 0:
+            logger.info(f"Chat history deleted with ID: {chat_history_id}")
+        return deleted_count > 0
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error deleting chat history: {e}")
+        raise
+
+
+async def delete_user_chat_history(db: AsyncSession, user_id: UUID) -> bool:
+    """Delete all chat history records for a user"""
+    try:
+        result = await db.execute(delete(ChatHistory).where(ChatHistory.user_id == user_id))
+        await db.commit()
+        deleted_count = result.rowcount
+        if deleted_count > 0:
+            logger.info(f"Deleted {deleted_count} chat history records for user: {user_id}")
+        return deleted_count > 0
+    except Exception as e:
+        await db.rollback()
+        logger.error(f"Error deleting user chat history: {e}")
+        raise
+
+
+# Utility functions
+async def get_user_with_chat_histories(db: AsyncSession, user_id: UUID) -> Optional[User]:
+    """Get a user with their chat histories"""
+    try:
+        result = await db.execute(
+            select(User)
+            .filter(User.id == user_id)
+        )
+        user = result.scalar_one_or_none()
+        return user
+    except Exception as e:
+        logger.error(f"Error getting user with chat histories: {e}")
+        raise
+
+
+async def get_user_with_documents(db: AsyncSession, user_id: UUID) -> Optional[User]:
+    """Get a user with their documents"""
+    try:
+        result = await db.execute(
+            select(User)
+            .filter(User.id == user_id)
+        )
+        user = result.scalar_one_or_none()
+        return user
+    except Exception as e:
+        logger.error(f"Error getting user with documents: {e}")
+        raise

src/db/models/chat_history.py

@@ -0,0 +1,28 @@
+"""
+ChatHistory model for the AI Backend with RAG + Authentication
+"""
+from sqlalchemy import Column, String, Text, ForeignKey, Index
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import relationship
+from uuid import uuid4
+from ...db.base import Base
+
+
+class ChatHistory(Base):
+    __tablename__ = "chat_history"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4, unique=True, nullable=False)
+    user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
+    query = Column(Text, nullable=False)
+    response = Column(Text, nullable=False)
+    context_used = Column(Text, nullable=True)  # JSON string of context snippets used
+
+    # Relationships
+    user = relationship("User", back_populates="chat_histories")
+
+    def __repr__(self):
+        return f"<ChatHistory(id={self.id}, user_id={self.user_id}, query='{self.query[:30]}...')>"
+
+# Create indexes
+Index('idx_chat_history_user_id', 'user_id')
+Index('idx_chat_history_created_at', 'created_at')

src/db/models/document.py

@@ -0,0 +1,31 @@
+"""
+Document model for the AI Backend with RAG + Authentication
+"""
+from sqlalchemy import Column, String, Text, ForeignKey, Index
+from sqlalchemy.dialects.postgresql import UUID, JSON
+from sqlalchemy.orm import relationship
+from uuid import uuid4
+from ...db.base import Base
+
+
+class Document(Base):
+    __tablename__ = "documents"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4, unique=True, nullable=False)
+    user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
+    title = Column(String(255), nullable=False)
+    content = Column(Text, nullable=False)
+    content_hash = Column(String(255), nullable=False, index=True)  # For deduplication
+    file_path = Column(String(500), nullable=True)  # Path if uploaded file
+    metadata = Column(JSON, nullable=True)  # Additional metadata as JSON
+
+    # Relationships
+    user = relationship("User", back_populates="documents")
+
+    def __repr__(self):
+        return f"<Document(id={self.id}, user_id={self.user_id}, title='{self.title}')>"
+
+# Create indexes
+Index('idx_document_user_id', 'user_id')
+Index('idx_document_content_hash', 'content_hash')
+Index('idx_document_title', 'title')

src/db/models/user.py

@@ -0,0 +1,28 @@
+"""
+User model for the AI Backend with RAG + Authentication
+"""
+from sqlalchemy import Column, String, Boolean, Text, Index
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import relationship
+from uuid import uuid4
+from ...db.base import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4, unique=True, nullable=False)
+    email = Column(String(255), unique=True, nullable=False, index=True)
+    hashed_password = Column(Text, nullable=False)
+    full_name = Column(String(255), nullable=True)
+    is_active = Column(Boolean, default=True, nullable=False)
+
+    # Relationships
+    chat_histories = relationship("ChatHistory", back_populates="user", cascade="all, delete-orphan")
+    documents = relationship("Document", back_populates="user", cascade="all, delete-orphan")
+
+    def __repr__(self):
+        return f"<User(id={self.id}, email='{self.email}', full_name='{self.full_name}')>"
+
+# Create indexes
+Index('idx_user_email', User.email)

src/embeddings/gemini_client.py

@@ -0,0 +1,335 @@
+"""
+Gemini API client for the AI Backend with RAG + Authentication
+Implements embedding generation and chat functionality using Google's Gemini API
+"""
+import google.generativeai as genai
+from google.generativeai import embedding
+from typing import List, Optional, Dict, Any
+import logging
+import time
+import asyncio
+from functools import wraps
+import re
+
+from ..config.settings import settings
+
+logger = logging.getLogger(__name__)
+
+# Initialize the Gemini API client with the API key from settings
+genai.configure(api_key=settings.gemini_api_key)
+
+
+class GeminiClient:
+    """
+    Client class to handle both Gemini embedding and chat operations
+    """
+
+    def __init__(self):
+        # Use the text-embedding-004 model for embeddings
+        self.embedding_model_name = "text-embedding-004"
+        # Use the Gemini 1.5 Flash model for chat responses (faster and more cost-effective)
+        self.chat_model_name = "gemini-1.5-flash-001"  # Updated model name
+        self.client = genai
+        self.max_retries = 3
+        self.retry_delay = 1  # seconds
+
+        # Initialize the chat model
+        self.chat_model = genai.GenerativeModel(self.chat_model_name)
+
+    # EMBEDDING METHODS
+    async def generate_embedding(self, text: str) -> Optional[List[float]]:
+        """
+        Generate embedding for the given text using Gemini text-embedding-004 model
+        """
+        for attempt in range(self.max_retries):
+            try:
+                # Generate the embedding using the Gemini API
+                result = await asyncio.get_event_loop().run_in_executor(
+                    None,
+                    lambda: genai.embed_content(
+                        model=self.embedding_model_name,
+                        content=text,
+                        task_type="RETRIEVAL_DOCUMENT",  # Optimal for RAG applications
+                        title="Document"  # Title can help with embedding quality
+                    )
+                )
+
+                embedding_values = result['embedding']
+
+                # Verify the embedding has the correct dimensions (1536 for text-embedding-004)
+                if len(embedding_values) != 1536:
+                    logger.warning(f"Generated embedding has {len(embedding_values)} dimensions, expected 1536")
+                    return None
+
+                logger.info(f"Successfully generated embedding for text of length {len(text)}")
+                return embedding_values
+
+            except Exception as e:
+                logger.warning(f"Attempt {attempt + 1} failed to generate embedding: {e}")
+
+                if attempt == self.max_retries - 1:
+                    # Last attempt failed
+                    logger.error(f"Failed to generate embedding after {self.max_retries} attempts: {e}")
+                    return None
+
+                # Wait before retrying
+                await asyncio.sleep(self.retry_delay * (2 ** attempt))  # Exponential backoff
+
+        return None
+
+    async def generate_embeddings_batch(self, texts: List[str]) -> Optional[List[List[float]]]:
+        """
+        Generate embeddings for a batch of texts
+        """
+        embeddings = []
+
+        for text in texts:
+            embedding = await self.generate_embedding(text)
+            if embedding is None:
+                logger.error(f"Failed to generate embedding for text: {text[:50]}...")
+                return None
+            embeddings.append(embedding)
+
+        return embeddings
+
+    # CHAT METHODS
+    async def generate_chat_response(
+        self,
+        query: str,
+        context: Optional[List[Dict[str, Any]]] = None,
+        conversation_history: Optional[List[Dict[str, str]]] = None
+    ) -> Optional[str]:
+        """
+        Generate a chat response using Gemini 1.5 Flash/Pro with RAG context
+        """
+        for attempt in range(self.max_retries):
+            try:
+                # Format the prompt with context and query
+                formatted_prompt = self._format_rag_prompt(query, context, conversation_history)
+
+                # Safety settings to moderate content
+                safety_settings = [
+                    {
+                        "category": "HARM_CATEGORY_DANGEROUS_CONTENT",
+                        "threshold": "BLOCK_MEDIUM_AND_ABOVE"
+                    },
+                    {
+                        "category": "HARM_CATEGORY_HATE_SPEECH",
+                        "threshold": "BLOCK_MEDIUM_AND_ABOVE"
+                    },
+                    {
+                        "category": "HARM_CATEGORY_HARASSMENT",
+                        "threshold": "BLOCK_MEDIUM_AND_ABOVE"
+                    },
+                    {
+                        "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
+                        "threshold": "BLOCK_MEDIUM_AND_ABOVE"
+                    }
+                ]
+
+                # Generate response using the chat model
+                response = await self.chat_model.generate_content_async(
+                    formatted_prompt,
+                    safety_settings=safety_settings,
+                    generation_config={
+                        "temperature": 0.3,  # Lower temperature for more consistent responses
+                        "max_output_tokens": 800,  # Limit response length
+                        "candidate_count": 1
+                    }
+                )
+
+                # Extract the text response
+                if response and response.text:
+                    logger.info(f"Successfully generated chat response for query: {query[:50]}...")
+                    return response.text.strip()
+                else:
+                    logger.warning("Gemini returned empty response")
+                    return None
+
+            except Exception as e:
+                logger.warning(f"Attempt {attempt + 1} failed to generate chat response: {e}")
+
+                if attempt == self.max_retries - 1:
+                    # Last attempt failed
+                    logger.error(f"Failed to generate chat response after {self.max_retries} attempts: {e}")
+                    return None
+
+                # Wait before retrying
+                await asyncio.sleep(self.retry_delay * (2 ** attempt))  # Exponential backoff
+
+        return None
+
+    def _format_rag_prompt(
+        self,
+        query: str,
+        context: Optional[List[Dict[str, Any]]] = None,
+        conversation_history: Optional[List[Dict[str, str]]] = None
+    ) -> str:
+        """
+        Format the prompt with RAG context and conversation history
+        """
+        prompt_parts = []
+
+        # Add system context
+        prompt_parts.append(
+            "You are an AI assistant that helps users by answering questions based on provided context. "
+            "Use only the information provided in the context to answer the questions. "
+            "If the context doesn't contain the information needed to answer the question, say so clearly. "
+            "Be helpful, accurate, and concise in your responses."
+        )
+
+        # Add conversation history if available
+        if conversation_history:
+            prompt_parts.append("\nPrevious conversation:")
+            for msg in conversation_history[-4:]:  # Use last 4 messages to avoid exceeding token limits
+                role = msg.get("role", "user")
+                content = msg.get("content", "")
+                prompt_parts.append(f"{role.capitalize()}: {content}")
+
+        # Add retrieved context if available
+        if context:
+            prompt_parts.append("\nContext for answering the question:")
+            for i, ctx in enumerate(context[:5]):  # Use top 5 context snippets
+                chunk_text = ctx.get("payload", {}).get("chunk_text", "") if isinstance(ctx, dict) else str(ctx)
+                # Clean up the chunk text if it contains the "..." marker from storage
+                if chunk_text.endswith("..."):
+                    # If it was truncated when stored, we don't have the full text
+                    # But we can still use what we have
+                    pass
+                prompt_parts.append(f"Context {i+1}: {chunk_text}")
+
+        # Add the current query
+        prompt_parts.append(f"\nQuestion: {query}")
+        prompt_parts.append("Answer:")
+
+        return "\n".join(prompt_parts)
+
+    async def moderate_content(self, text: str) -> Dict[str, Any]:
+        """
+        Moderate content using Gemini's safety features
+        """
+        try:
+            # Use the chat model to analyze content safety
+            response = await self.chat_model.generate_content_async(
+                f"Analyze the following text for safety issues: {text}",
+                safety_settings=[
+                    {
+                        "category": "HARM_CATEGORY_DANGEROUS_CONTENT",
+                        "threshold": "BLOCK_ONLY_HIGH"
+                    },
+                    {
+                        "category": "HARM_CATEGORY_HATE_SPEECH",
+                        "threshold": "BLOCK_ONLY_HIGH"
+                    },
+                    {
+                        "category": "HARM_CATEGORY_HARASSMENT",
+                        "threshold": "BLOCK_ONLY_HIGH"
+                    },
+                    {
+                        "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
+                        "threshold": "BLOCK_ONLY_HIGH"
+                    }
+                ]
+            )
+
+            # Return safety analysis
+            return {
+                "is_safe": True,
+                "text": text,
+                "moderation_applied": False  # Gemini handles moderation internally
+            }
+        except Exception as e:
+            logger.error(f"Content moderation error: {e}")
+            return {
+                "is_safe": False,
+                "text": text,
+                "moderation_applied": True,
+                "error": str(e)
+            }
+
+
+# Global instance of GeminiClient
+gemini_client = GeminiClient()
+
+
+def get_gemini_client() -> GeminiClient:
+    """Get the Gemini client instance (for both embeddings and chat)"""
+    return gemini_client
+
+
+# Embedding functions (backward compatibility)
+async def generate_embedding(text: str) -> Optional[List[float]]:
+    """
+    Generate embedding for the given text using the configured Gemini model
+    """
+    return await gemini_client.generate_embedding(text)
+
+
+async def generate_embeddings_batch(texts: List[str]) -> Optional[List[List[float]]]:
+    """
+    Generate embeddings for a batch of texts
+    """
+    return await gemini_client.generate_embeddings_batch(texts)
+
+
+# Chat functions
+async def generate_chat_response(
+    query: str,
+    context: Optional[List[Dict[str, Any]]] = None,
+    conversation_history: Optional[List[Dict[str, str]]] = None
+) -> Optional[str]:
+    """
+    Generate a chat response using Gemini with RAG context
+    """
+    return await gemini_client.generate_chat_response(query, context, conversation_history)
+
+
+async def moderate_content(text: str) -> Dict[str, Any]:
+    """
+    Moderate content using Gemini's safety features
+    """
+    return await gemini_client.moderate_content(text)
+
+
+# Decorator for rate limiting (basic implementation)
+def rate_limit(calls_per_second: float = 10):
+    """
+    Decorator to implement basic rate limiting
+    Google Gemini API has rate limits, so we need to be respectful
+    """
+    min_interval = 1.0 / calls_per_second
+    last_called = [0.0]
+
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            elapsed = time.time() - last_called[0]
+            left_to_wait = min_interval - elapsed
+            if left_to_wait > 0:
+                await asyncio.sleep(left_to_wait)
+            ret = await func(*args, **kwargs)
+            last_called[0] = time.time()
+            return ret
+        return wrapper
+    return decorator
+
+
+# Apply rate limiting to the main functions
+@rate_limit(calls_per_second=10)  # Adjust based on your API quota
+async def generate_embedding_with_rate_limit(text: str) -> Optional[List[float]]:
+    """
+    Generate embedding with rate limiting applied
+    """
+    return await generate_embedding(text)
+
+
+@rate_limit(calls_per_second=5)  # Lower rate limit for chat as it's more resource intensive
+async def generate_chat_response_with_rate_limit(
+    query: str,
+    context: Optional[List[Dict[str, Any]]] = None,
+    conversation_history: Optional[List[Dict[str, str]]] = None
+) -> Optional[str]:
+    """
+    Generate chat response with rate limiting applied
+    """
+    return await generate_chat_response(query, context, conversation_history)

src/embeddings/processor.py

@@ -0,0 +1,303 @@
+"""
+Embedding processor for the AI Backend with RAG + Authentication
+Implements text preprocessing, caching, and document chunking for embeddings
+"""
+import hashlib
+import asyncio
+from typing import List, Optional, Tuple, Dict
+import logging
+from uuid import UUID
+
+from ..config.settings import settings
+from .gemini_client import generate_embedding, generate_embeddings_batch
+from ..qdrant.operations import get_vector_operations
+from ..db import crud
+from ..config.database import get_db_session
+
+logger = logging.getLogger(__name__)
+
+# Maximum characters per chunk (Gemini has token limits)
+MAX_CHUNK_SIZE = 2000
+OVERLAP_SIZE = 200  # Overlap between chunks to maintain context
+
+
+class EmbeddingProcessor:
+    """
+    Processor class to handle embedding workflows including preprocessing,
+    caching, and document chunking
+    """
+
+    def __init__(self):
+        self.vector_ops = get_vector_operations()
+        # Simple in-memory cache (in production, use Redis or similar)
+        self.cache: Dict[str, List[float]] = {}
+
+    def _generate_content_hash(self, content: str) -> str:
+        """
+        Generate a hash for content to use for caching and deduplication
+        """
+        return hashlib.sha256(content.encode('utf-8')).hexdigest()
+
+    def _preprocess_text(self, text: str) -> str:
+        """
+        Preprocess text by cleaning and normalizing
+        """
+        if not text or not isinstance(text, str):
+            raise ValueError("Input text must be a non-empty string")
+
+        # Remove extra whitespace
+        text = ' '.join(text.split())
+
+        # Validate text length
+        if len(text) > 1000000:  # 1M characters max
+            logger.warning(f"Text is very long ({len(text)} chars), consider pre-processing")
+
+        return text.strip()
+
+    def _chunk_text(self, text: str, chunk_size: int = MAX_CHUNK_SIZE, overlap: int = OVERLAP_SIZE) -> List[str]:
+        """
+        Split text into overlapping chunks to maintain context
+        """
+        if len(text) <= chunk_size:
+            return [text]
+
+        chunks = []
+        start = 0
+
+        while start < len(text):
+            end = start + chunk_size
+
+            # If we're near the end, include the rest
+            if end > len(text):
+                end = len(text)
+                start = max(0, end - chunk_size)
+
+            chunk = text[start:end]
+
+            # If this isn't the last chunk, try to break at sentence boundary
+            if end < len(text):
+                # Look for sentence endings to break at
+                sentence_end = max(
+                    chunk.rfind('.'),
+                    chunk.rfind('!'),
+                    chunk.rfind('?'),
+                    chunk.rfind('\n')
+                )
+
+                if sentence_end > chunk_size // 2:  # Only if it's not too early
+                    end = start + sentence_end + 1
+                    chunk = text[start:end]
+
+            chunks.append(chunk)
+            start = end - overlap
+
+        return chunks
+
+    async def _get_from_cache(self, content_hash: str) -> Optional[List[float]]:
+        """
+        Get embedding from cache if available
+        """
+        return self.cache.get(content_hash)
+
+    async def _save_to_cache(self, content_hash: str, embedding: List[float]):
+        """
+        Save embedding to cache
+        """
+        self.cache[content_hash] = embedding
+
+    async def process_single_text(self, text: str, user_id: UUID) -> Optional[List[float]]:
+        """
+        Process a single text for embedding with caching
+        """
+        try:
+            # Preprocess the text
+            processed_text = self._preprocess_text(text)
+
+            if not processed_text:
+                logger.warning("Text preprocessing resulted in empty string")
+                return None
+
+            # Generate content hash for caching
+            content_hash = self._generate_content_hash(processed_text)
+
+            # Check cache first
+            cached_embedding = await self._get_from_cache(content_hash)
+            if cached_embedding:
+                logger.info(f"Found embedding in cache for text of length {len(processed_text)}")
+                return cached_embedding
+
+            # Generate embedding using Gemini
+            embedding = await generate_embedding(processed_text)
+            if embedding is None:
+                logger.error(f"Failed to generate embedding for text of length {len(processed_text)}")
+                return None
+
+            # Save to cache
+            await self._save_to_cache(content_hash, embedding)
+
+            logger.info(f"Successfully processed embedding for text of length {len(processed_text)}")
+            return embedding
+
+        except Exception as e:
+            logger.error(f"Error processing single text: {e}")
+            return None
+
+    async def process_document(
+        self,
+        document_id: UUID,
+        user_id: UUID,
+        content: str,
+        title: Optional[str] = None,
+        metadata: Optional[Dict] = None
+    ) -> bool:
+        """
+        Process a document for embedding, including chunking and storage
+        """
+        try:
+            # Preprocess the content
+            processed_content = self._preprocess_text(content)
+
+            if not processed_content:
+                logger.warning("Document content preprocessing resulted in empty string")
+                return False
+
+            # Chunk the document if it's large
+            if len(processed_content) > MAX_CHUNK_SIZE:
+                chunks = self._chunk_text(processed_content)
+                logger.info(f"Document chunked into {len(chunks)} parts")
+            else:
+                chunks = [processed_content]
+
+            # Process each chunk
+            all_embeddings = []
+            chunk_payloads = []
+
+            for i, chunk in enumerate(chunks):
+                # Generate content hash for caching
+                content_hash = self._generate_content_hash(chunk)
+
+                # Check cache first
+                embedding = await self._get_from_cache(content_hash)
+                if embedding is None:
+                    # Generate embedding using Gemini
+                    embedding = await generate_embedding(chunk)
+                    if embedding is None:
+                        logger.error(f"Failed to generate embedding for chunk {i}")
+                        continue
+
+                    # Save to cache
+                    await self._save_to_cache(content_hash, embedding)
+
+                all_embeddings.append(embedding)
+
+                # Create payload for this chunk
+                chunk_payload = {
+                    "chunk_index": i,
+                    "chunk_text": chunk[:100] + "..." if len(chunk) > 100 else chunk,  # Store first 100 chars as reference
+                    "document_id": str(document_id),
+                    "user_id": str(user_id),
+                    "title": title or "Untitled Document",
+                    "total_chunks": len(chunks)
+                }
+
+                if metadata:
+                    chunk_payload.update(metadata)
+
+                chunk_payloads.append(chunk_payload)
+
+            # Store embeddings in Qdrant
+            if all_embeddings:
+                success = await self.vector_ops.batch_upsert_vectors(
+                    user_id=user_id,
+                    document_id=document_id,
+                    embeddings_list=all_embeddings,
+                    payloads_list=chunk_payloads
+                )
+
+                if success:
+                    logger.info(f"Successfully stored {len(all_embeddings)} embeddings for document {document_id}")
+                    return True
+                else:
+                    logger.error(f"Failed to store embeddings in Qdrant for document {document_id}")
+                    return False
+            else:
+                logger.warning("No embeddings were generated for the document")
+                return False
+
+        except Exception as e:
+            logger.error(f"Error processing document {document_id}: {e}")
+            return False
+
+    async def process_texts_batch(
+        self,
+        texts: List[str],
+        user_id: UUID
+    ) -> Optional[List[List[float]]]:
+        """
+        Process a batch of texts for embedding with caching
+        """
+        try:
+            embeddings = []
+
+            for text in texts:
+                embedding = await self.process_single_text(text, user_id)
+                if embedding is None:
+                    logger.error(f"Failed to process text: {text[:50]}...")
+                    return None
+                embeddings.append(embedding)
+
+            logger.info(f"Successfully processed batch of {len(texts)} texts")
+            return embeddings
+
+        except Exception as e:
+            logger.error(f"Error processing text batch: {e}")
+            return None
+
+    async def invalidate_cache_for_document(self, document_id: UUID):
+        """
+        Remove cached embeddings associated with a document
+        In a real implementation with Redis, this would be more sophisticated
+        """
+        # In our simple in-memory cache, we can't easily identify which cache entries
+        # belong to a specific document, so we'd need to implement a more sophisticated
+        # cache structure. For now, we'll just log the action.
+        logger.info(f"Cache invalidation requested for document {document_id} (not implemented in simple cache)")
+
+
+# Global instance of EmbeddingProcessor
+embedding_processor = EmbeddingProcessor()
+
+
+def get_embedding_processor() -> EmbeddingProcessor:
+    """Get the embedding processor instance"""
+    return embedding_processor
+
+
+async def process_single_text(text: str, user_id: UUID) -> Optional[List[float]]:
+    """
+    Process a single text for embedding with caching
+    """
+    return await embedding_processor.process_single_text(text, user_id)
+
+
+async def process_document(
+    document_id: UUID,
+    user_id: UUID,
+    content: str,
+    title: Optional[str] = None,
+    metadata: Optional[Dict] = None
+) -> bool:
+    """
+    Process a document for embedding, including chunking and storage
+    """
+    return await embedding_processor.process_document(document_id, user_id, content, title, metadata)
+
+
+async def process_texts_batch(
+    texts: List[str],
+    user_id: UUID
+) -> Optional[List[List[float]]]:
+    """
+    Process a batch of texts for embedding with caching
+    """
+    return await embedding_processor.process_texts_batch(texts, user_id)

src/main.py

@@ -0,0 +1,51 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+import logging
+import sys
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    handlers=[
+        logging.StreamHandler(sys.stdout)
+    ]
+)
+
+logger = logging.getLogger(__name__)
+
+# Import settings - this will validate environment variables on startup
+from .config.settings import settings
+
+app = FastAPI(
+    title="AI Backend with RAG + Authentication",
+    description="A scalable backend featuring authentication, RAG capabilities, and integration with external services",
+    version="1.0.0"
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # In production, replace with specific origins
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/")
+async def root():
+    return {"message": "AI Backend with RAG + Authentication is running!"}
+
+
+# Include all routes
+from .routes import auth, search, history, documents, health
+
+app.include_router(auth.router, prefix="/auth", tags=["authentication"])
+app.include_router(search.router, prefix="/search", tags=["search"])
+app.include_router(history.router, prefix="/history", tags=["history"])
+app.include_router(documents.router, prefix="/documents", tags=["documents"])
+app.include_router(health.router)
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host=settings.server_host, port=settings.server_port)

src/models/documents.py

@@ -0,0 +1,32 @@
+"""
+Document models for the AI Backend with RAG + Authentication
+Pydantic models for document-related request/response validation
+"""
+from pydantic import BaseModel, Field
+from typing import Optional, Dict, Any
+from uuid import UUID
+from datetime import datetime
+
+
+class DocumentCreate(BaseModel):
+    title: str = Field(..., min_length=1, max_length=255, description="Document title")
+    content: str = Field(..., min_length=1, description="Document content")
+    file_path: Optional[str] = Field(None, max_length=500, description="Path if uploaded file")
+    metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata")
+
+
+class DocumentResponse(BaseModel):
+    document_id: UUID
+    success: bool
+    message: str
+
+
+class DocumentUpdate(BaseModel):
+    title: Optional[str] = Field(None, min_length=1, max_length=255)
+    content: Optional[str] = Field(None, min_length=1)
+    metadata: Optional[Dict[str, Any]] = None
+
+
+class DocumentListResponse(BaseModel):
+    documents: list
+    total: int

src/models/search.py

@@ -0,0 +1,43 @@
+"""
+Search models for the AI Backend with RAG + Authentication
+Pydantic models for search-related request/response validation
+"""
+from pydantic import BaseModel, Field
+from typing import Optional, List, Dict, Any
+from uuid import UUID
+
+
+class SearchRequest(BaseModel):
+    query: str = Field(..., min_length=1, max_length=1000, description="Search query")
+    top_k: Optional[int] = Field(default=5, ge=1, le=20, description="Number of results to return")
+    filters: Optional[Dict[str, Any]] = Field(None, description="Additional filters for search")
+
+
+class SearchResult(BaseModel):
+    id: str
+    document_id: str
+    score: float
+    payload: Dict[str, Any]
+
+
+class SearchResponse(BaseModel):
+    results: List[SearchResult]
+    query: str
+    total_results: int
+
+
+class Message(BaseModel):
+    role: str = Field(..., pattern=r"^(user|assistant|system)$", description="Role of the message sender")
+    content: str = Field(..., min_length=1, description="Content of the message")
+
+
+class ChatRequest(BaseModel):
+    query: str = Field(..., min_length=1, max_length=1000, description="Chat query")
+    top_k: Optional[int] = Field(default=5, ge=1, le=10, description="Number of context results to retrieve")
+    conversation_history: Optional[List[Message]] = Field(None, description="Previous conversation messages")
+
+
+class ChatResponse(BaseModel):
+    response: str
+    sources: List[Dict[str, Any]]
+    context_used: List[Dict[str, Any]]

src/qdrant/client.py

@@ -0,0 +1,140 @@
+"""
+Qdrant client setup for the AI Backend with RAG + Authentication
+Implements Qdrant client initialization and connection management
+"""
+from qdrant_client import QdrantClient
+from qdrant_client.http import models
+from qdrant_client.http.exceptions import UnexpectedResponse
+from typing import Optional
+import logging
+
+from ..config.settings import settings
+
+logger = logging.getLogger(__name__)
+
+# Vector dimensions for Gemini text-embedding-004 model
+VECTOR_DIMENSIONS = 1536  # Standard for text-embedding-004
+
+class QdrantService:
+    """
+    Service class to manage Qdrant client and operations
+    """
+
+    def __init__(self):
+        self.client: Optional[QdrantClient] = None
+        self._initialize_client()
+
+    def _initialize_client(self):
+        """Initialize the Qdrant client with settings from configuration"""
+        try:
+            if settings.qdrant_api_key:
+                if settings.qdrant_url.startswith('https://'):
+                    # For cloud instances, use the URL directly with API key
+                    self.client = QdrantClient(
+                        url=settings.qdrant_url,
+                        api_key=settings.qdrant_api_key,
+                        timeout=10.0  # 10 second timeout
+                    )
+                else:
+                    # For local instances
+                    self.client = QdrantClient(
+                        url=settings.qdrant_url,
+                        api_key=settings.qdrant_api_key,
+                        timeout=10.0  # 10 second timeout
+                    )
+            else:
+                # If no API key is provided, connect without authentication
+                self.client = QdrantClient(
+                    url=settings.qdrant_url,
+                    timeout=10.0  # 10 second timeout
+                )
+
+            logger.info("Qdrant client initialized successfully")
+        except Exception as e:
+            logger.error(f"Failed to initialize Qdrant client: {e}")
+            raise
+
+    async def health_check(self) -> bool:
+        """Check if Qdrant server is accessible"""
+        try:
+            # Try to get cluster info as a health check
+            if self.client:
+                cluster_info = self.client.get_cluster_info()
+                logger.info(f"Qdrant health check passed. Cluster: {cluster_info}")
+                return True
+            return False
+        except Exception as e:
+            logger.error(f"Qdrant health check failed: {e}")
+            return False
+
+    def get_client(self) -> QdrantClient:
+        """Get the initialized Qdrant client"""
+        if self.client is None:
+            raise RuntimeError("Qdrant client not initialized")
+        return self.client
+
+
+# Global instance of QdrantService
+qdrant_service = QdrantService()
+
+
+def get_qdrant_client() -> QdrantClient:
+    """Get the Qdrant client instance"""
+    return qdrant_service.get_client()
+
+
+async def initialize_qdrant_collections():
+    """Initialize required collections in Qdrant"""
+    try:
+        client = get_qdrant_client()
+
+        # Check if the documents collection already exists
+        collections = client.get_collections()
+        collection_names = [collection.name for collection in collections.collections]
+
+        if "documents" not in collection_names:
+            # Create the documents collection with proper vector configuration
+            client.create_collection(
+                collection_name="documents",
+                vectors_config=models.VectorParams(
+                    size=VECTOR_DIMENSIONS,
+                    distance=models.Distance.COSINE  # Cosine distance is good for embeddings
+                )
+            )
+            logger.info("Created 'documents' collection in Qdrant")
+        else:
+            logger.info("Collection 'documents' already exists in Qdrant")
+
+        # Verify the collection has the correct configuration
+        collection_info = client.get_collection(collection_name="documents")
+        vector_config = collection_info.config.params.vectors
+        if hasattr(vector_config, 'size'):
+            actual_size = vector_config.size
+        else:
+            # Handle the case where vector_config is a dictionary
+            actual_size = vector_config['size'] if isinstance(vector_config, dict) else vector_config
+
+        if actual_size != VECTOR_DIMENSIONS:
+            logger.warning(f"Collection vector size is {actual_size}, expected {VECTOR_DIMENSIONS}")
+        else:
+            logger.info(f"Collection 'documents' has correct vector size: {VECTOR_DIMENSIONS}")
+
+        return True
+
+    except UnexpectedResponse as e:
+        logger.error(f"Qdrant API error during collection initialization: {e}")
+        return False
+    except Exception as e:
+        logger.error(f"Unexpected error during collection initialization: {e}")
+        return False
+
+
+# Initialize collections on module import
+async def setup_qdrant():
+    """Setup function to initialize Qdrant collections"""
+    success = await initialize_qdrant_collections()
+    if success:
+        logger.info("Qdrant setup completed successfully")
+    else:
+        logger.error("Qdrant setup failed")
+    return success