# runtime
## Native output
A local engine such as LiteRT is a C++ library loaded into the Python
process. It prints its messages straight to file descriptors 1 and 2,
the process’s standard output and standard error, so they never pass
through Python’s `print` or `logging`. When a turn fails, for example
because the prompt was longer than the model’s window, the only
explanation is often one of those lines. This section captures the
descriptors during a model call and keeps the lines that explain a
failure.
[`interesting`](https://vedicreader.github.io/ramabana/runtime.html#interesting)
picks those lines out of captured text. It drops the engine’s routine
startup messages, keeps lines that mention an error, a failure, an
exceeded limit or running out of memory, and returns the last `limit` of
them.
------------------------------------------------------------------------
source
### interesting
``` python
def interesting(
text, limit:int=4
):
```
*The lines of captured output a person should see: complaints, not
chatter.*
Engines have no log level to filter on, so
[`interesting`](https://vedicreader.github.io/ramabana/runtime.html#interesting)
matches words. In this log three lines are startup chatter and one says
why the call failed:
``` python
log = '''Created TensorFlow Lite XNNPACK delegate for CPU.
Loading model from /models/gemma-e4b
I0000 00:00:1730000000.000000 metal delegate ready
ERROR: input tokens 21014 exceeds max_num_tokens 16384
'''
interesting(log)
```
Repeated lines collapse into one, and only the most recent `limit` lines
are kept. An engine that fails on every step of a tool loop prints the
same error many times, and the user needs to see it once.
``` python
test_eq(interesting('ERROR: oom\nERROR: oom\nERROR: oom'), ['ERROR: oom'])
interesting('\n'.join(f'error {i}' for i in range(10)), limit=2)
```
[`captured`](https://vedicreader.github.io/ramabana/runtime.html#captured)
is a context manager that records what is written to descriptors 1 and 2
inside its `with` block. It copies the output rather than hiding it: the
bytes still reach the terminal, and the last 8,000 bytes of each
descriptor (`MAX_KEEP`) are kept in `cap.text`. `cap.problems` runs
[`interesting`](https://vedicreader.github.io/ramabana/runtime.html#interesting)
over that text.
Only one
[`captured`](https://vedicreader.github.io/ramabana/runtime.html#captured)
block is active at a time across threads, because two threads
redirecting the same descriptor would restore each other’s copies and
leave the real descriptor detached. A thread that cannot take the lock
within half a second runs its block without capture. Setting
`RAMABANA_NO_NATIVE_CAPTURE=1` turns capture off.
------------------------------------------------------------------------
source
### captured
``` python
def captured(
fds:tuple=(1, 2), enabled:NoneType=None
):
```
*Copy what is written to descriptors 1 and 2 into `text`, while the
output still reaches the terminal.*
------------------------------------------------------------------------
source
### capture
``` python
def capture(
fn, *a, **kw
):
```
*Call `fn` and return `(result, problems)`; an exception carries them
out too.*
`capture(fn, *args)` is the form to use around a single call. It calls
`fn` inside
[`captured`](https://vedicreader.github.io/ramabana/runtime.html#captured)
and returns a pair: the result, and the problem lines the engine
printed. A call can succeed and still print a warning worth showing, as
here, where the engine truncated its input:
``` python
def _engine_call():
os.write(2, b'Created TensorFlow Lite XNNPACK delegate for CPU.\n')
os.write(2, b'WARNING: input exceeds max_num_tokens 16384, truncating\n')
return 'a short reply'
capture(_engine_call)
```
When `fn` raises,
[`capture`](https://vedicreader.github.io/ramabana/runtime.html#capture)
attaches the problem lines to the exception as `native_output` and
raises it again. Native failures often arrive as an exception with an
empty message, and this keeps the real explanation with it:
``` python
def _engine_dies():
os.write(2, b'ERROR: failed to allocate 4.00 GiB for KV cache\n')
raise RuntimeError('')
try: capture(_engine_dies)
except RuntimeError as e: err = e
err.native_output
```
## Sizing a conversation
These functions decide when a conversation has grown too big for its
model. Each limit is a fraction of the context window rather than a
fixed number of tokens, so the same code works for a 4k-token local
model and a 200k-token hosted one. A fixed reserve would not: with 16k
tokens held back from a 4k window, every turn would be due for
compaction, and the agent would compact a two-message conversation
forever.
[`estimate_tokens`](https://vedicreader.github.io/ramabana/runtime.html#estimate_tokens)
counts the tokens in a text.
[`threshold`](https://vedicreader.github.io/ramabana/runtime.html#threshold)
gives the token count at which compaction becomes due, and
[`should_compact`](https://vedicreader.github.io/ramabana/runtime.html#should_compact)
compares a count with it.
[`halvings`](https://vedicreader.github.io/ramabana/runtime.html#halvings)
returns a budget followed by its halves, for retrying a request whose
size was only estimated.
------------------------------------------------------------------------
source
### should_compact
``` python
def should_compact(
used, ctx, reserve:NoneType=None
):
```
*Whether `used` tokens against a `ctx` window has crossed the line.*
------------------------------------------------------------------------
source
### threshold
``` python
def threshold(
ctx, reserve:NoneType=None
):
```
*The token count at which a conversation should be compacted, or None
when there is no window. `reserve` is config.toml’s `window.reserve`,
else `RESERVE`.*
------------------------------------------------------------------------
source
### halvings
``` python
def halvings(
budget, tries:int=3, floor:int=256
):
```
*A budget and its halvings, to retry a prompt whose fit was only
estimated.*
------------------------------------------------------------------------
source
### estimate_tokens
``` python
def estimate_tokens(
text, count:NoneType=None
):
```
*Tokens in `text`: exact via `count` when a tokenizer is at hand,
`CHARS_PER_TOKEN` otherwise.*
[`estimate_tokens`](https://vedicreader.github.io/ramabana/runtime.html#estimate_tokens)
uses the model’s own tokenizer when you pass one as `count`, and
otherwise divides the length of the text by `CHARS_PER_TOKEN` (3.25).
The estimate is rough, and close enough for budgeting.
``` python
estimate_tokens('the exporter writes one module per notebook'), estimate_tokens('', ), estimate_tokens('x' * 400)
```
`threshold(ctx)` holds back a reserve below the window: the full
`RESERVE` of 16,384 tokens for a large window, and at most a quarter of
the window for a smaller one. Without the quarter-window cap, a 4k model
would be due for compaction before its first turn.
``` python
threshold(200_000), threshold(32_768), threshold(4096)
```
``` python
test_eq(threshold(4096), 3072)
test_eq(threshold(0), None)
should_compact(3000, 4096), should_compact(3072, 4096)
```
## Reading a conversation
A backend stores its conversation as a list of message dictionaries in
urai’s format, which is the same for every model. Each message has a
`role` (`user`, `assistant` or `tool`) and `content`. An assistant
message that calls tools also carries `tool_calls`, and the result of
each call comes back as a `tool` message. Compaction reads this list
directly, using urai’s `resp_text` for a message’s text and `tool_rows`
for its tool calls.
The examples below use this four-message conversation: a request, a tool
call, the tool’s result, and the reply.
``` python
hist = [
{'role': 'user', 'content': 'the exporter duplicates cells, please fix it'},
{'role': 'assistant', 'content': 'Reading the exporter.',
'tool_calls': [{'function': {'name': 'view_file', 'arguments': {'path': 'nbdev/maker.py'}}}]},
{'role': 'tool', 'content': 'def _make_exists(self, cells):\n with self.fname.open("a") as f: ...'},
{'role': 'assistant', 'content': 'It appends, so a second export duplicates every named cell.'},
]
[(m.get('role', '?'), resp_text(m)[:40], [name for name, _ in tool_rows(m)]) for m in hist]
```
[`serialise`](https://vedicreader.github.io/ramabana/runtime.html#serialise)
writes a list of messages as the text a summarizing model reads. Each
message goes in a `` tag, with its tool calls named in a
`` tag, and each text is clipped to `mx` characters (2,000
by default). Tool results make up most of a conversation’s length and
little of its meaning, so clipping them costs little.
------------------------------------------------------------------------
source
### split_previous
``` python
def split_previous(
msgs
):
```
*Split off a leading summary:
`(previous_summary_or_None, remaining_msgs)`.*
------------------------------------------------------------------------
source
### serialise
``` python
def serialise(
msgs, mx:int=2000
):
```
*Messages as the tagged block the summarizer reads, each text clipped to
`mx`.*
``` python
print(serialise(hist))
```
Once a conversation has been compacted, its first message is a user
message that starts with `SUMMARY_PREFIX` and holds the summary.
[`split_previous`](https://vedicreader.github.io/ramabana/runtime.html#split_previous)
separates that summary from the messages after it. The next compaction
can then update the existing summary with the new messages, instead of
summarising the old summary as if it were ordinary conversation.
``` python
prev, rest = split_previous([{'role': 'user', 'content': SUMMARY_PREFIX + '## Goal\nfix the exporter'}] + hist)
prev, len(rest)
```
``` python
test_eq(split_previous(hist), (None, hist))
split_previous([])
```
[`summarise_prompt`](https://vedicreader.github.io/ramabana/runtime.html#summarise_prompt)
builds the request sent to the summarizing model. With no earlier
summary it asks for a new one in a fixed format: goal, constraints,
progress, key decisions, next steps and critical context. When
[`split_previous`](https://vedicreader.github.io/ramabana/runtime.html#split_previous)
finds an earlier summary, it asks for that summary to be updated with
the new messages instead. `extra` adds a line saying what the summary
should focus on.
The summary is usually written by a cheaper model with a smaller window,
so `max_tokens` caps the size of the request. When the whole request
does not fit, the earlier summary is cut to a quarter of the budget, and
the messages after it fill what is left, cut from the end. The
instructions come last and are kept whole while the budget can hold
them.
------------------------------------------------------------------------
source
### fold_summary
``` python
def fold_summary(
msgs, summariser, extra:str='', max_tokens:NoneType=None, count:NoneType=None
):
```
*Summarise `msgs` in runs that each fit `max_tokens`, every summary
updated with the next run; ’’ when a step answers nothing.*
------------------------------------------------------------------------
source
### transcript_runs
``` python
def transcript_runs(
msgs, budget, count:NoneType=None
):
```
*`msgs` in consecutive runs whose transcript fits `budget` tokens; a
message that alone is larger makes a run of its own.*
------------------------------------------------------------------------
source
### summarise_prompt
``` python
def summarise_prompt(
msgs, extra:str='', max_tokens:NoneType=None, count:NoneType=None
):
```
*The bounded prompt handed to the summarizer, choosing fresh or update
instructions.*
With no budget the request holds the whole conversation, one ``
per message, followed by the format instructions:
``` python
p = summarise_prompt(hist)
test_eq(p.count('source
### truncate_middle
``` python
def truncate_middle(
text, budget, count:NoneType=None, mark:str=' … '
):
```
*Keep both ends of text inside a token budget.*
------------------------------------------------------------------------
source
### surgical_history
``` python
def surgical_history(
msgs, policy:NoneType=None, count:NoneType=None
):
```
*Render old history as a compact, readable DSL while preserving tool
evidence.*
``` python
truncate_middle('nbs/01_runtime.ipynb::Compactor.compact::summarise_prompt', 12)
```
Each role has its own marker: `§` for the user, `»` for the assistant’s
text, `▶` for a tool call and `>` for its result. The lines of a tool
result are joined with `¶`.
``` python
print(surgical_history(hist))
```
`SURGICAL_POLICY` sets the budgets: 2,000 tokens for a user message, 150
for the assistant’s text, 60 for a tool call and 35 for a result. The
user’s own words are kept nearly whole, and tool results are cut
hardest. `policy` overrides any of them:
``` python
print(surgical_history(hist, policy={'result': 8, 'assistant': 20}))
```
## Reorienting the model
After compaction the model sees a summary where its conversation used to
be, and it can draw wrong conclusions from that.
[`reorient`](https://vedicreader.github.io/ramabana/runtime.html#reorient)
writes the reminder placed after the summary. It says whether the Python
kernel survived, so that the model neither rebuilds variables that still
exist nor uses ones that are gone. It tells the model that skill text it
read earlier is no longer in context and has to be read again with
`read_skill`, naming the skills when `skills` is given. It also tells
the model not to answer a finished request a second time: if the latest
request is complete, the model replies in one line and waits. `REORIENT`
is the reminder with the default arguments.
------------------------------------------------------------------------
source
### reorient
``` python
def reorient(
kernel_alive:bool=True, skills:tuple=()
):
```
*The system reminder placed after a compacted or otherwise rewritten
conversation.*
``` python
print(reorient(skills=('nbdev', 'fossick')))
```
When the kernel restarted, the reminder says the namespace is empty.
Getting this wrong would send the model looking for variables that no
longer exist, or rebuilding ones that do.
``` python
test_eq('do not re-import anything' in reorient(kernel_alive=False), False)
print(reorient(kernel_alive=False)[:180])
```
``` python
assert 'inspect_python' in reorient() and 'list_vars' not in reorient()
```
## Notices on the way in
Before a prompt reaches the model,
[`prompt_notices`](https://vedicreader.github.io/ramabana/runtime.html#prompt_notices)
checks it with a few string tests and returns notices: short
instructions that tell the model how to treat the prompt. They follow
the `UserPromptSubmit` hook in Answer.AI’s aai-coding prompts, and each
one answers a mistake models often make.
- A prompt ending in `?` is a question. The model answers it and stops,
without starting the work it implies (`Q_NOTICE`).
- A prompt containing “please read”, “read the” or “have a look at” asks
the model to read the target in full before answering (`READ_NOTICE`).
- A bare approval such as “go”, “ok” or “do it” approves only what the
user explicitly agreed to, not everything the model proposed earlier
(`APPROVAL_NOTICE`).
- A prompt starting with “BTW” is a side request. The model answers it
and then goes back to the task it was on (`BTW_NOTICE`).
- A prompt starting with a verb such as fix, add, run or rename, or
ending in `as NAME`, asks for action. The model does the work and
checks it before replying (`ACTION_NOTICE`).
Three one-character prompts are shortcuts and get only their own notice:
`.` continues the agreed task, `;` asks for the last answer in simpler
words, and `,` asks what the user needs to answer next.
------------------------------------------------------------------------
source
### notices_block
``` python
def notices_block(
prompt
):
```
*The notices for `prompt` as one reminder to append to it, or `''`.*
------------------------------------------------------------------------
source
### prompt_notices
``` python
def prompt_notices(
prompt
):
```
*Notices a submitted prompt earns, from aai-coding’s `UserPromptSubmit`
hook.*
A question gets `Q_NOTICE` and nothing else. A bare “go” gets
`APPROVAL_NOTICE`:
``` python
test_eq(prompt_notices('can you fix the exporter?'), [Q_NOTICE])
prompt_notices('go')
```
A prompt can earn several notices. Here an imperative gets one, a side
request that asks for a read gets two, and “thanks” gets none:
``` python
[len(prompt_notices(p)) for p in ('fix the exporter', 'BTW please read nbs/01_runtime.ipynb', 'thanks')]
```
[`notices_block`](https://vedicreader.github.io/ramabana/runtime.html#notices_block)
is the function the agent calls. It wraps the notices in one
`` block, ready to append to the prompt, or returns
`''` when there are none.
``` python
print(notices_block('ok'))
```
``` python
test_eq(notices_block('thanks'), '')
notices_block('thanks')
```
## Fitting a notebook into the prompt
A prompt can carry the user’s open notebook inside a `` tag,
one `` per cell. Each cell has a
[`compact`](https://vedicreader.github.io/ramabana/learned.html#compact)
attribute that the user sets: `keep`, `auto` or `discard`.
[`compact_notebook_context`](https://vedicreader.github.io/ramabana/runtime.html#compact_notebook_context)
leaves the prompt alone when it fits. When it does not, it removes
`discard` cells, then `auto` cells, each in notebook order, until the
prompt fits. It never removes a `keep` cell. A `` tag
records how many cells were left out.
`fits` is a function that says whether a prompt fits. The agent passes
the backend’s `fits`, so the decision is made against the real window of
the model in use.
------------------------------------------------------------------------
source
### compact_notebook_context
``` python
def compact_notebook_context(
prompt, fits
):
```
*Reduce a tagged notebook only when `prompt` does not fit. A `keep` cell
is never removed.*
A notebook of three cells, one with each setting, and a question about
it. When everything fits, the prompt comes back unchanged:
``` python
nb = '''
| print(load_everything()) # 900 lines of output |
df = read_csv('sales.csv') |
model.fit(df) # the cell the question is about |
why did the fit fail?'''
print(compact_notebook_context(nb, fits=lambda s: True))
```
With a lower limit the `discard` cell goes first, and
`` records it:
``` python
out = compact_notebook_context(nb, fits=lambda s: len(s) < 270)
print(out)
```
A lower limit still removes the `auto` cell as well. The `keep` cell
stays, even though the result is still over the limit, because the user
chose to keep it.
``` python
tight = compact_notebook_context(nb, fits=lambda s: len(s) < 150)
test_eq('compact="keep"' in tight, True)
test_eq('load_everything' in tight, False)
print(tight)
```
## The compactor
[`Compactor`](https://vedicreader.github.io/ramabana/runtime.html#compactor)
decides when a backend’s conversation is due for compaction, and carries
it out. It keeps the most recent messages as they are and replaces
everything older with a summary, followed by the
[`reorient`](https://vedicreader.github.io/ramabana/runtime.html#reorient)
reminder. The summary comes from a second model call, made through the
`summariser(prompt, sp)` function you pass to
[`compact`](https://vedicreader.github.io/ramabana/learned.html#compact).
With `strategy='surgical'` it uses
[`surgical_history`](https://vedicreader.github.io/ramabana/runtime.html#surgical_history)
instead and calls no model.
`due(backend)` says whether the backend has crossed its threshold.
[`compact`](https://vedicreader.github.io/ramabana/learned.html#compact)
returns the summary, or `''` when it did nothing, and
[`note`](https://vedicreader.github.io/ramabana/agent.html#note) says
what happened either way. `on_compact`, when given, is called with each
new summary. If the summarizer fails or returns nothing, the
conversation is left alone. When you give the summarizer’s window as
`summary_ctx`, no request is larger than it:
[`fold_summary`](https://vedicreader.github.io/ramabana/runtime.html#fold_summary)
splits a history too large for one request into runs that each fit
([`transcript_runs`](https://vedicreader.github.io/ramabana/runtime.html#transcript_runs)),
summarises the first, and updates that summary with each run after it,
so a history of any length compacts. A failed or empty call is retried
at half and then a quarter of that size.
Compaction lives in its own class rather than in a rishi callback
because it needs a second model call and then replaces the first model’s
history, and a callback does not do both.
------------------------------------------------------------------------
source
### Compactor
``` python
def Compactor(
reserve:NoneType=None, # None reads config.toml's `window.reserve`, else `RESERVE`, each time
keep_recent:NoneType=None, # None reads `window.keep_recent`, else `KEEP_RECENT`
auto:bool=True, # compact on crossing the threshold
kernel_alive:bool=True, # what the reorientation note may promise
on_compact:NoneType=None, # called with each new checkpoint
strategy:str='summary', # 'summary' or deterministic 'surgical'
):
```
*Decide when to compact, and do it.*
`budget(ctx)` is how many tokens of recent conversation survive
compaction: `KEEP_RECENT` (20,000), capped at half of what the window
holds besides the system prompt and tools (`overhead`). Without the cap,
a 4k model would count its whole conversation as recent, find nothing
older to summarise, and report
`everything is recent; nothing to compact` until the engine refused the
turn.
[`compact`](https://vedicreader.github.io/ramabana/learned.html#compact)
also keeps at most about half of the conversation, and the part it keeps
always starts at a user message.
``` python
c = Compactor()
c.budget(200_000), c.budget(32_768), c.budget(4096)
```
``` python
test_eq(c.budget(4096), 2048)
test_eq(c.note, 'not compacted')
c.strategy
```
Compacting needs a running backend and a summarizer.
[testing](04_testing.ipynb) defines doubles for both and runs a
compaction end to end.
## Answers, without the thinking
Reasoning models think before they answer. Some runtimes return the
thinking in a separate channel, and others leave it in the reply,
wrapped in `...`. The agent’s one-shot jobs read the
reply as data, such as a label, a summary or a completion to insert, so
the thinking has to come off first. A classifier that returned the
model’s reasoning would return no label at all.
[`answer_only`](https://vedicreader.github.io/ramabana/runtime.html#answer_only)
removes the thinking from a finished reply, using urai’s `split_think`.
------------------------------------------------------------------------
source
### answer_only
``` python
def answer_only(
text
):
```
*A one-shot reply with the model’s thinking removed, however the runtime
left it.*
In the simple case the thinking is wrapped in a pair of tags:
``` python
answer_only('7 times 6. Six sevens are 42.\n\n42')
```
Some chat templates open the `` block in the prompt itself, so
the reply starts inside the thinking and only the closing tag appears.
[`answer_only`](https://vedicreader.github.io/ramabana/runtime.html#answer_only)
handles that too. A reply that opens a block and never closes it,
because it hit the output cap, has no answer and gives `''`. A reply
with no tags comes back as it is.
``` python
answer_only('Six sevens are 42. They asked for only the number.\n\n\n42')
```
``` python
test_eq(answer_only('cut off at the cap and never closed'), '')
test_eq(answer_only('thinking, and then the cap hit\n'), 'thinking, and then the cap hit')
test_eq(answer_only('42'), '42')
answer_only('ayes')
```
A model whose template opens the thinking block does so at every step of
a turn: after each tool call, the next reply begins inside a new
thought.
[`prefills_think`](https://vedicreader.github.io/ramabana/runtime.html#prefills_think)
tells whether a model’s template does this, by rendering a one-message
prompt and looking for an unclosed ``. For such a model
[`RishiBackend`](https://vedicreader.github.io/ramabana/runtime.html#rishibackend)
passes the stream through a
[`ThinkFilter`](https://vedicreader.github.io/ramabana/runtime.html#thinkfilter),
which drops text until it sees ``, lets the rest through, and
starts dropping again after each tool call. Use
[`answer_only`](https://vedicreader.github.io/ramabana/runtime.html#answer_only)
on a finished reply and
[`ThinkFilter`](https://vedicreader.github.io/ramabana/runtime.html#thinkfilter)
on a stream. The filter also counts the characters of thinking
(`thought`) and of answer (`answer`), so the backend can report a turn
that spent all its output thinking. One-shot jobs avoid the problem by
asking the model not to think.
------------------------------------------------------------------------
source
### ThinkFilter
``` python
def ThinkFilter():
```
*Drop a template-opened thinking block out of a raw chunk stream. A tool
call re-arms it.*
------------------------------------------------------------------------
source
### prefills_think
``` python
def prefills_think(
chat
):
```
*Whether this model’s chat template opens a `` block and leaves
the model to close it.*
The closing tag can be split across chunks, and the filter still finds
it. The answer comes through whole:
``` python
def chunks(*texts): return [{'content': [{'type': 'text', 'text': t}]} for t in texts]
def texts(chunks): return ''.join(resp_text(c) for c in chunks)
f = ThinkFilter()
texts(f(chunks('Six sevens', ' are 42.\n\n\n', '42')))
```
``` python
test_eq(texts(ThinkFilter()(chunks('deliberating', '', 'yes'))), 'yes')
tool = {'content': [{'type': 'tool_call', 'name': 'search_code', 'arguments': {}}]}
out = list(ThinkFilter()(chunks('think 1', '', 'calling ') + [tool] + chunks('think 2', '', 'done')))
test_eq(texts(out), 'calling done')
test_eq(out[1], tool)
f = ThinkFilter()
test_eq(texts(f(chunks('cut off at the cap'))), '')
test_eq((f.thinking, f.thought, f.answer), (True, 18, 0))
f = ThinkFilter()
test_eq(texts(f(chunks('deliberating', '\n\n'))), '')
test_eq((f.thinking, f.answer), (False, 0))
```
## Usage
[`Usage`](https://vedicreader.github.io/ramabana/runtime.html#usage)
holds the token counts and cost of a turn: tokens in and out, cached
tokens, tokens written to the cache, reasoning tokens, the cost in
dollars, and the number of turns. Two
[`Usage`](https://vedicreader.github.io/ramabana/runtime.html#usage)
values add up, so a session’s total is `sum` of its turns, and
subtracting an earlier running total from a later one gives what was
used in between. Its repr is the one-line summary shown in the
terminal’s status bar.
------------------------------------------------------------------------
source
### Usage
``` python
def Usage(
model:str='', input:int=0, output:int=0, total:int=0, cached:int=0, cache_write:int=0, reasoning:int=0,
cost:float=0.0, turns:int=0
)->None:
```
*Token counts and cost for a turn or a session; values add up.*
``` python
a = Usage(model='anthropic/claude-sonnet-5', input=12_000, output=800, total=12_800, cached=9_000, cost=0.041, turns=1); a
```
The sum takes the model name from the right-hand value, so a session
that switched models reports the model it uses now:
``` python
b = Usage(model='openai/gpt-5.6-terra', input=3_000, output=200, total=3_200, cost=0.008, turns=1)
sum([a, b], None)
```
``` python
test_eq((a + None).total, a.total)
test_eq(sum([a, b], None).turns, 2)
a.dict()['cached']
```
## Backend
[`Backend`](https://vedicreader.github.io/ramabana/runtime.html#backend)
is the contract the agent uses to talk to a model. One backend holds one
conversation. It starts the model on first use, sends a message with
`send` or streams the reply with `stream`, asks one-shot questions with
`oneshot`, counts tokens, and replaces or restores the conversation’s
history. Subclasses supply the engine-specific methods whose names start
with an underscore.
[`RishiBackend`](https://vedicreader.github.io/ramabana/runtime.html#rishibackend),
further down, is the one the agent uses.
`send`, `stream` and `oneshot` do not raise when the model fails. A
model that cannot start, a stream that breaks and an empty reply all
become text:
[`note`](https://vedicreader.github.io/ramabana/agent.html#note) holds
the latest status, `problems` keeps the last 20 problem lines, and
`turn_error` says why the last turn failed. A caller can always show the
user what the backend returned.
The examples below run the base class itself, which has no engine, to
show these rules.
------------------------------------------------------------------------
source
### fit_content
``` python
def fit_content(
c, room, save, how:str='', count:NoneType=None, tail:int=0
):
```
*Message content `c` within `room` tokens: the newest pictures that fit
stay and the rest are saved and named, then text past the room becomes
an excerpt. The last `tail` characters of the last text part are always
kept.*
------------------------------------------------------------------------
source
### excerpt
``` python
def excerpt(
text, room, save, how:str='', count:NoneType=None, tail:int=0
):
```
*`text` cut to its head and its end within `room` tokens, the last
`tail` characters always kept; the whole goes to `save(data, suffix)`
and the note names the file.*
------------------------------------------------------------------------
source
### omitted
``` python
def omitted(
n, path, how:str=''
):
```
*The note an excerpt carries where `n` characters were cut: where the
whole is, and how to read it.*
------------------------------------------------------------------------
source
### trim_hist
``` python
def trim_hist(
hist, need, count:NoneType=None, keep:int=2, save:NoneType=None, how:str=''
):
```
*Free about
[`need`](https://vedicreader.github.io/ramabana/setup.html#need) tokens
from `hist` in place, the oldest tool results first and then the oldest
pictures; the last `keep` messages stay whole. With `save`, a trimmed
result is kept in a file its note names. Returns what was freed.*
------------------------------------------------------------------------
source
### msg_tokens
``` python
def msg_tokens(
m, count:NoneType=None
):
```
*One history message’s tokens: its text, its tool calls’ arguments, and
`IMG_TOKENS` a picture.*
------------------------------------------------------------------------
source
### img_tokens
``` python
def img_tokens():
```
------------------------------------------------------------------------
source
### margin
``` python
def margin():
```
------------------------------------------------------------------------
source
### Backend
``` python
def Backend(
spec, sp:str='', tools:tuple=(), approve:NoneType=None, tool_max_len:NoneType=None, shared:bool=False, **kw
):
```
*The contract a model engine meets: start, send or stream a turn, count
tokens, and replace history.*
The base class has no engine, so `start` fails. It returns `None` and
records the `NotImplementedError` in
[`note`](https://vedicreader.github.io/ramabana/agent.html#note) instead
of raising it:
``` python
b = Backend(resolve('gemma-e4b'))
b.start(), b.note
```
`send` on a backend that could not start returns the same note, and the
failure is in `problems`. A caller that never checks for errors still
shows the user something true:
``` python
test_eq(b.send('hello'), b.note)
b.problems
```
`pending_tokens(msg)` estimates what a message will add to the
conversation. Each image or audio part counts as
[`img_tokens()`](https://vedicreader.github.io/ramabana/runtime.html#img_tokens),
config.toml’s `window.image_tokens` or `IMG_TOKENS` (1,600), so a 120 KB
image is not counted as the tokens of its bytes:
``` python
test_eq(b.pending_tokens([bytes(120_000), 'what is this?']),
b.pending_tokens('what is this?') + IMG_TOKENS)
```
`problem` adds a line to `problems`. It skips a line identical to the
one before it and keeps the latest 20:
``` python
for _ in range(30): b.problem('ERROR: oom')
len(b.problems), b.problems[-1]
```
Before the model has started there is no tokenizer, so `count_tokens`
falls back to
[`estimate_tokens`](https://vedicreader.github.io/ramabana/runtime.html#estimate_tokens).
`used_tokens` is the size of the conversation so far, and `pct_full` is
that size as a share of the window:
``` python
b.count_tokens('a prompt of some length'), b.used_tokens, round(b.pct_full, 3)
```
`fits(msg)` says whether the conversation plus `msg` still fits in the
window, keeping back a quarter of the window for the reply, but no less
than 128 and no more than 1,024 tokens. This is the `fits` that
[`compact_notebook_context`](https://vedicreader.github.io/ramabana/runtime.html#compact_notebook_context)
receives. The `gemma-e4b` spec used here has a 16,384-token window, and
a spec with no window (`ctx=0`) fits anything:
``` python
big = 'x' * 200_000
b.fits('a short prompt'), b.fits(big)
```
``` python
test_eq(Backend(ModelSpec('unbounded', 'remote', 'v/m', ctx=0)).fits(big), True)
b.spec.ctx
```
`fits` charges the pending message
[`margin()`](https://vedicreader.github.io/ramabana/runtime.html#margin)
more than its estimate, config.toml’s `window.margin` or `EST_MARGIN`
(1.15), because a count from characters runs short on code and escapes.
A hosted chat’s `token_count` is what its last step reported, so after
`replace_hist` or `restore_hist`, `used_tokens` gives `window_tokens`,
an estimate of the briefing, the tool schemas and every message, until
the chat reports again. Without that, a compaction left the old count in
place, `fits` still failed, and each retry compacted again. A hosted
chat that has not reported a step yet counts `window_tokens` too, and
one not started yet counts the conversation it will restore, so the
first turn after a resume is measured before it goes out.
`trim(need)` frees tokens from the oldest tool results first, then the
oldest pictures, and leaves the newest two messages whole. With `spill`
set, a trimmed result is written whole to a file first, and its note
names the file and says how to read it (`how`). `shrink_recent(need)`
cuts the newest two messages a person or a tool wrote, largest first,
which `trim` never touches. `fit_content(content, room, save)` does the
cutting: pictures past what fits are saved and named, the newest kept,
and text past the room becomes
[`excerpt`](https://vedicreader.github.io/ramabana/runtime.html#excerpt),
its head and its end around a note such as
`…[52,000 characters omitted; full text at ; read it in pieces with view_file(path, start=N)]`.
Nothing is lost: the whole is in the file:
``` python
from urai import mk_msgs
t = trim_hist(h := mk_msgs(['look at this', 'done']) + [{'role': 'tool', 'tool_call_id': 'c', 'content': 'r' * 9000}] + mk_msgs(['and now', 'ok']), 100)
test_eq(h[2]['content'], TRIMMED)
assert t > 100 and h[-1]['content'] == 'ok'
```
``` python
saved = []
def save(data, suffix): saved.append(data); return Path(f'/tmp/full{len(saved)}{suffix}')
cut = fit_content('head ' + 'x' * 20_000 + ' end', 200, save, 'read it in pieces with view_file(path, start=N)')
assert cut.startswith('head') and cut.endswith(' end') and 'full text at /tmp/full1.txt' in cut and saved[0].endswith(b' end')
pics = fit_content([b'\x89PNG\r\n\x1a\n1', b'\x89PNG\r\n\x1a\n2', b'\x89PNG\r\n\x1a\n3', 'which?'], 2 * IMG_TOKENS, save)
test_eq(pics[0], b'\x89PNG\r\n\x1a\n3') # the newest picture stays
assert 'which?' in pics and '2 of the 3 pictures' in pics[-1]
print(cut[:60] + ' ... ' + cut[-160:])
```
`oneshot(prompt)` asks one question in a throwaway conversation and
returns the answer with any thinking removed. It cuts the prompt to fit
the window, keeping its end, and returns `if_busy` (by default `''`)
while another call is using the backend. This subclass stands in for a
model that leaves its thinking in the reply:
``` python
class Thinker(Backend):
def _start(self): return self
def _oneshot(self, prompt, sp, max_tokens): return 'They want one label.\n\n\nsuccess'
test_eq(Thinker(resolve('gemma-e4b')).oneshot('the tests pass'), 'success')
Thinker(resolve('gemma-e4b')).oneshot('the tests pass')
```
The history methods behave differently: `replace_hist`, `restore_hist`
and `revise_last_assistant` raise when no model is running, because
there is no conversation to change. `replace_hist(summary, keep)` is
what
[`Compactor`](https://vedicreader.github.io/ramabana/runtime.html#compactor)
calls. `snapshot_hist` and `restore_hist` save and restore a
conversation, for branching. `resume_hist` hands a saved conversation to
a backend that has not started yet, and applies it when the model
starts.
``` python
test_fail(lambda: b.replace_hist('summary'), contains='not running')
test_fail(lambda: b.restore_hist([]), contains='not running')
test_fail(lambda: b.revise_last_assistant('...'), contains='not running')
```
## RishiBackend
[`RishiBackend`](https://vedicreader.github.io/ramabana/runtime.html#rishibackend)
is the only implementation of
[`Backend`](https://vedicreader.github.io/ramabana/runtime.html#backend),
because rishi runs every model, local or hosted. It turns a
[`ModelSpec`](https://vedicreader.github.io/ramabana/core.html#modelspec)
(a model’s name, runtime, id and window, from [core](00_core.ipynb))
into a rishi `Chat`, converts rishi’s usage into
[`Usage`](https://vedicreader.github.io/ramabana/runtime.html#usage),
and handles what differs between runtimes:
- A LiteRT model gets its window as the engine’s token ceiling,
`max_num_tokens`, and a llama.cpp or ollama model gets it as `n_ctx`.
- A Claude model is asked for `high` reasoning effort (`CLAUDE_EFFORT`)
unless `RAMABANA_EFFORT` names another level.
- On MLX, one-shot questions run in a separate conversation on the same
engine, so a completion never sees earlier completions.
- A local model’s window is set from the window it was trained for,
capped by Ramabana’s limit for that model, and narrowed again if the
engine reports a smaller one.
- When a hosted service refuses the tool schemas, the backend moves that
model to the tags channel, where the schemas travel in the system
prompt and calls come back as `` blocks in the text, and
retries once.
- When a reply on the tags channel describes a tool call instead of
making one, the backend sends one reminder (`TAG_REMINDER`) asking the
model to make the call.
rishi adds two callbacks to a chat by default: a reminder on every
message, and a sliding window that forgets old messages without
summarising them.
[`RishiBackend`](https://vedicreader.github.io/ramabana/runtime.html#rishibackend)
removes both (`ROUTINE_CBS`), because
[`Compactor`](https://vedicreader.github.io/ramabana/runtime.html#compactor)
does that job. LiteRT runs the tool loop inside the engine, so the text
a model writes before a tool call never reaches Ramabana.
`QUIET_RUNTIMES` lists such runtimes, and `narrates` is False for them.
LiteRT models run on the GPU by default. rishi asks for the GPU, and
falls back to the CPU with a warning when the build has no GPU support.
`RAMABANA_LITERT_BACKEND=cpu` pins a model to the CPU, and `gpu` asks
for the GPU explicitly. A `backend=` argument, or
`eng_kw={'backend': ...}`, takes precedence over the variable.
`RISHI_LITERT_GPU=0` stops rishi asking for the GPU at all.
------------------------------------------------------------------------
source
### make_backend
``` python
def make_backend(
spec, **kw
):
```
------------------------------------------------------------------------
source
### **getattr**
``` python
def __getattr__(
name
):
```
------------------------------------------------------------------------
source
### RishiBackend
``` python
def RishiBackend(
*a, max_steps:int=40, **kw
):
```
*The
[`Backend`](https://vedicreader.github.io/ramabana/runtime.html#backend)
that runs every model, local or hosted, through rishi’s `Chat`.*
------------------------------------------------------------------------
source
### use_chat
``` python
def use_chat(
f
):
```
*Build model conversations with `f` for the duration, instead of rishi’s
`Chat`.*
`use_chat(f)` makes backends build their conversations with `f` instead
of rishi’s `Chat`, for the duration of a `with` block.
[testing](04_testing.ipynb) uses it to put in urai’s `CachedChat`, which
replays a real model’s recorded answers with no weights and no network.
Here `f` is a stub chat that answers from a script, so nothing is
downloaded:
``` python
from ramabana.core import ModelSpec
built = []
class StubChat:
def __init__(self, model, **kw): built.append((model, kw)); self.hist = []
def oneshot(self, prompt, sp='', think=None, max_tokens=None): return 'replayed'
def close(self): pass
with use_chat(StubChat):
be = make_backend(ModelSpec('stub', 'remote', 'openai/gpt-x', 8000))
assert be.start() is not None
test_eq(built[0][0], 'openai/gpt-x')
test_eq(be.oneshot('a cheap job'), 'replayed')
test_eq(be._usage().total, 0)
test_eq(_MK_CHAT, None)
```
Ramabana does not set LiteRT’s constrained decoding, since rishi turns
it on itself when there are tools. A `conv_kw` you pass goes through
unchanged:
``` python
def _a_tool(query: str) -> str:
"A tool."
return ''
kw = make_backend(resolve('gemma-e4b'), tools=[_a_tool])._runtime_kw()
assert 'enable_constrained_decoding' not in kw.get('conv_kw', {})
assert 'constrained_decoding_config' not in kw.get('conv_kw', {})
mine = dict(constrained_decoding_config='mine')
test_eq(make_backend(resolve('gemma-e4b'), tools=[_a_tool], conv_kw=mine)._runtime_kw()['conv_kw'], mine)
```
`make_backend(spec)` is the constructor to call, and it returns a
[`RishiBackend`](https://vedicreader.github.io/ramabana/runtime.html#rishibackend).
The old names `LlamaBackend` and `FastllmBackend` still work as aliases
of
[`RishiBackend`](https://vedicreader.github.io/ramabana/runtime.html#rishibackend),
with a `DeprecationWarning`:
``` python
from litert_lm import Backend as LB
r = make_backend(resolve('gemma-e4b'), eng_kw=dict(backend=LB.GPU()))
import warnings
from ramabana import runtime
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter('always')
assert runtime.LlamaBackend is runtime.FastllmBackend is runtime.RishiBackend
assert [x.category for x in w] == [DeprecationWarning, DeprecationWarning]
r.kind, type(r).__name__
```
('rishi', 'RishiBackend')
For a LiteRT spec the window goes to the engine as `max_num_tokens`. The
cells below also check that `RAMABANA_LITERT_BACKEND` is applied, and
that an explicit backend takes precedence over it:
``` python
r._runtime_kw()
```
``` python
r2 = make_backend(resolve('gemma-e4b'), tools=[interesting])
test_eq(r2._runtime_kw().get('conv_kw'), None)
r2._runtime_kw()['eng_kw']
```
``` python
import os
os.environ['RAMABANA_LITERT_BACKEND'] = 'gpu'
try:
k = make_backend(resolve('gemma-e4b'))._runtime_kw()
test_eq((type(k['backend']), 'backend' in k['eng_kw']), (type(LB.GPU()), False))
k2 = make_backend(resolve('gemma-e4b'), eng_kw=dict(backend=LB.CPU()))._runtime_kw()
test_eq((type(k2['eng_kw']['backend']), 'backend' in k2), (type(LB.CPU()), False))
finally: del os.environ['RAMABANA_LITERT_BACKEND']
```
None of this has started an engine. A backend starts its model on the
first `send`, `stream` or `oneshot`, so building one costs nothing:
``` python
r.ready, r.busy, r.note
```
## Runs
A run is one piece of model work that can be cancelled: the turn the
user started (`kind='root'`), or a sub-agent that the turn delegated to
(`kind='child'`, made with `child`). Runs form a tree. Cancelling a run
cancels everything under it, and the tool calls of the whole tree count
against one budget, through `draw`.
A run starts
[`pending`](https://vedicreader.github.io/ramabana/learned.html#pending),
becomes `running` with `start`, and ends `completed`, `failed`,
`cancelled`, `detached` or `terminated`. `cancel` asks the backend to
stop and waits up to `grace` seconds. A run whose backend does not stop
in time is detached: the caller treats it as over and ignores anything
the model sends later. `terminate` closes the backend outright.
A run also keeps what the user should see while it works. `status` is
the last line the model wrote before its latest tool call. `notes` keeps
up to 100 such messages for the now pane, the live view beside the
terminal ([pane](18_pane.ipynb)), and `write` appends a line to the
run’s transcript file when it has one. `tell` queues a message from the
user for a running sub-agent, and `drain_inbox` hands the queued
messages over. `air` returns a reminder to tell the user where the work
stands once 8 tool calls in a row have gone by with less than 100
characters said before them. `repeated` counts identical calls with
identical results in a row, so the agent can tell a model that it is
looping.
------------------------------------------------------------------------
source
### Run
``` python
def Run(
id:str, kind:str='root', question:str='', model:str='', parent:object=None, grace:float=0.25,
state:str='pending', started:float=0.0, ended:float=0.0, backend:object=None, log:object=None, status:str='',
answer:str=''
)->None:
```
*A foreground or delegated model call with bounded cancellation.*
------------------------------------------------------------------------
source
### said_before_call
``` python
def said_before_call(
hist
):
```
*The text of the message that made the latest tool call in `hist`, or
’’.*
------------------------------------------------------------------------
source
### status_line
``` python
def status_line(
text, n:int=80
):
```
*The last line of `text` worth showing as a status, without bullet or
bold, clipped to `n` chars.*
A run’s status comes from the text streamed since the last tool call
(`hear`), or, when nothing was streamed, from the message in the
backend’s history that made the call.
[`status_line`](https://vedicreader.github.io/ramabana/runtime.html#status_line)
takes the last line worth showing, without list markers or bold, cut to
80 characters:
``` python
class _Said:
hist = [{'role': 'user', 'content': 'go'},
{'role': 'assistant', 'content': 'Thought about it.\nReading the loader.', 'tool_calls': [{'id': '1'}]}]
r = Run('run_status')
r.start(_Said())
test_eq(r.on_call().status, 'Reading the loader.')
r.hear('Checking the '); r.hear('tests.\n> 🧠 hmm')
test_eq(r.on_call().status, 'Checking the tests.')
test_eq(status_line('x ' * 60), ('x ' * 40).strip())
test_eq(status_line('- **Reading** it.\n```'), 'Reading it.')
```
A run whose backend stops when asked ends `cancelled`. A run whose
backend cannot stop ends `detached` once its grace period runs out:
``` python
import threading, time
class _Blocking:
def __init__(self, stoppable=True): self.stoppable, self.released = stoppable, threading.Event()
def cancel(self):
if self.stoppable: self.released.set()
r = Run('run_cancelled', grace=.05)
r.start(_Blocking())
threading.Thread(target=lambda: (r.backend.released.wait(), r.finish()), daemon=True).start()
test_eq(r.cancel().state, 'cancelled')
r = Run('run_detached', grace=.01)
r.start(_Blocking(False))
test_eq(r.cancel().state, 'detached')
```
[`current_run()`](https://vedicreader.github.io/ramabana/runtime.html#current_run)
returns the run that the current thread or async task is working for,
and `run_context(run)` sets it for a `with` block. The agent sets it
around each turn and each sub-agent, so code deep inside a tool call can
find its run without having it passed down.
------------------------------------------------------------------------
source
### run_context
``` python
def run_context(
run
):
```
*Make `run` the current run inside a `with` block.*
------------------------------------------------------------------------
source
### current_run
``` python
def current_run():
```
Once its run is cancelled, a backend’s `stream` stops yielding, drops
what the model sends later, and does not retry:
``` python
class _CancelStream(Backend):
def __init__(self):
from ramabana.core import ModelSpec
super().__init__(ModelSpec('cancel-test', 'fake', 'fake/cancel', ctx=1000))
self.first, self.release, self.calls, self.recoveries = threading.Event(), threading.Event(), 0, 0
def _start(self): return self
def _stream(self, msg, **kw):
self.calls += 1
yield 'first'
self.first.set(); self.release.wait()
yield 'late'
def cancel(self): self.release.set(); return True
def _usage(self): return Usage()
def _recover(self, e): self.recoveries += 1; return True
be, run, chunks = _CancelStream(), Run('run_late', grace=.05), []
run.start(be)
t = threading.Thread(target=lambda: chunks.extend(be.stream('x', run=run)), daemon=True)
t.start(); assert be.first.wait(.2)
run.cancel(); t.join(.2)
test_eq(chunks, ['first'])
test_eq((be.calls, be.recoveries), (1, 0))
```
`terminate` ends a run even after it was detached, and closes its
backend:
``` python
class _Closable:
def __init__(self): self.closed = threading.Event()
def close(self): self.closed.set()
r = Run('run_terminated', grace=0)
r.start(_Closable()); r.detach(); r.terminate()
test_eq(r.state, 'terminated')
assert r.backend.closed.wait(.2)
```
`add_cb` registers a rishi callback class on a backend. It is applied to
the current chat and to any chat that replaces it later.
`takes_pictures` says whether an image returned by a tool can reach the
model: the model has to accept images and its transport has to carry
them. A model that cannot see a picture gets urai’s text note in its
place. `start` is redefined here so that it applies both when the model
starts.
------------------------------------------------------------------------
source
### Backend.start
``` python
def start():
```
------------------------------------------------------------------------
source
### Backend.takes_pictures
``` python
def takes_pictures():
```
*Whether a picture a tool returns can reach this model: it accepts
images and its transport carries them.*
------------------------------------------------------------------------
source
### Backend.add_cb
``` python
def add_cb(
cb
):
```
*Register one Rishi callback class, for this chat and for any that
replaces it.*
[`TokenLogger`](https://vedicreader.github.io/ramabana/runtime.html#tokenlogger)
is a rishi callback that prints the usage the provider reports after
each response. Set `TokenLogger.sink` to send the line somewhere other
than `print`. `CHAT_CALLBACKS` maps the names that
[`Agent.add_chat_callback`](https://vedicreader.github.io/ramabana/agent.html#agent.add_chat_callback)
accepts to their classes.
------------------------------------------------------------------------
source
### TokenLogger
``` python
def TokenLogger(
*args, **kwargs
):
```
*Report Rishi’s provider-reported usage after each response.*
|