Add notebook with evals

src/modules/evals.py

@@ -6,80 +6,151 @@ from sklearn.metrics import (
     accuracy_score,
     classification_report,
 )
-from tqdm import tqdm
+from tqdm.asyncio import tqdm as async_tqdm
+import asyncio
 
-from src.modules.vlm_inference import analyze_product_image
+from src.modules.vlm_inference import analyze_product_image_async
 from src.modules.data_processing import image_to_base64
+from pathlib import Path
 
+DATA_PATH = Path(__file__).parents[2] / "data"
 
-def run_inference_on_dataframe(
+
+async def _process_single_row(
+    row_data: dict,
+    model: str,
+    api_key: str,
+    provider: str,
+    semaphore: asyncio.Semaphore,
+) -> dict:
+    """
+    Process a single row with semaphore control
+
+    Args:
+        row_data: Dictionary with 'image' and 'id' keys
+        model: Model to use for inference
+        api_key: API key for the provider
+        provider: Provider to use
+        semaphore: Asyncio semaphore for rate limiting
+
+    Returns:
+        dict: Prediction result
+    """
+    async with semaphore:
+        try:
+            img_b64 = image_to_base64(row_data["image"])
+            prediction = await analyze_product_image_async(
+                image_url=img_b64,
+                model=model,
+                api_key=api_key,
+                provider=provider,
+            )
+            return {
+                "id": row_data["id"],
+                "pred_masterCategory": prediction.master_category,
+                "pred_gender": prediction.gender,
+                "pred_subCategory": prediction.sub_category,
+                "pred_description": prediction.description,
+            }
+        except Exception as e:
+            return {
+                "id": row_data["id"],
+                "pred_masterCategory": None,
+                "pred_gender": None,
+                "pred_subCategory": None,
+                "pred_description": f"Error: {str(e)}",
+            }
+
+
+async def run_inference_on_dataframe_async(
     df: pd.DataFrame,
     model: str = "accounts/fireworks/models/qwen2p5-vl-72b-instruct",
     api_key: Optional[str] = None,
     provider: str = "Fireworks",
-    image_col: str = "image",
-    id_col: str = "id",
+    max_concurrent_requests: int = 10,
 ) -> pd.DataFrame:
     """
-    Run VLM inference on entire dataframe of images
+    Run VLM inference on entire dataframe of images with concurrent requests
 
     Args:
         df: DataFrame containing images
         model: Model to use for inference
         api_key: API key for the provider
         provider: Provider to use (Fireworks or OpenAI)
-        image_col: Column name containing images
-        id_col: Column name containing image IDs
+        max_concurrent_requests: Maximum number of concurrent API requests (default: 10)
 
     Returns:
         pd.DataFrame: Results with columns:
             - id: Image ID
-            - pred_master_category: Predicted master category
+            - pred_masterCategory: Predicted master category
             - pred_gender: Predicted gender
-            - pred_sub_category: Predicted sub-category
+            - pred_subCategory: Predicted sub-category
             - pred_description: Predicted description
     """
+    # Create semaphore for rate limiting
+    semaphore = asyncio.Semaphore(max_concurrent_requests)
+
+    # Prepare all rows as dictionaries
+    rows_data = [
+        {"image": row.image, "id": row.id}
+        for row in df.itertuples(index=False, name="columns")
+    ]
+
+    # Create all tasks (coroutines, not awaited yet)
+    tasks = [
+        _process_single_row(row_data, model, api_key, provider, semaphore)
+        for row_data in rows_data
+    ]
+
+    _model = model.split("/")[-1]
+    # Run all tasks concurrently with progress bar
     results = []
+    for task in async_tqdm.as_completed(tasks, total=len(tasks)):
+        result = await task
+        results.append(result)
 
-    for idx, row_id, row_image in tqdm(
-        df.itertuples(index=True, name="columns"),
-        total=len(df),
-        desc="Running inference",
-    ):
-        try:
-            img_b64 = image_to_base64(row_image)
+        if len(results) % 10 == 0:
+            df_pred = pd.DataFrame(results)
+            file_name = DATA_PATH / f"df_pred_{provider}_{_model}.csv"
+            df_pred.to_csv(file_name, index=False)
 
-            prediction = analyze_product_image(
-                image_url=img_b64,
-                model=model,
-                api_key=api_key,
-                provider=provider,
-            )
+    # Final save
+    df_pred = pd.DataFrame(results)
+    file_name = DATA_PATH / f"df_pred_{provider}_{_model}.csv"
+    df_pred.to_csv(file_name, index=False)
 
-            results.append(
-                {
-                    "id": row_id,
-                    "pred_master_category": prediction.master_category,
-                    "pred_gender": prediction.gender,
-                    "pred_sub_category": prediction.sub_category,
-                    "pred_description": prediction.description,
-                }
-            )
+    print(f"\nPrediction successful, dataset saved to {file_name}")
+    return df_pred
 
-        except Exception as e:
-            print(f"Error processing row {idx} (ID: {row_id}): {e}")
-            # Append placeholder for failed predictions
-            results.append(
-                {
-                    "id": row_id,
-                    "pred_master_category": None,
-                    "pred_gender": None,
-                    "pred_sub_category": None,
-                    "pred_description": f"Error: {str(e)}",
-                }
-            )
 
-    return pd.DataFrame(results)
+def run_inference_on_dataframe(
+    df: pd.DataFrame,
+    model: str = "accounts/fireworks/models/qwen2p5-vl-72b-instruct",
+    api_key: Optional[str] = None,
+    provider: str = "Fireworks",
+    max_concurrent_requests: int = 10,
+) -> pd.DataFrame:
+    """
+    Run VLM inference on entire dataframe of images (sync wrapper for async function)
+
+    Args:
+        df: DataFrame containing images
+        model: Model to use for inference
+        api_key: API key for the provider
+        provider: Provider to use (Fireworks or OpenAI)
+        max_concurrent_requests: Maximum number of concurrent API requests (default: 10)
+
+    Returns:
+        pd.DataFrame: Results with columns:
+            - id: Image ID
+            - pred_masterCategory: Predicted master category
+            - pred_gender: Predicted gender
+            - pred_subCategory: Predicted sub-category
+            - pred_description: Predicted description
+    """
+    return asyncio.run(
+        run_inference_on_dataframe_async(df, model, api_key, provider, max_concurrent_requests)
+    )
 
 
 def calculate_metrics(
@@ -225,3 +296,29 @@ def create_evaluation_summary(results: dict) -> pd.DataFrame:
             )
 
     return pd.DataFrame(summary_data)
+
+
+def extract_metrics(results_dict, model_name):
+    """
+    Extract accuracy, precision, and recall for each category.
+
+    Args:
+        results_dict: Dictionary containing evaluation metrics
+        model_name: Name of the model for identification
+
+    Returns:
+        List of dictionaries with metrics per category
+    """
+    metrics_list = []
+
+    for category, metrics in results_dict.items():
+        metrics_list.append({
+            'model': model_name,
+            'category': category,
+            'accuracy': metrics['accuracy'],
+            'precision': metrics['precision'],
+            'recall': metrics['recall'],
+            'num_samples': metrics['num_samples']
+        })
+
+    return metrics_list

src/modules/vlm_inference.py

@@ -1,5 +1,5 @@
 import os
-from openai import OpenAI
+from openai import OpenAI, AsyncOpenAI
 from pydantic import BaseModel, Field
 from typing import Optional, Literal
 
@@ -86,13 +86,12 @@ def analyze_product_image(
     Returns:
         ProductClassification: Structured classification and description
     """
-    if provider == "Fireworks":
-        # Initialize OpenAI client
+    if provider.lower() in ["fireworks", "fireworksai"]:
         client = OpenAI(
             api_key=api_key or os.getenv("FIREWORKS_API_KEY"),
             base_url="https://api.fireworks.ai/inference/v1",
         )
-    elif provider == "OpenAI":
+    elif provider.lower() == "openai":
         client = OpenAI(
             api_key=api_key or os.getenv("OPENAI_API_KEY"),
         )
@@ -119,6 +118,56 @@ def analyze_product_image(
     return completion.choices[0].message.parsed
 
 
+async def analyze_product_image_async(
+    image_url: str,
+    model: str = "accounts/fireworks/models/qwen2p5-vl-72b-instruct",
+    api_key: Optional[str] = None,
+    provider: str = "Fireworks",
+) -> ProductClassification:
+    """
+    Async version of analyze_product_image for concurrent processing
+
+    Args:
+        image_url: URL or base64-encoded image string (with data:image prefix)
+        model: Model to use for inference (default: Qwen2.5 VL 72B)
+        api_key: API key (defaults to provider-specific env variable)
+        provider: Provider to use for inference (default: Fireworks)
+
+    Returns:
+        ProductClassification: Structured classification and description
+    """
+    if provider.lower() in ["fireworks", "fireworksai"]:
+        client = AsyncOpenAI(
+            api_key=api_key or os.getenv("FIREWORKS_API_KEY"),
+            base_url="https://api.fireworks.ai/inference/v1",
+        )
+    elif provider.lower() == "openai":
+        client = AsyncOpenAI(
+            api_key=api_key or os.getenv("OPENAI_API_KEY"),
+        )
+    else:
+        raise ValueError(f"Unknown provider: {provider}")
+
+    # Call the API with structured output
+    completion = await client.beta.chat.completions.parse(
+        model=model,
+        messages=[
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {
+                "role": "user",
+                "content": [
+                    {"type": "image_url", "image_url": {"url": image_url}},
+                    {"type": "text", "text": USER_PROMPT},
+                ],
+            },
+        ],
+        response_format=ProductClassification,
+    )
+
+    # Extract and return the structured output
+    return completion.choices[0].message.parsed
+
+
 def batch_analyze_products(
     image_urls: list[str],
     model: str = "accounts/fireworks/models/qwen2p5-vl-72b-instruct",