LM Studio ou Ollama : lequel choisir pour un LLM en local sur Mac ? Le comparatif mesuré

brew install ollama
brew install --cask lm-studio

# Ollama : contexte de 8 192 jetons pour tous les modèles chargés
OLLAMA_CONTEXT_LENGTH=8192 ollama serve &
ollama pull qwen3.5:9b            # format par défaut (NVFP4)
ollama pull qwen3.5:9b-q4_K_M     # variante GGUF

# LM Studio : la ligne de commande lms est livrée dans l'application
LMS="/Applications/LM Studio.app/Contents/Resources/app/.webpack/lms"
open -g -a "LM Studio"            # démarre le service local
"$LMS" get qwen/qwen3.5-9b --mlx -y
"$LMS" get "qwen/qwen3.5-9b@q4_k_m" --gguf -y
"$LMS" server start               # API compatible OpenAI sur le port 1234

"$LMS" import ~/.lmstudio/models/lmstudio-community/Qwen3.5-9B-GGUF/Qwen3.5-9B-Q4_K_M.gguf \
  --user-repo bench/qwen3.5-9b-gguf -L -y
"$LMS" load bench/qwen3.5-9b -c 8192 --identifier qwen-gguf -y

"""Compare deux serveurs locaux compatibles OpenAI (Ollama, LM Studio) sur le même modèle.

Mesures, réflexion désactivée (reasoning_effort = "none") :
  generation -> débit de génération (jetons/s) et temps avant le premier jeton
  lecture    -> vitesse de lecture d'un long prompt (~6 000 jetons)
  parallele  -> 4 requêtes simultanées : débit cumulé
  fonctions  -> sortie JSON contrainte par un schéma, appel d'outil

Usage : python bench.py <nom> <base_url> <modele>
  python bench.py ollama-nvfp4 http://localhost:11434/v1 qwen3.5:9b
  python bench.py lmstudio-mlx http://localhost:1234/v1 qwen-mlx
"""

import asyncio
import json
import statistics
import sys
import time
from pathlib import Path

from openai import APIStatusError, AsyncOpenAI

NOM, BASE_URL, MODELE = sys.argv[1:4]
client = AsyncOpenAI(base_url=BASE_URL, api_key="local", timeout=600)
SANS_REFLEXION = {"reasoning_effort": "none"}
REPETITIONS = 5

SUJETS = ["la virtualisation des serveurs", "la sauvegarde 3-2-1", "le chiffrement de bout en bout",
          "les conteneurs Docker", "la haute disponibilité d'une base de données"]
# Un long document technique synthétique, identique pour les deux serveurs.
DOCUMENT = "\n".join(
    f"Incident {i} : le service {['paiement', 'stock', 'facturation', 'messagerie'][i % 4]} a été "
    f"indisponible {i % 7 + 3} minutes à cause d'un certificat expiré sur le nœud {i % 5}; "
    f"redémarrage manuel, puis renouvellement automatique activé."
    for i in range(160)
)


async def appel(messages: list[dict], max_tokens: int, **extra) -> dict:
    debut = time.perf_counter()
    premier, texte, usage = None, [], None
    flux = await client.chat.completions.create(
        model=MODELE, messages=messages, max_tokens=max_tokens, temperature=0, stream=True,
        stream_options={"include_usage": True}, extra_body=SANS_REFLEXION, **extra)
    async for e in flux:
        if e.usage:
            usage = e.usage
        if e.choices and e.choices[0].delta.content:
            premier = premier or time.perf_counter() - debut
            texte.append(e.choices[0].delta.content)
    fin = time.perf_counter() - debut
    return {"premier": premier or fin, "total": fin, "entree": usage.prompt_tokens,
            "sortie": usage.completion_tokens, "texte": "".join(texte)}


async def generation() -> dict:
    mesures = []
    for i in range(REPETITIONS):
        r = await appel([{"role": "user", "content": f"Explique {SUJETS[i]} à un DSI en 350 mots."}], 600)
        mesures.append(r)
    return {
        "premier_jeton_s": round(statistics.median(m["premier"] for m in mesures), 2),
        "jetons_par_s": round(statistics.median(m["sortie"] / (m["total"] - m["premier"]) for m in mesures), 1),
    }


async def lecture() -> dict:
    mesures = []
    for i in range(3):
        # Une question différente à chaque fois, en tête, pour éviter le cache de prompt.
        r = await appel([{"role": "user", "content": f"Question {i} : combien d'incidents concernent le service "
                          f"{['paiement', 'stock', 'facturation'][i]} ?\n\n{DOCUMENT}"}], 40)
        mesures.append(r)
    return {
        "jetons_entree": mesures[0]["entree"],
        "lecture_jetons_par_s": round(statistics.median(m["entree"] / m["premier"] for m in mesures)),
        "premier_jeton_s": round(statistics.median(m["premier"] for m in mesures), 2),
    }


async def parallele(n: int = 4) -> dict:
    debut = time.perf_counter()
    rs = await asyncio.gather(*(appel([{"role": "user", "content": f"Explique {SUJETS[i % 5]} en 300 mots."}], 500)
                                for i in range(n)))
    mur = time.perf_counter() - debut
    return {"requetes": n, "debit_cumule_jetons_par_s": round(sum(r["sortie"] for r in rs) / mur, 1),
            "duree_s": round(mur, 1)}


async def fonctions() -> dict:
    schema = {"type": "object", "additionalProperties": False, "required": ["service", "duree_minutes"],
              "properties": {"service": {"type": "string"}, "duree_minutes": {"type": "integer"}}}
    try:
        r = await client.chat.completions.create(
            model=MODELE, temperature=0, extra_body=SANS_REFLEXION,
            messages=[{"role": "user", "content": "Le service paiement a été coupé 12 minutes ce matin. Extrais les faits."}],
            response_format={"type": "json_schema", "json_schema": {"name": "incident", "strict": True, "schema": schema}})
        json_ok = json.loads(r.choices[0].message.content) == {"service": "paiement", "duree_minutes": 12}
    except (TypeError, ValueError):
        json_ok = False
    except APIStatusError as e:  # serveur ou moteur qui ne prend pas en charge la sortie structurée
        json_ok = f"erreur {e.status_code} : {e.message[:60]}"
    outil = {"type": "function", "function": {
        "name": "ouvrir_ticket", "description": "Ouvre un ticket d'incident",
        "parameters": {"type": "object", "required": ["service", "priorite"],
                       "properties": {"service": {"type": "string"}, "priorite": {"type": "string", "enum": ["P1", "P2", "P3"]}}}}}
    try:
        r = await client.chat.completions.create(
            model=MODELE, temperature=0, extra_body=SANS_REFLEXION, tools=[outil],
            messages=[{"role": "user", "content": "Le service paiement est complètement en panne, ouvre un ticket P1."}])
        appels = r.choices[0].message.tool_calls or []
        outil_ok = bool(appels) and appels[0].function.name == "ouvrir_ticket" and \
            json.loads(appels[0].function.arguments) == {"service": "paiement", "priorite": "P1"}
    except APIStatusError as e:
        outil_ok = f"erreur {e.status_code} : {e.message[:60]}"
    return {"json_schema": json_ok, "appel_outil": outil_ok}


async def main() -> None:
    await appel([{"role": "user", "content": "Bonjour"}], 5)  # échauffement : modèle chargé, noyaux compilés
    resultats = {"nom": NOM, "modele": MODELE, "generation": await generation(), "lecture": await lecture(),
                 "parallele": await parallele(), "fonctions": await fonctions()}
    Path("resultats").mkdir(exist_ok=True)
    Path(f"resultats/{NOM}.json").write_text(json.dumps(resultats, ensure_ascii=False, indent=2))
    print(json.dumps(resultats, ensure_ascii=False))


if __name__ == "__main__":
    asyncio.run(main())

"""Charge tour à tour chaque configuration, mesure temps de chargement et mémoire, lance bench.py.

Les deux serveurs tournent déjà : `ollama serve` (port 11434) et `lms server start` (port 1234),
avec un contexte de 8 192 jetons des deux côtés (OLLAMA_CONTEXT_LENGTH=8192, lms load -c 8192).
"""

import json
import subprocess
import sys
import time
import urllib.request

import psutil

LMS = "/Applications/LM Studio.app/Contents/Resources/app/.webpack/lms"
CONFIGS = [
    ("lmstudio-mlx", "lmstudio", "qwen/qwen3.5-9b", "qwen-mlx"),
    ("lmstudio-gguf", "lmstudio", "bench/qwen3.5-9b", "qwen-gguf"),
    ("ollama-nvfp4", "ollama", "qwen3.5:9b", "qwen3.5:9b"),
    ("ollama-gguf", "ollama", "qwen3.5:9b-q4_K_M", "qwen3.5:9b-q4_K_M"),
]


def tout_decharger() -> None:
    subprocess.run([LMS, "unload", "--all"], capture_output=True)
    for ligne in subprocess.run(["ollama", "ps"], capture_output=True, text=True).stdout.splitlines()[1:]:
        subprocess.run(["ollama", "stop", ligne.split()[0]], capture_output=True)
    time.sleep(5)


def charger(serveur: str, chemin: str, identifiant: str) -> float:
    """Durée de chargement mesurée de l'extérieur, de la même façon pour les deux serveurs."""
    debut = time.perf_counter()
    if serveur == "lmstudio":
        sortie = subprocess.run([LMS, "load", chemin, "-c", "8192", "--identifier", identifiant, "-y"],
                                capture_output=True, text=True)
        if "loaded successfully" not in sortie.stdout + sortie.stderr:
            raise RuntimeError(sortie.stderr[-500:])
    else:
        requete = urllib.request.Request("http://localhost:11434/api/generate", method="POST",
                                         data=json.dumps({"model": chemin, "keep_alive": "30m"}).encode())
        urllib.request.urlopen(requete, timeout=600).read()
    return round(time.perf_counter() - debut, 1)


def main(selection: list[str]) -> None:
    for nom, serveur, chemin, identifiant in CONFIGS:
        if selection and nom not in selection:
            continue
        tout_decharger()
        avant = psutil.virtual_memory().used
        chargement = charger(serveur, chemin, identifiant)
        url = "http://localhost:1234/v1" if serveur == "lmstudio" else "http://localhost:11434/v1"
        bench = subprocess.run([sys.executable, "bench.py", nom, url, identifiant], capture_output=True, text=True)
        memoire = (psutil.virtual_memory().used - avant) / 2**30
        if bench.returncode:
            raise RuntimeError(bench.stderr[-800:])
        resultat = json.loads(bench.stdout.strip().splitlines()[-1])
        resultat.update(chargement_s=chargement, memoire_go=round(memoire, 1))
        open(f"resultats/{nom}.json", "w").write(json.dumps(resultat, ensure_ascii=False, indent=2))
        print(json.dumps(resultat, ensure_ascii=False))
    tout_decharger()


if __name__ == "__main__":
    main(sys.argv[1:])

"""Mémoire réelle du moteur d'inférence, après un prompt long (cache KV alloué).

Sur Mac, llama.cpp projette le fichier du modèle en mémoire (compté dans le RSS), alors que
MLX place les poids dans des allocations Metal (comptées dans l'empreinte physique, pas dans
le RSS). On retient donc le maximum des deux pour le processus du moteur.
"""

import re
import subprocess
import sys

import psutil
from openai import OpenAI

from comparer import CONFIGS, charger, tout_decharger

DOCUMENT = open("bench.py").read() * 2  # ~5 000 jetons, sous la limite de contexte de 8 192


def empreinte(pid: int) -> int:
    sortie = subprocess.run(["footprint", str(pid)], capture_output=True, text=True).stdout
    m = re.search(r"Footprint: ([\d.]+) (MB|GB)", sortie)
    return int(float(m.group(1)) * (2**20 if m.group(2) == "MB" else 2**30)) if m else 0


def moteur() -> tuple[str, float]:
    candidats = [p for p in psutil.process_iter(["pid", "name", "cmdline"])
                 if any(k in " ".join(p.info["cmdline"] or []) for k in ("llama-server", "ollama runner", ".lmstudio/.internal/utils/node"))]
    mesures = [(max(p.memory_info().rss, empreinte(p.pid)), p.info["name"]) for p in candidats]
    octets, nom = max(mesures)
    return nom, round(octets / 2**30, 1)


for nom, serveur, chemin, identifiant in CONFIGS:
    if len(sys.argv) > 1 and nom not in sys.argv[1:]:
        continue
    tout_decharger()
    charger(serveur, chemin, identifiant)
    url = "http://localhost:1234/v1" if serveur == "lmstudio" else "http://localhost:11434/v1"
    OpenAI(base_url=url, api_key="local").chat.completions.create(
        model=identifiant, max_tokens=20, extra_body={"reasoning_effort": "none"},
        messages=[{"role": "user", "content": "Résume ce code en une phrase.\n\n" + DOCUMENT}])
    processus, go = moteur()
    print(f"{nom:14} {go:5} Go  ({processus})")
tout_decharger()