diff --git a/contextforge/__init__.py b/contextforge/__init__.py index 0121143..52e1a5a 100644 --- a/contextforge/__init__.py +++ b/contextforge/__init__.py @@ -16,6 +16,7 @@ from .infinite_context import InfiniteContext, InfiniteContextStats from .layer import ContextForge +from .hyperdense import HyperDenseMemoryConfig, QuantizationBounds, SQ8VectorIndex from .providers import LLMProvider, get_provider from .tree import WorkingSet from .wiki import WikiCompilationResult, WikiIssue, WikiMemory @@ -24,7 +25,10 @@ "ContextForge", "InfiniteContext", "InfiniteContextStats", + "HyperDenseMemoryConfig", "LLMProvider", + "QuantizationBounds", + "SQ8VectorIndex", "WorkingSet", "WikiCompilationResult", "WikiIssue", diff --git a/contextforge/hyperdense.py b/contextforge/hyperdense.py new file mode 100644 index 0000000..6243acd --- /dev/null +++ b/contextforge/hyperdense.py @@ -0,0 +1,346 @@ +"""Hyper-dense memory primitives: SQ8 vectors, temporal metadata, compaction.""" + +from __future__ import annotations + +import hashlib +import math +import re +import time +from dataclasses import dataclass, field +from typing import Iterable, Literal, Optional, Sequence + +import numpy as np + +from .utils import estimate_tokens + +DistanceMetric = Literal["cosine", "euclidean"] + +_TOKEN_RE = re.compile(r"[A-Za-z0-9_./:-]+") +_MAX_TOKEN_WEIGHT_LENGTH = 24 +_STATE_PATTERNS = ( + re.compile(r"\b(?:file|path|module)\s*[:=]\s*([A-Za-z0-9_./\\-]+)", re.IGNORECASE), + re.compile(r"\b([A-Za-z_][A-Za-z0-9_]*)\s*(?:=|:=|->|became|changed to|updated to)", re.IGNORECASE), + re.compile(r"\b([A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)+)\b"), +) + + +@dataclass(frozen=True) +class HyperDenseMemoryConfig: + """Configuration for ContextForge's local hyper-dense memory engine.""" + + dimensions: int = 384 + semantic_top_k: int = 20 + search_multiplier: int = 4 + token_ceiling: int = 4096 + foundation_states: int = 1 + recent_states: int = 3 + distance_metric: DistanceMetric = "cosine" + enable_state_deduplication: bool = True + + def __post_init__(self) -> None: + if self.dimensions <= 0: + raise ValueError("dimensions must be positive") + if self.semantic_top_k <= 0: + raise ValueError("semantic_top_k must be positive") + if self.search_multiplier <= 0: + raise ValueError("search_multiplier must be positive") + if self.token_ceiling <= 0: + raise ValueError("token_ceiling must be positive") + if self.foundation_states < 0 or self.recent_states < 0: + raise ValueError("foundation_states and recent_states must be non-negative") + if self.distance_metric not in {"cosine", "euclidean"}: + raise ValueError("distance_metric must be 'cosine' or 'euclidean'") + + +@dataclass(frozen=True) +class QuantizationBounds: + """Per-dimension SQ8 calibration bounds for a batch of embeddings.""" + + minimum: np.ndarray + maximum: np.ndarray + + @property + def scale(self) -> np.ndarray: + span = self.maximum - self.minimum + return np.where(span == 0.0, 1.0, span / 255.0).astype(np.float32) + + +@dataclass(frozen=True) +class TemporalMetadata: + """Strict chronology metadata attached to every memory object.""" + + timestamp: float + turn_sequence: int + state_anchors: tuple[str, ...] = field(default_factory=tuple) + + +@dataclass(frozen=True) +class TemporalMemoryRecord: + """A retrieved memory fragment ready for chrono pass and compaction.""" + + key: int | str + token_estimate: int + timestamp: float + turn_sequence: int + state_anchors: tuple[str, ...] = field(default_factory=tuple) + + +def text_to_embedding(text: str, dimensions: int) -> np.ndarray: + """Create a deterministic, local semantic embedding with signed feature hashing.""" + if dimensions <= 0: + raise ValueError("dimensions must be positive") + vector = np.zeros(dimensions, dtype=np.float32) + if not text: + return vector + + for token in _TOKEN_RE.findall(text.lower()): + digest = hashlib.blake2b(token.encode("utf-8"), digest_size=8).digest() + bucket = int.from_bytes(digest[:4], "little") % dimensions + sign = 1.0 if digest[4] & 1 else -1.0 + weight = 1.0 + min(len(token), _MAX_TOKEN_WEIGHT_LENGTH) / _MAX_TOKEN_WEIGHT_LENGTH + vector[bucket] += sign * weight + + norm = float(np.linalg.norm(vector)) + if norm > 0.0 and math.isfinite(norm): + vector /= norm + return vector + + +def calibrate_bounds(embeddings: np.ndarray) -> QuantizationBounds: + """Calculate per-dimension min/max SQ8 calibration bounds for a batch.""" + matrix = _as_float_matrix(embeddings) + if matrix.size == 0 or matrix.shape[0] == 0: + raise ValueError("cannot calibrate an empty embedding batch") + return QuantizationBounds( + minimum=np.min(matrix, axis=0).astype(np.float32), + maximum=np.max(matrix, axis=0).astype(np.float32), + ) + + +def quantize(embeddings: np.ndarray, bounds: Optional[QuantizationBounds] = None) -> tuple[np.ndarray, QuantizationBounds]: + """Quantize FP32 embeddings to uint8 with batch SQ8 calibration.""" + matrix = _as_float_matrix(embeddings) + if matrix.size == 0 or matrix.shape[0] == 0: + raise ValueError("cannot quantize an empty embedding batch") + q_bounds = bounds or calibrate_bounds(matrix) + scale = q_bounds.scale + safe = np.where(scale == 0.0, 1.0, scale) + quantized = np.rint((matrix - q_bounds.minimum) / safe) + return np.clip(quantized, 0, 255).astype(np.uint8), q_bounds + + +def dequantize(quantized_vectors: np.ndarray, bounds: QuantizationBounds) -> np.ndarray: + """Restore SQ8 vectors to approximate FP32 values.""" + q_matrix = _as_uint8_matrix(quantized_vectors) + return (q_matrix.astype(np.float32) * bounds.scale) + bounds.minimum + + +def quantized_euclidean(query: np.ndarray, matrix: np.ndarray) -> np.ndarray: + """Integer-domain negative squared Euclidean score for uint8 vectors.""" + q = _as_uint8_vector(query).astype(np.int16) + m = _as_uint8_matrix(matrix).astype(np.int16) + diff = m - q + return -np.sum(diff.astype(np.int32) * diff.astype(np.int32), axis=1).astype(np.float32) + + +def quantized_cosine(query: np.ndarray, matrix: np.ndarray) -> np.ndarray: + """Integer-domain cosine similarity over SQ8 vectors.""" + q = _as_uint8_vector(query).astype(np.int32) + m = _as_uint8_matrix(matrix).astype(np.int32) + dot = np.sum(m * q, axis=1).astype(np.float32) + q_norm = math.sqrt(float(np.sum(q * q))) + m_norm = np.sqrt(np.sum(m * m, axis=1).astype(np.float32)) + denom = np.maximum(m_norm * max(q_norm, 1e-12), 1e-12) + return dot / denom + + +class SQ8VectorIndex: + """Local uint8 semantic index with deterministic embeddings and vectorized search.""" + + def __init__(self, config: Optional[HyperDenseMemoryConfig] = None) -> None: + self.config = config or HyperDenseMemoryConfig() + self._keys: list[int | str] = [] + self._matrix: Optional[np.ndarray] = None + self._bounds: Optional[QuantizationBounds] = None + + @property + def size(self) -> int: + return len(self._keys) + + @property + def byte_size(self) -> int: + return int(self._matrix.nbytes) if self._matrix is not None else 0 + + def clear(self) -> None: + self._keys.clear() + self._matrix = None + self._bounds = None + + def build(self, documents: Sequence[tuple[int | str, str]]) -> None: + """Build the SQ8 matrix from ``(key, text)`` documents.""" + self.clear() + if not documents: + return + embeddings = np.vstack([ + text_to_embedding(text, self.config.dimensions) for _, text in documents + ]).astype(np.float32) + self._matrix, self._bounds = quantize(embeddings) + self._keys = [key for key, _ in documents] + + def search(self, query: str, top_k: Optional[int] = None) -> list[tuple[int | str, float]]: + """Return top semantic matches using only quantized integer vectors.""" + if not query or self._matrix is None or self._bounds is None or not self._keys: + return [] + query_embedding = text_to_embedding(query, self.config.dimensions).reshape(1, -1) + query_quantized, _ = quantize(query_embedding, self._bounds) + metric = self.config.distance_metric + if metric == "euclidean": + scores = quantized_euclidean(query_quantized[0], self._matrix) + else: + scores = quantized_cosine(query_quantized[0], self._matrix) + limit = max(1, min(top_k or self.config.semantic_top_k, len(self._keys))) + order = np.argsort(scores)[::-1][:limit] + return [(self._keys[int(i)], float(scores[int(i)])) for i in order] + + +def normalize_temporal_metadata( + metadata: Optional[dict], + *, + timestamp: Optional[float] = None, + turn_sequence: Optional[int] = None, + fallback_sequence: int = 0, + content: str = "", + path: str = "", +) -> TemporalMetadata: + """Coerce user metadata into the strict temporal schema with safe fallbacks.""" + meta = dict(metadata or {}) + ts_raw = timestamp if timestamp is not None else meta.get("timestamp") + seq_raw = turn_sequence if turn_sequence is not None else meta.get("turn_sequence") + + try: + ts = float(ts_raw) if ts_raw is not None else time.time() + except (TypeError, ValueError): + ts = time.time() + if not math.isfinite(ts): + ts = time.time() + + try: + seq = int(seq_raw) if seq_raw is not None else int(fallback_sequence) + except (TypeError, ValueError): + seq = int(fallback_sequence) + + anchors = extract_state_anchors(content=content, path=path, metadata=meta) + return TemporalMetadata(timestamp=ts, turn_sequence=seq, state_anchors=tuple(sorted(anchors))) + + +def temporal_sort(records: Iterable[TemporalMemoryRecord]) -> list[TemporalMemoryRecord]: + """Sort records in strict chronological order.""" + return sorted(records, key=lambda r: (r.turn_sequence, r.timestamp, str(r.key))) + + +def deduplicate_latest_state(records: Iterable[TemporalMemoryRecord]) -> list[TemporalMemoryRecord]: + """Drop older fragments that update the same state anchor as a newer fragment.""" + ordered = temporal_sort(records) + seen: set[str] = set() + kept: list[TemporalMemoryRecord] = [] + for record in reversed(ordered): + anchors = set(record.state_anchors) + if anchors and seen.intersection(anchors): + continue + kept.append(record) + seen.update(anchors) + return temporal_sort(kept) + + +def compact_timeline( + records: Sequence[TemporalMemoryRecord], + token_ceiling: int, + *, + foundation_states: int = 1, + recent_states: int = 3, +) -> list[TemporalMemoryRecord]: + """Fill context backwards while preserving oldest foundation and newest states.""" + if token_ceiling <= 0: + raise ValueError("token_ceiling must be positive") + ordered = temporal_sort(records) + if not ordered: + return [] + total = sum(max(0, r.token_estimate) for r in ordered) + if total <= token_ceiling: + return ordered + + selected: dict[int | str, TemporalMemoryRecord] = {} + used = 0 + + def try_add(record: TemporalMemoryRecord, *, force: bool = False) -> None: + nonlocal used + if record.key in selected: + return + tokens = max(0, record.token_estimate) + if force or used + tokens <= token_ceiling or not selected: + selected[record.key] = record + used += tokens + + for record in ordered[:foundation_states]: + try_add(record, force=True) + for record in reversed(ordered[-recent_states:] if recent_states else []): + try_add(record, force=True) + + for record in reversed(ordered): + try_add(record) + + return temporal_sort(selected.values()) + + +def extract_state_anchors(content: str = "", path: str = "", metadata: Optional[dict] = None) -> set[str]: + """Extract state-tracking anchors from metadata, paths, variables, and update text.""" + anchors: set[str] = set() + if path: + anchors.add(f"path:{path.lower()}") + meta = metadata or {} + for key in ("state_anchor", "state_anchors", "variable", "variable_name", "file_path", "path"): + value = meta.get(key) + if value is None: + continue + if isinstance(value, (list, tuple, set)): + values = value + else: + values = [value] + for item in values: + text = str(item).strip().lower() + if text: + anchors.add(f"state:{text}") + + for pattern in _STATE_PATTERNS: + for match in pattern.findall(content or ""): + anchor = str(match).strip().lower().rstrip(".,;:)") + if len(anchor) >= 2: + anchors.add(f"state:{anchor}") + return anchors + + +def _as_float_matrix(values: np.ndarray) -> np.ndarray: + matrix = np.asarray(values, dtype=np.float32) + if matrix.ndim == 1: + matrix = matrix.reshape(1, -1) + if matrix.ndim != 2: + raise ValueError("expected a 1D or 2D embedding array") + return matrix + + +def _as_uint8_matrix(values: np.ndarray) -> np.ndarray: + matrix = np.asarray(values, dtype=np.uint8) + if matrix.ndim == 1: + matrix = matrix.reshape(1, -1) + if matrix.ndim != 2: + raise ValueError("expected a 1D or 2D quantized array") + return matrix + + +def _as_uint8_vector(values: np.ndarray) -> np.ndarray: + vector = np.asarray(values, dtype=np.uint8) + if vector.ndim == 2 and vector.shape[0] == 1: + vector = vector[0] + if vector.ndim != 1: + raise ValueError("expected a single quantized vector") + return vector diff --git a/contextforge/index.py b/contextforge/index.py index 4f81ddc..5b5d386 100644 --- a/contextforge/index.py +++ b/contextforge/index.py @@ -1,29 +1,40 @@ -"""MemoryIndex — in-memory inverted index for O(1) keyword → node lookup.""" +"""MemoryIndex — hyper-dense local memory index with SQ8 semantic retrieval.""" from __future__ import annotations import math +import json from collections import defaultdict from dataclasses import dataclass, field from typing import Optional +from .hyperdense import ( + HyperDenseMemoryConfig, + SQ8VectorIndex, + normalize_temporal_metadata, +) from .utils import extract_keywords +_SEMANTIC_SCORE_WEIGHT = 10.0 + @dataclass class IndexEntry: - """A single entry in the inverted index.""" + """A single entry in the compatibility keyword index.""" node_id: int path: str title: str category: str term_frequency: float = 0.0 + timestamp: float = 0.0 + turn_sequence: int = 0 + state_anchors: tuple[str, ...] = field(default_factory=tuple) @dataclass class SearchResult: - """A scored search result.""" + """A scored semantic search result with strict temporal metadata.""" node_id: int path: str @@ -31,16 +42,22 @@ class SearchResult: category: str score: float matched_terms: list[str] = field(default_factory=list) + timestamp: float = 0.0 + turn_sequence: int = 0 + state_anchors: tuple[str, ...] = field(default_factory=tuple) class MemoryIndex: - """In-memory inverted index built from KnowledgeTree nodes. + """In-memory SQ8 semantic index built from KnowledgeTree nodes. - Provides BM25-scored keyword search with O(1) term lookup. + Provides quantized semantic top-k retrieval with a small keyword + compatibility layer for direct lookup and lexical tie-breaking. Rebuilt from the database on startup; no persistence of its own. """ - def __init__(self) -> None: + def __init__(self, config: Optional[HyperDenseMemoryConfig] = None) -> None: + self.config = config or HyperDenseMemoryConfig() + self._semantic = SQ8VectorIndex(self.config) # term → list of IndexEntry self._index: dict[str, list[IndexEntry]] = defaultdict(list) # node_id → set of terms @@ -54,6 +71,8 @@ def __init__(self) -> None: # BM25 parameters self._k1: float = 1.5 self._b: float = 0.75 + # node_id → document payload for SQ8 rebuilds + self._documents: dict[int, tuple[str, str, str, str, float, int, tuple[str, ...]]] = {} @property def num_docs(self) -> int: @@ -63,6 +82,11 @@ def num_docs(self) -> int: def num_terms(self) -> int: return len(self._index) + @property + def quantized_bytes(self) -> int: + """RAM used by the SQ8 vector matrix.""" + return self._semantic.byte_size + def add_document( self, node_id: int, @@ -70,8 +94,30 @@ def add_document( title: str, category: str, content: str, + metadata: Optional[dict] = None, + timestamp: Optional[float] = None, + turn_sequence: Optional[int] = None, ) -> None: """Index a document's content.""" + temporal = normalize_temporal_metadata( + metadata, + timestamp=timestamp, + turn_sequence=turn_sequence, + fallback_sequence=node_id, + content=content, + path=path, + ) + self.remove_document(node_id) + self._documents[node_id] = ( + path, + title, + category, + content, + temporal.timestamp, + temporal.turn_sequence, + temporal.state_anchors, + ) + keywords = extract_keywords(content, top_k=50) title_keywords = extract_keywords(title, top_k=10) all_terms = set(keywords) | set(title_keywords) | {category.lower()} @@ -97,26 +143,34 @@ def add_document( title=title, category=category, term_frequency=count / doc_length if doc_length > 0 else 0, + timestamp=temporal.timestamp, + turn_sequence=temporal.turn_sequence, + state_anchors=temporal.state_anchors, ) self._index[term].append(entry) self._num_docs += 1 total_length = sum(self._doc_lengths.values()) self._avg_dl = total_length / self._num_docs if self._num_docs > 0 else 0 + self._rebuild_semantic() def remove_document(self, node_id: int) -> None: """Remove a document from the index.""" + existed = node_id in self._documents or node_id in self._node_terms terms = self._node_terms.pop(node_id, set()) for term in terms: self._index[term] = [e for e in self._index[term] if e.node_id != node_id] if not self._index[term]: del self._index[term] self._doc_lengths.pop(node_id, None) - self._num_docs = max(0, self._num_docs - 1) + self._documents.pop(node_id, None) + if existed: + self._num_docs = max(0, self._num_docs - 1) if self._num_docs > 0: self._avg_dl = sum(self._doc_lengths.values()) / self._num_docs else: self._avg_dl = 0 + self._rebuild_semantic() def search( self, @@ -124,17 +178,73 @@ def search( top_k: int = 10, category: Optional[str] = None, ) -> list[SearchResult]: - """Search the index using BM25 scoring. + """Search the SQ8 semantic index with lexical tie-breaking. - Returns the top-k results sorted by relevance score. + Returns top-k candidates sorted by semantic relevance. Callers that + assemble context should then run a chronological pass. """ terms = extract_keywords(query, top_k=15) - if not terms: + if not query.strip() or not self._documents: + return [] + + scores, matched, entry_info = self._keyword_scores(terms, category) + if terms and not scores: + return [] + + semantic_limit = max(top_k, top_k * self.config.search_multiplier) + semantic_hits = self._semantic.search(query, top_k=semantic_limit) + if not semantic_hits and not scores: return [] + combined: dict[int, float] = defaultdict(float) + for raw_id, semantic_score in semantic_hits: + node_id = int(raw_id) + doc = self._documents.get(node_id) + if not doc: + continue + if category and doc[2] != category: + continue + combined[node_id] += semantic_score * _SEMANTIC_SCORE_WEIGHT + + for node_id, keyword_score in scores.items(): + combined[node_id] += keyword_score + + if not combined: + return [] + + results: list[SearchResult] = [] + for node_id, score in sorted(combined.items(), key=lambda x: -x[1])[:top_k]: + doc = self._documents.get(node_id) + if not doc: + continue + path, title, doc_category, _content, timestamp, turn_sequence, anchors = doc + info = entry_info.get(node_id) + results.append( + SearchResult( + node_id=node_id, + path=info.path if info else path, + title=info.title if info else title, + category=info.category if info else doc_category, + score=score, + matched_terms=matched.get(node_id, []), + timestamp=timestamp, + turn_sequence=turn_sequence, + state_anchors=anchors, + ) + ) + + return results + + def _keyword_scores( + self, + terms: list[str], + category: Optional[str], + ) -> tuple[dict[int, float], dict[int, list[str]], dict[int, IndexEntry]]: scores: dict[int, float] = defaultdict(float) matched: dict[int, list[str]] = defaultdict(list) entry_info: dict[int, IndexEntry] = {} + if not terms: + return scores, matched, entry_info for term in terms: entries = self._index.get(term, []) @@ -160,21 +270,7 @@ def search( matched[entry.node_id].append(term) entry_info[entry.node_id] = entry - results = [] - for node_id, score in sorted(scores.items(), key=lambda x: -x[1])[:top_k]: - info = entry_info[node_id] - results.append( - SearchResult( - node_id=node_id, - path=info.path, - title=info.title, - category=info.category, - score=score, - matched_terms=matched[node_id], - ) - ) - - return results + return scores, matched, entry_info def lookup(self, term: str) -> list[IndexEntry]: """Direct O(1) lookup of a single term.""" @@ -187,15 +283,20 @@ def build_from_tree(self, tree) -> int: """ self.clear() rows = tree.conn.execute( - "SELECT id, path, title, content, category FROM knowledge_nodes" + "SELECT id, path, title, content, category, metadata_json FROM knowledge_nodes ORDER BY id" ).fetchall() for row in rows: + try: + metadata = json.loads(row[5] or "{}") + except json.JSONDecodeError: + metadata = {} self.add_document( node_id=row[0], path=row[1], title=row[2], category=row[4], content=row[3], + metadata=metadata, ) return len(rows) @@ -204,5 +305,14 @@ def clear(self) -> None: self._index.clear() self._node_terms.clear() self._doc_lengths.clear() + self._documents.clear() self._num_docs = 0 self._avg_dl = 0.0 + self._semantic.clear() + + def _rebuild_semantic(self) -> None: + documents = [ + (node_id, f"{path}\n{title}\n{category}\n{content}") + for node_id, (path, title, category, content, *_rest) in self._documents.items() + ] + self._semantic.build(documents) diff --git a/contextforge/infinite_context.py b/contextforge/infinite_context.py index b6eacde..1f3076a 100644 --- a/contextforge/infinite_context.py +++ b/contextforge/infinite_context.py @@ -129,9 +129,19 @@ async def query( """ budget = max_knowledge_tokens or self._dynamic_budget() - # 1–3: Load a working set via the tree - keywords = extract_keywords(message, top_k=15) - working_set = self.tree.get_working_set(keywords, max_tokens=budget) + # 1–3: Load a temporal, quantized semantic working set via the loader + original_loader_budget = self.loader.max_context_tokens + self.loader.max_context_tokens = budget + try: + loaded = self.loader.load(message, conversation_context=self._compacted_history) + finally: + self.loader.max_context_tokens = original_loader_budget + working_set = WorkingSet( + content=loaded.system_prefix, + node_paths=loaded.branch_paths, + total_tokens=loaded.total_tokens, + node_ids=[], + ) self._track_branches(working_set) # 4: Build messages @@ -265,12 +275,13 @@ def get_stats(self) -> InfiniteContextStats: active = self._current_active_tokens() disk = self._total_disk_tokens() total = disk if disk > 0 else 1 + compression_ratio = min(1.0, active / total) return InfiniteContextStats( active_tokens=active, disk_tokens=disk, permanent_tokens=self._permanent_tokens, - compression_ratio=active / total, + compression_ratio=compression_ratio, recycles=self._recycles, peak_active=self._peak_active, ) diff --git a/contextforge/layer.py b/contextforge/layer.py index 43c76f6..f42df92 100644 --- a/contextforge/layer.py +++ b/contextforge/layer.py @@ -352,6 +352,7 @@ def stats(self) -> dict: "knowledge_nodes": self._tree.total_nodes(), "index_terms": self._index.num_terms, "index_docs": self._index.num_docs, + "quantized_index_bytes": self._index.quantized_bytes, "sessions": len(self._sessions.list_sessions()), "cache": self._loader.cache_stats(), "infinite_context": { diff --git a/contextforge/loader.py b/contextforge/loader.py index 27a4382..0cf5d3c 100644 --- a/contextforge/loader.py +++ b/contextforge/loader.py @@ -6,6 +6,12 @@ from dataclasses import dataclass, field from typing import Optional +from .hyperdense import ( + TemporalMemoryRecord, + compact_timeline, + deduplicate_latest_state, + temporal_sort, +) from .index import MemoryIndex, SearchResult from .tree import KnowledgeTree from .utils import estimate_tokens, extract_keywords @@ -13,6 +19,7 @@ _WIKI_AUTO_BUDGET_RATIOS = (0.50, 0.75, 1.0) +_DATED_SOURCE_REF_BOOST = 35.0 _DATE_RE = re.compile(r"\b20\d{2}-\d{2}-\d{2}\b") _DAY_RE = re.compile(r"\bday[-_\s]*(\d{1,4})\b", re.IGNORECASE) _ENTITY_RE = re.compile(r"\b[A-Z][a-zA-Z]+(?:\s+[A-Z][a-zA-Z]+){0,2}\b") @@ -134,6 +141,7 @@ def load( top_k=20, category=category_hint, ) + results = self._temporal_chrono_pass(results) if not results: return LoadedContext( @@ -164,6 +172,7 @@ def load_multi( """ combined = f"{query} {conversation_context}".strip() results = self._index.search(query=combined, top_k=30) + results = self._temporal_chrono_pass(results) if not results: return [] @@ -175,7 +184,7 @@ def load_multi( contexts = [] for category, cat_results in by_category.items(): - selected = self._select_nodes(cat_results) + selected = self._select_nodes(self._temporal_chrono_pass(cat_results)) if selected: ctx = self._assemble(selected, label=category) contexts.append(ctx) @@ -384,7 +393,7 @@ def _prioritize_wiki_results( if path.startswith("wiki/timeline/") and dates: score += 20.0 if path.startswith("wiki/sources/") and dates: - score += 10.0 + score += _DATED_SOURCE_REF_BOOST if path.startswith("wiki/status/") and statuses: score += 2.0 if path.startswith("wiki/threads/") and (entities or terms): @@ -455,29 +464,33 @@ def _wiki_context_is_sufficient( return False def _select_nodes(self, results: list[SearchResult]) -> list[SearchResult]: - """Select nodes within the token budget, boosting cached branches.""" - budget = self._max_tokens - selected: list[SearchResult] = [] + """Select a chronological timeline within the hard token budget.""" + if not results: + return [] # Boost cached entries - scored = [] + candidates: list[SearchResult] = [] for r in results: - boost = 1.0 if r.path in self._cache: - boost = 1.5 self._cache[r.path].hits += 1 - scored.append((r, r.score * boost)) + candidates.append(r) - scored.sort(key=lambda x: -x[1]) + by_id = {r.node_id: r for r in candidates} + records: list[TemporalMemoryRecord] = [] - for r, _ in scored: + for r in candidates: node = self._tree.get(r.path) if not node: continue - if budget - node.token_estimate < 0 and selected: - continue - budget -= node.token_estimate - selected.append(r) + records.append( + TemporalMemoryRecord( + key=r.node_id, + token_estimate=node.token_estimate or estimate_tokens(node.content), + timestamp=r.timestamp, + turn_sequence=r.turn_sequence, + state_anchors=r.state_anchors, + ) + ) # Cache this branch if r.path not in self._cache: @@ -487,7 +500,36 @@ def _select_nodes(self, results: list[SearchResult]) -> list[SearchResult]: tokens=node.token_estimate, ) - return selected + compacted = compact_timeline( + records, + max(1, self._max_tokens), + foundation_states=self._index.config.foundation_states, + recent_states=self._index.config.recent_states, + ) + return [by_id[int(record.key)] for record in compacted if int(record.key) in by_id] + + def _temporal_chrono_pass(self, results: list[SearchResult]) -> list[SearchResult]: + """Sort semantic candidates chronologically and remove stale state updates.""" + if not results: + return [] + records = [ + TemporalMemoryRecord( + key=result.node_id, + token_estimate=0, + timestamp=result.timestamp, + turn_sequence=result.turn_sequence, + state_anchors=result.state_anchors, + ) + for result in results + ] + if self._index.config.enable_state_deduplication and not all( + result.category == WIKI_CATEGORY for result in results + ): + records = deduplicate_latest_state(records) + else: + records = temporal_sort(records) + by_id = {result.node_id: result for result in results} + return [by_id[int(record.key)] for record in records if int(record.key) in by_id] def _select_results_with_budget( self, diff --git a/contextforge/tree.py b/contextforge/tree.py index 925af80..7e1cf1b 100644 --- a/contextforge/tree.py +++ b/contextforge/tree.py @@ -3,7 +3,9 @@ from __future__ import annotations import os +import json import sqlite3 +import time from dataclasses import dataclass, field from pathlib import Path from typing import Optional @@ -122,8 +124,6 @@ def add( If the path already exists, updates the content. """ - import json - parent_id = None if parent_path: row = self.conn.execute( @@ -132,7 +132,10 @@ def add( if row: parent_id = row[0] - meta_json = json.dumps(metadata or {}) + meta = dict(metadata or {}) + meta.setdefault("timestamp", time.time()) + meta.setdefault("turn_sequence", self._next_turn_sequence()) + meta_json = json.dumps(meta) tok_est = estimate_tokens(content) existing = self.conn.execute( @@ -178,13 +181,11 @@ def add( category=category, parent_id=parent_id, token_estimate=tok_est, - metadata=metadata or {}, + metadata=meta, ) def get(self, path: str) -> Optional[KnowledgeNode]: """Retrieve a single node by path.""" - import json - row = self.conn.execute( "SELECT id, path, title, content, category, parent_id, token_estimate, metadata_json " "FROM knowledge_nodes WHERE path = ?", @@ -233,8 +234,6 @@ def get_children(self, path: str) -> list[KnowledgeNode]: parent = self.get(path) if not parent: return [] - import json - rows = self.conn.execute( "SELECT id, path, title, content, category, parent_id, token_estimate, metadata_json " "FROM knowledge_nodes WHERE parent_id = ? ORDER BY path", @@ -258,8 +257,6 @@ def get_branch(self, path: str) -> list[KnowledgeNode]: queue = [root.id] while queue: pid = queue.pop(0) - import json - rows = self.conn.execute( "SELECT id, path, title, content, category, parent_id, token_estimate, metadata_json " "FROM knowledge_nodes WHERE parent_id = ? ORDER BY path", @@ -316,6 +313,21 @@ def total_nodes(self) -> int: row = self.conn.execute("SELECT COUNT(*) FROM knowledge_nodes").fetchone() return row[0] if row else 0 + def _next_turn_sequence(self) -> int: + """Return the next monotonically increasing knowledge turn sequence.""" + rows = self.conn.execute("SELECT metadata_json FROM knowledge_nodes").fetchall() + max_sequence = -1 + for row in rows: + try: + meta = json.loads(row[0] or "{}") + except json.JSONDecodeError: + continue + try: + max_sequence = max(max_sequence, int(meta.get("turn_sequence", -1))) + except (TypeError, ValueError): + continue + return max_sequence + 1 + # -- bulk ingestion --------------------------------------------------- def ingest_directory( diff --git a/pyproject.toml b/pyproject.toml index 50ee26a..c3fc7b4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,7 @@ classifiers = [ dependencies = [ "httpx>=0.25", + "numpy>=1.26.4", ] [project.optional-dependencies] diff --git a/tests/test_hyperdense.py b/tests/test_hyperdense.py new file mode 100644 index 0000000..4f0e4a6 --- /dev/null +++ b/tests/test_hyperdense.py @@ -0,0 +1,107 @@ +"""Tests for SQ8 quantization and temporal memory ordering.""" + +from __future__ import annotations + +import numpy as np + +from contextforge.hyperdense import ( + HyperDenseMemoryConfig, + SQ8VectorIndex, + TemporalMemoryRecord, + compact_timeline, + deduplicate_latest_state, + dequantize, + quantize, + quantized_cosine, +) +from contextforge.index import MemoryIndex +from contextforge.loader import ProactiveLoader +from contextforge.tree import KnowledgeTree + + +def test_sq8_quantize_dequantize_and_cosine(): + embeddings = np.array( + [[0.0, 0.5, 1.0], [1.0, 0.25, -1.0]], + dtype=np.float32, + ) + q, bounds = quantize(embeddings) + assert q.dtype == np.uint8 + assert q.nbytes == embeddings.shape[0] * embeddings.shape[1] + restored = dequantize(q, bounds) + assert restored.shape == embeddings.shape + scores = quantized_cosine(q[0], q) + assert scores[0] >= scores[1] + + +def test_sq8_vector_index_searches_semantically(): + index = SQ8VectorIndex(HyperDenseMemoryConfig(dimensions=64)) + index.build([ + (1, "revenue profit financial quarter"), + (2, "python class function module"), + ]) + results = index.search("quarterly revenue", top_k=1) + assert results[0][0] == 1 + assert index.byte_size == 2 * 64 + + +def test_memory_index_wraps_temporal_metadata(): + index = MemoryIndex(HyperDenseMemoryConfig(dimensions=64)) + index.add_document( + 1, + "memory/day1", + "Day 1", + "memory", + "status = draft", + metadata={"timestamp": 10.0, "turn_sequence": 1}, + ) + result = index.search("status draft", top_k=1)[0] + assert result.timestamp == 10.0 + assert result.turn_sequence == 1 + assert "state:status" in result.state_anchors + + +def test_temporal_dedup_keeps_latest_state(): + records = [ + TemporalMemoryRecord(1, 10, 1.0, 1, ("state:status",)), + TemporalMemoryRecord(2, 10, 2.0, 2, ("state:status",)), + TemporalMemoryRecord(3, 10, 3.0, 3, ("state:other",)), + ] + deduped = deduplicate_latest_state(records) + assert [record.key for record in deduped] == [2, 3] + + +def test_compact_timeline_preserves_oldest_and_recent_states(): + records = [TemporalMemoryRecord(i, 10, float(i), i, ()) for i in range(6)] + compacted = compact_timeline(records, token_ceiling=30, foundation_states=1, recent_states=2) + keys = [record.key for record in compacted] + assert keys[0] == 0 + assert 4 in keys and 5 in keys + assert keys == sorted(keys) + + +def test_loader_chrono_pass_discards_stale_updates(): + tree = KnowledgeTree(":memory:") + tree.open() + try: + tree.add( + "memory/day1", + "Day 1", + "Project Condor status = blocked pending diligence.", + category="memory", + metadata={"timestamp": 1.0, "turn_sequence": 1, "state_anchor": "condor_status"}, + ) + tree.add( + "memory/day2", + "Day 2", + "Project Condor status = approved after Jamie review.", + category="memory", + metadata={"timestamp": 2.0, "turn_sequence": 2, "state_anchor": "condor_status"}, + ) + idx = MemoryIndex(HyperDenseMemoryConfig(dimensions=64)) + idx.build_from_tree(tree) + loader = ProactiveLoader(tree, idx, max_context_tokens=1000) + loaded = loader.load("What is Project Condor status?") + assert "approved" in loaded.system_prefix + assert "blocked pending" not in loaded.system_prefix + finally: + tree.close()