src/dream_engine/helix_memory.py

import numpy as np
import pickle
from pathlib import Path
from typing import List, Tuple, Dict, Optional

class HelixMemory:
    def __init__(self, storage_path: Path):
        self.storage_path = storage_path
        self._load()

    def _load(self) -> None:
        if self.storage_path.exists():
            with self.storage_path.open("rb") as f:
                self.entries: List[Tuple[int, np.ndarray, int, dict]] = pickle.load(f)
        else:
            self.entries = []

    def _save(self) -> None:
        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
        with self.storage_path.open("wb") as f:
            pickle.dump(self.entries, f)

    def add(self, hv: np.ndarray, meta: Optional[dict] = None) -> None:
        meta = meta or {}
        for i, (code, proto, cnt, old_meta) in enumerate(self.entries):
            sim = np.mean(hv == proto)
            if sim > 0.85:
                merged = {**old_meta, **meta}
                self.entries[i] = (code, proto, cnt + 1, merged)
                self._save()
                return

        new_code = len(self.entries)
        self.entries.append((new_code, hv.copy(), 1, meta))
        self._save()

    def retrieve(self, hv: np.ndarray, top_k: int = 3) -> List[Tuple[np.ndarray, dict]]:
        sims = [(np.mean(hv == proto), proto, meta) for _, proto, _, meta in self.entries]
        sims.sort(key=lambda x: x[0], reverse=True)
        return [(proto, meta) for _, proto, meta in sims[:top_k]]

    def reconstruct(self, code_id: int) -> np.ndarray:
        for cid, proto, _, _ in self.entries:
            if cid == code_id:
                return proto.copy()
        raise KeyError(f"Helix code {code_id} not found")

    def most_uncertain(self) -> Tuple[int, np.ndarray]:
        if not self.entries:
            raise RuntimeError("HelixMemory empty")
        entry = min(self.entries, key=lambda e: e[2])
        return entry[0], entry[1]