Upload 34 files

Dockerfile

@@ -0,0 +1,60 @@
+# Hugging Face Spaces - Docker SDK
+# Schema-Agnostic Database Chatbot with RAG
+
+FROM python:3.11-slim
+
+# Set working directory
+WORKDIR /app
+
+# Set environment variables
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONPATH=/app \
+    HF_HOME=/app/.cache \
+    TRANSFORMERS_CACHE=/app/.cache/transformers \
+    SENTENCE_TRANSFORMERS_HOME=/app/.cache/sentence_transformers
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    curl \
+    git \
+    libpq-dev \
+    && rm -rf /var/lib/apt/lists/* \
+    && apt-get clean
+
+# Create a non-root user for security
+RUN useradd -m -u 1000 appuser
+
+# Create cache directories with proper permissions
+RUN mkdir -p /app/.cache/sentence_transformers /app/.cache/transformers /app/faiss_index \
+    && chown -R appuser:appuser /app
+
+# Copy requirements first for better caching
+COPY --chown=appuser:appuser requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY --chown=appuser:appuser . .
+
+# Switch to non-root user
+USER appuser
+
+# Expose Streamlit port (HF Spaces expects 7860)
+EXPOSE 7860
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
+    CMD curl --fail http://localhost:7860/_stcore/health || exit 1
+
+# Run Streamlit
+CMD ["streamlit", "run", "app.py", \
+    "--server.port=7860", \
+    "--server.address=0.0.0.0", \
+    "--server.enableCORS=true", \
+    "--server.enableXsrfProtection=false", \
+    "--browser.gatherUsageStats=false", \
+    "--server.fileWatcherType=none"]

README.md

@@ -1,11 +1,111 @@
----
-title: DB Chatbot
-emoji: 🌖
-colorFrom: gray
-colorTo: yellow
-sdk: docker
-pinned: false
-license: mit
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: Database Copilot
+emoji: 🤖
+colorFrom: blue
+colorTo: purple
+sdk: docker
+pinned: false
+license: mit
+app_port: 7860
+---
+
+# 🤖 Database Copilot
+
+A production-grade, **schema-agnostic chatbot** that connects to **any** database (MySQL, PostgreSQL, or SQLite) and provides intelligent querying through **RAG** (Retrieval-Augmented Generation) and **Text-to-SQL**.
+
+**🆓 Powered by Groq for FREE LLM inference!**
+
+## 🌟 Features
+
+- **Multi-Database Support**: Works with **MySQL**, **PostgreSQL**, and **SQLite**
+- **Schema-Agnostic**: Works with ANY database schema - no hardcoding required
+- **Dynamic Introspection**: Automatically discovers tables, columns, and relationships
+- **Hybrid Query Routing**: Intelligently routes queries to RAG or SQL based on intent
+- **Semantic Search (RAG)**: FAISS-based vector search for text content
+- **Text-to-SQL**: LLM-powered SQL generation with dialect-specific syntax
+- **Security First**: Read-only queries, SQL validation, table whitelisting
+- **FREE LLM**: Uses Groq API (free tier) with Llama 3.3, Mixtral, and Gemma models
+
+## 🚀 Getting Started
+
+### 1. Configure Secrets
+
+This Space requires the following secrets to be set in your Hugging Face Space settings:
+
+**Required:**
+| Secret Name | Description |
+|------------|-------------|
+| `GROQ_API_KEY` | Your Groq API key ([Get FREE key](https://console.groq.com)) |
+
+**Database Configuration (choose one):**
+
+#### For MySQL:
+| Secret Name | Description |
+|------------|-------------|
+| `DB_TYPE` | Set to `mysql` |
+| `DB_HOST` | MySQL server hostname |
+| `DB_PORT` | MySQL port (default: 3306) |
+| `DB_DATABASE` | Database name |
+| `DB_USERNAME` | Database username |
+| `DB_PASSWORD` | Database password |
+
+#### For PostgreSQL:
+| Secret Name | Description |
+|------------|-------------|
+| `DB_TYPE` | Set to `postgresql` |
+| `DB_HOST` | PostgreSQL server hostname |
+| `DB_PORT` | PostgreSQL port (default: 5432) |
+| `DB_DATABASE` | Database name |
+| `DB_USERNAME` | Database username |
+| `DB_PASSWORD` | Database password |
+
+#### For SQLite:
+| Secret Name | Description |
+|------------|-------------|
+| `DB_TYPE` | Set to `sqlite` |
+| `SQLITE_PATH` | Path to SQLite database file |
+
+**Optional:**
+| Secret Name | Description | Default |
+|------------|-------------|---------|
+| `GROQ_MODEL` | Groq model to use | `llama-3.3-70b-versatile` |
+| `DB_SSL_CA` | Path to SSL CA certificate | None |
+
+### 2. Connect & Use
+
+1. Click **"Connect & Initialize"** in the sidebar
+2. Click **"Index Text Data"** to enable semantic search
+3. Start asking questions about your data!
+
+## 💬 Example Queries
+
+**Semantic Search (RAG):**
+- "What products are related to electronics?"
+- "Tell me about customer feedback on shipping"
+
+**Structured Queries (SQL):**
+- "How many orders were placed last month?"
+- "Show me the top 10 customers by revenue"
+
+**Hybrid:**
+- "Find customers who complained about delivery and show their order count"
+
+## 🔒 Security
+
+- **Read-Only Transactions**: All queries run in read-only mode
+- **SQL Validation**: Only SELECT statements allowed
+- **Forbidden Keywords**: INSERT, UPDATE, DELETE, DROP, etc. are blocked
+- **Table Whitelisting**: Only discovered tables are queryable
+- **Automatic LIMIT**: All queries have LIMIT clauses enforced
+
+## 🆓 Why Groq?
+
+[Groq](https://console.groq.com) provides **FREE API access** with incredibly fast inference:
+- **Llama 3.3 70B** - Best quality, state-of-the-art
+- **Llama 3.1 8B Instant** - Fastest responses
+- **Mixtral 8x7B** - Great for code and SQL
+- **Gemma 2 9B** - Google's efficient model
+
+## 📝 License
+
+MIT License

app.py

@@ -0,0 +1,336 @@
+"""
+Schema-Agnostic Database Chatbot - Streamlit Application
+
+A production-grade chatbot that connects to ANY MySQL database
+and provides intelligent querying through RAG and Text-to-SQL.
+
+Uses Groq for FREE LLM inference!
+"""
+
+import os
+from pathlib import Path
+
+# Load .env FIRST before any other imports
+from dotenv import load_dotenv
+load_dotenv(Path(__file__).parent / ".env")
+
+import streamlit as st
+import uuid
+from datetime import datetime
+
+# Page config must be first
+st.set_page_config(
+    page_title="Database Copilot",
+    page_icon="🤖",
+    layout="wide",
+    initial_sidebar_state="expanded"
+)
+
+# Imports
+from config import config
+from database import get_db, get_schema, get_introspector
+from llm import create_llm_client
+from chatbot import create_chatbot, DatabaseChatbot
+from memory import create_memory, create_enhanced_memory, EnhancedChatMemory
+
+
+# Groq models (all FREE!)
+GROQ_MODELS = [
+    "llama-3.3-70b-versatile",
+    "llama-3.1-8b-instant",
+    "mixtral-8x7b-32768",
+    "gemma2-9b-it"
+]
+
+
+def init_session_state():
+    """Initialize Streamlit session state."""
+    if "session_id" not in st.session_state:
+        st.session_state.session_id = str(uuid.uuid4())
+    
+    if "messages" not in st.session_state:
+        st.session_state.messages = []
+    
+    if "chatbot" not in st.session_state:
+        st.session_state.chatbot = None
+    
+    if "initialized" not in st.session_state:
+        st.session_state.initialized = False
+    
+    if "user_id" not in st.session_state:
+        st.session_state.user_id = "default"
+    
+    if "enable_summarization" not in st.session_state:
+        st.session_state.enable_summarization = True
+    
+    if "summary_threshold" not in st.session_state:
+        st.session_state.summary_threshold = 10
+        
+    if "memory" not in st.session_state:
+        st.session_state.memory = create_enhanced_memory(
+            st.session_state.session_id, 
+            user_id=st.session_state.user_id,
+            enable_summarization=st.session_state.enable_summarization,
+            summary_threshold=st.session_state.summary_threshold
+        )
+        # Clear temporary memory on fresh load/reload
+        st.session_state.memory.clear_user_history()
+    
+    if "indexed" not in st.session_state:
+        st.session_state.indexed = False
+
+
+def render_sidebar():
+    """Render the configuration sidebar."""
+    with st.sidebar:
+        st.title("⚙️ Settings")
+        
+        # User Profile
+        st.subheader("👤 User Profile")
+        user_id = st.text_input(
+            "User ID / Name", 
+            value=st.session_state.get("user_id", "default"),
+            key="user_id_input",
+            help="Your unique ID for private memory storage"
+        )
+        if user_id != st.session_state.get("user_id"):
+            # USER ID CHANGE - Same behavior as "New Chat":
+            # 1. Clear temporary memory (session history) for clean start
+            # 2. Permanent memory remains UNTOUCHED (per-user storage)
+            st.session_state.user_id = user_id
+            st.session_state.session_id = str(uuid.uuid4())  # New session
+            st.session_state.messages = []  # Clear UI chat history
+            
+            # Create memory for new user and clear their temp history (fresh start)
+            st.session_state.memory = create_enhanced_memory(
+                st.session_state.session_id, 
+                user_id=user_id,
+                enable_summarization=st.session_state.enable_summarization,
+                summary_threshold=st.session_state.summary_threshold
+            )
+            st.session_state.memory.clear_user_history()  # Clears _chatbot_memory, NOT _chatbot_permanent_memory_v2
+            st.rerun()
+        
+        st.divider()
+        
+        # Initialize Button
+        if st.button("🚀 Connect & Initialize", use_container_width=True, type="primary"):
+            with st.spinner("Connecting to database..."):
+                success = initialize_chatbot()
+                if success:
+                    st.success("✅ Connected!")
+                    st.rerun()
+        
+        # Index Button (after initialization)
+        if st.session_state.initialized:
+            if st.button("📚 Index Text Data", use_container_width=True):
+                with st.spinner("Indexing text data..."):
+                    index_data()
+                    st.success("✅ Indexed!")
+                    st.rerun()
+        
+        st.divider()
+        
+        # Status
+        st.subheader("📊 Status")
+        if st.session_state.initialized:
+            st.success("Database: Connected")
+            schema = get_schema()
+            st.info(f"Tables: {len(schema.tables)}")
+            
+            if st.session_state.indexed:
+                from rag import get_rag_engine
+                engine = get_rag_engine()
+                st.info(f"Indexed Docs: {engine.document_count}")
+        else:
+            st.warning("Not connected")
+        
+        # New Chat (Context Switch)
+        # New Chat (Context Switch)
+        if st.button("➕ New Chat", use_container_width=True, type="secondary"):
+            # Clear previous session from DB
+            if "memory" in st.session_state and st.session_state.memory:
+                st.session_state.memory.clear()
+            
+            st.session_state.messages = []
+            st.session_state.session_id = str(uuid.uuid4())  # Generate new session ID
+            
+            # Preserve current user ID and memory settings
+            current_user = st.session_state.get("user_id", "default")
+            st.session_state.memory = create_enhanced_memory(
+                st.session_state.session_id, 
+                user_id=current_user,
+                enable_summarization=st.session_state.enable_summarization,
+                summary_threshold=st.session_state.summary_threshold
+            )
+            # Set LLM client if available
+            if "llm" in st.session_state and st.session_state.llm:
+                st.session_state.memory.set_llm_client(st.session_state.llm)
+            st.rerun()
+
+
+def initialize_chatbot() -> bool:
+    """Initialize the chatbot using environment variables."""
+    try:
+        # Use Groq as default provider (from environment)
+        api_key = os.getenv("GROQ_API_KEY", "")
+        model = os.getenv("GROQ_MODEL", "llama-3.3-70b-versatile")
+        
+        if not api_key:
+            st.error("GROQ_API_KEY not configured. Please set it in your .env file.")
+            return False
+        
+        llm = create_llm_client("groq", api_key=api_key, model=model)
+        
+        # Create and initialize chatbot
+        chatbot = create_chatbot(llm)
+        
+        # Explicitly set LLM client (also configures router and sql_generator)
+        chatbot.set_llm_client(llm)
+        
+        success, msg = chatbot.initialize()
+        
+        if success:
+            st.session_state.chatbot = chatbot
+            st.session_state.llm = llm  # Store LLM separately too
+            st.session_state.initialized = True
+            
+            # Set LLM client on memory for summarization
+            if hasattr(st.session_state.memory, 'set_llm_client'):
+                st.session_state.memory.set_llm_client(llm)
+            
+            return True
+        else:
+            st.error(f"Initialization failed: {msg}")
+            return False
+            
+    except Exception as e:
+        st.error(f"Error: {str(e)}")
+        return False
+
+
+def index_data():
+    """Index text data from the database."""
+    if st.session_state.chatbot:
+        progress = st.progress(0)
+        status = st.empty()
+        
+        schema = get_schema()
+        total_tables = len(schema.tables)
+        indexed = 0
+        
+        def progress_callback(table_name, docs):
+            nonlocal indexed
+            indexed += 1
+            progress.progress(indexed / total_tables)
+            status.text(f"Indexed {table_name}: {docs} documents")
+        
+        total_docs = st.session_state.chatbot.index_text_data(progress_callback)
+        st.session_state.indexed = True
+        status.text(f"Total: {total_docs} documents indexed")
+
+
+def render_schema_explorer():
+    """Render schema explorer in an expander."""
+    if not st.session_state.initialized:
+        return
+    
+    with st.expander("📋 Database Schema", expanded=False):
+        schema = get_schema()
+        
+        for table_name, table_info in schema.tables.items():
+            with st.container():
+                st.markdown(f"**{table_name}** ({table_info.row_count or '?'} rows)")
+                
+                cols = []
+                for col in table_info.columns:
+                    pk = "🔑" if col.is_primary_key else ""
+                    txt = "📝" if col.is_text_type else ""
+                    cols.append(f"`{col.name}` {col.data_type} {pk}{txt}")
+                
+                st.caption(" | ".join(cols))
+                st.divider()
+
+
+def render_chat_interface():
+    """Render the main chat interface."""
+    st.title("🤖 Database Copilot")
+    st.caption("Schema-agnostic chatbot powered by Groq (FREE!)")
+    
+    # Schema explorer
+    render_schema_explorer()
+    
+    # Chat container
+    chat_container = st.container()
+    
+    with chat_container:
+        # Display messages
+        for msg in st.session_state.messages:
+            with st.chat_message(msg["role"]):
+                st.markdown(msg["content"])
+                
+                # Show metadata for assistant messages
+                if msg["role"] == "assistant" and "metadata" in msg:
+                    meta = msg["metadata"]
+                    if meta.get("query_type"):
+                        st.caption(f"Query type: {meta['query_type']}")
+                    if meta.get("sql_query"):
+                        with st.expander("SQL Query"):
+                            st.code(meta["sql_query"], language="sql")
+    
+    # Chat input
+    if prompt := st.chat_input("Ask about your data..."):
+        if not st.session_state.initialized:
+            st.error("Please connect to a database first!")
+            return
+        
+        # Add user message
+        st.session_state.messages.append({"role": "user", "content": prompt})
+        st.session_state.memory.add_message("user", prompt)
+        
+        with st.chat_message("user"):
+            st.markdown(prompt)
+        
+        # Get response
+        with st.chat_message("assistant"):
+            with st.spinner("Thinking..."):
+                response = st.session_state.chatbot.chat(
+                    prompt, 
+                    st.session_state.memory
+                )
+                
+                st.markdown(response.answer)
+                
+                # Show metadata
+                if response.query_type != "general":
+                    st.caption(f"Query type: {response.query_type}")
+                
+                if response.sql_query:
+                    with st.expander("SQL Query"):
+                        st.code(response.sql_query, language="sql")
+                
+                if response.sql_results:
+                    with st.expander("Results"):
+                        st.dataframe(response.sql_results)
+        
+        # Save to memory
+        st.session_state.messages.append({
+            "role": "assistant",
+            "content": response.answer,
+            "metadata": {
+                "query_type": response.query_type,
+                "sql_query": response.sql_query
+            }
+        })
+        st.session_state.memory.add_message("assistant", response.answer)
+
+
+def main():
+    """Main application entry point."""
+    init_session_state()
+    render_sidebar()
+    render_chat_interface()
+
+
+if __name__ == "__main__":
+    main()

chatbot.py

@@ -0,0 +1,391 @@
+"""
+Chatbot Core - Main orchestrator for the schema-agnostic database chatbot.
+
+Combines all components:
+- Schema introspection
+- Query routing
+- RAG retrieval
+- SQL generation & execution
+- Response generation
+"""
+
+import logging
+from typing import Dict, Any, List, Optional, Tuple
+from dataclasses import dataclass
+
+from database import get_db, get_schema, get_introspector
+from rag import get_rag_engine
+from sql import get_sql_generator, get_sql_validator
+from llm import create_llm_client, LLMClient
+from router import get_query_router, QueryType
+from memory import ChatMemory, EnhancedChatMemory, create_memory
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ChatResponse:
+    """Response from the chatbot."""
+    answer: str
+    query_type: str
+    sources: List[Dict[str, Any]] = None
+    sql_query: Optional[str] = None
+    sql_results: Optional[List[Dict]] = None
+    error: Optional[str] = None
+    
+    def __post_init__(self):
+        if self.sources is None:
+            self.sources = []
+
+
+class DatabaseChatbot:
+    """Main chatbot class orchestrating all components."""
+    
+    RESPONSE_PROMPT = """You are a helpful database assistant. Answer the user's question based on the provided context.
+
+IMPORTANT: Use the conversation history to understand follow-up questions. If the user refers to "it", "that", "the product", etc., look at the previous messages to understand what they're referring to.
+
+{context}
+
+USER QUESTION: {question}
+
+INSTRUCTIONS:
+- Answer ONLY based on the provided context AND conversation history
+- Do NOT use outside knowledge, general assumptions, or hallucinate facts
+- If the context doesn't contain the answer, explicitly state that the information is not available in the database
+- Resolve pronouns using previous messages
+- Be concise but complete
+- Format data nicely
+
+YOUR RESPONSE:"""
+
+    def __init__(self, llm_client: Optional[LLMClient] = None):
+        self.db = get_db()
+        self.introspector = get_introspector()
+        self.rag_engine = get_rag_engine()
+        # Pass database type to SQL generator for dialect-specific SQL
+        db_type = self.db.db_type.value
+        self.sql_generator = get_sql_generator(db_type)
+        self.sql_validator = get_sql_validator()
+        self.router = get_query_router()
+        self.llm_client = llm_client
+        
+        self._schema_initialized = False
+        self._rag_initialized = False
+    
+    def set_llm_client(self, llm_client: LLMClient):
+        """Configure the LLM client."""
+        self.llm_client = llm_client
+        self.sql_generator.set_llm_client(llm_client)
+        self.router.set_llm_client(llm_client)
+    
+    def initialize(self) -> Tuple[bool, str]:
+        """Initialize the chatbot by introspecting the database."""
+        try:
+            # Test connection
+            success, msg = self.db.test_connection()
+            if not success:
+                return False, f"Database connection failed: {msg}"
+            
+            # Introspect schema
+            schema = self.introspector.introspect(force_refresh=True)
+            
+            # Configure SQL validator with discovered tables
+            self.sql_validator.set_allowed_tables(schema.table_names)
+            
+            self._schema_initialized = True
+            
+            return True, f"Initialized with {len(schema.tables)} tables"
+            
+        except Exception as e:
+            logger.error(f"Initialization failed: {e}")
+            return False, str(e)
+    
+    def index_text_data(self, progress_callback=None) -> int:
+        """Index all text data for RAG."""
+        if not self._schema_initialized:
+            raise RuntimeError("Chatbot not initialized. Call initialize() first.")
+        
+        schema = get_schema()
+        total_docs = 0
+        
+        for table_name, table_info in schema.tables.items():
+            text_cols = [c.name for c in table_info.text_columns]
+            if not text_cols:
+                continue
+            
+            pk = table_info.primary_keys[0] if table_info.primary_keys else None
+            cols_to_select = text_cols + ([pk] if pk else [])
+            
+            query = f"SELECT {', '.join(cols_to_select)} FROM {table_name} LIMIT 1000"
+            
+            try:
+                rows = self.db.execute_query(query)
+                docs = self.rag_engine.index_table(table_name, rows, text_cols, pk)
+                total_docs += docs
+                
+                if progress_callback:
+                    progress_callback(table_name, docs)
+                    
+            except Exception as e:
+                logger.warning(f"Failed to index {table_name}: {e}")
+        
+        self.rag_engine.save()
+        self._rag_initialized = True
+        
+        return total_docs
+    
+    def chat(self, query: str, memory: Optional[ChatMemory] = None) -> ChatResponse:
+        """Process a user query and return a response."""
+        if not self._schema_initialized:
+            return ChatResponse(answer="Chatbot not initialized.", query_type="error",
+                              error="Call initialize() first")
+        
+        if not self.llm_client:
+            return ChatResponse(answer="LLM not configured.", query_type="error",
+                              error="Configure LLM client first")
+        
+        try:
+            schema = get_schema()
+            schema_context = schema.to_context_string()
+            
+            # Check for memory commands
+            # Check for memory commands
+            # Check for memory commands using regex for flexibility
+            import re
+            save_pattern = re.compile(r"(?:please\s+)?(?:save|remember|memorize)\s+(?:this|that)?\s*(?:to\s+(?:main\s+)?memory)?\s*(?:that)?\s*:?\s*(.*)", re.IGNORECASE)
+            match = save_pattern.match(query.strip())
+            
+            # Check if it looks like a command (starts with command words)
+            is_command = bool(match) and (
+                query.lower().startswith(("save", "remember", "memorize")) or 
+                "saved to" in query.lower() # specific user case "saved to main memory"
+            )
+
+            if is_command and memory:
+                content_to_save = match.group(1).strip() if match else ""
+                
+                # If specific content is provided (e.g. "Remember that I like pizza")
+                if content_to_save:
+                    # Save the explicit content
+                    success = memory.save_permanent_context(content_to_save)
+                    if success:
+                        return ChatResponse(answer=f"💾 I've saved to your permanent memory: '{content_to_save}'", query_type="memory")
+                    else:
+                        return ChatResponse(answer="❌ Failed to save to permanent memory. Please try again.", query_type="memory")
+
+                # If no content (e.g. "Save this"), save the previous conversation turn
+                elif len(memory.messages) >= 2:
+                    # [-1] is current command ("save to memory")
+                    # [-2] is previous assistant response
+                    # [-3] is previous user query (context for the response)
+                    
+                    msgs_to_save = []
+                    # We try to grab the last QA pair: User Prompt + AI Response
+                    # memory.messages structure: [User, AI, User, AI, User(current)]
+                    
+                    if len(memory.messages) >= 3:
+                        msg_user = memory.messages[-3]
+                        msg_ai = memory.messages[-2]
+                        
+                        # Verify roles to ensure we are saving a Q&A pair
+                        if msg_user.role == "user" and msg_ai.role == "assistant":
+                            msgs_to_save = [msg_user, msg_ai]
+                            
+                    if msgs_to_save:
+                        # Format: "User: ... | Assistant: ..."
+                        context_str = f"User: {msgs_to_save[0].content} | Assistant: {msgs_to_save[1].content}"
+                        success = memory.save_permanent_context(context_str)
+                        if success:
+                            return ChatResponse(answer="💾 I've saved our last exchange to your permanent memory.", query_type="memory")
+                        else:
+                            return ChatResponse(answer="❌ Failed to save to permanent memory.", query_type="memory")
+                    else:
+                        return ChatResponse(answer="⚠️ I couldn't find a clear previous exchange to save. Try saying 'Remember that [fact]'.", query_type="memory")
+                else:
+                    return ChatResponse(answer="⚠️ Nothing previous to save. Tell me something to remember first!", query_type="memory")
+
+            # Route the query
+            routing = self.router.route(query, schema_context)
+            
+            # Get chat history for context
+            history = memory.get_context_messages(5) if memory else []
+            
+            # Process based on route
+            if routing.query_type == QueryType.RAG:
+                return self._handle_rag(query, history)
+            elif routing.query_type == QueryType.SQL:
+                return self._handle_sql(query, schema_context, history)
+            elif routing.query_type == QueryType.HYBRID:
+                return self._handle_hybrid(query, schema_context, history)
+            else:
+                return self._handle_general(query, history)
+                
+        except Exception as e:
+            logger.error(f"Chat error: {e}")
+            return ChatResponse(answer=f"Error: {str(e)}", query_type="error", error=str(e))
+    
+    def _handle_rag(self, query: str, history: List[Dict]) -> ChatResponse:
+        """Handle RAG-based query."""
+        context = self.rag_engine.get_context(query, top_k=5)
+        
+        prompt = self.RESPONSE_PROMPT.format(context=f"RELEVANT DATA:\n{context}", question=query)
+        
+        messages = self._construct_messages(
+            "You are a helpful database assistant.",
+            history, 
+            prompt
+        )
+        
+        answer = self.llm_client.chat(messages)
+        
+        return ChatResponse(answer=answer, query_type="rag",
+                          sources=[{"type": "semantic_search", "context": context[:500]}])
+    
+    def _handle_sql(self, query: str, schema_context: str, history: List[Dict]) -> ChatResponse:
+        """Handle SQL-based query."""
+        sql, explanation = self.sql_generator.generate(query, schema_context, history)
+        
+        # Validate SQL
+        is_valid, msg, sanitized_sql = self.sql_validator.validate(sql)
+        if not is_valid:
+            return ChatResponse(answer=f"Could not generate safe query: {msg}",
+                              query_type="sql", error=msg)
+        
+        # Execute query
+        try:
+            results = self.db.execute_query(sanitized_sql)
+        except Exception as e:
+            return ChatResponse(answer=f"Query execution failed: {e}",
+                              query_type="sql", sql_query=sanitized_sql, error=str(e))
+        
+        # SMART FALLBACK: If SQL returns nothing, it might be a semantic issue (e.g. wrong column)
+        # We try RAG as a fallback if SQL found nothing
+        if not results:
+            logger.info(f"SQL returned no results for query: '{query}'. Falling back to RAG.")
+            rag_response = self._handle_rag(query, history)
+            
+            # Combine the info: "I couldn't find an exact match in the rows, but here is what I found semantically:"
+            rag_response.answer = f"I couldn't find a direct match using a database query, but here is what I found in the product descriptions:\n\n{rag_response.answer}"
+            rag_response.query_type = "hybrid_fallback"
+            rag_response.sql_query = sanitized_sql
+            return rag_response
+
+        # Generate response
+        context = f"SQL QUERY:\n{sanitized_sql}\n\nRESULTS:\n{self._format_results(results)}"
+        prompt = self.RESPONSE_PROMPT.format(context=context, question=query)
+        
+        messages = self._construct_messages(
+            "You are a helpful database assistant.",
+            history, 
+            prompt
+        )
+        
+        answer = self.llm_client.chat(messages)
+        
+        return ChatResponse(answer=answer, query_type="sql",
+                          sql_query=sanitized_sql, sql_results=results[:10])
+    
+    def _handle_hybrid(self, query: str, schema_context: str, history: List[Dict]) -> ChatResponse:
+        """Handle hybrid RAG + SQL query."""
+        # Get RAG context
+        rag_context = self.rag_engine.get_context(query, top_k=3)
+        
+        # Try SQL as well
+        sql_context = ""
+        sql_query = None
+        try:
+            sql, _ = self.sql_generator.generate(query, schema_context, history)
+            is_valid, _, sanitized_sql = self.sql_validator.validate(sql)
+            if is_valid:
+                results = self.db.execute_query(sanitized_sql)
+                sql_context = f"\nSQL RESULTS:\n{self._format_results(results)}"
+                sql_query = sanitized_sql
+        except Exception as e:
+            logger.debug(f"SQL part of hybrid failed: {e}")
+        
+        context = f"SEMANTIC SEARCH RESULTS:\n{rag_context}{sql_context}"
+        prompt = self.RESPONSE_PROMPT.format(context=context, question=query)
+        
+        messages = self._construct_messages(
+            "You are a helpful database assistant.",
+            history, 
+            prompt
+        )
+        
+        answer = self.llm_client.chat(messages)
+        
+        return ChatResponse(answer=answer, query_type="hybrid", sql_query=sql_query)
+    
+    def _construct_messages(self, system_instruction: str, history: List[Dict], user_content: str) -> List[Dict]:
+        """Construct message list, merging system messages from history."""
+        # Check if first history item is a system message (from memory)
+        additional_context = ""
+        filtered_history = []
+        
+        for msg in history:
+            if msg.get("role") == "system":
+                additional_context += f"\n\n{msg.get('content')}"
+            else:
+                filtered_history.append(msg)
+                
+        full_system_prompt = f"{system_instruction}{additional_context}"
+        
+        messages = [{"role": "system", "content": full_system_prompt}]
+        messages.extend(filtered_history)
+        messages.append({"role": "user", "content": user_content})
+        
+        return messages
+
+    def _handle_general(self, query: str, history: List[Dict]) -> ChatResponse:
+        """Handle conversation."""
+        # Use a strict prompt for general conversation as well to prevent hallucinations
+        strict_system_prompt = (
+            "You are a helpful database assistant.\n"
+            "INSTRUCTIONS:\n"
+            "- Answer ONLY based on the conversation history and any context provided within it.\n"
+            "- Do NOT use outside knowledge, general assumptions, or hallucinate facts.\n"
+            "- If the answer is not in the history or context, state that you don't have that information.\n"
+            "- Be concise."
+        )
+        
+        messages = self._construct_messages(
+            strict_system_prompt,
+            history, 
+            query
+        )
+        answer = self.llm_client.chat(messages)
+        return ChatResponse(answer=answer, query_type="general")
+    
+    def _format_results(self, results: List[Dict], max_rows: int = 10) -> str:
+        """Format SQL results for display."""
+        if not results:
+            return "No results found."
+        
+        rows = results[:max_rows]
+        lines = []
+        
+        # Header
+        headers = list(rows[0].keys())
+        lines.append(" | ".join(headers))
+        lines.append("-" * len(lines[0]))
+        
+        # Rows
+        for row in rows:
+            values = [str(v)[:50] for v in row.values()]
+            lines.append(" | ".join(values))
+        
+        if len(results) > max_rows:
+            lines.append(f"... and {len(results) - max_rows} more rows")
+        
+        return "\n".join(lines)
+    
+    def get_schema_summary(self) -> str:
+        """Get a summary of the database schema."""
+        if not self._schema_initialized:
+            return "Schema not loaded."
+        return get_schema().to_context_string()
+
+
+def create_chatbot(llm_client: Optional[LLMClient] = None) -> DatabaseChatbot:
+    return DatabaseChatbot(llm_client)

config.py

@@ -0,0 +1,280 @@
+"""
+Configuration module for the Schema-Agnostic Database Chatbot.
+
+This module handles all configuration including:
+- Database connection settings (MySQL, PostgreSQL, SQLite)
+- LLM provider settings (Groq / OpenAI / Local LLaMA)
+- Embedding model configuration
+- Security settings
+"""
+
+import os
+from pathlib import Path
+from dataclasses import dataclass, field
+from typing import Optional, List
+from enum import Enum
+
+# Load .env file BEFORE any os.getenv calls
+from dotenv import load_dotenv
+env_path = Path(__file__).parent / ".env"
+load_dotenv(env_path)
+
+
+class DatabaseType(Enum):
+    """Supported database types."""
+    MYSQL = "mysql"
+    POSTGRESQL = "postgresql"
+    SQLITE = "sqlite"
+
+
+class LLMProvider(Enum):
+    """Supported LLM providers."""
+    GROQ = "groq"  # FREE!
+    OPENAI = "openai"
+    LOCAL_LLAMA = "local_llama"
+
+
+class EmbeddingProvider(Enum):
+    """Supported embedding providers."""
+    OPENAI = "openai"
+    SENTENCE_TRANSFORMERS = "sentence_transformers"
+
+
+@dataclass
+class DatabaseConfig:
+    """
+    Database configuration supporting MySQL, PostgreSQL, and SQLite.
+    
+    All sensitive values are loaded from environment variables.
+    """
+    # Database type (mysql, postgresql, sqlite)
+    db_type: DatabaseType = field(
+        default_factory=lambda: DatabaseType(os.getenv("DB_TYPE", "mysql").lower())
+    )
+    
+    # Common connection settings (for MySQL/PostgreSQL)
+    host: str = field(default_factory=lambda: os.getenv("DB_HOST", os.getenv("MYSQL_HOST", "")))
+    port: int = field(default_factory=lambda: int(os.getenv("DB_PORT", os.getenv("MYSQL_PORT", "3306"))))
+    database: str = field(default_factory=lambda: os.getenv("DB_DATABASE", os.getenv("MYSQL_DATABASE", "")))
+    username: str = field(default_factory=lambda: os.getenv("DB_USERNAME", os.getenv("MYSQL_USERNAME", "")))
+    password: str = field(default_factory=lambda: os.getenv("DB_PASSWORD", os.getenv("MYSQL_PASSWORD", "")))
+    
+    # SSL configuration
+    ssl_ca: Optional[str] = field(default_factory=lambda: os.getenv("DB_SSL_CA", os.getenv("MYSQL_SSL_CA", None)))
+    
+    # SQLite-specific: path to database file
+    sqlite_path: str = field(default_factory=lambda: os.getenv("SQLITE_PATH", "./chatbot.db"))
+    
+    @property
+    def connection_string(self) -> str:
+        """Generate SQLAlchemy connection string based on database type."""
+        if self.db_type == DatabaseType.SQLITE:
+            # SQLite uses file path
+            return f"sqlite:///{self.sqlite_path}"
+        
+        elif self.db_type == DatabaseType.POSTGRESQL:
+            # PostgreSQL connection string
+            base_url = f"postgresql+psycopg2://{self.username}:{self.password}@{self.host}:{self.port}/{self.database}"
+            if self.ssl_ca:
+                return f"{base_url}?sslmode=verify-full&sslrootcert={self.ssl_ca}"
+            return base_url
+        
+        else:  # MySQL (default)
+            # MySQL connection string
+            base_url = f"mysql+pymysql://{self.username}:{self.password}@{self.host}:{self.port}/{self.database}"
+            if self.ssl_ca:
+                return f"{base_url}?ssl_ca={self.ssl_ca}"
+            return base_url
+    
+    def is_configured(self) -> bool:
+        """Check if all required database settings are configured."""
+        if self.db_type == DatabaseType.SQLITE:
+            # SQLite only needs a valid path
+            return bool(self.sqlite_path)
+        else:
+            # MySQL/PostgreSQL need host, database, username, password
+            return all([self.host, self.database, self.username, self.password])
+    
+    @property
+    def is_mysql(self) -> bool:
+        """Check if using MySQL."""
+        return self.db_type == DatabaseType.MYSQL
+    
+    @property
+    def is_postgresql(self) -> bool:
+        """Check if using PostgreSQL."""
+        return self.db_type == DatabaseType.POSTGRESQL
+    
+    @property
+    def is_sqlite(self) -> bool:
+        """Check if using SQLite."""
+        return self.db_type == DatabaseType.SQLITE
+
+
+@dataclass
+class LLMConfig:
+    """LLM configuration for query routing and response generation."""
+    provider: LLMProvider = field(
+        default_factory=lambda: LLMProvider(os.getenv("LLM_PROVIDER", "openai"))
+    )
+    openai_api_key: str = field(default_factory=lambda: os.getenv("OPENAI_API_KEY", ""))
+    openai_model: str = field(default_factory=lambda: os.getenv("OPENAI_MODEL", "gpt-4o-mini"))
+    
+    # Local LLaMA settings
+    local_model_path: str = field(
+        default_factory=lambda: os.getenv("LOCAL_MODEL_PATH", "")
+    )
+    local_model_name: str = field(
+        default_factory=lambda: os.getenv("LOCAL_MODEL_NAME", "llama-2-7b-chat")
+    )
+    
+    # Generation parameters
+    temperature: float = 0.1  # Low temperature for more deterministic outputs
+    max_tokens: int = 1024
+    
+    def is_configured(self) -> bool:
+        """Check if LLM is properly configured."""
+        if self.provider == LLMProvider.OPENAI:
+            return bool(self.openai_api_key)
+        return bool(self.local_model_path)
+
+
+@dataclass
+class EmbeddingConfig:
+    """Embedding model configuration for RAG."""
+    provider: EmbeddingProvider = field(
+        default_factory=lambda: EmbeddingProvider(
+            os.getenv("EMBEDDING_PROVIDER", "sentence_transformers")
+        )
+    )
+    
+    # OpenAI embedding settings
+    openai_embedding_model: str = "text-embedding-3-small"
+    
+    # Sentence Transformers settings
+    st_model_name: str = field(
+        default_factory=lambda: os.getenv(
+            "EMBEDDING_MODEL", 
+            "sentence-transformers/all-MiniLM-L6-v2"
+        )
+    )
+    
+    # Embedding dimensions (varies by model)
+    embedding_dim: int = 384  # Default for all-MiniLM-L6-v2
+
+
+@dataclass
+class SecurityConfig:
+    """Security settings for SQL validation and execution."""
+    
+    # SQL operations whitelist - ONLY SELECT allowed
+    allowed_operations: List[str] = field(default_factory=lambda: ["SELECT"])
+    
+    # Dangerous keywords that should never appear in queries
+    forbidden_keywords: List[str] = field(default_factory=lambda: [
+        "INSERT", "UPDATE", "DELETE", "DROP", "CREATE", "ALTER",
+        "TRUNCATE", "GRANT", "REVOKE", "EXECUTE", "EXEC",
+        "INTO OUTFILE", "INTO DUMPFILE", "LOAD_FILE",
+        "INFORMATION_SCHEMA.USER_PRIVILEGES"
+    ])
+    
+    # Maximum number of rows to return
+    max_result_rows: int = 100
+    
+    # Default LIMIT clause if not specified
+    default_limit: int = 50
+
+
+@dataclass
+class RAGConfig:
+    """RAG (Retrieval-Augmented Generation) configuration."""
+    
+    # FAISS index settings
+    faiss_index_path: str = "./faiss_index"
+    
+    # Number of top results to retrieve
+    top_k: int = 5
+    
+    # Minimum similarity score for relevance
+    similarity_threshold: float = 0.3
+    
+    # Text columns to consider for RAG (common across database types)
+    text_column_types: List[str] = field(default_factory=lambda: [
+        # MySQL types
+        "TEXT", "MEDIUMTEXT", "LONGTEXT", "TINYTEXT", "VARCHAR", "CHAR",
+        # PostgreSQL types
+        "CHARACTER VARYING", "CHARACTER", 
+        # SQLite types (SQLite is flexible but these are common)
+        "CLOB", "NVARCHAR", "NCHAR"
+    ])
+    
+    # Minimum character length to consider a column for RAG
+    min_text_length: int = 50
+    
+    # Chunk size for long text documents
+    chunk_size: int = 500
+    chunk_overlap: int = 50
+
+
+@dataclass
+class ChatConfig:
+    """Chat and memory configuration."""
+    
+    # Short-term memory (in session)
+    max_session_messages: int = 20
+    
+    # Long-term memory table name (will be created if not exists)
+    memory_table_name: str = "_chatbot_memory"
+    
+    # Number of recent messages to include in context
+    context_messages: int = 5
+
+
+class AppConfig:
+    """
+    Main application configuration aggregator.
+    
+    Combines all configuration sections and provides
+    validation methods.
+    """
+    
+    def __init__(self):
+        self.database = DatabaseConfig()
+        self.llm = LLMConfig()
+        self.embedding = EmbeddingConfig()
+        self.security = SecurityConfig()
+        self.rag = RAGConfig()
+        self.chat = ChatConfig()
+    
+    def validate(self) -> tuple[bool, List[str]]:
+        """
+        Validate all configuration settings.
+        
+        Returns:
+            tuple: (is_valid, list of error messages)
+        """
+        errors = []
+        
+        if not self.database.is_configured():
+            db_type = self.database.db_type.value.upper()
+            if self.database.is_sqlite:
+                errors.append("SQLite configuration incomplete. Check SQLITE_PATH environment variable.")
+            else:
+                errors.append(f"{db_type} configuration incomplete. Check DB_* environment variables.")
+        
+        if not self.llm.is_configured():
+            errors.append(
+                f"LLM configuration incomplete for provider: {self.llm.provider.value}. "
+                "Check API keys or model paths."
+            )
+        
+        return len(errors) == 0, errors
+    
+    @classmethod
+    def from_env(cls) -> "AppConfig":
+        """Create configuration from environment variables."""
+        return cls()
+
+
+# Global configuration instance
+config = AppConfig.from_env()

database/__init__.py

@@ -0,0 +1,30 @@
+"""
+Database module for the Schema-Agnostic Chatbot.
+
+Provides:
+- Database connection management
+- Dynamic schema introspection
+- Safe query execution
+"""
+
+from .connection import DatabaseConnection, get_db, db_connection
+from .schema_introspector import (
+    SchemaIntrospector,
+    SchemaInfo,
+    TableInfo,
+    ColumnInfo,
+    get_introspector,
+    get_schema
+)
+
+__all__ = [
+    "DatabaseConnection",
+    "get_db",
+    "db_connection",
+    "SchemaIntrospector",
+    "SchemaInfo",
+    "TableInfo",
+    "ColumnInfo",
+    "get_introspector",
+    "get_schema"
+]

database/connection.py

@@ -0,0 +1,231 @@
+"""
+Database Connection Module - Multi-Database Support.
+
+This module provides:
+- SQLAlchemy engine and session management for MySQL, PostgreSQL, and SQLite
+- Connection pooling (for MySQL/PostgreSQL)
+- SSL/TLS support
+- Connection health checking
+"""
+
+import logging
+from contextlib import contextmanager
+from typing import Optional, Generator
+from sqlalchemy import create_engine, text, event
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import sessionmaker, Session
+from sqlalchemy.pool import QueuePool, StaticPool
+from sqlalchemy.exc import OperationalError, SQLAlchemyError
+
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from config import DatabaseConfig, DatabaseType, config
+
+logger = logging.getLogger(__name__)
+
+
+class DatabaseConnection:
+    """
+    Manages database connections with connection pooling.
+    
+    Supports MySQL, PostgreSQL, and SQLite.
+    """
+    
+    def __init__(self, db_config: Optional[DatabaseConfig] = None):
+        """
+        Initialize database connection manager.
+        
+        Args:
+            db_config: Database configuration. Uses global config if not provided.
+        """
+        self.config = db_config or config.database
+        self._engine: Optional[Engine] = None
+        self._session_factory: Optional[sessionmaker] = None
+    
+    def _create_engine(self) -> Engine:
+        """
+        Create SQLAlchemy engine with appropriate settings for each database type.
+        
+        Returns:
+            Configured SQLAlchemy Engine instance
+        """
+        connect_args = {}
+        
+        if self.config.db_type == DatabaseType.SQLITE:
+            # SQLite-specific settings
+            # Use StaticPool for SQLite to handle multi-threading
+            connect_args["check_same_thread"] = False
+            
+            engine = create_engine(
+                self.config.connection_string,
+                poolclass=StaticPool,  # SQLite works best with StaticPool
+                connect_args=connect_args,
+                echo=False
+            )
+            
+            # Enable foreign keys for SQLite
+            @event.listens_for(engine, "connect")
+            def set_sqlite_pragma(dbapi_connection, connection_record):
+                cursor = dbapi_connection.cursor()
+                cursor.execute("PRAGMA foreign_keys=ON")
+                cursor.close()
+            
+        elif self.config.db_type == DatabaseType.POSTGRESQL:
+            # PostgreSQL-specific settings
+            if self.config.ssl_ca:
+                connect_args["sslmode"] = "verify-full"
+                connect_args["sslrootcert"] = self.config.ssl_ca
+            
+            engine = create_engine(
+                self.config.connection_string,
+                poolclass=QueuePool,
+                pool_size=5,
+                max_overflow=10,
+                pool_timeout=30,
+                pool_recycle=1800,
+                pool_pre_ping=True,
+                connect_args=connect_args,
+                echo=False
+            )
+            
+        else:  # MySQL (default)
+            # MySQL-specific settings (SSL for Aiven)
+            if self.config.ssl_ca:
+                connect_args["ssl"] = {
+                    "ca": self.config.ssl_ca,
+                    "check_hostname": True,
+                    "verify_mode": True
+                }
+            
+            engine = create_engine(
+                self.config.connection_string,
+                poolclass=QueuePool,
+                pool_size=5,
+                max_overflow=10,
+                pool_timeout=30,
+                pool_recycle=1800,
+                pool_pre_ping=True,
+                connect_args=connect_args,
+                echo=False
+            )
+        
+        return engine
+    
+    @property
+    def engine(self) -> Engine:
+        """Get or create the SQLAlchemy engine."""
+        if self._engine is None:
+            self._engine = self._create_engine()
+        return self._engine
+    
+    @property
+    def session_factory(self) -> sessionmaker:
+        """Get or create the session factory."""
+        if self._session_factory is None:
+            self._session_factory = sessionmaker(
+                bind=self.engine,
+                autocommit=False,
+                autoflush=False
+            )
+        return self._session_factory
+    
+    @property
+    def db_type(self) -> DatabaseType:
+        """Get the current database type."""
+        return self.config.db_type
+    
+    @contextmanager
+    def get_session(self) -> Generator[Session, None, None]:
+        """
+        Context manager for database sessions.
+        
+        Yields:
+            SQLAlchemy Session instance
+            
+        Example:
+            with db.get_session() as session:
+                result = session.execute(text("SELECT * FROM users"))
+        """
+        session = self.session_factory()
+        try:
+            yield session
+            session.commit()
+        except SQLAlchemyError as e:
+            session.rollback()
+            logger.error(f"Database session error: {e}")
+            raise
+        finally:
+            session.close()
+    
+    def execute_query(self, query: str, params: Optional[dict] = None) -> list:
+        """
+        Execute a read-only SQL query and return results.
+        
+        Args:
+            query: SQL query string (must be SELECT)
+            params: Optional query parameters for parameterized queries
+            
+        Returns:
+            List of result rows as dictionaries
+        """
+        with self.get_session() as session:
+            result = session.execute(text(query), params or {})
+            # Convert rows to dictionaries for easier handling
+            columns = result.keys()
+            return [dict(zip(columns, row)) for row in result.fetchall()]
+
+    def execute_write(self, query: str, params: Optional[dict] = None) -> bool:
+        """
+        Execute a write operation (INSERT, UPDATE, DELETE, CREATE).
+        
+        Args:
+            query: SQL query string
+            params: Optional query parameters
+            
+        Returns:
+            bool: True if successful
+        """
+        with self.get_session() as session:
+            session.execute(text(query), params or {})
+            session.commit()
+            return True
+    
+    def test_connection(self) -> tuple[bool, str]:
+        """
+        Test database connectivity.
+        
+        Returns:
+            tuple: (success: bool, message: str)
+        """
+        try:
+            with self.get_session() as session:
+                result = session.execute(text("SELECT 1 as health_check"))
+                row = result.fetchone()
+                if row and row[0] == 1:
+                    db_type = self.config.db_type.value.upper()
+                    return True, f"{db_type} connection successful"
+                return False, "Unexpected result from health check query"
+        except OperationalError as e:
+            logger.error(f"Database connection failed: {e}")
+            return False, f"Connection failed: {str(e)}"
+        except Exception as e:
+            logger.error(f"Unexpected error during connection test: {e}")
+            return False, f"Unexpected error: {str(e)}"
+    
+    def close(self):
+        """Close all connections and dispose of the engine."""
+        if self._engine:
+            self._engine.dispose()
+            self._engine = None
+            self._session_factory = None
+            logger.info("Database connections closed")
+
+
+# Create a global database connection instance
+db_connection = DatabaseConnection()
+
+
+def get_db() -> DatabaseConnection:
+    """Get the global database connection instance."""
+    return db_connection

database/schema_introspector.py

@@ -0,0 +1,648 @@
+"""
+Dynamic Schema Introspection Module - Multi-Database Support.
+
+This module is the CORE of the schema-agnostic design.
+It dynamically discovers:
+- All tables in the database
+- All columns with their data types
+- Primary keys and foreign keys
+- Text-like columns for RAG indexing
+- Relationships between tables
+
+Supports MySQL, PostgreSQL, and SQLite.
+NEVER hardcodes any table or column names.
+"""
+
+import logging
+from dataclasses import dataclass, field
+from typing import List, Dict, Optional, Any
+from sqlalchemy import text, inspect
+from sqlalchemy.engine import Engine
+
+from .connection import get_db
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ColumnInfo:
+    """Information about a single database column."""
+    name: str
+    data_type: str
+    is_nullable: bool
+    is_primary_key: bool
+    max_length: Optional[int] = None
+    default_value: Optional[str] = None
+    comment: Optional[str] = None
+    
+    @property
+    def is_text_type(self) -> bool:
+        """Check if this column contains text data suitable for RAG."""
+        text_types = [
+            # MySQL
+            'text', 'mediumtext', 'longtext', 'tinytext', 'varchar', 'char', 'json',
+            # PostgreSQL
+            'character varying', 'character', 'text', 'json', 'jsonb',
+            # SQLite (column affinity - TEXT)
+            'clob', 'nvarchar', 'nchar', 'ntext'
+        ]
+        data_type_lower = self.data_type.lower().split('(')[0].strip()
+        return data_type_lower in text_types
+    
+    @property
+    def is_numeric(self) -> bool:
+        """Check if this column contains numeric data."""
+        numeric_types = [
+            # Common across databases
+            'int', 'integer', 'bigint', 'smallint', 'tinyint',
+            'decimal', 'numeric', 'float', 'double', 'real',
+            # PostgreSQL specific
+            'double precision', 'serial', 'bigserial', 'smallserial',
+            # SQLite (NUMERIC affinity)
+            'bool', 'boolean'
+        ]
+        data_type_lower = self.data_type.lower().split('(')[0].strip()
+        return data_type_lower in numeric_types
+
+
+@dataclass
+class TableInfo:
+    """Complete information about a database table."""
+    name: str
+    columns: List[ColumnInfo] = field(default_factory=list)
+    primary_keys: List[str] = field(default_factory=list)
+    foreign_keys: Dict[str, str] = field(default_factory=dict)  # column -> referenced_table.column
+    row_count: Optional[int] = None
+    comment: Optional[str] = None
+    
+    @property
+    def text_columns(self) -> List[ColumnInfo]:
+        """Get columns suitable for text/RAG indexing."""
+        return [col for col in self.columns if col.is_text_type]
+    
+    @property
+    def column_names(self) -> List[str]:
+        """Get list of all column names."""
+        return [col.name for col in self.columns]
+    
+    def get_column(self, name: str) -> Optional[ColumnInfo]:
+        """Get column info by name."""
+        for col in self.columns:
+            if col.name.lower() == name.lower():
+                return col
+        return None
+
+
+@dataclass
+class SchemaInfo:
+    """Complete database schema information."""
+    database_name: str
+    tables: Dict[str, TableInfo] = field(default_factory=dict)
+    
+    @property
+    def table_names(self) -> List[str]:
+        """Get list of all table names."""
+        return list(self.tables.keys())
+    
+    @property
+    def all_text_columns(self) -> List[tuple]:
+        """Get all text columns across all tables as (table, column) tuples."""
+        result = []
+        for table_name, table_info in self.tables.items():
+            for col in table_info.text_columns:
+                result.append((table_name, col.name))
+        return result
+    
+    def to_context_string(self) -> str:
+        """
+        Generate a natural language description of the schema.
+        This is used as context for the LLM.
+        """
+        lines = [f"Database: {self.database_name}", ""]
+        lines.append("Available Tables:")
+        lines.append("-" * 40)
+        
+        for table_name, table_info in self.tables.items():
+            lines.append(f"\nTable: {table_name}")
+            if table_info.comment:
+                lines.append(f"  Description: {table_info.comment}")
+            if table_info.row_count is not None:
+                lines.append(f"  Approximate rows: {table_info.row_count}")
+            
+            lines.append("  Columns:")
+            for col in table_info.columns:
+                pk_marker = " [PRIMARY KEY]" if col.is_primary_key else ""
+                nullable = " (nullable)" if col.is_nullable else " (required)"
+                lines.append(f"    - {col.name}: {col.data_type}{pk_marker}{nullable}")
+                if col.comment:
+                    lines.append(f"      Comment: {col.comment}")
+            
+            if table_info.foreign_keys:
+                lines.append("  Foreign Keys:")
+                for col, ref in table_info.foreign_keys.items():
+                    lines.append(f"    - {col} -> {ref}")
+        
+        return "\n".join(lines)
+    
+    def to_sql_ddl(self) -> str:
+        """
+        Generate SQL-like DDL representation of the schema.
+        Useful for SQL generation context.
+        """
+        ddl_lines = []
+        
+        for table_name, table_info in self.tables.items():
+            ddl_lines.append(f"CREATE TABLE {table_name} (")
+            
+            col_defs = []
+            for col in table_info.columns:
+                col_def = f"  {col.name} {col.data_type}"
+                if col.is_primary_key:
+                    col_def += " PRIMARY KEY"
+                if not col.is_nullable:
+                    col_def += " NOT NULL"
+                col_defs.append(col_def)
+            
+            ddl_lines.append(",\n".join(col_defs))
+            ddl_lines.append(");\n")
+        
+        return "\n".join(ddl_lines)
+
+
+class SchemaIntrospector:
+    """
+    Dynamically introspects database schema.
+    
+    This is the key component that enables schema-agnostic operation.
+    It queries database system catalogs to discover the complete schema.
+    Supports MySQL, PostgreSQL, and SQLite.
+    """
+    
+    # System tables to exclude from introspection
+    SYSTEM_TABLES = {
+        '_chatbot_memory',  # Our own chat history table
+        '_chatbot_permanent_memory_v2',
+        '_chatbot_user_summaries',
+        'schema_migrations',
+        'flyway_schema_history',
+        # SQLite internal tables
+        'sqlite_sequence',
+        'sqlite_stat1',
+        'sqlite_stat4'
+    }
+    
+    def __init__(self, engine: Optional[Engine] = None):
+        """
+        Initialize the introspector.
+        
+        Args:
+            engine: SQLAlchemy engine. Uses global connection if not provided.
+        """
+        self.db = get_db()
+        self._cached_schema: Optional[SchemaInfo] = None
+    
+    def introspect(self, force_refresh: bool = False) -> SchemaInfo:
+        """
+        Perform complete schema introspection.
+        
+        Args:
+            force_refresh: If True, bypass cache and re-introspect
+            
+        Returns:
+            SchemaInfo object with complete schema details
+        """
+        if self._cached_schema is not None and not force_refresh:
+            return self._cached_schema
+        
+        logger.info("Starting schema introspection...")
+        
+        # Get database name
+        db_name = self._get_database_name()
+        
+        # Get all user tables
+        tables = self._get_tables()
+        
+        schema = SchemaInfo(database_name=db_name)
+        
+        for table_name in tables:
+            if table_name in self.SYSTEM_TABLES:
+                continue
+            # Also skip tables that start with underscore (internal tables)
+            if table_name.startswith('_chatbot'):
+                continue
+                
+            table_info = self._introspect_table(table_name)
+            if table_info:
+                schema.tables[table_name] = table_info
+        
+        self._cached_schema = schema
+        logger.info(f"Schema introspection complete. Found {len(schema.tables)} tables.")
+        
+        return schema
+    
+    def _get_database_name(self) -> str:
+        """Get the current database name."""
+        db_type = self.db.db_type
+        
+        try:
+            if db_type.value == "sqlite":
+                # For SQLite, return the database file name
+                return self.db.config.sqlite_path.split('/')[-1]
+            elif db_type.value == "postgresql":
+                result = self.db.execute_query("SELECT current_database() as db_name")
+                return result[0]['db_name'] if result else "unknown"
+            else:  # MySQL
+                result = self.db.execute_query("SELECT DATABASE() as db_name")
+                return result[0]['db_name'] if result else "unknown"
+        except Exception as e:
+            logger.error(f"Error getting database name: {e}")
+            return "unknown"
+    
+    def _get_tables(self) -> List[str]:
+        """
+        Get all user tables from the database.
+        Uses database-specific queries for comprehensive discovery.
+        """
+        db_type = self.db.db_type
+        
+        try:
+            if db_type.value == "sqlite":
+                query = """
+                    SELECT name as table_name
+                    FROM sqlite_master 
+                    WHERE type='table' 
+                    AND name NOT LIKE 'sqlite_%'
+                    ORDER BY name
+                """
+                result = self.db.execute_query(query)
+                return [row['table_name'] for row in result]
+                
+            elif db_type.value == "postgresql":
+                query = """
+                    SELECT table_name 
+                    FROM information_schema.tables 
+                    WHERE table_schema = 'public'
+                    AND table_type = 'BASE TABLE'
+                    ORDER BY table_name
+                """
+                result = self.db.execute_query(query)
+                return [row['table_name'] for row in result]
+                
+            else:  # MySQL
+                query = """
+                    SELECT TABLE_NAME 
+                    FROM INFORMATION_SCHEMA.TABLES 
+                    WHERE TABLE_SCHEMA = DATABASE()
+                    AND TABLE_TYPE = 'BASE TABLE'
+                    ORDER BY TABLE_NAME
+                """
+                result = self.db.execute_query(query)
+                return [row['TABLE_NAME'] for row in result]
+                
+        except Exception as e:
+            logger.error(f"Error getting tables: {e}")
+            return []
+    
+    def _introspect_table(self, table_name: str) -> Optional[TableInfo]:
+        """
+        Get complete information about a specific table.
+        
+        Args:
+            table_name: Name of the table to introspect
+            
+        Returns:
+            TableInfo object or None if table doesn't exist
+        """
+        try:
+            # Get column information
+            columns = self._get_columns(table_name)
+            
+            # Get primary keys
+            primary_keys = self._get_primary_keys(table_name)
+            
+            # Get foreign keys
+            foreign_keys = self._get_foreign_keys(table_name)
+            
+            # Get approximate row count (fast estimation)
+            row_count = self._get_row_count(table_name)
+            
+            # Get table comment (not available in SQLite)
+            comment = self._get_table_comment(table_name)
+            
+            # Mark primary key columns
+            for col in columns:
+                col.is_primary_key = col.name in primary_keys
+            
+            return TableInfo(
+                name=table_name,
+                columns=columns,
+                primary_keys=primary_keys,
+                foreign_keys=foreign_keys,
+                row_count=row_count,
+                comment=comment
+            )
+            
+        except Exception as e:
+            logger.error(f"Error introspecting table {table_name}: {e}")
+            return None
+    
+    def _get_columns(self, table_name: str) -> List[ColumnInfo]:
+        """Get all columns for a table."""
+        db_type = self.db.db_type
+        
+        try:
+            if db_type.value == "sqlite":
+                query = f"PRAGMA table_info('{table_name}')"
+                result = self.db.execute_query(query)
+                
+                columns = []
+                for row in result:
+                    columns.append(ColumnInfo(
+                        name=row['name'],
+                        data_type=row['type'] or 'TEXT',  # SQLite columns can have no type
+                        is_nullable=row['notnull'] == 0,
+                        is_primary_key=row['pk'] == 1,
+                        max_length=None,
+                        default_value=row['dflt_value'],
+                        comment=None  # SQLite doesn't support column comments
+                    ))
+                return columns
+                
+            elif db_type.value == "postgresql":
+                query = """
+                    SELECT 
+                        column_name,
+                        data_type,
+                        is_nullable,
+                        column_default,
+                        character_maximum_length,
+                        col_description(
+                            (SELECT oid FROM pg_class WHERE relname = :table_name),
+                            ordinal_position
+                        ) as column_comment
+                    FROM information_schema.columns
+                    WHERE table_schema = 'public'
+                    AND table_name = :table_name
+                    ORDER BY ordinal_position
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                
+                columns = []
+                for row in result:
+                    columns.append(ColumnInfo(
+                        name=row['column_name'],
+                        data_type=row['data_type'],
+                        is_nullable=row['is_nullable'] == 'YES',
+                        is_primary_key=False,  # Will be set later
+                        max_length=row['character_maximum_length'],
+                        default_value=row['column_default'],
+                        comment=row.get('column_comment')
+                    ))
+                return columns
+                
+            else:  # MySQL
+                query = """
+                    SELECT 
+                        COLUMN_NAME,
+                        COLUMN_TYPE,
+                        IS_NULLABLE,
+                        COLUMN_DEFAULT,
+                        CHARACTER_MAXIMUM_LENGTH,
+                        COLUMN_COMMENT
+                    FROM INFORMATION_SCHEMA.COLUMNS
+                    WHERE TABLE_SCHEMA = DATABASE()
+                    AND TABLE_NAME = :table_name
+                    ORDER BY ORDINAL_POSITION
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                
+                columns = []
+                for row in result:
+                    columns.append(ColumnInfo(
+                        name=row['COLUMN_NAME'],
+                        data_type=row['COLUMN_TYPE'],
+                        is_nullable=row['IS_NULLABLE'] == 'YES',
+                        is_primary_key=False,  # Will be set later
+                        max_length=row['CHARACTER_MAXIMUM_LENGTH'],
+                        default_value=row['COLUMN_DEFAULT'],
+                        comment=row['COLUMN_COMMENT'] if row['COLUMN_COMMENT'] else None
+                    ))
+                return columns
+                
+        except Exception as e:
+            logger.error(f"Error getting columns for {table_name}: {e}")
+            return []
+    
+    def _get_primary_keys(self, table_name: str) -> List[str]:
+        """Get primary key columns for a table."""
+        db_type = self.db.db_type
+        
+        try:
+            if db_type.value == "sqlite":
+                query = f"PRAGMA table_info('{table_name}')"
+                result = self.db.execute_query(query)
+                return [row['name'] for row in result if row['pk'] > 0]
+                
+            elif db_type.value == "postgresql":
+                query = """
+                    SELECT a.attname as column_name
+                    FROM pg_index i
+                    JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
+                    WHERE i.indrelid = :table_name::regclass
+                    AND i.indisprimary
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                return [row['column_name'] for row in result]
+                
+            else:  # MySQL
+                query = """
+                    SELECT COLUMN_NAME
+                    FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
+                    WHERE TABLE_SCHEMA = DATABASE()
+                    AND TABLE_NAME = :table_name
+                    AND CONSTRAINT_NAME = 'PRIMARY'
+                    ORDER BY ORDINAL_POSITION
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                return [row['COLUMN_NAME'] for row in result]
+                
+        except Exception as e:
+            logger.error(f"Error getting primary keys for {table_name}: {e}")
+            return []
+    
+    def _get_foreign_keys(self, table_name: str) -> Dict[str, str]:
+        """Get foreign key relationships for a table."""
+        db_type = self.db.db_type
+        
+        try:
+            if db_type.value == "sqlite":
+                query = f"PRAGMA foreign_key_list('{table_name}')"
+                result = self.db.execute_query(query)
+                return {
+                    row['from']: f"{row['table']}.{row['to']}"
+                    for row in result
+                }
+                
+            elif db_type.value == "postgresql":
+                query = """
+                    SELECT
+                        kcu.column_name,
+                        ccu.table_name AS foreign_table_name,
+                        ccu.column_name AS foreign_column_name
+                    FROM information_schema.table_constraints AS tc
+                    JOIN information_schema.key_column_usage AS kcu
+                        ON tc.constraint_name = kcu.constraint_name
+                        AND tc.table_schema = kcu.table_schema
+                    JOIN information_schema.constraint_column_usage AS ccu
+                        ON ccu.constraint_name = tc.constraint_name
+                        AND ccu.table_schema = tc.table_schema
+                    WHERE tc.constraint_type = 'FOREIGN KEY'
+                    AND tc.table_name = :table_name
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                return {
+                    row['column_name']: f"{row['foreign_table_name']}.{row['foreign_column_name']}"
+                    for row in result
+                }
+                
+            else:  # MySQL
+                query = """
+                    SELECT 
+                        COLUMN_NAME,
+                        REFERENCED_TABLE_NAME,
+                        REFERENCED_COLUMN_NAME
+                    FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
+                    WHERE TABLE_SCHEMA = DATABASE()
+                    AND TABLE_NAME = :table_name
+                    AND REFERENCED_TABLE_NAME IS NOT NULL
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                return {
+                    row['COLUMN_NAME']: f"{row['REFERENCED_TABLE_NAME']}.{row['REFERENCED_COLUMN_NAME']}"
+                    for row in result
+                }
+                
+        except Exception as e:
+            logger.error(f"Error getting foreign keys for {table_name}: {e}")
+            return {}
+    
+    def _get_row_count(self, table_name: str) -> Optional[int]:
+        """
+        Get approximate row count for a table.
+        Uses different strategies per database.
+        """
+        db_type = self.db.db_type
+        
+        try:
+            if db_type.value == "sqlite":
+                # SQLite doesn't have stats table, use max rowid for estimation
+                query = f"SELECT MAX(rowid) as row_count FROM \"{table_name}\""
+                result = self.db.execute_query(query)
+                return result[0]['row_count'] if result and result[0]['row_count'] else 0
+                
+            elif db_type.value == "postgresql":
+                # Use pg_stat_user_tables for fast estimation
+                query = """
+                    SELECT n_live_tup as row_count
+                    FROM pg_stat_user_tables
+                    WHERE relname = :table_name
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                return result[0]['row_count'] if result else None
+                
+            else:  # MySQL
+                query = """
+                    SELECT TABLE_ROWS
+                    FROM INFORMATION_SCHEMA.TABLES
+                    WHERE TABLE_SCHEMA = DATABASE()
+                    AND TABLE_NAME = :table_name
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                return result[0]['TABLE_ROWS'] if result else None
+                
+        except Exception as e:
+            logger.error(f"Error getting row count for {table_name}: {e}")
+            return None
+    
+    def _get_table_comment(self, table_name: str) -> Optional[str]:
+        """Get table comment/description."""
+        db_type = self.db.db_type
+        
+        try:
+            if db_type.value == "sqlite":
+                # SQLite doesn't support table comments
+                return None
+                
+            elif db_type.value == "postgresql":
+                query = """
+                    SELECT obj_description(:table_name::regclass, 'pg_class') as table_comment
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                comment = result[0]['table_comment'] if result else None
+                return comment if comment else None
+                
+            else:  # MySQL
+                query = """
+                    SELECT TABLE_COMMENT
+                    FROM INFORMATION_SCHEMA.TABLES
+                    WHERE TABLE_SCHEMA = DATABASE()
+                    AND TABLE_NAME = :table_name
+                """
+                result = self.db.execute_query(query, {"table_name": table_name})
+                comment = result[0]['TABLE_COMMENT'] if result else None
+                return comment if comment else None
+                
+        except Exception as e:
+            logger.error(f"Error getting table comment for {table_name}: {e}")
+            return None
+    
+    def get_text_columns_for_rag(self, min_length: int = 50) -> List[Dict[str, Any]]:
+        """
+        Get all text columns suitable for RAG indexing.
+        
+        Args:
+            min_length: Minimum max_length for varchar columns to be considered
+            
+        Returns:
+            List of dicts with table name, column name, and metadata
+        """
+        schema = self.introspect()
+        text_columns = []
+        
+        for table_name, table_info in schema.tables.items():
+            for col in table_info.columns:
+                if col.is_text_type:
+                    # Skip very short varchar columns
+                    if col.max_length and col.max_length < min_length:
+                        continue
+                    
+                    text_columns.append({
+                        "table": table_name,
+                        "column": col.name,
+                        "data_type": col.data_type,
+                        "primary_keys": table_info.primary_keys,
+                        "max_length": col.max_length
+                    })
+        
+        return text_columns
+    
+    def refresh_cache(self) -> SchemaInfo:
+        """Force refresh the cached schema."""
+        return self.introspect(force_refresh=True)
+
+
+# Global introspector instance
+_introspector: Optional[SchemaIntrospector] = None
+
+
+def get_introspector() -> SchemaIntrospector:
+    """Get or create the global schema introspector."""
+    global _introspector
+    if _introspector is None:
+        _introspector = SchemaIntrospector()
+    return _introspector
+
+
+def get_schema() -> SchemaInfo:
+    """Convenience function to get the current schema."""
+    return get_introspector().introspect()

llm/__init__.py

@@ -0,0 +1,17 @@
+"""LLM module exports."""
+
+from .client import (
+    LLMClient,
+    GroqClient,
+    OpenAIClient,
+    LocalLLaMAClient,
+    create_llm_client
+)
+
+__all__ = [
+    "LLMClient",
+    "GroqClient",
+    "OpenAIClient",
+    "LocalLLaMAClient",
+    "create_llm_client"
+]

llm/client.py

@@ -0,0 +1,188 @@
+"""
+LLM Client - Unified interface for Groq, OpenAI, and local models.
+
+Groq is the DEFAULT provider (free tier available).
+"""
+
+import logging
+from abc import ABC, abstractmethod
+from typing import List, Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class LLMClient(ABC):
+    """Abstract base class for LLM clients."""
+    
+    @abstractmethod
+    def chat(self, messages: List[Dict[str, str]]) -> str:
+        pass
+    
+    @abstractmethod
+    def is_available(self) -> bool:
+        pass
+
+
+class GroqClient(LLMClient):
+    """
+    Groq API client - FREE and FAST inference.
+    
+    Available models:
+    - llama-3.3-70b-versatile (recommended)
+    - llama-3.1-8b-instant (faster)
+    - mixtral-8x7b-32768
+    - gemma2-9b-it
+    """
+    
+    AVAILABLE_MODELS = [
+        "llama-3.3-70b-versatile",
+        "llama-3.1-70b-versatile", 
+        "llama-3.1-8b-instant",
+        "llama3-70b-8192",
+        "llama3-8b-8192",
+        "mixtral-8x7b-32768",
+        "gemma2-9b-it"
+    ]
+    
+    def __init__(
+        self,
+        api_key: str,
+        model: str = "llama-3.3-70b-versatile",
+        temperature: float = 0.1,
+        max_tokens: int = 1024
+    ):
+        self.api_key = api_key
+        self.model = model
+        self.temperature = temperature
+        self.max_tokens = max_tokens
+        self._client = None
+    
+    @property
+    def client(self):
+        if self._client is None:
+            from groq import Groq
+            self._client = Groq(api_key=self.api_key)
+        return self._client
+    
+    def chat(self, messages: List[Dict[str, str]]) -> str:
+        response = self.client.chat.completions.create(
+            model=self.model,
+            messages=messages,
+            temperature=self.temperature,
+            max_tokens=self.max_tokens
+        )
+        return response.choices[0].message.content
+    
+    def is_available(self) -> bool:
+        try:
+            # Simple test call
+            self.client.models.list()
+            return True
+        except Exception as e:
+            logger.warning(f"Groq availability check failed: {e}")
+            return False
+
+
+class OpenAIClient(LLMClient):
+    """OpenAI API client (paid)."""
+    
+    def __init__(
+        self,
+        api_key: str,
+        model: str = "gpt-4o-mini",
+        temperature: float = 0.1,
+        max_tokens: int = 1024
+    ):
+        self.api_key = api_key
+        self.model = model
+        self.temperature = temperature
+        self.max_tokens = max_tokens
+        self._client = None
+    
+    @property
+    def client(self):
+        if self._client is None:
+            from openai import OpenAI
+            self._client = OpenAI(api_key=self.api_key)
+        return self._client
+    
+    def chat(self, messages: List[Dict[str, str]]) -> str:
+        response = self.client.chat.completions.create(
+            model=self.model,
+            messages=messages,
+            temperature=self.temperature,
+            max_tokens=self.max_tokens
+        )
+        return response.choices[0].message.content
+    
+    def is_available(self) -> bool:
+        try:
+            self.client.models.list()
+            return True
+        except Exception:
+            return False
+
+
+class LocalLLaMAClient(LLMClient):
+    """Local LLaMA/Phi model client via transformers."""
+    
+    def __init__(
+        self,
+        model_name: str = "microsoft/Phi-3-mini-4k-instruct",
+        temperature: float = 0.1,
+        max_tokens: int = 1024
+    ):
+        self.model_name = model_name
+        self.temperature = temperature
+        self.max_tokens = max_tokens
+        self._pipeline = None
+    
+    @property
+    def pipeline(self):
+        if self._pipeline is None:
+            from transformers import pipeline
+            logger.info(f"Loading local model: {self.model_name}")
+            self._pipeline = pipeline(
+                "text-generation",
+                model=self.model_name,
+                torch_dtype="auto",
+                device_map="auto"
+            )
+        return self._pipeline
+    
+    def chat(self, messages: List[Dict[str, str]]) -> str:
+        output = self.pipeline(
+            messages,
+            max_new_tokens=self.max_tokens,
+            temperature=self.temperature,
+            do_sample=True
+        )
+        return output[0]["generated_text"][-1]["content"]
+    
+    def is_available(self) -> bool:
+        try:
+            _ = self.pipeline
+            return True
+        except Exception:
+            return False
+
+
+def create_llm_client(provider: str = "groq", **kwargs) -> LLMClient:
+    """
+    Factory function to create LLM client.
+    
+    Args:
+        provider: "groq" (default, free), "openai", or "local"
+        **kwargs: Provider-specific arguments
+        
+    Returns:
+        Configured LLMClient instance
+    """
+    if provider == "groq":
+        return GroqClient(**kwargs)
+    elif provider == "openai":
+        return OpenAIClient(**kwargs)
+    elif provider == "local":
+        return LocalLLaMAClient(**kwargs)
+    else:
+        raise ValueError(f"Unknown provider: {provider}. Use 'groq', 'openai', or 'local'")

memory.py

@@ -0,0 +1,760 @@
+"""
+Chat Memory - Short-term and long-term memory management.
+
+Supports MySQL, PostgreSQL, and SQLite with dialect-specific DDL.
+"""
+
+import logging
+import json
+from typing import List, Dict, Any, Optional
+from datetime import datetime
+from dataclasses import dataclass
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ChatMessage:
+    role: str  # "user" or "assistant"
+    content: str
+    timestamp: datetime = None
+    metadata: Dict[str, Any] = None
+    
+    def __post_init__(self):
+        if self.timestamp is None:
+            self.timestamp = datetime.now()
+        if self.metadata is None:
+            self.metadata = {}
+    
+    def to_dict(self) -> Dict[str, str]:
+        return {"role": self.role, "content": self.content}
+
+
+def get_memory_table_ddl(db_type: str) -> str:
+    """Get the DDL for chat memory table based on database type."""
+    if db_type == "postgresql":
+        return """
+        CREATE TABLE IF NOT EXISTS _chatbot_memory (
+            id SERIAL PRIMARY KEY,
+            session_id VARCHAR(255) NOT NULL,
+            user_id VARCHAR(255) NOT NULL DEFAULT 'default',
+            role VARCHAR(50) NOT NULL,
+            content TEXT NOT NULL,
+            metadata JSONB,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+        """
+    elif db_type == "sqlite":
+        return """
+        CREATE TABLE IF NOT EXISTS _chatbot_memory (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT NOT NULL,
+            user_id TEXT NOT NULL DEFAULT 'default',
+            role TEXT NOT NULL,
+            content TEXT NOT NULL,
+            metadata TEXT,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+        """
+    else:  # MySQL
+        return """
+        CREATE TABLE IF NOT EXISTS _chatbot_memory (
+            id INT AUTO_INCREMENT PRIMARY KEY,
+            session_id VARCHAR(255) NOT NULL,
+            user_id VARCHAR(255) NOT NULL DEFAULT 'default',
+            role VARCHAR(50) NOT NULL,
+            content TEXT NOT NULL,
+            metadata JSON,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+            INDEX idx_session (session_id),
+            INDEX idx_user (user_id),
+            INDEX idx_created (created_at)
+        )
+        """
+
+
+def get_permanent_memory_ddl(db_type: str) -> str:
+    """Get the DDL for permanent memory table based on database type."""
+    if db_type == "postgresql":
+        return """
+        CREATE TABLE IF NOT EXISTS _chatbot_permanent_memory_v2 (
+            id SERIAL PRIMARY KEY,
+            user_id VARCHAR(255) NOT NULL DEFAULT 'default',
+            content TEXT NOT NULL,
+            tags VARCHAR(255),
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+        """
+    elif db_type == "sqlite":
+        return """
+        CREATE TABLE IF NOT EXISTS _chatbot_permanent_memory_v2 (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id TEXT NOT NULL DEFAULT 'default',
+            content TEXT NOT NULL,
+            tags TEXT,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+        """
+    else:  # MySQL
+        return """
+        CREATE TABLE IF NOT EXISTS _chatbot_permanent_memory_v2 (
+            id INT AUTO_INCREMENT PRIMARY KEY,
+            user_id VARCHAR(255) NOT NULL DEFAULT 'default',
+            content TEXT NOT NULL,
+            tags VARCHAR(255),
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+            INDEX idx_user (user_id)
+        )
+        """
+
+
+def get_summary_table_ddl(db_type: str) -> str:
+    """Get the DDL for summary table based on database type."""
+    if db_type == "postgresql":
+        return """
+        CREATE TABLE IF NOT EXISTS _chatbot_user_summaries (
+            id SERIAL PRIMARY KEY,
+            user_id VARCHAR(255) NOT NULL UNIQUE,
+            summary TEXT NOT NULL,
+            message_count INT DEFAULT 0,
+            last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+        """
+    elif db_type == "sqlite":
+        return """
+        CREATE TABLE IF NOT EXISTS _chatbot_user_summaries (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id TEXT NOT NULL UNIQUE,
+            summary TEXT NOT NULL,
+            message_count INTEGER DEFAULT 0,
+            last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+        """
+    else:  # MySQL
+        return """
+        CREATE TABLE IF NOT EXISTS _chatbot_user_summaries (
+            id INT AUTO_INCREMENT PRIMARY KEY,
+            user_id VARCHAR(255) NOT NULL,
+            summary TEXT NOT NULL,
+            message_count INT DEFAULT 0,
+            last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+            UNIQUE KEY idx_user (user_id)
+        )
+        """
+
+
+def get_upsert_summary_query(db_type: str) -> str:
+    """Get the upsert query for summary based on database type."""
+    if db_type == "postgresql":
+        return """
+            INSERT INTO _chatbot_user_summaries 
+            (user_id, summary, message_count, last_updated)
+            VALUES (:user_id, :summary, :message_count, CURRENT_TIMESTAMP)
+            ON CONFLICT (user_id) 
+            DO UPDATE SET 
+                summary = EXCLUDED.summary, 
+                message_count = EXCLUDED.message_count,
+                last_updated = CURRENT_TIMESTAMP
+        """
+    elif db_type == "sqlite":
+        return """
+            INSERT INTO _chatbot_user_summaries 
+            (user_id, summary, message_count, last_updated)
+            VALUES (:user_id, :summary, :message_count, CURRENT_TIMESTAMP)
+            ON CONFLICT(user_id) 
+            DO UPDATE SET 
+                summary = excluded.summary, 
+                message_count = excluded.message_count,
+                last_updated = CURRENT_TIMESTAMP
+        """
+    else:  # MySQL
+        return """
+            INSERT INTO _chatbot_user_summaries 
+            (user_id, summary, message_count)
+            VALUES (:user_id, :summary, :message_count)
+            ON DUPLICATE KEY UPDATE 
+                summary = :summary, 
+                message_count = :message_count,
+                last_updated = CURRENT_TIMESTAMP
+        """
+
+
+class ChatMemory:
+    """Manages chat history with short-term and long-term storage."""
+    
+    def __init__(self, session_id: str, user_id: str = "default", max_messages: int = 20, db_connection=None):
+        self.session_id = session_id
+        self.user_id = user_id
+        self.max_messages = max_messages
+        self.db = db_connection
+        self.messages: List[ChatMessage] = []
+        self._db_type = None
+        
+        if self.db:
+            self._db_type = self.db.db_type.value
+            self._ensure_tables()
+    
+    def _ensure_tables(self):
+        """Create memory tables if they don't exist."""
+        try:
+            memory_ddl = get_memory_table_ddl(self._db_type)
+            permanent_ddl = get_permanent_memory_ddl(self._db_type)
+            
+            self.db.execute_write(memory_ddl)
+            self.db.execute_write(permanent_ddl)
+            
+            # Create indexes for SQLite and PostgreSQL (MySQL creates them inline)
+            if self._db_type in ("sqlite", "postgresql"):
+                self._create_indexes()
+            
+            # Migration: Ensure user_id column exists (MySQL only for legacy support)
+            if self._db_type == "mysql":
+                self._migrate_mysql_user_id()
+                
+        except Exception as e:
+            logger.warning(f"Failed to create memory tables: {e}")
+    
+    def _create_indexes(self):
+        """Create indexes for SQLite and PostgreSQL."""
+        try:
+            if self._db_type == "sqlite":
+                self.db.execute_write("CREATE INDEX IF NOT EXISTS idx_memory_session ON _chatbot_memory(session_id)")
+                self.db.execute_write("CREATE INDEX IF NOT EXISTS idx_memory_user ON _chatbot_memory(user_id)")
+                self.db.execute_write("CREATE INDEX IF NOT EXISTS idx_memory_created ON _chatbot_memory(created_at)")
+                self.db.execute_write("CREATE INDEX IF NOT EXISTS idx_permanent_user ON _chatbot_permanent_memory_v2(user_id)")
+            elif self._db_type == "postgresql":
+                self.db.execute_write("CREATE INDEX IF NOT EXISTS idx_memory_session ON _chatbot_memory(session_id)")
+                self.db.execute_write("CREATE INDEX IF NOT EXISTS idx_memory_user ON _chatbot_memory(user_id)")
+                self.db.execute_write("CREATE INDEX IF NOT EXISTS idx_memory_created ON _chatbot_memory(created_at)")
+                self.db.execute_write("CREATE INDEX IF NOT EXISTS idx_permanent_user ON _chatbot_permanent_memory_v2(user_id)")
+        except Exception as e:
+            logger.debug(f"Index creation (may already exist): {e}")
+    
+    def _migrate_mysql_user_id(self):
+        """Migrate MySQL table to include user_id column if missing."""
+        try:
+            check_query = """
+                SELECT COLUMN_NAME 
+                FROM INFORMATION_SCHEMA.COLUMNS 
+                WHERE TABLE_SCHEMA = :db_name 
+                AND TABLE_NAME = '_chatbot_memory' 
+                AND COLUMN_NAME = 'user_id'
+            """
+            db_name = self.db.config.database
+            result = self.db.execute_query(check_query, {"db_name": db_name})
+            
+            if not result:
+                self.db.execute_write("ALTER TABLE _chatbot_memory ADD COLUMN user_id VARCHAR(255) NOT NULL DEFAULT 'default' AFTER session_id")
+                self.db.execute_write("CREATE INDEX idx_user ON _chatbot_memory(user_id)")
+                logger.info("Migrated _chatbot_memory to include user_id")
+        except Exception as e:
+            logger.debug(f"Migration check failed: {e}")
+    
+    def add_message(self, role: str, content: str, metadata: Dict = None):
+        """Add a message to memory and optionally persist it."""
+        msg = ChatMessage(role=role, content=content, metadata=metadata)
+        self.messages.append(msg)
+        
+        # Trim if exceeds max (short-term)
+        if len(self.messages) > self.max_messages:
+            self.messages = self.messages[-self.max_messages:]
+            
+        # Persist to DB (session history)
+        if self.db:
+            try:
+                query = """
+                    INSERT INTO _chatbot_memory (session_id, user_id, role, content, metadata)
+                    VALUES (:session_id, :user_id, :role, :content, :metadata)
+                """
+                self.db.execute_write(query, {
+                    "session_id": self.session_id,
+                    "user_id": self.user_id,
+                    "role": role,
+                    "content": content,
+                    "metadata": json.dumps(metadata) if metadata else None
+                })
+            except Exception as e:
+                logger.warning(f"Failed to persist message: {e}")
+    
+    def save_permanent_context(self, content: str, tags: str = "user_saved"):
+        """Save specific context explicitly to permanent memory for this user."""
+        if not self.db:
+            return False, "No database connection"
+        
+        try:
+            query = """
+                INSERT INTO _chatbot_permanent_memory_v2 (user_id, content, tags)
+                VALUES (:user_id, :content, :tags)
+            """
+            self.db.execute_write(query, {
+                "user_id": self.user_id,
+                "content": content, 
+                "tags": tags
+            })
+            return True, "Context saved to permanent memory"
+        except Exception as e:
+            logger.error(f"Failed to save permanent context: {e}")
+            return False, str(e)
+
+    def get_permanent_context(self, limit: int = 5) -> List[str]:
+        """Retrieve recent permanent context for this user only."""
+        if not self.db:
+            return []
+            
+        try:
+            # Use database-agnostic LIMIT syntax
+            query = """
+                SELECT content FROM _chatbot_permanent_memory_v2 
+                WHERE user_id = :user_id 
+                ORDER BY created_at DESC LIMIT :limit
+            """
+            rows = self.db.execute_query(query, {
+                "user_id": self.user_id,
+                "limit": limit
+            })
+            return [row['content'] for row in rows]
+        except Exception as e:
+            logger.warning(f"Failed to load permanent context: {e}")
+            return []
+    
+    def get_messages(self, limit: Optional[int] = None) -> List[Dict[str, str]]:
+        """Get messages for LLM context."""
+        msgs = self.messages if limit is None else self.messages[-limit:]
+        return [m.to_dict() for m in msgs]
+    
+    def get_context_messages(self, count: int = 5) -> List[Dict[str, str]]:
+        """Get recent messages plus permanent context for injection."""
+        # Get short-term session messages
+        context = self.get_messages(limit=count)
+        
+        # Inject permanent memory if available
+        perm_docs = self.get_permanent_context(limit=3)
+        if perm_docs:
+            perm_context = f"IMPORTANT CONTEXT FOR USER '{self.user_id}':\n" + "\n".join(perm_docs)
+            # Add as a system note at the start
+            context.insert(0, {"role": "system", "content": perm_context})
+            
+        return context
+    
+    def clear(self):
+        """Clear current session memory and remove from DB (temporary history)."""
+        self.messages = []
+        
+        if self.db:
+            try:
+                # Delete temporary messages for this session
+                query = "DELETE FROM _chatbot_memory WHERE session_id = :session_id"
+                self.db.execute_write(query, {"session_id": self.session_id})
+                logger.info(f"Cleared session memory for {self.session_id}")
+            except Exception as e:
+                logger.warning(f"Failed to clear memory from DB: {e}")
+
+    def clear_user_history(self):
+        """Clear ALL temporary history for this user (across all sessions)."""
+        self.messages = []
+        if self.db:
+            try:
+                query = "DELETE FROM _chatbot_memory WHERE user_id = :user_id"
+                self.db.execute_write(query, {"user_id": self.user_id})
+                logger.info(f"Cleared all temporary history for user: {self.user_id}")
+            except Exception as e:
+                logger.warning(f"Failed to clear user history from DB: {e}")
+
+
+class ConversationSummaryMemory:
+    """
+    Per-user conversation summary memory using LLM for summarization.
+    
+    This class maintains a running summary of the conversation, updating it
+    periodically (when message count exceeds threshold). This dramatically
+    reduces token usage while preserving context for long conversations.
+    
+    Features:
+    - Automatic summarization when threshold is reached
+    - Per-user summary storage in database
+    - Combines summary + recent messages for optimal context
+    - Lazy summarization (only when needed)
+    """
+    
+    SUMMARIZATION_PROMPT = """You are a conversation summarizer. Create a concise summary of the conversation below that captures:
+1. Key topics discussed
+2. Important facts or preferences mentioned by the user
+3. Any decisions or conclusions reached
+4. Context needed for follow-up questions
+
+Keep the summary under 300 words but include all important details.
+
+CONVERSATION:
+{conversation}
+
+SUMMARY:"""
+
+    INCREMENTAL_SUMMARY_PROMPT = """You are a conversation summarizer. Update the existing summary to incorporate new messages.
+
+EXISTING SUMMARY:
+{existing_summary}
+
+NEW MESSAGES:
+{new_messages}
+
+Create an updated, comprehensive summary that:
+1. Incorporates new information from the recent messages
+2. Retains important context from the existing summary
+3. Removes redundant or outdated information
+4. Stays under 300 words
+
+UPDATED SUMMARY:"""
+
+    def __init__(
+        self, 
+        user_id: str, 
+        session_id: str, 
+        db_connection=None,
+        llm_client=None,
+        summary_threshold: int = 10,  # Summarize every N messages
+        recent_messages_count: int = 5  # Keep this many recent messages verbatim
+    ):
+        self.user_id = user_id
+        self.session_id = session_id
+        self.db = db_connection
+        self.llm = llm_client
+        self.summary_threshold = summary_threshold
+        self.recent_messages_count = recent_messages_count
+        self._db_type = None
+        
+        self._cached_summary: Optional[str] = None
+        self._messages_since_summary: int = 0
+        
+        if self.db:
+            self._db_type = self.db.db_type.value
+            self._ensure_tables()
+            self._load_state()
+    
+    def _ensure_tables(self):
+        """Create summary table if it doesn't exist."""
+        try:
+            ddl = get_summary_table_ddl(self._db_type)
+            self.db.execute_write(ddl)
+        except Exception as e:
+            logger.warning(f"Failed to create summary table: {e}")
+    
+    def _load_state(self):
+        """Load existing summary state from database (per-user, not per-session)."""
+        try:
+            query = """
+                SELECT summary, message_count FROM _chatbot_user_summaries 
+                WHERE user_id = :user_id
+            """
+            rows = self.db.execute_query(query, {
+                "user_id": self.user_id
+            })
+            if rows:
+                self._cached_summary = rows[0].get('summary')
+                self._messages_since_summary = 0  # Reset since we loaded
+                logger.debug(f"Loaded summary for user {self.user_id}")
+        except Exception as e:
+            logger.warning(f"Failed to load summary state: {e}")
+    
+    def set_llm_client(self, llm_client):
+        """Set the LLM client for summarization."""
+        self.llm = llm_client
+    
+    def on_message_added(self, message_count: int):
+        """
+        Called after a message is added to track when to summarize.
+        
+        Args:
+            message_count: Current total number of messages in the conversation
+        """
+        self._messages_since_summary += 1
+        
+        # Check if we should summarize
+        if self._messages_since_summary >= self.summary_threshold:
+            self._trigger_summarization()
+    
+    def _trigger_summarization(self):
+        """Trigger summarization of the conversation."""
+        if not self.llm:
+            logger.warning("Cannot summarize: No LLM client configured")
+            return
+        
+        if not self.db:
+            logger.warning("Cannot summarize: No database connection")
+            return
+        
+        try:
+            # Get messages that need to be summarized
+            query = """
+                SELECT role, content FROM _chatbot_memory 
+                WHERE user_id = :user_id AND session_id = :session_id
+                ORDER BY created_at ASC
+            """
+            rows = self.db.execute_query(query, {
+                "user_id": self.user_id,
+                "session_id": self.session_id
+            })
+            
+            if not rows:
+                return
+            
+            # Format conversation for summarization
+            conversation_text = self._format_messages_for_summary(rows)
+            
+            # Generate summary
+            if self._cached_summary:
+                # Incremental update
+                prompt = self.INCREMENTAL_SUMMARY_PROMPT.format(
+                    existing_summary=self._cached_summary,
+                    new_messages=conversation_text
+                )
+            else:
+                # Fresh summary
+                prompt = self.SUMMARIZATION_PROMPT.format(conversation=conversation_text)
+            
+            messages = [
+                {"role": "system", "content": "You are a helpful assistant that creates concise conversation summaries."},
+                {"role": "user", "content": prompt}
+            ]
+            
+            summary = self.llm.chat(messages)
+            
+            # Save to database
+            self._save_summary(summary, len(rows))
+            
+            self._cached_summary = summary
+            self._messages_since_summary = 0
+            
+            logger.info(f"Generated summary for user {self.user_id}")
+            
+        except Exception as e:
+            logger.error(f"Summarization failed: {e}")
+    
+    def _format_messages_for_summary(self, messages: List[Dict]) -> str:
+        """Format messages as text for summarization."""
+        lines = []
+        for msg in messages:
+            role = msg.get('role', 'unknown').upper()
+            content = msg.get('content', '')
+            lines.append(f"{role}: {content}")
+        return "\n\n".join(lines)
+    
+    def _save_summary(self, summary: str, message_count: int):
+        """Save or update summary in database (per-user)."""
+        try:
+            query = get_upsert_summary_query(self._db_type)
+            self.db.execute_write(query, {
+                "user_id": self.user_id,
+                "summary": summary,
+                "message_count": message_count
+            })
+        except Exception as e:
+            logger.error(f"Failed to save summary: {e}")
+    
+    def get_summary(self) -> Optional[str]:
+        """Get the current conversation summary."""
+        return self._cached_summary
+    
+    def get_context_for_llm(self, recent_messages: List[Dict[str, str]]) -> List[Dict[str, str]]:
+        """
+        Get optimized context for LLM calls.
+        
+        Combines the summary (if available) with recent messages for optimal
+        token usage while maintaining context.
+        
+        Args:
+            recent_messages: List of recent messages to include verbatim
+            
+        Returns:
+            List of messages with summary prepended as system context
+        """
+        context_messages = []
+        
+        # Add summary as system context if available
+        if self._cached_summary:
+            summary_context = f"""CONVERSATION SUMMARY (previous context):
+{self._cached_summary}
+
+Use this summary to understand the conversation history and context for follow-up questions."""
+            context_messages.append({
+                "role": "system",
+                "content": summary_context
+            })
+        
+        # Add recent messages verbatim
+        context_messages.extend(recent_messages[-self.recent_messages_count:])
+        
+        return context_messages
+    
+    def force_summarize(self):
+        """Force immediate summarization regardless of threshold."""
+        self._trigger_summarization()
+    
+    def clear_summary(self):
+        """Clear the summary for this user."""
+        self._cached_summary = None
+        self._messages_since_summary = 0
+        
+        if self.db:
+            try:
+                query = "DELETE FROM _chatbot_user_summaries WHERE user_id = :user_id"
+                self.db.execute_write(query, {
+                    "user_id": self.user_id
+                })
+                logger.info(f"Cleared summary for user: {self.user_id}")
+            except Exception as e:
+                logger.warning(f"Failed to clear summary: {e}")
+    
+    def clear_all_user_summaries(self):
+        """Clear all summaries for this user (alias for clear_summary since it's now per-user)."""
+        self.clear_summary()
+
+
+class EnhancedChatMemory(ChatMemory):
+    """
+    Enhanced ChatMemory with integrated conversation summarization.
+    
+    Combines the standard ChatMemory functionality with ConversationSummaryMemory
+    for automatic summarization and optimized context retrieval.
+    """
+    
+    def __init__(
+        self, 
+        session_id: str, 
+        user_id: str = "default", 
+        max_messages: int = 20, 
+        db_connection=None,
+        llm_client=None,
+        enable_summarization: bool = True,
+        summary_threshold: int = 10
+    ):
+        super().__init__(session_id, user_id, max_messages, db_connection)
+        
+        self.enable_summarization = enable_summarization
+        self.summary_memory: Optional[ConversationSummaryMemory] = None
+        
+        if enable_summarization:
+            self.summary_memory = ConversationSummaryMemory(
+                user_id=user_id,
+                session_id=session_id,
+                db_connection=db_connection,
+                llm_client=llm_client,
+                summary_threshold=summary_threshold
+            )
+    
+    def set_llm_client(self, llm_client):
+        """Set the LLM client for summarization."""
+        if self.summary_memory:
+            self.summary_memory.set_llm_client(llm_client)
+    
+    def add_message(self, role: str, content: str, metadata: Dict = None):
+        """Add a message and trigger summarization check."""
+        super().add_message(role, content, metadata)
+        
+        # Notify summary memory of new message
+        if self.summary_memory:
+            self.summary_memory.on_message_added(len(self.messages))
+    
+    def get_context_messages(self, count: int = 5) -> List[Dict[str, str]]:
+        """
+        Get context messages with summary integration.
+        
+        If summarization is enabled and a summary exists, it will be 
+        prepended to provide historical context while keeping recent
+        messages verbatim.
+        """
+        # Get base context from parent
+        base_context = super().get_context_messages(count)
+        
+        # If summarization is enabled, use summary-enhanced context
+        if self.summary_memory and self.summary_memory.get_summary():
+            # Filter out system messages from base context (we'll add summary separately)
+            filtered = [m for m in base_context if m.get("role") != "system"]
+            
+            # Get summary-enhanced context
+            enhanced = self.summary_memory.get_context_for_llm(filtered)
+            
+            # Re-add permanent memory context if it was present
+            for msg in base_context:
+                if msg.get("role") == "system" and "IMPORTANT CONTEXT" in msg.get("content", ""):
+                    enhanced.insert(0, msg)
+            
+            return enhanced
+        
+        return base_context
+    
+    def get_summary(self) -> Optional[str]:
+        """Get the current conversation summary."""
+        if self.summary_memory:
+            return self.summary_memory.get_summary()
+        return None
+    
+    def force_summarize(self):
+        """Force immediate summarization."""
+        if self.summary_memory:
+            self.summary_memory.force_summarize()
+    
+    def clear(self):
+        """Clear session memory but KEEP the summary (long-term memory)."""
+        super().clear()
+        # NOTE: Summary is intentionally NOT cleared here
+        # Summary acts as long-term memory that persists across chat sessions
+    
+    def clear_with_summary(self):
+        """Clear session memory AND the summary (full reset)."""
+        super().clear()
+        if self.summary_memory:
+            self.summary_memory.clear_summary()
+    
+    def clear_user_history(self):
+        """Clear all user temp history but KEEP summaries."""
+        super().clear_user_history()
+        # NOTE: Summaries are intentionally NOT cleared
+        # They persist as long-term memory for the user
+    
+    def clear_all_including_summaries(self):
+        """Clear ALL user data including summaries (complete wipe)."""
+        super().clear_user_history()
+        if self.summary_memory:
+            self.summary_memory.clear_all_user_summaries()
+
+
+def create_memory(session_id: str, user_id: str = "default", max_messages: int = 20) -> ChatMemory:
+    """Create a standard ChatMemory instance."""
+    from database import get_db
+    db = get_db()
+    return ChatMemory(session_id=session_id, user_id=user_id, max_messages=max_messages, db_connection=db)
+
+
+def create_enhanced_memory(
+    session_id: str, 
+    user_id: str = "default", 
+    max_messages: int = 20,
+    llm_client=None,
+    enable_summarization: bool = True,
+    summary_threshold: int = 10
+) -> EnhancedChatMemory:
+    """
+    Create an EnhancedChatMemory with summarization support.
+    
+    Args:
+        session_id: Unique session identifier
+        user_id: User identifier for per-user memory isolation
+        max_messages: Maximum messages to keep in short-term memory
+        llm_client: LLM client for summarization (can be set later)
+        enable_summarization: Whether to enable automatic summarization
+        summary_threshold: Summarize after this many messages
+        
+    Returns:
+        EnhancedChatMemory instance with summarization capabilities
+    """
+    from database import get_db
+    db = get_db()
+    return EnhancedChatMemory(
+        session_id=session_id,
+        user_id=user_id,
+        max_messages=max_messages,
+        db_connection=db,
+        llm_client=llm_client,
+        enable_summarization=enable_summarization,
+        summary_threshold=summary_threshold
+    )

rag/__init__.py

@@ -0,0 +1,20 @@
+"""RAG module exports."""
+
+from .embeddings import (
+    EmbeddingProvider,
+    SentenceTransformerEmbedding,
+    OpenAIEmbedding,
+    get_embedding_provider,
+    create_embedding_provider
+)
+from .document_processor import Document, DocumentProcessor, get_document_processor
+from .vector_store import VectorStore, get_vector_store
+from .rag_engine import RAGEngine, get_rag_engine
+
+__all__ = [
+    "EmbeddingProvider", "SentenceTransformerEmbedding", "OpenAIEmbedding",
+    "get_embedding_provider", "create_embedding_provider",
+    "Document", "DocumentProcessor", "get_document_processor",
+    "VectorStore", "get_vector_store",
+    "RAGEngine", "get_rag_engine"
+]

rag/document_processor.py

@@ -0,0 +1,122 @@
+"""
+Document Processor for RAG.
+
+Converts database rows into semantic documents for embedding.
+"""
+
+import logging
+import hashlib
+from dataclasses import dataclass, field
+from typing import List, Dict, Any, Optional, Generator
+import re
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class Document:
+    """Semantic document from the database."""
+    id: str
+    content: str
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    table_name: str = ""
+    column_name: str = ""
+    primary_key_value: Optional[str] = None
+    chunk_index: int = 0
+    total_chunks: int = 1
+    
+    def __post_init__(self):
+        if not self.id:
+            hash_input = f"{self.table_name}:{self.column_name}:{self.primary_key_value}:{self.chunk_index}"
+            self.id = hashlib.md5(hash_input.encode()).hexdigest()
+    
+    def to_context_string(self) -> str:
+        source = f"[Source: {self.table_name}.{self.column_name}"
+        if self.primary_key_value:
+            source += f" (id: {self.primary_key_value})"
+        source += "]"
+        return f"{source}\n{self.content}"
+
+
+class TextChunker:
+    """Splits long text into overlapping chunks."""
+    
+    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+        self.sentence_pattern = re.compile(r'(?<=[.!?])\s+(?=[A-Z])')
+    
+    def chunk_text(self, text: str) -> List[str]:
+        if not text or len(text) <= self.chunk_size:
+            return [text] if text else []
+        
+        sentences = self.sentence_pattern.split(text)
+        chunks = []
+        current_chunk = []
+        current_length = 0
+        
+        for sentence in sentences:
+            sentence = sentence.strip()
+            if not sentence:
+                continue
+            
+            if current_length + len(sentence) + 1 > self.chunk_size:
+                if current_chunk:
+                    chunks.append(' '.join(current_chunk))
+                current_chunk = [sentence]
+                current_length = len(sentence)
+            else:
+                current_chunk.append(sentence)
+                current_length += len(sentence) + 1
+        
+        if current_chunk:
+            chunks.append(' '.join(current_chunk))
+        
+        return chunks if chunks else [text]
+
+
+class DocumentProcessor:
+    """Converts database rows into semantic documents."""
+    
+    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
+        self.chunker = TextChunker(chunk_size, chunk_overlap)
+    
+    def process_row(
+        self, row: Dict[str, Any], table_name: str,
+        text_columns: List[str], primary_key_column: Optional[str] = None
+    ) -> List[Document]:
+        documents = []
+        pk_value = str(row.get(primary_key_column, "")) if primary_key_column else None
+        
+        for column_name in text_columns:
+            text = row.get(column_name)
+            if not text or not isinstance(text, str):
+                continue
+            
+            text = text.strip()
+            if not text:
+                continue
+            
+            chunks = self.chunker.chunk_text(text)
+            for i, chunk in enumerate(chunks):
+                doc = Document(
+                    id="", content=chunk, table_name=table_name,
+                    column_name=column_name, primary_key_value=pk_value,
+                    chunk_index=i, total_chunks=len(chunks),
+                    metadata={"table": table_name, "column": column_name, "pk": pk_value}
+                )
+                documents.append(doc)
+        
+        return documents
+    
+    def process_rows(
+        self, rows: List[Dict[str, Any]], table_name: str,
+        text_columns: List[str], primary_key_column: Optional[str] = None
+    ) -> Generator[Document, None, None]:
+        for row in rows:
+            for doc in self.process_row(row, table_name, text_columns, primary_key_column):
+                yield doc
+
+
+def get_document_processor(chunk_size: int = 500, chunk_overlap: int = 50) -> DocumentProcessor:
+    return DocumentProcessor(chunk_size, chunk_overlap)

rag/embeddings.py

@@ -0,0 +1,206 @@
+"""
+Embedding Generation Module.
+
+Supports:
+- Sentence Transformers (local, free)
+- OpenAI Embeddings (cloud, paid)
+
+Configurable via environment variables.
+"""
+
+import logging
+from abc import ABC, abstractmethod
+from typing import List, Optional
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+class EmbeddingProvider(ABC):
+    """Abstract base class for embedding providers."""
+    
+    @abstractmethod
+    def embed_text(self, text: str) -> np.ndarray:
+        """Generate embedding for a single text."""
+        pass
+    
+    @abstractmethod
+    def embed_texts(self, texts: List[str]) -> np.ndarray:
+        """Generate embeddings for multiple texts."""
+        pass
+    
+    @property
+    @abstractmethod
+    def dimension(self) -> int:
+        """Return the embedding dimension."""
+        pass
+
+
+class SentenceTransformerEmbedding(EmbeddingProvider):
+    """
+    Sentence Transformers embedding provider.
+    
+    Uses local models, no API key required.
+    Default: all-MiniLM-L6-v2 (384 dimensions)
+    """
+    
+    def __init__(self, model_name: str = "sentence-transformers/all-MiniLM-L6-v2"):
+        """
+        Initialize the Sentence Transformer model.
+        
+        Args:
+            model_name: HuggingFace model name
+        """
+        self.model_name = model_name
+        self._model = None
+        self._dimension = None
+    
+    @property
+    def model(self):
+        """Lazy load the model."""
+        if self._model is None:
+            try:
+                from sentence_transformers import SentenceTransformer
+                logger.info(f"Loading embedding model: {self.model_name}")
+                self._model = SentenceTransformer(self.model_name)
+                self._dimension = self._model.get_sentence_embedding_dimension()
+                logger.info(f"Model loaded. Embedding dimension: {self._dimension}")
+            except ImportError:
+                raise ImportError(
+                    "sentence-transformers is required. Install with: pip install sentence-transformers"
+                )
+        return self._model
+    
+    @property
+    def dimension(self) -> int:
+        """Get embedding dimension."""
+        if self._dimension is None:
+            _ = self.model  # Force model load
+        return self._dimension
+    
+    def embed_text(self, text: str) -> np.ndarray:
+        """Generate embedding for a single text."""
+        return self.model.encode(text, convert_to_numpy=True)
+    
+    def embed_texts(self, texts: List[str]) -> np.ndarray:
+        """Generate embeddings for multiple texts."""
+        return self.model.encode(texts, convert_to_numpy=True, show_progress_bar=len(texts) > 100)
+
+
+class OpenAIEmbedding(EmbeddingProvider):
+    """
+    OpenAI embedding provider.
+    
+    Uses OpenAI API, requires API key.
+    Default: text-embedding-3-small (1536 dimensions)
+    """
+    
+    DIMENSION_MAP = {
+        "text-embedding-3-small": 1536,
+        "text-embedding-3-large": 3072,
+        "text-embedding-ada-002": 1536
+    }
+    
+    def __init__(self, api_key: str, model_name: str = "text-embedding-3-small"):
+        """
+        Initialize OpenAI embedding client.
+        
+        Args:
+            api_key: OpenAI API key
+            model_name: OpenAI embedding model name
+        """
+        self.api_key = api_key
+        self.model_name = model_name
+        self._client = None
+        self._dimension = self.DIMENSION_MAP.get(model_name, 1536)
+    
+    @property
+    def client(self):
+        """Lazy load the OpenAI client."""
+        if self._client is None:
+            try:
+                from openai import OpenAI
+                self._client = OpenAI(api_key=self.api_key)
+            except ImportError:
+                raise ImportError(
+                    "openai is required. Install with: pip install openai"
+                )
+        return self._client
+    
+    @property
+    def dimension(self) -> int:
+        """Get embedding dimension."""
+        return self._dimension
+    
+    def embed_text(self, text: str) -> np.ndarray:
+        """Generate embedding for a single text."""
+        response = self.client.embeddings.create(
+            input=text,
+            model=self.model_name
+        )
+        return np.array(response.data[0].embedding, dtype=np.float32)
+    
+    def embed_texts(self, texts: List[str]) -> np.ndarray:
+        """Generate embeddings for multiple texts (batch)."""
+        # OpenAI API supports batching up to 2048 inputs
+        batch_size = 100
+        all_embeddings = []
+        
+        for i in range(0, len(texts), batch_size):
+            batch = texts[i:i + batch_size]
+            response = self.client.embeddings.create(
+                input=batch,
+                model=self.model_name
+            )
+            embeddings = [np.array(d.embedding, dtype=np.float32) for d in response.data]
+            all_embeddings.extend(embeddings)
+        
+        return np.array(all_embeddings)
+
+
+def create_embedding_provider(
+    provider_type: str = "sentence_transformers",
+    model_name: Optional[str] = None,
+    api_key: Optional[str] = None
+) -> EmbeddingProvider:
+    """
+    Factory function to create the appropriate embedding provider.
+    
+    Args:
+        provider_type: "sentence_transformers" or "openai"
+        model_name: Model name (optional, uses defaults)
+        api_key: API key for OpenAI (required if using OpenAI)
+        
+    Returns:
+        Configured EmbeddingProvider instance
+    """
+    if provider_type == "openai":
+        if not api_key:
+            raise ValueError("OpenAI API key is required for OpenAI embeddings")
+        return OpenAIEmbedding(
+            api_key=api_key,
+            model_name=model_name or "text-embedding-3-small"
+        )
+    else:
+        return SentenceTransformerEmbedding(
+            model_name=model_name or "sentence-transformers/all-MiniLM-L6-v2"
+        )
+
+
+# Global embedding provider instance
+_embedding_provider: Optional[EmbeddingProvider] = None
+
+
+def get_embedding_provider() -> EmbeddingProvider:
+    """Get or create the global embedding provider."""
+    global _embedding_provider
+    if _embedding_provider is None:
+        # Default to sentence transformers (free, local)
+        _embedding_provider = SentenceTransformerEmbedding()
+    return _embedding_provider
+
+
+def set_embedding_provider(provider: EmbeddingProvider):
+    """Set the global embedding provider."""
+    global _embedding_provider
+    _embedding_provider = provider

rag/rag_engine.py

@@ -0,0 +1,120 @@
+"""
+RAG Engine - Orchestrates the retrieval-augmented generation pipeline.
+
+Handles:
+- Automatic indexing of text columns from the database
+- Semantic retrieval using FAISS
+- Context building for the LLM
+"""
+
+import logging
+from typing import List, Dict, Any, Optional, Tuple
+
+from .document_processor import Document, get_document_processor
+from .vector_store import VectorStore, get_vector_store
+from .embeddings import get_embedding_provider
+
+logger = logging.getLogger(__name__)
+
+
+class RAGEngine:
+    """Main RAG engine for semantic retrieval from database text."""
+    
+    def __init__(self, vector_store: Optional[VectorStore] = None):
+        self.vector_store = vector_store or get_vector_store()
+        self.doc_processor = get_document_processor()
+        self.indexed_tables: Dict[str, bool] = {}
+    
+    def index_table(
+        self,
+        table_name: str,
+        rows: List[Dict[str, Any]],
+        text_columns: List[str],
+        primary_key_column: Optional[str] = None
+    ) -> int:
+        """
+        Index text data from a table.
+        
+        Returns:
+            Number of documents indexed
+        """
+        documents = list(self.doc_processor.process_rows(
+            rows, table_name, text_columns, primary_key_column
+        ))
+        
+        if documents:
+            self.vector_store.add_documents(documents)
+            self.indexed_tables[table_name] = True
+            logger.info(f"Indexed {len(documents)} documents from {table_name}")
+        
+        return len(documents)
+    
+    def search(
+        self, 
+        query: str, 
+        top_k: int = 5,
+        table_filter: Optional[List[str]] = None
+    ) -> List[Tuple[Document, float]]:
+        """
+        Search for relevant documents.
+        
+        Args:
+            query: Search query
+            top_k: Number of results
+            table_filter: Optional list of tables to search in
+            
+        Returns:
+            List of (document, score) tuples
+        """
+        results = self.vector_store.search(query, top_k=top_k * 2)
+        
+        if table_filter:
+            results = [
+                (doc, score) for doc, score in results 
+                if doc.table_name in table_filter
+            ]
+        
+        return results[:top_k]
+    
+    def get_context(
+        self, 
+        query: str, 
+        top_k: int = 5,
+        table_filter: Optional[List[str]] = None
+    ) -> str:
+        """
+        Get formatted context for LLM from search results.
+        """
+        results = self.search(query, top_k, table_filter)
+        
+        if not results:
+            return "No relevant information found in the database."
+        
+        context_parts = []
+        for doc, score in results:
+            context_parts.append(doc.to_context_string())
+        
+        return "\n\n---\n\n".join(context_parts)
+    
+    def clear_index(self):
+        """Clear the entire index."""
+        self.vector_store.clear()
+        self.indexed_tables = {}
+    
+    def save(self):
+        """Save the index to disk."""
+        self.vector_store.save()
+    
+    @property
+    def document_count(self) -> int:
+        return len(self.vector_store)
+
+
+_rag_engine: Optional[RAGEngine] = None
+
+
+def get_rag_engine() -> RAGEngine:
+    global _rag_engine
+    if _rag_engine is None:
+        _rag_engine = RAGEngine()
+    return _rag_engine

rag/vector_store.py

@@ -0,0 +1,173 @@
+"""
+FAISS Vector Store for RAG.
+
+Manages the FAISS index for semantic search over database text content.
+"""
+
+import logging
+import pickle
+import os
+from typing import List, Dict, Any, Optional, Tuple
+import numpy as np
+
+try:
+    import faiss
+except ImportError:
+    faiss = None
+
+from .document_processor import Document
+from .embeddings import get_embedding_provider, EmbeddingProvider
+
+logger = logging.getLogger(__name__)
+
+
+class VectorStore:
+    """FAISS-based vector store for semantic search."""
+    
+    def __init__(
+        self, 
+        embedding_provider: Optional[EmbeddingProvider] = None,
+        index_path: str = "./faiss_index"
+    ):
+        if faiss is None:
+            raise ImportError("faiss-cpu is required. Install with: pip install faiss-cpu")
+        
+        self.embedding_provider = embedding_provider or get_embedding_provider()
+        self.index_path = index_path
+        self.dimension = self.embedding_provider.dimension
+        
+        self.index: Optional[faiss.IndexFlatIP] = None
+        self.documents: List[Document] = []
+        self.id_to_idx: Dict[str, int] = {}
+        
+        self._initialize_index()
+    
+    def _initialize_index(self):
+        """Initialize or load the FAISS index."""
+        index_file = os.path.join(self.index_path, "index.faiss")
+        docs_file = os.path.join(self.index_path, "documents.pkl")
+        
+        if os.path.exists(index_file) and os.path.exists(docs_file):
+            try:
+                # Check file size - if 0 something is wrong
+                if os.path.getsize(index_file) > 0:
+                    self.index = faiss.read_index(index_file)
+                    with open(docs_file, 'rb') as f:
+                        self.documents = pickle.load(f)
+                    self.id_to_idx = {doc.id: i for i, doc in enumerate(self.documents)}
+                    
+                    # Verify index dimension matches expected
+                    if self.index.d != self.dimension:
+                        logger.warning(f"Index dimension mismatch: {self.index.d} != {self.dimension}. Resetting.")
+                        raise ValueError("Dimension mismatch")
+                        
+                    logger.info(f"Loaded index with {len(self.documents)} documents")
+                    return
+            except (Exception, RuntimeError) as e:
+                logger.warning(f"Failed to load index (might be corrupted or memory error): {e}")
+                # If loading fails, we should probably backup the broken files or just overwrite
+                if os.path.exists(index_file):
+                    try:
+                        os.rename(index_file, index_file + ".bak")
+                        os.rename(docs_file, docs_file + ".bak")
+                    except:
+                        pass
+        
+        # Create new index (Inner Product for cosine similarity with normalized vectors)
+        self.index = faiss.IndexFlatIP(self.dimension)
+        self.documents = []
+        self.id_to_idx = {}
+        logger.info(f"Created new FAISS index with dimension {self.dimension}")
+    
+    def add_documents(self, documents: List[Document], batch_size: int = 100):
+        """Add documents to the vector store."""
+        if not documents:
+            return
+        
+        new_docs = [doc for doc in documents if doc.id not in self.id_to_idx]
+        if not new_docs:
+            logger.info("No new documents to add")
+            return
+        
+        logger.info(f"Adding {len(new_docs)} documents to index")
+        
+        for i in range(0, len(new_docs), batch_size):
+            batch = new_docs[i:i + batch_size]
+            texts = [doc.content for doc in batch]
+            
+            embeddings = self.embedding_provider.embed_texts(texts)
+            
+            # Normalize for cosine similarity
+            faiss.normalize_L2(embeddings)
+            
+            start_idx = len(self.documents)
+            self.index.add(embeddings)
+            
+            for j, doc in enumerate(batch):
+                self.documents.append(doc)
+                self.id_to_idx[doc.id] = start_idx + j
+        
+        logger.info(f"Index now contains {len(self.documents)} documents")
+    
+    def search(
+        self, query: str, top_k: int = 5, threshold: float = 0.0
+    ) -> List[Tuple[Document, float]]:
+        """Search for similar documents."""
+        if not self.documents:
+            return []
+        
+        query_embedding = self.embedding_provider.embed_text(query)
+        query_embedding = query_embedding.reshape(1, -1).astype(np.float32)
+        faiss.normalize_L2(query_embedding)
+        
+        k = min(top_k, len(self.documents))
+        scores, indices = self.index.search(query_embedding, k)
+        
+        results = []
+        for score, idx in zip(scores[0], indices[0]):
+            if idx >= 0 and score >= threshold:
+                results.append((self.documents[idx], float(score)))
+        
+        return results
+    
+    def save(self):
+        """Save the index to disk."""
+        os.makedirs(self.index_path, exist_ok=True)
+        
+        index_file = os.path.join(self.index_path, "index.faiss")
+        docs_file = os.path.join(self.index_path, "documents.pkl")
+        
+        faiss.write_index(self.index, index_file)
+        with open(docs_file, 'wb') as f:
+            pickle.dump(self.documents, f)
+        
+        logger.info(f"Saved index with {len(self.documents)} documents")
+    
+    def clear(self):
+        """Clear the index."""
+        self.index = faiss.IndexFlatIP(self.dimension)
+        self.documents = []
+        self.id_to_idx = {}
+        
+        # Delete files
+        index_file = os.path.join(self.index_path, "index.faiss")
+        docs_file = os.path.join(self.index_path, "documents.pkl")
+        
+        for f in [index_file, docs_file]:
+            if os.path.exists(f):
+                os.remove(f)
+        
+        logger.info("Index cleared")
+    
+    def __len__(self) -> int:
+        return len(self.documents)
+
+
+_vector_store: Optional[VectorStore] = None
+
+
+def get_vector_store() -> VectorStore:
+    global _vector_store
+    if _vector_store is None:
+        _vector_store = VectorStore()
+    return _vector_store

requirements.txt

@@ -0,0 +1,31 @@
+# Schema-Agnostic Database Chatbot
+# Multi-Database Support: MySQL, PostgreSQL, SQLite
+
+# Core dependencies
+streamlit>=1.30.0
+sqlalchemy>=2.0.0
+
+# Database drivers
+pymysql>=1.1.0       # MySQL driver
+psycopg2-binary>=2.9.9  # PostgreSQL driver
+# SQLite is built into Python - no driver needed
+
+# RAG dependencies
+faiss-cpu>=1.7.4
+sentence-transformers>=2.2.2
+
+# LLM dependencies
+groq>=0.4.0  # FREE API!
+openai>=1.0.0  # Optional, for OpenAI provider
+
+# For local models (optional)
+# transformers>=4.36.0
+# torch>=2.0.0
+
+# SQL parsing and validation
+sqlparse>=0.4.4
+
+# Utilities
+python-dotenv>=1.0.0
+numpy>=1.24.0
+pandas>=2.0.0

router.py

@@ -0,0 +1,164 @@
+"""
+Query Router - Decides between RAG, SQL, or hybrid approach.
+
+Analyzes user intent and routes to the appropriate handler.
+"""
+
+import logging
+from enum import Enum
+from typing import Dict, Any, Optional, Tuple, List
+from dataclasses import dataclass
+
+logger = logging.getLogger(__name__)
+
+
+class QueryType(Enum):
+    RAG = "rag"           # Semantic search in text
+    SQL = "sql"           # Structured query
+    HYBRID = "hybrid"     # Both RAG and SQL
+    GENERAL = "general"   # General conversation
+
+
+@dataclass
+class RoutingDecision:
+    query_type: QueryType
+    confidence: float
+    reasoning: str
+    suggested_tables: List[str] = None
+    
+    def __post_init__(self):
+        if self.suggested_tables is None:
+            self.suggested_tables = []
+
+
+class QueryRouter:
+    """Routes queries to appropriate handlers based on intent analysis."""
+    
+    ROUTING_PROMPT = """Analyze this user query and determine the best approach to answer it.
+
+DATABASE SCHEMA:
+{schema}
+
+USER QUERY: {query}
+
+Determine if this query needs:
+1. RAG - Semantic search through text content (searching for meanings, concepts, descriptions)
+2. SQL - Structured database query (counting, filtering, aggregating, specific lookups)
+3. HYBRID - Both semantic search and structured query
+4. GENERAL - General conversation not requiring database access
+
+Respond in this exact format:
+TYPE: [RAG|SQL|HYBRID|GENERAL]
+CONFIDENCE: [0.0-1.0]
+TABLES: [comma-separated list of relevant tables, or NONE]
+REASONING: [brief explanation]"""
+
+    def __init__(self, llm_client=None):
+        self.llm_client = llm_client
+    
+    def set_llm_client(self, llm_client):
+        self.llm_client = llm_client
+    
+    def route(self, query: str, schema_context: str) -> RoutingDecision:
+        """Analyze query and determine routing."""
+        if not self.llm_client:
+            # Fallback to simple heuristics
+            return self._heuristic_route(query)
+        
+        prompt = self.ROUTING_PROMPT.format(schema=schema_context, query=query)
+        
+        try:
+            response = self.llm_client.chat([
+                {"role": "system", "content": "You are a query routing assistant."},
+                {"role": "user", "content": prompt}
+            ])
+            return self._parse_routing_response(response)
+        except Exception as e:
+            logger.warning(f"LLM routing failed: {e}, using heuristics")
+            return self._heuristic_route(query)
+    
+    def _parse_routing_response(self, response: str) -> RoutingDecision:
+        """Parse LLM routing response."""
+        lines = response.strip().split('\n')
+        
+        query_type = QueryType.GENERAL
+        confidence = 0.5
+        tables = []
+        reasoning = ""
+        
+        for line in lines:
+            line = line.strip()
+            if line.startswith("TYPE:"):
+                type_str = line.replace("TYPE:", "").strip().upper()
+                query_type = QueryType[type_str] if type_str in QueryType.__members__ else QueryType.GENERAL
+            elif line.startswith("CONFIDENCE:"):
+                try:
+                    confidence = float(line.replace("CONFIDENCE:", "").strip())
+                except ValueError:
+                    confidence = 0.5
+            elif line.startswith("TABLES:"):
+                tables_str = line.replace("TABLES:", "").strip()
+                if tables_str.upper() != "NONE":
+                    tables = [t.strip() for t in tables_str.split(",")]
+            elif line.startswith("REASONING:"):
+                reasoning = line.replace("REASONING:", "").strip()
+        
+        return RoutingDecision(query_type, confidence, reasoning, tables)
+    
+    def _heuristic_route(self, query: str) -> RoutingDecision:
+        """Simple heuristic-based routing when LLM is unavailable."""
+        query_lower = query.lower()
+        
+        # SQL keywords - for structured data retrieval
+        sql_keywords = [
+            'how many', 'count', 'total', 'average', 'sum', 'max', 'min',
+            'list all', 'show all', 'find all', 'get all', 'between',
+            'greater than', 'less than', 'equal to', 'top', 'bottom',
+            # Data listing patterns
+            'what products', 'what customers', 'what orders', 'what items',
+            'show me', 'list', 'display', 'give me', 'get me',
+            'all products', 'all customers', 'all orders',
+            'products do you have', 'customers do you have',
+            'from new york', 'from chicago', 'from los angeles',
+            # Specific lookups
+            'price of', 'cost of', 'stock of', 'quantity',
+            'where', 'which', 'who'
+        ]
+        
+        # RAG keywords - for semantic/conceptual questions
+        rag_keywords = [
+            'what is the policy', 'explain', 'describe', 'tell me about',
+            'meaning of', 'definition', 'why', 'how does', 'what does',
+            'similar to', 'return policy', 'shipping policy', 'warranty',
+            'support', 'help with', 'information about', 'details about'
+        ]
+        
+        sql_score = sum(1 for kw in sql_keywords if kw in query_lower)
+        rag_score = sum(1 for kw in rag_keywords if kw in query_lower)
+        
+        # Boost SQL score for common listing patterns
+        if any(word in query_lower for word in ['products', 'customers', 'orders', 'items']):
+            if any(word in query_lower for word in ['what', 'show', 'list', 'all', 'have']):
+                sql_score += 2
+        
+        if sql_score > rag_score:
+            return RoutingDecision(QueryType.SQL, 0.8, "SQL query for data retrieval")
+        elif rag_score > sql_score:
+            return RoutingDecision(QueryType.RAG, 0.8, "Semantic search for concepts")
+        elif sql_score > 0 and rag_score > 0:
+            return RoutingDecision(QueryType.HYBRID, 0.6, "Mixed query type")
+        else:
+            # Default to SQL for simple questions about data
+            if any(word in query_lower for word in ['products', 'customers', 'orders']):
+                return RoutingDecision(QueryType.SQL, 0.6, "Default to SQL for data tables")
+            return RoutingDecision(QueryType.RAG, 0.5, "Default to semantic search")
+
+
+_router: Optional[QueryRouter] = None
+
+
+def get_query_router() -> QueryRouter:
+    global _router
+    if _router is None:
+        _router = QueryRouter()
+    return _router

sql/__init__.py

@@ -0,0 +1,9 @@
+"""SQL module exports."""
+
+from .validator import SQLValidator, SQLValidationError, get_sql_validator
+from .generator import SQLGenerator, get_sql_generator
+
+__all__ = [
+    "SQLValidator", "SQLValidationError", "get_sql_validator",
+    "SQLGenerator", "get_sql_generator"
+]

sql/generator.py

@@ -0,0 +1,159 @@
+"""
+Text-to-SQL Generator - Multi-Database Support.
+
+Uses LLM to generate SQL queries from natural language,
+with dynamic schema context. Supports MySQL, PostgreSQL, and SQLite.
+"""
+
+import logging
+from typing import Optional, Dict, Any, List, Tuple
+import re
+
+logger = logging.getLogger(__name__)
+
+
+def get_sql_dialect(db_type: str) -> str:
+    """Get the SQL dialect name for the given database type."""
+    dialects = {
+        "mysql": "MySQL",
+        "postgresql": "PostgreSQL",
+        "sqlite": "SQLite"
+    }
+    return dialects.get(db_type, "SQL")
+
+
+def get_dialect_specific_hints(db_type: str) -> str:
+    """Get database-specific hints for SQL generation."""
+    if db_type == "postgresql":
+        return """
+PostgreSQL-SPECIFIC NOTES:
+- Use ILIKE for case-insensitive pattern matching (instead of LIKE)
+- String concatenation uses || operator
+- Use LIMIT at the end of queries
+- Boolean values are TRUE/FALSE (not 1/0)
+- Use double quotes for identifiers with special chars, single quotes for strings
+"""
+    elif db_type == "sqlite":
+        return """
+SQLite-SPECIFIC NOTES:
+- LIKE is case-insensitive for ASCII characters by default
+- Use || for string concatenation
+- No ILIKE - use LIKE (case-insensitive) or GLOB (case-sensitive)
+- Use LIMIT at the end of queries
+- Boolean values are 1/0
+- Uses strftime() for date functions instead of DATE_FORMAT
+"""
+    else:  # MySQL
+        return """
+MySQL-SPECIFIC NOTES:
+- LIKE is case-insensitive for non-binary strings
+- Use CONCAT() for string concatenation
+- Use LIMIT at the end of queries
+- Boolean values are 1/0
+- Use backticks for identifiers with special chars, single quotes for strings
+"""
+
+
+class SQLGenerator:
+    """Generates SQL queries from natural language using LLM."""
+    
+    SYSTEM_PROMPT_TEMPLATE = """You are a SQL expert. Generate {dialect} SELECT queries based on user questions.
+
+RULES:
+1. ONLY generate SELECT statements.
+2. NEVER use INSERT, UPDATE, DELETE, DROP, CREATE, ALTER, or TRUNCATE.
+3. Always include a LIMIT clause (max 50 rows unless specified).
+4. Use table and column names EXACTLY as shown in the schema.
+5. AMBIGUITY: If the user asks for a category, type, or specific value, and you are unsure which column it belongs to:
+   - Check multiple likely columns (e.g., `category`, `sub_category`, `type`, `description`).
+   - Use pattern matching for flexibility.
+   - Use `OR` to combine multiple column checks.
+6. DATA AWARENESS: In footwear databases, specific types like 'Formal', 'Casual', or 'Sports' often appear in `sub_category` OR `category`. Check both if available.
+7. Return ONLY the SQL query, no explanations.
+
+{dialect_hints}
+
+DATABASE SCHEMA:
+{schema}
+
+Generate a single {dialect} SELECT query to answer the user's question."""
+
+    def __init__(self, llm_client=None, db_type: str = "mysql"):
+        self.llm_client = llm_client
+        self.db_type = db_type
+    
+    def set_llm_client(self, llm_client):
+        self.llm_client = llm_client
+    
+    def set_db_type(self, db_type: str):
+        """Set the database type for SQL generation."""
+        self.db_type = db_type
+    
+    def generate(
+        self,
+        question: str,
+        schema_context: str,
+        chat_history: Optional[List[Dict[str, str]]] = None
+    ) -> Tuple[str, str]:
+        """
+        Generate SQL from natural language.
+        
+        Returns:
+            Tuple of (sql_query, explanation)
+        """
+        if not self.llm_client:
+            raise ValueError("LLM client not configured")
+        
+        dialect = get_sql_dialect(self.db_type)
+        dialect_hints = get_dialect_specific_hints(self.db_type)
+        
+        system_prompt = self.SYSTEM_PROMPT_TEMPLATE.format(
+            dialect=dialect,
+            dialect_hints=dialect_hints,
+            schema=schema_context
+        )
+        
+        messages = [{"role": "system", "content": system_prompt}]
+        
+        if chat_history:
+            for msg in chat_history[-3:]:  # Last 3 exchanges for context
+                messages.append(msg)
+        
+        messages.append({"role": "user", "content": question})
+        
+        response = self.llm_client.chat(messages)
+        
+        # Extract SQL from response
+        sql = self._extract_sql(response)
+        
+        return sql, response
+    
+    def _extract_sql(self, response: str) -> str:
+        """Extract SQL query from LLM response."""
+        # Look for SQL in code blocks
+        code_block = re.search(r'```(?:sql)?\\s*(.*?)```', response, re.DOTALL | re.IGNORECASE)
+        if code_block:
+            return code_block.group(1).strip()
+        
+        # Look for SELECT statement
+        select_match = re.search(
+            r'(SELECT\\s+.+?(?:;|$))', 
+            response, 
+            re.DOTALL | re.IGNORECASE
+        )
+        if select_match:
+            return select_match.group(1).strip().rstrip(';')
+        
+        return response.strip()
+
+
+_generator: Optional[SQLGenerator] = None
+
+
+def get_sql_generator(db_type: str = "mysql") -> SQLGenerator:
+    global _generator
+    if _generator is None:
+        _generator = SQLGenerator(db_type=db_type)
+    else:
+        _generator.set_db_type(db_type)
+    return _generator

sql/validator.py

@@ -0,0 +1,163 @@
+"""
+SQL Validator - Security layer for SQL queries.
+
+Ensures ONLY safe SELECT queries are executed.
+Validates against whitelist and blocks dangerous operations.
+"""
+
+import logging
+import re
+from typing import List, Tuple, Optional, Set
+import sqlparse
+from sqlparse.sql import Statement, Token, Identifier, IdentifierList
+from sqlparse.tokens import Keyword, DML
+
+logger = logging.getLogger(__name__)
+
+
+class SQLValidationError(Exception):
+    """Raised when SQL validation fails."""
+    pass
+
+
+class SQLValidator:
+    """Validates SQL queries for safety before execution."""
+    
+    FORBIDDEN_KEYWORDS = {
+        'INSERT', 'UPDATE', 'DELETE', 'DROP', 'CREATE', 'ALTER',
+        'TRUNCATE', 'GRANT', 'REVOKE', 'EXECUTE', 'EXEC',
+        'INTO OUTFILE', 'INTO DUMPFILE', 'LOAD_FILE', 'LOAD DATA'
+    }
+    
+    FORBIDDEN_PATTERNS = [
+        r'INTO\s+OUTFILE',
+        r'INTO\s+DUMPFILE', 
+        r'LOAD_FILE\s*\(',
+        r'LOAD\s+DATA',
+        r';\s*(?:DROP|DELETE|UPDATE|INSERT)',  # Multi-statement attacks
+        r'--',  # SQL comments (potential injection)
+        r'/\*.*\*/',  # Block comments
+    ]
+    
+    def __init__(self, allowed_tables: Optional[Set[str]] = None, max_limit: int = 100):
+        self.allowed_tables = allowed_tables or set()
+        self.max_limit = max_limit
+        self._compiled_patterns = [re.compile(p, re.IGNORECASE) for p in self.FORBIDDEN_PATTERNS]
+    
+    def set_allowed_tables(self, tables: List[str]):
+        """Set the whitelist of allowed tables."""
+        self.allowed_tables = set(tables)
+    
+    def validate(self, sql: str) -> Tuple[bool, str, Optional[str]]:
+        """
+        Validate SQL query for safety.
+        
+        Returns:
+            Tuple of (is_valid, message, sanitized_sql)
+        """
+        if not sql or not sql.strip():
+            return False, "Empty SQL query", None
+        
+        sql = sql.strip()
+        
+        # Check for forbidden patterns
+        for pattern in self._compiled_patterns:
+            if pattern.search(sql):
+                return False, f"Forbidden pattern detected in query", None
+        
+        # Parse SQL
+        try:
+            parsed = sqlparse.parse(sql)
+        except Exception as e:
+            return False, f"Failed to parse SQL: {e}", None
+        
+        if not parsed:
+            return False, "Failed to parse SQL query", None
+        
+        # Only allow single statements
+        if len(parsed) > 1:
+            return False, "Multiple SQL statements not allowed", None
+        
+        statement = parsed[0]
+        
+        # Check statement type
+        stmt_type = statement.get_type()
+        if stmt_type != 'SELECT':
+            return False, f"Only SELECT statements allowed, got: {stmt_type}", None
+        
+        # Check for forbidden keywords in tokens
+        sql_upper = sql.upper()
+        for keyword in self.FORBIDDEN_KEYWORDS:
+            if keyword in sql_upper:
+                return False, f"Forbidden keyword detected: {keyword}", None
+        
+        # Extract and validate tables
+        tables = self._extract_tables(statement)
+        if self.allowed_tables:
+            invalid_tables = tables - self.allowed_tables
+            if invalid_tables:
+                return False, f"Access denied to tables: {invalid_tables}", None
+        
+        # Ensure LIMIT clause exists
+        sanitized = self._ensure_limit(sql)
+        
+        return True, "Query validated successfully", sanitized
+    
+    def _extract_tables(self, statement: Statement) -> Set[str]:
+        """Extract table names from a SELECT statement using regex."""
+        tables = set()
+        sql = str(statement)
+        
+        # Use regex to find tables after FROM and JOIN
+        # Pattern: FROM table_name or JOIN table_name
+        from_pattern = re.compile(
+            r'\bFROM\s+([a-zA-Z_][a-zA-Z0-9_]*)',
+            re.IGNORECASE
+        )
+        join_pattern = re.compile(
+            r'\bJOIN\s+([a-zA-Z_][a-zA-Z0-9_]*)',
+            re.IGNORECASE
+        )
+        
+        # Find all FROM tables
+        for match in from_pattern.finditer(sql):
+            tables.add(match.group(1))
+        
+        # Find all JOIN tables
+        for match in join_pattern.finditer(sql):
+            tables.add(match.group(1))
+        
+        return tables
+    
+    def _ensure_limit(self, sql: str) -> str:
+        """Ensure the query has a LIMIT clause."""
+        sql_upper = sql.upper()
+        
+        if 'LIMIT' in sql_upper:
+            # Check if limit is too high
+            limit_match = re.search(r'LIMIT\s+(\d+)', sql_upper)
+            if limit_match:
+                current_limit = int(limit_match.group(1))
+                if current_limit > self.max_limit:
+                    # Replace with max limit
+                    sql = re.sub(
+                        r'LIMIT\s+\d+',
+                        f'LIMIT {self.max_limit}',
+                        sql,
+                        flags=re.IGNORECASE
+                    )
+            return sql
+        else:
+            # Add LIMIT clause
+            sql = sql.rstrip(';').strip()
+            return f"{sql} LIMIT {self.max_limit}"
+
+
+_validator: Optional[SQLValidator] = None
+
+
+def get_sql_validator() -> SQLValidator:
+    global _validator
+    if _validator is None:
+        _validator = SQLValidator()
+    return _validator