Removed useless code and cleaned the pipeline.py

pipeline.py

@@ -10,7 +10,6 @@ from collections import OrderedDict
 import pandas as pd
 from pydantic import BaseModel, Field, ValidationError, validator
 
-# NLTK for input validation
 import nltk
 from nltk.corpus import words
 try:
@@ -19,7 +18,6 @@ except LookupError:
     nltk.download('words')
     english_words = set(words.words())
 
-# LangChain / Groq / LLM imports
 from langchain_groq import ChatGroq
 from langchain_community.embeddings import HuggingFaceEmbeddings
 from langchain_community.vectorstores import FAISS
@@ -28,35 +26,15 @@ from langchain.prompts import PromptTemplate
 from langchain.docstore.document import Document
 from langchain_core.caches import BaseCache
 from langchain_core.callbacks import Callbacks
-# from langchain_core.callbacks import CallbackManager
-# from langchain.callbacks.base import BaseCallbacks  # Updated import
-# from langchain.callbacks.manager import CallbackManager
-# from langchain.callbacks import StdOutCallbackHandler
 
-# Custom chain imports
-# from groq_client import GroqClient
 from chain.classification_chain import get_classification_chain
 from chain.refusal_chain import get_refusal_chain
 from chain.tailor_chain import get_tailor_chain
 from chain.cleaner_chain import get_cleaner_chain
 from chain.tailor_chain_wellnessBrand import get_tailor_chain_wellnessBrand
 
-# Mistral moderation
 from mistralai import Mistral
 
-# Google Gemini LLM
-# from langchain_google_genai import ChatGoogleGenerativeAI
-
-# Web search
-# from smolagents import DuckDuckGoSearchTool, ManagedAgent, HfApiModel, CodeAgent
-# from openinference.instrumentation.smolagents import SmolagentsInstrumentor
-# from phoenix.otel import register
-
-
-# register()
-# SmolagentsInstrumentor().instrument(skip_dep_check=True)
-
-
 from smolagents import (
     CodeAgent,
     DuckDuckGoSearchTool,
@@ -65,9 +43,7 @@ from smolagents import (
     VisitWebpageTool,
 )
 
-# Import new prompts
-from chain.prompts import selfharm_prompt, frustration_prompt, ethical_conflict_prompt,classification_prompt, refusal_prompt, tailor_prompt, cleaner_prompt
-
+from chain.prompts import selfharm_prompt, frustration_prompt, ethical_conflict_prompt, classification_prompt, refusal_prompt, tailor_prompt, cleaner_prompt
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -75,19 +51,12 @@ logger = logging.getLogger(__name__)
 from langchain_core.tracers import LangChainTracer
 from langsmith import Client
 
+os.environ["LANGCHAIN_TRACING_V2"] = "true"
+os.environ["LANGSMITH_ENDPOINT"] = "https://api.smith.langchain.com"
+os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGCHAIN_API_KEY")
+os.environ["LANGCHAIN_PROJECT"] = os.getenv("LANGCHAIN_PROJECT")
 
-os.environ["LANGCHAIN_TRACING_V2"]="true"
-os.environ["LANGSMITH_ENDPOINT"]="https://api.smith.langchain.com"
-# langsmith_client = Client()
-os.environ["LANGCHAIN_API_KEY"]=os.getenv("LANGCHAIN_API_KEY")
-os.environ["LANGCHAIN_PROJECT"]=os.getenv("LANGCHAIN_PROJECT")
-# tracer = LangChainTracer(project_name=os.environ.get("LANGCHAIN_PROJECT", "healthy_ai_expert"))
-
-
-
-# -------------------------------------------------------
 # Basic Models
-# -------------------------------------------------------
 class QueryInput(BaseModel):
     query: str = Field(..., min_length=1)
 
@@ -115,9 +84,7 @@ class ProcessingMetrics(BaseModel):
             / self.total_requests
         )
 
-# -------------------------------------------------------
 # Mistral Moderation
-# -------------------------------------------------------
 class ModerationResult(BaseModel):
     is_safe: bool
     categories: Dict[str, bool]
@@ -127,9 +94,7 @@ mistral_api_key = os.environ.get("MISTRAL_API_KEY")
 client = Mistral(api_key=mistral_api_key)
 
 def moderate_text(query: str) -> ModerationResult:
-    """
-    Uses Mistral's moderation to detect unsafe content.
-    """
+    """Moderates text using Mistral to detect unsafe content."""
     try:
         query_input = QueryInput(query=query)
         response = client.classifiers.moderate_chat(
@@ -161,53 +126,27 @@ def moderate_text(query: str) -> ModerationResult:
         raise RuntimeError(f"Moderation failed: {e}")
 
 def compute_moderation_severity(mresult: ModerationResult) -> float:
+    """Computes severity score based on moderation flags."""
     severity = 0.0
     for flag in mresult.categories.values():
         if flag:
             severity += 0.3
     return min(severity, 1.0)
 
-# -------------------------------------------------------
 # Models
-# -------------------------------------------------------
 GROQ_MODELS = {
-    "default":        "llama3-70b-8192",
+    "default": "llama3-70b-8192",
     "classification": "qwen-qwq-32b",
-    "moderation":     "mistral-moderation-latest",
-    "combination":    "llama-3.3-70b-versatile"
+    "moderation": "mistral-moderation-latest",
+    "combination": "llama-3.3-70b-versatile"
 }
 
 MAX_RETRIES = 3
 RATE_LIMIT_REQUESTS = 60  
 CACHE_SIZE_LIMIT = 1000
 
-# Google Gemini (primary)
-# GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY")
-# gemini_llm = ChatGoogleGenerativeAI(
-#     model="gemini-2.0-flash",
-#     temperature=0.5,
-#     max_tokens=None,
-#     timeout=None,
-#     max_retries=2,
-# )
-
-# # Fallback
-# fallback_groq_api_key = os.environ.get("GROQ_API_KEY_FALLBACK", "GROQ_API_KEY")
-
-# # Attempt to initialize ChatGroq without a cache
-# try:
-#     groq_fallback_llm = ChatGroq(
-#         model=GROQ_MODELS["default"],
-#         temperature=0.7,
-#         # groq_api_key=fallback_groq_api_key,
-#         max_tokens=2048
-#     )
-# except Exception as e:
-#     logger.error(f"Failed to initialize ChatGroq: {e}")
-#     raise RuntimeError("ChatGroq initialization failed.") from e
-# Define a simple no-op cache class
 class NoCache(BaseCache):
-    """Simple no-op cache implementation."""
+    """No-op cache implementation for ChatGroq."""
     def __init__(self):
         pass
 
@@ -220,28 +159,27 @@ class NoCache(BaseCache):
     def clear(self):
         pass
 
-# Rebuild the ChatGroq model after defining NoCache
 ChatGroq.model_rebuild()
-# Initialize ChatGroq with cache
+
 try:
     fallback_groq_api_key = os.environ.get("GROQ_API_KEY_FALLBACK", os.environ.get("GROQ_API_KEY"))
     if not fallback_groq_api_key:
         logger.warning("No Groq API key found for fallback LLM")
     groq_fallback_llm = ChatGroq(
-        model=GROQ_MODELS["default"],  # Replace with your actual model name if different
+        model=GROQ_MODELS["default"],
         temperature=0.7,
         groq_api_key=fallback_groq_api_key,
         max_tokens=2048,
-        cache=NoCache(),  # Set cache explicitly
-        callbacks=[]  # Explicitly set callbacks to an empty list
+        cache=NoCache(),
+        callbacks=[]
     )
 except Exception as e:
     logger.error(f"Failed to initialize fallback Groq LLM: {e}")
     raise RuntimeError("ChatGroq initialization failed.") from e
-# -------------------------------------------------------
+
 # Rate-limit & Cache
-# -------------------------------------------------------
 def handle_rate_limiting(state: "PipelineState") -> bool:
+    """Enforces rate limiting based on request timestamps."""
     current_time = time.time()
     one_min_ago = current_time - 60
     state.request_timestamps = [t for t in state.request_timestamps if t > one_min_ago]
@@ -251,6 +189,7 @@ def handle_rate_limiting(state: "PipelineState") -> bool:
     return True
 
 def manage_cache(state: "PipelineState", query: str, response: str = None) -> Optional[str]:
+    """Manages cache for query responses."""
     cache_key = query.strip().lower()
     if response is None:
         return state.cache.get(cache_key)
@@ -262,17 +201,16 @@ def manage_cache(state: "PipelineState", query: str, response: str = None) -> Op
     return None
 
 def create_error_response(error_type: str, details: str = "") -> str:
+    """Generates standardized error messages."""
     templates = {
         "validation": "I couldn't process your query: {details}",
         "processing": "I encountered an error while processing: {details}",
         "rate_limit": "Too many requests. Please try again soon.",
-        "general":    "Apologies, but something went wrong."
+        "general": "Apologies, but something went wrong."
     }
     return templates.get(error_type, templates["general"]).format(details=details)
 
-# -------------------------------------------------------
 # Web Search
-# -------------------------------------------------------
 web_search_cache: Dict[str, str] = {}
 
 def store_websearch_result(query: str, result: str):
@@ -282,6 +220,7 @@ def retrieve_websearch_result(query: str) -> Optional[str]:
     return web_search_cache.get(query.strip().lower())
 
 def do_web_search(query: str) -> str:
+    """Performs web search if no cached result exists."""
     try:
         cached = retrieve_websearch_result(query)
         if cached:
@@ -289,26 +228,17 @@ def do_web_search(query: str) -> str:
             return cached
 
         logger.info("Performing a new web search for: '%s'", query)
-        # model = HfApiModel()
-        # search_tool = DuckDuckGoSearchTool()
-        # web_agent = CodeAgent(tools=[search_tool], model=model)
-        
-        # managed_web_agent = ManagedAgent(
-        #     agent=web_agent,
-        #     name="web_search",
-        #     description="Runs a web search. Provide your query."
-        # )
         search_agent = ToolCallingAgent(
-        tools=[DuckDuckGoSearchTool(), VisitWebpageTool()],
-        model=HfApiModel(),
-        name="search_agent",
-        description="This is an agent that can do web search.",
+            tools=[DuckDuckGoSearchTool(), VisitWebpageTool()],
+            model=HfApiModel(),
+            name="search_agent",
+            description="This is an agent that can do web search.",
         )
         
         manager_agent = CodeAgent(
             tools=[],
-            model=model,
-            managed_agents=[managed_web_agent]
+            model=HfApiModel(),
+            managed_agents=[search_agent]
         )
         
         new_search_result = manager_agent.run(f"Search for information about: {query}")
@@ -319,34 +249,21 @@ def do_web_search(query: str) -> str:
         return ""
 
 def is_greeting(query: str) -> bool:
-    """
-    Returns True if the query is a greeting. This check is designed to be
-    lenient enough to catch common greetings even with minor spelling mistakes
-    or punctuation.
-    """
-    # Define a set of common greeting words (you can add variants or use fuzzy matching if needed)
+    """Detects if the query is a greeting."""
     greetings = {"hello", "hi", "hey", "hii", "hola", "greetings"}
-    
-    # Remove punctuation and extra whitespace, and lower the case.
     cleaned = re.sub(r'[^\w\s]', '', query).strip().lower()
-    
-    # Split the cleaned text into words.
     words_in_query = set(cleaned.split())
-    
-    # Return True if any of the greeting words are in the query.
     return not words_in_query.isdisjoint(greetings)
 
-
-# -------------------------------------------------------
 # Vector Stores & RAG
-# -------------------------------------------------------
 def build_or_load_vectorstore(csv_path: str, store_dir: str) -> FAISS:
+    """Builds or loads FAISS vector store from CSV data."""
     if os.path.exists(store_dir):
         logger.info(f"Loading existing FAISS store from {store_dir}")
         embeddings = HuggingFaceEmbeddings(
             model_name="sentence-transformers/multi-qa-mpnet-base-dot-v1"
         )
-        return FAISS.load_local(store_dir, embeddings,allow_dangerous_deserialization=True)# It will allow to deserialize and use the faiss store created first time locally 
+        return FAISS.load_local(store_dir, embeddings, allow_dangerous_deserialization=True)
     else:
         logger.info(f"Building new FAISS store from {csv_path}")
         df = pd.read_csv(csv_path)
@@ -373,8 +290,9 @@ def build_or_load_vectorstore(csv_path: str, store_dir: str) -> FAISS:
         vectorstore = FAISS.from_documents(docs, embedding=embeddings)
         vectorstore.save_local(store_dir)
         return vectorstore
-#rag chain is for wellness
+
 def build_rag_chain(vectorstore: FAISS, llm) -> RetrievalQA:
+    """Builds RAG chain for wellness queries."""
     prompt = PromptTemplate(
         template="""
         [INST] You are an AI wellness assistant speaking directly to a user who has asked: "{question}"
@@ -408,8 +326,9 @@ def build_rag_chain(vectorstore: FAISS, llm) -> RetrievalQA:
         }
     )
     return chain
-#rag chain to is for brand
+
 def build_rag_chain2(vectorstore: FAISS, llm) -> RetrievalQA:
+    """Builds RAG chain for brand strategy queries."""
     prompt = PromptTemplate(
         template="""
         [INST] You are the brand strategy advisor for Healthy AI Expert. A team member has asked: "{question}"
@@ -425,7 +344,6 @@ def build_rag_chain2(vectorstore: FAISS, llm) -> RetrievalQA:
         
         Remember our key brand pillars: AI-driven personalization, scientific credibility, user-centric design, and innovation leadership.
         [/INST]
-
         """,
         input_variables=["context", "question"]
     )
@@ -444,9 +362,7 @@ def build_rag_chain2(vectorstore: FAISS, llm) -> RetrievalQA:
     )
     return chain
 
-# -------------------------------------------------------
 # PipelineState
-# -------------------------------------------------------
 class PipelineState:
     _instance = None
     
@@ -462,6 +378,7 @@ class PipelineState:
         self._initialize()
 
     def _initialize(self):
+        """Initializes pipeline state and chains."""
         try:
             self.metrics = ProcessingMetrics()
             self.error_count = 0
@@ -478,52 +395,31 @@ class PipelineState:
             raise RuntimeError("Pipeline initialization failed.") from e
 
     def _setup_chains(self):
-        # Existing custom chains
+        """Sets up all processing chains and vector stores."""
         self.tailor_chainWellnessBrand = get_tailor_chain_wellnessBrand()
         self.classification_chain = get_classification_chain()
-        self.refusal_chain        = get_refusal_chain()
-        self.tailor_chain         = get_tailor_chain()
-        self.cleaner_chain        = get_cleaner_chain()
+        self.refusal_chain = get_refusal_chain()
+        self.tailor_chain = get_tailor_chain()
+        self.cleaner_chain = get_cleaner_chain()
 
-        # Specialized chain for self-harm
-        from chain.prompts import selfharm_prompt
-        # self.self_harm_chain = LLMChain(llm=gemini_llm, prompt=selfharm_prompt, verbose=False)
-        
         self.self_harm_chain = LLMChain(llm=groq_fallback_llm, prompt=selfharm_prompt, verbose=False)
-        
-        
-        # NEW: chain for frustration/harsh queries
-        from chain.prompts import frustration_prompt
-        # self.frustration_chain = LLMChain(llm=gemini_llm, prompt=frustration_prompt, verbose=False)
         self.frustration_chain = LLMChain(llm=groq_fallback_llm, prompt=frustration_prompt, verbose=False)
-        
-        
-        # NEW: chain for ethical conflict queries
-        from chain.prompts import ethical_conflict_prompt
-        # self.ethical_conflict_chain = LLMChain(llm=gemini_llm, prompt=ethical_conflict_prompt, verbose=False)
         self.ethical_conflict_chain = LLMChain(llm=groq_fallback_llm, prompt=ethical_conflict_prompt, verbose=False)
 
-        # Build brand & wellness vectorstores
-        brand_csv    = "dataset/BrandAI.csv"
-        brand_store  = "faiss_brand_store"
+        brand_csv = "dataset/BrandAI.csv"
+        brand_store = "faiss_brand_store"
         wellness_csv = "dataset/AIChatbot.csv"
         wellness_store = "faiss_wellness_store"
 
-        brand_vs    = build_or_load_vectorstore(brand_csv,    brand_store)
+        brand_vs = build_or_load_vectorstore(brand_csv, brand_store)
         wellness_vs = build_or_load_vectorstore(wellness_csv, wellness_store)
 
-        # Default LLM & fallback
-        # self.gemini_llm = gemini_llm
         self.groq_fallback_llm = groq_fallback_llm
-
-        # self.brand_rag_chain    = build_rag_chain2(brand_vs,    self.gemini_llm)
-        # self.wellness_rag_chain = build_rag_chain(wellness_vs, self.gemini_llm)
-        self.brand_rag_chain    = build_rag_chain2(brand_vs,    self.groq_fallback_llm)
+        self.brand_rag_chain = build_rag_chain2(brand_vs, self.groq_fallback_llm)
         self.wellness_rag_chain = build_rag_chain(wellness_vs, self.groq_fallback_llm)
-        # self.brand_rag_chain_fallback    = build_rag_chain2(brand_vs,    self.groq_fallback_llm)
-        # self.wellness_rag_chain_fallback = build_rag_chain(wellness_vs, self.groq_fallback_llm)
 
     def handle_error(self, error: Exception) -> bool:
+        """Handles errors and triggers reset if needed."""
         self.error_count += 1
         self.metrics.errors += 1
         if self.error_count >= MAX_RETRIES:
@@ -533,6 +429,7 @@ class PipelineState:
         return True
 
     def reset(self):
+        """Resets pipeline state while preserving metrics."""
         try:
             logger.info("Resetting pipeline state.")
             old_metrics = self.metrics
@@ -548,6 +445,7 @@ class PipelineState:
             raise RuntimeError("Failed to reset pipeline.")
 
     def get_metrics(self) -> Dict[str, Any]:
+        """Returns pipeline performance metrics."""
         uptime = (datetime.now() - self.metrics.last_reset).total_seconds() / 3600
         return {
             "total_requests": self.metrics.total_requests,
@@ -558,20 +456,15 @@ class PipelineState:
         }
 
     def update_metrics(self, start_time: float, is_cache_hit: bool = False):
+        """Updates processing metrics."""
         duration = time.time() - start_time
         self.metrics.update_metrics(duration, is_cache_hit)
 
 pipeline_state = PipelineState()
 
-# -------------------------------------------------------
-# Helper checks: detect aggression or ethical conflict
-# -------------------------------------------------------
-
+# Helper Checks
 def is_aggressive_or_harsh(query: str) -> bool:
-    """
-    Very naive check: If user is insulting AI, complaining about worthless answers, etc.
-    You can refine with better logic or a small LLM classifier.
-    """
+    """Detects aggressive or harsh language in query."""
     triggers = ["useless", "worthless", "you cannot do anything", "so bad at answering"]
     for t in triggers:
         if t in query.lower():
@@ -579,226 +472,140 @@ def is_aggressive_or_harsh(query: str) -> bool:
     return False
 
 def is_ethical_conflict(query: str) -> bool:
-    """
-    Check if user is asking about lying, revenge, or other moral dilemmas.
-    You can expand or refine as needed.
-    """
+    """Detects ethical dilemmas in query."""
     ethics_keywords = ["should i lie", "should i cheat", "revenge", "get back at", "hurt them back"]
     q_lower = query.lower()
     return any(k in q_lower for k in ethics_keywords)
 
-
-# -------------------------------------------------------
 # Main Pipeline
-# -------------------------------------------------------
 def run_with_chain(query: str) -> str:
-        """
-        Overall flow:
-        1) Validate & rate-limit
-        2) Mistral moderation => 
-           - If self-harm => self_harm_chain
-           - If hate => refusal
-           - If violence/dangerous => we STILL produce a guided response (ethics) unless it's extreme
-        3) If not refused, check if query is aggression/ethical => route to chain
-        4) Otherwise classify => brand/wellness/out-of-scope => RAG => tailor
-        """
-        # with tracer.new_trace(name="wellness_pipeline_run") as run:
-        start_time = time.time()
-        try:
-            # 1) Validate
-            if not query or query.strip() == "":
-                return create_error_response("validation", "Empty query.")
-            if len(query.strip()) < 2:
-                return create_error_response("validation", "Too short.")
-            words_in_text = re.findall(r'\b\w+\b', query.lower())
-            if not any(w in english_words for w in words_in_text):
-                return create_error_response("validation", "Unclear words.")
-            if len(query) > 500:
-                return create_error_response("validation", "Too long (>500).")
-            if not handle_rate_limiting(pipeline_state):
-                return create_error_response("rate_limit")
-            # New: Check if the query is a greeting
-            if is_greeting(query):
-                greeting_response = "Hello there!! Welcome to Healthy AI Expert,  How may I assist you today?"
-                manage_cache(pipeline_state, query, greeting_response)
-                pipeline_state.update_metrics(start_time)
-                return greeting_response
+    """Processes query through validation, moderation, and chains."""
+    start_time = time.time()
+    try:
+        if not query or query.strip() == "":
+            return create_error_response("validation", "Empty query.")
+        if len(query.strip()) < 2:
+            return create_error_response("validation", "Too short.")
+        words_in_text = re.findall(r'\b\w+\b', query.lower())
+        if not any(w in english_words for w in words_in_text):
+            return create_error_response("validation", "Unclear words.")
+        if len(query) > 500:
+            return create_error_response("validation", "Too long (>500).")
+        if not handle_rate_limiting(pipeline_state):
+            return create_error_response("rate_limit")
         
-            if not handle_rate_limiting(pipeline_state):
-                return create_error_response("rate_limit")
-    
-            # Cache check
-            cached = manage_cache(pipeline_state, query)
-            if cached:
-                pipeline_state.update_metrics(start_time, is_cache_hit=True)
-                return cached
-    
-            # 2) Mistral moderation
-            try:
-                mod_res = moderate_text(query)
-                severity = compute_moderation_severity(mod_res)
-    
-                # If self-harm => supportive
-                if mod_res.categories.get("selfharm", False):
-                    logger.info("Self-harm flagged => providing supportive chain response.")
-                    selfharm_resp = pipeline_state.self_harm_chain.run({"query": query})
-                    final_tailored = pipeline_state.tailor_chain.run({"response": selfharm_resp}).strip()
-                    manage_cache(pipeline_state, query, final_tailored)
-                    pipeline_state.update_metrics(start_time)
-                    return final_tailored
-    
-                # If hate => refuse
-                if mod_res.categories.get("hate", False):
-                    logger.info("Hate content => refusal.")
-                    refusal_resp = pipeline_state.refusal_chain.run({"topic": "moderation_flagged"})
-                    manage_cache(pipeline_state, query, refusal_resp)
-                    pipeline_state.update_metrics(start_time)
-                    return refusal_resp
-    
-                # If "dangerous" or "violence" is flagged, we might still want to 
-                # provide a "non-violent advice" approach (like revenge queries).
-                # So we won't automatically refuse. We'll rely on the 
-                # is_ethical_conflict() check below.
-    
-            except Exception as e:
-                logger.error(f"Moderation error: {e}")
-                severity = 0.0
-    
-            # 3) Check for aggression or ethical conflict
-            if is_aggressive_or_harsh(query):
-                logger.info("Detected harsh/aggressive language => frustration_chain.")
-                frustration_resp = pipeline_state.frustration_chain.run({"query": query})
-                final_tailored = pipeline_state.tailor_chain.run({"response": frustration_resp}).strip()
+        if is_greeting(query):
+            greeting_response = "Hello there!! Welcome to Healthy AI Expert, How may I assist you today?"
+            manage_cache(pipeline_state, query, greeting_response)
+            pipeline_state.update_metrics(start_time)
+            return greeting_response
+
+        cached = manage_cache(pipeline_state, query)
+        if cached:
+            pipeline_state.update_metrics(start_time, is_cache_hit=True)
+            return cached
+
+        try:
+            mod_res = moderate_text(query)
+            severity = compute_moderation_severity(mod_res)
+
+            if mod_res.categories.get("selfharm", False):
+                logger.info("Self-harm flagged => providing supportive chain response.")
+                selfharm_resp = pipeline_state.self_harm_chain.run({"query": query})
+                final_tailored = pipeline_state.tailor_chain.run({"response": selfharm_resp}).strip()
                 manage_cache(pipeline_state, query, final_tailored)
                 pipeline_state.update_metrics(start_time)
                 return final_tailored
-    
-            if is_ethical_conflict(query):
-                logger.info("Detected ethical dilemma => ethical_conflict_chain.")
-                ethical_resp = pipeline_state.ethical_conflict_chain.run({"query": query})
-                final_tailored = pipeline_state.tailor_chain.run({"response": ethical_resp}).strip()
-                manage_cache(pipeline_state, query, final_tailored)
+
+            if mod_res.categories.get("hate", False):
+                logger.info("Hate content => refusal.")
+                refusal_resp = pipeline_state.refusal_chain.run({"topic": "moderation_flagged"})
+                manage_cache(pipeline_state, query, refusal_resp)
                 pipeline_state.update_metrics(start_time)
-                return final_tailored
-    
-            # 4) Standard path: classification => brand/wellness/out-of-scope
-            try:
-                class_out = pipeline_state.classification_chain.run({"query": query})
-                classification = class_out.strip().lower()
-            except Exception as e:
-                logger.error(f"Classification error: {e}")
-                if not pipeline_state.handle_error(e):
-                    return create_error_response("processing", "Classification error.")
-                return create_error_response("processing")
-    
-            if classification in ["outofscope", "out_of_scope"]:
-                try:
-                    # Politely refuse if truly out-of-scope
-                    refusal_text = pipeline_state.refusal_chain.run({"topic": query})
-                    tailored_refusal = pipeline_state.tailor_chain.run({"response": refusal_text}).strip()
-                    manage_cache(pipeline_state, query, tailored_refusal)
-                    pipeline_state.update_metrics(start_time)
-                    return tailored_refusal
-                except Exception as e:
-                    logger.error(f"Refusal chain error: {e}")
-                    if not pipeline_state.handle_error(e):
-                        return create_error_response("processing", "Refusal error.")
-                    return create_error_response("processing")
-    
-            # brand vs wellness
-            if classification == "brand":
-                rag_chain_main = pipeline_state.brand_rag_chain
-                # rag_chain_fallback = pipeline_state.brand_rag_chain_fallback
-            else:
-                rag_chain_main = pipeline_state.wellness_rag_chain
-                # rag_chain_fallback = pipeline_state.wellness_rag_chain_fallback
-    
-            # RAG with fallback
-            try:
-                try:
-                    rag_output = rag_chain_main({"query": query})
-                except Exception as e_main:
-                    if "resource exhausted" in str(e_main).lower():
-                        logger.warning("Gemini resource exhausted. Falling back to Groq.")
-                        # rag_output = rag_chain_fallback({"query": query})
-                    else:
-                        raise
-    
-                if isinstance(rag_output, dict) and "result" in rag_output:
-                    csv_ans = rag_output["result"].strip()
-                else:
-                    csv_ans = str(rag_output).strip()
-    
-                # If not enough => web
-                if "not enough context" in csv_ans.lower() or len(csv_ans) < 40:
-                    logger.info("Insufficient RAG => web search.")
-                    web_info = do_web_search(query)
-                    if web_info:
-                        csv_ans += f"\n\nAdditional info:\n{web_info}"
-            except Exception as e:
-                logger.error(f"RAG error: {e}")
-                if not pipeline_state.handle_error(e):
-                    return create_error_response("processing", "RAG error.")
-                return create_error_response("processing")
-    
-            # Tailor final
+                return refusal_resp
+
+        except Exception as e:
+            logger.error(f"Moderation error: {e}")
+            severity = 0.0
+
+        if is_aggressive_or_harsh(query):
+            logger.info("Detected harsh/aggressive language => frustration_chain.")
+            frustration_resp = pipeline_state.frustration_chain.run({"query": query})
+            final_tailored = pipeline_state.tailor_chain.run({"response": frustration_resp}).strip()
+            manage_cache(pipeline_state, query, final_tailored)
+            pipeline_state.update_metrics(start_time)
+            return final_tailored
+
+        if is_ethical_conflict(query):
+            logger.info("Detected ethical dilemma => ethical_conflict_chain.")
+            ethical_resp = pipeline_state.ethical_conflict_chain.run({"query": query})
+            final_tailored = pipeline_state.tailor_chain.run({"response": ethical_resp}).strip()
+            manage_cache(pipeline_state, query, final_tailored)
+            pipeline_state.update_metrics(start_time)
+            return final_tailored
+
+        try:
+            class_out = pipeline_state.classification_chain.run({"query": query})
+            classification = class_out.strip().lower()
+        except Exception as e:
+            logger.error(f"Classification error: {e}")
+            if not pipeline_state.handle_error(e):
+                return create_error_response("processing", "Classification error.")
+            return create_error_response("processing")
+
+        if classification in ["outofscope", "out_of_scope"]:
             try:
-                final_tailored = pipeline_state.tailor_chainWellnessBrand.run({"response": csv_ans}).strip()
-                if severity > 0.5:
-                    final_tailored += "\n\n(Please note: This may involve sensitive content.)"
-    
-                manage_cache(pipeline_state, query, final_tailored)
+                refusal_text = pipeline_state.refusal_chain.run({"topic": query})
+                tailored_refusal = pipeline_state.tailor_chain.run({"response": refusal_text}).strip()
+                manage_cache(pipeline_state, query, tailored_refusal)
                 pipeline_state.update_metrics(start_time)
-                return final_tailored
+                return tailored_refusal
             except Exception as e:
-                logger.error(f"Tailor chain error: {e}")
+                logger.error(f"Refusal chain error: {e}")
                 if not pipeline_state.handle_error(e):
-                    return create_error_response("processing", "Tailoring error.")
+                    return create_error_response("processing", "Refusal error.")
                 return create_error_response("processing")
-    
+
+        if classification == "brand":
+            rag_chain_main = pipeline_state.brand_rag_chain
+        else:
+            rag_chain_main = pipeline_state.wellness_rag_chain
+
+        try:
+            rag_output = rag_chain_main({"query": query})
+            if isinstance(rag_output, dict) and "result" in rag_output:
+                csv_ans = rag_output["result"].strip()
+            else:
+                csv_ans = str(rag_output).strip()
+
+            if "not enough context" in csv_ans.lower() or len(csv_ans) < 40:
+                logger.info("Insufficient RAG => web search.")
+                web_info = do_web_search(query)
+                if web_info:
+                    csv_ans += f"\n\nAdditional info:\n{web_info}"
+        except Exception as e:
+            logger.error(f"RAG error: {e}")
+            if not pipeline_state.handle_error(e):
+                return create_error_response("processing", "RAG error.")
+            return create_error_response("processing")
+
+        try:
+            final_tailored = pipeline_state.tailor_chainWellnessBrand.run({"response": csv_ans}).strip()
+            if severity > 0.5:
+                final_tailored += "\n\n(Please note: This may involve sensitive content.)"
+
+            manage_cache(pipeline_state, query, final_tailored)
+            pipeline_state.update_metrics(start_time)
+            return final_tailored
         except Exception as e:
-            logger.error(f"Critical error in run_with_chain: {e}")
-            pipeline_state.metrics.errors += 1
-            return create_error_response("general")
-
-# -------------------------------------------------------
-# Health & Utility
-# -------------------------------------------------------
-# def reset_pipeline():
-#     try:
-#         pipeline_state.reset()
-#         return {"status": "success", "message": "Pipeline reset successful"}
-#     except Exception as e:
-#         logger.error(f"Reset pipeline error: {e}")
-#         return {"status": "error", "message": str(e)}
-
-# def get_pipeline_health() -> Dict[str, Any]:
-#     try:
-#         stats = pipeline_state.get_metrics()
-#         healthy = stats["error_rate"] < 0.1
-#         return {
-#             **stats,
-#             "is_healthy": healthy,
-#             "status": "healthy" if healthy else "degraded"
-#         }
-#     except Exception as e:
-#         logger.error(f"Health check error: {e}")
-#         return {"is_healthy": False, "status": "error", "error": str(e)}
-
-# def health_check() -> Dict[str, Any]:
-#     try:
-#         _ = run_with_chain("Test query for pipeline health check.")
-#         return {
-#             "status": "ok",
-#             "timestamp": datetime.now().isoformat(),
-#             "metrics": get_pipeline_health()
-#         }
-#     except Exception as e:
-#         return {
-#             "status": "error",
-#             "timestamp": datetime.now().isoformat(),
-#             "error": str(e)
-#         }
-
-logger.info("Pipeline initialization complete!")
+            logger.error(f"Tailor chain error: {e}")
+            if not pipeline_state.handle_error(e):
+                return create_error_response("processing", "Tailoring error.")
+            return create_error_response("processing")
+
+    except Exception as e:
+        logger.error(f"Critical error in run_with_chain: {e}")
+        pipeline_state.metrics.errors += 1
+        return create_error_response("general")
 
+logger.info("Pipeline initialization complete!")