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

refactor(tools): fold local STT loader/gates and TTS normalizer passes into table-driven substitutions

Browse Source
This commit is contained in:
Teknium
2026-09-03 00:52:11 -07:00
parent 97b554f200
commit 404a324ea5
2 changed files with 39 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/transcription_local.py
+19 -51
View File
@@ -36,12 +36,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,7 +48,7 @@ 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 "
@@ -99,16 +95,14 @@ _CUDA_LIB_ERROR_MARKERS = (
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 ""
@@ -143,25 +137,18 @@ def _load_local_whisper_model(model_name: str, device: str = "auto", compute_typ
# 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,7 +167,6 @@ 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] = {
@@ -192,30 +178,24 @@ def build_local_transcribe_kwargs(stt_config: Optional[Dict[str, Any]] = None) -
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:
@@ -225,11 +205,10 @@ def _is_hallucinated_segment(segment: Any, no_speech_threshold: float, logprob_t
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 +217,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()
@@ -259,37 +236,28 @@ def _transcribe_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),
)
# 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/tts_text_normalize.py
+20 -44
View File
@@ -31,6 +31,7 @@ _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"),
@@ -56,7 +57,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,23 +73,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,48 +95,35 @@ 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:
@@ -150,7 +134,6 @@ def smooth_whitespace_for_tts(text: str) -> str:
"""
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] = []
@@ -180,15 +163,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()
@@ -219,11 +198,9 @@ def flatten_newlines_for_payload(text: str) -> str:
"""
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()
@@ -265,7 +242,6 @@ def _strip_markdown_for_tts(text: str) -> str:
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()
for pattern, repl in _LEGACY_TTS_STRIP_STEPS:
text = pattern.sub(repl, text)
return text.strip()