feat: onnx intent classifier replacing fuzzy match

config.py

@@ -41,3 +41,9 @@ TTS_BANDPASS_HIGH_HZ = 7000
 TTS_REVERB_WET = 0.20
 TTS_REVERB_DECAY_MS = 100
 TTS_PITCH_SEMITONES = 0
+
+# Семантический матчинг команд через эмбеддинги MiniLM-L6-v2 (ONNX, CPU).
+# Порог — косинусная близость [0..1]. На вход поступает уже отфильтрованная
+# фраза (без алиасов и vа_tbr); 0.45 эмпирически отделяет валидные команды
+# от шума, оставляя запас на разговорные перефразировки.
+INTENT_SIMILARITY_THRESHOLD = 0.45

intent.py

@@ -0,0 +1,79 @@
+import os
+import numpy as np
+import onnxruntime as ort
+from tokenizers import Tokenizer
+
+import config
+
+
+MODEL_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "models", "all-MiniLM-L6-v2")
+MAX_TOKENS = 128
+
+
+def _mean_pool(last_hidden: np.ndarray, attention_mask: np.ndarray) -> np.ndarray:
+    mask = attention_mask.astype(np.float32)[..., None]
+    summed = (last_hidden * mask).sum(axis=1)
+    counts = np.clip(mask.sum(axis=1), a_min=1e-9, a_max=None)
+    return summed / counts
+
+
+def _l2_normalize(x: np.ndarray) -> np.ndarray:
+    norm = np.linalg.norm(x, axis=-1, keepdims=True)
+    return x / np.clip(norm, a_min=1e-12, a_max=None)
+
+
+class IntentClassifier:
+    def __init__(self, model_dir: str = MODEL_DIR):
+        tok_path = os.path.join(model_dir, "tokenizer.json")
+        onnx_path = os.path.join(model_dir, "model.onnx")
+        self._tokenizer = Tokenizer.from_file(tok_path)
+        self._tokenizer.enable_truncation(max_length=MAX_TOKENS)
+        self._tokenizer.enable_padding(pad_id=0, pad_token="[PAD]")
+        so = ort.SessionOptions()
+        so.intra_op_num_threads = max(1, (os.cpu_count() or 2) // 2)
+        so.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
+        self._session = ort.InferenceSession(onnx_path, sess_options=so, providers=["CPUExecutionProvider"])
+        self._input_names = {inp.name for inp in self._session.get_inputs()}
+        self._cache: dict[str, np.ndarray] = {}
+
+    def embed(self, texts: list[str]) -> np.ndarray:
+        if not texts:
+            return np.zeros((0, 384), dtype=np.float32)
+        encs = self._tokenizer.encode_batch(texts)
+        ids = np.array([e.ids for e in encs], dtype=np.int64)
+        mask = np.array([e.attention_mask for e in encs], dtype=np.int64)
+        feeds = {"input_ids": ids, "attention_mask": mask}
+        if "token_type_ids" in self._input_names:
+            feeds["token_type_ids"] = np.zeros_like(ids)
+        outputs = self._session.run(None, feeds)
+        last_hidden = outputs[0]
+        pooled = _mean_pool(last_hidden, mask)
+        return _l2_normalize(pooled).astype(np.float32)
+
+    def prime(self, candidates: dict[str, list[str]]) -> None:
+        self._cache.clear()
+        for cmd_id, phrases in candidates.items():
+            if not phrases:
+                continue
+            self._cache[cmd_id] = self.embed(list(phrases))
+
+    def match(self, utterance: str, candidates: dict[str, list[str]]) -> dict:
+        utterance = (utterance or "").strip()
+        if not utterance:
+            return {"cmd": "", "score": 0.0}
+        for cmd_id, phrases in candidates.items():
+            if cmd_id not in self._cache and phrases:
+                self._cache[cmd_id] = self.embed(list(phrases))
+        query = self.embed([utterance])[0]
+        best_cmd = ""
+        best_score = -1.0
+        for cmd_id, mat in self._cache.items():
+            sims = mat @ query
+            top = float(sims.max()) if sims.size else -1.0
+            if top > best_score:
+                best_score = top
+                best_cmd = cmd_id
+        threshold = getattr(config, "INTENT_SIMILARITY_THRESHOLD", 0.45)
+        if best_score < threshold:
+            return {"cmd": "", "score": max(0.0, best_score)}
+        return {"cmd": best_cmd, "score": max(0.0, best_score)}

main.py

@@ -27,6 +27,7 @@ from rich import print
 
 import config
 import tts
+from intent import IntentClassifier
 
 # some consts
 CDIR = os.getcwd()
@@ -60,6 +61,9 @@ VAD_FRAME_MS = 30
 VAD_FRAME_BYTES = int(samplerate * VAD_FRAME_MS / 1000) * 2
 vad = webrtcvad.Vad(config.VAD_AGGRESSIVENESS)
 
+intent_classifier = IntentClassifier()
+intent_classifier.prime({c: v['phrases'] for c, v in VA_CMD_LIST.items()})
+
 
 def gpt_answer():
     global message_log
@@ -140,9 +144,10 @@ def va_respond(voice: str):
 
     print(cmd)
 
+    min_percent = int(round(config.INTENT_SIMILARITY_THRESHOLD * 100))
     if len(cmd['cmd'].strip()) <= 0:
         return False
-    elif cmd['percent'] < 70 or cmd['cmd'] not in VA_CMD_LIST.keys():
+    elif cmd['percent'] < min_percent or cmd['cmd'] not in VA_CMD_LIST.keys():
         # play("not_found")
         # tts.va_speak("Что?")
         if fuzz.ratio(voice.join(voice.split()[:1]).strip(), "скажи") > 75:
@@ -179,15 +184,9 @@ def filter_cmd(raw_voice: str):
 
 
 def recognize_cmd(cmd: str):
-    rc = {'cmd': '', 'percent': 0}
-    for c, v in VA_CMD_LIST.items():
-        for x in v['phrases']:
-            vrt = fuzz.ratio(cmd, x)
-            if vrt > rc['percent']:
-                rc['cmd'] = c
-                rc['percent'] = vrt
-
-    return rc
+    candidates = {c: v['phrases'] for c, v in VA_CMD_LIST.items()}
+    res = intent_classifier.match(cmd, candidates)
+    return {'cmd': res['cmd'], 'percent': int(round(res['score'] * 100))}
 
 
 def _set_mute(state: int):