tkrmagid/javis_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 20 Wiki Activity
Files
de5384d166361a0df17d37340e7845e91a40c1e3
javis_bot/bridge/server.py
javis-bot de5384d166 feat: show per-stage timing (듣기/LLM/TTS) in the transcript channel 
The transcript channel only showed STT and LLM seconds. Add wall-clock
start/end times and durations for listening, LLM and TTS so it's obvious
what takes long; STT surfaces as the gap between listening end and LLM start.

- bridge: emit llm_start_ms/llm_end_ms on meta and tts_*_ms on the end event
- bot: capture the listening window, assemble full timing after the stream,
  and render a per-stage breakdown in the transcript message

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-13 23:58:49 +09:00

536 lines
22 KiB
Python
Raw Blame History

"""
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, has_speech
except ImportError: # script-relative when run as ``bridge/server.py``
from text_utils import split_sentences
from stt_filter import filter_speech_segments, has_speech
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")
# Pre-STT speech gate (Silero VAD). Tunable for the Discord mic without a code
# change: raise VAD_THRESHOLD to reject more noise, lower it to catch quieter
# speech. VAD_MIN_SPEECH_MS is the shortest run of speech that counts (a brief
# loud blip shorter than this never reaches Whisper). Set VAD_ENABLED=0 to fall
# back to the old behaviour (always transcribe, rely on the post-filter only).
VAD_ENABLED = os.environ.get("VAD_ENABLED", "1") not in ("0", "false", "False")
VAD_THRESHOLD = float(os.environ.get("VAD_THRESHOLD", "0.4"))
VAD_MIN_SPEECH_MS = int(os.environ.get("VAD_MIN_SPEECH_MS", "200"))
# Lock STT to a single language (this deployment is Korean-only). Skipping
# Whisper's language auto-detect both fixes occasional mis-detection (e.g. a
# Korean phrase decoded as Chinese) and shaves a little latency. Empty = auto.
STT_LANGUAGE = os.environ.get("STT_LANGUAGE", "ko").strip() or None
# 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)
# Pre-STT speech gate: don't even invoke Whisper unless there is real speech
# in the clip. Noise or a brief loud blip (no actual speech) is dropped here,
# before transcription, so the model never gets a chance to hallucinate a
# phrase from it. Fail-open inside has_speech() keeps a real utterance from
# being swallowed if the VAD is unavailable.
if VAD_ENABLED and not has_speech(
audio,
16000,
threshold=VAD_THRESHOLD,
min_speech_duration_ms=VAD_MIN_SPEECH_MS,
log=lambda m: print(f"[bridge] {m}", flush=True),
):
print("[bridge] no speech detected (VAD) — skipping STT", flush=True)
return {"text": "", "language": None, "note": "음성 아님(VAD 차단)"}
segments, info = _whisper.transcribe(audio, beam_size=1, language=STT_LANGUAGE)
# Second line of defence: drop non-speech / hallucinated segments by
# Whisper's own no_speech_prob. The no_speech_prob hard cutoff (plus the VAD
# pre-gate above) is what rejects noise/hallucinations. The avg_logprob
# CONFIDENCE floor is deliberately OFF by default (STT_MIN_CONFIDENCE=0):
# short, accented, or quiet real speech over a Discord mic scores very low
# avg_logprob (e.g. the wake word "자비스" at 0.0-0.3) and a confidence floor
# silently eats it, making the bot need many tries to hear one utterance.
# Raise STT_MIN_CONFIDENCE only if hallucinations slip past the no_speech gate.
no_speech_threshold = float(os.environ.get("STT_NO_SPEECH_THRESHOLD", str(getattr(_cfg, "whisper_no_speech_threshold", 0.5))))
min_confidence = float(os.environ.get("STT_MIN_CONFIDENCE", "0.0"))
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()
note = "ok" if text else "인식 실패(빈 결과/필터)"
return {"text": text, "language": getattr(info, "language", None), "note": note}
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():
import time
def now_ms() -> int:
# Wall-clock epoch ms so the Node side can line these up against its
# own Date.now() capture timestamps (same host, same clock).
return int(time.time() * 1000)
t0 = time.monotonic()
stt = transcribe(raw)
t_stt = time.monotonic()
transcript = stt.get("text", "")
if not transcript:
yield json.dumps({"type": "meta", "transcript": "", "language": stt.get("language"),
"reply": "", "error": stt.get("error"),
"note": stt.get("note", "빈 결과"),
"stt_sec": round(t_stt - t0, 1), "broadcast_action": None}) + "\n"
yield json.dumps({"type": "end"}) + "\n"
return
llm_start_ms = now_ms()
result = think(transcript, stt.get("language"), broadcasting)
t_think = time.monotonic()
llm_end_ms = now_ms()
reply = result.get("reply", "")
yield json.dumps({
"type": "meta",
"transcript": transcript,
"language": stt.get("language"),
"reply": reply,
"error": result.get("error"),
"note": "ok" if reply.strip() else "답변 없음",
"stt_sec": round(t_stt - t0, 1),
"think_sec": round(t_think - t_stt, 1),
# Wall-clock LLM window (epoch ms) for the transcript-channel timing
# breakdown. STT shows up as the gap between the Node-side capture
# end and llm_start_ms.
"llm_start_ms": llm_start_ms,
"llm_end_ms": llm_end_ms,
"broadcast_action": result.get("broadcast_action"),
}) + "\n"
tts_total = 0.0
tts_start_ms = None
tts_end_ms = None
for seq, sentence in enumerate(split_sentences(reply)):
ts = time.monotonic()
if tts_start_ms is None:
tts_start_ms = now_ms()
audio = synthesize(sentence)
tts_total += time.monotonic() - ts
tts_end_ms = now_ms()
if audio:
yield json.dumps({
"type": "audio",
"seq": seq,
"audio_b64": base64.b64encode(audio).decode("ascii"),
}) + "\n"
# The end event carries TTS timing because synthesis happens AFTER the
# meta line (it is pipelined sentence-by-sentence).
yield json.dumps({
"type": "end",
"tts_sec": round(tts_total, 1),
"tts_start_ms": tts_start_ms,
"tts_end_ms": tts_end_ms,
}) + "\n"
print(
f"[bridge] ⏱️ turn stt={t_stt - t0:.1f}s think(LLM)={t_think - t_stt:.1f}s "
f"tts={tts_total:.1f}s total={time.monotonic() - t0:.1f}s replylen={len(reply)} "
f"transcript={transcript[:40]!r}",
flush=True,
)
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()
Reference in New Issue View Git Blame Copy Permalink