Skip to content

mirascope.core.cohere.stream

The CohereStream class for convenience around streaming LLM calls.

Usage Documentation

Streams

CohereStream 

CohereStream(
    *,
    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[CohereCallResponse, CohereCallResponseChunk, ChatMessage, ChatMessage, ChatMessage, ChatMessage, CohereTool, Tool, CohereDynamicConfig, CohereCallParams, ChatStreamEndEventFinishReason]

A class for convenience around streaming Cohere LLM calls.

Example:

from mirascope.core import prompt_template
from mirascope.core.cohere import cohere_call


@cohere_call("command-r-plus", stream=True)
@prompt_template("Recommend a {genre} book")
def recommend_book(genre: str):
    ...


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

Constructs the call response from a consumed CohereStream.

Raises:

Type Description
ValueError

if the stream has not yet been consumed.
Source code in mirascope/core/cohere/stream.py 
 77
 78
 79
 80
 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
def construct_call_response(self) -> CohereCallResponse:
    """Constructs the call response from a consumed CohereStream.

    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."
        )
    if not self.input_tokens and not self.output_tokens:
        meta = None
    else:
        meta = ApiMeta(
            billed_units=ApiMetaBilledUnits(
                input_tokens=self.input_tokens, output_tokens=self.output_tokens
            )
        )
    completion = NonStreamedChatResponse(
        generation_id=self.id,
        text=self.message_param.message,
        meta=meta,
        finish_reason=self.finish_reasons[0] if self.finish_reasons else None,
    )

    return CohereCallResponse(
        metadata=self.metadata,
        response=completion,
        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,
    )