core

Backend-agnostic Chat: dispatch by model name, callbacks, HITL tools, streaming, structured output, usage tracking.

Usage tracking

UsageStats tallies prompt/completion/cached tokens per turn. chat.use is the last turn; backends merge into chat.usage. cost/model are always set so local and hosted usage share one type.

---

source

UsageStats

def UsageStats(
    prompt_tokens:int=0, completion_tokens:int=0, total_tokens:int=0, n:int=0, cached_tokens:int=0, cost:float=0.0,
    model:NoneType=None
):

Token usage for a chat turn. cached_tokens is the part of the prompt served from a KV/prefix cache. cost/model are always present (cost is 0 for local inference) so a harness merging local and hosted usage can carry one type instead of two.

from fastcore.test import test_eq, test_fail, test_close
a = UsageStats(prompt_tokens=10, completion_tokens=5, total_tokens=15, n=1)
b = UsageStats(prompt_tokens=3, completion_tokens=2, total_tokens=20, n=1)
c = a + b
test_eq((c.prompt_tokens, c.completion_tokens, c.n), (13, 7, 2))
assert "in=13" in repr(c) and "out=7" in repr(c)
test_eq((a + None).prompt_tokens, 10)

# cost/model default to 0/None and stay out of the repr until something sets them
test_eq((a.cost, a.model), (0.0, None))
assert 'cost' not in repr(a) and 'model' not in repr(a)
d = UsageStats(prompt_tokens=1, cost=0.02, model='gemma-4-e2b') + UsageStats(prompt_tokens=1, cost=0.03)
test_eq(d.cost, 0.05)
test_eq(d.model, 'gemma-4-e2b')                       # first non-None model wins
assert 'cost=$0.0500' in repr(d) and 'model=gemma-4-e2b' in repr(d)

# cached_tokens sums like the other counters, and only shows up once a backend reports one
test_eq(a.cached_tokens, 0)
assert 'cached' not in repr(a)
e = UsageStats(prompt_tokens=100, cached_tokens=90) + UsageStats(prompt_tokens=100, cached_tokens=80)
test_eq(e.cached_tokens, 170)
assert 'cached=170' in repr(e)

Callbacks

Subclass ChatCallback, hook before_send / after_response / before_tool_calls / after_tool_calls. order controls sequencing; run_cbs dispatches. State lives on self.chat via GetAttr.

---

source

run_cbs

def run_cbs(
    chat, event
):

Dispatch event to enabled callbacks in order; forward any yielded stream items.

---

source

ChatCallback

def ChatCallback(
    *args, **kwargs
):

Base chat callback; reads chat state via GetAttr (self.turn_msg -> chat.turn_msg).

class _Dummy: pass
class _A(ChatCallback):
    order = 20
    def before_send(self): self.chat.log.append('A')
class _B(ChatCallback):
    order = 10
    def before_send(self):
        self.chat.log.append('B')
        yield {'text': 'from-B'}
d = _Dummy(); d.log = []
a, b = _A(), _B(); a.chat = d; b.chat = d
d.cbs = L(a, b)
out = list(run_cbs(d, 'before_send'))
test_eq(d.log, ['B', 'A'])
test_eq(out, [{'text': 'from-B'}])
test_eq(repr(a), '_A')

Streaming display

StreamFormatter turns token chunks into markdown for notebooks. display_stream(gen) renders live; adisplay_stream is the async twin.

---

source

display_stream

def display_stream(
    chunks
):

Progressively render a markdown-chunk stream (e.g. chat(msg, stream=True)) live in a notebook; returns the full markdown.

---

source

StreamFormatter

def StreamFormatter(
    mx:int=2000, showthink:bool=True
):

Format a litert response stream to markdown; thinking streams as a blockquote.

---

source

mk_tr_details

def mk_tr_details(
    name, args, result, mx:int=2000
):

<details> JSON block for a completed tool call.

---

source

tc_summary_

def tc_summary_(
    name, args, result:NoneType=None
):

One-line <code> summary of a tool call.

---

source

Resp

def Resp(
    *args, **kwargs
):

A litert response dict that renders as markdown (thinking + text + tool calls/responses) in notebooks.

---

source

quote_

def quote_(
    text
):

Render text as a markdown blockquote with a Thinking header (renders in any md engine).

---

source

thought

def thought(
    resp
):

The model’s thinking (channels.thought) for a litert response/chunk, or ’’.

---

source

resp_text

def resp_text(
    resp
):

Join text parts of a litert response/chunk dict.

Built-in callbacks

UsageCallback accumulates tokens. ToolReminderCallback nudges the model when tools exist but weren’t called. TruncationCallback clips over-long tool results before they hit the context.

---

source

TruncationCallback

def TruncationCallback(
    max_tokens
):

Flag turn_res['truncated'] when the reply reaches max_tokens output tokens (best-effort).

---

source

truncated

def truncated(
    resp
):

Whether resp was flagged as cut off at the token cap.

OpenAI-style messages

Helpers to build OpenAI-shaped message dicts (mk_oai_content, mk_oai_msg, mk_oai_msgs) used by llama.cpp and as the canonical interchange format. Media items become placeholders in past turns via strip_media.

---

source

is_media

def is_media(
    p
):

Is p an image or audio content part?

---

source

mk_oai_msgs

def mk_oai_msgs(
    msgs
):

Normalize a list of messages to OpenAI-style dicts.

---

source

mk_oai_msg

def mk_oai_msg(
    content, role:str='user'
):

Create an OpenAI-style message dict from str/bytes/list/dict.

---

source

mk_oai_content

def mk_oai_content(
    o
):

Convert o to an OpenAI-style content part (text, a base64 image_url, or an input_audio).

from fastcore.test import test_eq, test_fail

test_eq(mk_oai_msg("hello"), {'role': 'user', 'content': 'hello'})
test_eq(mk_oai_msg(["a", "b"]), {'role': 'user', 'content': 'a\nb'})
test_eq(mk_oai_msg({'role': 'assistant', 'content': 'hi'}), {'role': 'assistant', 'content': 'hi'})
test_eq(mk_oai_msgs(['hi', {'role': 'assistant', 'content': 'hello'}]),
        [{'role': 'user', 'content': 'hi'}, {'role': 'assistant', 'content': 'hello'}])
png = b'\x89PNG\r\n\x1a\n' + b'0' * 8
assert mk_oai_msg([png, 'what is this?'])['content'][0]['type'] == 'image_url'
test_eq(mk_oai_content(b'RIFF0000WAVE')['type'], 'input_audio')          # audio is sniffed too
test_fail(lambda: mk_oai_content(b'%PDF-1.4 junk'), contains='text, image, and audio')

Tool schemas

mk_toolspec(fn) reads signatures and docstrings into a JSON schema for the model.

---

source

mk_toolspec

def mk_toolspec(
    f
):

OpenAI-style tool spec for callable f (via fastcore.funccall.get_schema); spec dicts pass through.

def _add(
    a:int, # first addend
    b:int=0 # second addend
):
    'Add two integers.'
    return a + b

s = mk_toolspec(_add)
test_eq(s['type'], 'function')
test_eq(s['function']['name'], '_add')
test_eq(s['function']['parameters']['required'], ['a'])
test_eq(mk_toolspec(s), s)                      # spec dicts pass through

# `get_schema` adds a stray `title` for classes/dataclasses (the `structured` path); it must not reach
# the rendered tool listing or llama.cpp's JSON-schema grammar builder
from dataclasses import dataclass

@dataclass
class _Point:
    'A 2d point'
    x: int  # the x coord
    y: int  # the y coord

p = mk_toolspec(_Point)
test_eq(p['function']['name'], '_Point')
test_eq(p['function']['parameters']['required'], ['x', 'y'])
assert 'title' not in p['function']['parameters'], p['function']['parameters']

Think tags and tool-call tags

split_think separates <think> from answer text. Tag-mode backends parse <tool_call> blocks via parse_tool_tags / mk_tag_tc.

---

source

render_prompt

def render_prompt(
    hist, sp:str=''
):

A conversation as one block of text, for a transport that takes a prompt rather than a message list.

---

source

est_tokens

def est_tokens(
    text
):

Rough tokens in text, for a transport with no tokenizer of its own.

---

source

tag_tools_sp

def tag_tools_sp(
    toolspecs, sp:str='',
    template:str='\n\n# Tools\n\nYou can call the functions below. Their signatures are given as JSON schemas inside <tools></tools>:\n\n<tools>\n{tools}\n</tools>\n\nTo call one, emit a JSON object with the function\'s name and its arguments inside <tool_call></tool_call>, and then stop and wait for the result:\n\n<tool_call>\n{{"name": "the_function_name", "arguments": {{"first": "value"}}}}\n</tool_call>\n\nCall one function at a time. Do not describe the call in prose as well as emitting it, and never invent a result -- the real one comes back in the next message.'
):

sp plus the tag protocol and toolspecs as JSON, for a transport that can’t carry tools.

---

source

parse_tool_tags

def parse_tool_tags(
    text
):

Parse Hermes/Qwen-style <tool_call>{json}</tool_call> blocks; returns (clean_text, tool_calls).

---

source

mk_tag_tc

def mk_tag_tc(
    s
):

Build a tool_call dict from the JSON inside a <tool_call> block, or None.

---

source

split_think

def split_think(
    text
):

Split <think>...</think> blocks out of text; returns (clean_text, thought).

test_eq(split_think('<think>hmm</think>\n\nhi'), ('hi', 'hmm'))
test_eq(split_think('no tags'), ('no tags', ''))
test_eq(split_think('<think>cut off mid-'), ('', 'cut off mid-'))
txt, tcs = parse_tool_tags('calling now\n<tool_call>{"name": "add", "arguments": {"a": 1}}</tool_call>')
test_eq(txt, 'calling now')
test_eq(tcs[0]['function'], {'name': 'add', 'arguments': {'a': 1}})
test_eq(parse_tool_tags('<tool_call>not json</tool_call>')[1], [])

spec = mk_toolspec(_add)
sp = tag_tools_sp([spec], 'Be terse.')
assert sp.startswith('Be terse.') and '<tool_call>' in sp and '"name": "_add"' in sp
test_eq(tag_tools_sp([], 'Be terse.'), 'Be terse.')      # no tools, nothing to teach
# what the prompt asks for is what the parser reads back
test_eq(parse_tool_tags(sp)[1][0]['function'], {'name': 'the_function_name', 'arguments': {'first': 'value'}})

Response normalization

norm_resp maps backend completion dicts to Resp. to_oai_msg converts canonical history entries back to wire messages.

---

source

strip_media

def strip_media(
    m
):

Replace media parts with a text placeholder and collapse content to a string, so past-turn media isn’t re-encoded.

---

source

sum_usage

def sum_usage(
    us
):

Sum OpenAI usage dicts (ignoring Nones); None if nothing to sum. A model name, if any, is carried through rather than summed.

---

source

to_oai_msg

def to_oai_msg(
    m
):

Project a history entry to an OpenAI-style wire message: drop rishi-only keys, re-encode tool-call args as JSON.

---

source

norm_resp

def norm_resp(
    r
):

Normalize an OpenAI-style chat completion to a rishi-style Resp dict.

---

source

parse_args

def parse_args(
    a
):

Parse OpenAI JSON-string tool arguments to a dict (dicts pass through).

raw = {'choices': [{'message': {'role': 'assistant',
                               'content': '<think>2+3</think>The answer is 5.\n<tool_call>{"name": "add", "arguments": {"a": 2, "b": 3}}</tool_call>',
                               'tool_calls': [{'id': 'x1', 'type': 'function',
                                               'function': {'name': 'mul', 'arguments': '{"a": 4}'}}]},
                    'finish_reason': 'stop'}],
       'usage': {'prompt_tokens': 10, 'completion_tokens': 5, 'total_tokens': 15}}
r = norm_resp(raw)
test_eq(resp_text(r), 'The answer is 5.')
test_eq(thought(r), '2+3')
test_eq([tc['function']['name'] for tc in r['tool_calls']], ['mul', 'add'])
test_eq(r['tool_calls'][0]['function']['arguments'], {'a': 4})
assert not truncated(r)

o = to_oai_msg(r)
assert 'channels' not in o and 'usage' not in o
test_eq(o['tool_calls'][0]['function']['arguments'], '{"a": 4}')
test_eq(to_oai_msg({'role': 'tool', 'tool_call_id': 't1', 'name': 'add', 'content': '5'}),
        {'role': 'tool', 'tool_call_id': 't1', 'name': 'add', 'content': '5'})
test_eq(sum_usage([{'prompt_tokens': 1, 'completion_tokens': 2, 'total_tokens': 3}, None,
                    {'prompt_tokens': 10, 'completion_tokens': 20, 'total_tokens': 30}]),
        {'prompt_tokens': 11, 'completion_tokens': 22, 'total_tokens': 33, 'cached_tokens': 0})
test_eq(sum_usage([{'cached_tokens': 5}, {'cached_tokens': 7}])['cached_tokens'], 12)
test_eq(sum_usage([None, None]), None)
test_eq(sum_usage([{'model': 'a'}, {'model': 'b'}])['model'], 'a')   # carried, not summed
assert 'model' not in sum_usage([{'prompt_tokens': 1}])

Streaming text splitter

StreamSplit incrementally splits a token stream on think/tool/answer boundaries for live display.

---

source

acc_tc

def acc_tc(
    acc, deltas
):

Fold streamed OpenAI tool_calls deltas into acc (a list of partial tool_call dicts).

---

source

StreamSplit

def StreamSplit():

Stateful splitter for streamed text: <think> -> thought chunks, <tool_call> blocks held back and parsed.

def _run_split(deltas):
    sp, chunks = StreamSplit(), []
    for d in deltas: chunks += list(sp.feed(d))
    chunks += list(sp.finish())
    return sp, chunks

# tags split across chunk boundaries
sp, chunks = _run_split(['<thi', 'nk>let me', ' see</th', 'ink>\n\nans', 'wer: 5 <tool', '_call>{"name": "add", "arguments": {"a": 1}}</tool_call>'])
test_eq(sp.thought, 'let me see')
test_eq(sp.text, 'answer: 5 ')
test_eq(sp.tool_calls[0]['function'], {'name': 'add', 'arguments': {'a': 1}})
assert all(('channels' in c) or ('content' in c) for c in chunks)

# plain text passes through; unterminated think flushes to thought
sp, _ = _run_split(['just ', 'text'])
test_eq((sp.text, sp.thought, sp.tool_calls), ('just text', '', []))
sp, _ = _run_split(['<think>never closed'])
test_eq(sp.thought, 'never closed')

# a lone '<' that never becomes a tag is still emitted
sp, _ = _run_split(['a < b', ' and c'])
test_eq(sp.text, 'a < b and c')

acc = []
acc_tc(acc, [{'index': 0, 'id': 'c1', 'function': {'name': 'add', 'arguments': ''}}])
acc_tc(acc, [{'index': 0, 'function': {'arguments': '{"a"'}}])
acc_tc(acc, [{'index': 0, 'function': {'arguments': ': 2}'}}])
test_eq(acc, [{'id': 'c1', 'type': 'function', 'function': {'name': 'add', 'arguments': '{"a": 2}'}}])

Shared built-in callbacks

UsageCallback, ToolReminderCallback, and SlidingWindowCallback register on every backend by default (default_cbs=True). Drop or replace with remove_cb / default_cbs=False.

---

source

ToolReminderCallback

def ToolReminderCallback(
    tool_reminder:str='\n<system-reminder>After every tool call result, briefly summarise in prose what you found before continuing or calling another tool.</system-reminder>'
):

Inject a tool-summary reminder into the outgoing message when tools are registered.

---

source

UsageCallback

def UsageCallback(
    *args, **kwargs
):

Fold each turn’s usage block (summed across tool rounds) into chat.use.

Chat and dispatch

Reusing a token prefix

common_prefix_len finds shared token/id prefixes so backends can reuse KV state between turns.

---

source

common_prefix_len

def common_prefix_len(
    a, b
):

Length of the longest common prefix of sequences a and b.

test_eq(common_prefix_len([1, 2, 3, 4], [1, 2, 9]), 2)
test_eq(common_prefix_len([1, 2], [1, 2, 3]), 2)       # one is a prefix of the other
test_eq(common_prefix_len([], [1, 2]), 0)
test_eq(common_prefix_len([9], [1]), 0)

---

source

get_runtime

def get_runtime(
    nm
):

The Chat subclass for runtime nm (imports the backend module lazily).

---

source

resolve_runtime

def resolve_runtime(
    model:NoneType=None, runtime:NoneType=None, model_path:NoneType=None
):

Resolve (runtime, model) from an explicit runtime, a runtime/ prefix, or the id/path shape.

---

source

infer_runtime

def infer_runtime(
    model
):

Guess a runtime from the shape of a model id or path (.litertlm vs .gguf), else None.

---

source

split_runtime

def split_runtime(
    model
):

Split 'runtime/model' into (runtime, model); the prefix must name a known runtime, else (None, model).

Model capabilities

model_caps(model) says what a model accepts and hands back without loading it, so a picker can label everything it lists. Cloud from fastllm’s bundled model table, GGUF from an mmproj beside the weights, MLX from the towers in config.json.

Caps.source records who answered — the table leaves the modality fields empty on rows it still carries supports_vision for, so a caller needs to tell unknown from no.

---

source

model_caps

def model_caps(
    model:NoneType=None, runtime:NoneType=None, model_path:NoneType=None
):

What model accepts and what it returns, resolved without loading it.

---

source

Caps

def Caps(
    inp:tuple=('text',), out:tuple=('text',), tools:tuple=(), source:str='default'
)->None:

What a model can be sent, and what it can send back.

c = Caps(('text', 'image'), ('text', 'image'), source='litellm')
test_eq((c.gen_image, c.gen_video, c.known), (True, False, True))
test_eq(c.fmt(), 'in: text image \u00b7 out: text image')
assert c.accepts('image') and not c.accepts('audio')

test_eq(Caps(source='litellm').fmt(), '')
test_eq(Caps().known, False)
test_eq(Caps().fmt(), 'modalities unknown')

# a chat model reaches image generation through the Responses tool, which the table's
# `supported_output_modalities` never says -- gpt-5.6-luna reports text-out and draws anyway
luna = _tbl_caps('openai/gpt-5.6-luna')
test_eq(luna.out, ('text',))
test_eq(luna.tools, ('image',))
assert luna.gen_image
assert 'via tool: image' in luna.fmt()

test_eq(_fallback_caps('claude-opus-4-5'), Caps(('text', 'image'), ('text',), (), 'fallback'))
test_eq(_fallback_caps('gpt-5'), None)
test_eq(model_caps(runtime='litert', model='google/gemma-3n-E2B-it-litert-lm').inp,
        ('text', 'image', 'audio'))

Tool calls and tool results

ToolCall records name/args/id. mk_tool_res_msg builds the tool-result message appended after execution.

---

source

mk_tool_res_msgs

def mk_tool_res_msgs(
    tcs, results
):

Canonical tool-result messages for several calls at once.

---

source

mk_tool_res_msg

def mk_tool_res_msg(
    tc, result
):

Canonical role='tool' message carrying result back for tool call tc.

---

source

tc_name

def tc_name(
    tc
):

Name of a tool call, however it was built.

---

source

ToolCall

def ToolCall(
    name:str='', arguments:NoneType=None, id:NoneType=None, server:bool=False
):

One tool call in canonical form. A dict subclass, so anything that indexes it keeps working.

tc = ToolCall('add', {'a': 1, 'b': 2})
test_eq((tc.name, tc.arguments), ('add', {'a': 1, 'b': 2}))
test_eq(tc['function']['name'], 'add')          # still just a dict underneath
assert tc['id'].startswith('call_')
test_eq(ToolCall('x', id='fixed')['id'], 'fixed')
test_eq(tc.server, False)
assert 'server' not in tc                       # not carried onto the wire unless it's true
assert ToolCall('web_search', server=True).server

test_eq(mk_tool_res_msg(tc, 3),
        {'role': 'tool', 'tool_call_id': tc['id'], 'name': 'add', 'content': '3'})
test_eq([m['content'] for m in mk_tool_res_msgs([tc, tc], [1, 2])], ['1', '2'])
test_eq(tc_name({'function': {'name': 'f'}}), 'f'); test_eq(tc_name({}), '')

Tool-call budget and context recovery

max_steps caps tool calls per turn; past the cap calls are denied and final_prompt asks for a prose answer.

If the context overflows mid-turn, recover_context truncates oldest tool results, rebuilds backend state, and retries once before raising ContextWindowExceededError.

---

source

is_ctx_error

def is_ctx_error(
    chat, e
):

Best-effort: does e, raised from a backend step, look like the context window filling up?

---

source

ContextWindowExceededError

def ContextWindowExceededError(
    *args, **kwargs
):

Raised when a turn can’t finish because the context window filled up and truncate-and-retry recovery also failed.

---

source

Chat

def Chat(
    *args, **kwargs
):

Backend-agnostic chat: Chat(model) dispatches to the litert/llama/mlx subclass by runtime/model shape.

Reconfigure and one-shots

reconfigure(sp=, tools=) changes briefing or tools without clearing hist.

oneshot(prompt, …) is a stateless side call (classify-sized jobs). think=False matters on reasoning models with tiny max_tokens budgets.

The tool loop

ToolLoopMixin runs the Python-side tool loop for llama/MLX/cursor: approval, budget, optional parallel_tools, and context recovery. Backends supply _model_step / _stream_step and a tool namespace.

---

source

ToolLoopMixin

def ToolLoopMixin(
    *args, **kwargs
):

The Python-side tool loop, for backends that get tool calls back as data (llama, mlx).

The backend supplies one wire call - _model_step(max_output_tokens) -> Resp and _stream_step(max_output_tokens), a generator of chunk dicts that leaves the merged Resp on self._step_res - plus a ns tool namespace. Backends holding conversation state of their own also override _recreate_conv.

Keeping the context window from filling up

SlidingWindowCallback drops whole message groups from the middle of hist when pct_full crosses a threshold — keeping early and recent turns, never splitting a tool call from its result. summarize=True replaces evicted turns with one summary call. Registered by default on every backend.

---

source

SlidingWindowCallback

def SlidingWindowCallback(
    threshold:float=0.9, # evict once the context is this full
    keep_first:int=2, # leading groups to anchor (the task, usually)
    keep_last:int=8, # trailing groups to keep (the live thread)
    summarize:bool=False, # spend one model call to replace the dropped middle with a summary
    mx:int=4000, # chars of dropped conversation to feed the summarizer
):

Evict the middle of hist before a turn that would overflow the context window.

Needs chat.ctx_limit to be set - without a limit there is nothing to measure pct_full against, and the callback stays out of the way.

---

source

evict_middle

def evict_middle(
    hist, keep_first:int=2, keep_last:int=8
):

Drop whole groups from the middle of hist, keeping the first and last few. Returns (new_hist, dropped_msgs).

---

source

msg_groups

def msg_groups(
    hist
):

Split hist into atomic groups: an assistant message with tool_calls stays with the tool results that answer it.

Testing the tool loop without a model

The hide cells below drive ToolLoopMixin with a fake backend — no weights required.

---

source

adisplay_stream

async def adisplay_stream(
    chunks
):

Async twin of display_stream: drive an async chunk stream - or the awaitable that yields one (achat(msg, stream=True)) - to completion, live-rendering in a notebook when possible; returns the full markdown.

---

source

AsyncChat

def AsyncChat(
    model:NoneType=None, runtime:NoneType=None, **kw
):

Async twin of Chat for either backend; blocking calls run in a worker thread.

Draining an async stream

AsyncChat(msg, stream=True) returns an async iterator; .value holds the final Resp. Sync: wrap chat(msg, stream=True) in SaveReturn and read .value after iteration.

Managing callbacks

add_cb, remove_cb (by instance or class), and per-call cbs= on a single turn.

Human-in-the-loop tool approval

approve(tool_call) -> bool runs before each tool. hitl_policy maps tool names to approved / dont_run / check. browser=True routes check to Leela’s approval card (LEELA_URL, default http://127.0.0.1:5001).

---

source

hitl_policy

def hitl_policy(
    modes, ask:NoneType=None, browser:bool=False
):

Build an approve(tool_call) from per-tool modes; optionally ask in the Leela browser.

---

source

browser_approval

def browser_approval(
    url:NoneType=None, timeout:int=300
):

An approval callback that asks through a running Leela web IDE.

Running python from replies

PyFenceCallback runs the last python fence through `safepyrun`, feeds aresult block back, and loops until prose or done. Same approve gate as tools. chat.run_py(code) executes directly in the persistent sandbox.

---

source

PyFenceCallback

def PyFenceCallback(
    max_rounds:int=5, done:NoneType=None, pyrun:callable=None
):

Run ``python fences, feed results back, loop untildone(chat)(default: no fence).max_rounds` is a safety cap.

---

source

output_matches

def output_matches(
    expected
):

done policy: stop once the last code output contains str(expected) (e.g. the answer to match against).

---

source

task_complete

def task_complete(
    chat
):

done policy: judge (via classify, isolated) whether the latest result completes the request.

---

source

Chat.run_py

def run_py(
    code, ban_defs:bool=False, g:NoneType=None
):

Run code in this chat’s sandboxed, persistent namespace; return stdout + last-expr repr.

---

source

sync_iter

def sync_iter(
    agen_fn
):

Drive the async generator returned by agen_fn() from sync code, yielding its items.

The whole generator runs on one event loop in one background thread, rather than a fresh loop per item: a streaming HTTP response is bound to the loop that opened it, so pumping it with repeated asyncio.run calls would tear the connection down mid-stream.

---

source

run_coro

def run_coro(
    coro
):

Run an awaitable to completion from sync code, even inside a running event loop.

---

source

mk_result_fence

def mk_result_fence(
    out
):

Feed a code result back, prompting a prose answer or more code.

---

source

matches_

def matches_(
    actual, expected
):

True if actual contains any value in expected (a scalar, or a list of accepted values).

---

source

extract_fence

def extract_fence(
    text, tag:str='answer'
):

Contents of the last ``<tag> fence intext`, else the whole stripped text.

---

source

extract_code

def extract_code(
    text
):

Code of the last ``python fence intext`, else None.

Utilities: classify, structured

classify and structured run isolated one-shot turns on the same engine without touching hist.

---

source

Chat.structured

def structured(
    prompt, schema, sp:str='Reply with only a JSON object matching the schema.'
):

One-shot structured output; returns schema(...), rebuilding nested dataclasses.

---

source

Chat.classify

def classify(
    text, labels, sp:str='Reply with only the single best label and nothing else.'
):

One-shot label for text, run stateless (isolated from the conversation).

Grading answers

chat.check(question, expected) pulls an ``answer fence and grades with [matches_](https://vedicreader.github.io/rishi/core.html#matches_) by default. Passgrade_fn,llm_judge=True, orjudge=another [Chat`](https://vedicreader.github.io/rishi/core.html#chat) for model grading.

---

source

Chat.check

def check(
    question, expected, grade_fn:function=matches_, llm_judge:bool=False, judge:NoneType=None, tag:str='answer',
    sp:str='Answer the question, then put your final answer inside a ```answer fence.'
):

Ask question stateless, extract the ``<tag> answer, and grade it againstexpected`.

---

source

Chat.grades

def grades(
    question, expected, actual
):

LLM-as-judge (on this chat’s engine): is actual a correct answer to question given reference expected?

from fastcore.test import test_eq, test_fail
test_eq(split_runtime('llama/Qwen/Qwen3-4B-GGUF'), ('llama', 'Qwen/Qwen3-4B-GGUF'))
test_eq(split_runtime('litert-community/gemma-4-E2B-it-litert-lm'), (None, 'litert-community/gemma-4-E2B-it-litert-lm'))
test_eq(infer_runtime('Qwen/Qwen3-0.6B-GGUF'), 'llama'); test_eq(infer_runtime('/m/x.litertlm'), 'litert')
test_eq(infer_runtime('mlx-community/Qwen3-4B-4bit'), 'mlx')
test_eq(split_runtime('mlx/Qwen3-4B'), ('mlx', 'Qwen3-4B'))
test_eq(resolve_runtime('llama/my'), ('llama','my')); test_eq(resolve_runtime(), ('litert', None))
test_eq(resolve_runtime('mlx-community/Qwen3-4B-4bit')[0], 'mlx')
test_fail(lambda: resolve_runtime('gemma-4-E2B'), contains="Can't tell which backend")
test_fail(lambda: resolve_runtime('x', runtime='vllm'), contains='Unknown runtime')
from dataclasses import dataclass
@dataclass
class _A: year:int; month:int
@dataclass
class _P: name:str; age:_A
test_eq(_mk_obj(_P, {'name':'Alice','age':{'year':1995,'month':6}}).age, _A(1995,6))

class _FakeChat(Chat):
    _runtime='litert'
    def __init__(self, **kw): self.ctx_limit=100; self._setup(**kw)
    def mk_msgs(self, msgs): return list(msgs or [])
    def _oneshot(self, prompt, sp='', think=None, max_tokens=None):
        self.asked = think; return 'positive because good'
    def _structured_call(self, prompt, schema, sp): return {'name':'Alice','age':{'year':1995,'month':6}}
    def _send(self, msg, mot=None): return Resp({'role':'assistant','content':'ok'})
    def close(self): pass
c=_FakeChat(sp='hi')
test_eq(c.classify('great', ['positive','negative']), 'positive')
test_eq(c.asked, False)                                      # a label is a cheap job, so: don't think
test_eq(c.oneshot('hi', think=True, max_tokens=8), 'positive because good')
test_eq(c.structured('x', _P).age, _A(1995,6))
test_eq(run_coro(AsyncChat(c)('go')), Resp({'role':'assistant','content':'ok'}))

# dispatch runs the *subclass's* __new__, so a backend can re-route further (mlx -> mlx-vlm)
import sys, types
class _FakeSub(Chat):
    _runtime = 'fakert'
    def __new__(cls, model=None, **kw): return object.__new__(_FakeDeeper if kw.get('deep') else cls)
class _FakeDeeper(_FakeSub): pass
_m = types.ModuleType('_fake_backend'); _m._FakeSub = _FakeSub
sys.modules['_fake_backend'] = _m
runtimes['fakert'] = ('_fake_backend', '_FakeSub')
try:
    test_eq(type(Chat.__new__(Chat, 'x', runtime='fakert')), _FakeSub)
    test_eq(type(Chat.__new__(Chat, 'x', runtime='fakert', deep=True)), _FakeDeeper)
    test_eq(type(_FakeSub.__new__(_FakeSub)), _FakeSub)          # direct construction is untouched
finally: del runtimes['fakert'], sys.modules['_fake_backend']

# hosted model names route to the remote backend, and local shapes still win
test_eq(infer_runtime('claude-sonnet-4-5'), 'remote')
test_eq(infer_runtime('gpt-5.5'), 'remote')
test_eq(infer_runtime('anthropic/claude-opus-4-5'), 'remote')
test_eq(infer_runtime('mlx-community/Qwen3-4B-4bit'), 'mlx')      # a local repo, not a hosted name
test_eq(split_runtime('remote/claude-sonnet-4-5'), ('remote', 'claude-sonnet-4-5'))

# sync_iter drains an async generator from sync code, and propagates its exceptions
async def _agen():
    for i in range(3): yield i
test_eq(list(sync_iter(_agen)), [0, 1, 2])
async def _aboom():
    yield 1
    raise ValueError('boom')
test_fail(lambda: list(sync_iter(_aboom)), contains='boom')

Recording a chat for CI replay

RecordCache stores turn results in a diskcache directory; replays skip the engine entirely. A miss raises unless RISHI_RECORD_CHAT=1 (or record=True). Keys hash model, briefing, tools, and conversation (KEY_VERSION bumps invalidate all recordings). CachedChat wraps Chat for docs/CI. Transient errors (timeouts, rate limits) are never recorded.

---

source

CachedChat

def CachedChat(
    model:NoneType=None, # anything `Chat` takes; part of the key
    path:NoneType=None, # the diskcache directory; None -> `CHAT_CACHE`
    record:NoneType=None, # let a miss reach a real model; None -> $RISHI_RECORD_CHAT
    sp:str='', # system prompt, part of the key
    tools:NoneType=None, # tool names are part of the key; the real chat gets the tools themselves
    **kw
):

A Chat whose replies are recorded to disk and replayed on a second ask; a replay builds no engine.

---

source

RecordCache

def RecordCache(
    path:NoneType=None, # the diskcache directory; None -> `CHAT_CACHE`
    record:NoneType=None, # let a miss run for real; None -> the environment
    env:str='RISHI_RECORD_CHAT', # what says a miss may run
    version:NoneType=None, # part of every key; None -> `KEY_VERSION`
):

Record what a call returned the first time and replay it every time after; the primitive under CachedChat.

---

source

is_transient

def is_transient(
    e
):

Is e a failure of the moment rather than of the ask? Those are not worth remembering.

Installing the skill

mv_skill_md() copies bundled skill.md into standard agent skill directories (.claude/skills/rishi/, .agents/skills/rishi/). dry_run=True prints paths only.

---

source

mv_skill_md

def mv_skill_md(
    dry_run:bool=True, dir:NoneType=None
):

Copy the bundled skill.md into .claude and .agents skill dirs so a harness can load the rishi skill.

---

source

repo_root

def repo_root()->Path:

Root of the current git repository, or None if not in one.