rigging.message

This module covers core message objects and handling.

Role = t.Literal['system', 'user', 'assistant'] module-attribute

The role of a message. Can be 'system', 'user', or 'assistant'.

Message(role: Role, content: str | list[str | Content], parts: t.Sequence[ParsedMessagePart] | None = None, **kwargs: t.Any)

Bases: BaseModel

Represents a message with role, content, and parsed message parts.

Note

Historically, content was a string, but multi-modal LLMs require us to have a more structured content representation.

For interface stability, content will remain a property accessor for the text of a message, but the "real" content is available in all_content. During serialization, we rename all_content to content for compatibility.

Source code in rigging/message.py

def __init__(
    self,
    role: Role,
    content: str | list[str | Content],
    parts: t.Sequence[ParsedMessagePart] | None = None,
    **kwargs: t.Any,
):
    if isinstance(content, str):
        content = dedent(content)
    else:
        content = [ContentText(text=dedent(part)) if isinstance(part, str) else part for part in content]

    super().__init__(role=role, all_content=content, parts=parts or [], **kwargs)

all_content: str | list[Content] = Field('', repr=False) class-attribute instance-attribute

Interior str content or structured content parts.

content: str property writable

The content of the message.

If the interior of the message content is stored as a list of Content objects, this property will return the concatenated text of any ContentText parts.

models: list[Model] property

Returns a list of models parsed from the message.

parts: list[ParsedMessagePart] = Field(default_factory=list) class-attribute instance-attribute

The parsed message parts.

role: Role instance-attribute

The role of the message.

uuid: UUID = Field(default_factory=uuid4, repr=False) class-attribute instance-attribute

The unique identifier for the message.

apply(**kwargs: str) -> Message

Applies the given keyword arguments with string templating to the content of the message.

Uses string.Template.safe_substitute underneath.

Note

This call produces a clone of the message, leaving the original message unchanged.

Parameters:

• **kwargs (str, default: {} ) –

  Keyword arguments to substitute in the message content.

Source code in rigging/message.py

def apply(self, **kwargs: str) -> Message:
    """
    Applies the given keyword arguments with string templating to the content of the message.

    Uses [string.Template.safe_substitute](https://docs.python.org/3/library/string.html#string.Template.safe_substitute) underneath.

    Note:
        This call produces a clone of the message, leaving the original message unchanged.

    Args:
        **kwargs: Keyword arguments to substitute in the message content.
    """
    new = self.clone()
    template = string.Template(new.content)
    new.content = template.safe_substitute(**kwargs)
    return new

apply_to_list(messages: t.Sequence[Message], **kwargs: str) -> list[Message] classmethod

Helper function to apply keyword arguments to a list of Message objects.

Source code in rigging/message.py

@classmethod
def apply_to_list(cls, messages: t.Sequence[Message], **kwargs: str) -> list[Message]:
    """Helper function to apply keyword arguments to a list of Message objects."""
    return [message.apply(**kwargs) for message in messages]

clone() -> Message

Creates a copy of the message.

Source code in rigging/message.py

def clone(self) -> Message:
    """Creates a copy of the message."""
    return Message(self.role, self.content, parts=copy.deepcopy(self.parts))

fit(message: t.Union[Message, MessageDict, str]) -> Message classmethod

Helper function to convert various common types to a Message object.

Source code in rigging/message.py

@classmethod
def fit(cls, message: t.Union[Message, MessageDict, str]) -> Message:
    """Helper function to convert various common types to a Message object."""
    if isinstance(message, str):
        return cls(role="user", content=message)
    return cls(**message) if isinstance(message, dict) else message.model_copy(deep=True)

fit_as_list(messages: t.Sequence[MessageDict] | t.Sequence[Message] | MessageDict | Message | str) -> list[Message] classmethod

Helper function to convert various common types to a strict list of Message objects.

Source code in rigging/message.py

@classmethod
def fit_as_list(
    cls, messages: t.Sequence[MessageDict] | t.Sequence[Message] | MessageDict | Message | str
) -> list[Message]:
    """Helper function to convert various common types to a strict list of Message objects."""
    if isinstance(messages, (Message, dict, str)):
        return [cls.fit(messages)]
    return [cls.fit(message) for message in messages]

force_str_content() -> Message

Forces the content of the message to be a string by stripping any structured content parts like images.

Returns:

• Message –

  The modified message.

Source code in rigging/message.py

def force_str_content(self) -> Message:
    """
    Forces the content of the message to be a string by stripping
    any structured content parts like images.

    Returns:
        The modified message.
    """
    self.all_content = self.content
    return self

from_model(models: Model | t.Sequence[Model], role: Role = 'user', suffix: str | None = None) -> Message classmethod

Create a Message object from one or more Model objects.

Parameters:

• models (Model | Sequence[Model]) –

  The Model object(s) to convert to a Message.

• role (Role, default: 'user' ) –

  The role of the Message.

• suffix (str | None, default: None ) –

  A suffix to append to the content.

Returns:

• Message –

  The created Message object.

Source code in rigging/message.py

@classmethod
def from_model(
    cls: type[Message], models: Model | t.Sequence[Model], role: Role = "user", suffix: str | None = None
) -> Message:
    """
    Create a Message object from one or more Model objects.

    Args:
        models: The Model object(s) to convert to a Message.
        role: The role of the Message.
        suffix: A suffix to append to the content.

    Returns:
        The created Message object.
    """
    parts: list[ParsedMessagePart] = []
    content: str = ""
    for model in models if isinstance(models, list) else [models]:
        text_form = model.to_pretty_xml()
        slice_ = slice(len(content), len(content) + len(text_form))
        content += f"{text_form}\n"
        parts.append(ParsedMessagePart(model=model, slice_=slice_))

    if suffix is not None:
        content += f"\n{suffix}"

    return cls(role=role, content=content, parts=parts)

parse(model_type: type[ModelT]) -> ModelT

Parses a model from the message content.

Parameters:

• model_type (type[ModelT]) –

  The type of model to parse.

Returns:

• ModelT –

  The parsed model.

Raises:

• ValueError –

  If no models of the given type are found and fail_on_missing is set to True.

Source code in rigging/message.py

def parse(self, model_type: type[ModelT]) -> ModelT:
    """
    Parses a model from the message content.

    Args:
        model_type: The type of model to parse.

    Returns:
        The parsed model.

    Raises:
        ValueError: If no models of the given type are found and `fail_on_missing` is set to `True`.
    """
    return self.try_parse_many(model_type, fail_on_missing=True)[0]

parse_many(*types: type[ModelT]) -> list[ModelT]

Parses multiple models of the specified non-identical types from the message content.

Parameters:

• *types (type[ModelT], default: () ) –

  The types of models to parse.

Returns:

• list[ModelT] –

  A list of parsed models.

Raises:

• MissingModelError –

  If any of the models are missing.

Source code in rigging/message.py

def parse_many(self, *types: type[ModelT]) -> list[ModelT]:
    """
    Parses multiple models of the specified non-identical types from the message content.

    Args:
        *types: The types of models to parse.

    Returns:
        A list of parsed models.

    Raises:
        MissingModelError: If any of the models are missing.
    """
    return self.try_parse_many(*types, fail_on_missing=True)

parse_set(model_type: type[ModelT], minimum: int | None = None) -> list[ModelT]

Parses a set of models of the specified identical type from the message content.

Parameters:

• model_type (type[ModelT]) –

  The type of models to parse.

• minimum (int | None, default: None ) –

  The minimum number of models required.

Returns:

• list[ModelT] –

  A list of parsed models.

Raises:

• MissingModelError –

  If the minimum number of models is not met.

Source code in rigging/message.py

def parse_set(self, model_type: type[ModelT], minimum: int | None = None) -> list[ModelT]:
    """
    Parses a set of models of the specified identical type from the message content.

    Args:
        model_type: The type of models to parse.
        minimum: The minimum number of models required.

    Returns:
        A list of parsed models.

    Raises:
        MissingModelError: If the minimum number of models is not met.
    """
    return self.try_parse_set(model_type, minimum=minimum, fail_on_missing=True)

strip(model_type: type[Model], *, fail_on_missing: bool = False) -> list[ParsedMessagePart]

Removes and returns a list of ParsedMessagePart objects from the message that match the specified model type.

Parameters:

• model_type (type[Model]) –

  The type of model to match.

• fail_on_missing (bool, default: False ) –

  If True, raises a TypeError if no matching model is found.

Returns:

• list[ParsedMessagePart] –

  A list of removed ParsedMessagePart objects.

Raises:

• TypeError –

  If no matching model is found and fail_on_missing is True.

Source code in rigging/message.py

def strip(self, model_type: type[Model], *, fail_on_missing: bool = False) -> list[ParsedMessagePart]:
    """
    Removes and returns a list of ParsedMessagePart objects from the message that match the specified model type.

    Args:
        model_type: The type of model to match.
        fail_on_missing: If True, raises a TypeError if no matching model is found.

    Returns:
        A list of removed ParsedMessagePart objects.

    Raises:
        TypeError: If no matching model is found and fail_on_missing is True.
    """
    removed: list[ParsedMessagePart] = []
    for part in self.parts[:]:
        if isinstance(part.model, model_type):
            self._remove_part(part)
            removed.append(part)

    if not removed and fail_on_missing:
        raise TypeError(f"Could not find <{model_type.__xml_tag__}> ({model_type.__name__}) in message")

    return removed

to_openai_spec() -> dict[str, t.Any]

Converts the message to the OpenAI-compatible JSON format. This should be the primary way to serialize a message for use with APIs.

Returns:

• dict[str, Any] –

  The serialized message.

Source code in rigging/message.py

def to_openai_spec(self) -> dict[str, t.Any]:
    """
    Converts the message to the OpenAI-compatible JSON format. This should
    be the primary way to serialize a message for use with APIs.

    Returns:
        The serialized message.
    """
    # `all_content` will be moved to `content`
    return self.model_dump(include={"role", "all_content"})

try_parse(model_type: type[ModelT]) -> ModelT | None

Tries to parse a model from the message content.

Parameters:

• model_type (type[ModelT]) –

  The type of model to search for.

Returns:

• ModelT | None –

  The first model that matches the given model type, or None if no match is found.

Source code in rigging/message.py

def try_parse(self, model_type: type[ModelT]) -> ModelT | None:
    """
    Tries to parse a model from the message content.

    Args:
        model_type: The type of model to search for.

    Returns:
        The first model that matches the given model type, or None if no match is found.
    """
    return next(iter(self.try_parse_many(model_type)), None)

try_parse_many(*types: type[ModelT], fail_on_missing: bool = False) -> list[ModelT]

Tries to parse multiple models from the content of the message.

Parameters:

• *types (type[ModelT], default: () ) –

  The types of models to parse.

• fail_on_missing (bool, default: False ) –

  Whether to raise an exception if a model type is missing.

Returns:

• list[ModelT] –

  A list of parsed models.

Raises:

• MissingModelError –

  If a model type is missing and fail_on_missing is True.

Source code in rigging/message.py

def try_parse_many(self, *types: type[ModelT], fail_on_missing: bool = False) -> list[ModelT]:
    """
    Tries to parse multiple models from the content of the message.

    Args:
        *types: The types of models to parse.
        fail_on_missing: Whether to raise an exception if a model type is missing.

    Returns:
        A list of parsed models.

    Raises:
        MissingModelError: If a model type is missing and `fail_on_missing` is True.
    """
    model: ModelT
    parsed: list[tuple[ModelT, slice]] = try_parse_many(self.content, *types, fail_on_missing=fail_on_missing)
    for model, slice_ in parsed:
        self._add_part(ParsedMessagePart(model=model, slice_=slice_))
    self._sync_parts()
    return [p[0] for p in parsed]

try_parse_set(model_type: type[ModelT], minimum: int | None = None, fail_on_missing: bool = False) -> list[ModelT]

Tries to parse a set of models from the message content.

Parameters:

• model_type (type[ModelT]) –

  The type of model to parse.

• minimum (int | None, default: None ) –

  The minimum number of models expected.

• fail_on_missing (bool, default: False ) –

  Whether to raise an exception if models are missing.

Returns:

• list[ModelT] –

  The parsed models.

Raises:

• MissingModelError –

  If the number of parsed models is less than the minimum required.

Source code in rigging/message.py

def try_parse_set(
    self, model_type: type[ModelT], minimum: int | None = None, fail_on_missing: bool = False
) -> list[ModelT]:
    """
    Tries to parse a set of models from the message content.

    Args:
        model_type: The type of model to parse.
        minimum: The minimum number of models expected.
        fail_on_missing: Whether to raise an exception if models are missing.

    Returns:
        The parsed models.

    Raises:
        MissingModelError: If the number of parsed models is less than the minimum required.
    """
    models = self.try_parse_many(model_type, fail_on_missing=fail_on_missing)
    if minimum is not None and len(models) < minimum:
        raise MissingModelError(f"Expected at least {minimum} {model_type.__name__} in message")
    return models

MessageDict

Bases: TypedDict

Helper to represent a rigging.message.Message as a dictionary.

content: str | list[t.Any] instance-attribute

The content of the message.

role: Role instance-attribute

The role of the message.

ParsedMessagePart

Bases: BaseModel

Represents a parsed message part.

model: SerializeAsAny[Model] instance-attribute

The rigging/pydantic model associated with the message part.

slice_: slice instance-attribute

The slice representing the range into the message content.