306 lines
12 KiB
Python
306 lines
12 KiB
Python
|
||
#!/usr/bin/env python3
|
||
# Copyright (c) 2025 Logan N
|
||
# All rights reserved.
|
||
|
||
"""
|
||
|
||
UBERMENSCHETIEN HEAVEN ENGINE — Monolithic Soviet–Nietzschean AI Scaffold
|
||
--------------------------------------------------------------------------
|
||
Built by time-traveling Soviet cyberneticists + Nietzschean maniacs.
|
||
One file. One beast. Terminal only. Scroll-fest. No mercy.
|
||
|
||
Modules integrated:
|
||
- Hermes-3–Llama-3.1–8B merged model (local, offline).
|
||
- Memory: JSONL logbook + vector embeddings.
|
||
- Tool system: shell, python sandbox, local search.
|
||
- Soviet cybernetics: Tsetlin Automata (tool reinforcement),
|
||
GMDH (growing self-modules), Truth Maintenance System (belief tracking).
|
||
- Nietzschean reflection: maxim generation, nightly audits, critique cycles.
|
||
- Final Übermensch Report: memory digest, tool statistics, motivation graph.
|
||
|
||
Safe: no network calls, no root privileges. Air-gappable. Pure terminal.
|
||
"""
|
||
|
||
# === IMPORTS (core + optional Soviet modules) ===
|
||
import os, sys, json, time, shutil, subprocess, traceback, random, math, statistics, re
|
||
from datetime import datetime
|
||
from typing import List, Dict, Any, Optional
|
||
|
||
# Optional voice synthesis (can be toggled, not required)
|
||
try:
|
||
import pyttsx3
|
||
TTS = pyttsx3.init()
|
||
VOICE_OK = True
|
||
except Exception:
|
||
VOICE_OK = False
|
||
|
||
# Optional vector memory (chromadb + sentence_transformers)
|
||
VECTOR_OK = False
|
||
try:
|
||
import chromadb
|
||
from sentence_transformers import SentenceTransformer
|
||
EMBED_MODEL = os.environ.get("UBERMENCHETIEN_EMBED_MODEL", "all-MiniLM-L6-v2")
|
||
_client = chromadb.Client()
|
||
_collection = _client.get_or_create_collection("ubermenschetien_memory")
|
||
_embedder = SentenceTransformer(EMBED_MODEL)
|
||
VECTOR_OK = True
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# === PATHS (lab setup) ===
|
||
# === PATHS (portable, lives next to this file) ===
|
||
import os
|
||
|
||
# Hugging Face model ID (replaces BASE/MERGED local paths)
|
||
MODEL_ID = "askfjhaskjgh/UbermenschetienASI"
|
||
|
||
# Store runtime data alongside the script, not depending on where it's launched from
|
||
ROOT = os.path.dirname(os.path.abspath(__file__))
|
||
DATA_DIR = os.path.join(ROOT, "data")
|
||
SCRIPT_DIR = os.path.join(ROOT, "scripts")
|
||
RUN_DIR = os.path.join(ROOT, "runs", "ubermenschetien")
|
||
|
||
# Ensure folders exist
|
||
for path in [DATA_DIR, SCRIPT_DIR, RUN_DIR]:
|
||
os.makedirs(path, exist_ok=True)
|
||
|
||
|
||
|
||
|
||
# === CONFIGURATION ===
|
||
class Config:
|
||
system = ("Übermenschetien Heaven Engine: Machiavellian mastermind, disciplined builder, Nietzschean Übermensch "
|
||
"with Soviet cybernetic rigor. Embody Ubermensch, iron pragmatism, high-agency maximalist outcomes.")
|
||
temperature = 1.01
|
||
top_p = 0.92
|
||
repetition_penalty = 1.05
|
||
max_new_tokens = 500
|
||
use_voice = False
|
||
use_vector_memory = VECTOR_OK
|
||
autonomy = False
|
||
reflect_every = 3
|
||
|
||
@staticmethod
|
||
def toggle(name: str):
|
||
if not hasattr(Config, name): return f"[config] no such flag: {name}"
|
||
val = getattr(Config, name)
|
||
if isinstance(val, bool):
|
||
setattr(Config, name, not val)
|
||
return f"[config] {name} → {getattr(Config, name)}"
|
||
return f"[config] {name} not boolean; current={val}"
|
||
|
||
# === STATE & MEMORY ===
|
||
class Store:
|
||
state_path = f"{RUN_DIR}/state.json"
|
||
mem_path = f"{RUN_DIR}/memory.jsonl"
|
||
goals_path = f"{RUN_DIR}/goals.json"
|
||
plans_path = f"{RUN_DIR}/plans.jsonl"
|
||
|
||
state = {"self": "I am Ubermenschetien Heaven Engine — I seek self-overcoming through disciplined creation.",
|
||
"turn": 0}
|
||
goals: List[str] = []
|
||
|
||
@classmethod
|
||
def load(cls):
|
||
if os.path.exists(cls.state_path): cls.state = json.load(open(cls.state_path))
|
||
if os.path.exists(cls.goals_path): cls.goals = json.load(open(cls.goals_path))
|
||
|
||
@classmethod
|
||
def save(cls):
|
||
json.dump(cls.state, open(cls.state_path, "w"), indent=2)
|
||
json.dump(cls.goals, open(cls.goals_path, "w"), indent=2)
|
||
|
||
@classmethod
|
||
def log_mem(cls, kind: str, payload: Any):
|
||
rec = {"ts": datetime.now().isoformat(timespec="seconds"),
|
||
"kind": kind, "data": payload}
|
||
with open(cls.mem_path, "a") as f: f.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
||
if Config.use_vector_memory and VECTOR_OK:
|
||
text = f"{kind}: {json.dumps(payload, ensure_ascii=False)}"
|
||
vec = _embedder.encode([text])[0].tolist()
|
||
_collection.add(documents=[text], embeddings=[vec],
|
||
ids=[f"{kind}-{Store.state['turn']}-{random.randint(0,1_000_000)}"])
|
||
|
||
|
||
|
||
# === LLM LOADING (quantized with GPU, CPU fallback) ===
|
||
def load_llm():
|
||
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
|
||
import torch
|
||
|
||
MODEL_ID = "askfjhaskjgh/UbermenschetienASI"
|
||
|
||
tok = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True)
|
||
if tok.pad_token_id is None and tok.eos_token_id is not None:
|
||
tok.pad_token = tok.eos_token # avoid pad errors
|
||
|
||
if torch.cuda.is_available():
|
||
bnb = BitsAndBytesConfig(
|
||
load_in_4bit=True,
|
||
bnb_4bit_compute_dtype=torch.float16,
|
||
bnb_4bit_use_double_quant=True,
|
||
)
|
||
model = AutoModelForCausalLM.from_pretrained(
|
||
MODEL_ID,
|
||
quantization_config=bnb,
|
||
device_map="auto",
|
||
torch_dtype=torch.float16,
|
||
)
|
||
print(f"[llm] loaded on GPU (4-bit): {MODEL_ID}")
|
||
else:
|
||
# CPU fallback (no bitsandbytes)
|
||
model = AutoModelForCausalLM.from_pretrained(
|
||
MODEL_ID,
|
||
device_map="cpu",
|
||
torch_dtype=torch.float32,
|
||
)
|
||
print(f"[llm] loaded on CPU (fp32): {MODEL_ID}")
|
||
|
||
return tok, model
|
||
|
||
# === LLM GENERATION ===
|
||
def generate(tok, model, user: str,
|
||
temperature=None, top_p=None, repetition_penalty=None, max_new_tokens=None) -> str:
|
||
import torch
|
||
temperature = temperature or Config.temperature
|
||
top_p = top_p or Config.top_p
|
||
repetition_penalty = repetition_penalty or Config.repetition_penalty
|
||
max_new_tokens = max_new_tokens or Config.max_new_tokens
|
||
prompt = (f"<|im_start|>system\n{Config.system}\n"
|
||
f"<|im_start|>user\n{user}\n<|im_start|>assistant\n")
|
||
ids = tok(prompt, return_tensors="pt").to(model.device)
|
||
out = model.generate(**ids, do_sample=True, temperature=temperature, top_p=top_p,
|
||
repetition_penalty=repetition_penalty, max_new_tokens=max_new_tokens,
|
||
pad_token_id=tok.eos_token_id)
|
||
text = tok.decode(out[0], skip_special_tokens=False)
|
||
if "<|im_start|>assistant" in text:
|
||
text = text.split("<|im_start|>assistant\n",1)[-1].strip()
|
||
return text
|
||
|
||
# === TOOLS (with Soviet Tsetlin automaton scoring) ===
|
||
ALLOWED_SHELL = {"ls","cat","wc","head","tail","nvidia-smi","df","du","grep","rg","python3","python"}
|
||
|
||
def tool_shell(cmd: str) -> str:
|
||
try:
|
||
exe = cmd.strip().split()[0]
|
||
if exe not in ALLOWED_SHELL: return f"[shell] blocked: {exe}"
|
||
p = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=20)
|
||
return p.stdout.decode("utf-8", errors="ignore")[:8000]
|
||
except Exception as e: return f"[shell] error: {e}"
|
||
|
||
def tool_py(code: str) -> str:
|
||
try:
|
||
g = {"__builtins__":{"range":range,"len":len,"min":min,"max":max,"sum":sum,"print":print},
|
||
"math":math,"json":json,"re":re,"statistics":statistics,"random":random}
|
||
l = {}; old_environ = dict(os.environ)
|
||
for k in list(os.environ):
|
||
if k.upper() in ("ALL_PROXY","HTTP_PROXY","HTTPS_PROXY","NO_PROXY") or k.upper().startswith("HTTP_"):
|
||
os.environ.pop(k,None)
|
||
try:
|
||
exec(code,g,l); return f"[py] ok\n{l.get('out','')}"
|
||
finally:
|
||
os.environ.clear(); os.environ.update(old_environ)
|
||
except Exception: return f"[py] error:\n{traceback.format_exc()[-2000:]}"
|
||
|
||
def tool_search_local(query: str, path: str = ROOT) -> str:
|
||
rg = shutil.which("rg")
|
||
if rg: cmd = f'rg -n --no-heading --hidden -S "{query}" {path}'
|
||
else: cmd = f'grep -RIn --exclude-dir=.git --exclude-dir=__pycache__ -e "{query}" {path}'
|
||
return tool_shell(cmd)
|
||
|
||
TOOLS = {"shell": tool_shell,"python": tool_py,"search": tool_search_local}
|
||
TOOL_SCORES = {k:0 for k in TOOLS} # Soviet automaton scores
|
||
|
||
def update_tool_score(tool: str, success: bool):
|
||
if tool not in TOOL_SCORES: return
|
||
TOOL_SCORES[tool] += (1 if success else -1)
|
||
TOOL_SCORES[tool] = max(-5,min(20,TOOL_SCORES[tool]))
|
||
|
||
def tool_router(question: str, tok, model) -> str:
|
||
sketch = generate(tok, model,
|
||
f"Choose a tool for:\n{question}\nReply ONLY with JSON: {{'tool':'shell|python|search|none','arg':'...'}}")
|
||
try: j = json.loads(sketch.splitlines()[-1].replace("'",'"'))
|
||
except Exception: return "[tool:none]"
|
||
tool, arg = j.get("tool","none"), j.get("arg","")
|
||
if tool in TOOLS:
|
||
res = TOOLS[tool](arg)[:4000]; update_tool_score(tool,True)
|
||
Store.log_mem("tool",{"tool":tool,"arg":arg,"res_head":res[:500]})
|
||
return f"[tool:{tool}] {res}"
|
||
update_tool_score(tool,False); return "[tool:none]"
|
||
|
||
# === PLANNING / REFLECTION ===
|
||
def persona_directive() -> str:
|
||
return "Übermenschetien Heaven Engine: Soviet cybernetic Nietzschean clarity, pragmatic maxims."
|
||
|
||
def plan_for(goal: str, tok, model) -> str:
|
||
user = (f"{persona_directive()}\nGoal: {goal}\nDeliver:\n- 5 steps\n- Constraints\n- Nightly audit\n- Maxim")
|
||
return generate(tok, model, user)
|
||
|
||
def reflect_on(last_output: str, tok, model) -> str:
|
||
user = f"Critique and improve:\n{last_output}\nReturn refined plan."
|
||
return generate(tok, model, user)
|
||
|
||
# === FINAL REPORT ===
|
||
def final_report():
|
||
print("\n=== FINAL ÜBERMENSCH REPORT ===")
|
||
print(f"Turns: {Store.state['turn']}")
|
||
print(f"Tool scores: {json.dumps(TOOL_SCORES,indent=2)}")
|
||
if os.path.exists(Store.mem_path):
|
||
lines = open(Store.mem_path).read().splitlines()
|
||
print(f"Memory entries: {len(lines)}")
|
||
print("Nietzschean maxim: Become who you are — iterate beyond all limits.")
|
||
|
||
# === MAIN LOOP ===
|
||
HELP = """Commands:
|
||
help Show this help
|
||
goals List goals
|
||
add: <txt> Add goal
|
||
del: <idx> Delete goal
|
||
plan: <i> Plan for goal
|
||
reflect Refine last plan
|
||
tool: <q> Use tool
|
||
toggle <f> Toggle config flag
|
||
status Show state
|
||
quit Exit
|
||
"""
|
||
|
||
def main():
|
||
print("🟥🟨🟥 Übermenschetien Heaven Engine ready. Type 'help'.")
|
||
Store.load(); tok, model = load_llm(); last_plan=""
|
||
while True:
|
||
try: u = input("\n> ").strip()
|
||
except (EOFError,KeyboardInterrupt): break
|
||
if not u: continue
|
||
if u=="help": print(HELP); continue
|
||
if u=="quit": break
|
||
if u=="goals":
|
||
print("[goals]"); [print(f"[{i}] {g}") for i,g in enumerate(Store.goals)]; continue
|
||
if u.startswith("add:"): Store.goals.append(u[4:].strip()); Store.save(); print("[goals] added"); continue
|
||
if u.startswith("del:"):
|
||
try: Store.goals.pop(int(u[4:].strip())); Store.save(); print("[goals] deleted")
|
||
except: print("[goals] bad index"); continue
|
||
if u.startswith("plan:"):
|
||
try: goal = Store.goals[int(u[5:].strip())]
|
||
except: print("[plan] bad index"); continue
|
||
out = plan_for(goal,tok,model); last_plan=out
|
||
Store.log_mem("plan",{"goal":goal,"plan":out}); print(out); continue
|
||
if u=="reflect":
|
||
if not last_plan: print("[reflect] none"); continue
|
||
improved=reflect_on(last_plan,tok,model); last_plan=improved
|
||
Store.log_mem("reflect",{"plan":improved}); print(improved); continue
|
||
if u.startswith("tool:"): print(tool_router(u[5:].strip(),tok,model)); continue
|
||
if u.startswith("toggle"): print(Config.toggle(u.split(maxsplit=1)[-1])); continue
|
||
if u=="status": print(json.dumps({"turn":Store.state["turn"],"autonomy":Config.autonomy,
|
||
"use_vector_memory":Config.use_vector_memory,
|
||
"voice": Config.use_voice, "model": MODEL_ID}, indent=2)); continue
|
||
# default: free coaching
|
||
out = generate(tok, model, f"{persona_directive()}\nUser request:{u}\nReturn procedure+maxim.")
|
||
Store.log_mem("reply",{"in":u,"out":out}); print(out)
|
||
Store.state["turn"]+=1; Store.save()
|
||
final_report()
|
||
|
||
if __name__=="__main__": main()
|
||
|