24b1b7a6e1
web (Vite+React19+TS, Cloudflare Pages 배포): - 디자인토큰(세이지틸/테라코타 SSOT), 앱셸, 공통 UI 프리미티브 - 7화면: 로그인/학습자홈/상담세션/회기리뷰/교수자/관리자/설정 - ClientAvatar: SVG 반구상 흉상 4상태 + RMS 립싱크 + 6파라미터 정서 - 회기리뷰는 외부 레퍼런스 디자인을 Vignette 토큰으로 리스킨 api (FastAPI): - 게이트웨이 /v1/generate·/v1/stream 어댑터(상주풀/EngineSession 보존) - services: 페르소나 L0~L6 빌더 / 결정론 상태머신 / 가드레일 / 턴 오케스트레이터 / 회기간 메모리 / 평가AI / 음성 / RAG - store: DB off 폴백(in-memory), sessions 실구현 검증: - web: node22 tsc+vite build 통과(node23 segfault 회피), Pages 배포 200 - api: app.main import 통과 - 핫픽스: Topbar initials undefined-safe (undefined.trim 크래시) - E2E: 서연(P1) 상담 1턴 — 좋은/나쁜 상담에 차등 반응 실증
370 lines
16 KiB
Python
370 lines
16 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 math
|
|
from dataclasses import dataclass, field
|
|
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"
|
|
|
|
# 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(slots=True)
|
|
class TTSChunk:
|
|
"""TTS 스트림 1청크 + 립싱크 힌트(설계 §4.3 RMS 1채널)."""
|
|
|
|
audio: bytes
|
|
rms: float = 0.0 # 0~1, 입 열림(scaleY) 매핑용 근사 진폭
|
|
seq: int = 0
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
# 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,
|
|
)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
# 립싱크 RMS 근사 (설계 §4.3 — 정밀 viseme 안 함, 진폭 1채널)
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
def estimate_chunk_rms(chunk: bytes) -> float:
|
|
"""오디오 청크 바이트 에너지로 RMS(0~1) 근사.
|
|
|
|
압축 포맷(mp3) 바이트를 PCM 디코딩 없이 근사한다(의존성 0). 평균 바이트 편차를
|
|
0~1 로 정규화 → 프론트가 데드존(0.04)·지수평활(τ≈180ms) 적용해 입 열림에 매핑.
|
|
NOTE: 정밀 진폭이 필요하면 프론트 Web Audio AnalyserNode 가 재계산(설계 §4.3 권장).
|
|
이 힌트는 서버측 보조(네트워크 끊김/저사양 폴백)다.
|
|
"""
|
|
if not chunk:
|
|
return 0.0
|
|
# 128 중심 편차의 RMS(8bit 가정 근사). mp3 프레임이라 정밀치 아님(상대값).
|
|
n = len(chunk)
|
|
acc = 0
|
|
# 과샘플 비용 회피 — 최대 2048 바이트만 샘플링
|
|
step = max(1, n // 2048)
|
|
cnt = 0
|
|
for i in range(0, n, step):
|
|
d = chunk[i] - 128
|
|
acc += d * d
|
|
cnt += 1
|
|
if cnt == 0:
|
|
return 0.0
|
|
rms = math.sqrt(acc / cnt) / 128.0
|
|
return max(0.0, min(1.0, rms))
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
# OpenAI 음성 서비스
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
class VoiceService:
|
|
"""OpenAI STT/TTS 어댑터. 앱 수명주기 동안 1 인스턴스 재사용(httpx 풀 공유)."""
|
|
|
|
def __init__(self, api_key: Optional[str] = None, base_url: str = OPENAI_BASE_URL) -> None:
|
|
self._api_key = (api_key if api_key is not None else settings.openai_api_key) or ""
|
|
self._base_url = base_url.rstrip("/")
|
|
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)
|
|
|
|
@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 전파.
|
|
"""
|
|
if not text or not text.strip():
|
|
return
|
|
payload: dict[str, object] = {
|
|
"model": model,
|
|
"voice": voice.openai_voice,
|
|
"input": text,
|
|
"response_format": response_format,
|
|
"speed": _clamp_speed(voice.rate),
|
|
}
|
|
# gpt-4o-mini-tts 계열은 instructions(표현 지시) 지원. tts-1 은 무시됨.
|
|
if voice.instructions and model.startswith("gpt-4o"):
|
|
payload["instructions"] = voice.instructions
|
|
|
|
seq = 0
|
|
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, rms=estimate_chunk_rms(chunk), seq=seq)
|
|
seq += 1
|
|
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_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
|
|
seq = 0
|
|
for i in range(0, len(data), 4096):
|
|
chunk = data[i : i + 4096]
|
|
yield TTSChunk(audio=chunk, rms=estimate_chunk_rms(chunk), seq=seq)
|
|
seq += 1
|
|
|
|
|
|
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
|
|
|
|
|
|
# 앱 전역 싱글톤 (main lifespan 이 startup/shutdown — Foundation 이 관리하거나
|
|
# 라우트가 lazy 사용). engine_client 패턴과 동일.
|
|
voice_service = VoiceService()
|
|
|
|
|
|
__all__ = [
|
|
"VoiceUnavailable",
|
|
"VoicePreset",
|
|
"TranscriptResult",
|
|
"TTSChunk",
|
|
"VoiceService",
|
|
"voice_service",
|
|
"resolve_voice",
|
|
"estimate_chunk_rms",
|
|
"PRESET_TO_OPENAI_VOICE",
|
|
"PERSONA_CODE_TO_PRESET",
|
|
"DEFAULT_OPENAI_VOICE",
|
|
"STT_MODEL",
|
|
"TTS_MODEL",
|
|
]
|