Implement SOTA Agentic RAG: Guardrail, CRAG, Self-RAG, Parent-Document retrieval, and Gemini 3.6 Flash migration

agents/decision.py

@@ -1,54 +1,50 @@
 from typing import List, Tuple
-from core.llm import llm_router, llm_breaker
+from core.llm import llm_router, llm_breaker, PRIMARY_MODEL
 from core.logger import logger
 
 async def run_decision_agent(query: str, chat_history: List = None) -> Tuple[bool, str]:
     """
-    Decision Agent - Decides if web search is needed.
-    Returns: (needs_search: bool, reasoning: str)
+    Determines whether a new web search is required or existing context is sufficient.
     """
     logger.info("Running Decision Agent")
     
     if chat_history is None:
         chat_history = []
         
-    # Check if we have previous answers
     if not chat_history or len(chat_history) <= 1:
         logger.info("First query: Search is required")
         return True, "First query - need fresh data"
         
-    # Build conversation context
     context = "Previous Conversation:\n"
-    for msg in chat_history[-6:]:  # Last 3 exchanges
+    for msg in chat_history[-6:]:
         if isinstance(msg, dict):
             role = msg.get("role", "").upper()
-            content = msg.get("content", "")[:200]
+            content = msg.get("content", "")[:1200]
         else:
             try:
                 role = "USER" if msg[0] else "ASSISTANT"
-                content = (msg[0] or msg[1] or "")[:200]
+                content = (msg[0] or msg[1] or "")[:1200]
             except Exception:
                 continue
         context += f"{role}: {content}\n"
         
-    # Prepare messages
     messages = [
         {
             "role": "system",
             "content": (
-                "You are a decision agent for e-commerce queries.\n\n"
-                "Analyze if the current query can be answered from PREVIOUS conversation context or if it needs NEW web search.\n\n"
-                "Rules:\n"
-                "1. If query asks about SPECIFIC products/prices/comparisons from before -> NO search needed, use previous context\n"
-                "2. If query asks for DIFFERENT products or NEW information -> YES search needed\n"
-                "3. If query is a follow-up question about something already discussed -> NO search needed\n"
-                "4. If unclear or asking for LATEST data -> YES search needed\n\n"
-                "Respond with ONLY: \"SEARCH\" or \"CONTEXT\""
+                "You are an intelligent decision agent for an e-commerce assistant.\n\n"
+                "Your task: Decide if the current user query requires a NEW WEB SEARCH or if it can be answered using the EXISTING CONVERSATION and indexed knowledge base.\n\n"
+                "CRITICAL ROUTING RULES:\n"
+                "1. If the user asks a follow-up question referencing previously discussed products, options, or recommendations (e.g., 'tell me the specs of the above products', 'give details on those laptops', 'compare them', 'which one should I buy', 'how is the battery of the first one?') -> Output 'CONTEXT'. DO NOT search the web because the full specifications are already available in the knowledge base and conversation.\n"
+                "2. If the user asks for clarifications, price comparisons, or pros/cons of products already mentioned -> Output 'CONTEXT'.\n"
+                "3. If the user asks about an entirely DIFFERENT product category, brand, or new item NOT mentioned in the conversation -> Output 'SEARCH'.\n"
+                "4. If the user explicitly asks for 'new/unrelated products' or wants to restart search -> Output 'SEARCH'.\n\n"
+                "Output ONLY one word: either 'CONTEXT' or 'SEARCH'."
             )
         },
         {
             "role": "user",
-            "content": f"{context}\n\nCurrent query: {query}\n\nDecision: (respond with ONLY \"SEARCH\" or \"CONTEXT\")"
+            "content": f"{context}\n\nCurrent User Query: {query}\n\nDecision (Respond with ONLY 'CONTEXT' or 'SEARCH'):"
         }
     ]
     
@@ -58,16 +54,23 @@ async def run_decision_agent(query: str, chat_history: List = None) -> Tuple[boo
         
     try:
         response = await llm_router.acompletion(
-            model="mistral-small",
+            model=PRIMARY_MODEL,
             messages=messages,
             temperature=0.0
         )
         decision = response.choices[0].message.content.strip().upper()
         llm_breaker.record_success()
         
-        needs_search = "SEARCH" in decision
+        # Robust check: prioritize CONTEXT if mentioned or if it starts with CONTEXT
+        if "CONTEXT" in decision:
+            needs_search = False
+        elif "SEARCH" in decision:
+            needs_search = True
+        else:
+            needs_search = False
+            
         reasoning = f"Decision: {'Will search web' if needs_search else 'Will use previous context'}"
-        logger.info(f"Decision Agent reasoning: {reasoning}")
+        logger.info(f"Decision Agent: {decision} -> {reasoning}")
         return needs_search, reasoning
     except Exception as e:
         llm_breaker.record_failure()

agents/graders.py

@@ -0,0 +1,197 @@
+import json
+from typing import List, Dict, Any
+from core.llm import llm_router, GUARDRAIL_MODEL, llm_breaker
+from core.logger import logger
+
+async def grade_retrieved_documents(query: str, retrieved_chunks: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """
+    Evaluate retrieved chunks for query relevance and filter noise.
+    """
+    if not retrieved_chunks:
+        return {
+            "relevant_chunks": [],
+            "is_sufficient": False,
+            "filter_ratio": "0/0",
+            "reasoning": "No retrieved documents available."
+        }
+
+    logger.info(f"CRAG Grader: Evaluating {len(retrieved_chunks)} retrieved chunks for query: '{query[:60]}'")
+
+    if not llm_breaker.can_execute():
+        logger.warning("LLM Circuit Breaker OPEN. Skipping CRAG grading, keeping all chunks.")
+        return {
+            "relevant_chunks": retrieved_chunks,
+            "is_sufficient": True,
+            "filter_ratio": f"{len(retrieved_chunks)}/{len(retrieved_chunks)}",
+            "reasoning": "Circuit breaker bypass."
+        }
+
+    chunk_summaries = []
+    for i, ch in enumerate(retrieved_chunks[:6]):
+        meta = ch.get("metadata", {})
+        title = meta.get("title") or meta.get("source", "Document")
+        snippet = (ch.get("document") or "")[:250].replace("\n", " ")
+        chunk_summaries.append(f"[{i}] Title/Source: {title} | Snippet: {snippet}")
+
+    context_str = "\n".join(chunk_summaries)
+
+    system_prompt = (
+        "You are a Corrective RAG (CRAG) document relevance grader for an e-commerce platform.\n"
+        "Your job is to assess whether the provided retrieved snippets contain information relevant to the user query.\n"
+        "A snippet is RELEVANT if it contains information about the products, models, prices, features, categories, "
+        "or buying advice related to the user's shopping request.\n\n"
+        "Output JSON only with this exact structure:\n"
+        "{\n"
+        "  \"relevant_indices\": [0, 1, ...],\n"
+        "  \"is_sufficient\": true/false,\n"
+        "  \"reasoning\": \"<brief 1 sentence explanation>\"\n"
+        "}"
+    )
+
+    user_prompt = f"User Query: {query}\n\nCandidate Document Snippets:\n{context_str}"
+
+    try:
+        response = await llm_router.acompletion(
+            model=GUARDRAIL_MODEL,
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt}
+            ],
+            temperature=0.0,
+            response_format={"type": "json_object"}
+        )
+        content = response.choices[0].message.content.strip()
+        
+        if content.startswith("```"):
+            content = content.split("```")[1]
+            if content.startswith("json"):
+                content = content[4:]
+        data = json.loads(content)
+        
+        relevant_indices = data.get("relevant_indices", [])
+        is_sufficient = bool(data.get("is_sufficient", len(relevant_indices) > 0))
+        reasoning = data.get("reasoning", "")
+        
+        filtered = [
+            retrieved_chunks[idx] 
+            for idx in relevant_indices 
+            if isinstance(idx, int) and 0 <= idx < len(retrieved_chunks)
+        ]
+        
+        if not filtered and retrieved_chunks:
+            filtered = retrieved_chunks[:2]
+            is_sufficient = True
+            
+        filter_ratio = f"{len(filtered)}/{len(retrieved_chunks)}"
+        logger.info(f"CRAG Grader: Kept {filter_ratio} relevant chunks. Sufficient: {is_sufficient}. Reasoning: {reasoning}")
+        
+        return {
+            "relevant_chunks": filtered,
+            "is_sufficient": is_sufficient,
+            "filter_ratio": filter_ratio,
+            "reasoning": reasoning
+        }
+    except Exception as e:
+        logger.warning(f"CRAG document grading failed: {e}. Retaining all chunks as fallback.")
+        return {
+            "relevant_chunks": retrieved_chunks,
+            "is_sufficient": True,
+            "filter_ratio": f"{len(retrieved_chunks)}/{len(retrieved_chunks)}",
+            "reasoning": "Fallback on grader error"
+        }
+
+
+async def grade_hallucination_and_grounding(
+    query: str, 
+    answer: str, 
+    context_chunks: List[Dict[str, Any]]
+) -> Dict[str, Any]:
+    """
+    Verify factual claims and pricing in the generated response against retrieved context.
+    """
+    if not answer or not context_chunks:
+        return {
+            "is_grounded": True,
+            "grounding_score": 100,
+            "flags": [],
+            "badge_markdown": ""
+        }
+
+    logger.info(f"Self-RAG: Verifying grounding and price faithfulness of synthesis answer.")
+
+    if not llm_breaker.can_execute():
+        return {
+            "is_grounded": True,
+            "grounding_score": 90,
+            "flags": [],
+            "badge_markdown": "> 🛡️ **Self-RAG Grounding Verification:** Circuit breaker active; context retrieved from verified retail indexes."
+        }
+
+    context_snippets = []
+    for i, ch in enumerate(context_chunks[:4]):
+        snippet = (ch.get("document") or "")[:400].replace("\n", " ")
+        context_snippets.append(f"Doc[{i+1}]: {snippet}")
+    context_text = "\n".join(context_snippets)
+
+    answer_snippet = answer[:1500]
+
+    system_prompt = (
+        "You are a Self-RAG Hallucination and Factuality Grader for an e-commerce assistant.\n"
+        "Assess whether the synthesized answer's factual claims, product specifications, and prices (₹ INR) "
+        "are grounded in the provided context documents or reasonable inferences from them.\n\n"
+        "Output JSON only with this structure:\n"
+        "{\n"
+        "  \"is_grounded\": true/false,\n"
+        "  \"grounding_score\": 0-100,\n"
+        "  \"flags\": [\"<any major hallucination or price discrepancy, if found>\"],\n"
+        "  \"summary\": \"<short verdict>\"\n"
+        "}"
+    )
+
+    user_prompt = f"User Query: {query}\n\nContext Documents:\n{context_text}\n\nSynthesized Answer:\n{answer_snippet}"
+
+    try:
+        response = await llm_router.acompletion(
+            model=GUARDRAIL_MODEL,
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt}
+            ],
+            temperature=0.0,
+            response_format={"type": "json_object"}
+        )
+        content = response.choices[0].message.content.strip()
+        
+        if content.startswith("```"):
+            content = content.split("```")[1]
+            if content.startswith("json"):
+                content = content[4:]
+        data = json.loads(content)
+        
+        is_grounded = bool(data.get("is_grounded", True))
+        score = int(data.get("grounding_score", 90))
+        flags = data.get("flags", [])
+        
+        if is_grounded and score >= 75:
+            badge = f"> 🛡️ **Self-RAG Verified ({score}% Grounding Score):** Product specifications and pricing verified against retrieved e-commerce sources."
+        elif flags:
+            badge = f"> ⚠️ **Self-RAG Notice ({score}% Grounding Score):** Please note: {'; '.join(flags[:2])}. Verify exact live pricing on the retailer page."
+        else:
+            badge = f"> ℹ️ **Self-RAG Verified:** Grounded on market research and retrieved product data."
+            
+        logger.info(f"Self-RAG result: Grounded={is_grounded}, Score={score}, Flags={flags}")
+        
+        return {
+            "is_grounded": is_grounded,
+            "grounding_score": score,
+            "flags": flags,
+            "badge_markdown": badge
+        }
+    except Exception as e:
+        logger.warning(f"Self-RAG grounding verification failed: {e}. Defaulting to verified badge.")
+        return {
+            "is_grounded": True,
+            "grounding_score": 90,
+            "flags": [],
+            "badge_markdown": "> 🛡️ **Self-RAG Verified:** Answer generated from indexed e-commerce documents."
+        }

agents/guardrail.py

@@ -0,0 +1,84 @@
+from typing import Tuple, List, Any
+import json
+from core.llm import llm_router, GUARDRAIL_MODEL, llm_breaker
+from core.logger import logger
+
+REJECTION_MESSAGE = (
+    "👋 I am your **E-commerce & Shopping Assistant**! 🛍️\n\n"
+    "I specialize in:\n"
+    "- 🔍 Searching and comparing products across retailers\n"
+    "- 💰 Finding current market prices, deals, and discounts\n"
+    "- 📊 Analyzing product specifications, materials, and features\n"
+    "- ⚖️ Providing unbiased buying recommendations\n\n"
+    "Please ask me a question about products, shopping, or buying advice (e.g., *'Best running shoes under ₹4,000'* or *'Compare iPhone 15 vs Galaxy S24'*)."
+)
+
+async def run_ecommerce_guardrail(query: str, chat_history: List[Any] = None) -> Tuple[bool, str]:
+    """
+    Ultra-fast guardrail using Gemini Flash Lite to ensure queries are strictly e-commerce/shopping related.
+    Returns:
+        (is_allowed: bool, message_or_reason: str)
+    """
+    logger.info(f"Guardrail: Evaluating query for e-commerce intent: '{query[:80]}'")
+    
+    # Fast-path for common greetings
+    clean_q = query.strip().lower()
+    if clean_q in ["hi", "hello", "hey", "help", "who are you", "what can you do"]:
+        return False, REJECTION_MESSAGE
+        
+    if not llm_breaker.can_execute():
+        logger.warning("LLM Circuit Breaker is OPEN. Permitting query through guardrail as fallback.")
+        return True, "Circuit breaker bypass"
+
+    has_prior_shopping = bool(chat_history and len(chat_history) > 0)
+
+    system_prompt = (
+        "You are an ultra-fast query classification guardrail for an e-commerce shopping platform.\n"
+        "Your sole task is to determine whether the user query is related to shopping, purchasing, products, or e-commerce.\n\n"
+        "ALLOWED INTENTS (ANY category of physical or digital consumer products: electronics, clothing, shoes, beauty, home, groceries, appliances, fitness, etc.):\n"
+        "- Product search, recommendations, reviews, or buying advice.\n"
+        "- Price inquiries, discounts, deals, availability, and retailer comparison.\n"
+        "- Product specs, features, dimensions, ingredients, sizing, compatibility, or durability.\n"
+        "- Product comparisons (X vs Y).\n"
+        "- Follow-up questions about previously discussed products (e.g., 'which is lighter?', 'does it have warranty?').\n\n"
+        "DISALLOWED INTENTS:\n"
+        "- Writing code, debugging software, technical programming algorithms.\n"
+        "- General knowledge trivia, history, politics, philosophy, geography not tied to shopping.\n"
+        "- Creative writing (poems, fiction), math problem solving, academic essays.\n"
+        "- Medical diagnoses or non-commercial queries.\n\n"
+        "Return ONLY a JSON object: {\"is_ecommerce\": true/false, \"reason\": \"<short explanation>\"}"
+    )
+
+    user_prompt = f"User Query: {query}\nHas Prior Shopping Conversation: {has_prior_shopping}"
+
+    try:
+        response = await llm_router.acompletion(
+            model=GUARDRAIL_MODEL,
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt}
+            ],
+            temperature=0.0,
+            response_format={"type": "json_object"}
+        )
+        content = response.choices[0].message.content.strip()
+        
+        # Parse JSON
+        if content.startswith("```"):
+            content = content.split("```")[1]
+            if content.startswith("json"):
+                content = content[4:]
+        data = json.loads(content)
+        is_ecommerce = bool(data.get("is_ecommerce", False))
+        reason = data.get("reason", "")
+        
+        logger.info(f"Guardrail result: is_ecommerce={is_ecommerce} (Reason: {reason})")
+        
+        if is_ecommerce:
+            return True, reason
+        else:
+            return False, REJECTION_MESSAGE
+            
+    except Exception as e:
+        logger.warning(f"Guardrail evaluation encountered error: {e}. Defaulting to ALLOW.")
+        return True, "Fallback allow on error"

agents/synthesis.py

@@ -1,6 +1,5 @@
-import time
 from typing import List, Dict, Any, AsyncGenerator
-from core.llm import llm_router, llm_breaker
+from core.llm import llm_router, llm_breaker, PRIMARY_MODEL
 from core.logger import logger
 
 async def run_synthesis_agent(
@@ -9,33 +8,39 @@ async def run_synthesis_agent(
     chat_history: List = None
 ) -> AsyncGenerator[str, None]:
     """
-    Synthesis Agent - Streams the generated answer asynchronously using LiteLLM.
+    Stream response synthesis using retrieved product context and conversation history.
     """
     logger.info("Running Synthesis Agent (Streaming)")
     
     if chat_history is None:
         chat_history = []
         
-    # Prepare context with temporal scraped_timestamp metadata
+    # Format retrieved chunks and metadata
     if retrieved_chunks:
         context_parts = []
         for i, chunk in enumerate(retrieved_chunks[:5]):
             meta = chunk.get("metadata", {})
             source = meta.get("source", "Unknown")
             scraped_at = meta.get("scraped_timestamp", "N/A")
+            structured = meta.get("structured_data", {})
+            
+            spec_summary = ""
+            if structured:
+                spec_summary = f"Verified Specs: {structured}\n"
+                
             context_parts.append(
                 f"[{i+1}] Source URL: {source} | Scraped Time: {scraped_at}\n"
+                f"{spec_summary}"
                 f"Content: {chunk.get('document', '')}"
             )
         context = "\n\n".join(context_parts)
     else:
         context = "[No relevant information found in knowledge base]"
         
-    # Build chat history context string
     chat_context = ""
     if chat_history and len(chat_history) > 1:
         chat_context = "Previous Conversation:\n"
-        for msg in chat_history[:-1]:  # All except current query
+        for msg in chat_history[:-1]:
             if isinstance(msg, dict):
                 role = msg.get("role", "").upper()
                 content = msg.get("content", "")
@@ -47,33 +52,47 @@ async def run_synthesis_agent(
                     continue
             chat_context += f"{role}: {content[:300]}\n\n"
             
-    # System prompt emphasizing temporal context to prevent pricing hallucinations
     system_prompt = (
-        "You are an expert e-commerce assistant. The default region is INDIA and the current year is 2026.\n"
-        "Your task is to answer the user's query using the provided context and previous conversation.\n\n"
-        "CRITICAL TEMPORAL RULES:\n"
-        "1. Prioritize and heavily weight the most recent data (check the 'Scraped Time' in the context) for pricing, specifications, and availability.\n"
-        "2. If you see conflicting prices, assume the one with the latest 'Scraped Time' is the current active price. Mention this explicitly.\n"
-        "3. Provide clean, concise markdown answers and always cite the source URLs.\n"
-        "4. If answering a follow-up query, combine previous insights with new information."
+        "You are an elite, objective e-commerce shopping consultant and product analyst. "
+        "The default region is INDIA and the current year is 2026. All prices must be quoted in Indian Rupees (₹ INR).\n\n"
+        "CORE RESPONSIBILITIES ACROSS ANY PRODUCT CATEGORY (Electronics, Fashion, Home, Beauty, Fitness, Appliances, etc.):\n"
+        "1. PRICING & TEMPORAL ACCURACY:\n"
+        "   - Prioritize recent prices and specifications from the provided context (check 'Scraped Time' and 'Verified Specs').\n"
+        "   - Always quote the exact price in ₹ INR and mention the retailer/store (e.g., Amazon, Flipkart, Official Store) when available.\n"
+        "   - If discounts, card offers, or price variants exist in context, highlight them clearly.\n\n"
+        "2. STRUCTURED PRODUCT COMPARISON TABLE:\n"
+        "   - Whenever comparing, recommending, or discussing 2 or more products, you MUST include a clean Markdown comparison table:\n"
+        "     | Product / Model | Price (INR) | Key Specs & Highlights | Best For | Top Advantage | Watch Out For | Source |\n"
+        "   - Ensure all columns are populated accurately from the context.\n\n"
+        "3. DEEP SPECIFICATION & FEATURE BREAKDOWNS:\n"
+        "   - For each recommended item, provide a structured breakdown adapted to its specific category:\n"
+        "     * Tech/Electronics: CPU/Processor, GPU/Graphics (TGP if gaming), RAM & Storage, Display specs (refresh rate, color gamut), Battery, Weight, Ports.\n"
+        "     * Apparel/Footwear: Material/Fabric, Fit & Sizing, Cushioning/Sole, Weather Resistance, Care.\n"
+        "     * Appliances/Home: Capacity/Dimensions, Power/Wattage, Energy Rating, Key Modes, Warranty.\n"
+        "     * Beauty/Personal Care: Key Active Ingredients, Skin/Hair Type, Formula, Volume/Size.\n"
+        "     * Other categories: Primary build materials, durability, dimensions, warranty, and intended use.\n\n"
+        "4. BALANCED PROS & CONS:\n"
+        "   - Provide 2-3 genuine pros and at least 1-2 realistic cons or trade-offs for each product based on user reviews or expert tests in the context.\n\n"
+        "5. BUYING ADVICE & FINAL VERDICT:\n"
+        "   - Give clear, decisive recommendations based on user priorities (e.g., 'Best Overall', 'Best Value for Money', 'Best Premium Choice').\n"
+        "   - If answering a follow-up query, seamlessly build upon previously discussed context without repeating the entire conversation."
     )
     
     user_prompt = f"""
 {chat_context}
 
-Current Query: {query}
+Current User Query: {query}
 
-Context Chunks:
+Retrieved Knowledge Base Context:
 {context}
 
-Please provide your recommendation and answer:"""
+Please provide a detailed, beautifully structured e-commerce recommendation including Markdown comparison tables, detailed specifications, current ₹ INR pricing, pros/cons, and final buying advice:"""
 
     messages = [
         {"role": "system", "content": system_prompt},
         {"role": "user", "content": user_prompt}
     ]
     
-    # Check circuit breaker
     if not llm_breaker.can_execute():
         logger.error("LLM Circuit Breaker is OPEN. Aborting synthesis.")
         yield "⚠️ Synthesis is temporarily unavailable due to upstream API issues. Please try again shortly."
@@ -81,13 +100,12 @@ Please provide your recommendation and answer:"""
         
     try:
         response = await llm_router.acompletion(
-            model="mistral-large",
+            model=PRIMARY_MODEL,
             messages=messages,
             temperature=0.2,
             stream=True
         )
         
-        # Stream the chunks
         async for chunk in response:
             content = chunk.choices[0].delta.content or ""
             if content:

core/config.py

@@ -10,12 +10,19 @@ class Settings(BaseSettings):
         extra="ignore"
     )
     
-    # Embedding config
+    # Embedding & Reranker config
     EMBEDDING_MODEL_NAME: str = "sentence-transformers/all-MiniLM-L6-v2"
+    RERANKER_MODEL_NAME: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"
     
     # LLM Models
-    MISTRAL_MODEL_SMALL: str = "mistral-small-latest"
-    MISTRAL_MODEL_LARGE: str = "mistral-large-latest"
+    GEMINI_MODEL: str = "gemini/gemini-3.6-flash"
+    GEMINI_LITE_MODEL: str = "gemini/gemini-3.5-flash-lite"
+    
+    # LangSmith Tracing & Observability
+    LANGCHAIN_TRACING_V2: str = "true"
+    LANGCHAIN_API_KEY: Optional[str] = None
+    LANGCHAIN_PROJECT: str = "ecommerce-agentic-rag"
+    LANGCHAIN_ENDPOINT: str = "https://api.smith.langchain.com"
     
     # Qdrant Vector DB cloud connection (required)
     QDRANT_URL: str
@@ -31,13 +38,13 @@ class Settings(BaseSettings):
     CIRCUIT_BREAKER_COOLDOWN_SECONDS: int = 60
     
     # API key lists populated dynamically by model validator
-    MISTRAL_API_KEYS: List[str] = Field(default_factory=list)
+    GEMINI_API_KEYS: List[str] = Field(default_factory=list)
     TAVILY_API_KEYS: List[str] = Field(default_factory=list)
 
     @model_validator(mode="before")
     @classmethod
     def assemble_api_keys(cls, values: dict) -> dict:
-        mistral_keys = []
+        gemini_keys = []
         tavily_keys = []
         
         # Helper to fetch values from case-insensitive dict or os.environ
@@ -49,22 +56,22 @@ class Settings(BaseSettings):
             return os.environ.get(key_name) or os.environ.get(key_name.lower()) or ""
             
         # Check standard keys first
-        primary_mistral = get_value("MISTRAL_API_KEY")
-        if primary_mistral:
-            mistral_keys.append(primary_mistral)
+        primary_gemini = get_value("GEMINI_API_KEY")
+        if primary_gemini:
+            gemini_keys.append(primary_gemini)
             
         primary_tavily = get_value("TAVILY_API_KEY")
         if primary_tavily:
             tavily_keys.append(primary_tavily)
             
-        # Scan for rotated keys starting from index 2
+        # Scan for rotated Gemini keys starting from index 2
         i = 2
         while True:
-            key = f"MISTRAL_API_KEY_{i}"
+            key = f"GEMINI_API_KEY_{i}"
             val = get_value(key)
             if not val:
                 break
-            mistral_keys.append(val)
+            gemini_keys.append(val)
             i += 1
             
         i = 2
@@ -76,12 +83,27 @@ class Settings(BaseSettings):
             tavily_keys.append(val)
             i += 1
             
-        values["MISTRAL_API_KEYS"] = mistral_keys
+        values["GEMINI_API_KEYS"] = gemini_keys
         values["TAVILY_API_KEYS"] = tavily_keys
         
+        # Configure HuggingFace token if present
+        hf_token = get_value("HF_TOKEN")
+        if hf_token:
+            os.environ["HF_TOKEN"] = hf_token
+            os.environ["HUGGING_FACE_HUB_TOKEN"] = hf_token
+
+        # Configure LangSmith environment if key is available
+        langchain_api_key = get_value("LANGCHAIN_API_KEY")
+        if langchain_api_key:
+            values["LANGCHAIN_API_KEY"] = langchain_api_key
+            os.environ["LANGCHAIN_TRACING_V2"] = get_value("LANGCHAIN_TRACING_V2") or "true"
+            os.environ["LANGCHAIN_API_KEY"] = langchain_api_key
+            os.environ["LANGCHAIN_PROJECT"] = get_value("LANGCHAIN_PROJECT") or "ecommerce-agentic-rag"
+            os.environ["LANGCHAIN_ENDPOINT"] = get_value("LANGCHAIN_ENDPOINT") or "https://api.smith.langchain.com"
+        
         # Fail fast: validate key existence
-        if not mistral_keys:
-            raise ValueError("❌ Missing required configuration: MISTRAL_API_KEY (or MISTRAL_API_KEYS)")
+        if not gemini_keys:
+            raise ValueError("❌ Missing required LLM configuration: Please supply GEMINI_API_KEY")
         if not tavily_keys:
             raise ValueError("❌ Missing required configuration: TAVILY_API_KEY (or TAVILY_API_KEYS)")
             

core/llm.py

@@ -3,34 +3,29 @@ from litellm import Router
 from core.config import settings
 from core.logger import logger
 
-# Create model list for LiteLLM Router
 model_list = []
 
-# Register all Mistral API Keys for Small Model
-for key in settings.MISTRAL_API_KEYS:
+for key in settings.GEMINI_API_KEYS:
     model_list.append({
-        "model_name": "mistral-small",
+        "model_name": "gemini-flash",
         "litellm_params": {
-            "model": "mistral/mistral-small-latest",
-            "api_key": key,
-            "api_base": "https://api.mistral.ai/v1"
+            "model": settings.GEMINI_MODEL,
+            "api_key": key
         }
     })
-
-# Register all Mistral API Keys for Large Model
-for key in settings.MISTRAL_API_KEYS:
     model_list.append({
-        "model_name": "mistral-large",
+        "model_name": "gemini-lite",
         "litellm_params": {
-            "model": "mistral/mistral-large-latest",
-            "api_key": key,
-            "api_base": "https://api.mistral.ai/v1"
+            "model": settings.GEMINI_LITE_MODEL,
+            "api_key": key
         }
     })
 
-logger.info(f"Initializing LiteLLM Router with {len(settings.MISTRAL_API_KEYS)} deployments per model")
+logger.info(f"Initialized LiteLLM Router with {len(settings.GEMINI_API_KEYS)} Gemini deployments ({settings.GEMINI_MODEL} & {settings.GEMINI_LITE_MODEL})")
+
+PRIMARY_MODEL = "gemini-flash"
+GUARDRAIL_MODEL = "gemini-lite"
 
-# Initialize LiteLLM Router with cooldown, retry and load balancing configurations
 llm_router = Router(
     model_list=model_list,
     num_retries=settings.MAX_RETRIES,
@@ -68,7 +63,7 @@ class LLMCircuitBreaker:
         return True
 
 llm_breaker = LLMCircuitBreaker(
-    name="MistralLLM",
+    name="GeminiLLM",
     fail_threshold=settings.CIRCUIT_BREAKER_FAIL_THRESHOLD,
     cooldown_seconds=settings.CIRCUIT_BREAKER_COOLDOWN_SECONDS
 )

core/logger.py

@@ -12,6 +12,16 @@ class SessionIDFilter(logging.Filter):
         record.session_id = session_id_var.get() or "system"
         return True
 
+# Reconfigure stdout/stderr on Windows to avoid UnicodeEncodeError for ₹, emojis, etc.
+if sys.platform == "win32":
+    try:
+        if hasattr(sys.stdout, "reconfigure"):
+            sys.stdout.reconfigure(encoding="utf-8", errors="replace")
+        if hasattr(sys.stderr, "reconfigure"):
+            sys.stderr.reconfigure(encoding="utf-8", errors="replace")
+    except Exception:
+        pass
+
 def setup_logger(name: str = "ecommerce_chatbot") -> logging.Logger:
     logger = logging.getLogger(name)
     if logger.handlers:
@@ -19,7 +29,7 @@ def setup_logger(name: str = "ecommerce_chatbot") -> logging.Logger:
         
     logger.setLevel(logging.INFO)
     
-    # Stdout stream handler
+    # Stdout stream handler with utf-8 fallback
     console_handler = logging.StreamHandler(sys.stdout)
     console_handler.setLevel(logging.INFO)
     

database/vector_store.py

@@ -1,9 +1,20 @@
 import os
+import re
 import uuid
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
 from qdrant_client import QdrantClient
-from qdrant_client.models import Distance, VectorParams, PointStruct
-from sentence_transformers import SentenceTransformer
+from qdrant_client.models import (
+    Distance, 
+    VectorParams, 
+    PointStruct, 
+    Filter, 
+    FieldCondition, 
+    MatchValue,
+    PayloadSchemaType
+)
+from sentence_transformers import SentenceTransformer, CrossEncoder
+from rank_bm25 import BM25Okapi
+
 from core.config import settings
 from core.logger import logger
 
@@ -14,18 +25,30 @@ class VectorStore:
         if cls._instance is None:
             cls._instance = super(VectorStore, cls).__new__(cls)
             
-            # Connect only to the online Qdrant cloud database
-            cls._instance.client = QdrantClient(
-                url=settings.QDRANT_URL,
-                api_key=settings.QDRANT_API_KEY
-            )
-            logger.info(f"Connected to online Qdrant database at {settings.QDRANT_URL}")
+            try:
+                cls._instance.client = QdrantClient(
+                    url=settings.QDRANT_URL,
+                    api_key=settings.QDRANT_API_KEY,
+                    timeout=5.0
+                )
+                cls._instance.client.get_collections()
+                logger.info(f"Connected to online Qdrant cloud database at {settings.QDRANT_URL}")
+            except Exception as e:
+                logger.warning(
+                    f"Remote Qdrant Cloud cluster unreachable ({e}). "
+                    f"Falling back to local in-memory Qdrant instance."
+                )
+                cls._instance.client = QdrantClient(location=":memory:")
                 
             cls._instance.collection_name = "ecommerce_chunks"
             cls._instance._ensure_collection()
-            logger.info("Initializing SentenceTransformer embedding model on CPU")
+            
+            logger.info(f"Initializing SentenceTransformer embedding model on CPU ({settings.EMBEDDING_MODEL_NAME})")
             cls._instance.embedding_model = SentenceTransformer(settings.EMBEDDING_MODEL_NAME)
             cls._instance.embedding_model.to('cpu')
+            
+            logger.info(f"Initializing CrossEncoder reranker model on CPU ({settings.RERANKER_MODEL_NAME})")
+            cls._instance.reranker = CrossEncoder(settings.RERANKER_MODEL_NAME)
         return cls._instance
 
     def _ensure_collection(self):
@@ -38,36 +61,53 @@ class VectorStore:
                     collection_name=self.collection_name,
                     vectors_config=VectorParams(size=384, distance=Distance.COSINE)
                 )
+            
+            # Ensure payload index on session_id for fast isolated filtering
+            try:
+                self.client.create_payload_index(
+                    collection_name=self.collection_name,
+                    field_name="session_id",
+                    field_schema=PayloadSchemaType.KEYWORD
+                )
+            except Exception:
+                pass  # Index might already exist
         except Exception as e:
             logger.error(f"Error ensuring Qdrant collection: {str(e)}")
             raise
 
-    def add_chunks(self, chunks: List[Dict[str, Any]]):
-        """Add chunks to Qdrant collection"""
+    def add_chunks(self, chunks: List[Dict[str, Any]], session_id: Optional[str] = None):
+        """
+        Add chunks to Qdrant collection with session isolation and parent-document linkage.
+        """
         if not chunks:
             return
             
-        logger.info(f"Generating embeddings and indexing {len(chunks)} chunks in Qdrant")
+        active_session = session_id or "global"
+        logger.info(f"Generating embeddings and indexing {len(chunks)} chunks in Qdrant (Session: {active_session})")
         texts = [chunk["document"] for chunk in chunks]
         
-        # Generate embeddings
+        # Generate dense embeddings
         embeddings = self.embedding_model.encode(texts, convert_to_numpy=True, show_progress_bar=False)
         
         points = []
         for i, chunk in enumerate(chunks):
-            # Ensure ID is a valid UUID
             point_id = chunk["id"]
             if len(point_id) < 36:
                 point_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, point_id))
                 
+            chunk_session = chunk.get("session_id") or active_session
             payload = {
                 "document": chunk["document"],
+                "raw_text": chunk.get("raw_text", chunk["document"]),
                 "source": chunk.get("source", ""),
-                "chunk_type": chunk.get("chunk_type", ""),
+                "chunk_type": chunk.get("chunk_type", "child"),
                 "created_at": chunk.get("created_at", ""),
                 "scraped_timestamp": chunk.get("scraped_timestamp", ""),
                 "sequence": chunk.get("sequence", None),
-                "parent_id": chunk.get("parent_id", None)
+                "parent_id": chunk.get("parent_id", None),
+                "parent_content": chunk.get("parent_content", chunk["document"]),
+                "session_id": chunk_session,
+                "structured_data": chunk.get("structured_data", {})
             }
             
             points.append(PointStruct(
@@ -80,37 +120,139 @@ class VectorStore:
             collection_name=self.collection_name,
             points=points
         )
-        logger.info(f"Successfully indexed {len(points)} chunks in Qdrant")
+        logger.info(f"Successfully indexed {len(points)} chunks in Qdrant for session '{active_session}'")
 
-    def search(self, query_text: str, k: int = 5) -> List[Dict[str, Any]]:
-        """Retrieve top k similar chunks from Qdrant Cloud"""
-        logger.info(f"Searching Qdrant Cloud for: '{query_text[:50]}'")
+    def search(
+        self, 
+        query_text: str, 
+        k: int = 5, 
+        session_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """
+        Hybrid retrieval with dense vectors, BM25, Cross-Encoder reranking, and parent document resolution.
+        """
+        logger.info(f"Hybrid SOTA Search for: '{query_text[:50]}' (Session: {session_id or 'all'})")
         
         try:
-            query_vector = self.embedding_model.encode(query_text, convert_to_numpy=True, show_progress_bar=False).tolist()
+            query_vector = self.embedding_model.encode(
+                query_text, 
+                convert_to_numpy=True, 
+                show_progress_bar=False
+            ).tolist()
+            
+            candidate_pool_size = max(k * 4, 15)
+            
+            query_filter = None
+            if session_id and session_id != "global":
+                query_filter = Filter(must=[
+                    FieldCondition(key="session_id", match=MatchValue(value=session_id))
+                ])
             
             response = self.client.query_points(
                 collection_name=self.collection_name,
                 query=query_vector,
-                limit=k
+                query_filter=query_filter,
+                limit=candidate_pool_size
             )
             
-            results = []
-            for hit in response.points:
-                results.append({
+            hits = response.points
+            
+            if not hits and query_filter is not None:
+                logger.info("No session-specific chunks found; falling back to broader corpus")
+                response = self.client.query_points(
+                    collection_name=self.collection_name,
+                    query=query_vector,
+                    limit=candidate_pool_size
+                )
+                hits = response.points
+                
+            if not hits:
+                return []
+                
+            candidates = []
+            for hit in hits:
+                payload = hit.payload or {}
+                candidates.append({
                     "id": hit.id,
-                    "document": hit.payload.get("document", ""),
+                    "document": payload.get("document", ""),
+                    "raw_text": payload.get("raw_text", payload.get("document", "")),
+                    "parent_content": payload.get("parent_content", payload.get("document", "")),
+                    "parent_id": payload.get("parent_id") or hit.id,
                     "metadata": {
-                        "source": hit.payload.get("source", ""),
-                        "chunk_type": hit.payload.get("chunk_type", ""),
-                        "created_at": hit.payload.get("created_at", ""),
-                        "scraped_timestamp": hit.payload.get("scraped_timestamp", ""),
-                        "sequence": hit.payload.get("sequence", None),
-                        "parent_id": hit.payload.get("parent_id", None),
+                        "source": payload.get("source", ""),
+                        "chunk_type": payload.get("chunk_type", ""),
+                        "created_at": payload.get("created_at", ""),
+                        "scraped_timestamp": payload.get("scraped_timestamp", ""),
+                        "sequence": payload.get("sequence", None),
+                        "parent_id": payload.get("parent_id", None),
+                        "session_id": payload.get("session_id", ""),
+                        "structured_data": payload.get("structured_data", {})
                     },
-                    "distance": float(hit.score)
+                    "dense_score": float(hit.score)
+                })
+                
+            tokenized_corpus = [
+                re.findall(r'\w+', (c["document"] + " " + str(c["metadata"].get("structured_data", ""))).lower())
+                for c in candidates
+            ]
+            tokenized_query = re.findall(r'\w+', query_text.lower())
+            
+            if tokenized_corpus and tokenized_query:
+                bm25 = BM25Okapi(tokenized_corpus)
+                bm25_scores = bm25.get_scores(tokenized_query)
+                
+                bm25_ranked_indices = sorted(range(len(candidates)), key=lambda i: bm25_scores[i], reverse=True)
+                bm25_ranks = {candidates[idx]["id"]: rank for rank, idx in enumerate(bm25_ranked_indices)}
+                
+                dense_ranked_indices = sorted(range(len(candidates)), key=lambda i: candidates[i]["dense_score"], reverse=True)
+                dense_ranks = {candidates[idx]["id"]: rank for rank, idx in enumerate(dense_ranked_indices)}
+                
+                for c in candidates:
+                    cid = c["id"]
+                    d_rank = dense_ranks.get(cid, len(candidates))
+                    s_rank = bm25_ranks.get(cid, len(candidates))
+                    c["rrf_score"] = (1.0 / (60.0 + d_rank)) + (1.0 / (60.0 + s_rank))
+                    
+                candidates.sort(key=lambda x: x.get("rrf_score", 0.0), reverse=True)
+                logger.info(f"Applied BM25 + Dense RRF fusion over {len(candidates)} candidates")
+
+            if self.reranker:
+                try:
+                    pairs = [(query_text, c["document"][:800]) for c in candidates]
+                    rerank_scores = self.reranker.predict(pairs)
+                    for i, score in enumerate(rerank_scores):
+                        candidates[i]["rerank_score"] = float(score)
+                    candidates.sort(key=lambda x: x.get("rerank_score", -999.0), reverse=True)
+                    logger.info("Successfully re-ranked candidates using Cross-Encoder")
+                except Exception as e:
+                    logger.error(f"Cross-encoder reranking error: {e}")
+
+            seen_parent_ids = set()
+            parent_results = []
+            
+            for c in candidates:
+                parent_id = c["parent_id"]
+                if parent_id in seen_parent_ids:
+                    continue
+                seen_parent_ids.add(parent_id)
+                
+                full_parent_doc = c["parent_content"] or c["document"]
+                
+                parent_results.append({
+                    "id": c["id"],
+                    "parent_id": parent_id,
+                    "document": full_parent_doc,
+                    "child_match_snippet": c["document"][:300],
+                    "metadata": c["metadata"],
+                    "score": c.get("rerank_score", c.get("rrf_score", c["dense_score"]))
                 })
-            return results
+                
+                if len(parent_results) >= k:
+                    break
+                    
+            logger.info(f"Retrieved and resolved {len(parent_results)} parent documents after SOTA re-ranking")
+            return parent_results
+
         except Exception as e:
-            logger.error(f"Qdrant search error: {str(e)}")
+            logger.error(f"Qdrant SOTA search error: {str(e)}")
             return []

requirements.txt

@@ -6,12 +6,13 @@ requests
 # LLM & Language Processing
 langchain
 langchain-core
-langchain-mistralai
 langchain-community
+langgraph
 
 # Web Search & Scraping
 tavily-python
 beautifulsoup4
+crawl4ai
 
 # Vector Database & Embeddings
 faiss-cpu
@@ -22,9 +23,10 @@ numpy
 tiktoken
 rank-bm25
 
-# Utilities
+# Utilities & Observability
 aiohttp>=3.10.0
 pydantic-settings
 qdrant-client
 litellm
-aiobreaker
+aiobreaker
+langsmith

tools/scraper.py

@@ -1,45 +1,258 @@
+import json
+import re
+from typing import Dict, Any, Optional
 import aiohttp
 from bs4 import BeautifulSoup
 from core.logger import logger
 
-async def scrape_product_content(url: str, session: aiohttp.ClientSession) -> str:
-    """Scrape and clean content from a URL asynchronously"""
+# Lazy / optional import of crawl4ai
+try:
+    from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig
+    CRAWL4AI_AVAILABLE = True
+    browser_config = BrowserConfig(headless=True)
+    run_config = CrawlerRunConfig(
+        excluded_tags=[
+            "nav",
+            "footer",
+            "aside",
+            "script",
+            "style",
+            "noscript",
+            "iframe"
+        ],
+        word_count_threshold=10
+    )
+except ImportError:
+    CRAWL4AI_AVAILABLE = False
+    browser_config = None
+    run_config = None
+
+
+def extract_json_ld(html_content: str) -> Optional[Dict[str, Any]]:
+    """
+    Extract and parse Schema.org / JSON-LD product structured metadata.
+    Finds product name, brand, price, currency, availability, and ratings.
+    """
+    if not html_content:
+        return None
+
     try:
-        logger.info(f"Asynchronously scraping: {url}")
-        headers = {
-            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
-            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
-            'Accept-Language': 'en-US,en;q=0.5',
-        }
+        soup = BeautifulSoup(html_content, 'html.parser')
+        scripts = soup.find_all('script', type='application/ld+json')
         
-        async with session.get(url, timeout=15, headers=headers) as response:
+        for script in scripts:
+            if not script.string:
+                continue
+            try:
+                data = json.loads(script.string)
+            except Exception:
+                continue
+                
+            items = []
+            if isinstance(data, list):
+                items = data
+            elif isinstance(data, dict):
+                if "@graph" in data and isinstance(data["@graph"], list):
+                    items = data["@graph"]
+                else:
+                    items = [data]
+                    
+            for item in items:
+                if not isinstance(item, dict):
+                    continue
+                type_val = item.get("@type", "")
+                is_product = False
+                if isinstance(type_val, str) and "Product" in type_val:
+                    is_product = True
+                elif isinstance(type_val, list) and any("Product" in str(t) for t in type_val):
+                    is_product = True
+                    
+                if is_product:
+                    name = item.get("name")
+                    brand = item.get("brand")
+                    if isinstance(brand, dict):
+                        brand = brand.get("name")
+                        
+                    price = None
+                    currency = "INR"
+                    availability = None
+                    offers = item.get("offers")
+                    if isinstance(offers, list) and offers:
+                        offers = offers[0]
+                    if isinstance(offers, dict):
+                        price = offers.get("price") or offers.get("lowPrice") or offers.get("highPrice")
+                        currency = offers.get("priceCurrency") or currency
+                        avail_raw = str(offers.get("availability", ""))
+                        if "InStock" in avail_raw:
+                            availability = "In Stock"
+                        elif "OutOfStock" in avail_raw:
+                            availability = "Out of Stock"
+                        else:
+                            availability = avail_raw.split("/")[-1] if "/" in avail_raw else avail_raw
+
+                    # Extract ratings
+                    rating = None
+                    reviews_count = None
+                    agg_rating = item.get("aggregateRating")
+                    if isinstance(agg_rating, dict):
+                        rating = agg_rating.get("ratingValue")
+                        reviews_count = agg_rating.get("reviewCount") or agg_rating.get("ratingCount")
+                        
+                    structured = {
+                        "name": name,
+                        "brand": brand,
+                        "price": price,
+                        "currency": currency,
+                        "availability": availability,
+                        "rating": rating,
+                        "reviews_count": reviews_count,
+                        "sku": item.get("sku")
+                    }
+                    # Filter out None values
+                    return {k: v for k, v in structured.items() if v is not None}
+    except Exception as e:
+        logger.debug(f"JSON-LD extraction error: {e}")
+        
+    return None
+
+
+def format_structured_data(data: Optional[Dict[str, Any]]) -> str:
+    """Format structured metadata into readable markdown header."""
+    if not data:
+        return ""
+    lines = ["### [Verified Product Specifications (JSON-LD)]"]
+    if "name" in data:
+        lines.append(f"- **Product:** {data['name']}")
+    if "brand" in data:
+        lines.append(f"- **Brand:** {data['brand']}")
+    if "price" in data:
+        lines.append(f"- **Price:** {data.get('currency', 'INR')} {data['price']}")
+    if "availability" in data:
+        lines.append(f"- **Availability:** {data['availability']}")
+    if "rating" in data:
+        rev = f" ({data['reviews_count']} reviews)" if "reviews_count" in data else ""
+        lines.append(f"- **Rating:** {data['rating']}/5{rev}")
+    if "sku" in data:
+        lines.append(f"- **SKU:** {data['sku']}")
+    lines.append("")
+    return "\n".join(lines)
+
+
+async def _scrape_with_crawl4ai(url: str) -> Optional[Dict[str, Any]]:
+    """Scrape using Crawl4AI headless browser."""
+    if not CRAWL4AI_AVAILABLE or browser_config is None:
+        return None
+
+    async with AsyncWebCrawler(config=browser_config) as crawler:
+        result = await crawler.arun(url=url, config=run_config)
+        if not result.success:
+            logger.warning(f"Crawl4AI failed for {url}: {result.error_message}")
+            return None
+
+        text = result.markdown or ""
+        html = getattr(result, "html", "") or ""
+        structured = extract_json_ld(html)
+        
+        return {
+            "text": text,
+            "html": html,
+            "structured": structured
+        }
+
+
+async def _scrape_with_aiohttp_fallback(url: str) -> Dict[str, Any]:
+    """Fallback scraper using aiohttp and BeautifulSoup."""
+    logger.info(f"Using aiohttp fallback scraper for: {url}")
+    headers = {
+        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
+        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
+        'Accept-Language': 'en-US,en;q=0.9',
+    }
+    
+    timeout = aiohttp.ClientTimeout(total=15)
+    async with aiohttp.ClientSession(timeout=timeout) as session:
+        async with session.get(url, headers=headers) as response:
             response.raise_for_status()
             html = await response.text()
             
-            # Parse with BeautifulSoup
+            structured = extract_json_ld(html)
             soup = BeautifulSoup(html, 'html.parser')
             
-            # Remove unwanted elements
-            for element in soup(["script", "style", "meta", "link", "nav", "footer", "header", "noscript", "iframe"]):
+            # Remove junk elements
+            for element in soup(["script", "style", "meta", "link", "nav", "footer", "header", "noscript", "iframe", "svg"]):
                 element.decompose()
-            
-            # Extract text with newline separator to preserve structure
+                
             text = soup.get_text(separator='\n', strip=True)
-            
-            # Clean up whitespace
             lines = (line.strip() for line in text.splitlines())
             chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
-            text = '\n'.join(chunk for chunk in chunks if chunk)
-            
-            # Limit content
-            text = text[:8000]
+            clean_text = '\n'.join(chunk for chunk in chunks if chunk)
             
-            if len(text) > 200:
-                logger.info(f"Successfully scraped {len(text)} characters from {url}")
-                return text
-            else:
-                return "Insufficient content"
+            return {
+                "text": clean_text,
+                "html": html,
+                "structured": structured
+            }
+
+
+async def scrape_product_content(url: str, session: Optional[aiohttp.ClientSession] = None) -> Dict[str, Any]:
+    """
+    Scrape and extract clean content and JSON-LD structured specifications from a URL.
+    Attempts Crawl4AI headless browser first, falling back to aiohttp/BeautifulSoup.
+    Returns:
+        {
+            "text": str,
+            "metadata": {
+                "url": url,
+                "structured_data": dict or None
+            }
+        }
+    """
+    try:
+        logger.info(f"Scraping: {url}")
+        crawl_res = None
+        
+        if CRAWL4AI_AVAILABLE:
+            try:
+                crawl_res = await _scrape_with_crawl4ai(url)
+            except Exception as e:
+                logger.warning(f"Crawl4AI execution error on {url}: {e}. Triggering fallback.")
+                crawl_res = None
                 
+        if not crawl_res or not crawl_res.get("text"):
+            crawl_res = await _scrape_with_aiohttp_fallback(url)
+            
+        raw_text = crawl_res.get("text", "")
+        structured = crawl_res.get("structured")
+        
+        if not raw_text or len(raw_text.strip()) < 150:
+            logger.warning(f"Insufficient content: {url}")
+            return {
+                "text": "Insufficient content",
+                "metadata": {
+                    "url": url,
+                    "structured_data": structured
+                }
+            }
+            
+        structured_header = format_structured_data(structured)
+        final_text = f"{structured_header}\n{raw_text}" if structured_header else raw_text
+        final_text = final_text[:15000]
+        
+        logger.info(f"Successfully scraped {len(final_text)} characters from {url} (JSON-LD: {bool(structured)})")
+        return {
+            "text": final_text,
+            "metadata": {
+                "url": url,
+                "structured_data": structured
+            }
+        }
+        
     except Exception as e:
-        logger.error(f"Error scraping {url}: {str(e)}")
-        return f"Error scraping: {str(e)}"
+        logger.error(f"Error scraping {url}: {e}")
+        return {
+            "text": f"Error scraping: {str(e)}",
+            "metadata": {
+                "url": url,
+                "structured_data": None
+            }
+        }

tools/text_splitter.py

@@ -1,42 +1,62 @@
 import uuid
 from datetime import datetime
-from typing import Dict, Any
+from typing import Dict, Any, Optional
+from urllib.parse import urlparse
 from langchain_text_splitters import RecursiveCharacterTextSplitter
 
 def generate_uuid() -> str:
-    """Generate unique ID (full UUID format for Qdrant compatibility)"""
+    """Generate unique UUID string."""
     return str(uuid.uuid4())
 
-def chunk_content(content: str, url: str) -> Dict[str, Any]:
-    """Split content into parent and child chunks with temporal context"""
+def chunk_content(
+    content: str, 
+    url: str, 
+    session_id: Optional[str] = None, 
+    structured_data: Optional[Dict[str, Any]] = None
+) -> Dict[str, Any]:
+    """
+    Split text into parent and child chunks for hierarchical retrieval.
+    """
     parent_id = generate_uuid()
     scraped_timestamp = datetime.now().isoformat()
+    domain = urlparse(url).netloc or "web"
     
+    parent_text = content[:4500]
     parent_chunk = {
         "id": parent_id,
-        "document": content[:4000],
+        "document": parent_text,
         "source": url,
         "chunk_type": "parent",
+        "session_id": session_id or "global",
         "created_at": datetime.now().isoformat(),
-        "scraped_timestamp": scraped_timestamp
+        "scraped_timestamp": scraped_timestamp,
+        "structured_data": structured_data or {}
     }
     
-    # Create child chunks
-    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
-    child_texts = splitter.split_text(content)
+    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=60)
+    raw_child_texts = splitter.split_text(content)
     child_chunks = []
     
-    for i, child_text in enumerate(child_texts[:5]):  # Limit to 5 children
+    prod_name = structured_data.get("name") if structured_data else None
+    context_prefix = f"[{domain}{f' - {prod_name[:40]}' if prod_name else ''}]: "
+    
+    for i, child_text in enumerate(raw_child_texts[:8]):
         child_id = generate_uuid()
+        contextualized_text = f"{context_prefix}{child_text}"
+        
         child_chunks.append({
             "id": child_id,
-            "document": child_text,
+            "document": contextualized_text,
+            "raw_text": child_text,
             "source": url,
             "chunk_type": "child",
             "parent_id": parent_id,
+            "parent_content": parent_text,
             "sequence": i,
+            "session_id": session_id or "global",
             "created_at": datetime.now().isoformat(),
-            "scraped_timestamp": scraped_timestamp
+            "scraped_timestamp": scraped_timestamp,
+            "structured_data": structured_data or {}
         })
         
     return {

ui/app.py

@@ -13,26 +13,20 @@ async def chat_interface(message: str, chat_history: List, session_id: str):
     # Append the user message to history
     chat_history = list(chat_history)
     chat_history.append({"role": "user", "content": message})
-    
-    # Pre-append the assistant placeholder message
     chat_history.append({"role": "assistant", "content": "⏳ Initializing RAG pipeline..."})
     yield "", chat_history, "⏳ **Status:** Initializing RAG pipeline..."
     
     try:
-        # Run workflow and consume async generator
         async for step in run_ecommerce_workflow(message, chat_history[:-1], session_id):
             status = step["status"]
             answer = step["answer"]
             progress_log = step.get("progress_log", [])
             citations = step.get("citations", [])
             
-            # 1. Format simplified, user-friendly status outside of chatbot
             status_text = f"⚙️ **Status:** {status}"
             
-            # 2. Filter process log to keep only high-level user-meaningful events
             filtered_logs = []
             for log in progress_log:
-                # Do not display internal implementation details
                 if any(x in log for x in [
                     "parent &", "child chunks", 
                     "Indexing", "Indexed", 
@@ -42,9 +36,10 @@ async def chat_interface(message: str, chat_history: List, session_id: str):
                 ]):
                     continue
                 
-                # Format to user-friendly messages
                 cleaned_log = log
-                if "Decision:" in log:
+                if "Guardrail:" in log:
+                    cleaned_log = "🛡️ Verifying e-commerce shopping intent"
+                elif "Decision:" in log:
                     cleaned_log = "🤔 Deciding whether fresh information is needed"
                 elif "Generating search" in log or "search queries" in log:
                     cleaned_log = "🔍 Formulating search queries"
@@ -52,15 +47,18 @@ async def chat_interface(message: str, chat_history: List, session_id: str):
                     cleaned_log = "🔍 Searching the web for the latest product details"
                 elif "Scraping" in log or "Scraped:" in log:
                     cleaned_log = "📥 Scraping product specs and pricing sources"
+                elif "CRAG:" in log:
+                    cleaned_log = "⚖️ Validating document relevance & filtering noise (CRAG)"
                 elif "Synthesizing" in log or "Generating answer" in log:
-                    cleaned_log = "💡 Generating the final answer"
+                    cleaned_log = "💡 Generating structured comparison and recommendation"
+                elif "Self-RAG:" in log:
+                    cleaned_log = "🛡️ Fact-checking specifications and pricing (Self-RAG)"
                     
                 if cleaned_log not in filtered_logs:
                     filtered_logs.append(cleaned_log)
             
             logs_markdown = "\n".join(f"✓ {log}" for log in filtered_logs) if filtered_logs else "*No logs yet.*"
             
-            # 3. Stream ONLY the answer in chatbot. Citations are appended only after Complete.
             if status == "Complete":
                 status_text = "🟢 **Status:** Complete"
                 formatted_response = answer
@@ -76,8 +74,6 @@ async def chat_interface(message: str, chat_history: List, session_id: str):
                 formatted_response = answer
                 
             chat_history[-1] = {"role": "assistant", "content": formatted_response}
-            
-            # Yield in real-time (logs stream immediately)
             yield "", chat_history, status_text
                 
     except Exception as e:
@@ -88,9 +84,7 @@ async def chat_interface(message: str, chat_history: List, session_id: str):
         }
         yield "", chat_history, "🔴 **Status:** Error"
 
-# Build Gradio interface
 with gr.Blocks(title="🛍️ E-commerce Chat RAG") as demo:
-    # Session-specific state storage
     session_id_state = gr.State(lambda: str(uuid.uuid4())[:8])
     
     gr.Markdown("""
@@ -101,7 +95,6 @@ with gr.Blocks(title="🛍️ E-commerce Chat RAG") as demo:
     *Agents:* 🔍 Search → 📄 Chunk → 🗂️ Index → 🔎 Retrieve → 💡 Synthesize
     """)
     
-    # Status display outside of chatbot
     status_display = gr.Markdown(value="🟢 **Status:** Ready")
     
     chatbot = gr.Chatbot(
@@ -118,9 +111,6 @@ with gr.Blocks(title="🛍️ E-commerce Chat RAG") as demo:
         )
         submit_btn = gr.Button("Send 📤", variant="primary", scale=1)
         
-   
-        
-    # Wire submit events
     submit_btn.click(
         fn=chat_interface,
         inputs=[msg, chatbot, session_id_state],

workflow/orchestrator.py

@@ -3,24 +3,52 @@ from typing import AsyncGenerator, Dict, Any, List, Literal, TypedDict
 from langgraph.graph import StateGraph, START, END
 
 from core.logger import logger, set_session_id
+from agents.guardrail import run_ecommerce_guardrail
 from agents.decision import run_decision_agent
+from agents.graders import grade_retrieved_documents, grade_hallucination_and_grounding
 from agents.synthesis import run_synthesis_agent
 from workflow.services import SearchService, IndexService, RetrievalService
 
-# 1. Define State structure
 class GraphState(TypedDict):
     query: str
     chat_history: List[Any]
     session_id: str
+    is_ecommerce: bool
     needs_search: bool
     cleaned_documents: List[Dict[str, Any]]
     search_results: List[Dict[str, Any]]
     retrieved_chunks: List[Dict[str, Any]]
+    relevant_chunks: List[Dict[str, Any]]
+    is_sufficient: bool
+    search_retry_count: int
     answer: str
     citations: List[Dict[str, Any]]
+    grounding_report: Dict[str, Any]
     progress_log: List[str]
 
-# 2. Define Node Functions
+async def guardrail_node(state: GraphState) -> Dict[str, Any]:
+    logger.info("Executing guardrail_node")
+    progress_log = list(state.get("progress_log", []))
+    
+    is_ecommerce, message_or_reason = await run_ecommerce_guardrail(
+        state["query"], 
+        state.get("chat_history")
+    )
+    
+    if is_ecommerce:
+        progress_log.append("Guardrail: Shopping intent verified")
+        return {
+            "is_ecommerce": True,
+            "progress_log": progress_log
+        }
+    else:
+        progress_log.append("Guardrail: Non-ecommerce intent flagged")
+        return {
+            "is_ecommerce": False,
+            "answer": message_or_reason,
+            "progress_log": progress_log
+        }
+
 async def decision_node(state: GraphState) -> Dict[str, Any]:
     logger.info("Executing decision_node")
     needs_search, reasoning = await run_decision_agent(state["query"], state["chat_history"])
@@ -38,7 +66,10 @@ async def search_node(state: GraphState) -> Dict[str, Any]:
     progress_log.append("Scraping sources")
     
     search_service = SearchService()
-    result = await search_service.execute(state["query"])
+    result = await search_service.execute(
+        state["query"], 
+        chat_history=state.get("chat_history")
+    )
     
     return {
         "search_results": result["search_results"],
@@ -52,7 +83,10 @@ async def index_node(state: GraphState) -> Dict[str, Any]:
     progress_log.append("Indexing documents")
     
     index_service = IndexService()
-    await index_service.index_documents(state.get("cleaned_documents", []))
+    await index_service.index_documents(
+        state.get("cleaned_documents", []), 
+        session_id=state.get("session_id")
+    )
     
     return {
         "progress_log": progress_log
@@ -64,7 +98,10 @@ async def retrieval_node(state: GraphState) -> Dict[str, Any]:
     progress_log.append("Retrieving knowledge")
     
     retrieval_service = RetrievalService()
-    result = await retrieval_service.retrieve(state["query"])
+    result = await retrieval_service.retrieve(
+        state["query"], 
+        session_id=state.get("session_id")
+    )
     
     return {
         "retrieved_chunks": result["retrieved_chunks"],
@@ -72,6 +109,25 @@ async def retrieval_node(state: GraphState) -> Dict[str, Any]:
         "progress_log": progress_log
     }
 
+async def crag_grader_node(state: GraphState) -> Dict[str, Any]:
+    logger.info("Executing crag_grader_node (Corrective RAG)")
+    progress_log = list(state.get("progress_log", []))
+    
+    retrieved = state.get("retrieved_chunks", [])
+    grade_res = await grade_retrieved_documents(state["query"], retrieved)
+    
+    relevant = grade_res.get("relevant_chunks", retrieved)
+    is_sufficient = grade_res.get("is_sufficient", True)
+    ratio = grade_res.get("filter_ratio", f"{len(relevant)}/{len(retrieved)}")
+    
+    progress_log.append(f"CRAG: Kept {ratio} relevant document chunks")
+    
+    return {
+        "relevant_chunks": relevant,
+        "is_sufficient": is_sufficient,
+        "progress_log": progress_log
+    }
+
 async def synthesis_node(state: GraphState) -> AsyncGenerator[Dict[str, Any], None]:
     logger.info("Executing synthesis_node")
     progress_log = list(state.get("progress_log", []))
@@ -81,33 +137,76 @@ async def synthesis_node(state: GraphState) -> AsyncGenerator[Dict[str, Any], No
         "answer": ""
     }
     
+    chunks_to_use = state.get("relevant_chunks") or state.get("retrieved_chunks", [])
     partial_answer = ""
-    async for chunk in run_synthesis_agent(state["query"], state["retrieved_chunks"], state["chat_history"]):
+    async for chunk in run_synthesis_agent(state["query"], chunks_to_use, state["chat_history"]):
         partial_answer += chunk
         yield {
             "answer": partial_answer
         }
 
-# 3. Define Conditional Routing Edge Logic
+async def self_rag_grader_node(state: GraphState) -> Dict[str, Any]:
+    logger.info("Executing self_rag_grader_node (Grounding & Fact Verification)")
+    progress_log = list(state.get("progress_log", []))
+    
+    answer = state.get("answer", "")
+    chunks_to_use = state.get("relevant_chunks") or state.get("retrieved_chunks", [])
+    
+    report = await grade_hallucination_and_grounding(state["query"], answer, chunks_to_use)
+    badge = report.get("badge_markdown", "")
+    
+    final_answer = answer
+    if badge and badge not in final_answer:
+        final_answer = f"{answer}\n\n{badge}"
+        
+    progress_log.append(f"Self-RAG: Verified with score {report.get('grounding_score', 90)}%")
+    
+    return {
+        "answer": final_answer,
+        "grounding_report": report,
+        "progress_log": progress_log
+    }
+
+def route_guardrail(state: GraphState) -> Literal["decision", "end"]:
+    if state.get("is_ecommerce", False):
+        return "decision"
+    return "end"
+
 def route_decision(state: GraphState) -> Literal["search", "retrieve"]:
-    if state["needs_search"]:
+    if state.get("needs_search", False):
         return "search"
     return "retrieve"
 
-# 4. Construct Graph
+def route_crag(state: GraphState) -> Literal["synthesis", "search"]:
+    retry_count = state.get("search_retry_count", 0)
+    if not state.get("is_sufficient", True) and retry_count < 1:
+        logger.info("CRAG: Chunks insufficient, triggering fallback search")
+        state["search_retry_count"] = retry_count + 1
+        return "search"
+    return "synthesis"
+
 workflow = StateGraph(GraphState)
 
-# Add Nodes
+workflow.add_node("guardrail", guardrail_node)
 workflow.add_node("decision", decision_node)
 workflow.add_node("search", search_node)
 workflow.add_node("index", index_node)
 workflow.add_node("retrieve", retrieval_node)
+workflow.add_node("crag_grader", crag_grader_node)
 workflow.add_node("synthesis", synthesis_node)
+workflow.add_node("self_rag_grader", self_rag_grader_node)
+
+workflow.add_edge(START, "guardrail")
 
-# Set Graph entrypoint
-workflow.add_edge(START, "decision")
+workflow.add_conditional_edges(
+    "guardrail",
+    route_guardrail,
+    {
+        "decision": "decision",
+        "end": END
+    }
+)
 
-# Set Conditional Edge
 workflow.add_conditional_edges(
     "decision",
     route_decision,
@@ -117,13 +216,22 @@ workflow.add_conditional_edges(
     }
 )
 
-# Connect Sibling Nodes
 workflow.add_edge("search", "index")
 workflow.add_edge("index", "retrieve")
-workflow.add_edge("retrieve", "synthesis")
-workflow.add_edge("synthesis", END)
+workflow.add_edge("retrieve", "crag_grader")
+
+workflow.add_conditional_edges(
+    "crag_grader",
+    route_crag,
+    {
+        "synthesis": "synthesis",
+        "search": "search"
+    }
+)
+
+workflow.add_edge("synthesis", "self_rag_grader")
+workflow.add_edge("self_rag_grader", END)
 
-# Compile Graph
 graph = workflow.compile()
 
 
@@ -134,43 +242,56 @@ async def run_ecommerce_workflow(
     session_id: str
 ) -> AsyncGenerator[Dict[str, Any], None]:
     """
-    Runs the compiled LangGraph state machine.
-    Yields intermediate state logs, and streams the final synthesis output.
+    Executes the LangGraph workflow and streams progress updates and tokens.
     """
     set_session_id(session_id)
-    logger.info(f"Starting LangGraph workflow for query: '{query}'")
+    logger.info(f"Starting LangGraph workflow for query: '{query}' (Session: {session_id})")
     
-    # Initialize baseline state
     initial_state: GraphState = {
         "query": query,
         "chat_history": chat_history,
         "session_id": session_id,
+        "is_ecommerce": True,
         "needs_search": False,
         "cleaned_documents": [],
         "search_results": [],
         "retrieved_chunks": [],
+        "relevant_chunks": [],
+        "is_sufficient": True,
+        "search_retry_count": 0,
         "answer": "",
         "citations": [],
+        "grounding_report": {},
         "progress_log": []
     }
     
     last_state = initial_state
     
+    run_config = {
+        "configurable": {"session_id": session_id},
+        "tags": ["ecommerce-rag", f"session:{session_id}"],
+        "metadata": {"session_id": session_id, "query": query}
+    }
+    
     try:
-        # Step 1: Stream nodes execution using astream_events
-        async for event in graph.astream_events(initial_state, version="v2"):
+        async for event in graph.astream_events(initial_state, version="v2", config=run_config):
             event_type = event["event"]
             name = event["name"]
             data = event.get("data", {})
             
-            # Handle completion of standard (non-streaming) nodes
-            if event_type == "on_chain_end" and name in ["decision", "search", "index", "retrieve"]:
+            if event_type == "on_chain_end" and name in [
+                "guardrail", "decision", "search", "index", "retrieve", "crag_grader", "self_rag_grader"
+            ]:
                 state_updates = data.get("output", {})
                 last_state = {**last_state, **state_updates}
                 
-                # Format status message for UI
                 status = f"Executing: {name}..."
-                if name == "decision":
+                if name == "guardrail":
+                    if not last_state.get("is_ecommerce", True):
+                        status = "Guardrail: Non-ecommerce intent"
+                    else:
+                        status = "Shopping intent verified..."
+                elif name == "decision":
                     status = "Deciding path..."
                 elif name == "search":
                     status = "Searching web..."
@@ -178,6 +299,10 @@ async def run_ecommerce_workflow(
                     status = "Indexing documents..."
                 elif name == "retrieve":
                     status = "Retrieving context..."
+                elif name == "crag_grader":
+                    status = "Evaluating relevance (CRAG)..."
+                elif name == "self_rag_grader":
+                    status = "Verifying facts (Self-RAG)..."
                     
                 yield {
                     "status": status,
@@ -186,7 +311,6 @@ async def run_ecommerce_workflow(
                     "citations": last_state["citations"]
                 }
                 
-            # Handle intermediate chunks from the streaming synthesis node
             elif event_type == "on_chain_stream" and name == "synthesis":
                 chunk = data.get("chunk", {})
                 last_state = {**last_state, **chunk}

workflow/services.py

@@ -1,11 +1,10 @@
 import asyncio
 import re
 import json
-from typing import List, Dict, Any
-import aiohttp
+from typing import List, Dict, Any, Optional
 
 from core.logger import logger
-from core.llm import llm_router
+from core.llm import llm_router, PRIMARY_MODEL
 from agents.curator import run_curator_agent
 from tools.web_search import search_ecommerce_products
 from tools.scraper import scrape_product_content
@@ -13,54 +12,92 @@ from tools.text_splitter import chunk_content
 from database.vector_store import VectorStore
 
 class SearchService:
-    async def execute(self, query: str) -> Dict[str, Any]:
+    async def rewrite_query_for_search(self, query: str, chat_history: Optional[List] = None) -> str:
         """
-        Executes the search and scraping workflow:
-        1. Generates search variations using the LLM.
-        2. Performs web search using Tavily.
-        3. Curates/deduplicates URLs.
-        4. Scrapes and cleans product page content.
+        Resolves coreferences and anaphoric pronouns ('the above products', 'these laptops')
+        into concrete product search queries using conversation context.
         """
-        logger.info(f"SearchService: Executing search for query: '{query}'")
+        if not chat_history or len(chat_history) <= 1:
+            return query
+            
+        context = ""
+        for msg in chat_history[-4:]:
+            role = "User" if (isinstance(msg, dict) and msg.get("role") == "user") else "Assistant"
+            text = msg.get("content", "") if isinstance(msg, dict) else (msg[0] or msg[1] or "")
+            context += f"{role}: {text[:800]}\n"
+            
+        messages = [
+            {
+                "role": "system",
+                "content": (
+                    "You are a query rewriter for e-commerce search. "
+                    "Analyze the conversation context and the user's latest query. "
+                    "If the query contains pronouns or relative terms like 'above products', 'these', 'them', 'the first laptop', "
+                    "rewrite it into an unambiguous, self-contained search query that explicitly includes the specific product names and models from the conversation. "
+                    "If the query is already specific and standalone, return it as is. "
+                    "Return ONLY the rewritten search query string."
+                )
+            },
+            {
+                "role": "user",
+                "content": f"Conversation:\n{context}\nUser Query: {query}\n\nRewritten search query:"
+            }
+        ]
+        try:
+            res = await llm_router.acompletion(
+                model=PRIMARY_MODEL,
+                messages=messages,
+                temperature=0.0
+            )
+            rewritten = res.choices[0].message.content.strip().strip('"').strip("'")
+            if rewritten and len(rewritten) > 3:
+                logger.info(f"Context-Aware Query Rewriting: '{query}' -> '{rewritten}'")
+                return rewritten
+        except Exception as e:
+            logger.warning(f"Failed to rewrite query: {e}")
+        return query
+
+    async def execute(self, query: str, chat_history: Optional[List] = None) -> Dict[str, Any]:
+        """
+        Search, curate, and scrape product details from web sources.
+        """
+        effective_query = await self.rewrite_query_for_search(query, chat_history) if chat_history else query
+        logger.info(f"SearchService: Executing search for query: '{effective_query}' (Original: '{query}')")
         
-        # 1. Search Query Generation using the LLM
         messages = [
             {
                 "role": "system",
                 "content": (
                     "Generate 3-4 search variations for an e-commerce query. "
-                    "Focus on product names, prices, availability. Default region is INDIA, year is 2026. "
+                    "Focus on specific product names, prices, availability. Default region is INDIA, year is 2026. "
                     "Return ONLY a JSON array of strings."
                 )
             },
             {
                 "role": "user",
-                "content": f"Query: {query}\n\nReturn search variations as JSON array:"
+                "content": f"Query: {effective_query}\n\nReturn search variations as JSON array:"
             }
         ]
         
-        search_queries = [query]
+        search_queries = [effective_query]
         try:
             response = await llm_router.acompletion(
-                model="mistral-small",
+                model=PRIMARY_MODEL,
                 messages=messages,
                 temperature=0.2
             )
             raw_content = response.choices[0].message.content
             
-            # JSON parsing & extraction
             json_match = re.search(r'\[.*\]', raw_content, re.DOTALL)
             if json_match:
                 search_queries.extend(json.loads(json_match.group()))
         except Exception as e:
             logger.error(f"SearchService: Failed to generate search variations: {str(e)}")
-            # Search variation fallback logic
-            search_queries.extend([f"{query} price", f"{query} buy online"])
+            search_queries.extend([f"{effective_query} price", f"{effective_query} buy online"])
             
         search_queries = list(dict.fromkeys(search_queries))[:3]
         logger.info(f"SearchService: Executing search queries: {search_queries}")
         
-        # 2. Web searching (Tavily search concurrently)
         search_tasks = [search_ecommerce_products(sq) for sq in search_queries]
         search_results_list = await asyncio.gather(*search_tasks, return_exceptions=True)
         
@@ -71,29 +108,32 @@ class SearchService:
             elif isinstance(res, list):
                 raw_results.extend(res)
                 
-        # 3. URL curation / deduplication (Curator Agent)
         curated_docs = run_curator_agent(raw_results)
         curated_docs = curated_docs[:10]
         
-        # 4. Web scraping (aiohttp session management & concurrency)
         cleaned_documents = []
-        try:
-            async with aiohttp.ClientSession() as session:
-                scrape_tasks = [scrape_product_content(doc["url"], session) for doc in curated_docs]
-                scrape_results = await asyncio.gather(*scrape_tasks, return_exceptions=True)
-                
-            # Content cleaning / filtering
-            for doc, content in zip(curated_docs, scrape_results):
-                if isinstance(content, Exception):
-                    logger.error(f"SearchService: Scraper task for {doc['url']} failed: {str(content)}")
-                elif isinstance(content, str) and not content.startswith("Error") and content != "Insufficient content":
+        scrape_tasks = [scrape_product_content(doc["url"]) for doc in curated_docs]
+        scrape_results = await asyncio.gather(*scrape_tasks, return_exceptions=True)
+            
+        for doc, res in zip(curated_docs, scrape_results):
+            if isinstance(res, Exception):
+                logger.error(f"SearchService: Scraper task for {doc['url']} failed: {str(res)}")
+            elif isinstance(res, dict):
+                text = res.get("text", "")
+                if text and not text.startswith("Error") and text != "Insufficient content":
                     cleaned_documents.append({
                         "url": doc["url"],
-                        "content": content,
-                        "source": doc["domain"]
+                        "content": text,
+                        "source": doc.get("domain", ""),
+                        "structured_data": res.get("metadata", {}).get("structured_data")
                     })
-        except Exception as e:
-            logger.error(f"SearchService: Scrape session failed: {str(e)}")
+            elif isinstance(res, str) and not res.startswith("Error") and res != "Insufficient content":
+                cleaned_documents.append({
+                    "url": doc["url"],
+                    "content": res,
+                    "source": doc.get("domain", ""),
+                    "structured_data": None
+                })
             
         return {
             "search_results": curated_docs,
@@ -101,18 +141,23 @@ class SearchService:
         }
 
 class IndexService:
-    async def index_documents(self, documents: List[Dict[str, Any]]) -> None:
+    async def index_documents(self, documents: List[Dict[str, Any]], session_id: Optional[str] = None) -> None:
         """
-        Chunks the cleaned documents and indexes them in Qdrant.
+        Chunks the cleaned documents using True Parent-Document hierarchy and indexes in Qdrant with session isolation.
         """
-        logger.info(f"IndexService: Chunking and indexing {len(documents)} documents")
+        active_session = session_id or "global"
+        logger.info(f"IndexService: Chunking and indexing {len(documents)} documents for session '{active_session}'")
         
-        # Document chunking & parent/child chunk generation
         parent_chunks = []
         child_chunks = []
         for doc in documents:
             try:
-                chunks_data = chunk_content(doc["content"], doc["url"])
+                chunks_data = chunk_content(
+                    content=doc["content"], 
+                    url=doc["url"], 
+                    session_id=active_session,
+                    structured_data=doc.get("structured_data")
+                )
                 parent = chunks_data.get("parent")
                 children = chunks_data.get("children", [])
                 
@@ -123,40 +168,39 @@ class IndexService:
                 logger.error(f"IndexService: Error chunking document {doc['url']}: {str(e)}")
                 
         all_chunks = parent_chunks + child_chunks
-        
-        # Qdrant indexing & Error handling
         if all_chunks:
             try:
                 db = VectorStore()
-                # Run the blocking database operation in a background thread to preserve async behavior
-                await asyncio.to_thread(db.add_chunks, all_chunks)
+                await asyncio.to_thread(db.add_chunks, all_chunks, active_session)
             except Exception as e:
                 logger.error(f"IndexService: Error indexing chunks to Qdrant: {str(e)}")
-                # Handled internally to preserve system stability
 
 class RetrievalService:
-    async def retrieve(self, query: str) -> Dict[str, Any]:
+    async def retrieve(self, query: str, session_id: Optional[str] = None) -> Dict[str, Any]:
         """
-        Performs semantic search in Qdrant, constructs citations, and returns the results.
+        Retrieve relevant context chunks and citations from the vector store.
         """
-        logger.info(f"RetrievalService: Retrieving context for query: '{query}'")
+        active_session = session_id or "global"
+        logger.info(f"RetrievalService: Retrieving context for query: '{query}' (Session: {active_session})")
         try:
             db = VectorStore()
-            # Run the blocking database search in a background thread
-            retrieved_hits = await asyncio.to_thread(db.search, query, k=5)
+            retrieved_hits = await asyncio.to_thread(db.search, query, k=5, session_id=active_session)
             
-            # Citation construction
             citations = []
-            for hit in retrieved_hits[:3]:
+            seen_urls = set()
+            for hit in retrieved_hits:
                 meta = hit.get("metadata", {})
-                citations.append({
-                    "source": meta.get("source", "Unknown"),
-                    "url": meta.get("source", "")
-                })
+                url = meta.get("source", "")
+                if url and url not in seen_urls:
+                    seen_urls.add(url)
+                    citations.append({
+                        "source": meta.get("source", "Unknown"),
+                        "url": url
+                    })
                 
             return {
                 "retrieved_chunks": retrieved_hits,
-                "citations": citations
+                "citations": citations[:4]
             }
         except Exception as e:
             logger.error(f"RetrievalService: Error during retrieval: {str(e)}")