feat: lots of speech stuff

This commit is contained in:
Brooklyn Nicholson
2026-05-01 19:28:02 -05:00
parent 9f3d393a4d
commit d5d7b5c6dc
41 changed files with 1405 additions and 361 deletions
+110 -4
View File
@@ -11,6 +11,7 @@ Provides speech-to-text transcription with six providers:
- **mistral** — Mistral Voxtral Transcribe API, requires ``MISTRAL_API_KEY``.
- **xai** — xAI Grok STT API, requires ``XAI_API_KEY``. High accuracy,
Inverse Text Normalization, diarization, 21 languages.
- **elevenlabs** — ElevenLabs Scribe API, requires ``ELEVENLABS_API_KEY``.
Used by the messaging gateway to automatically transcribe voice messages
sent by users on Telegram, Discord, WhatsApp, Slack, and Signal.
@@ -84,6 +85,7 @@ DEFAULT_LOCAL_STT_LANGUAGE = "en"
DEFAULT_STT_MODEL = os.getenv("STT_OPENAI_MODEL", "whisper-1")
DEFAULT_GROQ_STT_MODEL = os.getenv("STT_GROQ_MODEL", "whisper-large-v3-turbo")
DEFAULT_MISTRAL_STT_MODEL = os.getenv("STT_MISTRAL_MODEL", "voxtral-mini-latest")
DEFAULT_ELEVENLABS_STT_MODEL = os.getenv("STT_ELEVENLABS_MODEL", "scribe_v2")
LOCAL_STT_COMMAND_ENV = "HERMES_LOCAL_STT_COMMAND"
LOCAL_STT_LANGUAGE_ENV = "HERMES_LOCAL_STT_LANGUAGE"
COMMON_LOCAL_BIN_DIRS = ("/opt/homebrew/bin", "/usr/local/bin")
@@ -91,6 +93,7 @@ COMMON_LOCAL_BIN_DIRS = ("/opt/homebrew/bin", "/usr/local/bin")
GROQ_BASE_URL = os.getenv("GROQ_BASE_URL", "https://api.groq.com/openai/v1")
OPENAI_BASE_URL = os.getenv("STT_OPENAI_BASE_URL", "https://api.openai.com/v1")
XAI_STT_BASE_URL = os.getenv("XAI_STT_BASE_URL", "https://api.x.ai/v1")
ELEVENLABS_STT_BASE_URL = os.getenv("ELEVENLABS_STT_BASE_URL", "https://api.elevenlabs.io/v1")
SUPPORTED_FORMATS = {".mp3", ".mp4", ".mpeg", ".mpga", ".m4a", ".wav", ".webm", ".ogg", ".aac", ".flac"}
LOCAL_NATIVE_AUDIO_FORMATS = {".wav", ".aiff", ".aif"}
@@ -268,9 +271,17 @@ def _get_provider(stt_config: dict) -> str:
)
return "none"
if provider == "elevenlabs":
if get_env_value("ELEVENLABS_API_KEY"):
return "elevenlabs"
logger.warning(
"STT provider 'elevenlabs' configured but ELEVENLABS_API_KEY not set"
)
return "none"
return provider # Unknown — let it fail downstream
# --- Auto-detect (no explicit provider): local > groq > openai > mistral > xai -
# --- Auto-detect (no explicit provider): local > groq > openai > mistral > xai > elevenlabs -
if _HAS_FASTER_WHISPER:
return "local"
@@ -288,6 +299,9 @@ def _get_provider(stt_config: dict) -> str:
if get_env_value("XAI_API_KEY"):
logger.info("No local STT available, using xAI Grok STT API")
return "xai"
if get_env_value("ELEVENLABS_API_KEY"):
logger.info("No local STT available, using ElevenLabs Scribe STT API")
return "elevenlabs"
return "none"
# ---------------------------------------------------------------------------
@@ -781,6 +795,92 @@ def _transcribe_xai(file_path: str, model_name: str) -> Dict[str, Any]:
return {"success": False, "transcript": "", "error": f"xAI STT transcription failed: {e}"}
# ---------------------------------------------------------------------------
# Provider: ElevenLabs (Scribe STT API)
# ---------------------------------------------------------------------------
def _transcribe_elevenlabs(file_path: str, model_name: str) -> Dict[str, Any]:
"""Transcribe using ElevenLabs Scribe STT API."""
api_key = get_env_value("ELEVENLABS_API_KEY")
if not api_key:
return {"success": False, "transcript": "", "error": "ELEVENLABS_API_KEY not set"}
stt_config = _load_stt_config()
elevenlabs_config = stt_config.get("elevenlabs", {})
base_url = str(
elevenlabs_config.get("base_url")
or get_env_value("ELEVENLABS_STT_BASE_URL")
or ELEVENLABS_STT_BASE_URL
).strip().rstrip("/")
language_code = str(elevenlabs_config.get("language_code") or "").strip()
tag_audio_events = is_truthy_value(elevenlabs_config.get("tag_audio_events", False))
diarize = is_truthy_value(elevenlabs_config.get("diarize", False))
try:
import requests
data: Dict[str, str] = {
"model_id": model_name,
"tag_audio_events": "true" if tag_audio_events else "false",
"diarize": "true" if diarize else "false",
}
if language_code:
data["language_code"] = language_code
with open(file_path, "rb") as audio_file:
response = requests.post(
f"{base_url}/speech-to-text",
headers={"xi-api-key": api_key},
files={"file": (Path(file_path).name, audio_file)},
data=data,
timeout=120,
)
if response.status_code != 200:
detail = ""
try:
err_body = response.json()
error_value = err_body.get("detail") or err_body.get("error")
if isinstance(error_value, dict):
detail = str(error_value.get("message") or error_value)
elif error_value:
detail = str(error_value)
else:
detail = response.text[:300]
except Exception:
detail = response.text[:300]
return {
"success": False,
"transcript": "",
"error": f"ElevenLabs STT API error (HTTP {response.status_code}): {detail}",
}
result = response.json()
transcript_text = _extract_transcript_text(result)
if not transcript_text:
return {
"success": False,
"transcript": "",
"error": "ElevenLabs STT returned empty transcript",
}
logger.info(
"Transcribed %s via ElevenLabs Scribe (%s, %d chars)",
Path(file_path).name,
model_name,
len(transcript_text),
)
return {"success": True, "transcript": transcript_text, "provider": "elevenlabs"}
except PermissionError:
return {"success": False, "transcript": "", "error": f"Permission denied: {file_path}"}
except Exception as e:
logger.error("ElevenLabs STT transcription failed: %s", e, exc_info=True)
return {"success": False, "transcript": "", "error": f"ElevenLabs STT transcription failed: {e}"}
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
@@ -792,7 +892,7 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A
Provider priority:
1. User config (``stt.provider`` in config.yaml)
2. Auto-detect: local faster-whisper (free) > Groq (free tier) > OpenAI (paid)
2. Auto-detect: local > Groq > OpenAI > Mistral > xAI > ElevenLabs
Args:
file_path: Absolute path to the audio file to transcribe.
@@ -854,6 +954,11 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A
model_name = model or "grok-stt"
return _transcribe_xai(file_path, model_name)
if provider == "elevenlabs":
elevenlabs_cfg = stt_config.get("elevenlabs", {})
model_name = model or elevenlabs_cfg.get("model_id", DEFAULT_ELEVENLABS_STT_MODEL)
return _transcribe_elevenlabs(file_path, model_name)
# No provider available
return {
"success": False,
@@ -862,8 +967,9 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A
"No STT provider available. Install faster-whisper for free local "
f"transcription, configure {LOCAL_STT_COMMAND_ENV} or install a local whisper CLI, "
"set GROQ_API_KEY for free Groq Whisper, set MISTRAL_API_KEY for Mistral "
"Voxtral Transcribe, set XAI_API_KEY for xAI Grok STT, or set VOICE_TOOLS_OPENAI_KEY "
"or OPENAI_API_KEY for the OpenAI Whisper API."
"Voxtral Transcribe, set XAI_API_KEY for xAI Grok STT, set ELEVENLABS_API_KEY "
"for ElevenLabs Scribe, or set VOICE_TOOLS_OPENAI_KEY or OPENAI_API_KEY for "
"the OpenAI Whisper API."
),
}