fix(cpu-thread): optimize cpu processing for local

agent/classifier.py

@@ -6,9 +6,9 @@
 """
 
 import logging
-import os
+from typing import Any
 
-from engines.local_slm import LocalSLMEngine
+from engines.local_slm import ClassifierEngine
 
 logger = logging.getLogger(__name__)
 
@@ -61,10 +61,10 @@
 _grammar = None
 
 
-def _get_grammar():
+def _get_grammar() -> Any:
     global _grammar
     if _grammar is None:
-        from llama_cpp import LlamaGrammar
+        from llama_cpp import LlamaGrammar  # type: ignore
 
         _grammar = LlamaGrammar.from_string(_GRAMMAR_STR)
     return _grammar
@@ -78,7 +78,7 @@ def classify(prompt: str) -> str:
         logger.debug("Classifier: long context (%d chars) → %s", len(prompt), ROUTE_API_LONG)
         return ROUTE_API_LONG
 
-    engine = LocalSLMEngine.get_instance()
+    engine = ClassifierEngine.get_instance()
     raw = engine.generate(
         prompt=f"Task:\n{prompt}\n\nLabel:",
         system_prompt=_SYSTEM_PROMPT,

engines/local_slm.py

@@ -23,7 +23,9 @@ def get_instance(cls) -> "LocalSLMEngine":
 
     def __init__(self, model_path: str) -> None:
         n_ctx = int(os.environ.get("LOCAL_N_CTX", "2048"))
-        n_threads = int(os.environ.get("LOCAL_N_THREADS", "2"))
+        # Dynamic thread scaling: fully utilize AMD compute instance
+        default_threads = str(max(2, (os.cpu_count() or 4) - 1))
+        n_threads = int(os.environ.get("LOCAL_N_THREADS", default_threads))
         # Default n_gpu_layers to -1 (automatic metal) on Mac/Jupyter, but 0 in production
         # We can dynamically set this via environment variables.
         n_gpu_layers = int(os.environ.get("LOCAL_N_GPU_LAYERS", "0"))
@@ -75,3 +77,23 @@ def generate(self, prompt: str, system_prompt: str = "", max_tokens: int = 250,
             return "__ESCALATE__"
 
         return text
+
+
+class ClassifierEngine(LocalSLMEngine):
+    """
+    Separate singleton to maintain an isolated KV cache for the Classifier.
+    Because llama.cpp uses mmap, weights (2.1GB) are shared in memory with LocalSLMEngine.
+    This entirely eliminates the KV cache thrashing and lock contention between classification and generation.
+    """
+
+    _instance: Optional["ClassifierEngine"] = None
+    _init_lock = threading.Lock()
+
+    @classmethod
+    def get_instance(cls) -> "ClassifierEngine":
+        if cls._instance is None:
+            with cls._init_lock:
+                if cls._instance is None:
+                    model_path = os.environ.get("LOCAL_MODEL_PATH", "models/qwen2.5-3b-instruct-q4_k_m.gguf")
+                    cls._instance = cls(model_path=model_path)
+        return cls._instance

engines/remote_llm.py

@@ -103,6 +103,7 @@ def select_remote_model(category: str) -> str:
 # ---------------------------------------------------------------------------
 class RemoteLLMEngine:
     def __init__(self) -> None:
+        self._semaphore: asyncio.Semaphore | None = None
         self.api_key = os.environ.get("FIREWORKS_API_KEY", "")
         raw_url = os.environ.get("FIREWORKS_BASE_URL", "https://api.fireworks.ai/inference/v1")
 
@@ -184,19 +185,26 @@ async def generate(
             }
 
         timeout = aiohttp.ClientTimeout(total=20, connect=5)
-        async with aiohttp.ClientSession(timeout=timeout) as session:
-            async with session.post(endpoint, headers=headers, json=payload) as response:
-                if response.status != 200:
-                    text = await response.text()
-                    logger.error("Fireworks API error (status %d): %s", response.status, text)
-                    response.raise_for_status()
-
-                data = await response.json()
-                try:
-                    if is_chat:
-                        return str(data["choices"][0]["message"]["content"]).strip()
-                    else:
-                        return str(data["choices"][0]["text"]).strip()
-                except KeyError as e:
-                    logger.error("Response structure: %s", data)
-                    raise e
+        # Use a class-level or module-level semaphore to limit concurrency
+        # Initialize it lazily in an async context since asyncio.Semaphore needs an active event loop
+        if self._semaphore is None:
+            self._semaphore = asyncio.Semaphore(15)
+
+        sem: asyncio.Semaphore = self._semaphore
+        async with sem:
+            async with aiohttp.ClientSession(timeout=timeout) as session:
+                async with session.post(endpoint, headers=headers, json=payload) as response:
+                    if response.status != 200:
+                        text = await response.text()
+                        logger.error("Fireworks API error (status %d): %s", response.status, text)
+                        response.raise_for_status()
+
+                    data = await response.json()
+                    try:
+                        if is_chat:
+                            return str(data["choices"][0]["message"]["content"]).strip()
+                        else:
+                            return str(data["choices"][0]["text"]).strip()
+                    except KeyError as e:
+                        logger.error("Response structure: %s", data)
+                        raise e