remote

Hosted models through fastllm — same Chat API with API keys in the environment (ANTHROPIC_API_KEY, OPENAI_API_KEY, GEMINI_API_KEY, …).

from fastcore.test import test_eq, test_fail, test_close

Messages

fastllm’s canonical message is a Msg(role, content=[Part, ...]), where a Part is typed: text, thinking, tool_use, tool_result, input_image/input_audio (whose payload is a data URL in part.text). rishi’s canonical history is OpenAI-shaped dicts. Neither is richer than the other in practice, so this is a straight two-way mapping - and it is where media fidelity across a backend hop actually gets decided, so images and audio are carried through as data URLs rather than collapsed to a placeholder.

Note thinking round-trips properly here: rishi keeps it in channels.thought, fastllm keeps it as a thinking Part, and each converts to the other.

---

source

to_hist

def to_hist(
    m
):

A fastllm Msg -> canonical rishi history dicts (a tool Msg can hold several results).

---

source

to_msg

def to_msg(
    m
):

One canonical rishi history dict -> a fastllm Msg.

# a round trip through fastllm's Msg keeps text, thinking, tool calls and tool results intact
h = [{'role': 'user', 'content': 'what is 2+2?'},
     {'role': 'assistant', 'content': 'let me add', 'channels': {'thought': 'hmm'},
      'tool_calls': [ToolCall('add', {'a': 2, 'b': 2}, id='c1')]},
     {'role': 'tool', 'tool_call_id': 'c1', 'name': 'add', 'content': '4'},
     {'role': 'assistant', 'content': 'it is 4'}]
msgs = [to_msg(m) for m in h]
test_eq([m.role for m in msgs], ['user', 'assistant', 'tool', 'assistant'])
back = [x for m in msgs for x in to_hist(m)]
test_eq(back[0], {'role': 'user', 'content': 'what is 2+2?'})
test_eq(back[1]['channels'], {'thought': 'hmm'})
test_eq(back[1]['tool_calls'][0].name, 'add')
test_eq(back[1]['tool_calls'][0].arguments, {'a': 2, 'b': 2})
test_eq(back[2], {'role': 'tool', 'tool_call_id': 'c1', 'name': 'add', 'content': '4'})
test_eq(back[3], {'role': 'assistant', 'content': 'it is 4'})

# media survives the hop as a data URL, rather than being collapsed to a placeholder
png = b'\x89PNG\r\n\x1a\n' + b'0' * 8
um = mk_oai_msg([png, 'what is this?'])
m = to_msg(um)
test_eq([p.type for p in m.content], ['input_image', 'text'])
assert m.content[0].text.startswith('data:image/png;base64,')
rt = to_hist(m)[0]
test_eq(rt['content'][1]['image_url']['url'], m.content[0].text)

# audio too
wav = mk_oai_msg([b'RIFF0000WAVE', 'transcribe'])
am = to_msg(wav)
test_eq([p.type for p in am.content], ['input_audio', 'text'])
test_eq(to_hist(am)[0]['content'][1]['input_audio']['format'], 'wav')

# a server-side tool call keeps its flag across the hop
sm = to_msg({'role': 'assistant', 'content': '', 'tool_calls': [ToolCall('web_search', {}, id='s1', server=True)]})
assert to_hist(sm)[0]['tool_calls'][0].server

Responses and usage

norm_completion maps fastllm completions to rishi Resp. UsageStats.cached_tokens is populated when the provider reports prompt caching.

---

source

norm_completion

def norm_completion(
    comp
):

fastllm Completion -> rishi Resp, with <tool_call> tags read out of the text as core.norm_resp does.

---

source

norm_usage

def norm_usage(
    u, model:NoneType=None
):

fastllm Usage -> rishi UsageStats.

RemoteChat

Async wire via run_coro / sync_iter. Hosted-only passthrough: tool_choice, reasoning_effort, and tool_mode='tags'|'native'. Provider-run tools return server=True and are recorded, not executed locally.

---

source

RemoteChat

def RemoteChat(
    model:NoneType=None, runtime:NoneType=None, model_path:NoneType=None, api_key:NoneType=None,
    base_url:NoneType=None, vendor_name:NoneType=None, api_name:NoneType=None, sp:str='', messages:NoneType=None,
    tools:NoneType=None, ctx_limit:NoneType=None, approve:NoneType=None, tool_max_len:NoneType=None,
    max_steps:int=10, parallel_tools:bool=False, max_parallel_tools:NoneType=None,
    final_prompt:str="You've reached the tool-call budget for this turn. Stop calling tools and answer with what you already have.",
    tool_choice:NoneType=None, reasoning_effort:NoneType=None, temp:NoneType=None, max_output_tokens:int=4096,
    retries:int=2, comp_kw:NoneType=None, cbs:NoneType=None, default_cbs:bool=True,
    tool_mode:str='native', # 'native' sends schemas on the wire; 'tags' puts them in the system prompt
):

Chat against a hosted model through fastllm - the same rishi.core.Chat API as the local backends.

def _add(a: int, b: int) -> int:
    'Add a and b.'
    return a + b

# native puts the schemas on the wire; tags puts them in the system prompt and leaves the field empty
kw = RemoteChat('gpt-5.1', tools=[_add], sp='Be terse.')._kw()
test_eq([t['function']['name'] for t in kw['tools']], ['_add'])
test_eq(kw['system'], 'Be terse.')

kw = RemoteChat('gpt-5.1', tools=[_add], sp='Be terse.', tool_mode='tags', tool_choice='required')._kw()
assert 'tools' not in kw and 'tool_choice' not in kw
assert kw['system'].startswith('Be terse.') and '"name": "_add"' in kw['system']

# ...and a tag call in the reply text comes back as a real tool call, with the prose left behind
def _comp(text):
    return Completion(model='m', usage=Usage(prompt_tokens=1, completion_tokens=2),
                      message=Msg(role='assistant', content=[Part(type=PartType.text, text=text)]))
r = norm_completion(_comp('on it\n<tool_call>{"name": "_add", "arguments": {"a": 1, "b": 2}}</tool_call>'))
test_eq(resp_text(r), 'on it')
test_eq((r['tool_calls'][0].name, r['tool_calls'][0].arguments), ('_add', {'a': 1, 'b': 2}))
test_eq(norm_completion(_comp('just prose')).get('tool_calls'), None)

Against a real API

Chat('gpt-5.1'), Chat('anthropic/claude-sonnet-4-5'), or runtime='remote' — routing only; you need a vendor key.

from fastllm.acomplete import acomplete               # undo the test patch above

chat = Chat('gpt-4.1', sp='You are concise.')
test_eq(chat.runtime, 'remote')
r = chat('Reply with exactly: pong')
assert 'pong' in resp_text(r).lower()
print(chat.use)

total=22|in=20|out=2|turns=1|model=gpt-4.1

display_stream(Chat('gpt-5.1')('Write two sentences about the monsoon.', stream=True))

The monsoon is a seasonal wind system that brings heavy rainfall to regions such as South Asia, Southeast Asia, and parts of Africa. It is crucial for agriculture and water resources but can also cause severe flooding and landslides.

'The monsoon is a seasonal wind system that brings heavy rainfall to regions such as South Asia, Southeast Asia, and parts of Africa. It is crucial for agriculture and water resources but can also cause severe flooding and landslides.'

th = Chat('gpt-5.1', reasoning_effort='high', max_output_tokens=2048)
r = th('A bat and ball cost $1.10, the bat is $1 more than the ball. How much is the ball?')
print(thought(r)[:400]); print('---'); print(resp_text(r))

---
Let the price of the ball be \(x\) dollars.  
Then the bat costs \(x + 1\) dollars.

Total cost:
\[
x + (x + 1) = 1.10
\]
\[
2x + 1 = 1.10
\]
\[
2x = 0.10
\]
\[
x = 0.05
\]

The ball costs **5 cents**.

img = Path('images.jpeg').read_bytes()
print(resp_text(Chat('gpt-5.1')([img, 'What is in this image? One sentence.'])))

A German Shepherd dog is standing on an outdoor path with its tongue out.

# the whole point: start local, finish hosted, with one history
from rishi.llama import LlamaChat, qwen3_17b

local = LlamaChat(qwen3_17b, n_ctx=4096)
local('My name is Karthik and my favourite number is 17. Remember both.')
hist = local.hist
local.close()

remote = Chat('gpt-5.1', messages=hist)
r = remote('What is my name and my favourite number?')
assert 'karthik' in resp_text(r).lower() and '17' in resp_text(r)
resp_text(r)

/Users/71293/code/personal/orgs/ramabana/.venv/lib/python3.13/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html
  from .autonotebook import tqdm as notebook_tqdm
llama_context: n_ctx_seq (4096) < n_ctx_train (40960) -- the full capacity of the model will not be utilized

'Your name is Karthik, and your favourite number is 17.'