Threads and messages

Thread

Thread(t=None, inst='', join_sep='\n')

A sequence of messages alternating between IN ("user" role) and OUT ("assistant" role).

Stores a special initial INST information (known as "system" role in ChatML) providing instructions to the model. Some models don't use system instructions - in those cases it's prepended to first IN message.

Messages are kept in a strict IN,OUT,IN,OUT,... order. To enforce this, if two IN messages are added, the second just appends to the text of the first or to its image list.

Parameters:

Name	Type	Description	Default
t	Optional[Union[Self, list, Msg, dict, tuple, str]]	Optionally initialize from a Thread, list[Msg], list[ChatML format dict], list[tuple], list[str], Msg, ChatML format dict, tuple or str.	None
inst	str	Instructions text. If inst arg is not set and t is a Thread, its inst will be used.	''
join_sep	str	Separator used when message text needs to be joined. Defaults to "\n".	'\n'

Raises:

Type	Description
TypeError	On invalid args passed.

Source code in sibila/thread.py

def __init__(self,
             t: Optional[Union[Self,list,Msg,dict,tuple,str]] = None,
             inst: str = "",
             join_sep: str = "\n"):
    """
    Args:
        t: Optionally initialize from a Thread, list[Msg], list[ChatML format dict], list[tuple], list[str], Msg, ChatML format dict, tuple or str.
        inst: Instructions text. If inst arg is not set and t is a Thread, its inst will be used.
        join_sep: Separator used when message text needs to be joined. Defaults to "\\n".

    Raises:
        TypeError: On invalid args passed.
    """

    self._msgs = []
    self.inst = Msg.make_INST(inst)
    self.join_sep = join_sep

    if t is not None:
        self.concat(t)

inst instance-attribute

inst = make_INST(inst)

System instructions in an Msg of kind INST, defaults to empty text.

init_INST_IN

init_INST_IN(inst_text, in_text, in_images=None)

Initialize Thread with instructions and an IN message.

Parameters:

Name	Type	Description	Default
inst_text	str	Instructions text.	required
in_text	str	Text for IN message.	required
in_images	Optional[Union[list, str, dict]]	An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.	None

Source code in sibila/thread.py

def init_INST_IN(self,
                 inst_text: str,
                 in_text: str,
                 in_images: Optional[Union[list,str,dict]] = None):
    """Initialize Thread with instructions and an IN message.

    Args:
        inst_text: Instructions text.
        in_text: Text for IN message.
        in_images: An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.
    """
    self.clear()
    self.inst.text = inst_text
    self.add_IN(in_text, in_images)

add_IN

add_IN(in_text, in_images=None)

Appends an IN message to Thread.

Parameters:

Name	Type	Description	Default
in_text	str	Text for IN message.	required
in_images	Optional[Union[list, str, dict]]	An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.	None

Source code in sibila/thread.py

def add_IN(self,
           in_text: str,
           in_images: Optional[Union[list,str,dict]] = None):
    """Appends an IN message to Thread.

    Args:
        in_text: Text for IN message.
        in_images: An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.
    """
    self.add(Msg.Kind.IN, in_text, in_images)

add_OUT

add_OUT(out_text, out_images=None)

Appends an OUT message to Thread.

Parameters:

Name	Type	Description	Default
out_text	str	Text for OUT message.	required
out_images	Optional[Union[list, str, dict]]	An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.	None

Source code in sibila/thread.py

def add_OUT(self,
            out_text: str,
            out_images: Optional[Union[list,str,dict]] = None):
    """Appends an OUT message to Thread.

    Args:
        out_text: Text for OUT message.
        out_images: An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.
    """
    self.add(Msg.Kind.OUT, out_text, out_images)

add_OUT_IN

add_OUT_IN(
    out_text, in_text, *, out_images=None, in_images=None
)

Appends an OUT message followed by an IN message.

Parameters:

Name	Type	Description	Default
out_text	str	Text for OUT message.	required
in_text	str	Text for IN message.	required
out_images	Optional[Union[list, str, dict]]	An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.	None
in_images	Optional[Union[list, str, dict]]	Optional list of IN message images.	None

Source code in sibila/thread.py

def add_OUT_IN(self,
               out_text: str,
               in_text: str,
               *,
               out_images: Optional[Union[list,str,dict]] = None,
               in_images: Optional[Union[list,str,dict]] = None):
    """Appends an OUT message followed by an IN message.

    Args:
        out_text: Text for OUT message.
        in_text: Text for IN message.
        out_images: An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.
        in_images: Optional list of IN message images.
    """        
    self.add(Msg.Kind.OUT, out_text, out_images)
    self.add(Msg.Kind.IN, in_text, in_images)

make_INST_IN staticmethod

make_INST_IN(inst_text, in_text, in_images=None)

Return an initialized Thread with instructions and an IN message.

Parameters:

Name	Type	Description	Default
inst_text	str	Instructions text.	required
in_text	str	Text for IN message.	required
in_images	Optional[Union[list, str, dict]]	An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.	None

Source code in sibila/thread.py

@staticmethod
def make_INST_IN(inst_text: str,
                 in_text: str,
                 in_images: Optional[Union[list,str,dict]] = None) -> 'Thread':
    """Return an initialized Thread with instructions and an IN message.

    Args:
        inst_text: Instructions text.
        in_text: Text for IN message.
        in_images: An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.
    """

    thread = Thread(inst=inst_text)
    thread.add_IN(in_text, in_images)
    return thread

make_IN staticmethod

make_IN(in_text, in_images=None)

Return an initialized Thread with an IN message.

Parameters:

Name	Type	Description	Default
in_text	str	Text for IN message.	required
in_images	Optional[Union[list, str, dict]]	An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.	None

Source code in sibila/thread.py

@staticmethod
def make_IN(in_text: str,
            in_images: Optional[Union[list,str,dict]] = None) -> 'Thread':
    """Return an initialized Thread with an IN message.

    Args:
        in_text: Text for IN message.
        in_images: An array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with "url" key and others. If url arg is not a valid URL, it will be loaded and converted to a data: URL.
    """

    thread = Thread()
    thread.add_IN(in_text, in_images)
    return thread

clone

clone()

Return a copy of current Thread.

Returns:

Type	Description
Self	A copy of this Thread.

Source code in sibila/thread.py

def clone(self) -> Self:
    """Return a copy of current Thread.

    Returns:
        A copy of this Thread.
    """
    return Thread(self)

clear

clear(clear_inst=True)

Delete all messages and clear inst.

Source code in sibila/thread.py

def clear(self,
          clear_inst: bool = True):
    """Delete all messages and clear inst."""
    self._msgs = []
    if clear_inst:
        self.inst.text = ""

load

load(path, clear)

Load this Thread from a JSON file.

Parameters:

Name	Type	Description	Default
path	str	Path of file to load.	required
clear	bool	Should thread be cleared of messages, including INST? If not will concatenate with existing ones.	required

Source code in sibila/thread.py

def load(self,
         path: str,
         clear: bool):
    """Load this Thread from a JSON file.

    Args:
        path: Path of file to load.
        clear: Should thread be cleared of messages, including INST? If not will concatenate with existing ones.
    """

    with open(path, 'r', encoding='utf-8') as f:
        js = f.read()
    state = json.loads(js)

    if clear:
        self.clear()

    th = self.from_dict(state)
    self.concat(th)

save

save(path)

Serialize this Thread to a JSON file.

Parameters:

Name	Type	Description	Default
path	str	Path of file to save into.	required

Source code in sibila/thread.py

def save(self,
         path: str):
    """Serialize this Thread to a JSON file.

    Args:
        path: Path of file to save into.
    """

    state = self.as_dict()

    json_str = json.dumps(state, indent=2, default=vars)

    with open(path, 'w', encoding='utf-8') as f:
        f.write(json_str)

from_dict staticmethod

from_dict(state)

Deserialize a Thread from a dict.

Source code in sibila/thread.py

@staticmethod
def from_dict(state: dict) -> 'Thread':
    """Deserialize a Thread from a dict."""

    th = Thread()
    for dic in state["_msgs"]:
        th.add(Msg.from_dict(dic))
    th.inst = Msg.from_dict(state["inst"])
    th.join_sep = state["join_sep"]

    return th

as_dict

as_dict()

Serialize this Thread to a dict.

Source code in sibila/thread.py

def as_dict(self) -> dict:
    """Serialize this Thread to a dict."""

    state = {"_msgs": [],
             "inst": self.inst.as_dict(),
             "join_sep": self.join_sep}

    for msg in self._msgs:
        state["_msgs"].append(msg.as_dict()) # type: ignore[attr-defined]

    return state

as_chatml

as_chatml(include_INST=True)

Returns Thread as a list of ChatML messages.

Returns:

Type	Description
list[dict]	A list of ChatML dict elements with "role" and "content" keys.

Source code in sibila/thread.py

def as_chatml(self,
              include_INST: bool = True) -> list[dict]:
    """Returns Thread as a list of ChatML messages.

    Returns:
        A list of ChatML dict elements with "role" and "content" keys.
    """
    msgs = []

    if self.inst.text and include_INST:
        msgs.append(self.inst.as_chatml())

    for msg in self._msgs:
        msgs.append(msg.as_chatml())

    return msgs

Trim

Flags for Thread trimming.

NONE class-attribute instance-attribute

NONE = 0

No trimming.

INST class-attribute instance-attribute

INST = 1

Can remove INST message.

IN class-attribute instance-attribute

IN = 2

Can remove IN messages.

OUT class-attribute instance-attribute

OUT = 4

Can remove OUT messages.

KEEP_FIRST_IN class-attribute instance-attribute

KEEP_FIRST_IN = 1024

If trimming IN messages, never remove first one.

KEEP_FIRST_OUT class-attribute instance-attribute

KEEP_FIRST_OUT = 2048

If trimming OUT messages, never remove first one.

trim

trim(trim_flags, max_token_len, thread_token_len_fn)

Trim context by selectively removing older messages until thread fits max_token_len.

Parameters:

Name	Type	Description	Default
trim_flags	Trim	Flags to guide selection of which messages to remove.	required
max_token_len	int	Cut messages until size is lower than this number. Defaults to None.	required
thread_token_len_fn	Callable	A function that returns token count for a passed Thread.	required

Example of a thread_token_len_fn that counts 1 char = 1 token: def thread_token_len_fn(thread: Thread) -> int: total = len(thread.inst.text) for msg in thread: total += len(msg.text) if msg.images: total += len(str(msg.images)) return total

Returns:

Type	Description
int	Trimming result: 1=trimmed messages to max_token_len, 0: no trimming was needed, -1: Unable to trim to max_token_len.

Source code in sibila/thread.py

def trim(self,
         trim_flags: Trim,
         max_token_len: int,
         thread_token_len_fn: Callable
         ) -> int:
    """Trim context by selectively removing older messages until thread fits max_token_len.

    Args:
        trim_flags: Flags to guide selection of which messages to remove.
        max_token_len: Cut messages until size is lower than this number. Defaults to None.
        thread_token_len_fn: A function that returns token count for a passed Thread.

    Example of a thread_token_len_fn that counts 1 char = 1 token:
        def thread_token_len_fn(thread: Thread) -> int:
            total = len(thread.inst.text)
            for msg in thread:
                total += len(msg.text)
                if msg.images:
                    total += len(str(msg.images))
            return total

    Returns:
        Trimming result: 1=trimmed messages to max_token_len, 0: no trimming was needed, -1: Unable to trim to max_token_len.
    """

    if trim_flags == Thread.Trim.NONE: # no trimming
        return 0

    thread = self.clone()

    any_trim = False

    while True:

        curr_len = thread_token_len_fn(thread)

        if curr_len <= max_token_len:
            break

        logger.debug(f"len={curr_len} / max={max_token_len}")

        if thread.inst.text and trim_flags & Thread.Trim.INST:
            thread.inst.text = ""
            any_trim = True
            logger.debug(f"Cutting INST {thread.inst.text[:40]}")
            continue

        # cut first possible message, starting from oldest first ones
        trimmed = False
        in_index = out_index = 0

        for index,msg in enumerate(thread):

            if msg.kind == Msg.Kind.IN:
                if trim_flags & Thread.Trim.IN:
                    if not (trim_flags & Thread.Trim.KEEP_FIRST_IN and in_index == 0):
                        del thread[index]
                        trimmed = True
                        logger.debug(f"Cutting IN {msg.text[:40]}")
                        break
                in_index += 1

            elif msg.kind == Msg.Kind.OUT:
                if trim_flags & Thread.Trim.OUT:                        
                    if not (trim_flags & Thread.Trim.KEEP_FIRST_OUT and out_index == 0):
                        del thread[index]
                        trimmed = True
                        logger.debug(f"Cutting OUT {msg.text[:40]}")
                        break
                out_index += 1

        if not trimmed:
            # all thread messages were cycled but not a single could be cut, so size remains the same
            # arriving here we did all we could for trim_flags but could not remove any more
            return -1
        else:
            any_trim = True

    # while end


    if any_trim:
        self._msgs = thread._msgs
        self.inst = thread.inst

    return int(any_trim)

add

add(t, text=None, images=None)

Add a message to Thread.

Accepts any of these argument combinations

t=Msg, ChatML format dict, tuple or str --or-- t=kind, text[, images]

Parameters:

Name	Type	Description	Default
t	Union[Msg, dict, tuple, str, Kind]	One of Msg, ChatML format dict, tuple or str, or Msg.Kind.	required
text	Optional[str]	Message text, only if t=Msg.Kind.	None
images	Optional[Union[list, str, dict]]	only if t=Msg.Kind or t=str-> an array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with keys "url" and any other keys like "detail". If url arg is not a valid URL, it will be loaded and converted to a data URL.	None

Source code in sibila/thread.py

def add(self, 
        t: Union[Msg,dict,tuple,str,Msg.Kind],
        text: Optional[str] = None,
        images: Optional[Union[list,str,dict]] = None):

    """Add a message to Thread.

    Accepts any of these argument combinations:
        t=Msg, ChatML format dict, tuple or str
        --or--
        t=kind, text[, images]

    Args:
        t: One of Msg, ChatML format dict, tuple or str, or Msg.Kind.
        text: Message text, only if t=Msg.Kind.
        images: only if t=Msg.Kind or t=str-> an array (or its first element) of either an str (a file path, will be loaded and converted to a data: URL) or a dict with keys "url" and any other keys like "detail". If url arg is not a valid URL, it will be loaded and converted to a data URL.
    """

    if text is not None:
        if not isinstance(t, Msg.Kind):
            raise TypeError("When arg 'text' is given, first arg must be of type Msg.Kind")

        msg = Msg(t, text, images)

    else: # add from t arg
        if isinstance(t, dict): # ChatML formatted dict
            msg = Msg.from_chatml(t)


        elif isinstance(t, tuple):
            msg = Msg(self.next_kind,
                      *t)

        elif isinstance(t, str): # simple text
            msg = Msg(self.next_kind,
                      t,
                      images)

        elif isinstance(t, Msg):
            msg = t.clone()

        else:
            raise TypeError("Arg 't' must be one of: Msg, ChatML format dict, tuple or str")


    # now append to list
    if msg.kind == Msg.Kind.INST:
        self.inst.join_same_kind(msg, self.join_sep)

    else:
        if not len(self._msgs) or msg.kind == self.next_kind: # next different kind or empty
            self._msgs.append(msg)
        else: # new msg is of same kind as last existing message: join/append to it
            last = self._msgs[-1]
            last.join_same_kind(msg, self.join_sep)

concat

concat(t)

Concatenate to current Thread: another Thread, list[Msg], list[ChatML format dict], list[str], Msg, ChatML format dict or str.

if last message in self is the same kind of first in t, their text, images, etc will be joined.

Parameters:

Name	Type	Description	Default
t	Union[Self, list, Msg, dict, tuple, str]	A Thread, list[Msg], list[ChatML format dict], list[str], Msg, ChatML format dict or str.	required

Source code in sibila/thread.py

def concat(self,
           t: Union[Self,list,Msg,dict,tuple,str]):
    """Concatenate to current Thread: another Thread, list[Msg], list[ChatML format dict], list[str], Msg, ChatML format dict or str.

    if last message in self is the same kind of first in t, their text, images, etc will be joined.

    Args:
        t: A Thread, list[Msg], list[ChatML format dict], list[str], Msg, ChatML format dict or str.
    """
    if isinstance(t, Thread):
        for msg in t:
            self.add(msg)

        self.inst.join_same_kind(t.inst, self.join_sep)

    else:
        if not isinstance(t, list):
            t = [t]
        for msg in t:
            self.add(msg)

get_iter

get_iter(include_set_inst)

Return an iterator that can be used to cycle over messages. include_set_inst: If inst message is set, include it before all others.

Source code in sibila/thread.py

def get_iter(self,
             include_set_inst: bool):
    """Return an iterator that can be used to cycle over messages.
    include_set_inst: If inst message is set, include it before all others.
    """
    class MsgIter:
        def __init__(self, 
                     thread: Thread,
                     include_inst: bool):
            self.thread = thread
            self.curr = -1 - int(include_inst)

        def __iter__(self):
            return self

        def __next__(self):
            self.curr += 1
            if self.curr == -1:
                return self.thread.inst
            elif self.curr < len(self.thread):
                return self.thread[self.curr]
            else:
                raise StopIteration

    return MsgIter(self,
                   include_set_inst and bool(self.inst.text))

has_images property

has_images

next_kind property

next_kind

Get kind of next new message that can be added to thread .

Returns:

Type	Description
Kind	Kind of last message or Msg.Kind.IN if empty.

has_text_lower

has_text_lower(text_lower)

Can the lowercase text be found in one of the messages?

Parameters:

Name	Type	Description	Default
text_lower	str	The lowercase text to search for in messages.	required

Returns:

Type	Description
bool	True if such text was found.

Source code in sibila/thread.py

def has_text_lower(self,
                   text_lower: str) -> bool:
    """Can the lowercase text be found in one of the messages?

    Args:
        text_lower: The lowercase text to search for in messages.

    Returns:
        True if such text was found.
    """
    for msg in self._msgs:
        if text_lower in msg.text.lower():
            return True

    return False        

join_sep instance-attribute

join_sep = join_sep

Separator used when message text needs to be joined. Defaults to '\n'

__add__

__add__(other)

Source code in sibila/thread.py

def __add__(self,
            other: Union[Self,list, Msg, dict, str]) -> Self:
    out = self.clone()
    out.concat(other)
    return out

__iter__

__iter__()

Source code in sibila/thread.py

def __iter__(self):
    # Default iterator doesn't include inst message.
    return self.get_iter(False)

Msg dataclass

Kind

Enumeration for kinds of messages in a Thread.

IN class-attribute instance-attribute

IN = 'IN'

Input message, from user.

OUT class-attribute instance-attribute

OUT = 'OUT'

Model output message.

INST class-attribute instance-attribute

INST = 'INST'

Initial model instructions.

as_chatml_role

as_chatml_role()

Source code in sibila/thread.py

def as_chatml_role(self: Self) -> str:
    CHATML_FROM_KIND: dict = {Msg.Kind.IN: "user", Msg.Kind.OUT: "assistant", Msg.Kind.INST: "system"}
    return CHATML_FROM_KIND.get(self) # type: ignore[return-value]

from_chatml_role staticmethod

from_chatml_role(role)

Source code in sibila/thread.py

@staticmethod
def from_chatml_role(role: str) -> 'Msg.Kind':
    KIND_FROM_CHATML: dict = {"user": Msg.Kind.IN, "assistant": Msg.Kind.OUT, "system": Msg.Kind.INST}
    kind = KIND_FROM_CHATML.get(role)
    if kind is None:
        raise ValueError(f"Unknown ChatML role '{role}'")
    else:
        return kind

flip staticmethod

flip(kind)

Source code in sibila/thread.py

@staticmethod
def flip(kind: 'Msg.Kind') -> 'Msg.Kind':
    return Msg.Kind.OUT if kind is Msg.Kind.IN else Msg.Kind.IN

__repr__

__repr__()

Source code in sibila/thread.py

def __repr__(self):
    return repr(self.value)

kind instance-attribute

kind

Message kind.

text instance-attribute

text

Message text (mandatory).

images class-attribute instance-attribute

images = None

List of images in message. An entry must have a 'url' key, but any other keys can be added. Key 'url' key must be a remote url (https,http) or a 'data:' base64-encoded url.

make_IN staticmethod

make_IN(text, images=None)

Source code in sibila/thread.py

@staticmethod
def make_IN(text: str,
            images: Optional[Union[list,str,dict]] = None) -> 'Msg':
    return Msg(Msg.Kind.IN,
               text,
               images)

make_OUT staticmethod

make_OUT(text, images=None)

Source code in sibila/thread.py

@staticmethod
def make_OUT(text: str,
             images: Optional[Union[list,str,dict]] = None) -> 'Msg':
    return Msg(Msg.Kind.OUT, 
               text, 
               images)

make_INST staticmethod

make_INST(text, images=None)

Source code in sibila/thread.py

@staticmethod
def make_INST(text: str,
              images: Optional[Union[list,str,dict]] = None) -> 'Msg':
    return Msg(Msg.Kind.INST, 
               text,
               images)

clone

clone()

Source code in sibila/thread.py

def clone(self) -> 'Msg':
    return Msg(self.kind, self.text, self.images)

from_dict staticmethod

from_dict(dic)

Source code in sibila/thread.py

@staticmethod
def from_dict(dic: dict) -> 'Msg':
    return Msg(kind=Msg.Kind(dic["kind"]),
               text=dic["text"],
               images=dic["images"])

as_dict

as_dict()

Return Msg as a dict.

Source code in sibila/thread.py

def as_dict(self) -> dict:
    """Return Msg as a dict."""
    return {"kind": self.kind.value, # kind as string
            "text": self.text,
            "images": self.images}

from_chatml staticmethod

from_chatml(dic, join_sep='\n')

Source code in sibila/thread.py

@staticmethod
def from_chatml(dic: dict,
                join_sep:str = "\n") -> 'Msg':

    role = dic.get("role")
    if role is None:
        raise ValueError(f"Key 'role' not found in {dic}")

    kind = Msg.Kind.from_chatml_role(role)

    content = dic.get("content")
    if content is None:
        raise ValueError(f"Bad 'content' key in {dic}")

    text = ''
    images = []
    if isinstance(content, list):
        for cont in content:
            if not isinstance(cont, dict) or "type" not in cont:
                raise TypeError(f"ChatML list entries must be of type dict and include a 'type' key in {cont}")

            if cont["type"] == "text":
                text = join_text(text, cont["text"], join_sep)

            elif cont["type"] == "image_url":
                image = cont["image_url"]
                if "url" not in image:
                    raise TypeError(f"ChatML image_url entries must include a 'url' key in {cont}")
                images.append(image)

    elif isinstance(content, str):
        text = content

    else:
        raise TypeError(f"ChatML content must have str or dict type in {content}")

    return Msg(kind, 
               text,
               images if images else None)

as_chatml

as_chatml()

Returns message in a ChatML dict.

Returns:

Type	Description
dict	A ChatML dict with "role" and "content" keys.

Source code in sibila/thread.py

def as_chatml(self) -> dict:
    """Returns message in a ChatML dict.

    Returns:
        A ChatML dict with "role" and "content" keys.
    """

    role = self.kind.as_chatml_role()

    if self.images:
        chatml_msg = {
            "role": role, 
            "content": [
                {"type": "text", "text": self.text},
            ]}

        for image in self.images:
            if "url" not in image:
                raise ValueError(f"Image without 'url' key at {image}")

            image_url = {"url": image["url"]}
            if "detail" in image:
                image_url["detail"] = image["detail"]

            chatml_msg["content"].append( # type: ignore[attr-defined]
                {"type": "image_url", "image_url": image_url}
            )
        return chatml_msg
    else:
        return {"role": role, "content": self.text}