perf(api): refactor data insertion to enhance performance

app/api/endpoints/analysis.py

@@ -2,12 +2,14 @@ from typing import Any, List, Dict
 import uuid
 from datetime import datetime
 from fastapi import APIRouter, HTTPException, status, Request
-from trendspy import Trends
 
 import asyncio
 from motor.motor_asyncio import AsyncIOMotorClient
 from bson import ObjectId
 
+import pandas as pd
+from trendspy import Trends
+
 from app.core.config import settings
 from app.core.clients import qstash_client
 from app.schemas.analysis_schema import (
@@ -46,7 +48,7 @@ async def fetch_repr_comments(entity_id):
     if not source_ids:
         return {"positive": [], "neutral": [], "negative": []}
 
-    # Fetch 2 comments for each sentiment
+    # Fetch new comments for each sentiment
     sentiments = ["positive", "neutral", "negative"]
     comment_tasks = []
     limit = settings.REPRESENTATIVE_COMMENTS_LIMIT
@@ -418,7 +420,7 @@ async def process_on_demand_job(request: Request):
         texts_to_predict = [comment.get("text", "") for comment in batch_comments]
 
         # Process one small batch at a time
-        batch_predictions = sentiment_service.predict_batch(texts_to_predict)
+        batch_predictions = sentiment_service.predict(texts_to_predict)
         all_predictions.extend(batch_predictions)
         print(f"  - Processed batch {i // batch_size + 1}...")
 

app/core/config.py

@@ -61,6 +61,9 @@ class Settings(BaseSettings):
     ON_DEMAND_COMMENTS_PER_VIDEO: int = 100
     ON_DEMAND_TOTAL_COMMENTS: int = 500
 
+    # Inference Batch Size
+    INFERENCE_BATCH_SIZE: int = 32
+
     # Pydantic model configuration to load from .env file
     model_config = SettingsConfigDict(
         env_file=".env", env_file_encoding="utf-8", extra="ignore"

app/scripts/consumer_job.py

@@ -41,7 +41,7 @@ def process_message_batch(
         return
 
     # --- 2. Perform Batch Sentiment Analysis ---
-    predictions = sentiment_service.predict_batch(texts_to_predict)
+    predictions = sentiment_service.predict(texts_to_predict)
 
     # --- 3. Save data to Database ---
     video_id_cache: Dict[str, ObjectId] = {}

app/services/sentiment_service.py

@@ -1,8 +1,6 @@
 import os
-from pathlib import Path
 from typing import List, Dict, Any
-
-# from app.core.config import settings
+from app.core.config import settings
 
 import torch
 import numpy as np
@@ -46,12 +44,6 @@ class SentimentService:
             self.device
         )
 
-        # self.tokenizer = AutoTokenizer.from_pretrained(model_source)
-        # self.config = AutoConfig.from_pretrained(model_source)
-        # self.model = AutoModelForSequenceClassification.from_pretrained(
-        #     model_source
-        # ).to(self.device)
-
         self.model.eval()  # set model to inference mode
         print("Sentiment model loaded successfully.")
 
@@ -68,9 +60,10 @@ class SentimentService:
             new_text.append(t)
         return " ".join(new_text)
 
-    def predict_batch(self, texts: List[str]) -> List[Dict[str, Any]]:
+    def predict(self, texts: List[str]) -> List[Dict[str, Any]]:
         """
-        Predict sentiment for a batch of texts (batch size is assumed to be small).
+        Predict sentiment for a batch of texts, splitting into sub-batches
+        for efficiency on CPU.
         """
         # Preprocess all texts
         preprocessed_texts = [self._preprocess_text(text) for text in texts]
@@ -84,34 +77,45 @@ class SentimentService:
 
         indices, texts_to_predict = zip(*non_empty_texts_with_indices)
 
-        # Tokenize the batch
-        encoded_inputs = self.tokenizer(
-            list(texts_to_predict),
-            return_tensors="pt",
-            padding=True,
-            truncation=True,
-            max_length=512,
-        ).to(self.device)
-
-        # Run inference
-        with torch.no_grad():
-            outputs = self.model(**encoded_inputs)
-            logits = outputs.logits.detach().cpu().numpy()
+        # --- Define batch size for CPU ---
+        batch_size = settings.INFERENCE_BATCH_SIZE
 
-        # Explicitly clear intermediate tensors from VRAM
-        del encoded_inputs, outputs
-        torch.cuda.empty_cache()
-
-        # Apply softmax to get probabilities
-        probs = softmax(logits, axis=1)
-
-        # Map predictions to labels with highest probability
         predictions = []
-        for prob in probs:
-            max_idx = int(np.argmax(prob))
-            predictions.append(
-                {"label": self.config.id2label[max_idx], "score": float(prob[max_idx])}
-            )
+        # --- Process in chunks ---
+        for start in range(0, len(texts_to_predict), batch_size):
+            sub_texts = texts_to_predict[start : start + batch_size]
+
+            # Tokenize
+            encoded_inputs = self.tokenizer(
+                list(sub_texts),
+                return_tensors="pt",
+                padding=True,
+                truncation=True,
+                max_length=512,
+            ).to(self.device)
+
+            # Inference
+            with torch.no_grad():
+                outputs = self.model(**encoded_inputs)
+                logits = outputs.logits.detach().cpu().numpy()
+
+            # Clear memory
+            del encoded_inputs, outputs
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+
+            # Softmax + map to labels
+            probs = softmax(logits, axis=1)
+            for prob in probs:
+                max_idx = int(np.argmax(prob))
+                predictions.append(
+                    {
+                        "label": self.config.id2label[max_idx],
+                        "score": float(prob[max_idx]),
+                    }
+                )
+
+            print(f"  - Processed batch {start // batch_size + 1}...")
 
         # Map predictions back to their original positions
         final_results: List[Dict[str, Any] | None] = [None] * len(texts)