| """ | |
| Cognitive Evolution Log for Berna R7. | |
| Records every event in the model's cognitive lifecycle: | |
| - birth: new cell created when new pattern discovered | |
| - split: existing cell divides due to saturation | |
| - pattern: new knowledge signature detected | |
| - milestone: notable thresholds crossed | |
| Persistent (JSONL), queryable, narrative-generating, feedable to model. | |
| """ | |
| import json | |
| import hashlib | |
| from pathlib import Path | |
| from datetime import datetime, timezone | |
| from typing import List, Optional | |
| class CognitiveLog: | |
| def __init__(self, log_dir, model_uuid="mother-001"): | |
| self.log_dir = Path(log_dir) | |
| self.log_dir.mkdir(parents=True, exist_ok=True) | |
| self.model_uuid = model_uuid | |
| self.jsonl_path = self.log_dir / f"{model_uuid}_events.jsonl" | |
| self.narrative_path = self.log_dir / f"{model_uuid}_narrative.txt" | |
| self.events = [] | |
| self.MAX_EVENTS = 10000 | |
| self._load_existing() | |
| def _load_existing(self): | |
| if self.jsonl_path.exists(): | |
| with open(self.jsonl_path, "r", encoding="utf-8") as f: | |
| for line in f: | |
| try: | |
| self.events.append(json.loads(line)) | |
| except Exception: | |
| pass | |
| def record(self, event_type, **fields): | |
| event = { | |
| "timestamp": datetime.now(timezone.utc).isoformat(), | |
| "step": fields.pop("step", None), | |
| "type": event_type, | |
| "model_uuid": self.model_uuid, | |
| } | |
| event.update(fields) | |
| self.events.append(event) | |
| if len(self.events) > self.MAX_EVENTS: | |
| self.events = self.events[-self.MAX_EVENTS:] | |
| with open(self.jsonl_path, "a", encoding="utf-8") as f: | |
| f.write(json.dumps(event, ensure_ascii=False) + "\n") | |
| return event | |
| def record_birth(self, cell_id, parent_id, pattern_sig, step, layer_idx): | |
| return self.record("birth", step=step, cell_id=cell_id, | |
| parent_id=parent_id, pattern_sig=pattern_sig, | |
| layer_idx=layer_idx) | |
| def record_split(self, parent_id, child_a, child_b, S, step, layer_idx): | |
| return self.record("split", step=step, parent_id=parent_id, | |
| child_a=child_a, child_b=child_b, | |
| S=round(float(S), 4), layer_idx=layer_idx) | |
| def record_pattern(self, pattern_sig, cell_ids, step, count): | |
| return self.record("pattern", step=step, pattern_sig=pattern_sig, | |
| cell_ids=cell_ids, seen_count=count) | |
| def record_milestone(self, name, value, step): | |
| return self.record("milestone", step=step, name=name, value=value) | |
| def recent(self, n=20, event_type=None): | |
| if event_type is None: | |
| return self.events[-n:] | |
| return [e for e in self.events if e["type"] == event_type][-n:] | |
| def count_by_type(self): | |
| counts = {} | |
| for e in self.events: | |
| counts[e["type"]] = counts.get(e["type"], 0) + 1 | |
| return counts | |
| def to_narrative(self, max_events=50): | |
| events = self.events[-max_events:] | |
| lines = [] | |
| lines.append(f"=== Berna {self.model_uuid} — Cognitive Chronicle ===") | |
| lines.append(f"Total events: {len(self.events)}") | |
| lines.append(f"Breakdown: {self.count_by_type()}") | |
| lines.append("") | |
| for e in events: | |
| t = e["type"] | |
| if t == "birth": | |
| lines.append(f"[step {e['step']}] Born: {e['cell_id']} " | |
| f"(from {e.get('parent_id','seed')}) — " | |
| f"knows pattern {str(e.get('pattern_sig','?'))[:8]}") | |
| elif t == "split": | |
| lines.append(f"[step {e['step']}] Split: {e['parent_id']} → " | |
| f"{e['child_a']} + {e['child_b']} (S={e['S']})") | |
| elif t == "pattern": | |
| lines.append(f"[step {e['step']}] New pattern: " | |
| f"{str(e['pattern_sig'])[:8]} " | |
| f"(seen {e['seen_count']}×, matched " | |
| f"{len(e.get('cell_ids', []))} cells)") | |
| elif t == "milestone": | |
| lines.append(f"[step {e['step']}] Milestone: " | |
| f"{e['name']} = {e['value']}") | |
| return "\n".join(lines) | |
| def write_narrative(self, path=None): | |
| text = self.to_narrative() | |
| p = Path(path) if path else self.narrative_path | |
| p.write_text(text, encoding="utf-8") | |
| return p | |
| def sample_for_training(self, n=2): | |
| """Short self-narrative snippet to prepend to training samples.""" | |
| births = self.recent(n=n, event_type="birth") | |
| splits = self.recent(n=n, event_type="split") | |
| lines = ["I am Berna R7, a self-growing language model."] | |
| if births: | |
| ids = ", ".join(e["cell_id"] for e in births) | |
| lines.append(f"Recently I gave birth to new cells: {ids}.") | |
| if splits: | |
| ids = ", ".join(e["parent_id"] for e in splits) | |
| lines.append(f"I also split {ids} to refine my knowledge.") | |
| if not births and not splits: | |
| lines.append("I am learning language, mathematics, and code.") | |
| return " ".join(lines) | |
| def pattern_signature(activations, top_k=50): | |
| """Binary fingerprint of top-K activated neurons, hashed.""" | |
| import numpy as np | |
| a = np.abs(np.asarray(activations, dtype=np.float32)) | |
| mean_a = a.mean(axis=0) | |
| if mean_a.size == 0: | |
| return "empty" | |
| k = min(top_k, mean_a.size) | |
| top_idx = np.argsort(mean_a)[-k:] | |
| fp = np.zeros(mean_a.size, dtype=np.uint8) | |
| fp[top_idx] = 1 | |
| return hashlib.sha256(fp.tobytes()).hexdigest()[:16] | |
| def hamming_similarity(fp_a, fp_b): | |
| """Similarity ∈ [0,1] between two fingerprints (hex strings).""" | |
| import numpy as np | |
| try: | |
| a = np.frombuffer(bytes.fromhex(fp_a), dtype=np.uint8) | |
| b = np.frombuffer(bytes.fromhex(fp_b), dtype=np.uint8) | |
| except Exception: | |
| return 0.0 | |
| n = min(len(a), len(b)) | |
| if n == 0: | |
| return 0.0 | |
| matches = np.sum(a[:n] == b[:n]) | |
| return float(matches / n) | |
| if __name__ == "__main__": | |
| import numpy as np | |
| log = CognitiveLog("/tmp/berna_test_log") | |
| log.record_birth("c-001", "seed", "abc123", 1, 0) | |
| log.record_split("c-000", "c-000-a", "c-000-b", 0.87, 100, 0) | |
| log.record_milestone("first_split", 1, 100) | |
| print(log.to_narrative()) | |
| print() | |
| a = np.random.randn(32, 128).astype(np.float32) | |
| b = np.random.randn(32, 128).astype(np.float32) | |
| print("fp_a:", pattern_signature(a)) | |
| print("fp_b:", pattern_signature(b)) | |
| print("sim:", hamming_similarity(pattern_signature(a), pattern_signature(b))) | |