feat: PipBoy rename, /lore tab fix, run.py status check loop, crash test script
This commit is contained in:
Corback
@@ -0,0 +1,326 @@
|
||||
"""
|
||||
LLM Crash Test — Fallout: Venice of Wasteland
|
||||
Test de performance des deux LLM (MJ 14b + PNJ 7b) en conditions réelles.
|
||||
|
||||
Modes:
|
||||
--mode simultane : les deux LLM tournent en parallèle (threads)
|
||||
--mode decale : push progressif PNJ → MJ (pipeline producteur/consommateur)
|
||||
--mode solo-mj : seulement le 14b
|
||||
--mode solo-pnj : seulement le 7b
|
||||
|
||||
Usage:
|
||||
python llm_crash_test.py --mode simultane --rounds 5
|
||||
python llm_crash_test.py --mode decale --rounds 10 --pnj-count 3
|
||||
"""
|
||||
|
||||
import os, sys, time, json, argparse, threading, queue
|
||||
import urllib.request
|
||||
from datetime import datetime
|
||||
|
||||
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")
|
||||
MODEL_MJ = os.getenv("MODEL_MJ", "qwen2.5:14b")
|
||||
MODEL_PNJ = os.getenv("MODEL_PNJ", "qwen2.5:7b")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prompts de test réalistes (contexte Fallout Venice)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
PNJ_PROMPTS = [
|
||||
"Tu es Remy Tureaud, marchand créole de Pearl River. Réponds en 2 phrases à : 'T'as des piles ?'",
|
||||
"Tu es Sœur Eulalie, soigneuse de L'Union. Réponds en 2 phrases à : 'J'ai une blessure par balle.'",
|
||||
"Tu es Jacques-Henri, garde de la Régie. Réponds en 2 phrases à : 'Laisse-moi passer le checkpoint.'",
|
||||
"Tu es Mama Voodoo, goule du Grand Krewe. Réponds en 2 phrases à : 'C'est quoi ce tatouage ?'",
|
||||
"Tu es un Écumeur anonyme. Réponds en 2 phrases à : 'On veut juste passer.'",
|
||||
]
|
||||
|
||||
MJ_PROMPTS = [
|
||||
"Résume en 3 phrases la situation politique entre L'Union et la CdA à New Orleans post-apo.",
|
||||
"Décris en 3 phrases une rencontre de nuit dans les bayous de Pearl River.",
|
||||
"En 3 phrases, quelles sont les conséquences d'un blocus sur la route de Baton Rouge ?",
|
||||
"Décris en 3 phrases l'ambiance d'un marché noir dans le Vieux Carré de NOLA.",
|
||||
"En 3 phrases, comment réagit le Grand Krewe si des étrangers fouillent leurs ruines ?",
|
||||
]
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# LLM call
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def call_llm(model: str, prompt: str, timeout: int = 60) -> dict:
|
||||
"""Appel Ollama, retourne {tokens, elapsed, error, text}."""
|
||||
t0 = time.time()
|
||||
body = json.dumps({
|
||||
"model": model,
|
||||
"prompt": prompt,
|
||||
"stream": False,
|
||||
"options": {"num_predict": 150, "temperature": 0.7},
|
||||
}).encode()
|
||||
req = urllib.request.Request(
|
||||
f"{OLLAMA_URL}/api/generate", data=body, method="POST",
|
||||
headers={"Content-Type": "application/json"}
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
data = json.loads(r.read())
|
||||
elapsed = time.time() - t0
|
||||
tokens = data.get("eval_count", 0)
|
||||
return {
|
||||
"model": model,
|
||||
"tokens": tokens,
|
||||
"elapsed": round(elapsed, 2),
|
||||
"tok_s": round(tokens / elapsed, 1) if elapsed > 0 else 0,
|
||||
"error": None,
|
||||
"text": data.get("response", "")[:120],
|
||||
}
|
||||
except Exception as e:
|
||||
return {
|
||||
"model": model,
|
||||
"tokens": 0,
|
||||
"elapsed": round(time.time() - t0, 2),
|
||||
"tok_s": 0,
|
||||
"error": str(e)[:80],
|
||||
"text": "",
|
||||
}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Modes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def run_solo(model: str, prompts: list[str], rounds: int) -> list[dict]:
|
||||
results = []
|
||||
for i in range(rounds):
|
||||
prompt = prompts[i % len(prompts)]
|
||||
print(f" [{model}] Round {i+1}/{rounds}...", end="", flush=True)
|
||||
r = call_llm(model, prompt)
|
||||
print(f" {r['tok_s']} tok/s | {r['elapsed']}s | {'ERR: '+r['error'] if r['error'] else 'OK'}")
|
||||
results.append(r)
|
||||
return results
|
||||
|
||||
|
||||
def run_simultane(rounds: int, pnj_count: int = 2) -> dict:
|
||||
"""Les deux LLM tournent en parallèle via threads."""
|
||||
print(f"\n[SIMULTANE] {MODEL_MJ} + {MODEL_PNJ} x{pnj_count} PNJ — {rounds} rounds\n")
|
||||
all_results = {"mj": [], "pnj": []}
|
||||
lock = threading.Lock()
|
||||
|
||||
def mj_worker(round_idx: int):
|
||||
prompt = MJ_PROMPTS[round_idx % len(MJ_PROMPTS)]
|
||||
r = call_llm(MODEL_MJ, prompt)
|
||||
with lock:
|
||||
all_results["mj"].append(r)
|
||||
status = f"ERR: {r['error']}" if r['error'] else f"{r['tok_s']} tok/s"
|
||||
print(f" [MJ 14b] R{round_idx+1} → {status} ({r['elapsed']}s)")
|
||||
|
||||
def pnj_worker(round_idx: int, pnj_idx: int):
|
||||
prompt = PNJ_PROMPTS[(round_idx * pnj_count + pnj_idx) % len(PNJ_PROMPTS)]
|
||||
r = call_llm(MODEL_PNJ, prompt)
|
||||
with lock:
|
||||
all_results["pnj"].append(r)
|
||||
status = f"ERR: {r['error']}" if r['error'] else f"{r['tok_s']} tok/s"
|
||||
print(f" [PNJ 7b] R{round_idx+1} PNJ{pnj_idx+1} → {status} ({r['elapsed']}s)")
|
||||
|
||||
t_start = time.time()
|
||||
for i in range(rounds):
|
||||
threads = []
|
||||
t = threading.Thread(target=mj_worker, args=(i,))
|
||||
threads.append(t)
|
||||
for j in range(pnj_count):
|
||||
t = threading.Thread(target=pnj_worker, args=(i, j))
|
||||
threads.append(t)
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
print()
|
||||
|
||||
total = time.time() - t_start
|
||||
return {**all_results, "total_elapsed": round(total, 1)}
|
||||
|
||||
|
||||
def run_decale(rounds: int, pnj_count: int = 3) -> dict:
|
||||
"""Pipeline producteur/consommateur : les PNJ accumulent des infos → poussé vers le MJ."""
|
||||
print(f"\n[DÉCALÉ] {pnj_count} PNJ 7b → buffer → MJ 14b — {rounds} rounds\n")
|
||||
pnj_queue: queue.Queue = queue.Queue()
|
||||
all_results = {"mj": [], "pnj": [], "latency_pnj_to_mj": []}
|
||||
stop_flag = threading.Event()
|
||||
|
||||
def pnj_producer():
|
||||
idx = 0
|
||||
while not stop_flag.is_set():
|
||||
prompt = PNJ_PROMPTS[idx % len(PNJ_PROMPTS)]
|
||||
r = call_llm(MODEL_PNJ, prompt)
|
||||
r["produced_at"] = time.time()
|
||||
pnj_queue.put(r)
|
||||
with threading.Lock():
|
||||
status = f"ERR: {r['error']}" if r['error'] else f"{r['tok_s']} tok/s"
|
||||
print(f" [PNJ 7b ] prod #{idx+1} → {status} ({r['elapsed']}s) | queue={pnj_queue.qsize()}")
|
||||
idx += 1
|
||||
|
||||
def mj_consumer():
|
||||
consumed = 0
|
||||
while consumed < rounds:
|
||||
# Attendre qu'il y ait au moins 1 item dans la queue (ou timeout)
|
||||
try:
|
||||
pnj_result = pnj_queue.get(timeout=90)
|
||||
except queue.Empty:
|
||||
print(" [MJ 14b] Timeout attente PNJ")
|
||||
break
|
||||
|
||||
latency = time.time() - pnj_result["produced_at"]
|
||||
# Construire le prompt MJ à partir du résultat PNJ
|
||||
pnj_text = pnj_result.get("text", "(vide)")
|
||||
mj_prompt = (
|
||||
f"Un PNJ vient de parler : \"{pnj_text}\"\n"
|
||||
f"En tant que MJ, en 2 phrases, quelle est la conséquence narrative ?"
|
||||
)
|
||||
r = call_llm(MODEL_MJ, mj_prompt)
|
||||
r["latency_from_pnj"] = round(latency, 2)
|
||||
all_results["mj"].append(r)
|
||||
all_results["latency_pnj_to_mj"].append(round(latency + r["elapsed"], 2))
|
||||
status = f"ERR: {r['error']}" if r['error'] else f"{r['tok_s']} tok/s"
|
||||
print(f" [MJ 14b ] cons #{consumed+1} → {status} ({r['elapsed']}s) | pipeline={round(latency+r['elapsed'],1)}s")
|
||||
consumed += 1
|
||||
pnj_queue.task_done()
|
||||
print()
|
||||
|
||||
# Lancer N threads PNJ + 1 thread MJ
|
||||
pnj_threads = [threading.Thread(target=pnj_producer, daemon=True) for _ in range(pnj_count)]
|
||||
mj_thread = threading.Thread(target=mj_consumer)
|
||||
|
||||
t_start = time.time()
|
||||
for t in pnj_threads:
|
||||
t.start()
|
||||
mj_thread.start()
|
||||
mj_thread.join()
|
||||
stop_flag.set()
|
||||
|
||||
# Récupérer tous les résultats PNJ dans la queue restante
|
||||
while not pnj_queue.empty():
|
||||
all_results["pnj"].append(pnj_queue.get())
|
||||
|
||||
total = time.time() - t_start
|
||||
return {**all_results, "total_elapsed": round(total, 1)}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Rapport
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def print_report(results: dict, mode: str):
|
||||
def stats(lst: list[dict]) -> dict:
|
||||
if not lst:
|
||||
return {}
|
||||
ok = [r for r in lst if not r.get("error")]
|
||||
err = len(lst) - len(ok)
|
||||
if not ok:
|
||||
return {"calls": len(lst), "errors": err}
|
||||
tok_s = [r["tok_s"] for r in ok]
|
||||
elaps = [r["elapsed"] for r in ok]
|
||||
return {
|
||||
"calls": len(lst),
|
||||
"errors": err,
|
||||
"avg_tok_s": round(sum(tok_s) / len(tok_s), 1),
|
||||
"min_tok_s": min(tok_s),
|
||||
"max_tok_s": max(tok_s),
|
||||
"avg_elapsed": round(sum(elaps) / len(elaps), 1),
|
||||
"total_tokens": sum(r["tokens"] for r in ok),
|
||||
}
|
||||
|
||||
sep = "=" * 60
|
||||
print(f"\n{sep}")
|
||||
print(f" RAPPORT — MODE {mode.upper()}")
|
||||
print(f" {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
print(sep)
|
||||
|
||||
if "mj" in results and results["mj"]:
|
||||
s = stats(results["mj"])
|
||||
print(f"\n MJ 14b ({MODEL_MJ})")
|
||||
for k, v in s.items():
|
||||
print(f" {k:20s} : {v}")
|
||||
|
||||
if "pnj" in results and results["pnj"]:
|
||||
s = stats(results["pnj"])
|
||||
print(f"\n PNJ 7b ({MODEL_PNJ})")
|
||||
for k, v in s.items():
|
||||
print(f" {k:20s} : {v}")
|
||||
|
||||
if "latency_pnj_to_mj" in results and results["latency_pnj_to_mj"]:
|
||||
lat = results["latency_pnj_to_mj"]
|
||||
print(f"\n Pipeline PNJ→MJ (latence totale)")
|
||||
print(f" avg_pipeline_s : {round(sum(lat)/len(lat), 1)}")
|
||||
print(f" min_pipeline_s : {min(lat)}")
|
||||
print(f" max_pipeline_s : {max(lat)}")
|
||||
|
||||
print(f"\n Durée totale : {results.get('total_elapsed', '?')}s")
|
||||
|
||||
# Score de viabilité
|
||||
mj_ok = not any(r.get("error") for r in results.get("mj", []))
|
||||
pnj_ok = not any(r.get("error") for r in results.get("pnj", []))
|
||||
mj_speed = stats(results.get("mj", [])).get("avg_tok_s", 0)
|
||||
pnj_speed = stats(results.get("pnj", [])).get("avg_tok_s", 0)
|
||||
|
||||
print(f"\n DIAGNOSTIC")
|
||||
print(f" MJ stable : {'✓' if mj_ok else '✗ ERREURS'}")
|
||||
print(f" PNJ stable : {'✓' if pnj_ok else '✗ ERREURS'}")
|
||||
if mj_speed:
|
||||
viable = "✓ VIABLE" if mj_speed > 8 else "⚠ LENT (< 8 tok/s)" if mj_speed > 3 else "✗ TROP LENT"
|
||||
print(f" MJ débit : {mj_speed} tok/s → {viable}")
|
||||
if pnj_speed:
|
||||
viable = "✓ VIABLE" if pnj_speed > 15 else "⚠ LENT" if pnj_speed > 5 else "✗ TROP LENT"
|
||||
print(f" PNJ débit : {pnj_speed} tok/s → {viable}")
|
||||
print(sep)
|
||||
|
||||
# Export JSON
|
||||
report_path = f"/tmp/crash_test_{mode}_{int(time.time())}.json"
|
||||
try:
|
||||
with open(report_path, "w") as f:
|
||||
json.dump({"mode": mode, "results": results,
|
||||
"stats": {"mj": stats(results.get("mj",[])),
|
||||
"pnj": stats(results.get("pnj",[]))}}, f, indent=2)
|
||||
print(f"\n Rapport JSON : {report_path}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="LLM Crash Test — Fallout Venice")
|
||||
parser.add_argument("--mode", choices=["simultane","decale","solo-mj","solo-pnj"],
|
||||
default="simultane")
|
||||
parser.add_argument("--rounds", type=int, default=5, help="Nombre de rounds MJ")
|
||||
parser.add_argument("--pnj-count", type=int, default=2, help="PNJ parallèles (modes simultane/decale)")
|
||||
args = parser.parse_args()
|
||||
|
||||
print(f"=== LLM CRASH TEST — {args.mode.upper()} ===")
|
||||
print(f"Ollama : {OLLAMA_URL}")
|
||||
print(f"MJ : {MODEL_MJ}")
|
||||
print(f"PNJ : {MODEL_PNJ}")
|
||||
print(f"Rounds : {args.rounds} | PNJ parallèles : {args.pnj_count}")
|
||||
|
||||
# Vérifier que Ollama répond
|
||||
try:
|
||||
with urllib.request.urlopen(f"{OLLAMA_URL}/api/tags", timeout=5) as r:
|
||||
models = [m["name"] for m in json.loads(r.read()).get("models", [])]
|
||||
print(f"Modèles disponibles : {', '.join(models[:5])}")
|
||||
except Exception as e:
|
||||
print(f"[ERR] Ollama inaccessible : {e}")
|
||||
sys.exit(1)
|
||||
|
||||
t_global = time.time()
|
||||
|
||||
if args.mode == "simultane":
|
||||
results = run_simultane(args.rounds, args.pnj_count)
|
||||
elif args.mode == "decale":
|
||||
results = run_decale(args.rounds, args.pnj_count)
|
||||
elif args.mode == "solo-mj":
|
||||
r = run_solo(MODEL_MJ, MJ_PROMPTS, args.rounds)
|
||||
results = {"mj": r, "total_elapsed": round(time.time() - t_global, 1)}
|
||||
elif args.mode == "solo-pnj":
|
||||
r = run_solo(MODEL_PNJ, PNJ_PROMPTS, args.rounds)
|
||||
results = {"pnj": r, "total_elapsed": round(time.time() - t_global, 1)}
|
||||
|
||||
print_report(results, args.mode)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user