Update classify.py

classify.py

@@ -1,5 +1,5 @@
 """
-classify.py — 3-Tier Hybrid Pipeline (V3 — Latency-Tracked)
+classify.py — 3-Tier Hybrid Pipeline (V3 — Latency-Tracked & Parallelized)
 
 Architecture:
   LegacyCRM → LLM directly
@@ -10,11 +10,13 @@ Changes in V3:
   - Pipeline summary with p50/p95 per tier
   - Defensive: LLM timeout + retry baked in via processor_llm
   - classify_logs returns richer result dict
+  - 🚀 Added ThreadPoolExecutor for Parallel LLM Processing (Zero Lag)
 """
 from __future__ import annotations
 import time
 import statistics
 import pandas as pd
+from concurrent.futures import ThreadPoolExecutor
 from processor_regex import classify_with_regex
 from processor_bert  import classify_batch as bert_batch
 from processor_llm   import classify_with_llm
@@ -60,7 +62,6 @@ def classify_logs(logs: list[tuple[str, str]]) -> list[dict]:
     bert_indices  = []
     entry_times   = [time.perf_counter()] * n  # approximate per-log start
 
-    t_route_start = time.perf_counter()
     for i, (source, log_msg) in enumerate(logs):
         entry_times[i] = time.perf_counter()
         if source == LEGACY_SOURCE:
@@ -90,14 +91,22 @@ def classify_logs(logs: list[tuple[str, str]]) -> list[dict]:
             else:
                 llm_indices.append(idx)
 
-    # ── Step 3: LLM (LegacyCRM + BERT fallback) ────────────────────────────
-    for i in llm_indices:
-        _, log_msg = logs[i]
-        t0    = time.perf_counter()
-        label = classify_with_llm(log_msg)
-        t1    = time.perf_counter()
-        tier  = "LLM" if logs[i][0] == LEGACY_SOURCE else "LLM (fallback)"
-        results[i] = _make_result(label, tier, None, (t1 - t0) * 1000)
+    # ── Step 3: LLM (Parallel Concurrency Fix) ──────────────────────────────
+    if llm_indices:
+        def parallel_llm(idx):
+            src, msg = logs[idx]
+            t_llm_0 = time.perf_counter()
+            label = classify_with_llm(msg)
+            t_llm_ms = (time.perf_counter() - t_llm_0) * 1000
+            tier = "LLM" if src == LEGACY_SOURCE else "LLM (fallback)"
+            return idx, _make_result(label, tier, None, t_llm_ms)
+
+        # 🚨 GOOGLE-LEVEL FIX: 20 threads API calls ek saath marenge!
+        with ThreadPoolExecutor(max_workers=20) as executor:
+            llm_results = list(executor.map(parallel_llm, llm_indices))
+
+        for idx, res in llm_results:
+            results[idx] = res
 
     return results
 
@@ -195,4 +204,4 @@ if __name__ == "__main__":
 
     print("\n🏷️  Label distribution:")
     for label, count in sorted(summary["label_counts"].items(), key=lambda x: -x[1]):
-        print(f"  • {label}: {count}")
+        print(f"  • {label}: {count}")