Skip to content

Incomplete inputs and outputs (Streaming)

You might be looking for...

This section contains information on the data structures for streaming (also known as "incompletes"). We have designed these data structures to be able to add them to each other seamlessly with the + operator.

Please see below on how to use them.

Visual explanation

This visualization should help you understand the class hierarchy of the messages:

Message class hierarchy

You can see in this image that the classes with a "I" in front of their name have incomplete counterparts, which you can find below.

How it works

If you want to implement streaming responses, the most important complication is that you need to handle incomplete messages, and you need to determine how to add them to each other.

The following classes can help:

  1. They can be instantiated without any arguments, and have generally much looser requirements than their 'complete' counterparts.
  2. They can be added to each other with the + operator. The logic of how the addition is performed is custom to each class. They can also be added to a None object, which is useful for implementing streaming responses.
  3. They have a complete method that converts them to their 'complete' counterparts.

In practice, it looks like this:

from conatus.models.inputs_outputs.content_parts import (
    IncompleteAssistantAIMessageContentTextPart as IncompleteTextPart,
)

text_part = IncompleteTextPart(text="Hello")
text_part2 = IncompleteTextPart(text=" world")

assert (text_part + text_part2) == IncompleteTextPart(text="Hello world")
assert (text_part2 + text_part) == IncompleteTextPart(text=" worldHello")
assert (text_part + IncompleteTextPart()) == text_part

You can also make sure that the addition is "smart", in that it will merge content parts that are continuation, but not if they are distinct content parts.

from conatus.models.inputs_outputs.messages import IncompleteAssistantAIMessage
from conatus.models.inputs_outputs.content_parts import (
    IncompleteAssistantAIMessageContentToolCallPart as IncompleteToolCallPart,
)
from conatus.models.inputs_outputs.tool_calls import IncompleteAIToolCall

# We have two stream events...
tool_call_chunk_0a = IncompleteToolCallPart(
    uid="call_0", tool_call=IncompleteAIToolCall(returned_arguments='{"a": ')
)
tool_call_chunk_0b = IncompleteToolCallPart(
    uid=None, tool_call=IncompleteAIToolCall(returned_arguments='1}')
)

# That are functionally equivalent to this complete tool call:
tool_call_chunk_0_complete = IncompleteToolCallPart(
    uid="call_0", tool_call=IncompleteAIToolCall(returned_arguments='{"a": 1}')
)

# ...and this one, which is different:
tool_call_chunk_1 = IncompleteToolCallPart(
    uid="call_1", tool_call=IncompleteAIToolCall(returned_arguments='{"b": 2}')
)
chunk0a = IncompleteAssistantAIMessage(content=[tool_call_chunk_0a])
chunk0b = IncompleteAssistantAIMessage(content=[tool_call_chunk_0b])
chunk1 = IncompleteAssistantAIMessage(content=[tool_call_chunk_1])

# We can add the two chunks together to get the complete tool call:
assert (chunk0a + chunk0b) == IncompleteAssistantAIMessage(
    content=[tool_call_chunk_0_complete]
)

# But if we add a different chunk, it will not be added:
assert (chunk0a + chunk0b + chunk1) == IncompleteAssistantAIMessage(
    content=[tool_call_chunk_0_complete, tool_call_chunk_1]
)

Base class for all incomplete data structures

conatus.models.inputs_outputs.common.T module-attribute

T = TypeVar('T')

Type variable for the Addable class.

In this case, T is the type of the complete data structure that can be obtained from the incomplete data structure.

conatus.models.inputs_outputs.common.Addable

Bases: ABC

Addable class.

__add__ abstractmethod

__add__(other: Self | None) -> Self

Add an incomplete tool call to an incomplete tool call.

PARAMETER DESCRIPTION
other

The other incomplete tool call.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete tool calls.

Source code in conatus/models/inputs_outputs/common.py
@abstractmethod
def __add__(self, other: Self | None) -> Self:
    """Add an incomplete tool call to an incomplete tool call.

    Args:
        other: The other incomplete tool call.

    Returns:
        The sum of the two incomplete tool calls.
    """

__radd__

__radd__(other: Self | None) -> Self

Add an incomplete tool call to an incomplete tool call.

PARAMETER DESCRIPTION
other

The other incomplete tool call.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete tool calls.

Source code in conatus/models/inputs_outputs/common.py
def __radd__(self, other: Self | None) -> Self:
    """Add an incomplete tool call to an incomplete tool call.

    Args:
        other: The other incomplete tool call.

    Returns:
        The sum of the two incomplete tool calls.
    """
    return self.__add__(other)

conatus.models.inputs_outputs.common.Incomplete

Bases: Addable, ABC, Generic[T]

Incomplete class.

In addition to the methods defined in Addable , this class defines a complete method, which converts the incomplete data structure to the complete data structure.

complete abstractmethod

complete() -> T

Complete the incomplete data structure.

RETURNS DESCRIPTION
T

The complete data structure.

Source code in conatus/models/inputs_outputs/common.py
@abstractmethod
def complete(self) -> T:
    """Complete the incomplete data structure.

    Returns:
        The complete data structure.
    """

Incomplete data structures

conatus.models.inputs_outputs.response.IncompleteAIResponse dataclass

IncompleteAIResponse(
    prompt: AIPrompt[OutputSchemaType] = AIPrompt[
        OutputSchemaType
    ](),
    message_received: IncompleteAssistantAIMessage = IncompleteAssistantAIMessage(),
    finish_reason: FinishReasons | None = None,
    usage: CompletionUsage | None = None,
    uid: str | None = None,
)

Bases: Incomplete[AIResponse[OutputSchemaType]]

An incomplete AI response.

This is a convenience class that allows you to build an AI response incrementally.

For its complete counterpart, see AIResponse .

ATTRIBUTE DESCRIPTION
prompt

The prompt that was used to create the response.

TYPE: AIPrompt[OutputSchemaType]

message_received

The message received from the AI.

TYPE: IncompleteAssistantAIMessage

finish_reason

The reason the model stopped generating tokens.

TYPE: FinishReasons | None

usage

The usage statistics of the AI response.

TYPE: CompletionUsage | None

uid

The unique identifier of the AI response.

TYPE: str | None

messages_sent property

messages_sent: list[AIMessage]

The messages sent to the AI.

This is the same as the AIPrompt.messages field in the AIPrompt that was used to create the response. This is purely a convenience property.

__add__

__add__(other: Self | None) -> Self

Add an incomplete AI response to another incomplete AI response.

PARAMETER DESCRIPTION
other

The other incomplete AI response.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete AI responses.

Source code in conatus/models/inputs_outputs/response.py
@override
def __add__(self, other: Self | None) -> Self:
    """Add an incomplete AI response to another incomplete AI response.

    Args:
        other: The other incomplete AI response.

    Returns:
        The sum of the two incomplete AI responses.
    """
    if other is None:
        return self
    return type(self)(
        prompt=self.prompt,
        message_received=self.message_received + other.message_received,
        finish_reason=self.finish_reason or other.finish_reason,
        usage=(
            self.usage
            if other.usage is None
            else other.usage
            if self.usage is None
            else self.usage + other.usage
        ),
        uid=self.uid or other.uid,
    )

complete

complete(
    *,
    output_schema_was_converted_to_item_object: bool = False
) -> AIResponse[OutputSchemaType]

Complete the incomplete AI response.

PARAMETER DESCRIPTION
output_schema_was_converted_to_item_object

Whether the output schema was converted to an item object.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
AIResponse[OutputSchemaType]

The complete AI response.

Source code in conatus/models/inputs_outputs/response.py
@override
def complete(
    self,
    *,
    output_schema_was_converted_to_item_object: bool = False,
) -> AIResponse[OutputSchemaType]:
    """Complete the incomplete AI response.

    Args:
        output_schema_was_converted_to_item_object: Whether the output
            schema was converted to an item object.

    Returns:
        The complete AI response.
    """
    return AIResponse[OutputSchemaType](
        prompt=self.prompt,
        message_received=self.message_received.complete(),
        finish_reason=self.finish_reason,
        usage=self.usage,
        uid=self.uid,
        output_schema_was_converted_to_item_object=output_schema_was_converted_to_item_object,
    )

__radd__

__radd__(other: Self | None) -> Self

Add an incomplete tool call to an incomplete tool call.

PARAMETER DESCRIPTION
other

The other incomplete tool call.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete tool calls.

Source code in conatus/models/inputs_outputs/common.py
def __radd__(self, other: Self | None) -> Self:
    """Add an incomplete tool call to an incomplete tool call.

    Args:
        other: The other incomplete tool call.

    Returns:
        The sum of the two incomplete tool calls.
    """
    return self.__add__(other)

conatus.models.inputs_outputs.messages.IncompleteAssistantAIMessage dataclass

IncompleteAssistantAIMessage(
    content: list[
        IncompleteAssistantAIMessageContentPart
        | AssistantAIMessageContentToolCallPart
    ] = list(),
    refusal: str | None = None,
    role: Literal["assistant"] = "assistant",
)

Bases: Incomplete[AssistantAIMessage]

Incomplete assistant message.

This is a helper class to help with the addition of assistant messages: - You can add them to other assistant messages - If you want to finish an incomplete assistant message, you can use the complete method.

ATTRIBUTE DESCRIPTION
content

The content of the incomplete assistant message.

TYPE: list[IncompleteAssistantAIMessageContentPart | AssistantAIMessageContentToolCallPart]

refusal

The refusal of the incomplete assistant message.

TYPE: str | None

role

The role of the incomplete assistant message. Always "assistant".

TYPE: Literal['assistant']

all_text property

all_text: str | None

Get all the text from the assistant message.

tool_call_content_parts property

Get all the tool calls from the assistant message.

tool_call_content_parts_local_execution property

tool_call_content_parts_local_execution: list[
    IncompleteAssistantAIMessageContentToolCallPart
]

Get all the tool calls requiring local execution.

complete

complete() -> AssistantAIMessage

Convert an incomplete assistant message to a complete one.

RETURNS DESCRIPTION
AssistantAIMessage

The complete assistant message.

Source code in conatus/models/inputs_outputs/messages.py
@override
def complete(self) -> AssistantAIMessage:
    """Convert an incomplete assistant message to a complete one.

    Returns:
        The complete assistant message.
    """
    return AssistantAIMessage(
        content=[part.complete() for part in self.content],
        refusal=self.refusal,
        role=self.role,
    )

__radd__

__radd__(other: Self | None) -> Self

Add an incomplete tool call to an incomplete tool call.

PARAMETER DESCRIPTION
other

The other incomplete tool call.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete tool calls.

Source code in conatus/models/inputs_outputs/common.py
def __radd__(self, other: Self | None) -> Self:
    """Add an incomplete tool call to an incomplete tool call.

    Args:
        other: The other incomplete tool call.

    Returns:
        The sum of the two incomplete tool calls.
    """
    return self.__add__(other)

conatus.models.inputs_outputs.content_parts.IncompleteAssistantAIMessageContentPart module-attribute

The content part of an assistant message when streaming.

This is the same as AssistantAIMessageContentPart , but with laxer requirements.

conatus.models.inputs_outputs.content_parts.IncompleteAssistantAIMessageContentTextPart dataclass

IncompleteAssistantAIMessageContentTextPart(
    text: str | None = None,
    uid: str | None = None,
    type: Literal["text"] = "text",
)

Bases: Incomplete[AssistantAIMessageContentTextPart]

The content part of an assistant message that contains text.

ATTRIBUTE DESCRIPTION
text

The text of the assistant message content part.

TYPE: str | None

uid

The ID of the assistant message content part.
This should be given by the AI provider.

TYPE: str | None

type

The type of the assistant message content part. Always "text".

TYPE: Literal['text']

complete

Convert an incomplete text part to a complete text part.

RETURNS DESCRIPTION
AssistantAIMessageContentTextPart

The complete text part.

Source code in conatus/models/inputs_outputs/content_parts.py
@override
def complete(self) -> AssistantAIMessageContentTextPart:
    """Convert an incomplete text part to a complete text part.

    Returns:
        The complete text part.
    """
    return AssistantAIMessageContentTextPart(
        text=self.text or "",
        uid=self.uid,
    )

__radd__

__radd__(other: Self | None) -> Self

Add an incomplete tool call to an incomplete tool call.

PARAMETER DESCRIPTION
other

The other incomplete tool call.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete tool calls.

Source code in conatus/models/inputs_outputs/common.py
def __radd__(self, other: Self | None) -> Self:
    """Add an incomplete tool call to an incomplete tool call.

    Args:
        other: The other incomplete tool call.

    Returns:
        The sum of the two incomplete tool calls.
    """
    return self.__add__(other)

conatus.models.inputs_outputs.content_parts.IncompleteAssistantAIMessageContentToolCallPart dataclass

IncompleteAssistantAIMessageContentToolCallPart(
    tool_call: IncompleteAIToolCall,
    uid: str | None = None,
    type: Literal["tool_call"] = "tool_call",
)

Bases: Incomplete[AssistantAIMessageContentToolCallPart]

The content part of an assistant message that contains a tool call.

ATTRIBUTE DESCRIPTION
tool_call

The tool call of the assistant message content part.

TYPE: IncompleteAIToolCall

uid

The ID of the assistant message content part.
This should be given by the AI provider.

TYPE: str | None

type

The type of the assistant message content part. Always "tool_call".

TYPE: Literal['tool_call']

complete

Convert an incomplete tool call to a complete tool call.

RETURNS DESCRIPTION
AssistantAIMessageContentToolCallPart

The complete tool call.

Source code in conatus/models/inputs_outputs/content_parts.py
@override
def complete(self) -> AssistantAIMessageContentToolCallPart:
    """Convert an incomplete tool call to a complete tool call.

    Returns:
        The complete tool call.
    """
    return AssistantAIMessageContentToolCallPart(
        tool_call=self.tool_call.complete(),
        uid=self.uid,
    )

__radd__

__radd__(other: Self | None) -> Self

Add an incomplete tool call to an incomplete tool call.

PARAMETER DESCRIPTION
other

The other incomplete tool call.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete tool calls.

Source code in conatus/models/inputs_outputs/common.py
def __radd__(self, other: Self | None) -> Self:
    """Add an incomplete tool call to an incomplete tool call.

    Args:
        other: The other incomplete tool call.

    Returns:
        The sum of the two incomplete tool calls.
    """
    return self.__add__(other)

conatus.models.inputs_outputs.content_parts.IncompleteAssistantAIMessageContentRefusalPart dataclass

IncompleteAssistantAIMessageContentRefusalPart(
    refusal: str | None,
    uid: str | None = None,
    type: Literal["refusal"] = "refusal",
)

Bases: Incomplete[AssistantAIMessageContentRefusalPart]

The content part of an assistant message that contains a refusal.

ATTRIBUTE DESCRIPTION
refusal

The refusal of the assistant message content part.

TYPE: str | None

uid

The ID of the assistant message content part.
This should be given by the AI provider.

TYPE: str | None

type

The type of the assistant message content part. Always "refusal".

TYPE: Literal['refusal']

complete

Convert an incomplete refusal to a complete refusal.

RETURNS DESCRIPTION
AssistantAIMessageContentRefusalPart

The complete refusal.

Source code in conatus/models/inputs_outputs/content_parts.py
@override
def complete(self) -> AssistantAIMessageContentRefusalPart:
    """Convert an incomplete refusal to a complete refusal.

    Returns:
        The complete refusal.
    """
    return AssistantAIMessageContentRefusalPart(
        refusal=self.refusal or "",
        uid=self.uid,
    )

__radd__

__radd__(other: Self | None) -> Self

Add an incomplete tool call to an incomplete tool call.

PARAMETER DESCRIPTION
other

The other incomplete tool call.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete tool calls.

Source code in conatus/models/inputs_outputs/common.py
def __radd__(self, other: Self | None) -> Self:
    """Add an incomplete tool call to an incomplete tool call.

    Args:
        other: The other incomplete tool call.

    Returns:
        The sum of the two incomplete tool calls.
    """
    return self.__add__(other)

conatus.models.inputs_outputs.content_parts.IncompleteAssistantAIMessageContentReasoningPart dataclass

IncompleteAssistantAIMessageContentReasoningPart(
    reasoning: str | None,
    uid: str | None = None,
    is_redacted: bool = False,
    model_name: str | None = None,
    index: int | None = None,
    type: Literal["reasoning"] = "reasoning",
)

Bases: Incomplete[AssistantAIMessageContentReasoningPart]

The content part of an assistant message that contains reasoning.

ATTRIBUTE DESCRIPTION
reasoning

The reasoning of the assistant message content part.

TYPE: str | None

uid

The ID of the assistant message content part.
This should be given by the AI provider.

TYPE: str | None

is_redacted

Whether the reasoning is redacted. Supported by Anthropic.

TYPE: bool

model_name

The name of the model that produced the reasoning. This is important when switching between models.

TYPE: str | None

index

The index of the assistant message content part.
This is useful when streaming. Supported by Anthropic and OpenAI.

TYPE: int | None

type

The type of the assistant message content part. Always "reasoning".

TYPE: Literal['reasoning']

complete

Convert an incomplete reasoning to a complete reasoning.

Source code in conatus/models/inputs_outputs/content_parts.py
@override
def complete(self) -> AssistantAIMessageContentReasoningPart:
    """Convert an incomplete reasoning to a complete reasoning."""
    return AssistantAIMessageContentReasoningPart(
        reasoning=self.reasoning or "",
        uid=self.uid,
        is_redacted=self.is_redacted,
        model_name=self.model_name,
        index=self.index,
    )

__radd__

__radd__(other: Self | None) -> Self

Add an incomplete tool call to an incomplete tool call.

PARAMETER DESCRIPTION
other

The other incomplete tool call.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete tool calls.

Source code in conatus/models/inputs_outputs/common.py
def __radd__(self, other: Self | None) -> Self:
    """Add an incomplete tool call to an incomplete tool call.

    Args:
        other: The other incomplete tool call.

    Returns:
        The sum of the two incomplete tool calls.
    """
    return self.__add__(other)

Incomplete tool calls

conatus.models.inputs_outputs.tool_calls.IncompleteAIToolCall dataclass

IncompleteAIToolCall(
    call_id: str | None = None,
    name: str | None = None,
    returned_arguments: str | None = None,
    type: Literal["function"] = "function",
    index: int | str | None = None,
    requires_local_execution: bool = True,
    could_be_structured_output: bool = False,
)

Bases: Incomplete[AIToolCall]

Incomplete tool call.

For more information on its complete counterpart, see AIToolCall.

call_id class-attribute instance-attribute

call_id: str | None = None

The ID of the tool call.

This is the ID of the tool call as given by the AI provider. Some providers require that tool responses reference the ID of the tool call.

name class-attribute instance-attribute

name: str | None = None

The name of the tool call.

returned_arguments class-attribute instance-attribute

returned_arguments: str | None = None

The arguments of the tool call, as returned by the AI.

type class-attribute instance-attribute

type: Literal['function'] = 'function'

The type of the tool call. Always "function".

index class-attribute instance-attribute

index: int | str | None = None

The index of the tool call.

We assume that in the case of additions, the index is the same between self and other. It's the responsibility of the caller to ensure that this is the case.

requires_local_execution class-attribute instance-attribute

requires_local_execution: bool = True

Whether the tool call requires local execution.

If False, the tool calls is meant to be executed on the server side.

could_be_structured_output class-attribute instance-attribute

could_be_structured_output: bool = False

Whether the tool could be a structured output in disguise.

Some AI providers (e.g. Anthropic) will return a structured output in the form of a tool call.

complete

complete() -> AIToolCall

Create an AIToolCall from an incomplete tool call.

RETURNS DESCRIPTION
AIToolCall

The complete tool call.

Source code in conatus/models/inputs_outputs/tool_calls.py
@override
def complete(self) -> AIToolCall:
    """Create an AIToolCall from an incomplete tool call.

    Returns:
        The complete tool call.
    """
    return AIToolCall(
        call_id=self.call_id,
        name=self.name or "",
        returned_arguments=self.returned_arguments or "",
        type=self.type,
        requires_local_execution=self.requires_local_execution,
        could_be_structured_output=self.could_be_structured_output,
    )

__radd__

__radd__(other: Self | None) -> Self

Add an incomplete tool call to an incomplete tool call.

PARAMETER DESCRIPTION
other

The other incomplete tool call.

TYPE: Self | None

RETURNS DESCRIPTION
Self

The sum of the two incomplete tool calls.

Source code in conatus/models/inputs_outputs/common.py
def __radd__(self, other: Self | None) -> Self:
    """Add an incomplete tool call to an incomplete tool call.

    Args:
        other: The other incomplete tool call.

    Returns:
        The sum of the two incomplete tool calls.
    """
    return self.__add__(other)