from fastllm.chat import mk_msgs, mk_msg, tool_dtls_tag, hist2fmt
from fastcore.test import *
import random, json
from IPython.display import Markdownhist
Imports
def tool():
"dummmy tool"
passrandom.seed(6)Overview
| Function | Purpose | Input → Output |
|---|---|---|
render_outputs_ai |
Render Jupyter outputs as plain text | list[dict] → str |
ai_fmt |
Format an AI reply for chat history | str → str |
Message.to_xml |
Convert message to XML | Message → str |
Message.to_media |
Gather media from message | Message → list[str\|bytes] |
Message.to_parts |
Convert message to history parts | Message → list[str\|bytes] |
dlg2hist |
Main entry point | list[Message], dict → hist |
Test helpers
mk is a convenience function to create conformant dummy data for testing. For msg_type=='code' cells, output text is a serialized structure following nbformat v4. For creating tests, we have a helper function jwrap that wraps output text in this jupyter v4 format.
dname = 'backend'
dlg = Dialog(dname)jwrap
def jwrap(
txts:list=['']
):Wrap plain txt in a minimal Jupyter-style cell output
def mk(msg_type: str, content=None, output='', id=None, attachments=None):
"Return a `Message`, ensuring `output` is always valid JSON. Uses timestamp-based id if not provided."
if msg_type==scode: output = jwrap(output)
if msg_type==sprompt: output = prompt_output(output)
res = Message(msg_type=msg_type, output=output, attachments=attachments)
res.content = content or f'{msg_type}>{res.id}'
res.dlg = dlg # to avoid weakref issue
return resjwrap("test output")[{'output_type': 'stream', 'name': 'stdout', 'text': 'test output'}]
Code cells automatically jwrap the output value:
mk('code', 'print("hello")', 'hello\n')print(“hello”) ⇒ [{‘output_type’: ‘stream’, ‘name’: ‘stdout’, ‘text’: ‘hello’}]
- id: _29f88512
- meta: {}
- msg_type: code
mk('prompt', 'What is 2+2?', 'The answer is 4')What is 2+2? ⇒ [{‘output_type’: ‘display_data’, ‘metadata’: {‘is_ai_res’: True}, ‘data’: {‘text/markdown’: ‘The answer is 4’}}]
- id: _004af0bf
- meta: {}
- msg_type: prompt
Rendering outputs for the AI
The model sees a plain-text rendering of Jupyter outputs: streams concatenated as they would appear on a terminal screen, ANSI codes stripped, markdown preferred over HTML, images excluded (they travel separately, as media parts). The Jupyter-output primitives — preferred_out, preferred_msg_out, concat_streams, and the screen-semantics cleaning they build on — live in fastcore.nbio and fastcore.xtras; this module only adds the choices specific to model-facing text.
join_out
def join_out(
d
):Join Jupyter’s list-of-lines output data into one string
render_outputs_ai
def render_outputs_ai(
outputs, renderers:NoneType=None
):Plain-text rendering of a Jupyter output list for LLM context
render_output_ai
def render_output_ai(
out, renderers:NoneType=None
):Plain-text rendering of one Jupyter output, as the AI sees it; renderers overrides/extends the per-mime table
Each output type has a sensible plain-text form; note the stream applying the \r overwrite, and the error rendering its traceback:
test_eq(render_outputs_ai([{'output_type': 'stream', 'name': 'stdout', 'text': 'Hello\rBye\n'}]), 'Bye\n') # chkstyle: ignore
test_eq(render_output_ai({'output_type': 'execute_result', 'data': {'text/plain': '"42"'}}), '"42"')
test_eq(render_output_ai({'output_type': 'display_data', 'data': {'text/markdown': '**bold**'}}), '**bold**')
test_eq(render_output_ai({'output_type': 'display_data', 'data': {'text/latex': '$x^2$'}}), '$x^2$') # verbatim by default
test_eq(render_output_ai({'output_type': 'display_data', 'data': {'text/latex': '$x^2$'}}, renderers={'text/latex': str.upper}), '$X^2$')
test_eq(render_output_ai({'output_type': 'error', 'traceback': ['Error: something broke']}), 'Error: something broke')
test_eq(render_output_ai({'output_type': 'display_data', 'data': {'text/html': '<b>hi</b>', 'text/plain': 'hi'}}), '<b>hi</b>')
assert not render_output_ai({'output_type': 'execute_result', 'data': {'text/plain': ''}})Message AI views
try_eval
def try_eval(
s, typ:str | None=None
):Like literal_eval, but wraps in dynamic class named typ if succeeds, and returns s if fails
ai_output is the model-facing text of a message’s output, computed through render_out and cached in _ai_rend. render_out computes only the AI side; a UI (like Solveit) overrides it on its Message subclass to compute and cache its HTML rendering too, exposed through properties of its own.
Message.ai_output
def ai_output():Call self as a function.
Message.render_out
def render_out():Compute and cache the for-AI rendering of this message’s output
test_eq(Message('1+1', msg_type=scode, output=code_output('2')).ai_output, '2')
test_eq(Message('1+1', msg_type=scode).ai_output, '') # No output yet
m = Message('test', msg_type=sprompt, output=prompt_output('Ok!\n\n*[Response interrupted]*\n'))
test_eq(m.ai_output, 'Ok!')
test_eq(Message('empty', msg_type=sprompt, output=[]).ai_output, '<output result="pending" reason="incomplete"/>')Message.local_time
def local_time():Localized time_run for hosts that stamp one; ’’ when absent
to_local_time
def to_local_time(
time_str, tz:str='UTC'
):Convert ISO time string to local timezone
test_eq(to_local_time(''), '')
test_eq(to_local_time(None), '')
result = to_local_time('2026-01-27T12:00:00+00:00', 'UTC')
assert '2026-01-27' in result
m = Message('test', msg_type=snote)
test_eq(m.local_time(), '')
m.time_run = '2026-01-27T12:00:00+00:00'
assert m.local_time()Message.todict
def todict():Call self as a function.
Media
Media in the LLM context is represented by an XML metadata tag followed by raw bytes.
For successfully loaded media, the context contains:
- a
<media>tag withid,type, andmimeattributes - the raw media bytes immediately after the tag
If media cannot be loaded or decoded, the context contains a <media-unavailable> tag instead of silently dropping it. This lets the AI tell the user that the media was referenced but was not available.
id= is:
- UUID for attachments referenced via
attachment:uuidin markdown - variable name for media in variables
- content hash for code output images
type= is one of:
output— media from code cell outputvariable— media data in a variablecontent— media embedded in markdown message content
media_item
def media_item(
media_type, media_id, data, aim_info, mime:NoneType=None, max_im_sz:NoneType=None, prep:NoneType=None,
unavail_msg:str='An unsupported media type was included in the context: {mime}. Please instruct the user to switch to a model that supports this media type.'
):Load/prepare/package media, returning [xml_tag, data] or [media-unavailable].
detect_mime('https://image.jpg')A helper function and small dummy img for testing purposes.
def _mk_img(imgb):
"Image bytes from a base64 str"
return base64.b64decode(imgb)_imgb64 = b'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAAAAAA6fptVAAAACklEQVR4nGP4DwABAQEAsTj2FAAAAABJRU5ErkJggg=='Valid media should produce a <media> tag plus bytes
png = _mk_img(_imgb64)
res = media_item('content', 'img', png, dict(supports_vision=True))
test_eq(res[0], '<media id="img" type="content" mime="image/png"></media>')
test(res[1], bytes, isinstance)Expected media-loading failures should produce <media-unavailable>
for data in (None, 'not bytes', b'not media'): test(media_item('content', 'x', data, {})[0], '<media-unavailable', operator.contains)
def bad_b64_loader(): raise binascii.Error('bad base64')
test(media_item('output', 'bad', bad_b64_loader, {})[0], '<media-unavailable', operator.contains)def bad_loader(): raise RuntimeError('boom')
with expect_fail(Exception, 'boom'): media_item('content', 'x', bad_loader, {})test(media_item('content', 'img', png, dict(supports_vision=False))[0], "An unsupported media type was", operator.contains)
test(media_item('content', 'img', png, dict(supports_vision=False))[0], 'mime="image/png"', operator.contains)_media_atts
These are attachments found in a message such as the ones uploaded from the clipboard or the screenshot of the last output.
Message.prep_img
def prep_img(
data, mime, max_im_sz:NoneType=None
):Prepare image bytes before they enter LLM context; the base class passes them through unchanged
def parse_att_id(s):
"Extract id attribute from XML tag string"
return re.search(r' id="([^"]*)"', s).group(1)_mk_img(_imgb64)[:40]b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x00\x00\x00\x00:~\x9bU\x00\x00\x00\nIDA'
Only valid image attachments should be included:
att1,att2,att3 = Attachment(png, 'image/png'), Attachment(b'not an image', 'text/plain'), Attachment(_mk_img(_imgb64), 'image/png')
m = Message(f' and ', attachments=[att1, att2, att3])
res = _media_atts(m, dict(supports_vision=True))test_eq(len(m.attachments), 3)
test_eq(len(res), 4)
test_eq(L(res[::2]).map(parse_att_id), [att1.id, att3.id])Only the attachments referenced in the message content should be included:
m = Message(f'Only ', attachments=[att1,att3])
res = _media_atts(m, dict(supports_vision=True))
test_eq(len(m.attachments), 2)
test_eq(L(res[::2]).map(parse_att_id), [att3.id])bad = Attachment(b'not an image', 'image/png')
res = _media_atts(Message(f'', attachments=[bad]), dict(supports_vision=True))
test(res[0], '<media-unavailable', operator.contains)These are images from the message outputs, such as a code output displaying a plot
test_eq(_img_id(_mk_img(_imgb64)), _img_id(_mk_img(_imgb64)))
_img_id(_mk_img(_imgb64))'4P52II3F'
img_out = mk_code_output({'image/png': base64.b64encode(png).decode()})
m = Message(msg_type='code', output=img_out)res = _img_output(m, dict(supports_vision=True), None)
test_eq(len(res), 2)
test_eq(res[0], _media_xml('output', _img_id(png), 'image/png'))m = Message(msg_type='code', output=mk_code_output({'image/png': 'not-base64'}))
res = _img_output(m, dict(supports_vision=True), None)
test(res[0], '<media-unavailable', operator.contains)Message XML Helpers
These methods convert Message objects into XML for LLM context, using the shared item2xml grammar (per-type tags, content bare inside, an <out> section when output is present). Prompts render as their bare content via the prompt_txt host hook: in history, the reply is the next turn, not part of the message.
to_xml combines the msg input and output. A prompt renders through prompt_txt — bare content here, with last marking the live request; a host that wants an envelope around its prompts patches prompt_txt alone:
Message.to_xml
def to_xml(
last:bool=False
):Create XML representation of this message
Message.prompt_txt
def prompt_txt(
last:bool=False
):History text for a prompt message: the content, bare. Hosts patch this to add their own envelope
# Combines content and output into full XML representation
m = mk('code', 'print("hi")', 'hi\n')
test_eq(m.to_xml(), f'<code id="{m.id}">print("hi")<out>hi\n</out></code>')
n = mk('note', '# hey')
test_eq(n.to_xml(), f'<markdown id="{n.id}"># hey</markdown>') # notes render as markdown; no out, no extra tags'<message type="code" id="_624f8907" time=""><cell-source>print("hi")</cell-source><output>hi\n</output></message>'
# Prompts render bare
test_eq(mk('prompt', 'question').to_xml(), 'question')
test_eq(mk('prompt', 'question').prompt_txt(last=True), 'question')Message.to_media
def to_media(
aim_info, max_im_sz:NoneType=None
):Aggregate media from attachments, media_extra, and code outputs
Message.media_extra
def media_extra(
aim_info, max_im_sz:NoneType=None
):Additional media sources contributed by subclasses (e.g. app-specific file references)
# No media for plain messages
test_eq(mk('note', 'hello').to_media({}), [])# Code output images
m = Message(msg_type='code', output=mk_code_output({'image/png': base64.b64encode(png).decode()}))
media = m.to_media(dict(supports_vision=True))
test_eq(len(media), 2) # xml tag, bytesMessage.to_parts
def to_parts(
aim_info:dict, last:bool=False
):Convert message to a list of history parts: media, then XML text.
att = Attachment(_mk_img(_imgb64), 'image/png')
m = mk('prompt', f'What is in ?', attachments=[att])
res = m.to_parts(dict(supports_vision=True))
['...bytes...' if isinstance(o,bytes) else o for o in res]['<media id="f4ec230f-412a-4287-b7ff-0a4660e1f5e1" type="content" mime="image/png"></media>',
'...bytes...',
'What is in ?']
dlg2hist
def dlg2hist(
dlg, # A `Dialog`, or iterable of messages, ending with a prompt
aim_info:dict, # Model capability dict for media handling
):Convert dlg to LLM history. The final prompt renders with last=True.
Dialogs to canonical messages
dlg2hist produces alternating user parts and AI reply strings. For surgery we usually want the replies’ tool calls back as structured messages, and fastllm’s fmt2hist inverts the reply format exactly. dlg2chat composes the two into the canonical form every provider conversion starts from, and rejects duplicate tool-call ids, which hand-authored dialogs can accidentally produce. Thinking blocks survive only in the stored reply string: explode/implode (reply2dlg/dlg2reply) and this transmission projection both drop them, a deliberate loss since nothing downstream keeps them.
dlg2chat
def dlg2chat(
dlg, # A `Dialog`, ending with a prompt
aim_info:NoneType=None, # Model capability dict for media handling; images enabled if None
):Canonical fastllm messages for dlg, with each reply’s tool calls recovered as real parts
The conversion tests below need replies carrying tool-call details blocks, so first a fixture that builds one in fastllm’s reply format.
def _mk_dtl(func, args, result, tid='x1'):
d = json.dumps(dict(id=tid, server=False, call=dict(function=func, arguments=args), result=result))
return f"{tool_dtls_tag}\n<summary><code>{func}(...)</code></summary>\n\n```json\n{d}\n```\n\n</details>"
py_dtl = _mk_dtl('python', {'code':'1+1'}, '2')ddlg = Dialog('canon')
ddlg.mk_message('What is 1+1?', msg_type=sprompt, output=f"Check:\n\n{py_dtl}\n\nIt is 2.")
cm = dlg2chat(ddlg)
test_eq([m.role for m in cm], ['user', 'assistant', 'tool', 'assistant'])
test_eq(cm[1].content[1].data['name'], 'python')One reply often holds several sequential tool calls, each issued after reading the previous result. fmt2hist returns them batched into a single call message and a single result message, which reads as one parallel volley; dlg2chat explodes the batch back into strict call/result alternation, the shape live sessions record.
py_dtl2 = _mk_dtl('python', {'code':'2+2'}, '4', 'x2')
sdlg = Dialog('seq')
sdlg.mk_message('Two steps.', msg_type=sprompt, output=f"{py_dtl}\n\n{py_dtl2}\n\nDone.")
sm = dlg2chat(sdlg)
test_eq([(m.role, len(m.content)) for m in sm], [('user',1), ('assistant',1), ('tool',1), ('assistant',1), ('tool',1), ('assistant',1)])
test_eq(sm[2].content[0].data['id'], sm[1].content[0].data['id'])
test_eq(sm[4].content[0].data['id'], sm[3].content[0].data['id'])bad = Dialog('dup')
bad.mk_message('a', msg_type=sprompt, output=py_dtl)
bad.mk_message('b', msg_type=sprompt, output=py_dtl)
with expect_fail(ValueError, 'duplicate tool call id'): dlg2chat(bad)raw = [
mk('prompt', 'hello there', output='Hello! How can I help?'),
mk('note', 'a note between turns'),
mk('code', output="aaa", content="'some source text'"),
mk('prompt', 'add the numbers', output='They sum to 4.'),
mk('note'),
mk('note'),
mk('code', output="333"),
mk('prompt'), # empty output on purpose
mk('prompt', 'summarize the conversation so far', output='We greeted, then added some numbers.'),
mk('note', 'below we insert an image'),
mk('code', 'img=..some code to read image bytes'),
mk('prompt', 'Please tell me what you see in the image', output='I can see...'),
mk('note', 'lets add an image note'),
mk('prompt', 'Please tell me what you see in the 2 images in the above msg',output='i see two puppies!'),
mk('prompt', 'And one final question.')]aim_info = dict(supports_vision=True)A realistic message mix — prompts with and without preceding notes and code, an empty AI response, media references — checks the turn structure:
*hist,p,_ = dlg2hist(raw, aim_info)Lets take a look at the history:
for idx, turn in enumerate(mk_msgs(hist)):
display(Markdown(f"**======turn: {idx}======**"))
display(turn)======turn: 0======
Msg
- role:
user
Part (text)
hello $x
- data:
None
======turn: 1======
Msg
- role:
assistant
Part (text)
Hello! I can see the variable x
- data:
None
======turn: 2======
Msg
- role:
user
Part (text)
not_variable1
- data:
None
Part (text)
not_variable2’- data:
None
Part (text)
hi hi $y
- data:
None
======turn: 3======
Msg
- role:
assistant
Part (text)
Hello! I can see the variable y
- data:
None
======turn: 4======
Msg
- role:
user
Part (text)
- data:
None
Part (text)
- data:
None
Part (text)
- data:
None
Part (text)
prompt>_7140d653
- data:
None
======turn: 5======
Msg
- role:
assistant
Part (text)
======turn: 6======
Msg
- role:
user
Part (text)
$x, $y
- data:
None
======turn: 7======
Msg
- role:
assistant
Part (text)
Hello! I can see the variables x and y
- data:
None
======turn: 8======
Msg
- role:
user
Part (text)
- data:
None
Part (text)
- data:
None
Part (text)
Please tell me what you see in $img
- data:
None
======turn: 9======
Msg
- role:
assistant
Part (text)
I can see…
- data:
None
======turn: 10======
Msg
- role:
user
Part (text)
- data:
None
Part (text)
Please tell me what you see in the 2 images in the above msg
- data:
None
======turn: 11======
Msg
- role:
assistant
Part (text)
i see two puppies!
- data:
None
p['Whats variable $`y`?']
test_eq(len(hist),12) # `raw` has 6 prompts which means 12 user+ai turnshist[6]['$`x`, $`y`']
# the last prompt is marked `last`, but core rendering is identical: bare content
test_eq(p, [raw[-1].content])
test_eq(hist[6], [raw[8].content])A raw message carrying a rec_kind tag in its meta is session bookkeeping some codec kept for reversibility (e.g. llmsurgery.ant’s system records), not conversation content, so history projections skip it:
tr = [mk('note', 'context'), mk('prompt', 'q', output='r')]
h1 = dlg2hist(tr, aim_info)
trd = Dialog('trd'); trd.messages = list(tr)
test_eq(dlg2hist(trd, aim_info), h1) # a Dialog and its message list produce the same history
tr.insert(1, Message('', msg_type=sraw, meta=dict(rec_kind='system', rec=dict(type='system'))))
test_eq(dlg2hist(tr, aim_info), h1) # tagged raws never leak into history# test each user input for expected length
test_eq(len(hist[0]),1) # prompt w/o preceding msgs
test_eq(len(hist[2]),3) # prompt w preceding note and code
test_eq(len(hist[4]),4) # prompt w preceding notes (2) and code
test_eq(len(hist[6]),1) # prompt w/o preceding msgs
test_eq(len(hist[8]),3) # [note, code, prompt]
test_eq(len(hist[10]),2) # [note, prompt]# Test each AI output for expected length
for i in (1,3,7,9,11): test(len(hist[i]),0,operator.gt)
test_eq(hist[5],'<output result="pending" reason="incomplete"/>') # empty ai response