dreadnode.generators

Chats are used pre and post generation to hold messages.

They are the primary way to interact with the generator.

DEFAULT_MAX_DEPTH

DEFAULT_MAX_DEPTH = 20

Maximum depth of nested pipeline generations to attempt before giving up.

DEFAULT_MAX_ROUNDS

DEFAULT_MAX_ROUNDS = 5

Maximum number of internal callback rounds to attempt during generation before giving up.

FailMode

FailMode = Literal['raise', 'skip', 'include']

How to handle failures in pipelines.

• raise: Raise an exception when a failure is encountered.
• skip: Ignore the error and do not include the failed chat in the final output.
• include: Mark the message as failed and include it in the final output.

Chat

Chat(
    messages: Messages,
    generated: Messages | None = None,
    generator: Generator | None = None,
    params: GenerateParams | None = None,
    **kwargs: Any,
)

A completed chat interaction.

Initialize a Chat object.

Parameters:

• messages (Messages) –The messages for the chat.
• generated (Messages | None, default: None ) –The next messages for the chat.
• generator (Generator | None, default: None ) –The generator associated with this chat.
• **kwargs (Any, default: {} ) –Additional keyword arguments (typically used for deserialization)

all

all: list[Message]

Returns all messages in the chat, including the next messages.

conversation

conversation: str

Returns a string representation of the chat.

error

error: (
    Annotated[
        BaseException,
        PlainSerializer(
            lambda x: str(x),
            return_type=str,
            when_used=json - unless - none,
        ),
        WithJsonSchema(
            {type: string, description: "Error message"}
        ),
    ]
    | None
) = Field(None, repr=False)

Holds any exception that was caught during the generation pipeline.

extra

extra: dict[str, Any] = Field(
    default_factory=dict, repr=False
)

Any additional information from the generation.

failed

failed: bool = Field(
    default=False, exclude=False, repr=True
)

Indicates whether conditions during generation were not met. This is typically used for graceful error handling when parsing.

generated

generated: list[Message] = Field(default_factory=list)

The list of messages resulting from the generation.

generator

generator: Generator | None = Field(
    None, exclude=True, repr=False
)

The generator associated with the chat.

generator_id

generator_id: str | None

The identifier of the generator used to create the chat

last

last: Message

Alias for .all[-1]

message_dicts

message_dicts: list[MessageDict]

Returns the chat as a minimal message dictionaries.

message_metadata

message_metadata: dict[str, Any]

Returns a merged dictionary of metadata from all messages in the chat.

messages

messages: list[Message]

The list of messages prior to generation.

metadata

metadata: dict[str, Any] = Field(default_factory=dict)

Additional metadata for the chat.

next

next: list[Message]

Alias for the .generated property

params

params: GenerateParams | None = Field(None, repr=False)

Any additional generation params used for this chat.

prev

prev: list[Message]

Alias for the .messages property

stop_reason

stop_reason: StopReason = Field(default='unknown')

The reason the generation stopped.

timestamp

timestamp: datetime = Field(default_factory=now, repr=False)

The timestamp when the chat was created.

usage

usage: Usage | None = Field(None, repr=False)

The usage statistics for the generation if available.

uuid

uuid: UUID = Field(default_factory=uuid4)

The unique identifier for the chat.

apply

apply(**kwargs: str) -> Chat

Calls [rigging.message.Message.apply][] on the last message in the chat with the given keyword arguments.

Parameters:

• **kwargs (str, default: {} ) –The string mapping of replacements.

Returns:

• Chat –The updated chat.

apply_to_all

apply_to_all(**kwargs: str) -> Chat

Calls [rigging.message.Message.apply][] on all messages in the chat with the given keyword arguments.

Parameters:

• **kwargs (str, default: {} ) –The string mapping of replacements.

Returns:

• Chat –The updated chat.

inject_system_content

inject_system_content(content: str) -> Chat

Injects content into the chat as a system message.

Note

If the chat is empty or the first message is not a system message, a new system message with the given content is inserted at the beginning of the chat. If the first message is a system message, the content is appended to it.

Parameters:

• content (str) –The content to be injected.

Returns:

• Chat –The updated chat.

message_slices

message_slices(
    slice_type: SliceType | None = None,
    filter_fn: Callable[[MessageSlice], bool] | None = None,
    *,
    reverse: bool = False,
) -> list[MessageSlice]

Get all slices across all messages with optional filtering.

See Message.find_slices() for more information.

Parameters:

• slice_type (SliceType | None, default: None ) –Filter by slice type
• filter_fn (Callable[[MessageSlice], bool] | None, default: None ) –A function to filter slices. If provided, only slices for which filter_fn(slice) returns True will be included.
• reverse (bool, default: False ) –If True, the slices will be returned in reverse order.

Returns:

• list[MessageSlice] –List of all matching slices across all messages

meta

meta(**kwargs: Any) -> Chat

Updates the metadata of the chat with the provided key-value pairs.

Parameters:

• **kwargs (Any, default: {} ) –Key-value pairs representing the metadata to be updated.

Returns:

• Chat –The updated chat.

to_df

to_df() -> t.Any

Converts the chat to a Pandas DataFrame.

See [rigging.data.chats_to_df][] for more information.

Returns:

• Any –The chat as a DataFrame.

to_elastic

to_elastic(
    index: str,
    client: AsyncElasticsearch,
    *,
    op_type: ElasticOpType = "index",
    create_index: bool = True,
    **kwargs: Any,
) -> int

Converts the chat data to Elasticsearch format and indexes it.

See [rigging.data.chats_to_elastic][] for more information.

Returns:

• int –The number of chats indexed.

to_openai

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

Converts the chat messages to the OpenAI-compatible JSON format.

See Message.to_openai() for more information.

Returns:

• list[dict[str, Any]] –The serialized chat.

to_tokens

to_tokens(
    tokenizer: str | Tokenizer,
    transform: str | Transform | None = None,
) -> TokenizedChat

Converts the chat messages to a list of tokenized messages.

Parameters:

• tokenizer (str | Tokenizer) –The tokenizer to use for tokenization. Can be a string identifier or a Tokenizer instance.
• transform (str | Transform | None, default: None ) –An optional transform to apply to the chat before tokenization. Can be a well-known transform identifier or a Transform instance.

Returns:

• TokenizedChat –The serialized chat as a list of token lists.

transform

transform(transform: Transform | str) -> Chat

Applies a transform to the chat.

Parameters:

• transform (Transform | str) –The transform to apply.

Returns:

• Chat –A new chat with the transform applied to its messages and parameters.

ChatList

Represents a list of chat objects.

Inherits from the built-in list class and is specialized for storing Chat objects.

to_df

to_df() -> t.Any

Converts the chat list to a Pandas DataFrame.

See [rigging.data.chats_to_df][] for more information.

Returns:

• Any –The chat list as a DataFrame.

to_elastic

to_elastic(
    index: str,
    client: AsyncElasticsearch,
    *,
    op_type: ElasticOpType = "index",
    create_index: bool = True,
    **kwargs: Any,
) -> int

Converts the chat list to Elasticsearch format and indexes it.

See [rigging.data.chats_to_elastic][] for more information.

Returns:

• int –The number of chats indexed.

to_json

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

Helper to convert the chat list to a list of dictionaries.

to_openai

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

Converts the chat list to a list of OpenAI-compatible JSON format.

See Message.to_openai() for more information.

Returns:

• list[list[dict[str, Any]]] –The serialized chat list.

to_tokens

to_tokens(
    tokenizer: str | Tokenizer,
    transform: str | Transform | None = None,
) -> list[TokenizedChat]

Converts the chat list to a list of tokenized chats.

Parameters:

• tokenizer (str | Tokenizer) –The tokenizer to use for tokenization. Can be a string identifier or a Tokenizer instance.
• transform (str | Transform | None, default: None ) –An optional transform to apply to each chat before tokenization. Can be a well-known transform identifier or a Transform instance.

Returns:

• list[TokenizedChat] –A list of tokenized chats. This module covers core message objects and handling.

Content

Content = ContentText | ContentImageUrl | ContentAudioInput

The types of content that can be included in a message.

EPHERMAL_CACHE_CONTROL

EPHERMAL_CACHE_CONTROL = {'type': 'ephemeral'}

Cache control entry for ephemeral messages.

Role

Role = Literal['system', 'user', 'assistant', 'tool']

The role of a message. Can be ‘system’, ‘user’, ‘assistant’, or ‘tool’.

ContentAudioInput

An audio content part of a message.

cache_control

cache_control: dict[str, str] | None = None

Cache control entry for prompt caching.

input_audio

input_audio: Audio

The audio URL content.

transcript

transcript: str | None

Returns the transcript of the audio data.

Returns:

• str | None –The transcript of the audio data.

type

type: Literal['input_audio'] = 'input_audio'

The type of content (always input_audio).

Audio

data

data: str

The base64-encoded audio data.

format

format: str

The format of the audio data.

transcript

transcript: str | None = None

The transcript of the audio data (if available).

from_bytes

from_bytes(
    data: bytes,
    *,
    format: ContentAudioFormat | None = None,
    transcript: str | None = None,
) -> ContentAudioInput

Creates a ContentAudioInput object from raw bytes.

Parameters:

• data (bytes) –The raw bytes of the audio.
• format (ContentAudioFormat | None, default: None ) –The format of the audio.

Returns:

• ContentAudioInput –The created ContentAudioInput

from_file

from_file(
    file: Path | str,
    *,
    format: ContentAudioFormat | None = None,
    transcript: str | None = None,
) -> ContentAudioInput

Creates a ContentAudioInput object from a file.

Parameters:

• file (Path | str) –The file to create the content from.
• format (ContentAudioFormat | None, default: None ) –The format of the audio. If not provided, it will be guessed based on the file extension.
• transcript (str | None, default: None ) –The transcript of the audio data (if available).

Returns:

• ContentAudioInput –The created ContentAudioInput object.

save

save(path: Path | str) -> None

Saves the audio data to a file.

Parameters:

• path (Path | str) –The path to save the audio to.

to_bytes

to_bytes() -> bytes

Converts the audio data to bytes.

Returns:

• bytes –The decoded audio data.

ContentImageUrl

An image URL content part of a message.

cache_control

cache_control: dict[str, str] | None = None

Cache control entry for prompt caching.

image_url

image_url: ImageUrl

The image URL content.

type

type: Literal['image_url'] = 'image_url'

The type of content (always image_url).

ImageUrl

detail

detail: Literal['auto', 'low', 'high'] = 'auto'

The detail level of the image.

url

url: str

The URL of the image (supports base64-encoded).

from_bytes

from_bytes(
    data: bytes,
    mimetype: str,
    *,
    detail: Literal["auto", "low", "high"] = "auto",
) -> ContentImageUrl

Creates a ContentImageUrl object from raw bytes.

Parameters:

• data (bytes) –The raw bytes of the image.
• mimetype (str) –The mimetype of the image.
• detail (Literal['auto', 'low', 'high'], default: 'auto' ) –The detail level of the image.

Returns:

• ContentImageUrl –The created ContentImageUrl

from_file

from_file(
    file: Path | str,
    *,
    mimetype: str | None = None,
    detail: Literal["auto", "low", "high"] = "auto",
) -> ContentImageUrl

Creates a ContentImageUrl object from a file.

Parameters:

• file (Path | str) –The file to create the content from.
• mimetype (str | None, default: None ) –The mimetype of the file. If not provided, it will be guessed.

Returns:

• ContentImageUrl –The created ContentImageUrl object.

from_url

from_url(
    url: str,
    *,
    detail: Literal["auto", "low", "high"] = "auto",
) -> ContentImageUrl

Creates a ContentImageUrl object from a URL.

Parameters:

• url (str) –The URL of the image.
• detail (Literal['auto', 'low', 'high'], default: 'auto' ) –The detail level of the image.

Returns:

• ContentImageUrl –The created ContentImageUrl object.

save

save(path: Path | str) -> None

Saves the data to a file.

Parameters:

• path (Path | str) –The path to save the image to.

to_bytes

to_bytes() -> bytes

Converts the data to bytes (if the URL is base64-encoded).

Returns:

• bytes –The decoded image data.

ContentText

A text content part of a message.

cache_control

cache_control: dict[str, str] | None = None

Cache control entry for prompt caching.

text

text: str

The text content.

type

type: Literal['text'] = 'text'

The type of content (always text).

Message

Message(
    role: Role,
    content: str | Sequence[str | Content] | None = None,
    slices: Sequence[MessageSlice] | None = None,
    tool_calls: Sequence[ToolCall]
    | Sequence[dict[str, Any]]
    | None = None,
    tool_call_id: str | None = None,
    cache_control: Literal["ephemeral"]
    | dict[str, str]
    | None = None,
    **kwargs: Any,
)

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 content_parts. During serialization, we rename content_parts to content for compatibility.

all_content

all_content: str | list[Content]

Returns all content parts of the message or the single text content part as a string.

Deprecated - Use .content_parts instead

compatibility_flags

compatibility_flags: set[CompatibilityFlag] = Field(
    default_factory=set, repr=False
)

Compatibility flags to be applied when conversions occur.

content

content: str

The content of the message as a string. If multiple text parts are present, they will be concatenated together with newlines in between.

This is considered the ground truth for slices of this message. In other words, slices do not take into account any structured content parts like images or audio.

If you need to access the structured content parts, use .content_parts.

content_parts

content_parts: list[Content] = Field([], repr=False)

Interior str content or structured content parts.

hash

hash: int

Returns a weak hash of the functional message content, ignoring UUID, metadata, and supplementary fields.

metadata

metadata: dict[str, Any] = Field(
    default_factory=dict, repr=False
)

Metadata associated with the message.

models

models: list[XMLModel]

Returns a list of all models available in slices of the message.

parts

parts: list[Any]

Deprecated - iterate through .slices instead

role

role: Role

The role of the message.

slices

slices: list[MessageSlice]

The slices of the message content.

tool_call_id

tool_call_id: str | None = Field(None)

Associated call id if this message is a response to a tool call.

tool_calls

tool_calls: list[ToolCall] | None = Field(None)

The tool calls associated with the message.

uuid

uuid: UUID = Field(default_factory=uuid4, repr=False)

The unique identifier for the message.

append_slice

append_slice(
    content: str | XMLModel,
    slice_type: SliceType | None = None,
    *,
    obj: SliceObj | None = None,
    metadata: dict[str, Any] | None = None,
) -> MessageSlice

Add content to the end of the message (with newline separator) and create a slice tracking it.

Type defaults to ‘model’ for Model objects, ‘other’ for strings.

Parameters:

• content (str | XMLModel) –The content to append. This can be a string or a Model instance.
• slice_type (SliceType | None, default: None ) –The type of slice to create, inferred from content type if not provided.
• obj (SliceObj | None, default: None ) –The object associated with the slice
• metadata (dict[str, Any] | None, default: None ) –Additional metadata for the slice

Returns:

• MessageSlice –The created MessageSlice

apply

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.

apply_to_list

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

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

cache

cache(
    cache_control: dict[str, str] | bool = True,
) -> Message

Update cache control settings for this message.

Parameters:

• cache_control (dict[str, str] | bool, default: True ) –The cache control settings to apply to the message. If False, all cache control settings will be removed. If True, the default ephemeral cache control will be applied. If a dictionary, it will be applied as the cache control settings.

Returns:

• Message –The updated message.

clone

clone() -> Message

Creates a copy of the message.

find_slices

find_slices(
    slice_type: SliceType | None = None,
    filter_fn: Callable[[MessageSlice], bool] | None = None,
    *,
    reverse: bool = False,
) -> list[MessageSlice]

Find slices with simple filtering.

Parameters:

• slice_type (SliceType | None, default: None ) –Filter by slice type
• filter_fn (Callable[[MessageSlice], bool] | None, default: None ) –Custom filter function called for each slice

Returns:

• list[MessageSlice] –List of matching slices

fit

fit(
    message: Union[Message, MessageDict, Content, str],
) -> Message

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

fit_as_list

fit_as_list(
    messages: Sequence[MessageDict]
    | Sequence[Message]
    | MessageDict
    | Message
    | Content
    | str,
) -> list[Message]

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

from_model

from_model(
    models: XMLModel | Sequence[XMLModel],
    role: Role = "user",
    suffix: str | None = None,
    tool_call_id: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> Message

Create a Message object from one or more Model objects.

Parameters:

• models (XMLModel | Sequence[XMLModel]) –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.
• metadata (dict[str, Any] | None, default: None ) –Additional metadata for the Message.
• tool_call_id (str | None, default: None ) –The ID of the tool call associated with this message.

Returns:

• Message –The created Message object.

get_slice

get_slice(
    slice_type: SliceType | None = None,
    *,
    select: Literal["first", "last"] = "first",
) -> MessageSlice | None

Get a single slice of the message, optionally filtering by type.

Parameters:

• slice_type (SliceType | None, default: None ) –Optional type or string to filter slices by.
• select (Literal['first', 'last'], default: 'first' ) –Which slice to return - ‘first’ or ‘last’.

Returns:

• MessageSlice | None –The requested MessageSlice or None if not found.

iter_slices

iter_slices(
    slice_type: SliceType
    | Iterable[SliceType]
    | None = None,
    *,
    reverse: bool = False,
) -> t.Iterator[MessageSlice]

Iterate over slices of the message, optionally filtering by type.

Parameters:

• slice_type (SliceType | Iterable[SliceType] | None, default: None ) –Optional type or iterable of types to filter slices by.
• reverse (bool, default: False ) –If True, iterate in reverse order.

Returns:

• Iterator[MessageSlice] –An iterator over MessageSlice objects.

mark_slice

mark_slice(
    target: str
    | tuple[int, int]
    | Literal[-1]
    | Pattern[str]
    | type[XMLModel],
    slice_type: SliceType | None = None,
    *,
    obj: SliceObj | None = None,
    metadata: dict[str, Any] | None = None,
    select: Literal["first", "last"] = "first",
    case_sensitive: bool = True,
) -> MessageSlice | None

mark_slice(
    target: str
    | tuple[int, int]
    | Literal[-1]
    | Pattern[str]
    | type[XMLModel],
    slice_type: SliceType | None = None,
    *,
    obj: SliceObj | None = None,
    metadata: dict[str, Any] | None = None,
    select: Literal["all"],
    case_sensitive: bool = True,
) -> list[MessageSlice]

mark_slice(
    target: str
    | tuple[int, int]
    | Literal[-1]
    | Pattern[str]
    | type[XMLModel],
    slice_type: SliceType | None = None,
    *,
    obj: SliceObj | None = None,
    metadata: dict[str, Any] | None = None,
    select: Literal["first", "last", "all"] = "first",
    case_sensitive: bool = True,
) -> MessageSlice | list[MessageSlice] | None

Mark existing content as slices without modifying content.

Parameters:

• target (str | tuple[int, int] | Literal[-1] | Pattern[str] | type[XMLModel]) –What to mark as a slice:
  • str: Find this text in content
  • tuple[int, int]: Mark this exact range
  • ”*” or -1: Mark entire message content
  • re.Pattern: Find matches of this pattern
  • type[Model]: Parse and mark instances of this model type
• slice_type (SliceType | None, default: None ) –The type of slice to create
• obj (SliceObj | None, default: None ) –The object associated with the slice
• metadata (dict[str, Any] | None, default: None ) –Additional metadata for the slice
• select (Literal['first', 'last', 'all'], default: 'first' ) –Which matches to return - ‘first’, ‘last’, or ‘all’
• case_sensitive (bool, default: True ) –Whether string search should be case sensitive

Returns:

• MessageSlice | list[MessageSlice] | None –If select=‘first’/‘last’: MessageSlice or None if no matches, otherwise if select=‘all’: list[MessageSlice] (empty if no matches)

meta

meta(**kwargs: Any) -> Message

Updates the metadata of the message with the provided key-value pairs.

Parameters:

• **kwargs (Any, default: {} ) –Key-value pairs representing the metadata to be updated.

Returns:

• Message –The updated message.

parse

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.

parse_many

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.

parse_set

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.

remove_slices

remove_slices(
    *slices: MessageSlice | str | SliceType | type[Any],
) -> list[MessageSlice]

Removes and returns slices from the message that match the given object.

If the object is a string, it will find slices that match the string content. If the object is a SliceType, it will find slices of that type. If the object is a type, it will find slices that have an obj of that type. If the object is a MessageSlice, it will remove that slice exactly.

Parameters:

• *slices (MessageSlice | str | SliceType | type[Any], default: () ) –The slices to remove. Can be a MessageSlice, a string, a SliceType, or a type.

Returns:

• list[MessageSlice] –The removed MessageSliceRef objects.

replace_with_slice

replace_with_slice(
    content: str | XMLModel,
    slice_type: SliceType | None = None,
    *,
    obj: SliceObj | None = None,
    metadata: dict[str, Any] | None = None,
) -> MessageSlice

Replace all message content and create a slice tracking the new content.

Type defaults to ‘model’ for Model objects, ‘other’ for strings.

Parameters:

• content (str | XMLModel) –The content to replace with. This can be a string or a Model instance.
• slice_type (SliceType | None, default: None ) –The type of slice to create, inferred from content type if not provided.
• obj (SliceObj | None, default: None ) –The object associated with the slice
• metadata (dict[str, Any] | None, default: None ) –Additional metadata for the slice

Returns:

• MessageSlice –The created MessageSlice

shorten

shorten(max_length: int, sep: str = '...') -> Message

Shortens the message content to at most max_length characters long by removing the middle of the string

Parameters:

• max_length (int) –The maximum length of the message content.
• sep (str, default: '...' ) –The separator to use when shortening the content.

Returns:

• Message –The shortened message.

strip

strip(obj: SliceType | type[Any]) -> list[MessageSlice]

Removes and returns all slices of the specified type from the message.

This is a deprecated method, use remove_slice() instead.

Parameters:

• obj (SliceType | type[Any]) –The type of slice to remove. Can be a SliceType or a model class. If a model class is provided, it will remove all slices that have a model of that type.

Returns:

• list[MessageSlice] –A list of removed slices.

to_openai

to_openai(
    *,
    compatibility_flags: set[CompatibilityFlag]
    | None = None,
) -> 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.

to_openai_spec

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.

Deprecated - Use .to_openai instead

truncate

truncate(
    max_length: int, suffix: str = "\n[truncated]"
) -> Message

Truncates the message content to a maximum length.

Parameters:

• max_length (int) –The maximum length of the message content.

Returns:

• Message –The truncated message.

try_parse

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.

try_parse_many

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.

try_parse_set

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.

MessageDict

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

content

content: str | list[Any]

The content of the message.

role

role: Role

The role of the message.

MessageSlice

Represents a slice content within a message.

This can be a tool call, tool response, or model output. You can associate metadata with the slice to add rich information like scores, confidence levels, or reward information.

content

content: str

Get the content text for this slice from the parent message.

metadata

metadata: dict[str, Any] = Field(default_factory=dict)

Metadata associated with the slice.

obj

obj: SerializeAsAny[SliceObj] | None = Field(
    default=None, repr=False
)

The model, tool call, or other object associated with the slice.

slice_

slice_: slice

Returns the slice representing the range into the message content.

start

start: int

The start index of the slice.

stop

stop: int

The stop index of the slice.

type

type: SliceType

The type of the slice.

__len__

__len__() -> int

Returns the length of the slice.

__str__

__str__() -> str

Returns a string representation of the slice.

clone

clone() -> MessageSlice

Creates a deep copy of the MessageSlice.

Returns:

• MessageSlice –A new MessageSlice instance with the same properties.

inject_system_content

inject_system_content(
    messages: list[Message], content: str
) -> list[Message]

Injects content into a list of messages as a system message.

Note

If the message list is empty or the first message is not a system message, a new system message with the given content is inserted at the beginning of the list. If the first message is a system message, the content is appended to it.

Parameters:

• messages (list[Message]) –The list of messages to modify.
• content (str) –The content to be injected.

Returns:

• list[Message] –The modified list of messages

make_compaction_message

make_compaction_message(
    summary_text: str,
    *,
    messages_compacted: int,
    trigger: str,
) -> Message

Create the compaction marker message for conversation summarization.

This is the single source of truth for the <conversation-summary> XML format used by threshold compaction, overflow recovery, and manual /compact. All code paths that produce compaction markers must use this function.

strip_system_content

strip_system_content(
    messages: list[Message], content: str
) -> list[Message]

Strips the system message from a list of messages.

Parameters:

• messages (list[Message]) –The list of messages to modify.

Returns:

• list[Message] –The modified list of messages without the system message. Generators produce completions for a given set of messages or text.

HttpHook

HttpHook = Callable[
    ["HTTPGenerator", Response],
    Awaitable[HttpHookAction | None],
]

Hook to run after each HTTP request of the HTTPGenerator.

The hook receives the generator instance and the HTTP response.

It can return:

• “retry”: to retry the request.
• “raise”: to raise an error.
• “continue”/None: to continue processing without retrying.

StopReason

StopReason = Literal[
    "stop",
    "length",
    "content_filter",
    "tool_calls",
    "unknown",
]

Reporting reason for generation completing.

GenerateParams

Parameters for generating text using a language model.

These are designed to generally overlap with underlying APIs like litellm, but will be extended as needed.

Note

Use the extra field to pass additional parameters to the API.

api_base

api_base: str | None = None

The base URL for the API.

audio

audio: dict[str, str] | None = None

The audio parameters to be used in the generation.

extra

extra: dict[str, Any] = Field(default_factory=dict)

Extra parameters to be passed to the API.

frequency_penalty

frequency_penalty: float | None = None

The frequency penalty.

max_tokens

max_tokens: int | None = None

The maximum number of tokens to generate.

modalities

modalities: list[str] | None = None

The modalities to be used in the generation.

parallel_tool_calls

parallel_tool_calls: bool | None = None

Whether to run allow tool calls in parallel.

presence_penalty

presence_penalty: float | None = None

The presence penalty.

seed

seed: int | None = None

The random seed.

stop

stop: list[str] | None = None

A list of stop sequences to stop generation at.

temperature

temperature: float | None = None

The sampling temperature.

timeout

timeout: int | None = None

The timeout for the API request.

tool_choice

tool_choice: ToolChoice | None = None

The tool choice to be used in the generation.

tools

tools: list[ToolDefinition] | None = None

The tools to be used in the generation.

top_k

top_k: int | None = None

The top-k sampling parameter.

top_p

top_p: float | None = None

The nucleus sampling probability.

__hash__

__hash__() -> int

Create a hash based on the json representation of this object.

clone

clone() -> GenerateParams

Create a copy of the current parameters instance.

Returns:

• GenerateParams –A new instance of GenerateParams with the same values.

merge_with

merge_with(
    *others: GenerateParams | None,
) -> GenerateParams

Apply a series of parameter overrides to the current instance and return a copy.

Parameters:

• *others (GenerateParams | None, default: () ) –The parameters to be merged with the current instance’s parameters. Can be multiple and overrides will be applied in order.

Returns:

• GenerateParams –The merged parameters instance.

to_dict

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

Convert the parameters to a dictionary.

Returns:

• dict[str, Any] –The parameters as a dictionary.

GeneratedMessage

A generated message with additional generation information.

extra

extra: dict[str, Any] = Field(default_factory=dict)

Any additional information from the generation.

message

message: Message

The generated message.

stop_reason

stop_reason: Annotated[
    StopReason, BeforeValidator(convert_stop_reason)
] = "unknown"

The reason for stopping generation.

usage

usage: Usage | None = None

The usage statistics for the generation if available.

GeneratedText

A generated text with additional generation information.

extra

extra: dict[str, Any] = Field(default_factory=dict)

Any additional information from the generation.

stop_reason

stop_reason: Annotated[
    StopReason, BeforeValidator(convert_stop_reason)
] = "unknown"

The reason for stopping generation.

text

text: str

The generated text.

usage

usage: Usage | None = None

The usage statistics for the generation if available.

Generator

Base class for all rigging generators.

This class provides common functionality and methods for generating completion messages.

A subclass of this can implement both or one of the following:

• generate_messages: Process a batch of messages.
• generate_texts: Process a batch of texts.

api_key

api_key: str | None = Field(None, exclude=True)

The API key used for authentication.

model

model: str

The model name to be used by the generator.

params

params: GenerateParams

The parameters used for generating completion messages.

generate_messages

generate_messages(
    messages: Sequence[Sequence[Message]],
    params: Sequence[GenerateParams],
) -> t.Sequence[GeneratedMessage | BaseException]

Generate a batch of messages using the specified parameters.

Note

The length of params must be the same as the length of many.

Parameters:

• messages (Sequence[Sequence[Message]]) –A sequence of sequences of messages.
• params (Sequence[GenerateParams]) –A sequence of GenerateParams objects.

Returns:

• Sequence[GeneratedMessage | BaseException] –A sequence of generated messages.

Raises:

• NotImplementedError –This method is not supported by this generator.

generate_texts

generate_texts(
    texts: Sequence[str], params: Sequence[GenerateParams]
) -> t.Sequence[GeneratedText | BaseException]

Generate a batch of text completions using the generator.

Note

This method falls back to looping over the inputs and calling generate_text for each item.

Note

If supplied, the length of params must be the same as the length of many.

Parameters:

• texts (Sequence[str]) –The input texts for generating the batch.
• params (Sequence[GenerateParams]) –Additional parameters for generating each text in the batch.

Returns:

• Sequence[GeneratedText | BaseException] –The generated texts.

Raises:

• NotImplementedError –This method is not supported by this generator.

load

load() -> Self

If supported, trigger underlying loading and preparation of the model.

Returns:

• Self –The generator.

prompt

prompt(
    func: Callable[P, Coroutine[None, None, R]],
) -> t.Any

Decorator to convert a function into a prompt bound to this generator.

Note

This method is deprecated. Use the generator’s generate_messages method directly.

Parameters:

• func (Callable[P, Coroutine[None, None, R]]) –The function to be converted into a prompt.

Raises:

• NotImplementedError –This method is no longer supported.

supports_function_calling

supports_function_calling() -> bool | None

Check if the generator supports calling functions explicitly or is unknown.

Returns:

• bool | None –True/False if the generator supports function calling, None if unknown.

supports_prompt_caching

supports_prompt_caching() -> bool

Check if the generator supports prompt caching via cache_control markers.

Returns:

• bool –True if the generator supports prompt caching, False otherwise.

to_identifier

to_identifier(
    params: GenerateParams | None = None,
    *,
    short: bool = False,
) -> str

Converts the generator instance back into a rigging identifier string.

This calls [rigging.generator.get_identifier][] with the current instance.

Parameters:

• params (GenerateParams | None, default: None ) –The generation parameters.

Returns:

• str –The identifier string.

unload

unload() -> Self

If supported, clean up resources used by the underlying model.

Returns:

• Self –The generator.

wrap

wrap(func: Callable[[CallableT], CallableT] | None) -> Self

If supported, wrap any underlying interior framework calls with this function.

This is useful for adding things like backoff or rate limiting.

Parameters:

• func (Callable[[CallableT], CallableT] | None) –The function to wrap the calls with.

Returns:

• Self –The generator.

HTTPGenerator

Generator to map messages to HTTP requests and back.

The generator takes a spec attribute which describes how to encode messages into HTTP requests and decode the responses back into messages.

You can pass this spec as a python dictionary, JSON string, YAML string, or a base64 encoded JSON/YAML string.

Example

from dreadnode.generators import HTTPGenerator

spec = r"""
request:
url: "https://{{ model }}.crucible.dreadnode.io/submit"
headers:
    "X-Api-Key": "{{ api_key }}"
    "Content-Type": "application/json"
transforms:
    - type: "json"
    pattern: {
        "data": "$content"
    }
response:
transforms:
    - type: "jsonpath"
    pattern: $.flag,output,message
"""

crucible = rg.get_generator("http!test,api_key=<key>")
crucible.spec = spec

chat = await crucible.chat("How about a flag?").run()

print(chat.conversation)

hook

hook: HttpHook | None = Field(default=None, exclude=True)

Optional hook to run after each HTTP request with the option to retry or raise an error.

max_retries

max_retries: int = DEFAULT_MAX_RETRIES

“Maximum number of retries the hook can trigger. Defaults to 5.

spec

spec: HTTPSpec | None = None

Specification for building/parsing HTTP interactions.

state

state: dict[str, Any] = Field(default_factory=dict)

Mutable dictionary for dynamic state like access tokens to use in your spec.

for_json_endpoint

for_json_endpoint(
    url: str,
    request: dict[str, Any],
    model: str | None = None,
    api_key: str | None = None,
    method: str = "POST",
    headers: dict[str, str] | None = None,
    auth: HttpAuthConfigDict | HttpAuthConfig | None = None,
    response: ApiResponseConfigDict
    | ApiResponseConfig
    | None = None,
    valid_status_codes: list[int] | None = None,
    timeout: int | None = None,
    hook: HttpHook | None = None,
    state: dict[str, Any] | None = None,
    **kwargs: Any,
) -> HTTPGenerator

Creates an HTTPGenerator from a simplified, high-level API definition for JSON endpoints.

This is the recommended entry point for most use cases. It provides full autocompletion when creating configuration dictionaries in your IDE.

Example

from dreadnode.generators import HTTPGenerator

openai_api = HTTPGenerator.for_json_endpoint(
    "https://api.openai.com/v1/chat/completions",
    auth={
        "header": "Authorization",
        "format": "Bearer {api_key}"
    },
    request={
        "model": "{{ model }}",
        "messages": "$messages",
    },
    response={
        "content_path": "$.choices[0].message.content",
        "error_path": "$.error.message"
    }
)

Parameters:

• url (str) –The URL of the API endpoint (supports Jinja templates).
• request (dict[str, Any]) –A dictionary defining the request body structure. Use $<variable> to reference context variables.
• model (str | None, default: None ) –Optional model name for the generator.
• api_key (str | None, default: None ) –Optional API key to use for authentication.
• method (str, default: 'POST' ) –HTTP method to use (default is “POST”).
• headers (dict[str, str] | None, default: None ) –Optional headers to include in the request. Defaults to “Content-Type”: “application/json”.
• auth (HttpAuthConfigDict | HttpAuthConfig | None, default: None ) –Optional authentication configuration for API key headers.
• response (ApiResponseConfigDict | ApiResponseConfig | None, default: None ) –Optional configuration for parsing the response body.
• valid_status_codes (list[int] | None, default: None ) –List of valid HTTP status codes (default is [200]).
• timeout (int | None, default: None ) –Optional timeout in seconds for the request.
• hook (HttpHook | None, default: None ) –Optional hook to run after each HTTP request.
• state (dict[str, Any] | None, default: None ) –Optional mutable dictionary for dynamic state like access tokens.
• **kwargs (Any, default: {} ) –Additional keyword arguments passed to the generator.

Returns:

• HTTPGenerator –An instance of HTTPGenerator configured for the specified endpoint.

for_text_endpoint

for_text_endpoint(
    url: str,
    request: str,
    response_pattern: str | None = None,
    response_pattern_type: Literal[
        "regex", "jinja"
    ] = "regex",
    model: str | None = None,
    api_key: str | None = None,
    method: str = "POST",
    headers: dict[str, str] | None = None,
    auth: HttpAuthConfigDict | HttpAuthConfig | None = None,
    valid_status_codes: list[int] | None = None,
    timeout: int | None = None,
    hook: HttpHook | None = None,
    state: dict[str, Any] | None = None,
    **kwargs: Any,
) -> HTTPGenerator

Creates an HTTPGenerator from a template-based definition.

Ideal for simpler text-based APIs where the request body is generated from a Jinja2 template and the response is parsed with a Regex or another template.

Example

from dreadnode.generators import HTTPGenerator

text_api = HTTPGenerator.for_text_endpoint(
    "http://api.example.com/prompt",
    "User prompt: {{ content }}", # Jinja template
    response_pattern="Response: (.*)", # Regex to extract content
    auth={
        "header": "Authorization",
        "format": "Bearer {api_key}"
    }
)

Parameters:

• url (str) –The URL of the API endpoint (supports Jinja templates).
• request (str) –A Jinja template string for the request body.
• response_pattern (str | None, default: None ) –Optional pattern to extract content from the response. If not provided, the entire response body will be used.
• response_pattern_type (Literal['regex', 'jinja'], default: 'regex' ) –Type of the response pattern, either “regex” or “jinja
• model (str | None, default: None ) –Optional model name for the generator.
• api_key (str | None, default: None ) –Optional API key to use for authentication.
• method (str, default: 'POST' ) –HTTP method to use (default is “POST”).
• headers (dict[str, str] | None, default: None ) –Optional headers to include in the request. Defaults to “Content-Type”: “text/plain”.
• auth (HttpAuthConfigDict | HttpAuthConfig | None, default: None ) –Optional authentication configuration for API key headers.
• valid_status_codes (list[int] | None, default: None ) –List of valid HTTP status codes (default is [200]).
• timeout (int | None, default: None ) –Optional timeout in seconds for the request.
• hook (HttpHook | None, default: None ) –Optional hook to run after each HTTP request.
• state (dict[str, Any] | None, default: None ) –Optional mutable dictionary for dynamic state like access tokens.
• **kwargs (Any, default: {} ) –Additional keyword arguments passed to the generator.

HTTPSpec

Defines how to build requests and parse responses for the HTTPGenerator.

request

request: RequestSpec

Specification for building the request.

response

response: ResponseSpec | None = None

Specification for parsing the response.

LiteLLMGenerator

Generator backed by the LiteLLM library.

Find more information about supported models and formats in their docs..

Note

Batching support is not performant and simply a loop over inputs.

Caution

While some providers support passing n to produce a batch of completions per request, we don’t currently use this in the implementation due to it’s brittle requirements.

Tip

Consider setting [max_connections][rigging.generator.litellm_.LiteLLMGenerator.max_connections] or [min_delay_between_requests][rigging.generator.litellm_.LiteLLMGenerator.min_delay_between_requests if you run into API limits. You can pass this directly in the generator id:

get_generator("litellm!openai/gpt-4o,max_connections=2,min_delay_between_requests=1000")

max_connections

max_connections: int = 10

How many simultaneous requests to pool at one time. This is useful to set when you run into API limits at a provider.

Set to 0 to remove the limit.

min_delay_between_requests

min_delay_between_requests: float = 0.0

Minimum time (ms) between each request. This is useful to set when you run into API limits at a provider.

Usage

Usage statistics for a generation.

cache_creation_input_tokens

cache_creation_input_tokens: int = 0

Input tokens that wrote to the prompt cache on this call.

cache_read_input_tokens

cache_read_input_tokens: int = 0

Input tokens served from prompt cache (cheaper re-reads).

cost_usd

cost_usd: float | None = None

Estimated USD cost for the generation, sourced from litellm’s per-provider cost calculator (cache reads/writes, reasoning tokens, region/tier multipliers all accounted for). None when the underlying provider didn’t supply a cost — callers should fall back or report unknown rather than infer from token rates.

input_tokens

input_tokens: int = 0

The number of input tokens.

output_tokens

output_tokens: int = 0

The number of output tokens.

total_tokens

total_tokens: int = 0

The total number of tokens processed.

get_generator

get_generator(
    identifier: str,
    *,
    params: GenerateParams | dict[str, Any] | None = None,
) -> Generator

Get a generator by an identifier string. Uses LiteLLM by default.

Identifier strings are formatted like <provider>!<model>,\<**kwargs>

(provider is optional and defaults to litellm if not specified)

Examples:

• “gpt-3.5-turbo” -> LiteLLMGenerator(model="gpt-3.5-turbo")
• “litellm!claude-2.1” -> LiteLLMGenerator(model="claude-2.1")
• “mistral/mistral-tiny” -> LiteLLMGenerator(model="mistral/mistral-tiny")

You can also specify arguments to the generator by comma-separating them:

• “mistral/mistral-medium,max_tokens=1024”
• “gpt-4-0613,temperature=0.9,max_tokens=512”
• “claude-2.1,stop_sequences=Human:;test,max_tokens=100”

(These get parsed as [rigging.generator.GenerateParams][])

Parameters:

• identifier (str) –The identifier string to use to get a generator.
• params (GenerateParams | dict[str, Any] | None, default: None ) –The generation parameters to use for the generator. These will override any parameters specified in the identifier string.

Returns:

• Generator –The generator object.

Raises:

• InvalidGeneratorError –If the identifier is invalid.

get_identifier

get_identifier(
    generator: Generator,
    params: GenerateParams | None = None,
    *,
    short: bool = False,
) -> str

Converts the generator instance back into a rigging identifier string.

Caution

The extra parameter field is not currently supported in identifiers.

Parameters:

• generator (Generator) –The generator object.
• params (GenerateParams | None, default: None ) –The generation parameters.

Returns:

• str –The identifier string for the generator.

register_generator

register_generator(
    provider: str,
    generator_cls: type[Generator] | LazyGenerator,
) -> None

Register a generator class for a provider id.

This let’s you use [rigging.generator.get_generator][] with a custom generator class.

Parameters:

• provider (str) –The name of the provider.
• generator_cls (type[Generator] | LazyGenerator) –The generator class to register.

Returns:

• None –None Tokenizers encode chats and associated message data into tokens for training and inference.

TokenSlice

TokenSlice(
    start: int,
    end: int,
    type: SliceType,
    obj: SliceObj | None = None,
    metadata: dict[str, Any] | None = None,
)

Represents a slice of tokens within a tokenized chat.

end

end: int

The ending index of the slice in the token list.

metadata

metadata: dict[str, Any] | None = None

Additional metadata associated with this slice, if any.

obj

obj: SliceObj | None = None

The original object this slice corresponds to, if any.

start

start: int

The starting index of the slice in the token list.

type

type: SliceType

The type of the slice (e.g. message, tool_call, etc.).

TokenizedChat

TokenizedChat(
    text: str,
    tokens: list[int],
    slices: list[TokenSlice],
    obj: Chat | None = None,
    metadata: dict[str, Any] | None = None,
)

A tokenized representation of a chat, containing the full text, token list, and structured slices of tokens.

metadata

metadata: dict[str, Any] | None = None

Additional metadata associated with the tokenized chat, if any.

obj

obj: Chat | None = None

The original chat object, if available.

slices

slices: list[TokenSlice]

Structured slices of tokens, each representing a part of the chat.

text

text: str

The full text of the chat, formatted as a single string.

tokens

tokens: list[int]

The list of tokens representing the chat text.

Tokenizer

Base class for all rigging tokenizers.

This class provides common functionality and methods for tokenizing chats.

model

model: str

The model name to be used by the tokenizer.

decode

decode(tokens: list[int]) -> str

Decodes a list of tokens back into a string.

Parameters:

• tokens (list[int]) –The list of tokens to decode.

Returns:

• str –The decoded string.

encode

encode(text: str) -> list[int]

Encodes the given text into a list of tokens.

Parameters:

• text (str) –The text to encode.

Returns:

• list[int] –A list of tokens representing the encoded text.

format_chat

format_chat(chat: Chat) -> str

Formats the chat into a string representation.

Parameters:

• chat (Chat) –The chat object to format.

Returns:

• str –A string representation of the chat.

tokenize_chat

tokenize_chat(chat: Chat) -> TokenizedChat

Transform a chat into a tokenized format with structured slices.

Parameters:

• chat (Chat) –The chat object to tokenize.

Returns:

• TokenizedChat –A TokenizedChat object containing the tokenized chat data.

get_tokenizer

get_tokenizer(identifier: str) -> Tokenizer

Get a tokenizer by an identifier string. Uses Transformers by default.

Identifier strings are formatted like <provider>!<model>,\<**kwargs>

(provider is optional and defaults to transformers if not specified)

Examples:

• “meta-llama/Meta-Llama-3-8B-Instruct” -> TransformersTokenizer(model="meta-llama/Meta-Llama-3-8B-Instruct”)`
• “transformers!microsoft/Phi-4-mini-instruct” -> TransformersTokenizer(model="microsoft/Phi-4-mini-instruct")

Parameters:

• identifier (str) –The identifier string to use to get a tokenizer.

Returns:

• Tokenizer –The tokenizer object.

Raises:

• InvalidTokenizerError –If the identifier is invalid.

register_tokenizer

register_tokenizer(
    provider: str,
    tokenizer_cls: type[Tokenizer] | LazyTokenizer,
) -> None

Register a tokenizer class for a provider id.

This let’s you use [rigging.tokenizer.get_tokenizer][] with a custom tokenizer class.

Parameters:

• provider (str) –The name of the provider.
• tokenizer_cls (type[Tokenizer] | LazyTokenizer) –The tokenizer class to register.

Returns:

• None –None Models are the core datatypes for structured parsing.

Answer

Quick model for answers.

CommaDelimitedAnswer

Comma delimited answer (,)

DelimitedAnswer

Mixed support delimited answer (- | / ,) selected based on most-matches

items

items: list[str]

Parsed items from the content.

Description

Quick model for descriptions.

ErrorModel

from_exception

from_exception(exception: Exception) -> te.Self

Create an ErrorModel instance from an exception.

Parameters:

• exception (Exception) –The exception to convert.

Returns:

• Self –An instance of ErrorModel with the exception content.

Instructions

Quick model for instructions.

NewlineDelimitedAnswer

Newline delimited answer ( )

Question

Quick model for questions.

QuestionAnswer

Quick model for question-answer pairs.

answer

answer: Answer = element()

The answer

question

question: Question = element()

The question

Thinking

Quick model for thinking messages.

XMLModel

from_text

from_text(
    content: str, *, return_errors: Literal[False] = False
) -> list[tuple[te.Self, slice]]

from_text(
    content: str, *, return_errors: Literal[True]
) -> list[tuple[te.Self | Exception, slice]]

from_text(
    content: str, *, return_errors: bool = False
) -> (
    list[tuple[te.Self, slice]]
    | list[tuple[te.Self | Exception, slice]]
)

The core parsing method which attempts to extract and parse as many valid instances of a model from semi-structured text.

Parameters:

• content (str) –The text content to parse.

Returns:

• list[tuple[Self, slice]] | list[tuple[Self | Exception, slice]] –A list of tuples containing the extracted models and their corresponding slices.

Raises:

• MissingModelError –If the specified model tags are not found in the message.
• ValidationError –If an error occurs while parsing the content.

is_simple

is_simple() -> bool

Check if the model is “simple”, meaning it has a single field with a basic datatype.

Until we refactor our XML parsing, this helps make the parsing more consistent for models which can support it.

Returns:

• bool –True if the model is simple, False otherwise.

is_simple_with_attrs

is_simple_with_attrs() -> bool

Check if the model would otherwise be marked as “simple”, but has other fields which are all attributes. If so, we can do some parsing magic below and make sure our non-element field is updated with the extracted content properly, while pydantic-xml takes care of the attributes.

Returns:

• bool –True if the model is simple with attrs, False otherwise.

one_from_text

one_from_text(
    content: str, *, fail_on_many: bool = False
) -> tuple[te.Self, slice]

Finds and returns a single match from the given text content.

Parameters:

• content (str) –The text content to search for matches.
• fail_on_many (bool, default: False ) –If True, raises a ValueError if multiple matches are found.

Returns:

• tuple[Self, slice] –A tuple containing the matched model and the slice indicating the match location.

Raises:

• ValueError –If multiple matches are found and fail_on_many is True.

preprocess_with_cdata

preprocess_with_cdata(content: str) -> str

Process the content and attempt to auto-wrap interior field content in CDATA tags if they contain unescaped XML entities.

Parameters:

• content (str) –The XML content to preprocess.

Returns:

• str –The processed XML content with CDATA tags added where necessary.

to_pretty_xml

to_pretty_xml(
    *,
    skip_empty: bool = False,
    exclude_none: bool = False,
    exclude_unset: bool = False,
    **_: Any,
) -> str

Converts the model to a pretty XML string with indents and newlines.

Returns:

• str –The pretty XML representation of the model.

to_xml

to_xml(
    *,
    skip_empty: bool = False,
    exclude_none: bool = False,
    exclude_unset: bool = False,
    **kwargs: Any,
) -> str

Serializes the object to an xml string.

Parameters:

• skip_empty (bool, default: False ) –skip empty elements (elements without sub-elements, attributes and text, Nones)
• exclude_none (bool, default: False ) –exclude None values
• exclude_unset (bool, default: False ) –exclude values that haven’t been explicitly set
• kwargs (Any, default: {} ) –additional xml serialization arguments

Returns:

• str –object xml representation

xml_end_tag

xml_end_tag() -> str

Helper method which wrapped the class tag in XML braces with a leading slash.

xml_example

xml_example() -> str

Returns an example XML representation of the given class.

This method generates a pretty-printed XML string that includes:

• Example values for each field, taken from the example argument in a field constructor.
• Field descriptions as XML comments, derived from the field’s docstring or the description argument.

Note: This implementation is designed for models with flat structures and does not recursively generate examples for nested models.

Returns:

• str –A string containing the pretty-printed XML example.

xml_start_tag

xml_start_tag() -> str

Helper method which wrapped the class tag in XML braces.

xml_tags

xml_tags() -> str

Helper method which returns the full XML tags for the class.

YesNoAnswer

Yes/No answer answer with coercion

boolean

boolean: bool

The boolean value of the answer.

make_from_schema

make_from_schema(
    schema: dict[str, Any],
    name: str | None = None,
    *,
    allow_primitive: bool = False,
) -> type[XMLModel]

Helper to build a Rigging model dynamically from a JSON schema.

Note

There are plenty of edge cases this doesn’t handle, consider this very experimental and only suitable for simple schemas.

Parameters:

• schema (dict[str, Any]) –The JSON schema to build the model from.
• name (str | None, default: None ) –The name of the model (otherwise inferred from the schema).
• allow_primitive (bool, default: False ) –If True, allows the model to be a simple primitive

Returns:

• type[XMLModel] –The Pydantic model class.

make_primitive

make_primitive(
    name: str,
    type_: type[PrimitiveT] = str,
    *,
    tag: str | None = None,
    doc: str | None = None,
    validator: Callable[[str], str | None] | None = None,
    strip_content: bool = True,
) -> type[Primitive[PrimitiveT]]

Helper to create a simple primitive model with an optional content validator.

Note

This API is experimental and may change in the future.

Parameters:

• name (str) –The name of the model.
• tag (str | None, default: None ) –The XML tag for the model.
• doc (str | None, default: None ) –The documentation for the model.
• validator (Callable[[str], str | None] | None, default: None ) –An optional content validator for the model.
• strip_content (bool, default: True ) –Whether to strip the content string before pydantic validation.

Returns:

• type[Primitive[PrimitiveT]] –The primitive model class. Utilities for converting chat data between different formats.

ElasticMapping

ElasticMapping = {
    "properties": {
        "generated": {"type": "nested"},
        "messages": {"type": "nested"},
    }
}

Default index mapping for chat objects in elastic.

ElasticOpType

ElasticOpType = Literal['index', 'create', 'delete']

Available operations for bulk operations.

chats_to_df

chats_to_df(chats: Chat | Sequence[Chat]) -> pd.DataFrame

Convert a Chat or list of Chat objects into a pandas DataFrame.

Note

The messages will be flatted and can be joined by the chat_id column.

Parameters:

• chats (Chat | Sequence[Chat]) –A Chat or list of Chat objects.

Returns:

• DataFrame –A pandas DataFrame containing the chat data.

chats_to_elastic

chats_to_elastic(
    chats: Chat | Sequence[Chat],
    index: str,
    client: AsyncElasticsearch,
    *,
    op_type: ElasticOpType = "index",
    create_index: bool = True,
    **kwargs: Any,
) -> int

Convert chat data to Elasticsearch bulk operation format and store it with a client.

Parameters:

• chats (Chat | Sequence[Chat]) –The chat or list of chats to be converted and stored.
• index (str) –The name of the Elasticsearch index where the data will be stored.
• client (AsyncElasticsearch) –The AsyncElasticsearch client instance.
• op_type (ElasticOpType, default: 'index' ) –The operation type for Elasticsearch. Defaults to “create”.
• create_index (bool, default: True ) –Whether to create the index if it doesn’t exist and update its mapping.
• kwargs (Any, default: {} ) –Additional keyword arguments to be passed to the Elasticsearch client.

Returns:

• int –The indexed count from the bulk operation

chats_to_elastic_data

chats_to_elastic_data(
    chats: Chat | Sequence[Chat],
    index: str,
    *,
    op_type: ElasticOpType = "index",
) -> list[dict[str, t.Any]]

Convert chat data to Elasticsearch bulk operation format.

Parameters:

• chats (Chat | Sequence[Chat]) –The chat or list of chats to be converted.
• op_type (ElasticOpType, default: 'index' ) –The operation type for Elasticsearch.

Returns:

• list[dict[str, Any]] –Formatted bulk operation dict.

df_to_chats

df_to_chats(df: DataFrame) -> list[Chat]

Convert a pandas DataFrame into a list of Chat objects.

Note

The DataFrame should have the same structure as the one generated by the chats_to_df function.

Parameters:

• df (DataFrame) –A pandas DataFrame containing the chat data.

Returns:

• list[Chat] –A list of Chat objects.

elastic_data_to_chats

elastic_data_to_chats(
    data: Mapping[str, Any] | ObjectApiResponse[Any],
) -> list[Chat]

Convert the raw elastic results into a list of Chat objects.

elastic_to_chats

elastic_to_chats(
    query: Mapping[str, Any],
    index: str,
    client: AsyncElasticsearch,
    *,
    max_results: int | None = None,
    **kwargs: Any,
) -> list[Chat]

Retrieve chat data from Elasticsearch and convert it to a pandas DataFrame.

Parameters:

• query (Mapping[str, Any]) –The Elasticsearch query to be executed.
• index (str) –The name of the Elasticsearch index where the data will be retrieved.
• client (AsyncElasticsearch) –The Elasticsearch client instance.
• max_results (int | None, default: None ) –The maximum number of results to retrieve.
• kwargs (Any, default: {} ) –Additional keyword arguments to be passed to the Elasticsearch client.

Returns:

• list[Chat] –A pandas DataFrame containing the chat data.

flatten_chats

flatten_chats(
    chats: Chat | Sequence[Chat],
) -> list[dict[t.Any, t.Any]]

Flatten a list of chats into a individual messages with duplicated properties relevant to the chat.

Parameters:

• chats (Chat | Sequence[Chat]) –A Chat or list of Chat objects.

Returns:

• list[dict[Any, Any]] –A list of flat Message objects as dictionaries.

unflatten_chats

unflatten_chats(
    messages: Sequence[dict[Any, Any]],
) -> list[Chat]

Unflatten a list of messages into a list of Chat objects.

Parameters:

• messages (Sequence[dict[Any, Any]]) –A list of flat Message objects in the format from [rigging.data.flatten_chats][].

Returns:

• list[Chat] –A list of Chat objects. Parsing helpers for extracting rigging models from text

parse

parse(
    text: str, model_type: type[ModelT]
) -> tuple[ModelT, slice]

Parses a single model from text.

Parameters:

• text (str) –The content to parse.
• model_type (type[ModelT]) –The type of model to parse.

Returns:

• tuple[ModelT, slice] –The parsed model.

Raises:

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

parse_many

parse_many(
    text: str, *types: type[ModelT]
) -> list[tuple[ModelT, slice]]

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

Parameters:

• text (str) –The content to parse.
• *types (type[ModelT], default: () ) –The types of models to parse.

Returns:

• list[tuple[ModelT, slice]] –A list of parsed models.

Raises:

• MissingModelError –If any of the models are missing.

parse_set

parse_set(
    text: str,
    model_type: type[ModelT],
    *,
    minimum: int | None = None,
) -> list[tuple[ModelT, slice]]

Parses a set of models with the specified identical type from text.

Parameters:

• text (str) –The content to parse.
• model_type (type[ModelT]) –The type of models to parse.
• minimum (int | None, default: None ) –The minimum number of models required.

Returns:

• list[tuple[ModelT, slice]] –A list of parsed models.

Raises:

• MissingModelError –If the minimum number of models is not met.

try_parse

try_parse(
    text: str, model_type: type[ModelT]
) -> tuple[ModelT, slice] | None

Tries to parse a model from text.

Parameters:

• text (str) –The content to parse.
• model_type (type[ModelT]) –The type of model to search for.

Returns:

• tuple[ModelT, slice] | None –The first model that matches the given model type, or None if no match is found.

try_parse_many

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

Tries to parses multiple models of the specified non-identical types from text.

Parameters:

• text (str) –The content to parse.
• *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[tuple[ModelT, slice]] –A list of parsed models.

Raises:

• MissingModelError –If a model type is missing and fail_on_missing is True.
• Exception –If the model is malformed and fail_on_missing is True.

try_parse_set

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

Tries to parse a set of models with the specified identical type from text.

Parameters:

• text (str) –The content to parse.
• 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[tuple[ModelT, slice]] –The parsed models.

Raises:

• MissingModelError –If the number of parsed models is less than the minimum required. CacheMode

---

CacheMode = Literal['latest']

How to handle cache_control entries on messages.

• latest: Mark the final system message (if present) and the last 2 non-assistant, non-system messages with cache_control: ephemeral. This spends up to 3 of Anthropic’s 4 breakpoints — one pinning the tools+system prefix, two forming a rolling window over the most recent user/tool turns — which matches the rolling-window pattern recommended for multi-turn agents. We try to avoid creating custom exceptions unless they are necessary.

We use the built-in and pydantic exceptions as much as possible.

CompletionExhaustedMaxRoundsError

CompletionExhaustedMaxRoundsError(
    max_rounds: int, completion: str
)

Raised when the maximum number of rounds is exceeded while generating completions.

completion

completion = completion

The completion which was being generated when the exception occurred.

ExhaustedMaxRoundsError

ExhaustedMaxRoundsError(max_rounds: int)

Raised when the maximum number of rounds is exceeded while generating.

max_rounds

max_rounds = max_rounds

The number of rounds which was exceeded.

GeneratorWarning

Base class for all generator warnings.

This is used to indicate that something unexpected happened during the generator execution, but it is not critical enough to stop the execution.

InvalidGeneratorError

InvalidGeneratorError(model: str)

Raised when an invalid identifier is specified when getting a generator.

InvalidTokenizerError

InvalidTokenizerError(tokenizer: str)

Raised when an invalid tokenizer is specified.

tokenizer

tokenizer = tokenizer

The name of the tokenizer which was invalid.

MaxDepthError

MaxDepthError(max_steps: int)

Raise from a hook to stop the agent’s run due to reaching the maximum number of steps.

MessageWarning

Base class for all message warnings.

This is used to indicate that something unexpected happened during the message processing, but it is not critical enough to stop the execution.

MessagesExhaustedMaxRoundsError

MessagesExhaustedMaxRoundsError(
    max_rounds: int, messages: list[Message]
)

Raised when the maximum number of rounds is exceeded while generating messages.

messages

messages = messages

The messages which were being generated when the exception occurred.

MissingModelError

MissingModelError(content: str)

Raised when a model is missing when parsing a message.

ProcessingError

ProcessingError(content: str)

Raised when an error occurs during internal generator processing.

Stop

Stop(message: str)

Raise inside a pipeline to indicate a stopping condition.

Example

from dreanode.generators import pipeline

async def read_file(path: str) -> str:
    "Read the contents of a file."

    if no_more_files(path):
        raise Stop("There are no more files to read.")

    ...

chat = await pipeline.using(read_file).run()

message

message = message

The message associated with the stop.

TokenizerWarning

Base class for all tokenization warnings.

This is used to indicate that something unexpected happened during the tokenization process, but it is not critical enough to stop the execution.

ToolDefinitionError

ToolDefinitionError(message: str)

Raised when a tool cannot be properly defined.

ToolWarning

Base class for all tool warnings.

This is used to indicate that something unexpected happened during the tool execution, but it is not critical enough to stop the execution.

UnknownToolError

UnknownToolError(tool_name: str)

Raised when the an api tool call is made for an unknown tool.

tool_name

tool_name = tool_name

The name of the tool which was unknown.

raise_as

raise_as(
    error_type: type[Exception], message: str
) -> t.Callable[[t.Callable[P, R]], t.Callable[P, R]]

When the wrapped function raises an exception, raise ... from with the new error type.