MistralInterface

llm.MistralInterface — sync & async Mistral AI interface with native streaming (with a fallback).

Constructor

MistralInterface(
    model_name: str = config.mistral_model,  # "mistral-large-latest"
    api_key: Optional[str] = None,
    **kwargs,
)

Imports from mistralai import Mistral and creates a single Mistral(api_key=self.api_key) client (there is no separate sync/async client pair here — one client object handles both, via separate method calls like chat.complete vs chat.complete_async). Raises ImportError("mistralai library is required. Install it using: pip install mistralai") if missing.

Per the class docstring, supported models include: mistral-small-latest, mistral-medium-latest, mistral-large-latest, open-mistral-7b, open-mixtral-8x7b, open-mixtral-8x22b, codestral-latest.

Message building: _build_messages(prompt, messages)

@staticmethod
def _build_messages(prompt, messages):
    """Resolve prompt / messages into a messages list (mirrors OpenAI interface)."""
    if messages is not None:
        return messages
    if prompt is not None:
        return [{"role": "user", "content": prompt}]
    raise ValueError("Either 'prompt' or 'messages' must be provided.")

Identical pattern to OpenAI/Anthropic/Groq/AWS's helper of the same name — no role translation is needed here, since Mistral's chat API already uses the same flat OpenAI-style {"role": ..., "content": "..."} shape. messages wins if both are given.

generate(messages=None, prompt=None, max_tokens=2048, temperature=0.3, tools=None, tool_choice=None, **kwargs) -> Union[str, dict]

msgs = self._build_messages(prompt, messages)
if tools:
    kwargs["tools"] = to_mistral_tools(tools)
    if tool_choice is not None:
        kwargs["tool_choice"] = tool_choice
response = self._client.chat.complete(
    model=self.model_name,
    messages=msgs,
    max_tokens=max_tokens, temperature=temperature, **kwargs,
)
message = response.choices[0].message
calls = parse_openai_tool_calls(message)
if calls:
    return {"content": message.content, "tool_calls": tool_calls_to_dicts(calls), "stop_reason": response.choices[0].finish_reason}
return message.content

Parameters

Returns: If the model doesn't call a tool, response.choices[0].message.content (str) on success — same as without tools. If the model calls one or more tools, parse_openai_tool_calls (the same helper OpenAI/Groq use, since Mistral's message.tool_calls[i].id/.function.name/.function.arguments shape matches theirs) extracts the calls and this method returns {"content": ..., "tool_calls": [...], "stop_reason": response.choices[0].finish_reason} instead. On exception (including a missing prompt/messages, and including any exception raised while building or parsing tool calls): f"Error: {e}" (a string, not a dict like OpenAI/Anthropic/Groq/AWS) — also prints f"❌ Mistral error: {e}" to stdout before returning. Because _build_messages(prompt, messages) is called inside this same try/except, a missing prompt/messages follows the same path: it does not raise ValueError out to the caller here — it's caught and returned as "Error: Either 'prompt' or 'messages' must be provided.", unlike Gemini's synchronous generate (which does let that ValueError propagate). Note that only the error path returns a string-shaped error on this provider — the tool-call success dict has the same {"content", "tool_calls", "stop_reason"} shape as every other provider. See Tool Calling.

generate_async(messages=None, prompt=None, max_tokens=2048, temperature=0.3, tools=None, tool_choice=None, **kwargs) -> Union[str, dict]

Same contract, calling await self._client.chat.complete_async(...). On exception, returns f"Error: {e}" — silently, with no print() call (unlike the sync path). Same as generate, a missing prompt/messages is caught and returned as an error string, not raised. Same tools/tool_choice handling and tool-call return dict as generate.

astream(messages=None, prompt=None, max_tokens=2048, temperature=0.3, **kwargs) -> AsyncIterator[str]

msgs = self._build_messages(prompt, messages)
stream = await self._client.chat.stream_async(...)   # await first, then iterate — no `async with`
async for chunk in stream:
    delta = chunk.data.choices[0].delta.content
    if delta:
        yield delta

The source comments (in Arabic) note this specific pattern — await the stream call first and then iterate the result directly, rather than using async with — was arrived at after debugging; using async with here previously did not work as expected with this SDK version. astream does not accept tools/tool_choice as named parameters and never reconstructs tool-call deltas from the native stream.

On any exception during native streaming — including a missing prompt/messages, since _build_messages is called inside this same try — this method falls back to a full non-streaming generation (via generate_async, passed the original messages=/prompt= again) and yields it word-by-word:

except Exception as e:
    try:
        response = await self.generate_async(messages=messages, prompt=prompt, max_tokens=max_tokens, temperature=temperature, **kwargs)
        for word in response.split(" "):
            yield word + " "
    except Exception as e2:
        yield f"Error: {e2}"

Because generate_async itself never raises for a missing prompt/messages (see above — it returns an error string instead), this fallback path will successfully yield "Error:"-prefixed words rather than raising, even when astream() is called with neither argument.

So a caller consuming astream() on MistralInterface cannot assume tokens are arriving incrementally — if the native stream call fails for any reason mid-stream-setup, the method transparently switches to word-chunking a fully-generated response instead, with no signal to the caller other than the (unavoidable) change in delivery cadence.

Example

import asyncio
from zerda.llm import MistralInterface

async def main():
    llm = MistralInterface(api_key="...")
    async for token in llm.astream(prompt="اكتب قصيدة عن الذكاء الاصطناعي"):
        print(token, end="", flush=True)

asyncio.run(main())

acleanup()

async def acleanup(self):
    try:
        await self._client.close()
    except Exception:
        pass

Silently swallows any exception on close — unlike other providers, this one doesn't even log a warning on cleanup failure. Note: there is no synchronous cleanup() method defined on this class (unlike OpenAI/Anthropic/Groq/AWS) — only the async acleanup().

Related APIs

• BaseLLMInterface
• Configuration — mistral_model is one of the fields this class actually reads.
• Tool Calling — the unified tools/tool_choice contract; note Mistral's plain-string error path is a documented exception to the otherwise-uniform dict shapes.