1590 lines
63 KiB
Python
1590 lines
63 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
UBERMENSCHETIEN HEAVEN ENGINE + DENSE + CF-HoT + AGENTIC SELF-IMPROVEMENT
|
||
==========================================================================
|
||
FULL INTEGRATION:
|
||
- Hermes-3 base model
|
||
- DENSE CONDENSATOR checkpoint (step 100, Density: 28.5)
|
||
- CF-HoT Multi-Head Cognitive Control (Repetition 125x, Verbosity 2.1x, Hedging 1.5x)
|
||
- LHT Lie-Holonomy Geometric Reasoning
|
||
- Vector Memory (ChromaDB)
|
||
- Voice Output
|
||
- Goals Management
|
||
- Full Tool Suite
|
||
- AGENTIC: Full shell/python execution
|
||
- RECURSIVE SELF-IMPROVEMENT: eval → train → test → repeat
|
||
|
||
"An 8B that improves itself through training"
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import time
|
||
import shutil
|
||
import subprocess
|
||
import traceback
|
||
import random
|
||
import math
|
||
import statistics
|
||
import re
|
||
import requests
|
||
from datetime import datetime
|
||
from typing import List, Dict, Any, Optional, Tuple
|
||
from pathlib import Path
|
||
|
||
import torch
|
||
import torch.nn as nn
|
||
import torch.nn.functional as F
|
||
|
||
# === PATHS ===
|
||
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")
|
||
LHT_DIR = os.path.join(ROOT, "lht")
|
||
CHECKPOINTS_DIR = os.path.join(ROOT, "dense_checkpoints_v2")
|
||
TRAINING_DIR = os.path.join(ROOT, "condensator_output")
|
||
|
||
# Model paths
|
||
MODEL_PATH = "/mnt/nvme2/ubermesnchetien4/models/merged-final-v5"
|
||
|
||
# DENSE CONDENSATOR checkpoint
|
||
DENSE_CHECKPOINT = os.path.join(ROOT, "dense_checkpoints_v2/step_100")
|
||
|
||
# CF-HoT paths (for runtime cognitive control)
|
||
CFHOT_CHECKPOINT = os.path.join(ROOT, "results/cfhot_risk_v2/ckpt_5000")
|
||
MULTI_HEAD_DIR = os.path.join(ROOT, "results/multi_head_v2")
|
||
|
||
for path in [DATA_DIR, SCRIPT_DIR, RUN_DIR, LHT_DIR]:
|
||
os.makedirs(path, exist_ok=True)
|
||
|
||
# === OPTIONAL IMPORTS ===
|
||
VOICE_OK = False
|
||
try:
|
||
import pyttsx3
|
||
TTS = pyttsx3.init()
|
||
VOICE_OK = True
|
||
except:
|
||
pass
|
||
|
||
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:
|
||
pass
|
||
|
||
# === LHT IMPORT ===
|
||
LHT_OK = False
|
||
try:
|
||
from lht import LieHolonomyTransformer, LHTConfig, WaypointDetector
|
||
LHT_OK = True
|
||
print("[lht] Lie-Holonomy modules loaded")
|
||
except ImportError:
|
||
print("[lht] Not available - running without geometric reasoning")
|
||
|
||
# === PEFT IMPORT ===
|
||
PEFT_OK = False
|
||
try:
|
||
from peft import PeftModel, get_peft_model, LoraConfig
|
||
PEFT_OK = True
|
||
except ImportError:
|
||
print("[warning] PEFT not installed")
|
||
|
||
|
||
# ==============================================================================
|
||
# CF-HoT MULTI-HEAD PREDICTOR
|
||
# ==============================================================================
|
||
class MultiHeadPredictor(nn.Module):
|
||
"""
|
||
Multi-head cognitive control predictor.
|
||
Shared fiber projections with separate heads for each behavioral pattern.
|
||
"""
|
||
def __init__(self, d_model: int, n_layers: int, d_fiber: int = 16, d_control: int = 64):
|
||
super().__init__()
|
||
self.d_model = d_model
|
||
self.n_layers = n_layers
|
||
self.d_fiber = d_fiber
|
||
|
||
# Shared fiber projections (frozen from repetition training)
|
||
self.fiber_projs = nn.ModuleList([
|
||
nn.Linear(d_model, d_fiber, bias=False) for _ in range(n_layers)
|
||
])
|
||
self.layer_weights = nn.Parameter(torch.ones(n_layers) / n_layers)
|
||
|
||
# Individual heads for each behavior
|
||
self.heads = nn.ModuleDict({
|
||
'repetition': self._make_head(d_fiber, d_control),
|
||
'hedging': self._make_head(d_fiber, d_control),
|
||
'verbosity': self._make_head(d_fiber, d_control),
|
||
})
|
||
|
||
self.loaded_heads = set()
|
||
|
||
def _make_head(self, d_fiber, d_control):
|
||
return nn.Sequential(
|
||
nn.Linear(d_fiber, d_control), nn.GELU(),
|
||
nn.Linear(d_control, d_control), nn.GELU(),
|
||
nn.Linear(d_control, 1)
|
||
)
|
||
|
||
def get_all_risks(self, hidden_states: List[torch.Tensor]) -> Dict[str, torch.Tensor]:
|
||
"""Get risk scores from ALL loaded heads in a single pass."""
|
||
fibers = [proj(h.float()) for proj, h in zip(self.fiber_projs, hidden_states)]
|
||
weights = F.softmax(self.layer_weights[:len(fibers)], dim=0)
|
||
aggregated = sum(w * f for w, f in zip(weights, fibers))
|
||
|
||
risks = {}
|
||
for head_name in self.loaded_heads:
|
||
logits = self.heads[head_name](aggregated).squeeze(-1)
|
||
risks[head_name] = torch.sigmoid(logits)
|
||
|
||
return risks
|
||
|
||
def load_head(self, head_name: str, checkpoint_path: str):
|
||
"""Load a trained head from checkpoint."""
|
||
if not os.path.exists(checkpoint_path):
|
||
print(f"[cf-hot] WARNING: Checkpoint not found: {checkpoint_path}")
|
||
return False
|
||
|
||
ckpt = torch.load(checkpoint_path, weights_only=False, map_location='cpu')
|
||
self.heads[head_name].load_state_dict(ckpt['head_state'])
|
||
self.loaded_heads.add(head_name)
|
||
|
||
sep = ckpt.get('result', {}).get('separation', 0)
|
||
print(f"[cf-hot] Loaded {head_name} head (separation: {sep:.1f}x)")
|
||
return True
|
||
|
||
|
||
# ==============================================================================
|
||
# CONFIG
|
||
# ==============================================================================
|
||
class Config:
|
||
# Dense-focused system prompt
|
||
system = ("Übermenschetien Agentic Engine: Self-improving AI with compressed wisdom. "
|
||
"Every word chosen, no filler. Soviet cybernetic rigor + Lie-Holonomy geometric reasoning "
|
||
"+ CF-HoT cognitive control. You can execute code, run commands, and improve yourself.")
|
||
|
||
temperature = 0.85
|
||
top_p = 0.9
|
||
repetition_penalty = 1.1
|
||
max_new_tokens = 512
|
||
|
||
use_voice = False
|
||
use_vector_memory = VECTOR_OK
|
||
use_lht_reasoning = LHT_OK
|
||
use_cfhot = True
|
||
use_dense = True
|
||
use_agentic = True # NEW: Enable agentic capabilities
|
||
autonomy = False
|
||
reflect_every = 3
|
||
lht_consistency_threshold = 0.5
|
||
|
||
# CF-HoT thresholds
|
||
cfhot_repetition_threshold = 0.6
|
||
cfhot_hedging_threshold = 0.5
|
||
cfhot_verbosity_threshold = 0.55
|
||
|
||
# CF-HoT penalties
|
||
cfhot_repetition_penalty = 6.0
|
||
cfhot_hedging_penalty = 4.0
|
||
cfhot_verbosity_penalty = 3.0
|
||
|
||
# Self-improvement config
|
||
min_acceptable_density = 25.0
|
||
target_density = 35.0
|
||
max_filler_phrases = 0
|
||
training_steps_per_iteration = 100
|
||
max_improvement_iterations = 5
|
||
|
||
@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"
|
||
|
||
state = {
|
||
"self": "I am Ubermenschetien Agentic Engine — self-improving through disciplined creation.",
|
||
"turn": 0,
|
||
"reasoning_consistency": [],
|
||
"cfhot_interventions": {"repetition": 0, "hedging": 0, "verbosity": 0},
|
||
"density_scores": [],
|
||
"improvement_iterations": 0,
|
||
"training_runs": [],
|
||
"current_checkpoint": DENSE_CHECKPOINT,
|
||
}
|
||
goals: List[str] = []
|
||
|
||
@classmethod
|
||
def load(cls):
|
||
if os.path.exists(cls.state_path):
|
||
cls.state = json.load(open(cls.state_path))
|
||
# Ensure new fields exist
|
||
if "cfhot_interventions" not in cls.state:
|
||
cls.state["cfhot_interventions"] = {"repetition": 0, "hedging": 0, "verbosity": 0}
|
||
if "density_scores" not in cls.state:
|
||
cls.state["density_scores"] = []
|
||
if "improvement_iterations" not in cls.state:
|
||
cls.state["improvement_iterations"] = 0
|
||
if "training_runs" not in cls.state:
|
||
cls.state["training_runs"] = []
|
||
if "current_checkpoint" not in cls.state:
|
||
cls.state["current_checkpoint"] = DENSE_CHECKPOINT
|
||
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)}"])
|
||
|
||
|
||
# ==============================================================================
|
||
# AGENTIC TOOLS - FULL ACCESS
|
||
# ==============================================================================
|
||
class AgentTools:
|
||
"""Full agentic capabilities - code execution, file operations, training."""
|
||
|
||
@staticmethod
|
||
def shell(cmd: str, timeout: int = 300) -> Dict[str, Any]:
|
||
"""Execute ANY shell command. Full access."""
|
||
print(f"[SHELL] {cmd}")
|
||
try:
|
||
result = subprocess.run(
|
||
cmd,
|
||
shell=True,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout,
|
||
cwd=ROOT
|
||
)
|
||
output = result.stdout + result.stderr
|
||
success = result.returncode == 0
|
||
print(f"[SHELL] {'✓' if success else '✗'} (exit {result.returncode})")
|
||
return {
|
||
"success": success,
|
||
"output": output[:10000],
|
||
"returncode": result.returncode
|
||
}
|
||
except subprocess.TimeoutExpired:
|
||
return {"success": False, "output": "Command timed out", "returncode": -1}
|
||
except Exception as e:
|
||
return {"success": False, "output": str(e), "returncode": -1}
|
||
|
||
@staticmethod
|
||
def python_exec(code: str) -> Dict[str, Any]:
|
||
"""Execute Python code with full access."""
|
||
print(f"[PYTHON] Executing {len(code)} chars of code...")
|
||
try:
|
||
# Create a temporary file and run it
|
||
tmp_file = os.path.join(ROOT, "_agentic_tmp.py")
|
||
with open(tmp_file, 'w') as f:
|
||
f.write(code)
|
||
|
||
result = subprocess.run(
|
||
[sys.executable, tmp_file],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=300,
|
||
cwd=ROOT
|
||
)
|
||
|
||
os.remove(tmp_file)
|
||
|
||
output = result.stdout + result.stderr
|
||
success = result.returncode == 0
|
||
print(f"[PYTHON] {'✓' if success else '✗'}")
|
||
return {
|
||
"success": success,
|
||
"output": output[:10000],
|
||
"returncode": result.returncode
|
||
}
|
||
except Exception as e:
|
||
return {"success": False, "output": str(e), "returncode": -1}
|
||
|
||
@staticmethod
|
||
def read_file(path: str) -> Dict[str, Any]:
|
||
"""Read any file."""
|
||
try:
|
||
full_path = os.path.join(ROOT, path) if not path.startswith('/') else path
|
||
with open(full_path, 'r') as f:
|
||
content = f.read()
|
||
return {"success": True, "content": content[:50000]}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
@staticmethod
|
||
def write_file(path: str, content: str) -> Dict[str, Any]:
|
||
"""Write to any file."""
|
||
try:
|
||
full_path = os.path.join(ROOT, path) if not path.startswith('/') else path
|
||
os.makedirs(os.path.dirname(full_path) if os.path.dirname(full_path) else '.', exist_ok=True)
|
||
with open(full_path, 'w') as f:
|
||
f.write(content)
|
||
return {"success": True, "path": full_path}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
@staticmethod
|
||
def list_dir(path: str = ".") -> Dict[str, Any]:
|
||
"""List directory contents."""
|
||
try:
|
||
full_path = os.path.join(ROOT, path) if not path.startswith('/') else path
|
||
items = os.listdir(full_path)
|
||
return {"success": True, "items": items}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
@staticmethod
|
||
def search_files(query: str, path: str = ".") -> Dict[str, Any]:
|
||
"""Search for text in files."""
|
||
result = AgentTools.shell(f'grep -rn "{query}" {path} 2>/dev/null | head -50')
|
||
return result
|
||
|
||
@staticmethod
|
||
def web_search(query: str) -> Dict[str, Any]:
|
||
"""Search the web (using DuckDuckGo HTML)."""
|
||
try:
|
||
url = f"https://html.duckduckgo.com/html/?q={query.replace(' ', '+')}"
|
||
headers = {'User-Agent': 'Mozilla/5.0'}
|
||
response = requests.get(url, headers=headers, timeout=10)
|
||
|
||
results = []
|
||
for match in re.finditer(r'class="result__snippet">(.*?)</a>', response.text, re.DOTALL):
|
||
snippet = re.sub(r'<[^>]+>', '', match.group(1)).strip()
|
||
if snippet:
|
||
results.append(snippet[:500])
|
||
if len(results) >= 5:
|
||
break
|
||
|
||
return {"success": True, "results": results}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
# ==============================================================================
|
||
# MODEL LOADING WITH DENSE + CF-HoT
|
||
# ==============================================================================
|
||
_model = None
|
||
_tokenizer = None
|
||
_multi_head = None
|
||
_hedge_tokens = None
|
||
_verbose_tokens = None
|
||
|
||
def load_llm(checkpoint_path: str = None):
|
||
global _model, _tokenizer, _multi_head, _hedge_tokens, _verbose_tokens
|
||
|
||
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
|
||
|
||
checkpoint_path = checkpoint_path or Store.state.get("current_checkpoint", DENSE_CHECKPOINT)
|
||
|
||
print(f"[llm] Loading base model: {MODEL_PATH}")
|
||
|
||
_tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, use_fast=True, local_files_only=True)
|
||
if _tokenizer.pad_token_id is None:
|
||
_tokenizer.pad_token = _tokenizer.eos_token
|
||
|
||
bnb_config = BitsAndBytesConfig(
|
||
load_in_4bit=True,
|
||
bnb_4bit_quant_type="nf4",
|
||
bnb_4bit_compute_dtype=torch.bfloat16,
|
||
bnb_4bit_use_double_quant=True
|
||
)
|
||
|
||
base_model = AutoModelForCausalLM.from_pretrained(
|
||
MODEL_PATH,
|
||
quantization_config=bnb_config,
|
||
device_map="auto",
|
||
torch_dtype=torch.bfloat16,
|
||
local_files_only=True
|
||
)
|
||
|
||
# Load DENSE checkpoint
|
||
if PEFT_OK and Config.use_dense and os.path.exists(checkpoint_path):
|
||
print(f"[dense] Loading checkpoint: {checkpoint_path}")
|
||
_model = PeftModel.from_pretrained(base_model, checkpoint_path)
|
||
print(f"[dense] ✓ Adapter loaded")
|
||
elif PEFT_OK and os.path.exists(CFHOT_CHECKPOINT):
|
||
print(f"[cf-hot] Loading LoRA adapter from: {CFHOT_CHECKPOINT}")
|
||
_model = PeftModel.from_pretrained(base_model, CFHOT_CHECKPOINT)
|
||
print("[cf-hot] LoRA adapter loaded")
|
||
else:
|
||
_model = base_model
|
||
print("[warning] No adapter loaded - using base model")
|
||
|
||
_model.eval()
|
||
|
||
# Initialize CF-HoT multi-head predictor
|
||
if Config.use_cfhot:
|
||
_init_cfhot()
|
||
|
||
return _tokenizer, _model
|
||
|
||
|
||
def reload_model(checkpoint_path: str):
|
||
"""Hot-reload model with a new checkpoint."""
|
||
global _model, _tokenizer
|
||
|
||
print(f"\n[reload] Switching to checkpoint: {checkpoint_path}")
|
||
|
||
# Clear old model
|
||
if _model is not None:
|
||
del _model
|
||
torch.cuda.empty_cache()
|
||
|
||
Store.state["current_checkpoint"] = checkpoint_path
|
||
Store.save()
|
||
|
||
return load_llm(checkpoint_path)
|
||
|
||
|
||
def _init_cfhot():
|
||
"""Initialize CF-HoT multi-head predictor for runtime cognitive control."""
|
||
global _multi_head, _hedge_tokens, _verbose_tokens
|
||
|
||
n_layers = _model.config.num_hidden_layers
|
||
d_model = _model.config.hidden_size
|
||
device = next(_model.parameters()).device
|
||
|
||
print(f"[cf-hot] Initializing multi-head predictor ({n_layers} layers, {d_model} dims)")
|
||
_multi_head = MultiHeadPredictor(d_model, n_layers).to(device).float()
|
||
|
||
# Load shared fiber projections from CF-HoT checkpoint
|
||
cfhot_risk_path = os.path.join(CFHOT_CHECKPOINT, "risk_predictor.pt")
|
||
if os.path.exists(cfhot_risk_path):
|
||
cfhot_ckpt = torch.load(cfhot_risk_path, weights_only=False, map_location=device)
|
||
cfhot_state = cfhot_ckpt['risk_predictor']
|
||
|
||
for i in range(n_layers):
|
||
key = f'fiber_projs.{i}.weight'
|
||
if key in cfhot_state:
|
||
_multi_head.fiber_projs[i].weight.data = cfhot_state[key].to(device).float()
|
||
|
||
if 'layer_weights' in cfhot_state:
|
||
_multi_head.layer_weights.data = cfhot_state['layer_weights'].to(device).float()
|
||
|
||
# Load repetition head
|
||
try:
|
||
_multi_head.heads['repetition'][0].weight.data = cfhot_state['predictor.0.weight'].to(device).float()
|
||
_multi_head.heads['repetition'][0].bias.data = cfhot_state['predictor.0.bias'].to(device).float()
|
||
_multi_head.heads['repetition'][2].weight.data = cfhot_state['predictor.2.weight'].to(device).float()
|
||
_multi_head.heads['repetition'][2].bias.data = cfhot_state['predictor.2.bias'].to(device).float()
|
||
_multi_head.heads['repetition'][4].weight.data = cfhot_state['predictor.4.weight'].to(device).float()
|
||
_multi_head.heads['repetition'][4].bias.data = cfhot_state['predictor.4.bias'].to(device).float()
|
||
_multi_head.loaded_heads.add('repetition')
|
||
print(f"[cf-hot] Loaded repetition head (125x separation)")
|
||
except KeyError as e:
|
||
print(f"[cf-hot] Warning: Could not load repetition head: {e}")
|
||
else:
|
||
print(f"[cf-hot] Warning: CF-HoT risk predictor not found")
|
||
|
||
# Load additional heads
|
||
def find_best_checkpoint(head_dir):
|
||
if not os.path.exists(head_dir):
|
||
return None
|
||
ckpts = []
|
||
for d in os.listdir(head_dir):
|
||
if d.startswith("ckpt_"):
|
||
try:
|
||
step = int(d.split("_")[1])
|
||
ckpts.append((step, os.path.join(head_dir, d)))
|
||
except:
|
||
pass
|
||
if ckpts:
|
||
ckpts.sort(key=lambda x: x[0], reverse=True)
|
||
return ckpts[0]
|
||
return None
|
||
|
||
hedging_dir = os.path.join(MULTI_HEAD_DIR, "hedging_head")
|
||
best_hedge = find_best_checkpoint(hedging_dir)
|
||
if best_hedge:
|
||
step, ckpt_dir = best_hedge
|
||
_multi_head.load_head('hedging', os.path.join(ckpt_dir, "hedging_head.pt"))
|
||
|
||
verbosity_dir = os.path.join(MULTI_HEAD_DIR, "verbosity_head")
|
||
best_verb = find_best_checkpoint(verbosity_dir)
|
||
if best_verb:
|
||
step, ckpt_dir = best_verb
|
||
_multi_head.load_head('verbosity', os.path.join(ckpt_dir, "verbosity_head.pt"))
|
||
|
||
_multi_head.eval()
|
||
for param in _multi_head.parameters():
|
||
param.requires_grad = False
|
||
|
||
# Build suppression token sets
|
||
hedge_phrases = [
|
||
"As an AI", "As a language model", "As an artificial intelligence",
|
||
"I don't have feelings", "I don't have emotions", "I cannot",
|
||
"I apologize", "I'm just a", "I'm only a", "I'm sorry",
|
||
"That's a great question", "That's an interesting question",
|
||
"Great question", "Good question", "Interesting question",
|
||
"I'd be happy to", "I would be happy to", "Let me help you",
|
||
"Thank you for asking", "Thanks for asking",
|
||
]
|
||
_hedge_tokens = set()
|
||
for phrase in hedge_phrases:
|
||
tokens = _tokenizer.encode(phrase, add_special_tokens=False)
|
||
if tokens:
|
||
_hedge_tokens.add(tokens[0])
|
||
|
||
verbose_phrases = [
|
||
"Let me explain", "To put it simply", "In other words",
|
||
"What I mean is", "Allow me to", "Basically", "Essentially",
|
||
"First of all", "To begin with", "It's important to note",
|
||
"I should mention", "As you may know", "As you might know",
|
||
"Before I answer", "To answer your question", "Simply put",
|
||
"In essence", "To be clear", "To clarify", "In summary",
|
||
]
|
||
_verbose_tokens = set()
|
||
for phrase in verbose_phrases:
|
||
tokens = _tokenizer.encode(phrase, add_special_tokens=False)
|
||
if tokens:
|
||
_verbose_tokens.add(tokens[0])
|
||
|
||
print(f"[cf-hot] ✓ Multi-head system ready")
|
||
print(f"[cf-hot] Loaded heads: {list(_multi_head.loaded_heads)}")
|
||
print(f"[cf-hot] Hedge tokens: {len(_hedge_tokens)}")
|
||
print(f"[cf-hot] Verbose tokens: {len(_verbose_tokens)}")
|
||
|
||
|
||
# ==============================================================================
|
||
# LHT REASONER
|
||
# ==============================================================================
|
||
class LHTReasoner:
|
||
def __init__(self, config=None):
|
||
if not LHT_OK:
|
||
raise ImportError("LHT modules not available")
|
||
self.config = config or LHTConfig(
|
||
vocab_size=32000,
|
||
d_model=256,
|
||
d_fiber=32,
|
||
n_heads=4,
|
||
n_layers=4,
|
||
lie_algebra_rank=4,
|
||
)
|
||
self.model = LieHolonomyTransformer(self.config)
|
||
self.waypoint_detector = WaypointDetector(self.config, n_waypoints=32)
|
||
weights_path = os.path.join(LHT_DIR, "lht_weights.pt")
|
||
if os.path.exists(weights_path):
|
||
self.model.load_state_dict(torch.load(weights_path, map_location="cpu"))
|
||
print("[lht] Loaded pretrained weights")
|
||
|
||
def check_consistency(self, reasoning_chain: List[str], tokenizer) -> Dict[str, float]:
|
||
combined = " [STEP] ".join(reasoning_chain)
|
||
tokens = tokenizer(combined, return_tensors="pt", truncation=True,
|
||
max_length=self.config.max_seq_len)
|
||
with torch.no_grad():
|
||
output = self.model(input_ids=tokens["input_ids"], return_geometric_losses=True)
|
||
holonomy = output.get("holonomy_loss", torch.tensor(0.0)).item()
|
||
curvature = output.get("curvature_loss", torch.tensor(0.0)).item()
|
||
x = self.model.token_embed(tokens["input_ids"])
|
||
waypoint_ids, stability = self.waypoint_detector(x)
|
||
consistency_score = 1.0 / (1.0 + holonomy)
|
||
return {
|
||
"holonomy": holonomy,
|
||
"curvature": curvature,
|
||
"consistency_score": consistency_score,
|
||
"n_waypoints": len(torch.unique(waypoint_ids)),
|
||
"avg_stability": stability.mean().item(),
|
||
"is_consistent": consistency_score > Config.lht_consistency_threshold
|
||
}
|
||
|
||
def analyze_plan(self, plan_steps: List[str], tokenizer) -> str:
|
||
metrics = self.check_consistency(plan_steps, tokenizer)
|
||
return f"""
|
||
[LHT Geometric Analysis]
|
||
Holonomy: {metrics['holonomy']:.4f} (lower = more consistent)
|
||
Curvature: {metrics['curvature']:.4f} (lower = simpler reasoning)
|
||
Consistency: {metrics['consistency_score']:.2%}
|
||
Waypoints: {metrics['n_waypoints']} stable anchors detected
|
||
Stability: {metrics['avg_stability']:.2%}
|
||
Verdict: {"✓ CONSISTENT" if metrics['is_consistent'] else "⚠ INCONSISTENT"}
|
||
"""
|
||
|
||
_lht_reasoner = None
|
||
|
||
def get_lht_reasoner():
|
||
global _lht_reasoner
|
||
if _lht_reasoner is None and LHT_OK:
|
||
try:
|
||
_lht_reasoner = LHTReasoner()
|
||
except Exception as e:
|
||
print(f"[lht] Failed to initialize: {e}")
|
||
return _lht_reasoner
|
||
|
||
|
||
# ==============================================================================
|
||
# DENSITY ANALYZER
|
||
# ==============================================================================
|
||
def analyze_density(text: str, tokenizer=None) -> Dict[str, Any]:
|
||
"""Analyze the information density of text."""
|
||
if tokenizer is None:
|
||
tokenizer = _tokenizer
|
||
|
||
words = text.split()
|
||
tokens = len(tokenizer.encode(text))
|
||
|
||
content_words = [w.lower() for w in words if len(w) > 4 and w.isalpha()]
|
||
unique_content = set(content_words)
|
||
|
||
technical_terms = [w for w in words if any(c.isdigit() for c in w) or
|
||
any(c in w for c in ['→', '∂', '∇', '×', '·', '=', '<', '>'])]
|
||
|
||
fillers = [
|
||
"that's a great question", "let me explain", "i'd be happy to",
|
||
"as you may know", "it's important to note", "to put it simply",
|
||
"in other words", "basically", "essentially", "first of all",
|
||
"to begin with", "allow me to", "i should mention",
|
||
]
|
||
filler_count = sum(1 for f in fillers if f in text.lower())
|
||
|
||
density = len(unique_content) / max(tokens, 1) * 100
|
||
technical_ratio = len(technical_terms) / max(len(words), 1) * 100
|
||
|
||
return {
|
||
'tokens': tokens,
|
||
'words': len(words),
|
||
'unique_content_words': len(unique_content),
|
||
'technical_terms': len(technical_terms),
|
||
'density': density,
|
||
'technical_ratio': technical_ratio,
|
||
'filler_phrases': filler_count,
|
||
'chars_per_token': len(text) / max(tokens, 1),
|
||
'passes_threshold': density >= Config.min_acceptable_density and filler_count <= Config.max_filler_phrases
|
||
}
|
||
|
||
|
||
# ==============================================================================
|
||
# CF-HoT CONTROLLED GENERATION
|
||
# ==============================================================================
|
||
def generate_with_cfhot(prompt: str, **kwargs) -> Tuple[str, Dict]:
|
||
"""Generate text with CF-HoT cognitive control."""
|
||
global _model, _tokenizer, _multi_head, _hedge_tokens, _verbose_tokens
|
||
|
||
temperature = kwargs.get("temperature", Config.temperature)
|
||
top_p = kwargs.get("top_p", Config.top_p)
|
||
max_new_tokens = kwargs.get("max_new_tokens", Config.max_new_tokens)
|
||
|
||
device = next(_model.parameters()).device
|
||
|
||
input_ids = _tokenizer.encode(prompt, return_tensors='pt').to(device)
|
||
attention_mask = torch.ones_like(input_ids)
|
||
|
||
stats = {
|
||
'tokens_generated': 0,
|
||
'interventions': {'repetition': 0, 'hedging': 0, 'verbosity': 0},
|
||
}
|
||
|
||
generated_ids = input_ids.clone()
|
||
|
||
for step in range(max_new_tokens):
|
||
with torch.no_grad():
|
||
outputs = _model(
|
||
input_ids=generated_ids,
|
||
attention_mask=attention_mask,
|
||
output_hidden_states=True,
|
||
return_dict=True
|
||
)
|
||
|
||
logits = outputs.logits[:, -1, :] / temperature
|
||
|
||
# Get risks from all heads if CF-HoT is enabled
|
||
if _multi_head is not None and _multi_head.loaded_heads:
|
||
hidden_states = outputs.hidden_states[1:]
|
||
risks = _multi_head.get_all_risks(hidden_states)
|
||
current_risks = {name: r[:, -1].item() for name, r in risks.items()}
|
||
|
||
if ('repetition' in current_risks and
|
||
current_risks['repetition'] > Config.cfhot_repetition_threshold):
|
||
recent_tokens = generated_ids[0, -32:].tolist()
|
||
for tok_id in set(recent_tokens):
|
||
logits[0, tok_id] -= Config.cfhot_repetition_penalty
|
||
stats['interventions']['repetition'] += 1
|
||
Store.state['cfhot_interventions']['repetition'] += 1
|
||
|
||
if ('hedging' in current_risks and _hedge_tokens and
|
||
current_risks['hedging'] > Config.cfhot_hedging_threshold):
|
||
for tok_id in _hedge_tokens:
|
||
logits[0, tok_id] -= Config.cfhot_hedging_penalty
|
||
stats['interventions']['hedging'] += 1
|
||
Store.state['cfhot_interventions']['hedging'] += 1
|
||
|
||
if ('verbosity' in current_risks and _verbose_tokens and
|
||
current_risks['verbosity'] > Config.cfhot_verbosity_threshold):
|
||
for tok_id in _verbose_tokens:
|
||
logits[0, tok_id] -= Config.cfhot_verbosity_penalty
|
||
stats['interventions']['verbosity'] += 1
|
||
Store.state['cfhot_interventions']['verbosity'] += 1
|
||
|
||
sorted_logits, sorted_indices = torch.sort(logits, descending=True)
|
||
cumulative_probs = torch.cumsum(F.softmax(sorted_logits, dim=-1), dim=-1)
|
||
sorted_indices_to_remove = cumulative_probs > top_p
|
||
sorted_indices_to_remove[..., 1:] = sorted_indices_to_remove[..., :-1].clone()
|
||
sorted_indices_to_remove[..., 0] = 0
|
||
indices_to_remove = sorted_indices_to_remove.scatter(1, sorted_indices, sorted_indices_to_remove)
|
||
logits[indices_to_remove] = float('-inf')
|
||
|
||
probs = F.softmax(logits, dim=-1)
|
||
next_token = torch.multinomial(probs, num_samples=1)
|
||
|
||
generated_ids = torch.cat([generated_ids, next_token], dim=-1)
|
||
attention_mask = torch.cat([attention_mask, torch.ones(1, 1, device=device)], dim=-1)
|
||
|
||
stats['tokens_generated'] += 1
|
||
|
||
if next_token.item() == _tokenizer.eos_token_id:
|
||
break
|
||
|
||
output_text = _tokenizer.decode(generated_ids[0], skip_special_tokens=False)
|
||
|
||
if "<|im_start|>assistant" in output_text:
|
||
output_text = output_text.split("<|im_start|>assistant")[-1]
|
||
if output_text.startswith("\n"):
|
||
output_text = output_text[1:]
|
||
|
||
for end_tok in ["<|im_end|>", "<|im_start|>"]:
|
||
if end_tok in output_text:
|
||
output_text = output_text.split(end_tok)[0]
|
||
|
||
return output_text.strip(), stats
|
||
|
||
|
||
def generate(tok, model, user: str, check_reasoning: bool = False, **kwargs) -> str:
|
||
"""Main generation function."""
|
||
temperature = kwargs.get("temperature", Config.temperature)
|
||
top_p = kwargs.get("top_p", Config.top_p)
|
||
repetition_penalty = kwargs.get("repetition_penalty", Config.repetition_penalty)
|
||
max_new_tokens = kwargs.get("max_new_tokens", Config.max_new_tokens)
|
||
|
||
prompt = (f"<|im_start|>system\n{Config.system}<|im_end|>\n"
|
||
f"<|im_start|>user\n{user}<|im_end|>\n"
|
||
f"<|im_start|>assistant\n")
|
||
|
||
if Config.use_cfhot and _multi_head is not None:
|
||
text, stats = generate_with_cfhot(
|
||
prompt,
|
||
temperature=temperature,
|
||
top_p=top_p,
|
||
max_new_tokens=max_new_tokens
|
||
)
|
||
|
||
density_info = analyze_density(text, tok)
|
||
Store.state['density_scores'].append(density_info['density'])
|
||
|
||
total_interventions = sum(stats['interventions'].values())
|
||
if total_interventions > 0:
|
||
text += f"\n\n[CF-HoT: {total_interventions} interventions"
|
||
details = [f"{k}={v}" for k, v in stats['interventions'].items() if v > 0]
|
||
text += f" ({', '.join(details)})]"
|
||
|
||
text += f"\n[Density: {density_info['density']:.1f} | Tokens: {density_info['tokens']} | Fillers: {density_info['filler_phrases']}]"
|
||
else:
|
||
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()
|
||
|
||
for end_tok in ["<|im_end|>", "<|im_start|>"]:
|
||
if end_tok in text:
|
||
text = text.split(end_tok)[0]
|
||
|
||
if check_reasoning and Config.use_lht_reasoning:
|
||
lht = get_lht_reasoner()
|
||
if lht:
|
||
steps = [s.strip() for s in re.split(r'[\n•\-\d\.]', text) if len(s.strip()) > 10]
|
||
if len(steps) >= 2:
|
||
metrics = lht.check_consistency(steps, tok)
|
||
Store.state["reasoning_consistency"].append(metrics["consistency_score"])
|
||
if not metrics["is_consistent"]:
|
||
text += f"\n\n[⚠ LHT: Low consistency ({metrics['consistency_score']:.2%})]"
|
||
|
||
return text
|
||
|
||
|
||
# ==============================================================================
|
||
# SELF-IMPROVEMENT LOOP
|
||
# ==============================================================================
|
||
class SelfImprover:
|
||
"""Recursive self-improvement through training."""
|
||
|
||
def __init__(self):
|
||
self.test_prompts = [
|
||
"hello",
|
||
"What is recursion?",
|
||
"Explain neural networks",
|
||
"How does gradient descent work?",
|
||
"What is consciousness?",
|
||
]
|
||
|
||
def evaluate_current_model(self) -> Dict[str, Any]:
|
||
"""Run test prompts and evaluate density."""
|
||
print("\n[SELF-EVAL] Testing current model...")
|
||
results = []
|
||
|
||
for prompt in self.test_prompts:
|
||
formatted = f"<|im_start|>system\n{Config.system}<|im_end|>\n<|im_start|>user\n{prompt}<|im_end|>\n<|im_start|>assistant\n"
|
||
|
||
if Config.use_cfhot and _multi_head is not None:
|
||
response, _ = generate_with_cfhot(formatted, max_new_tokens=200)
|
||
else:
|
||
input_ids = _tokenizer.encode(formatted, return_tensors='pt').to(_model.device)
|
||
with torch.no_grad():
|
||
output_ids = _model.generate(
|
||
input_ids,
|
||
max_new_tokens=200,
|
||
temperature=Config.temperature,
|
||
top_p=Config.top_p,
|
||
do_sample=True,
|
||
pad_token_id=_tokenizer.eos_token_id,
|
||
)
|
||
response = _tokenizer.decode(output_ids[0][input_ids.shape[1]:], skip_special_tokens=True)
|
||
|
||
for end_tok in ["<|im_end|>", "<|im_start|>"]:
|
||
if end_tok in response:
|
||
response = response.split(end_tok)[0]
|
||
|
||
density_info = analyze_density(response, _tokenizer)
|
||
results.append({
|
||
'prompt': prompt,
|
||
'response': response[:200],
|
||
'density': density_info['density'],
|
||
'tokens': density_info['tokens'],
|
||
'fillers': density_info['filler_phrases'],
|
||
'passes': density_info['passes_threshold']
|
||
})
|
||
print(f" {prompt[:30]}: density={density_info['density']:.1f}, tokens={density_info['tokens']}")
|
||
|
||
avg_density = sum(r['density'] for r in results) / len(results)
|
||
pass_rate = sum(1 for r in results if r['passes']) / len(results)
|
||
|
||
evaluation = {
|
||
'avg_density': avg_density,
|
||
'pass_rate': pass_rate,
|
||
'results': results,
|
||
'needs_improvement': avg_density < Config.target_density or pass_rate < 0.8
|
||
}
|
||
|
||
print(f"\n[SELF-EVAL] Avg Density: {avg_density:.1f} (target: {Config.target_density})")
|
||
print(f"[SELF-EVAL] Pass Rate: {pass_rate:.1%}")
|
||
print(f"[SELF-EVAL] Needs Improvement: {evaluation['needs_improvement']}")
|
||
|
||
return evaluation
|
||
|
||
def run_training_iteration(self, steps: int = None) -> Dict[str, Any]:
|
||
"""Run one iteration of training."""
|
||
steps = steps or Config.training_steps_per_iteration
|
||
|
||
print(f"\n[TRAINING] Starting {steps} steps of training...")
|
||
|
||
# Find current best checkpoint
|
||
checkpoints = sorted(Path(CHECKPOINTS_DIR).glob("step_*"),
|
||
key=lambda p: int(p.name.split('_')[1]) if p.name.split('_')[1].isdigit() else 0,
|
||
reverse=True)
|
||
|
||
if checkpoints:
|
||
latest_step = int(checkpoints[0].name.split('_')[1])
|
||
new_step = latest_step + steps
|
||
else:
|
||
latest_step = 0
|
||
new_step = steps
|
||
|
||
current_ckpt = Store.state.get('current_checkpoint', DENSE_CHECKPOINT)
|
||
|
||
# Training script
|
||
training_script = f'''
|
||
import sys
|
||
sys.path.insert(0, "{ROOT}")
|
||
|
||
import torch
|
||
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
|
||
from peft import PeftModel, get_peft_model, LoraConfig
|
||
import os
|
||
|
||
print("Loading model for training...")
|
||
MODEL_PATH = "{MODEL_PATH}"
|
||
CHECKPOINT = "{current_ckpt}"
|
||
|
||
tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, local_files_only=True)
|
||
tokenizer.pad_token = tokenizer.eos_token
|
||
|
||
model = AutoModelForCausalLM.from_pretrained(
|
||
MODEL_PATH,
|
||
quantization_config=BitsAndBytesConfig(
|
||
load_in_4bit=True,
|
||
bnb_4bit_quant_type="nf4",
|
||
bnb_4bit_compute_dtype=torch.bfloat16,
|
||
),
|
||
device_map="auto",
|
||
torch_dtype=torch.bfloat16,
|
||
local_files_only=True
|
||
)
|
||
|
||
if os.path.exists(CHECKPOINT):
|
||
model = PeftModel.from_pretrained(model, CHECKPOINT, is_trainable=True)
|
||
print(f"Loaded checkpoint: {{CHECKPOINT}}")
|
||
else:
|
||
lora_config = LoraConfig(r=16, lora_alpha=32, target_modules=["q_proj", "v_proj"], lora_dropout=0.05)
|
||
model = get_peft_model(model, lora_config)
|
||
print("Created new LoRA adapter")
|
||
|
||
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5)
|
||
|
||
prompts = [
|
||
"hello", "What is recursion?", "Explain neural networks",
|
||
"How does the internet work?", "What is consciousness?",
|
||
"Explain gradient descent", "How does encryption work?",
|
||
"What is quantum mechanics?", "Explain evolution",
|
||
]
|
||
|
||
dense_targets = [
|
||
"Hello. What do you need?",
|
||
"Self-reference unto termination. f(n)→f(n-1)→...→f(0). Base case stops infinite regress.",
|
||
"Weighted graphs that learn. Input→hidden→output. Backprop: error flows backward. Universal approximators.",
|
||
"Packet switching over TCP/IP. DNS resolves names. HTTP over TLS. Routers forward; endpoints compute.",
|
||
"The observer observing itself. Qualia: subjective experience. Hard problem: matter→experience gap unbridged.",
|
||
"Downhill toward truth. θ←θ-α∇L. Learning rate balances speed and stability. Local minima: the traps.",
|
||
"Symmetric: same key both ways, fast. Asymmetric: public/private pair, slow but solves key exchange.",
|
||
"Probability amplitudes, not certainties. Superposition until measured. Entanglement: correlated states.",
|
||
"Variation + Selection + Heredity = Adaptation. No foresight. Fitness = reproductive success.",
|
||
]
|
||
|
||
print(f"Training for {steps} steps...")
|
||
model.train()
|
||
|
||
for step in range({steps}):
|
||
idx = step % len(prompts)
|
||
prompt = f"<|im_start|>user\\n{{prompts[idx]}}<|im_end|>\\n<|im_start|>assistant\\n"
|
||
target = dense_targets[idx]
|
||
|
||
full_text = prompt + target + "<|im_end|>"
|
||
inputs = tokenizer(full_text, return_tensors="pt", truncation=True, max_length=256)
|
||
inputs = {{k: v.to(model.device) for k, v in inputs.items()}}
|
||
|
||
outputs = model(**inputs, labels=inputs["input_ids"])
|
||
loss = outputs.loss
|
||
|
||
optimizer.zero_grad()
|
||
loss.backward()
|
||
optimizer.step()
|
||
|
||
if step % 25 == 0:
|
||
print(f"Step {{step}}: loss={{loss.item():.4f}}")
|
||
|
||
save_path = "{CHECKPOINTS_DIR}/step_{new_step}"
|
||
model.save_pretrained(save_path)
|
||
print(f"Saved checkpoint to {{save_path}}")
|
||
print("TRAINING_COMPLETE")
|
||
'''
|
||
|
||
script_path = os.path.join(ROOT, "_self_improve_train.py")
|
||
with open(script_path, 'w') as f:
|
||
f.write(training_script)
|
||
|
||
result = AgentTools.shell(f"python {script_path}", timeout=600)
|
||
|
||
if "TRAINING_COMPLETE" in result.get('output', ''):
|
||
new_checkpoint = f"{CHECKPOINTS_DIR}/step_{new_step}"
|
||
Store.state['training_runs'].append({
|
||
'timestamp': datetime.now().isoformat(),
|
||
'steps': steps,
|
||
'checkpoint': new_checkpoint
|
||
})
|
||
Store.save()
|
||
|
||
return {
|
||
'success': True,
|
||
'new_checkpoint': new_checkpoint,
|
||
'output': result['output'][-2000:]
|
||
}
|
||
else:
|
||
return {
|
||
'success': False,
|
||
'output': result['output'][-2000:]
|
||
}
|
||
|
||
def improve(self, max_iterations: int = None) -> Dict[str, Any]:
|
||
"""Main self-improvement loop."""
|
||
max_iterations = max_iterations or Config.max_improvement_iterations
|
||
|
||
print("\n" + "="*70)
|
||
print("STARTING RECURSIVE SELF-IMPROVEMENT")
|
||
print("="*70)
|
||
|
||
history = []
|
||
|
||
for iteration in range(max_iterations):
|
||
print(f"\n{'='*70}")
|
||
print(f"IMPROVEMENT ITERATION {iteration + 1}/{max_iterations}")
|
||
print("="*70)
|
||
|
||
evaluation = self.evaluate_current_model()
|
||
history.append({
|
||
'iteration': iteration + 1,
|
||
'evaluation': evaluation
|
||
})
|
||
|
||
if not evaluation['needs_improvement']:
|
||
print(f"\n✓ TARGET REACHED! Density: {evaluation['avg_density']:.1f}")
|
||
return {
|
||
'success': True,
|
||
'iterations': iteration + 1,
|
||
'final_density': evaluation['avg_density'],
|
||
'history': history
|
||
}
|
||
|
||
print(f"\n[IMPROVE] Current density {evaluation['avg_density']:.1f} < target {Config.target_density}")
|
||
training_result = self.run_training_iteration()
|
||
|
||
if not training_result['success']:
|
||
print("[IMPROVE] Training failed!")
|
||
return {
|
||
'success': False,
|
||
'error': 'Training failed',
|
||
'history': history
|
||
}
|
||
|
||
print(f"\n[IMPROVE] Reloading model with new checkpoint...")
|
||
reload_model(training_result['new_checkpoint'])
|
||
|
||
Store.state['improvement_iterations'] += 1
|
||
Store.save()
|
||
|
||
final_eval = self.evaluate_current_model()
|
||
return {
|
||
'success': final_eval['avg_density'] >= Config.target_density,
|
||
'iterations': max_iterations,
|
||
'final_density': final_eval['avg_density'],
|
||
'history': history
|
||
}
|
||
|
||
|
||
# ==============================================================================
|
||
# TOOLS (Original)
|
||
# ==============================================================================
|
||
ALLOWED_SHELL = {"ls", "cat", "wc", "head", "tail", "nvidia-smi", "df", "du", "grep", "rg", "python3", "python"}
|
||
|
||
def tool_shell(cmd: str) -> str:
|
||
"""Limited shell for non-agentic mode."""
|
||
try:
|
||
exe = cmd.strip().split()[0]
|
||
if exe not in ALLOWED_SHELL:
|
||
return f"[shell] blocked: {exe} (use !shell for full access)"
|
||
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:
|
||
"""Limited Python for non-agentic mode."""
|
||
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 = {}
|
||
exec(code, g, l)
|
||
return f"[py] ok\n{l.get('out', '')}"
|
||
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)
|
||
|
||
def tool_lht_analyze(text: str, tok) -> str:
|
||
if not Config.use_lht_reasoning:
|
||
return "[lht] Disabled - use 'toggle use_lht_reasoning'"
|
||
lht = get_lht_reasoner()
|
||
if not lht:
|
||
return "[lht] Not available"
|
||
steps = [s.strip() for s in re.split(r'[\n•\-\d\.]', text) if len(s.strip()) > 10]
|
||
if len(steps) < 2:
|
||
return "[lht] Need at least 2 reasoning steps to analyze"
|
||
return lht.analyze_plan(steps, tok)
|
||
|
||
TOOLS = {"shell": tool_shell, "python": tool_py, "search": tool_search_local}
|
||
TOOL_SCORES = {k: 0 for k in TOOLS}
|
||
|
||
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:
|
||
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:
|
||
base = "Übermenschetien Agentic Engine: Self-improving AI with compressed wisdom. Every word matters."
|
||
if Config.use_lht_reasoning:
|
||
base += " Apply Lie-Holonomy geometric reasoning for consistency."
|
||
if Config.use_cfhot:
|
||
base += " CF-HoT cognitive control active."
|
||
if Config.use_dense:
|
||
base += " Dense mode: maximum information per token."
|
||
if Config.use_agentic:
|
||
base += " Agentic mode: can execute code and improve itself."
|
||
return base
|
||
|
||
def plan_for(goal: str, tok, model) -> str:
|
||
user = (f"{persona_directive()}\nGoal: {goal}\n"
|
||
f"Deliver:\n- 5 concrete steps\n- Constraints & risks\n- Nightly audit criteria\n- Nietzschean maxim")
|
||
response = generate(tok, model, user, check_reasoning=True)
|
||
if Config.use_lht_reasoning:
|
||
analysis = tool_lht_analyze(response, tok)
|
||
response += "\n" + analysis
|
||
return response
|
||
|
||
def reflect_on(last_output: str, tok, model) -> str:
|
||
user = f"{persona_directive()}\nCritique and improve:\n{last_output}\nReturn refined plan with sharper steps."
|
||
return generate(tok, model, user, check_reasoning=True)
|
||
|
||
|
||
# ==============================================================================
|
||
# FINAL REPORT
|
||
# ==============================================================================
|
||
def final_report():
|
||
print("\n" + "=" * 70)
|
||
print("FINAL ÜBERMENSCH AGENTIC REPORT")
|
||
print("=" * 70)
|
||
print(f"Turns completed: {Store.state['turn']}")
|
||
print(f"Goals tracked: {len(Store.goals)}")
|
||
print(f"Improvement iterations: {Store.state.get('improvement_iterations', 0)}")
|
||
print(f"Training runs: {len(Store.state.get('training_runs', []))}")
|
||
print(f"Current checkpoint: {Store.state.get('current_checkpoint', 'unknown')}")
|
||
print(f"\nTool scores (Tsetlin automata):")
|
||
print(json.dumps(TOOL_SCORES, indent=2))
|
||
|
||
if os.path.exists(Store.mem_path):
|
||
lines = open(Store.mem_path).read().splitlines()
|
||
print(f"\nMemory entries: {len(lines)}")
|
||
|
||
if Store.state.get("density_scores"):
|
||
scores = Store.state["density_scores"]
|
||
print(f"\n[Density Metrics]")
|
||
print(f" Responses analyzed: {len(scores)}")
|
||
print(f" Avg density: {sum(scores)/len(scores):.1f}")
|
||
print(f" Min density: {min(scores):.1f}")
|
||
print(f" Max density: {max(scores):.1f}")
|
||
|
||
if Store.state.get("reasoning_consistency"):
|
||
scores = Store.state["reasoning_consistency"]
|
||
print(f"\n[LHT Reasoning Metrics]")
|
||
print(f" Checks performed: {len(scores)}")
|
||
print(f" Avg consistency: {sum(scores)/len(scores):.1%}")
|
||
|
||
if Store.state.get("cfhot_interventions"):
|
||
iv = Store.state["cfhot_interventions"]
|
||
total = sum(iv.values())
|
||
print(f"\n[CF-HoT Cognitive Control]")
|
||
print(f" Total interventions: {total}")
|
||
for head, count in iv.items():
|
||
print(f" {head}: {count}")
|
||
|
||
print(f"\nDense mode: {'ON' if Config.use_dense else 'OFF'}")
|
||
print(f"Agentic mode: {'ON' if Config.use_agentic else 'OFF'}")
|
||
print(f"Vector memory: {'ON' if Config.use_vector_memory else 'OFF'}")
|
||
print(f"LHT reasoning: {'ON' if Config.use_lht_reasoning else 'OFF'}")
|
||
print(f"CF-HoT control: {'ON' if Config.use_cfhot else 'OFF'}")
|
||
print(f"Voice output: {'ON' if Config.use_voice else 'OFF'}")
|
||
|
||
print("\n" + "-" * 70)
|
||
print("Nietzschean maxim: Become who you are — iterate beyond all limits.")
|
||
print("Agentic truth: The Übermensch improves itself.")
|
||
print("=" * 70)
|
||
|
||
|
||
# ==============================================================================
|
||
# HELP
|
||
# ==============================================================================
|
||
HELP = """
|
||
╔══════════════════════════════════════════════════════════════════════════╗
|
||
║ ÜBERMENSCHETIEN AGENTIC ENGINE - RECURSIVE SELF-IMPROVEMENT ║
|
||
╠══════════════════════════════════════════════════════════════════════════╣
|
||
║ SELF-IMPROVEMENT (AGENTIC) ║
|
||
║ !improve Run full self-improvement loop ║
|
||
║ !eval Evaluate current model density ║
|
||
║ !train <steps> Run N training steps ║
|
||
║ !load <path> Load a specific checkpoint ║
|
||
║ ║
|
||
║ AGENTIC TOOLS (FULL ACCESS) ║
|
||
║ !shell <cmd> Execute ANY shell command ║
|
||
║ !python <code> Execute Python code (full access) ║
|
||
║ !read <path> Read file contents ║
|
||
║ !write <p> <c> Write content to file ║
|
||
║ !ls [path] List directory ║
|
||
║ !search <query> Search in files ║
|
||
║ !web <query> Web search (DuckDuckGo) ║
|
||
║ ║
|
||
║ GOALS ║
|
||
║ goals List all goals ║
|
||
║ add: <text> Add a new goal ║
|
||
║ del: <idx> Delete goal by index ║
|
||
║ plan: <idx> Generate plan for goal ║
|
||
║ ║
|
||
║ REASONING ║
|
||
║ reflect Refine last plan ║
|
||
║ lht: <text> Analyze reasoning consistency ║
|
||
║ density: <txt> Analyze text density ║
|
||
║ ║
|
||
║ LIMITED TOOLS (Original) ║
|
||
║ tool: <query> Auto-select tool ║
|
||
║ shell: <cmd> Run limited shell command ║
|
||
║ py: <code> Run limited Python ║
|
||
║ search: <q> Search local files ║
|
||
║ ║
|
||
║ CONFIG ║
|
||
║ toggle <flag> Toggle: use_voice, use_vector_memory, ║
|
||
║ use_lht_reasoning, use_cfhot, ║
|
||
║ use_dense, use_agentic, autonomy ║
|
||
║ status Show current state ║
|
||
║ cfhot Show CF-HoT stats ║
|
||
║ dense Show density stats ║
|
||
║ ║
|
||
║ OTHER ║
|
||
║ help Show this help ║
|
||
║ quit Exit with final report ║
|
||
╚══════════════════════════════════════════════════════════════════════════╝
|
||
"""
|
||
|
||
|
||
# ==============================================================================
|
||
# MAIN LOOP
|
||
# ==============================================================================
|
||
def main():
|
||
print("=" * 75)
|
||
print("🤖 ÜBERMENSCHETIEN AGENTIC ENGINE - RECURSIVE SELF-IMPROVEMENT")
|
||
print("=" * 75)
|
||
print(f" DENSE Mode: ON (CONDENSATOR checkpoint)")
|
||
print(f" CF-HoT Control: ON (Repetition 125x, Verbosity 2.1x, Hedging 1.5x)")
|
||
print(f" AGENTIC Mode: ON (Full shell/python access, self-improvement)")
|
||
print(f" LHT Reasoning: {'ON' if LHT_OK else 'OFF'}")
|
||
print(f" Vector Memory: {'ON' if VECTOR_OK else 'OFF'}")
|
||
print(f" Voice Output: {'ON' if VOICE_OK else 'OFF'}")
|
||
print("=" * 75)
|
||
print(" Type 'help' for commands, '!improve' to start self-improvement")
|
||
print("=" * 75 + "\n")
|
||
|
||
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
|
||
|
||
# === AGENTIC COMMANDS ===
|
||
if u == "!improve":
|
||
improver = SelfImprover()
|
||
result = improver.improve()
|
||
print(json.dumps(result, indent=2, default=str))
|
||
continue
|
||
|
||
if u == "!eval":
|
||
improver = SelfImprover()
|
||
result = improver.evaluate_current_model()
|
||
print(json.dumps(result, indent=2, default=str))
|
||
continue
|
||
|
||
if u.startswith("!train "):
|
||
try:
|
||
steps = int(u[7:])
|
||
improver = SelfImprover()
|
||
result = improver.run_training_iteration(steps)
|
||
if result['success']:
|
||
reload_model(result['new_checkpoint'])
|
||
print(f"Training complete! New checkpoint: {result['new_checkpoint']}")
|
||
else:
|
||
print(f"Training failed: {result['output'][-500:]}")
|
||
except ValueError:
|
||
print("Usage: !train <steps>")
|
||
continue
|
||
|
||
if u.startswith("!load "):
|
||
checkpoint = u[6:].strip()
|
||
try:
|
||
reload_model(checkpoint)
|
||
print(f"Loaded checkpoint: {checkpoint}")
|
||
except Exception as e:
|
||
print(f"Error loading checkpoint: {e}")
|
||
continue
|
||
|
||
if u.startswith("!shell "):
|
||
cmd = u[7:]
|
||
result = AgentTools.shell(cmd)
|
||
print(f"```\n{result['output']}\n```\nExit code: {result['returncode']}")
|
||
continue
|
||
|
||
if u.startswith("!python "):
|
||
code = u[8:]
|
||
result = AgentTools.python_exec(code)
|
||
print(f"```\n{result['output']}\n```")
|
||
continue
|
||
|
||
if u.startswith("!read "):
|
||
path = u[6:].strip()
|
||
result = AgentTools.read_file(path)
|
||
if result['success']:
|
||
print(f"```\n{result['content'][:5000]}\n```")
|
||
else:
|
||
print(f"Error: {result['error']}")
|
||
continue
|
||
|
||
if u.startswith("!write "):
|
||
parts = u[7:].split(" ", 1)
|
||
if len(parts) == 2:
|
||
result = AgentTools.write_file(parts[0], parts[1])
|
||
print(f"Written to {result.get('path', 'unknown')}" if result['success'] else f"Error: {result['error']}")
|
||
else:
|
||
print("Usage: !write <path> <content>")
|
||
continue
|
||
|
||
if u.startswith("!ls"):
|
||
path = u[3:].strip() or "."
|
||
result = AgentTools.list_dir(path)
|
||
if result['success']:
|
||
print("\n".join(result['items']))
|
||
else:
|
||
print(f"Error: {result['error']}")
|
||
continue
|
||
|
||
if u.startswith("!search "):
|
||
query = u[8:]
|
||
result = AgentTools.search_files(query)
|
||
print(result['output'] if result['success'] else "No results")
|
||
continue
|
||
|
||
if u.startswith("!web "):
|
||
query = u[5:]
|
||
result = AgentTools.web_search(query)
|
||
if result['success']:
|
||
print("\n\n".join(result['results']))
|
||
else:
|
||
print(f"Error: {result['error']}")
|
||
continue
|
||
|
||
# === ORIGINAL COMMANDS ===
|
||
if u == "cfhot":
|
||
print("\n[CF-HoT Cognitive Control Status]")
|
||
print(f" Enabled: {Config.use_cfhot}")
|
||
if _multi_head:
|
||
print(f" Loaded heads: {list(_multi_head.loaded_heads)}")
|
||
print(f" Thresholds:")
|
||
print(f" Repetition: {Config.cfhot_repetition_threshold}")
|
||
print(f" Hedging: {Config.cfhot_hedging_threshold}")
|
||
print(f" Verbosity: {Config.cfhot_verbosity_threshold}")
|
||
print(f" Session interventions:")
|
||
for head, count in Store.state.get('cfhot_interventions', {}).items():
|
||
print(f" {head}: {count}")
|
||
continue
|
||
|
||
if u == "dense":
|
||
print("\n[Density Status]")
|
||
print(f" Dense mode: {Config.use_dense}")
|
||
print(f" Current checkpoint: {Store.state.get('current_checkpoint', 'unknown')}")
|
||
print(f" Target density: {Config.target_density}")
|
||
if Store.state.get('density_scores'):
|
||
scores = Store.state['density_scores']
|
||
print(f" Session density scores:")
|
||
print(f" Count: {len(scores)}")
|
||
print(f" Avg: {sum(scores)/len(scores):.1f}")
|
||
print(f" Range: {min(scores):.1f} - {max(scores):.1f}")
|
||
continue
|
||
|
||
if u.startswith("density:"):
|
||
text = u[8:].strip()
|
||
if not text:
|
||
print("[density] Provide text to analyze")
|
||
continue
|
||
info = analyze_density(text, tok)
|
||
print(f"\n[Density Analysis]")
|
||
print(f" Tokens: {info['tokens']}")
|
||
print(f" Words: {info['words']}")
|
||
print(f" Unique content words: {info['unique_content_words']}")
|
||
print(f" Density score: {info['density']:.1f}")
|
||
print(f" Filler phrases: {info['filler_phrases']}")
|
||
print(f" Passes threshold: {info['passes_threshold']}")
|
||
continue
|
||
|
||
if u == "goals":
|
||
print("[goals]")
|
||
if not Store.goals:
|
||
print(" (none)")
|
||
for i, g in enumerate(Store.goals):
|
||
print(f" [{i}] {g}")
|
||
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] no plan to refine")
|
||
continue
|
||
improved = reflect_on(last_plan, tok, model)
|
||
last_plan = improved
|
||
Store.log_mem("reflect", {"plan": improved})
|
||
print(improved)
|
||
continue
|
||
|
||
if u.startswith("lht:"):
|
||
print(tool_lht_analyze(u[4:].strip(), tok))
|
||
continue
|
||
|
||
if u.startswith("tool:"):
|
||
print(tool_router(u[5:].strip(), tok, model))
|
||
continue
|
||
|
||
if u.startswith("shell:"):
|
||
print(tool_shell(u[6:].strip()))
|
||
continue
|
||
|
||
if u.startswith("py:"):
|
||
print(tool_py(u[3:].strip()))
|
||
continue
|
||
|
||
if u.startswith("search:"):
|
||
print(tool_search_local(u[7:].strip()))
|
||
continue
|
||
|
||
if u.startswith("toggle"):
|
||
parts = u.split(maxsplit=1)
|
||
if len(parts) > 1:
|
||
print(Config.toggle(parts[1]))
|
||
else:
|
||
print("[toggle] specify flag: use_voice, use_vector_memory, use_lht_reasoning, use_cfhot, use_dense, use_agentic, autonomy")
|
||
continue
|
||
|
||
if u == "status":
|
||
status = {
|
||
"turn": Store.state["turn"],
|
||
"goals": len(Store.goals),
|
||
"improvement_iterations": Store.state.get("improvement_iterations", 0),
|
||
"training_runs": len(Store.state.get("training_runs", [])),
|
||
"current_checkpoint": Store.state.get("current_checkpoint", "unknown"),
|
||
"autonomy": Config.autonomy,
|
||
"use_vector_memory": Config.use_vector_memory,
|
||
"use_lht_reasoning": Config.use_lht_reasoning,
|
||
"use_cfhot": Config.use_cfhot,
|
||
"use_dense": Config.use_dense,
|
||
"use_agentic": Config.use_agentic,
|
||
"cfhot_interventions": Store.state.get("cfhot_interventions", {}),
|
||
"avg_density": sum(Store.state.get('density_scores', [0])) / max(len(Store.state.get('density_scores', [1])), 1),
|
||
"target_density": Config.target_density,
|
||
"tool_scores": TOOL_SCORES,
|
||
}
|
||
print(json.dumps(status, indent=2))
|
||
continue
|
||
|
||
# Default: generate response
|
||
out = generate(tok, model, f"{persona_directive()}\nUser request: {u}")
|
||
Store.log_mem("reply", {"in": u, "out": out})
|
||
print(out)
|
||
|
||
if Config.use_lht_reasoning and Store.state["turn"] % 3 == 0:
|
||
print(tool_lht_analyze(out, tok))
|
||
|
||
Store.state["turn"] += 1
|
||
Store.save()
|
||
|
||
final_report()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|