Le client MistralAzure en Python

Coder avec MistralAzure

Maintenant que vous avez déployé un endpoint Azure, passons à la pratique. Cette leçon détaille l’utilisation du client Python MistralAzure, les patterns de code courants et les spécificités à connaître pour développer efficacement avec Mistral sur Azure.

Installation et configuration

Installer le SDK

pip install mistralai

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

Initialisation du client

import os
from mistralai import MistralAzure

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

Le constructeur accepte deux paramètres obligatoires :

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

Chat Completions

Appel simple

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)

Avec paramètres avancés

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
)

Les paramètres de génération fonctionnent exactement comme avec l’API Mistral directe :

• 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

Pour les réponses longues, le streaming permet d’afficher le texte au fur et à mesure :

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

Le streaming est recommandé pour les applications interactives — l’utilisateur voit la réponse se construire en temps réel.

Appels asynchrones

Pour les applications web (FastAPI, Django async), utilisez le client asynchrone :

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 fonctionne de manière identique à l’API directe :

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, gérez les erreurs réseau et de rate limiting :

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 inclut un objet usage qui détaille les tokens consommés :

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}")

Utilisez ces métriques pour estimer vos coûts et optimiser vos prompts.

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