Incomplete inputs and outputs (Streaming)¶
You might be looking for...
- If you're looking for information on the
AIPromptandAIResponseclasses, see AI inputs and outputs: an overview. - If you're looking for information on the
AIMessageclasses, see AI messages. - If you're looking for information on content parts, see Content parts.
- If you're looking for information on the tool call data structures, see Tool calls.
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:
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:
- They can be instantiated without any arguments, and have generally much looser requirements than their 'complete' counterparts.
- 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 aNoneobject, which is useful for implementing streaming responses. - They have a
completemethod 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 an incomplete tool call to an incomplete tool call.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete tool call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete tool calls. |
Source code in conatus/models/inputs_outputs/common.py
__radd__
¶
Add an incomplete tool call to an incomplete tool call.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete tool call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete tool calls. |
Source code in conatus/models/inputs_outputs/common.py
conatus.models.inputs_outputs.common.Incomplete
¶
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:
|
message_received |
The message received from the AI. |
finish_reason |
The reason the model stopped generating tokens.
TYPE:
|
usage |
The usage statistics of the AI response.
TYPE:
|
uid |
The unique identifier of the AI response.
TYPE:
|
messages_sent
property
¶
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 an incomplete AI response to another incomplete AI response.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete AI response.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete AI responses. |
Source code in conatus/models/inputs_outputs/response.py
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:
|
| RETURNS | DESCRIPTION |
|---|---|
AIResponse[OutputSchemaType]
|
The complete AI response. |
Source code in conatus/models/inputs_outputs/response.py
__radd__
¶
Add an incomplete tool call to an incomplete tool call.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete tool call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete tool calls. |
Source code in conatus/models/inputs_outputs/common.py
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:
|
refusal |
The refusal of the incomplete assistant message.
TYPE:
|
role |
The role of the incomplete assistant message. Always
TYPE:
|
tool_call_content_parts
property
¶
tool_call_content_parts: list[
IncompleteAssistantAIMessageContentToolCallPart
]
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
__radd__
¶
Add an incomplete tool call to an incomplete tool call.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete tool call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete tool calls. |
Source code in conatus/models/inputs_outputs/common.py
conatus.models.inputs_outputs.content_parts.IncompleteAssistantAIMessageContentPart
module-attribute
¶
IncompleteAssistantAIMessageContentPart = (
IncompleteAssistantAIMessageContentTextPart
| IncompleteAssistantAIMessageContentToolCallPart
| IncompleteAssistantAIMessageContentRefusalPart
| IncompleteAssistantAIMessageContentReasoningPart
)
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:
|
uid |
The ID of the assistant message content part.
TYPE:
|
type |
The type of the assistant message content part. Always
TYPE:
|
complete
¶
complete() -> AssistantAIMessageContentTextPart
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
__radd__
¶
Add an incomplete tool call to an incomplete tool call.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete tool call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete tool calls. |
Source code in conatus/models/inputs_outputs/common.py
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:
|
uid |
The ID of the assistant message content part.
TYPE:
|
type |
The type of the assistant message content part. Always
TYPE:
|
complete
¶
complete() -> AssistantAIMessageContentToolCallPart
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
__radd__
¶
Add an incomplete tool call to an incomplete tool call.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete tool call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete tool calls. |
Source code in conatus/models/inputs_outputs/common.py
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:
|
uid |
The ID of the assistant message content part.
TYPE:
|
type |
The type of the assistant message content part. Always
TYPE:
|
complete
¶
complete() -> AssistantAIMessageContentRefusalPart
Convert an incomplete refusal to a complete refusal.
| RETURNS | DESCRIPTION |
|---|---|
AssistantAIMessageContentRefusalPart
|
The complete refusal. |
Source code in conatus/models/inputs_outputs/content_parts.py
__radd__
¶
Add an incomplete tool call to an incomplete tool call.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete tool call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete tool calls. |
Source code in conatus/models/inputs_outputs/common.py
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:
|
uid |
The ID of the assistant message content part.
TYPE:
|
is_redacted |
Whether the reasoning is redacted. Supported by Anthropic.
TYPE:
|
model_name |
The name of the model that produced the reasoning. This is important when switching between models.
TYPE:
|
index |
The index of the assistant message content part.
TYPE:
|
type |
The type of the assistant message content part. Always
TYPE:
|
complete
¶
complete() -> AssistantAIMessageContentReasoningPart
Convert an incomplete reasoning to a complete reasoning.
Source code in conatus/models/inputs_outputs/content_parts.py
__radd__
¶
Add an incomplete tool call to an incomplete tool call.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete tool call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete tool calls. |
Source code in conatus/models/inputs_outputs/common.py
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.
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
¶
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
__radd__
¶
Add an incomplete tool call to an incomplete tool call.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
The other incomplete tool call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
The sum of the two incomplete tool calls. |
