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

merge simp/r3-34-G pass2 into simp/r3-34

Browse Source
This commit is contained in:
Teknium
2026-09-03 01:21:15 -07:00
parent e9bed90bb3 ff44eca74f
commit 0f66267bf4
9 changed files with 344 additions and 737 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/transcription_audio.py
+31 -66
View File
@@ -14,6 +14,7 @@ import os
import shutil
import subprocess
import tempfile
from functools import partial
from pathlib import Path
from typing import Any, Dict, Optional
@@ -21,8 +22,7 @@ from hermes_cli._subprocess_compat import windows_hide_flags
from utils import is_truthy_value
from tools.transcription_common import (
COMMON_LOCAL_BIN_DIRS, LOCAL_NATIVE_AUDIO_FORMATS, MAX_FILE_SIZE, SUPPORTED_FORMATS,
_config_number, _error_result, _lazy_ensure_quietly, _process_error_detail,
)
_config_number, _error_result, _lazy_ensure_quietly, _process_error_detail)
# Log-record parity with the origin module.
logger = logging.getLogger("tools.transcription_tools")
@@ -37,16 +37,9 @@ def _find_binary(binary_name: str) -> Optional[str]:
return shutil.which(binary_name)
def _find_ffmpeg_binary() -> Optional[str]:
return _find_binary("ffmpeg")
def _find_ffprobe_binary() -> Optional[str]:
return _find_binary("ffprobe")
def _find_whisper_binary() -> Optional[str]:
return _find_binary("whisper")
_find_ffmpeg_binary = partial(_find_binary, "ffmpeg")
_find_ffprobe_binary = partial(_find_binary, "ffprobe")
_find_whisper_binary = partial(_find_binary, "whisper")
def _run_quiet(command: list, *, timeout: float, env: Optional[dict] = None) -> subprocess.CompletedProcess:
@@ -54,8 +47,7 @@ def _run_quiet(command: list, *, timeout: float, env: Optional[dict] = None) ->
return subprocess.run(
command, check=True, capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=timeout,
stdin=subprocess.DEVNULL, env=env, creationflags=windows_hide_flags(),
)
stdin=subprocess.DEVNULL, env=env, creationflags=windows_hide_flags())
# Shared encode profile for every STT-bound m4a (transcode and silence-trim):
@@ -65,19 +57,15 @@ _STT_M4A_ENCODE_ARGS = ("-vn", "-ac", "1", "-ar", "16000", "-c:a", "aac", "-b:a"
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 semantics."""
command = [ffmpeg, "-y", "-i", input_path]
if audio_filter:
command += ["-af", audio_filter]
command += [*_STT_M4A_ENCODE_ARGS, output_path]
_run_quiet(command, timeout=120)
filter_args = ["-af", audio_filter] if audio_filter else []
_run_quiet([ffmpeg, "-y", "-i", input_path, *filter_args, *_STT_M4A_ENCODE_ARGS, output_path],
timeout=120)
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; ``(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.
"""
Newer OpenAI models reject containers ``whisper-1`` accepted (notably Ogg/Opus voice notes) and
gateway downloads may carry a misleading extension."""
from tools.transcription_tools import _find_ffmpeg_binary, _run_ffmpeg_stt_encode
ffmpeg = _find_ffmpeg_binary()
if not ffmpeg:
@@ -121,13 +109,10 @@ def _validate_audio_source_file(file_path: str, *, enforce_size_limit: bool = Tr
def _validate_audio_file(file_path: str, *, enforce_size_limit: bool = True) -> Optional[Dict[str, Any]]:
"""Validate a supported, decoder-safe audio file."""
source_error = _validate_audio_source_file(file_path, enforce_size_limit=enforce_size_limit)
if source_error:
return source_error
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 None
if source_error or suffix.lower() in SUPPORTED_FORMATS:
return source_error
return _error_result(f"Unsupported format: {suffix}. Supported: {', '.join(sorted(SUPPORTED_FORMATS))}")
def _prepare_audio_for_transcription(file_path: str) -> tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
@@ -143,12 +128,10 @@ def _prepare_audio_for_transcription(file_path: str) -> tuple[Optional[str], Opt
return None, None, _error_result(
"Unsupported format: .silk. Install the optional 'pilk' dependency to enable WeChat voice transcription."
)
temp_dir = tempfile.mkdtemp(prefix="hermes-silk-")
converted_path = os.path.join(temp_dir, f"{audio_path.stem}.wav")
try:
import pilk
pilk.silk_to_wav(file_path, converted_path)
if not Path(converted_path).is_file() or Path(converted_path).stat().st_size == 0:
raise RuntimeError("pilk did not produce a readable WAV file")
@@ -165,11 +148,9 @@ def _prepare_local_audio(file_path: str, work_dir: str) -> tuple[Optional[str],
audio_path = Path(file_path)
if audio_path.suffix.lower() in LOCAL_NATIVE_AUDIO_FORMATS:
return file_path, None
ffmpeg = _find_ffmpeg_binary()
if not ffmpeg:
return None, "Local STT fallback requires ffmpeg for non-WAV inputs, but ffmpeg was not found"
converted_path = os.path.join(work_dir, f"{audio_path.stem}.wav")
try:
_run_quiet([ffmpeg, "-y", "-i", file_path, converted_path], timeout=300)
@@ -194,9 +175,7 @@ def _convert_caf_to_wav(file_path: str) -> Optional[str]:
("ffmpeg", [ffmpeg, "-y", "-i", file_path, wav_path] if ffmpeg else None),
("afconvert", [afconvert, file_path, wav_path, "-d", "LEI16", "-f", "WAVE"] if afconvert else None),
)
for label, command in candidates:
if not command:
continue
for label, command in ((label, cmd) for label, cmd in candidates if cmd):
try:
_run_quiet(command, timeout=300)
return wav_path
@@ -227,20 +206,17 @@ _CLOUD_TRIM_MIN_INPUT_SECONDS = 12.0
def _probe_audio_duration(file_path: str) -> Optional[float]:
"""Return the audio duration in seconds via ffprobe, or None.
Canonical sync probe; ``gateway/run.py._probe_audio_duration`` and the
Telegram adapter carry local variants — keep the command shape in sync.
"""
"""Return the audio duration in seconds via ffprobe, or None. Canonical sync probe;
``gateway/run.py._probe_audio_duration`` and the Telegram adapter carry local variants — keep
the command shape in sync."""
from tools.transcription_tools import _find_ffprobe_binary
ffprobe = _find_ffprobe_binary()
if not ffprobe:
return None
command = [
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())
probe = _run_quiet([ffprobe, "-v", "error", "-show_entries", "format=duration", "-of",
"default=noprint_wrappers=1:nokey=1", file_path], timeout=30)
return float(probe.stdout.strip())
except Exception: # noqa: BLE001 - probe is best-effort
return None
@@ -257,9 +233,7 @@ 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 (= upload the original).
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)
if not enabled:
@@ -274,19 +248,15 @@ def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> O
return None
name = Path(file_path).name
if original_duration < _CLOUD_TRIM_MIN_INPUT_SECONDS:
logger.debug(
"Cloud STT silence trim skipped for %s: %.1fs is below the %.0fs gate",
name, original_duration, _CLOUD_TRIM_MIN_INPUT_SECONDS,
)
logger.debug("Cloud STT silence trim skipped for %s: %.1fs is below the %.0fs gate",
name, original_duration, _CLOUD_TRIM_MIN_INPUT_SECONDS)
return None
keep_seconds = keep_ms / 1000.0
# 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}:"
f"stop_periods=-1:stop_threshold={threshold_db}dB:stop_silence={keep_seconds}"
)
f"stop_periods=-1:stop_threshold={threshold_db}dB:stop_silence={keep_seconds}")
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: output that is solely kept pause must never upload as "speech".
@@ -296,20 +266,15 @@ def _trim_silence_for_cloud_stt(file_path: str, stt_config: Dict[str, Any]) -> O
_run_ffmpeg_stt_encode(ffmpeg, file_path, trimmed_path, audio_filter=filter_expr)
trimmed_duration = _probe_audio_duration(trimmed_path)
if not trimmed_duration or trimmed_duration < min_result_seconds:
logger.debug(
"Cloud STT silence trim discarded for %s: trimmed result ~empty (%.2fs)", name, trimmed_duration or 0.0,
)
logger.debug("Cloud STT silence trim discarded for %s: trimmed result ~empty (%.2fs)",
name, trimmed_duration or 0.0)
return None
if trimmed_duration > original_duration * (1 - _CLOUD_TRIM_MIN_SAVING):
logger.debug(
"Cloud STT silence trim discarded for %s: saves <%.0f%% (%.1fs -> %.1fs)",
name, _CLOUD_TRIM_MIN_SAVING * 100, original_duration, trimmed_duration,
)
logger.debug("Cloud STT silence trim discarded for %s: saves <%.0f%% (%.1fs -> %.1fs)",
name, _CLOUD_TRIM_MIN_SAVING * 100, original_duration, trimmed_duration)
return None
logger.info(
"Trimmed silence from %s before cloud STT upload (%.1fs -> %.1fs, -%d%%)",
name, original_duration, trimmed_duration, round((1 - trimmed_duration / original_duration) * 100),
)
logger.info("Trimmed silence from %s before cloud STT upload (%.1fs -> %.1fs, -%d%%)", name,
original_duration, trimmed_duration, round((1 - trimmed_duration / original_duration) * 100))
keep_result = True
return trimmed_path
except Exception as exc: # noqa: BLE001 - trim is best-effort
tools/transcription_cloud.py
+50 -80
View File
@@ -21,11 +21,14 @@ from tools.transcription_audio import _transcode_audio_for_stt
from tools.transcription_common import (
DEFAULT_GROQ_STT_MODEL, DEFAULT_STT_MODEL, ELEVENLABS_STT_BASE_URL, GROQ_BASE_URL, GROQ_MODELS,
OPENAI_BASE_URL, OPENAI_MODELS, XAI_STT_BASE_URL, _error_result, _get_stt_section,
_lazy_ensure_quietly, _log_prompt_unsupported, _ok_result,
)
_lazy_ensure_quietly, _log_prompt_unsupported, _ok_result)
# Log-record parity with the origin module.
logger = logging.getLogger("tools.transcription_tools")
# Voxtral-style ``language xx <asr_text> ...`` prefix some SDK responses carry.
_ASR_TEXT_RE = re.compile(
r"\s*language\s+[\w.-]+(?:\s*<audio_language>[^<]*</audio_language>)?\s*<asr_text>\s*(?P<text>.*)",
flags=re.IGNORECASE | re.DOTALL)
def _has_xai_stt_credentials() -> bool:
@@ -35,10 +38,8 @@ def _has_xai_stt_credentials() -> bool:
def _with_openai_client(api_key: str, base_url: Optional[str], file_path: str, log_label: str, body):
"""Run ``body(client)`` on a fresh OpenAI SDK client (30s timeout, no retries); always closed.
Errors map to the shared envelope. APIConnectionError is checked before APITimeoutError
(its subclass) so timeouts report as connection errors, as they always have.
"""
Errors map to the shared envelope. APIConnectionError is checked before APITimeoutError (its
subclass) so timeouts report as connection errors, as they always have."""
try:
from openai import OpenAI
client = OpenAI(api_key=api_key, base_url=base_url, timeout=30, max_retries=0)
@@ -92,27 +93,24 @@ def _transcribe_groq(
language = language or _resolve_stt_language("groq")
def _run(client):
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, model=model_name,
response_format="text",
**_sdk_prompt_kwargs(language, prompt))
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,
prompt: Optional[str] = None,
) -> Dict[str, Any]:
"""Transcribe via the OpenAI ``audio.transcriptions.create`` SDK shape.
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.
"""
prompt: Optional[str] = None) -> Dict[str, Any]:
"""Transcribe via the OpenAI ``audio.transcriptions.create`` SDK shape, 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:
try:
@@ -137,8 +135,8 @@ def _transcribe_openai(
"model": model_name, "response_format": "text" if model_name == "whisper-1" else "json",
}
if language:
# gpt-transcribe takes a ``languages`` list and rejects the legacy field.
if model_name == "gpt-transcribe":
# gpt-transcribe takes a ``languages`` list and rejects the legacy field.
create_kwargs["extra_body"] = {"languages": [language]}
else:
create_kwargs["language"] = language
@@ -147,13 +145,11 @@ def _transcribe_openai(
create_kwargs["prompt"] = prompt
with open(path, "rb") as audio_file:
return client.audio.transcriptions.create(file=audio_file, **create_kwargs)
with tempfile.TemporaryDirectory(prefix="hermes-stt-") as work_dir:
try:
transcription = _create_transcription(file_path)
except BadRequestError as exc:
message = str(exc).lower()
if not any(k in message for k in ("unsupported", "corrupted", "invalid file")):
if not any(k in str(exc).lower() for k in ("unsupported", "corrupted", "invalid file")):
raise
# 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)
@@ -166,7 +162,6 @@ def _transcribe_openai(
logger.info("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)
@@ -181,24 +176,21 @@ def _transcribe_mistral(
try:
_lazy_ensure_quietly("stt.mistral")
from mistralai.client import 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")
result = client.audio.transcriptions.complete(
model=model_name, file={"content": audio_file, "file_name": Path(file_path).name},
**_sdk_prompt_kwargs(language, prompt),
)
transcript_text = _extract_transcript_text(result)
logger.info("Transcribed %s via Mistral API (%s, %d chars)",
Path(file_path).name, model_name, len(transcript_text))
return _ok_result(transcript_text, "mistral")
with Mistral(api_key=api_key) as client, open(file_path, "rb") as audio_file:
# Language: hook override > stt.mistral.language > stt.language > env > auto.
language = language or _resolve_stt_language("mistral")
result = client.audio.transcriptions.complete(
model=model_name, file={"content": audio_file, "file_name": Path(file_path).name},
**_sdk_prompt_kwargs(language, prompt))
transcript_text = _extract_transcript_text(result)
logger.info("Transcribed %s via Mistral API (%s, %d chars)",
Path(file_path).name, model_name, len(transcript_text))
return _ok_result(transcript_text, "mistral")
except Exception as e:
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
with open(file_path, "rb") as audio_file:
@@ -208,14 +200,11 @@ def _post_audio_multipart(url: str, headers: Dict[str, str], file_path: str, dat
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 multipart REST flow: ``post()`` -> ``log(text, body)`` -> ok envelope.
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);
exceptions -> ``_cloud_failure``.
"""
extract_text, log: Callable[[str, Dict[str, Any]], None]) -> Dict[str, Any]:
"""Shared multipart REST flow: ``post()`` -> ``log(text, body)`` -> ok envelope. 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); exceptions ->
``_cloud_failure``."""
try:
response = post()
if response.status_code != 200:
@@ -245,11 +234,9 @@ def _transcribe_xai(
# 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 = {"provider": "xai", "api_key": direct_api_key,
"base_url": str(get_env_value("XAI_BASE_URL") or "https://api.x.ai/v1").strip().rstrip("/")}
else:
creds = resolve_xai_http_credentials()
creds = {"provider": "xai", "api_key": direct_api_key,
"base_url": str(get_env_value("XAI_BASE_URL") or "https://api.x.ai/v1").strip().rstrip("/")
} if direct_api_key else resolve_xai_http_credentials()
api_key = str(creds.get("api_key") or "").strip()
if not api_key:
return _error_result("No xAI credentials found. Configure xAI OAuth in `hermes model` or set XAI_API_KEY")
@@ -258,10 +245,9 @@ def _transcribe_xai(
def _resolve_base_url(resolved_creds: Dict[str, str]) -> str:
# 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:
url = xai_config.get("base_url") or get_env_value("XAI_STT_BASE_URL") or resolved_creds.get("base_url")
url = resolved_creds.get("base_url")
if resolved_creds.get("provider") != "xai-oauth":
url = xai_config.get("base_url") or get_env_value("XAI_STT_BASE_URL") or url
return str(url or XAI_STT_BASE_URL).strip().rstrip("/")
# Language: hook override > stt.xai.language > stt.language > env.
@@ -270,9 +256,8 @@ def _transcribe_xai(
def _post() -> Any:
from tools.xai_http import hermes_xai_user_agent
data: Dict[str, str] = {"language": language} if language else {}
for flag, default in (("format", True), ("diarize", False)):
if is_truthy_value(xai_config.get(flag, default)):
data[flag] = "true"
data.update({flag: "true" for flag, default in (("format", True), ("diarize", False))
if is_truthy_value(xai_config.get(flag, default))})
def _post_transcription(bearer: str, endpoint_base_url: str):
headers = {"Authorization": f"Bearer {bearer}", "User-Agent": hermes_xai_user_agent()}
@@ -329,9 +314,7 @@ def _transcribe_elevenlabs(
"model_id": model_name,
"tag_audio_events": str(is_truthy_value(elevenlabs_config.get("tag_audio_events", False))).lower(),
"diarize": str(is_truthy_value(elevenlabs_config.get("diarize", False))).lower(),
}
if language_code:
data["language_code"] = language_code
**({"language_code": language_code} if language_code else {})}
return _post_audio_multipart(f"{base_url}/speech-to-text", {"xi-api-key": api_key}, file_path, data)
def _log(transcript_text: str, _body: Dict[str, Any]) -> None:
@@ -353,20 +336,16 @@ def _transcribe_deepinfra(
from hermes_cli.models import deepinfra_base_url, deepinfra_model_ids
# ``stt.deepinfra: null`` in YAML yields None, not {} — coalesce.
base_url = deepinfra_base_url(_get_stt_section(_load_stt_config(), "deepinfra"))
model_name = model_name or next(iter(deepinfra_model_ids("stt")), None)
if not model_name:
candidates = deepinfra_model_ids("stt")
if not candidates:
return _error_result(
"No DeepInfra STT model available. Pin one in config.yaml under stt.deepinfra.model, "
"or check connectivity to api.deepinfra.com so the live catalog can be fetched."
)
model_name = candidates[0]
return _error_result(
"No DeepInfra STT model available. Pin one in config.yaml under stt.deepinfra.model, "
"or check connectivity to api.deepinfra.com so the live catalog can be fetched.")
return _transcribe_openai(file_path, model_name, api_key=api_key, base_url=base_url,
provider_label="deepinfra", language=language, prompt=prompt)
# ---- OpenAI audio credential resolution -----------------------------------
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)."""
@@ -374,11 +353,9 @@ def _is_local_or_private_url(url: str) -> bool:
from urllib.parse import urlparse
import ipaddress
host = (urlparse(url).hostname or "").lower()
if not host:
return False
if host == "localhost" or host.endswith((".local", ".lan", ".internal")):
return True
addr = ipaddress.ip_address(host)
addr = ipaddress.ip_address(host) # raises for "" and non-IP hostnames
return addr.is_private or addr.is_loopback
except Exception: # unparsable URL or non-IP hostname
return False
@@ -403,8 +380,7 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]:
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,
)
resolve_managed_tool_gateway)
from tools.tool_backend_helpers import NOUS_MANAGED_PROVIDER, read_selection, selection_error
openai_cfg = _load_stt_config().get("openai") or {}
selected = read_selection("stt")
@@ -418,10 +394,8 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]:
if selected == NOUS_MANAGED_PROVIDER:
managed = _managed()
if managed is None:
raise ValueError(selection_error(
"stt", NOUS_MANAGED_PROVIDER,
"the Nous Tool Gateway is not available (not entitled or unreachable)",
))
raise ValueError(selection_error("stt", NOUS_MANAGED_PROVIDER,
"the Nous Tool Gateway is not available (not entitled or unreachable)"))
return managed
direct = _direct_openai_credentials(openai_cfg.get("api_key", ""), openai_cfg.get("base_url", ""))
if direct is not None:
@@ -429,8 +403,7 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]:
if selected is not None:
raise ValueError(selection_error(
"stt", selected,
"neither stt.openai.api_key in config nor VOICE_TOOLS_OPENAI_KEY/OPENAI_API_KEY is set",
))
"neither stt.openai.api_key in config nor VOICE_TOOLS_OPENAI_KEY/OPENAI_API_KEY is set"))
managed = _managed()
if managed is None:
message = "Neither stt.openai.api_key in config nor VOICE_TOOLS_OPENAI_KEY/OPENAI_API_KEY is set"
@@ -446,8 +419,5 @@ def _extract_transcript_text(transcription: Any) -> str:
if not isinstance(value, str) and isinstance(transcription, dict):
value = transcription.get("text")
text = (value if isinstance(value, str) else str(transcription)).strip()
match = re.match(
r"\s*language\s+[\w.-]+(?:\s*<audio_language>[^<]*</audio_language>)?\s*<asr_text>\s*(?P<text>.*)",
text, flags=re.IGNORECASE | re.DOTALL,
)
match = _ASR_TEXT_RE.match(text)
return match.group("text").strip() if match else text
tools/transcription_command.py
+43 -93
View File
@@ -12,6 +12,7 @@ from __future__ import annotations
import logging
import subprocess
import tempfile
from functools import partial
from pathlib import Path
from typing import Any, Dict, Optional
@@ -19,11 +20,9 @@ from tools.tts_command_provider import (
_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,
)
run_command_provider as _run_command_stt)
from tools.transcription_common import (
BUILTIN_STT_PROVIDERS, _error_result, _log_prompt_unsupported, _ok_result,
)
BUILTIN_STT_PROVIDERS, _error_result, _log_prompt_unsupported, _ok_result)
# Log-record parity with the origin module.
logger = logging.getLogger("tools.transcription_tools")
@@ -43,81 +42,57 @@ 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]:
"""``stt.providers.<name>`` (canonical), else ``stt.<name>`` for non-built-in names only."""
return _named_provider_config(stt_config, name, BUILTIN_STT_PROVIDERS)
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 _command_timeout(config, DEFAULT_COMMAND_STT_TIMEOUT_SECONDS)
def _get_command_stt_output_format(config: Dict[str, Any]) -> str:
return _command_output_format(config, COMMAND_STT_OUTPUT_FORMATS, DEFAULT_COMMAND_STT_OUTPUT_FORMAT)
# ``stt.providers.<name>`` (canonical), else ``stt.<name>`` for non-built-in names only.
_get_named_stt_provider_config = partial(_named_provider_config, builtins=BUILTIN_STT_PROVIDERS)
# The provider config if it is a command type; None for built-ins, ``none``, unknown.
_resolve_command_stt_provider_config = partial(_resolve_command_config,
reserved=_NON_COMMAND_STT_NAMES)
_get_command_stt_timeout = partial(_command_timeout, default=DEFAULT_COMMAND_STT_TIMEOUT_SECONDS)
_get_command_stt_output_format = partial(_command_output_format, formats=COMMAND_STT_OUTPUT_FORMATS,
default=DEFAULT_COMMAND_STT_OUTPUT_FORMAT)
def _read_command_stt_output(output_path: Path, stdout: str, fmt: str) -> str:
"""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()
except UnicodeDecodeError:
content = output_path.read_bytes().decode("utf-8", errors="replace").strip()
if content:
return content
if stdout and stdout.strip():
return stdout.strip()
content = (output_path.read_bytes().decode("utf-8", errors="replace").strip()
if output_path.exists() else "")
if content or (stdout or "").strip():
return content or stdout.strip()
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,
prompt: Optional[str] = None,
) -> Dict[str, Any]:
"""Transcribe via a user-declared ``stt.providers.<name>: type: command``.
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).
"""
prompt: Optional[str] = None) -> Dict[str, Any]:
"""Transcribe via a user-declared ``stt.providers.<name>: type: command``. 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
if prompt:
_log_prompt_unsupported(f"Command STT provider '{provider_name}'")
def fail(error: str) -> Dict[str, Any]:
return _error_result(error, provider=provider_name)
command_template = str(config.get("command") or "").strip()
if not command_template:
return fail(f"stt.providers.{provider_name}.command is not configured")
audio = Path(file_path).expanduser()
if not audio.exists():
return fail(f"Audio file not found: {file_path}")
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
)
model = model_override or config.get("model") or ""
language = (language_override or config.get("language")
or _resolve_stt_language(provider_name, stt_config) or DEFAULT_COMMAND_STT_LANGUAGE)
try:
with tempfile.TemporaryDirectory(prefix=f"hermes-cmd-stt-{provider_name}-") as tmpdir:
output_path = Path(tmpdir) / f"transcript.{output_format}"
placeholders = {
command = _render_command_stt_template(command_template, {
"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)
"language": str(language), "model": str(model_override or config.get("model") or ""),
})
logger.info("Transcribing %s via command STT provider '%s'...", audio.name, provider_name)
result = _run_command_stt(command, timeout, env_passthrough=_command_stt_env_passthrough(config))
transcript_text = _read_command_stt_output(output_path, result.stdout or "", output_format)
@@ -131,7 +106,6 @@ def _transcribe_command_stt(
return fail(str(exc))
except OSError as exc:
return fail(f"STT command provider '{provider_name}' failed: {exc}")
logger.info("Transcribed %s via command STT provider '%s' (%d chars)", audio.name, provider_name, len(transcript_text))
return _ok_result(transcript_text, provider_name)
@@ -144,8 +118,7 @@ def _unregistered_stt_provider_error(provider: str) -> Dict[str, Any]:
"installed STT plugins, or configure a command provider under "
f"`stt.providers.{key}.command`.",
provider=key,
error_type="provider_not_registered",
)
error_type="provider_not_registered")
def _dispatch_to_plugin_provider(
@@ -153,23 +126,18 @@ def _dispatch_to_plugin_provider(
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 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
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 _NON_COMMAND_STT_NAMES:
explicitly opted in via ``stt.provider``. Provider exceptions become the error envelope."""
key = (provider or "").lower().strip()
if not key or 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)):
return None
try:
from agent.transcription_registry import get_provider
from hermes_cli.plugins import _ensure_plugins_discovered
_ensure_plugins_discovered()
plugin_provider = get_provider(key)
if plugin_provider is None:
@@ -182,7 +150,6 @@ def _dispatch_to_plugin_provider(
return None
if plugin_provider is None:
return None
# ``is_available()`` MUST NOT raise per the ABC contract; defend anyway so
# a buggy plugin can't break dispatch for everyone.
try:
@@ -196,9 +163,7 @@ def _dispatch_to_plugin_provider(
logger.info("STT plugin provider '%s' reports not available; returning unavailability envelope.", key)
return _error_result(
f"STT plugin '{key}' is not available — check that its required credentials / dependencies are configured.",
provider=key,
)
provider=key)
logger.info("Transcribing with plugin STT provider '%s'...", key)
# The prompt travels via the ABC's ``**extra`` kwargs and is only sent when
# set, so pre-prompt providers see byte-identical calls on the no-prompt path.
@@ -208,7 +173,6 @@ def _dispatch_to_plugin_provider(
except Exception as exc: # noqa: BLE001
logger.warning("STT plugin provider '%s' raised: %s", key, exc, exc_info=True)
return _error_result(f"STT plugin '{key}' raised: {exc}", provider=key)
if not isinstance(result, dict):
return _error_result(f"STT plugin '{key}' returned a non-dict result", provider=key)
result.setdefault("provider", key)
@@ -229,16 +193,13 @@ _WHISPER_PROMPT_CAPPED_PROVIDERS = frozenset({"local", "openai", "groq", "deepin
def _enforce_prompt_length_limit(prompt: Optional[str], provider: str) -> Optional[str]:
"""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
if len(prompt) <= max_chars:
if not prompt or provider not in _WHISPER_PROMPT_CAPPED_PROVIDERS or len(prompt) <= max_chars:
return prompt
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:]
@@ -247,32 +208,24 @@ def _apply_pre_transcription_hook(
prompt: Optional[str], source: Optional[str],
) -> tuple[Optional[str], Optional[str], Optional[str]]:
"""Fire the ``pre_transcription`` plugin hook; returns ``(model, language_override, prompt)``.
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.
"""
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
if not has_hook("pre_transcription"):
return model, None, prompt
hook_results = invoke_hook(
"pre_transcription", file_path=file_path, provider=provider,
model=model, language=language, prompt=prompt, source=source,
)
model=model, language=language, prompt=prompt, source=source)
overrides: Dict[str, Any] = {}
for hook_result in hook_results:
if not isinstance(hook_result, dict):
continue
for key, value in hook_result.items():
for key, value in (hook_result.items() if isinstance(hook_result, dict) else ()):
if key == "file_path":
logger.warning(
"pre_transcription hook attempted to change "
"file_path (read-only) — ignoring the attempt."
)
"file_path (read-only) — ignoring the attempt.")
elif key not in _PRE_TRANSCRIPTION_MUTABLE_FIELDS:
logger.debug("pre_transcription hook returned unsupported field %r — ignoring.", key)
elif not isinstance(value, str):
@@ -281,13 +234,10 @@ def _apply_pre_transcription_hook(
)
else:
overrides[key] = value
if "model" in overrides:
model = overrides["model"]
# Hooks win over the static ``stt.prompt`` config; "" clears it.
if "prompt" in overrides:
# Hooks win over the static ``stt.prompt`` config; "" clears it.
prompt = overrides["prompt"] or None
return model, overrides.get("language") or None, prompt
return overrides.get("model", model), overrides.get("language") or None, prompt
except Exception as _hook_err: # noqa: BLE001 — hook plumbing is fail-open
logger.debug("pre_transcription hook error: %s", _hook_err)
return model, None, prompt
tools/transcription_common.py
+3 -6
View File
@@ -45,8 +45,7 @@ GROQ_MODELS = {"whisper-large-v3", "whisper-large-v3-turbo", "distil-whisper-lar
# (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",
})
"local", "local_command", "groq", "openai", "mistral", "xai", "elevenlabs", "deepinfra"})
# Built-in providers that upload audio to a remote API.
CLOUD_STT_PROVIDERS = frozenset(BUILTIN_STT_PROVIDERS - {"local", "local_command"})
@@ -62,10 +61,8 @@ def _ok_result(transcript: str, provider: str) -> Dict[str, Any]:
def _lazy_ensure_quietly(dep: str) -> None:
"""Best-effort ``tools.lazy_deps.ensure(dep, prompt=False)``; failures are swallowed.
prompt=False: a bare input() deadlocks under the interactive CLI where
prompt_toolkit owns stdin; installs are gated by ``security.allow_lazy_installs``.
"""
prompt=False: a bare input() deadlocks under the interactive CLI where prompt_toolkit owns
stdin; installs are gated by ``security.allow_lazy_installs``."""
try:
from tools.lazy_deps import ensure
ensure(dep, prompt=False)
tools/transcription_local.py
+32 -76
View File
@@ -23,8 +23,7 @@ from tools.transcription_audio import _run_quiet
from tools.transcription_common import (
DEFAULT_LOCAL_MODEL, DEFAULT_LOCAL_STT_LANGUAGE, GROQ_MODELS, LOCAL_STT_COMMAND_ENV,
OPENAI_MODELS, _config_number, _error_result, _log_prompt_unsupported, _ok_result,
_process_error_detail,
)
_process_error_detail)
# Log-record parity with the origin module.
logger = logging.getLogger("tools.transcription_tools")
@@ -36,12 +35,8 @@ def _get_local_command_template() -> Optional[str]:
if configured:
return configured
whisper_binary = _find_whisper_binary()
if whisper_binary:
return (
f"{shlex.quote(whisper_binary)} {{input_path}} --model {{model}} --output_format txt "
"--output_dir {output_dir} --language {language}"
)
return None
return (f"{shlex.quote(whisper_binary)} {{input_path}} --model {{model}} --output_format txt "
"--output_dir {output_dir} --language {language}") if whisper_binary else None
def _has_local_command() -> bool:
@@ -52,13 +47,12 @@ def _normalize_local_model(model_name: Optional[str]) -> str:
"""Return a valid faster-whisper size; cloud-only names (``whisper-1`` …) fall back to the default with a warning."""
if not model_name:
return DEFAULT_LOCAL_MODEL
if model_name in OPENAI_MODELS or model_name in GROQ_MODELS:
if model_name in OPENAI_MODELS | GROQ_MODELS:
logger.warning(
"STT model '%s' is a cloud-only name and cannot be used with the local "
"provider. Falling back to '%s'. Set stt.local.model to a valid "
"faster-whisper size (tiny, base, small, medium, large-v3).",
model_name, DEFAULT_LOCAL_MODEL,
)
model_name, DEFAULT_LOCAL_MODEL)
return DEFAULT_LOCAL_MODEL
return model_name
@@ -81,8 +75,7 @@ def _try_lazy_install_stt() -> bool:
"venv owner: `stat -c '%%u' '$(dirname $(dirname $(which python3)))'` "
"then `su - <owner> -c 'VIRTUAL_ENV=/opt/hermes/.venv "
"uv pip install faster-whisper==1.2.1'`",
exc,
)
exc)
return False
@@ -93,22 +86,19 @@ def _try_lazy_install_stt() -> bool:
_CUDA_LIB_ERROR_MARKERS = (
"libcublas", "libcudnn", "libcudart", "cannot be loaded", "cannot open shared object",
"no kernel image is available", "CUBLAS_STATUS_NOT_SUPPORTED", "no CUDA-capable device",
"CUDA driver version is insufficient",
)
"CUDA driver version is insufficient")
def _looks_like_cuda_lib_error(exc: BaseException) -> bool:
"""Heuristic: is this a missing/broken CUDA runtime library (not a legitimate runtime failure)?"""
msg = str(exc)
return any(marker in msg for marker in _CUDA_LIB_ERROR_MARKERS)
return any(marker in str(exc) for marker in _CUDA_LIB_ERROR_MARKERS)
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,
).strip()
return subprocess.check_output(["/usr/sbin/sysctl", "-n", name], stderr=subprocess.DEVNULL, text=True,
timeout=2).strip()
except Exception:
return ""
@@ -131,37 +121,27 @@ 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). Try the requested
config first; on a CUDA library load failure fall back to CPU + int8. Pass
``stt.local.device`` / ``compute_type`` to pin.
"""
"""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): 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:
# Importing ctranslate2 can itself abort on Apple Silicon/Rosetta when
# multiple Intel OpenMP runtimes are loaded — set before the import.
os.environ.setdefault("KMP_DUPLICATE_LIB_OK", "TRUE")
from faster_whisper import WhisperModel
if force_cpu:
logger.info(
"Apple Silicon/Rosetta detected — loading faster-whisper on CPU "
"(int8) to avoid native device autodetection crashes"
)
logger.info("Apple Silicon/Rosetta detected — loading faster-whisper on CPU "
"(int8) to avoid native device autodetection crashes")
return WhisperModel(model_name, device="cpu", compute_type="int8")
try:
return WhisperModel(model_name, device=device, compute_type=compute_type)
except Exception as exc:
if not _looks_like_cuda_lib_error(exc):
raise
logger.warning(
"faster-whisper CUDA load failed (%s) — falling back to CPU (int8). "
"Install the NVIDIA CUDA runtime (libcublas/libcudnn) to use GPU.",
exc,
)
logger.warning("faster-whisper CUDA load failed (%s) — falling back to CPU (int8). "
"Install the NVIDIA CUDA runtime (libcublas/libcudnn) to use GPU.", exc)
return WhisperModel(model_name, device="cpu", compute_type="int8")
@@ -180,56 +160,45 @@ def build_local_transcribe_kwargs(stt_config: Optional[Dict[str, Any]] = None) -
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 {}
# ``vad: null`` in YAML means "default on".
vad_enabled = local_cfg.get("vad", True)
kwargs: Dict[str, Any] = {
"beam_size": 5,
"condition_on_previous_text": False,
"vad_filter": vad_enabled is None or bool(vad_enabled),
}
"vad_filter": vad_enabled is None or bool(vad_enabled)}
if kwargs["vad_filter"]:
kwargs["vad_parameters"] = {
"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 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)
if forced_lang:
kwargs["language"] = forced_lang
initial_prompt = local_cfg.get("initial_prompt")
if isinstance(initial_prompt, str) and initial_prompt.strip():
kwargs["initial_prompt"] = initial_prompt
return kwargs
def _confidence_thresholds(local_cfg: Dict[str, Any]) -> tuple[float, float]:
"""Resolve (no_speech_prob, avg_logprob) gate thresholds from config."""
return (
_config_number(local_cfg, "no_speech_prob_threshold", _NO_SPEECH_PROB_THRESHOLD_DEFAULT),
_config_number(local_cfg, "logprob_threshold", _LOGPROB_THRESHOLD_DEFAULT),
)
return (_config_number(local_cfg, "no_speech_prob_threshold", _NO_SPEECH_PROB_THRESHOLD_DEFAULT),
_config_number(local_cfg, "logprob_threshold", _LOGPROB_THRESHOLD_DEFAULT))
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): non-speech AND low decode
confidence, so quiet-but-real speech survives. Unknown segment shapes are never dropped.
"""
"""True when a segment is very likely a silence hallucination. 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"))
avg_logprob = float(getattr(segment, "avg_logprob"))
return (float(segment.no_speech_prob) > no_speech_threshold
and float(segment.avg_logprob) < logprob_threshold)
except (AttributeError, TypeError, ValueError):
return False
return no_speech_prob > no_speech_threshold and avg_logprob < logprob_threshold
def _join_confident_segments(segments: Any, local_cfg: Dict[str, Any]) -> str:
@@ -238,11 +207,9 @@ def _join_confident_segments(segments: Any, local_cfg: Dict[str, Any]) -> str:
kept: list[str] = []
for segment in segments:
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, "avg_logprob", float("nan")),
)
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, "avg_logprob", float("nan")))
continue
kept.append(segment.text.strip())
return " ".join(kept).strip()
@@ -255,41 +222,30 @@ def _transcribe_local_command(
from tools.transcription_tools import _prepare_local_audio, _resolve_stt_language
if prompt:
_log_prompt_unsupported("STT provider 'local_command'")
command_template = _get_local_command_template()
if not command_template:
return _error_result(f"{LOCAL_STT_COMMAND_ENV} not configured and no local whisper binary was found")
# Language: hook override > stt.local.language > stt.language > env > "en".
language = language or _resolve_stt_language("local") or DEFAULT_LOCAL_STT_LANGUAGE
normalized_model = _normalize_local_model(model_name)
try:
with tempfile.TemporaryDirectory(prefix="hermes-local-stt-") as output_dir:
prepared_input, prep_error = _prepare_local_audio(file_path, output_dir)
if prep_error:
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),
)
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
_run_quiet(shlex.split(command), timeout=300, env=hermes_subprocess_env(inherit_credentials=False))
txt_files = sorted(Path(output_dir).glob("*.txt"))
if not txt_files:
return _error_result("Local STT command completed but did not produce a .txt transcript")
transcript_text = txt_files[0].read_text(encoding="utf-8").strip()
logger.info(
"Transcribed %s via local STT command (%s, %d chars)",
Path(file_path).name, normalized_model, len(transcript_text),
)
logger.info("Transcribed %s via local STT command (%s, %d chars)",
Path(file_path).name, normalized_model, len(transcript_text))
return _ok_result(transcript_text, "local_command")
except KeyError as e:
return _error_result(f"Invalid {LOCAL_STT_COMMAND_ENV} template, missing placeholder: {e}")
except subprocess.CalledProcessError as e:
tools/transcription_tools.py
+58 -114
View File
@@ -25,42 +25,36 @@ from hermes_cli._subprocess_compat import windows_hide_flags # noqa: F401 (imp
from utils import is_truthy_value
from tools.managed_tool_gateway import resolve_managed_tool_gateway # noqa: F401 (patched by tests)
from tools.tool_backend_helpers import ( # noqa: F401 (patched by tests; read lazily by transcription_cloud)
managed_nous_tools_enabled, nous_tool_gateway_unavailable_message, resolve_openai_audio_api_key,
)
managed_nous_tools_enabled, nous_tool_gateway_unavailable_message, resolve_openai_audio_api_key)
from tools.transcription_common import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
BUILTIN_STT_PROVIDERS, CLOUD_STT_PROVIDERS, DEFAULT_ELEVENLABS_STT_MODEL,
DEFAULT_GROQ_STT_MODEL, DEFAULT_LOCAL_MODEL, DEFAULT_MISTRAL_STT_MODEL, DEFAULT_PROVIDER,
DEFAULT_STT_MODEL, ELEVENLABS_STT_BASE_URL, GROQ_MODELS, LOCAL_STT_COMMAND_ENV,
LOCAL_STT_LANGUAGE_ENV, MAX_FILE_SIZE, OPENAI_MODELS, SUPPORTED_FORMATS, XAI_STT_BASE_URL,
_error_result, _get_stt_section, _ok_result,
)
_error_result, _get_stt_section, _ok_result)
from tools.transcription_audio import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
_CLOUD_TRIM_KEEP_MS_DEFAULT, _CLOUD_TRIM_MIN_INPUT_SECONDS, _CLOUD_TRIM_THRESHOLD_DB_DEFAULT,
_cloud_trim_settings, _convert_caf_to_wav, _find_ffmpeg_binary, _find_ffprobe_binary,
_find_whisper_binary, _prepare_audio_for_transcription, _prepare_local_audio,
_probe_audio_duration, _run_ffmpeg_stt_encode, _trim_silence_for_cloud_stt,
_validate_audio_file, _validate_audio_file_size, _validate_audio_source_file,
)
_validate_audio_file, _validate_audio_file_size, _validate_audio_source_file)
from tools.transcription_local import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
_LOGPROB_THRESHOLD_DEFAULT, _NO_SPEECH_PROB_THRESHOLD_DEFAULT, _get_idle_unload_seconds,
_get_local_command_template, _has_local_command, _is_hallucinated_segment,
_join_confident_segments, _load_local_whisper_model, _looks_like_cuda_lib_error,
_normalize_local_model, _transcribe_local_command, _try_lazy_install_stt,
build_local_transcribe_kwargs,
)
build_local_transcribe_kwargs)
from tools.transcription_cloud import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
_extract_transcript_text, _has_xai_stt_credentials, _is_local_or_private_url,
_resolve_openai_audio_client_config, _transcribe_deepinfra, _transcribe_elevenlabs,
_transcribe_groq, _transcribe_mistral, _transcribe_openai, _transcribe_xai,
)
_transcribe_groq, _transcribe_mistral, _transcribe_openai, _transcribe_xai)
from tools.transcription_command import ( # noqa: F401 (re-exported; tests patch tools.transcription_tools.<name>)
COMMAND_STT_OUTPUT_FORMATS, DEFAULT_COMMAND_STT_LANGUAGE, DEFAULT_COMMAND_STT_OUTPUT_FORMAT,
DEFAULT_COMMAND_STT_TIMEOUT_SECONDS, _PROMPT_CHARS_PER_TOKEN, _WHISPER_PROMPT_TOKEN_CAP,
_apply_pre_transcription_hook, _dispatch_to_plugin_provider, _enforce_prompt_length_limit,
_get_command_stt_output_format, _get_command_stt_timeout, _get_named_stt_provider_config,
_render_command_stt_template, _resolve_command_stt_provider_config, _run_command_stt,
_transcribe_command_stt, _unregistered_stt_provider_error,
)
_transcribe_command_stt, _unregistered_stt_provider_error)
logger = logging.getLogger(__name__)
@@ -91,10 +85,8 @@ def _safe_find_spec(module_name: str) -> bool:
return module_name in globals() or module_name in os.sys.modules
_HAS_FASTER_WHISPER = _safe_find_spec("faster_whisper")
_HAS_OPENAI = _safe_find_spec("openai")
_HAS_MISTRAL = _safe_find_spec("mistralai")
_HAS_PILK = _safe_find_spec("pilk")
_HAS_FASTER_WHISPER, _HAS_OPENAI, _HAS_MISTRAL, _HAS_PILK = map(
_safe_find_spec, ("faster_whisper", "openai", "mistralai", "pilk"))
# Local model singleton; the lock guards check-then-load against concurrent voice messages.
_local_model: Optional[object] = None
@@ -112,7 +104,6 @@ _IDLE_UNLOAD_CHECK_INTERVAL = 30 # seconds between idle checks
# ---- Config helpers -----------------------------------------------------
def _load_stt_config() -> dict:
"""Load the ``stt`` section from user config, falling back to defaults."""
try:
@@ -123,19 +114,16 @@ def _load_stt_config() -> dict:
def is_stt_enabled(stt_config: Optional[dict] = None) -> bool:
if stt_config is None:
stt_config = _load_stt_config()
return is_truthy_value(stt_config.get("enabled", True), default=True)
cfg = _load_stt_config() if stt_config is None else stt_config
return is_truthy_value(cfg.get("enabled", True), default=True)
def _resolve_stt_language(
provider_key: str, stt_config: Optional[Dict[str, Any]] = None, *, extra_keys: tuple = ()
) -> Optional[str]:
"""Language hint for an STT provider, first non-empty wins (never "").
``stt.<provider>.language`` (plus *extra_keys* aliases, e.g. ``language_code``) >
``stt.language`` > ``HERMES_LOCAL_STT_LANGUAGE`` env > None (provider auto-detects).
"""
"""Language hint for an STT provider, first non-empty wins (never ""): ``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()
provider_cfg = _get_stt_section(stt_config, provider_key)
@@ -143,10 +131,7 @@ def _resolve_stt_language(
if isinstance(stt_config, dict):
candidates.append(stt_config.get("language"))
candidates.append(os.getenv(LOCAL_STT_LANGUAGE_ENV))
for candidate in candidates:
if isinstance(candidate, str) and candidate.strip():
return candidate.strip()
return None
return next((c.strip() for c in candidates if isinstance(c, str) and c.strip()), None)
def _openai_audio_unavailable_reason() -> Optional[str]:
@@ -168,13 +153,11 @@ 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))
sdk_ok = (not needs_openai or _HAS_OPENAI) and (not needs_mistral or _HAS_MISTRAL)
return sdk_ok and bool(_resolve_provider_key(env_var, provider))
return probe
@@ -201,18 +184,15 @@ def _detect_local_backend() -> Optional[str]:
"""faster-whisper > local whisper CLI > lazy-installed faster-whisper; None when nothing local works."""
if _HAS_FASTER_WHISPER:
return "local"
if _has_local_command():
return "local_command"
return "local" if _try_lazy_install_stt() else None
return "local_command" if _has_local_command() else ("local" if _try_lazy_install_stt() else None)
def _resolve_explicit_local() -> str:
backend = _detect_local_backend()
if backend:
return backend
logger.warning("STT provider 'local' configured but unavailable "
"(install faster-whisper or set HERMES_LOCAL_STT_COMMAND)")
return "none"
if not backend:
logger.warning("STT provider 'local' configured but unavailable "
"(install faster-whisper or set HERMES_LOCAL_STT_COMMAND)")
return backend or "none"
def _resolve_explicit_local_command() -> str:
@@ -255,15 +235,13 @@ _CLOUD_PROVIDER_SPECS = {
"No local STT available, using ElevenLabs Scribe STT API"),
"deepinfra": (_has_deepinfra_key, _has_deepinfra_key,
"STT provider 'deepinfra' configured but DEEPINFRA_API_KEY not set (or openai package missing)",
"No local STT available, using DeepInfra Whisper API"),
}
"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,
}
"openai": _resolve_explicit_openai}
def _resolve_explicit_provider(provider: str) -> str:
@@ -273,12 +251,9 @@ def _resolve_explicit_provider(provider: str) -> str:
if resolver is not None:
return resolver()
spec = _CLOUD_PROVIDER_SPECS.get(provider)
if spec is None:
if spec is None or spec[0]():
return provider
probe, _auto, warning, _log = spec
if probe():
return provider
logger.warning(warning)
logger.warning(spec[2])
return "none"
@@ -314,7 +289,6 @@ def _get_provider(stt_config: dict) -> str:
# ---- Provider: local (faster-whisper) -----------------------------------
def _unload_local_model() -> None:
"""Release the cached local whisper model. Thread-safe via the model lock."""
global _local_model, _local_model_name
@@ -326,22 +300,17 @@ def _unload_local_model() -> None:
def _start_idle_unload_watcher(timeout_seconds: int) -> None:
"""Ensure the single idle-unload watcher thread is running.
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.
"""
"""Ensure the single idle-unload watcher thread is running. 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. Exits after unloading, when the timeout becomes 0, or when the model is gone."""
global _idle_unload_thread
with _idle_unload_mgmt_lock:
if _idle_unload_thread is not None and _idle_unload_thread.is_alive():
return
def _watch(initial_timeout=timeout_seconds):
while not _idle_unload_stop.is_set():
if _idle_unload_stop.wait(_IDLE_UNLOAD_CHECK_INTERVAL) or _local_model is None:
break
while not _idle_unload_stop.wait(_IDLE_UNLOAD_CHECK_INTERVAL) and _local_model is not None:
try:
timeout = _get_idle_unload_seconds(_load_stt_config().get("local") or {})
except Exception: # noqa: BLE001 - keep the seed value
@@ -351,7 +320,6 @@ def _start_idle_unload_watcher(timeout_seconds: int) -> None:
if time.monotonic() - _last_transcription_time >= timeout:
_unload_local_model()
break
_idle_unload_stop.clear()
_idle_unload_thread = threading.Thread(target=_watch, name="hermes-stt-idle-unload", daemon=True)
_idle_unload_thread.start()
@@ -364,10 +332,8 @@ def _touch_transcription_time() -> None:
def _get_or_load_local_model(model_name: str, local_cfg: Dict[str, Any]):
"""Cached faster-whisper model, (re)loaded under a double-checked lock when needed.
The returned strong reference stays valid even if the idle watcher nulls the global mid-transcription.
"""
"""Cached faster-whisper model, (re)loaded under a double-checked lock when needed. 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
if model is None or _local_model_name != model_name:
@@ -375,10 +341,8 @@ def _get_or_load_local_model(model_name: str, local_cfg: Dict[str, Any]):
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 pin a configuration where ``auto`` mis-detects.
_local_model = _load_local_whisper_model(
model_name, device=local_cfg.get("device", "auto"),
compute_type=local_cfg.get("compute_type", "auto"),
)
_local_model = _load_local_whisper_model(model_name, device=local_cfg.get("device", "auto"),
compute_type=local_cfg.get("compute_type", "auto"))
_local_model_name = model_name
model = _local_model
return model
@@ -390,8 +354,7 @@ def _replace_cached_model_on_cpu(model_name: str):
from faster_whisper import WhisperModel
model = WhisperModel(model_name, device="cpu", compute_type="int8")
with _local_model_lock:
_local_model = model
_local_model_name = model_name
_local_model, _local_model_name = model, model_name
return model
@@ -409,12 +372,10 @@ def _transcribe_local(
model = _get_or_load_local_model(model_name, local_cfg)
if model is None: # defensive: load failed without raising
return _error_result("Local whisper model failed to load")
transcribe_kwargs = build_local_transcribe_kwargs(stt_config)
# pre_transcription hook overrides win over config-resolved values.
if language:
transcribe_kwargs["language"] = language
if prompt:
transcribe_kwargs["initial_prompt"] = prompt
transcribe_kwargs = build_local_transcribe_kwargs(stt_config)
transcribe_kwargs.update({k: v for k, v in (("language", language), ("initial_prompt", prompt))
if v})
try:
segments, info = model.transcribe(file_path, **transcribe_kwargs)
except Exception as exc:
@@ -440,7 +401,6 @@ def _transcribe_local(
# ---- Public API ---------------------------------------------------------
def _read_block_error(file_path: str) -> Optional[Dict[str, Any]]:
"""Refuse to ship a credential store (auth.json, .env, OAuth tokens) to an STT provider.
Mirrors the image-gen / video-gen read guards."""
@@ -450,13 +410,10 @@ 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
) -> Dict[str, Any]:
"""Transcribe a validated audio file with the configured STT provider.
``model`` overrides the config default; ``source`` is a caller-surface label
(``"gateway"``, ``"voice_mode"``) forwarded to the ``pre_transcription`` hook only.
"""
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 default; ``source`` is a caller-surface label (``"gateway"``, ``"voice_mode"``) forwarded
to the ``pre_transcription`` hook only."""
# 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 = _read_block_error(file_path) or _validate_audio_file(file_path, enforce_size_limit=False)
@@ -470,11 +427,11 @@ def _transcribe_prepared_audio(
error = _validate_audio_file_size(Path(file_path))
if error:
return error
# Convert CAF (iMessage voice notes) to WAV for cloud STT providers.
if Path(file_path).suffix.lower() == ".caf" and provider not in ("local", "local_command"):
file_path = _convert_caf_to_wav(file_path)
if not file_path:
return _error_result("CAF audio could not be converted to WAV.")
# Convert CAF (iMessage voice notes) to WAV for cloud STT providers.
if Path(file_path).suffix.lower() == ".caf":
file_path = _convert_caf_to_wav(file_path)
if not file_path:
return _error_result("CAF audio could not be converted to WAV.")
# Best-effort pre-upload silence trim for built-in cloud providers.
trim_cleanup_dir: Optional[str] = None
if provider in CLOUD_STT_PROVIDERS:
@@ -498,8 +455,7 @@ _BUILTIN_MODEL_KEYS = {
"openai": ("openai", "model", DEFAULT_STT_MODEL, False),
"mistral": ("mistral", "model", DEFAULT_MISTRAL_STT_MODEL, False),
"elevenlabs": ("elevenlabs", "model_id", DEFAULT_ELEVENLABS_STT_MODEL, False),
"deepinfra": ("deepinfra", "model", "", True),
}
"deepinfra": ("deepinfra", "model", "", True)}
def _builtin_model_name(provider: str, stt_config: Dict[str, Any], model: Optional[str]) -> str:
@@ -515,13 +471,11 @@ def _builtin_model_name(provider: str, stt_config: Dict[str, Any], model: Option
def _dispatch_stt_provider(
file_path: str, provider: str, stt_config: Dict[str, Any], model: Optional[str] = None,
source: Optional[str] = None,
) -> Dict[str, Any]:
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; 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
prompt = prompt if isinstance(prompt, str) and prompt.strip() else None
# 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,
@@ -544,8 +498,7 @@ def _dispatch_stt_provider(
# Plugin backend: reads ``stt.<provider>`` like built-ins; the ``model`` argument overrides it.
plugin_result = _dispatch_to_plugin_provider(
file_path, provider, stt_config, model=model or _get_stt_section(stt_config, provider).get("model"),
language=language or _resolve_stt_language(provider, stt_config), prompt=prompt,
)
language=language or _resolve_stt_language(provider, stt_config), prompt=prompt)
return plugin_result if plugin_result is not None else _no_provider_error(provider, stt_config)
@@ -565,18 +518,13 @@ def _no_provider_error(provider: str, stt_config: Dict[str, Any]) -> Dict[str, A
"set GROQ_API_KEY for free Groq Whisper, set MISTRAL_API_KEY for Mistral "
"Voxtral Transcribe, configure xAI OAuth or set XAI_API_KEY for xAI Grok STT, "
"set ELEVENLABS_API_KEY for ElevenLabs Scribe, or set VOICE_TOOLS_OPENAI_KEY "
"or OPENAI_API_KEY for the OpenAI Whisper API."
)
"or OPENAI_API_KEY for the OpenAI Whisper API.")
def transcribe_audio(
file_path: str, model: Optional[str] = None, source: Optional[str] = None
) -> Dict[str, Any]:
"""Validate, preprocess supported inputs, and dispatch transcription.
``source`` is a caller-surface label (``"gateway"``, ``"voice_mode"``) forwarded to
the ``pre_transcription`` hook for observability only.
"""
file_path: str, model: Optional[str] = None, source: Optional[str] = None) -> Dict[str, Any]:
"""Validate, preprocess supported inputs, and dispatch transcription. ``source`` is a caller-surface
label (``"gateway"``, ``"voice_mode"``) forwarded to the ``pre_transcription`` hook only."""
# Secret-store refusal runs before ANY validation so the error names the real reason.
blocked = _read_block_error(file_path)
if blocked:
@@ -599,16 +547,12 @@ def transcribe_audio(
def transcribe_audio_local_fallback(file_path: str, model: Optional[str] = None) -> Dict[str, Any]:
"""Try an already-installed local STT backend without changing config.
Passive inbound-media recovery after the configured provider failed: never
lazy-installs or falls through to a cloud provider.
"""
"""Try an already-installed local STT backend without changing config: passive inbound-media
recovery after the configured provider failed — never lazy-installs or falls through to cloud."""
error = _validate_audio_file(file_path)
if error:
return error
local_cfg = _load_stt_config().get("local") or {}
local_model = model or local_cfg.get("model", DEFAULT_LOCAL_MODEL)
local_model = model or (_load_stt_config().get("local") or {}).get("model", DEFAULT_LOCAL_MODEL)
if _HAS_FASTER_WHISPER:
return _transcribe_local(file_path, _normalize_local_model(local_model))
if _has_local_command():
tools/tts_command_provider.py
+60 -130
View File
@@ -1,20 +1,12 @@
"""Shared runner for user-configured shell ("command") TTS/STT providers.
``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 -f {output_path} < {input_path}", output_format: wav}
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``.
``tts.providers.<name>: {type: command, command: "piper -f {output_path} < {input_path}"}``
(and the ``stt.`` twin): ``{placeholders}`` are shell-quoted for their surrounding quote
context, ``{{``/``}}`` stay literal. Owns the quote-aware rendering, the idle-timeout
process runner and the generic ``<section>.providers.<name>`` readers, re-imported by
``tts_tool``/``transcription_tools`` under their historical private names. TTS placeholders:
``{input_path}``/``{text_path}``, ``{output_path}``, ``{format}``, ``{voice}``, ``{model}``,
``{speed}``. Built-in provider names always win over a same-named ``providers`` entry.
"""
from __future__ import annotations
@@ -27,9 +19,12 @@ import subprocess
import tempfile
import threading
import time
from functools import partial
from pathlib import Path
from typing import Any, Dict, FrozenSet, Optional
from utils import is_truthy_value
def shell_quote_context(command_template: str, position: int) -> Optional[str]:
"""Return the shell quote char (``'``/``"``) active right before *position*, or None."""
@@ -48,10 +43,8 @@ def shell_quote_context(command_template: str, position: int) -> Optional[str]:
escaped = True
elif char == '"':
quote = None
elif char == "'":
quote = "'"
elif char == '"':
quote = '"'
elif char in ("'", '"'):
quote = char
elif char == "\\":
i += 1
i += 1
@@ -64,9 +57,7 @@ def quote_command_placeholder(value: str, quote_context: Optional[str]) -> str:
return value.replace("'", r"'\''")
if quote_context == '"':
return value.replace("\\", "\\\\").replace('"', r'\"').replace("$", r"\$").replace("`", r"\`")
if os.name == "nt":
return subprocess.list2cmdline([value])
return shlex.quote(value)
return subprocess.list2cmdline([value]) if os.name == "nt" else shlex.quote(value)
def render_command_template(command_template: str, placeholders: Dict[str, str]) -> str:
@@ -110,35 +101,29 @@ def terminate_command_process_tree(proc: subprocess.Popen) -> 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,
)
subprocess.run(["taskkill", "/F", "/T", "/PID", str(proc.pid)], stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, timeout=5, stdin=subprocess.DEVNULL)
except Exception:
proc.kill()
return
try:
import psutil # type: ignore
except ImportError:
proc.terminate()
try:
proc.wait(timeout=2)
except subprocess.TimeoutExpired:
proc.kill()
return
_signal_process_tree(psutil, proc, "terminate")
psutil = None
# Without psutil only the shell itself is signalled (children may survive).
signal = ((lambda m: getattr(proc, m)()) if psutil is None
else (lambda m: _signal_process_tree(psutil, proc, m)))
signal("terminate")
try:
proc.wait(timeout=2)
except subprocess.TimeoutExpired:
_signal_process_tree(psutil, proc, "kill")
signal("kill")
def command_env_passthrough(config: Dict[str, Any]) -> list:
"""``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()]
return [str(x).strip() for x in raw if str(x).strip()] if isinstance(raw, (list, tuple)) else []
def command_failure_detail(exc: subprocess.CalledProcessError) -> str:
@@ -151,30 +136,23 @@ def run_command_provider(
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
scrubbed = hermes_subprocess_env(inherit_credentials=False)
for key in env_passthrough or []:
value = os.environ.get(key)
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.
"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)
else:
popen_kwargs["start_new_session"] = True
proc = subprocess.Popen(command, **popen_kwargs, stdin=subprocess.DEVNULL)
# Own process group so the whole tree can be signalled on idle timeout. Lossy UTF-8 decode:
# locale-mismatched bytes must not raise in the reader threads.
group = ({"creationflags": getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)} if os.name == "nt"
else {"start_new_session": True})
proc = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
encoding="utf-8", errors="replace", env=delegated_child_subprocess_env(scrubbed),
stdin=subprocess.DEVNULL, **group)
output_queue: "queue.Queue[tuple[str, Optional[str]]]" = queue.Queue()
chunks: Dict[str, list[str]] = {"stdout": [], "stderr": []}
open_streams = {"stdout", "stderr"}
@@ -191,13 +169,10 @@ def run_command_provider(
finally:
output_queue.put((name, None))
readers = [
threading.Thread(target=read_stream, args=(name, stream), daemon=True)
for name, stream in (("stdout", proc.stdout), ("stderr", proc.stderr))
]
readers = [threading.Thread(target=read_stream, args=(name, stream), daemon=True)
for name, stream in (("stdout", proc.stdout), ("stderr", proc.stderr))]
for reader in readers:
reader.start()
deadline = time.monotonic() + timeout
timed_out = False
while open_streams:
@@ -214,61 +189,42 @@ def run_command_provider(
continue
chunks[name].append(chunk)
deadline = time.monotonic() + timeout
if not timed_out:
try:
proc.wait(timeout=max(0.0, deadline - time.monotonic()))
except subprocess.TimeoutExpired:
timed_out = True
if timed_out:
terminate_command_process_tree(proc)
for reader in readers:
reader.join(timeout=0.5)
while True:
try:
name, chunk = output_queue.get_nowait()
except queue.Empty:
break
while not output_queue.empty():
name, chunk = output_queue.get_nowait()
if chunk:
chunks[name].append(chunk)
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, stderr = "".join(chunks["stdout"]), "".join(chunks["stderr"])
if timed_out:
raise subprocess.TimeoutExpired(command, timeout, output=stdout, stderr=stderr) from (
subprocess.TimeoutExpired(command, timeout))
if proc.returncode:
raise subprocess.CalledProcessError(proc.returncode, command, output=stdout, stderr=stderr)
return subprocess.CompletedProcess(command, proc.returncode, stdout, stderr)
# ===========================================================================
# Generic ``<section>.providers.<name>`` config layer (TTS and STT share it)
# ===========================================================================
# ---- Generic ``<section>.providers.<name>`` config layer (TTS and STT share it) ----
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 = config.get(name)
section = config.get(name) if isinstance(config, dict) else None
return section if isinstance(section, dict) else {}
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.
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>.providers.<name>`` (canonical), else ``<section>.<name>`` for non-built-in names
only — refused for built-ins so a user's ``openai:`` block still means OpenAI, not a command."""
section = _get_provider_section(config, "providers").get(name)
if isinstance(section, dict):
return section
if name.lower() not in builtins:
return _get_provider_section(config, name)
return {}
return _get_provider_section(config, name) if name.lower() not in builtins else {}
def _is_command_provider_config(config: Dict[str, Any]) -> bool:
@@ -276,20 +232,15 @@ def _is_command_provider_config(config: Dict[str, Any]) -> bool:
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())
return ptype in ("", "command") and isinstance(command, str) and bool(command.strip())
def _resolve_command_config(
provider: str, config: Dict[str, Any], reserved: FrozenSet[str],
) -> Optional[Dict[str, Any]]:
provider: str, config: Dict[str, Any], reserved: FrozenSet[str]) -> Optional[Dict[str, Any]]:
"""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 reserved:
key = (provider or "").lower().strip()
if not key or key in reserved:
return None
named = _named_provider_config(config, key, reserved)
return named if _is_command_provider_config(named) else None
@@ -317,8 +268,7 @@ def _command_output_format(config: Dict[str, Any], formats: FrozenSet[str], defa
# 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",
})
"neutts", "kittentts", "piper", "deepinfra"})
DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS = 120
DEFAULT_COMMAND_TTS_OUTPUT_FORMAT = "mp3"
@@ -326,12 +276,9 @@ COMMAND_TTS_OUTPUT_FORMATS = frozenset({"mp3", "wav", "ogg", "flac", "m4a", "aac
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)
_get_named_provider_config = partial(_named_provider_config, builtins=BUILTIN_TTS_PROVIDERS)
_resolve_command_provider_config = partial(_resolve_command_config, reserved=BUILTIN_TTS_PROVIDERS)
_get_command_tts_timeout = partial(_command_timeout, default=DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS)
def _iter_command_providers(tts_config: Dict[str, Any]):
@@ -341,25 +288,17 @@ def _iter_command_providers(tts_config: Dict[str, Any]):
yield name, cfg
def _get_command_tts_timeout(config: Dict[str, Any]) -> float:
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:
"""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
suffix = Path(output_path).suffix.lower().strip().lstrip(".") if output_path else ""
if suffix in COMMAND_TTS_OUTPUT_FORMATS:
return suffix
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:
"""True only when the user explicitly opted in to voice delivery."""
value = config.get("voice_compatible", False)
if isinstance(value, str):
return value.strip().lower() in {"1", "true", "yes", "on"}
return bool(value)
return is_truthy_value(config.get("voice_compatible", False))
def _configured_command_tts_output_path(path: Path, config: Dict[str, Any]) -> Path:
@@ -371,36 +310,28 @@ def _generate_command_tts(
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; returns the audio path it wrote.
Raises ``ValueError`` for invalid provider config and ``RuntimeError`` for
timeouts / non-zero exits / empty output.
"""
Raises ``ValueError`` for bad provider config, ``RuntimeError`` for timeouts / bad exits / no output."""
command_template = str(config.get("command") or "").strip()
if not command_template:
raise ValueError(f"tts.providers.{provider_name}.command is not configured")
output = Path(output_path).expanduser()
output.parent.mkdir(parents=True, exist_ok=True)
if output.exists():
output.unlink()
timeout = _get_command_tts_timeout(config)
output_format = _get_command_tts_output_format(config, str(output))
speed = config.get("speed", tts_config.get("speed", ""))
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),
"format": _get_command_tts_output_format(config, str(output)),
"voice": str(config.get("voice", "")), "model": str(config.get("model", "")),
"speed": str(config.get("speed", tts_config.get("speed", ""))),
}
command = render_command_template(command_template, placeholders)
try:
# 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))
except subprocess.TimeoutExpired as exc:
raise RuntimeError(f"TTS provider '{provider_name}' timed out after {timeout:g}s") from exc
@@ -408,7 +339,6 @@ def _generate_command_tts(
raise RuntimeError(
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}")
return str(output)
tools/tts_streaming.py
+38 -86
View File
@@ -27,12 +27,9 @@ _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); seam over ``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
return _resolve_provider_key(env_var, provider_id) or ""
except Exception:
return get_env_value(env_var) or ""
@@ -42,10 +39,9 @@ def _gemini_key() -> str:
return _resolve_key("GEMINI_API_KEY", "gemini") or _resolve_key("GOOGLE_API_KEY", "gemini")
# 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.
# Interruption latch: a barge-in on a spoken reply 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.]"
_INTERRUPT_TTL_S = 120.0
_interrupted_at: Optional[float] = None
@@ -68,12 +64,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.
"""
"""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."""
def __init__(self, min_len: int = 20):
self.min_len = min_len
@@ -98,13 +91,12 @@ class SentenceChunker:
def flush(self) -> List[str]:
"""Drain the tail (end-of-text or long-idle flush)."""
tail = _THINK_BLOCK_RE.sub("", self.buf).strip()
self.buf = ""
tail, self.buf = _THINK_BLOCK_RE.sub("", self.buf).strip(), ""
return [tail] if tail else []
class StreamingTTSProvider(ABC):
"""Yields raw int16, little-endian, mono PCM chunks at ``sample_rate``."""
"""Yields raw int16, little-endian, mono PCM chunks at ``sample_rate`` (built-ins: 24 kHz)."""
sample_rate: int = 24000
channels: int = 1
@@ -152,23 +144,17 @@ _PROVIDER_PRIORITY: List[str] = ["elevenlabs", "gemini", "openai", "xai"]
def resolve_streaming_provider(
tts_config: Dict, preferred: Optional[str] = None
) -> Optional[StreamingTTSProvider]:
tts_config: Dict, preferred: Optional[str] = None) -> Optional[StreamingTTSProvider]:
"""Return a ready streamer for the *configured* provider, else ``None``.
``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.
"""
``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."""
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
return next((inst for name in _PROVIDER_PRIORITY
if (inst := _try_instantiate(name, tts_config))), None)
return _try_instantiate(pinned or (preferred or _get_provider(tts_config)).lower().strip(), tts_config)
@@ -187,8 +173,6 @@ def _capped(chunks: Iterator[bytes], label: str) -> Iterator[bytes]:
class ElevenLabsStreamer(StreamingTTSProvider):
"""ElevenLabs chunked HTTP → pcm_24000 (the original reference path)."""
sample_rate = 24000
@staticmethod
def available() -> bool:
return bool(_resolve_key("ELEVENLABS_API_KEY", "elevenlabs"))
@@ -198,45 +182,37 @@ class ElevenLabsStreamer(StreamingTTSProvider):
from tools.tts_tool_providers import (
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),
)
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),
)
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=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)),
output_format="pcm_24000")
def _openai_config_api_key() -> str:
"""Return ``tts.openai.api_key`` from config.yaml, or empty string."""
try:
openai_cfg = (_load_tts_config().get("openai") or {})
return (_load_tts_config().get("openai") or {}).get("api_key") or ""
except Exception:
return ""
return openai_cfg.get("api_key") or ""
@register("openai")
class OpenAIStreamer(StreamingTTSProvider):
"""OpenAI speech with ``response_format=pcm`` (24 kHz mono int16)."""
sample_rate = 24000
@staticmethod
def available() -> bool:
return bool(_openai_config_api_key() or resolve_openai_audio_api_key())
def stream(self, text: str) -> Iterator[bytes]:
from openai import OpenAI
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))
with client.audio.speech.with_streaming_response.create(
model=self.section.get("model", "gpt-4o-mini-tts"), voice=self.section.get("voice", "alloy"),
input=text, response_format="pcm",
@@ -248,8 +224,6 @@ class OpenAIStreamer(StreamingTTSProvider):
class GeminiStreamer(StreamingTTSProvider):
"""Gemini ``streamGenerateContent?alt=sse`` → SSE feed of base64 PCM chunks (24 kHz), bounded streamed body."""
sample_rate = 24000
@staticmethod
def available() -> bool:
return bool(_gemini_key())
@@ -257,13 +231,9 @@ class GeminiStreamer(StreamingTTSProvider):
def stream(self, text: str) -> Iterator[bytes]:
import base64
import json as _json
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
@@ -274,9 +244,7 @@ class GeminiStreamer(StreamingTTSProvider):
"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]:
@@ -288,13 +256,11 @@ class GeminiStreamer(StreamingTTSProvider):
if not line or not line.startswith("data: "):
continue
try:
event = _json.loads(line[len("data: "):])
parts = event["candidates"][0]["content"]["parts"]
parts = _json.loads(line[len("data: "):])["candidates"][0]["content"]["parts"]
except (ValueError, KeyError, IndexError, TypeError):
continue
for part in parts:
inline = part.get("inlineData") or part.get("inline_data") or {}
b64 = inline.get("data", "")
b64 = (part.get("inlineData") or part.get("inline_data") or {}).get("data", "")
if not b64:
continue
try:
@@ -308,21 +274,15 @@ class GeminiStreamer(StreamingTTSProvider):
@register("xai")
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. ``_collect_async`` bridges the async WS loop to the sync
iterator contract — the seam unit tests patch.
"""
sample_rate = 24000
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 (test
seam)."""
@staticmethod
def available() -> bool:
try:
from tools.xai_http import resolve_xai_http_credentials
creds = resolve_xai_http_credentials()
return bool(str(creds.get("api_key") or "").strip())
return bool(str(resolve_xai_http_credentials().get("api_key") or "").strip())
except Exception:
return False
@@ -332,26 +292,20 @@ class XAIStreamer(StreamingTTSProvider):
def _collect_async(self, text: str) -> List[bytes]:
import asyncio
return asyncio.run(self._drain_async(text))
async def _drain_async(self, text: str) -> List[bytes]:
return [frame async for frame in self._async_frames(text)]
async def _drain() -> List[bytes]:
return [frame async for frame in self._async_frames(text)]
return asyncio.run(_drain())
async def _async_frames(self, text: str):
import json as _json
import websockets
from tools.tts_tool_providers import DEFAULT_XAI_VOICE_ID
from tools.xai_http import resolve_xai_http_credentials
creds = resolve_xai_http_credentials()
api_key = str(creds.get("api_key") or "").strip()
api_key = str(resolve_xai_http_credentials().get("api_key") or "").strip()
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()
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:
@@ -367,15 +321,13 @@ class XAIStreamer(StreamingTTSProvider):
return
continue
etype = envelope.get("type")
if etype == "done":
return
if etype == "error":
logger.warning(
"xAI WS error envelope: %s", envelope.get("error") or envelope.get("message") or envelope,
)
if etype in ("done", "error"):
return
except Exception as exc:
if exc.__class__.__name__ == "ConnectionClosed":
return
logger.warning("xAI WS receive failed: %s", exc)
if exc.__class__.__name__ != "ConnectionClosed":
logger.warning("xAI WS receive failed: %s", exc)
return
tools/tts_text_normalize.py
+29 -86
View File
@@ -31,11 +31,11 @@ _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+")
_DEGREE_UNITS = (("C", "Celsius"), ("F", "Fahrenheit"))
# 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"),
)
(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),
@@ -47,8 +47,7 @@ _EMOJI_RE = re.compile(
"[\U0001F1E6-\U0001F1FF\U0001F300-\U0001F5FF\U0001F600-\U0001F64F\U0001F680-\U0001F6FF"
"\U0001F700-\U0001F77F\U0001F780-\U0001F7FF\U0001F800-\U0001F8FF\U0001F900-\U0001F9FF"
"\U0001FA00-\U0001FAFF☀-➿]+",
flags=re.UNICODE,
)
flags=re.UNICODE)
_VARIATION_SELECTOR_RE = re.compile("[︎️]")
@@ -56,7 +55,6 @@ def strip_markdown_for_tts(text: str) -> str:
"""Strip Markdown/Telegram formatting while preserving readable words."""
if not text:
return ""
text = html.unescape(str(text))
text = _MD_CODE_BLOCK_RE.sub(" ", text)
text = _MD_IMAGE_RE.sub(lambda m: f" {m.group(1)} " if m.group(1) else " ", text)
@@ -73,24 +71,20 @@ def strip_markdown_for_tts(text: str) -> str:
text = _MD_BLOCKQUOTE_RE.sub("", text)
text = _MD_LIST_ITEM_RE.sub("", text)
text = _MD_HR_RE.sub("", text)
# Leftover table pipes become pauses instead of a spoken "vertical bar".
text = _MD_TABLE_PIPE_RE.sub("; ", text)
return text
return _MD_TABLE_PIPE_RE.sub("; ", text)
def _normalize_temperature_ranges(text: str) -> str:
"""``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")):
for unit, word in _DEGREE_UNITS:
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,
)
text, flags=re.IGNORECASE)
return text
@@ -98,59 +92,43 @@ def normalize_symbols_for_tts(text: str) -> str:
"""Expand common symbols/shorthand into words a TTS engine reads well."""
if not text:
return ""
text = str(text)
text = re.sub("[   ]", " ", text) # non-breaking / thin spaces
text = text.replace("\u2212", "-") # minus sign
text = text.replace("…", "...") # ellipsis
text = re.sub("[   ]", " ", str(text)) # non-breaking / thin spaces
text = text.replace("\u2212", "-").replace("…", "...") # minus sign, ellipsis
text = _normalize_temperature_ranges(text)
# 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")):
for unit, word in _DEGREE_UNITS:
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")):
for unit, word in _DEGREE_UNITS:
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")
text = re.sub(r"(?<!\w)([-+]?\d+(?:\.\d+)?)\s*°", r"\1 degrees", text).replace("°", " degrees")
# Common weather/travel units.
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. 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 = re.sub("[•◦▪▫]", " ", text.replace("&", " and ")) # bullet glyphs
for symbol, word in (("→", " to "), ("⇒", " to "), ("≈", " about "), ("~", " about ")):
text = text.replace(symbol, word)
text = _VARIATION_SELECTOR_RE.sub("", text)
text = _EMOJI_RE.sub("", text)
return text
return _EMOJI_RE.sub("", _VARIATION_SELECTOR_RE.sub("", text))
def smooth_whitespace_for_tts(text: str) -> str:
"""Collapse visual formatting into calm spoken paragraphs.
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.
"""
"""Collapse visual formatting into calm spoken paragraphs. 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 ""
raw_lines = text.splitlines()
add_sentence_pauses = sum(1 for raw_line in raw_lines if raw_line.replace(_HEAD, "").strip()) > 1
lines: list[str] = []
@@ -161,7 +139,6 @@ def smooth_whitespace_for_tts(text: str) -> str:
if pending_heading is not None:
lines.append(pending_heading.rstrip(".:;,") + ".")
pending_heading = None
for raw_line in raw_lines:
is_heading = raw_line.rstrip().endswith(_HEAD)
line = raw_line.replace(_HEAD, "").strip()
@@ -180,15 +157,11 @@ def smooth_whitespace_for_tts(text: str) -> str:
if add_sentence_pauses and line[-1] not in ".!?;:":
line += "."
lines.append(line)
flush_pending()
text = "\n".join(lines)
text = re.sub(r"\n{3,}", "\n\n", text)
text = re.sub(r"[ \t]{2,}", " ", text)
text = re.sub(r"\s+([,.;:!?])", r"\1", text)
text = re.sub(r"([,.;:!?])([A-Za-z])", r"\1 \2", text)
text = re.sub(r"\.{4,}", "...", text)
for pattern, repl in ((r"\n{3,}", "\n\n"), (r"[ \t]{2,}", " "), (r"\s+([,.;:!?])", r"\1"),
(r"([,.;:!?])([A-Za-z])", r"\1 \2"), (r"\.{4,}", "...")):
text = re.sub(pattern, repl, text)
return text.strip()
@@ -212,27 +185,21 @@ def strip_nonspoken_blocks(text: str) -> str:
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 at the first newline; the
smoothing pass already terminates each line with punctuation, so this is safe.
"""
"""Collapse newlines into sentence breaks for single-line TTS payloads: some OpenAI-compatible
backends (e.g. Kokoro) truncate at the first newline; smoothing already ends each line with
punctuation, so this is safe."""
if not text:
return ""
text = re.sub(r"\n{2,}", ". ", text)
text = re.sub(r"(?<=[.!?;:,])\n", " ", text)
text = text.replace("\n", ". ")
text = re.sub(r"\.\s*\.", ".", text)
text = re.sub(r"[ \t]{2,}", " ", text)
for pattern, repl in ((r"\n{2,}", ". "), (r"(?<=[.!?;:,])\n", " "), (r"\n", ". "), (r"\.\s*\.", "."),
(r"[ \t]{2,}", " ")):
text = re.sub(pattern, repl, text)
return text.strip()
def prepare_spoken_text(text: str, max_chars: int | None = 4000) -> str:
"""Return a TTS-friendly script from assistant text (deterministic cleanup, not a rewrite).
Pipeline: non-spoken blocks > Markdown > symbols/units > line formatting into
sentence pauses > single line (for newline-sensitive providers), then ``max_chars``.
"""
Pipeline: non-spoken blocks > Markdown > symbols/units > line formatting into sentence
pauses > single line (for newline-sensitive providers), then ``max_chars``."""
spoken = text
for step in (strip_nonspoken_blocks, strip_markdown_for_tts, normalize_symbols_for_tts,
smooth_whitespace_for_tts, flatten_newlines_for_payload):
@@ -242,30 +209,6 @@ def prepare_spoken_text(text: str, max_chars: int | None = 4000) -> str:
return spoken
# Legacy regex fallback, only used if the shared normalizer raises.
_LEGACY_TTS_STRIP_STEPS = (
(re.compile(r'<think[\s>].*?</think>', flags=re.DOTALL), ' '),
(re.compile(r'```[\s\S]*?```'), ' '),
(re.compile(r'\[([^\]]+)\]\([^)]+\)'), r'\1'),
(re.compile(r'https?://\S+'), ''),
(re.compile(r'\*\*(.+?)\*\*'), r'\1'),
(re.compile(r'\*(.+?)\*'), r'\1'),
(re.compile(r'`(.+?)`'), r'\1'),
(re.compile(r'^#+\s*', flags=re.MULTILINE), ''),
(re.compile(r'^\s*[-*]\s+', flags=re.MULTILINE), ''),
(re.compile(r'---+'), ''),
# Emoji + variation selectors/ZWJ: providers speak them as awkward labels.
(re.compile('[\U0001F000-\U0001FAFF\u2600-\u27BF\uFE0F\u200D\U000E0020-\U000E007F]+'), ' '),
(re.compile(r'\n{3,}'), '\n\n'),
)
def _strip_markdown_for_tts(text: str) -> str:
"""``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:
pass
for pattern, repl in _LEGACY_TTS_STRIP_STEPS:
text = pattern.sub(repl, text)
return text.strip()
"""``prepare_spoken_text`` without a length cap (``tts_tool`` compatibility name)."""
return prepare_spoken_text(text, max_chars=None)