# models


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

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`](https://vedicreader.github.io/abhyasa/models.html#for_task)),
  the default ask
  ([`route_ask`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#fake), a
model for tests that runs offline.

Import the tools that this module uses.

### The model

Each surface gives its own `ask`.

<table>
<colgroup>
<col style="width: 50%" />
<col style="width: 50%" />
</colgroup>
<thead>
<tr>
<th>surface</th>
<th><code>ask</code></th>
</tr>
</thead>
<tbody>
<tr>
<td>web app and notebook</td>
<td><a
href="https://vedicreader.github.io/abhyasa/models.html#route_ask"><code>route_ask</code></a>:
each call goes through the routing of ramabana, on a backend that stays
open</td>
</tr>
<tr>
<td>the ramabana terminal agent</td>
<td><code>agent.oneshot</code>, so the routing of the agent applies</td>
</tr>
<tr>
<td>tests</td>
<td><a
href="https://vedicreader.github.io/abhyasa/models.html#fake"><code>Fake</code></a></td>
</tr>
</tbody>
</table>

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

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L34"
target="_blank" style="float:right; font-size:smaller">source</a>

### dflt_model

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

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

### JSON replies

[`parse_json`](https://vedicreader.github.io/abhyasa/models.html#parse_json)
finds the JSON in a reply.
[`ask_json`](https://vedicreader.github.io/abhyasa/models.html#ask_json)
asks again one time when the reply is not valid JSON.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L59"
target="_blank" style="float:right; font-size:smaller">source</a>

### ask_json

``` python
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.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L39"
target="_blank" style="float:right; font-size:smaller">source</a>

### parse_json

``` python
def parse_json(
    s:str
):
```

*The first JSON object or array in `s`, also inside a
\`\``fence. Raise`ValueError\` when there is none.*

[`parse_json`](https://vedicreader.github.io/abhyasa/models.html#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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L70"
target="_blank" style="float:right; font-size:smaller">source</a>

### Fake

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

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

A matching rule gives its reply, and
[`Fake`](https://vedicreader.github.io/abhyasa/models.html#fake) records
the call.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#ask_json)
asks again and gets the JSON.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#router)
reads it on each call, so a change has an effect on the next call.

<table>
<colgroup>
<col style="width: 50%" />
<col style="width: 50%" />
</colgroup>
<thead>
<tr>
<th>task</th>
<th>what the model does</th>
</tr>
</thead>
<tbody>
<tr>
<td><code>question</code></td>
<td>writes a question and its hidden model answer</td>
</tr>
<tr>
<td><code>grade</code></td>
<td>reads the conversation and grades it</td>
</tr>
<tr>
<td><code>hint</code></td>
<td>gives the next hint</td>
</tr>
<tr>
<td><code>lesson</code></td>
<td>teaches the smallest step after a miss</td>
</tr>
<tr>
<td><a
href="https://vedicreader.github.io/abhyasa/sources.html#extract"><code>extract</code></a></td>
<td>finds the concepts in a text</td>
</tr>
</tbody>
</table>

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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#route_ask)
below shows.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L88"
target="_blank" style="float:right; font-size:smaller">source</a>

### model_ask

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#model_label)
makes the name from the id, and
[`model_label`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#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”.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L112"
target="_blank" style="float:right; font-size:smaller">source</a>

### provider_label

``` python
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.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L105"
target="_blank" style="float:right; font-size:smaller">source</a>

### model_label

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

*ramabana’s
[`model_label`](https://vedicreader.github.io/abhyasa/models.html#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.

``` python
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.

``` python
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:

``` python
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:

``` python
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:

``` python
from fastcore.xtras import modified_env
```

``` python
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:

``` python
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:

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#_on_mac),
as
[`provider_label`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#where_chip)
gives the short text for the page.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L131"
target="_blank" style="float:right; font-size:smaller">source</a>

### where_chip

``` python
def where_chip(
    name:str
)->str:
```

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

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L123"
target="_blank" style="float:right; font-size:smaller">source</a>

### where_of

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#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`.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L139"
target="_blank" style="float:right; font-size:smaller">source</a>

### catalog

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

*The models that can run here: ramabana’s rows (`value`, `name`,
[`label`](https://vedicreader.github.io/abhyasa/sources.html#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.

``` python
import ramabana.core as rc
```

Two calls to
[`catalog`](https://vedicreader.github.io/abhyasa/models.html#catalog)
ask the providers one time.

``` python
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.

``` python
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.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L151"
target="_blank" style="float:right; font-size:smaller">source</a>

### one_each

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

*One row of
[`rows`](https://vedicreader.github.io/abhyasa/learner.html#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.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L209"
target="_blank" style="float:right; font-size:smaller">source</a>

### for_task

``` python
def for_task(
    ask, task:str
):
```

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

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L175"
target="_blank" style="float:right; font-size:smaller">source</a>

### Router

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L218"
target="_blank" style="float:right; font-size:smaller">source</a>

### ramabana_agent

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#_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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L242"
target="_blank" style="float:right; font-size:smaller">source</a>

### ramabana_oneshot

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

*Ask
[`prompt`](https://vedicreader.github.io/abhyasa/learner.html#prompt) on
the ramabana job `job`, pointed at the model `name`. Wait for a busy
backend. Raise
[`no_answer`](https://vedicreader.github.io/abhyasa/models.html#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.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L235"
target="_blank" style="float:right; font-size:smaller">source</a>

### no_answer

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#ramabana_stream)
is the twin of
[`ramabana_oneshot`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L251"
target="_blank" style="float:right; font-size:smaller">source</a>

### ramabana_stream

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

*The reply of
[`ramabana_oneshot`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#no_answer)
when no route answers.*

[`RouteAsk`](https://vedicreader.github.io/abhyasa/models.html#routeask)
is the default `ask` of the web app. `via(task, name, max_tokens)` gives
the `ask` of one task, and
[`Router.for_`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#route_ask)
gives a
[`RouteAsk`](https://vedicreader.github.io/abhyasa/models.html#routeask):
Leela imports this name.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L284"
target="_blank" style="float:right; font-size:smaller">source</a>

### route_ask

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

*The default `ask`: a
[`RouteAsk`](https://vedicreader.github.io/abhyasa/models.html#routeask)
on `model`.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L260"
target="_blank" style="float:right; font-size:smaller">source</a>

### RouteAsk

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#ramabana_worker)
makes a second ramabana `Agent`, for the gather agent of research. It is
never the one `Agent` of
[`ramabana_agent`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/ramabana/agent.html#token_rate)),
and keeps the estimate in `estimated`, so the dollar cap still holds.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L304"
target="_blank" style="float:right; font-size:smaller">source</a>

### ramabana_lead

``` python
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`](https://vedicreader.github.io/abhyasa/path.html#minutes) for
a sub-agent. With `web=False` it has no tool that reaches the web
(`WEB_TOOLS`).

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L293"
target="_blank" style="float:right; font-size:smaller">source</a>

### ramabana_worker

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`.

``` python
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.

``` python
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.

``` python
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.

``` python
import tempfile, shutil
```

``` python
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.

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

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

``` python
@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.

``` python
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.

``` python
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:

``` python
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.

``` python
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 ’’.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#ask_json)
never gets ’’.

``` python
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.

``` python
from ramabana.runtime import OneshotTimeout
```

``` python
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.

``` python
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:

``` python
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:

``` python
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`:

``` python
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.

``` python
from ramabana.core import register_model, Routing
```

``` python
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.

``` python
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`.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#route_ask)
gives a
[`RouteAsk`](https://vedicreader.github.io/abhyasa/models.html#routeask).

``` python
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.

``` python
from abhyasa.learner import update_settings, val
```

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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.

``` python
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`](https://vedicreader.github.io/abhyasa/sources.html#extract)
([`plan`](https://vedicreader.github.io/abhyasa/path.html#plan),
[`pick`](https://vedicreader.github.io/abhyasa/learner.html#pick),
`note`, `profile`) borrows the model, the reply cap and the job of an
older task until it has a model of its own.

``` python
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.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#withheld)),
and `off` lets it through. A model on this Mac gets the personal data.
When the route is not known,
[`guard`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L343"
target="_blank" style="float:right; font-size:smaller">source</a>

### guarded

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

*The `prepare` of a routed call: the prompt `p` and the system prompt
`sp` through
[`guard`](https://vedicreader.github.io/abhyasa/models.html#guard) for
the model of the backend `b`.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L337"
target="_blank" style="float:right; font-size:smaller">source</a>

### safe_query

``` python
def safe_query(
    q:str
)->str:
```

*The web query `q` as it is, through ramabana’s `query`;
[`Withheld`](https://vedicreader.github.io/abhyasa/models.html#withheld)
when it holds a secret or personal data.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L329"
target="_blank" style="float:right; font-size:smaller">source</a>

### guard

``` python
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`).*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L326"
target="_blank" style="float:right; font-size:smaller">source</a>

### Withheld

``` python
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:

``` python
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:

``` python
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:

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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.

<table>
<colgroup>
<col style="width: 12%" />
<col style="width: 12%" />
<col style="width: 12%" />
<col style="width: 12%" />
<col style="width: 12%" />
<col style="width: 12%" />
<col style="width: 12%" />
<col style="width: 12%" />
</colgroup>
<thead>
<tr>
<th>model</th>
<th>load</th>
<th>5 s</th>
<th>clip A</th>
<th>clip B, no hint / hint</th>
<th>terms, no hint / hint</th>
<th>Sanskrit CER</th>
<th>memory</th>
</tr>
</thead>
<tbody>
<tr>
<td><code>mlx-community/Qwen3-ASR-0.6B-8bit</code></td>
<td>3.8 s</td>
<td>0.14 s</td>
<td>6%</td>
<td>13% / 0%</td>
<td>5 / 9</td>
<td>16%</td>
<td>2.0 GB</td>
</tr>
<tr>
<td><code>mlx-community/whisper-large-v3-turbo-asr-fp16</code></td>
<td>4.8 s</td>
<td>0.68 s</td>
<td>6%</td>
<td>4% / 0%</td>
<td>8 / 9</td>
<td>10%</td>
<td>2.3 GB</td>
</tr>
<tr>
<td><code>mlx-community/Qwen3-ASR-1.7B-8bit</code></td>
<td>3.6 s</td>
<td>0.25 s</td>
<td>6%</td>
<td>6% / 0%</td>
<td>7 / 9</td>
<td>14%</td>
<td>3.6 GB</td>
</tr>
<tr>
<td><code>mlx-community/parakeet-tdt-0.6b-v3</code></td>
<td>4.9 s</td>
<td>0.15 s</td>
<td>6%</td>
<td>4% / 4%</td>
<td>7 / 7</td>
<td>Latin script</td>
<td>3.2 GB</td>
</tr>
<tr>
<td><code>mlx-community/Voxtral-Mini-4B-Realtime-2602-4bit</code></td>
<td>3.7 s</td>
<td>2.21 s</td>
<td>6%</td>
<td>6% / 6%</td>
<td>7 / 7</td>
<td>22%</td>
<td>5.5 GB</td>
</tr>
<tr>
<td><code>UsefulSensors/moonshine-base</code></td>
<td>3.9 s</td>
<td>0.07 s</td>
<td>100%</td>
<td>8% / 8%</td>
<td>6 / 6</td>
<td>Latin script</td>
<td>0.8 GB</td>
</tr>
<tr>
<td><code>Mediform/canary-1b-v2-mlx-q8</code></td>
<td>4.2 s</td>
<td>0.08 s</td>
<td>6%</td>
<td>4% / 4%</td>
<td>8 / 8</td>
<td>Latin script</td>
<td>1.4 GB</td>
</tr>
</tbody>
</table>

- “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()`](https://vedicreader.github.io/abhyasa/learner.html#cfg_dir).
The
[`settings`](https://vedicreader.github.io/abhyasa/learner.html#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.

``` python
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`](https://vedicreader.github.io/abhyasa/learner.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/learner.py#L1612"
target="_blank" style="float:right; font-size:smaller">source</a>

### current

``` python
def current()->tuple:
```

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

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L362"
target="_blank" style="float:right; font-size:smaller">source</a>

### resolve

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L391"
target="_blank" style="float:right; font-size:smaller">source</a>

### resident

``` python
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.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L380"
target="_blank" style="float:right; font-size:smaller">source</a>

### unload

``` python
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.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/learner.html#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`](https://vedicreader.github.io/abhyasa/models.html#set_openai_key)
writes the abhyasa file with `save_key` (atomic, mode 0600).
[`masked`](https://vedicreader.github.io/abhyasa/models.html#masked)
shows only the last 4 characters.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L449"
target="_blank" style="float:right; font-size:smaller">source</a>

### masked

``` python
def masked(
    key:str
)->str:
```

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L440"
target="_blank" style="float:right; font-size:smaller">source</a>

### set_openai_key

``` python
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`.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L432"
target="_blank" style="float:right; font-size:smaller">source</a>

### openai_key

``` python
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.

``` python
from ramabana.setup import keys_path
```

``` python
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:

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#set_openai_key)
refuses it.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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()`](https://vedicreader.github.io/abhyasa/learner.html#current),
so a change on the Settings page applies to the next segment.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L452"
target="_blank" style="float:right; font-size:smaller">source</a>

### transcribe

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#wav_seconds)
gives the length of a clip in seconds.
[`silence`](https://vedicreader.github.io/abhyasa/models.html#silence)
makes a clip of silence. The app uses this clip to warm a model.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L470"
target="_blank" style="float:right; font-size:smaller">source</a>

### silence

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

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

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L463"
target="_blank" style="float:right; font-size:smaller">source</a>

### wav_seconds

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#cached)
looks for its English model only.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L492"
target="_blank" style="float:right; font-size:smaller">source</a>

### cache_bytes

``` python
def cache_bytes(
    model:str
)->int:
```

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

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L485"
target="_blank" style="float:right; font-size:smaller">source</a>

### cached

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#model_size)
gives the size of a download: for the local voice, only its English
files.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L513"
target="_blank" style="float:right; font-size:smaller">source</a>

### search_models

``` python
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.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L507"
target="_blank" style="float:right; font-size:smaller">source</a>

### model_size

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

*The bytes that a download of `model` gets from Hugging Face.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L502"
target="_blank" style="float:right; font-size:smaller">source</a>

### is_stt

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#search_models)
gives only the speech to text models, and
[`model_size`](https://vedicreader.github.io/abhyasa/models.html#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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#remove)
deletes a model from the cache, but only a model that
[`fetch`](https://vedicreader.github.io/abhyasa/models.html#fetch)
downloaded. Other entries in the cache belong to other programs.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L531"
target="_blank" style="float:right; font-size:smaller">source</a>

### remove

``` python
def remove(
    model:str
):
```

*Delete `model` from the Hugging Face cache. Only a model that
[`fetch`](https://vedicreader.github.io/abhyasa/models.html#fetch)
downloaded.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L522"
target="_blank" style="float:right; font-size:smaller">source</a>

### fetch

``` python
def fetch(
    model:str
):
```

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

[`remove`](https://vedicreader.github.io/abhyasa/models.html#remove)
refuses a model that abhyasa did not download.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#fit_voice)
keeps a voice that the model has, and else gives the first voice of the
model.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L558"
target="_blank" style="float:right; font-size:smaller">source</a>

### fit_voice

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

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

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L556"
target="_blank" style="float:right; font-size:smaller">source</a>

### openai_voices

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#fit_voice)).
[`out_current`](https://vedicreader.github.io/abhyasa/models.html#out_current)
gives the voice of the settings, with its speed.
[`out_where`](https://vedicreader.github.io/abhyasa/models.html#out_where)
says where the voice runs.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L578"
target="_blank" style="float:right; font-size:smaller">source</a>

### out_where

``` python
def out_where(
    engine:str
)->str:
```

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L574"
target="_blank" style="float:right; font-size:smaller">source</a>

### out_current

``` python
def out_current()->tuple:
```

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

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L564"
target="_blank" style="float:right; font-size:smaller">source</a>

### out_resolve

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#by_letters)
shares the length of the audio among the words by their letters.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L581"
target="_blank" style="float:right; font-size:smaller">source</a>

### by_letters

``` python
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.

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

#### WAV

[`to_wav`](https://vedicreader.github.io/abhyasa/models.html#to_wav)
makes WAV bytes (mono, 16 bit) from the float samples of a model.
[`clean_wav`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L597"
target="_blank" style="float:right; font-size:smaller">source</a>

### clean_wav

``` python
def clean_wav(
    audio:bytes
)->bytes:
```

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

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L589"
target="_blank" style="float:right; font-size:smaller">source</a>

### to_wav

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L609"
target="_blank" style="float:right; font-size:smaller">source</a>

### pad_wav

``` python
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.

``` python
import numpy as np
```

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#_pocket)
loads the model one time into the `tts` slot of
[`resident`](https://vedicreader.github.io/abhyasa/models.html#resident),
and keeps the state of each voice that it used.
[`_pocket_say`](https://vedicreader.github.io/abhyasa/models.html#_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`](https://vedicreader.github.io/abhyasa/models.html#guard),
with
[`ramabana.guard.outgoing`](https://vedicreader.github.io/ramabana/guard.html#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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#speakerror)
with this line, so the learner never sees the raw text of the HTTP
client.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L646"
target="_blank" style="float:right; font-size:smaller">source</a>

### openai_why

``` python
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.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L643"
target="_blank" style="float:right; font-size:smaller">source</a>

### SpeakError

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#spoken)
gives, for each spoken word, the part that it reads, and
[`part_times`](https://vedicreader.github.io/abhyasa/models.html#part_times)
gives the start of each part from the start of its first spoken word.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L758"
target="_blank" style="float:right; font-size:smaller">source</a>

### spoken

``` python
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.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L750"
target="_blank" style="float:right; font-size:smaller">source</a>

### part_times

``` python
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.*

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L695"
target="_blank" style="float:right; font-size:smaller">source</a>

### tex_words

``` python
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.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#spoken),
and the times are for the parts of the line as the board shows them. A
local line gets its gap
([`pad_wav`](https://vedicreader.github.io/abhyasa/models.html#pad_wav)),
and its words share the time of its sound. An OpenAI line keeps its
audio as it is.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L773"
target="_blank" style="float:right; font-size:smaller">source</a>

### speak

``` python
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.

``` python
import numpy as np
```

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#speak) has
no engine. The browser voice makes no audio here.

``` python
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.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#speak)
itself sends `coral` for `ballad` with `tts-1-hd`: see
[`out_resolve`](https://vedicreader.github.io/abhyasa/models.html#out_resolve).)
No line has the raw text of the HTTP client or a link.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L785"
target="_blank" style="float:right; font-size:smaller">source</a>

### Speaker

``` python
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.

``` python
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.

``` python
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.

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L833"
target="_blank" style="float:right; font-size:smaller">source</a>

### Dictation

``` python
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`](https://vedicreader.github.io/abhyasa/models.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/vedicreader/abhyasa/blob/main/abhyasa/models.py#L869"
target="_blank" style="float:right; font-size:smaller">source</a>

### Speech

``` python
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.

``` python
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.

``` python
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.

``` python
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.

``` python
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.

``` python
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`.

``` python
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()
```

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