Skip to content

mirascope.core.vertex.stream

The VertexStream class for convenience around streaming LLM calls.

Usage Documentation

Streams

VertexStream 

VertexStream(
    *,
    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[VertexCallResponse, VertexCallResponseChunk, Content, Content, Content, Content, VertexTool, Tool, VertexDynamicConfig, VertexCallParams, FinishReason]

A class for convenience around streaming Vertex LLM calls.

Example:

from mirascope.core import prompt_template
from mirascope.core.vertex import vertex_call


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


stream = recommend_book("fantasy")  # returns `VertexStream` 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() -> VertexCallResponse

Constructs the call response from a consumed VertexStream.

Raises:

Type Description
ValueError

if the stream has not yet been consumed.
Source code in mirascope/core/vertex/stream.py 
 81
 82
 83
 84
 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
def construct_call_response(self) -> VertexCallResponse:
    """Constructs the call response from a consumed VertexStream.

    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 = GenerationResponse.from_dict(
        {
            "candidates": [
                Candidate.from_dict(
                    {
                        "finish_reason": self.finish_reasons[0]
                        if self.finish_reasons
                        else FinishReason.STOP,
                        "content": Content(
                            role=self.message_param.role,
                            parts=self.message_param.parts,
                        ).to_dict(),
                    }
                ).to_dict()
            ]
        }
    )
    return VertexCallResponse(
        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,
    )