185 lines
9.4 KiB
Python
185 lines
9.4 KiB
Python
|
"""Слот-автомат: какие факты раскрыты, какие вопросы заданы.
|
|||
|
|
|
|||
|
|
**Звонящий не выдаёт информацию сам.** Автомат держит состояние явно и отдаёт
|
|||
|
|
звонящему только раскрытые факты — иначе 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
|
|||
|
|
]
|