database chat external user's api

controller.py

@@ -5,16 +5,16 @@ import os
 import asyncio
 import threading
 import uuid
-import time
-from fastapi import FastAPI, HTTPException, Header, Request
+from fastapi import FastAPI, HTTPException, Header
 from fastapi.encoders import jsonable_encoder
-from typing import Dict, List, Optional
+from typing import Dict, List, Literal, Optional
+from fastapi.responses import FileResponse
 import numpy as np
 import pandas as pd
 from pandasai import SmartDataframe
 from langchain_groq.chat_models import ChatGroq
 from dotenv import load_dotenv
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ValidationError
 from csv_service import clean_data, extract_chart_filenames, generate_csv_data, get_csv_basic_info
 from urllib.parse import unquote
 from langchain_groq import ChatGroq
@@ -36,109 +36,43 @@ from fastapi.middleware.cors import CORSMiddleware
 import matplotlib
 matplotlib.use('Agg')
 
-# Load environment variables first
-load_dotenv()
-
-# Configure professional logging
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
-    datefmt='%Y-%m-%d %H:%M:%S'
-)
-logger = logging.getLogger(__name__)
-
-# Performance logger for timing critical operations
-perf_logger = logging.getLogger("performance")
-perf_logger.setLevel(logging.INFO)
-
-def log_performance(operation_name: str):
-    """Decorator to log operation performance"""
-    def decorator(func):
-        async def async_wrapper(*args, **kwargs):
-            start_time = time.time()
-            try:
-                result = await func(*args, **kwargs)
-                duration = time.time() - start_time
-                perf_logger.info(f"{operation_name} completed in {duration:.2f}s")
-                return result
-            except Exception as e:
-                duration = time.time() - start_time
-                perf_logger.error(f"{operation_name} failed after {duration:.2f}s: {str(e)}")
-                raise
-        
-        def sync_wrapper(*args, **kwargs):
-            start_time = time.time()
-            try:
-                result = func(*args, **kwargs)
-                duration = time.time() - start_time
-                perf_logger.info(f"{operation_name} completed in {duration:.2f}s")
-                return result
-            except Exception as e:
-                duration = time.time() - start_time
-                perf_logger.error(f"{operation_name} failed after {duration:.2f}s: {str(e)}")
-                raise
-        
-        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
-    return decorator
-
 # Initialize FastAPI app
-app = FastAPI(title="CSV Chat Service", version="1.0.0")
+app = FastAPI()
 
-logger.info("Initializing CSV Chat Service...")
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
 # Initialize the ProcessPoolExecutor
 max_cpus = os.cpu_count()
-logger.info(f"System CPUs detected: {max_cpus}, using {max_cpus-2} workers")
+logger.info(f"Max CPUs: {max_cpus}")
 
-# Ensure directories exist
-directories = ["/app/cache", "/app", "/app/generated_charts"]
-for directory in directories:
-    os.makedirs(directory, exist_ok=True)
-    logger.debug(f"Directory ensured: {directory}")
+# Ensure the cache directory exists
+os.makedirs("/app/cache", exist_ok=True)
 
-open("/app/pandasai.log", "a").close()
-logger.debug("PandasAI log file initialized")
+os.makedirs("/app", exist_ok=True)
+open("/app/pandasai.log", "a").close()  # Create the file if it doesn't exist
 
-# CORS Configuration - Optimized and Fixed
-image_file_path = os.getenv("IMAGE_FILE_PATH")
-image_not_found = os.getenv("IMAGE_NOT_FOUND")
-
-# Get allowed hosts from environment variable with fallback
-allowed_hosts_env = os.getenv("ALLOWED_HOSTS", "")
-if allowed_hosts_env:
-    allowed_hosts = [host.strip() for host in allowed_hosts_env.split(",") if host.strip()]
-else:
-    # Secure fallback: only allow your specific domain
-    allowed_hosts = ["https://chatcsvandpdf.vercel.app"]
+# Ensure the generated_charts directory exists
+os.makedirs("/app/generated_charts", exist_ok=True)
 
-logger.info(f"CORS configured for origins: {allowed_hosts}")
+load_dotenv()
 
-# Add CORS middleware with optimized configuration
+image_file_path = os.getenv("IMAGE_FILE_PATH")
+image_not_found = os.getenv("IMAGE_NOT_FOUND")
+allowed_hosts = os.getenv("ALLOWED_HOSTS", "").split(",")
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=allowed_hosts,
+    allow_origins=allowed_hosts, 
     allow_credentials=True,
-    allow_methods=["GET", "POST", "OPTIONS"],  # Only methods you actually use
-    allow_headers=["Content-Type", "Authorization", "Accept"],  # Specific headers only
-    expose_headers=["Content-Length"],
-    max_age=3600,  # Cache preflight requests for 1 hour
+    allow_methods=["*"], 
+    allow_headers=["*"],  
 )
 
-# Load environment variables with validation
-groq_api_keys = os.getenv("GROQ_API_KEYS", "").split(",")
+# Load environment variables
+groq_api_keys = os.getenv("GROQ_API_KEYS").split(",")
 model_name = os.getenv("GROQ_LLM_MODEL")
 
-if not groq_api_keys or not groq_api_keys[0]:
-    logger.critical("GROQ_API_KEYS not configured properly")
-    raise ValueError("GROQ API keys are required")
-
-if not model_name:
-    logger.critical("GROQ_LLM_MODEL not configured")
-    raise ValueError("Model name is required")
-
-logger.info(f"Service initialized with {len(groq_api_keys)} API keys and model: {model_name}")
-
-# Pydantic models remain the same
 class CsvUrlRequest(BaseModel):
     csv_url: str
     
@@ -149,7 +83,7 @@ class ImageRequest(BaseModel):
 class FileProps(BaseModel):
     fileName: str
     filePath: str
-    fileType: str
+    fileType: str  # 'csv' | 'image'
 
 class Files(BaseModel):
     csv_files: List[FileProps]
@@ -157,156 +91,118 @@ class Files(BaseModel):
 
 class FileBoxProps(BaseModel):
     files: Files
-
-# Optimized thread-safe key management
-class ApiKeyManager:
-    def __init__(self, keys: List[str], name: str):
-        self.keys = keys
-        self.name = name
-        self.current_index = 0
-        self.lock = threading.Lock()
-        self.exhausted = False
     
-    def get_next_key(self):
-        with self.lock:
-            if self.current_index >= len(self.keys):
-                self.exhausted = True
-                logger.warning(f"All {self.name} API keys exhausted")
-                return None
-            key = self.keys[self.current_index]
-            self.current_index += 1
-            logger.debug(f"Using {self.name} API key index: {self.current_index-1}")
-            return key
-    
-    def reset(self):
-        with self.lock:
-            self.current_index = 0
-            self.exhausted = False
-            logger.info(f"Reset {self.name} API key rotation")
-
-# Initialize key managers
-groq_key_manager = ApiKeyManager(groq_api_keys, "Groq")
-langchain_key_manager = ApiKeyManager(groq_api_keys, "Langchain")
-chart_key_manager = ApiKeyManager(groq_api_keys, "Chart")
-
-# Use a process pool for CPU-bound operations
-process_executor = ProcessPoolExecutor(max_workers=max_cpus-2)
-logger.info(f"ProcessPoolExecutor initialized with {max_cpus-2} workers")
-
-# Middleware to log requests
-@app.middleware("http")
-async def log_requests(request: Request, call_next):
-    start_time = time.time()
-    client_ip = request.client.host
-    method = request.method
-    url = str(request.url)
-    
-    logger.info(f"Request started: {method} {url} from {client_ip}")
-    
-    response = await call_next(request)
-    
-    process_time = time.time() - start_time
-    logger.info(f"Request completed: {method} {url} - Status: {response.status_code} - Duration: {process_time:.2f}s")
-    
-    return response
+dummy_response = FileBoxProps(
+            files=Files(
+                csv_files=[
+                    FileProps(
+                        fileName="sales_data.csv",
+                        filePath="/downloads/sales_data.csv",
+                        fileType="csv"
+                    ),
+                    FileProps(
+                        fileName="customer_data.csv",
+                        filePath="/downloads/customer_data.csv",
+                        fileType="csv"
+                    )
+                ],
+                image_files=[
+                    FileProps(
+                        fileName="chart.png",
+                        filePath="/downloads/chart.png",
+                        fileType="image"
+                    ),
+                    FileProps(
+                        fileName="graph.jpg",
+                        filePath="/downloads/graph.jpg",
+                        fileType="image"
+                    )
+                ]
+            )
+        )
+
+
+# Thread-safe key management for groq_chat
+current_groq_key_index = 0
+current_groq_key_lock = threading.Lock()
+
+# Thread-safe key management for langchain_csv_chat
+current_langchain_key_index = 0
+current_langchain_key_lock = threading.Lock()
 
-# ROOT CHECK
-@app.get("/")
-async def root():
-    logger.debug("Health check endpoint accessed")
-    return {"message": "CSV Chat-Service-1 is running", "status": "healthy"}
 
 # PING CHECK
 @app.get("/ping")
-async def ping():
-    logger.debug("Ping endpoint accessed")
-    return {"message": "Pong !!", "timestamp": time.time()}
-
-# CORS test endpoint
-@app.get("/api/cors-test")
-async def cors_test():
-    logger.info("CORS test endpoint accessed")
-    return {
-        "message": "CORS is working", 
-        "allowed_origins": allowed_hosts,
-        "timestamp": time.time()
-    }
+async def root():
+    return {"message": "Pong !!"}
+
+
 
 # BASIC KNOWLEDGE BASED ON CSV
+
+# Remove trailing slash from the URL otherwise it will redirect to GET method
 @app.post("/api/basic_csv_data")
-@log_performance("basic_csv_data")
 async def basic_csv_data(request: CsvUrlRequest):
     try:
         decoded_url = unquote(request.csv_url)
-        logger.info(f"Processing basic CSV data request for URL: {decoded_url[:100]}...")
-        
+        logger.info(f"Fetching CSV data from URL: {decoded_url}")
+        # csv_data = await get_csv_basic_info(decoded_url)    
+        # Run the synchronous function in a thread pool executor
         loop = asyncio.get_running_loop()
         csv_data = await loop.run_in_executor(
             process_executor, get_csv_basic_info, decoded_url
         )
-        
-        logger.info("CSV basic data retrieved successfully")
+        logger.info(f"CSV data fetched successfully: {csv_data}")
         return {"data": csv_data}
-        
     except Exception as e:
-        logger.error(f"Error in basic_csv_data: {str(e)}")
+        logger.error(f"Error while fetching CSV data: {e}")
         raise HTTPException(status_code=400, detail=f"Failed to retrieve CSV data: {str(e)}")
 
+
 # GET THE CHART FROM A SPECIFIC FILE PATH
 @app.post("/api/get-chart")
-@log_performance("get_chart")
 async def get_image(request: ImageRequest, authorization: str = Header(None)):
-    logger.info(f"Chart retrieval request for chat_id: {request.chat_id}")
-    
-    # Optimized auth check
-    if not authorization or not authorization.startswith("Bearer "):
-        logger.warning("Invalid authorization header format")
-        raise HTTPException(status_code=401, detail="Authorization header missing or invalid")
+    if not authorization:
+        raise HTTPException(status_code=401, detail="Authorization header missing")
 
-    token = authorization[7:]  # Remove "Bearer " prefix
-    if not token or token != os.getenv("AUTH_TOKEN"):
-        logger.warning("Authentication failed")
+    if not authorization.startswith("Bearer "):
+        raise HTTPException(status_code=401, detail="Invalid authorization header format")
+
+    token = authorization.split(" ")[1]
+    if not token:
+        raise HTTPException(status_code=401, detail="Token missing")
+    if token != os.getenv("AUTH_TOKEN"):
         raise HTTPException(status_code=403, detail="Invalid token")
 
     try:
+        logger.info("Groq Chat created a chat for the user query...")
         image_file_path = request.image_path
-        unique_file_name = f'{str(uuid.uuid4())}.png'
-        
-        logger.info(f"Uploading chart to Supabase: {unique_file_name}")
-        image_public_url = await upload_file_to_supabase(
-            image_file_path, unique_file_name, chat_id=request.chat_id
-        )
-        
-        # Clean up local file
-        if os.path.exists(image_file_path):
-            os.remove(image_file_path)
-            logger.debug(f"Local file cleaned up: {image_file_path}")
-        
-        logger.info("Chart uploaded successfully")
+        unique_file_name =f'{str(uuid.uuid4())}.png'
+        logger.info("Uploading the chart to supabase...")
+        image_public_url = await upload_file_to_supabase(f"{image_file_path}", unique_file_name, chat_id=request.chat_id)
+        logger.info("Image uploaded to Supabase and Image URL is... ", {image_public_url})
+        os.remove(image_file_path)
         return {"image_url": image_public_url}
-        
+        # return FileResponse(image_file_path, media_type="image/png")
     except Exception as e:
-        logger.error(f"Error in get_image: {str(e)}")
+        logger.error(f"Error: {e}")
         return {"answer": "error"}
+        
 
 # GET CSV DATA FOR GENERATING THE TABLE
 @app.post("/api/csv_data")
-@log_performance("csv_data_generation")
 async def get_csv_data(request: CsvUrlRequest):
     try:
         decoded_url = unquote(request.csv_url)
-        logger.info(f"Generating CSV table data for URL: {decoded_url[:100]}...")
-        
+        logger.info(f"Fetching CSV data from URL: {decoded_url}")
+        # csv_data = await generate_csv_data(decoded_url)     
         loop = asyncio.get_running_loop()
         csv_data = await loop.run_in_executor(
             process_executor, generate_csv_data, decoded_url
-        )
-        
-        logger.info("CSV table data generated successfully")
+        )   
         return csv_data
-        
     except Exception as e:
-        logger.error(f"Error in get_csv_data: {str(e)}")
+        logger.error(f"Error while fetching CSV data: {e}")
         raise HTTPException(status_code=400, detail=f"Failed to retrieve CSV data: {str(e)}")
 
 # EXECUTE THE PYTHON CODE 
@@ -315,112 +211,109 @@ class ExecutionRequest(BaseModel):
     csv_url: str = Field(..., alias="csv_url")
     codeExecutionPayload: CsvChatResult
 
+
 @app.post("/api/code_execution_csv")
-@log_performance("code_execution")
 async def code_execution_csv(
-    request_data: ExecutionRequest,
+    request_data: ExecutionRequest,  # Change from ExecutionRequest to dict to see raw input
     authorization: Optional[str] = Header(None)
 ):
-    logger.info(f"Code execution request for chat_id: {request_data.chat_id}")
-    
+    # Auth check remains the same
     expected_token = os.environ.get("AUTH_TOKEN")
     if not authorization or not expected_token or authorization.replace("Bearer ", "") != expected_token:
-        logger.warning("Unauthorized code execution attempt")
         raise HTTPException(status_code=401, detail="Unauthorized")
 
     try:
+        # First log the incoming request data
+        logger.info("Incoming request data:", request_data)
+            
+        # Rest of your processing logic...
         decoded_url = unquote(request_data.csv_url)
-        logger.info("Loading and cleaning CSV data for execution")
-        
         df = clean_data(decoded_url)
         executor = PythonExecutor(df)
-        
-        logger.info("Executing Python code payload")
-        formatted_output = await executor.process_response(
-            request_data.codeExecutionPayload, request_data.chat_id
-        )
-        
-        logger.info("Code execution completed successfully")
+        formatted_output = await executor.process_response(request_data.codeExecutionPayload, request_data.chat_id)
         return {"answer": formatted_output}
 
     except Exception as e:
-        logger.error(f"Code execution error: {str(e)}")
+        logger.info("Processing error:", str(e))
         return {"error": "Failed to execute request", "message": str(e)}
 
+
 # CHAT CODING STARTS FROM HERE
 
+# Modified groq_chat function with thread-safe key rotation
 def groq_chat(csv_url: str, question: str):
-    """Optimized groq chat with better error handling"""
-    logger.info("Starting Groq chat processing")
-    
-    while not groq_key_manager.exhausted:
-        current_api_key = groq_key_manager.get_next_key()
-        if not current_api_key:
-            break
+    global current_groq_key_index, current_groq_key_lock
+
+    while True:
+        with current_groq_key_lock:
+            if current_groq_key_index >= len(groq_api_keys):
+                return {"error": "All API keys exhausted."}
+            current_api_key = groq_api_keys[current_groq_key_index]
 
         try:
-            logger.debug("Loading and cleaning CSV data")
+           
             data = clean_data(csv_url)
-            
             llm = ChatGroq(model=model_name, api_key=current_api_key)
+            # Generate unique filename using UUID
             chart_filename = f"chart_{uuid.uuid4()}.png"
             chart_path = os.path.join("generated_charts", chart_filename)
             
-            logger.debug(f"Creating SmartDataframe with chart path: {chart_path}")
+            # Configure SmartDataframe with chart settings
             df = SmartDataframe(
                 data,
                 config={
                     'llm': llm,
-                    'save_charts': True,
+                    'save_charts': True,  # Enable chart saving
                     'open_charts': False,
-                    'save_charts_path': os.path.dirname(chart_path),
-                    'custom_chart_filename': chart_filename,
+                    'save_charts_path': os.path.dirname(chart_path),  # Directory to save
+                    'custom_chart_filename': chart_filename, # Unique filename
                     'enable_cache': False
                 }
             )
             
-            logger.debug("Processing chat query with SmartDataframe")
             answer = df.chat(question)
 
-            # Optimized response processing
+            # Process different response types
             if isinstance(answer, pd.DataFrame):
                 processed = answer.apply(handle_out_of_range_float).to_dict(orient="records")
             elif isinstance(answer, pd.Series):
                 processed = answer.apply(handle_out_of_range_float).to_dict()
-            elif isinstance(answer, (list, dict)):
-                processed = jsonable_encoder(answer)
+            elif isinstance(answer, list):
+                processed = [handle_out_of_range_float(item) for item in answer]
+            elif isinstance(answer, dict):
+                processed = {k: handle_out_of_range_float(v) for k, v in answer.items()}
             else:
                 processed = {"answer": str(handle_out_of_range_float(answer))}
 
-            logger.info("Groq chat processing completed successfully")
             return processed
 
         except Exception as e:
             error_message = str(e)
-            if "429" in error_message or "rate limit" in error_message.lower():
-                logger.warning(f"Rate limit hit, trying next API key")
-                continue
+            if error_message != "":
+                logger.warning("Rate limit exceeded. Switching to next API key.")
+                with current_groq_key_lock:
+                    current_groq_key_index += 1
+                    if current_groq_key_index >= len(groq_api_keys):
+                        return {"error": "All API keys exhausted."}
             else:
-                logger.error(f"Groq chat error: {error_message}")
+                logger.error("Error in groq_chat: %s", e)
                 return {"error": error_message}
 
-    logger.error("All Groq API keys exhausted")
-    return {"error": "All API keys exhausted."}
-
+# Modified langchain_csv_chat with thread-safe key rotation
 def langchain_csv_chat(csv_url: str, question: str, chart_required: bool):
-    """Optimized langchain chat with better key management"""
-    logger.info(f"Starting Langchain CSV chat (chart_required: {chart_required})")
-    
+    global current_langchain_key_index, current_langchain_key_lock, current_langchain_chart_key_index, current_langchain_chart_lock
+
     data = clean_data(csv_url)
     attempts = 0
 
     while attempts < len(groq_api_keys):
-        api_key = langchain_key_manager.get_next_key()
-        if not api_key:
-            break
-        
-        attempts += 1
-        logger.debug(f"Langchain attempt {attempts} with API key")
+        with current_langchain_key_lock:
+            if current_langchain_key_index >= len(groq_api_keys):
+                current_langchain_key_index = 0
+            api_key = groq_api_keys[current_langchain_key_index]
+            current_key = current_langchain_key_index
+            current_langchain_key_index += 1
+            attempts += 1
 
         try:
             llm = ChatGroq(model=model_name, api_key=api_key)
@@ -437,114 +330,95 @@ def langchain_csv_chat(csv_url: str, question: str, chart_required: bool):
                 llm,
                 data,
                 agent_type="tool-calling",
-                verbose=False,  # Reduce verbosity for performance
+                verbose=True,
                 allow_dangerous_code=True,
                 extra_tools=[tool],
                 return_intermediate_steps=True
             )
 
             prompt = _prompt_generator(question, chart_required, csv_url)
-            logger.debug("Invoking Langchain agent")
             result = agent.invoke({"input": prompt})
-            
-            logger.info("Langchain processing completed successfully")
             return result.get("output")
 
         except Exception as e:
             error_message = str(e)
-            if "429" in error_message or "rate limit" in error_message.lower():
-                logger.warning(f"Langchain rate limit hit, attempt {attempts}")
-                continue
+            if error_message != "":
+                with current_langchain_chart_lock:
+                    current_langchain_chart_key_index = (current_langchain_chart_key_index + 1)
+                    logger.warning(f"Rate limit exceeded. Switching to next API key: {groq_api_keys[current_langchain_chart_key_index]}")
             else:
-                logger.error(f"Langchain error on attempt {attempts}: {error_message}")
+                logger.error(f"Error with API key {api_key}: {error_message}")
                 return {"error": error_message}
 
-    logger.error("All Langchain API keys exhausted")
     return {"error": "All API keys exhausted"}
 
+# Async endpoint with non-blocking execution
 @app.post("/api/csv-chat")
-@log_performance("csv_chat")
 async def csv_chat(request: Dict, authorization: str = Header(None)):
-    """Optimized CSV chat endpoint"""
-    logger.info("Request Body: %s", request)
-    chat_id = request.get("chat_id", "unknown")
-    logger.info(f"CSV chat request started for chat_id: {chat_id}")
-    
-    # Optimized authorization check
+    # Authorization checks
     if not authorization or not authorization.startswith("Bearer "):
-        logger.warning("Invalid authorization format")
         raise HTTPException(status_code=401, detail="Invalid authorization")
     
-    token = authorization[7:]  # Remove "Bearer " prefix
+    token = authorization.split(" ")[1]
     if token != os.getenv("AUTH_TOKEN"):
-        logger.warning("Authentication failed")
         raise HTTPException(status_code=403, detail="Invalid token")
 
     try:
-        # Extract request parameters
         query = request.get("query")
         csv_url = request.get("csv_url")
         decoded_url = unquote(csv_url)
-        detailed_answer = request.get("detailed_answer", False)
+        detailed_answer = request.get("detailed_answer")
         conversation_history = request.get("conversation_history", [])
-        generate_report = request.get("generate_report", False)
+        generate_report = request.get("generate_report")
+        chat_id = request.get("chat_id")
         
-        logger.info(f"Processing query: '{query[:50]}...' (detailed: {detailed_answer}, report: {generate_report})")
-
-        # Report generation path
-        if generate_report:
-            logger.info("Generating CSV report")
+        if generate_report is True:
             report_files = await generate_csv_report(csv_url, query, chat_id, conversation_history)
             if report_files is not None:
-                logger.info("Report generated successfully")
                 return {"answer": jsonable_encoder(report_files)}
 
-        # Initial chat question path
         if if_initial_chat_question(query):
-            logger.info("Processing as initial chat question")
             answer = await asyncio.to_thread(
                 langchain_csv_chat, decoded_url, query, False
             )
-            logger.info("Initial chat question processed")
+            logger.info("langchain_answer:", answer)
             return {"answer": jsonable_encoder(answer)}
         
-        # Detailed answer orchestration
-        if detailed_answer:
-            logger.info("Processing with orchestrator for detailed answer")
-            orchestrator_answer = await asyncio.to_thread(
-                csv_orchestrator_chat, decoded_url, query, conversation_history, chat_id
-            )
-            if orchestrator_answer is not None:
-                logger.info("Orchestrator processing completed")
-                return {"answer": jsonable_encoder(orchestrator_answer)}
+        # Orchestrate the execution
+        if detailed_answer is True:
+           orchestrator_answer = await asyncio.to_thread(
+            csv_orchestrator_chat, decoded_url, query, conversation_history, chat_id
+           )
+           if orchestrator_answer is not None:
+              return {"answer": jsonable_encoder(orchestrator_answer)}
 
-        # Cerebras CSV agent processing
-        logger.info("Processing with Cerebras CSV agent")
-        result = await query_csv_agent(decoded_url, query, chat_id)
+        # Process with groq_chat first
+        # groq_answer = await asyncio.to_thread(groq_chat, decoded_url, query)
+        # logger.info("groq_answer:", groq_answer)
         
-        if result is not None and result != "":
-            logger.info("Cerebras agent processing successful")
+        result = await query_csv_agent(decoded_url, query, chat_id)
+        logger.info("cerebras csv answer == >", result)
+        if result is not None or result == "":
             return {"answer": result}
      
-        # Fallback to langchain
-        logger.info("Falling back to Langchain processing")
+        # if process_answer(groq_answer) == "Empty response received.":
+        #     return {"answer": "Sorry, I couldn't find relevant data..."}
+
+        # if process_answer(groq_answer):
         lang_answer = await asyncio.to_thread(
             langchain_csv_chat, decoded_url, query, False
         )
-        
-        if not process_answer(lang_answer):
-            logger.info("Langchain fallback successful")
-            return {"answer": jsonable_encoder(lang_answer)}
-        
-        logger.warning("All processing methods failed")
-        return {"answer": "error"}
+        if process_answer(lang_answer):
+            return {"answer": "error"}
+        return {"answer": jsonable_encoder(lang_answer)}
+
+        # return {"answer": jsonable_encoder(groq_answer)}
 
     except Exception as e:
-        logger.error(f"Critical error in csv_chat: {str(e)}")
+        logger.error(f"Error processing request: {str(e)}")
         return {"answer": "error"}
 
 def handle_out_of_range_float(value):
-    """Optimized float handling"""
     if isinstance(value, float):
         if np.isnan(value):
             return None
@@ -552,82 +426,125 @@ def handle_out_of_range_float(value):
             return "Infinity"
     return value
 
-# CHART GENERATION SECTION
+
+
+
+
+
+
+# CHART CODING STARTS FROM HERE
 
 instructions = """
+
 - Please ensure that each value is clearly visible, You may need to adjust the font size, rotate the labels, or use truncation to improve readability (if needed).
 - For multiple charts, put all of them in a single file.
 - Use colorblind-friendly palette
 - Read above instructions and follow them.
+
 """
 
+# Thread-safe configuration for chart endpoints
+current_groq_chart_key_index = 0
+current_groq_chart_lock = threading.Lock()
+
+# current_langchain_chart_key_index = 0
+# current_langchain_chart_lock = threading.Lock()
+
 def model():
-    """Get model instance with current API key"""
-    api_key = chart_key_manager.get_next_key()
-    if not api_key:
-        raise Exception("All API keys exhausted for chart generation")
+    global current_groq_chart_key_index, current_groq_chart_lock
+    with current_groq_chart_lock:
+        if current_groq_chart_key_index >= len(groq_api_keys):
+            raise Exception("All API keys exhausted for chart generation")
+        api_key = groq_api_keys[current_groq_chart_key_index]
     return ChatGroq(model=model_name, api_key=api_key)
 
 def groq_chart(csv_url: str, question: str):
-    """Optimized Groq chart generation"""
-    logger.info("Starting Groq chart generation")
+    global current_groq_chart_key_index, current_groq_chart_lock
     
     for attempt in range(len(groq_api_keys)):
-        api_key = chart_key_manager.get_next_key()
-        if not api_key:
-            break
-            
         try:
-            logger.debug(f"Chart generation attempt {attempt + 1}")
+            # Clean cache before processing
+            # cache_db_path = "/workspace/cache/cache_db_0.11.db"
+            # if os.path.exists(cache_db_path):
+            #     try:
+            #         os.remove(cache_db_path)
+            #     except Exception as e:
+            #         logger.info(f"Cache cleanup error: {e}")
+
             data = clean_data(csv_url)
-            llm = ChatGroq(model=model_name, api_key=api_key)
+            with current_groq_chart_lock:
+                current_api_key = groq_api_keys[current_groq_chart_key_index]
+            
+            llm = ChatGroq(model=model_name, api_key=current_api_key)
             
+            # Generate unique filename using UUID
             chart_filename = f"chart_{uuid.uuid4()}.png"
             chart_path = os.path.join("generated_charts", chart_filename)
             
+            # Configure SmartDataframe with chart settings
             df = SmartDataframe(
                 data,
                 config={
                     'llm': llm,
-                    'save_charts': True,
+                    'save_charts': True,  # Enable chart saving
                     'open_charts': False,
-                    'save_charts_path': os.path.dirname(chart_path),
-                    'custom_chart_filename': chart_filename,
+                    'save_charts_path': os.path.dirname(chart_path),  # Directory to save
+                    'custom_chart_filename': chart_filename,  # Unique filename
                     'enable_cache': False
                 }
             )
             
             answer = df.chat(question + instructions)
             
-            if not process_answer(answer):
-                logger.info("Groq chart generated successfully")
-                return answer
-            
+            if process_answer(answer):
+                return "Chart not generated"
+            return answer
+
         except Exception as e:
             error = str(e)
-            if "429" in error or "rate limit" in error.lower():
-                logger.warning(f"Rate limit hit on attempt {attempt + 1}")
-                continue
+            # if "429" in error:
+            if error != "":
+                with current_groq_chart_lock:
+                    current_groq_chart_key_index = (current_groq_chart_key_index + 1)
             else:
                 logger.error(f"Chart generation error: {error}")
                 return {"error": error}
     
-    logger.error("All API keys exhausted for Groq chart generation")
     return {"error": "All API keys exhausted for chart generation"}
 
+
+
+
+# Global locks for key rotation (chart endpoints)
+# current_groq_chart_key_index = 0
+# current_groq_chart_lock = threading.Lock()
+current_langchain_chart_key_index = 0
+current_langchain_chart_lock = threading.Lock()
+
+
+# Use a process pool to run CPU-bound charts generation
+process_executor = ProcessPoolExecutor(max_workers=max_cpus-2)
+
+# --- LANGCHAIN-BASED CHART GENERATION ---
 def langchain_csv_chart(csv_url: str, question: str, chart_required: bool):
-    """Optimized langchain chart generation"""
-    logger.info("Starting Langchain chart generation")
+    """
+    Generate a chart using the langchain-based method.
+    Modifications:
+      • No shared deletion of cache.
+      • Close matplotlib figures after generation.
+      • Return a list of full chart file paths.
+    """
+    global current_langchain_chart_key_index, current_langchain_chart_lock
     
     data = clean_data(csv_url)
 
     for attempt in range(len(groq_api_keys)):
-        api_key = chart_key_manager.get_next_key()
-        if not api_key:
-            break
-
         try:
-            logger.debug(f"Langchain chart attempt {attempt + 1}")
+            with current_langchain_chart_lock:
+                api_key = groq_api_keys[current_langchain_chart_key_index]
+                current_key = current_langchain_chart_key_index
+                current_langchain_chart_key_index = (current_langchain_chart_key_index + 1) % len(groq_api_keys)
+
             llm = ChatGroq(model=model_name, api_key=api_key)
             tool = PythonAstREPLTool(locals={
                 "df": data,
@@ -643,7 +560,7 @@ def langchain_csv_chart(csv_url: str, question: str, chart_required: bool):
                 llm,
                 data,
                 agent_type="tool-calling",
-                verbose=False,  # Reduced verbosity
+                verbose=True,
                 allow_dangerous_code=True,
                 extra_tools=[tool],
                 return_intermediate_steps=True
@@ -652,43 +569,46 @@ def langchain_csv_chart(csv_url: str, question: str, chart_required: bool):
             result = agent.invoke({"input": _prompt_generator(question, True, csv_url)})
             output = result.get("output", "")
 
-            # Clean up matplotlib figures
+            # Close figures to avoid interference
             plt.close('all')
 
-            # Extract chart filenames
+            # Extract chart filenames (assuming extract_chart_filenames returns a list)
             chart_files = extract_chart_filenames(output)
             if len(chart_files) > 0:
-                full_paths = [os.path.join(image_file_path, f) for f in chart_files]
-                logger.info(f"Langchain chart generated: {len(chart_files)} files")
-                return full_paths
+                # Return full paths (join with your image_file_path)
+                return [os.path.join(image_file_path, f) for f in chart_files]
+
+            if attempt < len(groq_api_keys) - 1:
+                logger.info(f"Langchain chart error (key {current_key}): {output}")
 
         except Exception as e:
             error_message = str(e)
-            if "429" in error_message or "rate limit" in error_message.lower():
-                logger.warning(f"Langchain chart rate limit hit on attempt {attempt + 1}")
-                continue
+            if error_message != "":
+                with current_langchain_chart_lock:
+                    current_langchain_chart_key_index = (current_langchain_chart_key_index + 1)
+                    logger.warning(f"Rate limit exceeded. Switching to next API key: {groq_api_keys[current_langchain_chart_key_index]}")
             else:
-                logger.error(f"Langchain chart error: {error_message}")
+                logger.error(f"Error with API key {api_key}: {error_message}")
                 return {"error": error_message}
     
-    logger.error("All API keys exhausted for Langchain chart generation")
+    logger.error("All API keys exhausted for chart generation")
     return "Chart generation failed after all retries"
 
+
+# --- FASTAPI ENDPOINT FOR CHART GENERATION ---
 @app.post("/api/csv-chart")
-@log_performance("csv_chart_generation")
 async def csv_chart(request: dict, authorization: str = Header(None)):
-    """Optimized CSV chart generation endpoint"""
-    chat_id = request.get("chat_id", "unknown")
-    logger.info(f"Chart generation request started for chat_id: {chat_id}")
-    
-    # Optimized authorization check
+    """
+    Endpoint for generating a chart from CSV data.
+    This endpoint uses a ProcessPoolExecutor to run the (CPU-bound) chart generation
+    functions in separate processes so that multiple requests can run in parallel.
+    """
+    # --- Authorization Check ---
     if not authorization or not authorization.startswith("Bearer "):
-        logger.warning("Invalid chart authorization")
         raise HTTPException(status_code=401, detail="Authorization required")
     
-    token = authorization[7:]  # Remove "Bearer " prefix
+    token = authorization.split(" ")[1]
     if token != os.getenv("AUTH_TOKEN"):
-        logger.warning("Chart authentication failed")
         raise HTTPException(status_code=403, detail="Invalid credentials")
 
     try:
@@ -697,261 +617,77 @@ async def csv_chart(request: dict, authorization: str = Header(None)):
         detailed_answer = request.get("detailed_answer", False)
         conversation_history = request.get("conversation_history", [])
         generate_report = request.get("generate_report", False)
+        chat_id = request.get("chat_id", "")
         
-        logger.info(f"Chart request: '{query[:50]}...' (detailed: {detailed_answer})")
-
-        # Report generation
-        if generate_report:
-            logger.info("Generating chart report")
+        if generate_report is True:
             report_files = await generate_csv_report(csv_url, query, chat_id, conversation_history)
             if report_files is not None:
-                logger.info("Chart report generated successfully")
                 return {"orchestrator_response": jsonable_encoder(report_files)}
 
         loop = asyncio.get_running_loop()
-        
-        # Initial chart question processing
+        # First, try the langchain-based method if the question qualifies
         if if_initial_chart_question(query):
-            logger.info("Processing as initial chart question")
             langchain_result = await loop.run_in_executor(
                 process_executor, langchain_csv_chart, csv_url, query, True
             )
-            
+            logger.info("Langchain chart result:", langchain_result)
             if isinstance(langchain_result, list) and len(langchain_result) > 0:
-                unique_file_name = f'{str(uuid.uuid4())}.png'
-                logger.info("Uploading initial chart to Supabase")
-                
-                image_public_url = await upload_file_to_supabase(
-                    langchain_result[0], unique_file_name, chat_id=chat_id
-                )
-                
-                # Clean up local file
-                if os.path.exists(langchain_result[0]):
-                    os.remove(langchain_result[0])
-                    logger.debug("Local chart file cleaned up")
-                
-                logger.info("Initial chart uploaded successfully")
+                unique_file_name =f'{str(uuid.uuid4())}.png'
+                logger.info("Uploading the chart to supabase...")
+                image_public_url = await upload_file_to_supabase(f"{langchain_result[0]}", unique_file_name, chat_id=chat_id)
+                logger.info("Image uploaded to Supabase and Image URL is... ", {image_public_url})
+                os.remove(langchain_result[0])
                 return {"image_url": image_public_url}
+               # return FileResponse(langchain_result[0], media_type="image/png")
                
-        # Detailed answer orchestration
-        if detailed_answer:
-            logger.info("Processing with orchestrator for detailed chart answer")
-            orchestrator_answer = await asyncio.to_thread(
-                csv_orchestrator_chat, csv_url, query, conversation_history, chat_id
-            )
-        
-            if orchestrator_answer is not None:
-                logger.info("Orchestrator chart processing completed")
-                return {"orchestrator_response": jsonable_encoder(orchestrator_answer)}
-        
-        # Cerebras processing
-        logger.info("Trying Cerebras AI for chart generation")
+        # Use orchestrator to handle the user's chart query first
+        if detailed_answer is True:
+           orchestrator_answer = await asyncio.to_thread(
+             csv_orchestrator_chat, csv_url, query, conversation_history, chat_id
+           )
+        
+           if orchestrator_answer is not None:
+             return {"orchestrator_response": jsonable_encoder(orchestrator_answer)}
+        
+        # Next, try the groq-based method
+        # groq_result = await loop.run_in_executor(
+        #     process_executor, groq_chart, csv_url, query
+        # )
+        # logger.info(f"Groq chart result: {groq_result}")
+        # if isinstance(groq_result, str) and groq_result != "Chart not generated":
+        #     unique_file_name =f'{str(uuid.uuid4())}.png'
+        #     logger.info("Uploading the chart to supabase...")
+        #     image_public_url = await upload_file_to_supabase(f"{groq_result}", unique_file_name, chat_id=chat_id)
+        #     logger.info("Image uploaded to Supabase and Image URL is... ", {image_public_url})
+        #     os.remove(groq_result)
+        #     return {"image_url": image_public_url}
+            # return FileResponse(groq_result, media_type="image/png")
+        
+        logger.info("Trying cerebras ai llama...")
         result = await query_csv_agent(csv_url, query, chat_id)
-        
+        logger.info("cerebras ai result ==>", result)
         if result is not None and result != "":
-            logger.info("Cerebras chart processing successful")
-            return {"orchestrator_response": jsonable_encoder(result)}
+         return {"orchestrator_response": jsonable_encoder(result)}
           
-        # Fallback to langchain
-        logger.info("Falling back to Langchain for chart generation")
+        # Fallback: try langchain-based again
+        logger.error("Cerebras ai llama response failed, trying langchain groq....")
         langchain_paths = await loop.run_in_executor(
             process_executor, langchain_csv_chart, csv_url, query, True
         )
-        
+        logger.info("Fallback langchain chart result:", langchain_paths)
         if isinstance(langchain_paths, list) and len(langchain_paths) > 0:
-            unique_file_name = f'{str(uuid.uuid4())}.png'
-            logger.info("Uploading fallback chart to Supabase")
-            
-            image_public_url = await upload_file_to_supabase(
-                langchain_paths[0], unique_file_name, chat_id=chat_id
-            )
-            
-            # Clean up local file
-            if os.path.exists(langchain_paths[0]):
-                os.remove(langchain_paths[0])
-                logger.debug("Fallback chart file cleaned up")
-            
-            logger.info("Fallback chart uploaded successfully")
-            return {"image_url": image_public_url}
+           unique_file_name =f'{str(uuid.uuid4())}.png'
+           logger.info("Uploading the chart to supabase...")
+           image_public_url = await upload_file_to_supabase(f"{langchain_paths[0]}", unique_file_name, chat_id=chat_id)
+           logger.info("Image uploaded to Supabase and Image URL is... ", {image_public_url})
+           os.remove(langchain_paths[0])
+           return {"image_url": image_public_url}
+           return FileResponse(langchain_paths[0], media_type="image/png")
         else:
-            logger.error("All chart generation methods failed")
-            return {"answer": "error"}
+           logger.error("All chart generation methods failed")
+           return {"answer": "error"}
 
     except Exception as e:
-        logger.error(f"Critical chart generation error: {str(e)}")
+        logger.error(f"Critical chart error: {str(e)}")
         return {"answer": "error"}
 
-# Additional optimizations and middleware
-
-@app.on_event("startup")
-async def startup_event():
-    """Application startup tasks"""
-    logger.info("=== CSV Chat Service Starting Up ===")
-    logger.info(f"Environment: {os.getenv('ENVIRONMENT', 'development')}")
-    logger.info(f"Available API keys: {len(groq_api_keys)}")
-    logger.info(f"Model: {model_name}")
-    logger.info(f"Max workers: {max_cpus-2}")
-    logger.info("=== Startup Complete ===")
-
-@app.on_event("shutdown")
-async def shutdown_event():
-    """Application shutdown cleanup"""
-    logger.info("=== CSV Chat Service Shutting Down ===")
-    try:
-        # Clean up process executor
-        process_executor.shutdown(wait=True)
-        logger.info("ProcessPoolExecutor shutdown complete")
-        
-        # Clean up matplotlib
-        plt.close('all')
-        logger.info("Matplotlib figures cleaned up")
-        
-    except Exception as e:
-        logger.error(f"Error during shutdown: {str(e)}")
-    
-    logger.info("=== Shutdown Complete ===")
-
-# Health check with detailed status
-@app.get("/api/health")
-async def health_check():
-    """Detailed health check endpoint"""
-    try:
-        health_status = {
-            "status": "healthy",
-            "timestamp": time.time(),
-            "service": "CSV Chat Service",
-            "version": "1.0.0",
-            "system": {
-                "cpu_count": max_cpus,
-                "workers": max_cpus-2,
-                "api_keys_available": len(groq_api_keys),
-                "model": model_name
-            },
-            "directories": {
-                "cache": os.path.exists("/app/cache"),
-                "charts": os.path.exists("/app/generated_charts"),
-                "app": os.path.exists("/app")
-            }
-        }
-        
-        logger.debug("Health check completed successfully")
-        return health_status
-        
-    except Exception as e:
-        logger.error(f"Health check failed: {str(e)}")
-        return {
-            "status": "unhealthy", 
-            "error": str(e),
-            "timestamp": time.time()
-        }
-
-# API key status endpoint (for monitoring)
-@app.get("/api/key-status")
-async def api_key_status(authorization: str = Header(None)):
-    """Monitor API key usage status"""
-    # Quick auth check
-    if not authorization or authorization.replace("Bearer ", "") != os.getenv("AUTH_TOKEN"):
-        raise HTTPException(status_code=401, detail="Unauthorized")
-    
-    try:
-        status = {
-            "groq_keys": {
-                "total": len(groq_api_keys),
-                "current_index": groq_key_manager.current_index,
-                "exhausted": groq_key_manager.exhausted
-            },
-            "langchain_keys": {
-                "total": len(groq_api_keys),
-                "current_index": langchain_key_manager.current_index,
-                "exhausted": langchain_key_manager.exhausted
-            },
-            "chart_keys": {
-                "total": len(groq_api_keys),
-                "current_index": chart_key_manager.current_index,
-                "exhausted": chart_key_manager.exhausted
-            },
-            "timestamp": time.time()
-        }
-        
-        logger.debug("API key status retrieved")
-        return status
-        
-    except Exception as e:
-        logger.error(f"Error getting key status: {str(e)}")
-        return {"error": str(e)}
-
-# Reset API keys endpoint (for emergency reset)
-@app.post("/api/reset-keys")
-async def reset_api_keys(authorization: str = Header(None)):
-    """Reset all API key managers"""
-    if not authorization or authorization.replace("Bearer ", "") != os.getenv("AUTH_TOKEN"):
-        raise HTTPException(status_code=401, detail="Unauthorized")
-    
-    try:
-        groq_key_manager.reset()
-        langchain_key_manager.reset()
-        chart_key_manager.reset()
-        
-        logger.info("All API key managers reset successfully")
-        return {
-            "message": "All API key managers reset",
-            "timestamp": time.time(),
-            "status": "success"
-        }
-        
-    except Exception as e:
-        logger.error(f"Error resetting keys: {str(e)}")
-        return {"error": str(e)}
-
-# Error handler for better error logging
-@app.exception_handler(Exception)
-async def global_exception_handler(request: Request, exc: Exception):
-    """Global exception handler with detailed logging"""
-    logger.error(f"Unhandled exception on {request.method} {request.url}: {str(exc)}")
-    return {"error": "Internal server error", "timestamp": time.time()}
-
-# Optimized utility functions
-def safe_file_cleanup(file_path: str):
-    """Safely remove file with logging"""
-    try:
-        if os.path.exists(file_path):
-            os.remove(file_path)
-            logger.debug(f"File cleaned up: {file_path}")
-            return True
-    except Exception as e:
-        logger.warning(f"Failed to cleanup file {file_path}: {str(e)}")
-    return False
-
-# Memory optimization: Periodic cleanup task
-@app.on_event("startup")
-async def setup_periodic_cleanup():
-    """Setup periodic cleanup of temporary files"""
-    async def cleanup_task():
-        while True:
-            try:
-                await asyncio.sleep(300)  # Run every 5 minutes
-                charts_dir = "/app/generated_charts"
-                if os.path.exists(charts_dir):
-                    # Clean up files older than 1 hour
-                    current_time = time.time()
-                    cleaned_count = 0
-                    
-                    for filename in os.listdir(charts_dir):
-                        file_path = os.path.join(charts_dir, filename)
-                        try:
-                            if os.path.getmtime(file_path) < current_time - 3600:  # 1 hour
-                                os.remove(file_path)
-                                cleaned_count += 1
-                        except Exception:
-                            continue
-                    
-                    if cleaned_count > 0:
-                        logger.info(f"Periodic cleanup: removed {cleaned_count} old chart files")
-                        
-            except Exception as e:
-                logger.error(f"Periodic cleanup error: {str(e)}")
-    
-    # Start cleanup task in background
-    asyncio.create_task(cleanup_task())
-    logger.info("Periodic cleanup task started")
-