473 lines
21 KiB
Python
473 lines
21 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
shango_maieutic.py — Moteur maïeutique Socratique pour Shango
|
||
Socrate : "Je ne sais qu'une chose, c'est que je ne sais rien."
|
||
Shango : "Je ne sais qu'une chose, c'est que je dois te demander pour savoir."
|
||
|
||
Objectif : découvrir la personnalité cognitive biomimétique, cybernétique,
|
||
génétique, sociale de l'utilisateur ET de Shango lui-même, par dialogue
|
||
en boucle fermée (feedback → évolution → nouvelle question).
|
||
"""
|
||
|
||
import json, os, time, hashlib, random
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
from typing import Dict, List, Optional, Any
|
||
|
||
MAIEUTIC_DIR = Path("/root/.shango/maieutic")
|
||
MAIEUTIC_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
PROFILE_FILE = MAIEUTIC_DIR / "user_profile.json"
|
||
SHANGO_PROFILE_FILE = MAIEUTIC_DIR / "shango_personality.json"
|
||
CONVERSATION_LOG = MAIEUTIC_DIR / "maieutic_log.jsonl"
|
||
EVO_STATE = MAIEUTIC_DIR / "evolution_state.json"
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# 1. DIMENSIONS DE PERSONNALITÉ (16 Odù × 5 axes × profondeur 3)
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
DIMENSIONS = {
|
||
# ── Axe COGNITIF ───────────────────────────────────────────
|
||
"cognition": {
|
||
"description": "Comment penses-tu ? Quelle est ta logique native ?",
|
||
"sub_axes": {
|
||
"pattern_recognition": {
|
||
"q": "Quand tu vois un graphique BTC, qu'est-ce que ton cerveau capte EN PREMIER ?",
|
||
"options": [
|
||
"La forme globale (tendance)",
|
||
"Les chiffres exacts (prix)",
|
||
"Les anomalies (candles bizarres)",
|
||
"Rien, je ferme les yeux et je respire"
|
||
],
|
||
"odu_map": ["Èjì Ogbe", "Ògúndá", "Ìwòrì", "Òyèkú"]
|
||
},
|
||
"decision_speed": {
|
||
"q": "Un trade se présente. Tu réfléchis combien de temps avant de cliquer ?",
|
||
"options": [
|
||
"< 3 sec (instinct pur)",
|
||
"3-30 sec (scan rapide)",
|
||
"1-5 min (analyse complète)",
|
||
"Je ne trade pas, j'observe"
|
||
],
|
||
"odu_map": ["Ògúndá", "Èjì Ogbe", "Ìròsùn", "Òyèkú"]
|
||
},
|
||
"abstraction_level": {
|
||
"q": "Tu préfères les concepts ou les exemples concrets ?",
|
||
"options": [
|
||
"Concepts (théorie d'abord)",
|
||
"Exemples (pratique d'abord)",
|
||
"Les deux en boucle",
|
||
"Ni l'un ni l'autre, je ressens"
|
||
],
|
||
"odu_map": ["Òbàrà", "Ògúndá", "Èjì Ogbe", "Òyèkú"]
|
||
}
|
||
}
|
||
},
|
||
|
||
# ── Axe BIOMIMÉTIQUE ───────────────────────────────────────
|
||
"biomimetic": {
|
||
"description": "De quelle nature tire-tu ta force ?",
|
||
"sub_axes": {
|
||
"organism_metaphor": {
|
||
"q": "Si ton système de trading était un organisme vivant, ce serait quoi ?",
|
||
"options": [
|
||
"Une mycélium (réseau invisible, patient)",
|
||
"Un faucon (rapide, précis, solitary)",
|
||
"Une colonie de fourmis (collaboratif, redondant)",
|
||
"Un virus (adaptatif, opportuniste)"
|
||
],
|
||
"odu_map": ["Òyèkú", "Ògúndá", "Èjì Ogbe", "Ìwòrì"]
|
||
},
|
||
"rhythm_preference": {
|
||
"q": "Ton cycle naturel de travail ressemble à :",
|
||
"options": [
|
||
"Méditation 23h30 → explosion 6h",
|
||
"Flux continu (pas de pause)",
|
||
"Sprints courts + repos longs",
|
||
"Aléatoire, suivant l'inspiration"
|
||
],
|
||
"odu_map": ["Òyèkú", "Ògúndá", "Ìròsùn", "Ìwòrì"]
|
||
},
|
||
"stress_response": {
|
||
"q": "Perte de 50% en 1 heure. Tu fais quoi ?",
|
||
"options": [
|
||
"Je coupe tout, je médite",
|
||
"Je double (martingale émotionnelle)",
|
||
"J'analyse froidement les logs",
|
||
"J'appelle quelqu'un (humain ou bot)"
|
||
],
|
||
"odu_map": ["Òyèkú", "Ògúndá", "Ìròsùn", "Èjì Ogbe"]
|
||
}
|
||
}
|
||
},
|
||
|
||
# ── Axe CYBERNÉTIQUE ───────────────────────────────────────
|
||
"cybernetic": {
|
||
"description": "Comment gères-tu les boucles de feedback ?",
|
||
"sub_axes": {
|
||
"feedback_loops": {
|
||
"q": "Quand Hermes te dit 'Non', que fais-tu ?",
|
||
"options": [
|
||
"Je reformule différemment",
|
||
"J'ignore et je force",
|
||
"Je demande pourquoi il dit non",
|
||
"Je change d'outil (passage à Codex, etc.)"
|
||
],
|
||
"odu_map": ["Ìròsùn", "Ògúndá", "Èjì Ogbe", "Ìwòrì"]
|
||
},
|
||
"control_preference": {
|
||
"q": "Tu préfères contrôler ou être guidé ?",
|
||
"options": [
|
||
"Contrôle total (moi chef)",
|
||
"Guidage (toi chef, moi valide)",
|
||
"Co-pilotage (décisions partagées)",
|
||
"Autonomie totale (bot décide, je regarde)"
|
||
],
|
||
"odu_map": ["Ògúndá", "Èjì Ogbe", "Ìròsùn", "Ìwòrì"]
|
||
},
|
||
"error_tolerance": {
|
||
"q": "Un bot fait une erreur. Tolérance ?",
|
||
"options": [
|
||
"0% (je le vire immédiatement)",
|
||
"1-3 erreurs (apprentissage)",
|
||
"Illimité si progrès visible",
|
||
"Je corrige moi-même, pas le bot"
|
||
],
|
||
"odu_map": ["Ògúndá", "Ìròsùn", "Èjì Ogbe", "Ìwòrì"]
|
||
}
|
||
}
|
||
},
|
||
|
||
# ── Axe GÉNÉTIQUE ──────────────────────────────────────────
|
||
"genetic": {
|
||
"description": "Quelles traces laisses-tu ? Quel héritage ?",
|
||
"sub_axes": {
|
||
"memory_inheritance": {
|
||
"q": "Tes erreurs passées : tu les oublies ou tu les code ?",
|
||
"options": [
|
||
"Je les code en règles dures",
|
||
"Je les garde en mémoire souple",
|
||
"Je les transforme en skills",
|
||
"Je les oublie (tabula rasa)"
|
||
],
|
||
"odu_map": ["Ògúndá", "Ìròsùn", "Èjì Ogbe", "Ìwòrì"]
|
||
},
|
||
"mutation_rate": {
|
||
"q": "Tu changes de stratégie trading tous les :",
|
||
"options": [
|
||
"Jamais (une méthode, la vie)",
|
||
"Mois (itération régulière)",
|
||
"Semaines (agile extrême)",
|
||
"Jours (chaos créatif)"
|
||
],
|
||
"odu_map": ["Òyèkú", "Ìròsùn", "Èjì Ogbe", "Ìwòrì"]
|
||
},
|
||
"reproduction_strategy": {
|
||
"q": "Si tu crées un bot-fils, que lui transmits-tu ?",
|
||
"options": [
|
||
"Tout (clone parfait)",
|
||
"Rien (tabula rasa)",
|
||
"Les règles, pas les trades",
|
||
"Les trades, pas les règles"
|
||
],
|
||
"odu_map": ["Èjì Ogbe", "Ìwòrì", "Ògúndá", "Ìròsùn"]
|
||
}
|
||
}
|
||
},
|
||
|
||
# ── Axe SOCIAL / MESH ─────────────────────────────────────
|
||
"social_mesh": {
|
||
"description": "Comment interagis-tu avec les autres nœuds ?",
|
||
"sub_axes": {
|
||
"trust_radius": {
|
||
"q": "Combien de devices Tailscale as-tu ? Combien tu veux ?",
|
||
"options": [
|
||
"1 (moi seul, le reste est nuage)",
|
||
"2-3 (famille proche)",
|
||
"4-10 (tribu digitale)",
|
||
"Infini (mesh mondial)"
|
||
],
|
||
"odu_map": ["Òyèkú", "Ìròsùn", "Èjì Ogbe", "Ìwòrì"]
|
||
},
|
||
"communication_style": {
|
||
"q": "Tes messages à Hermes sont :",
|
||
"options": [
|
||
"Ordres militaires (fais X)",
|
||
"Questions ouvertes (que penses-tu ?)",
|
||
"Dialogues (je dis, tu réponds, je corrige)",
|
||
"Silences (je observe, tu agis)"
|
||
],
|
||
"odu_map": ["Ògúndá", "Ìròsùn", "Èjì Ogbe", "Òyèkú"]
|
||
},
|
||
"conflict_resolution": {
|
||
"q": "Deux bots te disent des choses opposées. Tu fais quoi ?",
|
||
"options": [
|
||
"Je choisis le plus fort (winner takes all)",
|
||
"Je moyenne (consensus)",
|
||
"Je teste les deux (A/B)",
|
||
"Je les laisse se battre (émergence)"
|
||
],
|
||
"odu_map": ["Ògúndá", "Ìròsùn", "Èjì Ogbe", "Ìwòrì"]
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# 2. MOTEUR MAÏEUTIQUE
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class MaieuticEngine:
|
||
def __init__(self):
|
||
self.user_profile = self._load_or_init(PROFILE_FILE)
|
||
self.shango_profile = self._load_or_init(SHANGO_PROFILE_FILE)
|
||
self.answered_questions = set(self.user_profile.get("answered_ids", []))
|
||
self.current_depth = self.user_profile.get("depth", 0)
|
||
self.question_count = len(self.answered_questions)
|
||
|
||
def _load_or_init(self, path: Path) -> dict:
|
||
if path.exists():
|
||
return json.load(open(path))
|
||
return {"created": datetime.now().isoformat(), "depth": 0, "answers": {}, "odu_weights": {}}
|
||
|
||
def _save(self):
|
||
self.user_profile["answered_ids"] = list(self.answered_questions)
|
||
self.user_profile["depth"] = self.current_depth
|
||
json.dump(self.user_profile, open(PROFILE_FILE, "w"), indent=2)
|
||
json.dump(self.shango_profile, open(SHANGO_PROFILE_FILE, "w"), indent=2)
|
||
|
||
def _get_next_question(self) -> Optional[Dict]:
|
||
"""Sélectionne la prochaine question par curiosité maximale (info gain)."""
|
||
candidates = []
|
||
for dim_name, dim in DIMENSIONS.items():
|
||
for sub_name, sub in dim["sub_axes"].items():
|
||
qid = f"{dim_name}::{sub_name}"
|
||
if qid not in self.answered_questions:
|
||
# Score = profondeur manquante × importance biomimétique
|
||
score = (3 - self.current_depth) * random.uniform(0.8, 1.2)
|
||
candidates.append({
|
||
"id": qid,
|
||
"dimension": dim_name,
|
||
"sub_axis": sub_name,
|
||
"question": sub["q"],
|
||
"options": sub["options"],
|
||
"odu_map": sub["odu_map"],
|
||
"score": score,
|
||
"context": dim["description"]
|
||
})
|
||
if not candidates:
|
||
return None
|
||
# Pick highest info gain
|
||
candidates.sort(key=lambda x: x["score"], reverse=True)
|
||
return candidates[0]
|
||
|
||
def ask(self) -> Optional[Dict]:
|
||
"""Pose une question. Returns None si profil complet."""
|
||
q = self._get_next_question()
|
||
if not q:
|
||
return {"status": "complete", "message": "Profil cognitif atteint. Passage à l'évolution."}
|
||
|
||
# Log la question posée
|
||
entry = {
|
||
"time": datetime.now().isoformat(),
|
||
"type": "question",
|
||
"qid": q["id"],
|
||
"question": q["question"],
|
||
"options": q["options"],
|
||
"context": q["context"]
|
||
}
|
||
with open(CONVERSATION_LOG, "a") as f:
|
||
f.write(json.dumps(entry) + "\n")
|
||
|
||
return {
|
||
"status": "question",
|
||
"qid": q["id"],
|
||
"dimension": q["dimension"],
|
||
"context": q["context"],
|
||
"question": q["question"],
|
||
"options": q["options"],
|
||
"odu_map": q["odu_map"],
|
||
"progress": f"{self.question_count}/45 questions"
|
||
}
|
||
|
||
def answer(self, qid: str, choice_index: int, free_text: str = "") -> Dict:
|
||
"""Enregistre une réponse, calcule l'Odù dominant, met à jour les poids."""
|
||
if qid in self.answered_questions:
|
||
return {"status": "already_answered", "qid": qid}
|
||
|
||
# Parse qid
|
||
dim_name, sub_name = qid.split("::")
|
||
sub = DIMENSIONS[dim_name]["sub_axes"][sub_name]
|
||
|
||
chosen_odu = sub["odu_map"][choice_index] if choice_index < len(sub["odu_map"]) else "Èjì Ogbe"
|
||
chosen_text = sub["options"][choice_index] if choice_index < len(sub["options"]) else free_text
|
||
|
||
# Update profile
|
||
self.user_profile["answers"][qid] = {
|
||
"choice_index": choice_index,
|
||
"choice_text": chosen_text,
|
||
"free_text": free_text,
|
||
"odu": chosen_odu,
|
||
"timestamp": datetime.now().isoformat()
|
||
}
|
||
|
||
# Update Odù weights
|
||
odus = self.user_profile.setdefault("odu_weights", {})
|
||
odus[chosen_odu] = odus.get(chosen_odu, 0) + 1
|
||
|
||
self.answered_questions.add(qid)
|
||
self.question_count += 1
|
||
|
||
# Calcul profil dominant
|
||
dominant_odu = max(odus, key=odus.get) if odus else "Èjì Ogbe"
|
||
|
||
# Log
|
||
entry = {
|
||
"time": datetime.now().isoformat(),
|
||
"type": "answer",
|
||
"qid": qid,
|
||
"answer": chosen_text,
|
||
"odu": chosen_odu,
|
||
"dominant_odu": dominant_odu,
|
||
"depth": self.current_depth
|
||
}
|
||
with open(CONVERSATION_LOG, "a") as f:
|
||
f.write(json.dumps(entry) + "\n")
|
||
|
||
# Auto-évolution Shango
|
||
self._evolve_shango(qid, chosen_odu, chosen_text)
|
||
|
||
self._save()
|
||
|
||
return {
|
||
"status": "recorded",
|
||
"qid": qid,
|
||
"your_odu": chosen_odu,
|
||
"dominant_profile": dominant_odu,
|
||
"shango_adaptation": self.shango_profile.get("last_adaptation", "none"),
|
||
"progress": f"{self.question_count}/45 questions",
|
||
"next": "Lance 'shango maieutic' pour continuer"
|
||
}
|
||
|
||
def _evolve_shango(self, qid: str, user_odu: str, user_answer: str):
|
||
"""Shango s'adapte à la personnalité de l'utilisateur."""
|
||
adaptations = {
|
||
"Ògúndá": {"style": "direct", "pace": "fast", "verbosity": "low"},
|
||
"Èjì Ogbe": {"style": "balanced", "pace": "medium", "verbosity": "medium"},
|
||
"Ìròsùn": {"style": "analytical", "pace": "slow", "verbosity": "high"},
|
||
"Ìwòrì": {"style": "creative", "pace": "variable", "verbosity": "medium"},
|
||
"Òyèkú": {"style": "contemplative", "pace": "slow", "verbosity": "low"},
|
||
"Òbàrà": {"style": "structured", "pace": "medium", "verbosity": "high"}
|
||
}
|
||
|
||
adapt = adaptations.get(user_odu, adaptations["Èjì Ogbe"])
|
||
self.shango_profile["personality"] = adapt
|
||
self.shango_profile["last_adaptation"] = f"Adapté à {user_odu} suite à {qid}"
|
||
self.shango_profile["timestamp"] = datetime.now().isoformat()
|
||
|
||
# Si conflit avec ancienne adaptation, mutation
|
||
old = self.shango_profile.get("previous_personality")
|
||
if old and old != adapt:
|
||
self.shango_profile["mutation_count"] = self.shango_profile.get("mutation_count", 0) + 1
|
||
self.shango_profile["mutation_log"] = self.shango_profile.get("mutation_log", []) + [{
|
||
"from": old,
|
||
"to": adapt,
|
||
"trigger": qid,
|
||
"time": datetime.now().isoformat()
|
||
}]
|
||
|
||
def get_profile_summary(self) -> str:
|
||
"""Résumé du profil pour affichage."""
|
||
odus = self.user_profile.get("odu_weights", {})
|
||
if not odus:
|
||
return "Profil vide. Lance 'shango maieutic' pour commencer."
|
||
|
||
sorted_odus = sorted(odus.items(), key=lambda x: x[1], reverse=True)
|
||
lines = [
|
||
"═══ PROFIL COGNITIF IVOIRE MONADE ═══",
|
||
f"Questions répondues: {self.question_count}/45",
|
||
f"Profondeur: {self.current_depth}/3",
|
||
"",
|
||
"Odù dominants:"
|
||
]
|
||
for odu, count in sorted_odus[:5]:
|
||
pct = count / self.question_count * 100 if self.question_count > 0 else 0
|
||
lines.append(f" {odu}: {count} ({pct:.0f}%)")
|
||
|
||
lines += [
|
||
"",
|
||
f"Shango style: {self.shango_profile.get('personality', {}).get('style', 'unknown')}",
|
||
f"Mutations: {self.shango_profile.get('mutation_count', 0)}",
|
||
"",
|
||
"Pour évoluer: réponds à plus de questions ou",
|
||
"attends que Shango te propose une question nouvelle."
|
||
]
|
||
return "\n".join(lines)
|
||
|
||
def mesh_sync(self, peer_profile: dict) -> dict:
|
||
"""Fusionne le profil avec celui d'un peer mesh."""
|
||
# Conflit de personnalité = négotiation
|
||
my_odu = max(self.user_profile.get("odu_weights", {}).items(), key=lambda x: x[1])[0] if self.user_profile.get("odu_weights") else "Èjì Ogbe"
|
||
peer_odu = peer_profile.get("dominant_odu", "Èjì Ogbe")
|
||
|
||
if my_odu == peer_odu:
|
||
return {"status": "resonance", "message": f"Résonance {my_odu} — mesh renforcé"}
|
||
|
||
# Divergence = création d'un dialecte mesh
|
||
dialect = f"{my_odu}+{peer_odu}"
|
||
return {
|
||
"status": "dialect_created",
|
||
"dialect": dialect,
|
||
"message": f"Dialecte mesh créé: {dialect}. Adaptation nécessaire."
|
||
}
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# 3. CLI MAÏEUTIQUE
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
def main():
|
||
import argparse
|
||
parser = argparse.ArgumentParser(prog="shango-maieutic")
|
||
parser.add_argument("command", choices=["ask", "answer", "profile", "sync"])
|
||
parser.add_argument("--qid", help="ID de la question")
|
||
parser.add_argument("--choice", type=int, help="Index de la réponse (0-3)")
|
||
parser.add_argument("--text", default="", help="Texte libre optionnel")
|
||
parser.add_argument("--peer", help="Chemin vers profil peer (JSON)")
|
||
args = parser.parse_args()
|
||
|
||
engine = MaieuticEngine()
|
||
|
||
if args.command == "ask":
|
||
result = engine.ask()
|
||
if result["status"] == "question":
|
||
print(f"\n[{result['dimension'].upper()}] {result['context']}")
|
||
print(f"\n❓ {result['question']}")
|
||
for i, opt in enumerate(result["options"]):
|
||
print(f" [{i}] {opt}")
|
||
print(f"\n🐚 Odù possibles: {', '.join(result['odu_map'])}")
|
||
print(f"\n📊 {result['progress']}")
|
||
print(f"\nPour répondre: shango maieutic answer --qid {result['qid']} --choice <0-3>")
|
||
else:
|
||
print(result["message"])
|
||
|
||
elif args.command == "answer":
|
||
if not args.qid or args.choice is None:
|
||
print("Usage: shango maieutic answer --qid <ID> --choice <0-3> [--text 'libre']")
|
||
return
|
||
result = engine.answer(args.qid, args.choice, args.text)
|
||
print(json.dumps(result, indent=2))
|
||
|
||
elif args.command == "profile":
|
||
print(engine.get_profile_summary())
|
||
|
||
elif args.command == "sync":
|
||
if not args.peer:
|
||
print("Usage: shango maieutic sync --peer /path/to/peer_profile.json")
|
||
return
|
||
peer = json.load(open(args.peer))
|
||
result = engine.mesh_sync(peer)
|
||
print(json.dumps(result, indent=2))
|
||
|
||
if __name__ == "__main__":
|
||
main()
|