Initial commit of current state
This commit is contained in:
Adolfo Reyna
+272
@@ -0,0 +1,272 @@
|
||||
"""Tell the recognizer which words to expect, and fix the ones it still gets wrong.
|
||||
|
||||
Two mechanisms, deliberately separate, because they fail differently:
|
||||
|
||||
- **Biasing.** `SFSpeechRecognitionRequest.contextualStrings` nudges the decoder
|
||||
towards terms it would otherwise never produce. Measured at 23.5% -> 16.5%
|
||||
word error rate on technical speech. It is a hint, so it fails softly.
|
||||
- **Repair.** Deterministic substitutions applied after recognition, for
|
||||
mistakes that recur identically ("the coral voice" is always Kokoro). Exact
|
||||
and testable, but blind to context, so the rules must be specific.
|
||||
|
||||
Relevance beats coverage, and by more than expected. Eighteen apt terms measured
|
||||
16.5% WER; padding them out to 100 with names harvested from the project put it
|
||||
back to 23.5%, i.e. no better than no biasing at all. Padding with a thousand
|
||||
*random dictionary* words cost only 1.1 points, so the damage is not volume —
|
||||
it is that filenames and identifiers ("bot", "plan", "hack") are ordinary words
|
||||
the recognizer would happily produce anyway, and biasing towards them drags
|
||||
real speech onto them.
|
||||
|
||||
So auto-discovered terms are filtered to ones that are *not* ordinary English,
|
||||
capped tightly, and always ranked behind the hand-written list. The curated file
|
||||
is what carries the benefit; discovery is a small bonus with a real downside.
|
||||
"""
|
||||
|
||||
import re
|
||||
import subprocess
|
||||
from collections import OrderedDict
|
||||
from pathlib import Path
|
||||
|
||||
from loguru import logger
|
||||
|
||||
# Apple documents a ceiling of 100, but measurement says stay well under it:
|
||||
# quality decays fast once ordinary words get in.
|
||||
MAX_TERMS = 40
|
||||
|
||||
# Discovered terms are the risky kind, so they get a small share of the budget.
|
||||
MAX_DISCOVERED = 12
|
||||
|
||||
_SYSTEM_DICTIONARY = Path("/usr/share/dict/words")
|
||||
|
||||
# Words that are never worth biasing towards and only crowd out real terms.
|
||||
_STOPWORDS = {
|
||||
"and", "are", "but", "for", "from", "has", "have", "into", "not", "our",
|
||||
"out", "the", "that", "this", "was", "were", "will", "with", "you",
|
||||
"your", "main", "init", "self", "test", "tests", "src", "lib", "util",
|
||||
"utils", "readme", "true", "false", "none", "null", "class", "def",
|
||||
}
|
||||
|
||||
_IDENTIFIER = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]*$")
|
||||
_CAMEL_BOUNDARY = re.compile(r"(?<=[a-z0-9])(?=[A-Z])")
|
||||
# Things worth learning from Claude's replies: CamelCase, dotted paths,
|
||||
# snake_case, and words carrying digits. Ordinary prose is skipped.
|
||||
_TECHNICAL = re.compile(r"\b(?:[A-Za-z]+[._][A-Za-z0-9._-]+|[a-z]+[A-Z][A-Za-z]*|[A-Za-z]*\d[A-Za-z0-9]*)\b")
|
||||
|
||||
_SKIP_DIRS = {".git", ".venv", "__pycache__", "node_modules", ".cache", "dist", "build"}
|
||||
|
||||
|
||||
def _speakable(term: str) -> str:
|
||||
"""Turn an identifier into something a person could say.
|
||||
|
||||
Apple's guidance is one or two words per phrase, spoken without a pause, so
|
||||
`sounddevice_transport` is useless as-is but `sounddevice transport` is not.
|
||||
"""
|
||||
term = term.replace("_", " ").replace("-", " ").replace(".", " ")
|
||||
term = _CAMEL_BOUNDARY.sub(" ", term)
|
||||
return " ".join(term.split())
|
||||
|
||||
|
||||
class Vocabulary:
|
||||
"""Ranked terms for the recognizer, plus repair rules for its known mistakes.
|
||||
|
||||
Terms are ordered by how likely they are to matter: words the user wrote
|
||||
down, then the project around them, then whatever Claude has been talking
|
||||
about lately. Only the first ``limit`` survive.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
project_dir: Path | None = None,
|
||||
vocabulary_file: Path | None = None,
|
||||
corrections_file: Path | None = None,
|
||||
limit: int = MAX_TERMS,
|
||||
):
|
||||
self._limit = limit
|
||||
self._user: list[str] = []
|
||||
self._project: list[str] = []
|
||||
self._observed: OrderedDict[str, None] = OrderedDict()
|
||||
self._corrections: list[tuple[re.Pattern, str]] = []
|
||||
|
||||
if vocabulary_file and vocabulary_file.exists():
|
||||
self._user = _read_terms(vocabulary_file)
|
||||
logger.debug(f"Vocabulary: {len(self._user)} terms from {vocabulary_file.name}")
|
||||
if corrections_file and corrections_file.exists():
|
||||
self._corrections = _read_corrections(corrections_file)
|
||||
logger.debug(f"Vocabulary: {len(self._corrections)} repair rules")
|
||||
if project_dir:
|
||||
self._project = _terms_from_project(project_dir)
|
||||
|
||||
def add_terms(self, terms: list[str]):
|
||||
"""Add terms that rank alongside the hand-written ones.
|
||||
|
||||
For names the user certainly says — project names out of the brain —
|
||||
which are as authoritative as anything typed into vocabulary.txt.
|
||||
"""
|
||||
for term in terms:
|
||||
speakable = _speakable(term)
|
||||
if _worth_keeping(speakable) and speakable not in self._user:
|
||||
self._user.append(speakable)
|
||||
|
||||
def observe(self, text: str):
|
||||
"""Learn technical words from something Claude just said.
|
||||
|
||||
What the assistant is discussing is a good predictor of what the user is
|
||||
about to say back, so its replies feed the next recognition.
|
||||
"""
|
||||
for match in _TECHNICAL.findall(text or ""):
|
||||
term = _speakable(match)
|
||||
if _worth_keeping(term) and _is_distinctive(term):
|
||||
self._observed.pop(term, None) # move to most-recent
|
||||
self._observed[term] = None
|
||||
# Keep the tail bounded; only the newest can reach the term list anyway.
|
||||
while len(self._observed) > self._limit * 2:
|
||||
self._observed.popitem(last=False)
|
||||
|
||||
def terms(self) -> list[str]:
|
||||
"""The ranked, de-duplicated, capped term list.
|
||||
|
||||
What Claude said a moment ago predicts the next utterance better than an
|
||||
arbitrary filename does, so recent observations get a reserved share of
|
||||
the budget rather than queueing behind the whole project.
|
||||
"""
|
||||
budget = max(0, self._limit - len(self._user))
|
||||
discovered = min(budget, MAX_DISCOVERED)
|
||||
recent = list(reversed(self._observed))[: (discovered + 1) // 2]
|
||||
project = self._project[: discovered - len(recent)]
|
||||
ranked = [*self._user, *recent, *project]
|
||||
|
||||
seen: dict[str, str] = {}
|
||||
for term in ranked:
|
||||
key = term.lower()
|
||||
if key not in seen:
|
||||
seen[key] = term
|
||||
if len(seen) >= self._limit:
|
||||
break
|
||||
return list(seen.values())
|
||||
|
||||
def prompt_block(self, limit: int = 60) -> str:
|
||||
"""The stable terms, for Claude's system prompt.
|
||||
|
||||
Only the written-down and project-derived terms go here: the prompt is
|
||||
fixed for the session, so terms learned mid-conversation could not
|
||||
appear anyway.
|
||||
"""
|
||||
stable = list(dict.fromkeys([*self._user, *self._project]))[:limit]
|
||||
return ", ".join(stable)
|
||||
|
||||
def repair(self, text: str) -> str:
|
||||
"""Apply the substitution rules to a transcript."""
|
||||
for pattern, replacement in self._corrections:
|
||||
text = pattern.sub(replacement, text)
|
||||
return text
|
||||
|
||||
|
||||
def _worth_keeping(term: str) -> bool:
|
||||
return (
|
||||
len(term) >= 3
|
||||
and term.lower() not in _STOPWORDS
|
||||
and not term.isdigit()
|
||||
and len(term) <= 40
|
||||
)
|
||||
|
||||
|
||||
def _load_dictionary() -> frozenset[str]:
|
||||
try:
|
||||
return frozenset(w.strip().lower() for w in _SYSTEM_DICTIONARY.read_text().splitlines())
|
||||
except OSError:
|
||||
return frozenset()
|
||||
|
||||
|
||||
_ORDINARY_WORDS = _load_dictionary()
|
||||
|
||||
|
||||
def _is_distinctive(term: str) -> bool:
|
||||
"""Whether biasing towards this term could actually change an outcome.
|
||||
|
||||
Only words the recognizer would not already produce are worth boosting.
|
||||
Biasing towards ordinary English measurably drags correct speech onto the
|
||||
wrong word, which is how a harvested term list erased the entire benefit.
|
||||
"""
|
||||
words = term.lower().split()
|
||||
if len(words) > 1:
|
||||
return True # multi-word phrases are specific enough to be safe
|
||||
return words[0] not in _ORDINARY_WORDS
|
||||
|
||||
|
||||
def _read_terms(path: Path) -> list[str]:
|
||||
terms = []
|
||||
for line in path.read_text().splitlines():
|
||||
line = line.split("#", 1)[0].strip()
|
||||
if line and _worth_keeping(line):
|
||||
terms.append(line)
|
||||
return terms
|
||||
|
||||
|
||||
def _read_corrections(path: Path) -> list[tuple[re.Pattern, str]]:
|
||||
rules = []
|
||||
for lineno, raw in enumerate(path.read_text().splitlines(), 1):
|
||||
line = raw.split("#", 1)[0].strip()
|
||||
if not line:
|
||||
continue
|
||||
if "=>" not in line:
|
||||
logger.warning(f"{path.name}:{lineno}: expected 'heard => replacement', got {raw!r}")
|
||||
continue
|
||||
heard, replacement = (part.strip() for part in line.split("=>", 1))
|
||||
if not heard:
|
||||
continue
|
||||
# Word-bounded and case-insensitive so "coral voice" matches mid-sentence
|
||||
# but "chorale" never does.
|
||||
rules.append((re.compile(rf"\b{re.escape(heard)}\b", re.IGNORECASE), replacement))
|
||||
return rules
|
||||
|
||||
|
||||
def _terms_from_project(project_dir: Path) -> list[str]:
|
||||
"""Names from the code the user is most likely to talk about."""
|
||||
terms: list[str] = []
|
||||
|
||||
for path in sorted(project_dir.rglob("*")):
|
||||
if any(part in _SKIP_DIRS for part in path.parts):
|
||||
continue
|
||||
if path.is_file():
|
||||
term = _speakable(path.stem)
|
||||
if _worth_keeping(term) and _is_distinctive(term):
|
||||
terms.append(term)
|
||||
|
||||
terms.extend(_git_terms(project_dir))
|
||||
|
||||
for path in sorted(project_dir.glob("*.py")):
|
||||
try:
|
||||
source = path.read_text(errors="ignore")
|
||||
except OSError:
|
||||
continue
|
||||
for name in re.findall(r"^\s*(?:class|def)\s+([A-Za-z_][A-Za-z0-9_]*)", source, re.M):
|
||||
term = _speakable(name)
|
||||
if _worth_keeping(term) and _is_distinctive(term):
|
||||
terms.append(term)
|
||||
|
||||
return list(dict.fromkeys(terms))
|
||||
|
||||
|
||||
def _git_terms(project_dir: Path) -> list[str]:
|
||||
"""Branch and author names, which are spoken far more often than they're typed."""
|
||||
terms: list[str] = []
|
||||
commands = (
|
||||
["git", "branch", "--format=%(refname:short)"],
|
||||
["git", "log", "-40", "--format=%an"],
|
||||
)
|
||||
for command in commands:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
command, cwd=project_dir, capture_output=True, text=True, timeout=5
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
continue
|
||||
if result.returncode != 0:
|
||||
continue
|
||||
for line in result.stdout.splitlines():
|
||||
for piece in line.split():
|
||||
term = _speakable(piece)
|
||||
if _worth_keeping(term) and _is_distinctive(term):
|
||||
terms.append(term)
|
||||
return terms
|
||||
Reference in New Issue
Block a user