lct-hack/backend/app/dialog/slots.py

"""Слот-автомат: какие факты раскрыты, какие вопросы заданы.

**Звонящий не выдаёт информацию сам.** Автомат держит состояние явно и отдаёт
звонящему только раскрытые факты — иначе LLM услужливо назовёт адрес без вопроса
(docs/product/CALL-SIM.md).

Тот же автомат детерминированно отвечает на вопрос «спросил ли оператор про этаж»:
на нём стоят подсказки, полнота опроса в оценке (E1) и аналитика группы.
"""

import re
from dataclasses import dataclass, field

import numpy as np

from app.dialog.embeddings import Embedder
from app.scenarios.schema import ChecklistItem, Fact, Scenario

#: Как решается, что реплика — вопрос пункта чек-листа.
#:
#: Порог по косинусной близости не работает: на multilingual-e5-small настоящие
#: вопросы дают 0.79–0.95, а не-вопросы вроде «Оставайтесь на линии» — до 0.89.
#: Диапазоны перекрываются, и любой порог либо выдаёт адрес на «успокойтесь»,
#: либо не слышит «где вы находитесь?».
#:
#: Работает сравнение с ближайшим соседом: реплика засчитывается пункту, если
#: она ближе к одной из его формулировок, чем к любому не-вопросу из
#: checklists/common.yaml. На отложенных фразах (tests/test_slots_e5.py):
#: 19 из 20 вопросов распознано, 0 из 8 ложных срабатываний.
#:
#: MATCH_MARGIN — насколько ближе к пункту, чем к не-вопросу. Ноль даёт лучшее
#: распознавание при нуле ложных срабатываний; 0.03 теряет уже 5 вопросов из 20.
#: MATCH_FLOOR — нижняя граница на случай сценария без не-вопросов.
MATCH_MARGIN = 0.0
MATCH_FLOOR = 0.75

#: Реплика режется только по границам предложений, и знак остаётся при части.
#: «?» для e5 — сильный признак вопроса: без него «Куда ехать» ближе
#: к «Оставайтесь на линии», чем к пункту про адрес. По запятым и «и» не режем:
#: «Улица, дом?» распадается на обрывки, которые не похожи ни на что.
_SENTENCES = re.compile(r"(?<=[?!.;])\s+")


@dataclass
class TurnResult:
    """Что произошло в одной реплике оператора."""

    text: str
    matched: list[str] = field(default_factory=list)   # пункты чек-листа
    revealed: list[str] = field(default_factory=list)  # факты, раскрытые впервые
    repeated: list[str] = field(default_factory=list)  # факты, спрошенные повторно
    scores: dict[str, float] = field(default_factory=dict)

    @property
    def understood(self) -> bool:
        return bool(self.matched)


class SlotMachine:
    def __init__(
        self,
        scenario: Scenario,
        embedder: Embedder,
        floor: float = MATCH_FLOOR,
        margin: float = MATCH_MARGIN,
    ) -> None:
        self.scenario = scenario
        self.floor = floor
        self.margin = margin
        self._embedder = embedder

        self._facts: dict[str, Fact] = {fact.id: fact for fact in scenario.facts}
        self._items: list[ChecklistItem] = [item for item in scenario.checklist if item.question]

        # Каждая формулировка — отдельная строка; `_owner` помнит, чей это пункт.
        texts: list[str] = []
        self._owner: list[int] = []
        for index, item in enumerate(self._items):
            for text in [item.question, *item.examples]:
                texts.append(text)
                self._owner.append(index)
        self._anchors = embedder.embed(texts)
        self._not_questions = (
            embedder.embed(scenario.not_questions) if scenario.not_questions else None
        )

        # Какой пункт чек-листа какие факты раскрывает. Скрытые факты вопросом
        # не раскрываются никогда — только подходом.
        self._reveals: dict[str, list[str]] = {}
        for item in self._items:
            if item.fact and not self._facts[item.fact].hidden:
                self._reveals.setdefault(item.id, []).append(item.fact)
        for fact in scenario.facts:
            question = fact.reveal_on.question if fact.reveal_on else None
            if question and not fact.hidden and fact.id not in self._reveals.get(question, []):
                self._reveals.setdefault(question, []).append(fact.id)

        self.asked: list[str] = []
        self.revealed: list[str] = []

    # ── реплика оператора ──

    def _clauses(self, text: str) -> list[str]:
        parts = [part.strip() for part in _SENTENCES.split(text.strip()) if part.strip()]
        return parts or [text]

    def hear(self, text: str) -> TurnResult:
        """Сопоставить реплику с чек-листом и раскрыть заслуженные факты.

        Реплика режется на предложения: «Где вы? Есть кто внутри?» — два вопроса,
        и оба должны засчитаться.
        """
        result = TurnResult(text=text)
        clauses = self._embedder.embed(self._clauses(text))
        by_anchor = self._anchors @ clauses.T  # формулировки × части реплики

        # Лучшая близость каждого пункта к каждой части реплики.
        owner = np.array(self._owner)
        item_scores = np.stack(
            [by_anchor[owner == index].max(axis=0) for index in range(len(self._items))]
        )  # пункты × части

        # Насколько каждая часть похожа на не-вопрос.
        if self._not_questions is not None:
            not_question = (self._not_questions @ clauses.T).max(axis=0)
        else:
            not_question = np.full(clauses.shape[0], -1.0)

        # Часть реплики засчитывается только одному, ближайшему пункту:
        # «адрес» не должен заодно раскрыть «кто в квартире».
        winners: dict[int, float] = {}
        for part in range(clauses.shape[0]):
            best = int(np.argmax(item_scores[:, part]))
            score = float(item_scores[best, part])
            if score >= self.floor and score > not_question[part] + self.margin:
                winners[best] = max(score, winners.get(best, -1.0))

        for index, item in enumerate(self._items):
            result.scores[item.id] = float(item_scores[index].max())
            if index not in winners:
                continue
            result.matched.append(item.id)
            if item.id not in self.asked:
                self.asked.append(item.id)
            for fact_id in self._reveals.get(item.id, []):
                if fact_id in self.revealed:
                    if fact_id not in result.repeated:
                        result.repeated.append(fact_id)
                else:
                    self.revealed.append(fact_id)
                    result.revealed.append(fact_id)
        return result

    def reveal_by_approach(self, fact_id: str) -> bool:
        """Скрытый факт раскрывается подходом оператора, а не вопросом.
        Решение «создал ли оператор подход» принимает LLM (temperature=0) —
        автомат только фиксирует результат."""
        fact = self._facts.get(fact_id)
        if fact is None or fact_id in self.revealed:
            return False
        self.revealed.append(fact_id)
        return True

    def invalidate(self, fact_id: str) -> None:
        """Директива «адрес оказался неточным»: оператор обязан переспросить."""
        if fact_id in self.revealed:
            self.revealed.remove(fact_id)

    # ── что видят другие ──

    def revealed_facts(self) -> list[Fact]:
        """Единственное, что уходит в контекст звонящего."""
        return [self._facts[fact_id] for fact_id in self.revealed]

    def unasked(self) -> list[ChecklistItem]:
        """Неотработанные пункты по порядку чек-листа — источник подсказки."""
        return [item for item in self._items if item.id not in self.asked]

    def missing_required(self) -> list[str]:
        """Обязательные факты, которые оператор так и не добыл — основание E1."""
        return [
            fact_id
            for fact_id in self.scenario.ground_truth.required_facts
            if fact_id not in self.revealed
        ]