Skip to content

mistral.calls

A module for prompting Mistral API.

MistralCall

Bases: BaseCall[MistralCallResponse, MistralCallResponseChunk, MistralTool]

A class for" prompting Mistral's chat API.

Example:

from mirascope.mistral import MistralCall

class BookRecommender(MistralCall):
    prompt_template = "Please recommend a {genre} book"

    genre: str

response = BookRecommender(genre="fantasy").call()
print(response.content)
#> There are many great books to read, it ultimately depends...
Source code in mirascope/mistral/calls.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 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
116
117
118
119
120
121
122
123
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
class MistralCall(BaseCall[MistralCallResponse, MistralCallResponseChunk, MistralTool]):
    """A class for" prompting Mistral's chat API.

    Example:

    ```python
    from mirascope.mistral import MistralCall

    class BookRecommender(MistralCall):
        prompt_template = "Please recommend a {genre} book"

        genre: str

    response = BookRecommender(genre="fantasy").call()
    print(response.content)
    #> There are many great books to read, it ultimately depends...
    ```
    """

    call_params: ClassVar[MistralCallParams] = MistralCallParams()

    def messages(self) -> list[Message]:
        """Returns the template as a formatted list of messages."""
        return self._parse_messages(
            [MessageRole.SYSTEM, MessageRole.USER, MessageRole.ASSISTANT]
        )

    @retry
    def call(
        self, retries: Union[int, Retrying] = 1, **kwargs: Any
    ) -> MistralCallResponse:
        """Makes a call to the model using this `MistralCall` instance.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Returns:
            A `MistralCallResponse` instance.

        Raises:
            MistralException: raises any Mistral errors, see:
                https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
        """
        kwargs, tool_types = self._setup(kwargs, MistralTool)
        client = MistralClient(
            api_key=self.api_key,
            endpoint=self.base_url if self.base_url else ENDPOINT,
        )
        chat = client.chat
        if self.call_params.weave:
            chat = self.call_params.weave(chat)  # pragma: no cover
        if self.call_params.logfire:
            chat = self.call_params.logfire(
                chat,
                "mistral",
                response_type=MistralCallResponse,
                tool_types=tool_types,
            )  # pragma: no cover
        start_time = datetime.datetime.now().timestamp() * 1000
        completion = chat(messages=self.messages(), **kwargs)
        return MistralCallResponse(
            response=completion,
            tool_types=tool_types,
            start_time=start_time,
            cost=mistral_api_calculate_cost(completion.usage, completion.model),
            end_time=datetime.datetime.now().timestamp() * 1000,
        )

    @retry
    async def call_async(
        self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
    ) -> MistralCallResponse:
        """Makes an asynchronous call to the model using this `MistralCall` instance.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Returns:
            A `MistralCallResponse` instance.

        Raises:
            MistralException: raises any Mistral errors, see:
                https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
        """
        kwargs, tool_types = self._setup(kwargs, MistralTool)
        client = MistralAsyncClient(
            api_key=self.api_key,
            endpoint=self.base_url if self.base_url else ENDPOINT,
        )
        chat = client.chat
        if self.call_params.weave:
            chat = self.call_params.weave(chat)  # pragma: no cover
        elif self.call_params.logfire_async:
            chat = self.call_params.logfire_async(
                chat,
                "mistral",
                response_type=MistralCallResponse,
                tool_types=tool_types,
            )  # pragma: no cover

        start_time = datetime.datetime.now().timestamp() * 1000
        completion = await chat(messages=self.messages(), **kwargs)
        return MistralCallResponse(
            response=completion,
            tool_types=tool_types,
            start_time=start_time,
            end_time=datetime.datetime.now().timestamp() * 1000,
            cost=mistral_api_calculate_cost(completion.usage, completion.model),
        )

    @retry
    def stream(
        self, retries: Union[int, Retrying] = 1, **kwargs: Any
    ) -> Generator[MistralCallResponseChunk, None, None]:
        """Streams the response for a call using this `MistralCall` instance.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Returns:
            A `MistralCallResponseChunk` for each chunk of the response.

        Raises:
            MistralException: raises any Mistral errors, see:
                https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
        """
        kwargs, tool_types = self._setup(kwargs, MistralTool)
        client = MistralClient(
            api_key=self.api_key,
            endpoint=self.base_url if self.base_url else ENDPOINT,
        )
        chat_stream = client.chat_stream
        if self.call_params.logfire:
            chat_stream = self.call_params.logfire(
                chat_stream, "mistral", response_chunk_type=MistralCallResponseChunk
            )  # pragma: no cover
        stream = chat_stream(messages=self.messages(), **kwargs)

        for chunk in stream:
            yield MistralCallResponseChunk(chunk=chunk, tool_types=tool_types)

    @retry
    async def stream_async(
        self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
    ) -> AsyncGenerator[MistralCallResponseChunk, None]:
        """Streams the response for an asynchronous call using this `MistralCall`.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Returns:
            A `MistralCallResponseChunk` for each chunk of the response.

        Raises:
            MistralException: raises any Mistral errors, see:
                https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
        """
        kwargs, tool_types = self._setup(kwargs, MistralTool)
        client = MistralAsyncClient(
            api_key=self.api_key,
            endpoint=self.base_url if self.base_url else ENDPOINT,
        )
        chat_stream = client.chat_stream
        if self.call_params.logfire_async:
            chat_stream = self.call_params.logfire_async(
                chat_stream, "mistral", response_chunk_type=MistralCallResponseChunk
            )  # pragma: no cover
        stream = chat_stream(messages=self.messages(), **kwargs)
        async for chunk in stream:
            yield MistralCallResponseChunk(chunk=chunk, tool_types=tool_types)

call(retries=1, **kwargs)

Makes a call to the model using this MistralCall instance.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Returns:

Type Description
MistralCallResponse

A MistralCallResponse instance.

Raises:

Type Description
MistralException

raises any Mistral errors, see: https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
Source code in mirascope/mistral/calls.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@retry
def call(
    self, retries: Union[int, Retrying] = 1, **kwargs: Any
) -> MistralCallResponse:
    """Makes a call to the model using this `MistralCall` instance.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Returns:
        A `MistralCallResponse` instance.

    Raises:
        MistralException: raises any Mistral errors, see:
            https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
    """
    kwargs, tool_types = self._setup(kwargs, MistralTool)
    client = MistralClient(
        api_key=self.api_key,
        endpoint=self.base_url if self.base_url else ENDPOINT,
    )
    chat = client.chat
    if self.call_params.weave:
        chat = self.call_params.weave(chat)  # pragma: no cover
    if self.call_params.logfire:
        chat = self.call_params.logfire(
            chat,
            "mistral",
            response_type=MistralCallResponse,
            tool_types=tool_types,
        )  # pragma: no cover
    start_time = datetime.datetime.now().timestamp() * 1000
    completion = chat(messages=self.messages(), **kwargs)
    return MistralCallResponse(
        response=completion,
        tool_types=tool_types,
        start_time=start_time,
        cost=mistral_api_calculate_cost(completion.usage, completion.model),
        end_time=datetime.datetime.now().timestamp() * 1000,
    )

call_async(retries=1, **kwargs) async

Makes an asynchronous call to the model using this MistralCall instance.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Returns:

Type Description
MistralCallResponse

A MistralCallResponse instance.

Raises:

Type Description
MistralException

raises any Mistral errors, see: https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
Source code in mirascope/mistral/calls.py 
 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
124
125
126
127
128
@retry
async def call_async(
    self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
) -> MistralCallResponse:
    """Makes an asynchronous call to the model using this `MistralCall` instance.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Returns:
        A `MistralCallResponse` instance.

    Raises:
        MistralException: raises any Mistral errors, see:
            https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
    """
    kwargs, tool_types = self._setup(kwargs, MistralTool)
    client = MistralAsyncClient(
        api_key=self.api_key,
        endpoint=self.base_url if self.base_url else ENDPOINT,
    )
    chat = client.chat
    if self.call_params.weave:
        chat = self.call_params.weave(chat)  # pragma: no cover
    elif self.call_params.logfire_async:
        chat = self.call_params.logfire_async(
            chat,
            "mistral",
            response_type=MistralCallResponse,
            tool_types=tool_types,
        )  # pragma: no cover

    start_time = datetime.datetime.now().timestamp() * 1000
    completion = await chat(messages=self.messages(), **kwargs)
    return MistralCallResponse(
        response=completion,
        tool_types=tool_types,
        start_time=start_time,
        end_time=datetime.datetime.now().timestamp() * 1000,
        cost=mistral_api_calculate_cost(completion.usage, completion.model),
    )

messages()

Returns the template as a formatted list of messages.

Source code in mirascope/mistral/calls.py 
39
40
41
42
43
def messages(self) -> list[Message]:
    """Returns the template as a formatted list of messages."""
    return self._parse_messages(
        [MessageRole.SYSTEM, MessageRole.USER, MessageRole.ASSISTANT]
    )

stream(retries=1, **kwargs)

Streams the response for a call using this MistralCall instance.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Returns:

Type Description
Generator[MistralCallResponseChunk, None, None]

A MistralCallResponseChunk for each chunk of the response.

Raises:

Type Description
MistralException

raises any Mistral errors, see: https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
Source code in mirascope/mistral/calls.py 
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
157
158
159
160
@retry
def stream(
    self, retries: Union[int, Retrying] = 1, **kwargs: Any
) -> Generator[MistralCallResponseChunk, None, None]:
    """Streams the response for a call using this `MistralCall` instance.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Returns:
        A `MistralCallResponseChunk` for each chunk of the response.

    Raises:
        MistralException: raises any Mistral errors, see:
            https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
    """
    kwargs, tool_types = self._setup(kwargs, MistralTool)
    client = MistralClient(
        api_key=self.api_key,
        endpoint=self.base_url if self.base_url else ENDPOINT,
    )
    chat_stream = client.chat_stream
    if self.call_params.logfire:
        chat_stream = self.call_params.logfire(
            chat_stream, "mistral", response_chunk_type=MistralCallResponseChunk
        )  # pragma: no cover
    stream = chat_stream(messages=self.messages(), **kwargs)

    for chunk in stream:
        yield MistralCallResponseChunk(chunk=chunk, tool_types=tool_types)

stream_async(retries=1, **kwargs) async

Streams the response for an asynchronous call using this MistralCall.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Returns:

Type Description
AsyncGenerator[MistralCallResponseChunk, None]

A MistralCallResponseChunk for each chunk of the response.

Raises:

Type Description
MistralException

raises any Mistral errors, see: https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
Source code in mirascope/mistral/calls.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@retry
async def stream_async(
    self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
) -> AsyncGenerator[MistralCallResponseChunk, None]:
    """Streams the response for an asynchronous call using this `MistralCall`.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Returns:
        A `MistralCallResponseChunk` for each chunk of the response.

    Raises:
        MistralException: raises any Mistral errors, see:
            https://github.com/mistralai/client-python/blob/main/src/mistralai/exceptions.py
    """
    kwargs, tool_types = self._setup(kwargs, MistralTool)
    client = MistralAsyncClient(
        api_key=self.api_key,
        endpoint=self.base_url if self.base_url else ENDPOINT,
    )
    chat_stream = client.chat_stream
    if self.call_params.logfire_async:
        chat_stream = self.call_params.logfire_async(
            chat_stream, "mistral", response_chunk_type=MistralCallResponseChunk
        )  # pragma: no cover
    stream = chat_stream(messages=self.messages(), **kwargs)
    async for chunk in stream:
        yield MistralCallResponseChunk(chunk=chunk, tool_types=tool_types)