digiops/realtime_voice_service
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
af2e78ce209da28751317b4e65103a023201360e
realtime_voice_service/core/session.py
T
Konturai DevOps af2e78ce20 gemma
2026-05-02 01:54:00 +05:00

2534 lines
94 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from __future__ import annotations
import asyncio
import contextlib
import datetime
import difflib
import io
import logging
import os
import re
import time
import wave
from collections.abc import AsyncGenerator
from collections.abc import Iterable
from enum import Enum
from realtime_voice_service.core.filler_audio import FillerAudioLibrary
from realtime_voice_service.core.kazakh_names import TOP_KAZAKH_NAMES
from realtime_voice_service.core.vad import BaseVAD, SileroVADDetector
from realtime_voice_service.providers.base import BaseLLM, BaseSTT, BaseSTTStream, BaseTTS, MockLLM, MockSTT, MockTTS
from realtime_voice_service.transports.base import BaseMediaTransport
LOGGER = logging.getLogger("uvicorn.error")
def _read_bounded_int_env(name: str, default: int, *, minimum: int, maximum: int) -> int:
raw = str(os.getenv(name, str(default))).strip()
try:
value = int(float(raw))
except Exception:
value = default
return max(minimum, min(value, maximum))
def _semantic_hold_ms() -> int:
return _read_bounded_int_env("SEMANTIC_ENDPOINTING_HOLD_MS", 700, minimum=200, maximum=2000)
def _filler_delay_ms() -> int:
return _read_bounded_int_env("REALTIME_VOICE_FILLER_DELAY_MS", 250, minimum=0, maximum=3000)
def _tts_chunk_soft_min_chars() -> int:
return _read_bounded_int_env("REALTIME_VOICE_TTS_CHUNK_SOFT_MIN_CHARS", 10, minimum=1, maximum=80)
def _tts_chunk_soft_min_words() -> int:
return _read_bounded_int_env("REALTIME_VOICE_TTS_CHUNK_SOFT_MIN_WORDS", 2, minimum=1, maximum=10)
SEMANTIC_ENDPOINTING_HOLD_MS = _semantic_hold_ms()
FILLER_AUDIO_DELAY_MS = _filler_delay_ms()
TTS_CHUNK_SOFT_MIN_CHARS = _tts_chunk_soft_min_chars()
TTS_CHUNK_SOFT_MIN_WORDS = _tts_chunk_soft_min_words()
_COURTESY_OR_FAREWELL_MARKERS: frozenset[str] = frozenset({
"спасибо",
"благодарю",
"до свидания",
"всего доброго",
"пока",
"до встречи",
"хорошего дня",
"ничего не нужно",
"ничего не надо",
"мне ничего не нужно",
"мне ничего не надо",
"больше ничего",
"это все",
"это всё",
"все понятно",
"всё понятно",
"вопросов нет",
"не надо",
"не нужно",
"goodbye",
"bye",
})
_TERMINAL_DIRECT_FAREWELL_MARKERS: frozenset[str] = frozenset({
"до свидания",
"всего доброго",
"пока",
"до встречи",
"хорошего дня",
"goodbye",
"bye",
})
_TERMINAL_CLOSE_REQUEST_MARKERS: frozenset[str] = frozenset({
"можно завершить",
"можем завершить",
"можете завершать",
"давайте завершим",
"давайте закончим",
"завершить разговор",
"закончить разговор",
"заканчиваем разговор",
"завершайте разговор",
"кладу трубку",
"положу трубку",
})
_TERMINAL_NO_MORE_HELP_MARKERS: frozenset[str] = frozenset({
"ничего не нужно",
"ничего не надо",
"мне ничего не нужно",
"мне ничего не надо",
"больше ничего",
"больше ничего не нужно",
"больше ничего не надо",
"ничего больше не нужно",
"ничего больше не надо",
"это все",
"это всё",
"на этом все",
"на этом всё",
"вопросов нет",
"больше вопросов нет",
})
_FAREWELL_QUERY_MARKERS: frozenset[str] = frozenset({
"вопрос",
"подскаж",
"скажи",
"почему",
"как",
"что",
"где",
"когда",
"можно",
"нужно ли",
"нужно еще",
"нужно ещё",
"еще",
"ещё",
"теперь",
"тогда",
"адрес",
"тариф",
"оператор",
"статус",
"проблем",
"помощ",
"помочь",
"узнать",
"нужен",
"нужна",
"нужны",
})
_FAREWELL_HARD_QUERY_MARKERS: frozenset[str] = frozenset(
marker for marker in _FAREWELL_QUERY_MARKERS if marker not in {"вопрос", "можно"}
)
FAREWELL_RESPONSE_TEXT = "Спасибо за обращение в DigiOps. Всего доброго, до свидания!"
def _contains_voice_phrase(normalized_text: str, phrase: str) -> bool:
phrase_parts = [re.escape(part) for part in phrase.split()]
if not phrase_parts:
return False
phrase_pattern = r"\s+".join(phrase_parts)
return re.search(
rf"(?<![^\W\d_]){phrase_pattern}(?![^\W\d_])",
normalized_text,
flags=re.UNICODE,
) is not None
def _ends_with_voice_phrase(normalized_text: str, phrase: str) -> bool:
phrase_parts = [re.escape(part) for part in phrase.split()]
if not phrase_parts:
return False
phrase_pattern = r"\s+".join(phrase_parts)
return re.search(
rf"(?<![^\W\d_]){phrase_pattern}(?:\s+(?:спасибо|благодарю))?$",
normalized_text,
flags=re.UNICODE,
) is not None
def _is_farewell_transcript(text: str) -> bool:
normalized = _voice_text_key(text)
return any(_contains_voice_phrase(normalized, marker) for marker in _COURTESY_OR_FAREWELL_MARKERS)
def _is_terminal_farewell(text: str) -> bool:
normalized = _voice_text_key(text)
has_direct_farewell = any(
_contains_voice_phrase(normalized, marker)
for marker in _TERMINAL_DIRECT_FAREWELL_MARKERS
)
has_close_request = any(
_contains_voice_phrase(normalized, marker)
for marker in _TERMINAL_CLOSE_REQUEST_MARKERS
)
has_no_more_help = any(
_ends_with_voice_phrase(normalized, marker)
for marker in _TERMINAL_NO_MORE_HELP_MARKERS
)
if not (has_direct_farewell or has_close_request or has_no_more_help):
return False
query_markers = _FAREWELL_QUERY_MARKERS if has_direct_farewell else _FAREWELL_HARD_QUERY_MARKERS
if any(marker in normalized for marker in query_markers):
return False
word_count = len(re.findall(r"[^\W\d_]+(?:[-'][^\W\d_]+)*", normalized, flags=re.UNICODE))
return word_count <= 16
SEMANTIC_CONTINUATION_TOKENS = {
"а",
"в",
"во",
"да",
"до",
"и",
"или",
"к",
"ко",
"на",
"но",
"ну",
"о",
"об",
"от",
"по",
"при",
"с",
"со",
"так",
"то",
"у",
"что",
"чтобы",
"эээ",
"ммм",
}
def _preview_text(text: str, *, limit: int = 160) -> str:
normalized = " ".join(str(text or "").split())
if len(normalized) <= limit:
return normalized
return f"{normalized[:limit]}..."
_KAZAKH_SPECIFIC_LETTERS = set("әғқңөұүһі")
_RUSSIAN_SPECIFIC_LETTERS = set("ыэъё")
_KAZAKH_LANGUAGE_MARKERS = frozenset({
"казахский",
"казахском",
"казахски",
"казакша",
"казак",
"қазақ",
"қазақша",
})
_RUSSIAN_LANGUAGE_MARKERS = frozenset({
"русский",
"русском",
"русски",
"орыс",
"орысша",
})
def _detect_preferred_session_language(text: str) -> str | None:
normalized = _voice_text_key(text)
if not normalized:
return None
words = set(normalized.split())
if words & _KAZAKH_LANGUAGE_MARKERS:
return "kk"
if words & _RUSSIAN_LANGUAGE_MARKERS:
return "ru"
if re.search(r"\bпо\s+русски\b", normalized):
return "ru"
if re.search(r"\bна\s+казахском\b", normalized):
return "kk"
if re.search(r"\bна\s+русском\b", normalized):
return "ru"
return None
def _detect_session_language(text: str) -> str | None:
normalized = str(text or "").strip().lower()
if not normalized:
return None
preferred_language = _detect_preferred_session_language(normalized)
if preferred_language:
return preferred_language
has_kazakh = any(ch in _KAZAKH_SPECIFIC_LETTERS for ch in normalized)
if has_kazakh:
return "kk"
has_russian = any(ch in _RUSSIAN_SPECIFIC_LETTERS for ch in normalized)
if has_russian:
return "ru"
cyrillic_chars = sum(1 for ch in normalized if "а" <= ch <= "я" or ch == "ё")
if cyrillic_chars >= 2:
return "ru"
return None
_TTS_LANGUAGE_CODE_MAP = {
"ru": "ru",
"kk": "kk",
}
_STT_YANDEX_LANGUAGE_MAP = {
"ru": "ru-RU",
"kk": "kk-KZ",
}
_STT_ELEVENLABS_LANGUAGE_MAP = {
"ru": "rus",
"kk": "kaz",
}
def _tts_language_for(session_language: str | None) -> str | None:
if not session_language:
return None
return _TTS_LANGUAGE_CODE_MAP.get(session_language.lower())
def _stt_language_for(session_language: str | None) -> str | None:
if not session_language:
return None
return session_language.lower() if session_language.lower() in {"ru", "kk"} else None
def _audio_duration_ms(audio_bytes: bytes, *, sample_rate_hz: int) -> int:
if not audio_bytes or sample_rate_hz <= 0:
return 0
return int(((len(audio_bytes) // 2) / float(sample_rate_hz)) * 1000.0)
def _audio_byte_count_duration_ms(byte_count: int, *, sample_rate_hz: int) -> int:
if byte_count <= 0 or sample_rate_hz <= 0:
return 0
return int(((byte_count // 2) / float(sample_rate_hz)) * 1000.0)
RU_DIGIT_WORDS = ("ноль", "один", "два", "три", "четыре", "пять", "шесть", "семь", "восемь", "девять")
RU_UNITS_MASCULINE = ("", "один", "два", "три", "четыре", "пять", "шесть", "семь", "восемь", "девять")
RU_UNITS_FEMININE = ("", "одна", "две", "три", "четыре", "пять", "шесть", "семь", "восемь", "девять")
RU_TEENS = (
"десять",
"одиннадцать",
"двенадцать",
"тринадцать",
"четырнадцать",
"пятнадцать",
"шестнадцать",
"семнадцать",
"восемнадцать",
"девятнадцать",
)
RU_TENS = ("", "", "двадцать", "тридцать", "сорок", "пятьдесят", "шестьдесят", "семьдесят", "восемьдесят", "девяносто")
RU_HUNDREDS = (
"",
"сто",
"двести",
"триста",
"четыреста",
"пятьсот",
"шестьсот",
"семьсот",
"восемьсот",
"девятьсот",
)
RU_DAY_ORDINALS = {
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: "тридцать первое",
}
RU_MONTHS_GENITIVE = {
1: "января",
2: "февраля",
3: "марта",
4: "апреля",
5: "мая",
6: "июня",
7: "июля",
8: "августа",
9: "сентября",
10: "октября",
11: "ноября",
12: "декабря",
}
def _ru_plural(value: int, one: str, few: str, many: str) -> str:
value = abs(value) % 100
if 11 <= value <= 19:
return many
last_digit = value % 10
if last_digit == 1:
return one
if 2 <= last_digit <= 4:
return few
return many
def _ru_under_1000(value: int, *, feminine: bool = False) -> list[str]:
value = max(min(int(value), 999), 0)
words: list[str] = []
hundreds = value // 100
if hundreds:
words.append(RU_HUNDREDS[hundreds])
remainder = value % 100
if 10 <= remainder <= 19:
words.append(RU_TEENS[remainder - 10])
return words
tens = remainder // 10
units = remainder % 10
if tens:
words.append(RU_TENS[tens])
if units:
words.append((RU_UNITS_FEMININE if feminine else RU_UNITS_MASCULINE)[units])
return words
def _ru_int_words(value: int) -> str:
value = int(value)
if value == 0:
return RU_DIGIT_WORDS[0]
if value < 0:
return f"минус {_ru_int_words(abs(value))}"
words: list[str] = []
billions = value // 1_000_000_000
if billions:
words.extend(_ru_under_1000(billions))
words.append(_ru_plural(billions, "миллиард", "миллиарда", "миллиардов"))
value %= 1_000_000_000
millions = value // 1_000_000
if millions:
words.extend(_ru_under_1000(millions))
words.append(_ru_plural(millions, "миллион", "миллиона", "миллионов"))
value %= 1_000_000
thousands = value // 1000
if thousands:
words.extend(_ru_under_1000(thousands, feminine=True))
words.append(_ru_plural(thousands, "тысяча", "тысячи", "тысяч"))
value %= 1000
if value:
words.extend(_ru_under_1000(value))
return " ".join(word for word in words if word)
def _ru_digit_sequence(raw: str) -> str:
return " ".join(RU_DIGIT_WORDS[int(char)] for char in raw if char.isdigit())
def _ru_number_words(raw: str) -> str:
compact = str(raw or "").strip().replace(" ", "")
decimal_match = re.fullmatch(r"(\d+)[,.](\d+)", compact)
if decimal_match:
integer_part = _ru_number_words(decimal_match.group(1))
fractional_part = _ru_digit_sequence(decimal_match.group(2))
return f"{integer_part} целых {fractional_part}"
digits = re.sub(r"\D+", "", compact)
if not digits:
return str(raw or "")
if digits.startswith("0") or len(digits) > 6:
return _ru_digit_sequence(digits)
return _ru_int_words(int(digits))
def _ru_genitive_number_words(raw: str) -> str:
digits = re.sub(r"\D+", "", str(raw or ""))
if not digits:
return _ru_number_words(raw)
value = int(digits)
small_genitive = {
1: "одного",
2: "двух",
3: "трех",
4: "четырех",
5: "пяти",
6: "шести",
7: "семи",
8: "восьми",
9: "девяти",
10: "десяти",
}
return small_genitive.get(value, _ru_number_words(raw))
def _ru_percent_phrase(raw: str) -> str:
words = _ru_number_words(raw)
compact = str(raw or "").strip().replace(" ", "")
if re.fullmatch(r"\d+[,.]\d+", compact):
return f"{words} процента"
digits = re.sub(r"\D+", "", compact)
if not digits or digits.startswith("0") or len(digits) > 6:
return f"{words} процентов"
value = int(digits)
return f"{words} {_ru_plural(value, 'процент', 'процента', 'процентов')}"
def _ru_date_words(day_raw: str, month_raw: str, year_raw: str) -> str:
day = int(day_raw)
month = int(month_raw)
year = int(year_raw)
day_words = RU_DAY_ORDINALS.get(day, _ru_number_words(str(day)))
month_words = RU_MONTHS_GENITIVE.get(month, _ru_number_words(str(month)))
return f"{day_words} {month_words} {_ru_number_words(str(year))}"
VOICE_PRONUNCIATION_REPLACEMENTS: tuple[tuple[str, str], ...] = (
(r"\bDigiOps\b", "ДиджиОпс"),
(r"\bA\.?I\.?\b", "эй-ай"),
(r"\bAI\b", "эй-ай"),
(r"\bCRM\b", "си-ар-эм"),
(r"\bB2B\b", "би-ту-би"),
(r"\bIT\b", "ай-ти"),
)
def _normalize_voice_pronunciation(text: str) -> str:
normalized = str(text or "")
for pattern, replacement in VOICE_PRONUNCIATION_REPLACEMENTS:
normalized = re.sub(pattern, replacement, normalized, flags=re.IGNORECASE)
return normalized
def _ru_phone_words(raw: str) -> str:
payload = str(raw or "")
digits = re.sub(r"\D+", "", payload)
if len(digits) < 7:
return _ru_number_words(payload)
prefix = "плюс " if payload.strip().startswith("+") else ""
return f"{prefix}{_ru_digit_sequence(digits)}".strip()
def _normalize_voice_numbers(text: str) -> str:
normalized = str(text or "")
normalized = re.sub(
r"(?<!\w)(\+?\d(?:[\s()\-]*\d){6,})(?!\w)",
lambda match: _ru_phone_words(match.group(1)),
normalized,
)
normalized = re.sub(
r"(?<!\w)(\d{1,2})[.](\d{1,2})[.](\d{2,4})(?!\w)",
lambda match: _ru_date_words(match.group(1), match.group(2), match.group(3)),
normalized,
)
normalized = re.sub(
r"(?<!\w)(\d+(?:[,.]\d+)?)\s*%",
lambda match: _ru_percent_phrase(match.group(1)),
normalized,
)
normalized = re.sub(
r"(?<!\w)(\d+)\s*:\s*(\d+)(?!\w)",
lambda match: f"{_ru_number_words(match.group(1))} {_ru_number_words(match.group(2))}",
normalized,
)
normalized = re.sub(
r"(?<!\w)(\d+[,.]\d+)\s*/\s*(\d+)(?!\w)",
lambda match: f"{_ru_number_words(match.group(1))} из {_ru_genitive_number_words(match.group(2))}",
normalized,
)
normalized = re.sub(
r"(?<!\w)(\d+)\s*/\s*(\d+)(?!\w)",
lambda match: f"{_ru_number_words(match.group(1))} из {_ru_genitive_number_words(match.group(2))}",
normalized,
)
normalized = re.sub(
r"(?<!\w)(\d+[,.]\d+)(?!\w)",
lambda match: _ru_number_words(match.group(1)),
normalized,
)
normalized = re.sub(
r"(?<!\w)(\d+)(?!\w)",
lambda match: _ru_number_words(match.group(1)),
normalized,
)
return normalized
def _combine_user_transcripts(parts: Iterable[str]) -> str:
cleaned = [re.sub(r"\s+", " ", str(part or "")).strip() for part in parts]
cleaned = [part for part in cleaned if part]
if len(cleaned) <= 1:
return cleaned[0] if cleaned else ""
normalized_parts: list[str] = []
for part in cleaned:
if normalized_parts and not normalized_parts[-1].endswith((".", "!", "?", "…")):
normalized_parts[-1] = f"{normalized_parts[-1]}."
normalized_parts.append(part)
return " ".join(normalized_parts)
def _sanitize_voice_text(text: str) -> str:
sanitized = str(text or "")
sanitized = re.sub(r"\[([^\]]+)\]\((?:https?://|www\.)[^)\s]+[^)]*\)", r"\1", sanitized)
sanitized = re.sub(r"(?:https?://|www\.)\S+", "", sanitized)
sanitized = sanitized.replace("[", "").replace("]", "").replace("(", "").replace(")", "")
sanitized = _normalize_voice_pronunciation(sanitized)
sanitized = re.sub(r"\s+", " ", sanitized).strip()
sanitized = _normalize_voice_numbers(sanitized)
sanitized = re.sub(r"\s+", " ", sanitized).strip()
return sanitized
DEFAULT_NAME_RETRY_TEXT = "Подскажите, пожалуйста, как я могу к вам обращаться?"
DEFAULT_PERSONALIZED_GREETING_TEMPLATE = '{name}, я работаю на основе искусственного интеллекта и постараюсь максимально внимательно разобраться с вашим вопросом. Чем я могу помочь?'
KAZAKH_NAME_RETRY_TEXT = "Өтінемін, сізге қалай жүгінсем болады?"
KAZAKH_PERSONALIZED_GREETING_TEMPLATE = "{name}, мен жасанды интеллект негізінде жұмыс істеймін және сұрағыңызды мұқият түсінуге тырысамын. Қалай көмектесе аламын?"
_NAME_CAPTURE_CONFUSION_ALIASES = {
"парень": "Арнур",
"арнер": "Арнур",
"ар нур": "Арнур",
"шомарт": "Жомарт",
"шумарт": "Жомарт",
}
def _voice_text_key(text: str | None) -> str:
compact = re.sub(r"[^\w\s]+", " ", str(text or "").lower(), flags=re.UNICODE)
return re.sub(r"\s+", " ", compact).strip()
_KAZAKH_NAME_BY_KEY = {_voice_text_key(name): name for name in TOP_KAZAKH_NAMES}
def _voice_letters_key(text: str | None) -> str:
return "".join(re.findall(r"[^\W\d_]+", str(text or "").lower(), flags=re.UNICODE))
def _canonical_name(text: str) -> str:
words = [part for part in re.split(r"\s+", text.strip()) if part]
return " ".join(word[:1].upper() + word[1:].lower() if len(word) > 1 else word.upper() for word in words)
def _looks_like_supported_name_text(text: str) -> bool:
return bool(
re.fullmatch(
r"[A-Za-zА-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]+(?:[ -][A-Za-zА-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]+){0,2}",
str(text or "").strip(),
)
)
def _correct_name_capture_candidate(name: str | None) -> str | None:
candidate = _canonical_name(str(name or "").strip())
if not candidate:
return None
key = _voice_text_key(candidate)
exact_match = _KAZAKH_NAME_BY_KEY.get(key)
if exact_match:
return exact_match
alias_match = _NAME_CAPTURE_CONFUSION_ALIASES.get(key)
if alias_match:
return alias_match
best_name = None
best_ratio = 0.0
for gazetteer_key, gazetteer_name in _KAZAKH_NAME_BY_KEY.items():
ratio = difflib.SequenceMatcher(None, key, gazetteer_key).ratio()
if ratio > best_ratio:
best_ratio = ratio
best_name = gazetteer_name
if best_name is None:
return candidate
threshold = 0.82 if len(key) >= 5 else 0.9
if best_ratio >= threshold:
LOGGER.info(
"name capture candidate corrected via Kazakh names: candidate=%r corrected=%r ratio=%.3f",
candidate,
best_name,
best_ratio,
)
return best_name
return candidate
def _normalize_name_candidate(text: str | None) -> str | None:
raw = str(text or "").strip(" \t\r\n,.;:!?\"'()[]{}")
if not raw or any(ch.isdigit() for ch in raw):
return None
words = re.findall(r"[^\W\d_]+(?:[-'][^\W\d_]+)*", raw, flags=re.UNICODE)
if not words or len(words) > 4:
return None
lowered = [_voice_text_key(word) for word in words]
stop_words = {
"меня",
"зовут",
"это",
"я",
"мое",
"моё",
"имя",
"менің",
"атым",
"аты",
"болады",
"язык",
"тіл",
}
stop_words.update(_KAZAKH_LANGUAGE_MARKERS)
stop_words.update(_RUSSIAN_LANGUAGE_MARKERS)
filtered = [word for word, lowered_word in zip(words, lowered) if lowered_word not in stop_words]
if not filtered:
return None
if len(filtered) == 2 and len(filtered[1]) == 1:
filtered = [filtered[0]]
elif len(filtered) == 2 and len(filtered[0]) == 1:
filtered = [filtered[1]]
invalid_tokens = {
"да",
"нет",
"алло",
"привет",
"здравствуйте",
"добрый",
"день",
"хочу",
"хотел",
"узнать",
"помощь",
"вопрос",
"тариф",
"статус",
"оператор",
"проблема",
"интернет",
"click",
"noise",
"beep",
"tap",
"tick",
"hum",
"um",
"uh",
"umm",
"угу",
"ага",
"эм",
"эмм",
"эммм",
"мм",
"ммм",
"мммм",
"м",
"ум",
"ээ",
"эээ",
"ээээ",
"аа",
"ааа",
"аааа",
"хм",
"хмм",
"хммм",
"звони",
"слушаю",
"слушаюсь",
"хорошо",
"подожди",
"керек",
"сұрақ",
"көмек",
}
invalid_tokens.update(_KAZAKH_LANGUAGE_MARKERS)
invalid_tokens.update(_RUSSIAN_LANGUAGE_MARKERS)
invalid_letter_tokens = {_voice_letters_key(token) for token in invalid_tokens}
filler_letters = {"а", "э", "е", "ё", "у", "о", "ы", "м", "m", "h"}
for word in filtered:
token_key = _voice_text_key(word)
letters_key = _voice_letters_key(word)
if token_key in invalid_tokens or letters_key in invalid_letter_tokens:
return None
if len(letters_key) >= 2 and len(set(letters_key)) == 1 and letters_key[0] in filler_letters:
return None
if letters_key in {"хм", "хмм", "хммм"}:
return None
if len(filtered) > 2:
return None
if len(filtered) == 2:
similarity = difflib.SequenceMatcher(
None,
_voice_text_key(filtered[0]),
_voice_text_key(filtered[1]),
).ratio()
if similarity >= 0.72:
return None
candidate = _canonical_name(" ".join(filtered))
if not _looks_like_supported_name_text(candidate):
return None
return candidate
def _name_followup_needed(text: str) -> bool:
normalized = _voice_text_key(text)
request_markers = (
"хотел",
"хочу",
"нужн",
"помог",
"вопрос",
"проблем",
"тариф",
"статус",
"оператор",
"адрес",
"график",
"интернет",
"керек",
"сұра",
"көмек",
)
return any(marker in normalized for marker in request_markers)
def _extract_name_candidate(text: str | None) -> tuple[str | None, bool]:
raw = str(text or "").strip()
if not raw:
return None, False
if any(marker in raw for marker in ("[", "]", "<", ">")):
return None, False
normalized = _voice_text_key(raw)
explicit_patterns = (
r"(?:меня\s+зовут|мо[её]\s+имя|my name is|i am|this is)\s+(.+)",
r"(?:менің\s+атым|аты[мң]?|mening atym)\s+(.+)",
)
cutoff_tokens = {
"мне",
"надо",
"нужно",
"хочу",
"хотел",
"узнать",
"график",
"работы",
"адрес",
"филиал",
"город",
"тариф",
"статус",
"оператор",
"вопрос",
"проблема",
"интернет",
"керек",
"сұра",
"көмек",
}
cutoff_tokens.update(_KAZAKH_LANGUAGE_MARKERS)
cutoff_tokens.update(_RUSSIAN_LANGUAGE_MARKERS)
for pattern in explicit_patterns:
match = re.search(pattern, normalized, flags=re.IGNORECASE)
if not match:
continue
tail = match.group(1).strip()
tail_words = re.findall(r"[^\W\d_]+(?:[-'][^\W\d_]+)*", tail, flags=re.UNICODE)
candidate_words: list[str] = []
for word in tail_words:
if _voice_text_key(word) in cutoff_tokens:
break
candidate_words.append(word)
if len(candidate_words) >= 3:
break
candidate = _normalize_name_candidate(" ".join(candidate_words)) or _normalize_name_candidate(tail)
if candidate:
return candidate, False
candidate = _normalize_name_candidate(raw)
if candidate:
lower_candidate = candidate.lower()
stopwords = {
"здравствуйте",
"привет",
"алло",
"да",
"нет",
"добрый",
"день",
"вопрос",
"интернет",
"у",
"меня",
}
candidate_words = set(lower_candidate.split())
if candidate_words.issubset(stopwords) or len(lower_candidate) < 2:
return None, False
word_count = len(re.findall(r"[^\W\d_]+(?:[-'][^\W\d_]+)*", raw, flags=re.UNICODE))
if len(candidate.split()) == 1 and len(candidate) < 3:
return None, False
if word_count <= 2 and not _name_followup_needed(raw):
return candidate, False
return candidate, True
return None, False
def _voice_start_name_outcome(text: str | None) -> tuple[str, str | None]:
raw = str(text or "").strip()
if not raw:
return "name_not_obtained", None
candidate, needs_followup = _extract_name_candidate(raw)
if candidate:
corrected_candidate = _correct_name_capture_candidate(candidate) or candidate
if corrected_candidate != candidate:
LOGGER.info(
"name capture candidate normalized: candidate=%r corrected=%r transcript=%r",
candidate,
corrected_candidate,
_preview_text(raw),
)
candidate = corrected_candidate
if candidate and not needs_followup:
return "name_obtained", candidate
if candidate:
return "name_followup_required", candidate
if _name_followup_needed(raw):
return "name_followup_required", None
return "name_not_obtained", None
def _voice_short_name(name: str | None) -> str | None:
canonical = _normalize_name_candidate(name)
if not canonical:
raw = str(name or "").strip()
if not raw:
return None
canonical = _canonical_name(raw)
short_name = canonical.split(" ", 1)[0].strip()
return short_name or None
class SessionState(str, Enum):
LISTENING = "LISTENING"
USER_SPEAKING = "USER_SPEAKING"
ASSISTANT_THINKING = "ASSISTANT_THINKING"
ASSISTANT_SPEAKING = "ASSISTANT_SPEAKING"
class GenerationInterrupted(RuntimeError):
pass
class TextChunker:
_HARD_BOUNDARY_CHARS = ".!?\n"
_SOFT_BOUNDARY_CHARS = ",:"
_SOFT_BOUNDARY_TOKENS = ("\u2014",)
def __init__(
self,
*,
soft_min_chars: int = TTS_CHUNK_SOFT_MIN_CHARS,
soft_min_words: int = TTS_CHUNK_SOFT_MIN_WORDS,
) -> None:
self._buffer = ""
self._soft_min_chars = max(soft_min_chars, 1)
self._soft_min_words = max(soft_min_words, 1)
def feed(self, fragment: str) -> list[str]:
if not fragment:
return []
self._buffer += fragment
return self._extract_ready_chunks()
def flush(self) -> list[str]:
chunks = self._extract_ready_chunks()
tail = self._buffer.strip()
self._buffer = ""
if tail:
chunks.append(tail)
return chunks
def reset(self) -> None:
self._buffer = ""
def _extract_ready_chunks(self) -> list[str]:
chunks: list[str] = []
while True:
next_chunk = self._pop_next_chunk()
if next_chunk is None:
break
chunks.append(next_chunk)
return chunks
def _pop_next_chunk(self) -> str | None:
for index, char in enumerate(self._buffer):
if char in self._HARD_BOUNDARY_CHARS:
return self._consume_chunk(index + 1)
if char in self._SOFT_BOUNDARY_CHARS or char in self._SOFT_BOUNDARY_TOKENS:
candidate = self._buffer[: index + 1].strip()
if self._is_soft_chunk_ready(candidate):
return self._consume_chunk(index + 1)
return None
def _consume_chunk(self, end_index: int) -> str | None:
while end_index < len(self._buffer) and self._buffer[end_index] in self._HARD_BOUNDARY_CHARS:
end_index += 1
while end_index < len(self._buffer) and self._buffer[end_index].isspace():
end_index += 1
chunk = self._buffer[:end_index].strip()
self._buffer = self._buffer[end_index:]
return chunk or None
def _is_soft_chunk_ready(self, candidate: str) -> bool:
if len(candidate) < self._soft_min_chars:
return False
return len(candidate.split()) >= self._soft_min_words
class CallSession:
def __init__(
self,
*,
session_id: str,
transport: BaseMediaTransport,
vad: BaseVAD | None = None,
stt: BaseSTT | None = None,
llm: BaseLLM | None = None,
tts: BaseTTS | None = None,
filler_audio: FillerAudioLibrary | None = None,
initial_greeting_text: str | None = None,
initial_greeting_segments: Iterable[tuple[str, str | None]] | None = None,
) -> None:
self.session_id = session_id
self.transport = transport
self.state = SessionState.LISTENING
self.generation_epoch = 0
self.interruptions: list[str] = []
self.last_latency_ms: dict[str, int] = {}
self._vad = vad or SileroVADDetector(sample_rate_hz=transport.sample_rate_hz)
self._stt = stt or MockSTT(sample_rate_hz=transport.sample_rate_hz)
self._llm = llm or MockLLM()
self._tts = tts or MockTTS(sample_rate_hz=transport.sample_rate_hz)
self._conversation: list[tuple[str, str]] = []
self._pending_unanswered_texts: list[str] = []
self._pending_unanswered_audio: list[bytes] = []
self._active_turn_epoch: int | None = None
self._active_user_audio: bytes | None = None
self._active_user_transcript: str | None = None
self._active_answer_audio_started = False
self._active_unanswered_captured = False
self._assistant_task: asyncio.Task[None] | None = None
self._filler_task: asyncio.Task[None] | None = None
self._greeting_task: asyncio.Task[None] | None = None
self._sentence_queue: asyncio.Queue[str | None] | None = None
self._closed = False
self._filler_audio = filler_audio
self._initial_greeting_segments = self._normalize_initial_greeting_segments(
initial_greeting_segments,
fallback_text=initial_greeting_text,
)
self._initial_greeting_text = _combine_user_transcripts(text for text, _ in self._initial_greeting_segments)
self._initial_greeting_discarded_bytes = 0
self._awaiting_customer_name = bool(self._initial_greeting_segments)
self._customer_name: str | None = None
self._name_retry_text = DEFAULT_NAME_RETRY_TEXT
self._personalized_greeting_template = DEFAULT_PERSONALIZED_GREETING_TEMPLATE
self._live_stt_stream: BaseSTTStream | None = None
self._latest_partial_transcript = ""
self._session_language: str | None = None
self._default_vad_silence_timeout_ms = getattr(self._vad, "default_speech_end_silence_ms", 550)
self._semantic_hold_silence_timeout_ms = max(self._default_vad_silence_timeout_ms, SEMANTIC_ENDPOINTING_HOLD_MS)
LOGGER.info(
"realtime session %s initialized: protocol=%s sample_rate=%s frame_ms=%s frame_bytes=%s "
"initial_greeting=%s default_vad_silence_ms=%s semantic_hold_ms=%s",
self.session_id,
self.transport.protocol,
self.transport.sample_rate_hz,
self.transport.frame_duration_ms,
self.transport.frame_bytes,
bool(self._initial_greeting_text),
self._default_vad_silence_timeout_ms,
self._semantic_hold_silence_timeout_ms,
)
@staticmethod
def _normalize_initial_greeting_segments(
segments: Iterable[tuple[str, str | None]] | None,
*,
fallback_text: str | None,
) -> tuple[tuple[str, str | None], ...]:
normalized_segments: list[tuple[str, str | None]] = []
for text, language_code in segments or ():
normalized_text = str(text or "").strip()
if not normalized_text:
continue
normalized_language = str(language_code or "").strip() or None
normalized_segments.append((normalized_text, normalized_language))
if normalized_segments:
return tuple(normalized_segments)
normalized_fallback = str(fallback_text or "").strip()
if normalized_fallback:
return ((normalized_fallback, None),)
return ()
@property
def conversation(self) -> tuple[tuple[str, str], ...]:
return tuple(self._conversation)
async def run(self) -> None:
if self._assistant_task is not None:
raise RuntimeError("CallSession.run() can only be called once per session")
LOGGER.info("realtime session %s run started", self.session_id)
try:
self._start_initial_greeting()
await self.media_loop()
finally:
LOGGER.info("realtime session %s run stopping", self.session_id)
await self.stop()
async def media_loop(self) -> None:
while not self._closed:
audio_chunk = await self.transport.receive_audio()
if audio_chunk is None:
LOGGER.info("realtime session %s transport closed", self.session_id)
break
if not audio_chunk:
continue
if self._initial_greeting_in_progress():
self._initial_greeting_discarded_bytes += len(audio_chunk)
continue
vad_result = self._vad.feed(audio_chunk)
if vad_result.is_speech and self.state != SessionState.USER_SPEAKING:
if self.state in {SessionState.ASSISTANT_THINKING, SessionState.ASSISTANT_SPEAKING}:
self.interrupt("barge-in")
self._set_state(
SessionState.USER_SPEAKING,
reason=f"speech detected prob={vad_result.speech_probability:.3f}",
)
if vad_result.speech_started:
LOGGER.info(
"realtime session %s VAD speech_started: prob=%.3f chunk_bytes=%s current_timeout_ms=%s",
self.session_id,
vad_result.speech_probability,
len(audio_chunk),
getattr(self._vad, "current_speech_end_silence_ms", self._default_vad_silence_timeout_ms),
)
await self._start_live_stt_stream()
elif self._live_stt_stream is not None:
await self._push_live_stt_audio(audio_chunk)
if vad_result.speech_ended and vad_result.utterance_audio:
speech_end_monotonic = time.perf_counter()
LOGGER.info(
"realtime session %s VAD speech_ended: utterance_bytes=%s utterance_ms=%s "
"chunk_bytes=%s current_timeout_ms=%s",
self.session_id,
len(vad_result.utterance_audio),
_audio_duration_ms(vad_result.utterance_audio, sample_rate_hz=self.transport.sample_rate_hz),
len(audio_chunk),
getattr(self._vad, "current_speech_end_silence_ms", self._default_vad_silence_timeout_ms),
)
self._set_state(SessionState.ASSISTANT_THINKING, reason="speech end detected")
live_stt_stream = self._detach_live_stt_stream()
self._start_assistant_turn(
epoch=self.generation_epoch,
stt_stream=live_stt_stream,
utterance_audio=vad_result.utterance_audio,
speech_end_monotonic=speech_end_monotonic,
)
def _initial_greeting_in_progress(self) -> bool:
return self._greeting_task is not None and not self._greeting_task.done()
def interrupt(self, reason: str = "interrupt") -> int:
self.generation_epoch += 1
self.interruptions.append(reason)
if self.state in {SessionState.ASSISTANT_THINKING, SessionState.ASSISTANT_SPEAKING}:
self._capture_unanswered_user_turn(reason=reason)
self._clear_sentence_queue()
self.transport.clear_buffer()
if self._greeting_task is not None and not self._greeting_task.done():
self._greeting_task.cancel()
if self._filler_task is not None and not self._filler_task.done():
self._filler_task.cancel()
if self._assistant_task is not None and not self._assistant_task.done():
self._capture_unanswered_user_turn(reason="superseded by new assistant turn")
self._assistant_task.cancel()
LOGGER.info(
"realtime session %s interrupted: epoch=%s reason=%s state=%s conversation_entries=%s",
self.session_id,
self.generation_epoch,
reason,
self.state.value,
len(self._conversation),
)
return self.generation_epoch
async def stop(self) -> None:
if self._closed:
return
self._closed = True
LOGGER.info(
"realtime session %s stop requested: state=%s epoch=%s interruptions=%s conversation_entries=%s",
self.session_id,
self.state.value,
self.generation_epoch,
self.interruptions,
len(self._conversation),
)
self._vad.reset()
self._reset_semantic_endpointing()
self._clear_sentence_queue()
self.transport.clear_buffer()
await self._cancel_live_stt_stream()
if self._greeting_task is not None and not self._greeting_task.done():
self._greeting_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._greeting_task
if self._filler_task is not None and not self._filler_task.done():
self._filler_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._filler_task
if self._assistant_task is not None and not self._assistant_task.done():
self._assistant_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._assistant_task
await self.transport.close()
async def _close_after_farewell(self) -> None:
if self._closed:
return
self._closed = True
LOGGER.info(
"realtime session %s closing transport after farewell: state=%s epoch=%s conversation_entries=%s",
self.session_id,
self.state.value,
self.generation_epoch,
len(self._conversation),
)
await self._cancel_live_stt_stream()
await self.transport.close()
def _start_assistant_turn(
self,
*,
epoch: int,
stt_stream: BaseSTTStream | None,
utterance_audio: bytes,
speech_end_monotonic: float,
) -> None:
if self._assistant_task is not None and not self._assistant_task.done():
self._capture_unanswered_user_turn(reason="superseded by new assistant turn")
self._assistant_task.cancel()
LOGGER.info(
"realtime session %s assistant turn scheduled: epoch=%s utterance_bytes=%s utterance_ms=%s live_stt=%s",
self.session_id,
epoch,
len(utterance_audio),
_audio_duration_ms(utterance_audio, sample_rate_hz=self.transport.sample_rate_hz),
stt_stream is not None,
)
self._assistant_task = asyncio.create_task(
self._run_assistant_turn(
epoch=epoch,
stt_stream=stt_stream,
utterance_audio=utterance_audio,
speech_end_monotonic=speech_end_monotonic,
),
name=f"{self.session_id}-assistant-{epoch}",
)
async def _run_assistant_turn(
self,
*,
epoch: int,
stt_stream: BaseSTTStream | None,
utterance_audio: bytes,
speech_end_monotonic: float,
) -> None:
sentence_queue: asyncio.Queue[str | None] | None = None
playback_task: asyncio.Task[None] | None = None
assistant_fragments: list[str] = []
actually_spoken_chunks: list[str] = []
assistant_audio_started = [False]
transcript = ""
filler_started = asyncio.Event()
self._set_active_unanswered_turn(epoch=epoch, utterance_audio=utterance_audio)
try:
LOGGER.info(
"realtime session %s assistant turn started: epoch=%s utterance_bytes=%s live_stt=%s",
self.session_id,
epoch,
len(utterance_audio),
stt_stream is not None,
)
transcript = await self._resolve_transcript(
stt_stream=stt_stream,
utterance_audio=utterance_audio,
)
self._ensure_generation(epoch)
self._log_latency(
"stt_latency",
speech_end_monotonic,
epoch=epoch,
message="speech end -> transcript ready",
)
transcript = transcript.strip()
if not transcript:
LOGGER.info("realtime session %s produced empty transcript", self.session_id)
self._set_state(SessionState.LISTENING, reason="empty transcript")
return
pending_unanswered = await self._consume_pending_unanswered_transcripts()
if pending_unanswered:
original_transcript = transcript
transcript = _combine_user_transcripts([*pending_unanswered, transcript])
LOGGER.info(
"realtime session %s merged unanswered user turns: epoch=%s pending=%s "
"current=%r merged=%r",
self.session_id,
epoch,
len(pending_unanswered),
_preview_text(original_transcript),
_preview_text(transcript),
)
self._active_user_transcript = transcript
if self._session_language is None:
detected_language = _detect_session_language(transcript)
if detected_language:
self._session_language = detected_language
LOGGER.info(
"realtime session %s language locked: epoch=%s language=%s transcript=%r",
self.session_id,
epoch,
detected_language,
_preview_text(transcript),
)
LOGGER.info(
"realtime session %s transcript accepted: epoch=%s chars=%s text=%r",
self.session_id,
epoch,
len(transcript),
_preview_text(transcript),
)
self._conversation.append(("user", transcript))
LOGGER.info(
"realtime session %s conversation append user: entries=%s",
self.session_id,
len(self._conversation),
)
if _is_terminal_farewell(transcript):
LOGGER.info(
"realtime session %s farewell detected before assistant response: epoch=%s transcript=%r",
self.session_id,
epoch,
_preview_text(transcript),
)
await self._play_prepared_response(
epoch=epoch,
response_text=FAREWELL_RESPONSE_TEXT,
actually_spoken_chunks=actually_spoken_chunks,
)
self._ensure_generation(epoch)
self._commit_assistant_context(
llm_generated_text=FAREWELL_RESPONSE_TEXT,
actually_spoken_chunks=actually_spoken_chunks,
interrupted=False,
)
await self._close_after_farewell()
return
name_collection_response = self._build_name_collection_response(transcript)
if name_collection_response is not None:
LOGGER.info(
"realtime session %s name collection response prepared: epoch=%s name=%r text=%r",
self.session_id,
epoch,
self._customer_name,
_preview_text(name_collection_response),
)
await self._play_prepared_response(
epoch=epoch,
response_text=name_collection_response,
actually_spoken_chunks=actually_spoken_chunks,
)
self._ensure_generation(epoch)
self._commit_assistant_context(
llm_generated_text=name_collection_response,
actually_spoken_chunks=actually_spoken_chunks,
interrupted=False,
)
self._set_state(SessionState.LISTENING, reason="name collection completed")
return
if _is_terminal_farewell(transcript):
LOGGER.info(
"realtime session %s farewell response prepared: epoch=%s text=%r",
self.session_id,
epoch,
_preview_text(FAREWELL_RESPONSE_TEXT),
)
await self._play_prepared_response(
epoch=epoch,
response_text=FAREWELL_RESPONSE_TEXT,
actually_spoken_chunks=actually_spoken_chunks,
)
self._ensure_generation(epoch)
self._commit_assistant_context(
llm_generated_text=FAREWELL_RESPONSE_TEXT,
actually_spoken_chunks=actually_spoken_chunks,
interrupted=False,
)
await self._close_after_farewell()
return
sentence_queue = asyncio.Queue()
self._sentence_queue = sentence_queue
playback_task = asyncio.create_task(
self._stream_tts_pipeline(
epoch=epoch,
sentence_queue=sentence_queue,
actually_spoken_chunks=actually_spoken_chunks,
answer_audio_started=assistant_audio_started,
),
name=f"{self.session_id}-playback-{epoch}",
)
llm_started_monotonic = time.perf_counter()
chunker = TextChunker()
first_token_seen = False
filler_trigger_task = asyncio.create_task(
self._trigger_filler_audio_after_delay(
epoch=epoch,
tool_name=None,
started_monotonic=llm_started_monotonic,
filler_started=filler_started,
suppress=_is_terminal_farewell(transcript) or _is_farewell_transcript(transcript),
),
name=f"{self.session_id}-filler-delay-{epoch}",
)
async for event in self._llm.generate_stream(
transcript,
self._build_llm_context(),
language_code=self._session_language,
):
self._ensure_generation(epoch)
if event.type == "tool_call_start":
LOGGER.info(
"realtime session %s llm tool_call_start: epoch=%s name=%s tool_call_id=%s filler_started=%s",
self.session_id,
epoch,
event.name,
event.tool_call_id,
filler_started.is_set(),
)
if not filler_trigger_task.done():
filler_trigger_task.cancel()
if not filler_started.is_set():
self._start_filler_audio(
epoch=epoch,
tool_name=event.name,
started_monotonic=llm_started_monotonic,
filler_started=filler_started,
)
continue
token = str(event.content or "")
if not token:
continue
if not first_token_seen:
first_token_seen = True
if not filler_trigger_task.done():
filler_trigger_task.cancel()
self._log_latency(
"ttft",
llm_started_monotonic,
epoch=epoch,
message="llm request -> first token",
)
assistant_fragments.append(token)
await self._enqueue_chunks(
epoch=epoch,
sentence_queue=sentence_queue,
chunks=chunker.feed(token),
)
assistant_text = "".join(assistant_fragments).strip()
LOGGER.info(
"realtime session %s llm stream completed: epoch=%s generated_chars=%s spoken_chunks_so_far=%s",
self.session_id,
epoch,
len(assistant_text),
len(actually_spoken_chunks),
)
if not assistant_text:
await self._finish_sentence_queue(sentence_queue)
LOGGER.warning("realtime session %s llm produced no text", self.session_id)
self._set_state(SessionState.LISTENING, reason="empty llm response")
return
await self._enqueue_chunks(
epoch=epoch,
sentence_queue=sentence_queue,
chunks=chunker.flush(),
)
await self._finish_sentence_queue(sentence_queue)
if playback_task is not None:
await playback_task
self._ensure_generation(epoch)
self._commit_assistant_context(
llm_generated_text=assistant_text,
actually_spoken_chunks=actually_spoken_chunks,
interrupted=False,
)
LOGGER.info(
"realtime session %s assistant turn completed: epoch=%s generated_chars=%s spoken_chunks=%s",
self.session_id,
epoch,
len(assistant_text),
len(actually_spoken_chunks),
)
self._set_state(SessionState.LISTENING, reason="assistant turn completed")
except GenerationInterrupted:
if assistant_audio_started[0] or self._active_answer_audio_started:
self._commit_assistant_context(
llm_generated_text="".join(assistant_fragments).strip(),
actually_spoken_chunks=actually_spoken_chunks,
interrupted=True,
)
else:
self._capture_unanswered_user_turn(reason="stale generation before answer audio")
LOGGER.info(
"realtime session %s ignored stale generation %s",
self.session_id,
epoch,
)
except asyncio.CancelledError:
if assistant_audio_started[0] or self._active_answer_audio_started:
self._commit_assistant_context(
llm_generated_text="".join(assistant_fragments).strip(),
actually_spoken_chunks=actually_spoken_chunks,
interrupted=not self._closed,
)
elif not self._closed:
self._capture_unanswered_user_turn(reason="cancelled before answer audio")
LOGGER.info(
"realtime session %s cancelled generation %s",
self.session_id,
epoch,
)
raise
except Exception:
LOGGER.exception(
"realtime session %s generation %s failed",
self.session_id,
epoch,
)
if epoch == self.generation_epoch and not self._closed:
self._set_state(SessionState.LISTENING, reason="assistant generation failed")
finally:
filler_trigger_task = locals().get("filler_trigger_task")
if filler_trigger_task is not None and not filler_trigger_task.done():
filler_trigger_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await filler_trigger_task
if stt_stream is not None:
with contextlib.suppress(Exception):
await stt_stream.cancel()
self._clear_sentence_queue(sentence_queue)
if self._filler_task is not None and not self._filler_task.done():
self._filler_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._filler_task
self._filler_task = None
if playback_task is not None and not playback_task.done():
playback_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await playback_task
current_task = asyncio.current_task()
if self._assistant_task is current_task:
self._clear_active_unanswered_turn()
self._assistant_task = None
def _set_active_unanswered_turn(self, *, epoch: int, utterance_audio: bytes) -> None:
self._active_turn_epoch = epoch
self._active_user_audio = utterance_audio
self._active_user_transcript = None
self._active_answer_audio_started = False
self._active_unanswered_captured = False
def _clear_active_unanswered_turn(self) -> None:
self._active_turn_epoch = None
self._active_user_audio = None
self._active_user_transcript = None
self._active_answer_audio_started = False
self._active_unanswered_captured = False
def _capture_unanswered_user_turn(self, *, reason: str) -> None:
if self._closed or self._active_unanswered_captured or self._active_answer_audio_started:
return
captured_kind = ""
text = str(self._active_user_transcript or "").strip()
if text:
if self._conversation and self._conversation[-1] == ("user", text):
self._conversation.pop()
LOGGER.info(
"realtime session %s removed unanswered user turn from context: entries=%s text=%r",
self.session_id,
len(self._conversation),
_preview_text(text),
)
if not self._pending_unanswered_texts or self._pending_unanswered_texts[-1] != text:
self._pending_unanswered_texts.append(text)
captured_kind = "text"
elif self._active_user_audio:
self._pending_unanswered_audio.append(self._active_user_audio)
captured_kind = "audio"
else:
return
self._active_unanswered_captured = True
self._active_user_audio = None
self._active_user_transcript = None
LOGGER.info(
"realtime session %s captured unanswered user turn: epoch=%s kind=%s reason=%s "
"pending_texts=%s pending_audio=%s",
self.session_id,
self._active_turn_epoch,
captured_kind,
reason,
len(self._pending_unanswered_texts),
len(self._pending_unanswered_audio),
)
async def _consume_pending_unanswered_transcripts(self) -> list[str]:
pending_texts = [text for text in self._pending_unanswered_texts if text.strip()]
pending_audio = list(self._pending_unanswered_audio)
self._pending_unanswered_texts = []
self._pending_unanswered_audio = []
for audio_bytes in pending_audio:
if not audio_bytes:
continue
self._maybe_dump_audio(audio_bytes)
try:
LOGGER.info(
"realtime session %s resolving pending unanswered audio: bytes=%s ms=%s",
self.session_id,
len(audio_bytes),
_audio_duration_ms(audio_bytes, sample_rate_hz=self.transport.sample_rate_hz),
)
pending_transcript = (
await self._stt.transcribe(
audio_bytes,
keyterms=self._name_capture_keyterms() if self._awaiting_customer_name else None,
language_code=(
self._name_capture_language_code()
if self._awaiting_customer_name
else _stt_language_for(self._session_language)
),
force_batch=self._awaiting_customer_name,
)
).strip()
except Exception:
LOGGER.exception("realtime session %s failed to transcribe pending unanswered audio", self.session_id)
continue
if pending_transcript:
pending_texts.append(pending_transcript)
if pending_texts:
LOGGER.info(
"realtime session %s pending unanswered transcripts ready: count=%s preview=%r",
self.session_id,
len(pending_texts),
_preview_text(_combine_user_transcripts(pending_texts)),
)
return pending_texts
def _name_capture_keyterms(self) -> tuple[str, ...]:
return TOP_KAZAKH_NAMES
def _name_capture_language_code(self) -> str | None:
raw = (
os.getenv("STT_NAME_CAPTURE_LANGUAGE_CODE", "").strip()
or os.getenv("STT_NAME_CAPTURE_LANGUAGE", "").strip()
)
return raw or None
def _build_name_collection_response(self, transcript: str) -> str | None:
if not self._awaiting_customer_name:
return None
if self._session_language is None:
detected_language = _detect_session_language(transcript)
if detected_language:
self._session_language = detected_language
LOGGER.info(
"realtime session %s language locked during name collection: language=%s transcript=%r",
self.session_id,
detected_language,
_preview_text(transcript),
)
status, name_value = _voice_start_name_outcome(transcript)
LOGGER.info(
"realtime session %s name collection outcome: status=%s name=%r transcript=%r",
self.session_id,
status,
name_value,
_preview_text(transcript),
)
if status == "name_obtained" and name_value:
short_name = _voice_short_name(name_value) or str(name_value).strip()
if short_name:
self._customer_name = short_name
self._awaiting_customer_name = False
return self._personalized_greeting_text(short_name)
return self._name_retry_text_for_current_language()
def _name_retry_text_for_current_language(self) -> str:
if self._session_language == "kk":
return KAZAKH_NAME_RETRY_TEXT
return self._name_retry_text
def _personalized_greeting_text(self, name: str) -> str:
if self._session_language == "kk":
return KAZAKH_PERSONALIZED_GREETING_TEMPLATE.format(name=name)
return self._personalized_greeting_template.format(name=name)
def _build_llm_context(self) -> list[object]:
context: list[object] = list(self._conversation)
if self._customer_name:
context.append(
{
"role": "system",
"content": (
f"Имя пользователя: {self._customer_name}. "
"Не повторяй имя в каждом ответе. Используй имя только когда это естественно и полезно: "
"после первого знакомства, при уточнении, важном подтверждении, извинении или прощании. "
"В обычных ответах не обращайся по имени."
),
}
)
return context
async def _play_prepared_response(
self,
*,
epoch: int,
response_text: str,
actually_spoken_chunks: list[str],
) -> None:
sentence_queue: asyncio.Queue[str | None] = asyncio.Queue()
self._sentence_queue = sentence_queue
playback_task = asyncio.create_task(
self._stream_tts_pipeline(
epoch=epoch,
sentence_queue=sentence_queue,
actually_spoken_chunks=actually_spoken_chunks,
),
name=f"{self.session_id}-prepared-playback-{epoch}",
)
try:
chunker = TextChunker()
await self._enqueue_chunks(
epoch=epoch,
sentence_queue=sentence_queue,
chunks=chunker.feed(response_text),
)
await self._enqueue_chunks(
epoch=epoch,
sentence_queue=sentence_queue,
chunks=chunker.flush(),
)
await self._finish_sentence_queue(sentence_queue)
await playback_task
finally:
self._clear_sentence_queue(sentence_queue)
if not playback_task.done():
playback_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await playback_task
def _start_initial_greeting(self) -> None:
if not self._initial_greeting_segments or self._closed:
return
if self._greeting_task is not None and not self._greeting_task.done():
return
self._greeting_task = asyncio.create_task(
self._play_initial_greeting(
epoch=self.generation_epoch,
greeting_text=self._initial_greeting_text,
greeting_segments=self._initial_greeting_segments,
),
name=f"{self.session_id}-greeting-{self.generation_epoch}",
)
LOGGER.info(
"realtime session %s initial greeting scheduled: epoch=%s segments=%s chars=%s text=%r",
self.session_id,
self.generation_epoch,
[(language, len(text)) for text, language in self._initial_greeting_segments],
len(self._initial_greeting_text),
_preview_text(self._initial_greeting_text),
)
async def _play_initial_greeting(
self,
*,
epoch: int,
greeting_text: str,
greeting_segments: Iterable[tuple[str, str | None]],
) -> None:
first_audio_seen = False
started_monotonic = time.perf_counter()
cached_greeting = (
self._filler_audio.pick("initial_greeting", allow_fallback=False)
if self._filler_audio is not None
else None
)
if cached_greeting:
await self._play_cached_initial_greeting(
epoch=epoch,
greeting_text=greeting_text,
pcm_audio=cached_greeting,
started_monotonic=started_monotonic,
)
return
try:
audio_chunk_count = 0
audio_byte_count = 0
for segment_text, language_code in greeting_segments:
async def one_shot_text_stream() -> AsyncGenerator[str, None]:
yield segment_text
LOGGER.info(
"realtime session %s initial greeting segment start: epoch=%s language=%s chars=%s text=%r",
self.session_id,
epoch,
language_code,
len(segment_text),
_preview_text(segment_text),
)
audio_segment_chunk_count = 0
async for audio_chunk in self._tts.synthesize_stream(
one_shot_text_stream(),
language_code=language_code,
):
self._ensure_generation(epoch)
if not audio_chunk:
continue
audio_segment_chunk_count += 1
audio_chunk_count += 1
audio_byte_count += len(audio_chunk)
if not first_audio_seen:
first_audio_seen = True
self._log_latency(
"initial_greeting_ttfa",
started_monotonic,
epoch=epoch,
message="session start -> initial greeting first audio",
)
self._set_state(SessionState.ASSISTANT_SPEAKING, reason="initial greeting started")
await self.transport.send_audio(audio_chunk)
if audio_segment_chunk_count == 0:
LOGGER.warning(
"realtime session %s initial greeting segment produced NO audio: language=%s",
self.session_id,
language_code,
)
else:
LOGGER.info(
"realtime session %s initial greeting segment completed: language=%s chunks=%s",
self.session_id,
language_code,
audio_segment_chunk_count,
)
await self.transport.flush_audio()
self._ensure_generation(epoch)
LOGGER.info(
"realtime session %s initial greeting audio completed: epoch=%s chunks=%s bytes=%s",
self.session_id,
epoch,
audio_chunk_count,
audio_byte_count,
)
if self._initial_greeting_discarded_bytes:
LOGGER.info(
"realtime session %s ignored inbound audio during initial greeting: bytes=%s ms=%s",
self.session_id,
self._initial_greeting_discarded_bytes,
_audio_byte_count_duration_ms(
self._initial_greeting_discarded_bytes,
sample_rate_hz=self.transport.sample_rate_hz,
),
)
self._initial_greeting_discarded_bytes = 0
if not self._conversation or self._conversation[-1] != ("assistant", greeting_text):
self._conversation.append(("assistant", greeting_text))
LOGGER.info(
"realtime session %s conversation append initial greeting: entries=%s",
self.session_id,
len(self._conversation),
)
if self.state == SessionState.ASSISTANT_SPEAKING and self._assistant_task is None:
self._set_state(SessionState.LISTENING, reason="initial greeting completed")
except GenerationInterrupted:
LOGGER.info("realtime session %s initial greeting interrupted", self.session_id)
except asyncio.CancelledError:
raise
except Exception:
LOGGER.exception("realtime session %s initial greeting failed", self.session_id)
if self.state == SessionState.ASSISTANT_SPEAKING and self._assistant_task is None:
self._set_state(SessionState.LISTENING, reason="initial greeting failed")
finally:
current_task = asyncio.current_task()
if self._greeting_task is current_task:
self._greeting_task = None
def _ensure_generation(self, epoch: int) -> None:
if self._closed or epoch != self.generation_epoch:
raise GenerationInterrupted(f"stale generation {epoch}")
def _log_latency(
self,
metric_name: str,
started_monotonic: float,
*,
epoch: int,
message: str,
) -> None:
latency_ms = max(int((time.perf_counter() - started_monotonic) * 1000.0), 0)
self.last_latency_ms[metric_name] = latency_ms
LOGGER.info(
"realtime session %s %s=%sms epoch=%s (%s)",
self.session_id,
metric_name,
latency_ms,
epoch,
message,
)
async def _enqueue_chunks(
self,
*,
epoch: int,
sentence_queue: asyncio.Queue[str | None],
chunks: Iterable[str],
) -> None:
for chunk in chunks:
self._ensure_generation(epoch)
normalized = chunk.strip()
if not normalized:
continue
normalized = _sanitize_voice_text(normalized)
if not normalized:
LOGGER.info(
"realtime session %s skipped non-voice tts chunk after sanitization: epoch=%s raw=%r",
self.session_id,
epoch,
_preview_text(chunk),
)
continue
LOGGER.info(
"realtime session %s enqueue_tts_text: epoch=%s chars=%s preview=%r",
self.session_id,
epoch,
len(normalized),
_preview_text(normalized),
)
await sentence_queue.put(normalized)
async def _finish_sentence_queue(self, sentence_queue: asyncio.Queue[str | None]) -> None:
await sentence_queue.put(None)
async def _stream_tts_pipeline(
self,
*,
epoch: int,
sentence_queue: asyncio.Queue[str | None],
actually_spoken_chunks: list[str],
answer_audio_started: list[bool] | None = None,
) -> None:
first_audio_seen = False
tts_started_monotonic: list[float | None] = [None]
try:
audio_chunk_count = 0
audio_byte_count = 0
async for audio_chunk in self._tts.synthesize_stream(
self._iter_tts_text_chunks(
epoch=epoch,
sentence_queue=sentence_queue,
started_holder=tts_started_monotonic,
actually_spoken_chunks=actually_spoken_chunks,
),
language_code=_tts_language_for(self._session_language),
):
self._ensure_generation(epoch)
if not audio_chunk:
continue
audio_chunk_count += 1
audio_byte_count += len(audio_chunk)
if not first_audio_seen:
first_audio_seen = True
if answer_audio_started is not None:
answer_audio_started[0] = True
self._active_answer_audio_started = True
self._log_latency(
"ttfa",
tts_started_monotonic[0] or time.perf_counter(),
epoch=epoch,
message="tts request -> first audio",
)
self._set_state(SessionState.ASSISTANT_SPEAKING, reason="assistant playback started")
await self.transport.send_audio(audio_chunk)
await self.transport.flush_audio()
LOGGER.info(
"realtime session %s tts pipeline completed: epoch=%s audio_chunks=%s audio_bytes=%s spoken_chunks=%s",
self.session_id,
epoch,
audio_chunk_count,
audio_byte_count,
len(actually_spoken_chunks),
)
finally:
if not self._closed:
self.transport.clear_buffer()
async def _iter_tts_text_chunks(
self,
*,
epoch: int,
sentence_queue: asyncio.Queue[str | None],
started_holder: list[float | None],
actually_spoken_chunks: list[str],
) -> AsyncGenerator[str, None]:
while True:
self._ensure_generation(epoch)
sentence = await sentence_queue.get()
if sentence is None:
return
normalized = sentence.strip()
if not normalized:
continue
await self._wait_for_filler_audio()
if started_holder[0] is None:
started_holder[0] = time.perf_counter()
actually_spoken_chunks.append(normalized)
LOGGER.info(
"realtime session %s tts_text_yield: epoch=%s spoken_index=%s chars=%s preview=%r",
self.session_id,
epoch,
len(actually_spoken_chunks),
len(normalized),
_preview_text(normalized),
)
yield normalized
async def _wait_for_filler_audio(self) -> None:
filler_task = self._filler_task
if filler_task is None or filler_task.done():
return
with contextlib.suppress(asyncio.CancelledError):
await filler_task
async def _trigger_filler_audio_after_delay(
self,
*,
epoch: int,
tool_name: str | None,
started_monotonic: float,
filler_started: asyncio.Event,
suppress: bool = False,
) -> None:
if suppress:
return
if FILLER_AUDIO_DELAY_MS <= 0:
self._start_filler_audio(
epoch=epoch,
tool_name=tool_name,
started_monotonic=started_monotonic,
filler_started=filler_started,
)
return
await asyncio.sleep(FILLER_AUDIO_DELAY_MS / 1000.0)
self._ensure_generation(epoch)
if self._active_answer_audio_started or filler_started.is_set():
return
self._start_filler_audio(
epoch=epoch,
tool_name=tool_name,
started_monotonic=started_monotonic,
filler_started=filler_started,
)
def _start_filler_audio(
self,
*,
epoch: int,
tool_name: str | None,
started_monotonic: float,
filler_started: asyncio.Event,
) -> None:
if filler_started.is_set():
LOGGER.info(
"realtime session %s filler audio skipped: already started for current turn epoch=%s tool=%s",
self.session_id,
epoch,
tool_name or "generic",
)
return
if self._filler_task is not None and not self._filler_task.done():
return
filler_started.set()
LOGGER.info(
"realtime session %s filler audio scheduled: epoch=%s tool=%s",
self.session_id,
epoch,
tool_name or "generic",
)
self._filler_task = asyncio.create_task(
self._play_filler_audio(
epoch=epoch,
tool_name=tool_name,
started_monotonic=started_monotonic,
filler_started=filler_started,
),
name=f"{self.session_id}-filler-{epoch}",
)
async def _play_filler_audio(
self,
*,
epoch: int,
tool_name: str | None,
started_monotonic: float,
filler_started: asyncio.Event,
) -> None:
if self._filler_audio is None:
LOGGER.info("realtime session %s filler audio skipped: no library", self.session_id)
filler_started.clear()
return
filler_pcm = self._filler_audio.pick(tool_name)
if not filler_pcm:
LOGGER.info(
"realtime session %s filler audio skipped: no clip tool=%s",
self.session_id,
tool_name or "generic",
)
filler_started.clear()
return
first_audio_seen = False
sent_chunks = 0
sent_bytes = 0
for offset in range(0, len(filler_pcm), max(self.transport.frame_bytes * 4, self.transport.frame_bytes)):
self._ensure_generation(epoch)
audio_chunk = filler_pcm[offset : offset + max(self.transport.frame_bytes * 4, self.transport.frame_bytes)]
if not audio_chunk:
continue
sent_chunks += 1
sent_bytes += len(audio_chunk)
if not first_audio_seen:
first_audio_seen = True
self._log_latency(
"filler_ttfa",
started_monotonic,
epoch=epoch,
message=f"tool call -> filler audio started ({tool_name or 'generic'})",
)
self._set_state(SessionState.ASSISTANT_SPEAKING, reason="filler audio playback started")
await self.transport.send_audio(audio_chunk)
await self.transport.flush_audio()
LOGGER.info(
"realtime session %s filler audio completed: epoch=%s tool=%s chunks=%s bytes=%s",
self.session_id,
epoch,
tool_name or "generic",
sent_chunks,
sent_bytes,
)
def _clear_sentence_queue(self, sentence_queue: asyncio.Queue[str | None] | None = None) -> None:
queue = sentence_queue if sentence_queue is not None else self._sentence_queue
if queue is None:
return
removed = 0
while True:
try:
queue.get_nowait()
removed += 1
except asyncio.QueueEmpty:
break
with contextlib.suppress(asyncio.QueueFull):
queue.put_nowait(None)
LOGGER.info(
"realtime session %s sentence queue cleared: removed=%s owns_queue=%s",
self.session_id,
removed,
queue is self._sentence_queue,
)
if queue is self._sentence_queue:
self._sentence_queue = None
def _set_state(self, state: SessionState, *, reason: str | None = None) -> None:
if self.state == state:
return
previous_state = self.state
self.state = state
suffix = f" ({reason})" if reason else ""
LOGGER.info(
"realtime session %s state %s -> %s%s",
self.session_id,
previous_state.value,
state.value,
suffix,
)
async def _start_live_stt_stream(self) -> None:
await self._cancel_live_stt_stream()
self._reset_semantic_endpointing()
if self._awaiting_customer_name:
LOGGER.info(
"realtime session %s live STT stream skipped for name capture; will use batch STT with keyterms",
self.session_id,
)
return
try:
LOGGER.info("realtime session %s live STT stream start requested", self.session_id)
live_stt_stream = await self._stt.start_stream(
partial_callback=self._handle_partial_transcript,
keyterms=None,
language_code=_stt_language_for(self._session_language),
)
except Exception:
LOGGER.exception("realtime session %s failed to start live STT stream", self.session_id)
return
if live_stt_stream is None:
LOGGER.info("realtime session %s live STT stream unavailable; will use batch STT", self.session_id)
return
self._live_stt_stream = live_stt_stream
seed_audio = self._vad.current_utterance_audio()
if seed_audio:
try:
LOGGER.info(
"realtime session %s live STT seed audio: bytes=%s ms=%s",
self.session_id,
len(seed_audio),
_audio_duration_ms(seed_audio, sample_rate_hz=self.transport.sample_rate_hz),
)
await live_stt_stream.push_audio(seed_audio)
except Exception:
LOGGER.exception("realtime session %s failed to seed live STT stream", self.session_id)
await self._cancel_live_stt_stream()
async def _push_live_stt_audio(self, audio_chunk: bytes) -> None:
live_stt_stream = self._live_stt_stream
if live_stt_stream is None or not audio_chunk:
return
try:
await live_stt_stream.push_audio(audio_chunk)
except Exception:
LOGGER.exception("realtime session %s live STT push failed", self.session_id)
await self._cancel_live_stt_stream()
def _detach_live_stt_stream(self) -> BaseSTTStream | None:
live_stt_stream = self._live_stt_stream
self._live_stt_stream = None
self._latest_partial_transcript = ""
self._reset_semantic_endpointing()
LOGGER.info(
"realtime session %s live STT stream detached: present=%s",
self.session_id,
live_stt_stream is not None,
)
return live_stt_stream
async def _cancel_live_stt_stream(self) -> None:
live_stt_stream = self._detach_live_stt_stream()
if live_stt_stream is None:
return
LOGGER.info("realtime session %s live STT stream canceling", self.session_id)
with contextlib.suppress(Exception):
await live_stt_stream.cancel()
async def _resolve_transcript(
self,
*,
stt_stream: BaseSTTStream | None,
utterance_audio: bytes,
) -> str:
self._maybe_dump_audio(utterance_audio)
if stt_stream is not None:
try:
LOGGER.info("realtime session %s resolving transcript via live STT stream", self.session_id)
transcript = (await stt_stream.finish()).strip()
if transcript:
return transcript
except Exception:
LOGGER.exception("realtime session %s live STT stream failed; falling back to batch STT", self.session_id)
LOGGER.info(
"realtime session %s resolving transcript via batch STT: utterance_bytes=%s utterance_ms=%s",
self.session_id,
len(utterance_audio),
_audio_duration_ms(utterance_audio, sample_rate_hz=self.transport.sample_rate_hz),
)
return await self._stt.transcribe(
utterance_audio,
keyterms=self._name_capture_keyterms() if self._awaiting_customer_name else None,
language_code=(
self._name_capture_language_code()
if self._awaiting_customer_name
else _stt_language_for(self._session_language)
),
force_batch=self._awaiting_customer_name,
)
def _maybe_dump_audio(self, audio_bytes: bytes) -> None:
if str(os.getenv("ENABLE_AUDIO_DUMP", "")).strip().lower() not in {"1", "true", "yes", "on"}:
return
if not audio_bytes:
return
try:
dump_dir = os.getenv("AUDIO_DUMP_DIR", "debug_audio").strip() or "debug_audio"
os.makedirs(dump_dir, exist_ok=True)
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")
safe_session_id = re.sub(r"[^A-Za-z0-9_.-]+", "_", self.session_id)
filename = os.path.join(dump_dir, f"utterance_{safe_session_id}_{timestamp}.wav")
dump_sample_rate_hz = self.transport.sample_rate_hz
with wave.open(filename, "wb") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(dump_sample_rate_hz)
wav_file.writeframes(audio_bytes)
LOGGER.info(
"realtime session %s audio dumped: path=%s bytes=%s sample_rate=%s channels=1 sample_width=16bit",
self.session_id,
filename,
len(audio_bytes),
dump_sample_rate_hz,
)
except Exception:
LOGGER.exception("realtime session %s failed to dump audio", self.session_id)
async def _play_cached_initial_greeting(
self,
*,
epoch: int,
greeting_text: str,
pcm_audio: bytes,
started_monotonic: float,
) -> None:
audio_chunk_count = 0
audio_byte_count = 0
step = max(self.transport.frame_bytes * 4, self.transport.frame_bytes)
first_audio_seen = False
try:
for offset in range(0, len(pcm_audio), step):
self._ensure_generation(epoch)
audio_chunk = pcm_audio[offset : offset + step]
if not audio_chunk:
continue
audio_chunk_count += 1
audio_byte_count += len(audio_chunk)
if not first_audio_seen:
first_audio_seen = True
self._log_latency(
"initial_greeting_cached_ttfa",
started_monotonic,
epoch=epoch,
message="session start -> cached initial greeting first audio",
)
self._set_state(SessionState.ASSISTANT_SPEAKING, reason="cached initial greeting started")
await self.transport.send_audio(audio_chunk)
await self.transport.flush_audio()
self._ensure_generation(epoch)
LOGGER.info(
"realtime session %s cached initial greeting completed: epoch=%s chunks=%s bytes=%s",
self.session_id,
epoch,
audio_chunk_count,
audio_byte_count,
)
if self._initial_greeting_discarded_bytes:
LOGGER.info(
"realtime session %s ignored inbound audio during cached initial greeting: bytes=%s ms=%s",
self.session_id,
self._initial_greeting_discarded_bytes,
_audio_byte_count_duration_ms(
self._initial_greeting_discarded_bytes,
sample_rate_hz=self.transport.sample_rate_hz,
),
)
self._initial_greeting_discarded_bytes = 0
if not self._conversation or self._conversation[-1] != ("assistant", greeting_text):
self._conversation.append(("assistant", greeting_text))
LOGGER.info(
"realtime session %s conversation append cached initial greeting: entries=%s",
self.session_id,
len(self._conversation),
)
if self.state == SessionState.ASSISTANT_SPEAKING and self._assistant_task is None:
self._set_state(SessionState.LISTENING, reason="cached initial greeting completed")
except asyncio.CancelledError:
raise
except Exception:
LOGGER.exception("realtime session %s cached initial greeting failed", self.session_id)
async def _handle_partial_transcript(self, partial_text: str) -> None:
normalized = " ".join(str(partial_text or "").strip().lower().split())
if not normalized or self._closed or self._live_stt_stream is None:
return
self._latest_partial_transcript = normalized
LOGGER.info(
"realtime session %s partial transcript received: chars=%s text=%r",
self.session_id,
len(normalized),
_preview_text(normalized),
)
target_timeout_ms = (
self._semantic_hold_silence_timeout_ms
if self._should_hold_endpointing(normalized)
else self._default_vad_silence_timeout_ms
)
current_timeout_ms = getattr(self._vad, "current_speech_end_silence_ms", self._default_vad_silence_timeout_ms)
if target_timeout_ms == current_timeout_ms:
return
self._vad.set_speech_end_silence_ms(target_timeout_ms)
LOGGER.info(
"realtime session %s semantic endpointing timeout=%sms partial=%r",
self.session_id,
target_timeout_ms,
normalized[-80:],
)
def _reset_semantic_endpointing(self) -> None:
self._latest_partial_transcript = ""
self._vad.restore_default_speech_end_silence_ms()
def _should_hold_endpointing(self, partial_text: str) -> bool:
normalized = partial_text.strip().lower()
if not normalized:
return False
if normalized.endswith(("...", "…", ",", ":", ";", "-", "—")):
return True
if len(normalized) < 12:
return True
words = normalized.split()
if len(words) < 3:
return True
last_word = words[-1].strip(".,!?;:()[]{}\"'")
if last_word in SEMANTIC_CONTINUATION_TOKENS:
return True
if len(last_word) <= 2 and len(words) < 5:
return True
if normalized.endswith((".", "!", "?")):
return False
return len(words) < 6
def _commit_assistant_context(
self,
*,
llm_generated_text: str,
actually_spoken_chunks: list[str],
interrupted: bool,
) -> None:
if self._closed:
return
spoken_text = self._compose_assistant_history_text(
actually_spoken_chunks=actually_spoken_chunks,
fallback_text=llm_generated_text,
interrupted=interrupted,
)
if not spoken_text:
return
self._conversation.append(("assistant", spoken_text))
LOGGER.info(
"realtime session %s conversation append assistant: entries=%s interrupted=%s "
"generated_chars=%s spoken_chars=%s spoken_chunks=%s text=%r",
self.session_id,
len(self._conversation),
interrupted,
len(llm_generated_text),
len(spoken_text),
len(actually_spoken_chunks),
_preview_text(spoken_text),
)
@staticmethod
def _compose_assistant_history_text(
*,
actually_spoken_chunks: list[str],
fallback_text: str,
interrupted: bool,
) -> str:
spoken_text = " ".join(chunk.strip() for chunk in actually_spoken_chunks if chunk and chunk.strip()).strip()
if not spoken_text and not interrupted:
spoken_text = fallback_text.strip()
if not spoken_text:
return ""
if interrupted:
return f"{spoken_text} [interrupted]"
return spoken_text
Reference in New Issue View Git Blame Copy Permalink