532 lines
21 KiB
Python
532 lines
21 KiB
Python
"""음성 캐스케이드 — OpenAI STT(전사) + TTS(멀티보이스) 어댑터.
|
||
|
||
MASTERPLAN '음성 필수'(한신대 요구) / DESIGN_CONCEPT §5.2(음성 오브 4상태) / §4.3(립싱크 RMS):
|
||
STT : OpenAI /v1/audio/transcriptions (gpt-4o-transcribe | whisper-1). 학습자 음성 → 텍스트.
|
||
TTS : OpenAI /v1/audio/speech (gpt-4o-mini-tts | tts-1). 내담자 텍스트 → 음성(페르소나 voice).
|
||
|
||
설계 원칙(이 모듈의 경계):
|
||
- 순수 어댑터: httpx 로 OpenAI 음성 엔드포인트만 호출한다. 상담 로직(orchestrator)·상태머신은
|
||
호출부(routes/voice.py)가 조립한다. 여기는 "오디오↔텍스트" 변환 + voice preset 매핑만.
|
||
- API 키 없으면 명확히 degraded: is_available()=False, 호출 시 VoiceUnavailable.
|
||
절대 크래시·무한대기 금지(라우트가 503/close 로 변환).
|
||
- 립싱크 힌트: TTS 오디오 청크를 흘리며 RMS(진폭) 힌트를 같이 산출(설계 §4.3 — viseme 정밀
|
||
매칭 안 함, RMS 1채널). PCM 디코딩 의존성 없이 바이트 에너지 근사로 임시 RMS 추정.
|
||
|
||
페르소나 voice preset(persona/*.json voice.preset, 설계 §4.6) → OpenAI voice 매핑은
|
||
PRESET_TO_OPENAI_VOICE 테이블이 흡수. 새 preset 추가는 이 테이블만 손대면 된다.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import AsyncIterator, Optional
|
||
|
||
import httpx
|
||
|
||
from ..config import settings
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# OpenAI 음성 엔드포인트/모델 상수
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
OPENAI_BASE_URL = "https://api.openai.com/v1"
|
||
STT_ENDPOINT = "/audio/transcriptions"
|
||
TTS_ENDPOINT = "/audio/speech"
|
||
|
||
# STT 모델: gpt-4o-transcribe(고품질) — 미가용 폴백은 whisper-1.
|
||
STT_MODEL = "gpt-4o-transcribe"
|
||
STT_MODEL_FALLBACK = "whisper-1"
|
||
# TTS 모델: gpt-4o-mini-tts(저지연·표현력) — 폴백 tts-1.
|
||
TTS_MODEL = "gpt-4o-mini-tts"
|
||
TTS_MODEL_FALLBACK = "tts-1"
|
||
|
||
# 전사 언어 힌트(상담은 한국어). OpenAI 는 ISO-639-1.
|
||
STT_LANGUAGE = "ko"
|
||
|
||
# TTS 출력 포맷: 브라우저 MediaSource/<audio> 친화. 스트리밍은 mp3/opus 청크.
|
||
TTS_RESPONSE_FORMAT = "mp3"
|
||
|
||
# End-of-turn readiness default for cascaded STT providers.
|
||
EOT_SILENCE_THRESHOLD_MS = 1200
|
||
|
||
_REPO_ROOT = Path(__file__).resolve().parents[4]
|
||
POC_SAMPLE_TTS_PRESET = "soft-young-fem"
|
||
POC_SAMPLE_TTS_DEFAULT_DIR = (
|
||
_REPO_ROOT / "docs" / "voice-art" / "p1-seoyeon-higgs-v3-20260627"
|
||
)
|
||
POC_SAMPLE_TTS_CHUNK_SIZE = 4096
|
||
_POC_SAMPLE_TTS_DEFAULT_SAMPLE = "p1_seoyeon_01_depressed_slow"
|
||
_POC_SAMPLE_TTS_KEYWORDS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
||
(
|
||
"p1_seoyeon_03_anxious_guarded",
|
||
("엄마", "비밀", "말하지", "불안", "무서", "걱정", "들키", "싫", "갈래"),
|
||
),
|
||
(
|
||
"p1_seoyeon_02_tired_flat",
|
||
("잠", "피곤", "무거", "아무것도", "지쳐", "힘들", "에너지"),
|
||
),
|
||
(
|
||
"p1_seoyeon_05_recovered_lively",
|
||
("오늘은", "친구", "웃", "괜찮았", "좋았", "해냈"),
|
||
),
|
||
(
|
||
"p1_seoyeon_04_rapport_relief",
|
||
("괜찮", "들어", "고마", "선생님", "편해", "조금", "말해"),
|
||
),
|
||
)
|
||
|
||
# OpenAI 공식 voice 풀(2026 기준): alloy, ash, ballad, coral, echo, fable,
|
||
# nova, onyx, sage, shimmer, verse. 페르소나 톤별로 골라 매핑한다.
|
||
_OPENAI_VOICES = {
|
||
"alloy", "ash", "ballad", "coral", "echo", "fable",
|
||
"nova", "onyx", "sage", "shimmer", "verse",
|
||
}
|
||
DEFAULT_OPENAI_VOICE = "sage"
|
||
|
||
# ── 페르소나 voice preset(설계 §4.6) → OpenAI voice ──────────────────────────
|
||
# preset 명명: <톤><연령><성별> 조합(soft-young-fem 등). 새 페르소나는 여기에만 추가.
|
||
PRESET_TO_OPENAI_VOICE: dict[str, str] = {
|
||
# 청소년 여성(P1 서연) — 부드럽고 톤 높은
|
||
"soft-young-fem": "coral",
|
||
# 성인 남성(P2 민재) — 차분·안정·약간 긴장
|
||
"calm-adult-male": "ash",
|
||
# 성인 여성(P3 지우) — 따뜻하지만 지친
|
||
"warm-adult-fem": "shimmer",
|
||
# 범용 폴백 프리셋
|
||
"neutral": "sage",
|
||
}
|
||
|
||
# ── 페르소나 code(P1/P2/P3) → 기본 preset (persona 카드에 voice 필드 없을 때) ──
|
||
# persona.py 시드는 voice 필드를 갖지 않으므로 code 로 기본 preset 을 정한다.
|
||
# DB/JSON 페르소나가 voice.preset 을 직접 주면 그걸 우선한다(resolve_voice 참조).
|
||
PERSONA_CODE_TO_PRESET: dict[str, str] = {
|
||
"P1": "soft-young-fem",
|
||
"P2": "calm-adult-male",
|
||
"P3": "warm-adult-fem",
|
||
}
|
||
|
||
# TTS rate(말 속도) 페르소나 기본값(설계 §4.6 voice.rate). 1.0=표준.
|
||
PRESET_RATE: dict[str, float] = {
|
||
"soft-young-fem": 0.96,
|
||
"calm-adult-male": 1.0,
|
||
"warm-adult-fem": 0.98,
|
||
"neutral": 1.0,
|
||
}
|
||
|
||
|
||
class VoiceUnavailable(RuntimeError):
|
||
"""음성 미설정/장애. 라우트가 503/WS close(degraded) 로 변환."""
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class VoicePreset:
|
||
"""해석된 음성 프리셋(페르소나 → OpenAI 파라미터)."""
|
||
|
||
preset: str # 논리 preset 명(soft-young-fem 등)
|
||
openai_voice: str # OpenAI voice 파라미터
|
||
rate: float = 1.0 # 말 속도(speed)
|
||
instructions: Optional[str] = None # gpt-4o-mini-tts 표현 지시(선택)
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class TranscriptResult:
|
||
"""STT 결과."""
|
||
|
||
text: str
|
||
language: Optional[str] = None
|
||
model: str = STT_MODEL
|
||
duration: Optional[float] = None
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class EndOfTurnDecision:
|
||
"""Provider-neutral readiness signal for a completed learner utterance."""
|
||
|
||
ready: bool
|
||
transcript_ready: bool
|
||
silence_ready: bool
|
||
silence_ms: int
|
||
threshold_ms: int
|
||
reason: str
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class TTSChunk:
|
||
"""TTS 스트림 1청크(오디오 바이트). 립싱크는 프론트 Web Audio AnalyserNode가 자체 산출."""
|
||
|
||
audio: bytes
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# voice preset 해석 (페르소나 → OpenAI 파라미터)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
def resolve_voice(
|
||
*,
|
||
persona_code: Optional[str] = None,
|
||
preset: Optional[str] = None,
|
||
instructions: Optional[str] = None,
|
||
) -> VoicePreset:
|
||
"""페르소나 code 또는 명시 preset → OpenAI voice 파라미터로 해석.
|
||
|
||
우선순위: 명시 preset > persona_code 기본 preset > 'neutral'.
|
||
알 수 없는 preset 은 DEFAULT_OPENAI_VOICE 로 안전 폴백(크래시 없음).
|
||
"""
|
||
chosen = preset
|
||
if not chosen and persona_code:
|
||
chosen = PERSONA_CODE_TO_PRESET.get(persona_code.upper())
|
||
if not chosen:
|
||
chosen = "neutral"
|
||
|
||
openai_voice = PRESET_TO_OPENAI_VOICE.get(chosen, DEFAULT_OPENAI_VOICE)
|
||
if openai_voice not in _OPENAI_VOICES:
|
||
openai_voice = DEFAULT_OPENAI_VOICE
|
||
rate = PRESET_RATE.get(chosen, 1.0)
|
||
return VoicePreset(
|
||
preset=chosen,
|
||
openai_voice=openai_voice,
|
||
rate=rate,
|
||
instructions=instructions,
|
||
)
|
||
|
||
|
||
# 비언어 지문 패턴: (…)·（…）·[…]·【…】. 내담자 발화의 무대지시(고개 끄덕/한숨/침묵 등).
|
||
_STAGE_DIRECTION_RE = re.compile(r"[\(（\[【][^\)）\]】]*[\)）\]】]")
|
||
|
||
|
||
def speakable_text(text: str) -> str:
|
||
"""TTS로 읽을 텍스트만 남긴다 — 비언어 지문((고개 살짝 끄덕)·（한숨）·[침묵])을 제거.
|
||
|
||
지문은 자막/회기리뷰에 남고 아바타 애니메이션이 표현하며, 음성으로는 읽지 않는다.
|
||
지문만으로 이뤄진 발화(예: "(침묵)")는 빈 문자열을 반환 → 합성 생략.
|
||
"""
|
||
if not text:
|
||
return ""
|
||
stripped = _STAGE_DIRECTION_RE.sub(" ", text)
|
||
# 말줄임표/중복 공백 정리 + 고아 구두점 앞 공백 제거
|
||
stripped = re.sub(r"\s+", " ", stripped)
|
||
stripped = re.sub(r"\s+([,.!?…」』】)])", r"\1", stripped)
|
||
return stripped.strip()
|
||
|
||
|
||
def build_tts_payload(
|
||
text: str,
|
||
voice: VoicePreset,
|
||
*,
|
||
model: str = TTS_MODEL,
|
||
response_format: str = TTS_RESPONSE_FORMAT,
|
||
) -> dict[str, object]:
|
||
"""Build the deterministic OpenAI TTS payload for a resolved voice preset."""
|
||
payload: dict[str, object] = {
|
||
"model": model,
|
||
"voice": voice.openai_voice,
|
||
"input": text,
|
||
"response_format": response_format,
|
||
"speed": _clamp_speed(voice.rate),
|
||
}
|
||
if voice.instructions and model.startswith("gpt-4o"):
|
||
payload["instructions"] = voice.instructions
|
||
return payload
|
||
|
||
|
||
def assess_end_of_turn(
|
||
*,
|
||
transcript_text: Optional[str],
|
||
transcript_final: bool,
|
||
silence_ms: Optional[int],
|
||
silence_threshold_ms: int = EOT_SILENCE_THRESHOLD_MS,
|
||
) -> EndOfTurnDecision:
|
||
"""Return whether final STT text plus observed silence is enough to run a turn."""
|
||
observed_silence = _nonnegative_int(silence_ms)
|
||
threshold = max(0, _nonnegative_int(silence_threshold_ms))
|
||
has_text = bool((transcript_text or "").strip())
|
||
transcript_ready = bool(transcript_final and has_text)
|
||
silence_ready = observed_silence >= threshold
|
||
ready = transcript_ready and silence_ready
|
||
if ready:
|
||
reason = "ready"
|
||
elif not has_text:
|
||
reason = "empty_transcript"
|
||
elif not transcript_final:
|
||
reason = "final_transcript_pending"
|
||
else:
|
||
reason = "silence_threshold_pending"
|
||
return EndOfTurnDecision(
|
||
ready=ready,
|
||
transcript_ready=transcript_ready,
|
||
silence_ready=silence_ready,
|
||
silence_ms=observed_silence,
|
||
threshold_ms=threshold,
|
||
reason=reason,
|
||
)
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# OpenAI 음성 서비스
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
class VoiceService:
|
||
"""OpenAI STT/TTS 어댑터. 앱 수명주기 동안 1 인스턴스 재사용(httpx 풀 공유)."""
|
||
|
||
def __init__(
|
||
self,
|
||
api_key: Optional[str] = None,
|
||
base_url: Optional[str] = None,
|
||
*,
|
||
poc_sample_tts_enabled: Optional[bool] = None,
|
||
environment: Optional[str] = None,
|
||
poc_sample_tts_dir: Optional[str | Path] = None,
|
||
) -> None:
|
||
self._api_key = (api_key if api_key is not None else settings.openai_api_key) or ""
|
||
self._base_url = (base_url or settings.openai_base_url or OPENAI_BASE_URL).rstrip("/")
|
||
self._environment = environment if environment is not None else settings.environment
|
||
self._poc_sample_tts_enabled = (
|
||
bool(settings.voice_poc_sample_tts_enabled)
|
||
if poc_sample_tts_enabled is None
|
||
else bool(poc_sample_tts_enabled)
|
||
)
|
||
sample_dir_value: str | Path = (
|
||
poc_sample_tts_dir
|
||
if poc_sample_tts_dir is not None
|
||
else (settings.voice_poc_sample_tts_dir or POC_SAMPLE_TTS_DEFAULT_DIR)
|
||
)
|
||
sample_dir = Path(sample_dir_value)
|
||
if not sample_dir.is_absolute():
|
||
sample_dir = _REPO_ROOT / sample_dir
|
||
self._poc_sample_tts_dir = sample_dir
|
||
self._client: Optional[httpx.AsyncClient] = None
|
||
|
||
# ── 수명주기 ──────────────────────────────────────────
|
||
async def startup(self) -> None:
|
||
if not self._api_key:
|
||
return # 키 없으면 클라이언트도 안 띄움(degraded). 라우트가 503 처리.
|
||
self._client = httpx.AsyncClient(
|
||
base_url=self._base_url,
|
||
headers={"Authorization": f"Bearer {self._api_key}"},
|
||
timeout=httpx.Timeout(60.0, connect=10.0),
|
||
)
|
||
|
||
async def shutdown(self) -> None:
|
||
if self._client is not None:
|
||
await self._client.aclose()
|
||
self._client = None
|
||
|
||
def is_available(self) -> bool:
|
||
"""음성 기능 가용 여부(키 설정됨). 라우트가 핸드셰이크에서 검사."""
|
||
return bool(self._api_key)
|
||
|
||
def tts_provider(self) -> str:
|
||
if self._poc_sample_tts_available():
|
||
return "p1-sample-poc"
|
||
if self._api_key:
|
||
return "openai"
|
||
if self._poc_sample_tts_enabled and self._environment != "dev":
|
||
return "disabled-non-dev"
|
||
return "unavailable"
|
||
|
||
def poc_sample_tts_available(self) -> bool:
|
||
return self._poc_sample_tts_available()
|
||
|
||
def _poc_sample_tts_available(self) -> bool:
|
||
return (
|
||
self._poc_sample_tts_enabled
|
||
and self._environment == "dev"
|
||
and self._poc_sample_path(_POC_SAMPLE_TTS_DEFAULT_SAMPLE).is_file()
|
||
)
|
||
|
||
def _should_use_poc_sample_tts(self, voice: VoicePreset) -> bool:
|
||
return (
|
||
self._poc_sample_tts_enabled
|
||
and self._environment == "dev"
|
||
and voice.preset == POC_SAMPLE_TTS_PRESET
|
||
)
|
||
|
||
def _poc_sample_path(self, sample_id: str) -> Path:
|
||
return self._poc_sample_tts_dir / f"{sample_id}.mp3"
|
||
|
||
@property
|
||
def _http(self) -> httpx.AsyncClient:
|
||
if not self._api_key:
|
||
raise VoiceUnavailable("OPENAI_API_KEY 미설정 — 음성 기능 degraded")
|
||
if self._client is None:
|
||
# lazy 보강(테스트/지연 startup 대비)
|
||
self._client = httpx.AsyncClient(
|
||
base_url=self._base_url,
|
||
headers={"Authorization": f"Bearer {self._api_key}"},
|
||
timeout=httpx.Timeout(60.0, connect=10.0),
|
||
)
|
||
return self._client
|
||
|
||
# ── STT (transcriptions) ─────────────────────────────
|
||
async def transcribe(
|
||
self,
|
||
audio: bytes,
|
||
*,
|
||
filename: str = "audio.webm",
|
||
content_type: str = "audio/webm",
|
||
language: str = STT_LANGUAGE,
|
||
model: str = STT_MODEL,
|
||
) -> TranscriptResult:
|
||
"""오디오 바이트 → 텍스트 전사(OpenAI /audio/transcriptions).
|
||
|
||
클라가 보낸 webm/opus(또는 wav/mp3) 청크를 multipart 로 OpenAI 에 올린다.
|
||
키 없으면 VoiceUnavailable, OpenAI 오류는 그대로 RuntimeError 로 전파(라우트가 처리).
|
||
"""
|
||
if not audio:
|
||
return TranscriptResult(text="", model=model)
|
||
files = {"file": (filename, audio, content_type)}
|
||
data = {
|
||
"model": model,
|
||
"language": language,
|
||
"response_format": "json",
|
||
}
|
||
try:
|
||
r = await self._http.post(STT_ENDPOINT, files=files, data=data)
|
||
if r.status_code == 404 and model != STT_MODEL_FALLBACK:
|
||
# 모델 미가용(계정 권한) → whisper-1 폴백 1회
|
||
data["model"] = STT_MODEL_FALLBACK
|
||
r = await self._http.post(STT_ENDPOINT, files=files, data=data)
|
||
r.raise_for_status()
|
||
except VoiceUnavailable:
|
||
raise
|
||
except httpx.HTTPStatusError as e:
|
||
raise RuntimeError(f"STT {e.response.status_code}: {e.response.text[:200]}") from e
|
||
except httpx.HTTPError as e:
|
||
raise RuntimeError(f"STT transport error: {e}") from e
|
||
|
||
body = r.json()
|
||
return TranscriptResult(
|
||
text=(body.get("text") or "").strip(),
|
||
language=body.get("language"),
|
||
model=str(data["model"]),
|
||
duration=body.get("duration"),
|
||
)
|
||
|
||
# ── TTS (speech) — 스트리밍 ──────────────────────────
|
||
async def synthesize_stream(
|
||
self,
|
||
text: str,
|
||
voice: VoicePreset,
|
||
*,
|
||
model: str = TTS_MODEL,
|
||
response_format: str = TTS_RESPONSE_FORMAT,
|
||
) -> AsyncIterator[TTSChunk]:
|
||
"""텍스트 → 음성 스트리밍(OpenAI /audio/speech). 청크 + RMS 힌트 yield.
|
||
|
||
설계 §5.2 'speaking' 상태: 오디오 청크를 흘리며 진폭 힌트(립싱크)를 같이 보낸다.
|
||
키 없으면 VoiceUnavailable. OpenAI 오류는 RuntimeError 전파.
|
||
"""
|
||
# 비언어 지문((고개 끄덕)·（한숨）·[침묵])은 음성으로 읽지 않는다. 자막엔 남고
|
||
# 아바타 애니메이션이 표현한다. 지문만 있는 발화는 합성 생략(빈 오디오).
|
||
text = speakable_text(text)
|
||
if not text:
|
||
return
|
||
if self._should_use_poc_sample_tts(voice):
|
||
async for chunk in self._synthesize_poc_sample_tts(text):
|
||
yield chunk
|
||
return
|
||
payload = build_tts_payload(
|
||
text,
|
||
voice,
|
||
model=model,
|
||
response_format=response_format,
|
||
)
|
||
|
||
try:
|
||
async with self._http.stream("POST", TTS_ENDPOINT, json=payload) as r:
|
||
if r.status_code == 404 and model != TTS_MODEL_FALLBACK:
|
||
# 모델 미가용 → tts-1 폴백(비스트림 재시도). instructions 제거.
|
||
payload["model"] = TTS_MODEL_FALLBACK
|
||
payload.pop("instructions", None)
|
||
await r.aclose()
|
||
async for c in self._synthesize_fallback(payload):
|
||
yield c
|
||
return
|
||
r.raise_for_status()
|
||
async for chunk in r.aiter_bytes(chunk_size=4096):
|
||
if not chunk:
|
||
continue
|
||
yield TTSChunk(audio=chunk)
|
||
except VoiceUnavailable:
|
||
raise
|
||
except httpx.HTTPStatusError as e:
|
||
text_body = ""
|
||
try:
|
||
text_body = (await e.response.aread()).decode("utf-8", "ignore")[:200]
|
||
except Exception:
|
||
pass
|
||
raise RuntimeError(f"TTS {e.response.status_code}: {text_body}") from e
|
||
except httpx.HTTPError as e:
|
||
raise RuntimeError(f"TTS transport error: {e}") from e
|
||
|
||
async def _synthesize_poc_sample_tts(self, text: str) -> AsyncIterator[TTSChunk]:
|
||
sample_id = self._select_poc_sample_id(text)
|
||
sample_path = self._poc_sample_path(sample_id)
|
||
try:
|
||
data = sample_path.read_bytes()
|
||
except OSError as e:
|
||
raise VoiceUnavailable(f"P1 sample TTS asset is missing: {sample_path}") from e
|
||
for i in range(0, len(data), POC_SAMPLE_TTS_CHUNK_SIZE):
|
||
chunk = data[i : i + POC_SAMPLE_TTS_CHUNK_SIZE]
|
||
if chunk:
|
||
yield TTSChunk(audio=chunk)
|
||
|
||
def _select_poc_sample_id(self, text: str) -> str:
|
||
normalized = text.casefold()
|
||
for sample_id, keywords in _POC_SAMPLE_TTS_KEYWORDS:
|
||
if any(keyword.casefold() in normalized for keyword in keywords):
|
||
return sample_id
|
||
return _POC_SAMPLE_TTS_DEFAULT_SAMPLE
|
||
|
||
async def _synthesize_fallback(self, payload: dict[str, object]) -> AsyncIterator[TTSChunk]:
|
||
"""tts-1 폴백(비스트림 POST → 전체 바이트를 청크로 분할)."""
|
||
try:
|
||
r = await self._http.post(TTS_ENDPOINT, json=payload)
|
||
r.raise_for_status()
|
||
except httpx.HTTPStatusError as e:
|
||
raise RuntimeError(f"TTS(fallback) {e.response.status_code}: {e.response.text[:200]}") from e
|
||
except httpx.HTTPError as e:
|
||
raise RuntimeError(f"TTS(fallback) transport error: {e}") from e
|
||
data = r.content
|
||
for i in range(0, len(data), 4096):
|
||
chunk = data[i : i + 4096]
|
||
yield TTSChunk(audio=chunk)
|
||
|
||
|
||
def _clamp_speed(rate: float) -> float:
|
||
"""OpenAI speed 허용범위 [0.25, 4.0] 클램프."""
|
||
try:
|
||
return max(0.25, min(4.0, float(rate)))
|
||
except (TypeError, ValueError):
|
||
return 1.0
|
||
|
||
|
||
def _nonnegative_int(value: object) -> int:
|
||
try:
|
||
return max(0, int(value)) # type: ignore[arg-type]
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
|
||
|
||
# 앱 전역 싱글톤 (main lifespan 이 startup/shutdown — Foundation 이 관리하거나
|
||
# 라우트가 lazy 사용). engine_client 패턴과 동일.
|
||
voice_service = VoiceService()
|
||
|
||
|
||
__all__ = [
|
||
"VoiceUnavailable",
|
||
"VoicePreset",
|
||
"TranscriptResult",
|
||
"EndOfTurnDecision",
|
||
"TTSChunk",
|
||
"VoiceService",
|
||
"voice_service",
|
||
"resolve_voice",
|
||
"build_tts_payload",
|
||
"assess_end_of_turn",
|
||
"EOT_SILENCE_THRESHOLD_MS",
|
||
"PRESET_TO_OPENAI_VOICE",
|
||
"PERSONA_CODE_TO_PRESET",
|
||
"DEFAULT_OPENAI_VOICE",
|
||
"STT_MODEL",
|
||
"TTS_MODEL",
|
||
]
|