voice

Speech to text on this machine, for think-aloud answers.

This module turns a spoken answer into text. With the local engines, the audio does not go off the machine. The web app records while you talk. It cuts the audio into segments at your pauses and sends each segment here. This module transcribes the segments one at a time, in order. The text goes into the think-aloud box while you continue to talk.

These are the imports of the module.

Engines

There are three engines. gemma and mlx run on this machine. openai sends the audio to OpenAI.

• gemma runs a Gemma 4 model on LiteRT through the model layer. E4B is the default. E2B is faster.
• mlx runs any speech to text model that mlx-audio can load: Whisper, Parakeet, Qwen3-ASR, Voxtral, Canary, Moonshine and more. Install it with pip install abhyasa[whisper].
• openai uses the OpenAI transcription API. It is off until you set a key. It sends the audio off the machine.

Each engine gets a hint: the name of the concept and its terms. Gemma reads it in its prompt. Whisper reads it as its initial prompt, and Qwen3-ASR as hot words. OpenAI reads it as its prompt.

The table gives the numbers on this machine (an M-series Mac), with each model warm. “Clip A” is the English test clip of the model layer (rishi/nbs/speech.wav). “Clip B” is a clip in a cloned voice with Nyāya terms in IAST and technical terms. “Terms” is the fraction of the 9 key terms in clip B that the engine wrote correctly. “5 s” is the time for one 5-second segment.

model	load	5 s	clip A	clip B, no hint / hint	terms, no hint / hint	Sanskrit CER	memory
litert-community/gemma-4-E4B-it-litert-lm	5.0 s	0.69 s	9%	0% / 0%	9 / 9	12%	1.5 GB
litert-community/gemma-4-E2B-it-litert-lm	2.6 s	0.38 s	14%	6% / 2%	7 / 8	13%	1.3 GB
mlx-community/whisper-large-v3-turbo-asr-fp16	4.8 s	0.68 s	6%	4% / 0%	8 / 9	10%	2.3 GB
mlx-community/parakeet-tdt-0.6b-v3	4.9 s	0.15 s	6%	4% / 4%	7 / 7	Latin script	3.2 GB
mlx-community/Qwen3-ASR-0.6B-8bit	3.8 s	0.14 s	6%	13% / 0%	5 / 9	16%	2.0 GB
mlx-community/Qwen3-ASR-1.7B-8bit	3.6 s	0.25 s	6%	6% / 0%	7 / 9	14%	3.6 GB
mlx-community/Voxtral-Mini-4B-Realtime-2602-4bit	3.7 s	2.21 s	6%	6% / 6%	7 / 7	22%	5.5 GB
UsefulSensors/moonshine-base	3.9 s	0.07 s	100%	8% / 8%	6 / 6	Latin script	0.8 GB
Mediform/canary-1b-v2-mlx-q8	4.2 s	0.08 s	6%	4% / 4%	8 / 8	Latin script	1.4 GB
Su-śrotā (NeMo, CPU, in vachaka)	46 s	0.3 s for 9 s	Devanagari only	Devanagari only	—	0%	1.7 GB

• “load” is a cold start of a cached model in a new process, with the Python imports. The first download takes longer.
• “memory” is the MLX peak for the mlx models, and the growth of the process for Gemma on LiteRT.
• “Sanskrit CER” is the character error rate on a 9-second Nyāya sūtra (from vachaka), after the vachaka normalization. vachaka chose that clip with Su-śrotā, so the 0% of Su-śrotā is not a fair result.
• Moonshine gave no text for the 24-second clip A. It is good only for short segments.
• Su-śrotā writes only Devanagari, and it needs NeMo and torch on Python 3.12 or older. It is not an engine here. For Sanskrit speech, use Gemma or Whisper with the hint, or vachaka.

Gemma 4 E4B stays the default: it has no errors on clip B, it writes all 9 terms without a hint, and one 5-second segment takes 0.7 s. For more speed, use Parakeet (0.15 s) or Qwen3-ASR 0.6B with the hint (0.14 s, all 9 terms).

STT_SP is the system prompt for the Gemma engine. The other constants are the default models. ALIASES keeps the old engine names: whisper and parakeet are the mlx engine with a fixed model.

Settings

The settings are in settings.json in cfg_dir(). The web app changes them, and the next segment uses the change. You do not have to restart. The models section is for the model routing of the tutor, later. This module does not read it.

---

source

update_settings

def update_settings(
    section:str, **kw
)->dict:

Change some keys in one section of the settings.

---

source

save_settings

def save_settings(
    d:dict
)->dict:

Write all the settings.

---

source

settings

def settings()->dict:

The saved settings over the defaults.

---

source

cfg_dir

def cfg_dir()->pathlib.Path:

The folder for abhyasa’s settings and keys: $ABHYASA_CONFIG, else ~/.config/abhyasa.

Without a file, you get the defaults. A change to one key keeps the other keys.

test_eq(settings()['stt'], dict(engine='gemma', model='', segment=True))
update_settings('stt', segment=False)
test_eq(settings()['stt'], dict(engine='gemma', model='', segment=False))
update_settings('stt', segment=True)

current gives the engine and model in use. $ABHYASA_STT has priority over the settings. It can be an engine (gemma), an old name (whisper), or an engine and a model (mlx:mlx-community/Qwen3-ASR-0.6B-8bit).

---

source

current

def current()->tuple:

The engine and model in use: $ABHYASA_STT, else the settings.

---

source

resolve

def resolve(
    engine:str, model:str=''
)->tuple:

The engine and model for a name: aliases become mlx, engine:model splits, an empty model gets the default.

The settings give Gemma E4B by default. The environment variable has priority.

test_eq(current(), ('gemma', DFLT_GEMMA))
test_eq(resolve('parakeet'), ('mlx', DFLT_PARAKEET))
test_eq(resolve('mlx:mlx-community/Qwen3-ASR-0.6B-8bit'), ('mlx', 'mlx-community/Qwen3-ASR-0.6B-8bit'))
os.environ['ABHYASA_STT'] = 'whisper'
test_eq(current(), ('mlx', DFLT_WHISPER))
del os.environ['ABHYASA_STT']

One model in memory

resident keeps one speech model loaded. When the app asks for a different model, it frees the old model first. Two large models in memory at the same time can make the Mac slow.

---

source

resident

def resident(
    key, load, close:NoneType=None
):

The loaded speech model for key: when key is new, free the old model, then load() the new one.

---

source

unload

def unload():

Free the loaded speech model.

A new key frees the old model. The same key does not load again.

freed = []
resident('a', lambda: 'A', freed.append)
test_eq(resident('a', lambda: 'other'), 'A')
test_eq(resident('b', lambda: 'B'), 'B')
test_eq(freed, ['A'])
unload()

Each engine is a function that takes a WAV path, a hint and a model, and returns text. Gemma keeps one LiteRT engine loaded, and makes a new conversation for each segment. Then one segment does not change the next.

The mlx engine gives the hint to a model that accepts one: Whisper as initial_prompt, Qwen3-ASR as hotwords.

A Whisper-like model gets initial_prompt. A Qwen3-like model gets a list of hot words. A model without either gets nothing.

class _W:
    def generate(self, audio, initial_prompt=None): pass
class _Q:
    def generate(self, audio, hotwords=None): pass
test_eq(_hint_kw(_W(), 'Nyāya, hetu'), dict(initial_prompt='Nyāya, hetu'))
test_eq(_hint_kw(_Q(), 'Nyāya, hetu'), dict(hotwords=['Nyāya', 'hetu']))
test_eq(_hint_kw(_W(), ''), {})

OpenAI

The openai engine posts the WAV to the OpenAI transcription API, with the hint as its prompt. http_post is httpx.post unless a test puts a fake in its place.

The key comes from the first of these places that has OPENAI_API_KEY: the environment, the abhyasa keys.env, then the ramabana keys.env (~/.config/ramabana/keys.env). abhyasa only reads the ramabana file. set_openai_key writes the abhyasa file with mode 0600. An empty key removes the file. masked shows only the last 4 characters.

---

source

masked

def masked(
    key:str
)->str:

---

source

set_openai_key

def set_openai_key(
    key:str=''
):

Write key to abhyasa’s keys.env (mode 0600); an empty key removes it. A key with other than letters, digits, - and _ raises ValueError.

---

source

openai_key

def openai_key()->tuple:

The OpenAI key and where it is from; ('', '') when there is none.

A key in the abhyasa file comes back with its source, and the file has mode 0600. When the abhyasa file has no key, abhyasa reads the ramabana file.

RAMABANA_KEYS = Path(tempfile.mkdtemp())/'keys.env'
test_eq(openai_key(), ('', ''))
RAMABANA_KEYS.write_text('export OPENAI_API_KEY="sk-ramabana-0000"\n')
test_eq(openai_key(), ('sk-ramabana-0000', str(RAMABANA_KEYS)))
set_openai_key('sk-test-1234')
test_eq(openai_key(), ('sk-test-1234', str(cfg_dir()/'keys.env')))
test_eq(oct((cfg_dir()/'keys.env').stat().st_mode & 0o777), '0o600')
test_eq(masked('sk-test-1234'), '…1234')

The engine sends the model, the hint and the WAV, with the key in the header. This test uses a fake HTTP client.

class _R:
    def raise_for_status(self): pass
    def json(self): return dict(text='hetu and sādhya')
sent = {}
def fake_post(url, **kw): sent.update(url=url, **kw); return _R()
http_post = fake_post
p = Path(tempfile.mkdtemp())/'a.wav'; p.write_bytes(b'RIFF')
test_eq(_openai(p, 'hetu'), 'hetu and sādhya')
test_eq(sent['data'], dict(model=DFLT_OPENAI, response_format='json', prompt='hetu'))
test_eq(sent['headers']['Authorization'], 'Bearer sk-test-1234')
set_openai_key('')
RAMABANA_KEYS.unlink()
test_fail(lambda: _openai(p), contains='no OpenAI key')
http_post = None

A key with a newline or a space can add a second line to the file. Thus set_openai_key refuses it.

test_fail(lambda: set_openai_key('sk-a\nOTHER=1'), contains='only letters')
test_fail(lambda: set_openai_key('sk a b c d e'), contains='only letters')

Transcribe

transcribe accepts a path or bytes. It writes bytes to a temporary file and removes the file after use. Without an engine, it uses current(), so a change on the Settings page applies to the next segment.

---

source

transcribe

def transcribe(
    audio, engine:str=None, hint:str='', model:str=None
)->str:

Text for a WAV file or WAV bytes. hint lists words that the speaker can use, such as the name of the concept.

wav_seconds gives the length of a clip in seconds. silence makes a clip of silence. The app uses this clip to warm a model.

---

source

silence

def silence(
    secs:float=0.5
)->bytes:

WAV bytes of silence: 16 kHz, mono, 16 bit.

---

source

wav_seconds

def wav_seconds(
    audio:bytes
)->float:

The length of WAV bytes, in seconds; 0 when they are not WAV.

Half a second of silence is 0.5 seconds. Bytes that are not WAV give 0.

test_eq(wav_seconds(silence()), .5)
test_eq(wav_seconds(b'nope'), 0.)

Models

KNOWN lists the models that we measured (see the table above). The Settings page shows them first. cached tells if a model is in the Hugging Face cache. It does not use the network.

---

source

cache_bytes

def cache_bytes(
    model:str
)->int:

The bytes of model in the Hugging Face cache so far, with the files that are still downloading.

---

source

cached

def cached(
    model:str
)->bool:

True when model is in the Hugging Face cache.

search_models looks up mlx-community speech to text models on Hugging Face, with their size. It drops text to speech, aligners and diarization models, which the mlx engine cannot use.

---

source

search_models

def search_models(
    q:str='', limit:int=15, api:NoneType=None
)->list:

mlx-community speech to text models that match q, most downloaded first, with their size.

---

source

model_size

def model_size(
    model:str, api:NoneType=None
)->int:

The bytes of all the files of model on Hugging Face.

---

source

is_stt

def is_stt(
    model:str
)->bool:

True when the repo name looks like a speech to text model.

The filter keeps speech to text and drops the others.

test_eq([is_stt(m) for m in ('mlx-community/Qwen3-ASR-0.6B-8bit', 'mlx-community/whisper-tiny-asr-fp16', 'mlx-community/Qwen3-TTS-12Hz-1.7B-Base-bf16', 'mlx-community/silero-vad')],
        [True, True, False, False])

fetch downloads a model and notes it in the settings. remove deletes a model from the cache, but only a model that fetch downloaded. Other entries in the cache belong to other programs.

---

source

remove

def remove(
    model:str
):

Delete model from the Hugging Face cache. Only a model that fetch downloaded.

---

source

fetch

def fetch(
    model:str
):

Download model into the Hugging Face cache; note it when it was not there before.

remove refuses a model that abhyasa did not download.

test_fail(lambda: remove(DFLT_GEMMA), contains='stays')

Segments in order

The browser sends one segment each time you stop to think. Speech has one worker thread for all the model calls. Thus only one model is in memory, and the segments run one at a time, in the order they came. A Dictation collects the segments of one answer. next gives the text in the order of the segment numbers, also when a slow request arrives late. When the browser sends a segment again (after an error), the code does not use it two times.

---

source

Dictation

def Dictation(
    speech, hint:str=''
):

The segments of one spoken answer: each is transcribed on the speech worker, and the text comes out in order.

Speech owns the worker. warm loads the model in use with half a second of silence, so the first segment does not wait for it. use changes the engine and model, saves the change, and warms the new model.

---

source

Speech

def Speech(
    fn:NoneType=None
):

One worker thread for every speech model call, and the open dictations.

Tests with a fake engine

This fake engine is slow on even segments, so the requests finish out of order. The text must still come out in order.

calls = []
def slow_fake(audio, hint=''):
    n = int(audio.decode())
    calls.append(n)
    time.sleep(.05 if n % 2 == 0 else 0)
    return f'segment {n} ({hint})'
sp = Speech(slow_fake)
d = sp.runs[sp.start('hetu')]

Segments 0 to 5 arrive out of order, from several threads. Segment 3 comes two times.

order = [2, 0, 1, 3, 3, 5, 4]
with ThreadPoolExecutor(4) as ex: added = list(ex.map(lambda n: d.add(n, str(n).encode()), order))
test_eq(sum(added), 6)
d.stop(6)
out = []
while not (r := d.next(5)).get('done'): out.append(r)
test_eq([r['seq'] for r in out], list(range(6)))
test_eq(out[0]['text'], 'segment 0 (hetu)')
test_eq(sorted(calls), list(range(6)))

All the model calls ran on the one worker thread, one at a time.

seen = set()
sp.submit(lambda: seen.add(threading.current_thread().name)).result()
test_eq(len(seen), 1)
test_eq(d.pending, 0)

An engine error gives an empty text with the error. It does not stop the segments after it.

def flaky(audio, hint=''):
    if audio == b'bad': raise RuntimeError('no model')
    return 'ok'
sp = Speech(flaky)
d = sp.runs[sp.start()]
d.add(0, b'bad'); d.add(1, b'good'); d.stop(2)
test_eq(d.next(5), dict(seq=0, text='', error='no model'))
test_eq(d.next(5), dict(seq=1, text='ok'))
test_eq(d.next(5), dict(done=True))

Without stop, next waits and then gives None. use saves the choice and warms the new model with silence.

d = sp.runs[sp.start()]
test_is(d.next(.05), None)
heard = []
sp = Speech(lambda a, hint='': heard.append(wav_seconds(a)) or '')
sp.use('parakeet').result()
test_eq(current(), ('mlx', DFLT_PARAKEET))
test_eq(heard, [.5])
assert sp.status.startswith('ready'), sp.status
update_settings('stt', engine='gemma', model='')

Real models

These tests use a real clip and real models. They are slow. Run them with nbdev-test --flags slow.

clip = first(p for p in (Path('../../rishi/nbs/speech.wav'), Path('../../../rishi/nbs/speech.wav')) if p.exists())   # from the repo or a worktree
if clip:
    for e in ('gemma', 'parakeet', 'mlx:mlx-community/Qwen3-ASR-0.6B-8bit'): print(e, transcribe(clip, e, hint='Tennyson'))
    unload()