working ingredient analysis finetune needed

data/Food_Aditives_E_numbers.csv

@@ -1,4 +1,4 @@
-E Numbers,Name of Aditive,Description,Example Use
+E Numbers,Name of Additive,Description,Example Use
 E100,Curcumin,Naturally occurring orange/yellow colour Extracted from the spice turmeric,"Used in pastries, confectionery, sauces, and soups"
 E101,Riboflavin or lactoflavin (Vitamin B2),Naturally occurring B-group vitamin Usually obtained from yeast or produced synthetically,Enrichment and fortification of food Added to processed cheese as yellow/orange colour
 E102,Tartrazine,Widely used yellow/orange colour,"Found in soft drinks, cakes, biscuits, puddings, meat products, sauces, tinned and packet convenience foods and confectionery"

interfaces/productModels.py

@@ -0,0 +1,6 @@
+from typing import List
+from pydantic import BaseModel
+
+# Add this class to define the request body structure
+class ProductIngredientsRequest(BaseModel):
+    ingredients: List[str]

routers/analysis.py

@@ -1,13 +1,17 @@
+from datetime import datetime
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 from typing import List, Dict, Any
+from db.models import User
 from interfaces.ingredientModels import IngredientAnalysisResult, IngredientRequest
+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.repositories import IngredientRepository
 
 from services.ingredientFinderAgent import IngredientInfoAgentLangGraph
+from services.productAnalyzerAgent import analyze_product_ingredients
 
 
 router = APIRouter()
@@ -55,15 +59,59 @@ async def process_ingredient_endpoint(request: IngredientRequest, db: Session =
         raise HTTPException(status_code=500, detail="Internal Server Error")
 
 
-# @router.post("/process_ingredients")
-# def process_ingredients_endpoint(ingredients: List[str], db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
-#     log_info("process_ingredients_endpoint called")
-#     print(ingredients)
-#     try:
-#         # result = process_ingredients(db, ingredients, current_user.id)
-#         result = None
-#         log_info("process_ingredients_endpoint completed successfully")
-#         return result
-#     except Exception as e:
-#         log_error(f"Error in process_ingredients_endpoint: {str(e)}")
-#         raise HTTPException(status_code=500, detail=str(e))
+@router.post("/process_product_ingredients", response_model=Dict[str, Any])
+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)
+        
+        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)
+        
+        # Step 2: Generate aggregate analysis with product analyzer agent
+        
+        product_analysis = await analyze_product_ingredients(
+            ingredients_data=ingredient_results,
+            user_preferences={
+                "user_id": current_user.id,
+                "allergies": current_user.preferences[0].allergens if current_user.preferences else None,
+                "dietary_restrictions": current_user.preferences[0].dietary_restrictions if current_user.preferences else None
+            } if current_user else {}
+        )
+        
+        # Step 3: Prepare final response
+        result = {
+            "ingredients_count": len(ingredients),
+            "processed_ingredients": ingredient_results,
+            "overall_analysis": product_analysis,
+            "user_id": current_user.id if current_user else None,
+            "timestamp": datetime.now().isoformat()
+        }
+        
+        log_info("process_ingredients_endpoint completed successfully")
+        return result
+        
+    except Exception as e:
+        log_error(f"Error in process_ingredients_endpoint: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))

services/ingredientFinderAgent.py

@@ -36,7 +36,7 @@ def search_local_db(ingredient: str) -> Dict[str, Any]:
     """Search local database for ingredient information. E number database scrapped"""
     logger.info(f"Searching local DB for: {ingredient}")
     if additives_df is not None:
-        match = additives_df[additives_df['Name of Aditive'].str.contains(ingredient, case=False, na=False)]
+        match = additives_df[additives_df['Name of Additive'].str.contains(ingredient, case=False, na=False, regex=False)]
         if not match.empty:
             return {"source": "Local DB", "found": True, "data": match.iloc[0].to_dict()}
     return {"source": "Local DB", "found": False, "data": None}

services/productAnalyzerAgent.py

@@ -0,0 +1,179 @@
+import os
+from typing import List, Dict, Any, Optional
+from dotenv import load_dotenv
+from langchain_core.messages import HumanMessage
+from langchain_google_genai import ChatGoogleGenerativeAI
+from logger_manager import logger
+from interfaces.ingredientModels import IngredientAnalysisResult
+
+# Load environment variables
+load_dotenv()
+
+async def analyze_product_ingredients(
+    ingredients_data: List[IngredientAnalysisResult],
+    user_preferences: Optional[Dict[str, Any]] = None
+) -> Dict[str, Any]:
+    """
+    Analyze multiple ingredients to provide a comprehensive product analysis
+    for AR display, considering user preferences and dietary restrictions.
+    """
+    logger.info(f"Analyzing product with {len(ingredients_data)} ingredients")
+    
+    # Initialize LLM
+    api_key = os.getenv("LLM_API_KEY")
+    model_name = os.getenv("LLM_MODEL_NAME", "gemini-2.0-flash")
+    
+    llm = ChatGoogleGenerativeAI(
+        google_api_key=api_key,
+        model=model_name,
+        temperature=0.2  # Lower temperature for more factual responses
+    )
+    
+    # Prepare ingredient data for the prompt
+    ingredients_summary = []
+    for i, ingredient in enumerate(ingredients_data):
+        ingredient_info = f"""
+Ingredient {i+1}: {ingredient.name}
+Safety Rating: {ingredient.safety_rating}/10
+Diet Type: {ingredient.diet_type if hasattr(ingredient, 'diet_type') else 'Unknown'}
+Allergic Info: {', '.join(ingredient.allergic_info) if hasattr(ingredient, 'allergic_info') and ingredient.allergic_info else 'None known'}
+Health Effects: {', '.join(ingredient.health_effects) if ingredient.health_effects else 'Unknown'}
+Description: {ingredient.description[:200] + '...' if len(ingredient.description) > 200 else ingredient.description}
+"""
+        ingredients_summary.append(ingredient_info)
+    
+    # Add user preferences context if available
+    user_context = ""
+    if user_preferences:
+        allergies = user_preferences.get("allergies", "None specified")
+        diet = user_preferences.get("dietary_restrictions", "None specified")
+        user_context = f"""
+User has the following preferences:
+- Dietary Restrictions: {diet}
+- Allergies: {allergies}
+"""
+    
+    # Create the analysis prompt
+    analysis_prompt = f"""
+# PRODUCT INGREDIENT ANALYSIS TASK
+
+You are an expert food scientist and nutritionist analyzing a product's ingredients.
+Based on the detailed information about each ingredient below, provide a comprehensive
+analysis that would be helpful for a consumer viewing this in an AR application.
+
+## INGREDIENTS INFORMATION:
+{''.join(ingredients_summary)}
+
+{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
+4. Usage Recommendations: Provide safe consumption limits or usage guidance
+5. Health Insights: Summarize health benefits and concerns
+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.),
+  "allergy_warnings": (array of strings),
+  "usage_recommendations": (string with specific guidance),
+  "health_insights": {{
+    "benefits": (array of strings),
+    "concerns": (array of strings)
+  }},
+  "ingredient_interactions": (array of strings),
+  "key_takeaway": (string)
+}}
+
+Only include factual information based on the provided data. If information is unavailable for any field, use appropriate default values.
+"""
+    
+    logger.info("Sending product analysis prompt to LLM")
+    
+    try:
+        # Process with LLM
+        message = HumanMessage(content=analysis_prompt)
+        llm_response = llm.invoke([message])
+        analysis_text = llm_response.content
+        
+        # Extract JSON from response
+        import json
+        import re
+        
+        # Find JSON in the response using regex
+        json_match = re.search(r'({.*})', analysis_text.replace('\n', ' '), re.DOTALL)
+        
+        if json_match:
+            try:
+                analysis = json.loads(json_match.group(0))
+                logger.info("Successfully parsed product analysis")
+                return analysis
+            except json.JSONDecodeError as e:
+                logger.error(f"JSON parsing error: {e}")
+                # Return a simplified analysis on error
+                return {
+                    "overall_safety_score": calculate_average_safety(ingredients_data),
+                    "error": "Failed to parse complete analysis",
+                    "ingredient_count": len(ingredients_data),
+                    "key_takeaway": "Analysis error occurred, please check individual ingredients"
+                }
+        else:
+            logger.error("Could not find JSON in LLM response")
+            return {
+                "overall_safety_score": calculate_average_safety(ingredients_data),
+                "error": "Failed to generate structured analysis",
+                "ingredient_count": len(ingredients_data)
+            }
+    
+    except Exception as e:
+        logger.error(f"Error in product analysis: {e}")
+        # Fallback analysis based on simple calculations
+        return generate_fallback_analysis(ingredients_data)
+
+
+def calculate_average_safety(ingredients_data: List[IngredientAnalysisResult]) -> float:
+    """Calculate average safety score from ingredients."""
+    safety_scores = [i.safety_rating for i in ingredients_data if i.safety_rating is not None]
+    if not safety_scores:
+        return 5.0  # Default middle value
+    return round(sum(safety_scores) / len(safety_scores), 1)
+
+
+def generate_fallback_analysis(ingredients_data: List[IngredientAnalysisResult]) -> Dict[str, Any]:
+    """Generate a basic analysis when LLM processing fails."""
+    # Extract known allergens
+    allergens = []
+    for ingredient in ingredients_data:
+        if hasattr(ingredient, 'allergic_info') and ingredient.allergic_info:
+            allergens.extend(ingredient.allergic_info)
+    
+    # Determine diet type based on ingredients
+    diet_types = []
+    all_vegan = all(getattr(i, 'diet_type', '') == 'vegan' for i in ingredients_data 
+                    if hasattr(i, 'diet_type') and i.diet_type)
+    all_vegetarian = all(getattr(i, 'diet_type', '') in ['vegan', 'vegetarian'] 
+                         for i in ingredients_data if hasattr(i, 'diet_type') and i.diet_type)
+    
+    if all_vegan:
+        diet_types.append("Vegan")
+    if all_vegetarian:
+        diet_types.append("Vegetarian")
+    
+    # Calculate safety score
+    safety_score = calculate_average_safety(ingredients_data)
+    
+    return {
+        "overall_safety_score": safety_score,
+        "suitable_diet_types": diet_types,
+        "allergy_warnings": list(set(allergens)),
+        "usage_recommendations": "Please refer to product packaging for usage guidelines",
+        "health_insights": {
+            "benefits": [],
+            "concerns": ["Analysis system encountered an error, please check individual ingredients"]
+        },
+        "key_takeaway": f"Product has {len(ingredients_data)} ingredients with average safety score of {safety_score}/10"
+    }