feat(voice): real Discord listen+speak loop (STT->echo->TTS)
The bot (dave/bot.mjs) previously only joined the channel and counted audio frames — it never fed STT or spoke back. Wire the real loop: - Node bot: buffer each speaker's Opus->PCM utterance until AfterSilence, wrap as WAV, POST to the Python voice-turn endpoint, then play the returned reply wav into the channel via an AudioPlayer (ffmpeg->Opus). Skips its own audio, dedupes overlapping subscriptions, and ignores sub-0.35s noise. - Python: new `python -m wsai --voice-server` serves /api/voice-turn — decode the uploaded utterance, GPU faster-whisper STT, produce a reply (echo of what was heard for now), GPU MeloTTS synth, return the reply wav (recognised/reply text ride along as X-Heard/X-Reply headers). Both engines pre-warmed; turns show in the dashboard feed. MeloTTS.synth() extracted for direct wav reuse. Echo mode verifies listening+speaking+GPU recognition entirely in Discord; the Claude brain is the next slice. Verified the endpoint round-trip: utterance wav -> correct Korean X-Heard/X-Reply + a WAVE reply on device=cuda. 12 tests pass, node --check clean. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
EJClaw
86
dave/bot.mjs
86
dave/bot.mjs
@@ -19,6 +19,7 @@
|
||||
// "Speak" voice permissions. See dave/README or run `node bot.mjs --invite`.
|
||||
|
||||
import fs from 'node:fs';
|
||||
import { Readable } from 'node:stream';
|
||||
import {
|
||||
Client,
|
||||
GatewayIntentBits,
|
||||
@@ -29,6 +30,10 @@ import {
|
||||
entersState,
|
||||
VoiceConnectionStatus,
|
||||
EndBehaviorType,
|
||||
createAudioPlayer,
|
||||
createAudioResource,
|
||||
StreamType,
|
||||
NoSubscriberBehavior,
|
||||
} from '@discordjs/voice';
|
||||
import prism from 'prism-media';
|
||||
|
||||
@@ -47,10 +52,61 @@ const TOKEN = process.env.DISCORD_BOT_TOKEN || env.DISCORD_BOT_TOKEN;
|
||||
const GUILD_ID = process.env.GUILD_ID || env.GUILD_ID || '1352269198297923648';
|
||||
const CHANNEL_ID = process.env.CHANNEL_ID || env.CHANNEL_ID || '1352269198914621465';
|
||||
const RUN_MS = process.env.RUN_MS != null ? Number(process.env.RUN_MS) : 0; // 0 = stay forever
|
||||
// Python STT+TTS voice-turn endpoint (run `python -m wsai --voice-server`).
|
||||
const VOICE_ENDPOINT = process.env.WSAI_VOICE_ENDPOINT || env.WSAI_VOICE_ENDPOINT
|
||||
|| 'http://127.0.0.1:8787/api/voice-turn';
|
||||
// Ignore utterances shorter than this many PCM bytes (48kHz*2ch*2B = 192000 B/s),
|
||||
// so key clicks / brief noise don't trigger a turn. ~0.35s.
|
||||
const MIN_UTTERANCE_BYTES = Number(process.env.WSAI_MIN_UTTERANCE_BYTES || 67000);
|
||||
|
||||
const t0 = Date.now();
|
||||
const log = (...a) => console.log(`[+${String(Date.now() - t0).padStart(6)}ms]`, ...a);
|
||||
|
||||
// PCM s16le -> WAV container (so the Python side can ffmpeg-decode it).
|
||||
function wavHeader(dataLen, sampleRate = 48000, channels = 2, bits = 16) {
|
||||
const blockAlign = channels * bits / 8;
|
||||
const b = Buffer.alloc(44);
|
||||
b.write('RIFF', 0); b.writeUInt32LE(36 + dataLen, 4); b.write('WAVE', 8);
|
||||
b.write('fmt ', 12); b.writeUInt32LE(16, 16); b.writeUInt16LE(1, 20);
|
||||
b.writeUInt16LE(channels, 22); b.writeUInt32LE(sampleRate, 24);
|
||||
b.writeUInt32LE(sampleRate * blockAlign, 28); b.writeUInt16LE(blockAlign, 32);
|
||||
b.writeUInt16LE(bits, 34); b.write('data', 36); b.writeUInt32LE(dataLen, 40);
|
||||
return b;
|
||||
}
|
||||
|
||||
// Speak audio into the voice channel. Feed the reply wav bytes through ffmpeg
|
||||
// (StreamType.Arbitrary) so @discordjs/voice re-encodes to Opus.
|
||||
let voicePlayer = null;
|
||||
function playReply(wavBytes) {
|
||||
if (!voicePlayer || !wavBytes || wavBytes.length === 0) return;
|
||||
const resource = createAudioResource(Readable.from(wavBytes), { inputType: StreamType.Arbitrary });
|
||||
voicePlayer.play(resource);
|
||||
}
|
||||
|
||||
// One utterance: PCM -> WAV -> POST to Python -> play the reply back.
|
||||
async function handleUtterance(userId, pcm) {
|
||||
if (pcm.length < MIN_UTTERANCE_BYTES) {
|
||||
log(`utterance too short user=${userId} bytes=${pcm.length} — skip`);
|
||||
return;
|
||||
}
|
||||
const wav = Buffer.concat([wavHeader(pcm.length), pcm]);
|
||||
let resp;
|
||||
try {
|
||||
resp = await fetch(VOICE_ENDPOINT, {
|
||||
method: 'POST', headers: { 'Content-Type': 'audio/wav' }, body: wav,
|
||||
});
|
||||
} catch (e) {
|
||||
log(`voice-turn POST failed (is \`python -m wsai --voice-server\` running?): ${e.message}`);
|
||||
return;
|
||||
}
|
||||
if (!resp.ok) { log(`voice-turn HTTP ${resp.status}`); return; }
|
||||
const heard = decodeURIComponent(resp.headers.get('X-Heard') || '');
|
||||
const reply = decodeURIComponent(resp.headers.get('X-Reply') || '');
|
||||
const buf = Buffer.from(await resp.arrayBuffer());
|
||||
log(`heard="${heard}" reply="${reply}" replyWav=${buf.length}B`);
|
||||
playReply(buf);
|
||||
}
|
||||
|
||||
// ---------- invite URL helper ----------
|
||||
// A bot cannot add itself to a server; a server admin must click an OAuth2 invite
|
||||
// URL once. Print it so the user can authorise the bot with voice permissions.
|
||||
@@ -118,24 +174,40 @@ client.once('clientReady', async () => {
|
||||
}
|
||||
log(`✅ JOINED & READY. channel=${CHANNEL_ID} — staying connected, listening for speakers…`);
|
||||
|
||||
// ---------- receive path (partial M2) ----------
|
||||
// Playback path (bot speaks): one player, subscribed to the connection.
|
||||
voicePlayer = createAudioPlayer({ behaviors: { noSubscriber: NoSubscriberBehavior.Play } });
|
||||
voicePlayer.on('error', (e) => log(`player error: ${e.message}`));
|
||||
connection.subscribe(voicePlayer);
|
||||
log(`voice endpoint: ${VOICE_ENDPOINT}`);
|
||||
|
||||
// ---------- receive path: capture each utterance and run the voice turn ----
|
||||
const receiver = connection.receiver;
|
||||
const active = new Set(); // userIds with an in-flight subscription (avoid dupes)
|
||||
receiver.speaking.on('start', (userId) => {
|
||||
if (userId === client.user.id || active.has(userId)) return; // skip self / dupes
|
||||
active.add(userId);
|
||||
if (!perUser.has(userId)) perUser.set(userId, { opusPackets: 0, pcmFrames: 0 });
|
||||
log(`SPEAKING start user=${userId}`);
|
||||
const opusStream = receiver.subscribe(userId, {
|
||||
end: { behavior: EndBehaviorType.AfterSilence, duration: 500 },
|
||||
// End the utterance after a short silence so natural pauses don't cut words.
|
||||
end: { behavior: EndBehaviorType.AfterSilence, duration: 800 },
|
||||
});
|
||||
// Decode Opus -> 48kHz stereo s16le PCM (what faster-whisper will consume, downsampled to 16k in M3).
|
||||
// Decode Opus -> 48kHz stereo s16le PCM, buffered until the utterance ends.
|
||||
const decoder = new prism.opus.Decoder({ rate: 48000, channels: 2, frameSize: 960 });
|
||||
const chunks = [];
|
||||
opusStream.on('data', () => { perUser.get(userId).opusPackets++; });
|
||||
opusStream.pipe(decoder);
|
||||
decoder.on('data', () => {
|
||||
const s = perUser.get(userId);
|
||||
s.pcmFrames++;
|
||||
if (s.pcmFrames === 1 || s.pcmFrames % 200 === 0) log(`rtp: user=${userId} opus=${s.opusPackets} pcm=${s.pcmFrames}`);
|
||||
decoder.on('data', (d) => {
|
||||
chunks.push(d);
|
||||
const s = perUser.get(userId); s.pcmFrames++;
|
||||
});
|
||||
decoder.on('error', (e) => log(`decode error user=${userId}: ${e.message}`));
|
||||
decoder.on('end', () => {
|
||||
active.delete(userId);
|
||||
const pcm = Buffer.concat(chunks);
|
||||
log(`utterance end user=${userId} pcm=${pcm.length}B — running voice turn`);
|
||||
handleUtterance(userId, pcm).catch((e) => log(`voice turn error: ${e.message}`));
|
||||
});
|
||||
});
|
||||
|
||||
connection.on(VoiceConnectionStatus.Disconnected, () => {
|
||||
|
||||
@@ -100,6 +100,47 @@ def _run_stt_test(host: str, port: int) -> None:
|
||||
dash.stop()
|
||||
|
||||
|
||||
def _run_voice_server(host: str, port: int) -> None:
|
||||
"""Serve the STT+TTS voice-turn endpoint that the Discord bot (dave/bot.mjs)
|
||||
calls: it POSTs a captured utterance wav and gets back the reply wav to play
|
||||
into the voice channel. Both STT and TTS run on the GPU and are pre-warmed.
|
||||
The same page also shows the live turn feed. Echo mode for now (the reply is
|
||||
what was heard); the Claude brain can be added as the next slice."""
|
||||
import time
|
||||
|
||||
from .backends.melo import MeloTTS
|
||||
from .backends.whisper import WhisperSTT
|
||||
from .dashboard import Dashboard
|
||||
from .monitor import Monitor
|
||||
|
||||
monitor = Monitor()
|
||||
stt = WhisperSTT()
|
||||
tts = MeloTTS()
|
||||
dash = Dashboard(monitor, host=host, port=port, stt=stt, tts=tts)
|
||||
dash.start()
|
||||
monitor.set_components({"source": "none", "vision": "none", "stt": "whisper",
|
||||
"brain": "echo", "tts": "melo"})
|
||||
monitor.set_status(running=True, listening=False)
|
||||
monitor.log("info", "디스코드 음성 서버 시작 — STT+TTS GPU 워밍업 중…")
|
||||
print("\n STT+TTS 워밍업 중… (모델 로드 + CUDA 예열)")
|
||||
dash.warm()
|
||||
sdev = getattr(stt, "resolved_device", None) or "?"
|
||||
monitor.set_status(listening=True)
|
||||
monitor.log("info", f"음성 서버 준비 완료 (STT device={sdev}). 디스코드 봇 연결 대기.")
|
||||
|
||||
shown = host if host not in ("0.0.0.0", "") else _lan_ip()
|
||||
print(f"\n 음성 서버 준비 완료 (STT device: {sdev})")
|
||||
print(f" 대시보드/상태: http://{shown}:{port}")
|
||||
print(f" 봇 연결 엔드포인트: http://127.0.0.1:{port}/api/voice-turn\n")
|
||||
try:
|
||||
while True:
|
||||
time.sleep(3600)
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
dash.stop()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(prog="wsai")
|
||||
ap.add_argument("--voice", action="store_true", help="eyes-free voice loop (STT -> Brain -> TTS)")
|
||||
@@ -108,6 +149,8 @@ def main() -> None:
|
||||
help="keep generating mock demo utterances forever (off by default)")
|
||||
ap.add_argument("--stt-test", action="store_true",
|
||||
help="serve the dashboard with a live GPU STT recognition test")
|
||||
ap.add_argument("--voice-server", action="store_true",
|
||||
help="serve STT+TTS voice-turn endpoint for the Discord bot (dave/bot.mjs)")
|
||||
ap.add_argument("--live", action="store_true", help="capture screen + Claude backends")
|
||||
ap.add_argument("--env", action="store_true", help="build from WSAI_* env vars")
|
||||
ap.add_argument("--port", type=int, default=int(os.environ.get("WSAI_DASHBOARD_PORT", "8787")),
|
||||
@@ -126,6 +169,10 @@ def main() -> None:
|
||||
_run_stt_test(args.host, args.port)
|
||||
return
|
||||
|
||||
if args.voice_server:
|
||||
_run_voice_server(args.host, args.port)
|
||||
return
|
||||
|
||||
if args.dashboard:
|
||||
# Default to the eyes-free voice preset for the demo; env can override.
|
||||
settings = Settings.from_env() if args.env else Settings.voice()
|
||||
|
||||
@@ -131,11 +131,13 @@ class MeloTTS:
|
||||
self.load_ms = info.get("ms")
|
||||
log.info("melo worker ready in %s ms on %s", self.load_ms, info.get("device"))
|
||||
|
||||
async def speak(self, reply: Reply) -> None:
|
||||
async def synth(self, text: str) -> str:
|
||||
"""Synthesize `text` to a wav and return its path (no sink). Reusable by
|
||||
callers that want the wav directly (e.g. the Discord voice bridge)."""
|
||||
await self._ensure()
|
||||
self._n += 1
|
||||
out = str(self.out_dir / f"tts-{self._n:06d}.wav")
|
||||
req = json.dumps({"text": reply.text, "out": out, "speed": self.speed})
|
||||
req = json.dumps({"text": text, "out": out, "speed": self.speed})
|
||||
s = time.monotonic()
|
||||
async with self._lock:
|
||||
assert self._proc and self._proc.stdin and self._proc.stdout
|
||||
@@ -148,7 +150,11 @@ class MeloTTS:
|
||||
if not res.get("ok"):
|
||||
raise RuntimeError(f"melo synth failed: {res.get('error')}")
|
||||
log.debug("synth %d ms (worker %s ms)", int((time.monotonic() - s) * 1000), res.get("ms"))
|
||||
await self.sink(res["out"], reply)
|
||||
return res["out"]
|
||||
|
||||
async def speak(self, reply: Reply) -> None:
|
||||
out = await self.synth(reply.text)
|
||||
await self.sink(out, reply)
|
||||
|
||||
async def aclose(self) -> None:
|
||||
if self._proc is not None and self._proc.returncode is None:
|
||||
|
||||
@@ -58,9 +58,51 @@ def _make_handler(dash: "Dashboard"):
|
||||
path = self.path.split("?", 1)[0]
|
||||
if path == "/api/stt":
|
||||
self._handle_stt()
|
||||
elif path == "/api/voice-turn":
|
||||
self._handle_voice_turn()
|
||||
else:
|
||||
self._send(404, b"not found", "text/plain; charset=utf-8")
|
||||
|
||||
def _read_body(self) -> bytes:
|
||||
try:
|
||||
length = int(self.headers.get("Content-Length", "0"))
|
||||
except ValueError:
|
||||
length = 0
|
||||
return self.rfile.read(length) if length > 0 else b""
|
||||
|
||||
def _handle_voice_turn(self) -> None:
|
||||
"""Discord voice bridge: utterance wav in -> reply wav out. The
|
||||
recognised/reply text ride along as URL-encoded response headers so
|
||||
the bot can log them; the body is the reply audio to play back."""
|
||||
import urllib.parse
|
||||
|
||||
if dash.stt is None or dash.tts is None:
|
||||
self._send(503, json.dumps({"ok": False, "error": "voice loop not enabled"}).encode(),
|
||||
"application/json; charset=utf-8")
|
||||
return
|
||||
raw = self._read_body()
|
||||
if not raw:
|
||||
self._send(400, json.dumps({"ok": False, "error": "empty upload"}).encode(),
|
||||
"application/json; charset=utf-8")
|
||||
return
|
||||
try:
|
||||
res = dash.voice_turn(raw)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.exception("voice-turn failed")
|
||||
self._send(500, json.dumps({"ok": False, "error": f"{type(exc).__name__}: {exc}"},
|
||||
ensure_ascii=False).encode(), "application/json; charset=utf-8")
|
||||
return
|
||||
body = res["wav"]
|
||||
self.send_response(200 if body else 204)
|
||||
self.send_header("Content-Type", "audio/wav")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.send_header("X-Heard", urllib.parse.quote(res.get("heard", "")))
|
||||
self.send_header("X-Reply", urllib.parse.quote(res.get("reply", "")))
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.end_headers()
|
||||
if body:
|
||||
self.wfile.write(body)
|
||||
|
||||
def _handle_stt(self) -> None:
|
||||
"""Accept an uploaded audio blob (mic recording or file), run it
|
||||
through the real GPU STT, and return the recognised text."""
|
||||
@@ -130,18 +172,19 @@ class Dashboard:
|
||||
"""
|
||||
|
||||
def __init__(self, monitor: Monitor, host: str = "0.0.0.0", port: int = 8787,
|
||||
stt=None) -> None:
|
||||
stt=None, tts=None) -> None:
|
||||
self.monitor = monitor
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.stt = stt
|
||||
self.tts = tts
|
||||
self._server: ThreadingHTTPServer | None = None
|
||||
self._thread: threading.Thread | None = None
|
||||
self._loop = None
|
||||
self._loop_thread: threading.Thread | None = None
|
||||
|
||||
def start(self) -> None:
|
||||
if self.stt is not None:
|
||||
if self.stt is not None or self.tts is not None:
|
||||
self._start_loop()
|
||||
handler = _make_handler(self)
|
||||
self._server = ThreadingHTTPServer((self.host, self.port), handler)
|
||||
@@ -178,10 +221,72 @@ class Dashboard:
|
||||
return fut.result(timeout=timeout)
|
||||
|
||||
def warm(self) -> None:
|
||||
"""Pre-start the STT worker (loads + warms the GPU) so the first web
|
||||
recognition is instant instead of paying model-load + CUDA autotune."""
|
||||
"""Pre-start the STT/TTS workers (loads + warms the GPU) so the first
|
||||
recognition/synth is instant instead of paying model-load + CUDA autotune."""
|
||||
if self.stt is not None:
|
||||
self._submit(self.stt._ensure())
|
||||
if self.tts is not None:
|
||||
self._submit(self.tts._ensure())
|
||||
|
||||
def voice_turn(self, audio_bytes: bytes) -> dict:
|
||||
"""One Discord voice turn: decode the uploaded utterance, recognise it
|
||||
on the GPU, produce a reply (echo of what was heard for now), synthesise
|
||||
it on the GPU, and return {heard, reply, wav} where wav is the reply
|
||||
audio bytes for the bot to play back into the channel."""
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
import uuid
|
||||
|
||||
if self.stt is None or self.tts is None:
|
||||
raise RuntimeError("voice_turn needs both STT and TTS")
|
||||
updir = os.path.expanduser("~/.cache/wsai/uploads")
|
||||
os.makedirs(updir, exist_ok=True)
|
||||
stem = os.path.join(updir, uuid.uuid4().hex)
|
||||
src, wav = stem + ".bin", stem + ".wav"
|
||||
with open(src, "wb") as f:
|
||||
f.write(audio_bytes)
|
||||
turn = self.monitor.turn(source="discord")
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
subprocess.run(
|
||||
["ffmpeg", "-y", "-i", src, "-ar", "16000", "-ac", "1", wav],
|
||||
check=True, capture_output=True,
|
||||
)
|
||||
heard = self._submit(self.stt.transcribe(wav)) or ""
|
||||
turn.heard(heard or "(빈 결과)")
|
||||
reply_text = heard.strip()
|
||||
if not reply_text:
|
||||
# Nothing recognised (silence/noise): skip TTS, tell the bot.
|
||||
turn.finish()
|
||||
return {"heard": heard, "reply": "", "wav": b""}
|
||||
out_path = self._submit(self.tts.synth(reply_text))
|
||||
with open(out_path, "rb") as f:
|
||||
reply_wav = f.read()
|
||||
ms = int((time.monotonic() - t0) * 1000)
|
||||
turn.replied(reply_text)
|
||||
step = turn.step("STT+TTS(GPU)")
|
||||
step.ok, step.ms = True, float(ms)
|
||||
turn._steps.append(step)
|
||||
turn.finish()
|
||||
try:
|
||||
os.remove(out_path)
|
||||
except OSError:
|
||||
pass
|
||||
return {"heard": heard, "reply": reply_text, "wav": reply_wav}
|
||||
except subprocess.CalledProcessError as exc:
|
||||
turn.finish(error="ffmpeg decode failed")
|
||||
err = exc.stderr.decode("utf-8", "replace")[-300:] if exc.stderr else str(exc)
|
||||
raise RuntimeError(f"ffmpeg: {err}") from exc
|
||||
except Exception as exc:
|
||||
turn.finish(error=str(exc))
|
||||
raise
|
||||
finally:
|
||||
for p in (src, wav):
|
||||
try:
|
||||
os.remove(p)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def transcribe_upload(self, audio_bytes: bytes) -> dict:
|
||||
"""ffmpeg-normalise an uploaded blob to 16 kHz mono wav, transcribe it
|
||||
|
||||
Reference in New Issue
Block a user