""" Capacity Test — Fallout Venice of Wasteland Série de tests pour mesurer la capacité de charge LLM. Simule N joueurs simultanés et mesure : - Temps de réponse par joueur (P50, P95, max) - Débit total (tokens/s global) - Temps d'attente queue (joueur 1 vs joueur N) - Viabilité pour le mode temps-réel (1h/tick) Usage : python3 capacity_test.py python3 capacity_test.py --quick (skip 50-player test) python3 capacity_test.py --output /path/to/results.json """ import os, sys, time, json, argparse, threading, statistics 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") OUTPUT = os.getenv("CRASH_RESULTS_PATH", "/home/ubuntu/fallout-venice/src/config/crash_results.json") # --------------------------------------------------------------------------- # Prompts variés pour éviter le cache # --------------------------------------------------------------------------- PNJ_PROMPTS = [ "Tu es un garde de La Régie. Réponds en 1 phrase à : 'Laisse-moi passer.'", "Tu es un marchand créole de Pearl River. Réponds en 1 phrase à : 'T'as du carburant ?'", "Tu es une goule du Grand Krewe. Réponds en 1 phrase à : 'Que veut le Baron ?'", "Tu es un Écumeur. Réponds en 1 phrase à : 'On veut juste traverser le fleuve.'", "Tu es Sœur Eulalie de L'Union. Réponds en 1 phrase à : 'J'ai été mordu.'", "Tu es un technicien de la CdA. Réponds en 1 phrase à : 'Répare mon Pip-Boy.'", "Tu es un scavenger neutre. Réponds en 1 phrase à : 'Combien pour cette armure ?'", "Tu es un informateur du Syndicat. Réponds en 1 phrase à : 'Qui contrôle la Route 61 ?'", "Tu es un fermier de Oak Plantation. Réponds en 1 phrase à : 'Vous travaillez pour qui ?'", "Tu es un passeur de Laplace. Réponds en 1 phrase à : 'Combien pour traverser ?'", ] MJ_PROMPTS = [ "En 2 phrases, décris la réaction de la foule quand les joueurs entrent dans le marché de NOLA.", "En 2 phrases, décris l'ambiance de Pearl River à l'aube après une nuit de combat.", "En 2 phrases, que se passe-t-il quand le Grand Krewe envoie un émissaire aux joueurs ?", "En 2 phrases, décris les conséquences d'un vol de convoi de la Régie.", "En 2 phrases, comment réagit L'Union quand elle apprend qu'un de ses membres a trahi ?", ] # --------------------------------------------------------------------------- # HTTP # --------------------------------------------------------------------------- def http_post(url: str, body: dict, timeout: int = 180) -> dict: data = json.dumps(body).encode() req = urllib.request.Request(url, data=data, method="POST", headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()) def call_llm(model: str, prompt: str, player_id: int = 0) -> dict: t0 = time.time() try: res = http_post(f"{OLLAMA_URL}/api/generate", { "model": model, "prompt": prompt, "stream": False, "options": {"num_predict": 80, "temperature": 0.7, "seed": player_id}, }) elapsed = time.time() - t0 tokens = res.get("eval_count", 0) return { "player_id": player_id, "model": model, "tokens": tokens, "elapsed": round(elapsed, 2), "tok_s": round(tokens / elapsed, 2) if elapsed > 0 else 0, "queued_s": round(res.get("load_duration", 0) / 1e9, 2), "error": None, "enqueued_at": t0, } except Exception as e: return { "player_id": player_id, "model": model, "tokens": 0, "elapsed": round(time.time() - t0, 2), "tok_s": 0, "queued_s": 0, "error": str(e)[:80], "enqueued_at": t0, } # --------------------------------------------------------------------------- # Scénarios # --------------------------------------------------------------------------- def scenario_warmup() -> dict: """1 appel MJ + 1 appel PNJ pour chauffer les modèles.""" print(" [WARMUP] Chargement des modèles en RAM...") r_mj = call_llm(MODEL_MJ, MJ_PROMPTS[0], 0) r_pnj = call_llm(MODEL_PNJ, PNJ_PROMPTS[0], 0) ok = "✓" if not r_mj["error"] and not r_pnj["error"] else "✗" print(f" {ok} MJ: {r_mj['elapsed']}s ({r_mj['tok_s']} t/s) " f"PNJ: {r_pnj['elapsed']}s ({r_pnj['tok_s']} t/s)") return {"mj_warmup": r_mj, "pnj_warmup": r_pnj} def scenario_n_players(n: int, rounds: int = 3) -> dict: """ Simule N joueurs envoyant une action simultanément. Chaque joueur → 1 appel PNJ 7b (en parallèle via threads). Après tous les PNJ → 1 appel MJ 14b (narrateur du tick). """ print(f"\n [{n} JOUEURS] {rounds} round(s) — {n} PNJ en parallèle + 1 MJ après") all_rounds = [] for r in range(rounds): round_results = {"pnj": [], "mj": None} t_round_start = time.time() # Phase 1 : N joueurs en parallèle lock = threading.Lock() threads = [] def player_action(pid: int): prompt = PNJ_PROMPTS[pid % len(PNJ_PROMPTS)] res = call_llm(MODEL_PNJ, prompt, pid) with lock: round_results["pnj"].append(res) for i in range(n): t = threading.Thread(target=player_action, args=(i,)) threads.append(t) t_first = time.time() for t in threads: t.start() for t in threads: t.join() t_all_pnj_done = time.time() pnj_total_wall = round(t_all_pnj_done - t_first, 2) # Phase 2 : MJ synthétise mj_prompt = ( f"Tick {r+1}, {n} joueurs ont agi simultanément dans NOLA. " + MJ_PROMPTS[r % len(MJ_PROMPTS)] ) mj_res = call_llm(MODEL_MJ, mj_prompt, -1) round_results["mj"] = mj_res t_round_total = round(time.time() - t_round_start, 2) # Stats PNJ de ce round pnj_ok = [x for x in round_results["pnj"] if not x["error"]] pnj_elapsed = sorted([x["elapsed"] for x in pnj_ok]) p50 = statistics.median(pnj_elapsed) if pnj_elapsed else 0 p95 = pnj_elapsed[int(len(pnj_elapsed) * 0.95)] if len(pnj_elapsed) > 1 else (pnj_elapsed[0] if pnj_elapsed else 0) status = "✓" if not mj_res["error"] and len(pnj_ok) == n else f"⚠ {n - len(pnj_ok)} erreurs" print(f" R{r+1}: PNJ wall={pnj_total_wall}s P50={p50}s P95={p95}s | " f"MJ={mj_res['elapsed']}s | total={t_round_total}s | {status}") all_rounds.append({ "round": r + 1, "pnj": round_results["pnj"], "mj": mj_res, "pnj_wall_time": pnj_total_wall, "round_total_time": t_round_total, "pnj_p50": p50, "pnj_p95": p95, }) return {"n_players": n, "rounds": all_rounds} def compute_stats(scenario_result: dict) -> dict: n = scenario_result["n_players"] rounds = scenario_result["rounds"] all_pnj_elapsed = [x["elapsed"] for r in rounds for x in r["pnj"] if not x.get("error")] all_pnj_wall = [r["pnj_wall_time"] for r in rounds] all_mj_elapsed = [r["mj"]["elapsed"] for r in rounds if not r["mj"].get("error")] all_total = [r["round_total_time"] for r in rounds] all_p50 = [r["pnj_p50"] for r in rounds] all_p95 = [r["pnj_p95"] for r in rounds] errors = sum(1 for r in rounds for x in r["pnj"] if x.get("error")) def avg(lst): return round(sum(lst) / len(lst), 2) if lst else 0 tick_duration_s = avg(all_total) viable_realtime = tick_duration_s < 3600 viable_5min = tick_duration_s < 300 return { "n_players": n, "pnj_avg_elapsed_s": avg(all_pnj_elapsed), "pnj_avg_wall_s": avg(all_pnj_wall), "pnj_avg_p50_s": avg(all_p50), "pnj_avg_p95_s": avg(all_p95), "mj_avg_elapsed_s": avg(all_mj_elapsed), "tick_total_avg_s": tick_duration_s, "tick_total_min_s": min(all_total) if all_total else 0, "tick_total_max_s": max(all_total) if all_total else 0, "errors": errors, "viable_realtime_1h": viable_realtime, "viable_5min_tick": viable_5min, "actions_per_hour": int(3600 / tick_duration_s) if tick_duration_s > 0 else 0, } # --------------------------------------------------------------------------- # Rapport # --------------------------------------------------------------------------- SEP = "=" * 70 def print_report(warmup: dict, scenarios: list[dict], stats_list: list[dict]): print(f"\n{SEP}") print(f" RAPPORT CAPACITÉ — FALLOUT VENICE — {datetime.now().strftime('%d/%m/%Y %H:%M')}") print(SEP) w_mj = warmup.get("mj_warmup", {}) w_pnj = warmup.get("pnj_warmup", {}) print(f"\n BASELINE (warm)") print(f" MJ 14b : {w_mj.get('elapsed','?')}s | {w_mj.get('tok_s','?')} tok/s") print(f" PNJ 7b : {w_pnj.get('elapsed','?')}s | {w_pnj.get('tok_s','?')} tok/s") print(f"\n {'N':>4} {'PNJ wall':>10} {'P50':>7} {'P95':>7} {'MJ':>7} {'TICK total':>11} {'Actions/h':>10} {'1h viable':>10} ERR") print(f" {'-'*4} {'-'*10} {'-'*7} {'-'*7} {'-'*7} {'-'*11} {'-'*10} {'-'*10} {'-'*4}") for s in stats_list: v1h = "✓" if s["viable_realtime_1h"] else "✗" err = str(s["errors"]) if s["errors"] else "0" print(f" {s['n_players']:>4} " f"{s['pnj_avg_wall_s']:>9.1f}s " f"{s['pnj_avg_p50_s']:>6.1f}s " f"{s['pnj_avg_p95_s']:>6.1f}s " f"{s['mj_avg_elapsed_s']:>6.1f}s " f"{s['tick_total_avg_s']:>10.1f}s " f"{s['actions_per_hour']:>10} " f"{v1h:>10} " f"{err:>4}") print(f"\n LÉGENDE") print(f" PNJ wall : durée réelle pour que TOUS les PNJ répondent (parallèle)") print(f" P50 / P95 : médiane / 95e percentile du temps d'attente par joueur") print(f" TICK total : PNJ wall + MJ (= durée d'un tick complet)") print(f" Actions/h : combien de ticks complets tiennent en 1h réelle") print(f" 1h viable : tick < 1h (compatible mode temps-réel)") print(f"\n RECOMMANDATIONS") for s in stats_list: n = s["n_players"] t = s["tick_total_avg_s"] a = s["actions_per_hour"] if a >= 6: msg = f"✓ {n} joueurs → 1 action toutes les {round(t/60,1)} min — jouable" elif a >= 1: msg = f"⚠ {n} joueurs → 1 action toutes les {round(t/60,1)} min — lent mais viable en temps réel" else: msg = f"✗ {n} joueurs → tick de {round(t/60,1)} min — dépasse 1h, non viable en temps réel" print(f" {msg}") print(f"\n{SEP}") # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description="Capacity test LLM Fallout Venice") parser.add_argument("--quick", action="store_true", help="Skip 50-player test") parser.add_argument("--output", default=OUTPUT) args = parser.parse_args() print(f"\n{SEP}") print(f" CAPACITY TEST — FALLOUT VENICE OF WASTELAND") print(f" Ollama : {OLLAMA_URL}") print(f" MJ : {MODEL_MJ} | PNJ : {MODEL_PNJ}") print(f" Début : {datetime.now().strftime('%H:%M:%S')}") print(SEP) # Vérifier Ollama 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"\n Modèles dispos : {', '.join(models)}") except Exception as e: print(f" [ERR] Ollama inaccessible : {e}") sys.exit(1) # 1. Warmup (toujours) print(f"\n{SEP}") print(" PHASE 0 — WARMUP (chargement modèles en RAM)") print(SEP) warmup = scenario_warmup() # 2. Série de tests configs = [ {"n": 1, "rounds": 3}, # baseline solo {"n": 5, "rounds": 3}, # petit groupe {"n": 10, "rounds": 2}, # groupe moyen ] if not args.quick: configs.append({"n": 50, "rounds": 1}) # stress scenarios = [] stats_list = [] for cfg in configs: n, rounds = cfg["n"], cfg["rounds"] print(f"\n{SEP}") print(f" PHASE — {n} JOUEUR{'S' if n > 1 else ''} SIMULTANÉ{'S' if n > 1 else ''} ({rounds} round{'s' if rounds > 1 else ''})") print(SEP) result = scenario_n_players(n, rounds) s = compute_stats(result) scenarios.append(result) stats_list.append(s) # 3. Rapport terminal print_report(warmup, scenarios, stats_list) # 4. Export JSON output = { "test_type": "capacity", "mode": "capacity_test", "timestamp": datetime.now().isoformat(), "ollama_url": OLLAMA_URL, "model_mj": MODEL_MJ, "model_pnj": MODEL_PNJ, "warmup": warmup, "scenarios": scenarios, "stats": stats_list, "file_date": datetime.now().strftime("%d/%m/%Y %H:%M:%S"), } try: with open(args.output, "w") as f: json.dump(output, f, indent=2, ensure_ascii=False) print(f"\n Résultats écrits → {args.output}") except Exception as e: print(f" [WARN] Impossible d'écrire le fichier : {e}") if __name__ == "__main__": main()