"""
PHANTOM AI — Kognitivní Chatbot (HuggingFace Space)
POLARIS 0.18.13 | HELIX-9 | NexusServer (NOCTUS) | GoalVector | TranslationLayer | v2.1
"""
import os, json, time, threading, math, shutil
from pathlib import Path
import numpy as np
import gradio as gr
# ══════════════════════════════════════════════════════════════════════════════
# 1. NEUROMORPHIC MEMORY
# ══════════════════════════════════════════════════════════════════════════════
BANK_DIR = Path("neuromorphic_v1")
_DATA_DIR = Path("neuromorphic_data")
CHANNEL_NAMES = ['episodic', 'semantic', 'procedural', 'working', 'consolidated']
LAYER_NAMES = ['active', 'consolidated', 'archive']
N_CHANNELS, N_LAYERS = 5, 3
EMBED_DIM = 256
try:
import faiss
HAS_FAISS = True
except ImportError:
HAS_FAISS = False
def _assemble_from_chunks():
manifest_path = _DATA_DIR / "manifest.json"
if not manifest_path.exists():
print("[DB] manifest.json nenalezen v neuromorphic_data/")
return
manifest = json.loads(manifest_path.read_text())
BANK_DIR.mkdir(exist_ok=True)
for stem, info in manifest.items():
out_path = BANK_DIR / f"{stem}.npy"
if out_path.exists():
continue
print(f"[DB] Sestavuji {stem} ({len(info['chunks'])} chunků)...")
pieces = []
for cname in info["chunks"]:
chunk_path = _DATA_DIR / cname
if not chunk_path.exists():
print(f"[DB] chybí chunk {cname}")
return
loaded = np.load(str(chunk_path), allow_pickle=True)
key = "data" if "data" in loaded else list(loaded.keys())[0]
piece = loaded[key]
pieces.append(piece)
arr = np.concatenate(pieces, axis=0)
if info.get("dtype") == "float16":
arr = arr.astype(np.float32)
np.save(str(out_path), arr)
print(f"[DB] → {out_path} ({arr.shape})")
meta_src = _DATA_DIR / "meta.json"
meta_dst = BANK_DIR / "meta.json"
if meta_src.exists() and not meta_dst.exists():
shutil.copy(meta_src, meta_dst)
print("[DB] Sestavení hotovo.")
_sentinel = BANK_DIR / "meta.json"
if not _sentinel.exists():
print("[DB] Banka nenalezena — sestavuji z neuromorphic_data/...")
_assemble_from_chunks()
else:
print(f"[DB] Banka nalezena v {BANK_DIR}/")
texts_bank = [[[] for _ in range(N_LAYERS)] for _ in range(N_CHANNELS)]
domains_bank = [[[] for _ in range(N_LAYERS)] for _ in range(N_CHANNELS)]
hexa_bank = [[[] for _ in range(N_LAYERS)] for _ in range(N_CHANNELS)]
faiss_idx = [[None] * N_LAYERS for _ in range(N_CHANNELS)]
np_embs = [[[] for _ in range(N_LAYERS)] for _ in range(N_CHANNELS)]
channel_basis = None
enc = None
pol_W = None
total_entries = 0
meta = {}
def load_bank():
global texts_bank, domains_bank, hexa_bank, faiss_idx, np_embs
global channel_basis, total_entries, meta, enc, pol_W
if not BANK_DIR.exists():
print("BANK_DIR nenalezen — spusť sestavení nejdřív")
return
try:
meta = json.loads((BANK_DIR / "meta.json").read_text())
cb_path = BANK_DIR / "channel_basis.npy"
if cb_path.exists():
channel_basis = np.load(str(cb_path))
bb_path = BANK_DIR / "backbone_W.npy"
if bb_path.exists():
pol_W = np.load(str(bb_path)).astype(np.float32)
except Exception as e:
print(f"Meta load error: {e}")
n = 0
for ch in range(N_CHANNELS):
for layer in range(N_LAYERS):
prefix = str(BANK_DIR / f"ch{ch}_layer{layer}")
t_path = prefix + "_texts.npy"
if not Path(t_path).exists():
continue
texts_bank[ch][layer] = list(np.load(t_path, allow_pickle=True))
domains_bank[ch][layer] = list(np.load(prefix + "_domains.npy", allow_pickle=True))
h_path = prefix + "_hexa.npy"
if Path(h_path).exists():
hexa_bank[ch][layer] = list(np.load(h_path))
f_path = prefix + "_faiss.index"
if HAS_FAISS and Path(f_path).exists():
faiss_idx[ch][layer] = faiss.read_index(f_path)
e_path = prefix + "_embs.npy"
if Path(e_path).exists():
np_embs[ch][layer] = np.load(e_path)
n += len(texts_bank[ch][layer])
total_entries = n
print(f"Banka načtena: {n:,} záznamů")
try:
from cogcore.encoders.hash_projection import HashProjectionEncoder, HashEncoderConfig
enc_obj = HashProjectionEncoder(HashEncoderConfig(dim=4096))
enc = enc_obj
print("cogcore OK")
except Exception as e:
print(f"cogcore nedostupný (hash fallback): {e}")
def embed(text: str) -> np.ndarray:
if enc is not None and pol_W is not None:
X = enc.encode_batch([text])
e = (X @ pol_W)[0].astype(np.float32)
elif enc is not None:
X = enc.encode_batch([text])
e = X[0, :EMBED_DIM].astype(np.float32)
else:
rng = np.random.default_rng(hash(text) % (2**31))
e = rng.standard_normal(EMBED_DIM).astype(np.float32)
norm = np.linalg.norm(e)
return e / norm if norm > 1e-9 else e
def search_memory(query: str, top_k: int = 10,
channels=None, layers=None) -> list:
q = embed(query)
chs = channels if channels else list(range(N_CHANNELS))
lys = layers if layers else [0, 1]
results = []
for ch in chs:
for l in lys:
n_items = len(texts_bank[ch][l])
if n_items == 0:
continue
k = min(top_k, n_items)
if HAS_FAISS and faiss_idx[ch][l] is not None:
D, I = faiss_idx[ch][l].search(q.reshape(1, -1), k)
for score, idx in zip(D[0], I[0]):
if idx >= 0:
hexa = hexa_bank[ch][l][idx] if idx < len(hexa_bank[ch][l]) else None
results.append({
"text": texts_bank[ch][l][idx],
"score": float(score),
"channel": CHANNEL_NAMES[ch],
"layer": LAYER_NAMES[l],
"domain": domains_bank[ch][l][idx] if idx < len(domains_bank[ch][l]) else "",
"hexa": int(hexa) if hexa is not None else None,
})
elif len(np_embs[ch][l]) > 0:
embs = np_embs[ch][l] if isinstance(np_embs[ch][l], np.ndarray) else np.vstack(np_embs[ch][l])
sc = embs @ q
for idx in np.argsort(sc)[::-1][:k]:
hexa = hexa_bank[ch][l][idx] if idx < len(hexa_bank[ch][l]) else None
results.append({
"text": texts_bank[ch][l][idx],
"score": float(sc[idx]),
"channel": CHANNEL_NAMES[ch],
"layer": LAYER_NAMES[l],
"domain": domains_bank[ch][l][idx] if idx < len(domains_bank[ch][l]) else "",
"hexa": int(hexa) if hexa is not None else None,
})
results.sort(key=lambda x: x["score"], reverse=True)
return results[:top_k]
# ══════════════════════════════════════════════════════════════════════════════
# 2. NEXUSSERVER (NOCTUS) — rezonanční bus 32D
# ══════════════════════════════════════════════════════════════════════════════
class PhantomNexus:
def __init__(self, dim: int = 32):
self.dim = dim
self._bus = np.zeros(dim, dtype=np.float64)
self._lock = threading.Lock()
self._tick_count = 0
self._tick_ms_avg = 0.0
self._resonance = 0.0
self._coherence = 1.0
self._phase_velocity = 0.0
self._entropy = 0.0
self._running = False
self._thread = None
self._rng = np.random.default_rng(42)
def tick(self, signal: np.ndarray = None) -> dict:
t0 = time.perf_counter()
with self._lock:
noise = self._rng.standard_normal(self.dim) * 0.05
self._bus = self._bus * 0.95 + noise
if signal is not None:
sig = np.asarray(signal, dtype=np.float64)
if sig.ndim == 1:
n = min(len(sig), self.dim)
self._bus[:n] += sig[:n] * 0.3
mean_val = float(np.mean(self._bus))
std_val = float(np.std(self._bus))
self._resonance = float(abs(mean_val))
self._coherence = float(1.0 - std_val / max(std_val, 1e-9))
delta = float(np.linalg.norm(noise))
self._phase_velocity = delta
hist, _ = np.histogram(self._bus, bins=8, density=True)
hist = hist + 1e-12
hist /= hist.sum()
self._entropy = float(-np.sum(hist * np.log2(hist + 1e-12)))
self._tick_count += 1
elapsed_ms = (time.perf_counter() - t0) * 1000.0
self._tick_ms_avg = 0.95 * self._tick_ms_avg + 0.05 * elapsed_ms
return self.status()
def _loop(self):
interval = 1.0 / 20.0
while self._running:
self.tick()
time.sleep(interval)
def start(self):
if self._running:
return
self._running = True
self._thread = threading.Thread(target=self._loop, daemon=True)
self._thread.start()
def stop(self):
self._running = False
def status(self) -> dict:
with self._lock:
return {
"resonance": self._resonance,
"coherence": self._coherence,
"phase_velocity": self._phase_velocity,
"entropy": self._entropy,
"tick_count": self._tick_count,
"tick_ms_avg": self._tick_ms_avg,
}
# ══════════════════════════════════════════════════════════════════════════════
# 3. EMOTIONAL STATE
# ══════════════════════════════════════════════════════════════════════════════
class EmotionalState:
_POSITIVE_KW = {
"skvěl", "výborn", "perfekt", "super", "krásn", "báječn", "úžasn",
"dobr", "ano", "správn", "díky", "děkuj", "zajímav", "chytr",
"mocn", "radost", "štěst", "lásk", "pomoc", "řešen",
}
_NEGATIVE_KW = {
"špatn", "chyb", "probl", "nevím", "bohužel", "nemohu", "není",
"neexistuj", "selhán", "chybí", "chybějí", "smutn", "strach",
"nebezpeč", "varován", "kriz", "nefunguj", "nevyřeš",
}
def __init__(self):
self.DA = 0.5
self.NE = 0.4
self.CORT = 0.3
self.OXT = 0.5
self._5HT = 0.5
self.ACh = 0.6
self.valence = 0.0
self.arousal = 0.3
self.stress = 0.2
def _text_valence(self, text: str) -> float:
t = text.lower()
pos = sum(1 for kw in self._POSITIVE_KW if kw in t)
neg = sum(1 for kw in self._NEGATIVE_KW if kw in t)
total = pos + neg
return 0.0 if total == 0 else (pos - neg) / total
def update(self, text: str, context_scores: list):
text_val = self._text_valence(text)
avg_score = float(np.clip(np.mean(context_scores), 0.0, 1.0)) if context_scores else 0.0
length_factor = min(len(text) / 200.0, 1.0)
combined_val = float(np.clip(0.5 * text_val + 0.5 * (avg_score * 2.0 - 1.0), -1.0, 1.0))
alpha = 0.3
self.valence = (1 - alpha) * self.valence + alpha * combined_val
self.arousal = (1 - alpha) * self.arousal + alpha * (0.4 + 0.6 * length_factor)
if combined_val > 0:
self.DA = float(np.clip(self.DA + alpha * combined_val * 0.4, 0.0, 1.0))
self.OXT = float(np.clip(self.OXT + alpha * 0.1, 0.0, 1.0))
self._5HT = float(np.clip(self._5HT + alpha * 0.05, 0.0, 1.0))
self.CORT = float(np.clip(self.CORT - alpha * 0.1, 0.0, 1.0))
else:
self.CORT = float(np.clip(self.CORT + alpha * abs(combined_val) * 0.5, 0.0, 1.0))
self.NE = float(np.clip(self.NE + alpha * 0.2, 0.0, 1.0))
self.DA = float(np.clip(self.DA - alpha * 0.1, 0.0, 1.0))
self.stress = float(np.clip(0.6 * self.CORT + 0.4 * (1 - self._5HT), 0.0, 1.0))
decay = 0.02
self.DA = float(self.DA * (1 - decay) + 0.5 * decay)
self.NE = float(self.NE * (1 - decay) + 0.4 * decay)
self.CORT = float(self.CORT * (1 - decay) + 0.3 * decay)
self.OXT = float(self.OXT * (1 - decay) + 0.5 * decay)
self._5HT = float(self._5HT * (1 - decay) + 0.5 * decay)
self.ACh = float(self.ACh * (1 - decay) + 0.6 * decay)
def snapshot(self) -> dict:
return {
"valence": round(self.valence, 3),
"arousal": round(self.arousal, 3),
"stress": round(self.stress, 3),
"DA": round(self.DA, 3),
"NE": round(self.NE, 3),
"CORT": round(self.CORT, 3),
"OXT": round(self.OXT, 3),
"5HT": round(self._5HT, 3),
"ACh": round(self.ACh, 3),
}
# ══════════════════════════════════════════════════════════════════════════════
# 4. INTROSPECTION ENGINE
# ══════════════════════════════════════════════════════════════════════════════
class IntrospectionEngine:
_INTROSPECTIVE_KW = [
"jak funguje", "co jsi", "kdo jsi", "vysvětli", "co víš",
"jak pracuje", "co umíš", "jak myslíš", "co si myslíš",
"jak se cítíš", "co je phantom", "co je helix",
]
def evaluate(self, text: str, context_hits: list,
emotional_snap: dict, nexus_status: dict) -> dict:
tl = text.lower()
is_intro = any(kw in tl for kw in self._INTROSPECTIVE_KW)
steps = []
word_count = len(text.split())
steps.append(f"DECOMPOSE: dotaz obsahuje {word_count} slov, "
f"{'introspektivní' if is_intro else 'faktický'} charakter")
n_hits = len(context_hits)
if n_hits > 0:
top_score = context_hits[0]["score"]
channels_used = list({h["channel"] for h in context_hits})
steps.append(f"SEARCH: nalezeno {n_hits} záznamů, "
f"nejlepší skóre {top_score:.3f}, "
f"kanály: {', '.join(channels_used)}")
else:
steps.append("SEARCH: žádné záznamy nenalezeny")
if context_hits:
top3_scores = [h["score"] for h in context_hits[:3]]
kq = float(np.clip(np.mean(top3_scores), 0.0, 1.0))
else:
kq = 0.0
emo_note = ""
if emotional_snap.get("DA", 0.5) > 0.7:
emo_note = "vysoká aktivace DA → optimistický kontext"
elif emotional_snap.get("CORT", 0.3) > 0.7:
emo_note = "vysoký kortizol → zvýšená opatrnost"
else:
emo_note = "klidný emocionální stav"
steps.append(f"EVALUATE: knowledge_quality={kq:.2f}, {emo_note}")
nexus_res = nexus_status.get("resonance", 0.0)
steps.append(f"RENDER: nexus resonance={nexus_res:.3f}, "
f"syntéza {'z kontextu' if kq > 0.3 else 'generická'}")
if kq > 0.7:
czech_summary = f"Kvalitní znalost (skóre {kq:.2f}). Syntetizováno z {n_hits} záznamů."
elif kq > 0.3:
czech_summary = f"Částečná znalost (skóre {kq:.2f}). Odpověď může být nekompletní."
else:
czech_summary = "Relevantní znalost nenalezena. Odpověď je generická."
if is_intro:
czech_summary = "Introspektivní dotaz. " + czech_summary
return {
"knowledge_quality": round(kq, 3),
"czech_summary": czech_summary,
"is_introspective": is_intro,
"reasoning_steps": steps,
}
# ══════════════════════════════════════════════════════════════════════════════
# 5. HELIX-9 STATUS
# ══════════════════════════════════════════════════════════════════════════════
class Helix9Status:
N_HEADS = 9
N_LAYERS = 6
D_MODEL = 512
PARAMS_M = 35.8
SPIRAL = True
def __init__(self):
rng = np.random.default_rng(2024)
self._head_proj = rng.standard_normal((self.N_HEADS, EMBED_DIM)).astype(np.float32)
norms = np.linalg.norm(self._head_proj, axis=1, keepdims=True)
self._head_proj /= np.maximum(norms, 1e-9)
self._spiral_phases = np.linspace(0, 2 * math.pi, self.N_HEADS, endpoint=False)
def activate(self, emb: np.ndarray) -> np.ndarray:
emb_f = np.asarray(emb, dtype=np.float32)
raw = self._head_proj @ emb_f
if self.SPIRAL:
spiral_mod = np.cos(self._spiral_phases + float(np.sum(emb_f[:8])))
raw = raw + 0.2 * spiral_mod
return (1.0 / (1.0 + np.exp(-raw * 3.0))).astype(np.float32)
# ══════════════════════════════════════════════════════════════════════════════
# 5.5 GOAL VECTOR — sledování záměru konverzace
# ══════════════════════════════════════════════════════════════════════════════
class GoalVector:
"""
256D cílový vektor — reprezentuje záměr konverzace.
Aktualizuje se pomocí EMA z každého dotazu.
Určuje dominantní kognitivní cíl z kanálového směrování.
"""
_GOAL_LABELS = {
"semantic": "Faktický dotaz",
"episodic": "Kontextuální vzpomínka",
"procedural": "Navigace / Akce",
"working": "Krátkodobý pracovní cíl",
"consolidated": "Hluboká introspekce",
}
def __init__(self, dim: int = EMBED_DIM):
self._vec = np.zeros(dim, dtype=np.float32)
self._prev = np.zeros(dim, dtype=np.float32)
self._goal = "Neurčeno"
self._align = 0.0
self._drift = 0.0
self._iter = 0
def update(self, query_emb: np.ndarray, hits: list):
alpha = 0.35
self._prev = self._vec.copy()
self._vec = (1 - alpha) * self._vec + alpha * query_emb
norm = np.linalg.norm(self._vec)
if norm > 1e-9:
self._vec /= norm
if self._iter > 0:
self._drift = float(np.linalg.norm(self._vec - self._prev))
if hits:
ch_cnt: dict = {}
for h in hits:
ch = h.get("channel", "semantic")
ch_cnt[ch] = ch_cnt.get(ch, 0) + 1
dom = max(ch_cnt, key=ch_cnt.get)
self._goal = self._GOAL_LABELS.get(dom, "Obecný dotaz")
self._align = float(np.dot(self._vec, query_emb))
self._iter += 1
def status(self) -> dict:
return {
"goal": self._goal,
"alignment": round(self._align, 3),
"drift": round(self._drift, 3),
"iteration": self._iter,
}
# ══════════════════════════════════════════════════════════════════════════════
# 5.6 TRANSLATION LAYER — vstupní/výstupní překladová vrstva
# ══════════════════════════════════════════════════════════════════════════════
class TranslationLayer:
"""
Sleduje transformaci na vstupním a výstupním kanálu:
Vstupní: text → hash embedding → 256D POLARIS vektor → kanálové směrování
Výstupní: retrieved paměti → extrakce vět → český text
"""
def __init__(self):
self._in_chars = 0
self._in_norm = 0.0
self._ch_dist: dict = {}
self._out_words = 0
self._compression = 0.0
self._sources = 0
def encode(self, text: str, emb: np.ndarray, hits: list):
self._in_chars = len(text)
self._in_norm = float(np.linalg.norm(emb))
ch_cnt: dict = {}
for h in hits:
ch = h.get("channel", "?")
ch_cnt[ch] = ch_cnt.get(ch, 0) + 1
self._ch_dist = ch_cnt
self._sources = len(hits)
def decode(self, response: str, hits: list):
self._out_words = len(response.split())
total_src = sum(len(str(h.get("text", ""))) for h in hits[:3])
self._compression = round(len(response) / max(total_src, 1), 3)
def status(self) -> dict:
return {
"in_chars": self._in_chars,
"in_norm": round(self._in_norm, 4),
"ch_dist": self._ch_dist,
"out_words": self._out_words,
"compression": self._compression,
"sources": self._sources,
}
# ══════════════════════════════════════════════════════════════════════════════
# 6. SYNTHESIS
# ══════════════════════════════════════════════════════════════════════════════
def _bar(value: float, width: int = 10) -> str:
v = float(np.clip(value, 0.0, 1.0))
filled = round(v * width)
return "█" * filled + "░" * (width - filled)
def synthesize_response(user_msg: str, context_hits: list,
emotional_snap: dict, intro_result: dict,
goal_status: dict) -> str:
if not context_hits:
return ("Nemohu najít relevantní informaci k vašemu dotazu. "
"Zkuste přeformulovat otázku nebo se zeptejte na jiné téma.")
max_score = context_hits[0]["score"]
if max_score < 0.3:
base = ("Nemohu najít relevantní informaci k vašemu dotazu. "
"Moje paměťová banka neobsahuje dostatečně relevantní záznamy.")
else:
top3 = context_hits[:3]
sentences = []
for hit in top3:
txt = str(hit["text"])
if len(txt) > 300:
txt = txt[:300].rsplit(" ", 1)[0] + "..."
sentences.append(txt)
goal = goal_status.get("goal", "Obecný dotaz")
align = goal_status.get("alignment", 0.0)
if intro_result.get("is_introspective"):
preamble = ("Jsem PHANTOM AI — kognitivní systém s neuromorphní pamětí, "
"HELIX-9 architekturou a NexusServer rezonančním busem. "
"Zde jsou relevantní informace z mé paměti:\n\n")
elif goal == "Navigace / Akce":
preamble = "K dosažení tohoto cíle doporučuji:\n\n"
elif goal == "Hluboká introspekce":
preamble = "Na základě hluboké paměťové analýzy:\n\n"
elif align > 0.7:
preamble = f"Relevantní nalezeno (goal: {goal}):\n\n"
else:
preamble = "Na základě mé paměťové banky:\n\n"
base = preamble + " ".join(sentences)
da = emotional_snap.get("DA", 0.5)
cort = emotional_snap.get("CORT", 0.3)
suffix = ""
if da > 0.7:
suffix = "\n\n*[DA↑ Optimistický tón — systém v pozitivním stavu.]*"
elif cort > 0.7:
suffix = "\n\n*[CORT↑ Zvýšená opatrnost — doporučuji kritické ověření.]*"
elif cort > 0.5:
suffix = "\n\n*[Systém zaznamenal mírně vyšší stresovou hladinu.]*"
return base + suffix
# ══════════════════════════════════════════════════════════════════════════════
# RENDER FUNKCE
# ══════════════════════════════════════════════════════════════════════════════
def render_nexus_md(status: dict) -> str:
tick = status.get("tick_count", 0)
res = status.get("resonance", 0.0)
coh = status.get("coherence", 1.0)
pv = status.get("phase_velocity", 0.0)
ent = status.get("entropy", 0.0)
ms = status.get("tick_ms_avg", 0.0)
return (
f"**📡 NexusServer — tick #{tick}**\n\n"
f"Resonance: `{res:.3f}` {_bar(res)}\n\n"
f"Coherence: `{coh:.3f}` {_bar(coh)}\n\n"
f"Phase vel: `{pv:.3f}`\n\n"
f"Entropy: `{ent:.2f} bit`\n\n"
f"Avg tick: `{ms:.1f} ms`"
)
def render_emotion_md(snap: dict) -> str:
val = snap.get("valence", 0.0)
aro = snap.get("arousal", 0.3)
stre = snap.get("stress", 0.2)
da = snap.get("DA", 0.5)
ne = snap.get("NE", 0.4)
cort = snap.get("CORT", 0.3)
oxt = snap.get("OXT", 0.5)
sht = snap.get("5HT", 0.5)
ach = snap.get("ACh", 0.6)
val_bar = (val + 1.0) / 2.0
return (
f"**🧠 Emoční stav**\n\n"
f"Valence: `{val:+.3f}` {_bar(val_bar)}\n\n"
f"Arousal: `{aro:.3f}` {_bar(aro)}\n\n"
f"Stres: `{stre:.3f}` {_bar(stre)}\n\n"
f"---\n\n"
f"DA (Dopamin): `{da:.2f}` {_bar(da, 8)}\n\n"
f"NE (Noradrenalin): `{ne:.2f}` {_bar(ne, 8)}\n\n"
f"CORT (Kortizol): `{cort:.2f}` {_bar(cort, 8)}\n\n"
f"OXT (Oxytocin): `{oxt:.2f}` {_bar(oxt, 8)}\n\n"
f"5HT (Serotonin): `{sht:.2f}` {_bar(sht, 8)}\n\n"
f"ACh (Acetylcholin):`{ach:.2f}` {_bar(ach, 8)}"
)
def render_helix_md(activations: np.ndarray) -> str:
lines = ["**🔬 HELIX-9 aktivace hlav**\n"]
for i, v in enumerate(activations):
lines.append(f"Hlava {i+1}: `{v:.3f}` {_bar(float(v), 12)}")
lines.append(f"\n*n_heads=9 | n_layers=6 | d_model=512 | {Helix9Status.PARAMS_M}M params*")
return "\n\n".join(lines)
def render_intro_md(intro: dict) -> str:
kq = intro.get("knowledge_quality", 0.0)
summ = intro.get("czech_summary", "")
is_i = intro.get("is_introspective", False)
steps = intro.get("reasoning_steps", [])
intro_flag = "✅ Ano" if is_i else "❌ Ne"
steps_md = "\n".join(f" {j+1}. {s}" for j, s in enumerate(steps))
return (
f"**🔍 Introspekce — POLARIS analýza**\n\n"
f"Knowledge Quality: `{kq:.3f}` {_bar(kq)}\n\n"
f"Introspektivní: {intro_flag}\n\n"
f"**Souhrn:** {summ}\n\n"
f"**Kroky uvažování:**\n{steps_md}"
)
def render_goal_md(gs: dict) -> str:
goal = gs.get("goal", "Neurčeno")
align = gs.get("alignment", 0.0)
drift = gs.get("drift", 0.0)
itr = gs.get("iteration", 0)
align_n = float(np.clip((align + 1) / 2, 0, 1))
return (
f"**🎯 Goal Vector**\n\n"
f"Cíl: **{goal}**\n\n"
f"Alignment: `{align:.3f}` {_bar(align_n)}\n\n"
f"Drift: `{drift:.3f}` {_bar(min(drift * 5, 1.0))}\n\n"
f"Iterace: `{itr}`"
)
def render_translation_md(ts: dict) -> str:
in_c = ts.get("in_chars", 0)
in_n = ts.get("in_norm", 0.0)
ch = ts.get("ch_dist", {})
out_w = ts.get("out_words", 0)
comp = ts.get("compression", 0.0)
src = ts.get("sources", 0)
ch_str = " | ".join(f"{k}:{v}" for k, v in ch.items()) if ch else "—"
return (
f"**🔄 Překladová vrstva**\n\n"
f"**Vstupní kanál →**\n\n"
f" text `{in_c}` znaků → norma `{in_n:.4f}`\n\n"
f"**Kanálové směrování:**\n\n"
f" {ch_str}\n\n"
f"**← Výstupní kanál**\n\n"
f" `{src}` zdrojů → `{out_w}` slov → komprese `{comp:.3f}`"
)
# ══════════════════════════════════════════════════════════════════════════════
# INICIALIZACE modulů
# ══════════════════════════════════════════════════════════════════════════════
nexus = PhantomNexus()
nexus.start()
emotional = EmotionalState()
helix9 = Helix9Status()
intro_engine = IntrospectionEngine()
goal_vec = GoalVector()
translation_layer = TranslationLayer()
load_bank()
# ══════════════════════════════════════════════════════════════════════════════
# 7. CHAT LOGIKA
# ══════════════════════════════════════════════════════════════════════════════
def phantom_respond(user_msg: str, history: list):
"""
Vrací: (history, nexus_md, emo_md, helix_md, intro_md, goal_md, translation_md)
"""
if not user_msg or not user_msg.strip():
return (
history or [],
render_nexus_md(nexus.status()),
render_emotion_md(emotional.snapshot()),
render_helix_md(np.full(9, 0.5, dtype=np.float32)),
"**🔍 Introspekce**\n\nZadejte zprávu pro zahájení analýzy.",
render_goal_md(goal_vec.status()),
render_translation_md(translation_layer.status()),
)
try:
emb = embed(user_msg)
context_hits = search_memory(user_msg, top_k=8)
translation_layer.encode(user_msg, emb, context_hits)
nexus_tick = nexus.tick(emb[:32])
scores = [h["score"] for h in context_hits]
emotional.update(user_msg, scores)
emo_snap = emotional.snapshot()
head_acts = helix9.activate(emb)
intro_result = intro_engine.evaluate(user_msg, context_hits, emo_snap, nexus_tick)
goal_vec.update(emb, context_hits)
response = synthesize_response(user_msg, context_hits, emo_snap, intro_result, goal_vec.status())
translation_layer.decode(response, context_hits)
history = list(history) if history else []
history.append({"role": "user", "content": user_msg})
history.append({"role": "assistant", "content": response})
return (
history,
render_nexus_md(nexus_tick),
render_emotion_md(emo_snap),
render_helix_md(head_acts),
render_intro_md(intro_result),
render_goal_md(goal_vec.status()),
render_translation_md(translation_layer.status()),
)
except Exception as exc:
import traceback
traceback.print_exc()
history = list(history) if history else []
history.append({"role": "user", "content": user_msg})
history.append({"role": "assistant", "content": f"⚠️ Chyba: {exc}"})
return (
history,
render_nexus_md(nexus.status()),
render_emotion_md(emotional.snapshot()),
render_helix_md(np.full(9, 0.5, dtype=np.float32)),
f"**Chyba:** {exc}",
render_goal_md(goal_vec.status()),
render_translation_md(translation_layer.status()),
)
def _respond_and_clear(user_msg: str, history: list):
"""
Standardní Gradio 5 vzor: chatbot jako vstup I výstup.
history = aktuální hodnota chatbotu (list message dicts).
"""
try:
history = list(history) if history else []
if not user_msg or not user_msg.strip():
return "", history
emb = embed(user_msg)
context_hits = search_memory(user_msg, top_k=8)
translation_layer.encode(user_msg, emb, context_hits)
nexus_tick = nexus.tick(emb[:32])
scores = [h["score"] for h in context_hits]
emotional.update(user_msg, scores)
emo_snap = emotional.snapshot()
helix9.activate(emb)
intro_result = intro_engine.evaluate(user_msg, context_hits, emo_snap, nexus_tick)
goal_vec.update(emb, context_hits)
response = synthesize_response(user_msg, context_hits, emo_snap, intro_result, goal_vec.status())
translation_layer.decode(response, context_hits)
history.append({"role": "user", "content": user_msg})
history.append({"role": "assistant", "content": response})
return "", history
except Exception as exc:
import traceback; traceback.print_exc()
history = list(history) if history else []
history.append({"role": "user", "content": str(user_msg)})
history.append({"role": "assistant", "content": f"⚠️ Chyba: {exc}"})
return "", history
def _new_chat_ui(history: list, sessions: list):
ns, dd = _save_session(history, sessions)
return ns, dd, []
def _load_ui(sel: str, sessions: list):
return _load_session(sel, sessions)
def refresh_panels():
"""Pravidelná aktualizace panelů bez chat interakce (Timer)."""
generic_emb = np.zeros(EMBED_DIM, dtype=np.float32)
return (
render_nexus_md(nexus.status()),
render_emotion_md(emotional.snapshot()),
render_helix_md(helix9.activate(generic_emb)),
render_goal_md(goal_vec.status()),
render_translation_md(translation_layer.status()),
)
# ══════════════════════════════════════════════════════════════════════════════
# SESSION MANAGEMENT (sidebar historie chatů)
# ══════════════════════════════════════════════════════════════════════════════
def _save_session(history: list, sessions: list):
if not history:
return sessions, gr.Dropdown(choices=[s["title"] for s in sessions], value=None)
first_user = next((m["content"] for m in history if m["role"] == "user"), "Chat")
title = (first_user[:45] + "…") if len(first_user) > 45 else first_user
# Nevkládej duplicitu posledního záznamu
if sessions and sessions[-1].get("messages") == history:
return sessions, gr.Dropdown(choices=[s["title"] for s in sessions], value=None)
ns = list(sessions) + [{"title": title, "messages": list(history)}]
return ns, gr.Dropdown(choices=[s["title"] for s in ns], value=None)
def _new_chat(history: list, sessions: list):
ns, dd = _save_session(history, sessions)
blank_panels = refresh_panels()
return (ns, dd, [],
blank_panels[0], blank_panels[1], blank_panels[2],
"**🔍 Introspekce**\n\nNový chat — zadejte zprávu.",
blank_panels[3], blank_panels[4])
def _load_session(selected_title: str, sessions: list):
if not selected_title:
return []
for s in sessions:
if s["title"] == selected_title:
return s["messages"]
return []
# ══════════════════════════════════════════════════════════════════════════════
# 8. GRADIO UI
# ══════════════════════════════════════════════════════════════════════════════
_INITIAL_PANELS = refresh_panels()
_INITIAL_INTRO_MD = (
"**🔍 Introspekce — POLARIS analýza**\n\n"
"Zadejte zprávu pro zahájení kognitivní analýzy.\n\n"
"*DECOMPOSE → SEARCH → EVALUATE → RENDER*"
)
CSS = """
.phantom-header { text-align: center; padding: 6px 0 2px 0; }
.sidebar-box { border: 1px solid #444; border-radius: 8px; padding: 8px; min-height: 300px; }
.panel-box { border: 1px solid #333; border-radius: 6px; padding: 8px; margin-bottom: 4px; }
"""
with gr.Blocks(
title="PHANTOM AI — Kognitivní Chatbot",
theme=gr.themes.Soft(),
css=CSS,
) as demo:
sessions_state = gr.State([])
# ── Header ────────────────────────────────────────────────────────────────
gr.Markdown(
f""""""
)
# ── Hlavní layout: sidebar | chat | panely ────────────────────────────────
with gr.Row(equal_height=False):
# ── Levý sidebar: historie chatů ──────────────────────────────────────
with gr.Column(scale=1, min_width=190, elem_classes=["sidebar-box"]):
gr.Markdown("### 📚 Historie chatů")
sessions_dd = gr.Dropdown(
choices=[],
label="Minulé konverzace",
interactive=True,
allow_custom_value=False,
value=None,
)
load_btn = gr.Button("📂 Načíst", size="sm", variant="secondary")
new_btn = gr.Button("➕ Nový chat", size="sm", variant="primary")
gr.Markdown(
"Při novém chatu se aktuální konverzace automaticky uloží.",
)
# ── Střed: chatbot ────────────────────────────────────────────────────
with gr.Column(scale=3):
chatbot = gr.Chatbot(
type="messages",
value=[],
label="💬 PHANTOM AI Chat",
height=500,
show_label=True,
placeholder=(
"Vítejte v PHANTOM AI! Jsem kognitivní chatbot s neuromorphní pamětí.\n"
"Zeptejte se na cokoli..."
),
bubble_full_width=False,
)
with gr.Row():
msg_input = gr.Textbox(
placeholder="Napište zprávu a stiskněte Enter nebo Odeslat…",
label="",
scale=6,
autofocus=True,
show_label=False,
container=False,
lines=1,
)
send_btn = gr.Button("Odeslat ▶", variant="primary", scale=1, min_width=90)
# ── Pravý sloupec: živé panely ────────────────────────────────────────
with gr.Column(scale=2):
refresh_btn = gr.Button("🔄 Obnovit panely", size="sm", variant="secondary")
nexus_md_out = gr.Markdown(
value=_INITIAL_PANELS[0],
elem_classes=["panel-box"],
)
emo_md_out = gr.Markdown(
value=_INITIAL_PANELS[1],
elem_classes=["panel-box"],
)
with gr.Accordion("🔬 HELIX-9 + 🎯 Goal Vector", open=False):
helix_md_out = gr.Markdown(
value=_INITIAL_PANELS[2],
elem_classes=["panel-box"],
)
goal_md_out = gr.Markdown(
value=_INITIAL_PANELS[3],
elem_classes=["panel-box"],
)
translation_md_out = gr.Markdown(
value=_INITIAL_PANELS[4],
elem_classes=["panel-box"],
)
# ── Introspekce accordion ─────────────────────────────────────────────────
with gr.Accordion("🔍 Introspekce + kontext (kliknutím rozbalit)", open=False):
intro_md_out = gr.Markdown(value=_INITIAL_INTRO_MD)
# ── Odeslat — standardní Gradio 5 vzor: chatbot jako vstup i výstup ───────
send_btn.click(
fn=_respond_and_clear,
inputs=[msg_input, chatbot],
outputs=[msg_input, chatbot],
)
msg_input.submit(
fn=_respond_and_clear,
inputs=[msg_input, chatbot],
outputs=[msg_input, chatbot],
)
# ── Ruční refresh panelů ─────────────────────────────────────────────────
refresh_btn.click(
fn=refresh_panels,
outputs=[nexus_md_out, emo_md_out, helix_md_out, goal_md_out, translation_md_out],
)
# ── Nový chat ─────────────────────────────────────────────────────────────
new_btn.click(
fn=_new_chat_ui,
inputs=[chatbot, sessions_state],
outputs=[sessions_state, sessions_dd, chatbot],
)
# ── Načíst chat ze sidebaru ───────────────────────────────────────────────
load_btn.click(
fn=_load_ui,
inputs=[sessions_dd, sessions_state],
outputs=[chatbot],
)
# ══════════════════════════════════════════════════════════════════════════════
# 9. SPUŠTĚNÍ — HF Spaces spustí demo.launch() automaticky
# ══════════════════════════════════════════════════════════════════════════════
demo.launch()