This commit is contained in:
Samir Medjaher 2026-09-13 08:11:35 +02:00 committed by GitHub
commit 1875d7924c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 11621 additions and 0 deletions

View File

@ -0,0 +1,131 @@
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import List
@dataclass
class CortexIntent:
intent: str
confidence: str = "medium"
route_reason: str = "default"
needs_web_search: bool = False
needs_vault_search: bool = False
dashboard_context: str = ""
def get(self, key, default=None):
return getattr(self, key, default)
DASHBOARD_CONTEXT = """Contexte dashboard Cortex:
- URL locale: http://127.0.0.1:8765/
- Dashboard GPU/brain: brain_gpu.html
- Sidecar chat à droite
- Onglet Playtest
- Onglet Consortium
- APIs utiles: /api/cortex/judges, /api/cortex/homeostasis, /api/chat
"""
def _norm(text: str) -> str:
return (text or "").lower().strip()
def detect_intent(message: str) -> CortexIntent:
m = _norm(message)
if m.startswith("/code") and any(k in m for k in [
"playtest", "app", "application", "html", "interface",
"calculatrice", "todo", "kanban", "dashboard"
]):
return CortexIntent(
intent="playtest_code_task",
confidence="high",
route_reason="playtest_builder_direct",
dashboard_context=DASHBOARD_CONTEXT,
)
if any(k in m for k in [
"recherche web", "cherche sur le web", "actualité", "actualités",
"actu", "news", "récent", "récente", "aujourd'hui", "maintenant",
"dernières nouvelles", "latest"
]):
return CortexIntent(
intent="recent_web_search",
confidence="high",
route_reason="needs_web_search",
needs_web_search=True,
)
if any(k in m for k in [
"playtest", "playtest intégré", "dashboard", "brain_gpu",
"sidecar", "consortium", "juges", "judges", "homeostasis"
]):
return CortexIntent(
intent="playtest_dashboard_help",
confidence="high",
route_reason="dashboard_context_injected",
dashboard_context=DASHBOARD_CONTEXT,
)
if any(k in m for k in [
"vault", "mémoire", "memoire", "obsidian", "souviens",
"tu vois le projet", "projet de site", "site comores", "comores",
"workspace", "fichier local", "dans le repo", "dans les fichiers"
]):
return CortexIntent(
intent="local_project_search",
confidence="high",
route_reason="needs_vault_or_file_search",
needs_vault_search=True,
)
if any(k in m for k in [
"debug", "raisonne", "analyse profonde", "planifie",
"architecture", "pourquoi", "diagnostic"
]):
return CortexIntent(
intent="deep_reason",
confidence="medium",
route_reason="deep_reason_requested",
)
if any(k in m for k in [
"tu es qui", "présente toi", "présente-toi", "qui es-tu",
"c'est quoi cortex", "qui t'a créé"
]):
return CortexIntent(
intent="identity",
confidence="high",
route_reason="identity_or_presentation",
)
if any(k in m for k in [
"code", "patch", "modifie", "corrige", "commande",
"powershell", "python", "git"
]):
return CortexIntent(
intent="code_task",
confidence="medium",
route_reason="code_or_task_execution",
)
return CortexIntent(
intent="simple_chat",
confidence="medium",
route_reason="simple_chat",
)
def metadata(intent: CortexIntent, backend: str = "minimax_fast", tools_used: List[str] | None = None, evidence_count: int = 0) -> dict:
return {
"intent": intent.intent,
"tools_used": tools_used or [],
"evidence_count": evidence_count,
"backend": backend,
"route_reason": intent.route_reason,
"confidence": intent.confidence,
"needs_web_search": intent.needs_web_search,
"needs_vault_search": intent.needs_vault_search,
}

View File

@ -0,0 +1,393 @@
"""
cortex_kv_quantize.py Quantization KV cache TurboQuant-inspired pour LLM local.
(Distinct de cortex_quantize.py qui s'occupe du graphe TF-IDF.)
État actuel (2026-04) :
- TurboQuant 3-bit (Google) pas encore mergé dans llama.cpp / LM Studio mainstream.
- DISPONIBLE MAINTENANT dans llama.cpp : Q8_0 (~25% saving), Q4_0 (~50% saving)
sur le KV cache K et V.
- LM Studio expose les options dans Advanced Configuration KV Cache Quantization.
Sam veut tester maintenant donc on applique la plus agressive disponible (Q4_0)
qui est déjà éprouvée et stable, et on prépare le swap vers Q3 dès qu'il arrive.
Pipeline :
1. detect_loaded_model() : interroge LM Studio
2. recommend() : choisit Q4_0 si stable, sinon Q8_0
3. measure_latency() : mesure baseline avant changement
4. apply_lmstudio_config() : génère un guide précis pour LM Studio UI
5. After Sam applies re-measure_latency() pour valider
Endpoint serveur : GET /api/cortex/kv_quantize
"""
try:
from lmstudio_response import extract_lmstudio_content
except Exception:
from scripts.brain.lmstudio_response import extract_lmstudio_content
import json
import time
from pathlib import Path
import urllib.request
import urllib.error
REPO_ROOT = Path(r"H:\Code\Paperclip")
STATE_FILE = REPO_ROOT / ".cortex-kv-quantize-state.json"
GUIDE_FILE = REPO_ROOT / ".cortex-kv-quantize-guide.md"
LATENCY_LOG = REPO_ROOT / ".cortex-kv-quantize-latency.jsonl"
LMSTUDIO_API = "http://localhost:1234/v1"
# Modèles connus (heuristique)
KNOWN_MODELS = {
"qwen3.6-35b-a3b": {"params_b": 35, "n_layers": 64, "hidden": 5120,
"n_kv_heads": 8, "head_dim": 128, "moe": True},
"qwen3.5-27b": {"params_b": 27, "n_layers": 56, "hidden": 4480,
"n_kv_heads": 8, "head_dim": 128, "moe": False},
"llama-3-70b": {"params_b": 70, "n_layers": 80, "hidden": 8192,
"n_kv_heads": 8, "head_dim": 128, "moe": False},
"llama-3-8b": {"params_b": 8, "n_layers": 32, "hidden": 4096,
"n_kv_heads": 8, "head_dim": 128, "moe": False},
"mistral-7b": {"params_b": 7, "n_layers": 32, "hidden": 4096,
"n_kv_heads": 8, "head_dim": 128, "moe": False},
}
DTYPE_BYTES = {
"fp16": 2.0,
"q8_0": 1.0,
"q4_0": 0.5,
"q4_1": 0.5,
"q3_turboquant": 0.375, # placeholder — pas encore dispo
}
DTYPE_QUALITY = {
"fp16": {"loss_pct": 0.0, "stable": True, "available": True,
"note": "baseline, pas de quantization"},
"q8_0": {"loss_pct": 0.1, "stable": True, "available": True,
"note": "perte imperceptible · safe par défaut"},
"q4_0": {"loss_pct": 1.5, "stable": True, "available": True,
"note": "agressif · ~50% saving · valide chat/RAG"},
"q4_1": {"loss_pct": 1.0, "stable": True, "available": True,
"note": "mieux que q4_0 sur tâches sensibles"},
"q3_turboquant": {"loss_pct": 0.3, "stable": False, "available": False,
"note": "Google TurboQuant · pas encore dans llama.cpp · ETA 2026-Q3"},
}
# Quantization des POIDS GGUF — c'est ÇA qui mange la VRAM (pas le KV cache).
# Sur qwen35b par exemple : poids fp16=70GB, Q4_K_M=~17GB, Q3_K_M=~13GB, IQ3_XXS=~10GB.
WEIGHT_QUANT_PROFILES = {
"fp16": {"size_ratio": 1.00, "loss_pct": 0.0, "speed_factor": 0.5,
"note": "baseline · jamais utilisé en local sauf sur petit modèle"},
"Q8_0": {"size_ratio": 0.53, "loss_pct": 0.05, "speed_factor": 1.0,
"note": "presque sans perte · 47% VRAM en moins"},
"Q6_K": {"size_ratio": 0.41, "loss_pct": 0.15, "speed_factor": 1.05,
"note": "excellent compromis · perte invisible"},
"Q5_K_M": {"size_ratio": 0.36, "loss_pct": 0.30, "speed_factor": 1.1,
"note": "très bon · standard production"},
"Q4_K_M": {"size_ratio": 0.30, "loss_pct": 0.80, "speed_factor": 1.15,
"note": "default LM Studio · bon ratio"},
"Q4_K_S": {"size_ratio": 0.28, "loss_pct": 1.20, "speed_factor": 1.18,
"note": "un peu plus petit que Q4_K_M"},
"Q3_K_M": {"size_ratio": 0.22, "loss_pct": 2.50, "speed_factor": 1.25,
"note": "perte commence à se voir · OK pour chat simple"},
"IQ3_XXS": {"size_ratio": 0.19, "loss_pct": 3.00, "speed_factor": 1.30,
"note": "improved quantization · moins de perte que Q3_K_M à taille égale"},
"Q2_K": {"size_ratio": 0.18, "loss_pct": 5.00, "speed_factor": 1.35,
"note": "extrême · perte notable · à éviter sauf VRAM critique"},
"IQ2_M": {"size_ratio": 0.16, "loss_pct": 4.50, "speed_factor": 1.35,
"note": "improved Q2 · plus stable que Q2_K"},
}
def detect_loaded_model() -> dict:
try:
with urllib.request.urlopen(f"{LMSTUDIO_API}/models", timeout=3) as r:
data = json.loads(r.read().decode())
models = [m["id"] for m in data.get("data", []) if not m["id"].startswith("text-embedding")]
return {"ok": True, "models": models, "lmstudio_up": True}
except Exception as e:
return {"ok": False, "error": str(e), "lmstudio_up": False}
def model_meta(model_id: str) -> dict:
mid = (model_id or "").lower()
for key, meta in KNOWN_MODELS.items():
if key in mid:
return {**meta, "id": model_id, "matched_key": key}
import re, math
m = re.search(r'(\d+)b', mid)
params_b = int(m.group(1)) if m else 7
return {"params_b": params_b,
"n_layers": max(24, int(4 * math.log2(max(2, params_b * 8)))),
"hidden": max(2048, int(params_b ** 0.5 * 1024)),
"n_kv_heads": 8, "head_dim": 128, "moe": False,
"id": model_id, "matched_key": "(heuristique)"}
def kv_cache_size_bytes(meta: dict, n_ctx: int, dtype: str = "fp16") -> int:
bytes_per = DTYPE_BYTES.get(dtype, 2.0)
return int(2 * n_ctx * meta["n_layers"] * meta["n_kv_heads"]
* meta["head_dim"] * bytes_per)
def weight_size_gb(params_b: float, profile: str) -> float:
"""Taille des poids selon profile de quantization (en GB)."""
p = WEIGHT_QUANT_PROFILES.get(profile, {"size_ratio": 1.0})
# 2 bytes par param en fp16, ratio appliqué selon quantization
return params_b * 2.0 * p["size_ratio"]
def weights_recommend(params_b: float, target_vram_gb: float = 12.0) -> dict:
"""Choisis la quantization de poids la plus QUALITATIVE qui tient dans target_vram_gb.
On laisse une réserve pour KV cache + activations."""
# Réserve : ~2 GB pour KV (avec quantization Q4) + 1 GB activations + 0.5 GB système
reserve_gb = 3.5
budget_gb = max(1.0, target_vram_gb - reserve_gb)
options = []
for prof, info in WEIGHT_QUANT_PROFILES.items():
size_gb = weight_size_gb(params_b, prof)
options.append({
"profile": prof, "size_gb": round(size_gb, 2),
"fits_target": size_gb <= budget_gb,
"loss_pct": info["loss_pct"], "speed_factor": info["speed_factor"],
"note": info["note"],
})
options.sort(key=lambda x: x["size_gb"], reverse=True) # plus gros d'abord
# Recommandation : la plus QUALITATIVE qui tient (loss_pct le plus bas qui fit)
fitting = [o for o in options if o["fits_target"]]
if fitting:
# Parmi celles qui tiennent, choisis la plus qualitative (loss_pct min)
recommended = min(fitting, key=lambda x: x["loss_pct"])
else:
# Aucune ne tient → la plus petite
recommended = min(options, key=lambda x: x["size_gb"])
return {
"params_b": params_b, "target_vram_gb": target_vram_gb,
"budget_for_weights_gb": round(budget_gb, 2),
"options": options, "recommended": recommended,
}
def recommend(target_vram_gb: float = 12.0, n_ctx: int = 8192,
model_id: str | None = None) -> dict:
detected = detect_loaded_model()
model_id = model_id or (detected.get("models") or ["qwen3.6-35b-a3b"])[0]
meta = model_meta(model_id)
options = []
for dtype in ["fp16", "q8_0", "q4_1", "q4_0", "q3_turboquant"]:
size_bytes = kv_cache_size_bytes(meta, n_ctx, dtype)
size_gb = size_bytes / (1024**3)
q = DTYPE_QUALITY[dtype]
options.append({
"dtype": dtype, "size_gb": round(size_gb, 2),
"fits_target": size_gb <= target_vram_gb,
"loss_pct": q["loss_pct"], "stable": q["stable"],
"available_now": q["available"], "note": q["note"],
})
# Choisis la plus agressive ET stable ET disponible ET qui tient
recommended = None
for opt in reversed(options):
if opt["fits_target"] and opt["stable"] and opt["available_now"]:
recommended = opt; break
if not recommended:
recommended = next((o for o in options if o["available_now"]), options[0])
fp16_gb = next(o["size_gb"] for o in options if o["dtype"] == "fp16")
save_gb = fp16_gb - recommended["size_gb"]
return {
"ts": time.time(),
"model": meta, "n_ctx": n_ctx, "target_vram_gb": target_vram_gb,
"options": options,
"recommended": recommended,
"save_gb_vs_fp16": round(save_gb, 2),
"save_pct_vs_fp16": round(save_gb / max(0.001, fp16_gb) * 100, 1),
"lmstudio_status": detected,
"future_q3_turboquant": {
"size_gb": round(kv_cache_size_bytes(meta, n_ctx, "q3_turboquant")/(1024**3), 2),
"available_now": False, "eta": "2026-Q3",
"tracking": "https://research.google/blog/turboquant-redefining-ai-efficiency-with-extreme-compression/",
},
}
def measure_latency(prompt: str = "Bonjour, dis-moi en 1 phrase ce qu'est Python.",
max_tokens: int = 60, n_runs: int = 2) -> dict:
"""Mesure latence p50 sur N runs. Logue dans LATENCY_LOG pour comparer avant/après."""
detected = detect_loaded_model()
if not detected.get("ok") or not detected.get("models"):
return {"ok": False, "error": "LM Studio inactif"}
model_id = detected["models"][0]
durations = []
for i in range(n_runs):
t0 = time.time()
try:
payload = json.dumps({
"model": model_id,
"messages": [
{"role": "system", "content": "/no_think\nOutput only the final answer. No hidden reasoning. No markdown unless explicitly requested."},
{"role": "user", "content": "/no_think\n" + prompt},
],
"max_tokens": max_tokens,
"temperature": 0.1,
}).encode("utf-8")
req = urllib.request.Request(
f"{LMSTUDIO_API}/chat/completions", data=payload,
headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=120) as r:
resp = json.loads(r.read().decode())
dur = time.time() - t0
tokens = (resp.get("usage") or {}).get("completion_tokens") or max_tokens
durations.append({"run": i, "duration_s": round(dur, 2),
"tokens": tokens,
"tokens_per_s": round(tokens / max(0.01, dur), 1)})
except Exception as e:
durations.append({"run": i, "error": str(e)})
successful = [d for d in durations if "duration_s" in d]
p50 = sorted(d["duration_s"] for d in successful)[len(successful)//2] if successful else None
avg_tps = (sum(d["tokens_per_s"] for d in successful) / len(successful)
if successful else None)
result = {
"ok": bool(successful), "ts": time.time(), "model": model_id,
"n_runs": n_runs, "p50_s": p50, "avg_tokens_per_s": round(avg_tps, 1) if avg_tps else None,
"details": durations,
}
try:
with LATENCY_LOG.open("a", encoding="utf-8") as f:
f.write(json.dumps(result, ensure_ascii=False) + "\n")
except Exception: pass
return result
def apply_lmstudio_config(plan: dict) -> dict:
"""Génère un guide markdown précis (LM Studio n'expose pas d'API config)."""
rec = plan.get("recommended") or {}
dtype = rec.get("dtype", "q8_0")
model_id = (plan.get("model") or {}).get("id", "?")
save_gb = plan.get("save_gb_vs_fp16", 0)
save_pct = plan.get("save_pct_vs_fp16", 0)
body = f"""# Application KV cache quantization — `{dtype}`
Modèle ciblé : `{model_id}`
Économie attendue : **{save_gb} GB VRAM** (~{save_pct}% du KV cache)
Perte qualité estimée : ~{rec.get('loss_pct', 0)}% perplexity
## Étapes dans LM Studio (UI)
1. Décharge le modèle s'il tourne (icône Eject à côté du modèle dans Models).
2. Clic sur le modèle **Settings** (engrenage).
3. Section **Advanced Configuration** **KV Cache Quantization**.
4. Sélectionner :
- **K cache type** : `{dtype}`
- **V cache type** : `{dtype}`
5. **Save & Reload** le modèle.
6. Vérifier dans Performance : VRAM doit baisser de ~{save_pct}%.
## Test latence avant/après (script auto)
```bash
# AVANT — baseline (relancer plusieurs fois pour stabiliser)
python scripts/brain/cortex_kv_quantize.py latency
# Applique le changement dans LM Studio (manuel, voir au-dessus)
# APRÈS — vérifie tokens/s et vérification fonctionnelle
python scripts/brain/cortex_kv_quantize.py latency
# Comparer
python scripts/brain/cortex_kv_quantize.py compare
```
Le script log dans `.cortex-kv-quantize-latency.jsonl` chaque mesure pour
comparer p50 et tokens/s avant/après.
## Si la qualité chute trop
Re-applique en remontant : `q4_0` `q4_1` `q8_0` `fp16` (baseline).
## Quand TurboQuant Q3 sera mergé dans llama.cpp
Le module détecte automatiquement (relance `recommend()`) et propose le swap.
"""
try:
GUIDE_FILE.write_text(body, encoding="utf-8")
except Exception as e:
return {"ok": False, "error": str(e)}
return {"ok": True, "guide_path": str(GUIDE_FILE),
"save_gb": save_gb, "save_pct": save_pct, "dtype": dtype}
def compare_latencies(n_recent: int = 6) -> dict:
"""Compare les N dernières mesures pour montrer l'effet de la quantization."""
if not LATENCY_LOG.exists(): return {"ok": False, "error": "no measurements yet"}
lines = LATENCY_LOG.read_text(encoding="utf-8", errors="replace").splitlines()[-n_recent:]
runs = []
for ln in lines:
try: runs.append(json.loads(ln))
except Exception: pass
if len(runs) < 2: return {"ok": False, "error": "need at least 2 measurements"}
return {
"ok": True, "n_compared": len(runs),
"first": {"ts": runs[0]["ts"], "p50_s": runs[0].get("p50_s"),
"tokens_per_s": runs[0].get("avg_tokens_per_s")},
"last": {"ts": runs[-1]["ts"], "p50_s": runs[-1].get("p50_s"),
"tokens_per_s": runs[-1].get("avg_tokens_per_s")},
"speedup": (round(runs[0]["p50_s"] / max(0.01, runs[-1]["p50_s"]), 2)
if (runs[0].get("p50_s") and runs[-1].get("p50_s")) else None),
"all_runs": runs,
}
def full_recommend(target_vram_gb: float = 12.0, n_ctx: int = 8192) -> dict:
"""Recommandation complète : KV cache + poids + estimation VRAM totale."""
kv = recommend(target_vram_gb=target_vram_gb, n_ctx=n_ctx)
params_b = kv["model"].get("params_b", 7)
wts = weights_recommend(params_b, target_vram_gb=target_vram_gb)
rec_kv = kv.get("recommended", {})
rec_wt = wts.get("recommended", {})
# Total VRAM estimée avec les recommandations
total_with_rec = rec_kv.get("size_gb", 0) + rec_wt.get("size_gb", 0) + 1.0 # +activations
total_baseline = (kv["options"][0]["size_gb"] +
next(o["size_gb"] for o in wts["options"] if o["profile"] == "fp16") + 1.0)
saving = total_baseline - total_with_rec
return {
**kv,
"weights": wts,
"vram_estimation": {
"weights_gb": rec_wt.get("size_gb", 0),
"kv_cache_gb": rec_kv.get("size_gb", 0),
"activations_gb": 1.0,
"total_estimated_gb": round(total_with_rec, 2),
"vs_fp16_baseline_gb": round(total_baseline, 2),
"savings_gb": round(saving, 2),
},
"speedup_factor_estimated": rec_wt.get("speed_factor", 1.0),
}
def snapshot() -> dict:
plan = full_recommend()
apply_rep = apply_lmstudio_config(plan)
state = {**plan, "apply": apply_rep}
try:
STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False),
encoding="utf-8")
except Exception: pass
return state
if __name__ == "__main__":
import sys
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
cmd = sys.argv[1] if len(sys.argv) > 1 else "snapshot"
if cmd == "detect":
print(json.dumps(detect_loaded_model(), indent=2))
elif cmd == "recommend":
target = float(sys.argv[2]) if len(sys.argv) > 2 else 12.0
print(json.dumps(recommend(target_vram_gb=target), indent=2, ensure_ascii=False))
elif cmd == "apply":
plan = recommend()
print(json.dumps(apply_lmstudio_config(plan), indent=2))
elif cmd == "latency":
print(json.dumps(measure_latency(), indent=2, ensure_ascii=False))
elif cmd == "compare":
print(json.dumps(compare_latencies(), indent=2, ensure_ascii=False))
else:
print(json.dumps(snapshot(), indent=2, ensure_ascii=False))

View File

@ -0,0 +1,502 @@
"""
cortex_self_dev.py Boucle d'auto-développement de Cortex avec garde-fous.
Pipeline :
1. Reçoit un objectif en langage naturel ("ajoute un endpoint X", "fix bug Y")
2. Demande au router v2 de proposer un patch (modèles gratuits d'abord)
3. Parse la proposition (nouveau contenu de fichier ou diff)
4. Crée une branche cortex/dev/<timestamp>-<slug>
5. Applique le patch
6. Lance test_smoke complet
7. Si tests passent commit, sinon arrêt pour inspection
Le LLM ne touche JAMAIS le disque directement. Tout passe par cortex_tools
qui valide les chemins, et toute modif est isolée dans une branche git éphémère.
"""
try:
from lmstudio_response import extract_lmstudio_content
except Exception:
from scripts.brain.lmstudio_response import extract_lmstudio_content
try:
from lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
except Exception:
from scripts.brain.lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
import datetime as dt
import json
import re
import subprocess
import sys
import time
import urllib.request
from pathlib import Path
REPO_ROOT = Path(r"H:\Code\Paperclip")
ROUTER_URL = "http://127.0.0.1:18900/route_v2"
LM_STUDIO_URL = get_lmstudio_config()["base_url"] + "/v1"
SELFDEV_LOG = REPO_ROOT / ".cortex-self-dev.log"
GUARDRAILS_FILE = REPO_ROOT / "scripts" / "brain" / "cortex_self_dev_guardrails.json"
DEFAULT_GUARDRAILS = {
"enabled": True,
"allowed_path_prefixes": ["scripts/brain/"],
"blocked_path_fragments": [
".env", "secrets", "cookies", "token", "password", ".venv",
"node_modules", ".git"
],
"max_context_chars": 4200,
"max_files_per_change": 2,
"max_shrink_ratio": 0.55,
"require_explicit_path_in_goal": True,
"valid_tests": ["router", "serve", "memory", "tts", "cortex"],
"test_aliases": {"voice": "tts", "self_dev": "cortex", "identity": "cortex"},
"commit_only_applied_files": True,
"auto_apply_risk_threshold": "low"
}
def _load_guardrails() -> dict:
if not GUARDRAILS_FILE.exists():
GUARDRAILS_FILE.write_text(
json.dumps(DEFAULT_GUARDRAILS, ensure_ascii=False, indent=2),
encoding="utf-8",
)
return dict(DEFAULT_GUARDRAILS)
try:
data = json.loads(GUARDRAILS_FILE.read_text(encoding="utf-8"))
merged = dict(DEFAULT_GUARDRAILS)
merged.update(data if isinstance(data, dict) else {})
return merged
except Exception:
return dict(DEFAULT_GUARDRAILS)
# Import des tools sûrs
sys.path.insert(0, str(REPO_ROOT / "scripts" / "brain"))
import cortex_tools as ct
def _log(msg: str):
line = f"[{dt.datetime.now().isoformat(timespec='seconds')}] {msg}"
print(line, flush=True)
try:
with open(SELFDEV_LOG, "a", encoding="utf-8") as f: f.write(line + "\n")
except Exception: pass
def _slugify(text: str, max_len: int = 30) -> str:
s = re.sub(r'[^\w\s-]', '', text.lower())
s = re.sub(r'[-\s]+', '-', s).strip('-')
return s[:max_len] or "task"
# ─── Génération du patch via router v2 ───────────────────────────────────────
PROPOSAL_PROMPT = """Tu es Cortex, un assistant de développement Python qui peut modifier son propre code.
Objectif demandé par Sam :
{goal}
Contexte (fichiers pertinents) :
{context}
Format de réponse OBLIGATOIRE uniquement un bloc JSON, rien d'autre :
{{
"analysis": "explication courte du diagnostic et de l'approche",
"files": [
{{"path": "chemin/relatif/au/repo/fichier.py", "content": "CONTENU COMPLET DU FICHIER (pas un diff)"}}
],
"tests": ["tts", "router", "memory"]
}}
Règles strictes :
- Tous les chemins sont relatifs à H:\\Code\\Paperclip
- "content" doit être le fichier COMPLET (on remplace, pas un patch)
- "tests" liste les suites smoke à lancer (router|serve|memory|tts|cortex) au minimum les zones touchées
- Préserve l'encodage UTF-8 et les apostrophes françaises
- N'ajoute aucune dépendance externe sans justification
- Garde la cohérence avec le code existant (style, imports)
Réponds UNIQUEMENT avec le JSON, sans markdown."""
def _gather_context(goal: str, max_files: int = 2, max_chars_per_file: int = 3000,
total_budget: int = 7000) -> str:
"""Grep mots-clés du goal, retourne fichiers pertinents tronqués pour rester
sous total_budget chars (les free models digèrent mal au-delà de ~10K)."""
keywords = [w for w in re.findall(r'\w{4,}', goal) if w.lower() not in
{"dans", "pour", "avec", "tous", "comme", "doit", "code", "file",
"fichier", "ajouter", "ajoute", "ajout"}][:4]
if not keywords: return ""
relevant_files = {}
# Priorité absolue aux chemins explicitement cités par Sam / Cortex.
for raw_path in re.findall(r'(?:scripts|packages|server|ui|doc|docs)[\w./\\-]+\.\w+', goal):
rel = raw_path.replace("\\", "/")
if (REPO_ROOT / rel).exists():
relevant_files[rel] = 100
for kw in keywords:
r = ct.search(kw, "scripts", max_results=8)
for m in r.get("matches", []):
relevant_files[m["file"]] = relevant_files.get(m["file"], 0) + 1
top_files = sorted(relevant_files.items(), key=lambda x: -x[1])[:max_files]
parts, total = [], 0
for fname, _hits in top_files:
budget = min(max_chars_per_file, total_budget - total)
if budget < 500: break
f = ct.read_file(str(REPO_ROOT / fname), max_bytes=budget)
if f.get("ok"):
block = f"### {fname}\n```python\n{f['content']}\n```"
parts.append(block)
total += len(block)
return "\n\n".join(parts) if parts else "(pas de contexte trouvé)"
def _normalize_tests(tests, files) -> list[str]:
guardrails = _load_guardrails()
valid = set(guardrails.get("valid_tests", DEFAULT_GUARDRAILS["valid_tests"]))
aliases = guardrails.get("test_aliases", DEFAULT_GUARDRAILS["test_aliases"])
out = []
for t in tests or []:
key = aliases.get(str(t), str(t))
if key in valid and key not in out:
out.append(key)
touched = " ".join(files or [])
if "scripts/brain/" in touched and "cortex" not in out:
out.append("cortex")
return out or ["cortex"]
def _ask_router(prompt: str, timeout: int = 240) -> dict:
"""Pose la question au router v2, retourne le JSON parsé du modèle."""
try:
payload = json.dumps({"text": prompt, "role": "code"}).encode("utf-8")
req = urllib.request.Request(ROUTER_URL, data=payload,
headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read().decode())
except Exception as e:
return {"error": str(e)}
def _ask_lm_studio_direct(prompt: str, timeout: int = 180) -> dict:
"""Fallback direct quand le router v2 retourne inject=True.
Le self-dev a besoin d'une réponse modèle structurée. Un résultat
{"inject": true, "text": prompt} signifie seulement "à envoyer à Claude",
pas "patch proposé". On utilise alors le modèle local chargé dans LM Studio.
"""
try:
with urllib.request.urlopen(f"{LM_STUDIO_URL}/models", timeout=5) as r:
models = json.loads(r.read().decode()).get("data", [])
model = select_lmstudio_model(
task_type="complex_self_dev",
requested_model=os.environ.get("SELF_DEV_MODEL", get_lmstudio_config()["deep_model"]),
automatic=True,
available_models=[m.get("id", "") for m in models],
)
payload = {
"model": model,
"messages": [
{"role": "system", "content": "/no_think\nOutput only the final answer. No hidden reasoning. No markdown unless explicitly requested."},
{"role": "user", "content": "/no_think\n" + prompt},
],
"temperature": 0.1,
"max_tokens": 6000,
"stream": False,
}
payload = json.dumps(add_lmstudio_ttl(payload)).encode("utf-8")
req = urllib.request.Request(
f"{LM_STUDIO_URL}/chat/completions",
data=payload,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=timeout) as r:
d = json.loads(r.read().decode())
content = extract_lmstudio_content(d["choices"][0], expect_json=True)
return {"backend": f"lm_studio:{model}", "response": content}
except Exception as e:
return {"error": str(e)}
def _parse_proposal(response_text: str) -> dict | None:
"""Extrait le JSON de la réponse modèle (peut contenir du texte autour)."""
if not response_text: return None
# Cherche un bloc JSON {...}
for m in re.finditer(r'\{[\s\S]*\}', response_text):
candidate = m.group(0)
try:
d = json.loads(candidate)
if isinstance(d, dict) and "files" in d:
return d
except json.JSONDecodeError: continue
return None
def _parse_goal_proposal(response_text: str) -> dict | None:
"""Extrait le JSON court {goal,rationale,risk} de la boucle autonome."""
if not response_text:
return None
for m in re.finditer(r'\{[\s\S]*\}', response_text):
try:
d = json.loads(m.group(0))
except json.JSONDecodeError:
continue
if isinstance(d, dict) and d.get("goal"):
return d
return None
# ─── Application sécurisée ──────────────────────────────────────────────────
def propose_and_apply(goal: str, dry_run: bool = False) -> dict:
"""Boucle complète : propose → branche → applique → test → commit ou inspection.
Retourne un rapport structuré."""
started = time.time()
report = {"goal": goal, "started_at": dt.datetime.now().isoformat(),
"steps": [], "outcome": "pending"}
def step(name, **data):
entry = {"name": name, "ts": time.time() - started, **data}
report["steps"].append(entry)
_log(f" step: {name} {data}")
_log(f"=== self_dev: {goal[:80]} ===")
guardrails = _load_guardrails()
report["guardrails_file"] = str(GUARDRAILS_FILE)
if not guardrails.get("enabled", True):
report["outcome"] = "guardrails_disabled"
step("guardrails_disabled")
return report
# 1. Récupérer contexte pertinent
context = _gather_context(goal, max_files=int(guardrails.get("max_files_per_change", 2)),
max_chars_per_file=1800,
total_budget=int(guardrails.get("max_context_chars", 4200)))
step("context_gathered", chars=len(context))
# 2. Demander une proposition au router v2
prompt = PROPOSAL_PROMPT.format(goal=goal, context=context)
rv = _ask_router(prompt)
if "error" in rv:
report["outcome"] = "router_error"
report["error"] = rv["error"]
step("router_failed", error=rv["error"])
return report
if rv.get("inject") and not rv.get("response"):
step("router_inject_only", backend=rv.get("backend"))
rv = _ask_lm_studio_direct(prompt)
if "error" in rv:
report["outcome"] = "model_unavailable"
report["error"] = rv["error"]
step("lm_studio_failed", error=rv["error"])
return report
raw = rv.get("response") or rv.get("text") or ""
backend = rv.get("backend", "?")
step("proposal_received", backend=backend, chars=len(raw))
# 3. Parser le JSON
proposal = _parse_proposal(raw)
if not proposal:
report["outcome"] = "parse_failed"
report["raw_excerpt"] = raw[:500]
step("parse_failed", excerpt=raw[:200])
return report
files = proposal.get("files", [])
tests = _normalize_tests(proposal.get("tests", ["router", "serve", "memory"]),
[f.get("path", "") for f in files])
report["analysis"] = proposal.get("analysis", "")
report["files_planned"] = [f.get("path") for f in files]
step("proposal_parsed", files=len(files), tests=tests)
if guardrails.get("require_explicit_path_in_goal", True):
explicit_paths = set(p.replace("\\", "/") for p in re.findall(r'(?:scripts|packages|server|ui|doc|docs)[\w./\\-]+\.\w+', goal))
planned_paths = set(str(f.get("path", "")).replace("\\", "/") for f in files)
if not planned_paths or not planned_paths.issubset(explicit_paths):
report["outcome"] = "guardrail_refused"
step("guardrail_refused", reason="planned paths are not explicitly named in goal",
planned=list(planned_paths), explicit=list(explicit_paths))
return report
if dry_run:
report["outcome"] = "dry_run"
return report
# 4. Créer branche dédiée
branch = f"cortex/dev/{dt.datetime.now().strftime('%Y%m%d-%H%M%S')}-{_slugify(goal)}"
br = ct.git_branch(branch)
if not br.get("ok"):
report["outcome"] = "branch_failed"
step("branch_failed", err=br.get("stderr"))
return report
step("branch_created", branch=branch)
# 5. Appliquer les modifications
applied = []
for f in files:
path = f.get("path")
content = f.get("content", "")
if not path or not content:
continue
norm_path = path.replace("\\", "/")
allowed = guardrails.get("allowed_path_prefixes", [])
blocked = guardrails.get("blocked_path_fragments", [])
if allowed and not any(norm_path.startswith(p) for p in allowed):
step("write_refused", path=path, reason="outside allowed_path_prefixes")
continue
if any(b.lower() in norm_path.lower() for b in blocked):
step("write_refused", path=path, reason="blocked_path_fragments")
continue
# Vérification chemin sûr
try:
full_path = REPO_ROOT / path
ct._safe_path(str(full_path))
except PermissionError as e:
step("write_refused", path=path, reason=str(e))
continue
if full_path.exists():
old_size = full_path.stat().st_size
shrink = float(guardrails.get("max_shrink_ratio", 0.55))
if old_size > 1000 and len(content.encode("utf-8")) < old_size * shrink:
step("write_refused", path=path,
reason=f"proposal would shrink existing file from {old_size} to {len(content.encode('utf-8'))} bytes")
continue
w = ct.write_file(str(full_path), content)
if w.get("ok"):
applied.append(path)
step("file_written", path=path, size=w.get("size"))
else:
step("write_failed", path=path, error=w.get("error"))
report["files_applied"] = applied
if not applied:
report["outcome"] = "no_files_applied"
step("stopped_for_review", reason="no file passed guardrails")
return report
# 6. Lancer les tests smoke
test_results = {}
all_pass = True
for suite in tests:
sm = ct.run_smoke(suite, timeout=240)
test_results[suite] = sm
step("smoke_run", suite=suite, ok=sm.get("ok"),
passed=sm.get("passed"), total=sm.get("total"))
if not sm.get("ok"):
all_pass = False
report["tests"] = test_results
# 7. Commit ou rollback
if all_pass:
cm = ct.git_commit_paths(f"Cortex self-dev: {goal[:60]}", applied,
only_if_smoke_passes=False)
if cm.get("ok"):
report["outcome"] = "applied"
step("committed", branch=branch)
# Mémorise sémantiquement la compétence acquise (modulaire, indexée
# automatiquement dans le graphe → rappel par retrieve_context la
# prochaine fois qu'un goal sémantiquement proche arrive).
try:
import cortex_learned_skills as _cls
short_name = goal.strip()[:60].rstrip(".:")
rem = _cls.remember(name=short_name, goal=goal,
outcome="applied",
applied_files=applied,
tests=test_results,
tags=["self_dev", "applied"])
step("skill_remembered", ok=rem.get("ok"), path=rem.get("path"),
slug=rem.get("slug"))
except Exception as _ee:
step("skill_remember_failed", error=str(_ee))
else:
report["outcome"] = "commit_failed"
step("commit_failed", error=cm.get("stderr"))
else:
report["outcome"] = "tests_failed_left_for_review"
step("stopped_for_review", reason="smoke tests failed", files=applied)
report["duration_s"] = round(time.time() - started, 1)
_log(f"=== self_dev: outcome={report['outcome']} ({report['duration_s']}s) ===")
return report
# ─── Auto-génération de goals : boucle de curiosité ─────────────────────────
CURIOSITY_PROMPT = """Tu es Cortex, en train d'analyser ton propre fonctionnement.
Voici les statistiques de ton router v2 sur les 50 derniers échanges :
{stats}
Voici les 3 derniers échecs notables (smoke tests, parse errors, escalades vers Claude) :
{failures}
Ton objectif : proposer UN seul micro-objectif d'amélioration concret et testable.
Le but est de réduire un échec récurrent OU augmenter ta robustesse OU améliorer la qualité.
Format obligatoire UNIQUEMENT JSON :
{{
"goal": "description d'une seule action concrète (ex: 'ajouter un timeout configurable dans X', 'fix typo dans Y')",
"rationale": "pourquoi ce changement aide",
"risk": "low" ou "medium"
}}
Privilégie les tâches "low" risk pour une boucle automatique. Réponds UNIQUEMENT avec le JSON."""
def _read_recent_failures() -> str:
"""Scanne le log self-dev récent pour collecter les échecs."""
if not SELFDEV_LOG.exists(): return "(aucun échec récent)"
try:
lines = SELFDEV_LOG.read_text(encoding="utf-8", errors="replace").splitlines()[-200:]
failures = [l for l in lines if any(w in l.lower() for w in
["fail", "error", "rolled_back", "rollback", "exception"])][-5:]
return "\n".join(failures) if failures else "(aucun échec dans les logs récents)"
except Exception:
return "(impossible de lire les logs)"
def _read_v2_stats() -> str:
try:
with urllib.request.urlopen("http://127.0.0.1:18900/v2_state", timeout=3) as r:
d = json.loads(r.read().decode())
lines = [f"threshold={d.get('threshold')}"]
for k, v in d.get("models", {}).items():
lines.append(f" {k}: priority={v['priority']:.2f}, wins={v['wins']}/{v['calls']}, "
f"avg_lat={v.get('avg_latency', 0):.1f}s, "
f"cooldown={v.get('in_cooldown')}")
return "\n".join(lines)
except Exception as e:
return f"(stats unavailable: {e})"
def autonomous_iteration(risk_threshold: str = "low") -> dict:
"""Une itération autonome : génère un goal, le filtre par risque, applique.
Retourne le rapport. risk_threshold: 'low' refuse medium/high."""
_log("=== autonomous iteration start ===")
stats = _read_v2_stats()
failures = _read_recent_failures()
prompt = CURIOSITY_PROMPT.format(stats=stats, failures=failures)
rv = _ask_router(prompt)
raw = rv.get("response") or rv.get("text") or ""
proposal = _parse_goal_proposal(raw) or {}
goal = proposal.get("goal", "")
risk = proposal.get("risk", "medium")
if not goal:
return {"outcome": "no_goal_generated", "raw": raw[:300]}
if risk_threshold == "low" and risk != "low":
return {"outcome": "risk_too_high", "goal": goal, "risk": risk,
"rationale": proposal.get("rationale")}
_log(f" autonomous goal: {goal} (risk={risk})")
return propose_and_apply(goal, dry_run=False)
# ─── CLI ─────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
if len(sys.argv) < 2:
print("Usage:\n cortex_self_dev.py '<goal>' [--dry-run]\n cortex_self_dev.py --autonomous [low|medium]")
sys.exit(1)
if sys.argv[1] == "--autonomous":
risk = sys.argv[2] if len(sys.argv) > 2 else "low"
rep = autonomous_iteration(risk_threshold=risk)
else:
goal = sys.argv[1]
dry = "--dry-run" in sys.argv
rep = propose_and_apply(goal, dry_run=dry)
print("\n=== RAPPORT ===")
print(json.dumps(rep, ensure_ascii=False, indent=2)[:5000])

View File

@ -0,0 +1,70 @@
@echo off
REM ─────────────────────────────────────────────────────────────────────
REM Cortex — lanceur serveur dashboard avec auto-restart sur crash.
REM À déposer sur le bureau ou via raccourci .lnk.
REM ─────────────────────────────────────────────────────────────────────
title Cortex serveur (auto-restart)
color 0A
cd /d h:\Code\Paperclip
REM LM Studio policy for Cortex auto tasks
set LMSTUDIO_BASE_URL=http://127.0.0.1:1234
set LMSTUDIO_FAST_MODEL=qwen2.5-7b-instruct
set LMSTUDIO_DEEP_MODEL=qwen3.6-35b-a3b
set LMSTUDIO_EMBED_MODEL=text-embedding-nomic-embed-text-v1.5
set LMSTUDIO_TTL=300
set LMSTUDIO_ALLOW_DEEP_AUTO=0
set LMSTUDIO_USE_NATIVE_API=0
set LMSTUDIO_JIT_ENABLED=0
set LM_STUDIO_EXE=G:\Lmstudio\LM Studio\LM Studio.exe
set LM_STUDIO_MODELS_URL=http://127.0.0.1:1234/v1/models
set CHROME_EXE=C:\Program Files\Google\Chrome\Application\chrome.exe
set CORTEX_CHROME_PROFILE=%USERPROFILE%\.paperclip\chrome-cortex
set CORTEX_URL=http://127.0.0.1:8765/
set CORTEX_GPU_URL=http://127.0.0.1:8765/gpu
echo Verification LM Studio...
curl.exe %LM_STUDIO_MODELS_URL% --max-time 2 >nul 2>&1
if errorlevel 1 (
echo LM Studio non detecte - lancement...
if exist "%LM_STUDIO_EXE%" (
start "" "%LM_STUDIO_EXE%"
timeout /t 8 /nobreak >nul
) else (
echo ATTENTION: LM Studio.exe introuvable: %LM_STUDIO_EXE%
)
)
REM Tue les listeners port 8765 résiduels avant de démarrer (évite "port busy")
for /f "tokens=5" %%P in ('netstat -ano ^| findstr "LISTENING" ^| findstr ":8765 "') do (
echo Cleanup ancien listener pid %%P...
taskkill /F /PID %%P >nul 2>&1
)
set RESTART_COUNT=0
:loop
echo.
echo [%date% %time%] Cortex demarre (run #%RESTART_COUNT%)...
echo URL : http://localhost:8765/gpu
echo.
if "%RESTART_COUNT%"=="0" (
if exist "%CHROME_EXE%" (
echo Preparation Chrome...
if not exist "%CORTEX_CHROME_PROFILE%" mkdir "%CORTEX_CHROME_PROFILE%" >nul 2>&1
start "" cmd /c "timeout /t 6 /nobreak >nul && \"%CHROME_EXE%\" --user-data-dir=\"%CORTEX_CHROME_PROFILE%\" --new-window \"%CORTEX_URL%\" \"%CORTEX_GPU_URL%\""
) else (
echo Chrome introuvable: %CHROME_EXE%
)
)
python scripts\brain\dashboard\serve.py
set EXITCODE=%errorlevel%
set /a RESTART_COUNT+=1
echo.
echo [%date% %time%] Cortex termine (code %EXITCODE%) - relance dans 3s...
echo (Ctrl+C dans les 3 secondes pour stopper definitivement)
echo.
timeout /t 3 /nobreak >nul
goto loop

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

808
scripts/brain/llm_router.py Normal file
View File

@ -0,0 +1,808 @@
"""
llm_router.py Router LLM intelligent.
Reçoit les requêtes et les route vers le bon modèle selon la disponibilité.
Port : 19000
API : POST /route {"text": "...", "context": "voice|code|analysis"}
GET /status modèle actif, usage, backends disponibles
Backends prioritaires :
1. Claude (Max) via VS Code inject
2. GPT-4/5 via OpenAI API (si clé dispo)
3. qwen local via LM Studio (http://localhost:1234)
4. ollama via http://localhost:11434
"""
try:
from lmstudio_response import extract_lmstudio_content
except Exception:
from scripts.brain.lmstudio_response import extract_lmstudio_content
try:
from lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
except Exception:
from scripts.brain.lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
import json, os, re, socket, sys, time, threading, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
_LOCK = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
_LOCK.bind(("127.0.0.1", 19000))
except OSError:
print("[llm_router] déjà en cours — quitte.")
sys.exit(0)
PORT = 18900
MODEL_INFO = {
"claude": {"name": "Claude Sonnet 4.6", "cost": "Max sub", "iq": 90, "type": "subscription"},
"codex": {"name": "GPT-5.5", "cost": "OAI sub", "iq": 95, "type": "subscription"},
"lm_studio": {"name": "qwen3.6-35b", "cost": "free/local","iq": 72, "type": "local"},
"ollama": {"name": "Ollama local", "cost": "free/local","iq": 65, "type": "local"},
"big_pickle": {"name": "Big Pickle", "cost": "free", "iq": 70, "type": "opencode"},
"minimax_m2.5": {"name": "Minimax M2.5", "cost": "free", "iq": 75, "type": "opencode"},
"gpt_5_nano": {"name": "GPT-5 nano", "cost": "free", "iq": 65, "type": "opencode"},
"hy3_preview": {"name": "HY3 Preview", "cost": "free", "iq": 70, "type": "opencode"},
"nemotron_3_super": {"name": "Nemotron 3 Super", "cost": "free", "iq": 75, "type": "opencode"},
}
VAULT = Path(r"C:\Users\Smedj\Documents\Obsidian Vault")
COOKIES_FILE = Path.home() / ".claude" / ".claude-cookies.json"
ORG_UUID = "952c1bc7-5fd1-4f7c-83db-a020932db2ab"
LM_STUDIO = get_lmstudio_config()["base_url"]
OLLAMA = "http://localhost:11434"
# ─── État global ──────────────────────────────────────────────────────────────
_state = {
"active_backend": "claude",
"usage": {},
"backends": {},
"request_count": 0,
"last_route": None,
}
_lock = threading.Lock()
# ─── Sonde backends ───────────────────────────────────────────────────────────
def probe_claude_usage() -> dict:
"""Lit le quota Claude Max."""
try:
cookies = json.loads(COOKIES_FILE.read_text()) if COOKIES_FILE.exists() else {}
sk = cookies.get("sessionKey", "")
if not sk:
return {}
req = urllib.request.Request(
f"https://claude.ai/api/organizations/{ORG_UUID}/usage",
headers={"Cookie": f"sessionKey={sk}", "User-Agent": "Mozilla/5.0",
"Accept": "application/json", "Referer": "https://claude.ai/settings/usage",
"sec-fetch-site": "same-origin", "sec-fetch-mode": "cors"}
)
with urllib.request.urlopen(req, timeout=6) as r:
return json.loads(r.read().decode())
except Exception:
return {}
def probe_lm_studio() -> dict:
"""Vérifie si LM Studio est accessible et quel modèle est chargé."""
try:
req = urllib.request.Request(f"{LM_STUDIO}/v1/models")
with urllib.request.urlopen(req, timeout=2) as r:
d = json.loads(r.read().decode())
models = [m["id"] for m in d.get("data", [])]
return {"available": True, "models": models}
except Exception:
return {"available": False, "models": []}
def probe_ollama() -> dict:
"""Vérifie si Ollama est accessible."""
try:
req = urllib.request.Request(f"{OLLAMA}/api/tags")
with urllib.request.urlopen(req, timeout=2) as r:
d = json.loads(r.read().decode())
models = [m["name"] for m in d.get("models", [])]
return {"available": True, "models": models}
except Exception:
return {"available": False, "models": []}
def probe_codex() -> dict:
"""Vérifie si Codex CLI est disponible."""
import shutil
if shutil.which("codex"):
return {"available": True, "model": "gpt-5.5", "type": "subscription"}
return {"available": False}
_claude_rate_limited = False # True dès qu'une erreur rate-limit est reçue
_rate_limit_until = 0.0 # timestamp de fin de blocage estimé
def report_rate_limit(backend: str, reset_seconds: int = 3600):
"""Appelé quand Claude retourne une erreur 429/rate-limit."""
global _claude_rate_limited, _rate_limit_until
if backend == "claude":
_claude_rate_limited = True
_rate_limit_until = time.time() + reset_seconds
print(f"[router] ⚠ Claude rate-limité — switch automatique", flush=True)
def decide_backend(usage: dict) -> str:
"""Claude jusqu'à l'erreur réelle, puis Codex, puis qwen."""
global _claude_rate_limited, _rate_limit_until
# Lever le flag si le délai est passé
if _claude_rate_limited and time.time() > _rate_limit_until:
_claude_rate_limited = False
print("[router] Claude rate-limit expiré — retour Claude", flush=True)
if not _claude_rate_limited:
return "claude"
# Claude bloqué → modèles gratuits OpenCode d'abord (benchmark: big_pickle > minimax > codex)
for model_id in ["big_pickle", "minimax_m2.5", "gpt_5_nano", "hy3_preview", "nemotron_3_super"]:
m = _state["backends"].get(model_id, {})
if m.get("available"):
print(f"[router] → {model_id} (gratuit OpenCode)", flush=True)
return model_id
# Codex en dernier recours (abonnement payant)
cdx = _state["backends"].get("codex", {})
if cdx.get("available"):
print("[router] → Codex gpt-5.5 (dernier recours payant)", flush=True)
return "codex"
# LM Studio local
lms = _state["backends"].get("lm_studio", {})
if lms.get("available") and lms.get("models"):
print("[router] → LM Studio qwen", flush=True)
return "lm_studio"
return "claude"
def route_to_codex(text: str) -> str:
"""Appelle Codex CLI gpt-5.5 via subprocess (mode non-interactif)."""
result = subprocess.run(
["codex", "exec", "--model", "gpt-5.5", text],
capture_output=True, text=True, timeout=120,
encoding="utf-8", errors="replace"
)
out = result.stdout.strip()
# Codex ajoute du bruit (timestamps, token counts) — extraire la vraie réponse
lines = [l for l in out.splitlines() if l and not l.startswith("2026") and "tokens used" not in l]
return "\n".join(lines).strip() or result.stderr.strip()
OPENCODE_FREE = {
"big_pickle": "opencode/big-pickle",
"minimax_m2.5": "opencode/minimax-m2.5-free",
"gpt_5_nano": "opencode/gpt-5-nano",
"hy3_preview": "opencode/hy3-preview-free",
"nemotron_3_super": "opencode/nemotron-3-super-free",
}
OPENCODE_CMD = r"C:\Users\Smedj\AppData\Roaming\npm\opencode.cmd"
def probe_opencode_models() -> dict:
"""Vérifie si opencode est disponible."""
available = Path(OPENCODE_CMD).exists()
if available:
return {k: {"available": True, "model": v, "cost": "free"} for k, v in OPENCODE_FREE.items()}
return {k: {"available": False} for k in OPENCODE_FREE}
def probe_all():
"""Sonde périodique de tous les backends."""
while True:
usage = probe_claude_usage()
lms = probe_lm_studio()
cdx = probe_codex()
oc = probe_opencode_models()
with _lock:
_state["usage"] = usage
_state["backends"] = {"codex": cdx, "lm_studio": lms, **oc}
_state["active_backend"] = decide_backend(usage)
time.sleep(60)
def auto_benchmark_loop():
"""Auto-test périodique pour mettre à jour les pondérations dynamiques.
Tourne toutes les heures avec un set de questions standard."""
BENCH_QUESTIONS = [
"Quelle est la capitale du Japon ?",
"Combien font 17 fois 23 ?",
"Cite trois langages de programmation modernes.",
"Qu'est-ce que la quantification d'un modèle LLM ?",
"En une phrase, c'est quoi le RAG ?",
]
# Attendre 5 min après démarrage avant 1er benchmark
time.sleep(300)
while True:
try:
print(f"[v2 auto-bench] début round", flush=True)
for q in BENCH_QUESTIONS:
try:
route_v2(q)
time.sleep(10) # respiration entre questions
except Exception as e:
print(f"[v2 auto-bench] err: {e}", flush=True)
with _runtime_lock:
stats = {k: dict(v) for k, v in _model_runtime.items()}
print(f"[v2 auto-bench] fin — stats: {stats}", flush=True)
except Exception as e:
print(f"[v2 auto-bench] crash loop: {e}", flush=True)
time.sleep(3600) # 1h entre rounds
# ─── Routing requêtes ─────────────────────────────────────────────────────────
def route_to_local(text: str, backend: str) -> str:
"""Envoie une requête à un backend local (LM Studio ou Ollama)."""
if backend == "lm_studio":
lms = _state["backends"].get("lm_studio", {})
model = select_lmstudio_model(
task_type="short",
automatic=True,
available_models=lms.get("models", []),
)
url = f"{LM_STUDIO}/v1/chat/completions"
payload = {
"model": model,
"messages": [
{"role": "system", "content": "/no_think\nAnswer directly. Do not expose reasoning."},
{"role": "user", "content": "/no_think\n" + text},
],
"temperature": 0.7,
"max_tokens": 800,
"stream": False,
}
payload = json.dumps(add_lmstudio_ttl(payload)).encode("utf-8")
req = urllib.request.Request(url, data=payload,
headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=120) as r:
d = json.loads(r.read().decode())
return extract_lmstudio_content(d["choices"][0], expect_json=False)
elif backend == "ollama":
oll = _state["backends"].get("ollama", {})
model = oll.get("models", ["llama3.2"])[0]
url = f"{OLLAMA}/api/generate"
payload = json.dumps({"model": model, "prompt": text, "stream": False}).encode()
req = urllib.request.Request(url, data=payload,
headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=120) as r:
d = json.loads(r.read().decode())
return d.get("response", "")
return ""
# ─── v2 : parallèle + juge + cascade FrugalGPT ────────────────────────────────
import concurrent.futures, subprocess, difflib
JUDGE_THRESHOLD_BASE = 7.0 # score min de base
SELF_CONSISTENCY_K = 0.65 # ratio Jaccard pour consensus
COOLDOWN_FAILS = 3 # N échecs consécutifs → cooldown du modèle
COOLDOWN_DURATION = 1800 # 30 min de cooldown
BENCHMARK_LOG = Path(r"C:\Users\Smedj\Documents\Obsidian Vault\.vault-llm-benchmark-iag.json")
# État dynamique des modèles
_model_runtime = {
k: {"consecutive_fails": 0, "cooldown_until": 0.0, "wins": 0, "calls": 0, "avg_latency": 0.0}
for k in OPENCODE_FREE
}
_runtime_lock = threading.Lock()
def _is_in_cooldown(model_key: str) -> bool:
with _runtime_lock:
return _model_runtime[model_key]["cooldown_until"] > time.time()
def _record_call(model_key: str, success: bool, latency: float, won: bool = False):
with _runtime_lock:
s = _model_runtime[model_key]
s["calls"] += 1
# Moyenne mobile latence
s["avg_latency"] = 0.8 * s["avg_latency"] + 0.2 * latency if s["avg_latency"] else latency
if success:
s["consecutive_fails"] = 0
else:
s["consecutive_fails"] += 1
if s["consecutive_fails"] >= COOLDOWN_FAILS:
s["cooldown_until"] = time.time() + COOLDOWN_DURATION
print(f"[v2] {model_key} en cooldown {COOLDOWN_DURATION//60}min", flush=True)
if won:
s["wins"] += 1
def _model_priority(model_key: str) -> float:
"""Score de priorité : winrate * 0.7 + (1 - latency_norm) * 0.3."""
with _runtime_lock:
s = _model_runtime[model_key]
if s["calls"] < 3:
return 0.5 # neutre tant qu'on n'a pas de données
winrate = s["wins"] / s["calls"]
# Latency normalisée vs moyenne globale (10s baseline)
latency_score = max(0, 1 - (s["avg_latency"] / 30.0))
return 0.7 * winrate + 0.3 * latency_score
def _adaptive_threshold() -> float:
"""Seuil du juge ajusté selon quota Claude : si quota élevé, plus permissif (économise)."""
usage = _state.get("usage", {})
five_h = (usage.get("five_hour") or {}).get("utilization", 50)
seven_d = (usage.get("seven_day") or {}).get("utilization", 50)
pressure = max(five_h, seven_d)
# > 80% : très permissif (6.0), <30% : exigeant (8.0), milieu : 7.0
if pressure >= 80: return 6.0
if pressure >= 60: return 6.5
if pressure <= 30: return 8.0
return JUDGE_THRESHOLD_BASE
def _is_simple_question(text: str) -> bool:
"""Heuristique : courte, factuelle, pas de code, pas de demande complexe."""
t = text.strip()
if len(t) > 150: return False
if re.search(r'```|def |class |function|import |const |let |var |implement|architecture', t, re.I):
return False
if re.search(r'(comment|pourquoi|explique|détaille|analyse|compare|liste|résume)', t, re.I):
return False
# Question factuelle simple : "Quelle...?", "Qui...?", "Combien...?", "Quand...?"
if re.search(r'\b(quel|qui|combien|quand|où|c\'est quoi|how|what|who|when|where)\b', t, re.I):
return True
return False
_opencode_semaphore = threading.Semaphore(2) # max 2 opencode concurrents
def _call_opencode(model_key: str, text: str, timeout: int = 60) -> tuple[str, str | None, float]:
"""Appelle opencode avec prompt via stdin. Sémaphore global évite la saturation."""
t0 = time.time()
if not _opencode_semaphore.acquire(timeout=timeout):
return (model_key, None, time.time() - t0)
try:
model_id = OPENCODE_FREE[model_key]
r = subprocess.run(
[OPENCODE_CMD, "run", "--model", model_id, "-"], # "-" = lire stdin
input=text,
capture_output=True, text=True, timeout=timeout,
encoding="utf-8", errors="replace"
)
lines = [l for l in r.stdout.splitlines()
if l.strip() and not l.startswith(">") and "\x1b" not in l and "build" not in l.lower()]
response = "\n".join(lines).strip()
if not response:
# Fallback : si stdin pas supporté par cette version, retomber sur arg
r2 = subprocess.run(
[OPENCODE_CMD, "run", "--model", model_id, text],
capture_output=True, text=True, timeout=timeout,
encoding="utf-8", errors="replace"
)
lines = [l for l in r2.stdout.splitlines()
if l.strip() and not l.startswith(">") and "\x1b" not in l and "build" not in l.lower()]
response = "\n".join(lines).strip()
if not response:
return (model_key, None, time.time() - t0)
return (model_key, response, time.time() - t0)
except Exception as e:
print(f"[v2] {model_key} err: {e}", flush=True)
return (model_key, None, time.time() - t0)
finally:
try: _opencode_semaphore.release()
except: pass
def _call_lm_studio(prompt: str, timeout: int = 60, max_tokens: int = 800) -> str | None:
try:
lms = _state["backends"].get("lm_studio", {})
if not lms.get("available"): return None
model = select_lmstudio_model(
task_type="eval",
automatic=True,
available_models=lms.get("models", []),
)
url = f"{LM_STUDIO}/v1/chat/completions"
payload = {
"model": model,
"messages": [
{"role": "system", "content": "/no_think\nOutput only the final answer. No hidden reasoning. No markdown unless explicitly requested."},
{"role": "user", "content": "/no_think\n" + prompt},
],
"temperature": 0.3, "max_tokens": max_tokens, "stream": False,
}
payload = json.dumps(add_lmstudio_ttl(payload)).encode()
req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=timeout) as r:
return extract_lmstudio_content(json.loads(r.read().decode())["choices"][0], expect_json=False)
except Exception as e:
print(f"[v2] lm_studio err: {e}", flush=True)
return None
def _normalize(text: str) -> set:
"""Tokenise pour Jaccard similarity."""
return set(re.findall(r'\w+', text.lower())) if text else set()
def _jaccard(a: set, b: set) -> float:
if not a or not b: return 0.0
return len(a & b) / len(a | b)
def _self_consistency(responses: dict) -> str | None:
"""Si plusieurs modèles donnent une réponse similaire, retourne celle-là.
responses: {model_key: response_text}"""
valid = [(k, v) for k, v in responses.items() if v]
if len(valid) < 2: return None
norms = {k: _normalize(v) for k, v in valid}
# Compter pour chaque réponse combien d'autres sont similaires
best_key, best_count = None, 0
for k1, n1 in norms.items():
count = sum(1 for k2, n2 in norms.items() if k1 != k2 and _jaccard(n1, n2) >= SELF_CONSISTENCY_K)
if count > best_count:
best_count, best_key = count, k1
# Consensus si au moins 2 modèles agree (donc count >= 1 pour le best)
if best_count >= 1:
return best_key
return None
def _extract_scores(raw: str, n_letters: int) -> dict:
"""Extrait robustement {letter: score} depuis n'importe quel format de juge LLM.
Cherche d'abord JSON, puis patterns markdown / table / inline."""
if not raw: return {}
expected = [chr(65 + i) for i in range(n_letters)]
result = {}
# Tentative 1 : JSON valide
for m in re.finditer(r'\{[^{}]+\}', raw):
try:
d = json.loads(m.group(0))
for k, v in d.items():
kk = k.strip().upper()[:1]
if kk in expected:
try: result[kk] = float(v)
except: pass
if result: return result
except: continue
# Tentative 2 : patterns "A: 8", "A → 8", "A | 8", "A**: **8/10", "A) 7"
for letter in expected:
# Patterns variés
patterns = [
rf'\*?\*?{letter}\*?\*?\s*[:→\-=)|]+\s*\*?\*?(\d+(?:\.\d+)?)\s*\*?\*?(?:\s*/\s*10)?',
rf'(?:^|\W){letter}\s*[:|]\s*(\d+(?:\.\d+)?)',
rf'Réponse\s+{letter}\s*[:\-]?\s*\*?\*?(\d+(?:\.\d+)?)',
]
for pat in patterns:
m = re.search(pat, raw, re.MULTILINE | re.IGNORECASE)
if m:
try:
score = float(m.group(1))
if 0 <= score <= 10:
result[letter] = score
break
except: pass
return result
def _judge_one(judge_key: str, question: str, shuffled_responses: list, judge_call_fn) -> dict:
"""Un juge note les réponses anonymisées. Retourne {letter: score}."""
items = "\n\n".join(f"### Réponse {chr(65+i)}\n{v}" for i, (_, v) in enumerate(shuffled_responses))
prompt = (
f"Tu es un juge. Note chaque réponse de 0 à 10 selon : exactitude factuelle, pertinence, clarté concise. "
f"Ne récompense pas la verbosité.\n\n"
f"## Question\n{question}\n\n"
f"## Réponses anonymes\n{items}\n\n"
f"Format strict (1 ligne par réponse) :\n"
f"A: 8\nB: 6\nC: 9\n..."
)
try:
raw = judge_call_fn(prompt)
if not raw: return {}
scores = _extract_scores(raw, len(shuffled_responses))
if not scores:
print(f"[v2] judge {judge_key} no scores parsed, raw: {raw[:200]!r}", flush=True)
return scores
except Exception as e:
print(f"[v2] judge {judge_key} err: {e}", flush=True)
return {}
def _panel_judge(question: str, responses: dict, include_claude: bool = False) -> dict:
"""Conglomérat de juges anonymisé. Tronque candidats à 300 chars, limite à 3 juges rapides + qwen."""
valid = [(k, v) for k, v in responses.items() if v]
if len(valid) < 2: return {}
import random as _rnd
# Tronquer chaque candidat à 300 chars pour réduire taille prompt
truncated = [(k, (v[:300] + ("..." if len(v) > 300 else ""))) for k, v in valid]
shuffled = truncated.copy(); _rnd.shuffle(shuffled)
letter_to_key = {chr(65 + i): k for i, (k, _) in enumerate(shuffled)}
# Panel : qwen externe (si LM Studio actif) + top-2 free models
free_judges = sorted(responses.keys(), key=_model_priority, reverse=True)[:2]
lms_state = _state["backends"].get("lm_studio", {})
if lms_state.get("available"):
judges = ["qwen_external"] + free_judges
else:
judges = free_judges # skip qwen si LM Studio down — évite timeout
all_judgments = {}
with concurrent.futures.ThreadPoolExecutor(max_workers=len(judges)) as ex:
futures = {}
for judge in judges:
if judge == "qwen_external":
fn = lambda p: _call_lm_studio(p, max_tokens=150, timeout=30)
else:
fn = lambda p, jk=judge: _call_opencode(jk, p, timeout=30)[1]
futures[ex.submit(_judge_one, judge, question, shuffled, fn)] = judge
for f in concurrent.futures.as_completed(futures, timeout=60):
judge = futures[f]
try:
scores = f.result()
all_judgments[judge] = scores
if scores:
print(f"[v2 panel] {judge}{scores}", flush=True)
except Exception as e:
all_judgments[judge] = {}
print(f"[v2 panel] {judge} crash: {e}", flush=True)
# Aggréger anti-auto-favoritisme
aggregated = {k: [] for k, _ in valid}
for judge, scores in all_judgments.items():
for letter, score in scores.items():
target_key = letter_to_key.get(letter)
if not target_key or judge == target_key: continue
aggregated[target_key].append(score)
final = {k: sum(v)/len(v) for k, v in aggregated.items() if v}
print(f"[v2 panel] {len(judges)} juges, agrégé: {final}", flush=True)
return final
# Alias rétro-compatibilité
def _judge_responses(question: str, responses: dict) -> dict:
return _panel_judge(question, responses)
def _log_v2_round(question: str, responses: dict, latencies: dict, scores: dict, winner: str):
"""Log chaque round dans .vault-llm-benchmark-iag.json pour benchmark continu."""
import datetime as _dt
log_file = Path(r"C:\Users\Smedj\Documents\Obsidian Vault\.vault-llm-benchmark-iag.json")
try:
existing = json.loads(log_file.read_text(encoding="utf-8")) if log_file.exists() else {"rounds": []}
if not isinstance(existing, dict) or "rounds" not in existing: existing = {"rounds": []}
existing["rounds"].append({
"ts": _dt.datetime.now().isoformat(),
"question": question[:200],
"responses": {k: (v[:300] if v else None) for k, v in responses.items()},
"latencies": latencies,
"scores": scores,
"winner": winner,
})
# Garder les 200 derniers rounds
existing["rounds"] = existing["rounds"][-200:]
log_file.write_text(json.dumps(existing, ensure_ascii=False, indent=2), encoding="utf-8")
except Exception as e:
print(f"[v2] log err: {e}", flush=True)
def _record_winner(winner: str, path: str, scores: dict | None = None):
"""Met à jour l'état v2 visible dans /status pour l'UI."""
with _lock:
_state["v2_last_winner"] = winner
_state["v2_last_path"] = path
_state["v2_last_scores"] = scores or {}
_state["v2_last_ts"] = time.time()
def route_v2(text: str) -> dict:
"""Pipeline v2 intelligent :
- Skip cooldowned models
- Question simple top-priority model only (1 appel, pas de juge)
- Question complexe parallèle + self-consistency + juge adaptatif + cascade
"""
# Filtrer les modèles en cooldown
available = [k for k in OPENCODE_FREE
if _state["backends"].get(k, {}).get("available") and not _is_in_cooldown(k)]
if not available:
return {"backend": "claude", "inject": True, "text": text, "v2_path": "no_free_available"}
# Trier par priorité dynamique (winrate + latency)
available.sort(key=_model_priority, reverse=True)
threshold = _adaptive_threshold()
simple = _is_simple_question(text)
# ─── Question simple : un seul modèle (le top-priority) ─────────────────
if simple:
top = available[0]
print(f"[v2] simple Q → solo {top} (threshold={threshold})", flush=True)
_, resp, lat = _call_opencode(top, text, timeout=30)
if resp:
_record_call(top, success=True, latency=lat, won=True)
_log_v2_round(text, {top: resp}, {top: lat}, {}, top)
_record_winner(top, "simple_solo")
return {"backend": top, "inject": False, "response": resp,
"v2_path": "simple_solo", "model_priority": _model_priority(top)}
_record_call(top, success=False, latency=lat)
# Échec solo → retombe sur parallèle
print(f"[v2] solo {top} failed → fallback parallel", flush=True)
# ─── Question complexe : parallèle ───────────────────────────────────────
print(f"[v2] parallel → {available} (threshold={threshold})", flush=True)
responses, latencies = {}, {}
with concurrent.futures.ThreadPoolExecutor(max_workers=len(available)) as ex:
futures = {ex.submit(_call_opencode, k, text): k for k in available}
try:
for f in concurrent.futures.as_completed(futures, timeout=70):
k, resp, lat = f.result()
responses[k] = resp; latencies[k] = lat
_record_call(k, success=resp is not None, latency=lat)
except concurrent.futures.TimeoutError:
print(f"[v2] timeout global, on prend ce qu'on a", flush=True)
valid = {k: v for k, v in responses.items() if v}
print(f"[v2] {len(valid)}/{len(available)} valides", flush=True)
if not valid:
return {"backend": "claude", "inject": True, "text": text, "v2_path": "all_free_failed"}
# Self-consistency
consensus = _self_consistency(valid)
if consensus:
print(f"[v2] consensus → {consensus}", flush=True)
_record_call(consensus, True, latencies.get(consensus, 0), won=True)
_log_v2_round(text, responses, latencies, {}, consensus)
_record_winner(consensus, "consensus")
return {"backend": consensus, "inject": False, "response": valid[consensus],
"v2_path": "consensus", "candidates": list(valid.keys())}
# Juge avec seuil adaptatif
scores = _judge_responses(text, valid)
if scores:
winner_key, winner_score = max(scores.items(), key=lambda x: x[1])
print(f"[v2] judge scores={scores} winner={winner_key}@{winner_score} threshold={threshold}", flush=True)
if winner_score >= threshold:
_record_call(winner_key, True, latencies.get(winner_key, 0), won=True)
_record_call(winner_key, True, latencies.get(winner_key, 0), won=True)
_log_v2_round(text, responses, latencies, scores, winner_key)
_record_winner(winner_key, "judge_pass", scores)
return {"backend": winner_key, "inject": False, "response": valid[winner_key],
"v2_path": "judge_pass", "score": winner_score, "all_scores": scores,
"threshold": threshold}
# PAS de cascade Claude : Sam préfère économiser le quota (Claude Code l'utilise déjà).
# On va direct sur lm_studio (qwen) puis fallback best free.
# Codex fallback
cdx = _state["backends"].get("codex", {})
if cdx.get("available"):
try:
resp = route_to_codex(text)
_log_v2_round(text, responses, latencies, scores, "codex")
_record_winner("codex", "escalate_codex", scores)
return {"backend": "codex", "inject": False, "response": resp, "v2_path": "escalate_codex"}
except Exception: pass
# Dernier recours : meilleur free même si score bas
best_free = max(valid.keys(), key=lambda k: scores.get(k, 0))
_record_call(best_free, True, latencies.get(best_free, 0), won=True)
_log_v2_round(text, responses, latencies, scores, best_free)
_record_winner(best_free, "fallback_best_free", scores)
return {"backend": best_free, "inject": False, "response": valid[best_free],
"v2_path": "fallback_best_free", "warning": "no flagship available"}
# ─── HTTP Handler ─────────────────────────────────────────────────────────────
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/v2_state":
# Snapshot d'abord (lock court), priorité calculée après (sans lock pour éviter deadlock)
with _runtime_lock:
snapshot = {k: dict(v) for k, v in _model_runtime.items()}
stats = {}
for k, v in snapshot.items():
prio = _model_priority(k) # acquiert le lock à nouveau, OK car sorti
stats[k] = {**v, "in_cooldown": v["cooldown_until"] > time.time(),
"priority": prio}
payload = {
"models": stats,
"threshold": _adaptive_threshold(),
"claude_rate_limited": _claude_rate_limited,
"self_consistency_k": SELF_CONSISTENCY_K,
}
data = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Content-Length", str(len(data)))
self.end_headers(); self.wfile.write(data)
return
if self.path == "/status":
with _lock:
data = json.dumps({**_state, "models": MODEL_INFO, "claude_rate_limited": _claude_rate_limited}, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
else:
self.send_error(404)
def do_POST(self):
if self.path == "/route_v2":
length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(length).decode("utf-8-sig"))
text = body.get("text", "")
try:
result = route_v2(text)
except Exception as e:
print(f"[v2] err: {e}", flush=True)
result = {"backend": "claude", "inject": True, "text": text, "error": str(e)}
data = json.dumps(result, ensure_ascii=False).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers(); self.wfile.write(data)
return
if self.path == "/route":
length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(length).decode("utf-8-sig"))
text = body.get("text", "")
context = body.get("context", "voice")
with _lock:
backend = _state["active_backend"]
_state["request_count"] += 1
_state["last_route"] = {"backend": backend, "context": context, "ts": time.time()}
print(f"[router] → {backend} | {context} | {text[:50]!r}", flush=True)
if backend == "claude":
result = {"backend": "claude", "inject": True, "text": text}
elif backend in OPENCODE_FREE:
try:
model_id = OPENCODE_FREE[backend]
r = subprocess.run(
[OPENCODE_CMD, "run", "--model", model_id, text],
capture_output=True, text=True, timeout=60,
encoding="utf-8", errors="replace"
)
lines = [l for l in r.stdout.splitlines()
if l.strip() and not l.startswith(">") and "\x1b" not in l and "build" not in l.lower()]
response = "\n".join(lines).strip() or "Pas de réponse"
result = {"backend": backend, "inject": False, "response": response}
except Exception as e:
result = {"backend": "claude", "inject": True, "text": text, "fallback": True}
elif backend == "codex":
try:
response = route_to_codex(text)
result = {"backend": "codex", "inject": False, "response": response}
except Exception as e:
print(f"[router] ✗ codex: {e} → fallback claude", flush=True)
result = {"backend": "claude", "inject": True, "text": text, "fallback": True}
else:
try:
response = route_to_local(text, backend)
result = {"backend": backend, "inject": False, "response": response}
except Exception as e:
print(f"[router] ✗ {backend}: {e} → fallback claude", flush=True)
result = {"backend": "claude", "inject": True, "text": text, "fallback": True}
data = json.dumps(result, ensure_ascii=False).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
else:
self.send_error(404)
def log_message(self, fmt, *args):
pass # Silence les logs HTTP
# ─── Main ─────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== LLM Router ===", flush=True)
# Init immédiate des backends sans attendre le thread
with _lock:
_state["backends"] = {**probe_opencode_models(), "codex": probe_codex(), "lm_studio": {}, "ollama": {}}
threading.Thread(target=probe_all, daemon=True).start()
threading.Thread(target=auto_benchmark_loop, daemon=True).start()
with _lock:
print(f"[router] Backend actif : {_state['active_backend']}", flush=True)
free = [k for k,v in _state['backends'].items() if v.get('available') and v.get('cost')=='free']
print(f"[router] Modèles gratuits : {free}", flush=True)
print(f"[router] Écoute sur port {PORT}", flush=True)
server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()

View File

@ -0,0 +1,110 @@
import os
FAST_TASKS = {
"json",
"extract",
"classify",
"memory",
"vault",
"consolidate",
"synthesize_short",
"tooltip",
"short",
"eval",
"healthcheck",
}
DEEP_TASKS = {
"deep_reason",
"long_synthesis",
"manual_chat",
"explicit_deep",
"complex_self_dev",
}
def _env_flag(name: str, default: str = "0") -> bool:
return str(os.getenv(name, default)).strip().lower() in {"1", "true", "yes", "on"}
def normalize_model_id(model_id):
model_id = (model_id or "").strip()
if ":" in model_id:
base, suffix = model_id.rsplit(":", 1)
if suffix.isdigit():
return base
return model_id
def get_lmstudio_config():
return {
"base_url": os.getenv("LMSTUDIO_BASE_URL", "http://127.0.0.1:1234").rstrip("/"),
"fast_model": os.getenv("LMSTUDIO_FAST_MODEL", "qwen2.5-7b-instruct"),
"deep_model": os.getenv("LMSTUDIO_DEEP_MODEL", "qwen3.6-35b-a3b"),
"embed_model": os.getenv("LMSTUDIO_EMBED_MODEL", "text-embedding-nomic-embed-text-v1.5"),
"ttl": int(os.getenv("LMSTUDIO_TTL", "300")),
"allow_deep_auto": _env_flag("LMSTUDIO_ALLOW_DEEP_AUTO", "0"),
"use_native_api": _env_flag("LMSTUDIO_USE_NATIVE_API", "0"),
"jit_enabled": _env_flag("LMSTUDIO_JIT_ENABLED", "0"),
}
def is_deep_model(model_id):
cfg = get_lmstudio_config()
return normalize_model_id(model_id) == normalize_model_id(cfg["deep_model"])
def is_embedding_model(model_id):
cfg = get_lmstudio_config()
return normalize_model_id(model_id) == normalize_model_id(cfg["embed_model"])
def task_uses_fast(task_type):
return (task_type or "").strip().lower() in FAST_TASKS
def task_uses_deep(task_type):
return (task_type or "").strip().lower() in DEEP_TASKS
def select_lmstudio_model(task_type=None, requested_model=None, automatic=True, available_models=None):
cfg = get_lmstudio_config()
fast_model = cfg["fast_model"]
deep_model = cfg["deep_model"]
if requested_model:
selected = requested_model
elif task_uses_deep(task_type):
selected = deep_model
else:
selected = fast_model if automatic or task_uses_fast(task_type) else deep_model
normalized_available = {
normalize_model_id(model_id)
for model_id in (available_models or [])
}
if automatic and is_deep_model(selected) and not cfg["allow_deep_auto"]:
if normalize_model_id(fast_model) in normalized_available:
return fast_model
raise RuntimeError("local_fast_model_unavailable")
if normalized_available:
normalized_selected = normalize_model_id(selected)
if normalized_selected in normalized_available:
return selected
if automatic and normalize_model_id(fast_model) in normalized_available:
return fast_model
if automatic:
raise RuntimeError("local_fast_model_unavailable")
raise RuntimeError("requested_local_model_unavailable")
return selected
def add_lmstudio_ttl(payload):
cfg = get_lmstudio_config()
ttl = cfg["ttl"]
if ttl > 0:
payload["ttl"] = ttl
return payload

View File

@ -0,0 +1,38 @@
import json
def extract_lmstudio_content(choice, expect_json=False):
"""
Extracts usable content from LM Studio responses.
Workaround for Qwen reasoning models:
sometimes structured JSON is returned in message.reasoning_content
while message.content is empty.
"""
msg = choice.get("message", {}) or {}
content = (msg.get("content") or "").strip()
reasoning = (msg.get("reasoning_content") or "").strip()
finish = choice.get("finish_reason")
if finish == "length":
raise RuntimeError("lmstudio_output_truncated")
if content:
return content
if expect_json and finish == "stop" and reasoning:
if reasoning.startswith("{") or reasoning.startswith("["):
try:
json.loads(reasoning)
return reasoning
except Exception as e:
raise RuntimeError(f"reasoning_content_not_valid_json: {e}")
if reasoning:
raise RuntimeError("empty_content_with_reasoning_output")
raise RuntimeError("empty_lmstudio_output")
def parse_lmstudio_json(choice):
text = extract_lmstudio_content(choice, expect_json=True)
return json.loads(text)

View File

@ -0,0 +1,637 @@
"""
vault_brain.py Hybrid retrieval engine for the Obsidian second brain.
Architecture (per current SOTA):
- Episodic memory = 07 - Ingested/ (raw events, append-only, bi-temporal)
- Semantic memory = 08 - Semantic/ (extracted facts, Mem0-style)
- Procedural memory = 02 - Operations/ (runbooks, curated)
- Curated memory = 01/03/04/06 (decisions, topology, protocol)
Retrieval = hybrid BM25 (sparse) + nomic-embed-text v1.5 (dense, 768d, local
via LM Studio) fused with Reciprocal Rank Fusion (RRF, k=60). Index lives in
one SQLite file with FTS5 + a vec(768) BLOB column.
Citations:
- Tulving (1972) episodic vs semantic vs procedural memory taxonomy
- Cormack, Clarke, Buettcher (2009) Reciprocal Rank Fusion (default k=60)
- Lin et al. (2024) RRF beats BM25 / dense alone on MS MARCO
- Nussbaum et al. (2024) nomic-embed-text-v1.5, 768d, MTEB SOTA at size
- Chhikara et al. (2025) Mem0: append-only single-pass extraction (arXiv:2504.19413)
- Packer et al. (2024) MemGPT/Letta hierarchical memory (arXiv:2310.08560)
- Gutierrez et al. (2024) HippoRAG: PageRank-style traversal (arXiv:2405.14831)
Usage:
python vault_brain.py index # full reindex
python vault_brain.py index --incremental # only changed files
python vault_brain.py query "how is Atlas configured" --k 10
python vault_brain.py query "..." --format json
python vault_brain.py stats # index stats
"""
try:
from lmstudio_response import extract_lmstudio_content
except Exception:
from scripts.brain.lmstudio_response import extract_lmstudio_content
try:
from lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
except Exception:
from scripts.brain.lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
import argparse
import hashlib
import json
import math
import os
import re
import sqlite3
import struct
import sys
import time
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable
# Force stdout to UTF-8 on Windows so non-ASCII chars in vault content don't crash printing
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
# ─── Configuration ────────────────────────────────────────────────────────────
VAULT_PATH = Path(os.environ.get("VAULT_PATH", r"C:\Users\Smedj\Documents\Obsidian Vault"))
INDEX_PATH = VAULT_PATH / ".vault-brain.sqlite"
EMBED_URL = get_lmstudio_config()["base_url"] + "/v1/embeddings"
EMBED_MODEL = os.environ.get("EMBED_MODEL", get_lmstudio_config()["embed_model"])
EMBED_DIM = 768 # nomic-embed-text-v1.5
# Chunking: target ~500 tokens, overlap 50. Tokens approximated as 4 chars.
CHUNK_TARGET_CHARS = 2000
CHUNK_OVERLAP_CHARS = 200
# Reciprocal Rank Fusion constant. k=60 from Cormack et al. 2009 — the canonical default.
RRF_K = 60
# Folders to exclude from indexing (binary, daemon state, .obsidian internals)
EXCLUDED_DIRS = {".obsidian", ".trash", ".vault-brain.sqlite"}
EXCLUDED_FILES = {".vault-ingest-state.json"}
@dataclass
class Chunk:
file_path: str # vault-relative
chunk_idx: int
text: str
sha: str
folder: str # top-level folder, e.g. "07 - Ingested"
source: str # derived: "episodic" | "semantic" | "procedural" | "curated"
tags: list[str]
event_at: str | None # ISO from frontmatter if present
# ─── Embedding client (LM Studio, nomic-embed-text-v1.5) ──────────────────────
def embed_batch(texts: list[str], retries: int = 3) -> list[list[float]]:
"""Call LM Studio /v1/embeddings. Single request per call (LM Studio supports batches)."""
payload = json.dumps({"model": EMBED_MODEL, "input": texts}).encode("utf-8")
req = urllib.request.Request(EMBED_URL, data=payload, headers={"Content-Type": "application/json"})
last_err = None
for attempt in range(retries):
try:
with urllib.request.urlopen(req, timeout=120) as resp:
d = json.loads(resp.read().decode("utf-8"))
return [item["embedding"] for item in d["data"]]
except Exception as e:
last_err = e
time.sleep(0.5 * (attempt + 1))
raise RuntimeError(f"embed_batch failed after {retries} retries: {last_err}")
def cosine(a: list[float], b: list[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
na = math.sqrt(sum(x * x for x in a))
nb = math.sqrt(sum(x * x for x in b))
return dot / (na * nb) if na and nb else 0.0
def pack_vec(v: list[float]) -> bytes:
return struct.pack(f"{len(v)}f", *v)
def unpack_vec(b: bytes) -> list[float]:
n = len(b) // 4
return list(struct.unpack(f"{n}f", b))
# ─── Frontmatter & chunking ──────────────────────────────────────────────────
FRONTMATTER_RE = re.compile(r"^---\n(.*?)\n---\n", re.DOTALL)
def parse_frontmatter(text: str) -> tuple[dict, str]:
m = FRONTMATTER_RE.match(text)
if not m:
return {}, text
fm_block = m.group(1)
body = text[m.end() :]
fm = {}
for line in fm_block.splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if ":" not in line:
continue
k, _, v = line.partition(":")
v = v.strip()
# Crude YAML — handles "key: \"val\"", "key: [a, b, c]", "key: val"
if v.startswith('"') and v.endswith('"'):
v = v[1:-1].replace('\\"', '"')
elif v.startswith("[") and v.endswith("]"):
inner = v[1:-1]
v = [item.strip().strip('"') for item in inner.split(",") if item.strip()]
fm[k.strip()] = v
return fm, body
def folder_to_source(folder: str) -> str:
"""Map vault top-level folder to memory category."""
if folder.startswith("07 - Ingested"):
return "episodic"
if folder.startswith("08 - Semantic"):
return "semantic"
if folder.startswith("02 - Operations"):
return "procedural"
return "curated"
def chunk_text(text: str) -> list[str]:
"""Char-based chunker with overlap. Cheap and robust for markdown."""
if len(text) <= CHUNK_TARGET_CHARS:
return [text] if text.strip() else []
chunks = []
start = 0
while start < len(text):
end = min(start + CHUNK_TARGET_CHARS, len(text))
# Prefer to break at paragraph
if end < len(text):
break_at = text.rfind("\n\n", start, end)
if break_at > start + CHUNK_TARGET_CHARS // 2:
end = break_at
chunks.append(text[start:end].strip())
if end >= len(text):
break
start = end - CHUNK_OVERLAP_CHARS
return [c for c in chunks if c]
def iter_vault_chunks() -> Iterable[Chunk]:
"""Walk the vault, yield chunks with metadata."""
for root, dirs, files in os.walk(VAULT_PATH):
# Prune excluded dirs in-place
dirs[:] = [d for d in dirs if d not in EXCLUDED_DIRS]
for fname in files:
if not fname.endswith(".md"):
continue
if fname in EXCLUDED_FILES:
continue
fpath = Path(root) / fname
rel = fpath.relative_to(VAULT_PATH).as_posix()
top_folder = rel.split("/", 1)[0]
try:
text = fpath.read_text(encoding="utf-8", errors="replace")
except Exception:
continue
fm, body = parse_frontmatter(text)
tags = fm.get("tags", []) if isinstance(fm.get("tags"), list) else []
event_at = fm.get("event_at") or fm.get("created") or None
source = folder_to_source(top_folder)
for i, chunk in enumerate(chunk_text(body)):
sha = hashlib.sha256(f"{rel}::{i}::{chunk}".encode("utf-8")).hexdigest()[:16]
yield Chunk(rel, i, chunk, sha, top_folder, source, tags, event_at)
# ─── Index lifecycle ──────────────────────────────────────────────────────────
SCHEMA = """
CREATE TABLE IF NOT EXISTS chunks (
rowid INTEGER PRIMARY KEY,
file_path TEXT NOT NULL,
chunk_idx INTEGER NOT NULL,
sha TEXT NOT NULL UNIQUE,
folder TEXT NOT NULL,
source TEXT NOT NULL,
tags TEXT, -- JSON array
event_at TEXT,
text TEXT NOT NULL,
embedding BLOB -- 768f packed; NULL if not yet embedded
);
CREATE INDEX IF NOT EXISTS idx_chunks_path ON chunks(file_path);
CREATE INDEX IF NOT EXISTS idx_chunks_folder ON chunks(folder);
CREATE INDEX IF NOT EXISTS idx_chunks_source ON chunks(source);
CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(
text,
file_path UNINDEXED,
folder UNINDEXED,
tokenize = 'unicode61 remove_diacritics 2'
);
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT
);
"""
def open_index() -> sqlite3.Connection:
conn = sqlite3.connect(INDEX_PATH)
conn.executescript(SCHEMA)
conn.commit()
return conn
def cmd_index(args):
full = not args.incremental
conn = open_index()
cur = conn.cursor()
t0 = time.time()
if full:
cur.execute("DELETE FROM chunks")
cur.execute("DELETE FROM chunks_fts")
conn.commit()
# Collect all current SHAs
new_chunks: list[Chunk] = []
seen_shas: set[str] = set()
for ch in iter_vault_chunks():
seen_shas.add(ch.sha)
new_chunks.append(ch)
if args.incremental:
existing = {row[0] for row in cur.execute("SELECT sha FROM chunks").fetchall()}
# Delete chunks no longer present
to_delete = existing - seen_shas
if to_delete:
placeholders = ",".join("?" * len(to_delete))
cur.execute(f"DELETE FROM chunks WHERE sha IN ({placeholders})", list(to_delete))
cur.execute(f"DELETE FROM chunks_fts WHERE rowid IN (SELECT rowid FROM chunks WHERE sha IN ({placeholders}))", list(to_delete))
# Filter to only new chunks
new_chunks = [c for c in new_chunks if c.sha not in existing]
print(f"[index] {len(new_chunks)} new chunk(s) to embed (full={full})")
inserted = 0
BATCH = 32 # LM Studio handles ~32-64 inputs cleanly
for i in range(0, len(new_chunks), BATCH):
batch = new_chunks[i : i + BATCH]
try:
embeddings = embed_batch([c.text[:8000] for c in batch]) # truncate to 8K chars per chunk for safety
except Exception as e:
print(f"[index] WARN: embed batch failed at {i}: {e}", file=sys.stderr)
embeddings = [None] * len(batch)
for c, emb in zip(batch, embeddings):
blob = pack_vec(emb) if emb else None
try:
cur.execute(
"INSERT OR IGNORE INTO chunks (file_path, chunk_idx, sha, folder, source, tags, event_at, text, embedding) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(c.file_path, c.chunk_idx, c.sha, c.folder, c.source, json.dumps(c.tags), c.event_at, c.text, blob),
)
rowid = cur.lastrowid
cur.execute(
"INSERT INTO chunks_fts (rowid, text, file_path, folder) VALUES (?, ?, ?, ?)",
(rowid, c.text, c.file_path, c.folder),
)
inserted += 1
except Exception as e:
print(f"[index] WARN: insert failed for {c.file_path}#{c.chunk_idx}: {e}", file=sys.stderr)
if i and i % (BATCH * 10) == 0:
conn.commit()
print(f"[index] {inserted}/{len(new_chunks)} indexed (elapsed {time.time()-t0:.1f}s)")
cur.execute("INSERT OR REPLACE INTO meta (key, value) VALUES ('last_index_at', ?)", (str(int(time.time())),))
conn.commit()
print(f"[index] done. {inserted} chunk(s) added in {time.time()-t0:.1f}s")
cmd_stats(args)
def cmd_stats(args):
conn = open_index()
cur = conn.cursor()
rows = cur.execute("SELECT source, COUNT(*) FROM chunks GROUP BY source").fetchall()
total = cur.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
embedded = cur.execute("SELECT COUNT(*) FROM chunks WHERE embedding IS NOT NULL").fetchone()[0]
distinct_files = cur.execute("SELECT COUNT(DISTINCT file_path) FROM chunks").fetchone()[0]
last = cur.execute("SELECT value FROM meta WHERE key='last_index_at'").fetchone()
print(f"\n=== Vault Brain Index ===")
print(f" files indexed: {distinct_files}")
print(f" total chunks: {total}")
print(f" with embeddings: {embedded} ({100*embedded/max(total,1):.1f}%)")
print(f" by source:")
for s, n in rows:
print(f" {s:12} {n}")
if last:
from datetime import datetime
print(f" last index: {datetime.fromtimestamp(int(last[0])).isoformat()}")
# ─── Retrieval ────────────────────────────────────────────────────────────────
def fts_query_escape(query: str) -> str:
"""FTS5 MATCH expects safe tokens. Strip control chars, quote each word."""
words = re.findall(r"\w+", query, flags=re.UNICODE)
if not words:
return '""'
# OR all tokens, prefix-match the last for query continuation
return " OR ".join(f'"{w}"*' for w in words)
def search_bm25(conn: sqlite3.Connection, query: str, k: int) -> list[tuple[int, float]]:
"""Return list of (rowid, bm25_score) for top-k matches. Lower score = better in bm25()."""
fts = fts_query_escape(query)
rows = conn.execute(
"SELECT rowid, bm25(chunks_fts) AS score FROM chunks_fts WHERE chunks_fts MATCH ? ORDER BY score LIMIT ?",
(fts, k),
).fetchall()
return [(rid, sc) for rid, sc in rows]
def search_dense(conn: sqlite3.Connection, query_vec: list[float], k: int, source_filter: str | None = None) -> list[tuple[int, float]]:
"""Return list of (rowid, cosine) for top-k by cosine similarity."""
sql = "SELECT rowid, embedding FROM chunks WHERE embedding IS NOT NULL"
params: list = []
if source_filter:
sql += " AND source = ?"
params.append(source_filter)
rows = conn.execute(sql, params).fetchall()
scored = []
for rid, blob in rows:
v = unpack_vec(blob)
scored.append((rid, cosine(query_vec, v)))
scored.sort(key=lambda x: x[1], reverse=True)
return scored[:k]
def rrf_fuse(rankings: list[list[tuple[int, float]]], k: int = RRF_K) -> dict[int, float]:
"""Reciprocal Rank Fusion. Returns rowid -> fused score.
Reference: Cormack, Clarke & Buettcher, "Reciprocal Rank Fusion outperforms
Condorcet and individual rank learning methods", SIGIR 2009.
https://plg.uwaterloo.ca/~gvcormack/cormacksigir09-rrf.pdf
"""
fused: dict[int, float] = {}
for ranking in rankings:
for rank, (rowid, _score) in enumerate(ranking, start=1):
fused[rowid] = fused.get(rowid, 0.0) + 1.0 / (k + rank)
return fused
# Centrality boost: combine RRF score with PageRank centrality.
# We use a multiplicative boost log(1 + γ·centrality·N) so that high-centrality
# nodes get a small lift but the dominant signal stays the query-specific RRF.
# γ is tuned so a centrality at the 95th percentile contributes ~0.1 to log(1+x).
# Reference (motivation): Gutierrez et al., "HippoRAG: Neurobiologically Inspired
# Long-Term Memory for LLMs", NeurIPS 2024 — uses Personalized PageRank to boost
# graph-aware retrieval. We use static (non-personalized) PageRank which is a
# weaker but cheaper proxy when the graph is small.
import math as _math
CENTRALITY_GAMMA = 1000.0 # tuned for our pi values which are O(1e-3) max
def fuse_with_centrality(
fused_rrf: dict[int, float],
centrality_by_rowid: dict[int, float],
gamma: float = CENTRALITY_GAMMA,
) -> dict[int, float]:
"""Multiply RRF score by (1 + log(1 + γ·c)) where c is the file's centrality.
Strictly increasing in c, sub-linear so dominant high-centrality items don't
swamp the query-specific signal."""
out: dict[int, float] = {}
for rid, score in fused_rrf.items():
c = centrality_by_rowid.get(rid, 0.0) or 0.0
boost = 1.0 + _math.log(1.0 + gamma * c)
out[rid] = score * boost
return out
def hyde_expand(query: str, llm_url: str | None = None, model: str | None = None) -> str | None:
"""HyDE — Hypothetical Document Embeddings (Gao, Ma, Lin, Callan, ACL 2023).
Reference: Gao, L., Ma, X., Lin, J., Callan, J. (2023). "Precise Zero-Shot
Dense Retrieval without Relevance Labels." Proceedings of ACL 2023.
https://aclanthology.org/2023.acl-long.99/
arXiv: https://arxiv.org/abs/2212.10496
Asks a local LLM to generate a hypothetical answer to the query. We then
embed THAT answer instead of the query the dense bottleneck filters
out hallucinations and the embedding sits in the answer-space, much closer
to relevant retrievable chunks.
"""
url = llm_url or get_lmstudio_config()["base_url"] + "/v1/chat/completions"
try:
with urllib.request.urlopen(get_lmstudio_config()["base_url"] + "/v1/models", timeout=5) as resp:
available_models = [m["id"] for m in json.loads(resp.read().decode("utf-8")).get("data", [])]
model = select_lmstudio_model(
task_type="vault",
requested_model=model or os.environ.get("HYDE_MODEL", get_lmstudio_config()["fast_model"]),
automatic=True,
available_models=available_models,
)
except Exception:
return None
prompt = (
"You are answering an information-retrieval probe. Write a short, factual "
"paragraph (3-5 sentences) that DIRECTLY answers the user's query, as if you "
"had perfect knowledge. Do not refuse, do not hedge, do not add caveats — write "
"the answer that the ideal source document would contain. No greetings, no "
"meta-commentary.\n\nQuery: " + query
)
payload = {
"model": model,
"messages": [
{"role": "system", "content": "/no_think\nOutput only the final answer. No hidden reasoning. No markdown unless explicitly requested."},
{"role": "user", "content": "/no_think\n" + prompt},
],
"temperature": 0.4,
"max_tokens": 400,
}
payload = json.dumps(add_lmstudio_ttl(payload)).encode("utf-8")
req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=120) as resp:
d = json.loads(resp.read().decode("utf-8"))
text = extract_lmstudio_content(d["choices"][0], expect_json=False)
return text or None
except Exception:
return None
def cmd_query(args):
query = args.query
k = args.k
conn = open_index()
use_centrality = not args.no_centrality
# HyDE: optionally expand the query into a hypothetical answer first
qvec_query = None
qvec_hyde = None
hyde_text = None
try:
qvec_query = embed_batch([query])[0]
except Exception as e:
print(f"[query] dense (query) skipped: {e}", file=sys.stderr)
if args.hyde:
hyde_text = hyde_expand(query)
if hyde_text:
try:
qvec_hyde = embed_batch([hyde_text])[0]
except Exception as e:
print(f"[query] dense (hyde) skipped: {e}", file=sys.stderr)
qvec = qvec_query # default
bm25_top = search_bm25(conn, query, k=k * 3)
# Dense rankings: combine query embedding and HyDE embedding via RRF (best of both)
dense_query_top = search_dense(conn, qvec_query, k=k * 3) if qvec_query is not None else []
dense_hyde_top = search_dense(conn, qvec_hyde, k=k * 3) if qvec_hyde is not None else []
if qvec_hyde is not None:
dense_top = dense_query_top # keep dense_top variable for compat
else:
dense_top = dense_query_top
_rankings = [bm25_top, dense_query_top]
if dense_hyde_top:
_rankings.append(dense_hyde_top)
fused_rrf = rrf_fuse(_rankings)
bm25 = bm25_top # downstream display compat
dense = dense_query_top
if not fused_rrf:
print("(no results)")
return
# Centrality boost (graph-aware retrieval, HippoRAG-inspired)
centrality_by_rowid = {}
community_by_rowid = {}
if use_centrality:
cols = {r[1] for r in conn.execute("PRAGMA table_info(chunks)").fetchall()}
if "centrality" in cols:
rids = list(fused_rrf.keys())
placeholders = ",".join("?" * len(rids))
extra_col = ", community_id" if "community_id" in cols else ""
cent_rows = conn.execute(
f"SELECT rowid, centrality{extra_col} FROM chunks WHERE rowid IN ({placeholders})",
rids,
).fetchall()
for row in cent_rows:
rid = row[0]
centrality_by_rowid[rid] = row[1] or 0.0
if extra_col:
community_by_rowid[rid] = row[2]
# Per-community z-score: how central is this chunk WITHIN its community?
# Avoids the flat-distribution problem when the global graph is dense.
# Reference for community-aware ranking: Edge et al., GraphRAG (2024),
# which uses Leiden communities to scope retrieval. We do the simpler
# community-z-score variant.
if community_by_rowid:
from statistics import mean, pstdev
# Aggregate centrality per community (across the whole index, not just candidates)
global_stats: dict[int, tuple[float, float]] = {}
for cid in set(c for c in community_by_rowid.values() if c is not None):
rows_c = conn.execute(
"SELECT centrality FROM chunks WHERE community_id = ? AND centrality IS NOT NULL",
(cid,)
).fetchall()
vals = [r[0] for r in rows_c if r[0] is not None]
if len(vals) >= 2:
m = mean(vals); sd = pstdev(vals) or 1e-9
global_stats[cid] = (m, sd)
# Recompute "effective centrality" as z-score in own community
eff_centrality: dict[int, float] = {}
for rid, c in centrality_by_rowid.items():
cid = community_by_rowid.get(rid)
if cid in global_stats:
m, sd = global_stats[cid]
# Sigmoid-mapped z-score → bounded boost
z = (c - m) / sd
eff_centrality[rid] = max(0.0, c) * (1.0 + 0.2 * z)
else:
eff_centrality[rid] = c
fused = fuse_with_centrality(fused_rrf, eff_centrality)
else:
fused = fuse_with_centrality(fused_rrf, centrality_by_rowid)
else:
fused = fused_rrf
else:
fused = fused_rrf
top = sorted(fused.items(), key=lambda x: x[1], reverse=True)[:k]
rowids = [rid for rid, _ in top]
placeholders = ",".join("?" * len(rowids))
chunks = {
row[0]: row
for row in conn.execute(
f"SELECT rowid, file_path, chunk_idx, source, folder, event_at, text FROM chunks WHERE rowid IN ({placeholders})",
rowids,
)
}
results = []
for rid, score in top:
c = chunks.get(rid)
if not c:
continue
rid2, fp, ci, src, folder, ev, text = c
# Component scores for transparency
bm25_rank = next((i + 1 for i, (rr, _) in enumerate(bm25_top) if rr == rid), None)
dense_rank = next((i + 1 for i, (rr, _) in enumerate(dense_top) if rr == rid), None)
snippet = text[:300].replace("\n", " ")
results.append({
"score": round(score, 4),
"rrf_only": round(fused_rrf.get(rid, 0.0), 4),
"centrality": round(centrality_by_rowid.get(rid, 0.0), 6),
"bm25_rank": bm25_rank,
"dense_rank": dense_rank,
"source": src,
"folder": folder,
"file": fp,
"chunk": ci,
"event_at": ev,
"snippet": snippet,
})
if args.format == "json":
print(json.dumps(results, indent=2, ensure_ascii=False))
else:
for i, r in enumerate(results, 1):
print(f"\n[{i}] score={r['score']:.4f} rrf={r['rrf_only']:.4f} cent={r['centrality']:.2e} bm25_rank={r['bm25_rank']} dense_rank={r['dense_rank']} source={r['source']}")
print(f" {r['file']} (chunk {r['chunk']})")
print(f" {r['snippet']}...")
# ─── Main ────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(prog="vault_brain")
sub = parser.add_subparsers(dest="cmd", required=True)
p_idx = sub.add_parser("index", help="(Re)build the index")
p_idx.add_argument("--incremental", action="store_true", help="Only embed new/changed chunks")
p_q = sub.add_parser("query", help="Hybrid search (BM25 + dense + RRF + centrality)")
p_q.add_argument("query", help="Query text")
p_q.add_argument("-k", type=int, default=10, help="Number of results")
p_q.add_argument("--format", choices=["text", "json"], default="text")
p_q.add_argument("--no-centrality", action="store_true", help="Disable PageRank centrality boost")
p_q.add_argument("--hyde", action="store_true", help="Enable HyDE: generate hypothetical answer via local LLM, fuse its embedding into the dense ranking (Gao et al., ACL 2023)")
sub.add_parser("stats", help="Index statistics")
args = parser.parse_args()
if args.cmd == "index":
cmd_index(args)
elif args.cmd == "query":
cmd_query(args)
elif args.cmd == "stats":
cmd_stats(args)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,261 @@
"""
vault_consolidate.py Mem0-style episodic semantic extraction.
Reads recent notes from `07 - Ingested/` and extracts atomic facts into
`08 - Semantic/<topic>/<fact-id>.md`. Single-pass ADD-only per Chhikara et al.
(Mem0, arXiv:2504.19413, 2025): never UPDATE/DELETE accumulate, don't overwrite.
Each extracted fact is:
- atomic (one claim per note)
- traceable (frontmatter `derived_from:` lists source notes)
- bi-temporal (event_at = source event time, captured_at = extraction time)
- typed (`fact_type` = decision | preference | constraint | observation | reference)
Uses qwen3.6-35b-a3b via LM Studio (local, no quota cost) as the extraction LLM.
Usage:
python vault_consolidate.py --since 24h --dry-run # preview
python vault_consolidate.py --since 24h --apply # write notes
python vault_consolidate.py --since 7d --max-files 200
"""
try:
from lmstudio_response import extract_lmstudio_content
except Exception:
from scripts.brain.lmstudio_response import extract_lmstudio_content
try:
from lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
except Exception:
from scripts.brain.lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
import argparse
import datetime as dt
import hashlib
import json
import os
import re
import time
import urllib.request
from pathlib import Path
VAULT_PATH = Path(os.environ.get("VAULT_PATH", r"C:\Users\Smedj\Documents\Obsidian Vault"))
INGESTED_DIR = VAULT_PATH / "07 - Ingested"
SEMANTIC_DIR = VAULT_PATH / "08 - Semantic"
LM_STUDIO_URL = get_lmstudio_config()["base_url"] + "/v1/chat/completions"
EXTRACT_MODEL = os.environ.get("EXTRACT_MODEL", get_lmstudio_config()["fast_model"])
STATE_FILE = VAULT_PATH / ".vault-consolidate-state.json"
EXTRACTION_PROMPT = """You are an information-extraction module for a long-term agent memory.
Read the input note and produce a JSON array of atomic FACTS that should be remembered for future sessions. Each fact must be:
- ATOMIC: one claim per fact, fully self-contained
- TYPED: fact_type ["decision","preference","constraint","observation","reference"]
- DURABLE: would still be useful in 3 months, not session-trivia
- SHORT: 1-2 sentences
Skip the note entirely if it contains nothing durable (small talk, "ok", trivial confirmations, build output, errors that were resolved within the same session). Output `[]` in that case.
Respond with ONLY a valid JSON array. No prose, no markdown fences. Schema:
[
{
"fact_type": "decision|preference|constraint|observation|reference",
"topic": "short-slug-for-folder",
"title": "Imperative title under 80 chars",
"claim": "1-2 sentence durable claim",
"tags": ["topic/x","agent/y"]
}
]
Input note:
---
"""
def call_llm(note_text: str, retries: int = 2) -> list[dict]:
"""Call qwen via LM Studio chat/completions. Return parsed JSON list or []."""
try:
with urllib.request.urlopen(get_lmstudio_config()["base_url"] + "/v1/models", timeout=5) as resp:
available_models = [m["id"] for m in json.loads(resp.read().decode("utf-8")).get("data", [])]
model = select_lmstudio_model(
task_type="consolidate",
requested_model=EXTRACT_MODEL,
automatic=True,
available_models=available_models,
)
except Exception as e:
print(f"[consolidate] LM Studio model selection failed: {e}")
return []
payload = {
"model": model,
"messages": [
{"role": "user", "content": EXTRACTION_PROMPT + note_text[:8000]}, # cap input for safety
],
"temperature": 0.1,
"max_tokens": 2000,
}
payload = json.dumps(add_lmstudio_ttl(payload)).encode("utf-8")
req = urllib.request.Request(LM_STUDIO_URL, data=payload, headers={"Content-Type": "application/json"})
last_err = None
for attempt in range(retries):
try:
with urllib.request.urlopen(req, timeout=180) as resp:
d = json.loads(resp.read().decode("utf-8"))
content = extract_lmstudio_content(d["choices"][0], expect_json=True)
# Remove ```json fences if model added them despite instructions
content = re.sub(r"^```(?:json)?\s*", "", content)
content = re.sub(r"\s*```$", "", content)
if not content or content == "[]":
return []
facts = json.loads(content)
if not isinstance(facts, list):
return []
return facts
except json.JSONDecodeError as e:
last_err = f"JSON: {e}; raw: {content[:200]}"
except Exception as e:
last_err = str(e)
time.sleep(0.5 * (attempt + 1))
print(f"[consolidate] LLM call failed: {last_err}", file=sys.stderr if False else None)
return []
def load_state() -> dict:
if STATE_FILE.exists():
try:
return json.loads(STATE_FILE.read_text(encoding="utf-8"))
except Exception:
pass
return {"processed_files": {}} # path -> last_consolidated_iso
def save_state(state: dict):
tmp = STATE_FILE.with_suffix(".tmp")
tmp.write_text(json.dumps(state, indent=2), encoding="utf-8")
tmp.replace(STATE_FILE)
def parse_since(arg: str) -> dt.datetime:
"""Accept '24h', '7d', '30m'. Return UTC cutoff."""
m = re.match(r"^(\d+)([hdm])$", arg)
if not m:
raise ValueError(f"--since must be like '24h', '7d', '30m', got {arg!r}")
n, unit = int(m.group(1)), m.group(2)
delta = {"h": dt.timedelta(hours=n), "d": dt.timedelta(days=n), "m": dt.timedelta(minutes=n)}[unit]
return dt.datetime.now(dt.timezone.utc) - delta
def extract_event_at(text: str) -> dt.datetime | None:
m = re.search(r"^event_at:\s*\"?([^\"\n]+)\"?", text, re.MULTILINE)
if not m:
return None
try:
return dt.datetime.fromisoformat(m.group(1).replace("Z", "+00:00"))
except Exception:
return None
def write_fact(fact: dict, source_rel: str, event_at: dt.datetime | None) -> Path:
topic = re.sub(r"[^\w\-]", "-", fact.get("topic", "general")).lower().strip("-")[:40] or "general"
title = fact.get("title", "untitled")[:80]
slug = re.sub(r"[^\w\-]", "-", title).strip("-")[:60]
fact_id = hashlib.sha256(f"{source_rel}::{title}".encode()).hexdigest()[:10]
folder = SEMANTIC_DIR / topic
folder.mkdir(parents=True, exist_ok=True)
path = folder / f"{fact_id}-{slug}.md"
if path.exists():
return path # idempotent — already extracted
captured_at = dt.datetime.now(dt.timezone.utc).isoformat()
event_iso = event_at.isoformat() if event_at else captured_at
fact_type = fact.get("fact_type", "observation")
raw_tags = fact.get("tags", []) or []
tags = [str(t) for t in raw_tags] + [f"fact/{fact_type}", f"topic/{topic}", "ingested-fact"]
fm_lines = [
"---",
f'type: "semantic-fact"',
f'fact_type: "{fact_type}"',
f'topic: "{topic}"',
f'event_at: "{event_iso}"',
f'captured_at: "{captured_at}"',
f'derived_from: ["{source_rel}"]',
f'tags: [{", ".join(json.dumps(t) for t in tags)}]',
"---",
"",
f"# {title}",
"",
fact.get("claim", "(no claim)"),
"",
f"---",
f"Derived from [[{source_rel}]] on {captured_at}.",
]
path.write_text("\n".join(fm_lines), encoding="utf-8")
return path
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--since", default="24h", help="Relative window: 24h, 7d, 30m")
ap.add_argument("--apply", action="store_true", help="Actually write fact notes (default is dry-run)")
ap.add_argument("--max-files", type=int, default=500, help="Cap files processed per run")
ap.add_argument("--source-glob", default="*.md", help="Glob within 07 - Ingested/")
args = ap.parse_args()
cutoff = parse_since(args.since)
state = load_state()
candidates = []
if INGESTED_DIR.exists():
for f in INGESTED_DIR.rglob(args.source_glob):
if not f.is_file():
continue
rel = f.relative_to(VAULT_PATH).as_posix()
if rel in state["processed_files"]:
continue
try:
text = f.read_text(encoding="utf-8", errors="replace")
except Exception:
continue
ev = extract_event_at(text)
if ev and ev < cutoff:
continue
candidates.append((f, rel, text, ev))
candidates = candidates[: args.max_files]
print(f"[consolidate] {len(candidates)} candidate ingested note(s) since {args.since} (cutoff: {cutoff.isoformat()})")
print(f"[consolidate] mode: {'APPLY' if args.apply else 'DRY-RUN'}")
total_facts = 0
extracted_files = 0
t0 = time.time()
for i, (f, rel, text, ev) in enumerate(candidates, 1):
# Trim frontmatter from the LLM input — we already have the metadata
body = re.sub(r"^---\n.*?\n---\n", "", text, count=1, flags=re.DOTALL)
if len(body.strip()) < 80:
continue
facts = call_llm(body)
if facts:
extracted_files += 1
for fact in facts:
if not isinstance(fact, dict):
continue
if args.apply:
p = write_fact(fact, rel, ev)
total_facts += 1
else:
print(f" [{rel}] {fact.get('fact_type','?'):12} {fact.get('topic','?'):20} {fact.get('title','')[:60]}")
total_facts += 1
if args.apply:
state["processed_files"][rel] = dt.datetime.now(dt.timezone.utc).isoformat()
if i % 20 == 0:
save_state(state)
print(f"[consolidate] {i}/{len(candidates)} processed, {total_facts} facts so far ({time.time()-t0:.0f}s)")
if args.apply:
save_state(state)
print(f"\n[consolidate] done. {extracted_files} files yielded {total_facts} facts in {time.time()-t0:.0f}s")
if args.apply:
print(f"[consolidate] state: {STATE_FILE}")
if __name__ == "__main__":
import sys
main()

268
scripts/brain/vault_eval.py Normal file
View File

@ -0,0 +1,268 @@
"""
vault_eval.py Evaluation harness for the vault brain.
Measures:
- Recall@k (k=1, 5, 10, 20)
- MRR (Mean Reciprocal Rank)
- Per-strategy: BM25-only, dense-only, hybrid (RRF)
Generates synthetic queries from existing curated docs (decisions, runbook,
topology) by extracting representative phrases via a small LLM call. Each
synthetic query is paired with its source file as ground truth we then check
whether retrieval surfaces that file in the top-K.
Citations:
- Voorhees (1999) TREC: Reciprocal Rank metric
- Cormack et al. (2009) Recall@k methodology for fusion ranking
- Lin et al. (2024) RRF baseline for MS MARCO
Usage:
python vault_eval.py generate --out queries.json # build synthetic test set
python vault_eval.py run queries.json # run all 3 strategies
"""
try:
from lmstudio_response import extract_lmstudio_content
except Exception:
from scripts.brain.lmstudio_response import extract_lmstudio_content
try:
from lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
except Exception:
from scripts.brain.lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
import argparse
import json
import os
import random
import re
import sys
import time
import urllib.request
from pathlib import Path
# Reuse the brain functions
sys.path.insert(0, str(Path(__file__).parent))
from vault_brain import (
VAULT_PATH, open_index, embed_batch, search_bm25, search_dense, rrf_fuse,
fuse_with_centrality, EMBED_URL,
)
GENERATE_PROMPT = """Given the following note from an engineering knowledge base, generate 2 SHORT user queries (each 5-12 words) that a future user would plausibly type to find this exact information. The queries should:
- be in natural language, not just keywords
- be specific enough that THIS note is the right answer
- vary in style (one keyword-heavy, one paraphrased)
Output EXACTLY two lines, no numbering, no prose. Each line is one query.
Note (file: {file_path}):
---
{snippet}
---"""
def call_llm(prompt: str, model: str | None = None) -> str:
try:
with urllib.request.urlopen(get_lmstudio_config()["base_url"] + "/v1/models", timeout=5) as resp:
available_models = [m["id"] for m in json.loads(resp.read().decode("utf-8")).get("data", [])]
model = select_lmstudio_model(
task_type="eval",
requested_model=model or os.environ.get("EVAL_MODEL", get_lmstudio_config()["fast_model"]),
automatic=True,
available_models=available_models,
)
except Exception as e:
return f"__error__ {e}"
payload = {
"model": model,
"messages": [
{"role": "system", "content": "/no_think\nOutput only the final answer. No hidden reasoning. No markdown unless explicitly requested."},
{"role": "user", "content": "/no_think\n" + prompt},
],
"temperature": 0.4,
"max_tokens": 200,
}
payload = json.dumps(add_lmstudio_ttl(payload)).encode("utf-8")
req = urllib.request.Request(
get_lmstudio_config()["base_url"] + "/v1/chat/completions",
data=payload, headers={"Content-Type": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=120) as resp:
d = json.loads(resp.read().decode("utf-8"))
return extract_lmstudio_content(d["choices"][0], expect_json=False)
except Exception as e:
return f"__error__ {e}"
def cmd_generate(args):
"""Sample N curated files, generate 2 queries each."""
conn = open_index()
rows = conn.execute(
"SELECT DISTINCT file_path FROM chunks WHERE source IN ('curated','procedural','semantic') AND chunk_idx = 0 LIMIT 500"
).fetchall()
files = [r[0] for r in rows]
random.seed(42)
sample = random.sample(files, min(args.n, len(files)))
queries = []
for i, fp in enumerate(sample, 1):
row = conn.execute(
"SELECT text FROM chunks WHERE file_path=? AND chunk_idx=0", (fp,)
).fetchone()
if not row:
continue
snippet = row[0][:1500]
prompt = GENERATE_PROMPT.format(file_path=fp, snippet=snippet)
out = call_llm(prompt)
if out.startswith("__error__"):
print(f"[generate] WARN {fp}: {out}", file=sys.stderr)
continue
lines = [l.strip(" -*0123456789.") for l in out.splitlines() if l.strip()]
for q in lines[:2]:
if 4 <= len(q.split()) <= 20:
queries.append({"query": q, "expected_file": fp})
print(f"[generate] {i}/{len(sample)} {fp[:50]}: {len(lines[:2])} queries")
Path(args.out).write_text(json.dumps(queries, indent=2, ensure_ascii=False), encoding="utf-8")
print(f"\n[generate] wrote {len(queries)} queries to {args.out}")
def evaluate(queries: list[dict], strategy: str, k_values: list[int]) -> dict:
"""Run a strategy across queries, return metrics."""
conn = open_index()
metrics = {f"recall@{k}": 0 for k in k_values}
metrics["mrr"] = 0.0
metrics["n"] = len(queries)
metrics["queries_with_hit"] = 0
max_k = max(k_values)
for q in queries:
query, expected = q["query"], q["expected_file"]
needs_dense = strategy in ("dense", "hybrid", "hybrid_centrality")
needs_bm25 = strategy in ("bm25", "hybrid", "hybrid_centrality")
try:
qvec = embed_batch([query])[0] if needs_dense else None
except Exception:
qvec = None
bm25 = search_bm25(conn, query, k=max_k * 3) if needs_bm25 else []
dense = search_dense(conn, qvec, k=max_k * 3) if (qvec and needs_dense) else []
if strategy == "bm25":
ranked = [rid for rid, _ in bm25]
elif strategy == "dense":
ranked = [rid for rid, _ in dense]
elif strategy == "hybrid":
fused = rrf_fuse([bm25, dense])
ranked = [rid for rid, _ in sorted(fused.items(), key=lambda x: x[1], reverse=True)]
elif strategy == "hybrid_centrality":
fused = rrf_fuse([bm25, dense])
# Load centrality for the candidate rowids
rids = list(fused.keys())
if rids:
placeholders = ",".join("?" * len(rids))
cent = {r[0]: r[1] for r in conn.execute(
f"SELECT rowid, centrality FROM chunks WHERE rowid IN ({placeholders})", rids
).fetchall()}
fused = fuse_with_centrality(fused, cent)
ranked = [rid for rid, _ in sorted(fused.items(), key=lambda x: x[1], reverse=True)]
else:
raise ValueError(strategy)
if not ranked:
continue
# Map rowid -> file_path (top-K only)
top_rids = ranked[:max_k]
if not top_rids:
continue
placeholders = ",".join("?" * len(top_rids))
rows = conn.execute(
f"SELECT rowid, file_path FROM chunks WHERE rowid IN ({placeholders})", top_rids
).fetchall()
rid_to_file = {rid: fp for rid, fp in rows}
ranked_files = [rid_to_file.get(rid, "") for rid in top_rids]
# First match position
try:
pos = next(i for i, fp in enumerate(ranked_files, 1) if fp == expected)
metrics["mrr"] += 1.0 / pos
metrics["queries_with_hit"] += 1
for k in k_values:
if pos <= k:
metrics[f"recall@{k}"] += 1
except StopIteration:
pass
n = max(metrics["n"], 1)
metrics["mrr"] = round(metrics["mrr"] / n, 4)
for k in k_values:
metrics[f"recall@{k}"] = round(metrics[f"recall@{k}"] / n, 4)
return metrics
def cmd_run(args):
queries = json.loads(Path(args.queries).read_text(encoding="utf-8"))
print(f"[eval] {len(queries)} queries loaded from {args.queries}")
k_values = [1, 5, 10, 20]
results = {}
strategies = args.strategies.split(",") if args.strategies else ["bm25", "dense", "hybrid", "hybrid_centrality"]
for strategy in strategies:
t0 = time.time()
metrics = evaluate(queries, strategy, k_values)
elapsed = time.time() - t0
print(f"\n=== Strategy: {strategy} ({elapsed:.1f}s) ===")
for k in k_values:
print(f" Recall@{k:>2}: {metrics[f'recall@{k}']:.4f}")
print(f" MRR : {metrics['mrr']:.4f}")
print(f" hits : {metrics['queries_with_hit']}/{metrics['n']}")
results[strategy] = metrics
# Pretty markdown report saved to vault for transparency
report_path = VAULT_PATH / "06 - Agents" / f"eval-{time.strftime('%Y-%m-%d')}.md"
report_path.parent.mkdir(parents=True, exist_ok=True)
lines = [
"---",
f'type: "evaluation"',
f'created: "{time.strftime("%Y-%m-%d")}"',
f'tags: ["evaluation","retrieval","brain"]',
"---",
"",
f"# Vault Brain Retrieval Evaluation — {time.strftime('%Y-%m-%d')}",
"",
f"Test set: **{len(queries)} synthetic queries** generated from curated docs (decisions, runbook, topology).",
"",
"| Strategy | Recall@1 | Recall@5 | Recall@10 | Recall@20 | MRR |",
"|---|---|---|---|---|---|",
]
for s in strategies:
m = results[s]
lines.append(f"| {s} | {m['recall@1']:.4f} | {m['recall@5']:.4f} | {m['recall@10']:.4f} | {m['recall@20']:.4f} | {m['mrr']:.4f} |")
lines += ["", "## Interpretation", "",
"- **BM25** (sparse, lexical) — strong on exact identifiers, weak on paraphrase.",
"- **Dense** (nomic-embed-text v1.5, 768d) — strong on semantic match, weak on rare tokens.",
"- **Hybrid (RRF k=60)** — should beat both; if it doesn't, the index has gaps or the queries are too easy.",
"",
"Citations:",
"- Cormack, Clarke, Buettcher (2009) — *Reciprocal Rank Fusion outperforms Condorcet and individual rank learning methods*",
"- Lin et al. (2024) — RRF baseline reproductions on MS MARCO",
"- Voorhees (1999) — Mean Reciprocal Rank, TREC-8",
""]
report_path.write_text("\n".join(lines), encoding="utf-8")
print(f"\n[eval] report written to {report_path}")
print(json.dumps(results, indent=2))
def main():
ap = argparse.ArgumentParser()
sub = ap.add_subparsers(dest="cmd", required=True)
p_g = sub.add_parser("generate")
p_g.add_argument("--out", default=str(VAULT_PATH / ".eval-queries.json"))
p_g.add_argument("-n", type=int, default=30)
p_r = sub.add_parser("run")
p_r.add_argument("queries")
p_r.add_argument("--strategies", default="", help="Comma-list, e.g. bm25,dense,hybrid,hybrid_centrality")
args = ap.parse_args()
if args.cmd == "generate":
cmd_generate(args)
elif args.cmd == "run":
cmd_run(args)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,339 @@
"""
vault_synthesizer.py The brain that writes about itself.
For each Label-Propagation community of size >= MIN_SIZE, generate a
"community summary" note that:
1. Lists all members of the cluster (wikilinks)
2. Identifies the dominant theme (top TextRank phrases shared)
3. Asks the local LLM (qwen3.6 via LM Studio) to write a 5-sentence
synthesis describing what this cluster is about
4. Cites the most central member as the "anchor" of the cluster
5. Drops a note in 08 - Semantic/communities/
This makes the brain an *active* participant: it discovers structure (via
LPA), then describes that structure in human-readable form. New synthesis
notes themselves become high-degree hubs in the graph (each links to all
their cluster members), which dramatically improves PageRank centrality
and navigability.
Theoretical foundation (peer-reviewed):
- Park, J. S. et al. (2023). "Generative Agents." UIST 2023.
DOI: 10.1145/3586183.3606763. The "reflection" component of generative
agents writes higher-level abstractions over recent episodic memories.
This synthesizer applies the same pattern to graph communities.
- Raghavan, U. N., Albert, R., Kumara, S. (2007). "Near linear time
algorithm to detect community structures." Physical Review E 76(3),
036106. DOI: 10.1103/PhysRevE.76.036106. (Provides the communities
we summarize.)
- Mihalcea, R., Tarau, P. (2004). "TextRank: Bringing Order into Texts."
EMNLP 2004. (Provides the dominant phrases per community.)
Usage:
python vault_synthesizer.py --apply # synthesize all communities >= 5
python vault_synthesizer.py --apply --min-size 10 # only big clusters
python vault_synthesizer.py --apply --max-clusters 20 # cap LLM calls
"""
try:
from lmstudio_response import extract_lmstudio_content
except Exception:
from scripts.brain.lmstudio_response import extract_lmstudio_content
try:
from lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
except Exception:
from scripts.brain.lmstudio_policy import add_lmstudio_ttl, get_lmstudio_config, select_lmstudio_model
import argparse
import datetime as dt
import json
import os
import re
import sys
import time
import urllib.request
from collections import Counter, defaultdict
from pathlib import Path
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
VAULT_PATH = Path(os.environ.get("VAULT_PATH", r"C:\Users\Smedj\Documents\Obsidian Vault"))
COMMUNITIES_FILE = VAULT_PATH / ".vault-communities.json"
KEYPHRASES_FILE = VAULT_PATH / ".vault-keyphrases.json"
PAGERANK_FILE = VAULT_PATH / ".vault-pagerank.json"
COMMUNITY_DIR = VAULT_PATH / "08 - Semantic" / "communities"
LM_STUDIO_URL = get_lmstudio_config()["base_url"] + "/v1/chat/completions"
SYNTH_MODEL = os.environ.get("SYNTH_MODEL", get_lmstudio_config()["deep_model"])
# Tiered model preference for synthesis. Per the operator's correction (2026-04-29):
# Claude (this session) and GPT-5.5 (codex CLI) are flagships; qwen3.6 is fallback /
# token-saver for when both flagships are tapped out (cf. 04 - Quota & Cost/Strategy).
#
# This module only handles the qwen3.6 path (fastest to plumb, no auth). The claude-direct
# path is invoked by Claude itself writing the synthesis in-session (see existing
# community-XXXX notes written 2026-04-29). The gpt-5.5 path uses codex CLI in --exec mode.
PROMPT_FOR_FLAGSHIP_PATH = """If you want Claude or GPT-5.5 to write this synthesis directly
instead of calling qwen3.6, paste the prompt below into the active session:
---
{prompt}
---
Then write the resulting markdown into:
{out_path}
"""
SYNTH_PROMPT = """You are summarizing a cluster of related notes from an engineering knowledge base.
The cluster has the following dominant phrases (most-shared TextRank keywords across notes):
{phrases}
The cluster contains {n} notes. Here are titles of representative members:
{titles}
Write EXACTLY:
1. A 1-line theme name (under 8 words) that captures what this cluster is about.
2. A 4-5 sentence synthesis describing the common thread, recurring decisions, and any tensions.
3. One concrete TODO that an agent should pick up to either resolve a tension or extend the cluster.
Format strictly as YAML:
theme: "..."
synthesis: "..."
todo: "..."
Do not add prose, explanations, or markdown fences. Output ONLY the YAML."""
def call_llm(prompt: str, retries: int = 2, timeout: int = 90) -> str | None:
try:
with urllib.request.urlopen(get_lmstudio_config()["base_url"] + "/v1/models", timeout=5) as resp:
available_models = [m["id"] for m in json.loads(resp.read().decode("utf-8")).get("data", [])]
model = select_lmstudio_model(
task_type="long_synthesis",
requested_model=SYNTH_MODEL,
automatic=True,
available_models=available_models,
)
except Exception as e:
print(f"[synth] LM Studio model selection failed: {e}", file=sys.stderr)
return None
payload = {
"model": model,
"messages": [
{"role": "system", "content": "/no_think\nOutput only the final answer. No hidden reasoning. No markdown unless explicitly requested."},
{"role": "user", "content": "/no_think\n" + prompt},
],
"temperature": 0.3,
"max_tokens": 600,
}
payload = json.dumps(add_lmstudio_ttl(payload)).encode("utf-8")
for attempt in range(retries):
try:
req = urllib.request.Request(LM_STUDIO_URL, data=payload, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=timeout) as resp:
d = json.loads(resp.read().decode("utf-8"))
return extract_lmstudio_content(d["choices"][0], expect_json=False)
except Exception as e:
if attempt == retries - 1:
print(f"[synth] LLM call failed: {e}", file=sys.stderr)
time.sleep(1)
return None
def parse_synth_yaml(raw: str) -> dict | None:
if not raw:
return None
raw = raw.strip()
# Strip code fences
raw = re.sub(r"^```(?:yaml)?\s*", "", raw)
raw = re.sub(r"\s*```$", "", raw)
out = {}
for line in raw.splitlines():
m = re.match(r'\s*(theme|synthesis|todo)\s*:\s*"?(.+?)"?\s*$', line)
if m:
out[m.group(1)] = m.group(2).strip().strip('"').strip("'")
if "theme" in out and "synthesis" in out:
return out
return None
def build_community_phrases(community_paths: list[str], keyphrases: dict[str, list[str]],
top: int = 8) -> list[tuple[str, int]]:
counter = Counter()
for path in community_paths:
for p in keyphrases.get(path, []):
counter[p] += 1
return counter.most_common(top)
def render_note(community_id: int, paths: list[str], theme: str, synthesis: str, todo: str,
top_phrases: list[tuple[str, int]], anchor: str | None) -> str:
lines = [
"---",
f'type: "community-synthesis"',
f'community_id: {community_id}',
f'created: "{dt.datetime.now(dt.timezone.utc).isoformat()}"',
f'member_count: {len(paths)}',
f'tags: ["community-synthesis","auto-generated","layer-2"]',
"---",
"",
f"# Community {community_id}{theme}",
"",
"## Synthesis",
"",
synthesis,
"",
"## TODO",
"",
f"- {todo}",
"",
"## Anchor",
"",
f"- [[{anchor[:-3] if anchor and anchor.endswith('.md') else anchor}]]" if anchor else "*(no anchor)*",
"",
"## Dominant phrases",
"",
]
for p, c in top_phrases:
lines.append(f"- `{p}` ({c} notes)")
lines += ["", "## Members", ""]
for p in paths[:60]:
target = p[:-3] if p.endswith(".md") else p
lines.append(f"- [[{target.replace(chr(92), '/')}]]")
if len(paths) > 60:
lines.append(f"- *(...and {len(paths) - 60} more — see graph)*")
lines += [
"",
"---",
"",
"*Auto-synthesized from Label-Propagation community detection ([Raghavan, Albert & Kumara, "
"Phys. Rev. E 76(3), 2007](https://doi.org/10.1103/PhysRevE.76.036106)) over the densified wikilink graph. "
"Synthesis written by `qwen3.6-35b-a3b` (local, no API cost).*",
]
return "\n".join(lines)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--apply", action="store_true")
ap.add_argument("--min-size", type=int, default=5)
ap.add_argument("--max-clusters", type=int, default=15)
ap.add_argument("--rewrite", action="store_true")
ap.add_argument("--emit-prompts", action="store_true",
help="Don't call any LLM — print prompts for Claude/GPT to fill in interactively. "
"Use this in active sessions where the operator has Claude or GPT-5.5 ready.")
ap.add_argument("--via-codex", action="store_true",
help="Call OpenAI Codex CLI (gpt-5.5) instead of LM Studio (qwen3.6). "
"Faster but uses ChatGPT quota.")
args = ap.parse_args()
if not COMMUNITIES_FILE.exists():
print("[synth] missing communities — run vault_communities.py detect")
sys.exit(1)
if not KEYPHRASES_FILE.exists():
print("[synth] missing keyphrases — run vault_textrank.py extract")
sys.exit(1)
com = json.loads(COMMUNITIES_FILE.read_text(encoding="utf-8"))
nodes = com["nodes"]
labels = com["labels"]
keyphrases = json.loads(KEYPHRASES_FILE.read_text(encoding="utf-8"))
pr = {}
if PAGERANK_FILE.exists():
pr = json.loads(PAGERANK_FILE.read_text(encoding="utf-8")).get("pagerank", {})
# Group nodes by community
members: dict[int, list[str]] = defaultdict(list)
for i, lbl in enumerate(labels):
members[lbl].append(nodes[i])
eligible = sorted(
[(cid, paths) for cid, paths in members.items() if len(paths) >= args.min_size],
key=lambda x: -len(x[1]),
)
print(f"[synth] {len(eligible)} communities with size >= {args.min_size}")
eligible = eligible[: args.max_clusters]
print(f"[synth] processing top {len(eligible)} (capped by --max-clusters)")
if args.apply:
COMMUNITY_DIR.mkdir(parents=True, exist_ok=True)
written = 0
for cid, paths in eligible:
out_path = COMMUNITY_DIR / f"community-{cid:04d}.md"
if out_path.exists() and not args.rewrite:
print(f"[synth] community {cid}: file exists, skip (use --rewrite to overwrite)")
continue
top_phrases = build_community_phrases(paths, keyphrases, top=8)
if not top_phrases:
print(f"[synth] community {cid}: no phrases, skipping")
continue
# Pick anchor = highest-PageRank member
anchor = max(paths, key=lambda p: pr.get(p, 0.0)) if pr else paths[0]
# Sample titles for prompt (first chunk = title)
titles = []
for p in paths[:8]:
stem = Path(p).stem.replace("-", " ").replace("_", " ")
stem = re.sub(r"^\d{2}-\d{2}-\d{2}\s*", "", stem)
titles.append(f" - {stem[:80]}")
prompt = SYNTH_PROMPT.format(
phrases=", ".join(f"`{p}`" for p, _ in top_phrases),
n=len(paths),
titles="\n".join(titles),
)
if args.emit_prompts:
# Print the prompt so Claude or GPT-5.5 (in active session) can fill it in.
print("=" * 80)
print(PROMPT_FOR_FLAGSHIP_PATH.format(prompt=prompt, out_path=out_path))
print("=" * 80)
continue
if args.via_codex:
print(f"[synth] community {cid:4d}: routing via Codex CLI (gpt-5.5)…")
try:
import subprocess as _sp
t0 = time.time()
res = _sp.run(["codex", "exec", "--skip-git-repo-check", prompt],
capture_output=True, text=True, timeout=120)
raw = res.stdout.strip()
elapsed = time.time() - t0
except Exception as e:
print(f"[synth] codex failed: {e}; falling back to qwen3.6")
t0 = time.time()
raw = call_llm(prompt)
elapsed = time.time() - t0
else:
print(f"[synth] community {cid:4d} ({len(paths)} notes): calling qwen3.6 (fallback tier)…")
t0 = time.time()
raw = call_llm(prompt)
elapsed = time.time() - t0
if not raw:
print(f"[synth] community {cid}: LLM returned nothing ({elapsed:.0f}s)")
continue
parsed = parse_synth_yaml(raw)
if not parsed:
print(f"[synth] community {cid}: parse failed; raw[:120]={raw[:120]!r}")
continue
theme = parsed.get("theme", "(untitled)")
synthesis = parsed.get("synthesis", "")
todo = parsed.get("todo", "")
print(f"[synth] community {cid}: '{theme}' ({elapsed:.0f}s)")
note = render_note(cid, paths, theme, synthesis, todo, top_phrases, anchor)
if args.apply:
out_path.write_text(note, encoding="utf-8")
written += 1
print()
print(f"[synth] wrote {written} synthesis note(s) to {COMMUNITY_DIR}")
print(f"[synth] mode: {'APPLIED' if args.apply else 'DRY-RUN'}")
if __name__ == "__main__":
main()