Skip to content

mirascope.core.gemini.stream

The GeminiStream class for convenience around streaming LLM calls.

Usage Documentation

Streams

GeminiStream 

GeminiStream(
    *,
    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: BaseStream[GeminiCallResponse, GeminiCallResponseChunk, ContentDict, ContentDict, ContentDict, ContentsType, GeminiTool, Tool, GeminiDynamicConfig, GeminiCallParams, FinishReason]

A class for convenience around streaming Gemini LLM calls.

Example:

from mirascope.core import prompt_template
from mirascope.core.gemini import gemini_call


@gemini_call("gemini-1.5-flash", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str):
    ...


stream = recommend_book("fantasy")  # returns `GeminiStream` instance
for chunk, _ in stream:
    print(chunk.content, end="", flush=True)
Source code in mirascope/core/base/stream.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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 property 

cost: float | None

Returns the cost of the call.

construct_call_response 

construct_call_response() -> GeminiCallResponse

Constructs the call response from a consumed GeminiStream.

Raises:

Type Description
ValueError

if the stream has not yet been consumed.
Source code in mirascope/core/gemini/stream.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def construct_call_response(self) -> GeminiCallResponse:
    """Constructs the call response from a consumed GeminiStream.

    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 = GenerateContentResponseType.from_response(
        GenerateContentResponse(
            candidates=[
                Candidate(
                    finish_reason=self.finish_reasons[0]
                    if self.finish_reasons
                    else Candidate.FinishReason.STOP,
                    content=Content(
                        role=self.message_param["role"],
                        parts=self.message_param["parts"],
                    ),
                )
            ]
        )
    )
    return GeminiCallResponse(
        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,
    )