Skip to content

mirascope.core.bedrock.stream

The BedrockStream class for convenience around streaming LLM calls.

Usage Documentation

Streams

BedrockStream 

BedrockStream(*args: Any, **kwargs: Any)

Bases: BaseStream[BedrockCallResponse, BedrockCallResponseChunk, UserMessageTypeDef, AssistantMessageTypeDef, ToolUseBlockMessageTypeDef, InternalBedrockMessageParam, BedrockTool, ToolTypeDef, AsyncBedrockDynamicConfig | BedrockDynamicConfig, BedrockCallParams, StopReasonType]

A class for convenience around streaming Bedrock LLM calls.

Example:

from mirascope.core import prompt_template
from mirascope.core.bedrock import bedrock_call


@bedrock_call("gpt-4o-mini", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str):
    ...


stream = recommend_book("fantasy")  # returns `BedrockStream` instance
for chunk, _ in stream:
    print(chunk.content, end="", flush=True)
Source code in mirascope/core/bedrock/stream.py 
83
84
85
86
87
88
89
90
91
92
def __init__(self, *args: Any, **kwargs: Any) -> None:  # noqa: ANN401
    super().__init__(*args, **kwargs)
    self._metadata: (
        ConverseStreamMetadataEventTypeDef
        | AsyncConverseStreamMetadataEventTypeDef
        | None
    ) = None
    self._response_metadata: (
        ResponseMetadataTypeDef | AsyncResponseMetadataTypeDef
    ) = _DEFAULT_RESPONSE_METADATA

cost property 

cost: float | None

Returns the cost of the call.

construct_call_response 

construct_call_response() -> BedrockCallResponse

Constructs the call response from a consumed BedrockStream.

Raises:

Type Description
ValueError

if the stream has not yet been consumed.
Source code in mirascope/core/bedrock/stream.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def construct_call_response(self) -> BedrockCallResponse:
    """Constructs the call response from a consumed BedrockStream.

    Raises:
        ValueError: if the stream has not yet been consumed.
    """
    if not hasattr(self, "message_param"):
        raise ValueError(
            "No stream response, check if the stream has been consumed."
        )
    response = ConverseResponseTypeDef(
        output={"message": cast(MessageOutputTypeDef, self.message_param)},
        stopReason=self.finish_reasons[0] if self.finish_reasons else "end_turn",
        usage=self.metadata.get("usage", {}),
        metrics=self.metadata.get("metrics", {}),
        additionalModelResponseFields={},
        trace=self.metadata.get("trace", {}),
        ResponseMetadata=self._response_metadata,
    )
    return BedrockCallResponse(
        metadata=self.metadata,
        response=response,
        tool_types=self.tool_types,
        prompt_template=self.prompt_template,
        fn_args=self.fn_args if self.fn_args else {},
        dynamic_config=self.dynamic_config,
        messages=self.messages,
        call_params=self.call_params,
        call_kwargs=self.call_kwargs,
        user_message_param=self.user_message_param,
        start_time=self.start_time,
        end_time=self.end_time,
    )