7c97310938
Звонящий не выдаёт данные сам: автомат держит раскрытые факты явно и отдаёт звонящему только их. Тест прогоняет длинную серию реплик и проверяет, что ни один нераскрытый факт не прозвучал. Главная находка — матчинг по порогу близости не работает. На multilingual-e5-small настоящие вопросы дают 0.79–0.95, а «Оставайтесь на линии» — до 0.89: диапазоны перекрываются, и любой порог либо выдаёт адрес на «успокойтесь», либо не слышит «где вы находитесь?». Решение — сравнение с ближайшим соседом: у пункта чек-листа несколько формулировок (examples), рядом общий список не-вопросов (checklists/common.yaml), и реплика засчитывается пункту, только если она ближе к нему, чем к любому не-вопросу. На отложенных фразах: 19 из 20 вопросов, 0 из 8 ложных срабатываний. Реплика режется только по границам предложений, со знаком: «?» для e5 — сильный признак вопроса, без него «Куда ехать» уходит к не-вопросам. Подсказка берёт неотработанный пункт из автомата. В живой сессии автомат начнёт слышать оператора, когда голосовой контур передаст ему stt.final. Ждёт ключа LLM: условие approach у скрытых фактов, звонящий своими словами, dialog/llm.py с кэшем.
72 lines
3.3 KiB
Python
72 lines
3.3 KiB
Python
"""Эмбеддинги для матчинга вопроса оператора к чек-листу.
|
||
|
||
Не regex: «на каком этаже?», «этаж какой?», «а этаж» пришлось бы перечислять
|
||
руками для каждого факта каждого сценария, а сценарии пишут методист и генератор
|
||
(docs/arch/STACK.md). Модель — multilingual-e5-small в ONNX: русский, CPU,
|
||
около 10 мс, и не отнимает процессор у распознавания речи.
|
||
"""
|
||
|
||
from pathlib import Path
|
||
from typing import Protocol
|
||
|
||
import numpy as np
|
||
|
||
|
||
class Embedder(Protocol):
|
||
def embed(self, texts: list[str]) -> np.ndarray:
|
||
"""Матрица нормированных векторов, строка на текст."""
|
||
...
|
||
|
||
|
||
def normalize(vectors: np.ndarray) -> np.ndarray:
|
||
norms = np.linalg.norm(vectors, axis=1, keepdims=True)
|
||
return vectors / np.clip(norms, 1e-12, None)
|
||
|
||
|
||
class E5Embedder:
|
||
"""multilingual-e5-small, квантованный ONNX.
|
||
|
||
Для симметричной задачи «вопрос ↔ вопрос» e5 требует префикс `query: `
|
||
с обеих сторон — без него близость фраз заметно проседает.
|
||
"""
|
||
|
||
PREFIX = "query: "
|
||
MAX_TOKENS = 128 # реплика оператора — одна-две фразы, длиннее не бывает
|
||
|
||
def __init__(self, model_dir: Path) -> None:
|
||
import onnxruntime as ort
|
||
from tokenizers import Tokenizer
|
||
|
||
model_path = model_dir / "model_quantized.onnx"
|
||
tokenizer_path = model_dir / "tokenizer.json"
|
||
if not model_path.exists() or not tokenizer_path.exists():
|
||
raise FileNotFoundError(
|
||
f"нет модели эмбеддингов в {model_dir} — запусти `make models`"
|
||
)
|
||
|
||
self._tokenizer = Tokenizer.from_file(str(tokenizer_path))
|
||
self._tokenizer.enable_truncation(max_length=self.MAX_TOKENS)
|
||
self._tokenizer.enable_padding()
|
||
|
||
options = ort.SessionOptions()
|
||
# Два потока: модель делит процессор с распознаванием и синтезом речи.
|
||
options.intra_op_num_threads = 2
|
||
self._session = ort.InferenceSession(
|
||
str(model_path), sess_options=options, providers=["CPUExecutionProvider"]
|
||
)
|
||
self._inputs = {item.name for item in self._session.get_inputs()}
|
||
|
||
def embed(self, texts: list[str]) -> np.ndarray:
|
||
encoded = self._tokenizer.encode_batch([self.PREFIX + text for text in texts])
|
||
input_ids = np.array([item.ids for item in encoded], dtype=np.int64)
|
||
attention = np.array([item.attention_mask for item in encoded], dtype=np.int64)
|
||
|
||
feed = {"input_ids": input_ids, "attention_mask": attention}
|
||
if "token_type_ids" in self._inputs:
|
||
feed["token_type_ids"] = np.zeros_like(input_ids)
|
||
|
||
hidden = self._session.run(None, feed)[0]
|
||
# Среднее по токенам без паддинга — так e5 обучалась.
|
||
mask = attention[..., None].astype(hidden.dtype)
|
||
pooled = (hidden * mask).sum(axis=1) / np.clip(mask.sum(axis=1), 1e-9, None)
|
||
return normalize(pooled)
|