pydantic_ai.models.mistral

Setup

For details on how to set up authentication with this model, see model configuration for Mistral.

NamedMistralModels module-attribute

NamedMistralModels = Literal[
    "mistral-large-latest",
    "mistral-small-latest",
    "codestral-latest",
    "mistral-moderation-latest",
]

Latest / most popular named Mistral models.

MistralModelName module-attribute

MistralModelName = Union[NamedMistralModels, str]

Possible Mistral model names.

Since Mistral supports a variety of date-stamped models, we explicitly list the most popular models but allow any name in the type hints. Since the Mistral docs for a full list.

MistralModel dataclass

Bases: Model

A model that uses Mistral.

Internally, this uses the Mistral Python client to interact with the API.

API Documentation

Source code in pydantic_ai_slim/pydantic_ai/models/mistral.py

@dataclass(init=False)
class MistralModel(Model):
    """A model that uses Mistral.

    Internally, this uses the [Mistral Python client](https://github.com/mistralai/client-python) to interact with the API.

    [API Documentation](https://docs.mistral.ai/)
    """

    model_name: MistralModelName
    client: Mistral = field(repr=False)

    def __init__(
        self,
        model_name: MistralModelName,
        *,
        api_key: str | Callable[[], str | None] | None = None,
        client: Mistral | None = None,
        http_client: AsyncHTTPClient | None = None,
    ):
        """Initialize a Mistral model.

        Args:
            model_name: The name of the model to use.
            api_key: The API key to use for authentication, if unset uses `MISTRAL_API_KEY` environment variable.
            client: An existing `Mistral` client to use, if provided, `api_key` and `http_client` must be `None`.
            http_client: An existing `httpx.AsyncClient` to use for making HTTP requests.
        """
        self.model_name = model_name

        if client is not None:
            assert http_client is None, 'Cannot provide both `mistral_client` and `http_client`'
            assert api_key is None, 'Cannot provide both `mistral_client` and `api_key`'
            self.client = client
        else:
            api_key = os.getenv('MISTRAL_API_KEY') if api_key is None else api_key
            self.client = Mistral(api_key=api_key, async_client=http_client or cached_async_http_client())

    async def agent_model(
        self,
        *,
        function_tools: list[ToolDefinition],
        allow_text_result: bool,
        result_tools: list[ToolDefinition],
    ) -> AgentModel:
        """Create an agent model, this is called for each step of an agent run from Pydantic AI call."""
        return MistralAgentModel(
            self.client,
            self.model_name,
            allow_text_result,
            function_tools,
            result_tools,
        )

    def name(self) -> str:
        return f'mistral:{self.model_name}'

__init__

__init__(
    model_name: MistralModelName,
    *,
    api_key: str | Callable[[], str | None] | None = None,
    client: Mistral | None = None,
    http_client: AsyncClient | None = None
)

Initialize a Mistral model.

Parameters:

Name	Type	Description	Default
model_name	MistralModelName	The name of the model to use.	required
api_key	str | Callable[[], str | None] | None	The API key to use for authentication, if unset uses MISTRAL_API_KEY environment variable.	None
client	Mistral | None	An existing Mistral client to use, if provided, api_key and http_client must be None.	None
http_client	AsyncClient | None	An existing httpx.AsyncClient to use for making HTTP requests.	None

Source code in pydantic_ai_slim/pydantic_ai/models/mistral.py

def __init__(
    self,
    model_name: MistralModelName,
    *,
    api_key: str | Callable[[], str | None] | None = None,
    client: Mistral | None = None,
    http_client: AsyncHTTPClient | None = None,
):
    """Initialize a Mistral model.

    Args:
        model_name: The name of the model to use.
        api_key: The API key to use for authentication, if unset uses `MISTRAL_API_KEY` environment variable.
        client: An existing `Mistral` client to use, if provided, `api_key` and `http_client` must be `None`.
        http_client: An existing `httpx.AsyncClient` to use for making HTTP requests.
    """
    self.model_name = model_name

    if client is not None:
        assert http_client is None, 'Cannot provide both `mistral_client` and `http_client`'
        assert api_key is None, 'Cannot provide both `mistral_client` and `api_key`'
        self.client = client
    else:
        api_key = os.getenv('MISTRAL_API_KEY') if api_key is None else api_key
        self.client = Mistral(api_key=api_key, async_client=http_client or cached_async_http_client())

agent_model async

agent_model(
    *,
    function_tools: list[ToolDefinition],
    allow_text_result: bool,
    result_tools: list[ToolDefinition]
) -> AgentModel

Create an agent model, this is called for each step of an agent run from Pydantic AI call.

Source code in pydantic_ai_slim/pydantic_ai/models/mistral.py

async def agent_model(
    self,
    *,
    function_tools: list[ToolDefinition],
    allow_text_result: bool,
    result_tools: list[ToolDefinition],
) -> AgentModel:
    """Create an agent model, this is called for each step of an agent run from Pydantic AI call."""
    return MistralAgentModel(
        self.client,
        self.model_name,
        allow_text_result,
        function_tools,
        result_tools,
    )

MistralAgentModel dataclass

Bases: AgentModel

Implementation of AgentModel for Mistral models.

Source code in pydantic_ai_slim/pydantic_ai/models/mistral.py

@dataclass
class MistralAgentModel(AgentModel):
    """Implementation of `AgentModel` for Mistral models."""

    client: Mistral
    model_name: str
    allow_text_result: bool
    function_tools: list[ToolDefinition]
    result_tools: list[ToolDefinition]
    json_mode_schema_prompt: str = """Answer in JSON Object, respect this following format:\n```\n{schema}\n```\n"""

    async def request(
        self, messages: list[ModelMessage], model_settings: ModelSettings | None
    ) -> tuple[ModelResponse, Cost]:
        """Make a non-streaming request to the model from Pydantic AI call."""
        response = await self._completions_create(messages, model_settings)
        return self._process_response(response), _map_cost(response)

    @asynccontextmanager
    async def request_stream(
        self, messages: list[ModelMessage], model_settings: ModelSettings | None
    ) -> AsyncIterator[EitherStreamedResponse]:
        """Make a streaming request to the model from Pydantic AI call."""
        response = await self._stream_completions_create(messages, model_settings)
        is_function_tool = True if self.function_tools else False
        async with response:
            yield await self._process_streamed_response(is_function_tool, self.result_tools, response)

    async def _completions_create(
        self, messages: list[ModelMessage], model_settings: ModelSettings | None
    ) -> MistralChatCompletionResponse:
        """Make a non-streaming request to the model."""
        model_settings = model_settings or {}
        response = await self.client.chat.complete_async(
            model=str(self.model_name),
            messages=list(chain(*(self._map_message(m) for m in messages))),
            n=1,
            tools=self._map_function_and_result_tools_definition() or UNSET,
            tool_choice=self._get_tool_choice(),
            stream=False,
            max_tokens=model_settings.get('max_tokens', UNSET),
            temperature=model_settings.get('temperature', UNSET),
            top_p=model_settings.get('top_p', 1),
            timeout_ms=_get_timeout_ms(model_settings.get('timeout')),
        )
        assert response, 'A unexpected empty response from Mistral.'
        return response

    async def _stream_completions_create(
        self,
        messages: list[ModelMessage],
        model_settings: ModelSettings | None,
    ) -> MistralEventStreamAsync[MistralCompletionEvent]:
        """Create a streaming completion request to the Mistral model."""
        response: MistralEventStreamAsync[MistralCompletionEvent] | None
        mistral_messages = list(chain(*(self._map_message(m) for m in messages)))

        model_settings = model_settings or {}

        if self.result_tools and self.function_tools or self.function_tools:
            # Function Calling Mode
            response = await self.client.chat.stream_async(
                model=str(self.model_name),
                messages=mistral_messages,
                n=1,
                tools=self._map_function_and_result_tools_definition() or UNSET,
                tool_choice=self._get_tool_choice(),
                temperature=model_settings.get('temperature', UNSET),
                top_p=model_settings.get('top_p', 1),
                max_tokens=model_settings.get('max_tokens', UNSET),
                timeout_ms=_get_timeout_ms(model_settings.get('timeout')),
            )

        elif self.result_tools:
            # Json Mode
            schema: dict[str, Any] | list[dict[str, Any]]
            if len(self.result_tools) == 1:
                schema = _generate_simple_json_schema(self.result_tools[0].parameters_json_schema)
            else:
                parameters_json_schemas = [tool.parameters_json_schema for tool in self.result_tools]
                schema = _generate_simple_json_schemas(parameters_json_schemas)

            mistral_messages.append(MistralUserMessage(content=self.json_mode_schema_prompt.format(schema=schema)))
            response = await self.client.chat.stream_async(
                model=str(self.model_name),
                messages=mistral_messages,
                response_format={'type': 'json_object'},
                stream=True,
            )

        else:
            # Stream Mode
            response = await self.client.chat.stream_async(
                model=str(self.model_name),
                messages=mistral_messages,
                stream=True,
            )
        assert response, 'A unexpected empty response from Mistral.'
        return response

    def _get_tool_choice(self) -> MistralToolChoiceEnum | None:
        """Get tool choice for the model.

        - "auto": Default mode. Model decides if it uses the tool or not.
        - "any": Select any tool.
        - "none": Prevents tool use.
        - "required": Forces tool use.
        """
        if not self.function_tools and not self.result_tools:
            return None
        elif not self.allow_text_result:
            return 'required'
        else:
            return 'auto'

    def _map_function_and_result_tools_definition(self) -> list[MistralTool] | None:
        """Map function and result tools to MistralTool format.

        Returns None if both function_tools and result_tools are empty.
        """
        all_tools: list[ToolDefinition] = self.function_tools + self.result_tools
        tools = [
            MistralTool(
                function=MistralFunction(name=r.name, parameters=r.parameters_json_schema, description=r.description)
            )
            for r in all_tools
        ]
        return tools if tools else None

    @staticmethod
    def _process_response(response: MistralChatCompletionResponse) -> ModelResponse:
        """Process a non-streamed response, and prepare a message to return."""
        if response.created:
            timestamp = datetime.fromtimestamp(response.created, tz=timezone.utc)
        else:
            timestamp = _now_utc()

        assert response.choices, 'Unexpected empty response choice.'
        choice = response.choices[0]

        parts: list[ModelResponsePart] = []
        if choice.message.content is not None:
            # Note: Check len to handle potential mismatch between function calls and responses from the API. (`msg: not the same number of function class and reponses`)
            if isinstance(choice.message.content, str) and len(choice.message.content) > 0:
                parts.append(TextPart(choice.message.content))
            elif isinstance(choice.message.content, list):
                for chunk in choice.message.content:
                    if isinstance(chunk, MistralTextChunk) and len(chunk.text) > 0:
                        parts.append(TextPart(chunk.text))
                    else:
                        raise Exception(
                            f'Implementation for ImageURLChunk and ReferenceChunk is not available for the moment: {type(chunk)}'
                        )

        if isinstance(choice.message.tool_calls, list):
            for c in choice.message.tool_calls:
                if isinstance(c.function.arguments, str):
                    parts.append(ToolCallPart.from_json(c.function.name, c.function.arguments, c.id))
                else:
                    parts.append(ToolCallPart.from_dict(c.function.name, c.function.arguments, c.id))

        return ModelResponse(parts, timestamp=timestamp)

    @staticmethod
    async def _process_streamed_response(
        is_function_tools: bool,
        result_tools: list[ToolDefinition],
        response: MistralEventStreamAsync[MistralCompletionEvent],
    ) -> EitherStreamedResponse:
        """Process a streamed response, and prepare a streaming response to return."""
        start_cost = Cost()

        # Iterate until we get either `tool_calls` or `content` from the first chunk.
        while True:
            try:
                event = await response.__anext__()
                chunk = event.data
            except StopAsyncIteration as e:
                raise UnexpectedModelBehavior('Streamed response ended without content or tool calls') from e

            start_cost += _map_cost(chunk)

            if chunk.created:
                timestamp = datetime.fromtimestamp(chunk.created, tz=timezone.utc)
            else:
                timestamp = _now_utc()

            if chunk.choices:
                delta = chunk.choices[0].delta
                content = _map_delta_content(delta.content)

                tool_calls: list[MistralToolCall] | None = None
                if delta.tool_calls:
                    tool_calls = delta.tool_calls

                if content and result_tools:
                    return MistralStreamStructuredResponse(
                        is_function_tools,
                        {},
                        {c.name: c for c in result_tools},
                        response,
                        content,
                        timestamp,
                        start_cost,
                    )

                elif content:
                    return MistralStreamTextResponse(content, response, timestamp, start_cost)

                elif tool_calls:
                    return MistralStreamStructuredResponse(
                        is_function_tools,
                        {c.id if c.id else 'null': c for c in tool_calls},
                        {c.name: c for c in result_tools},
                        response,
                        None,
                        timestamp,
                        start_cost,
                    )

    @classmethod
    def _map_message(cls, message: ModelMessage) -> Iterable[MistralMessages]:
        """Just maps a `pydantic_ai.Message` to a `Mistral Message`."""
        if isinstance(message, ModelRequest):
            yield from cls._map_user_message(message)
        elif isinstance(message, ModelResponse):
            content_chunks: list[MistralContentChunk] = []
            tool_calls: list[MistralToolCall] = []

            for part in message.parts:
                if isinstance(part, TextPart):
                    content_chunks.append(MistralTextChunk(text=part.content))
                elif isinstance(part, ToolCallPart):
                    tool_calls.append(_map_pydantic_to_mistral_tool_call(part))
                else:
                    assert_never(part)
            yield MistralAssistantMessage(content=content_chunks, tool_calls=tool_calls)
        else:
            assert_never(message)

    @classmethod
    def _map_user_message(cls, message: ModelRequest) -> Iterable[MistralMessages]:
        for part in message.parts:
            if isinstance(part, SystemPromptPart):
                yield MistralSystemMessage(content=part.content)
            elif isinstance(part, UserPromptPart):
                yield MistralUserMessage(content=part.content)
            elif isinstance(part, ToolReturnPart):
                yield MistralToolMessage(
                    tool_call_id=part.tool_call_id,
                    content=part.model_response_str(),
                )
            elif isinstance(part, RetryPromptPart):
                if part.tool_name is None:
                    yield MistralUserMessage(content=part.model_response())
                else:
                    yield MistralToolMessage(
                        tool_call_id=part.tool_call_id,
                        content=part.model_response(),
                    )
            else:
                assert_never(part)

request async

request(
    messages: list[ModelMessage],
    model_settings: ModelSettings | None,
) -> tuple[ModelResponse, Cost]

Make a non-streaming request to the model from Pydantic AI call.

Source code in pydantic_ai_slim/pydantic_ai/models/mistral.py

async def request(
    self, messages: list[ModelMessage], model_settings: ModelSettings | None
) -> tuple[ModelResponse, Cost]:
    """Make a non-streaming request to the model from Pydantic AI call."""
    response = await self._completions_create(messages, model_settings)
    return self._process_response(response), _map_cost(response)

request_stream async

request_stream(
    messages: list[ModelMessage],
    model_settings: ModelSettings | None,
) -> AsyncIterator[EitherStreamedResponse]

Make a streaming request to the model from Pydantic AI call.

Source code in pydantic_ai_slim/pydantic_ai/models/mistral.py

@asynccontextmanager
async def request_stream(
    self, messages: list[ModelMessage], model_settings: ModelSettings | None
) -> AsyncIterator[EitherStreamedResponse]:
    """Make a streaming request to the model from Pydantic AI call."""
    response = await self._stream_completions_create(messages, model_settings)
    is_function_tool = True if self.function_tools else False
    async with response:
        yield await self._process_streamed_response(is_function_tool, self.result_tools, response)

MistralStreamTextResponse dataclass

Bases: StreamTextResponse

Implementation of StreamTextResponse for Mistral models.

Source code in pydantic_ai_slim/pydantic_ai/models/mistral.py

@dataclass
class MistralStreamTextResponse(StreamTextResponse):
    """Implementation of `StreamTextResponse` for Mistral models."""

    _first: str | None
    _response: MistralEventStreamAsync[MistralCompletionEvent]
    _timestamp: datetime
    _cost: Cost
    _buffer: list[str] = field(default_factory=list, init=False)

    async def __anext__(self) -> None:
        if self._first is not None and len(self._first) > 0:
            self._buffer.append(self._first)
            self._first = None
            return None

        chunk = await self._response.__anext__()
        self._cost += _map_cost(chunk.data)

        try:
            choice = chunk.data.choices[0]
        except IndexError:
            raise StopAsyncIteration()

        content = choice.delta.content
        if choice.finish_reason is None:
            assert content is not None, f'Expected delta with content, invalid chunk: {chunk!r}'
        if isinstance(content, str):
            self._buffer.append(content)
        elif isinstance(content, list):
            for chunk in content:
                if isinstance(chunk, MistralTextChunk):
                    self._buffer.append(chunk.text)
                else:
                    raise Exception(
                        f'Implementation for ImageURLChunk and ReferenceChunk is not available for the moment: {type(chunk)}'
                    )

    def get(self, *, final: bool = False) -> Iterable[str]:
        yield from self._buffer
        self._buffer.clear()

    def cost(self) -> Cost:
        return self._cost

    def timestamp(self) -> datetime:
        return self._timestamp

MistralStreamStructuredResponse dataclass

Bases: StreamStructuredResponse

Implementation of StreamStructuredResponse for Mistral models.

Source code in pydantic_ai_slim/pydantic_ai/models/mistral.py

@dataclass
class MistralStreamStructuredResponse(StreamStructuredResponse):
    """Implementation of `StreamStructuredResponse` for Mistral models."""

    _is_function_tools: bool
    _function_tools: dict[str, MistralToolCall]
    _result_tools: dict[str, ToolDefinition]
    _response: MistralEventStreamAsync[MistralCompletionEvent]
    _delta_content: str | None
    _timestamp: datetime
    _cost: Cost

    async def __anext__(self) -> None:
        chunk = await self._response.__anext__()
        self._cost += _map_cost(chunk.data)

        try:
            choice = chunk.data.choices[0]

        except IndexError:
            raise StopAsyncIteration()

        if choice.finish_reason is not None:
            raise StopAsyncIteration()

        delta = choice.delta
        content = _map_delta_content(delta.content)

        if self._function_tools and self._result_tools or self._function_tools:
            for new in delta.tool_calls or []:
                if current := self._function_tools.get(new.id or 'null'):
                    current.function = new.function
                else:
                    self._function_tools[new.id or 'null'] = new
        elif self._result_tools and content:
            if not self._delta_content:
                self._delta_content = content
            else:
                self._delta_content += content

    def get(self, *, final: bool = False) -> ModelResponse:
        calls: list[ModelResponsePart] = []

        if self._function_tools and self._result_tools or self._function_tools:
            for tool_call in self._function_tools.values():
                tool = _map_mistral_to_pydantic_tool_call(tool_call)
                calls.append(tool)
        elif self._delta_content and self._result_tools:
            # NOTE: Params set for the most efficient and fastest way.
            output_json = repair_json(self._delta_content, return_objects=True, skip_json_loads=True)
            assert isinstance(
                output_json, dict
            ), f'Expected repair_json as type dict, invalid type: {type(output_json)}'
            if output_json:
                for result_tool in self._result_tools.values():
                    # NOTE: Additional verification to prevent JSON validation to crash in `result.py`
                    # Ensures required parameters in the JSON schema are respected, especially for stream-based return types.
                    # For example, `return_type=list[str]` expects a 'response' key with value type array of str.
                    # when `{"response":` then `repair_json` sets `{"response": ""}` (type not found default str)
                    # when `{"response": {` then `repair_json` sets `{"response": {}}` (type found)
                    # This ensures it's corrected to `{"response": {}}` and other required parameters and type.
                    if not _validate_required_json_shema(output_json, result_tool.parameters_json_schema):
                        continue

                    tool = ToolCallPart.from_dict(
                        tool_name=result_tool.name,
                        args_dict=output_json,
                    )
                    calls.append(tool)

        return ModelResponse(calls, timestamp=self._timestamp)

    def cost(self) -> Cost:
        return self._cost

    def timestamp(self) -> datetime:
        return self._timestamp