Le client MistralAzure en Python

Coder avec MistralAzure

Votre endpoint Azure est déployé et répond ; il s’agit maintenant d’écrire le code qui l’exploitera durablement. Cette leçon détaille le client Python MistralAzure : son initialisation, les patterns que vous emploierez quotidiennement — appel simple, streaming, asynchrone, function calling — et les quelques spécificités qui distinguent Azure de l’API directe.

Installation et configuration

Le client Azure est inclus dans le package mistralai ; aucune dépendance supplémentaire n’est nécessaire.

pip install mistralai

L’initialisation ne demande que deux paramètres, tous deux obligatoires :

import os
from mistralai import MistralAzure

client = MistralAzure(
    azure_endpoint=os.environ["AZUREAI_ENDPOINT"],
    azure_api_key=os.environ["AZUREAI_API_KEY"]
)

• azure_endpoint — l’URL complète de votre endpoint Azure
• azure_api_key — votre clé d’authentification

Chat Completions

À partir de là, vous retrouvez l’ergonomie du client standard. Un message système pour cadrer le rôle du modèle, un message utilisateur pour la question, et la réponse se lit dans choices[0].message.content.

response = client.chat.complete(
    model="azureai",
    messages=[
        {"role": "system", "content": "Vous êtes un assistant technique spécialisé en Python."},
        {"role": "user", "content": "Comment implémenter un décorateur de cache ?"}
    ]
)

print(response.choices[0].message.content)

Les paramètres de génération se comportent exactement comme avec l’API Mistral directe. Une réponse trop imprévisible ou trop verbeuse se corrige ici, pas dans le prompt.

response = client.chat.complete(
    model="azureai",
    messages=[
        {"role": "user", "content": "Génère un haïku sur le cloud computing."}
    ],
    temperature=0.7,
    max_tokens=150,
    top_p=0.95
)

• temperature — contrôle la créativité (0.0 = déterministe, 1.0 = créatif)
• max_tokens — limite la longueur de la réponse
• top_p — échantillonnage nucleus

Streaming

Sur une réponse de plusieurs centaines de mots, un appel classique laisse l’utilisateur devant un écran figé pendant plusieurs secondes. Le streaming supprime cette attente perçue en affichant le texte au fil de sa génération : c’est le mode par défaut dès qu’une interface humaine est en jeu.

stream = client.chat.stream(
    model="azureai",
    messages=[
        {"role": "user", "content": "Explique le fonctionnement d'un load balancer."}
    ]
)

for chunk in stream:
    content = chunk.data.choices[0].delta.content
    if content:
        print(content, end="", flush=True)
print()  # Nouvelle ligne à la fin

Notez le test if content : certains chunks arrivent vides et écriraient None à l’écran sans cette précaution.

Appels asynchrones

Dans un serveur web — FastAPI, Django async — un appel bloquant immobilise un worker pendant toute la durée de la génération, soit plusieurs secondes par requête. Le client asynchrone rend la main pendant l’attente et permet à un même processus de traiter des dizaines de requêtes concurrentes.

import asyncio
from mistralai import MistralAzure

client = MistralAzure(
    azure_endpoint=os.environ["AZUREAI_ENDPOINT"],
    azure_api_key=os.environ["AZUREAI_API_KEY"]
)

async def generate_response(prompt: str) -> str:
    response = await client.chat.complete_async(
        model="azureai",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# Exécution
result = asyncio.run(generate_response("Qu'est-ce que Kubernetes ?"))
print(result)

Function Calling

Le function calling se comporte à l’identique de l’API directe. Vous décrivez les outils disponibles, le modèle décide s’il en appelle un, et votre code exécute réellement la fonction — le modèle ne fait que formuler l’intention et les arguments.

import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Obtenir la météo d'une ville",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "Nom de la ville"}
                },
                "required": ["city"]
            }
        }
    }
]

response = client.chat.complete(
    model="azureai",
    messages=[{"role": "user", "content": "Quelle est la météo à Paris ?"}],
    tools=tools,
    tool_choice="auto"
)

# Vérifier si le modèle veut appeler une fonction
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    print(f"Fonction appelée : {tool_call.function.name}")
    print(f"Arguments : {args}")

Gestion des erreurs

En production, trois familles d’erreurs reviennent, et elles n’appellent pas la même réponse. Un 429 signale un rate limit et se traite par une nouvelle tentative après attente croissante. Un 401 traduit une clé invalide ou expirée : réessayer ne servira à rien, autant échouer immédiatement avec un message clair. Un 404 pointe presque toujours vers une variable AZUREAI_ENDPOINT mal renseignée. Distinguer ces cas vous évite de passer une heure sur une faute de frappe dans une URL.

import time
from mistralai import MistralAzure

def robust_call(client, messages, max_retries=3):
    """Appel avec retry exponentiel et gestion d'erreurs."""
    for attempt in range(max_retries):
        try:
            return client.chat.complete(
                model="azureai",
                messages=messages
            )
        except Exception as e:
            error_str = str(e)
            if "429" in error_str:
                wait = 2 ** attempt
                print(f"Rate limit — retry dans {wait}s")
                time.sleep(wait)
            elif "401" in error_str:
                raise ValueError("Clé API invalide ou expirée") from e
            elif "404" in error_str:
                raise ValueError("Endpoint non trouvé — vérifiez AZUREAI_ENDPOINT") from e
            else:
                if attempt == max_retries - 1:
                    raise
                time.sleep(1)
    raise RuntimeError("Échec après tous les retries")

Mesurer la consommation

Chaque réponse embarque un objet usage qui détaille les tokens consommés. Lisez-le systématiquement : c’est la seule mesure fiable pour estimer vos coûts et repérer un prompt système devenu obèse à force de retouches.

response = client.chat.complete(
    model="azureai",
    messages=[{"role": "user", "content": "Bonjour"}]
)

usage = response.usage
print(f"Tokens prompt  : {usage.prompt_tokens}")
print(f"Tokens réponse : {usage.completion_tokens}")
print(f"Tokens total   : {usage.total_tokens}")

Points clés à retenir

• Le client MistralAzure s’utilise comme le client standard, avec model="azureai"
• Le streaming (chat.stream) est recommandé pour les interfaces utilisateur
• Les appels asynchrones (complete_async) sont essentiels pour les serveurs web
• Le function calling et le JSON mode fonctionnent de manière identique à l’API directe
• Implémentez toujours un retry exponentiel pour gérer les erreurs 429