mirascope.core.base.stream

This module contains the base classes for streaming responses from LLMs.

BaseStream

BaseStream(
    *,
    stream: (
        Generator[
            tuple[
                _BaseCallResponseChunkT, _BaseToolT | None
            ],
            None,
            None,
        ]
        | AsyncGenerator[
            tuple[
                _BaseCallResponseChunkT, _BaseToolT | None
            ],
            None,
        ]
    ),
    metadata: Metadata,
    tool_types: list[type[_BaseToolT]] | None,
    call_response_type: type[_BaseCallResponseT],
    model: str,
    prompt_template: str | None,
    fn_args: dict[str, Any],
    dynamic_config: _BaseDynamicConfigT,
    messages: list[_MessageParamT],
    call_params: _BaseCallParamsT,
    call_kwargs: BaseCallKwargs[_ToolSchemaT]
)

Bases: Generic[_BaseCallResponseT, _BaseCallResponseChunkT, _UserMessageParamT, _AssistantMessageParamT, _ToolMessageParamT, _MessageParamT, _BaseToolT, _ToolSchemaT, _BaseDynamicConfigT, _BaseCallParamsT, _FinishReason], ABC

A base class for streaming responses from LLMs.

Source code in mirascope/core/base/stream.py

def __init__(
    self,
    *,
    stream: Generator[tuple[_BaseCallResponseChunkT, _BaseToolT | None], None, None]
    | AsyncGenerator[
        tuple[_BaseCallResponseChunkT, _BaseToolT | None],
        None,
    ],
    metadata: Metadata,
    tool_types: list[type[_BaseToolT]] | None,
    call_response_type: type[_BaseCallResponseT],
    model: str,
    prompt_template: str | None,
    fn_args: dict[str, Any],
    dynamic_config: _BaseDynamicConfigT,
    messages: list[_MessageParamT],
    call_params: _BaseCallParamsT,
    call_kwargs: BaseCallKwargs[_ToolSchemaT],
) -> None:
    """Initializes an instance of `BaseStream`."""
    self.content = ""
    self.stream = stream
    self.metadata = metadata
    self.tool_types = tool_types
    self.call_response_type = call_response_type
    self.model = model
    self.prompt_template = prompt_template
    self.fn_args = fn_args
    self.dynamic_config = dynamic_config
    self.messages = messages
    self.call_params = call_params
    self.call_kwargs = call_kwargs
    self.user_message_param = get_possible_user_message_param(messages)  # pyright: ignore [reportAttributeAccessIssue]

cost abstractmethod property

cost: float | None

Returns the cost of the stream.

tool_message_params

tool_message_params(
    tools_and_outputs: list[tuple[_BaseToolT, JsonableType]]
) -> list[_ToolMessageParamT]

Returns the tool message parameters for tool call results.

Parameters:

Name	Type	Description	Default
tools_and_outputs	list[tuple[_BaseToolT, JsonableType]]	The list of tools and their outputs from which the tool message parameters should be constructed.	required

Source code in mirascope/core/base/stream.py

def tool_message_params(
    self, tools_and_outputs: list[tuple[_BaseToolT, JsonableType]]
) -> list[_ToolMessageParamT]:
    """Returns the tool message parameters for tool call results.

    Args:
        tools_and_outputs: The list of tools and their outputs from which the tool
            message parameters should be constructed.
    """
    return self.call_response_type.tool_message_params(tools_and_outputs)

construct_call_response abstractmethod

construct_call_response() -> _BaseCallResponseT

Constructs the call response.

Source code in mirascope/core/base/stream.py

@abstractmethod
def construct_call_response(self) -> _BaseCallResponseT:
    """Constructs the call response."""
    ...