ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-34-G' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:12:47 -07:00
parent 992fd64386 7ed48064db
commit 0a6b5370e5
9 changed files with 560 additions and 1043 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/transcription_audio.py
+25 -47
View File
@@ -2,10 +2,9 @@
Binary discovery, the shared ffmpeg m4a encode (transcode + silence trim),
source/format validation, WeChat .silk decoding, CAF conversion and the
best-effort cloud pre-upload silence trim.
Split out of ``tools/transcription_tools.py``, which re-imports every name (patch
surface) and is imported lazily here so origin patches still intercept.
best-effort cloud pre-upload silence trim. Every name is re-imported by
``tools/transcription_tools.py`` (patch surface), which is imported lazily here
so origin patches still intercept.
"""
from __future__ import annotations
@@ -61,17 +60,11 @@ def _run_quiet(command: list, *, timeout: float, env: Optional[dict] = None) ->
# Shared encode profile for every STT-bound m4a (transcode and silence-trim):
# 16 kHz mono 32 kbps AAC, faststart. One owner so codec/bitrate never drift.
_STT_M4A_ENCODE_ARGS = (
"-vn", "-ac", "1", "-ar", "16000",
"-c:a", "aac", "-b:a", "32k", "-movflags", "+faststart",
)
_STT_M4A_ENCODE_ARGS = ("-vn", "-ac", "1", "-ar", "16000", "-c:a", "aac", "-b:a", "32k", "-movflags", "+faststart")
def _run_ffmpeg_stt_encode(ffmpeg: str, input_path: str, output_path: str, *, audio_filter: Optional[str] = None) -> None:
"""Run the shared STT m4a encode, optionally with an ``-af`` filter.
Raises on failure — callers own the error semantics (transcode reports, trim swallows).
"""
"""Run the shared STT m4a encode, optionally with an ``-af`` filter. Raises on failure; callers own the semantics."""
command = [ffmpeg, "-y", "-i", input_path]
if audio_filter:
command += ["-af", audio_filter]
@@ -80,11 +73,10 @@ def _run_ffmpeg_stt_encode(ffmpeg: str, input_path: str, output_path: str, *, au
def _transcode_audio_for_stt(file_path: str, work_dir: str) -> tuple[Optional[str], Optional[str]]:
"""Transcode to a compact 16 kHz mono AAC/m4a for STT upload.
"""Transcode to a compact 16 kHz mono AAC/m4a for STT upload; ``(converted_path, None)`` or ``(None, error)``.
Newer OpenAI models reject containers ``whisper-1`` accepted (notably Ogg/Opus
voice notes) and gateway downloads may carry a misleading extension.
Returns ``(converted_path, None)`` or ``(None, error)``.
"""
from tools.transcription_tools import _find_ffmpeg_binary, _run_ffmpeg_stt_encode
ffmpeg = _find_ffmpeg_binary()
@@ -110,16 +102,13 @@ def _validate_audio_file_size(audio_path: Path, *, enforce_size_limit: bool = Tr
except OSError as e:
return _error_result(f"Failed to access file: {e}")
if enforce_size_limit and file_size > MAX_FILE_SIZE:
return _error_result(
f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)"
)
return _error_result(f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)")
return None
def _validate_audio_source_file(file_path: str, *, enforce_size_limit: bool = True) -> Optional[Dict[str, Any]]:
"""Validate source path safety (and optionally size) before any decoder runs."""
audio_path = Path(file_path)
if os.path.islink(audio_path):
return _error_result(f"Path is a symbolic link: {file_path}")
if not audio_path.exists():
@@ -137,9 +126,7 @@ def _validate_audio_file(file_path: str, *, enforce_size_limit: bool = True) ->
suffix = Path(file_path).suffix
if suffix.lower() not in SUPPORTED_FORMATS:
return _error_result(
f"Unsupported format: {suffix}. Supported: {', '.join(sorted(SUPPORTED_FORMATS))}"
)
return _error_result(f"Unsupported format: {suffix}. Supported: {', '.join(sorted(SUPPORTED_FORMATS))}")
return None
@@ -150,8 +137,7 @@ def _prepare_audio_for_transcription(file_path: str) -> tuple[Optional[str], Opt
if audio_path.suffix.lower() != ".silk":
return file_path, None, None
if not _HAS_PILK:
# pilk is a tiny silk-v3 codec binding — lazy-install on first .silk
# voice note instead of bloating the base install.
# pilk is a tiny silk-v3 codec binding — lazy-installed on first .silk voice note.
_lazy_ensure_quietly("stt.silk")
if not _safe_find_spec("pilk"):
return None, None, _error_result(
@@ -221,23 +207,22 @@ def _convert_caf_to_wav(file_path: str) -> Optional[str]:
# ---- Cloud pre-upload silence trim --------------------------------------
#
# Local faster-whisper gets Silero VAD; cloud providers get the raw file, so
# every second of silence is paid for twice (upload + per-minute billing) and
# cloud Whisper hallucinates on it. Before uploading to a built-in cloud
# provider we collapse long pauses with ffmpeg's silenceremove, keeping
# ``stt.cloud_trim_keep_ms`` of each pause so word boundaries survive.
# Purely best-effort — ANY of these uploads the original untouched:
# ``stt.cloud_trim_silence: false``, ffmpeg/ffprobe missing, trim failure or
# timeout, a ~empty result (the provider, not a dB heuristic, decides "no
# speech"), or <10% saving. Command-type and plugin providers are NOT trimmed:
# they may wrap local CLIs that want the original bytes.
# Local faster-whisper gets Silero VAD; cloud providers get the raw file, so silence
# is paid for twice (upload + per-minute billing) and cloud Whisper hallucinates on
# it. Before uploading to a built-in cloud provider we collapse long pauses with
# ffmpeg's silenceremove, keeping ``stt.cloud_trim_keep_ms`` of each pause so word
# boundaries survive. Purely best-effort — ANY of these uploads the original:
# ``stt.cloud_trim_silence: false``, ffmpeg/ffprobe missing, trim failure/timeout, a
# ~empty result (the provider, not a dB heuristic, decides "no speech"), or <10%
# saving. Command-type and plugin providers are NOT trimmed: they may wrap local
# CLIs that want the original bytes.
_CLOUD_TRIM_THRESHOLD_DB_DEFAULT = -40 # audio below this level counts as silence
_CLOUD_TRIM_KEEP_MS_DEFAULT = 300 # how much of each pause survives the trim
_CLOUD_TRIM_MIN_SAVING = 0.10 # use the trimmed file only when >=10% shorter
_CLOUD_TRIM_MIN_RESULT_SECONDS = 0.3 # all-silence guard floor: never upload ~empty audio
# Below this the trim can't pay for itself (several providers bill a 10s
# minimum per request) and the encode would sit on the synchronous voice-note path.
# Below this the trim can't pay for itself (several providers bill a 10s minimum)
# and the encode would sit on the synchronous voice-note path.
_CLOUD_TRIM_MIN_INPUT_SECONDS = 12.0
@@ -252,10 +237,7 @@ def _probe_audio_duration(file_path: str) -> Optional[float]:
if not ffprobe:
return None
command = [
ffprobe, "-v", "error",
"-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1",
file_path,
ffprobe, "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", file_path,
]
try:
return float(_run_quiet(command, timeout=30).stdout.strip())
@@ -274,11 +256,9 @@ def _cloud_trim_settings(stt_config: Dict[str, Any]) -> tuple[bool, int, int]:
def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> Optional[str]:
"""Return a silence-trimmed copy of *file_path* for cloud upload, or None.
"""Return a silence-trimmed copy of *file_path* for cloud upload, or None (= upload the original).
``None`` always means "upload the original" (disabled, tools missing, clip
too short, trim failed, mostly silence, or not enough saving). On success
the caller owns deleting the returned file's parent directory.
On success the caller owns deleting the returned file's parent directory.
"""
from tools.transcription_tools import _find_ffmpeg_binary, _probe_audio_duration, _run_ffmpeg_stt_encode
enabled, threshold_db, keep_ms = _cloud_trim_settings(stt_config)
@@ -301,8 +281,7 @@ def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> O
return None
keep_seconds = keep_ms / 1000.0
# start_periods=1 strips leading silence; stop_periods=-1 collapses every
# interior/trailing silence, keeping ``keep_seconds`` of each pause.
# start_periods=1 strips leading silence; stop_periods=-1 collapses every interior/trailing silence.
filter_expr = (
f"silenceremove="
f"start_periods=1:start_threshold={threshold_db}dB:start_silence={keep_seconds}:"
@@ -310,8 +289,7 @@ def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> O
)
work_dir = tempfile.mkdtemp(prefix="hermes-stt-trim-")
trimmed_path = os.path.join(work_dir, f"{Path(file_path).stem or 'audio'}-trimmed.m4a")
# Scale the all-silence guard with keep_ms: an output consisting solely
# of kept pause must never be uploaded as "speech".
# Scale the all-silence guard with keep_ms: output that is solely kept pause must never upload as "speech".
min_result_seconds = max(_CLOUD_TRIM_MIN_RESULT_SECONDS, 2 * keep_seconds)
keep_result = False
try:
tools/transcription_cloud.py
+104 -168
View File
@@ -1,11 +1,10 @@
"""Cloud STT providers.
OpenAI-SDK-shaped backends (groq, openai, deepinfra), Mistral Voxtral, the REST
multipart backends (xAI, ElevenLabs), and OpenAI audio credential resolution
(config > keyless local server > env > managed Nous gateway).
Split out of ``tools/transcription_tools.py``, which re-imports every name (patch
surface) and is imported lazily here so origin patches still intercept.
OpenAI-SDK-shaped backends (groq, openai, deepinfra), Mistral Voxtral, REST multipart
backends (xAI, ElevenLabs), and OpenAI audio credential resolution (config > keyless
local server > env > managed Nous gateway). Every name is re-imported by
``tools/transcription_tools.py`` (patch surface), imported lazily here so origin
patches still intercept.
"""
from __future__ import annotations
@@ -14,7 +13,7 @@ import logging
import re
import tempfile
from pathlib import Path
from typing import Any, Dict, Optional
from typing import Any, Callable, Dict, Optional
from urllib.parse import urljoin
from utils import is_truthy_value
@@ -35,25 +34,17 @@ def _has_xai_stt_credentials() -> bool:
return bool(resolve_xai_http_credentials().get("api_key"))
def _close_client(client: Any) -> None:
close = getattr(client, "close", None)
if callable(close):
close()
def _with_openai_client(api_key: str, base_url: Optional[str], file_path: str, log_label: str, body):
"""Run ``body(client)`` against a fresh OpenAI SDK client (30s timeout, no SDK retries).
Always closes the client; any exception maps to the shared envelope via
:func:`_openai_sdk_failure`.
"""
"""Run ``body(client)`` on a fresh OpenAI SDK client (30s timeout, no retries); always closed, errors -> envelope."""
try:
from openai import OpenAI
client = OpenAI(api_key=api_key, base_url=base_url, timeout=30, max_retries=0)
try:
return body(client)
finally:
_close_client(client)
close = getattr(client, "close", None)
if callable(close):
close()
except Exception as e:
return _openai_sdk_failure(e, file_path, log_label)
@@ -69,8 +60,8 @@ def _cloud_failure(exc: BaseException, file_path: str, label: str, detail: Optio
def _openai_sdk_failure(exc: BaseException, file_path: str, log_label: str) -> Dict[str, Any]:
"""Map an OpenAI-SDK-shaped exception to the shared error envelope.
Order matters: APIConnectionError is checked before APITimeoutError (its
subclass) so timeouts report as connection errors, as they always have.
APIConnectionError is checked before APITimeoutError (its subclass) so timeouts
report as connection errors, as they always have.
"""
try:
from openai import APIError, APIConnectionError, APITimeoutError
@@ -78,33 +69,31 @@ def _openai_sdk_failure(exc: BaseException, file_path: str, log_label: str) -> D
APIError = APIConnectionError = APITimeoutError = ()
if isinstance(exc, PermissionError):
return _error_result(f"Permission denied: {file_path}")
if isinstance(exc, APIConnectionError):
return _error_result(f"Connection error: {exc}")
if isinstance(exc, APITimeoutError):
return _error_result(f"Request timeout: {exc}")
if isinstance(exc, APIError):
return _error_result(f"API error: {exc}")
for cls, label in ((APIConnectionError, "Connection error"), (APITimeoutError, "Request timeout"), (APIError, "API error")):
if isinstance(exc, cls):
return _error_result(f"{label}: {exc}")
logger.error("%s transcription failed: %s", log_label, exc, exc_info=True)
return _error_result(f"Transcription failed: {exc}")
def _transcribe_groq(
file_path: str,
model_name: str,
*,
language: Optional[str] = None,
prompt: Optional[str] = None,
) -> Dict[str, Any]:
"""Transcribe using Groq Whisper API (free tier available).
def _sdk_prompt_kwargs(language: Optional[str], prompt: Optional[str]) -> Dict[str, Any]:
"""``language``/``prompt`` create-kwargs, each only when set so the bare request stays byte-identical."""
kwargs: Dict[str, Any] = {}
if language:
kwargs["language"] = language
if prompt:
kwargs["prompt"] = prompt
return kwargs
Language: hook override > ``stt.groq.language`` > ``stt.language`` > env;
otherwise Groq auto-detects.
"""
def _transcribe_groq(
file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Transcribe via the Groq Whisper API; language: hook > ``stt.groq.language`` > ``stt.language`` > env > auto."""
from tools.transcription_tools import _HAS_OPENAI, _resolve_provider_key, _resolve_stt_language
api_key = _resolve_provider_key("GROQ_API_KEY", "groq")
if not api_key:
return _error_result("GROQ_API_KEY not set")
if not _HAS_OPENAI:
return _error_result("openai package not installed")
@@ -116,45 +105,26 @@ def _transcribe_groq(
language = language or _resolve_stt_language("groq")
def _run(client):
create_kwargs = {
"model": model_name,
"response_format": "text",
}
if language:
create_kwargs["language"] = language
if prompt:
# Only sent when set so the no-hook, no-config request stays byte-identical.
create_kwargs["prompt"] = prompt
create_kwargs = {"model": model_name, "response_format": "text", **_sdk_prompt_kwargs(language, prompt)}
with open(file_path, "rb") as audio_file:
transcription = client.audio.transcriptions.create(
file=audio_file,
**create_kwargs,
)
transcription = client.audio.transcriptions.create(file=audio_file, **create_kwargs)
transcript_text = str(transcription).strip()
logger.info("Transcribed %s via Groq API (%s, lang=%s, %d chars)",
Path(file_path).name, model_name, language or "auto", len(transcript_text))
return _ok_result(transcript_text, "groq")
return _with_openai_client(api_key, GROQ_BASE_URL, file_path, "Groq", _run)
def _transcribe_openai(
file_path: str,
model_name: str,
*,
api_key: Optional[str] = None,
base_url: Optional[str] = None,
provider_label: str = "openai",
language: Optional[str] = None,
file_path: str, model_name: str, *, api_key: Optional[str] = None,
base_url: Optional[str] = None, provider_label: str = "openai", language: Optional[str] = None,
prompt: Optional[str] = None,
) -> Dict[str, Any]:
"""Transcribe via the OpenAI ``audio.transcriptions.create`` SDK shape.
Shared backend for every OpenAI-compatible STT endpoint (DeepInfra etc.):
callers pass explicit ``api_key``/``base_url`` to skip the OpenAI-only auth
chain and a ``provider_label`` for the response's ``provider``.
Shared by every OpenAI-compatible endpoint (DeepInfra etc.): explicit ``api_key``/
``base_url`` skip the OpenAI-only auth chain; ``provider_label`` names the response's provider.
"""
from tools.transcription_tools import _HAS_OPENAI, _resolve_openai_audio_client_config, _resolve_stt_language
if api_key is None:
@@ -170,8 +140,7 @@ def _transcribe_openai(
if not _HAS_OPENAI:
return _error_result("openai package not installed")
# Auto-correct a Groq-only model on the native OpenAI path only —
# third-party endpoints may legitimately serve a whisper-large-v3 variant.
# Auto-correct a Groq-only model on the native OpenAI path only (third-party endpoints may serve it).
if provider_label == "openai" and model_name in GROQ_MODELS:
logger.info("Model %s not available on OpenAI, using %s", model_name, DEFAULT_STT_MODEL)
model_name = DEFAULT_STT_MODEL
@@ -188,14 +157,12 @@ def _transcribe_openai(
}
if language:
if model_name == "gpt-transcribe":
# gpt-transcribe replaces ``language`` with a ``languages``
# list and rejects requests sending the legacy field.
# gpt-transcribe takes a ``languages`` list and rejects the legacy field.
create_kwargs["extra_body"] = {"languages": [language]}
else:
create_kwargs["language"] = language
logger.debug("Using language hint '%s' for OpenAI STT", language)
if prompt:
# Only sent when set so the no-hook, no-config request stays byte-identical.
if prompt: # only when set so the bare request stays byte-identical
create_kwargs["prompt"] = prompt
return client.audio.transcriptions.create(**create_kwargs)
@@ -206,8 +173,7 @@ def _transcribe_openai(
message = str(exc).lower()
if not any(k in message for k in ("unsupported", "corrupted", "invalid file")):
raise
# Newer models reject some containers whisper-1 accepted
# (notably Ogg/Opus voice notes): transcode to m4a, retry once.
# Newer models reject containers whisper-1 accepted (Ogg/Opus voice notes): transcode, retry once.
converted_path, transcode_error = _transcode_audio_for_stt(file_path, work_dir)
if transcode_error:
return _error_result(transcode_error)
@@ -222,18 +188,13 @@ def _transcribe_openai(
"Transcribed %s via %s (%s, %d chars)",
Path(file_path).name, provider_label, model_name, len(transcript_text),
)
return _ok_result(transcript_text, provider_label)
return _with_openai_client(api_key, base_url, file_path, provider_label, _run)
def _transcribe_mistral(
file_path: str,
model_name: str,
*,
language: Optional[str] = None,
prompt: Optional[str] = None,
file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Transcribe with the ``mistralai`` SDK (``/v1/audio/transcriptions``); requires ``MISTRAL_API_KEY``."""
from tools.transcription_tools import _resolve_provider_key, _resolve_stt_language
@@ -247,17 +208,13 @@ def _transcribe_mistral(
with Mistral(api_key=api_key) as client:
with open(file_path, "rb") as audio_file:
# Language: hook override > stt.mistral.language > stt.language > env > auto.
language = language or _resolve_stt_language("mistral")
complete_kwargs: Dict[str, Any] = {
"model": model_name,
"file": {"content": audio_file, "file_name": Path(file_path).name},
**_sdk_prompt_kwargs(language, prompt),
}
# Language: hook override > stt.mistral.language > stt.language > env > auto.
language = language or _resolve_stt_language("mistral")
if language:
complete_kwargs["language"] = language
if prompt:
# Only sent when set so the no-hook, no-config request stays byte-identical.
complete_kwargs["prompt"] = prompt
result = client.audio.transcriptions.complete(**complete_kwargs)
transcript_text = _extract_transcript_text(result)
@@ -271,6 +228,9 @@ def _transcribe_mistral(
return _cloud_failure(e, file_path, "Mistral transcription", type(e).__name__)
# ---- REST multipart backends (xAI, ElevenLabs) ----------------------------
def _post_audio_multipart(url: str, headers: Dict[str, str], file_path: str, data: Dict[str, str]):
import requests
@@ -281,22 +241,17 @@ def _post_audio_multipart(url: str, headers: Dict[str, str], file_path: str, dat
)
def _http_error_detail(response, extract) -> str:
"""``extract(json_body)`` -> detail string, falling back to the first 300 chars of the body."""
try:
return extract(response.json()) or response.text[:300]
except Exception:
return response.text[:300]
def _rest_transcript(response, label: str, extract_detail, extract_text):
"""Turn a multipart STT response into ``(text, body, None)`` or ``(None, None, error_envelope)``.
"""Multipart STT response -> ``(text, body, None)`` or ``(None, None, error_envelope)``.
Non-200 -> ``"<label> API error (HTTP n): detail"``; empty text -> the
``no_speech`` envelope so callers treat silence as non-fatal.
Non-200 -> ``"<label> API error (HTTP n): detail"`` (JSON detail via *extract_detail*, else
the first 300 body chars); empty text -> the ``no_speech`` envelope (silence is non-fatal).
"""
if response.status_code != 200:
detail = _http_error_detail(response, extract_detail)
try:
detail = extract_detail(response.json()) or response.text[:300]
except Exception:
detail = response.text[:300]
return None, None, _error_result(f"{label} API error (HTTP {response.status_code}): {detail}")
body = response.json()
text = extract_text(body)
@@ -305,19 +260,23 @@ def _rest_transcript(response, label: str, extract_detail, extract_text):
return text, body, None
def _elevenlabs_error_detail(err_body: Dict[str, Any]) -> str:
error_value = err_body.get("detail") or err_body.get("error")
if isinstance(error_value, dict):
return str(error_value.get("message") or error_value)
return str(error_value) if error_value else ""
def _rest_provider(
file_path: str, provider: str, label: str, post: Callable[[], Any], extract_detail,
extract_text, log: Callable[[str, Dict[str, Any]], None],
) -> Dict[str, Any]:
"""Shared REST flow: ``post()`` -> envelope/transcript -> ``log(text, body)`` -> ok; exceptions -> ``_cloud_failure``."""
try:
transcript_text, body, error = _rest_transcript(post(), label, extract_detail, extract_text)
if error:
return error
log(transcript_text, body)
return _ok_result(transcript_text, provider)
except Exception as e:
return _cloud_failure(e, file_path, f"{label} transcription")
def _transcribe_xai(
file_path: str,
model_name: str,
*,
language: Optional[str] = None,
prompt: Optional[str] = None,
file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Transcribe via xAI ``POST /v1/stt`` (multipart). Supports ITN, diarization, word timestamps."""
from tools.transcription_tools import _load_stt_config, _resolve_stt_language, get_env_value
@@ -326,9 +285,8 @@ def _transcribe_xai(
if prompt:
_log_prompt_unsupported("STT provider 'xai'")
# STT is API-billed: prefer the explicit XAI_API_KEY over the general xAI
# OAuth/Grok-subscription credential, which may be valid for Grok yet hit
# personal-team spending-limit errors on /v1/stt.
# STT is API-billed: prefer the explicit XAI_API_KEY over the xAI OAuth/Grok-subscription
# credential, which may be valid for Grok yet hit spending-limit errors on /v1/stt.
direct_api_key = str(get_env_value("XAI_API_KEY") or "").strip()
if direct_api_key:
creds = {
@@ -346,8 +304,7 @@ def _transcribe_xai(
xai_config = stt_config.get("xai") or {}
def _resolve_base_url(resolved_creds: Dict[str, str]) -> str:
# OAuth bearers are pinned to the resolver-validated xAI origin;
# config/env base URL overrides only apply to API-key credentials.
# OAuth bearers are pinned to the resolver-validated origin; overrides apply to API keys only.
if resolved_creds.get("provider") == "xai-oauth":
url = resolved_creds.get("base_url")
else:
@@ -357,7 +314,7 @@ def _transcribe_xai(
# Language: hook override > stt.xai.language > stt.language > env.
language = language or _resolve_stt_language("xai", stt_config) or ""
try:
def _post() -> Any:
from tools.xai_http import hermes_xai_user_agent
data: Dict[str, str] = {}
@@ -375,7 +332,6 @@ def _transcribe_xai(
)
response = _post_transcription(api_key, _resolve_base_url(creds))
if response.status_code in {401, 403} and creds.get("provider") == "xai-oauth":
logger.info("xAI STT got HTTP %d; refreshing OAuth credentials and retrying once", response.status_code)
try:
@@ -387,30 +343,29 @@ def _transcribe_xai(
logger.warning(
"xAI STT OAuth refresh-and-retry after HTTP %d failed: %s", response.status_code, retry_exc,
)
return response
transcript_text, result, error = _rest_transcript(
response, "xAI STT",
lambda body: body.get("error", {}).get("message", ""),
lambda body: body.get("text", "").strip(),
)
if error:
return error
def _log(transcript_text: str, result: Dict[str, Any]) -> None:
logger.info(
"Transcribed %s via xAI Grok STT (lang=%s, %.1fs audio, %d chars)",
Path(file_path).name, result.get("language", language), result.get("duration", 0), len(transcript_text),
)
return _ok_result(transcript_text, "xai")
except Exception as e:
return _cloud_failure(e, file_path, "xAI STT transcription")
return _rest_provider(
file_path, "xai", "xAI STT", _post, lambda body: body.get("error", {}).get("message", ""),
lambda body: body.get("text", "").strip(), _log,
)
def _elevenlabs_error_detail(err_body: Dict[str, Any]) -> str:
error_value = err_body.get("detail") or err_body.get("error")
if isinstance(error_value, dict):
return str(error_value.get("message") or error_value)
return str(error_value) if error_value else ""
def _transcribe_elevenlabs(
file_path: str,
model_name: str,
*,
language: Optional[str] = None,
prompt: Optional[str] = None,
file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Transcribe using ElevenLabs Scribe STT API."""
from tools.transcription_tools import _load_stt_config, _resolve_provider_key, _resolve_stt_language, get_env_value
@@ -428,7 +383,8 @@ def _transcribe_elevenlabs(
).strip().rstrip("/")
# Language: hook override > stt.elevenlabs.language(_code) > stt.language.
language_code = language or _resolve_stt_language("elevenlabs", stt_config, extra_keys=("language_code",)) or ""
try:
def _post() -> Any:
data: Dict[str, str] = {
"model_id": model_name,
"tag_audio_events": str(is_truthy_value(elevenlabs_config.get("tag_audio_events", False))).lower(),
@@ -436,33 +392,23 @@ def _transcribe_elevenlabs(
}
if language_code:
data["language_code"] = language_code
return _post_audio_multipart(f"{base_url}/speech-to-text", {"xi-api-key": api_key}, file_path, data)
response = _post_audio_multipart(f"{base_url}/speech-to-text", {"xi-api-key": api_key}, file_path, data)
transcript_text, _body, error = _rest_transcript(
response, "ElevenLabs STT", _elevenlabs_error_detail, _extract_transcript_text,
)
if error:
return error
def _log(transcript_text: str, _body: Dict[str, Any]) -> None:
logger.info(
"Transcribed %s via ElevenLabs Scribe (%s, %d chars)",
Path(file_path).name, model_name, len(transcript_text),
)
return _ok_result(transcript_text, "elevenlabs")
except Exception as e:
return _cloud_failure(e, file_path, "ElevenLabs STT transcription")
return _rest_provider(
file_path, "elevenlabs", "ElevenLabs STT", _post, _elevenlabs_error_detail, _extract_transcript_text, _log,
)
def _transcribe_deepinfra(
file_path: str,
model_name: str,
*,
language: Optional[str] = None,
prompt: Optional[str] = None,
file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Resolve DeepInfra credentials/model (via the shared ``hermes_cli.models``
helpers), then delegate to :func:`_transcribe_openai`."""
"""Resolve DeepInfra credentials/model (shared ``hermes_cli.models`` helpers), then delegate to :func:`_transcribe_openai`."""
from tools.transcription_tools import _load_stt_config, _resolve_provider_key, _transcribe_openai
api_key = _resolve_provider_key("DEEPINFRA_API_KEY", "deepinfra")
if not api_key:
@@ -488,13 +434,12 @@ def _transcribe_deepinfra(
)
def _is_local_or_private_url(url: str) -> bool:
"""True for loopback/RFC-1918/LAN-internal hosts.
# ---- OpenAI audio credential resolution -----------------------------------
Decides whether an empty ``stt.openai.api_key`` is acceptable: local
OpenAI-compatible STT servers ignore the auth header, so users shouldn't
need a sham ``api_key: not-needed``.
"""
def _is_local_or_private_url(url: str) -> bool:
"""True for loopback/RFC-1918/LAN-internal hosts, where an empty ``stt.openai.api_key`` is acceptable
(local OpenAI-compatible servers ignore the auth header — no sham ``api_key: not-needed`` needed)."""
try:
from urllib.parse import urlparse
import ipaddress
@@ -511,11 +456,8 @@ def _is_local_or_private_url(url: str) -> bool:
def _direct_openai_credentials(cfg_api_key: str, cfg_base_url: str) -> Optional[tuple[str, str]]:
"""Direct-credential ladder: config key > keyless local base_url > env key; None if none apply.
A local OpenAI-compatible server needs no key — send a placeholder so the
SDK doesn't refuse to construct a client.
"""
"""Direct-credential ladder: config key > keyless local base_url (placeholder key so the SDK
constructs a client) > env key; None if none apply."""
from tools.transcription_tools import resolve_openai_audio_api_key
if cfg_api_key:
return cfg_api_key, (cfg_base_url or OPENAI_BASE_URL)
@@ -528,16 +470,10 @@ def _direct_openai_credentials(cfg_api_key: str, cfg_base_url: str) -> Optional[
def _resolve_openai_audio_client_config() -> tuple[str, str]:
"""Return ``(api_key, base_url)`` for the OpenAI STT client.
Strict selection semantics on the stored ``stt`` provider string:
- ``"nous"`` → managed gateway ONLY; unentitled/unreachable is a
selection-naming error (a direct OPENAI_API_KEY must NOT override it).
- any other stored provider → direct credentials ONLY; missing credentials
is a selection-naming error — no silent managed fallback.
- never-configured stt section → legacy ladder: direct credentials, then
the managed gateway.
"""
"""``(api_key, base_url)`` for the OpenAI STT client, strict on the stored ``stt`` selection:
``"nous"`` -> managed gateway ONLY (a direct OPENAI_API_KEY must NOT override it); any other
stored provider -> direct credentials ONLY (no silent managed fallback); never-configured ->
legacy ladder: direct credentials, then the managed gateway. Failures raise ValueError."""
from tools.transcription_tools import (
_load_stt_config, managed_nous_tools_enabled, nous_tool_gateway_unavailable_message,
resolve_managed_tool_gateway,
tools/transcription_command.py
+59 -155
View File
@@ -2,10 +2,9 @@
``stt.providers.<name>: type: command`` registry, plugin-registered
``TranscriptionProvider`` dispatch, and the ``pre_transcription`` hook that
threads prompt/language/model overrides into every backend.
Split out of ``tools/transcription_tools.py``, which re-imports every name (patch
surface) and is imported lazily here so origin patches still intercept.
threads prompt/language/model overrides into every backend. Every name is
re-imported by ``tools/transcription_tools.py`` (patch surface), which is
imported lazily here so origin patches still intercept.
"""
from __future__ import annotations
@@ -17,12 +16,13 @@ from pathlib import Path
from typing import Any, Dict, Optional
from tools.tts_command_provider import (
command_env_passthrough as _command_stt_env_passthrough,
render_command_template as _render_command_stt_template,
_command_output_format, _command_timeout, _is_command_provider_config as _is_command_stt_provider_config,
_named_provider_config, _resolve_command_config, command_env_passthrough as _command_stt_env_passthrough,
command_failure_detail, render_command_template as _render_command_stt_template,
run_command_provider as _run_command_stt,
)
from tools.transcription_common import (
BUILTIN_STT_PROVIDERS, _error_result, _get_stt_section, _log_prompt_unsupported, _ok_result,
BUILTIN_STT_PROVIDERS, _error_result, _log_prompt_unsupported, _ok_result,
)
# Log-record parity with the origin module.
@@ -31,86 +31,38 @@ logger = logging.getLogger("tools.transcription_tools")
# ---- Command-provider registry (``stt.providers.<name>: type: command``) ---
#
# Mirrors the TTS command-provider registry: same placeholder grammar,
# shell-quote-aware rendering and process-tree termination on timeout.
# Resolution order: built-in name (always wins) > stt.providers.<name> command
# > plugin-registered TranscriptionProvider > "No STT provider available".
# The single-env-var HERMES_LOCAL_STT_COMMAND escape hatch stays untouched via
# the built-in ``local_command`` path.
# Mirrors the TTS command-provider registry (same placeholder grammar, quote-aware
# rendering, process-tree termination on timeout). Resolution order: built-in name
# (always wins) > stt.providers.<name> command > plugin TranscriptionProvider >
# "No STT provider available". The single-env-var HERMES_LOCAL_STT_COMMAND escape
# hatch stays untouched via the built-in ``local_command`` path.
DEFAULT_COMMAND_STT_TIMEOUT_SECONDS = 300
DEFAULT_COMMAND_STT_LANGUAGE = "en"
DEFAULT_COMMAND_STT_OUTPUT_FORMAT = "txt"
COMMAND_STT_OUTPUT_FORMATS = frozenset({"txt", "json", "srt", "vtt"})
_NON_COMMAND_STT_NAMES = frozenset(BUILTIN_STT_PROVIDERS | {"none"})
def _get_named_stt_provider_config(
stt_config: Dict[str, Any],
name: str,
) -> Dict[str, Any]:
"""Return the config for a user-declared STT provider, or {}.
``stt.providers.<name>`` is canonical; ``stt.<name>`` is accepted for
back-compat only when *name* is not a built-in, so a user's ``stt.openai``
block still means the OpenAI provider. Built-in sections can't be mistaken
for command providers anyway: ``_is_command_stt_provider_config`` requires
an explicit ``command:``.
"""
providers = _get_stt_section(stt_config, "providers")
section = providers.get(name)
if isinstance(section, dict):
return section
if name.lower() not in BUILTIN_STT_PROVIDERS:
return _get_stt_section(stt_config, name)
return {}
def _get_named_stt_provider_config(stt_config: Dict[str, Any], name: str) -> Dict[str, Any]:
"""``stt.providers.<name>`` (canonical), else ``stt.<name>`` for non-built-in names only."""
return _named_provider_config(stt_config, name, BUILTIN_STT_PROVIDERS)
def _is_command_stt_provider_config(config: Dict[str, Any]) -> bool:
"""Return True when *config* declares a command-type STT provider."""
if not isinstance(config, dict):
return False
ptype = str(config.get("type") or "").strip().lower()
if ptype and ptype != "command":
return False
command = config.get("command")
return isinstance(command, str) and bool(command.strip())
def _resolve_command_stt_provider_config(
provider: str,
stt_config: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
"""Return the provider config if *provider* is a command type; None for built-ins, ``none``, unknown."""
if not provider:
return None
key = provider.lower().strip()
if key in BUILTIN_STT_PROVIDERS or key == "none":
return None
config = _get_named_stt_provider_config(stt_config, key)
return config if _is_command_stt_provider_config(config) else None
def _resolve_command_stt_provider_config(provider: str, stt_config: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""The provider config if *provider* is a command type; None for built-ins, ``none``, unknown."""
return _resolve_command_config(provider, stt_config, _NON_COMMAND_STT_NAMES)
def _get_command_stt_timeout(config: Dict[str, Any]) -> float:
"""Return timeout in seconds (``timeout`` > ``timeout_seconds``), falling back when invalid or <= 0."""
raw = config.get("timeout", config.get("timeout_seconds", DEFAULT_COMMAND_STT_TIMEOUT_SECONDS))
try:
value = float(raw)
except (TypeError, ValueError):
value = 0.0
return value if value > 0 else float(DEFAULT_COMMAND_STT_TIMEOUT_SECONDS)
return _command_timeout(config, DEFAULT_COMMAND_STT_TIMEOUT_SECONDS)
def _get_command_stt_output_format(config: Dict[str, Any]) -> str:
"""Return the validated output format (txt/json/srt/vtt)."""
raw = config.get("format") or config.get("output_format") or DEFAULT_COMMAND_STT_OUTPUT_FORMAT
fmt = str(raw).lower().strip().lstrip(".")
return fmt if fmt in COMMAND_STT_OUTPUT_FORMATS else DEFAULT_COMMAND_STT_OUTPUT_FORMAT
return _command_output_format(config, COMMAND_STT_OUTPUT_FORMATS, DEFAULT_COMMAND_STT_OUTPUT_FORMAT)
def _read_command_stt_output(output_path: Path, stdout: str, fmt: str) -> str:
"""Return the transcript: non-empty output file > non-empty stdout (curl one-liners) > RuntimeError.
JSON output is returned raw — users configure ``format: txt`` or post-process.
"""
"""Transcript: non-empty output file > non-empty stdout (curl one-liners) > RuntimeError. JSON is returned raw."""
if output_path.exists():
try:
content = output_path.read_text(encoding="utf-8").strip()
@@ -120,26 +72,18 @@ def _read_command_stt_output(output_path: Path, stdout: str, fmt: str) -> str:
return content
if stdout and stdout.strip():
return stdout.strip()
raise RuntimeError(
f"Command STT provider wrote no output file at {output_path} "
f"and produced no stdout"
)
raise RuntimeError(f"Command STT provider wrote no output file at {output_path} and produced no stdout")
def _transcribe_command_stt(
file_path: str,
provider_name: str,
config: Dict[str, Any],
stt_config: Dict[str, Any],
model_override: Optional[str] = None,
language_override: Optional[str] = None,
file_path: str, provider_name: str, config: Dict[str, Any], stt_config: Dict[str, Any],
model_override: Optional[str] = None, language_override: Optional[str] = None,
prompt: Optional[str] = None,
) -> Dict[str, Any]:
"""Transcribe via a user-declared ``stt.providers.<name>: type: command``.
Placeholders (all shell-quote-aware; ``{{``/``}}`` stay literal):
``{input_path}`` original audio path, ``{output_path}`` file to write the
transcript to, ``{output_dir}`` its parent, ``{format}`` txt/json/srt/vtt,
Placeholders (shell-quote-aware; ``{{``/``}}`` stay literal): ``{input_path}``,
``{output_path}`` (transcript file), ``{output_dir}``, ``{format}`` txt/json/srt/vtt,
``{language}`` (default ``en``), ``{model}`` (empty when unset).
"""
from tools.transcription_tools import _resolve_stt_language
@@ -160,10 +104,8 @@ def _transcribe_command_stt(
timeout = _get_command_stt_timeout(config)
output_format = _get_command_stt_output_format(config)
language = (
language_override
or config.get("language")
or _resolve_stt_language(provider_name, stt_config)
or DEFAULT_COMMAND_STT_LANGUAGE
language_override or config.get("language")
or _resolve_stt_language(provider_name, stt_config) or DEFAULT_COMMAND_STT_LANGUAGE
)
model = model_override or config.get("model") or ""
@@ -171,12 +113,9 @@ def _transcribe_command_stt(
with tempfile.TemporaryDirectory(prefix=f"hermes-cmd-stt-{provider_name}-") as tmpdir:
output_path = Path(tmpdir) / f"transcript.{output_format}"
placeholders = {
"input_path": str(audio.resolve()),
"output_path": str(output_path),
"output_dir": str(output_path.parent),
"format": output_format,
"language": str(language),
"model": str(model),
"input_path": str(audio.resolve()), "output_path": str(output_path),
"output_dir": str(output_path.parent), "format": output_format,
"language": str(language), "model": str(model),
}
command = _render_command_stt_template(command_template, placeholders)
logger.info("Transcribing %s via command STT provider '%s'...", audio.name, provider_name)
@@ -185,11 +124,9 @@ def _transcribe_command_stt(
except subprocess.TimeoutExpired:
return fail(f"STT command provider '{provider_name}' timed out after {timeout:g}s")
except subprocess.CalledProcessError as exc:
detail_parts = [
f"{stream}: {text.strip()}" for stream, text in (("stderr", exc.stderr), ("stdout", exc.stdout)) if text
]
detail = "; ".join(detail_parts) or "no command output"
return fail(f"STT command provider '{provider_name}' exited with code {exc.returncode}: {detail}")
return fail(
f"STT command provider '{provider_name}' exited with code {exc.returncode}: {command_failure_detail(exc)}"
)
except RuntimeError as exc:
return fail(str(exc))
except OSError as exc:
@@ -212,32 +149,22 @@ def _unregistered_stt_provider_error(provider: str) -> Dict[str, Any]:
def _dispatch_to_plugin_provider(
file_path: str,
provider: str,
stt_config: Optional[Dict[str, Any]] = None,
*,
model: Optional[str] = None,
language: Optional[str] = None,
prompt: Optional[str] = None,
file_path: str, provider: str, stt_config: Optional[Dict[str, Any]] = None, *,
model: Optional[str] = None, language: Optional[str] = None, prompt: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
"""Route to a plugin-registered transcription provider; None when no plugin claims the name.
Invariants (re-verified here even though the caller short-circuits first,
so a caller refactor can't silently break them): built-in names never reach
the registry; a same-name ``stt.providers.<name>: type: command`` wins over
a plugin. A matched plugin reporting ``is_available() == False`` returns an
error envelope — not None — because the user explicitly opted in via
``stt.provider`` and the generic fall-through message would mislead.
Provider exceptions become the standard error envelope.
Invariants re-verified here so a caller refactor can't break them: built-in names never
reach the registry; a same-name command provider wins over a plugin. A matched plugin with
``is_available() == False`` returns an error envelope — not None — because the user
explicitly opted in via ``stt.provider``. Provider exceptions become the error envelope.
"""
if not provider:
return None
key = provider.lower().strip()
if key in BUILTIN_STT_PROVIDERS or key == "none":
if key in _NON_COMMAND_STT_NAMES:
return None
if stt_config is not None and _is_command_stt_provider_config(
_get_named_stt_provider_config(stt_config, key)
):
if stt_config is not None and _is_command_stt_provider_config(_get_named_stt_provider_config(stt_config, key)):
return None
try:
from agent.transcription_registry import get_provider
@@ -288,24 +215,20 @@ def _dispatch_to_plugin_provider(
return result
# Fields a pre_transcription hook may mutate. ``file_path`` is read-only —
# attempts to change it are logged and dropped.
# Fields a pre_transcription hook may mutate; ``file_path`` is read-only (logged and dropped).
_PRE_TRANSCRIPTION_MUTABLE_FIELDS = ("prompt", "language", "model")
# Whisper-family models only use the final ~224 tokens of the prompt; longer
# values waste upload bytes and can trip stricter OpenAI-compatible servers.
# Enforced client-side (truncate with a warning, never error), ~4 chars/token.
# Whisper-family models only use the final ~224 tokens of the prompt; longer values
# waste upload bytes and can trip stricter OpenAI-compatible servers. Enforced
# client-side (truncate with a warning, never error), ~4 chars/token.
_WHISPER_PROMPT_TOKEN_CAP = 224
_PROMPT_CHARS_PER_TOKEN = 4
_WHISPER_PROMPT_CAPPED_PROVIDERS = frozenset({"local", "openai", "groq", "deepinfra"})
def _enforce_prompt_length_limit(prompt: Optional[str], provider: str) -> Optional[str]:
"""Truncate *prompt* to the whisper-family token cap, keeping the TAIL (fail-open).
Whisper conditions on the final context window, so the most recently
appended hints survive. Other providers own their own validation.
"""
"""Truncate *prompt* to the whisper-family token cap, keeping the TAIL (whisper conditions
on the final context window, so the newest hints survive). Other providers self-validate."""
if not prompt or provider not in _WHISPER_PROMPT_CAPPED_PROVIDERS:
return prompt
max_chars = _WHISPER_PROMPT_TOKEN_CAP * _PROMPT_CHARS_PER_TOKEN
@@ -314,35 +237,21 @@ def _enforce_prompt_length_limit(prompt: Optional[str], provider: str) -> Option
logger.warning(
"Transcription prompt is ~%d tokens; whisper-family provider '%s' "
"only uses the final ~%d — truncating to the last %d characters.",
len(prompt) // _PROMPT_CHARS_PER_TOKEN,
provider,
_WHISPER_PROMPT_TOKEN_CAP,
max_chars,
len(prompt) // _PROMPT_CHARS_PER_TOKEN, provider, _WHISPER_PROMPT_TOKEN_CAP, max_chars,
)
return prompt[-max_chars:]
def _apply_pre_transcription_hook(
*,
file_path: str,
provider: str,
model: Optional[str],
language: Optional[str],
prompt: Optional[str],
source: Optional[str],
*, file_path: str, provider: str, model: Optional[str], language: Optional[str],
prompt: Optional[str], source: Optional[str],
) -> tuple[Optional[str], Optional[str], Optional[str]]:
"""Fire the ``pre_transcription`` plugin hook and merge its results.
"""Fire the ``pre_transcription`` plugin hook; returns ``(model, language_override, prompt)``.
Gated on ``has_hook`` so the no-hook path never builds hook kwargs, and
fail-open: any hook-plumbing error leaves the dispatch untouched. Results
arrive in registration order (plugins discovered in sorted order) and are
applied field-by-field, so the last hook to write a field wins. Model
values are accepted as-is and flow through the same per-backend
normalization a caller-supplied model would.
Returns ``(model, language_override, prompt)``; ``language_override`` is
None unless a hook explicitly set ``language``, so backends keep their own
config/env language resolution.
Gated on ``has_hook`` (the no-hook path never builds kwargs) and fail-open: any
plumbing error leaves the dispatch untouched. Results apply field-by-field in
registration order (last hook wins). ``language_override`` is None unless a hook
explicitly set ``language``, so backends keep their own config/env resolution.
"""
try:
from hermes_cli.plugins import has_hook, invoke_hook
@@ -351,13 +260,8 @@ def _apply_pre_transcription_hook(
return model, None, prompt
hook_results = invoke_hook(
"pre_transcription",
file_path=file_path,
provider=provider,
model=model,
language=language,
prompt=prompt,
source=source,
"pre_transcription", file_path=file_path, provider=provider,
model=model, language=language, prompt=prompt, source=source,
)
overrides: Dict[str, Any] = {}
for hook_result in hook_results:
tools/transcription_common.py
+7 -14
View File
@@ -1,7 +1,7 @@
"""Constants, result envelopes and tiny config readers shared by every STT module.
Split out of ``tools/transcription_tools.py``; every name is re-imported there, so
``tools.transcription_tools.<name>`` keeps resolving (and monkeypatching) as before.
Every name is re-imported by ``tools/transcription_tools.py``, so
``tools.transcription_tools.<name>`` keeps resolving (and monkeypatching).
"""
from __future__ import annotations
@@ -11,6 +11,8 @@ import os
import subprocess # noqa: F401 (type annotation only)
from typing import Any, Dict
from tools.tts_command_provider import _get_provider_section as _get_stt_section # noqa: F401 (re-exported)
# Log-record parity with the origin module.
logger = logging.getLogger("tools.transcription_tools")
@@ -39,10 +41,9 @@ MAX_FILE_SIZE = 25 * 1024 * 1024 # 25 MB
OPENAI_MODELS = {"whisper-1", "gpt-4o-mini-transcribe", "gpt-4o-transcribe", "gpt-transcribe"}
GROQ_MODELS = {"whisper-large-v3", "whisper-large-v3-turbo", "distil-whisper-large-v3-en"}
# Providers with native handlers. Kept in sync with
# ``agent.transcription_registry._BUILTIN_NAMES`` (a regression test fails on
# drift); plugins may not register under these names and the dispatcher
# short-circuits them before command/plugin lookup.
# Providers with native handlers. Kept in sync with ``agent.transcription_registry._BUILTIN_NAMES``
# (a regression test fails on drift); plugins may not register under these names and the
# dispatcher short-circuits them before command/plugin lookup.
BUILTIN_STT_PROVIDERS = frozenset({
"local", "local_command", "groq", "openai", "mistral", "xai", "elevenlabs", "deepinfra",
})
@@ -59,14 +60,6 @@ def _ok_result(transcript: str, provider: str) -> Dict[str, Any]:
return {"success": True, "transcript": transcript, "provider": provider}
def _get_stt_section(stt_config: Dict[str, Any], name: str) -> Dict[str, Any]:
"""Return an stt sub-section if it's a dict, else an empty dict."""
if not isinstance(stt_config, dict):
return {}
section = stt_config.get(name)
return section if isinstance(section, dict) else {}
def _lazy_ensure_quietly(dep: str) -> None:
"""Best-effort ``tools.lazy_deps.ensure(dep, prompt=False)``; failures are swallowed.
tools/transcription_local.py
+29 -50
View File
@@ -2,11 +2,9 @@
faster-whisper loading (CUDA->CPU fallback, Apple Silicon pinning), the
anti-hallucination transcribe kwargs and segment gate, and the local whisper CLI
(``local_command``) provider. The cached-model singleton and its idle-unload
watcher stay in ``transcription_tools`` (they own the module state).
Split out of ``tools/transcription_tools.py``, which re-imports every name (patch
surface) and is imported lazily here so origin patches still intercept.
(``local_command``) provider. The cached-model singleton and idle-unload watcher
stay in ``transcription_tools`` (module state), which re-imports every name here
(patch surface) and is imported lazily so origin patches still intercept.
"""
from __future__ import annotations
@@ -69,9 +67,8 @@ def _try_lazy_install_stt() -> bool:
"""Lazy-install faster-whisper and re-check dynamically so it's usable without a restart."""
try:
from tools.lazy_deps import ensure
# prompt=False: a bare input() deadlocks under the interactive CLI where
# prompt_toolkit owns stdin; the install is already gated by
# security.allow_lazy_installs, so reaching here is opt-in.
# prompt=False: a bare input() deadlocks under the interactive CLI where prompt_toolkit
# owns stdin; the install is already gated by security.allow_lazy_installs.
ensure("stt.faster_whisper", prompt=False)
if _ilu.find_spec("faster_whisper"):
return True
@@ -89,10 +86,9 @@ def _try_lazy_install_stt() -> bool:
return False
# Substrings identifying a missing/unloadable CUDA runtime library: when
# ctranslate2 can't dlopen one of these the "auto" device picker has already
# committed to CUDA, so we fall back to CPU and reload. Deliberately narrow
# (library names + dlopen phrasing) so legitimate runtime failures like "CUDA
# Substrings identifying a missing/unloadable CUDA runtime library: the "auto" device
# picker has already committed to CUDA, so we fall back to CPU and reload. Deliberately
# narrow (library names + dlopen phrasing) so legitimate runtime failures like "CUDA
# out of memory" surface to the user instead of silently running on CPU.
_CUDA_LIB_ERROR_MARKERS = (
"libcublas", "libcudnn", "libcudart", "cannot be loaded", "cannot open shared object",
@@ -111,10 +107,7 @@ def _sysctl_value(name: str) -> str:
"""Return a sysctl value, or an empty string when unavailable."""
try:
return subprocess.check_output(
["/usr/sbin/sysctl", "-n", name],
stderr=subprocess.DEVNULL,
text=True,
timeout=2,
["/usr/sbin/sysctl", "-n", name], stderr=subprocess.DEVNULL, text=True, timeout=2,
).strip()
except Exception:
return ""
@@ -140,10 +133,10 @@ def _get_idle_unload_seconds(local_cfg: Dict[str, Any]) -> int:
def _load_local_whisper_model(model_name: str, device: str = "auto", compute_type: str = "auto"):
"""Load faster-whisper with graceful CUDA → CPU fallback.
``device="auto"`` picks CUDA whenever the ctranslate2 wheel ships CUDA libs,
even on hosts without the NVIDIA runtime (WSL2, headless servers, CPU-only
dev boxes). Try the requested config first; on a CUDA library load failure
fall back to CPU + int8. Pass ``stt.local.device`` / ``compute_type`` to pin.
``device="auto"`` picks CUDA whenever the ctranslate2 wheel ships CUDA libs, even
on hosts without the NVIDIA runtime (WSL2, headless servers). Try the requested
config first; on a CUDA library load failure fall back to CPU + int8. Pass
``stt.local.device`` / ``compute_type`` to pin.
"""
force_cpu = _should_force_faster_whisper_cpu()
if force_cpu:
@@ -172,23 +165,18 @@ def _load_local_whisper_model(model_name: str, device: str = "auto", compute_typ
return WhisperModel(model_name, device="cpu", compute_type="int8")
# Silence-hallucination hardening for local faster-whisper (whisper decodes
# junk like "You"/"Thank you." from pure silence). Three layers, all tunable
# under ``stt.local``: Silero VAD so silence never reaches the model
# (``vad: false`` restores raw behaviour for music/ambient audio);
# condition_on_previous_text=False so one hallucinated token can't seed a run;
# and the segment confidence gate in _is_hallucinated_segment.
# Silence-hallucination hardening for local faster-whisper (whisper decodes junk like
# "You"/"Thank you." from pure silence). Three layers, all tunable under ``stt.local``:
# Silero VAD so silence never reaches the model (``vad: false`` restores raw behaviour
# for music/ambient audio); condition_on_previous_text=False so one hallucinated token
# can't seed a run; and the segment confidence gate in _is_hallucinated_segment.
_VAD_MIN_SILENCE_MS_DEFAULT = 500
_NO_SPEECH_PROB_THRESHOLD_DEFAULT = 0.6
_LOGPROB_THRESHOLD_DEFAULT = -1.0
def build_local_transcribe_kwargs(stt_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Build the kwargs for EVERY local faster-whisper ``model.transcribe`` call.
Single owner for the anti-hallucination hardening — new local-whisper call
sites must go through here instead of hand-rolling kwargs.
"""
"""Kwargs for EVERY local faster-whisper ``model.transcribe`` call — single owner of the anti-hallucination hardening."""
from tools.transcription_tools import _load_stt_config, _resolve_stt_language
stt_config = stt_config if isinstance(stt_config, dict) else _load_stt_config()
local_cfg = stt_config.get("local") or {}
@@ -205,11 +193,10 @@ def build_local_transcribe_kwargs(stt_config: Optional[Dict[str, Any]] = None) -
"min_silence_duration_ms": _config_number(local_cfg, "vad_min_silence_ms", _VAD_MIN_SILENCE_MS_DEFAULT, int)
}
# Push the confidence gate into faster-whisper itself: its internal
# defaults drop low-confidence segments BEFORE our post-filter sees them,
# so without this the ``stt.local`` threshold knobs were dead for that
# first gate (non-English speech decodes at lower avg_logprob and was
# silently discarded). Same values feed both gates; defaults unchanged.
# Push the confidence gate into faster-whisper itself: its internal defaults drop
# low-confidence segments BEFORE our post-filter sees them, so the ``stt.local``
# threshold knobs were dead for that first gate (non-English speech decodes at
# lower avg_logprob and was silently discarded). Same values feed both gates.
kwargs["no_speech_threshold"], kwargs["log_prob_threshold"] = _confidence_thresholds(local_cfg)
forced_lang = _resolve_stt_language("local", stt_config)
@@ -234,9 +221,8 @@ def _confidence_thresholds(local_cfg: Dict[str, Any]) -> tuple[float, float]:
def _is_hallucinated_segment(segment: Any, no_speech_threshold: float, logprob_threshold: float) -> bool:
"""True when a segment is very likely a silence hallucination.
Conservative AND gate (openai-whisper's own heuristic): the model must BOTH
think the window is non-speech AND have decoded it with low confidence, so
quiet-but-real speech survives. Unknown segment shapes are never dropped.
Conservative AND gate (openai-whisper's own heuristic): non-speech AND low decode
confidence, so quiet-but-real speech survives. Unknown segment shapes are never dropped.
"""
try:
no_speech_prob = float(getattr(segment, "no_speech_prob"))
@@ -254,8 +240,7 @@ def _join_confident_segments(segments: Any, local_cfg: Dict[str, Any]) -> str:
if _is_hallucinated_segment(segment, no_speech_threshold, logprob_threshold):
logger.debug(
"Dropping probable hallucinated segment %r (no_speech_prob=%.3f, avg_logprob=%.3f)",
getattr(segment, "text", ""),
getattr(segment, "no_speech_prob", float("nan")),
getattr(segment, "text", ""), getattr(segment, "no_speech_prob", float("nan")),
getattr(segment, "avg_logprob", float("nan")),
)
continue
@@ -264,11 +249,7 @@ def _join_confident_segments(segments: Any, local_cfg: Dict[str, Any]) -> str:
def _transcribe_local_command(
file_path: str,
model_name: str,
*,
language: Optional[str] = None,
prompt: Optional[str] = None,
file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Run the configured local STT command template and read back a .txt transcript."""
from tools.transcription_tools import _prepare_local_audio, _resolve_stt_language
@@ -290,10 +271,8 @@ def _transcribe_local_command(
return _error_result(prep_error)
command = command_template.format(
input_path=shlex.quote(prepared_input),
output_dir=shlex.quote(output_dir),
language=shlex.quote(language),
model=shlex.quote(normalized_model),
input_path=shlex.quote(prepared_input), output_dir=shlex.quote(output_dir),
language=shlex.quote(language), model=shlex.quote(normalized_model),
)
# Scrub Hermes secrets from the child env (same policy as _run_command_stt).
from tools.environments.local import hermes_subprocess_env
tools/transcription_tools.py
+103 -182
View File
@@ -1,16 +1,13 @@
#!/usr/bin/env python3
"""Speech-to-text transcription used by the gateway for voice messages.
Built-in providers: local (faster-whisper, default/free), local_command, groq,
openai (also serves the managed ``nous`` selection), mistral, xai, elevenlabs,
deepinfra; plus user-declared command providers and plugin providers.
result = transcribe_audio("/path/to/audio.ogg") # {"success", "transcript", "error"?, "provider"?}
This module owns provider resolution, the dispatcher, and the cached local
model + idle-unload state. Backends live in sibling modules
(``transcription_{common,audio,local,cloud,command}``) and are re-imported
here so ``tools.transcription_tools.<name>`` stays the patch/import surface.
Built-in providers: local (faster-whisper, default/free), local_command, groq, openai
(also serves the managed ``nous`` selection), mistral, xai, elevenlabs, deepinfra; plus
user-declared command providers and plugin providers. ``transcribe_audio(path)`` returns
``{"success", "transcript", "error"?, "provider"?}``. This module owns provider resolution,
the dispatcher and the cached local model + idle-unload state; backends live in
``transcription_{common,audio,local,cloud,command}`` and are re-imported here so
``tools.transcription_tools.<name>`` stays the patch surface.
"""
import logging
@@ -69,11 +66,7 @@ logger = logging.getLogger(__name__)
def get_env_value(name, default=None):
"""Read env values through the live config module.
Resolved at call time: tests monkeypatch/restore ``hermes_cli.config.get_env_value``
around this module's import, so a cached import would go stale.
"""
"""Read env values through the live config module (resolved per call: tests monkeypatch it around import)."""
try:
from hermes_cli.config import get_env_value as _get_env_value
except ImportError:
@@ -83,9 +76,9 @@ def get_env_value(name, default=None):
def _resolve_provider_key(env_var: str, provider_id: str) -> str:
"""Resolve an STT API key via the shared voice-key resolver (config > env/.env > credential pool).
"""STT API key via the shared voice-key resolver (config > env/.env > credential pool).
Resolved at call time so tests that reload the helpers module see the live function.
Resolved per call so tests that reload the helpers module see the live function.
"""
try:
from tools.tool_backend_helpers import resolve_provider_secret
@@ -107,18 +100,14 @@ _HAS_MISTRAL = _safe_find_spec("mistralai")
_HAS_PILK = _safe_find_spec("pilk")
# Singleton for the local model — loaded once, reused across calls. The lock
# guards the check-then-load so two concurrent voice messages can't both
# download/load the model.
# Local model singleton; the lock guards check-then-load against concurrent voice messages.
_local_model: Optional[object] = None
_local_model_name: Optional[str] = None
_local_model_lock = threading.Lock()
# Idle unload: a single daemon thread checks _last_transcription_time and
# releases the model (hundreds of MB of RAM/VRAM) after a configurable idle
# period, then exits; the next voice message reloads and restarts it.
# _idle_unload_mgmt_lock serializes the start check so two concurrent
# transcriptions can't both observe "no watcher alive" and spawn duplicates.
# Idle unload: one daemon thread releases the model (hundreds of MB of RAM/VRAM) after a
# configurable idle period, then exits; the next voice message reloads and restarts it.
# _idle_unload_mgmt_lock serializes the start check so no duplicate watchers spawn.
_last_transcription_time: float = 0.0
_idle_unload_thread: Optional[threading.Thread] = None
_idle_unload_stop = threading.Event()
@@ -147,16 +136,12 @@ def is_stt_enabled(stt_config: Optional[dict] = None) -> bool:
def _resolve_stt_language(
provider_key: str,
stt_config: Optional[Dict[str, Any]] = None,
*,
extra_keys: tuple = (),
provider_key: str, stt_config: Optional[Dict[str, Any]] = None, *, extra_keys: tuple = ()
) -> Optional[str]:
"""Resolve the language hint for an STT provider; first non-empty wins.
"""Language hint for an STT provider, first non-empty wins (never "").
Order: ``stt.<provider>.language`` (plus *extra_keys* aliases, e.g. ElevenLabs'
``language_code``) > ``stt.language`` > ``HERMES_LOCAL_STT_LANGUAGE`` env >
None (provider auto-detects). Never returns "".
``stt.<provider>.language`` (plus *extra_keys* aliases, e.g. ``language_code``) >
``stt.language`` > ``HERMES_LOCAL_STT_LANGUAGE`` env > None (provider auto-detects).
"""
if stt_config is None:
stt_config = _load_stt_config()
@@ -192,6 +177,15 @@ def _is_local_stt_provider(provider: str, stt_config: Dict[str, Any]) -> bool:
# ---- Provider resolution ------------------------------------------------
def _has_key(env_var: str, provider: str, *, needs_openai: bool = False, needs_mistral: bool = False):
"""Availability probe factory: optional SDK flag AND a resolvable API key."""
def probe() -> bool:
if (needs_openai and not _HAS_OPENAI) or (needs_mistral and not _HAS_MISTRAL):
return False
return bool(_resolve_provider_key(env_var, provider))
return probe
def _has_xai_stt_credentials_quietly() -> bool:
try:
return _has_xai_stt_credentials()
@@ -199,30 +193,11 @@ def _has_xai_stt_credentials_quietly() -> bool:
return False
def _has_key(env_var: str, provider: str, *, needs_openai: bool = False, needs_mistral: bool = False):
"""Availability probe factory: optional SDK flag AND a resolvable API key."""
def probe() -> bool:
if needs_openai and not _HAS_OPENAI:
return False
if needs_mistral and not _HAS_MISTRAL:
return False
return bool(_resolve_provider_key(env_var, provider))
return probe
_has_groq_key = _has_key("GROQ_API_KEY", "groq", needs_openai=True)
_has_mistral_key = _has_key("MISTRAL_API_KEY", "mistral", needs_mistral=True)
_has_elevenlabs_key = _has_key("ELEVENLABS_API_KEY", "elevenlabs")
_has_deepinfra_key = _has_key("DEEPINFRA_API_KEY", "deepinfra", needs_openai=True)
def _resolve_explicit_openai() -> str:
if not _HAS_OPENAI:
logger.warning("STT provider 'openai' configured but no API key available")
return "none"
# Resolve directly rather than via the boolean probe so a managed
# openai-audio gateway outage is logged with its real reason, not a
# generic "no API key" hint.
# Resolved directly so a managed openai-audio gateway outage is logged with its real reason.
reason = _openai_audio_unavailable_reason()
if reason is None:
return "openai"
@@ -262,14 +237,18 @@ def _resolve_explicit_local_command() -> str:
return "none"
_has_groq_key = _has_key("GROQ_API_KEY", "groq", needs_openai=True)
_has_mistral_key = _has_key("MISTRAL_API_KEY", "mistral", needs_mistral=True)
_has_elevenlabs_key = _has_key("ELEVENLABS_API_KEY", "elevenlabs")
_has_deepinfra_key = _has_key("DEEPINFRA_API_KEY", "deepinfra", needs_openai=True)
# Cloud providers in AUTO-DETECT priority order:
# name -> (explicit-selection probe, auto-detect probe, explicit warning, auto-detect log)
# The two probes differ only for openai (auto-detect additionally requires the
# SDK) and xai (auto-detect must never raise). DeepInfra is LAST so a
# DEEPINFRA_API_KEY set for the chat surface never displaces an existing
# xAI/ElevenLabs auto-selection. Mistral only auto-selects when the SDK is
# already present — no lazy-install during passive auto-detection (explicit
# ``provider: mistral`` installs on first use).
# The probes differ only for openai (explicit has its own resolver in _EXPLICIT_RESOLVERS;
# auto-detect also requires the SDK) and xai (auto-detect must never raise). DeepInfra is
# LAST so a DEEPINFRA_API_KEY set for chat never displaces an xAI/ElevenLabs auto-selection.
# Mistral only auto-selects when the SDK is present — no lazy-install during passive
# auto-detection (explicit ``provider: mistral`` installs on first use).
_CLOUD_PROVIDER_SPECS = {
"groq": (
_has_groq_key, _has_groq_key,
@@ -278,13 +257,12 @@ _CLOUD_PROVIDER_SPECS = {
),
"openai": (
None, lambda: _HAS_OPENAI and _has_openai_audio_backend(),
None, # explicit openai has its own resolver (logs the real gateway reason)
None,
"No local STT available, using OpenAI Whisper API",
),
"mistral": (
_has_mistral_key, _has_mistral_key,
"STT provider 'mistral' configured but mistralai package "
"not installed or MISTRAL_API_KEY not set",
"STT provider 'mistral' configured but mistralai package not installed or MISTRAL_API_KEY not set",
"No local STT available, using Mistral Voxtral Transcribe API",
),
"xai": (
@@ -299,25 +277,25 @@ _CLOUD_PROVIDER_SPECS = {
),
"deepinfra": (
_has_deepinfra_key, _has_deepinfra_key,
"STT provider 'deepinfra' configured but DEEPINFRA_API_KEY not set "
"(or openai package missing)",
"STT provider 'deepinfra' configured but DEEPINFRA_API_KEY not set (or openai package missing)",
"No local STT available, using DeepInfra Whisper API",
),
}
# Explicit selections whose resolution is more than a probe + warning.
_EXPLICIT_RESOLVERS = {
"local": _resolve_explicit_local,
"local_command": _resolve_explicit_local_command,
"openai": _resolve_explicit_openai,
}
def _resolve_explicit_provider(provider: str) -> str:
"""Resolve an explicit ``stt.provider`` to a usable provider name or ``"none"``.
Unknown names pass through untouched so the dispatcher can fail with the
provider-not-registered message.
"""
if provider == "local":
return _resolve_explicit_local()
if provider == "local_command":
return _resolve_explicit_local_command()
if provider == "openai":
return _resolve_explicit_openai()
"""Explicit ``stt.provider`` -> usable name or ``"none"``; unknown names pass through untouched
so the dispatcher fails with the provider-not-registered message."""
resolver = _EXPLICIT_RESOLVERS.get(provider)
if resolver is not None:
return resolver()
spec = _CLOUD_PROVIDER_SPECS.get(provider)
if spec is None:
return provider
@@ -329,29 +307,21 @@ def _resolve_explicit_provider(provider: str) -> str:
def _get_provider(stt_config: dict) -> str:
"""Determine which STT provider to use.
An explicit ``stt.provider`` is honoured — no silent cloud fallback. With no
provider configured, auto-detect tries local > groq > openai > mistral > xai
> elevenlabs > deepinfra.
"""
"""Which STT provider to use: an explicit ``stt.provider`` is honoured (no silent cloud
fallback); otherwise auto-detect local > groq > openai > mistral > xai > elevenlabs > deepinfra."""
if not is_stt_enabled(stt_config):
return "none"
explicit = "provider" in stt_config
provider = stt_config.get("provider", DEFAULT_PROVIDER)
# The managed "Nous Subscription" selection is serviced by the OpenAI
# implementation, routed through the managed gateway by
# _resolve_openai_audio_client_config.
# The managed "Nous Subscription" selection is the OpenAI backend routed via the managed gateway.
if isinstance(provider, str) and provider.strip().lower() == "nous":
provider = "openai"
if explicit and provider == "local":
# Legacy DEFAULT_CONFIG seeded ``stt.provider: local`` on every install,
# so a merged-config "local" is not proof of a user pick. Only a raw
# config.yaml selection counts as explicit; otherwise autodetect (which
# prefers local first anyway).
# Legacy DEFAULT_CONFIG seeded ``stt.provider: local`` on every install, so only a
# raw config.yaml selection counts as explicit; otherwise autodetect (local-first anyway).
try:
from tools.tool_backend_helpers import read_selection
@@ -381,10 +351,7 @@ def _unload_local_model() -> None:
global _local_model, _local_model_name
with _local_model_lock:
if _local_model is not None:
logger.info(
"Unloading local whisper model '%s' after idle timeout",
_local_model_name or "unknown",
)
logger.info("Unloading local whisper model '%s' after idle timeout", _local_model_name or "unknown")
_local_model = None
_local_model_name = None
@@ -392,12 +359,10 @@ def _unload_local_model() -> None:
def _start_idle_unload_watcher(timeout_seconds: int) -> None:
"""Ensure the single idle-unload watcher thread is running.
Started only when none is alive (one lock + one ``is_alive()`` per
transcription). The loop re-reads ``stt.local.unload_after_idle_seconds``
every cycle so config edits apply within one interval; ``timeout_seconds``
seeds the first cycle so a just-written config is honored even if a
concurrent read races. After unloading, when the timeout becomes 0, or when
the model is already gone, the thread exits; the next transcription restarts it.
The loop re-reads ``stt.local.unload_after_idle_seconds`` every cycle so config edits
apply within one interval; ``timeout_seconds`` seeds the first cycle so a just-written
config is honored even if a concurrent read races. The thread exits after unloading,
when the timeout becomes 0, or when the model is already gone.
"""
global _idle_unload_thread
with _idle_unload_mgmt_lock:
@@ -412,9 +377,7 @@ def _start_idle_unload_watcher(timeout_seconds: int) -> None:
if _local_model is None:
break
try:
timeout = _get_idle_unload_seconds(
_load_stt_config().get("local") or {}
)
timeout = _get_idle_unload_seconds(_load_stt_config().get("local") or {})
except Exception: # noqa: BLE001 - keep the seed value
timeout = initial_timeout
if timeout <= 0:
@@ -424,9 +387,7 @@ def _start_idle_unload_watcher(timeout_seconds: int) -> None:
break
_idle_unload_stop.clear()
_idle_unload_thread = threading.Thread(
target=_watch, name="hermes-stt-idle-unload", daemon=True
)
_idle_unload_thread = threading.Thread(target=_watch, name="hermes-stt-idle-unload", daemon=True)
_idle_unload_thread.start()
@@ -437,11 +398,9 @@ def _touch_transcription_time() -> None:
def _get_or_load_local_model(model_name: str, local_cfg: Dict[str, Any]):
"""Return the cached faster-whisper model, (re)loading under the lock when needed.
"""Cached faster-whisper model, (re)loaded under a double-checked lock when needed.
Double-checked lock: concurrent voice messages must not both download/load.
The returned strong reference stays valid even if the idle watcher nulls the
module global mid-transcription.
The returned strong reference stays valid even if the idle watcher nulls the global mid-transcription.
"""
global _local_model, _local_model_name
model = _local_model
@@ -449,11 +408,9 @@ def _get_or_load_local_model(model_name: str, local_cfg: Dict[str, Any]):
with _local_model_lock:
if _local_model is None or _local_model_name != model_name:
logger.info("Loading faster-whisper model '%s' (first load downloads the model)...", model_name)
# stt.local.device / compute_type let users pin a configuration
# where ``auto`` mis-detects; the loader keeps the CUDA→CPU fallback.
# stt.local.device / compute_type pin a configuration where ``auto`` mis-detects.
_local_model = _load_local_whisper_model(
model_name,
device=local_cfg.get("device", "auto"),
model_name, device=local_cfg.get("device", "auto"),
compute_type=local_cfg.get("compute_type", "auto"),
)
_local_model_name = model_name
@@ -473,11 +430,7 @@ def _replace_cached_model_on_cpu(model_name: str):
def _transcribe_local(
file_path: str,
model_name: str,
*,
language: Optional[str] = None,
prompt: Optional[str] = None,
file_path: str, model_name: str, *, language: Optional[str] = None, prompt: Optional[str] = None
) -> Dict[str, Any]:
"""Transcribe using faster-whisper (local, free)."""
if not _HAS_FASTER_WHISPER and not _try_lazy_install_stt():
@@ -486,8 +439,7 @@ def _transcribe_local(
try:
stt_config = _load_stt_config()
local_cfg = stt_config.get("local") or {}
# Reset the idle timer BEFORE loading/transcribing so the watcher can't
# count a long in-flight transcription as idle time and unload mid-use.
# Reset the idle timer BEFORE loading so a long in-flight transcription isn't counted as idle.
_touch_transcription_time()
model = _get_or_load_local_model(model_name, local_cfg)
if model is None: # defensive: load failed without raising
@@ -503,9 +455,8 @@ def _transcribe_local(
try:
segments, info = model.transcribe(file_path, **transcribe_kwargs)
except Exception as exc:
# CUDA libs sometimes only fail at dlopen-on-first-use, AFTER the
# model loaded. Evict the poisoned cached model, reload on CPU and
# retry once — otherwise every later voice message fails until restart.
# CUDA libs can fail at dlopen-on-first-use, AFTER loading: evict the poisoned
# cached model, reload on CPU and retry once, else every later message fails.
if not _looks_like_cuda_lib_error(exc):
raise
logger.warning(
@@ -538,8 +489,7 @@ def _transcribe_local(
def _read_block_error(file_path: str) -> Optional[Dict[str, Any]]:
"""Refuse to feed a credential / secret store (auth.json, .env, OAuth tokens, ...)
to an STT provider, which would ship its plaintext to a third-party API.
"""Refuse to ship a credential store (auth.json, .env, OAuth tokens) to an STT provider.
Mirrors the image-gen / video-gen read guards."""
from agent.file_safety import get_read_block_error
blocked = get_read_block_error(file_path)
@@ -547,23 +497,19 @@ def _read_block_error(file_path: str) -> Optional[Dict[str, Any]]:
def _transcribe_prepared_audio(
file_path: str,
model: Optional[str] = None,
source: Optional[str] = None,
file_path: str, model: Optional[str] = None, source: Optional[str] = None
) -> Dict[str, Any]:
"""Transcribe a validated audio file with the configured STT provider.
``model`` overrides the config/provider default; ``source`` is a caller-surface
label (``"gateway"``, ``"voice_mode"``) forwarded to the ``pre_transcription``
hook for observability only. Returns the standard result envelope.
``model`` overrides the config default; ``source`` is a caller-surface label
(``"gateway"``, ``"voice_mode"``) forwarded to the ``pre_transcription`` hook only.
"""
blocked = _read_block_error(file_path)
if blocked:
return blocked
# Validate before provider resolution so invalid files cannot trigger
# provider setup or lazy installation. The remote-upload size cap is
# enforced below, only for non-local providers.
# Validate before provider resolution so invalid files can't trigger provider setup
# or lazy installation; the remote-upload size cap applies to non-local only.
error = _validate_audio_file(file_path, enforce_size_limit=False)
if error:
return error
@@ -599,9 +545,8 @@ def _transcribe_prepared_audio(
shutil.rmtree(trim_cleanup_dir, ignore_errors=True)
# Built-in provider -> (stt section, config key, default, treat-empty-as-missing).
# "local_command" shares the ``stt.local`` section; xAI takes no model parameter
# (the name is logging-only); deepinfra resolves from the live catalog when empty.
# Built-in provider -> (stt section, config key, default, treat-empty-as-missing). "local_command"
# shares ``stt.local``; xAI takes no model (logging-only); deepinfra uses the live catalog when empty.
_BUILTIN_MODEL_KEYS = {
"local": ("local", "model", DEFAULT_LOCAL_MODEL, False),
"local_command": ("local", "model", DEFAULT_LOCAL_MODEL, False),
@@ -626,29 +571,17 @@ def _builtin_model_name(provider: str, stt_config: Dict[str, Any], model: Option
return cfg.get(key, default)
def _builtin_handler(provider: str):
"""Handler for a built-in provider, looked up in this module at call time so tests may patch ``_transcribe_*``."""
if provider not in BUILTIN_STT_PROVIDERS:
return None
return globals()[f"_transcribe_{provider}"]
def _dispatch_stt_provider(
file_path: str,
provider: str,
stt_config: Dict[str, Any],
model: Optional[str] = None,
file_path: str, provider: str, stt_config: Dict[str, Any], model: Optional[str] = None,
source: Optional[str] = None,
) -> Dict[str, Any]:
"""Route *file_path* to the handler for *provider* (built-in > command > plugin)."""
# Static ``stt.prompt`` is the base; pre_transcription hook results mutate
# on top in registration order (last hook to set a field wins).
# Static ``stt.prompt`` is the base; hook results mutate on top (last hook to set a field wins).
prompt = stt_config.get("prompt")
if not isinstance(prompt, str) or not prompt.strip():
prompt = None
# The hook fires after provider resolution and BEFORE any backend is
# invoked; ``language`` stays None unless a hook overrides it.
# Fires after provider resolution and BEFORE any backend; ``language`` stays None unless a hook sets it.
model, language, prompt = _apply_pre_transcription_hook(
file_path=file_path, provider=provider, model=model,
language=_get_stt_section(stt_config, provider).get("language"),
@@ -656,16 +589,16 @@ def _dispatch_stt_provider(
)
prompt = _enforce_prompt_length_limit(prompt, provider)
handler = _builtin_handler(provider)
if handler is not None:
if provider in BUILTIN_STT_PROVIDERS:
# Looked up in this module at call time so tests may patch ``_transcribe_*``.
handler = globals()[f"_transcribe_{provider}"]
model_name = _builtin_model_name(provider, stt_config, model)
if provider in ("local", "local_command"):
model_name = _normalize_local_model(model_name)
return handler(file_path, model_name, language=language, prompt=prompt)
# User-declared command provider: after built-ins (so ``stt.providers.openai
# .command`` can't override the real handler) and BEFORE plugins, because
# config is more local than a plugin install (same precedence as TTS).
# Command providers: after built-ins (``stt.providers.openai.command`` can't override the
# real handler) and BEFORE plugins, since config is more local than a plugin install.
command_provider_config = _resolve_command_stt_provider_config(provider, stt_config)
if command_provider_config is not None:
return _transcribe_command_stt(
@@ -673,14 +606,11 @@ def _dispatch_stt_provider(
model_override=model, language_override=language, prompt=prompt,
)
# Plugin-registered backend. Plugins read per-provider config under
# ``stt.<provider>`` like built-ins; the ``model`` argument overrides it.
# Plugin backend: reads ``stt.<provider>`` like built-ins; the ``model`` argument overrides it.
plugin_cfg = _get_stt_section(stt_config, provider)
plugin_result = _dispatch_to_plugin_provider(
file_path, provider, stt_config,
model=model or plugin_cfg.get("model"),
language=language or _resolve_stt_language(provider, stt_config),
prompt=prompt,
file_path, provider, stt_config, model=model or plugin_cfg.get("model"),
language=language or _resolve_stt_language(provider, stt_config), prompt=prompt,
)
if plugin_result is not None:
return plugin_result
@@ -693,9 +623,7 @@ def _no_provider_error(provider: str, stt_config: Dict[str, Any]) -> Dict[str, A
if "provider" in stt_config and provider_key and provider_key not in BUILTIN_STT_PROVIDERS and provider_key != "none":
return _unregistered_stt_provider_error(provider_key)
# An explicit openai selection flattened to "none" carries a
# selection-specific reason (e.g. managed openai-audio gateway down);
# surface it with its remediation instead of the all-provider hint.
# An explicit openai selection flattened to "none" has a specific reason (e.g. managed gateway down).
if provider_key == "none" and str(stt_config.get("provider") or "") == "openai" and _HAS_OPENAI:
reason = _openai_audio_unavailable_reason()
if reason is not None:
@@ -712,24 +640,20 @@ def _no_provider_error(provider: str, stt_config: Dict[str, Any]) -> Dict[str, A
def transcribe_audio(
file_path: str,
model: Optional[str] = None,
source: Optional[str] = None,
file_path: str, model: Optional[str] = None, source: Optional[str] = None
) -> Dict[str, Any]:
"""Safely validate, preprocess supported inputs, and dispatch transcription.
"""Validate, preprocess supported inputs, and dispatch transcription.
``source`` is an optional caller-surface label (``"gateway"``, ``"voice_mode"``)
forwarded to the ``pre_transcription`` hook for observability only.
``source`` is a caller-surface label (``"gateway"``, ``"voice_mode"``) forwarded to
the ``pre_transcription`` hook for observability only.
"""
# Secret-store refusal runs before ANY validation so the error names the
# real reason rather than a format error.
# Secret-store refusal runs before ANY validation so the error names the real reason.
blocked = _read_block_error(file_path)
if blocked:
return blocked
# Cap .silk sources before the decoder runs (decoder safety); for all other
# inputs the upload cap is provider-scoped in _transcribe_prepared_audio,
# so local whisper can handle big files.
# Cap .silk sources before the decoder runs; for other inputs the upload cap is
# provider-scoped in _transcribe_prepared_audio so local whisper can take big files.
is_silk = Path(file_path).suffix.lower() == ".silk"
source_error = _validate_audio_source_file(file_path, enforce_size_limit=is_silk)
if source_error:
@@ -750,13 +674,12 @@ def transcribe_audio(
def transcribe_audio_local_fallback(
file_path: str,
model: Optional[str] = None,
file_path: str, model: Optional[str] = None
) -> Dict[str, Any]:
"""Try an already-installed local STT backend without changing config.
For passive inbound-media recovery after the configured provider failed:
never lazy-installs or falls through to a cloud provider.
Passive inbound-media recovery after the configured provider failed: never
lazy-installs or falls through to a cloud provider.
"""
error = _validate_audio_file(file_path)
if error:
@@ -770,5 +693,3 @@ def transcribe_audio_local_fallback(
if _has_local_command():
return _transcribe_local_command(file_path, _normalize_local_model(local_model))
return _error_result("No installed local STT backend is available.", provider="local")
tools/tts_command_provider.py
+111 -185
View File
@@ -1,25 +1,20 @@
"""Shared runner for user-configured shell ("command") TTS/STT providers.
Both ``tools.tts_tool`` and ``tools.transcription_tools`` let users declare a
provider as a shell command template with ``{placeholders}``. This module owns
the shell-quote-aware template rendering and the idle-timeout process runner
they share, plus the TTS side's ``tts.providers.<name>`` config layer. Each
origin module re-imports these under its historical private names.
TTS config shape::
``tools.tts_tool`` and ``tools.transcription_tools`` both let users declare a
provider as a shell command template with ``{placeholders}`` (``{{``/``}}`` stay
literal; values are shell-quoted for their surrounding quote context). This
module owns the quote-aware rendering, the idle-timeout process runner and the
generic ``<section>.providers.<name>`` config readers; each origin module
re-imports them under its historical private names. TTS config shape::
tts:
provider: piper-en
providers:
piper-en:
type: command
command: "piper -m ~/model.onnx -f {output_path} < {input_path}"
output_format: wav
piper-en: {type: command, command: "piper -f {output_path} < {input_path}", output_format: wav}
Placeholders: ``{input_path}``, ``{text_path}`` (alias), ``{output_path}``,
``{format}``, ``{voice}``, ``{model}``, ``{speed}``; ``{{``/``}}`` for literal
braces. Values are shell-quoted for their surrounding quote context. Built-in
provider names always win over a same-named entry under ``tts.providers``.
TTS placeholders: ``{input_path}``/``{text_path}``, ``{output_path}``, ``{format}``,
``{voice}``, ``{model}``, ``{speed}``. Built-in provider names always win over a
same-named entry under ``providers``.
"""
from __future__ import annotations
@@ -33,7 +28,7 @@ import tempfile
import threading
import time
from pathlib import Path
from typing import Any, Dict, Optional
from typing import Any, Dict, FrozenSet, Optional
def shell_quote_context(command_template: str, position: int) -> Optional[str]:
@@ -68,43 +63,26 @@ def quote_command_placeholder(value: str, quote_context: Optional[str]) -> str:
if quote_context == "'":
return value.replace("'", r"'\''")
if quote_context == '"':
return (
value
.replace("\\", "\\\\")
.replace('"', r'\"')
.replace("$", r"\$")
.replace("`", r"\`")
)
return value.replace("\\", "\\\\").replace('"', r'\"').replace("$", r"\$").replace("`", r"\`")
if os.name == "nt":
return subprocess.list2cmdline([value])
return shlex.quote(value)
def render_command_template(
command_template: str,
placeholders: Dict[str, str],
) -> str:
def render_command_template(command_template: str, placeholders: Dict[str, str]) -> str:
"""Replace ``{name}`` placeholders (quote-aware) while preserving ``{{``/``}}``."""
names = "|".join(re.escape(name) for name in placeholders)
pattern = re.compile(
rf"(?<!\$)(?:\{{\{{(?P<double>{names})\}}\}}|\{{(?P<single>{names})\}})"
)
pattern = re.compile(rf"(?<!\$)(?:\{{\{{(?P<double>{names})\}}\}}|\{{(?P<single>{names})\}})")
replacements: list[tuple[str, str]] = []
def replace_match(match: re.Match[str]) -> str:
name = match.group("double") or match.group("single")
token = f"__HERMES_CMD_PLACEHOLDER_{len(replacements)}__"
replacements.append((
token,
quote_command_placeholder(
placeholders[name],
shell_quote_context(command_template, match.start()),
),
))
quoted = quote_command_placeholder(placeholders[name], shell_quote_context(command_template, match.start()))
replacements.append((token, quoted))
return token
rendered = pattern.sub(replace_match, command_template)
rendered = rendered.replace("{{", "{").replace("}}", "}")
rendered = pattern.sub(replace_match, command_template).replace("{{", "{").replace("}}", "}")
for token, value in replacements:
rendered = rendered.replace(token, value)
return rendered
@@ -130,20 +108,15 @@ def terminate_command_process_tree(proc: subprocess.Popen) -> None:
"""Best-effort termination of a shell process and all of its children."""
if proc.poll() is not None:
return
if os.name == "nt":
try:
subprocess.run(
["taskkill", "/F", "/T", "/PID", str(proc.pid)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=5,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5, stdin=subprocess.DEVNULL,
)
except Exception:
proc.kill()
return
try:
import psutil # type: ignore
except ImportError:
@@ -153,40 +126,35 @@ def terminate_command_process_tree(proc: subprocess.Popen) -> None:
except subprocess.TimeoutExpired:
proc.kill()
return
_signal_process_tree(psutil, proc, "terminate")
try:
proc.wait(timeout=2)
return
except subprocess.TimeoutExpired:
pass
_signal_process_tree(psutil, proc, "kill")
_signal_process_tree(psutil, proc, "kill")
def command_env_passthrough(config: Dict[str, Any]) -> list:
"""Return the provider's ``env_passthrough`` allowlist.
The child env is scrubbed of Hermes secrets by default; this list names
variables copied back from the parent env so a trusted template (e.g. a
curl one-liner using its own API key) keeps working.
"""
"""``env_passthrough`` allowlist: parent env vars copied back into the secret-scrubbed child env."""
raw = config.get("env_passthrough")
if not isinstance(raw, (list, tuple)):
return []
return [str(item).strip() for item in raw if str(item).strip()]
def command_failure_detail(exc: subprocess.CalledProcessError) -> str:
"""``stderr: ...; stdout: ...`` for a failed command provider, or ``no command output``."""
parts = [f"{stream}: {text.strip()}" for stream, text in (("stderr", exc.stderr), ("stdout", exc.stdout)) if text]
return "; ".join(parts) or "no command output"
def run_command_provider(
command: str,
timeout: float,
env_passthrough: Optional[list] = None,
command: str, timeout: float, env_passthrough: Optional[list] = None,
) -> subprocess.CompletedProcess:
"""Run a command-provider shell command with process-tree idle cleanup.
``timeout`` is an IDLE timeout, reset whenever the command emits output —
a slow-but-alive provider survives, a silently stalled one is killed.
Child env is scrubbed of Hermes secrets while propagating delegated-child
lineage markers.
``timeout`` is an IDLE timeout, reset whenever the command emits output — a
slow-but-alive provider survives, a silently stalled one is killed. Child env
is scrubbed of Hermes secrets while propagating delegated-child lineage markers.
"""
from agent.delegation_context import delegated_child_subprocess_env
from tools.environments.local import hermes_subprocess_env
@@ -197,15 +165,9 @@ def run_command_provider(
if value is not None:
scrubbed[key] = value
popen_kwargs: Dict[str, Any] = {
"shell": True,
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
"text": True,
# Lossy UTF-8 decode: locale-mismatched bytes must not raise in the
# reader threads on non-UTF-8 Windows.
"encoding": "utf-8",
"errors": "replace",
"env": delegated_child_subprocess_env(scrubbed),
"shell": True, "stdout": subprocess.PIPE, "stderr": subprocess.PIPE, "text": True,
# Lossy UTF-8 decode: locale-mismatched bytes must not raise in the reader threads.
"encoding": "utf-8", "errors": "replace", "env": delegated_child_subprocess_env(scrubbed),
}
if os.name == "nt":
popen_kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
@@ -222,10 +184,7 @@ def run_command_provider(
read1 = getattr(getattr(stream, "buffer", None), "read1", None)
try:
while True:
if read1 is None:
chunk = stream.read(65536)
else:
chunk = read1(65536).decode(encoding, errors="replace")
chunk = stream.read(65536) if read1 is None else read1(65536).decode(encoding, errors="replace")
if not chunk:
break
output_queue.put((name, chunk))
@@ -273,63 +232,42 @@ def run_command_provider(
break
if chunk:
chunks[name].append(chunk)
stdout = "".join(chunks["stdout"])
stderr = "".join(chunks["stderr"])
stdout, stderr = "".join(chunks["stdout"]), "".join(chunks["stderr"])
try:
raise subprocess.TimeoutExpired(command, timeout)
except subprocess.TimeoutExpired as exc:
raise subprocess.TimeoutExpired(
command, timeout, output=stdout, stderr=stderr,
) from exc
stdout = "".join(chunks["stdout"])
stderr = "".join(chunks["stderr"])
raise subprocess.TimeoutExpired(command, timeout, output=stdout, stderr=stderr) from exc
stdout, stderr = "".join(chunks["stdout"]), "".join(chunks["stderr"])
if proc.returncode:
raise subprocess.CalledProcessError(
proc.returncode, command, output=stdout, stderr=stderr,
)
raise subprocess.CalledProcessError(proc.returncode, command, output=stdout, stderr=stderr)
return subprocess.CompletedProcess(command, proc.returncode, stdout, stderr)
# ===========================================================================
# TTS ``tts.providers.<name>`` config layer
# Generic ``<section>.providers.<name>`` config layer (TTS and STT share it)
# ===========================================================================
# Any ``tts.provider`` value NOT in this set refers to ``tts.providers.<name>``.
BUILTIN_TTS_PROVIDERS = frozenset({
"edge", "elevenlabs", "openai", "minimax", "xai", "mistral", "gemini",
"neutts", "kittentts", "piper", "deepinfra",
})
DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS = 120
DEFAULT_COMMAND_TTS_OUTPUT_FORMAT = "mp3"
COMMAND_TTS_OUTPUT_FORMATS = frozenset(
{"mp3", "wav", "ogg", "flac", "m4a", "aac", "amr", "opus"}
)
DEFAULT_COMMAND_TTS_MAX_TEXT_LENGTH = 5000
def _get_provider_section(tts_config: Dict[str, Any], name: str) -> Dict[str, Any]:
"""Return a provider config block if it's a dict, else an empty dict."""
if not isinstance(tts_config, dict):
def _get_provider_section(config: Dict[str, Any], name: str) -> Dict[str, Any]:
"""Return ``config[name]`` if it's a dict, else an empty dict."""
if not isinstance(config, dict):
return {}
section = tts_config.get(name)
section = config.get(name)
return section if isinstance(section, dict) else {}
def _get_named_provider_config(tts_config: Dict[str, Any], name: str) -> Dict[str, Any]:
"""Config dict for a user-declared provider, or {}.
def _named_provider_config(config: Dict[str, Any], name: str, builtins: FrozenSet[str]) -> Dict[str, Any]:
"""``<section>.providers.<name>`` (canonical), else ``<section>.<name>`` for non-built-in names only.
``tts.providers.<name>`` is canonical; ``tts.<name>`` is accepted as
back-compat only for non-built-in names (so a user's ``tts.openai`` block
still means the OpenAI provider, not a custom command).
The back-compat form is refused for built-ins so a user's ``openai:`` block
still means the OpenAI provider, not a custom command.
"""
section = _get_provider_section(tts_config, "providers").get(name)
section = _get_provider_section(config, "providers").get(name)
if isinstance(section, dict):
return section
if name.lower() not in BUILTIN_TTS_PROVIDERS:
return _get_provider_section(tts_config, name)
if name.lower() not in builtins:
return _get_provider_section(config, name)
return {}
@@ -344,59 +282,76 @@ def _is_command_provider_config(config: Dict[str, Any]) -> bool:
return isinstance(command, str) and bool(command.strip())
def _resolve_command_provider_config(
provider: str,
tts_config: Dict[str, Any],
def _resolve_command_config(
provider: str, config: Dict[str, Any], reserved: FrozenSet[str],
) -> Optional[Dict[str, Any]]:
"""The provider config when *provider* is a user-declared command provider.
None for built-in names (native handlers win), unknown names, or
non-command types.
"""
"""Config of a user-declared command provider; None for *reserved* names, unknown or non-command."""
if not provider:
return None
key = provider.lower().strip()
if key in BUILTIN_TTS_PROVIDERS:
if key in reserved:
return None
config = _get_named_provider_config(tts_config, key)
return config if _is_command_provider_config(config) else None
named = _named_provider_config(config, key, reserved)
return named if _is_command_provider_config(named) else None
def _command_timeout(config: Dict[str, Any], default: float) -> float:
"""Timeout in seconds (``timeout`` > ``timeout_seconds``); invalid or non-positive -> *default*."""
raw = config.get("timeout", config.get("timeout_seconds", default))
try:
value = float(raw)
except (TypeError, ValueError):
return float(default)
return value if value > 0 else float(default)
def _command_output_format(config: Dict[str, Any], formats: FrozenSet[str], default: str) -> str:
"""Validated ``format``/``output_format`` from *config*, else *default*."""
raw = config.get("format") or config.get("output_format") or default
fmt = str(raw).lower().strip().lstrip(".")
return fmt if fmt in formats else default
# ---- TTS ``tts.providers.<name>`` layer -----------------------------------
# Any ``tts.provider`` value NOT in this set refers to ``tts.providers.<name>``.
BUILTIN_TTS_PROVIDERS = frozenset({
"edge", "elevenlabs", "openai", "minimax", "xai", "mistral", "gemini",
"neutts", "kittentts", "piper", "deepinfra",
})
DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS = 120
DEFAULT_COMMAND_TTS_OUTPUT_FORMAT = "mp3"
COMMAND_TTS_OUTPUT_FORMATS = frozenset({"mp3", "wav", "ogg", "flac", "m4a", "aac", "amr", "opus"})
DEFAULT_COMMAND_TTS_MAX_TEXT_LENGTH = 5000
def _get_named_provider_config(tts_config: Dict[str, Any], name: str) -> Dict[str, Any]:
return _named_provider_config(tts_config, name, BUILTIN_TTS_PROVIDERS)
def _resolve_command_provider_config(provider: str, tts_config: Dict[str, Any]) -> Optional[Dict[str, Any]]:
return _resolve_command_config(provider, tts_config, BUILTIN_TTS_PROVIDERS)
def _iter_command_providers(tts_config: Dict[str, Any]):
"""Yield (name, config) pairs for every declared command-type provider."""
for name, cfg in _get_provider_section(tts_config, "providers").items():
if (
isinstance(name, str)
and name.lower() not in BUILTIN_TTS_PROVIDERS
and _is_command_provider_config(cfg)
):
if isinstance(name, str) and name.lower() not in BUILTIN_TTS_PROVIDERS and _is_command_provider_config(cfg):
yield name, cfg
def _get_command_tts_timeout(config: Dict[str, Any]) -> float:
"""Timeout in seconds; invalid or non-positive values fall back to the default."""
raw = config.get("timeout", config.get("timeout_seconds", DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS))
try:
value = float(raw)
except (TypeError, ValueError):
return float(DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS)
if value <= 0:
return float(DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS)
return value
return _command_timeout(config, DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS)
def _get_command_tts_output_format(
config: Dict[str, Any],
output_path: Optional[str] = None,
) -> str:
def _get_command_tts_output_format(config: Dict[str, Any], output_path: Optional[str] = None) -> str:
"""Validated output format: the output path's suffix wins, then ``format``/``output_format``."""
if output_path:
suffix = Path(output_path).suffix.lower().strip().lstrip(".")
if suffix in COMMAND_TTS_OUTPUT_FORMATS:
return suffix
raw = config.get("format") or config.get("output_format") or DEFAULT_COMMAND_TTS_OUTPUT_FORMAT
fmt = str(raw).lower().strip().lstrip(".")
return fmt if fmt in COMMAND_TTS_OUTPUT_FORMATS else DEFAULT_COMMAND_TTS_OUTPUT_FORMAT
return _command_output_format(config, COMMAND_TTS_OUTPUT_FORMATS, DEFAULT_COMMAND_TTS_OUTPUT_FORMAT)
def _is_command_tts_voice_compatible(config: Dict[str, Any]) -> bool:
@@ -413,23 +368,16 @@ def _configured_command_tts_output_path(path: Path, config: Dict[str, Any]) -> P
def _generate_command_tts(
text: str,
output_path: str,
provider_name: str,
config: Dict[str, Any],
tts_config: Dict[str, Any],
text: str, output_path: str, provider_name: str, config: Dict[str, Any], tts_config: Dict[str, Any],
) -> str:
"""Generate speech by running a user-configured shell command.
"""Generate speech by running a user-configured shell command; returns the audio path it wrote.
Returns the absolute path of the audio file the command wrote. Raises
``ValueError`` for invalid provider config and ``RuntimeError`` for
Raises ``ValueError`` for invalid provider config and ``RuntimeError`` for
timeouts / non-zero exits / empty output.
"""
command_template = str(config.get("command") or "").strip()
if not command_template:
raise ValueError(
f"tts.providers.{provider_name}.command is not configured"
)
raise ValueError(f"tts.providers.{provider_name}.command is not configured")
output = Path(output_path).expanduser()
output.parent.mkdir(parents=True, exist_ok=True)
@@ -443,46 +391,24 @@ def _generate_command_tts(
with tempfile.TemporaryDirectory() as tmpdir:
text_path = Path(tmpdir) / "input.txt"
text_path.write_text(text, encoding="utf-8")
placeholders = {
"input_path": str(text_path),
"text_path": str(text_path),
"output_path": str(output),
"format": output_format,
"voice": str(config.get("voice", "")),
"model": str(config.get("model", "")),
"speed": str(speed),
"input_path": str(text_path), "text_path": str(text_path), "output_path": str(output),
"format": output_format, "voice": str(config.get("voice", "")),
"model": str(config.get("model", "")), "speed": str(speed),
}
command = render_command_template(command_template, placeholders)
try:
# Resolved through the origin so tests patching
# ``tools.tts_tool._run_command_tts`` still intercept.
# Resolved through the origin so tests patching ``tools.tts_tool._run_command_tts`` still intercept.
from tools.tts_tool import _run_command_tts
_run_command_tts(
command,
timeout,
env_passthrough=command_env_passthrough(config),
)
_run_command_tts(command, timeout, env_passthrough=command_env_passthrough(config))
except subprocess.TimeoutExpired as exc:
raise RuntimeError(
f"TTS provider '{provider_name}' timed out after {timeout:g}s"
) from exc
raise RuntimeError(f"TTS provider '{provider_name}' timed out after {timeout:g}s") from exc
except subprocess.CalledProcessError as exc:
detail_parts = []
if exc.stderr:
detail_parts.append(f"stderr: {exc.stderr.strip()}")
if exc.stdout:
detail_parts.append(f"stdout: {exc.stdout.strip()}")
detail = "; ".join(detail_parts) or "no command output"
raise RuntimeError(
f"TTS provider '{provider_name}' exited with code "
f"{exc.returncode}: {detail}"
f"TTS provider '{provider_name}' exited with code {exc.returncode}: {command_failure_detail(exc)}"
) from exc
if not output.exists() or output.stat().st_size <= 0:
raise RuntimeError(
f"TTS provider '{provider_name}' produced no output at {output}"
)
raise RuntimeError(f"TTS provider '{provider_name}' produced no output at {output}")
return str(output)
tools/tts_streaming.py
+49 -123
View File
@@ -1,16 +1,11 @@
"""Provider-agnostic streaming TTS: sentence text → int16 PCM chunk iterator.
"""Provider-agnostic streaming TTS: sentence text → int16 mono PCM chunk iterator.
``stream_tts_to_speaker`` (``tools.tts_tool``) owns the sentence buffer,
sounddevice output and stop/queue protocol; this module owns the *provider*
half — turning one sentence into audio the moment it's ready so playback starts
on sentence one instead of after the whole reply.
One contract (int16 mono PCM at ``sample_rate``): **true streamers**
(`StreamingTTSProvider.stream`) wrap chunked APIs (ElevenLabs pcm_24000, OpenAI
pcm, …); providers with no chunked API (edge, the default) still get per-
*sentence* playback via the sync ``text_to_speech_tool`` path in the dispatcher.
Adding a streamer is ``@register("name")`` on a subclass; the dispatcher, config
gate (``tts.<name>.streaming``) and resolver come free.
``stream_tts_to_speaker`` (``tools.tts_tool``) owns the sentence buffer, sounddevice
output and stop/queue protocol; this module owns the *provider* half so playback
starts on sentence one. True streamers (``StreamingTTSProvider.stream``) wrap chunked
APIs; providers with no chunked API (edge, the default) get per-sentence playback via
the sync ``text_to_speech_tool`` path. Adding a streamer is ``@register("name")`` on
a subclass; the dispatcher, config gate (``tts.<name>.streaming``) and resolver come free.
"""
from __future__ import annotations
@@ -26,16 +21,14 @@ from tools.tts_tool import _get_provider, _load_tts_config, get_env_value
logger = logging.getLogger(__name__)
# Per-sentence PCM byte cap, mirroring the 16 MiB bounded-body invariant of the
# sync providers: a buggy or hostile endpoint must not feed unbounded audio.
# Per-sentence PCM byte cap, mirroring the sync providers' 16 MiB bounded-body invariant.
_STREAM_SENTENCE_BYTE_CAP = 16 * 1024 * 1024
def _resolve_key(env_var: str, provider_id: str) -> str:
"""Provider secret lookup (config > env/.env > credential pool).
"""Provider secret lookup (config > env/.env > credential pool); seam over ``tts_tool._resolve_provider_key``.
Monkeypatchable seam over ``tools.tts_tool._resolve_provider_key``. ALL
streaming-provider key lookups go through here — never bare ``get_env_value``.
ALL streaming-provider key lookups go through here — never bare ``get_env_value``.
"""
try:
from tools.tts_tool import _resolve_provider_key
@@ -49,17 +42,11 @@ def _gemini_key() -> str:
return _resolve_key("GEMINI_API_KEY", "gemini") or _resolve_key("GOOGLE_API_KEY", "gemini")
# ---------------------------------------------------------------------------
# Interruption latch — lets the model know it was cut off mid-speech
# ---------------------------------------------------------------------------
# When the user barges in on a spoken reply, the surface marks the latch; the
# next turn's submit path takes it and prepends SPEECH_INTERRUPTED_NOTE to the
# model-bound message (API-call local, never persisted). The TTL keeps a stale
# Interruption latch: when the user barges in on a spoken reply, the surface marks
# it; the next turn's submit path takes it and prepends SPEECH_INTERRUPTED_NOTE to
# the model-bound message (API-call local, never persisted). The TTL keeps a stale
# barge from annotating an unrelated message minutes later.
SPEECH_INTERRUPTED_NOTE = (
"[Note: the user interrupted your previous spoken reply before it finished.]"
)
SPEECH_INTERRUPTED_NOTE = "[Note: the user interrupted your previous spoken reply before it finished.]"
_INTERRUPT_TTL_S = 120.0
_interrupted_at: Optional[float] = None
@@ -83,10 +70,9 @@ _THINK_BLOCK_RE = re.compile(r"<think[\s>].*?</think>", flags=re.DOTALL)
class SentenceChunker:
"""Incremental sentence cutter for LLM token deltas.
Shared by the speaker pipeline and the speak-stream WebSocket so every
surface cuts speech identically. Strips ``<think>`` blocks (even split
across deltas) and merges fragments shorter than *min_len* into the
following sentence, so "Ha!" rides along instead of stalling as a tiny clip.
Shared by the speaker pipeline and the speak-stream WebSocket so every surface
cuts speech identically. Strips ``<think>`` blocks (even split across deltas) and
merges fragments shorter than *min_len* into the following sentence.
"""
def __init__(self, min_len: int = 20):
@@ -117,10 +103,6 @@ class SentenceChunker:
return [tail] if tail else []
# ---------------------------------------------------------------------------
# ABC + registry
# ---------------------------------------------------------------------------
class StreamingTTSProvider(ABC):
"""Yields raw int16, little-endian, mono PCM chunks at ``sample_rate``."""
@@ -149,7 +131,6 @@ def register(name: str) -> Callable[[type[StreamingTTSProvider]], type[Streaming
def _wrap(cls: type[StreamingTTSProvider]) -> type[StreamingTTSProvider]:
_REGISTRY[name] = cls
return cls
return _wrap
@@ -165,39 +146,30 @@ def _try_instantiate(name: str, tts_config: Dict) -> Optional[StreamingTTSProvid
return None
# Fallback priority for ``tts.streaming.provider: auto`` — best chunked
# latency/quality first. Deliberately hard-coded (a UX decision, not a config
# knob); edge is absent because it has no chunked-PCM API.
# Fallback priority for ``tts.streaming.provider: auto`` — best chunked latency/quality
# first. Deliberately hard-coded (a UX decision); edge is absent (no chunked-PCM API).
_PROVIDER_PRIORITY: List[str] = ["elevenlabs", "gemini", "openai", "xai"]
def resolve_streaming_provider(
tts_config: Dict,
preferred: Optional[str] = None,
tts_config: Dict, preferred: Optional[str] = None
) -> Optional[StreamingTTSProvider]:
"""Return a ready streamer for the *configured* provider, else ``None``.
1. ``tts.streaming.provider`` when set: a name pins that exact streamer
(or ``None`` if unusable); ``auto`` walks ``_PROVIDER_PRIORITY`` and
returns the first usable one.
2. Otherwise the configured TTS provider (or ``preferred``). ``None`` means
"no chunked API" — the dispatcher speaks per-sentence via the sync path,
preserving the user's chosen voice. We never silently swap providers
just to get streaming.
``tts.streaming.provider`` when set: a name pins that exact streamer (``None``
if unusable); ``auto`` returns the first usable in ``_PROVIDER_PRIORITY``.
Otherwise the configured TTS provider (or ``preferred``): ``None`` means "no
chunked API" — the dispatcher speaks per-sentence via the sync path, preserving
the user's chosen voice. We never silently swap providers just to get streaming.
"""
streaming_cfg = tts_config.get("streaming") or {}
pinned = str(streaming_cfg.get("provider") or "").lower().strip()
pinned = str((tts_config.get("streaming") or {}).get("provider") or "").lower().strip()
if pinned == "auto":
for name in _PROVIDER_PRIORITY:
inst = _try_instantiate(name, tts_config)
if inst is not None:
return inst
return None
if pinned:
return _try_instantiate(pinned, tts_config)
name = (preferred or _get_provider(tts_config)).lower().strip()
return _try_instantiate(name, tts_config)
return _try_instantiate(pinned or (preferred or _get_provider(tts_config)).lower().strip(), tts_config)
def _capped(chunks: Iterator[bytes], label: str) -> Iterator[bytes]:
@@ -206,16 +178,11 @@ def _capped(chunks: Iterator[bytes], label: str) -> Iterator[bytes]:
for chunk in chunks:
total += len(chunk)
if total > _STREAM_SENTENCE_BYTE_CAP:
logger.warning("%s exceeded %d bytes for one sentence; truncating",
label, _STREAM_SENTENCE_BYTE_CAP)
logger.warning("%s exceeded %d bytes for one sentence; truncating", label, _STREAM_SENTENCE_BYTE_CAP)
return
yield chunk
# ---------------------------------------------------------------------------
# Providers
# ---------------------------------------------------------------------------
@register("elevenlabs")
class ElevenLabsStreamer(StreamingTTSProvider):
"""ElevenLabs chunked HTTP → pcm_24000 (the original reference path)."""
@@ -229,25 +196,18 @@ class ElevenLabsStreamer(StreamingTTSProvider):
def stream(self, text: str) -> Iterator[bytes]:
from tools.tts_tool import _import_elevenlabs
from tools.tts_tool_providers import (
DEFAULT_ELEVENLABS_STREAMING_MODEL_ID,
DEFAULT_ELEVENLABS_VOICE_ID,
_elevenlabs_environment_kwargs,
DEFAULT_ELEVENLABS_STREAMING_MODEL_ID, DEFAULT_ELEVENLABS_VOICE_ID, _elevenlabs_environment_kwargs,
)
client = _import_elevenlabs()(
api_key=_resolve_key("ELEVENLABS_API_KEY", "elevenlabs"),
**_elevenlabs_environment_kwargs(self.section),
api_key=_resolve_key("ELEVENLABS_API_KEY", "elevenlabs"), **_elevenlabs_environment_kwargs(self.section),
)
voice_id = self.section.get("voice_id", DEFAULT_ELEVENLABS_VOICE_ID)
model_id = self.section.get(
"streaming_model_id",
self.section.get("model_id", DEFAULT_ELEVENLABS_STREAMING_MODEL_ID),
"streaming_model_id", self.section.get("model_id", DEFAULT_ELEVENLABS_STREAMING_MODEL_ID),
)
yield from client.text_to_speech.convert(
text=text,
voice_id=voice_id,
model_id=model_id,
output_format="pcm_24000",
text=text, voice_id=voice_id, model_id=model_id, output_format="pcm_24000",
)
@@ -275,30 +235,18 @@ class OpenAIStreamer(StreamingTTSProvider):
client = OpenAI(
api_key=(self.section.get("api_key") or resolve_openai_audio_api_key()),
base_url=(
self.section.get("base_url")
or get_env_value("OPENAI_BASE_URL")
or None
),
base_url=(self.section.get("base_url") or get_env_value("OPENAI_BASE_URL") or None),
)
model = self.section.get("model", "gpt-4o-mini-tts")
voice = self.section.get("voice", "alloy")
with client.audio.speech.with_streaming_response.create(
model=model,
voice=voice,
input=text,
response_format="pcm",
model=self.section.get("model", "gpt-4o-mini-tts"), voice=self.section.get("voice", "alloy"),
input=text, response_format="pcm",
) as response:
yield from _capped(response.iter_bytes(), "OpenAI streaming TTS")
@register("gemini")
class GeminiStreamer(StreamingTTSProvider):
"""Gemini ``streamGenerateContent?alt=sse`` → base64 PCM chunks (24 kHz).
``?alt=sse`` flips the response from one JSON blob to an SSE feed of
base64 PCM chunks. Uses requests with a bounded streamed body.
"""
"""Gemini ``streamGenerateContent?alt=sse`` → SSE feed of base64 PCM chunks (24 kHz), bounded streamed body."""
sample_rate = 24000
@@ -313,40 +261,27 @@ class GeminiStreamer(StreamingTTSProvider):
import requests
from tools.tts_tool_providers import (
DEFAULT_GEMINI_TTS_BASE_URL,
DEFAULT_GEMINI_TTS_MODEL,
DEFAULT_GEMINI_TTS_VOICE,
DEFAULT_GEMINI_TTS_BASE_URL, DEFAULT_GEMINI_TTS_MODEL, DEFAULT_GEMINI_TTS_VOICE,
)
api_key = _gemini_key()
model = str(self.section.get("model", DEFAULT_GEMINI_TTS_MODEL)).strip() or DEFAULT_GEMINI_TTS_MODEL
voice = str(self.section.get("voice", DEFAULT_GEMINI_TTS_VOICE)).strip() or DEFAULT_GEMINI_TTS_VOICE
base_url = str(
self.section.get("base_url")
or get_env_value("GEMINI_BASE_URL")
or DEFAULT_GEMINI_TTS_BASE_URL
self.section.get("base_url") or get_env_value("GEMINI_BASE_URL") or DEFAULT_GEMINI_TTS_BASE_URL
).strip().rstrip("/")
payload = {
"contents": [{"parts": [{"text": text}]}],
"generationConfig": {
"responseModalities": ["AUDIO"],
"speechConfig": {
"voiceConfig": {
"prebuiltVoiceConfig": {"voiceName": voice},
},
},
"speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": voice}}},
},
}
url = f"{base_url}/models/{model}:streamGenerateContent"
def _sse_chunks() -> Iterator[bytes]:
with requests.post(
url,
params={"alt": "sse", "key": api_key},
json=payload,
timeout=60,
stream=True,
url, params={"alt": "sse", "key": api_key}, json=payload, timeout=60, stream=True,
) as response:
response.raise_for_status()
for line in response.iter_lines(decode_unicode=True):
@@ -374,9 +309,9 @@ class GeminiStreamer(StreamingTTSProvider):
class XAIStreamer(StreamingTTSProvider):
"""xAI WebSocket TTS (``wss://api.x.ai/v1/tts``) → binary PCM frames (24 kHz mono int16).
Credentials route through ``resolve_xai_http_credentials`` (OAuth or
XAI_API_KEY), same as the sync path. The async WS loop is bridged to the
sync iterator contract via ``_collect_async`` — the seam unit tests patch.
Credentials route through ``resolve_xai_http_credentials`` (OAuth or XAI_API_KEY),
same as the sync path. ``_collect_async`` bridges the async WS loop to the sync
iterator contract — the seam unit tests patch.
"""
sample_rate = 24000
@@ -394,8 +329,6 @@ class XAIStreamer(StreamingTTSProvider):
def stream(self, text: str) -> Iterator[bytes]:
yield from _capped(iter(self._collect_async(text)), "xAI streaming TTS")
# -- async→sync bridge (test seam) ------------------------------------
def _collect_async(self, text: str) -> List[bytes]:
import asyncio
@@ -420,18 +353,10 @@ class XAIStreamer(StreamingTTSProvider):
if not api_key:
raise RuntimeError("No xAI credentials for streaming TTS")
voice = str(self.section.get("voice_id", DEFAULT_XAI_VOICE_ID)).strip() or DEFAULT_XAI_VOICE_ID
ws_url = str(
self.section.get("streaming_url") or "wss://api.x.ai/v1/tts"
).strip()
ws_url = str(self.section.get("streaming_url") or "wss://api.x.ai/v1/tts").strip()
async with websockets.connect(
ws_url, extra_headers={"Authorization": f"Bearer {api_key}"}
) as ws:
await ws.send(_json.dumps({
"text": text,
"voice_id": voice,
"response_format": "pcm",
}))
async with websockets.connect(ws_url, extra_headers={"Authorization": f"Bearer {api_key}"}) as ws:
await ws.send(_json.dumps({"text": text, "voice_id": voice, "response_format": "pcm"}))
try:
while True:
message = await ws.recv()
@@ -448,8 +373,9 @@ class XAIStreamer(StreamingTTSProvider):
if etype == "done":
return
if etype == "error":
logger.warning("xAI WS error envelope: %s",
envelope.get("error") or envelope.get("message") or envelope)
logger.warning(
"xAI WS error envelope: %s", envelope.get("error") or envelope.get("message") or envelope,
)
return
except Exception as exc:
if exc.__class__.__name__ == "ConnectionClosed":
tools/tts_text_normalize.py
+73 -119
View File
@@ -1,11 +1,8 @@
"""Utilities for preparing assistant text for speech synthesis.
"""Deterministic cleanup turning assistant Markdown into a spoken script.
The TTS provider should receive a spoken script, not raw chat Markdown. This
module centralises the lightweight, deterministic cleanup used by explicit TTS
calls and gateway auto-TTS replies.
Non-ASCII characters are written as escapes on purpose so the file stays free of
invisible/look-alike glyphs.
Shared by explicit TTS calls, gateway auto-TTS, voice-mode streaming and the web
dashboard. Non-ASCII characters are written as escapes on purpose so the file
stays free of invisible/look-alike glyphs.
"""
from __future__ import annotations
@@ -13,9 +10,9 @@ from __future__ import annotations
import html
import re
# Sentinel appended to former heading lines so smooth_whitespace_for_tts can
# fold a heading into the sentence that follows it ("Weather, it will be sunny")
# rather than leaving a bare "Weather." label that reads abruptly aloud.
# Sentinel appended to former heading lines so smooth_whitespace_for_tts folds the
# heading into the sentence after it ("Weather, it will be sunny") instead of a bare
# "Weather." label.
_HEAD = "\x00"
_MD_CODE_BLOCK_RE = re.compile(r"```[\s\S]*?```")
@@ -34,21 +31,22 @@ _MD_HR_RE = re.compile(r"^\s*[-*_]{3,}\s*$", flags=re.MULTILINE)
_MD_TABLE_PIPE_RE = re.compile(r"\s*\|\s*")
_URL_RE = re.compile(r"https?://\S+")
# Broad emoji / pictograph cleanup. Voice providers vary a lot here; most read
# emojis as awkward labels, so keep the speech script calm and literal.
# Unit suffix (regex, after a digit) -> spoken word; km/h variants before the bare "m".
_UNIT_WORDS = (
(r"km\s*/\s*h", "kilometres per hour"), (r"km/h", "kilometres per hour"),
(r"mm", "millimetres"), (r"cm", "centimetres"), (r"m", "metres"),
)
# Currency prefix (regex) -> spoken word; order matters (NZ$/A$/US$ before bare $).
_CURRENCY_WORDS = (
(r"NZ\$", "New Zealand dollars", re.IGNORECASE), (r"A\$", "Australian dollars", re.IGNORECASE),
(r"US\$", "US dollars", re.IGNORECASE), ("€", "euros", 0), ("£", "pounds", 0), (r"\$", "dollars", 0),
)
# Broad emoji / pictograph cleanup: most voice providers read emojis as awkward labels.
_EMOJI_RE = re.compile(
"["
"\U0001F1E6-\U0001F1FF"
"\U0001F300-\U0001F5FF"
"\U0001F600-\U0001F64F"
"\U0001F680-\U0001F6FF"
"\U0001F700-\U0001F77F"
"\U0001F780-\U0001F7FF"
"\U0001F800-\U0001F8FF"
"\U0001F900-\U0001F9FF"
"\U0001FA00-\U0001FAFF"
"☀-➿"
"]+",
"[\U0001F1E6-\U0001F1FF\U0001F300-\U0001F5FF\U0001F600-\U0001F64F\U0001F680-\U0001F6FF"
"\U0001F700-\U0001F77F\U0001F780-\U0001F7FF\U0001F800-\U0001F8FF\U0001F900-\U0001F9FF"
"\U0001FA00-\U0001FAFF☀-➿]+",
flags=re.UNICODE,
)
_VARIATION_SELECTOR_RE = re.compile("[︎️]")
@@ -70,34 +68,29 @@ def strip_markdown_for_tts(text: str) -> str:
text = _MD_ITALIC_RE.sub(r"\1", text)
text = _MD_UNDERSCORE_ITALIC_RE.sub(r"\1", text)
text = _MD_STRIKE_RE.sub(r"\1", text)
# Mark headings (do not just delete the marker): the whitespace pass folds a
# heading into the sentence after it so speech says "Weather, it will be
# sunny" instead of a clipped "Weather." then a separate sentence.
# Mark headings (do not just delete the marker): see _HEAD.
text = _MD_HEADING_LINE_RE.sub(lambda m: m.group(1).rstrip() + _HEAD, text)
text = _MD_BLOCKQUOTE_RE.sub("", text)
text = _MD_LIST_ITEM_RE.sub("", text)
text = _MD_HR_RE.sub("", text)
# Pipe tables are terrible read aloud. Turn any leftover pipes into pauses
# instead of letting a provider speak "vertical bar".
# Leftover table pipes become pauses instead of a spoken "vertical bar".
text = _MD_TABLE_PIPE_RE.sub("; ", text)
return text
def _normalize_temperature_ranges(text: str) -> str:
# 11-17 degrees C -> "11 to 17 degrees Celsius" (en/em dash or hyphen).
text = re.sub(
r"(?<!\w)([-+\u2212]?\d+(?:\.\d+)?)\s*[\u2013\u2014-]\s*([-+\u2212]?\d+(?:\.\d+)?)\s*°\s*C\b",
lambda m: f"{m.group(1).replace(chr(0x2212), '-')} to {m.group(2).replace(chr(0x2212), '-')} degrees Celsius",
text,
flags=re.IGNORECASE,
)
text = re.sub(
r"(?<!\w)([-+\u2212]?\d+(?:\.\d+)?)\s*[\u2013\u2014-]\s*([-+\u2212]?\d+(?:\.\d+)?)\s*°\s*F\b",
lambda m: f"{m.group(1).replace(chr(0x2212), '-')} to {m.group(2).replace(chr(0x2212), '-')} degrees Fahrenheit",
text,
flags=re.IGNORECASE,
)
"""``11-17°C`` -> ``11 to 17 degrees Celsius`` (en/em dash or hyphen; unicode minus normalized)."""
number = r"([-+\u2212]?\d+(?:\.\d+)?)"
for unit, word in (("C", "Celsius"), ("F", "Fahrenheit")):
text = re.sub(
r"(?<!\w)" + number + r"\s*[\u2013\u2014-]\s*" + number + r"\s*°\s*" + unit + r"\b",
lambda m, w=word: (
f"{m.group(1).replace(chr(0x2212), '-')} to {m.group(2).replace(chr(0x2212), '-')} degrees {w}"
),
text,
flags=re.IGNORECASE,
)
return text
@@ -112,44 +105,37 @@ def normalize_symbols_for_tts(text: str) -> str:
text = text.replace("…", "...") # ellipsis
text = _normalize_temperature_ranges(text)
# Temperatures with a number. Do this before generic degree handling.
text = re.sub(r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°\s*C\b", r"\1 degrees Celsius", text, flags=re.IGNORECASE)
text = re.sub(r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°\s*F\b", r"\1 degrees Fahrenheit", text, flags=re.IGNORECASE)
# Bare units with no leading number ("measured in degrees C").
text = re.sub(r"°\s*C\b", "degrees Celsius", text, flags=re.IGNORECASE)
text = re.sub(r"°\s*F\b", "degrees Fahrenheit", text, flags=re.IGNORECASE)
# Any remaining degree symbol (angles, stray cases).
# Temperatures with a number first, then bare units ("measured in degrees C"),
# then any remaining degree symbol (angles, stray cases).
for unit, word in (("C", "Celsius"), ("F", "Fahrenheit")):
text = re.sub(
r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°\s*" + unit + r"\b", r"\1 degrees " + word, text, flags=re.IGNORECASE,
)
for unit, word in (("C", "Celsius"), ("F", "Fahrenheit")):
text = re.sub(r"°\s*" + unit + r"\b", "degrees " + word, text, flags=re.IGNORECASE)
text = re.sub(r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°", r"\1 degrees", text)
text = text.replace("°", " degrees")
# Common weather/travel units.
text = re.sub(r"(?<=\d)\s*km\s*/\s*h\b", " kilometres per hour", text, flags=re.IGNORECASE)
text = re.sub(r"(?<=\d)\s*km/h\b", " kilometres per hour", text, flags=re.IGNORECASE)
text = re.sub(r"(?<=\d)\s*mm\b", " millimetres", text, flags=re.IGNORECASE)
text = re.sub(r"(?<=\d)\s*cm\b", " centimetres", text, flags=re.IGNORECASE)
text = re.sub(r"(?<=\d)\s*m\b", " metres", text, flags=re.IGNORECASE)
for pattern, word in _UNIT_WORDS:
text = re.sub(r"(?<=\d)\s*" + pattern + r"\b", " " + word, text, flags=re.IGNORECASE)
# Numeric rates only ("5/month" -> "5 per month"). Requiring digit-then-letter
# keeps "and/or", "N/A", "TCP/IP" and dates like "2026/06" intact.
text = re.sub(r"(?<=\d)\s*/\s*(?=[A-Za-z])", " per ", text)
# Money and percentages. The integer part must END in a digit so a trailing
# comma ("A$50, ...") is not swallowed into the spoken amount.
text = re.sub(r"NZ\$\s*([\d,]*\d(?:\.\d+)?)", r"\1 New Zealand dollars", text, flags=re.IGNORECASE)
text = re.sub(r"A\$\s*([\d,]*\d(?:\.\d+)?)", r"\1 Australian dollars", text, flags=re.IGNORECASE)
text = re.sub(r"US\$\s*([\d,]*\d(?:\.\d+)?)", r"\1 US dollars", text, flags=re.IGNORECASE)
text = re.sub(r"€\s*([\d,]*\d(?:\.\d+)?)", r"\1 euros", text)
text = re.sub(r"£\s*([\d,]*\d(?:\.\d+)?)", r"\1 pounds", text)
text = re.sub(r"\$\s*([\d,]*\d(?:\.\d+)?)", r"\1 dollars", text)
# Money and percentages. The integer part must END in a digit so a trailing
# comma ("A$50, ...") is not swallowed into the spoken amount. Prefixed
# currencies run first so "$" doesn't eat "NZ$".
for symbol, word, flags in _CURRENCY_WORDS:
text = re.sub(symbol + r"\s*([\d,]*\d(?:\.\d+)?)", r"\1 " + word, text, flags=flags)
text = re.sub(r"(?<=\d)\s*%", " percent", text)
# Operators and separators that commonly leak from formatted answers.
text = text.replace("&", " and ")
text = re.sub("[•◦▪▫]", " ", text) # bullet glyphs
text = text.replace("→", " to ") # ->
text = text.replace("⇒", " to ") # =>
text = text.replace("≈", " about ") # almost equal
text = text.replace("~", " about ")
for symbol, word in (("→", " to "), ("⇒", " to "), ("≈", " about "), ("~", " about ")):
text = text.replace(symbol, word)
text = _VARIATION_SELECTOR_RE.sub("", text)
text = _EMOJI_RE.sub("", text)
@@ -159,10 +145,8 @@ def normalize_symbols_for_tts(text: str) -> str:
def smooth_whitespace_for_tts(text: str) -> str:
"""Collapse visual formatting into calm spoken paragraphs.
A former heading line (marked with the _HEAD sentinel) folds into the next
content line as a spoken lead-in: "Weather" + "It will be sunny" becomes
"Weather, It will be sunny." A heading with no content after it becomes its
own short sentence.
A _HEAD-marked heading folds into the next content line as a lead-in ("Weather,
It will be sunny."); a heading with no content after it becomes its own sentence.
"""
if not text:
return ""
@@ -182,8 +166,7 @@ def smooth_whitespace_for_tts(text: str) -> str:
is_heading = raw_line.rstrip().endswith(_HEAD)
line = raw_line.replace(_HEAD, "").strip()
if not line:
# Hold a pending heading across blank lines so it still folds into
# the next real content line; otherwise just collapse the blank.
# Hold a pending heading across blank lines so it still folds into the next content line.
if pending_heading is None and lines and lines[-1] != "":
lines.append("")
continue
@@ -209,44 +192,30 @@ def smooth_whitespace_for_tts(text: str) -> str:
return text.strip()
# Reasoning blocks: models with ``/reasoning show`` enabled emit
# ``<think>...</think>`` blocks in the final assistant message. Users want to
# SEE reasoning, not hear it read aloud (#34213).
# ``/reasoning show`` emits ``<think>...</think>`` in the final message: users want to
# SEE reasoning, not hear it. An unterminated block (streaming cut-off) is also silenced.
_THINK_BLOCK_RE = re.compile(r"<think[\s>].*?</think>", flags=re.DOTALL | re.IGNORECASE)
# An unterminated block (streaming cut-off) should still not be spoken.
_THINK_BLOCK_OPEN_RE = re.compile(r"<think[\s>].*\Z", flags=re.DOTALL | re.IGNORECASE)
# Turn-end file-mutation verifier footer appended by run_agent.py
# (``_format_file_mutation_failure_footer``). It's a UI affordance — reading
# "warning file mutation verifier, 2 files were NOT modified..." aloud is
# noise (#40772). The footer is a ``⚠️ File-mutation verifier:`` header line
# followed by indented ``•`` bullet lines; strip the whole block.
_VERIFIER_FOOTER_RE = re.compile(
r"^\s*⚠️?\s*File-mutation verifier:.*(?:\n[ \t]+•.*)*",
flags=re.MULTILINE,
)
# run_agent.py's turn-end file-mutation verifier footer (a ``⚠️ File-mutation verifier:``
# header line plus indented ``•`` bullets) is a UI affordance, not speech.
_VERIFIER_FOOTER_RE = re.compile(r"^\s*⚠️?\s*File-mutation verifier:.*(?:\n[ \t]+•.*)*", flags=re.MULTILINE)
def strip_nonspoken_blocks(text: str) -> str:
"""Remove blocks that must never reach a speech provider.
Currently: ``<think>`` reasoning blocks and the end-of-turn
file-mutation verifier footer.
"""
"""Remove ``<think>`` reasoning blocks and the file-mutation verifier footer."""
if not text:
return ""
text = _THINK_BLOCK_RE.sub(" ", text)
text = _THINK_BLOCK_OPEN_RE.sub(" ", text)
text = _VERIFIER_FOOTER_RE.sub(" ", text)
for pattern in (_THINK_BLOCK_RE, _THINK_BLOCK_OPEN_RE, _VERIFIER_FOOTER_RE):
text = pattern.sub(" ", text)
return text
def flatten_newlines_for_payload(text: str) -> str:
"""Collapse newlines into sentence breaks for single-line TTS payloads.
Some OpenAI-compatible backends (e.g. Kokoro) truncate synthesis at the
first newline (#9004). The smoothing pass already terminates each line
with punctuation, so newlines can safely become plain spaces.
Some OpenAI-compatible backends (e.g. Kokoro) truncate at the first newline; the
smoothing pass already terminates each line with punctuation, so this is safe.
"""
if not text:
return ""
@@ -259,28 +228,20 @@ def flatten_newlines_for_payload(text: str) -> str:
def prepare_spoken_text(text: str, max_chars: int | None = 4000) -> str:
"""Return a TTS-friendly script from assistant text.
"""Return a TTS-friendly script from assistant text (deterministic cleanup, not a rewrite).
Deterministic cleanup, not a semantic rewrite: it removes ``<think>``
reasoning blocks and the file-mutation verifier footer, removes Markdown,
expands common symbols such as a degree-Celsius sign to "degrees Celsius",
turns visual line formatting into speakable sentence pauses, and flattens
the result to a single line so newline-sensitive providers (Kokoro) speak
the whole script.
Pipeline: non-spoken blocks > Markdown > symbols/units > line formatting into
sentence pauses > single line (for newline-sensitive providers), then ``max_chars``.
"""
spoken = strip_nonspoken_blocks(text)
spoken = strip_markdown_for_tts(spoken)
spoken = normalize_symbols_for_tts(spoken)
spoken = smooth_whitespace_for_tts(spoken)
spoken = flatten_newlines_for_payload(spoken)
spoken = text
for step in (strip_nonspoken_blocks, strip_markdown_for_tts, normalize_symbols_for_tts,
smooth_whitespace_for_tts, flatten_newlines_for_payload):
spoken = step(spoken)
if max_chars is not None and max_chars > 0 and len(spoken) > max_chars:
spoken = spoken[:max_chars].rstrip()
return spoken
# ===========================================================================
# Speech text cleanup (shared by voice-mode streaming and gateway auto-TTS)
# ===========================================================================
# Legacy regex fallback, only used if the shared normalizer raises.
_LEGACY_TTS_STRIP_STEPS = (
(re.compile(r'<think[\s>].*?</think>', flags=re.DOTALL), ' '),
@@ -300,14 +261,7 @@ _LEGACY_TTS_STRIP_STEPS = (
def _strip_markdown_for_tts(text: str) -> str:
"""Prepare text for speech via the shared cleaner in tts_text_normalize.
One cleaner for every TTS path (tool, gateway auto-TTS, voice-mode
streaming, web dashboard): strips <think> blocks, the verifier footer,
markdown and emoji; expands units; flattens newlines so newline-sensitive
providers (Kokoro) speak the whole script. Falls back to the legacy regex
pipeline if the normalizer ever fails.
"""
"""``prepare_spoken_text`` without a length cap; falls back to the legacy regex pipeline if it raises."""
try:
return prepare_spoken_text(text, max_chars=None)
except Exception: