Enhance ingredient processing with async support and rate limiting; update JWT expiration to 15 days; improve product analysis prompt clarity and scan history timestamp handling.

routers/analysis.py

@@ -1,5 +1,8 @@
+import asyncio
+import os
 from datetime import datetime
 from fastapi import APIRouter, Depends, HTTPException
+import pytz
 from sqlalchemy.orm import Session
 from typing import List, Dict, Any
 from db.models import User
@@ -7,12 +10,23 @@ from interfaces.ingredientModels import IngredientAnalysisResult, IngredientRequ
 from interfaces.productModels import ProductIngredientsRequest
 from services.auth_service import get_current_user
 from logger_manager import log_info, log_error,logger
-from db.database import get_db
+from db.database import get_db,SessionLocal
 from db.repositories import IngredientRepository
+from dotenv import load_dotenv
+from langsmith import traceable
 
 from services.ingredientFinderAgent import IngredientInfoAgentLangGraph
 from services.productAnalyzerAgent import analyze_product_ingredients
 
+# Load environment variables
+load_dotenv()
+
+# Get rate limit from environment variable or use default
+PARALLEL_RATE_LIMIT = int(os.getenv("PARALLEL_RATE_LIMIT", 10))
+log_info(f"Using parallel rate limit of {PARALLEL_RATE_LIMIT}")
+
+# Create a semaphore to limit concurrent API calls
+llm_semaphore = asyncio.Semaphore(PARALLEL_RATE_LIMIT)
 
 router = APIRouter()
 
@@ -31,6 +45,7 @@ def ingredient_db_to_pydantic(db_ingredient):
 
 # process single ingredient 
 @router.post("/process_ingredient", response_model=IngredientAnalysisResult)
+@traceable
 async def process_ingredient_endpoint(request: IngredientRequest, db: Session = Depends(get_db)):
     try:
         logger.info(f"Received request to process ingredient: {request.name}")
@@ -47,7 +62,13 @@ async def process_ingredient_endpoint(request: IngredientRequest, db: Session =
         
         # If not in database, get from agent
         ingredient_finder = IngredientInfoAgentLangGraph()
-        result = ingredient_finder.process_ingredient(request.name)
+        # run async function if found event loop already running use normal function 
+        try:
+            result = await ingredient_finder.process_ingredient_async(request.name)
+        except RuntimeError:
+            # If the event loop is not running, run the function normally
+            result = ingredient_finder.process_ingredient(request.name)
+                    
         
         # Save to database
         repo.create_ingredient(result)
@@ -58,37 +79,77 @@ async def process_ingredient_endpoint(request: IngredientRequest, db: Session =
         logger.error(f"Error processing ingredient: {e}")
         raise HTTPException(status_code=500, detail="Internal Server Error")
 
+async def process_single_ingredient(ingredient_name: str):
+    """Process a single ingredient asynchronously with rate limiting"""
+    log_info(f"Starting processing for ingredient (async): {ingredient_name}")
+    
+    # Create a new DB session for this specific task to avoid conflicts
+    session = SessionLocal()
+    
+    try:
+        # Check if ingredient exists in database
+        repo = IngredientRepository(session)
+        db_ingredient = repo.get_ingredient_by_name(ingredient_name)
+        
+        if db_ingredient:
+            log_info(f"Found existing ingredient in database: {ingredient_name}")
+            ingredient_data = ingredient_db_to_pydantic(db_ingredient)
+            return ingredient_data
+        else:
+            # Apply rate limiting for LLM calls only if not in database
+            async with llm_semaphore:
+                log_info(f"Acquired semaphore for: {ingredient_name}")
+                # Get from agent if not in database
+                log_info(f"Fetching ingredient from agent: {ingredient_name}")
+                # Create a new instance for thread safety
+                ingredient_finder = IngredientInfoAgentLangGraph()
+                
+                # FIXED: Use the async version directly instead of the sync wrapper
+                ingredient_data = await ingredient_finder.process_ingredient_async(ingredient_name)
+                
+                # Save to database for future use
+                repo.create_ingredient(ingredient_data)
+                log_info(f"Saved new ingredient to database: {ingredient_name}")
+                
+                return ingredient_data
+    except Exception as e:
+        log_error(f"Error processing ingredient {ingredient_name}: {str(e)}")
+        # Return a minimal result on error to avoid failing the entire batch
+        return IngredientAnalysisResult(
+            name=ingredient_name,
+            is_found=False,
+            safety_rating=0,
+            description=f"Error during processing: {str(e)}",
+            health_effects=["Error during processing"],
+            allergic_info=[],
+            diet_type="unknown",
+            details_with_source=[]
+        )
+    finally:
+        # Important: Close the session when done
+        session.close()
 
 @router.post("/process_product_ingredients", response_model=Dict[str, Any])
+@traceable
 async def process_ingredients_endpoint(product_ingredient: ProductIngredientsRequest, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
     log_info(f"process_ingredients_endpoint called for {len(product_ingredient.ingredients)} ingredients")
     ingredients = product_ingredient.ingredients
     try:
         # Step 1: Process individual ingredients
         ingredient_results = []
-        ingredient_finder = IngredientInfoAgentLangGraph()
-        repo = IngredientRepository(db)
+            
+        log_info(f"Starting parallel ingredient processing with rate limit {PARALLEL_RATE_LIMIT}")
         
+        # Create tasks for parallel processing
+        tasks = []
         for ingredient_name in ingredients:
-            log_info(f"Processing ingredient: {ingredient_name}")
-            
-            # Check if ingredient exists in database
-            db_ingredient = repo.get_ingredient_by_name(ingredient_name)
-            
-            if db_ingredient:
-                log_info(f"Found existing ingredient in database: {ingredient_name}")
-                ingredient_data = ingredient_db_to_pydantic(db_ingredient)
-            else:
-                # Get from agent if not in database
-                log_info(f"Fetching ingredient from agent: {ingredient_name}")
-                ingredient_data = ingredient_finder.process_ingredient(ingredient_name)
-                
-                # Save to database for future use
-                repo.create_ingredient(ingredient_data)
-                log_info(f"Saved new ingredient to database: {ingredient_name}")
-            
-            ingredient_results.append(ingredient_data)
+            task = process_single_ingredient(ingredient_name)
+            tasks.append(task)
         
+        # Execute tasks concurrently with rate limiting
+        ingredient_results = await asyncio.gather(*tasks)
+        log_info(f"Completed parallel processing of {len(ingredient_results)} ingredients")
+                
         # Step 2: Generate aggregate analysis with product analyzer agent
         
         product_analysis = await analyze_product_ingredients(
@@ -106,7 +167,7 @@ async def process_ingredients_endpoint(product_ingredient: ProductIngredientsReq
             "processed_ingredients": ingredient_results,
             "overall_analysis": product_analysis,
             "user_id": current_user.id if current_user else None,
-            "timestamp": datetime.now().isoformat()
+            "timestamp": datetime.now(tz=pytz.timezone('Asia/Kolkata')).isoformat()
         }
         
         log_info("process_ingredients_endpoint completed successfully")
@@ -114,4 +175,4 @@ async def process_ingredients_endpoint(product_ingredient: ProductIngredientsReq
         
     except Exception as e:
         log_error(f"Error in process_ingredients_endpoint: {str(e)}")
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail="Internal Server Error")

services/auth_service.py

@@ -64,7 +64,7 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None):
         if expires_delta:
             expire = datetime.utcnow() + expires_delta
         else:
-            expire = datetime.utcnow() + timedelta(minutes=15)
+            expire = datetime.utcnow() + timedelta(days==15)
         to_encode.update({"exp": expire})
         encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
         return encoded_jwt

services/ingredientFinderAgent.py

@@ -1,9 +1,13 @@
+import asyncio
+from functools import partial
 import os
 import json
 import traceback
 import requests
 import pandas as pd
 from dotenv import load_dotenv
+import aiohttp
+import time
 
 from typing import Dict, Any
 from langchain_google_genai import ChatGoogleGenerativeAI
@@ -29,6 +33,14 @@ else:
     logger.warning("Scraped database not found!")
 
 
+# Define a rate limit (adjust as needed)
+PUBCHEM_TIMEOUT = float(os.getenv("PUBCHEM_TIMEOUT", "2.0"))   # seconds
+PUBCHEM_MAX_RETRIES = int(os.getenv("PUBCHEM_MAX_RETRIES", "3"))  # Max retries
+
+# Rate limiting configuration
+DUCKDUCKGO_RATE_LIMIT_DELAY = float(os.getenv("DUCKDUCKGO_RATE_LIMIT_DELAY", "2.0"))  # Delay in seconds
+DUCKDUCKGO_MAX_RETRIES = int(os.getenv("DUCKDUCKGO_MAX_RETRIES", "3"))  # Max retries
+
 
 # Define tool functions
 @tool("search_local_db")
@@ -115,37 +127,51 @@ def search_usda(ingredient: str) -> Dict[str, Any]:
         logger.error(f"Error searching USDA: {e}")
         return {"source": "USDA FoodData Central", "found": False, "error": str(e)}
 
-@tool("search_pubchem")
-def search_pubchem(ingredient: str) -> Dict[str, Any]:
-    """Search PubChem for chemical information about the ingredient."""
+async def async_search_pubchem(ingredient: str) -> Dict[str, Any]:
+    """Asynchronously search PubChem for chemical information about the ingredient."""
     logger.info(f"Searching PubChem for: {ingredient}")
     
     try:
         pubchem_api = "https://pubchem.ncbi.nlm.nih.gov/rest/pug"
         
-        # First try to get compound information by name
-        search_url = f"{pubchem_api}/compound/name/{ingredient}/JSON"
-        response = requests.get(search_url, timeout=10)
-        
-        if response.status_code == 200:
-            data = response.json()
-            if "PC_Compounds" in data:
+        async with aiohttp.ClientSession() as session:
+            # First try to get compound information by name
+            search_url = f"{pubchem_api}/compound/name/{ingredient}/JSON"
+            
+            async def fetch_data(url: str, timeout: int = PUBCHEM_TIMEOUT, retry_count: int = 0):
+                try:
+                    async with session.get(url, timeout=timeout) as response:
+                        if response.status == 200:
+                            return await response.json()
+                        else:
+                            logger.warning(f"PubChem returned status: {response.status} for URL: {url}")
+                            return None
+                except asyncio.TimeoutError:
+                    if retry_count < PUBCHEM_MAX_RETRIES:
+                        delay = (2 ** retry_count) * 5  # Exponential backoff
+                        logger.warning(f"PubChem timeout for URL '{url}'. Retrying in {delay:.2f} seconds (attempt {retry_count + 1}/{PUBCHEM_MAX_RETRIES})")
+                        await asyncio.sleep(delay)
+                        return await fetch_data(url, timeout, retry_count + 1)  # Recursive retry
+                    else:
+                        logger.error(f"Max retries reached for PubChem timeout on URL: {url}")
+                        return None
+                except Exception as e:
+                    logger.error(f"PubChem error for URL '{url}': {e}")
+                    return None
+            
+            data = await fetch_data(search_url)
+            
+            if data and "PC_Compounds" in data:
                 compound_id = data["PC_Compounds"][0]["id"]["id"]["cid"]
                 
                 # Get more detailed information using the CID
                 property_url = f"{pubchem_api}/compound/cid/{compound_id}/property/MolecularFormula,MolecularWeight,IUPACName,InChI,InChIKey,CanonicalSMILES/JSON"
-                prop_response = requests.get(property_url, timeout=10)
-                properties_data = None
-                if prop_response.status_code == 200:
-                    properties_data = prop_response.json()
-                    
+                properties_data = await fetch_data(property_url)
+                
                 # Get classifications and categories
                 classification_url = f"{pubchem_api}/compound/cid/{compound_id}/classification/JSON"
-                class_response = requests.get(classification_url, timeout=10)
-                classification_data = None
-                if class_response.status_code == 200:
-                    classification_data = class_response.json()
-                    
+                classification_data = await fetch_data(classification_url)
+                
                 return {
                     "source": "PubChem",
                     "found": True,
@@ -155,13 +181,25 @@ def search_pubchem(ingredient: str) -> Dict[str, Any]:
                         "classification": classification_data
                     }
                 }
-        
-        return {"source": "PubChem", "found": False, "data": None}
+            
+            return {"source": "PubChem", "found": False, "data": None}
     
     except Exception as e:
         logger.error(f"Error searching PubChem: {e}")
         return {"source": "PubChem", "found": False, "error": str(e)}
 
+@tool("search_pubchem")
+def search_pubchem(ingredient: str) -> Dict[str, Any]:
+    """Search PubChem for chemical information about the ingredient."""
+    # Use asyncio.run to handle the async operation from synchronous code
+    try:
+        # For Python 3.7+
+        return asyncio.run(async_search_pubchem(ingredient))
+    except RuntimeError:
+        # If already in an event loop (e.g., in FastAPI)
+        loop = asyncio.get_event_loop()
+        return loop.run_until_complete(async_search_pubchem(ingredient))
+    
 @tool("search_wikipedia")
 def search_wikipedia(ingredient: str) -> Dict[str, Any]:
     """Search Wikipedia for ingredient information."""
@@ -211,6 +249,7 @@ def search_web(ingredient: str) -> Dict[str, Any]:
         search_queries = [f"{ingredient} food ingredient safety", f"{ingredient} E-number food additive",f"{ingredient}'s allergic information",f"is {ingredient} vegan,vegetarian or Non-vegetarian"]
         all_results = []
         for query in search_queries:
+            time.sleep(DUCKDUCKGO_RATE_LIMIT_DELAY)
             result = duckduckgo.run(query)
             if result:
                 all_results.append({"query": query, "result": result})
@@ -386,7 +425,7 @@ def analyze_ingredient(state: IngredientState) -> IngredientState:
         
         # Combine all source texts
         combined_data = "\n\n".join(source_texts)
-        logger.debug(f"Combined data for analysis:\n{combined_data[:500]}...(truncated)")
+        logger.info(f"Combined data for analysis:\n{combined_data[:500]}...(truncated)")
         
         # Create the analysis prompt
         analysis_prompt = f"""
@@ -414,7 +453,7 @@ def analyze_ingredient(state: IngredientState) -> IngredientState:
         - "diet_type" : (string from vegan,vegetarian,non-vegetarian,unknown)
         
         Only include factual information supported by the provided data. If information is 
-        unavailable for any field, use appropriate default values. But if information is "too obvious" you can fill appropriate information.
+        unavailable for any field, use appropriate default values. But if information is too obvious you can fill appropriate information just make sure only relevant data is there in the output.
         """
         
         # Process with LLM
@@ -547,10 +586,93 @@ def format_list_source(source_name: str, source_data: list) -> str:
     return source_text
 
 class IngredientInfoAgentLangGraph:
-    # Add this method to your IngredientInfoAgentLangGraph class
+    async def _fetch_data_from_source(self, tool_func, ingredient: str) -> Dict[str, Any]:
+        """Fetch data from a single source asynchronously."""
+        # Get tool name safely - handle both function tools and structured tools
+        if hasattr(tool_func, "name"):
+            # For structured tools
+            tool_name = tool_func.name
+        elif hasattr(tool_func, "__name__"):
+            # For function tools
+            tool_name = tool_func.__name__
+        else:
+            # Fallback
+            tool_name = str(tool_func).split()[0]
+        
+        source_name = tool_name.replace("search_", "").replace("_", " ").title()
+        logger.info(f"Searching {source_name} for {ingredient}")
+        
+        try:
+            # Run the tool function in a thread pool to avoid blocking
+            loop = asyncio.get_event_loop()
+            result = await loop.run_in_executor(None, partial(tool_func.invoke, ingredient))
+            
+            if result.get("found", False):
+                logger.info(f"{source_name} found data for {ingredient}")
+            return result
+        except Exception as e:
+            logger.error(f"Error in {source_name} search: {e}")
+            return {"source": source_name, "found": False, "error": str(e)}
+    
+    async def process_ingredient_async(self, ingredient: str) -> IngredientAnalysisResult:
+        """Process an ingredient using parallel data fetching."""
+        logger.info(f"=== Parallel processing for: {ingredient} ===")
+        
+        # Define all the tools to run in parallel
+        tools = [
+            search_local_db,
+            search_web,
+            search_wikipedia,
+            search_open_food_facts,
+            search_usda,
+            search_pubchem
+        ]
+        
+        # Create tasks for each tool
+        tasks = [self._fetch_data_from_source(tool, ingredient) for tool in tools]
+        
+        # Run all tasks concurrently and collect results
+        results = await asyncio.gather(*tasks)
+        
+        # Filter for successful results
+        sources_data = [result for result in results if not result.get("error")]
+        
+        # Create a state for analysis
+        state = {
+            "ingredient": ingredient,
+            "sources_data": sources_data,
+            "result": None,
+            "status": "ready_for_analysis",
+            "analysis_done": False,
+            "local_db_checked": True,
+            "web_search_done": True,
+            "wikipedia_checked": True,
+            "open_food_facts_checked": True,
+            "usda_checked": True,
+            "pubchem_checked": True
+        }
+        
+        # Run the analysis with the collected data
+        final_state = analyze_ingredient(state)
+        
+        # Extract the result or create a default
+        if final_state.get("result"):
+            logger.info(f"Analysis complete for {ingredient}")
+            return IngredientAnalysisResult(**final_state["result"])
+        else:
+            logger.info(f"No result in final state for {ingredient}, returning default")
+            return IngredientAnalysisResult(
+                name=ingredient, 
+                is_found=len(sources_data) > 0, 
+                details_with_source=sources_data
+            )
+        
     def process_ingredient(self, ingredient: str) -> IngredientAnalysisResult:
-        """Process an ingredient using direct sequential approach instead of LangGraph."""
-        logger.info(f"=== Direct sequential processing for: {ingredient} ===")
+        """
+        Process an ingredient using direct sequential approach instead of async.
+        This method provides compatibility with synchronous code.
+        """
+        logger.info(f"=== Sequential processing for: {ingredient} ===")
         
         # Initialize empty sources data
         sources_data = []
@@ -580,7 +702,7 @@ class IngredientInfoAgentLangGraph:
             sources_data.append(result)
             logger.info(f"Open Food Facts found data for {ingredient}")
         
-        # Optional - Add these if needed:
+        
         logger.info(f"Searching USDA for {ingredient}")
         result = search_usda.invoke(ingredient)
         if result.get("found", False):
@@ -622,7 +744,7 @@ class IngredientInfoAgentLangGraph:
                 is_found=len(sources_data) > 0, 
                 details_with_source=sources_data
             )
-
+        
 if __name__ == "__main__":
     agent = IngredientInfoAgentLangGraph()
     

services/productAnalyzerAgent.py

@@ -64,21 +64,22 @@ analysis that would be helpful for a consumer viewing this in an AR application.
 ## INGREDIENTS INFORMATION:
 {''.join(ingredients_summary)}
 
+## Also consider the following user preferences:
 {user_context}
 
 ## REQUIRED ANALYSIS:
 1. Overall Safety Score (1-10): Calculate this based on individual ingredient safety scores
-2. Suitable Diet Types: Determine if this product is suitable for vegans, vegetarians, etc.
-3. Allergy Warnings: Flag any potential allergens present
+2. Suitable Diet Types: Determine if this product is for vegan, vegetarian, or Non-Vegetarian
+3. Allergy Warnings: Flag any potential allergens present related to food not more than 5 combine if needed
 4. Usage Recommendations: Provide safe consumption limits or usage guidance
-5. Health Insights: Summarize health benefits and concerns
+5. Health Insights: Summarize health benefits and concerns of the product not more than 3 for each and also focus on health not other aspects, may combine if needed but keep short
 6. Ingredient Interactions: Note any ingredients that may interact when combined
 7. Key Takeaway: A single sentence summarizing if this product is recommended
 
 ## FORMAT YOUR RESPONSE AS JSON:
 {{
   "overall_safety_score": (number between 1-10),
-  "suitable_diet_types": (array of strings like "Vegan", "Vegetarian", etc.),
+  "suitable_diet_types": (strings from "Vegan", "Vegetarian", "Non-Vegetarian"),
   "allergy_warnings": (array of strings),
   "usage_recommendations": (string with specific guidance),
   "health_insights": {{
@@ -89,7 +90,7 @@ analysis that would be helpful for a consumer viewing this in an AR application.
   "key_takeaway": (string)
 }}
 
-Only include factual information based on the provided data. If information is unavailable for any field, use appropriate default values.
+Only include factual information based on the provided data. If information is unavailable for any field, use appropriate default values. If the data required is too obvious then give appropriate answer.
 """
     
     logger.info("Sending product analysis prompt to LLM")

services/scan_history.py

@@ -1,4 +1,5 @@
 from fastapi import HTTPException
+import pytz
 from sqlalchemy.orm import Session
 from db.models import ScanHistory
 from datetime import datetime
@@ -10,7 +11,7 @@ def record_scan(db: Session, user_id: int, product_id: int) -> ScanHistory:
         scan_entry = ScanHistory(
             user_id=user_id,
             product_id=product_id,
-            scan_date=datetime.utcnow()
+            scan_date=datetime.now(tz=pytz.timezone('Asia/Kolkata')),
         )
         db.add(scan_entry)
         db.commit()