39c7a22a12
The bridge transcribe path joined every Whisper segment unconditionally, so a
brief loud sound or background noise that momentarily opened the mic gate (no
real speech) still produced a transcript, and Whisper's noise hallucinations
("감사합니다", "MBC 뉴스", ...) made the bot reply to nothing.
Add bridge/stt_filter.py mirroring the desktop listener's _filter_noisy_segments
policy: a hard no_speech_prob cutoff (whisper_no_speech_threshold) plus an
avg_logprob confidence floor (whisper_min_confidence), both config-driven. Apply
it in transcribe() so only segments that look like human speech survive; a
noise-only turn yields an empty transcript and the existing empty-transcript
guard drops it with no reply. Add unit tests for the gate.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
459 lines
18 KiB
Python
459 lines
18 KiB
Python
"""
|
|
Jarvis Brain Bridge
|
|
===================
|
|
|
|
A thin local HTTP service that exposes the existing Jarvis "brain"
|
|
(speech-to-text + reply engine + text-to-speech) to the Node/bun Discord bot.
|
|
|
|
The Discord layer (``bot/``) is responsible for everything Discord-specific:
|
|
joining voice channels, capturing user audio, playing audio back, slash
|
|
commands, and streaming the VNC screen. It does NOT contain any AI logic.
|
|
Instead it calls this bridge:
|
|
|
|
POST /converse (multipart wav) -> { transcript, reply, audio_b64 }
|
|
POST /text (json {text}) -> { reply, audio_b64 }
|
|
POST /stt (multipart wav) -> { text, language }
|
|
POST /tts (json {text}) -> { audio_b64 }
|
|
GET /health -> { ok, brain, stt, tts }
|
|
|
|
This keeps the mature ~39k-line Python brain intact while letting Node own the
|
|
Discord/voice/video integration (which is only feasible in the Node ecosystem).
|
|
|
|
Run:
|
|
python -m bridge.server # from repo root
|
|
# or
|
|
BRIDGE_HOST=127.0.0.1 BRIDGE_PORT=8765 python bridge/server.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
import wave
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
# Ensure repo-root/src is importable (jarvis package lives in src/jarvis) and
|
|
# the repo root itself (so ``bridge.text_utils`` resolves whether this module is
|
|
# launched as ``python -m bridge.server`` or ``python bridge/server.py``).
|
|
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
_SRC = _REPO_ROOT / "src"
|
|
if str(_SRC) not in sys.path:
|
|
sys.path.insert(0, str(_SRC))
|
|
if str(_REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(_REPO_ROOT))
|
|
|
|
from flask import Flask, request, jsonify, Response, stream_with_context
|
|
|
|
try: # package-relative when imported as ``bridge.server``
|
|
from bridge.text_utils import split_sentences
|
|
from bridge.stt_filter import filter_speech_segments
|
|
except ImportError: # script-relative when run as ``bridge/server.py``
|
|
from text_utils import split_sentences
|
|
from stt_filter import filter_speech_segments
|
|
|
|
app = Flask(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Configuration (env-driven; see .env.example)
|
|
# ---------------------------------------------------------------------------
|
|
BRIDGE_HOST = os.environ.get("BRIDGE_HOST", "127.0.0.1")
|
|
BRIDGE_PORT = int(os.environ.get("BRIDGE_PORT", "8765"))
|
|
BRAIN_ENABLED = os.environ.get("JARVIS_BRAIN_ENABLED", "1") not in ("0", "false", "False")
|
|
TTS_ENABLED = os.environ.get("JARVIS_TTS_ENABLED", "1") not in ("0", "false", "False")
|
|
|
|
# TTS engine: "melo" (MeloTTS Korean speaker, the warm worker) is the primary
|
|
# voice; Piper is kept as a fallback if the worker is unreachable. Set
|
|
# TTS_ENGINE=piper to disable MeloTTS entirely.
|
|
TTS_ENGINE = os.environ.get("TTS_ENGINE", "melo").strip().lower()
|
|
MELO_WORKER_URL = os.environ.get("MELO_WORKER_URL", "http://127.0.0.1:8770")
|
|
MELO_TIMEOUT = float(os.environ.get("MELO_TIMEOUT", "30"))
|
|
# When MeloTTS is the engine, do NOT silently fall back to the English Piper
|
|
# voice on failure: speaking Korean text through an English voice produces
|
|
# mangled audio. Default is melo-only (return no audio on failure); set
|
|
# MELO_FALLBACK_PIPER=1 to opt into the Piper fallback.
|
|
MELO_FALLBACK_PIPER = os.environ.get("MELO_FALLBACK_PIPER", "0") in ("1", "true", "True", "yes", "on")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lazy singletons. The first request pays the model-load cost; afterwards the
|
|
# brain stays warm. A lock guards initialization so concurrent Discord events
|
|
# don't double-load Whisper.
|
|
# ---------------------------------------------------------------------------
|
|
_init_lock = threading.Lock()
|
|
_cfg = None
|
|
_db = None
|
|
_dialogue_memory = None
|
|
_whisper = None
|
|
_piper_voice = None
|
|
_brain_error: Optional[str] = None
|
|
|
|
|
|
def _ensure_brain():
|
|
"""Initialize cfg, db, dialogue memory, and Whisper once."""
|
|
global _cfg, _db, _dialogue_memory, _whisper, _brain_error
|
|
if _cfg is not None or _brain_error is not None:
|
|
return
|
|
with _init_lock:
|
|
if _cfg is not None or _brain_error is not None:
|
|
return
|
|
try:
|
|
from jarvis.config import load_settings
|
|
from jarvis.memory.db import Database
|
|
from jarvis.memory.conversation import DialogueMemory
|
|
from faster_whisper import WhisperModel
|
|
|
|
cfg = load_settings()
|
|
db = Database(cfg.db_path, cfg.sqlite_vss_path)
|
|
dialogue_memory = DialogueMemory(
|
|
inactivity_timeout=getattr(cfg, "dialogue_memory_timeout", 300.0),
|
|
max_interactions=20,
|
|
)
|
|
device = os.environ.get("WHISPER_DEVICE", "auto")
|
|
compute = os.environ.get("WHISPER_COMPUTE_TYPE", "auto")
|
|
try:
|
|
whisper = WhisperModel(cfg.whisper_model, device=device, compute_type=compute)
|
|
except Exception as ge:
|
|
# GPU not available / unsupported -> fall back to CPU so the
|
|
# bridge still works without a GPU passed to the container.
|
|
if device != "cpu":
|
|
print(f"[bridge] whisper device='{device}' failed ({ge}); falling back to CPU", flush=True)
|
|
whisper = WhisperModel(cfg.whisper_model, device="cpu", compute_type="int8")
|
|
else:
|
|
raise
|
|
|
|
_cfg, _db, _dialogue_memory, _whisper = cfg, db, dialogue_memory, whisper
|
|
print(f"[bridge] brain ready (chat={cfg.ollama_chat_model}, whisper={cfg.whisper_model})", flush=True)
|
|
except Exception as e: # pragma: no cover - depends on local models
|
|
_brain_error = f"{type(e).__name__}: {e}"
|
|
print(f"[bridge] brain init FAILED: {_brain_error}", flush=True)
|
|
|
|
|
|
def _ensure_piper():
|
|
"""Initialize the Piper TTS voice once (independent of the brain)."""
|
|
global _piper_voice
|
|
if _piper_voice is not None or not TTS_ENABLED:
|
|
return
|
|
with _init_lock:
|
|
if _piper_voice is not None:
|
|
return
|
|
try:
|
|
from piper import PiperVoice # piper-tts package
|
|
model_path = os.environ.get("TTS_PIPER_MODEL_PATH")
|
|
if not model_path:
|
|
# Fall back to jarvis' default piper model location.
|
|
from jarvis.output.tts import _get_default_piper_model_path # type: ignore
|
|
model_path = _get_default_piper_model_path()
|
|
if not model_path or not Path(model_path).exists():
|
|
raise FileNotFoundError(
|
|
f"Piper voice model not found at '{model_path}'. "
|
|
f"Set TTS_PIPER_MODEL_PATH in .env or run scripts/setup_models.sh"
|
|
)
|
|
_piper_voice = PiperVoice.load(model_path)
|
|
print(f"[bridge] piper TTS ready ({model_path})", flush=True)
|
|
except Exception as e: # pragma: no cover
|
|
print(f"[bridge] piper init failed (TTS disabled): {e}", flush=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Core operations
|
|
# ---------------------------------------------------------------------------
|
|
def _read_wav_pcm(raw: bytes) -> tuple[bytes, int]:
|
|
"""Decode an incoming WAV blob to mono 16-bit PCM @ its sample rate."""
|
|
with wave.open(io.BytesIO(raw), "rb") as wf:
|
|
sr = wf.getframerate()
|
|
frames = wf.readframes(wf.getnframes())
|
|
return frames, sr
|
|
|
|
|
|
def transcribe(wav_bytes: bytes) -> dict:
|
|
_ensure_brain()
|
|
if _whisper is None:
|
|
return {"text": "", "language": None, "error": _brain_error or "stt unavailable"}
|
|
import numpy as np
|
|
|
|
pcm, sr = _read_wav_pcm(wav_bytes)
|
|
audio = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) / 32768.0
|
|
# faster-whisper expects 16kHz mono float32; linearly resample if needed.
|
|
if sr != 16000 and audio.size:
|
|
n_out = int(round(audio.size * 16000 / sr))
|
|
if n_out > 0:
|
|
x_old = np.linspace(0.0, 1.0, num=audio.size, endpoint=False)
|
|
x_new = np.linspace(0.0, 1.0, num=n_out, endpoint=False)
|
|
audio = np.interp(x_new, x_old, audio).astype(np.float32)
|
|
segments, info = _whisper.transcribe(audio, beam_size=1)
|
|
# Speech gate: drop non-speech / hallucinated segments so a brief loud sound
|
|
# or background noise (mic blip with no real speech) does not become a
|
|
# transcript and make the bot reply to nothing. Mirrors the desktop
|
|
# listener's policy, driven by the same config thresholds.
|
|
no_speech_threshold = getattr(_cfg, "whisper_no_speech_threshold", 0.5)
|
|
min_confidence = getattr(_cfg, "whisper_min_confidence", 0.3)
|
|
kept = filter_speech_segments(
|
|
segments,
|
|
no_speech_threshold=no_speech_threshold,
|
|
min_confidence=min_confidence,
|
|
log=lambda m: print(f"[bridge] {m}", flush=True),
|
|
)
|
|
text = "".join(seg.text for seg in kept).strip()
|
|
return {"text": text, "language": getattr(info, "language", None)}
|
|
|
|
|
|
def think(text: str, language: Optional[str] = None, broadcasting: Optional[bool] = None) -> dict:
|
|
"""Run the Jarvis reply engine on a piece of text.
|
|
|
|
``broadcasting`` is the bot's live screen-share state for this turn; it is
|
|
stashed in request-scoped state so the search routing can pick Chrome vs
|
|
Gemini. If the reply engine calls the setBroadcast tool, the recorded
|
|
directive is returned as ``broadcast_action`` for the bot to act on.
|
|
"""
|
|
if not BRAIN_ENABLED:
|
|
return {"reply": text, "error": "brain disabled (JARVIS_BRAIN_ENABLED=0)"}
|
|
_ensure_brain()
|
|
if _cfg is None:
|
|
return {"reply": "", "error": _brain_error or "brain unavailable"}
|
|
try:
|
|
from jarvis.reply.engine import run_reply_engine
|
|
from jarvis.reply import turn_state
|
|
|
|
turn_state.reset()
|
|
turn_state.set_broadcasting(broadcasting)
|
|
|
|
# tts=None: we do our own Discord-side synthesis, the engine must not
|
|
# try to speak to a local speaker that doesn't exist in this process.
|
|
reply = run_reply_engine(
|
|
_db, _cfg, None, text, _dialogue_memory, language=language
|
|
)
|
|
reply = (reply or "").strip()
|
|
if reply:
|
|
_dialogue_memory.add_interaction(text, reply)
|
|
return {"reply": reply, "broadcast_action": turn_state.get_broadcast_action()}
|
|
except Exception as e: # pragma: no cover
|
|
return {"reply": "", "error": f"{type(e).__name__}: {e}"}
|
|
|
|
|
|
def _coerce_bool(value) -> Optional[bool]:
|
|
"""Parse a broadcasting flag from JSON (bool) or a query string."""
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool):
|
|
return value
|
|
return str(value).strip().lower() in ("1", "true", "yes", "on")
|
|
|
|
|
|
def _melo_synthesize(text: str) -> Optional[bytes]:
|
|
"""Synthesise via the warm MeloTTS worker (separate /opt/melo venv, Korean
|
|
speaker @ speed 1.5). Returns a 16-bit PCM WAV, or None on any failure so
|
|
the caller can fall back to Piper."""
|
|
import urllib.request
|
|
|
|
try:
|
|
req = urllib.request.Request(
|
|
f"{MELO_WORKER_URL}/synth",
|
|
data=json.dumps({"text": text}).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=MELO_TIMEOUT) as resp:
|
|
if resp.status == 200:
|
|
return resp.read()
|
|
print(f"[bridge] melo worker HTTP {resp.status}", flush=True)
|
|
except Exception as e: # pragma: no cover - worker may be down
|
|
print(f"[bridge] melo worker unreachable: {e}", flush=True)
|
|
return None
|
|
|
|
|
|
def _piper_synthesize(text: str) -> Optional[bytes]:
|
|
"""Fallback: synthesise with Piper (English voice). Returns WAV bytes."""
|
|
_ensure_piper()
|
|
if _piper_voice is None:
|
|
return None
|
|
buf = io.BytesIO()
|
|
with wave.open(buf, "wb") as wf:
|
|
# piper-tts API: synthesize_wav(text, wav_file) writes a full WAV;
|
|
# plain synthesize() returns AudioChunks and takes a SynthesisConfig
|
|
# (NOT a wav file) as its 2nd arg.
|
|
_piper_voice.synthesize_wav(text, wf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def _tts_ready() -> bool:
|
|
"""Whether the configured TTS voice can synthesise right now.
|
|
|
|
The bot polls this before logging in so the very first spoken reply is not
|
|
silently dropped while the voice is still warming up. For MeloTTS the worker
|
|
only binds its HTTP port AFTER the model is loaded (``main()`` warms the
|
|
model before ``serve_forever()``), so a successful /health ping is a precise
|
|
"voice is warm" signal. Piper loads on first synth and was never gated, so
|
|
it reports ready. TTS disabled means there is nothing to wait for.
|
|
"""
|
|
if not TTS_ENABLED:
|
|
return True
|
|
if TTS_ENGINE == "melo":
|
|
import urllib.request
|
|
|
|
try:
|
|
with urllib.request.urlopen(f"{MELO_WORKER_URL}/health", timeout=2) as resp:
|
|
return resp.status == 200
|
|
except Exception:
|
|
return False
|
|
return True
|
|
|
|
|
|
def synthesize(text: str) -> Optional[bytes]:
|
|
"""Synthesize text to a 16-bit PCM WAV. The primary voice is MeloTTS
|
|
(Korean speaker, speed 1.5) served by the warm melo worker; Piper is a
|
|
fallback if the worker is unavailable. Returns None if TTS is off."""
|
|
if not TTS_ENABLED or not text.strip():
|
|
return None
|
|
if TTS_ENGINE == "melo":
|
|
audio = _melo_synthesize(text)
|
|
if audio:
|
|
return audio
|
|
if not MELO_FALLBACK_PIPER:
|
|
# Melo-only: better silent than mangled English for Korean text.
|
|
print("[bridge] melo synth failed; no audio (Piper fallback disabled)", flush=True)
|
|
return None
|
|
print("[bridge] melo synth failed; falling back to Piper", flush=True)
|
|
return _piper_synthesize(text)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HTTP endpoints
|
|
# ---------------------------------------------------------------------------
|
|
@app.get("/health")
|
|
def health():
|
|
return jsonify(
|
|
{
|
|
"ok": True,
|
|
"brain_enabled": BRAIN_ENABLED,
|
|
"brain_ready": _cfg is not None,
|
|
"brain_error": _brain_error,
|
|
"tts_enabled": TTS_ENABLED,
|
|
"tts_engine": TTS_ENGINE,
|
|
"tts_ready": _tts_ready(),
|
|
}
|
|
)
|
|
|
|
|
|
@app.post("/stt")
|
|
def http_stt():
|
|
raw = request.get_data()
|
|
if not raw:
|
|
return jsonify({"error": "empty body; send a WAV blob"}), 400
|
|
return jsonify(transcribe(raw))
|
|
|
|
|
|
@app.post("/text")
|
|
def http_text():
|
|
data = request.get_json(silent=True) or {}
|
|
text = (data.get("text") or "").strip()
|
|
if not text:
|
|
return jsonify({"error": "missing 'text'"}), 400
|
|
result = think(text, data.get("language"), _coerce_bool(data.get("broadcasting")))
|
|
audio = synthesize(result.get("reply", ""))
|
|
if audio:
|
|
result["audio_b64"] = base64.b64encode(audio).decode("ascii")
|
|
return jsonify(result)
|
|
|
|
|
|
@app.post("/tts")
|
|
def http_tts():
|
|
data = request.get_json(silent=True) or {}
|
|
text = (data.get("text") or "").strip()
|
|
if not text:
|
|
return jsonify({"error": "missing 'text'"}), 400
|
|
audio = synthesize(text)
|
|
if not audio:
|
|
return jsonify({"error": "tts unavailable"}), 503
|
|
return jsonify({"audio_b64": base64.b64encode(audio).decode("ascii")})
|
|
|
|
|
|
@app.post("/converse")
|
|
def http_converse():
|
|
"""Full turn: speech in -> transcript -> reply -> speech out."""
|
|
raw = request.get_data()
|
|
if not raw:
|
|
return jsonify({"error": "empty body; send a WAV blob"}), 400
|
|
stt = transcribe(raw)
|
|
transcript = stt.get("text", "")
|
|
if not transcript:
|
|
return jsonify({"transcript": "", "reply": "", "audio_b64": None})
|
|
broadcasting = _coerce_bool(request.args.get("broadcasting"))
|
|
result = think(transcript, stt.get("language"), broadcasting)
|
|
audio = synthesize(result.get("reply", ""))
|
|
return jsonify(
|
|
{
|
|
"transcript": transcript,
|
|
"language": stt.get("language"),
|
|
"reply": result.get("reply", ""),
|
|
"error": result.get("error"),
|
|
"broadcast_action": result.get("broadcast_action"),
|
|
"audio_b64": base64.b64encode(audio).decode("ascii") if audio else None,
|
|
}
|
|
)
|
|
|
|
|
|
@app.post("/converse_stream")
|
|
def http_converse_stream():
|
|
"""Streaming full turn: speech in -> transcript -> reply -> speech out.
|
|
|
|
Reduces perceived latency by synthesising the reply one sentence at a time
|
|
and emitting each clip as soon as it is ready, so the Discord layer can play
|
|
the first sentence while the rest are still being spoken. The response is
|
|
newline-delimited JSON (NDJSON):
|
|
|
|
{"type":"meta","transcript":..,"language":..,"reply":..,"error":..,"broadcast_action":..}
|
|
{"type":"audio","seq":0,"audio_b64":..}
|
|
{"type":"audio","seq":1,"audio_b64":..}
|
|
{"type":"end"}
|
|
|
|
STT and the reply engine still run to completion before the meta line; only
|
|
TTS is pipelined. The non-streaming /converse endpoint is unchanged.
|
|
"""
|
|
raw = request.get_data()
|
|
if not raw:
|
|
return jsonify({"error": "empty body; send a WAV blob"}), 400
|
|
broadcasting = _coerce_bool(request.args.get("broadcasting"))
|
|
|
|
def gen():
|
|
stt = transcribe(raw)
|
|
transcript = stt.get("text", "")
|
|
if not transcript:
|
|
yield json.dumps({"type": "meta", "transcript": "", "language": stt.get("language"),
|
|
"reply": "", "error": stt.get("error"), "broadcast_action": None}) + "\n"
|
|
yield json.dumps({"type": "end"}) + "\n"
|
|
return
|
|
result = think(transcript, stt.get("language"), broadcasting)
|
|
reply = result.get("reply", "")
|
|
yield json.dumps({
|
|
"type": "meta",
|
|
"transcript": transcript,
|
|
"language": stt.get("language"),
|
|
"reply": reply,
|
|
"error": result.get("error"),
|
|
"broadcast_action": result.get("broadcast_action"),
|
|
}) + "\n"
|
|
for seq, sentence in enumerate(split_sentences(reply)):
|
|
audio = synthesize(sentence)
|
|
if audio:
|
|
yield json.dumps({
|
|
"type": "audio",
|
|
"seq": seq,
|
|
"audio_b64": base64.b64encode(audio).decode("ascii"),
|
|
}) + "\n"
|
|
yield json.dumps({"type": "end"}) + "\n"
|
|
|
|
return Response(stream_with_context(gen()), mimetype="application/x-ndjson")
|
|
|
|
|
|
def main():
|
|
print(f"[bridge] listening on http://{BRIDGE_HOST}:{BRIDGE_PORT}", flush=True)
|
|
# threaded=True so STT (slow) on one request doesn't block /health, etc.
|
|
app.run(host=BRIDGE_HOST, port=BRIDGE_PORT, threaded=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|