diff --git a/README.md b/README.md index 8d2d27e..a612084 100644 --- a/README.md +++ b/README.md @@ -171,8 +171,9 @@ ffmpeg 디코드 → 프레임. DAVE는 `@snazzah/davey`(Rust NAPI), 수신은 ` · 검증: `python -m wsai --voice`가 화면 없이 발화마다 응답을 낸다(테스트 포함). - [ ] **V1 · 진짜 TTS(입).** MeloTTS로 두뇌 응답을 실제 음성으로 합성. · 검증: 응답 텍스트가 .wav로 합성돼 들린다. -- [ ] **V2 · 진짜 STT(귀).** faster-whisper로 오디오를 텍스트로 전사. - · 검증: 오디오 파일 입력이 텍스트로 나온다(이후 디스코드 보이스 수신 오디오로 연결). +- [x] **V2 · 진짜 STT(귀) — 엔진 완성.** faster-whisper(상주 워커, `WSAI_STT=whisper`)로 wav를 한국어 텍스트로 전사. + · 검증: 실제 왕복(MeloTTS wav → whisper)에서 문장이 거의 그대로 복원됨. warm 전사 ~1.2s(CPU, small/int8). + · 남은 것: 디스코드 보이스 수신 오디오(`audio_source`)를 붙여 실시간 발화 스트림으로 연결(V4). - [ ] **V3 · 진짜 두뇌.** Claude OAuth(Haiku)로 대화 응답 생성. · 검증: 실제 발화에 자연스러운 답이 나온다. - [ ] **V4 · 음성 왕복 + barge-in.** 디스코드 보이스 수신→STT→Brain→TTS 스트리밍, 말 끊기, 지연 측정. diff --git a/requirements.txt b/requirements.txt index 991a286..6dc60d8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,10 +8,13 @@ # --- cloud eyes + brain (WSAI_VISION=claude / WSAI_BRAIN=claude) --- # anthropic -# --- planned voice backends (not yet implemented) --- -# faster-whisper # STT (input = Discord voice reception, not a local mic) -# (Opus decode for Discord voice audio, e.g. via the selfbot/ffmpeg path) -# (a Korean TTS engine, e.g. MeloTTS) +# --- voice backends (run in their OWN venvs; loaded as persistent workers) --- +# STT: faster-whisper (WSAI_STT=whisper) — installed in a dedicated venv, e.g. +# uv venv --python 3.12 /home/claude/jarvis-stt/whisper312 +# uv pip install --python /home/claude/jarvis-stt/whisper312/bin/python faster-whisper +# (override interpreter/model via WSAI_WHISPER_PYTHON / WSAI_WHISPER_MODEL) +# TTS: MeloTTS (WSAI_TTS=melo) — in /home/claude/jarvis-tts/melo311 +# (Opus decode for Discord voice audio, e.g. via the selfbot/ffmpeg path — pending) # --- dev --- # pytest diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 145bb0e..3a21b51 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -108,3 +108,29 @@ def test_history_is_bounded(): pipe._remember(f"u{i}", f"a{i}") assert len(pipe._history) == 3 assert pipe._history[-1] == ("u9", "a9") + + +def test_prewarm_warms_backends_and_survives_failure(): + """Backends exposing warmup() are preloaded at startup; a warmup that + raises is logged but must not abort the run (first-utterance latency is an + optimization, not a hard requirement).""" + warmed: list[str] = [] + + class WarmTTS(MockTTS): + async def warmup(self): + warmed.append("tts") + + class BoomWarmSTT(MockSTT): + async def warmup(self): + warmed.append("stt") + raise RuntimeError("model unavailable") + + pipe = Pipeline( + brain=MockBrain(), + stt=BoomWarmSTT(script=["안녕"], interval=0.01), + tts=WarmTTS(), + ) + + asyncio.run(asyncio.wait_for(pipe.run(), timeout=5)) + + assert "tts" in warmed and "stt" in warmed, "warmup() not called on backends" diff --git a/tests/test_whisper_stt.py b/tests/test_whisper_stt.py new file mode 100644 index 0000000..6620da3 --- /dev/null +++ b/tests/test_whisper_stt.py @@ -0,0 +1,60 @@ +"""Unit tests for the WhisperSTT source plumbing. + +These do NOT load a model or spawn the worker (that needs the whisper312 venv and +is exercised by the manual TTS->STT round trip). They pin the SpeechToText +contract: how `utterances()` turns an audio source into Utterances, and how it +behaves with no source wired yet. +""" + +import asyncio +from typing import AsyncIterator + +from wsai.backends.whisper import WhisperSTT +from wsai.interfaces import SpeechToText, Utterance + + +def _collect(stt: WhisperSTT) -> list[Utterance]: + async def run(): + return [u async for u in stt.utterances()] + + return asyncio.run(asyncio.wait_for(run(), timeout=5)) + + +async def _paths(items) -> AsyncIterator[str]: + for it in items: + yield it + + +def test_is_speech_to_text(): + assert isinstance(WhisperSTT(), SpeechToText) + + +def test_no_audio_source_yields_nothing(): + # Discord voice receiver not wired yet -> the loop idles and returns. + assert _collect(WhisperSTT(audio_source=None)) == [] + + +def test_utterances_transcribes_each_chunk(monkeypatch): + stt = WhisperSTT(audio_source=_paths(["a.wav", "b.wav"])) + + async def fake_transcribe(wav_path, *, language=None): + return f"text::{wav_path}" + + monkeypatch.setattr(stt, "transcribe", fake_transcribe) + + utts = _collect(stt) + assert [u.text for u in utts] == ["text::a.wav", "text::b.wav"] + assert all(u.source == "voice" for u in utts) + + +def test_empty_transcript_is_skipped(monkeypatch): + # Silence / VAD-filtered audio transcribes to "" and must not become a turn. + stt = WhisperSTT(audio_source=_paths(["silent.wav", "real.wav"])) + + async def fake_transcribe(wav_path, *, language=None): + return "" if wav_path == "silent.wav" else "안녕" + + monkeypatch.setattr(stt, "transcribe", fake_transcribe) + + utts = _collect(stt) + assert [u.text for u in utts] == ["안녕"] diff --git a/wsai/backends/melo.py b/wsai/backends/melo.py index 632f7b1..690d65a 100644 --- a/wsai/backends/melo.py +++ b/wsai/backends/melo.py @@ -17,6 +17,7 @@ Env: from __future__ import annotations import asyncio +import collections import json import logging import os @@ -59,6 +60,37 @@ class MeloTTS: self._lock = asyncio.Lock() self._n = 0 self.load_ms: int | None = None + # Keep the worker's most recent stderr lines so a crash reports its real + # cause instead of a bare JSONDecodeError. Bounded so it can't grow. + self._stderr_tail: collections.deque[str] = collections.deque(maxlen=40) + self._stderr_task: asyncio.Task | None = None + + async def _drain_stderr(self, stream: asyncio.StreamReader) -> None: + # The worker redirects fd1 -> fd2, so ALL library chatter lands on + # stderr. If we PIPE stderr but never read it, the OS pipe buffer fills + # and the worker blocks forever. So we continuously drain it and keep + # only the last few lines for diagnostics. + try: + while True: + line = await stream.readline() + if not line: + return + self._stderr_tail.append(line.decode(errors="replace").rstrip()) + except asyncio.CancelledError: + raise + except Exception: # draining must never crash the caller + return + + def _stderr_hint(self) -> str: + tail = "\n".join(self._stderr_tail) + return f" worker stderr tail:\n{tail}" if tail else " (worker produced no stderr)" + + async def warmup(self) -> None: + """Start and load the worker now so the first real utterance is warm. + + Called at pipeline startup so users don't wait ~7s (CPU model load) for + the very first spoken reply.""" + await self._ensure() async def _ensure(self) -> None: if self._proc is not None and self._proc.returncode is None: @@ -72,12 +104,29 @@ class MeloTTS: cwd=repo_root, env=env, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.PIPE, ) + self._stderr_tail.clear() + assert self._proc.stderr is not None + self._stderr_task = asyncio.create_task(self._drain_stderr(self._proc.stderr)) ready = await self._proc.stdout.readline() - info = json.loads(ready.decode()) + if not ready: # worker died before signalling ready + await self._proc.wait() + raise RuntimeError( + f"melo worker exited before ready (code {self._proc.returncode})." + f"{self._stderr_hint()}" + ) + try: + info = json.loads(ready.decode()) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"melo worker sent invalid ready line {ready!r}: {exc}." + f"{self._stderr_hint()}" + ) from exc if not info.get("ready"): - raise RuntimeError(f"melo worker failed to start: {info}") + raise RuntimeError( + f"melo worker failed to start: {info}.{self._stderr_hint()}" + ) self.load_ms = info.get("ms") log.info("melo worker ready in %s ms on %s", self.load_ms, info.get("device")) @@ -93,7 +142,7 @@ class MeloTTS: await self._proc.stdin.drain() resp = await self._proc.stdout.readline() if not resp: - raise RuntimeError("melo worker closed unexpectedly") + raise RuntimeError(f"melo worker closed unexpectedly.{self._stderr_hint()}") res = json.loads(resp.decode()) if not res.get("ok"): raise RuntimeError(f"melo synth failed: {res.get('error')}") @@ -107,4 +156,11 @@ class MeloTTS: await asyncio.wait_for(self._proc.wait(), timeout=5) except (ProcessLookupError, asyncio.TimeoutError): pass + if self._stderr_task is not None: + self._stderr_task.cancel() + try: + await self._stderr_task + except (asyncio.CancelledError, Exception): + pass + self._stderr_task = None self._proc = None diff --git a/wsai/backends/whisper.py b/wsai/backends/whisper.py new file mode 100644 index 0000000..19ebf81 --- /dev/null +++ b/wsai/backends/whisper.py @@ -0,0 +1,187 @@ +"""Real STT via faster-whisper, run as a persistent out-of-venv worker. + +faster-whisper needs its own interpreter (whisper312) and loading the model +costs several seconds, so we keep one worker process alive and stream +transcription requests to it (see whisper_worker.py for the protocol) — the +same warm-worker shape as MeloTTS on the TTS side. + +`transcribe(wav)` is the core engine: hand it a wav path, get the recognised +text back. It is what closes the voice round trip (TTS wav -> STT text) and what +the Discord voice path will call once per detected utterance. + +`utterances()` turns this into a SpeechToText source: it pulls finished-utterance +wav paths from an injected `audio_source` and yields a transcribed Utterance for +each. Until the Discord voice receiver is wired, `audio_source` is None and the +loop simply idles (like the eyes-free perception loop), while `transcribe()` +stays usable directly. + +Env: + WSAI_WHISPER_PYTHON interpreter with faster-whisper installed + (default: /home/claude/jarvis-stt/whisper312/bin/python) + WSAI_WHISPER_MODEL model size/name (default: small) + WSAI_WHISPER_DEVICE cpu | cuda (default cpu; cuda needs GPU approval) + WSAI_WHISPER_LANGUAGE forced language, e.g. ko (default ko; "" = autodetect) +""" + +from __future__ import annotations + +import asyncio +import collections +import json +import logging +import os +import time +from pathlib import Path +from typing import AsyncIterator + +from ..interfaces import Utterance + +log = logging.getLogger("wsai.stt.whisper") + +_DEFAULT_PYTHON = "/home/claude/jarvis-stt/whisper312/bin/python" + + +class WhisperSTT: + def __init__( + self, + *, + python: str | None = None, + model: str | None = None, + device: str | None = None, + language: str | None = None, + audio_source: AsyncIterator[str] | None = None, + ) -> None: + self.python = python or os.environ.get("WSAI_WHISPER_PYTHON", _DEFAULT_PYTHON) + self.model = model or os.environ.get("WSAI_WHISPER_MODEL", "small") + self.device = device or os.environ.get("WSAI_WHISPER_DEVICE", "cpu") + # "" means autodetect; a real code like "ko" forces the language. + env_lang = os.environ.get("WSAI_WHISPER_LANGUAGE", "ko") + self.language = language if language is not None else (env_lang or None) + self.audio_source = audio_source + self._proc: asyncio.subprocess.Process | None = None + self._lock = asyncio.Lock() + self.load_ms: int | None = None + # Keep the worker's most recent stderr so a crash reports its real cause + # instead of a bare JSONDecodeError. Bounded so it can't grow unbounded. + self._stderr_tail: collections.deque[str] = collections.deque(maxlen=40) + self._stderr_task: asyncio.Task | None = None + + async def _drain_stderr(self, stream: asyncio.StreamReader) -> None: + # The worker redirects fd1 -> fd2, so model/library chatter lands on + # stderr. If we PIPE but never read it, the pipe buffer fills and the + # worker blocks. So we drain continuously, keeping only the last lines. + try: + while True: + line = await stream.readline() + if not line: + return + self._stderr_tail.append(line.decode(errors="replace").rstrip()) + except asyncio.CancelledError: + raise + except Exception: # draining must never crash the caller + return + + def _stderr_hint(self) -> str: + tail = "\n".join(self._stderr_tail) + return f" worker stderr tail:\n{tail}" if tail else " (worker produced no stderr)" + + async def warmup(self) -> None: + """Load the model now so the first real utterance is transcribed warm.""" + await self._ensure() + + async def _ensure(self) -> None: + if self._proc is not None and self._proc.returncode is None: + return + env = { + **os.environ, + "WSAI_WHISPER_MODEL": self.model, + "WSAI_WHISPER_DEVICE": self.device, + } + if self.language: + env["WSAI_WHISPER_LANGUAGE"] = self.language + repo_root = str(Path(__file__).resolve().parents[2]) + self._proc = await asyncio.create_subprocess_exec( + self.python, "-m", "wsai.backends.whisper_worker", + cwd=repo_root, env=env, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + self._stderr_tail.clear() + assert self._proc.stderr is not None + self._stderr_task = asyncio.create_task(self._drain_stderr(self._proc.stderr)) + ready = await self._proc.stdout.readline() + if not ready: # worker died before signalling ready + await self._proc.wait() + raise RuntimeError( + f"whisper worker exited before ready (code {self._proc.returncode})." + f"{self._stderr_hint()}" + ) + try: + info = json.loads(ready.decode()) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"whisper worker sent invalid ready line {ready!r}: {exc}." + f"{self._stderr_hint()}" + ) from exc + if not info.get("ready"): + raise RuntimeError( + f"whisper worker failed to start: {info}.{self._stderr_hint()}" + ) + self.load_ms = info.get("ms") + log.info( + "whisper worker ready in %s ms on %s (model %s)", + self.load_ms, info.get("device"), info.get("model"), + ) + + async def transcribe(self, wav_path: str, *, language: str | None = None) -> str: + """Transcribe one wav file to text using the warm worker.""" + await self._ensure() + req: dict[str, object] = {"wav": wav_path} + lang = language if language is not None else self.language + if lang: + req["language"] = lang + s = time.monotonic() + async with self._lock: + assert self._proc and self._proc.stdin and self._proc.stdout + self._proc.stdin.write((json.dumps(req) + "\n").encode()) + await self._proc.stdin.drain() + resp = await self._proc.stdout.readline() + if not resp: + raise RuntimeError(f"whisper worker closed unexpectedly.{self._stderr_hint()}") + res = json.loads(resp.decode()) + if not res.get("ok"): + raise RuntimeError(f"whisper transcribe failed: {res.get('error')}") + log.debug( + "transcribe %d ms (worker %s ms): %s", + int((time.monotonic() - s) * 1000), res.get("ms"), res.get("text", "")[:60], + ) + return res.get("text", "") + + async def utterances(self) -> AsyncIterator[Utterance]: + """Yield an Utterance per finished-utterance wav from `audio_source`. + + With no audio source wired yet (Discord voice receiver pending) this + idles and returns, leaving the voice loop dormant but valid.""" + if self.audio_source is None: + return + async for wav_path in self.audio_source: + text = await self.transcribe(wav_path) + if text: + yield Utterance(text=text, ts=time.monotonic(), source="voice") + + async def aclose(self) -> None: + if self._proc is not None and self._proc.returncode is None: + try: + self._proc.terminate() + await asyncio.wait_for(self._proc.wait(), timeout=5) + except (ProcessLookupError, asyncio.TimeoutError): + pass + if self._stderr_task is not None: + self._stderr_task.cancel() + try: + await self._stderr_task + except (asyncio.CancelledError, Exception): + pass + self._stderr_task = None + self._proc = None diff --git a/wsai/backends/whisper_worker.py b/wsai/backends/whisper_worker.py new file mode 100644 index 0000000..3fa70f4 --- /dev/null +++ b/wsai/backends/whisper_worker.py @@ -0,0 +1,81 @@ +"""Persistent faster-whisper STT worker. + +faster-whisper (ctranslate2) lives in its own Python (whisper312); loading the +model takes seconds, so we load it ONCE here and then serve transcription +requests over stdin/stdout. This process is launched with the whisper312 +interpreter by wsai.backends.whisper.WhisperSTT. + +Like the MeloTTS worker, model/backend chatter could corrupt the JSON protocol, +so on startup we split the streams: a private duplicate of the original stdout +carries the protocol, and fd 1 is redirected to fd 2 so any library print lands +on stderr instead (where the parent drains it for diagnostics). + +Protocol (one JSON object per line, on the protocol channel): + <- {"wav": "/abs/path.wav", "language": "ko"} + -> {"ok": true, "text": "...", "language": "ko", "ms": 123} + -> {"ok": false, "error": "..."} +On startup, once the model is ready, it emits exactly one line: + -> {"ready": true, "ms": , "device": "cpu", "model": "small"} +""" + +import json +import os +import sys +import time + +# Split protocol from library noise BEFORE importing anything heavy. +_proto = os.fdopen(os.dup(1), "w", buffering=1) # private copy of real stdout +os.dup2(2, 1) # fd1 -> stderr, so stray library prints don't hit the protocol + + +def _emit(obj: dict) -> None: + _proto.write(json.dumps(obj, ensure_ascii=False) + "\n") + _proto.flush() + + +def _log(*a): + print(*a, file=sys.stderr, flush=True) + + +def main() -> None: + model_name = os.environ.get("WSAI_WHISPER_MODEL", "small") + device = os.environ.get("WSAI_WHISPER_DEVICE", "cpu") # "cpu" | "cuda" + # int8 on CPU keeps a small model fast; float16 is the usual CUDA choice. + compute = os.environ.get( + "WSAI_WHISPER_COMPUTE", "int8" if device == "cpu" else "float16" + ) + default_lang = os.environ.get("WSAI_WHISPER_LANGUAGE", "ko") or None + + t0 = time.monotonic() + from faster_whisper import WhisperModel # heavy import; only in whisper venv + + model = WhisperModel(model_name, device=device, compute_type=compute) + load_ms = int((time.monotonic() - t0) * 1000) + _emit({"ready": True, "ms": load_ms, "device": device, "model": model_name}) + _log(f"[whisper_worker] {model_name} ready in {load_ms} ms on {device}/{compute}") + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + req = json.loads(line) + wav = req["wav"] + language = req.get("language", default_lang) + s = time.monotonic() + segments, info = model.transcribe( + wav, + language=language, + beam_size=int(req.get("beam_size", 5)), + vad_filter=bool(req.get("vad_filter", True)), + ) + text = "".join(seg.text for seg in segments).strip() + ms = int((time.monotonic() - s) * 1000) + _emit({"ok": True, "text": text, "language": info.language, "ms": ms}) + except Exception as exc: # keep the worker alive across bad requests + _emit({"ok": False, "error": f"{type(exc).__name__}: {exc}"}) + _log(f"[whisper_worker] error: {exc}") + + +if __name__ == "__main__": + main() diff --git a/wsai/config.py b/wsai/config.py index 19e7369..bf3aaa9 100644 --- a/wsai/config.py +++ b/wsai/config.py @@ -18,7 +18,7 @@ from dataclasses import dataclass class Settings: source: str | None = "mock" # mock | mss | None (eyes-free) vision: str | None = "mock" # mock | claude | None (eyes-free) - stt: str | None = "mock" # mock | (whisper) | None + stt: str | None = "mock" # mock | whisper | None tts: str | None = "mock" # mock | melo | None brain: str = "mock" # mock | claude text: str | None = None # None | (discord) diff --git a/wsai/factory.py b/wsai/factory.py index ca322d0..780adb0 100644 --- a/wsai/factory.py +++ b/wsai/factory.py @@ -69,6 +69,10 @@ def _brain(s: Settings): def _stt(s: Settings): if s.stt in (None, "none"): return None + if s.stt == "whisper": + from .backends.whisper import WhisperSTT + + return WhisperSTT() from .backends.mock import MockSTT return MockSTT() diff --git a/wsai/pipeline.py b/wsai/pipeline.py index cca3943..884639d 100644 --- a/wsai/pipeline.py +++ b/wsai/pipeline.py @@ -136,6 +136,28 @@ class Pipeline: await self._handle(utt) # -- lifecycle --------------------------------------------------------- # + async def _prewarm(self) -> None: + """Load slow-to-start backends before the loops accept input. + + Real STT/TTS backends (faster-whisper, MeloTTS) load a model into a + persistent worker on first use — several seconds on CPU. Warming them + here means the first real utterance is answered warm (~1s) instead of + paying the cold model load mid-conversation.""" + warmers = [] + for comp, name in ((self.tts, "tts"), (self.stt, "stt")): + warmup = getattr(comp, "warmup", None) + if callable(warmup): + warmers.append((name, warmup)) + if not warmers: + return + for name, warmup in warmers: + try: + await warmup() + except Exception as exc: # a warm failure must not abort startup + log.warning("prewarm %s failed: %s", name, exc) + if self.monitor is not None: + self.monitor.log("error", f"{name} 예열 실패: {exc}") + async def run(self) -> None: # A TaskGroup (not bare gather) so that if ONE loop raises, the others # are cancelled and awaited before teardown. With plain gather the @@ -145,6 +167,7 @@ class Pipeline: if self.monitor is not None: self.monitor.set_status(running=True) self.monitor.log("info", "파이프라인 시작") + await self._prewarm() try: async with asyncio.TaskGroup() as tg: tg.create_task(self._perceive())