models

The models: the model for each task, the PII guard for the text that leaves the Mac, and speech.

This module holds the models that abhyasa talks to. It imports only from abhyasa.learner. It has three parts:

• The model for each task: the route of each task (for_task), the default ask (route_ask), the list of models, and where each one runs.
• The guard: ramabana’s PII guard. Text that leaves the Mac passes it first.
• Speech: speech to text and text to speech, on this Mac or with OpenAI.

The heavy model code (ramabana, rishi, mlx-audio) loads inside the functions that use it, so the import of this module stays light.

Import the modules that this module uses. The heavy ones load inside the functions that need them, so the import stays light.

The model for each task

One call shape for each model: ask(prompt, sp) -> str.

This module connects the tutor to a model. The tutor needs only one function, ask(prompt, sp), which returns text. The module also parses JSON replies and gives Fake, a model for tests that runs offline.

Import the tools that this module uses.

The model

Each surface gives its own ask.

surface	ask
web app and notebook	route_ask: each call goes through the routing of ramabana, on a backend that stays open
the ramabana terminal agent	agent.oneshot, so the routing of the agent applies
tests	Fake

The default model is --model, else $ABHYASA_MODEL, else DFLT_MODEL (claude/sonnet).

---

source

dflt_model

def dflt_model(
    model:str=None
)->str:

model, else $ABHYASA_MODEL, else the setting models.default, else DFLT_MODEL.

JSON replies

parse_json finds the JSON in a reply. ask_json asks again one time when the reply is not valid JSON.

---

source

ask_json

def ask_json(
    ask, prompt:str, sp:str='', tries:int=2, need:tuple=()
):

Ask for JSON and parse it. On a bad reply, or one with no text in a key of need, ask one more time with the error.

---

source

parse_json

def parse_json(
    s:str
):

The first JSON object or array in s, also inside a ``fence. RaiseValueError` when there is none.

parse_json reads JSON in a fence, JSON inside other text, and arrays. A fence inside a JSON string does not break it. It fails when there is no JSON.

test_eq(parse_json('```json\n{"a": 1}\n```'), {'a': 1})
test_eq(parse_json('Here it is: {"a": [1, 2]} thanks'), {'a': [1, 2]})
test_eq(parse_json('[{"x": 1}]'), [{'x': 1}])
test_fail(lambda: parse_json('nothing'), contains='no JSON')
test_eq(parse_json('{"answer": "Use ```python\\nx = 1\\n``` here."}'), {'answer': 'Use ```python\nx = 1\n``` here.'})
test_eq(parse_json('Sure. {"a": 1} and then {"b": 2}'), {'a': 1})

A model for tests

Fake replies from a dict of rules. The first rule with a key in the prompt gives the reply. It sends a dict or a list reply as JSON. It keeps each call in calls.

---

source

Fake

def Fake(
    rules:dict=None, dflt:str='{}'
):

A stand-in model for tests: Fake({'grade': {...}, 'question': {...}}).

A matching rule gives its reply, and Fake records the call.

f = Fake({'GRADE': dict(correct=True)})
test_eq(ask_json(f, 'GRADE this'), dict(correct=True))
test_eq(len(f.calls), 1)

After a reply that is not JSON, ask_json asks again and gets the JSON.

bad = Fake(dict(X=lambda p, sp: 'oops' if 'not valid' not in p else '{"ok": 1}'))
test_eq(ask_json(bad, 'X'), {'ok': 1})

With need, a reply must be one JSON object with text in each of those keys. A list, or an object with no question, gets one more try. A second bad reply raises a ValueError that names the key.

odd = Fake(dict(Q=lambda p, sp: '{"question": "What is a rank?"}' if 'must be one JSON object' in p else '[1, 2]'))
test_eq(ask_json(odd, 'Q', need=('question',)), {'question': 'What is a rank?'})
test_fail(lambda: ask_json(Fake(dflt={}), 'Q', need=('question',)), contains='question')

A model for each task

The tutor does five tasks, and each task can use its own model. A quick model can grade and give hints. A strong model can write questions and lessons. The Settings page saves the choice. Router reads it on each call, so a change has an effect on the next call.

task	what the model does
question	writes a question and its hidden model answer
grade	reads the conversation and grades it
hint	gives the next hint
lesson	teaches the smallest step after a miss
extract	finds the concepts in a text

Each task also has a reply cap: the most tokens that the model can write in one reply. TOKENS gives the default of each task. A question has options and a rubric, and an extract can name 15 concepts, so these two get the most. A hint is short. The Settings page can change each cap. Only the cap is a setting, because ramabana’s oneshot uses no other per-call value: it turns thinking off, and the time limit of a model is in ramabana’s config.toml. The model layer gives the cap to the API models (OpenAI, Codex, Gemini, Copilot) and to the models on this Mac. Claude Code does not use it for the reply, but ramabana uses it to fit the prompt in the window of the model.

model_ask gives an ask for any model name. The name can be any model that ramabana can run: Claude Code, Codex, OpenAI, Copilot, Gemini, or a model on this machine (LiteRT, mlx, llama.cpp, Ollama). The call goes through the routing of ramabana, as route_ask below shows.

---

source

model_ask

def model_ask(
    name:str
):

An ask(prompt, sp='') for model name, made once and kept.

Names for the model list

The Settings page shows a list of models. Each item has a name that people know, such as “Claude Sonnet” or “GPT-5.5”. The model id, such as claude/sonnet, is the value that the settings keep. ramabana’s model_label makes the name from the id, and model_label here adds “a local model:” before the name of a model that runs on this Mac. PROVIDERS gives the name of each group in the list, and provider_label reads it. ramabana’s is_local tells if a model runs on this Mac. For Ollama, it looks at the daemon host (the host in config.toml, else $OLLAMA_HOST), and it does not connect to it. Thus a model on an Ollama daemon on another computer has no “a local model” name, and its group is “Ollama on another computer”.

---

source

provider_label

def provider_label(
    provider:str, values:tuple=()
)->str:

The name of the provider group in the model list, for the model ids values. An Ollama daemon on another computer is not on this Mac.

---

source

model_label

def model_label(
    value:str, provider:str='', repo:str=''
)->str:

ramabana’s model_label of the model id value, with LOCAL before a model on this Mac; repo is the download of a local model, if known.

Each kind of id gets a plain name. A Claude Code tier gets the word Claude. A local model says so.

test_eq(model_label('claude/sonnet'), 'Claude Sonnet')
test_eq(model_label('claude/claude-opus-5-5'), 'Claude Opus 5.5')
test_eq(model_label('anthropic/claude-haiku-4-5-20251001'), 'Claude Haiku 4.5')
test_eq(model_label('codex/gpt-6-astra'), 'GPT-6 Astra')
test_eq(model_label('openai/gpt-5.4-mini'), 'GPT-5.4 Mini')
test_eq(model_label('gemini/gemini-3.5-flash'), 'Gemini 3.5 Flash')
test_eq(model_label('qwen-4b', 'mlx', 'mlx-community/Qwen3.5-4B-MLX-4bit'), 'a local model: Qwen 3.5 4B')
test_eq(model_label('gemma-e4b', 'litert', 'litert-community/gemma-4-E4B-it-litert-lm'), 'a local model: Gemma 4 E4B')
test_eq(model_label('my-model', 'azure'), 'my-model')

An OpenAI id gets the name that OpenAI uses. A snapshot date such as 2024-07-18 goes, so the name is the same as the name of the model.

test_eq(model_label('openai/gpt-4o-mini-2024-07-18'), 'GPT-4o mini')
test_eq(model_label('copilot/gpt-4-o-preview'), 'GPT-4o preview')
test_eq(model_label('openai/gpt-4o'), 'GPT-4o')
test_eq(model_label('openai/gpt-4.1-mini'), 'GPT-4.1 Mini')

The o-series names stay in lower case, with their hyphen:

test_eq(model_label('codex/o4-mini'), 'o4-mini')
test_eq(model_label('openai/o3'), 'o3')

One letter and a number stay together, as the maker writes them:

test_eq(model_label('copilot/kimi-k3'), 'Kimi K3')
test_eq(model_label('openrouter/qwen3-8b'), 'Qwen 3 8B')   # a longer word and a number still get a space

An Ollama model is local only when its daemon is on this Mac. With $OLLAMA_HOST on localhost, the model and its group are on this Mac:

from fastcore.xtras import modified_env

with modified_env(OLLAMA_HOST='localhost:11434'):
    test_eq(model_label('ollama/qwen3:8b'), 'a local model: Qwen 3 8B')
    test_eq(provider_label('ollama', ['ollama/qwen3:8b']), 'On this Mac (Ollama)')

With $OLLAMA_HOST on another computer, the model is not called local, and its group has a hosted name. The check reads only the host name: it does not connect:

with modified_env(OLLAMA_HOST='http://gpu-box.lan:11434'):
    test_eq(model_label('ollama/qwen3:8b'), 'Qwen 3 8B')
    test_eq(provider_label('ollama', ['ollama/qwen3:8b']), 'Ollama on another computer')

The other groups keep their names from PROVIDERS. A provider that is not in it gets its own name:

test_eq(provider_label('mlx', ['qwen-4b']), 'On this Mac (MLX)')
test_eq(provider_label('claude', ['claude/sonnet']), 'Claude')
test_eq(provider_label('groq'), 'Groq')
test_eq(provider_label(''), 'Other')

Where a model runs

where_of says where the model name runs: on this Mac or hosted, and the company of a hosted model, from ramabana’s vendor_of. It uses _on_mac, as provider_label does, so an Ollama daemon on another computer is hosted. When ramabana does not know the company, the name of the group stands in. An unknown model is hosted, as the guard takes it. where_chip gives the short text for the page.

---

source

where_chip

def where_chip(
    name:str
)->str:

on this Mac, or hosted · <company>, for the model name.

---

source

where_of

def where_of(
    name:str
)->dict:

Who runs the model name: model (its label), where (on this Mac or hosted) and who (the company of a hosted model, ramabana’s vendor_of). An unknown model is hosted, as the guard takes it.

A Claude model is hosted by Anthropic. A model on an Ollama daemon on this Mac stays here.

test_eq(where_chip('claude/haiku'), 'hosted · Anthropic')
test_eq(where_chip('copilot/gpt-6-sol'), 'hosted · GitHub')
with modified_env(OLLAMA_HOST='localhost:11434'): test_eq(where_chip('ollama/qwen3:8b'), HERE)
with modified_env(OLLAMA_HOST='http://gpu-box.lan:11434'): test_eq(where_chip('ollama/qwen3:8b'), 'hosted · Ollama on another computer')
test_eq(where_of(''), dict(where='hosted'))

catalog lists the models that can run here, for the Settings page: the rows of ramabana’s available_models(ttl=). Each row has a name from ramabana’s model_label, local, vendor and model_id. abhyasa adds “a local model:” before the name of a local row. The look-up asks each provider, so it takes some seconds. ramabana keeps the list in memory and on the disk: a new process shows the last list at once, and a list older than ttl seconds refreshes in the background. When the look-up fails, the list has only the models in FALLBACK.

---

source

catalog

def catalog(
    ttl:float=300.0
)->list:

The models that can run here: ramabana’s rows (value, name, label, provider, source, local, vendor, model_id), with LOCAL before the name of a local row. The last list comes at once; one older than ttl seconds refreshes in the background.

These tests use ramabana. slow is a stand-in for the slow look-up of ramabana (_model_rows): it counts its calls, and gives one row with the facts of ramabana (_facts). The tests keep the list in a temp folder.

import ramabana.core as rc

Two calls to catalog ask the providers one time.

calls, real, real_dir = [], rc._model_rows, rc.PROBE_DIR
def slow(include_legacy=False): calls.append(1); time.sleep(.5); return [rc._facts(dict(value='claude/sonnet', label='sonnet', provider='claude', source='login'))]
rc._model_rows, rc.PROBE_DIR = slow, Path(tempfile.mkdtemp())
rc.forget_probes()
test_eq([r['name'] for r in catalog()], ['Claude Sonnet'])
test_eq((catalog(), len(calls))[1], 1)

A new process has no list in memory. It gets the last list from the disk at once, with no slow call, and a refresh starts in the background.

rc.forget_probes()   # the memory copy goes, as in a new process; the disk copy stays
t = time.time(); rows = catalog()
assert time.time() - t < .3 and [r['name'] for r in rows] == ['Claude Sonnet']
for _ in range(50):
    if len(calls) == 2: break
    time.sleep(.05)
test_eq(len(calls), 2)

When the look-up fails, the list has the three Claude tiers, each with a name.

def broken(include_legacy=False): raise OSError('no network')
rc._model_rows = broken; rc.forget_probes(disk=True)
test_eq([r['name'] for r in catalog()], ['Claude Sonnet', 'Claude Haiku', 'Claude Opus'])
rc._model_rows, rc.PROBE_DIR = real, real_dir; rc.forget_probes()

A local row gets “a local model:” before its name. A hosted row keeps the name of ramabana.

row = lambda v, local: dict(value=v, label=v, provider='mlx' if local else 'claude', source='test', name='Qwen 4B' if local else 'Claude Haiku', local=local)
real_am = rc.available_models
rc.available_models = lambda include_legacy=False, ttl=None: [row('qwen-4b', True), row('claude/haiku', False)]
try: test_eq([r['name'] for r in catalog()], ['a local model: Qwen 4B', 'Claude Haiku'])
finally: rc.available_models = real_am

One row for each model

Some models come through two providers. For example, Claude Haiku 5.5 comes through Claude Code and through GitHub Copilot, and GPT-6.1 Sol through Codex, GitHub Copilot and the OpenAI API. Each route has its own id, so the list showed the same name two or three times. one_each keeps one row for each name: the route of the first provider in ramabana’s READY_ORDER (subscriptions first, then keys). The row keeps the other ids in also. A saved id (keep) always stays, so the list shows the choice that you made.

---

source

one_each

def one_each(
    rows:list, keep:str=''
)->list:

One row of rows for each model name, in the first place of that name: the saved id keep, else the route that ramabana tries first. also has the other ids. An alias with a version beside it goes into the also of the newest version.

Claude Haiku 5.5 shows one time, on Claude Code. With the Copilot id saved, the Copilot row stays.

dup = [dict(value='copilot/claude-haiku-5.5', name='Claude Haiku 5.5', provider='copilot'), dict(value='claude/sonnet', name='Claude Sonnet', provider='claude'),
       dict(value='claude/claude-haiku-5-5', name='Claude Haiku 5.5', provider='claude')]
test_eq([(r['value'], r['also']) for r in one_each(dup)], [('claude/claude-haiku-5-5', ['copilot/claude-haiku-5.5']), ('claude/sonnet', [])])
test_eq(one_each(dup, keep='copilot/claude-haiku-5.5')[0]['value'], 'copilot/claude-haiku-5.5')

An alias such as claude/opus and its version, such as Claude Opus 5.5, are one model. The list shows only the version, with the alias in also. A saved alias stays.

al = [dict(value='claude/opus', name='Claude Opus', provider='claude'), dict(value='claude/claude-opus-4-1', name='Claude Opus 4.1', provider='claude'),
      dict(value='claude/claude-opus-5-5', name='Claude Opus 5.5', provider='claude'), dict(value='codex/gpt-5', name='GPT-5', provider='codex'),
      dict(value='codex/gpt-5-mini', name='GPT-5 Mini', provider='codex')]
test_eq([(r['name'], r['also']) for r in one_each(al)], [('Claude Opus 4.1', []), ('Claude Opus 5.5', ['claude/opus']), ('GPT-5', []), ('GPT-5 Mini', [])])
test_eq([r['name'] for r in one_each(al, keep='claude/opus')][:2], ['Claude Opus', 'Claude Opus 4.1'])

Router is the ask the app gives the tutor. Called as it is, it uses the default model. for_(task) gives the ask for one task: the model the settings name for it, else the default. for_(task, name, job) gives the ask for task on the model name, on the ramabana job job. The Test button of the Settings page uses the job test:<task>, so a test does not move the job of the task. When the default routes by task, as RouteAsk below does, for_ lets it make the ask. cap(task) gives the reply cap of the task, read on each call. warm starts the backend of each task in the background.

---

source

for_task

def for_task(
    ask, task:str
):

The ask for task when ask routes by task, else ask itself.

---

source

Router

def Router(
    dflt, models:NoneType=None, tokens:NoneType=None
):

The tutor’s model: a default ask, and one model for each task that the settings name.

A router with no settings uses the default for every task. A task that names a model gets that model.

dflt = Fake(dflt='default')
r = Router(dflt, models=lambda: {})
test_eq(r('x'), 'default')
test_is(r.for_('grade'), dflt)
named = Fake(dflt='named'); named.model = 'fast'
_asks['fast'] = named
r = Router(dflt, models=lambda: dict(grade='fast'))
test_eq(r.for_('grade')('x'), 'named')
test_eq(r.for_('question')('x'), 'default')
test_is(for_task(dflt, 'grade'), dflt)

Every call through ramabana

ramabana is the agent layer on top of the model layer. Its Routing sends each job to a model. When a cloud model fails or gives no answer, Routing puts it aside for five minutes and sends the job to the next route: other cloud models, then a small model on this Mac. Its backends stay open, so a call does not pay the start time of a new Chat.

ramabana_agent makes one ramabana Agent for the process, on the first call. The Agent gets only what Agent.oneshot needs:

• NullHost: no folders, and each file operation is refused.
• readonly=True, no sub-agents, no extensions, no skills, no lessons: no tool can write, run code or go on the network.
• Approvals that refuse when nobody answers, and no saved rules.
• cfg=None: no history, plan or run log, so the Agent writes no files.
• A short briefing, so that it does not read skills or project files.

The turn of the Agent is the default model, so a task whose model fails goes to the default model first.

---

source

ramabana_agent

def ramabana_agent(
    turn:str=None
):

The one ramabana Agent of this process, made on first use by ramabana’s model_agent on the turn dflt_model(turn): no folders, no tool that acts, no files.

Each task is a ramabana job with the same name. _point sets the model of the job before each call, only when the settings changed it. A model that ramabana put aside stays aside until its time is up.

ramabana_oneshot asks one question on the job of a task, through ramabana’s Agent.oneshot with wait=True and strict=True. With wait, a call waits for a backend that another call holds, on the first route and on each fallback. With strict, a call that no route answers raises ramabana’s OneshotEmpty, with the routes that ramabana put aside or the note of the route.

A failure can take some time: ramabana tries each route in turn, then the small model on this Mac. It puts a failed cloud model aside for five minutes.

A route can also raise ramabana’s OneshotTimeout when a call goes past its deadline. no_answer changes each of the two errors into one RuntimeError that names the task, the model and why, so each caller catches one kind of error.

prepare lets the caller change the text for each route, as ramabana’s Agent.oneshot does. The guide uses it for its guard: the guard then knows if the model of the route is on this Mac. HOSTED is the model that a guard takes when it does not know the route.

---

source

ramabana_oneshot

def ramabana_oneshot(
    job:str, name:str, prompt:str, sp:str='', max_tokens:int=4096, prepare:NoneType=None
)->str:

Ask prompt on the ramabana job job, pointed at the model name. Wait for a busy backend. Raise no_answer when no route answers. prepare(prompt, sp, backend) gives the text for each route, so a guard knows the model that gets it.

---

source

no_answer

def no_answer(
    job:str, name:str, e
)->RuntimeError:

The RuntimeError for ramabana’s OneshotEmpty or OneshotTimeout e on the job job: it names the task, the model name (else the route of e) and why.

ramabana_stream is the twin of ramabana_oneshot that gives the reply as the model sends it, through ramabana’s Agent.oneshot_stream, with the same wait and strict, and it raises the same errors. The guard (prepare) runs before the stream starts. With fallback=True, a route that cannot stream gives its whole reply as one chunk, so the caller does not need a second path.

---

source

ramabana_stream

def ramabana_stream(
    job:str, name:str, prompt:str, sp:str='', max_tokens:int=4096, prepare:NoneType=None
):

The reply of ramabana_oneshot, one chunk at a time as the model sends it. A route that cannot stream gives the whole reply as one chunk. Raise no_answer when no route answers.

RouteAsk is the default ask of the web app. via(task, name, max_tokens) gives the ask of one task, and Router.for_ uses it. A call with no task goes to the job ask:<model>. warm starts the backend of each task, with no model call. route_ask gives a RouteAsk: Leela imports this name.

---

source

route_ask

def route_ask(
    model:str=None
)->__main__.RouteAsk:

The default ask: a RouteAsk on model.

---

source

RouteAsk

def RouteAsk(
    model:str=None
):

The default ask: each call goes through the routing of ramabana, on the job of its task.

The worker agent

ramabana_worker makes a second ramabana Agent, for the gather agent of research. It is never the one Agent of ramabana_agent, so a long gather never holds a one-shot call. It runs on the default routing (Routing(turn=dflt_model())). Its host is the shalya host (default: a NullHost, with no folders). The caller names its tools in tools, then adds them with add_tool. ramabana’s only_tools offers the model only those tools, and refuses a tool with another name. ramabana’s memory and plan tools are held back. It has no approvals listener, so a tool that writes is refused. The caps stop a turn after max_calls tool calls, max_steps model steps, max_tokens tokens or max_cost dollars. A model such as Claude Code on a plan reports no cost. Then ramabana estimates the dollars from the tokens at the price of the model (ramabana.agent.token_rate), and keeps the estimate in estimated, so the dollar cap still holds.

---

source

ramabana_lead

def ramabana_lead(
    read_roots:tuple=(), tools:list=(), sp:str='You are the source gatherer of abhyasa, a tutor.', turn:str=None,
    minutes:float=15, max_calls:int=300, max_steps:int=80, max_cost:float=5.0, max_tokens:int=2000000, web:bool=True
)->tuple:

(agent, host): a new ramabana Agent made by ramabana’s mk_agent, with its own read and web tools (LEAD_TOOLS) and its sub-agents (delegate_search, at most 4 at once in the whole tree), on the turn dflt_model(turn). Its sub-agents run on ramabana’s subagent route. The folders read_roots are open for reads only; its one folder that it may write is a new empty one. It has no Python (python='off'), no vault, and no approvals listener, so each tool that writes is refused. The model sees only LEAD_TOOLS and the tools named in tools, when the caller adds them. The caps: max_calls tool calls in the whole tree, max_steps model steps, max_tokens tokens and max_cost dollars for a turn, and minutes for a sub-agent. With web=False it has no tool that reaches the web (WEB_TOOLS).

---

source

ramabana_worker

def ramabana_worker(
    tools:list=(), host:NoneType=None, sp:str='You are the source gatherer of abhyasa, a tutor.', turn:str=None,
    max_calls:int=300, max_steps:int=80, max_cost:float=5.0, max_tokens:int=2000000
):

A new ramabana Agent, made by ramabana’s model_agent on the turn dflt_model(turn), with the shalya host (default: no folders). The model sees only the tools named in tools, when the caller adds them. Never the agent of ramabana_agent. A turn stops at max_tokens tokens, and at max_cost dollars (estimated from the tokens when the model reports no cost).

The worker offers only the tool that the test names and adds. It is a new Agent each time, and never the shared one. The caps reach the Agent.

def look(q:str, # what to look for
         ) -> str:
    "Look for `q`."
    return q
w = ramabana_worker(['look'], max_calls=50, max_tokens=1000)
w.add_tool(look)
test_eq([t.__name__ for t in w.tools], ['look'])
assert w is not ramabana_worker() and w is not ramabana_agent()
test_eq((w.readonly, w.subagents, w.tool_budget, w.max_cost, w.max_tokens), (True, False, 50, WORKER_DOLLARS, 1000))

The worker refuses a tool that tools does not name. A worker with no names offers no tool, so ramabana’s own tools stay out.

def write(path:str, # the file
          ) -> str:
    "Write `path`."
    return path
test_fail(lambda: w.add_tool(write), contains='not in only_tools')
test_eq(ramabana_worker().tools, [])

When the model reports no cost, ramabana estimates the dollars from the tokens with its own price table (est_rate is None), and keeps them in estimated. A new worker has estimated nothing, and has the default token cap.

w = ramabana_worker()
test_eq((w.est_rate, w.estimated, w.max_tokens), (None, 0., WORKER_TOKENS))

The lead of research is made by ramabana’s mk_agent. It offers ramabana’s read and web tools and delegate_search, and the tool that the test adds. It has no tool that writes a file or runs a command. The folder of the learner is open for reads only.

import tempfile, shutil

ro = tempfile.mkdtemp()
a, h = ramabana_lead([ro], ['look'], minutes=2)
a.add_tool(look)
names = {t.__name__ for t in a.tools}
assert {'view_file', 'read_url', 'web_search', 'delegate_search', 'look'} <= names, names
assert not names & {'run_shell', 'run_python', 'create_file', 'replace_text', 'git_commit', 'add_root'}, names
test_eq((h.read_roots, a.subagent_timeout, a.subagent_writes), ([str(Path(ro).resolve())], 120, False))
a.close(); shutil.rmtree(h.scratch, ignore_errors=True)
a, h = ramabana_lead([ro], ['look'], web=False)
assert not {t.__name__ for t in a.tools} & set(WEB_TOOLS)
a.close(); shutil.rmtree(h.scratch, ignore_errors=True)

Tests of the route

These tests use a ramabana Agent with fake backends from ramabana.testing, so no model runs. using(a) puts that Agent in place of the real one for one test. CI does not install the models extra, so these tests run only when ramabana is installed.

from contextlib import contextmanager
from collections import Counter
from concurrent.futures import ThreadPoolExecutor

from ramabana.core import resolve
from ramabana.testing import fake_agent, FakeBackend

@contextmanager
def using(a):
    "Use the `Agent` `a` in place of the real one, for one test."
    _ra['agent'] = a
    try: yield a
    finally: _ra.pop('agent', None)

Each task goes to the ramabana job with the same name. The task that the settings name gets its model, the other tasks get the default. Each call asks for the reply cap of its task, TOKENS[task], so a long question with its rubric is not cut.

a, be = fake_agent()
jobs, one = [], a.oneshot
a.oneshot = lambda p, sp='', job='oneshot', max_tokens=None, **kw: jobs.append((job, max_tokens)) or one(p, sp, job, max_tokens, **kw)
with using(a):
    r = Router(RouteAsk('claude/sonnet'), models=lambda: dict(grade='claude/haiku'), tokens=lambda: {})
    test_eq(r.for_('grade')('hi'), 'ONESHOT:hi')
    test_eq(r.for_('question').model, 'claude/sonnet')
    r.for_('question')('hi')
test_eq(jobs, [('grade', TOKENS['grade']), ('question', TOKENS['question'])])
test_eq((a.routing.policy['grade'], a.routing.policy['question']), ('claude/haiku', 'claude/sonnet'))

A change in the settings has an effect on the next call, with no restart. An empty choice goes back to the default.

a, be = fake_agent()
ms = dict(grade='claude/haiku')
with using(a):
    r = Router(RouteAsk('claude/sonnet'), models=lambda: ms)
    r.for_('grade')('hi'); test_eq(a.routing.policy['grade'], 'claude/haiku')
    ms['grade'] = 'claude/opus'; r.for_('grade')('hi'); test_eq(a.routing.policy['grade'], 'claude/opus')
    ms['grade'] = ''; r.for_('grade')('hi'); test_eq(a.routing.policy['grade'], 'claude/sonnet')

prepare gets each route. Here it writes the name of the model before the prompt, so the reply shows what the route got:

a, be = fake_agent()
with using(a): test_eq(RouteAsk('claude/sonnet').via('hint', prepare=lambda p, sp, b: (f'{b.spec.name} {p}', sp))('hi'), 'ONESHOT:fake hi')

Slow is a fake backend that takes a moment to answer. It sets inside when a call starts.

class Slow(FakeBackend):
    "A fake backend that answers after 0.2 s and sets `inside` when a call starts."
    def __init__(self, *a, **kw): super().__init__(*a, **kw); self.inside = threading.Event()
    def _oneshot(self, prompt, sp, max_tokens): self.inside.set(); time.sleep(0.2); return f'ONESHOT:{prompt}'

Without wait, ramabana’s Agent.oneshot does not wait for a busy backend: a second call at the same time gets ’’.

a, _ = fake_agent(); slow = Slow(); a._be_or_none = lambda job='turn': slow
with ThreadPoolExecutor(2) as ex:
    f1 = ex.submit(a.oneshot, 'one'); slow.inside.wait(); f2 = ex.submit(a.oneshot, 'two')
test_eq((f1.result(), f2.result()), ('ONESHOT:one', ''))

Through the route (wait=True), the second call waits for the first, and both calls get an answer.

slow.inside.clear()
with using(a), ThreadPoolExecutor(2) as ex:
    ask = RouteAsk('claude/sonnet').via('question')
    f1 = ex.submit(ask, 'one'); slow.inside.wait(); f2 = ex.submit(ask, 'two')
    test_eq((f1.result(), f2.result()), ('ONESHOT:one', 'ONESHOT:two'))

When no route gives an answer, the call raises. The error names the task, the model and the note of ramabana, so ask_json never gets ’’.

class Empty(FakeBackend):
    "A fake cloud backend that answers nothing."
    def _oneshot(self, prompt, sp, max_tokens): return ''
a, _ = fake_agent(); empty = Empty(resolve('claude/haiku')); a._be_or_none = lambda job='turn': empty
with using(a):
    ask = Router(RouteAsk('claude/sonnet'), models=lambda: dict(grade='claude/haiku')).for_('grade')
    test_fail(lambda: ask('hi'), contains='grade task')
    try: ask('hi')
    except RuntimeError as e: assert 'claude/haiku' in str(e) and 'answered nothing' in str(e), e

A route that goes past its deadline raises ramabana’s OneshotTimeout. The call raises the same RuntimeError, with the model and the seconds, and keeps the first error as its cause.

from ramabana.runtime import OneshotTimeout

a, _ = fake_agent()
def late(*a, **kw): raise OneshotTimeout('claude/haiku', 31.6, True)
a.oneshot = late
with using(a):
    try: RouteAsk('claude/haiku').via('grade')('hi'); assert False, 'no error'
    except RuntimeError as e:
        test_eq(str(e), 'The grade task got no answer from claude/haiku through ramabana: claude/haiku did not answer in 32 seconds.')
        assert isinstance(e.__cause__, OneshotTimeout)

Tests of the stream

Chunks is a fake backend that can stream: it sends its reply three letters at a time. ask.stream gives the chunks in order, and their join is the reply. The guard of prepare sees the route before the stream starts.

class Chunks(FakeBackend):
    "A fake backend that streams `ONESHOT:<prompt>` in chunks of 3 letters."
    supports_stream = True
    def _oneshot_stream(self, prompt, sp, max_tokens): return (f'ONESHOT:{prompt}'[i:i+3] for i in range(0, len(f'ONESHOT:{prompt}'), 3))
a, _ = fake_agent(); ch = Chunks(resolve('claude/haiku')); a._be_or_none = lambda job='turn': ch
with using(a): got = list(RouteAsk('claude/haiku').via('hint', prepare=lambda p, sp, b: (f'{b.spec.name} {p}', sp)).stream('hi'))
test_eq(''.join(got), 'ONESHOT:claude/haiku hi')
assert len(got) > 1

A backend that cannot stream (the default of FakeBackend) gives the whole reply of the blocking call as one chunk:

a, be = fake_agent()
with using(a): test_eq(list(RouteAsk('claude/sonnet').via('hint').stream('hi')), ['ONESHOT:hi'])

The stream lets go of its backend at the end, also when it stops early, so the next call on that backend does not wait:

a, _ = fake_agent(); a._be_or_none = lambda job='turn': ch
with using(a), ThreadPoolExecutor(1) as ex:
    list(RouteAsk('claude/haiku').via('hint').stream('one'))
    test_eq(ex.submit(RouteAsk('claude/haiku').via('hint'), 'two').result(timeout=2), 'ONESHOT:two')
    it = RouteAsk('claude/haiku').via('hint').stream('three'); next(it); it.close()   # a stream that stops early lets go too
    test_eq(ex.submit(RouteAsk('claude/haiku').via('hint'), 'four').result(timeout=2), 'ONESHOT:four')

A stream with no text raises the same error as a blocking call, and a timeout raises the same RuntimeError:

a, _ = fake_agent(); empty = Empty(resolve('claude/haiku')); empty.supports_stream = True
empty._oneshot_stream = lambda prompt, sp, max_tokens: (c for c in ())
a._be_or_none = lambda job='turn': empty
with using(a): test_fail(lambda: list(RouteAsk('claude/haiku').via('grade').stream('hi')), contains='grade task')
def late(*a, **kw): raise OneshotTimeout('claude/haiku', 31.6, True); yield
a.oneshot_stream = late
with using(a): test_fail(lambda: list(RouteAsk('claude/haiku').via('grade').stream('hi')), contains='did not answer in 32 seconds')

The next tests need ramabana’s real Routing, so that a failed model goes to its next route. routed_agent gives a fake Agent with the real Routing on two fake models: fake-main, the default, and fake-quick. They are hosted models with a fake key, so Routing takes them on any machine. make(spec) gives the fake backend of each model, and bes keeps them by name.

from ramabana.core import register_model, Routing

os.environ['ABHYASA_FAKE_KEY'] = 'not a real key'
for n in ('fake-main', 'fake-quick'): register_model(n, f'fakevendor/{n}', runtime='remote', api_key_env='ABHYASA_FAKE_KEY')

def routed_agent(make):
    "A fake `Agent` on the real `Routing`, with the turn on `fake-main`. Returns the `Agent` and the backends by model name."
    a, _ = fake_agent(routing=Routing(turn='fake-main'))
    del a._be_or_none   # the real one, which asks `Routing`
    bes = {}
    def be(job='turn'):
        s = a.routing.spec(job)
        if s.name not in bes: bes[s.name] = make(s)
        return bes[s.name]
    a._be = be
    return a, bes

A fallback waits for the backend that it goes to. Here fake-quick answers nothing, so the grade call goes to fake-main. A question call on fake-main starts while that fallback runs. It waits for the backend, and both calls get an answer.

a, bes = routed_agent(lambda s: Slow(s) if s.name != 'fake-quick' else Empty(s))
with using(a), ThreadPoolExecutor(2) as ex:
    g = ex.submit(Router(RouteAsk('fake-main'), models=lambda: dict(grade='fake-quick'), tokens=lambda: {}).for_('grade'), 'g')
    while 'fake-main' not in bes: time.sleep(0.01)
    bes['fake-main'].inside.wait()
    q = ex.submit(RouteAsk('fake-main').via('question'), 'q')
    test_eq((g.result(), q.result()), ('ONESHOT:g', 'ONESHOT:q'))

When no route answers, each route is tried one time. The last route is the small model on this Mac, gemma-e4b.

calls = Counter()
class Counted(Empty):
    "A fake backend that answers nothing and counts its calls."
    def _oneshot(self, prompt, sp, max_tokens): calls[self.spec.name] += 1; return ''
a, bes = routed_agent(Counted)
with using(a): test_fail(lambda: RouteAsk('fake-main').via('grade', 'fake-quick')('hi'), contains='grade task')
test_eq(calls['gemma-e4b'], 1)
test_eq(set(calls.values()), {1})

warm asks ramabana for the backend of each task in the background, and sends nothing to the model. It gives the thread. The jobs of WARM start too: the guide job, on the model of the screen guide (the model for hints when it has none). warm(['guide']) starts only the guide.

a, be = fake_agent(); seen = []
a._be_or_none = lambda job='turn': seen.append(job) or be
with using(a):
    Router(RouteAsk('claude/sonnet'), models=lambda: {}).warm().join()
test_eq(seen, [*TASKS, 'guide'])
test_eq(be.sent, [])
seen.clear()
with using(a): Router(RouteAsk('claude/sonnet'), models=lambda: dict(hint='claude/haiku')).warm(['guide']).join()
test_eq((seen, a.routing.policy['guide']), (['guide'], 'claude/haiku'))
with using(a): Router(RouteAsk('claude/sonnet'), models=lambda: dict(hint='claude/haiku', guide='claude/opus')).warm(['guide']).join()
test_eq(a.routing.policy['guide'], 'claude/opus')   # a model of its own wins

route_ask gives a RouteAsk.

assert isinstance(route_ask('claude/haiku'), RouteAsk) and route_ask('claude/haiku').model == 'claude/haiku'

The reply cap of each task

A reply cap comes from the settings table (tokens.<task>), from 1 to TOKENS_MAX. A cap that you typed in the file as text, such as "900", works too.

from abhyasa.learner import update_settings, val

update_settings('tokens', hint=' 900 ')
test_eq(val('tokens.hint'), 900)
update_settings('tokens', hint=str(TOKENS_MAX + 1))
test_eq(val('tokens.hint'), TOKENS_MAX)   # clamped to the top
update_settings('tokens', hint=None)

Router.cap gives the saved cap of a task, else the default of the task. A bad value in the settings file gives the default, so a call never fails because of it.

caps = dict(hint=300)
r = Router(Fake(), models=lambda: {}, tokens=lambda: caps)
test_eq(r.cap('hint'), 300)
test_eq(r.cap('grade'), TOKENS['grade'])
caps['hint'] = 'lots'; test_eq(r.cap('hint'), TOKENS['hint'])
caps['hint'] = 99999; test_eq(r.cap('hint'), TOKENS_MAX)          # out of range: the most

A task after extract (plan, pick, note, profile) borrows the model, the reply cap and the job of an older task until it has a model of its own.

class Via:
    model = 'base'
    def __call__(self, p, sp=''): return ''
    def via(self, job, name='', cap=None): return (job, name, cap)
ms = dict(extract='x/extract', hint='x/hint')
r = Router(Via(), models=lambda: ms, tokens=lambda: {})
test_eq((r.for_('plan'), r.for_('note')), (('extract', 'x/extract', 4096), ('hint', 'x/hint', 512)))
ms['plan'] = 'x/plan'
test_eq(r.for_('plan'), ('plan', 'x/plan', 4096))

The cap reaches Agent.oneshot, and a change has an effect on the next call.

a, be = fake_agent()
jobs, one = [], a.oneshot
a.oneshot = lambda p, sp='', job='oneshot', max_tokens=None, **kw: jobs.append((job, max_tokens)) or one(p, sp, job, max_tokens, **kw)
caps = dict(grade=777)
with using(a):
    r = Router(RouteAsk('claude/sonnet'), models=lambda: {}, tokens=lambda: caps)
    r.for_('grade')('hi'); caps['grade'] = ''; r.for_('grade')('hi')
test_eq(jobs, [('grade', 777), ('grade', TOKENS['grade'])])

The Test button of the Settings page asks on its own job, test:<task>, with the cap of the task. The job of the task stays where the settings put it.

a, be = fake_agent()
jobs, one = [], a.oneshot
a.oneshot = lambda p, sp='', job='oneshot', max_tokens=None, **kw: jobs.append((job, max_tokens)) or one(p, sp, job, max_tokens, **kw)
with using(a):
    r = Router(RouteAsk('claude/sonnet'), models=lambda: dict(grade='claude/haiku'), tokens=lambda: {})
    r.for_('grade')('hi')
    r.for_('grade', 'claude/opus', job='test:grade')('hi')
test_eq(jobs, [('grade', TOKENS['grade']), ('test:grade', TOKENS['grade'])])
test_eq((a.routing.policy['grade'], a.routing.policy['test:grade']), ('claude/haiku', 'claude/opus'))

The guard

Text that leaves the Mac passes the PII guard first.

Text that leaves the Mac passes ramabana’s guard (ramabana.guard) first. guard(s, model) is for a model call: the page context and the question, and the prompt of a vision call. It is outgoing for the model of the route. It always masks API keys, tokens and private keys as [SECRET]. For personal data it follows ramabana’s setting cli.pii, but only for a hosted model: redact masks it, refuse withholds the text (Withheld), and off lets it through. A model on this Mac gets the personal data. When the route is not known, guard takes a hosted model (HOSTED).

safe_query(q) is for a web query. It is ramabana’s query: a query with a secret or with personal data does not go, with any setting.

The guard reads text only. A figure image goes to the vision model as it is: no text guard can read pixels.

---

source

guarded

def guarded(
    p, sp, b
)->tuple:

The prepare of a routed call: the prompt p and the system prompt sp through guard for the model of the backend b.

---

source

safe_query

def safe_query(
    q:str
)->str:

The web query q as it is, through ramabana’s query; Withheld when it holds a secret or personal data.

---

source

guard

def guard(
    s, model:NoneType=None, mode:str=None
)->str:

s as it may go to model (a ramabana ModelSpec; default HOSTED), through ramabana’s outgoing, with the PII policy mode (default cli.pii).

---

source

Withheld

def Withheld(
    *args, **kwargs
):

The guard withheld the text: it holds personal data and ramabana’s cli.pii is refuse, or it is a web query with a secret or personal data.

A key never gets through. Plain text stays as it is:

test_eq(guard('key sk-test1234567890abcdefXYZ here'), 'key [SECRET] here')
test_eq(guard(r'The share \bar\alpha_t is left.'), r'The share \bar\alpha_t is left.')

With redact, a hosted model gets [EMAIL], and a model on this Mac gets the address. A key is masked for both:

local = AttrDict(name='mlx/gemma', local=True)
test_eq(guard('mail ada@example.com', mode='redact'), 'mail [EMAIL]')
test_eq(guard('mail ada@example.com sk-test1234567890abcdefXYZ', local, 'redact'), 'mail ada@example.com [SECRET]')

With refuse, text with personal data for a hosted model is withheld. A web query with a secret or personal data never goes:

test_fail(lambda: guard('mail ada@example.com', mode='refuse'), exc=Withheld)
test_fail(lambda: safe_query('alpha bar sk-test1234567890abcdefXYZ'), exc=Withheld)
test_fail(lambda: safe_query('alpha bar ada@example.com'), exc=Withheld)
test_eq(safe_query('alpha bar diffusion'), 'alpha bar diffusion')

Speech

Speech to text on this machine, for think-aloud answers, and a natural voice that reads the guide aloud.

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.

The module also speaks. speak turns one line of the screen guide into WAV audio, with the start time of each word, so the board can light each word as it is said.

These are the imports of the module.

Engines

There are two engines. mlx runs on this machine. openai sends the audio to OpenAI.

• mlx runs any speech to text model that mlx-audio can load. The default is Qwen3-ASR 0.6B (8 bit, 1.0 GB). Whisper large-v3-turbo is the second model on the Settings page, and the search finds the others. Install mlx-audio 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. Qwen3-ASR reads it as hot words, Whisper as its initial prompt, and OpenAI 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
mlx-community/Qwen3-ASR-0.6B-8bit	3.8 s	0.14 s	6%	13% / 0%	5 / 9	16%	2.0 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/Qwen3-ASR-1.7B-8bit	3.6 s	0.25 s	6%	6% / 0%	7 / 9	14%	3.6 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/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

• “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.
• “Sanskrit CER” is the character error rate on a 9-second Nyāya sūtra (from vachaka), after the vachaka normalization.
• Moonshine gave no text for the 24-second clip A. It is good only for short segments.
• For Sanskrit speech, use Qwen3-ASR or Whisper with the hint, or vachaka.

Qwen3-ASR 0.6B is the default. With the hint, it has no errors on clip B and it writes all 9 terms. One 5-second segment takes 0.14 s. Before 0.0.16, the default was Gemma 4 E4B on LiteRT. It had the same score, but it was 5 times slower, and its model was 3.7 GB.

DFLT_STT is the default model of the mlx engine. The other constants are the named models. ALIASES keeps the old engine names: whisper and parakeet are the mlx engine with a fixed model. gemma was the default engine before 0.0.16, so it becomes the mlx engine with the default model. POCKET is the Hugging Face repo of the local voice (see “A natural voice” below), and POCKET_FILES are the files of its English model and of its default voice. The repo has many languages, so abhyasa gets only these files. VOICE_FILES gives the files of the local voice.

Settings

The settings are in settings.json in cfg_dir(). The settings module keeps the table of values and the file; this module uses its stt section. The web app changes the settings, and the next segment uses the change. You do not have to restart.

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

test_eq({k: settings()['stt'][k] for k in ('engine', 'model', 'segment')}, dict(engine='mlx', model='', segment=True))
update_settings('stt', segment=False)
test_eq(val('stt.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 (mlx), an old name (whisper, gemma), or an engine and a model (mlx:mlx-community/whisper-large-v3-turbo-asr-fp16). An old settings file with gemma and a LiteRT model gets the default model. Nothing writes the file again.

---

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 Qwen3-ASR by default. An old gemma setting gives the same, also with its old model. The environment variable has priority.

test_eq(current(), ('mlx', DFLT_STT))
test_eq(resolve('gemma', 'litert-community/gemma-4-E4B-it-litert-lm'), ('mlx', DFLT_STT))
test_eq(resolve('parakeet'), ('mlx', DFLT_PARAKEET))
test_eq(resolve('mlx:mlx-community/whisper-tiny-asr-fp16'), ('mlx', 'mlx-community/whisper-tiny-asr-fp16'))
test_fail(lambda: resolve('espeak'), contains='unknown')
os.environ['ABHYASA_STT'] = 'whisper'
test_eq(current(), ('mlx', DFLT_WHISPER))
os.environ['ABHYASA_STT'] = 'gemma'
test_eq(current(), ('mlx', DFLT_STT))
del os.environ['ABHYASA_STT']

One model in memory

resident keeps one model loaded in each slot: stt for speech to text, tts for the speaking voice. When the app asks for a different model in a slot, it frees the old model of that slot first. Two large models for the same job in memory at the same time can make the Mac slow.

---

source

resident

def resident(
    key, load, close:NoneType=None, slot:str='stt'
):

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

---

source

unload

def unload(
    slot:str='stt'
):

Free the loaded model of slot: stt (speech to text) or tts (the speaking voice).

A new key frees the old model. The same key does not load again. A model in the tts slot does not free the model in the stt slot.

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'])
test_eq(resident('k', lambda: 'K', slot='tts'), 'K')
test_eq((resident('b', lambda: 'other'), freed), ('B', ['A']))
unload(); unload('tts')

Each engine is a function that takes a WAV path, a hint and a model, and returns text.

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. ramabana says where its file is ($RAMABANA_CONFIG, else ~/.config/ramabana/keys.env). abhyasa reads both files with ramabana’s read_key. It refuses a file that other users can read, it runs no !command value, and it sets nothing in the environment. abhyasa only reads the ramabana file. set_openai_key writes the abhyasa file with save_key (atomic, mode 0600). masked shows only the last 4 characters.

---

source

masked

def masked(
    key:str
)->str:

---

source

set_openai_key

def set_openai_key(
    key:str=''
):

Save key in abhyasa’s keys.env (mode 0600); an empty key clears 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. This test points ramabana to a temp folder.

from ramabana.setup import keys_path

os.environ['RAMABANA_CONFIG'] = str(Path(tempfile.mkdtemp())/'config.toml')
test_eq(openai_key(), ('', ''))
keys_path().write_text('export OPENAI_API_KEY="sk-ramabana-0000"\n'); keys_path().chmod(0o600)
test_eq(openai_key(), ('sk-ramabana-0000', str(keys_path())))
test_eq(os.getenv('OPENAI_API_KEY'), None)
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')

A !command value in a key file is not run, so it gives no key:

set_openai_key('')
keys_path().write_text("OPENAI_API_KEY='!echo sk-from-a-command'\n"); keys_path().chmod(0o600)
test_eq(openai_key(), ('', ''))
set_openai_key('sk-test-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('')
keys_path().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 the Settings page shows first: the two local models that we measured (see the table above), and two OpenAI models. cached tells if a model is in the Hugging Face cache. It does not use the network. The repo of the local voice (POCKET, below) has many languages, so for that repo cached looks for its English model only.

---

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; for the local voice, when its English model is there.

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. model_size gives the size of a download: for the local voice, only its English files.

---

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 that a download of model gets from 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', POCKET)],
        [True, True, False, False, False])

search_models gives only the speech to text models, and model_size counts only the English files of the local voice. This test uses a fake Hugging Face API, so it does not use the network.

class _Api:
    def list_models(self, **kw): return [AttrDict(id=i, downloads=n) for i, n in (('mlx-community/csm-1b', 9), ('mlx-community/whisper-tiny-asr-fp16', 8))]
    def model_info(self, m, files_metadata=True): return AttrDict(siblings=[AttrDict(rfilename=f, size=10) for f in (*POCKET_FILES, 'languages/french/model.safetensors')])
test_eq(search_models('', api=_Api()), [dict(model='mlx-community/whisper-tiny-asr-fp16', downloads=8, size=40)])
test_eq(model_size(POCKET, _Api()), 30)

fetch downloads a model and notes it in the settings. The local voice gets only its English model and its default voice: it loads them, so they are in memory for the first line. 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_WHISPER), contains='stays')

A natural voice

The board of the screen guide says one line for each step. The browser can say it with its own voice (speechSynthesis), but that voice sounds like a machine. speak gives a natural voice. There are two engines:

• pocket runs Kyutai’s Pocket TTS on this Mac, with the pocket-tts-mlx package. The download is 225 MB. It has 21 English voices, and it makes a line of 4 seconds in less than 1 second. Install it with pip install abhyasa[voice].
• openai sends the line to the OpenAI speech API. It is off until you set a key. Each line passes ramabana’s guard first.

browser is the third choice of the setting voice.out. Then the page speaks, and this module makes no audio. The default is pocket. Until its model is on this Mac, the browser speaks: Speaker refuses the line with the word “browser”, and the page uses its own voice.

Licences: the pocket-tts-mlx code is MIT. The Pocket TTS weights (kyutai/pocket-tts-without-voice-cloning, by Kyutai) are CC-BY-4.0.

These are the defaults. POCKET_VOICES are the 21 English voices of Pocket TTS. caro_davy is first: the owner chose it by ear.

Not each OpenAI voice works with each model. VOICES_OF gives the voices of each speech model, from OpenAI’s guide (https://developers.openai.com/api/docs/guides/text-to-speech, read on 2 Nov 2026): gpt-4o-mini-tts has all 13 voices, and tts-1 and tts-1-hd have a smaller set of 9. Thus ballad, verse, marin and cedar work only with gpt-4o-mini-tts. The first voice of each model is its default. Change this table, and only this table, when OpenAI changes its voices.

openai_voices gives the voices of a model. A model that is not in the table gets all the voices: then OpenAI’s error says what is wrong. fit_voice keeps a voice that the model has, and else gives the first voice of the model.

---

source

fit_voice

def fit_voice(
    model:str, voice:str
)->str:

voice when model has it, else the default voice of model.

---

source

openai_voices

def openai_voices(
    model:str
)->tuple:

tts-1 has no ballad, so it gets coral. gpt-4o-mini-tts keeps ballad. An unknown model keeps its voice.

test_eq(openai_voices('tts-1-hd'), TTS1_VOICES)
test_eq((fit_voice('tts-1', 'ballad'), fit_voice('tts-1', 'nova'), fit_voice(DFLT_SPEECH, 'ballad'), fit_voice('tts-9', 'ballad')), ('coral', 'nova', 'ballad', 'ballad'))

out_resolve fills in the default model and voice of an engine. Before 0.0.16 the local voices were kokoro and hf. Both become pocket. A local voice that the engine does not have becomes its default voice, and the model of a local engine is always its own repo. Thus an old settings file works, and nothing writes it again. For OpenAI, a voice that the model does not have becomes the default voice of the model (fit_voice). out_current gives the voice of the settings, with its speed. out_where says where the voice runs.

---

source

out_where

def out_where(
    engine:str
)->str:

---

source

out_current

def out_current()->tuple:

The voice in use: (engine, model, voice, speed) from the settings.

---

source

out_resolve

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

The engine, model and voice for a name; an empty model or voice gets the default of the engine. browser has no model.

The settings give Pocket with caro_davy. An old Hugging Face setting or an old Kokoro setting gives the same. The old model and voice stay in the file. OpenAI keeps its voice when its model has it: tts-1-hd with ballad speaks with coral. An unknown engine is an error.

test_eq(out_current(), ('pocket', POCKET, 'caro_davy', 1.))
test_eq(out_resolve('pocket', '', 'mary'), ('pocket', POCKET, 'mary'))
for old in (dict(out='kokoro', out_model='mlx-community/Kokoro-82M-bf16', out_voice='bm_george'), dict(out='hf', out_model='mlx-community/csm-1b', out_voice='')):
    update_settings('voice', **old)
    test_eq(out_current()[:3], ('pocket', POCKET, 'caro_davy'))
test_eq(out_resolve('kokoro', '', 'bm_george'), ('pocket', POCKET, 'caro_davy'))
update_settings('voice', out='pocket', out_model='', out_voice='')
test_eq(out_resolve('openai', '', 'coral'), ('openai', DFLT_SPEECH, 'coral'))
test_eq(out_resolve('openai', 'tts-1-hd', 'ballad'), ('openai', 'tts-1-hd', 'coral'))
test_fail(lambda: out_resolve('espeak'), contains='unknown')

The time of each word

The board lights each word when the voice says it. Pocket TTS and OpenAI give no word times. Thus by_letters shares the length of the audio among the words by their letters.

---

source

by_letters

def by_letters(
    text:str, secs:float
)->list:

A start time in seconds for each word of text in audio of secs seconds: each word takes a share by its letters.

Two words of 2 and 6 letters in 1 second: the first word starts at 0, the second after 3/10 of the second.

test_eq(by_letters('is hidden', 1.), [0., .3])
test_eq(by_letters('', 1.), [])

WAV

to_wav makes WAV bytes (mono, 16 bit) from the float samples of a model. clean_wav writes the header of a WAV again with its true length. OpenAI streams its WAV, so its header can give a wrong length.

---

source

clean_wav

def clean_wav(
    audio:bytes
)->bytes:

audio with a header that gives the true length of its samples.

---

source

to_wav

def to_wav(
    samples, rate:int
)->bytes:

WAV bytes (mono, 16 bit) for float samples from -1 to 1.

One second of samples at 24 kHz is one second of WAV. A header that says the data is very long becomes a header with the true length.

test_eq(wav_seconds(to_wav([0.] * 24000, 24000)), 1.)
import struct
w = to_wav([0.] * 2400, 24000)
bad = w[:4] + struct.pack('<I', 0xFFFFFFFF) + w[8:40] + struct.pack('<I', 0xFFFFFFFF) + w[44:]   # as a stream: the RIFF and data sizes are unknown
test_eq(wav_seconds(clean_wav(bad)), .1)

A gap at the end

A local voice ends its line with no gap. On a Bluetooth or AirPlay output the last part of the sound is still in the buffer when the page stops the audio at its ended event, so the line seems to stop too soon. pad_wav fades out the last PAD.fade seconds and adds PAD.tail seconds of silence after the line, and PAD.lead seconds before it, for an output that wakes up late. It also gives where the sound starts and ends in the new audio: Pocket TTS makes about half a second of quiet audio before and after the words, and the word times must fall on the words, not on the quiet parts.

---

source

pad_wav

def pad_wav(
    audio:bytes, lead:float=0.05, tail:float=0.35, fade:float=0.03
)->tuple:

audio with a fade-out and silence around it, and the start and end of its sound in seconds: (wav, (start, end)).

One second of sound gets 0.4 s more. The tail is silent, and the last sample of the sound is faded to zero. The sound starts after the lead, and ends where it ended.

import numpy as np

w, (t0, t1) = pad_wav(to_wav(np.full(24000, .5, np.float32), 24000))
test_eq(round(wav_seconds(w), 3), 1. + PAD.lead + PAD.tail)
with wave.open(__import__('io').BytesIO(w)) as f: s = np.frombuffer(f.readframes(f.getnframes()), '<i2')
test_eq((abs(s[-int(24000 * PAD.tail):]).max(), s[int(24000 * PAD.lead) + 23999], abs(s[:int(24000 * PAD.lead)]).max()), (0, 0, 0))
test_eq((t0, t1 <= PAD.lead + 1.), (PAD.lead, True))

Pocket TTS

_pocket loads the model one time into the tts slot of resident, and keeps the state of each voice that it used. _pocket_say makes the audio of one line. normalize=True makes the voices equally loud: some, such as stuart_bell and peter_yearsley, are quiet. Pocket TTS has no speed, so the speed setting is only for OpenAI. When pocket-tts-mlx is not installed, the error (POCKET_MISSING) says how to install it, and that the browser speaks until then.

OpenAI

The openai engine posts the line to the OpenAI speech API and asks for WAV. instructions can set the tone (“calm, like a teacher”); only the gpt-4o models read it. The owner’s rule is that each text that goes to a hosted model passes ramabana’s guard (guard, with ramabana.guard.outgoing). It masks a secret and refuses personal data when the privacy setting says so.

When the call fails, OpenAI sends JSON with error.message. openai_why makes it one plain line that says what to do: a voice that the model does not have, a model that OpenAI does not know, a bad key (401), no credit or too many requests (429), or an error on OpenAI’s side. The engine raises SpeakError with this line, so the learner never sees the raw text of the HTTP client.

---

source

openai_why

def openai_why(
    r, model:str='', voice:str=''
)->str:

One plain line for the failed OpenAI speech call r, from the error.message of its JSON body.

---

source

SpeakError

def SpeakError(
    *args, **kwargs
):

A natural voice did not say a line. The text is one plain line for the learner.

Maths as words

A voice reads $ym$ as “yum”, and \alpha as “backslash alpha”. spoken puts the maths of a line into words before any voice says it: $…$, $$…$$, \(…\), \[…\], a short name in back quotes, and a bare W_Q (one letter with a short subscript). tex_words reads a run of single letters letter by letter, a Greek letter by its name, the common operators, ^2 as “squared”, _m as “sub m”, \frac, \sqrt, \bar and \hat. Each table holds one kind of name. Prose does not change: “ym” with no $ stays “ym”. No small library on PyPI does this (kosha found none in the dependencies), so it is one function with tables.

The text on the board stays as it is. The board lights one part at a time: a $…$ is one part, and each other word is one part (SAY_PART, the same rule as rupaka’s board). Thus spoken gives, for each spoken word, the part that it reads, and part_times gives the start of each part from the start of its first spoken word.

---

source

spoken

def spoken(
    text:str
)->tuple:

The words to say for a line with maths, and for each spoken word the index of the part of the line that it reads.

---

source

part_times

def part_times(
    times:list, at:list, n:int
)->list:

The start of each of the n parts of a line, from the start of each spoken word and the part that it reads.

---

source

tex_words

def tex_words(
    tex:str
)->str:

TeX maths as words to say: y_m is ‘y sub m’, \alpha^2 is ‘alpha squared’, \frac{a}{b} is ‘a over b’.

The examples of the owner. Prose with “rhythm” and a bare “ym” stays as it is, and each prose word is its own part.

test_eq([spoken(s)[0] for s in ('$y_m$', '$ym$', r'$\alpha^2$', r'$\frac{a}{b}$')], ['y sub m', 'y m', 'alpha squared', 'a over b'])
test_eq(spoken('The rhythm of ym stays.'), ('The rhythm of ym stays.', [0, 1, 2, 3, 4]))
test_eq(spoken(r'Here $\hat{y} = \sqrt{x^2 + 1}$, and W_Q is a matrix.')[0], 'Here y hat equals the square root of x squared plus 1, and W sub Q is a matrix.')
test_eq(spoken(r'Use \(\bar{x}_i\) and `ym` now.')[0], 'Use x bar sub i and y m now.')
test_eq(spoken(r'$$\sum_{i=1}^{n} x_i^{-1}$$')[0], 'the sum from i equals 1 to n of x sub i inverse')
test_eq(spoken(r'$\mathbf{W}^T x$ and $\Sigma$')[0], 'W transpose x and capital sigma')

$q = xW_Q$ is one part on the board, so its five spoken words all read part 1. A part with no word of its own starts with the part before it.

s, at = spoken('So $q = xW_Q$ here.')
test_eq((s, at), ('So q equals x W sub Q here.', [0, 1, 1, 1, 1, 1, 1, 2]))
test_eq(part_times([0., .1, .2, .3, .4, .5, .6, .7], at, 3), [0., .1, .7])
test_eq(part_times([0., .5], [0, 2], 3), [0., 0., .5])

speak

speak gives the WAV bytes of one line and the start time of each of its words. Without an engine, it uses the voice of the settings. For the browser voice, it refuses: the page speaks then. The voice says the words of spoken, and the times are for the parts of the line as the board shows them. A local line gets its gap (pad_wav), and its words share the time of its sound. An OpenAI line keeps its audio as it is.

---

source

speak

def speak(
    text:str, engine:str=None, model:str=None, voice:str=None, speed:float=None, instructions:str=''
)->tuple:

WAV bytes for text with its maths in words, and the start time in seconds of each part of text (SAY_PART).

Tests with a fake model

A fake Pocket TTS: it gives 0.25 s of silence for each word, and it notes each voice state that it makes. The loader is a fake too, so the test downloads nothing.

import numpy as np

class _Pocket:
    sample_rate = 24000
    def __init__(self): self.said, self.voices = [], []
    def get_state_for_audio_prompt(self, v): self.voices.append(v); return dict(voice=v)
    def generate_audio(self, st, text, normalize=False):
        self.said.append((st['voice'], text, normalize))
        return np.zeros(6000 * len(text.split()), np.float32)
loads = []
_real_load, _pocket_load = _pocket_load, (lambda: loads.append(1) or (_Pocket(), {}))

speak gives one second of WAV for four words, with the gap around it, and a time for each word that goes up. The second line does not load the model again, and a voice state loads one time.

wav, words = speak('Noise goes in, slowly.', 'pocket')
test_eq(round(wav_seconds(wav), 3), 1. + PAD.lead + PAD.tail)   # one second of words, and the gap
test_eq(words, [round(PAD.lead + t, 3) for t in by_letters('Noise goes in, slowly.', 1.)])
test_eq(words, sorted(words))
speak('Again.', 'pocket', voice='mary'); speak('Once more.', 'pocket', voice='mary')
pk = _slots['tts'][0].obj[0]
test_eq((loads, pk.voices, pk.said[-1]), ([1], ['caro_davy', 'mary'], ('mary', 'Once more.', True)))

The settings choose the voice when speak has no engine. The browser voice makes no audio here.

update_settings('voice', out='pocket', out_voice='paul')
speak('hello')
test_eq(pk.said[-1][0], 'paul')
update_settings('voice', out='browser', out_voice='')
test_fail(lambda: speak('hello'), contains='browser')

The OpenAI engine refuses with no key. With a key, it sends the text after the guard, with the model, the voice and the tone. Here a fake guard puts a mark on each text, so the test sees that each text passed it.

real_guard, guard = guard, lambda s, model=None, mode=None: f'[{model.name}] {s}'
test_fail(lambda: speak('hello', 'openai'), contains='no OpenAI key')
set_openai_key('sk-test-1234')
class _W:
    status_code, content = 200, to_wav([0.] * 12000, 24000)
sent = {}
http_post = lambda url, **kw: sent.update(url=url, **kw) or _W()
wav, words = speak('Look at the noise.', 'openai', instructions='calm, like a teacher')
test_eq(sent['url'], OPENAI_SPEECH_URL)
test_eq(sent['json'], dict(model=DFLT_SPEECH, voice='marin', input='[openai] Look at the noise.', response_format='wav', speed=1., instructions='[openai] calm, like a teacher'))
test_eq((wav_seconds(wav), len(words)), (.5, 4))

A fake OpenAI that fails with a status and an error body. _fail gives the line that the learner sees.

class _Bad:
    def __init__(self, status, message='', code=None): self.status_code, self.body = status, dict(error=dict(message=message, type='invalid_request_error', code=code))
    def json(self): return self.body
def _fail(r, model=DFLT_SPEECH, voice='marin'):
    global http_post
    good, http_post = http_post, (lambda url, **kw: r)
    try: speak('Hi.', 'openai', model, voice)
    except SpeakError as e: return str(e)
    finally: http_post = good

A voice that the model does not have gives one line with the fix. An unknown model, a bad key, no credit, too many requests and an error of OpenAI each give their own line. (speak itself sends coral for ballad with tts-1-hd: see out_resolve.) No line has the raw text of the HTTP client or a link.

test_eq(openai_why(_Bad(400, "Invalid voice 'ballad' for model 'tts-1-hd'."), 'tts-1-hd', 'ballad'),
        'OpenAI: the voice Ballad does not work with tts-1-hd. Pick another voice or the model gpt-4o-mini-tts.')
test_eq(openai_why(_Bad(400, 'Invalid value for voice.'), DFLT_SPEECH, 'verse'), 'OpenAI: the voice Verse does not work with gpt-4o-mini-tts. Pick another voice or the model gpt-4o-mini-tts.')   # the table can be old: the message says so
test_eq(_fail(_Bad(404, 'The model `tts-9` does not exist.', 'model_not_found'), 'tts-9', 'nova'),
        'OpenAI: the model tts-9 does not work for speech with your key. Pick the model gpt-4o-mini-tts.')
test_eq(_fail(_Bad(401, 'Incorrect API key provided.', 'invalid_api_key')), 'OpenAI: the key does not work. Set a new key on the Settings page, under More in Speech to text.')
test_eq(_fail(_Bad(429, 'You exceeded your current quota.', 'insufficient_quota')), 'OpenAI: your account has no credit. Add credit at platform.openai.com, or pick another voice.')
test_eq(_fail(_Bad(429, 'Rate limit reached.', 'rate_limit_exceeded')), 'OpenAI: too many requests. Wait one minute, then try again.')
test_eq(_fail(_Bad(503, 'busy')), 'OpenAI did not answer (error 503). Try again later.')
test_eq(_fail(_Bad(400, 'Input is too long. Shorten it.')), 'OpenAI did not say the line: Input is too long.')
for s in [_fail(_Bad(c, 'x')) for c in (400, 401, 429, 503)]: assert 'Client error' not in s and 'http' not in s

tts-1 does not read instructions, so the engine does not send them. Then the test puts back the real guard, the real HTTP client and the real loader.

speak('Hi.', 'openai', 'tts-1', instructions='calm')
assert 'instructions' not in sent['json']
guard, http_post = real_guard, None
set_openai_key('')
_pocket_load = _real_load
unload('tts')

One worker for the voice

Speaker has one worker thread for all the calls of the speaking voice, separate from the speech to text worker. Thus a line of the board does not wait for a segment of dictation. It keeps the lines that it said in this run (at most most), by engine, model, voice, speed and text. The board asks for the same line again when the learner goes back a step, and that line then comes at once. Two requests for the same line share one call. warm loads the model in use with one short line, so the first line of a board does not wait for the load. The browser voice and OpenAI have nothing to load.

A model that is not on this Mac yet is not downloaded for a line: the Settings page shows its download. Until then, say refuses the line with the word “browser”, as for the browser voice, so the page and the Mac app speak with their own voice.

When the voice fails on a line, error keeps why, as one plain line, so the Settings page can show it under the choice. The next line that the voice says clears it.

---

source

Speaker

def Speaker(
    fn:NoneType=None, most:int=64, here:NoneType=None
):

One worker thread for the speaking voice, and the lines that it said in this run.

A fake voice counts its calls. The same line comes from the cache, and a new speed is a new line.

calls = []
def fake_speak(text, engine, model, voice, speed):
    calls.append((text, engine, voice, speed)); time.sleep(.02)
    return to_wav([0.] * 2400, 24000), by_letters(text, .1)
sk = Speaker(fake_speak, most=2, here=lambda m: True)
test_fail(lambda: sk.say('hi', engine='browser'), contains='browser')
r = sk.say('Look  here.', engine='pocket')
test_eq((r['cached'], r['where'], r['voice'], r['words']), (False, 'on this Mac', 'caro_davy', by_letters('Look here.', .1)))
test_eq(sk.say('Look here.', engine='pocket')['cached'], True)
sk.say('Look here.', engine='pocket', speed=1.5)
test_eq(len(calls), 2)

Two requests at the same time for a new line make one call. The cache keeps at most most lines: the oldest goes first.

with ThreadPoolExecutor(2) as ex: list(ex.map(lambda _: sk.say('Both at once.', engine='pocket'), range(2)))
test_eq(len(calls), 3)
test_eq(len(sk.lines), 2)
test_eq(sk.say('Look here.', engine='pocket')['cached'], False)

A failed line is not kept, and error says why until a line works. warm loads the voice of the settings one time, and does nothing for the browser voice. A model that is not here is not warmed, and a line for it says that the browser speaks.

def _flaky(text, *a):
    if text == 'x': raise SpeakError('OpenAI: too many requests. Wait one minute, then try again.')
    return fake_speak(text, *a)
sk = Speaker(_flaky, here=lambda m: True)
test_fail(lambda: sk.say('x', engine='pocket'))
test_eq((sk.lines, sk.error), ({}, 'OpenAI: too many requests. Wait one minute, then try again.'))
sk.say('y', engine='pocket'); test_eq(sk.error, '')
update_settings('voice', out='browser')
test_is(Speaker(fake_speak).warm(), None)
update_settings('voice', out='pocket')
sk = Speaker(fake_speak, here=lambda m: True); calls.clear()
sk.warm().result(); test_is(sk.warm(), None)
test_eq((sk.status, calls), ('ready', [('Ready.', 'pocket', 'caro_davy', 1.)]))
sk = Speaker(fake_speak, here=lambda m: False)
test_is(sk.warm(), None)
test_fail(lambda: sk.say('x'), contains='browser')

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='mlx', 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 ('mlx', 'whisper'): assert transcribe(clip, e, hint='Tennyson').strip(), e
    unload()

wav, words = speak('The forward process adds a little noise at each step.', 'pocket')
assert wav_seconds(wav) > 1 and words, words
unload('tts')