dialog

Messages, dialogs, and attachments: the data model for LLM conversations stored as notebooks

A dialog is a conversation with an LLM kept in a Jupyter notebook: markdown notes, runnable code cells with their outputs, and prompt/response pairs, all in one document that can be viewed, edited, diffed, and version-controlled like any other notebook. The format comes from Solveit, where people have been editing dialogs as their daily working medium for a couple of years; this library extracts the format and its data model so that other tools can read, build, and surgically edit dialogs too, and convert them to and from LLM chat history.

This module is the in-memory model: Message, Dialog, and Attachment. Reading and writing the .ipynb form, and converting dialogs to chat history, live in later modules.

from fastcore.test import *

Message types

Every message is one of four types. A note is markdown, a code message is executable and carries Jupyter outputs, a prompt holds a user request together with the AI response it produced, and raw is untyped text. These map onto notebook cell types when a dialog is written to disk.

Identity

Messages, dialogs, and attachments all carry a string id, and it’s convenient for code to treat an object and its id interchangeably: msg in dlg.messages, atts.remove('img1.png'), or comparing a message against a bare id string. add_id_hash adds __eq__ and __hash__ to a class based on one identifier attribute, so equality and set/dict membership follow the id.

source

add_id_hash

def add_id_hash(
    cls, attr
):

Adds __eq__ and __hash__ to a cls based on a str or int attr id

A tiny class shows the behavior. Before patching, neither in nor == sees two instances with the same id as equal:

class Type:
    def __init__(self, id): store_attr()

atts = [Type(str(i)) for i in range(5)]
test_eq(Type('0') in atts, False)
test_eq(Type('3') == atts[3], False)

After patching, instances compare by id, ids themselves compare against instances, and hashing follows suit so sets and dict keys behave:

add_id_hash(Type, 'id')
test_eq(Type('0') in atts, True)
test_eq(Type(10) in atts, False)
test_eq(Type('3') == atts[3], True)
test_eq('1' in atts, True)
test_eq(Type('4') == '4', True)
atts.remove('0')
test_eq(len(atts), 4)
atts.remove(Type('3'))
test_eq(len(atts), 3)
test_eq([o.id for o in atts], ['1','2','4'])
test_eq(len({Type(1), Type(2), Type(1)}), 2)
test_eq({Type(1):"data"}[Type(1)], "data")

Dialog

source

Msgs

def Msgs(
    items:NoneType=None, *rest, use_list:bool=False, match:NoneType=None
):

A list of Messages with a preview-per-line repr

A Dialog consists of a name, an ordered list of messages, and meta (the notebook-level metadata dict).

source

Dialog

def Dialog(
    name, # Dialog name, usually the file stem
    messages:NoneType=None, # Initial `Message`s
    meta:NoneType=None, # Notebook-level metadata dict, carried verbatim through save/load
):

A named, ordered list of messages

dlg = Dialog('test')
dlg

test

Message

Prompt and code outputs both use the same ipynb-compatible structure: a list of output dicts with output_type, data, and metadata fields. One output format means one rendering and serialization path for both.

For prompts, the AI response text is stored in data['text/markdown'] with metadata['is_ai_res']=True to distinguish it from other display outputs. The ai_res property on Message extracts this text.

source

prompt_output

def prompt_output(
    txt:str=''
):

Create prompt output list from AI response text

source

code_output

def code_output(
    result:str='', subtype:str='plain', mimetype:str='text'
):

Call self as a function.

source

mk_code_output

def mk_code_output(
    d
):

Call self as a function.

source

displayobj

def displayobj(
    data:str='', subtype:str='plain', mimetype:str='text', meta:NoneType=None
):

Call self as a function.

source

mk_displayobj

def mk_displayobj(
    d, meta:NoneType=None
):

Call self as a function.

source

mk_output

def mk_output(
    typ, d, meta:NoneType=None, **kw
):

Call self as a function.

A message is one turn in a conversation between a human, an AI, and an interpreter. The type names the addressee and the answer it expects. A prompt pairs a request with the AI’s reply. A code message pairs source with the interpreter’s outputs. Notes address everyone and raws address no one. The reply and the outputs are the same slot pointed at different interlocutors, which is why both live in output. The index page tells the full story.

A Message carries exactly what the ipynb spec provides for a cell: content (source), output, msg_type (cell type, plus the prompt convention), id, attachments, and meta — the cell’s metadata dict, carried verbatim through save/load, so annotations this library knows nothing about survive a round trip untouched. Hosts declare the metadata they want promoted to real attributes in meta_attrs (attribute name → metadata key); anything undeclared simply rides along in meta. Ids are four random hex bytes, coerced to str; hosts add any display prefix (solveit’s DOM _) at their own boundary. content, output, and msg_type are DepProp descriptors that invalidate cached renders on assignment, through the clear_out_cache/clear_inp_cache hooks: core’s only cache is _ai_rend (the for-AI rendering), and a subclass that memoizes more overrides the hooks to clear its own slots at the same moments. The dlg backlink is a weakref to avoid a circular reference.

source

Message

def Message(
    content:str='', # The message text: markdown, code, or a prompt's request
    dlg:NoneType=None, # The `Dialog` this message belongs to
    output:str='', # Jupyter-style output list (code and prompt messages), or ''
    id:NoneType=None, # Message id; 4 random hex bytes if None
    msg_type:str='code', # One of `smsg_types`: code, note, prompt, or raw
    attachments:NoneType=None, # `Attachment`s carried by the message
    meta:NoneType=None, # Remaining cell metadata, carried verbatim through save/load
    **xtras
):

An ipynb cell, or Solveit message (which adds the prompt type)

Dialog.msg_cls is assigned here so that every factory below constructs messages through it: a subclassed Dialog sets msg_cls to its own Message subclass and gets the right instances from all inherited code, and read_ipynb takes a cls so whole files load as a host’s types.

The repr and markdown display show the content, the output when there is one, and the flags:

m = Message('in', output='out', dlg=dlg, msg_type=snote)
m

in ⇒ out

  • id: _ef8642b4
  • meta: {}
  • msg_type: note

Prompt outputs normalize on the way in: a string assigned to a prompt message’s output, at construction or any time later, wraps through prompt_output, so ai_res and history rendering always see the one list format.

pm = Message('q', msg_type=sprompt, output='A reply')
test_eq(pm.ai_res, 'A reply')
pm.output = 'Edited reply'
test_eq(pm.output, prompt_output('Edited reply'))
test_eq(pm.ai_res, 'Edited reply')

source

Message.preview

def preview(
    maxlen:int=120, # Maximum characters in the line
):

Escaped one-line summary of this message, the row format for Msgs displays

preview is what Msgs renders per row: id, type, escaped content start, and a size hint for any reply or output. summ (used by the markdown reprs) is bounded the same way, so a message with a huge output can never flood a display:

pv = pm.preview()
assert pv.startswith(f'{pm.id}:prompt: q') and 'reply(12)' in pv
big = Message('x=1', msg_type=scode, output=code_output('y'*500))
assert 'out(' in big.preview() and len(big.summ) < 200
test_eq(len(Message('z'*300, msg_type=snote).preview(80)), 80)
dd = Dialog('t', [pm, big])
assert isinstance(dd.messages, Msgs)
test_eq(str(dd.messages.show(rows=1)).count('\n'), 1)  # one row plus the "more" marker

update assigns several attributes at once, skipping any left as UNSET, which is what route handlers and edit operations pass for fields the user didn’t touch:

source

Message.update

def update(
    **kwargs
):

Update message attributes with provided keyword arguments

Building dialogs

mk_message creates a message through msg_cls and inserts it at the start, end, or relative to another message (which may be given as a Message or an id):

source

Dialog.mk_message

def mk_message(
    content:str, # Message text
    idx:int=-1, # Insert position; -1 appends
    after:NoneType=None, # Insert after this `Message` or id
    before:NoneType=None, # Insert before this `Message` or id
    msg_type:str='code', # One of `smsg_types`
    output:str='', # Output list; code messages also accept a JSON string
    **kwargs
):

Make new message and insert it into notebook before/after specified cell, or start of list (idx=0) by default

dlg = Dialog('test')
msg = dlg.mk_message(content='1')
dlg.mk_message(content='0', before=msg) # 0 before 1
dlg.mk_message(content='2', after=msg) # 2 after 1
assert L(dlg.messages).attrgot('content') == ['0','1','2']

Outputs attach at creation time; code_output builds a plausible one by hand:

out = code_output('64')
test_msg = dlg.mk_message('8*8', output=out)
test_eq(test_msg.output, out)
test_msg

8*8 ⇒ [{‘output_type’: ‘execute_result’, ‘metadata’: {}, ‘data’: {‘text/plain’: ‘64’}, ‘execution_count’: 1}]

  • id: _126a40a4
  • meta: {}
  • msg_type: code

When pasting messages from a clipboard, we need to copy each message and place the copies one after the other at a specific point. mk_messages makes that convenient:

source

Dialog.mk_messages

def mk_messages(
    msgs, after:NoneType=None, before:NoneType=None
):

Make new messages and insert them sequentially into the notebook after/before a specific message.

Mimicking copy/paste: copy the first two messages, then paste them after the second one.

dlg = Dialog('copy_paste')
for i in '123': dlg.mk_message(i, msg_type=snote)
def _contents(dlg): return [m.content for m in dlg.messages]
clipboard = copy.copy(dlg.messages[:2])
dlg.mk_messages(clipboard, after=dlg.messages[1])
test_eq(_contents(dlg), ['1', '2', '1', '2', '3'])

source

Message.insert_after

def insert_after(
    msgs
):

Call self as a function.

dlg = Dialog('test')
msg1,msg2,msg3 = [dlg.mk_message(o) for o in ('first','second','third')]
msg1.insert_after([Message('inserted1'), Message('inserted2')])
test_eq([m.content for m in dlg.messages], ['first', 'inserted1', 'inserted2', 'second', 'third'])

Sections

A note starting with a markdown heading opens a section, which runs until the next heading of the same or higher level. header_info parses a content string’s first line into its heading (level, text) pair, following CommonMark (up to 3 leading spaces allowed), so hosts share one definition of “is a heading” for sections, collapsing, and outline views. header_level reads the level for a note message (0 for anything that isn’t a heading note), and section_msgs returns a heading plus its span within any pool of messages; the pool is an argument, not dlg.messages, so callers can pass a filtered list (for example with skipped messages hidden) and the span respects it.

source

section_msgs

def section_msgs(
    ms, head
):

head plus its section in ms: messages until the next heading of the same or higher level

source

Message.header_level

def header_level():

Heading level of a heading note message, else 0

source

header_info

def header_info(
    content
):

Markdown heading (level, text) of content’s first line; (0, None) if not a heading

sd = Dialog('sect')
h1,a,h2,b,h3 = [sd.mk_message(c, msg_type=t) for c,t in
    [('# Top',snote),('body',snote),('## Sub\ndetail below',snote),('x=1',scode),('# Next',snote)]]
test_eq([m.header_level() for m in sd.messages], [1,0,2,0,1])
test_eq(sd.mk_message('#not a heading', msg_type=snote).header_level(), 0)  # markdown needs the space
test_eq(header_info('## Sub  \nmore'), (2, 'Sub  '))  # (level, text) of the first line
test_eq(header_info('  ## Indented'), (2, 'Indented'))  # CommonMark allows up to 3 leading spaces
test_eq(header_info('    ## Code'), (0, None))  # 4+ spaces is an indented code block, not a heading
test_eq(sd.mk_message('   # Deep', msg_type=snote).header_level(), 1)
test_eq(section_msgs(sd.messages, h1), [h1,a,h2,b])  # stops before the next same-or-higher heading
test_eq(section_msgs(sd.messages, h2), [h2,b])  # body text under the heading doesn't change its level
test_eq(section_msgs(sd.messages, a), [a])  # a non-heading has no span

Inspecting outputs

has_error reports whether a code message’s outputs include an error, and get_output_mds pulls out any markdown an output displayed.

source

get_output_mds

def get_output_mds(
    output
):

Extract text/markdown content from code outputs (display_data and execute_result)

source

Message.has_error

def has_error():

Call self as a function.

err_out = [dict(output_type='error', ename='ZeroDivisionError', evalue='division by zero', traceback=[])]
test_eq(Message('1/0', msg_type=scode, output=err_out).has_error, True)
test_eq(Message('1+1', msg_type=scode, output=code_output('2')).has_error, False)
test_eq(Message('note', msg_type=snote).has_error, False)

Only display and result outputs carry markdown; streams (prints) do not:

out_stream = [dict(output_type='stream', name='stdout', text='plain\n')]
out_display = [dict(output_type='display_data', metadata={}, data={'text/markdown': '*hi*'})]
out_exec = [dict(output_type='execute_result', metadata={}, data={'text/markdown': '**lo**'}, execution_count=1)]
test_eq(get_output_mds(out_stream), [])
test_eq(get_output_mds(out_display + out_exec), ['*hi*', '**lo**'])
test_eq(get_output_mds(None), [])

Kernel outputs

A running kernel reports results as iopub messages, and output_from_msg reshapes one into the ipynb output dict a message stores (the same conversion nbformat.v4.output_from_msg does, without the nbformat dependency). The mk_jmsg family builds such messages, for tests and for hosts that synthesize outputs.

source

output_from_msg

def output_from_msg(
    jmsg
):

ipynb output dict for the Jupyter iopub message dict jmsg

source

mk_execresult

def mk_execresult(
    data, metadata:NoneType=None, execution_count:int=1
):

Call self as a function.

source

mk_dispdata

def mk_dispdata(
    data, metadata:NoneType=None, display_id:NoneType=None
):

Call self as a function.

source

mk_error

def mk_error(
    ename:str='Exception', evalue:str='', traceback:NoneType=None
):

Call self as a function.

source

mk_stream

def mk_stream(
    text, name:str='stdout'
):

Call self as a function.

source

mk_jmsg

def mk_jmsg(
    typ, metadata:NoneType=None, **kwargs
):

A minimal Jupyter iopub message dict of type typ, with kwargs as its content

test_eq(output_from_msg(mk_stream('hi\n')), dict(output_type='stream', name='stdout', text='hi\n'))
test_eq(output_from_msg(mk_execresult({'text/plain':'2'}))['execution_count'], 1)
test_eq(output_from_msg(mk_error('E', 'boom', ['tb']))['ename'], 'E')
with expect_fail(ValueError, 'Unrecognized'): output_from_msg(mk_jmsg('status'))

add_output converts and appends one message at a time as a run streams in, handling the protocol’s stateful parts: a pending clear_output(wait=True) clears when the next output arrives, a trailing semicolon on the source suppresses the execute_result, and a display-id update replaces the earlier output it names. Hosts hook in without overriding: trunc= transforms each output dict (e.g. truncation), and on_input= receives kernel input_request messages (e.g. to show an input form).

source

Message.add_output

def add_output(
    jmsg, # A Jupyter iopub message dict
    trunc:NoneType=None, # Optional transform applied to each new output dict, e.g. truncation
    on_input:NoneType=None, # Optional handler for kernel `input_request` messages
):

Convert iopub message jmsg to an output dict and add it to this message’s outputs

source

Message.clear_output

def clear_output(
    wait:bool=False
):

Clear outputs now, or on the next add_output if wait

km = Message('1+1', msg_type=scode, output=[])
km.add_output(mk_stream('hi\n'))
km.add_output(mk_execresult({'text/plain':'2'}))
test_eq([o['output_type'] for o in km.output], ['stream','execute_result'])
km.clear_output(wait=True)
test_eq(len(km.output), 2)  # nothing cleared yet
km.add_output(mk_stream('fresh\n'))
test_eq([o['text'] for o in km.output], ['fresh\n'])  # the pending clear ran first
test_eq(Message('2+2;', msg_type=scode, output=[]).add_output(mk_execresult({'text/plain':'4'})), None)  # semicolon suppresses

A display-id update replaces in place; the trunc and on_input hooks see each output and input request:

dm = Message('show', msg_type=scode, output=[])
dm.add_output(mk_dispdata({'text/plain':'v1'}, display_id='d1'))
dm.add_output(mk_jmsg('update_display_data', data={'text/plain':'v2'}, transient={'display_id':'d1'}))
test_eq([o['data']['text/plain'] for o in dm.output], ['v2'])  # updated in place
seen = []
dm.add_output(mk_stream('x'*10), trunc=lambda o: o | dict(text=o['text'][:4]))
test_eq(dm.output[-1]['text'], 'xxxx')
dm.add_output(mk_jmsg('input_request', prompt='Name?', password=False), on_input=seen.append)
test_eq(seen[0]['content']['prompt'], 'Name?')

Exported code

A message is exported when its content carries an nbdev #| export directive - the directive is the single source of truth, matching the file bytes (the settable exported property reads and edits the content, so hosts never keep a separate flag). dlg2py joins a dialog’s exported code into a python source string - the projection behind “export to .py” features.

source

dlg2py

def dlg2py(
    dlg
):

The exported code of dlg, as a python source string

source

strip_export

def strip_export(
    s
):

Content s without its leading #| export directive line, if any

xd = Dialog('exp')
e1 = xd.mk_message('#| export\ndef f(): pass', msg_type=scode)
e2 = xd.mk_message('def g(): pass', msg_type=scode)
xd.mk_message('def h(): pass', msg_type=scode)
xd.mk_message('#| export\nexported note', msg_type=snote)
assert e1.exported and not e2.exported
e2.exported = True
test_eq(e2.content, '#| export\ndef g(): pass')
e2.exported = True  # idempotent
test_eq(e2.content, '#| export\ndef g(): pass')
test_eq(dlg2py(xd), '#| export\ndef f(): pass\n\n#| export\ndef g(): pass')  # code only
e2.exported = not e2.exported  # toggling is just assignment
test_eq((e2.exported, e2.content), (False, 'def g(): pass'))
e1.exported = False
test_eq(e1.content, 'def f(): pass')  # any directive spelling removed

Merging combines several messages into one; merge_content is each message’s contribution, in terms the target type understands: merging into code keeps sources raw, while merging into a note renders a prompt as its request/response sections and fences code. Hosts override it for their own conventions (llmsurgery’s merge_msgs drives it):

source

Message.merge_content

def merge_content(
    mtype:str, # The merge target's msg_type
):

This message’s content contribution when merged into a mtype message; hosts override for their own conventions

pm2 = Message('Sum?', msg_type=sprompt, output='It is 3.')
test_eq(pm2.merge_content(snote), '## AI Prompt\nSum?\n## AI Response\nIt is 3.')
test_eq(pm2.merge_content(scode), 'Sum?')
test_eq(Message('x=1', msg_type=scode).merge_content(snote), '```python\nx=1\n```')

Validation

Serialization builds valid cells by construction, but nothing stops code from assigning a malformed output list, an unserializable meta value, or a duplicate id directly; such damage otherwise only surfaces as a confusing error at save time. validate raises a ValueError naming the message as soon as it’s called, and never repairs. These are the model-level rules; file-level structure is fastcore.nbio’s validate_nb’s business.

source

Message.validate

def validate():

Raise ValueError for problems that would break saving or rendering; returns self if fine

test_eq(Message('1+1', id=1, msg_type=scode, output=[]).validate().id, '1')  # ids coerce to str at construction
m = Message('1+1', msg_type=scode, output=code_output('2'))
test_eq(m.validate(), m)
m.output = [dict(no_type=1)]
with expect_fail(ValueError, 'bad output entry'): m.validate()
with expect_fail(ValueError, 'must not carry output'): Message('n', msg_type=snote, output='x').validate()
with expect_fail(ValueError, 'not JSON-serializable'): Message('m', msg_type=snote, meta=dict(x={1,2})).validate()

source

Dialog.validate

def validate():

Validate every message plus dialog-scope rules (unique ids, serializable meta); returns self if fine

vd = Dialog('v', [Message('a', msg_type=snote), Message('b', msg_type=snote)])
test_eq(vd.validate(), vd)
vd.messages.append(Message('c', msg_type=snote, id=vd.messages[0].id))
with expect_fail(ValueError, 'duplicate message id'): vd.validate()

Attachments

Attachments are named binary blobs (usually images) that ride along with a message and serialize into the standard Jupyter attachments dict. Ids follow the uuid4 shape; ruuid4 builds them from rtoken_hex so tests that seed random get deterministic ids.

source

Message.mk_attachment

def mk_attachment(
    data:bytes, content_type:str, id:str=''
):

Call self as a function.

source

Attachment

def Attachment(
    data:bytes, content_type:str, id:str=''
):

Base class for objects needing a basic __repr__

source

ruuid4

def ruuid4():

Generate a deterministic UUID4-like string using rtoken_hex

tiny_png is a minimal valid one-pixel PNG, exported for examples and tests that need real image bytes.

Copy/paste keeps attachments with their messages:

test_eq(Attachment(b'x', 'text/plain', id=42).id, '42')  # ids coerce to str at construction
dlg_attach = Dialog('attach_test')
att1 = Attachment(data=b'screenshot1', content_type='image/png', id='img1.png')
m1 = dlg_attach.mk_message('1. Message with screenshot', msg_type=snote, attachments=[att1])
m2 = dlg_attach.mk_message('2. No attachment', msg_type=snote)
pasted = dlg_attach.mk_messages(copy.copy(dlg_attach.messages[:1]), after=m2)
test_eq(len(dlg_attach.messages), 3)
test_eq(pasted[0].attachments[0].id, 'img1.png')
test_eq(pasted[0].attachments[0].data, b'screenshot1')

Serializing

source

Dialog.todict

def todict():

Call self as a function.