refactored vss

app/service/vector_service.py

@@ -281,8 +281,11 @@ class VectorService:
         metadata: List[Dict[str, Any]],
         namespace: str
     ):
-        """Store in DuckDB VSS (with corrected schema, no expires_at)"""
+        """Store in DuckDB VSS (with DataFrame fix)"""
         try:
+            import pandas as pd
+        
+            # Build records
             records = []
             for idx, (emb, meta) in enumerate(zip(embeddings, metadata)):
                 content = " ".join([str(v) for v in meta.values() if v])[:1000]
@@ -296,7 +299,13 @@ class VectorService:
                     "created_at": datetime.now().isoformat(),
                 })
         
-            # VSS native upsert - REMOVED expires_at column
+            if not records:
+                return
+        
+            # ✅ FIXED: Convert to DataFrame for DuckDB
+            records_df = pd.DataFrame(records)
+        
+            # Insert using DataFrame
             self.vector_conn.execute("""
                 INSERT INTO vector_store.embeddings 
                 (id, org_id, content, embedding, entity_type, created_at)
@@ -304,37 +313,17 @@ class VectorService:
                     id, org_id, content, 
                     embedding::FLOAT[384],
                     entity_type, created_at
-                FROM records
+                FROM records_df
                 ON CONFLICT (id) DO UPDATE SET
                     embedding = EXCLUDED.embedding,
                     content = EXCLUDED.content,
                     created_at = EXCLUDED.created_at
-            """, [records])
+            """)
         
-            logger.info(f"[✅ VECTOR] VSS: Stored {len(records)} vectors")
+            logger.info(f"[✅ VECTOR] VSS: Stored {len(records_df)} vectors")
         
         except Exception as e:
-            logger.error(f"[❌ VECTOR] VSS error: {e}")
-    def semantic_search(
-        self,
-        query_embedding: List[float],
-        top_k: int = 10,
-        min_score: float = 0.35,
-        days_back: int = 30
-    ) -> List[Dict[str, Any]]:
-        """🔍 VSS-accelerated search: Redis first, then VSS"""
-        redis_results = self._search_redis(query_embedding, top_k, min_score)
-        if redis_results:
-            logger.info(f"[SEARCH] Redis hit: {len(redis_results)} results")
-            return redis_results
-        
-        logger.info("[SEARCH] Redis miss, querying VSS...")
-        vss_results = self._search_vss(query_embedding, top_k, min_score, days_back)
-        
-        if vss_results:
-            self._warm_cache(vss_results[:3])
-        
-        return vss_results
+            logger.error(f"[❌ VECTOR] VSS error: {e}", exc_info=True)
     
     def _search_redis(self, query_emb: List[float], top_k: int, min_score: float) -> List[Dict]:
         """Fast Redis scan (no VSS, manual cosine)"""

app/tasks/analytics_worker.py

@@ -17,6 +17,7 @@ from app.schemas.org_schema import OrgSchema
 from app.service.column_embedding_service import ColumnEmbeddingService
 from app.service.vector_service import VectorService
 from app.engine.kpi_calculators.registry import get_kpi_calculator
+from app.engine.kpi_calculators.registry import get_kpi_calculator_async
 from app.service.embedding_service import EmbeddingService
 
 # Configure logging with request context
@@ -108,12 +109,11 @@ class AnalyticsWorker:
                 name=f"embed-{self.org_id}-{self.source_id}"
             )
             
-            # 🎯 STEP 7: Compute KPIs (CPU-bound, run in thread pool)
-            # REPLACE the KPI calculation block
+        
 
             # 🎯 STEP 7: Compute KPIs (CPU-bound, run in thread pool)
             industry = await self._get_industry()
-            calculator = await get_kpi_calculator(  # ✅ Make it async
+            calculator = await get_kpi_calculator_async(  # ✅ Make it async
                 industry=industry, 
                 org_id=self.org_id, 
                 df=df,