File size: 2,407 Bytes
ceb51d3
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
from __future__ import annotations

from dataclasses import dataclass
import re
from typing import Any


LOW_CONFIDENCE_THRESHOLD = 0.78


@dataclass(frozen=True)
class RouteResult:
    call_model: bool
    reason: str
    speech: str | None = None
    next_question_id: str | None = None


_TARGETED_CONFIRMATIONS: tuple[tuple[re.Pattern[str], str, str], ...] = (
    (
        re.compile(r"\b(?:dizz|busy|spinn|balance)\w*\b", re.I),
        "The audio may have mentioned dizziness or balance. Are you feeling dizzy—yes or no?",
        "symptom.dizziness.present",
    ),
    (
        re.compile(r"\b(?:breath|breathe|air|chest)\w*\b", re.I),
        "The audio may have mentioned breathing or your chest. Are you having unusual difficulty breathing—yes or no?",
        "symptom.breathing_difficulty.present",
    ),
    (
        re.compile(r"\b(?:watch|wash|wearable|sensor|connect)\w*\b", re.I),
        "The audio may have mentioned your wearable connection. Is your wearable connected right now?",
        "device.connection_status",
    ),
    (
        re.compile(r"\b(?:head|headache|pain)\w*\b", re.I),
        "The audio may have mentioned head pain. Do you have a headache—yes or no?",
        "symptom.headache.present",
    ),
)


def route_request(request: dict[str, Any]) -> RouteResult:
    """Bypass the LLM when the speech signal is not reliable enough to interpret."""
    if request.get("task_type") != "climber_checkin":
        return RouteResult(True, "non_speech_or_robot_request")
    transcript = request.get("transcript")
    if not isinstance(transcript, dict):
        return RouteResult(True, "no_transcript_gate")
    confidence = transcript.get("confidence")
    text = str(transcript.get("text", "")).strip()
    polarity_ambiguous = bool(transcript.get("polarity_ambiguous", False))
    if isinstance(confidence, (int, float)) and confidence >= LOW_CONFIDENCE_THRESHOLD and not polarity_ambiguous:
        return RouteResult(True, "speech_confidence_passed")

    for pattern, speech, question_id in _TARGETED_CONFIRMATIONS:
        if pattern.search(text):
            return RouteResult(False, "low_confidence_targeted_confirmation", speech, question_id)
    return RouteResult(
        False,
        "low_confidence_short_repeat",
        "The audio was unclear. Please repeat only your current symptom or status in one short sentence.",
        None,
    )