From ea834bc7dc81434f7dbcfd85900a1f86332a46ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Wed, 1 Apr 2026 12:44:40 +0200 Subject: [PATCH] breaking: remove BFS and MPFP graph retrieval strategies (#767) Remove the BFS spreading activation and MPFP (Multi-Path Fact Propagation) graph retrieval strategies, leaving link_expansion as the sole graph retrieval algorithm. Rename MPFPTimings to GraphRetrievalTimings and mpfp_timings field to graph_timings since the timing struct is used by LinkExpansionRetriever. Deleted: - hindsight-api-slim/hindsight_api/engine/search/mpfp_retrieval.py - hindsight-api-slim/tests/test_mpfp_retrieval.py Removed config: HINDSIGHT_API_MPFP_TOP_K_NEIGHBORS --- CLAUDE.md | 4 +- ...c4d5e6_add_memory_links_composite_index.py | 4 +- hindsight-api-slim/hindsight_api/config.py | 6 +- .../hindsight_api/engine/memory_engine.py | 34 +- .../hindsight_api/engine/search/__init__.py | 7 +- .../engine/search/graph_retrieval.py | 229 +---- .../engine/search/link_expansion_retrieval.py | 6 +- .../engine/search/mpfp_retrieval.py | 702 --------------- .../hindsight_api/engine/search/retrieval.py | 39 +- .../hindsight_api/engine/search/types.py | 4 +- .../tests/test_mpfp_retrieval.py | 819 ------------------ .../docs/developer/configuration.md | 9 +- .../version-0.3/developer/configuration.md | 8 +- .../version-0.4/developer/configuration.md | 9 +- .../references/developer/configuration.md | 9 +- 15 files changed, 65 insertions(+), 1824 deletions(-) delete mode 100644 hindsight-api-slim/hindsight_api/engine/search/mpfp_retrieval.py delete mode 100644 hindsight-api-slim/tests/test_mpfp_retrieval.py diff --git a/CLAUDE.md b/CLAUDE.md index 10dc2265..146f8f27 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -101,8 +101,8 @@ cd hindsight-control-plane && npm run dev **search/**: Multi-strategy retrieval - `retrieval.py`: Main retrieval orchestrator -- `graph_retrieval.py`: Entity/relationship graph traversal -- `mpfp_retrieval.py`: Multi-Path Fact Propagation retrieval +- `graph_retrieval.py`: Graph retrieval abstract base class +- `link_expansion_retrieval.py`: Link expansion graph retrieval - `fusion.py`: Reciprocal rank fusion for combining results - `reranking.py`: Cross-encoder reranking diff --git a/hindsight-api-slim/hindsight_api/alembic/versions/f1a2b3c4d5e6_add_memory_links_composite_index.py b/hindsight-api-slim/hindsight_api/alembic/versions/f1a2b3c4d5e6_add_memory_links_composite_index.py index 09100aeb..4c756a73 100644 --- a/hindsight-api-slim/hindsight_api/alembic/versions/f1a2b3c4d5e6_add_memory_links_composite_index.py +++ b/hindsight-api-slim/hindsight_api/alembic/versions/f1a2b3c4d5e6_add_memory_links_composite_index.py @@ -5,7 +5,7 @@ Revises: e0a1b2c3d4e5 Create Date: 2025-01-12 Add composite index on memory_links (from_unit_id, link_type, weight DESC) -to optimize MPFP graph traversal queries that need top-k edges per type. +to optimize graph traversal queries that need top-k edges per type. """ from collections.abc import Sequence @@ -26,7 +26,7 @@ def _get_schema_prefix() -> str: def upgrade() -> None: - """Add composite index for efficient MPFP edge loading.""" + """Add composite index for efficient graph retrieval edge loading.""" schema = _get_schema_prefix() # Create composite index for efficient top-k per (from_node, link_type) queries # This enables LATERAL joins to use index-only scans with early termination diff --git a/hindsight-api-slim/hindsight_api/config.py b/hindsight-api-slim/hindsight_api/config.py index 1f91e417..53dc6fa5 100644 --- a/hindsight-api-slim/hindsight_api/config.py +++ b/hindsight-api-slim/hindsight_api/config.py @@ -245,7 +245,6 @@ ENV_MCP_ENABLED_TOOLS = "HINDSIGHT_API_MCP_ENABLED_TOOLS" ENV_MCP_STATELESS = "HINDSIGHT_API_MCP_STATELESS" ENV_ENABLE_BANK_CONFIG_API = "HINDSIGHT_API_ENABLE_BANK_CONFIG_API" ENV_GRAPH_RETRIEVER = "HINDSIGHT_API_GRAPH_RETRIEVER" -ENV_MPFP_TOP_K_NEIGHBORS = "HINDSIGHT_API_MPFP_TOP_K_NEIGHBORS" ENV_RECALL_MAX_CONCURRENT = "HINDSIGHT_API_RECALL_MAX_CONCURRENT" ENV_RECALL_CONNECTION_BUDGET = "HINDSIGHT_API_RECALL_CONNECTION_BUDGET" ENV_RECALL_MAX_QUERY_TOKENS = "HINDSIGHT_API_RECALL_MAX_QUERY_TOKENS" @@ -451,8 +450,7 @@ DEFAULT_MCP_ENABLED = True DEFAULT_MCP_ENABLED_TOOLS: list[str] | None = None # None = all tools enabled DEFAULT_MCP_STATELESS = False # False = stateful (supports SSE/GET); True = stateless (POST-only) DEFAULT_ENABLE_BANK_CONFIG_API = True -DEFAULT_GRAPH_RETRIEVER = "link_expansion" # Options: "link_expansion", "mpfp", "bfs" -DEFAULT_MPFP_TOP_K_NEIGHBORS = 20 # Fan-out limit per node in MPFP graph traversal +DEFAULT_GRAPH_RETRIEVER = "link_expansion" DEFAULT_RECALL_MAX_CONCURRENT = 32 # Max concurrent recall operations per worker DEFAULT_RECALL_CONNECTION_BUDGET = 4 # Max concurrent DB connections per recall operation DEFAULT_RECALL_MAX_QUERY_TOKENS = 500 # Maximum tokens allowed in recall query @@ -743,7 +741,6 @@ class HindsightConfig: # Recall graph_retriever: str - mpfp_top_k_neighbors: int recall_max_concurrent: int recall_connection_budget: int recall_max_query_tokens: int @@ -1218,7 +1215,6 @@ class HindsightConfig: == "true", # Recall graph_retriever=os.getenv(ENV_GRAPH_RETRIEVER, DEFAULT_GRAPH_RETRIEVER), - mpfp_top_k_neighbors=int(os.getenv(ENV_MPFP_TOP_K_NEIGHBORS, str(DEFAULT_MPFP_TOP_K_NEIGHBORS))), recall_max_concurrent=int(os.getenv(ENV_RECALL_MAX_CONCURRENT, str(DEFAULT_RECALL_MAX_CONCURRENT))), recall_connection_budget=int( os.getenv(ENV_RECALL_CONNECTION_BUDGET, str(DEFAULT_RECALL_CONNECTION_BUDGET)) diff --git a/hindsight-api-slim/hindsight_api/engine/memory_engine.py b/hindsight-api-slim/hindsight_api/engine/memory_engine.py index 16c6f18f..2d87176a 100644 --- a/hindsight-api-slim/hindsight_api/engine/memory_engine.py +++ b/hindsight-api-slim/hindsight_api/engine/memory_engine.py @@ -2832,7 +2832,7 @@ class MemoryEngine(MemoryEngineInterface): "temporal": 0.0, "temporal_extraction": 0.0, } - all_mpfp_timings = [] + all_graph_timings = [] detected_temporal_constraint = None max_conn_wait = multi_result.max_conn_wait @@ -2894,25 +2894,25 @@ class MemoryEngine(MemoryEngineInterface): ) # Log graph retriever timing breakdown if available - if all_mpfp_timings: + if all_graph_timings: retriever_name = get_default_graph_retriever().name.upper() - mpfp_total = all_mpfp_timings[0] # Take first fact type's timing as representative - mpfp_parts = [ - f"db_queries={mpfp_total.db_queries}", - f"edge_load={mpfp_total.edge_load_time:.3f}s", - f"edges={mpfp_total.edge_count}", - f"patterns={mpfp_total.pattern_count}", + graph_total = all_graph_timings[0] # Take first fact type's timing as representative + graph_parts = [ + f"db_queries={graph_total.db_queries}", + f"edge_load={graph_total.edge_load_time:.3f}s", + f"edges={graph_total.edge_count}", + f"patterns={graph_total.pattern_count}", ] - if mpfp_total.seeds_time > 0.01: - mpfp_parts.append(f"seeds={mpfp_total.seeds_time:.3f}s") - if mpfp_total.fusion > 0.001: - mpfp_parts.append(f"fusion={mpfp_total.fusion:.3f}s") - if mpfp_total.fetch > 0.001: - mpfp_parts.append(f"fetch={mpfp_total.fetch:.3f}s") - log_buffer.append(f" [{retriever_name}] {', '.join(mpfp_parts)}") + if graph_total.seeds_time > 0.01: + graph_parts.append(f"seeds={graph_total.seeds_time:.3f}s") + if graph_total.fusion > 0.001: + graph_parts.append(f"fusion={graph_total.fusion:.3f}s") + if graph_total.fetch > 0.001: + graph_parts.append(f"fetch={graph_total.fetch:.3f}s") + log_buffer.append(f" [{retriever_name}] {', '.join(graph_parts)}") # Log detailed hop timing for debugging slow queries - if mpfp_total.hop_details: - for hd in mpfp_total.hop_details: + if graph_total.hop_details: + for hd in graph_total.hop_details: log_buffer.append( f" hop{hd['hop']}: exec={hd.get('exec_time', 0) * 1000:.0f}ms, " f"uncached={hd.get('uncached_after_filter', 0)}, " diff --git a/hindsight-api-slim/hindsight_api/engine/search/__init__.py b/hindsight-api-slim/hindsight_api/engine/search/__init__.py index 37734ccd..e17580bb 100644 --- a/hindsight-api-slim/hindsight_api/engine/search/__init__.py +++ b/hindsight-api-slim/hindsight_api/engine/search/__init__.py @@ -3,12 +3,11 @@ Search module for memory retrieval. Provides modular search architecture: - Retrieval: 4-way parallel (semantic + BM25 + graph + temporal) -- Graph retrieval: Pluggable strategies (BFS, PPR) +- Graph retrieval: Link expansion strategy - Reranking: Pluggable strategies (heuristic, cross-encoder) """ -from .graph_retrieval import BFSGraphRetriever, GraphRetriever -from .mpfp_retrieval import MPFPGraphRetriever +from .graph_retrieval import GraphRetriever from .reranking import CrossEncoderReranker from .retrieval import ( ParallelRetrievalResult, @@ -21,7 +20,5 @@ __all__ = [ "set_default_graph_retriever", "ParallelRetrievalResult", "GraphRetriever", - "BFSGraphRetriever", - "MPFPGraphRetriever", "CrossEncoderReranker", ] diff --git a/hindsight-api-slim/hindsight_api/engine/search/graph_retrieval.py b/hindsight-api-slim/hindsight_api/engine/search/graph_retrieval.py index 7cdde929..377b83cb 100644 --- a/hindsight-api-slim/hindsight_api/engine/search/graph_retrieval.py +++ b/hindsight-api-slim/hindsight_api/engine/search/graph_retrieval.py @@ -2,17 +2,15 @@ Graph retrieval strategies for memory recall. This module provides an abstraction for graph-based memory retrieval, -allowing different algorithms (BFS spreading activation, PPR, etc.) to be -swapped without changing the rest of the recall pipeline. +allowing different algorithms to be swapped without changing the rest +of the recall pipeline. """ import logging from abc import ABC, abstractmethod -from ..db_utils import acquire_with_retry -from ..memory_engine import fq_table -from .tags import TagGroup, TagsMatch, filter_results_by_tag_groups, filter_results_by_tags -from .types import MPFPTimings, RetrievalResult +from .tags import TagGroup, TagsMatch +from .types import GraphRetrievalTimings, RetrievalResult logger = logging.getLogger(__name__) @@ -29,7 +27,7 @@ class GraphRetriever(ABC): @property @abstractmethod def name(self) -> str: - """Return identifier for this retrieval strategy (e.g., 'bfs', 'mpfp').""" + """Return identifier for this retrieval strategy (e.g., 'link_expansion').""" pass @abstractmethod @@ -47,7 +45,7 @@ class GraphRetriever(ABC): tags: list[str] | None = None, # Visibility scope tags for filtering tags_match: TagsMatch = "any", # How to match tags: 'any' (OR) or 'all' (AND) tag_groups: list[TagGroup] | None = None, # Compound boolean tag filter groups - ) -> tuple[list[RetrievalResult], MPFPTimings | None]: + ) -> tuple[list[RetrievalResult], GraphRetrievalTimings | None]: """ Retrieve relevant facts via graph traversal. @@ -60,223 +58,10 @@ class GraphRetriever(ABC): query_text: Original query text (optional, for some strategies) semantic_seeds: Pre-computed semantic entry points (from semantic retrieval) temporal_seeds: Pre-computed temporal entry points (from temporal retrieval) - adjacency: Pre-loaded typed adjacency graph (optional, for MPFP) + adjacency: Pre-loaded typed adjacency graph (optional) tags: Optional list of tags for visibility filtering (OR matching) Returns: Tuple of (List of RetrievalResult with activation scores, optional timing info) """ pass - - -class BFSGraphRetriever(GraphRetriever): - """ - Graph retrieval using BFS-style spreading activation. - - Starting from semantic entry points, spreads activation through - the memory graph (entity, temporal, causal links) using breadth-first - traversal with decaying activation. - - This is the original Hindsight graph retrieval algorithm. - """ - - def __init__( - self, - entry_point_limit: int = 5, - entry_point_threshold: float = 0.5, - activation_decay: float = 0.8, - min_activation: float = 0.1, - batch_size: int = 20, - ): - """ - Initialize BFS graph retriever. - - Args: - entry_point_limit: Maximum number of entry points to start from - entry_point_threshold: Minimum semantic similarity for entry points - activation_decay: Decay factor per hop (activation *= decay) - min_activation: Minimum activation to continue spreading - batch_size: Number of nodes to process per batch (for neighbor fetching) - """ - self.entry_point_limit = entry_point_limit - self.entry_point_threshold = entry_point_threshold - self.activation_decay = activation_decay - self.min_activation = min_activation - self.batch_size = batch_size - - @property - def name(self) -> str: - return "bfs" - - async def retrieve( - self, - pool, - query_embedding_str: str, - bank_id: str, - fact_type: str, - budget: int, - query_text: str | None = None, - semantic_seeds: list[RetrievalResult] | None = None, - temporal_seeds: list[RetrievalResult] | None = None, - adjacency=None, # Not used by BFS - tags: list[str] | None = None, - tags_match: TagsMatch = "any", - tag_groups: list[TagGroup] | None = None, - ) -> tuple[list[RetrievalResult], MPFPTimings | None]: - """ - Retrieve facts using BFS spreading activation. - - Algorithm: - 1. Find entry points (top semantic matches above threshold) - 2. BFS traversal: visit neighbors, propagate decaying activation - 3. Boost causal links (causes, enables, prevents) - 4. Return visited nodes up to budget - - Note: BFS finds its own entry points via embedding search. - The semantic_seeds, temporal_seeds, and adjacency parameters are accepted - for interface compatibility but not used. - """ - async with acquire_with_retry(pool) as conn: - results = await self._retrieve_with_conn( - conn, - query_embedding_str, - bank_id, - fact_type, - budget, - tags=tags, - tags_match=tags_match, - tag_groups=tag_groups, - ) - return results, None - - async def _retrieve_with_conn( - self, - conn, - query_embedding_str: str, - bank_id: str, - fact_type: str, - budget: int, - tags: list[str] | None = None, - tags_match: TagsMatch = "any", - tag_groups: list[TagGroup] | None = None, - ) -> list[RetrievalResult]: - """Internal implementation with connection.""" - from .tags import build_tag_groups_where_clause, build_tags_where_clause_simple - - tags_clause = build_tags_where_clause_simple(tags, 6, match=tags_match) - tag_groups_param_start = 6 + (1 if tags else 0) - groups_clause, groups_params, _ = build_tag_groups_where_clause(tag_groups, tag_groups_param_start) - params = [query_embedding_str, bank_id, fact_type, self.entry_point_threshold, self.entry_point_limit] - if tags: - params.append(tags) - params.extend(groups_params) - - # Step 1: Find entry points - entry_points = await conn.fetch( - f""" - SELECT id, text, context, event_date, occurred_start, occurred_end, - mentioned_at, fact_type, document_id, chunk_id, tags, - 1 - (embedding <=> $1::vector) AS similarity - FROM {fq_table("memory_units")} - WHERE bank_id = $2 - AND embedding IS NOT NULL - AND fact_type = $3 - AND (1 - (embedding <=> $1::vector)) >= $4 - {tags_clause} - {groups_clause} - ORDER BY embedding <=> $1::vector - LIMIT $5 - """, - *params, - ) - - if not entry_points: - logger.debug( - f"[BFS] No entry points found for fact_type={fact_type} (tags={tags}, tags_match={tags_match})" - ) - return [] - - logger.debug( - f"[BFS] Found {len(entry_points)} entry points for fact_type={fact_type} " - f"(tags={tags}, tags_match={tags_match})" - ) - - # Step 2: BFS spreading activation - visited = set() - results = [] - queue = [(RetrievalResult.from_db_row(dict(r)), r["similarity"]) for r in entry_points] - budget_remaining = budget - - while queue and budget_remaining > 0: - # Collect a batch of nodes to process - batch_nodes = [] - batch_activations = {} - - while queue and len(batch_nodes) < self.batch_size and budget_remaining > 0: - current, activation = queue.pop(0) - unit_id = current.id - - if unit_id not in visited: - visited.add(unit_id) - budget_remaining -= 1 - current.activation = activation - results.append(current) - batch_nodes.append(current.id) - batch_activations[unit_id] = activation - - # Batch fetch neighbors - if batch_nodes and budget_remaining > 0: - max_neighbors = len(batch_nodes) * 20 - neighbors = await conn.fetch( - f""" - SELECT mu.id, mu.text, mu.context, mu.occurred_start, mu.occurred_end, - mu.mentioned_at, mu.fact_type, - mu.document_id, mu.chunk_id, mu.tags, mu.metadata, - ml.weight, ml.link_type, ml.from_unit_id - FROM {fq_table("memory_links")} ml - JOIN {fq_table("memory_units")} mu ON ml.to_unit_id = mu.id - WHERE ml.from_unit_id = ANY($1::uuid[]) - AND ml.weight >= $2 - AND mu.fact_type = $3 - ORDER BY ml.weight DESC - LIMIT $4 - """, - batch_nodes, - self.min_activation, - fact_type, - max_neighbors, - ) - - for n in neighbors: - neighbor_id = str(n["id"]) - if neighbor_id not in visited: - parent_id = str(n["from_unit_id"]) - parent_activation = batch_activations.get(parent_id, 0.5) - - # Boost causal links - link_type = n["link_type"] - base_weight = n["weight"] - - if link_type in ("causes", "caused_by"): - causal_boost = 2.0 - elif link_type in ("enables", "prevents"): - causal_boost = 1.5 - else: - causal_boost = 1.0 - - effective_weight = base_weight * causal_boost - new_activation = parent_activation * effective_weight * self.activation_decay - - if new_activation > self.min_activation: - neighbor_result = RetrievalResult.from_db_row(dict(n)) - queue.append((neighbor_result, new_activation)) - - # Apply tags filtering (BFS may traverse into memories that don't match tags criteria) - if tags: - results = filter_results_by_tags(results, tags, match=tags_match) - - # Apply compound tag group filtering (post-traversal) - if tag_groups: - results = filter_results_by_tag_groups(results, tag_groups) - - return results diff --git a/hindsight-api-slim/hindsight_api/engine/search/link_expansion_retrieval.py b/hindsight-api-slim/hindsight_api/engine/search/link_expansion_retrieval.py index 5e4652ec..caa0aa0a 100644 --- a/hindsight-api-slim/hindsight_api/engine/search/link_expansion_retrieval.py +++ b/hindsight-api-slim/hindsight_api/engine/search/link_expansion_retrieval.py @@ -29,7 +29,7 @@ from ..db_utils import acquire_with_retry from ..memory_engine import fq_table from .graph_retrieval import GraphRetriever from .tags import TagGroup, TagsMatch, filter_results_by_tag_groups, filter_results_by_tags -from .types import MPFPTimings, RetrievalResult +from .types import GraphRetrievalTimings, RetrievalResult logger = logging.getLogger(__name__) @@ -116,7 +116,7 @@ class LinkExpansionRetriever(GraphRetriever): tags: list[str] | None = None, tags_match: TagsMatch = "any", tag_groups: list[TagGroup] | None = None, - ) -> tuple[list[RetrievalResult], MPFPTimings | None]: + ) -> tuple[list[RetrievalResult], GraphRetrievalTimings | None]: """ Retrieve facts by expanding links from seeds. @@ -136,7 +136,7 @@ class LinkExpansionRetriever(GraphRetriever): Tuple of (results, timings) """ start_time = time.time() - timings = MPFPTimings(fact_type=fact_type) + timings = GraphRetrievalTimings(fact_type=fact_type) async with acquire_with_retry(pool) as conn: # Find seeds if not provided diff --git a/hindsight-api-slim/hindsight_api/engine/search/mpfp_retrieval.py b/hindsight-api-slim/hindsight_api/engine/search/mpfp_retrieval.py deleted file mode 100644 index e8d938cd..00000000 --- a/hindsight-api-slim/hindsight_api/engine/search/mpfp_retrieval.py +++ /dev/null @@ -1,702 +0,0 @@ -""" -Meta-Path Forward Push (MPFP) graph retrieval. - -A sublinear graph traversal algorithm for memory retrieval over heterogeneous -graphs with multiple edge types (semantic, temporal, causal, entity). - -Combines meta-path patterns from HIN literature with Forward Push local -propagation from Approximate PPR. - -Key properties: -- Sublinear in graph size (threshold pruning bounds active nodes) -- Lazy edge loading: only loads edges for frontier nodes, not entire graph -- Predefined patterns capture different retrieval intents -- All patterns run in parallel, results fused via RRF -- No LLM in the loop during traversal -""" - -import asyncio -import logging -from collections import defaultdict -from dataclasses import dataclass, field - -from ..db_utils import acquire_with_retry -from ..memory_engine import fq_table -from .graph_retrieval import GraphRetriever -from .tags import TagGroup, TagsMatch -from .types import MPFPTimings, RetrievalResult - -logger = logging.getLogger(__name__) - - -# ----------------------------------------------------------------------------- -# Data Classes -# ----------------------------------------------------------------------------- - - -@dataclass -class EdgeTarget: - """A neighbor node with its edge weight.""" - - node_id: str - weight: float - - -@dataclass -class EdgeCache: - """ - Cache for lazily-loaded edges. - - Grows per-hop as edges are loaded for frontier nodes. - Shared across patterns to avoid redundant loads. - Loads ALL edge types at once to minimize DB queries. - Thread-safe via asyncio lock to prevent redundant concurrent loads. - """ - - # edge_type -> from_node_id -> list of EdgeTarget - graphs: dict[str, dict[str, list[EdgeTarget]]] = field(default_factory=dict) - # Track which nodes have been fully loaded (all edge types) - _fully_loaded: set[str] = field(default_factory=set) - # Timing stats - db_queries: int = 0 - edge_load_time: float = 0.0 - # Detailed hop timing for debugging - hop_details: list[dict] = field(default_factory=list) - # Lock to prevent redundant concurrent loads - _lock: asyncio.Lock = field(default_factory=asyncio.Lock) - - def get_neighbors(self, edge_type: str, node_id: str) -> list[EdgeTarget]: - """Get neighbors for a node via a specific edge type.""" - return self.graphs.get(edge_type, {}).get(node_id, []) - - def get_normalized_neighbors(self, edge_type: str, node_id: str, top_k: int) -> list[EdgeTarget]: - """Get top-k neighbors with weights normalized to sum to 1.""" - neighbors = self.get_neighbors(edge_type, node_id)[:top_k] - if not neighbors: - return [] - - total = sum(n.weight for n in neighbors) - if total == 0: - return [] - - return [EdgeTarget(node_id=n.node_id, weight=n.weight / total) for n in neighbors] - - def is_fully_loaded(self, node_id: str) -> bool: - """Check if all edges for this node have been loaded.""" - return node_id in self._fully_loaded - - def get_uncached(self, node_ids: list[str]) -> list[str]: - """Get node IDs that haven't been fully loaded yet.""" - return [n for n in node_ids if not self.is_fully_loaded(n)] - - def add_all_edges(self, edges_by_type: dict[str, dict[str, list[EdgeTarget]]], all_queried: list[str]): - """ - Add loaded edges to the cache (all edge types at once). - - Args: - edges_by_type: Dict mapping edge_type -> from_node_id -> list of EdgeTarget - all_queried: All node IDs that were queried (marks them as fully loaded) - """ - for edge_type, edges in edges_by_type.items(): - if edge_type not in self.graphs: - self.graphs[edge_type] = {} - for node_id, neighbors in edges.items(): - self.graphs[edge_type][node_id] = neighbors - - # Mark all queried nodes as fully loaded (even if they have no edges) - self._fully_loaded.update(all_queried) - - -@dataclass -class PatternResult: - """Result from a single pattern traversal.""" - - pattern: list[str] - scores: dict[str, float] # node_id -> accumulated mass - - -@dataclass -class MPFPConfig: - """Configuration for MPFP algorithm.""" - - alpha: float = 0.15 # teleport/keep probability - threshold: float = 1e-6 # mass pruning threshold (lower = explore more) - top_k_neighbors: int = 20 # fan-out limit per node - - # Patterns from semantic seeds - patterns_semantic: list[list[str]] = field( - default_factory=lambda: [ - ["semantic", "semantic"], # topic expansion - ["entity", "temporal"], # entity timeline - ["semantic", "causes"], # reasoning chains (forward) - ["semantic", "caused_by"], # reasoning chains (backward) - ["entity", "semantic"], # entity context - ] - ) - - # Patterns from temporal seeds - patterns_temporal: list[list[str]] = field( - default_factory=lambda: [ - ["temporal", "semantic"], # what was happening then - ["temporal", "entity"], # who was involved then - ] - ) - - -@dataclass -class SeedNode: - """An entry point node with its initial score.""" - - node_id: str - score: float # initial mass (e.g., similarity score) - - -# ----------------------------------------------------------------------------- -# Lazy Edge Loading -# ----------------------------------------------------------------------------- - - -async def load_all_edges_for_frontier( - pool, - node_ids: list[str], - top_k_per_type: int = 20, -) -> dict[str, dict[str, list[EdgeTarget]]]: - """ - Load top-k edges per (node, edge_type) for frontier nodes. - - Uses a LATERAL join to efficiently fetch only the top-k edges per type, - avoiding loading hundreds of entity edges when only 20 are needed. - - Requires composite index: (from_unit_id, link_type, weight DESC) - - Args: - pool: Database connection pool - node_ids: Frontier node IDs to load edges for - top_k_per_type: Max edges to load per (node, link_type) pair - - Returns: - Dict mapping edge_type -> from_node_id -> list of EdgeTarget - """ - if not node_ids: - return {} - - async with acquire_with_retry(pool) as conn: - # Use LATERAL join to get top-k per (from_node, link_type) - # This leverages the composite index for efficient early termination - rows = await conn.fetch( - f""" - WITH frontier(node_id) AS (SELECT unnest($1::uuid[])) - SELECT f.node_id as from_unit_id, lt.link_type, edges.to_unit_id, edges.weight - FROM frontier f - CROSS JOIN (VALUES ('semantic'), ('temporal'), ('entity'), ('causes'), ('caused_by')) AS lt(link_type) - CROSS JOIN LATERAL ( - SELECT ml.to_unit_id, ml.weight - FROM {fq_table("memory_links")} ml - WHERE ml.from_unit_id = f.node_id - AND ml.link_type = lt.link_type - AND ml.weight >= 0.1 - ORDER BY ml.weight DESC - LIMIT $2 - ) edges - """, - node_ids, - top_k_per_type, - ) - - # Group by edge_type -> from_node -> neighbors - result: dict[str, dict[str, list[EdgeTarget]]] = defaultdict(lambda: defaultdict(list)) - for row in rows: - edge_type = row["link_type"] - from_id = str(row["from_unit_id"]) - to_id = str(row["to_unit_id"]) - weight = row["weight"] - result[edge_type][from_id].append(EdgeTarget(node_id=to_id, weight=weight)) - - # Convert nested defaultdicts to regular dicts - return {edge_type: dict(edges) for edge_type, edges in result.items()} - - -# ----------------------------------------------------------------------------- -# Core Algorithm (Async with Lazy Loading) -# ----------------------------------------------------------------------------- - - -@dataclass -class PatternState: - """State for a pattern traversal between hops.""" - - pattern: list[str] - hop_index: int - scores: dict[str, float] - frontier: dict[str, float] - - -def _init_pattern_state(seeds: list[SeedNode], pattern: list[str]) -> PatternState: - """Initialize pattern state from seeds.""" - if not seeds: - return PatternState(pattern=pattern, hop_index=0, scores={}, frontier={}) - - total_seed_score = sum(s.score for s in seeds) - if total_seed_score == 0: - total_seed_score = len(seeds) - - frontier = {s.node_id: s.score / total_seed_score for s in seeds} - return PatternState(pattern=pattern, hop_index=0, scores={}, frontier=frontier) - - -def _execute_hop(state: PatternState, cache: EdgeCache, config: MPFPConfig) -> set[str]: - """ - Execute ONE hop of traversal, return frontier nodes for next hop. - - This is a pure function that uses cached edges (no DB access). - Returns set of uncached nodes needed for next hop. - """ - if state.hop_index >= len(state.pattern): - return set() - - edge_type = state.pattern[state.hop_index] - - # Collect active nodes above threshold - active_nodes = [node_id for node_id, mass in state.frontier.items() if mass >= config.threshold] - if not active_nodes: - state.frontier = {} - return set() - - # Propagate mass using cached edges - next_frontier: dict[str, float] = {} - uncached_for_next: set[str] = set() - - for node_id, mass in state.frontier.items(): - if mass < config.threshold: - continue - - # Keep α portion for this node - state.scores[node_id] = state.scores.get(node_id, 0) + config.alpha * mass - - # Push (1-α) to neighbors - push_mass = (1 - config.alpha) * mass - neighbors = cache.get_normalized_neighbors(edge_type, node_id, config.top_k_neighbors) - - for neighbor in neighbors: - next_frontier[neighbor.node_id] = next_frontier.get(neighbor.node_id, 0) + push_mass * neighbor.weight - # Track if we'll need edges for this node in the next hop - if not cache.is_fully_loaded(neighbor.node_id): - uncached_for_next.add(neighbor.node_id) - - state.frontier = next_frontier - state.hop_index += 1 - - return uncached_for_next - - -def _finalize_pattern(state: PatternState, config: MPFPConfig) -> PatternResult: - """Finalize pattern by adding remaining frontier mass to scores.""" - for node_id, mass in state.frontier.items(): - if mass >= config.threshold: - state.scores[node_id] = state.scores.get(node_id, 0) + mass - - return PatternResult(pattern=state.pattern, scores=state.scores) - - -async def mpfp_traverse_hop_synchronized( - pool, - pattern_jobs: list[tuple[list[SeedNode], list[str]]], - config: MPFPConfig, - cache: EdgeCache, -) -> list[PatternResult]: - """ - Execute ALL patterns with hop-synchronized edge loading. - - Instead of running each pattern independently (causing multiple DB queries), - this function: - 1. Runs hop 1 for ALL patterns (using pre-warmed seed edges) - 2. Collects ALL unique hop-2 frontier nodes across patterns - 3. Pre-warms hop-2 edges in ONE query - 4. Runs hop 2 for ALL patterns - - This reduces DB queries from O(patterns * hops) to O(hops). - - Args: - pool: Database connection pool - pattern_jobs: List of (seeds, pattern) tuples - config: Algorithm parameters - cache: Shared edge cache (should be pre-warmed with seed edges) - - Returns: - List of PatternResult for each pattern - """ - import time - - # Initialize all pattern states - states = [_init_pattern_state(seeds, pattern) for seeds, pattern in pattern_jobs] - - # Determine max hops (all patterns should be same length, but be safe) - max_hops = max((len(p) for _, p in pattern_jobs), default=0) - - # Detailed timing for debugging - hop_times: list[dict] = [] - - # Execute hop-by-hop across ALL patterns - for hop in range(max_hops): - hop_start = time.time() - hop_timing = {"hop": hop, "patterns_executed": 0, "uncached_count": 0, "load_time": 0.0} - - # Execute this hop for all patterns, collect uncached nodes for next hop - all_uncached: set[str] = set() - exec_start = time.time() - for state in states: - if state.hop_index < len(state.pattern): - uncached = _execute_hop(state, cache, config) - all_uncached.update(uncached) - hop_timing["patterns_executed"] += 1 - hop_timing["exec_time"] = time.time() - exec_start - - # Pre-warm edges for ALL uncached nodes before next hop - hop_timing["uncached_count"] = len(all_uncached) - if all_uncached: - uncached_list = list(all_uncached - cache._fully_loaded) - hop_timing["uncached_after_filter"] = len(uncached_list) - if uncached_list: - load_start = time.time() - edges_by_type = await load_all_edges_for_frontier(pool, uncached_list, config.top_k_neighbors) - hop_timing["load_time"] = time.time() - load_start - cache.edge_load_time += hop_timing["load_time"] - cache.db_queries += 1 - cache.add_all_edges(edges_by_type, uncached_list) - hop_timing["edges_loaded"] = sum( - len(neighbors) for edges in edges_by_type.values() for neighbors in edges.values() - ) - - hop_timing["total_time"] = time.time() - hop_start - hop_times.append(hop_timing) - - # Store hop timing details in cache for logging - cache.hop_details = hop_times - - # Finalize all patterns - return [_finalize_pattern(state, config) for state in states] - - -async def mpfp_traverse_async( - pool, - seeds: list[SeedNode], - pattern: list[str], - config: MPFPConfig, - cache: EdgeCache, -) -> PatternResult: - """ - Async Forward Push traversal with lazy edge loading. - - NOTE: For better performance with multiple patterns, use mpfp_traverse_hop_synchronized(). - This function is kept for single-pattern use cases. - """ - if not seeds: - return PatternResult(pattern=pattern, scores={}) - - results = await mpfp_traverse_hop_synchronized(pool, [(seeds, pattern)], config, cache) - return results[0] if results else PatternResult(pattern=pattern, scores={}) - - -def rrf_fusion( - results: list[PatternResult], - k: int = 60, - top_k: int = 50, -) -> list[tuple[str, float]]: - """ - Reciprocal Rank Fusion to combine pattern results. - - Args: - results: List of pattern results - k: RRF constant (higher = more uniform weighting) - top_k: Number of results to return - - Returns: - List of (node_id, fused_score) tuples, sorted by score descending - """ - fused: dict[str, float] = {} - - for result in results: - if not result.scores: - continue - - # Rank nodes by their score in this pattern - ranked = sorted(result.scores.keys(), key=lambda n: result.scores[n], reverse=True) - - for rank, node_id in enumerate(ranked): - fused[node_id] = fused.get(node_id, 0) + 1.0 / (k + rank + 1) - - # Sort by fused score and return top-k - sorted_results = sorted(fused.items(), key=lambda x: x[1], reverse=True) - - return sorted_results[:top_k] - - -# ----------------------------------------------------------------------------- -# Database Loading -# ----------------------------------------------------------------------------- - - -async def fetch_memory_units_by_ids( - pool, - node_ids: list[str], - fact_type: str, -) -> list[RetrievalResult]: - """Fetch full memory unit details for a list of node IDs.""" - if not node_ids: - return [] - - async with acquire_with_retry(pool) as conn: - rows = await conn.fetch( - f""" - SELECT id, text, context, event_date, occurred_start, occurred_end, - mentioned_at, fact_type, document_id, chunk_id, tags, metadata - FROM {fq_table("memory_units")} - WHERE id = ANY($1::uuid[]) - AND fact_type = $2 - """, - node_ids, - fact_type, - ) - - return [RetrievalResult.from_db_row(dict(r)) for r in rows] - - -# ----------------------------------------------------------------------------- -# Graph Retriever Implementation -# ----------------------------------------------------------------------------- - - -class MPFPGraphRetriever(GraphRetriever): - """ - Graph retrieval using Meta-Path Forward Push with lazy edge loading. - - Runs predefined patterns in parallel from semantic and temporal seeds, - loading edges on-demand per hop instead of loading entire graph upfront. - """ - - def __init__(self, config: MPFPConfig | None = None): - """ - Initialize MPFP retriever. - - Args: - config: Algorithm configuration (uses defaults if None) - """ - if config is None: - # Read top_k_neighbors from global config - from ...config import get_config - - global_config = get_config() - config = MPFPConfig(top_k_neighbors=global_config.mpfp_top_k_neighbors) - self.config = config - - @property - def name(self) -> str: - return "mpfp" - - async def retrieve( - self, - pool, - query_embedding_str: str, - bank_id: str, - fact_type: str, - budget: int, - query_text: str | None = None, - semantic_seeds: list[RetrievalResult] | None = None, - temporal_seeds: list[RetrievalResult] | None = None, - adjacency=None, # Ignored - kept for interface compatibility - tags: list[str] | None = None, - tags_match: TagsMatch = "any", - tag_groups: list[TagGroup] | None = None, - ) -> tuple[list[RetrievalResult], MPFPTimings | None]: - """ - Retrieve facts using MPFP algorithm with lazy edge loading. - - Args: - pool: Database connection pool - query_embedding_str: Query embedding (used for fallback seed finding) - bank_id: Memory bank ID - fact_type: Fact type to filter - budget: Maximum results to return - query_text: Original query text (optional) - semantic_seeds: Pre-computed semantic entry points - temporal_seeds: Pre-computed temporal entry points - adjacency: Ignored (kept for interface compatibility) - tags: Optional list of tags for visibility filtering (OR matching) - - Returns: - Tuple of (List of RetrievalResult with activation scores, MPFPTimings) - """ - import time - - timings = MPFPTimings(fact_type=fact_type) - - # Convert seeds to SeedNode format - semantic_seed_nodes = self._convert_seeds(semantic_seeds, "similarity") - temporal_seed_nodes = self._convert_seeds(temporal_seeds, "temporal_score") - - # If no semantic seeds provided, fall back to finding our own - if not semantic_seed_nodes: - seeds_start = time.time() - semantic_seed_nodes = await self._find_semantic_seeds( - pool, - query_embedding_str, - bank_id, - fact_type, - tags=tags, - tags_match=tags_match, - tag_groups=tag_groups, - ) - timings.seeds_time = time.time() - seeds_start - logger.debug( - f"[MPFP] Found {len(semantic_seed_nodes)} semantic seeds for fact_type={fact_type} (tags={tags}, tags_match={tags_match})" - ) - - # Collect all pattern jobs - pattern_jobs = [] - - # Patterns from semantic seeds - for pattern in self.config.patterns_semantic: - if semantic_seed_nodes: - pattern_jobs.append((semantic_seed_nodes, pattern)) - - # Patterns from temporal seeds - for pattern in self.config.patterns_temporal: - if temporal_seed_nodes: - pattern_jobs.append((temporal_seed_nodes, pattern)) - - if not pattern_jobs: - logger.debug( - f"[MPFP] No pattern jobs (semantic_seeds={len(semantic_seed_nodes)}, temporal_seeds={len(temporal_seed_nodes)})" - ) - return [], timings - - timings.pattern_count = len(pattern_jobs) - - # Shared edge cache across all patterns - cache = EdgeCache() - - # Pre-warm cache with ALL seed node edges BEFORE running patterns - # This prevents redundant DB queries at hop 1 - all_seed_ids = list({s.node_id for seeds, _ in pattern_jobs for s in seeds}) - if all_seed_ids: - import time as time_module - - prewarm_start = time_module.time() - edges_by_type = await load_all_edges_for_frontier(pool, all_seed_ids, self.config.top_k_neighbors) - cache.edge_load_time += time_module.time() - prewarm_start - cache.db_queries += 1 - cache.add_all_edges(edges_by_type, all_seed_ids) - - # Run all patterns with HOP-SYNCHRONIZED edge loading - # This batches hop-2 edge loads across ALL patterns into ONE query - # Reduces DB queries from O(patterns * hops) to O(hops) - step_start = time.time() - pattern_results = await mpfp_traverse_hop_synchronized(pool, pattern_jobs, self.config, cache) - timings.traverse = time.time() - step_start - - # Record edge loading stats from cache - timings.edge_count = sum(len(neighbors) for g in cache.graphs.values() for neighbors in g.values()) - timings.db_queries = cache.db_queries - timings.edge_load_time = cache.edge_load_time - timings.hop_details = cache.hop_details - - # Fuse results - step_start = time.time() - fused = rrf_fusion(pattern_results, top_k=budget) - timings.fusion = time.time() - step_start - - if not fused: - logger.debug(f"[MPFP] No fused results after RRF fusion (pattern_count={len(pattern_results)})") - return [], timings - - # Get top result IDs - result_ids = [node_id for node_id, score in fused][:budget] - - # Fetch full details - step_start = time.time() - results = await fetch_memory_units_by_ids(pool, result_ids, fact_type) - timings.fetch = time.time() - step_start - - # Filter results by tags (graph traversal may have picked up unfiltered memories) - if tags: - from .tags import filter_results_by_tags - - results = filter_results_by_tags(results, tags, match=tags_match) - - # Apply compound tag group filtering (post-traversal) - if tag_groups: - from .tags import filter_results_by_tag_groups - - results = filter_results_by_tag_groups(results, tag_groups) - - timings.result_count = len(results) - - # Add activation scores from fusion - score_map = {node_id: score for node_id, score in fused} - for result in results: - result.activation = score_map.get(result.id, 0.0) - - # Sort by activation - results.sort(key=lambda r: r.activation or 0, reverse=True) - - return results, timings - - def _convert_seeds( - self, - seeds: list[RetrievalResult] | None, - score_attr: str, - ) -> list[SeedNode]: - """Convert RetrievalResult seeds to SeedNode format.""" - if not seeds: - return [] - - result = [] - for seed in seeds: - score = getattr(seed, score_attr, None) - if score is None: - score = seed.activation or seed.similarity or 1.0 - result.append(SeedNode(node_id=seed.id, score=score)) - - return result - - async def _find_semantic_seeds( - self, - pool, - query_embedding_str: str, - bank_id: str, - fact_type: str, - limit: int = 20, - threshold: float = 0.3, - tags: list[str] | None = None, - tags_match: TagsMatch = "any", - tag_groups: list[TagGroup] | None = None, - ) -> list[SeedNode]: - """Fallback: find semantic seeds via embedding search.""" - from .tags import build_tag_groups_where_clause, build_tags_where_clause_simple - - tags_clause = build_tags_where_clause_simple(tags, 6, match=tags_match) - tag_groups_param_start = 6 + (1 if tags else 0) - groups_clause, groups_params, _ = build_tag_groups_where_clause(tag_groups, tag_groups_param_start) - params = [query_embedding_str, bank_id, fact_type, threshold, limit] - if tags: - params.append(tags) - params.extend(groups_params) - - async with acquire_with_retry(pool) as conn: - rows = await conn.fetch( - f""" - SELECT id, 1 - (embedding <=> $1::vector) AS similarity - FROM {fq_table("memory_units")} - WHERE bank_id = $2 - AND embedding IS NOT NULL - AND fact_type = $3 - AND (1 - (embedding <=> $1::vector)) >= $4 - {tags_clause} - {groups_clause} - ORDER BY embedding <=> $1::vector - LIMIT $5 - """, - *params, - ) - - return [SeedNode(node_id=str(r["id"]), score=r["similarity"]) for r in rows] diff --git a/hindsight-api-slim/hindsight_api/engine/search/retrieval.py b/hindsight-api-slim/hindsight_api/engine/search/retrieval.py index bca52afd..78e6bed7 100644 --- a/hindsight-api-slim/hindsight_api/engine/search/retrieval.py +++ b/hindsight-api-slim/hindsight_api/engine/search/retrieval.py @@ -18,11 +18,10 @@ from typing import Optional from ...config import get_config from ..db_utils import acquire_with_retry from ..memory_engine import fq_table -from .graph_retrieval import BFSGraphRetriever, GraphRetriever +from .graph_retrieval import GraphRetriever from .link_expansion_retrieval import LinkExpansionRetriever -from .mpfp_retrieval import MPFPGraphRetriever from .tags import TagGroup, TagsMatch, build_tag_groups_where_clause, build_tags_where_clause_simple -from .types import MPFPTimings, RetrievalResult +from .types import GraphRetrievalTimings, RetrievalResult logger = logging.getLogger(__name__) @@ -46,7 +45,9 @@ class ParallelRetrievalResult: temporal: list[RetrievalResult] | None timings: dict[str, float] = field(default_factory=dict) temporal_constraint: tuple | None = None # (start_date, end_date) - mpfp_timings: list[MPFPTimings] = field(default_factory=list) # MPFP sub-step timings per fact type + graph_timings: list[GraphRetrievalTimings] = field( + default_factory=list + ) # Graph retrieval sub-step timings per fact type max_conn_wait: float = 0.0 # Maximum connection acquisition wait time across all methods @@ -72,15 +73,7 @@ def get_default_graph_retriever() -> GraphRetriever: if _default_graph_retriever is None: config = get_config() retriever_type = config.graph_retriever.lower() - if retriever_type == "mpfp": - _default_graph_retriever = MPFPGraphRetriever() - logger.info( - f"Using MPFP graph retriever (top_k_neighbors={_default_graph_retriever.config.top_k_neighbors})" - ) - elif retriever_type == "bfs": - _default_graph_retriever = BFSGraphRetriever() - logger.info("Using BFS graph retriever") - elif retriever_type == "link_expansion": + if retriever_type == "link_expansion": _default_graph_retriever = LinkExpansionRetriever() logger.info("Using LinkExpansion graph retriever") else: @@ -627,9 +620,11 @@ async def retrieve_all_fact_types_parallel( timings["temporal_combined"] = temporal_time # Step 3: Run graph retrieval for each fact type in parallel - async def run_graph_for_fact_type(ft: str) -> tuple[str, list[RetrievalResult], float, MPFPTimings | None]: + async def run_graph_for_fact_type( + ft: str, + ) -> tuple[str, list[RetrievalResult], float, GraphRetrievalTimings | None]: graph_start = time.time() - results, mpfp_timing = await retriever.retrieve( + results, graph_timing = await retriever.retrieve( pool=pool, query_embedding_str=query_embedding_str, bank_id=bank_id, @@ -642,7 +637,7 @@ async def retrieve_all_fact_types_parallel( tags_match=tags_match, tag_groups=tag_groups, ) - return ft, results, time.time() - graph_start, mpfp_timing + return ft, results, time.time() - graph_start, graph_timing # Run graph for all fact types in parallel graph_tasks = [run_graph_for_fact_type(ft) for ft in fact_types] @@ -651,7 +646,7 @@ async def retrieve_all_fact_types_parallel( # Organize results by fact type results_by_fact_type: dict[str, ParallelRetrievalResult] = {} max_conn_wait = conn_wait # Single connection for semantic+bm25+temporal - all_mpfp_timings: list[MPFPTimings] = [] + all_graph_timings: list[GraphRetrievalTimings] = [] for ft in fact_types: # Get semantic + bm25 results for this fact type @@ -660,14 +655,14 @@ async def retrieve_all_fact_types_parallel( # Find graph results for this fact type graph_results = [] graph_time = 0.0 - mpfp_timing = None + graph_timing = None for gr in graph_results_list: if gr[0] == ft: graph_results = gr[1] graph_time = gr[2] - mpfp_timing = gr[3] - if mpfp_timing: - all_mpfp_timings.append(mpfp_timing) + graph_timing = gr[3] + if graph_timing: + all_graph_timings.append(graph_timing) break # Get temporal results for this fact type from combined result @@ -688,7 +683,7 @@ async def retrieve_all_fact_types_parallel( "temporal_extraction": temporal_extraction_time, }, temporal_constraint=temporal_constraint, - mpfp_timings=[mpfp_timing] if mpfp_timing else [], + graph_timings=[graph_timing] if graph_timing else [], max_conn_wait=max_conn_wait, ) diff --git a/hindsight-api-slim/hindsight_api/engine/search/types.py b/hindsight-api-slim/hindsight_api/engine/search/types.py index 90edcf0e..f49741ee 100644 --- a/hindsight-api-slim/hindsight_api/engine/search/types.py +++ b/hindsight-api-slim/hindsight_api/engine/search/types.py @@ -11,8 +11,8 @@ from typing import Any @dataclass -class MPFPTimings: - """Timing breakdown for a single MPFP retrieval call.""" +class GraphRetrievalTimings: + """Timing breakdown for a single graph retrieval call.""" fact_type: str edge_count: int = 0 # Total edges loaded diff --git a/hindsight-api-slim/tests/test_mpfp_retrieval.py b/hindsight-api-slim/tests/test_mpfp_retrieval.py deleted file mode 100644 index a36464db..00000000 --- a/hindsight-api-slim/tests/test_mpfp_retrieval.py +++ /dev/null @@ -1,819 +0,0 @@ -""" -Tests for MPFP (Meta-Path Forward Push) graph retrieval. - -Tests cover: -1. EdgeCache - lazy caching behavior -2. mpfp_traverse_async - core traversal algorithm -3. load_edges_for_frontier - lazy edge loading -4. rrf_fusion - result fusion -5. MPFPGraphRetriever - full integration -""" - -import pytest -from unittest.mock import AsyncMock, MagicMock, patch -from datetime import datetime, timezone - -from hindsight_api.engine.search.mpfp_retrieval import ( - EdgeCache, - EdgeTarget, - MPFPConfig, - MPFPGraphRetriever, - PatternResult, - SeedNode, - load_all_edges_for_frontier, - mpfp_traverse_async, - rrf_fusion, -) -from hindsight_api.engine.search.types import RetrievalResult - - -class TestEdgeCache: - """Tests for the EdgeCache lazy loading cache.""" - - def test_empty_cache_returns_empty_neighbors(self): - """Empty cache should return empty list for any node.""" - cache = EdgeCache() - neighbors = cache.get_neighbors("semantic", "node-1") - assert neighbors == [] - - def test_is_fully_loaded_false_for_uncached(self): - """is_fully_loaded should return False for nodes not yet loaded.""" - cache = EdgeCache() - assert cache.is_fully_loaded("node-1") is False - - def test_add_all_edges_marks_as_fully_loaded(self): - """Adding edges should mark nodes as fully loaded.""" - cache = EdgeCache() - - edges_by_type = { - "semantic": {"node-1": [EdgeTarget("node-2", 0.8), EdgeTarget("node-3", 0.6)]}, - } - cache.add_all_edges(edges_by_type, ["node-1", "node-4"]) # node-4 has no edges - - assert cache.is_fully_loaded("node-1") is True - assert cache.is_fully_loaded("node-4") is True # Marked even with no edges - assert cache.is_fully_loaded("node-2") is False # Target, not source - - def test_get_neighbors_returns_added_edges(self): - """get_neighbors should return edges after add_all_edges.""" - cache = EdgeCache() - - edges_by_type = { - "semantic": {"node-1": [EdgeTarget("node-2", 0.8), EdgeTarget("node-3", 0.6)]}, - } - cache.add_all_edges(edges_by_type, ["node-1"]) - - neighbors = cache.get_neighbors("semantic", "node-1") - assert len(neighbors) == 2 - assert neighbors[0].node_id == "node-2" - assert neighbors[0].weight == 0.8 - - def test_get_uncached_filters_loaded_nodes(self): - """get_uncached should only return nodes not yet fully loaded.""" - cache = EdgeCache() - - # Load some nodes (all edge types) - cache.add_all_edges({"semantic": {"node-1": []}}, ["node-1", "node-2"]) - - # Check uncached - uncached = cache.get_uncached(["node-1", "node-2", "node-3", "node-4"]) - assert set(uncached) == {"node-3", "node-4"} - - def test_get_normalized_neighbors_normalizes_weights(self): - """get_normalized_neighbors should normalize weights to sum to 1.""" - cache = EdgeCache() - - edges_by_type = { - "semantic": { - "node-1": [ - EdgeTarget("node-2", 0.8), - EdgeTarget("node-3", 0.4), - EdgeTarget("node-4", 0.2), - ], - }, - } - cache.add_all_edges(edges_by_type, ["node-1"]) - - # Get top 2, normalized - neighbors = cache.get_normalized_neighbors("semantic", "node-1", top_k=2) - assert len(neighbors) == 2 - - # Weights should sum to 1 - total = sum(n.weight for n in neighbors) - assert abs(total - 1.0) < 0.001 - - # node-2 should have higher normalized weight than node-3 - assert neighbors[0].node_id == "node-2" - assert neighbors[1].node_id == "node-3" - # Original: 0.8 and 0.4, so normalized: 0.8/1.2 and 0.4/1.2 - assert abs(neighbors[0].weight - 0.8 / 1.2) < 0.001 - assert abs(neighbors[1].weight - 0.4 / 1.2) < 0.001 - - def test_different_edge_types_are_separate(self): - """Different edge types should be stored separately.""" - cache = EdgeCache() - - edges_by_type = { - "semantic": {"node-1": [EdgeTarget("node-2", 0.8)]}, - "temporal": {"node-1": [EdgeTarget("node-3", 0.5)]}, - } - cache.add_all_edges(edges_by_type, ["node-1"]) - - semantic_neighbors = cache.get_neighbors("semantic", "node-1") - temporal_neighbors = cache.get_neighbors("temporal", "node-1") - - assert len(semantic_neighbors) == 1 - assert semantic_neighbors[0].node_id == "node-2" - - assert len(temporal_neighbors) == 1 - assert temporal_neighbors[0].node_id == "node-3" - - -class TestRRFFusion: - """Tests for RRF (Reciprocal Rank Fusion).""" - - def test_empty_results(self): - """Empty results should return empty fusion.""" - fused = rrf_fusion([]) - assert fused == [] - - def test_single_pattern_ranking(self): - """Single pattern should preserve ranking order.""" - result = PatternResult( - pattern=["semantic"], - scores={"node-1": 0.9, "node-2": 0.7, "node-3": 0.5}, - ) - - fused = rrf_fusion([result], top_k=3) - assert len(fused) == 3 - # node-1 should be first (highest score) - assert fused[0][0] == "node-1" - assert fused[1][0] == "node-2" - assert fused[2][0] == "node-3" - - def test_multiple_patterns_boost_common_nodes(self): - """Nodes appearing in multiple patterns should get boosted.""" - result1 = PatternResult( - pattern=["semantic", "semantic"], - scores={"node-1": 0.9, "node-2": 0.7}, - ) - result2 = PatternResult( - pattern=["entity", "temporal"], - scores={"node-1": 0.8, "node-3": 0.6}, # node-1 in both - ) - - fused = rrf_fusion([result1, result2], top_k=3) - - # node-1 should be first (appears in both patterns) - assert fused[0][0] == "node-1" - # Its score should be higher than others - assert fused[0][1] > fused[1][1] - - def test_top_k_limits_results(self): - """top_k should limit the number of results.""" - result = PatternResult( - pattern=["semantic"], - scores={f"node-{i}": 1.0 / (i + 1) for i in range(10)}, - ) - - fused = rrf_fusion([result], top_k=3) - assert len(fused) == 3 - - def test_empty_pattern_scores_ignored(self): - """Patterns with empty scores should be ignored.""" - result1 = PatternResult(pattern=["semantic"], scores={}) - result2 = PatternResult( - pattern=["entity"], - scores={"node-1": 0.5}, - ) - - fused = rrf_fusion([result1, result2], top_k=3) - assert len(fused) == 1 - assert fused[0][0] == "node-1" - - -class TestMPFPTraverseAsync: - """Tests for the async MPFP traversal algorithm.""" - - @pytest.mark.asyncio - async def test_empty_seeds_returns_empty(self): - """Empty seeds should return empty result.""" - cache = EdgeCache() - config = MPFPConfig() - - result = await mpfp_traverse_async( - pool=None, # Not used when no seeds - seeds=[], - pattern=["semantic"], - config=config, - cache=cache, - ) - - assert result.scores == {} - - @pytest.mark.asyncio - async def test_single_hop_no_edges(self): - """Single hop with no edges should deposit mass at seeds.""" - cache = EdgeCache() - config = MPFPConfig(alpha=0.15, threshold=1e-6) - - # Pre-populate cache with empty edges for seed (marks as fully loaded) - cache.add_all_edges({}, ["seed-1"]) - - seeds = [SeedNode("seed-1", 1.0)] - - with patch( - "hindsight_api.engine.search.mpfp_retrieval.load_all_edges_for_frontier", - new_callable=AsyncMock, - return_value={}, - ): - result = await mpfp_traverse_async( - pool=MagicMock(), - seeds=seeds, - pattern=["semantic"], - config=config, - cache=cache, - ) - - # Seed should have alpha portion of its mass - assert "seed-1" in result.scores - assert result.scores["seed-1"] == pytest.approx(config.alpha, rel=0.01) - - @pytest.mark.asyncio - async def test_single_hop_with_edges(self): - """Single hop should spread mass to neighbors.""" - cache = EdgeCache() - config = MPFPConfig(alpha=0.15, threshold=1e-6, top_k_neighbors=10) - - seeds = [SeedNode("seed-1", 1.0)] - - # Pre-populate cache with seed edges (mimics pre-warming in retrieve()) - cache.add_all_edges( - { - "semantic": { - "seed-1": [ - EdgeTarget("neighbor-1", 0.8), - EdgeTarget("neighbor-2", 0.4), - ] - } - }, - ["seed-1"], - ) - - # Mock for loading neighbor edges (after hop 0) - async def mock_load_all_edges(pool, node_ids, top_k=20): - return {} - - with patch( - "hindsight_api.engine.search.mpfp_retrieval.load_all_edges_for_frontier", - side_effect=mock_load_all_edges, - ): - result = await mpfp_traverse_async( - pool=MagicMock(), - seeds=seeds, - pattern=["semantic"], - config=config, - cache=cache, - ) - - # Seed keeps alpha portion - assert "seed-1" in result.scores - assert result.scores["seed-1"] == pytest.approx(config.alpha, rel=0.01) - - # Neighbors get remaining mass (normalized) - assert "neighbor-1" in result.scores - assert "neighbor-2" in result.scores - - # neighbor-1 should get more (higher weight) - assert result.scores["neighbor-1"] > result.scores["neighbor-2"] - - @pytest.mark.asyncio - async def test_two_hops(self): - """Two-hop pattern should traverse through neighbors.""" - cache = EdgeCache() - config = MPFPConfig(alpha=0.15, threshold=1e-6, top_k_neighbors=10) - - seeds = [SeedNode("seed-1", 1.0)] - - # Pre-populate cache with seed edges (mimics pre-warming in retrieve()) - cache.add_all_edges( - {"semantic": {"seed-1": [EdgeTarget("hop1-node", 1.0)]}}, - ["seed-1"], - ) - - # Mock edge loading for hop 1 nodes - async def mock_load_all_edges(pool, node_ids, top_k=20): - edges: dict[str, dict[str, list[EdgeTarget]]] = {"semantic": {}} - if "hop1-node" in node_ids: - edges["semantic"]["hop1-node"] = [EdgeTarget("hop2-node", 1.0)] - return edges - - with patch( - "hindsight_api.engine.search.mpfp_retrieval.load_all_edges_for_frontier", - side_effect=mock_load_all_edges, - ): - result = await mpfp_traverse_async( - pool=MagicMock(), - seeds=seeds, - pattern=["semantic", "semantic"], # Two hops - config=config, - cache=cache, - ) - - # Should have scores for all three nodes - assert "seed-1" in result.scores - assert "hop1-node" in result.scores - assert "hop2-node" in result.scores - - @pytest.mark.asyncio - async def test_cache_reuse(self): - """Cache should prevent redundant edge loading for already-cached nodes.""" - cache = EdgeCache() - config = MPFPConfig(alpha=0.15, threshold=1e-6) - - # Pre-load cache (marks seed-1 AND neighbor-1 as fully loaded) - # neighbor-1 is also cached because after hop 0, the frontier contains neighbor-1 - # and the algorithm tries to pre-warm edges for the next hop - cache.add_all_edges( - {"semantic": {"seed-1": [EdgeTarget("neighbor-1", 1.0)], "neighbor-1": []}}, - ["seed-1", "neighbor-1"], - ) - - seeds = [SeedNode("seed-1", 1.0)] - - load_mock = AsyncMock(return_value={}) - - with patch( - "hindsight_api.engine.search.mpfp_retrieval.load_all_edges_for_frontier", - load_mock, - ): - await mpfp_traverse_async( - pool=MagicMock(), - seeds=seeds, - pattern=["semantic"], - config=config, - cache=cache, - ) - - # Should not call load_all_edges_for_frontier since all nodes are already cached - load_mock.assert_not_called() - - -class TestMPFPGraphRetriever: - """Tests for the MPFPGraphRetriever class.""" - - def test_name_is_mpfp(self): - """Retriever name should be 'mpfp'.""" - retriever = MPFPGraphRetriever() - assert retriever.name == "mpfp" - - def test_default_config(self): - """Default config should have expected patterns.""" - # Use explicit config to avoid global config dependency - config = MPFPConfig() - retriever = MPFPGraphRetriever(config=config) - - assert len(retriever.config.patterns_semantic) > 0 - assert len(retriever.config.patterns_temporal) > 0 - assert retriever.config.alpha == 0.15 - assert retriever.config.top_k_neighbors == 20 - - def test_custom_config(self): - """Custom config should be used.""" - config = MPFPConfig(alpha=0.3, top_k_neighbors=10) - retriever = MPFPGraphRetriever(config=config) - - assert retriever.config.alpha == 0.3 - assert retriever.config.top_k_neighbors == 10 - - def test_convert_seeds_from_retrieval_results(self): - """_convert_seeds should extract scores from RetrievalResult.""" - retriever = MPFPGraphRetriever() - - results = [ - RetrievalResult(id="id-1", text="text1", fact_type="world", similarity=0.9), - RetrievalResult(id="id-2", text="text2", fact_type="world", similarity=0.7), - ] - - seeds = retriever._convert_seeds(results, "similarity") - - assert len(seeds) == 2 - assert seeds[0].node_id == "id-1" - assert seeds[0].score == 0.9 - assert seeds[1].node_id == "id-2" - assert seeds[1].score == 0.7 - - def test_convert_seeds_empty(self): - """_convert_seeds should handle empty/None input.""" - retriever = MPFPGraphRetriever() - - assert retriever._convert_seeds(None, "similarity") == [] - assert retriever._convert_seeds([], "similarity") == [] - - @pytest.mark.asyncio - async def test_retrieve_no_seeds_returns_empty(self): - """Retrieve with no seeds should return empty results.""" - # Use explicit config to avoid global config dependency - config = MPFPConfig() - retriever = MPFPGraphRetriever(config=config) - - # Mock _find_semantic_seeds to return empty - with patch.object(retriever, "_find_semantic_seeds", new_callable=AsyncMock, return_value=[]): - results, timings = await retriever.retrieve( - pool=MagicMock(), - query_embedding_str="[0.1, 0.2]", - bank_id="test", - fact_type="world", - budget=10, - ) - - assert results == [] - assert timings is not None - assert timings.pattern_count == 0 - - @pytest.mark.asyncio - async def test_retrieve_with_semantic_seeds(self): - """Retrieve with semantic seeds should run patterns and return results.""" - # Use explicit config to avoid global config dependency - config = MPFPConfig() - retriever = MPFPGraphRetriever(config=config) - - semantic_seeds = [ - RetrievalResult(id="seed-1", text="seed text", fact_type="world", similarity=0.9), - ] - - # Mock the internal functions - # mpfp_traverse_hop_synchronized returns a list of PatternResult (one per pattern) - async def mock_traverse(*args, **kwargs): - return [PatternResult(pattern=["semantic"], scores={"seed-1": 0.5, "result-1": 0.3})] - - async def mock_fetch(pool, node_ids, fact_type): - return [ - RetrievalResult(id="seed-1", text="seed text", fact_type="world"), - RetrievalResult(id="result-1", text="result text", fact_type="world"), - ] - - with ( - patch( - "hindsight_api.engine.search.mpfp_retrieval.mpfp_traverse_hop_synchronized", - side_effect=mock_traverse, - ), - patch( - "hindsight_api.engine.search.mpfp_retrieval.fetch_memory_units_by_ids", - side_effect=mock_fetch, - ), - patch( - "hindsight_api.engine.search.mpfp_retrieval.load_all_edges_for_frontier", - new_callable=AsyncMock, - return_value={}, - ), - ): - results, timings = await retriever.retrieve( - pool=MagicMock(), - query_embedding_str="[0.1, 0.2]", - bank_id="test", - fact_type="world", - budget=10, - semantic_seeds=semantic_seeds, - ) - - assert len(results) == 2 - assert timings is not None - assert timings.pattern_count > 0 - - -@pytest.mark.asyncio -async def test_mpfp_integration(memory, request_context): - """Integration test: MPFP retrieval with real database.""" - bank_id = f"test_mpfp_{datetime.now(timezone.utc).timestamp()}" - - try: - # Store memories with entity relationships - await memory.retain_async( - bank_id=bank_id, - content="Alice works at TechCorp as a software engineer", - context="employee info", - request_context=request_context, - ) - await memory.retain_async( - bank_id=bank_id, - content="TechCorp is located in San Francisco", - context="company info", - request_context=request_context, - ) - await memory.retain_async( - bank_id=bank_id, - content="Bob is Alice's manager at TechCorp", - context="employee info", - request_context=request_context, - ) - await memory.retain_async( - bank_id=bank_id, - content="San Francisco has many tech companies", - context="city info", - request_context=request_context, - ) - - # Query should find related facts via graph traversal - from hindsight_api.engine.memory_engine import Budget - - result = await memory.recall_async( - bank_id=bank_id, - query="Tell me about Alice", - fact_type=["world"], - budget=Budget.MID, - max_tokens=2048, - request_context=request_context, - ) - - # Should return results - assert result.results is not None - assert len(result.results) > 0 - - # Should find Alice-related facts - fact_texts = [f.text for f in result.results] - alice_facts = [t for t in fact_texts if "Alice" in t or "TechCorp" in t] - assert len(alice_facts) > 0, f"Should find Alice-related facts, got: {fact_texts}" - - print(f"\n✓ MPFP integration test passed! Found {len(result.results)} facts") - - finally: - await memory.delete_bank(bank_id, request_context=request_context) - - -@pytest.mark.asyncio -async def test_mpfp_lazy_loading_efficiency(memory, request_context): - """Test that MPFP loads edges lazily, not upfront.""" - bank_id = f"test_mpfp_lazy_{datetime.now(timezone.utc).timestamp()}" - - try: - # Store many memories to create a larger graph - for i in range(20): - await memory.retain_async( - bank_id=bank_id, - content=f"Fact number {i} about topic {i % 5}", - context=f"context {i}", - request_context=request_context, - ) - - from hindsight_api.engine.memory_engine import Budget - - # Query - MPFP should only load edges for relevant frontier nodes - result = await memory.recall_async( - bank_id=bank_id, - query="topic 0", - fact_type=["world"], - budget=Budget.LOW, - max_tokens=1024, - enable_trace=True, - request_context=request_context, - ) - - assert result.results is not None - - # Check trace for timing info - if result.trace: - print(f"\n✓ MPFP lazy loading test passed!") - print(f" - Facts returned: {len(result.results)}") - - finally: - await memory.delete_bank(bank_id, request_context=request_context) - - -# ============================================================================ -# MPFP Performance Benchmark Tests -# ============================================================================ -# These tests require an external database with a large memory bank to be useful. -# Set EXTERNAL_DATABASE_URL and BENCHMARK_BANK_ID environment variables to run. -# Example: -# EXTERNAL_DATABASE_URL=postgresql://user:pass@host:port/db \ -# BENCHMARK_BANK_ID=load-test \ -# pytest tests/test_mpfp_retrieval.py::test_mpfp_edge_loading_performance -v -s - - -import os -import asyncpg - -EXTERNAL_DATABASE_URL = os.environ.get("EXTERNAL_DATABASE_URL") -BENCHMARK_BANK_ID = os.environ.get("BENCHMARK_BANK_ID", "load-test") - -requires_external_db = pytest.mark.skipif( - EXTERNAL_DATABASE_URL is None, - reason="EXTERNAL_DATABASE_URL not set - skipping external DB benchmark", -) - - -@requires_external_db -@pytest.mark.asyncio -async def test_mpfp_edge_loading_performance(): - """ - Benchmark MPFP edge loading performance. - - This test measures the performance of the LATERAL query optimization - for loading edges in the MPFP graph traversal algorithm. - - Set EXTERNAL_DATABASE_URL to point to a database with existing data. - Set BENCHMARK_BANK_ID to specify which bank to query (default: load-test). - - Example usage: - EXTERNAL_DATABASE_URL=postgresql://hindsight:hindsight@localhost:5435/hindsight \ - BENCHMARK_BANK_ID=load-test \ - pytest tests/test_mpfp_retrieval.py::test_mpfp_edge_loading_performance -v -s - """ - import time - - # Connect to external database - pool = await asyncpg.create_pool(EXTERNAL_DATABASE_URL, min_size=2, max_size=10) - - try: - # Get some sample node IDs from the database - async with pool.acquire() as conn: - # First check how many links exist - stats = await conn.fetchrow(""" - SELECT - count(*) as total_links, - count(DISTINCT from_unit_id) as unique_sources - FROM memory_links - """) - print(f"\n📊 Database Stats:") - print(f" Total links: {stats['total_links']:,}") - print(f" Unique sources: {stats['unique_sources']:,}") - - # Get edge distribution by type - type_stats = await conn.fetch(""" - SELECT link_type, count(*) as cnt, - round(avg(weight)::numeric, 3) as avg_weight - FROM memory_links - GROUP BY link_type - ORDER BY cnt DESC - """) - print(f"\n Edge distribution:") - for row in type_stats: - print(f" - {row['link_type']}: {row['cnt']:,} (avg_weight={row['avg_weight']})") - - # Get sample frontier nodes (from memory_units in the benchmark bank) - # bank_id is the text primary key in banks table - frontier_rows = await conn.fetch(""" - SELECT id FROM memory_units - WHERE bank_id = $1 - LIMIT 100 - """, BENCHMARK_BANK_ID) - - if not frontier_rows: - pytest.skip(f"No memory units found for bank '{BENCHMARK_BANK_ID}'") - - frontier_node_ids = [str(row['id']) for row in frontier_rows] - print(f"\n🎯 Testing with {len(frontier_node_ids)} frontier nodes from bank '{BENCHMARK_BANK_ID}'") - - # Test 1: Original query approach (all edges, no per-type limit) - async with pool.acquire() as conn: - start = time.time() - original_rows = await conn.fetch(""" - SELECT ml.from_unit_id, ml.to_unit_id, ml.link_type, ml.weight - FROM memory_links ml - WHERE ml.from_unit_id = ANY($1::uuid[]) - AND ml.weight >= 0.1 - ORDER BY ml.from_unit_id, ml.link_type, ml.weight DESC - """, frontier_node_ids) - original_time = time.time() - start - original_count = len(original_rows) - - # Test 2: New LATERAL query approach (top-k per type) - async with pool.acquire() as conn: - start = time.time() - lateral_rows = await conn.fetch(""" - WITH frontier(node_id) AS (SELECT unnest($1::uuid[])) - SELECT f.node_id as from_unit_id, lt.link_type, edges.to_unit_id, edges.weight - FROM frontier f - CROSS JOIN (VALUES ('semantic'), ('temporal'), ('entity'), ('causes'), ('caused_by')) AS lt(link_type) - CROSS JOIN LATERAL ( - SELECT ml.to_unit_id, ml.weight - FROM memory_links ml - WHERE ml.from_unit_id = f.node_id - AND ml.link_type = lt.link_type - AND ml.weight >= 0.1 - ORDER BY ml.weight DESC - LIMIT 20 - ) edges - """, frontier_node_ids) - lateral_time = time.time() - start - lateral_count = len(lateral_rows) - - # Print results - print(f"\n⏱️ Performance Comparison ({len(frontier_node_ids)} nodes):") - print(f"\n Original (all edges):") - print(f" - Time: {original_time * 1000:.2f}ms") - print(f" - Rows: {original_count:,}") - print(f" - Rows/node: {original_count / len(frontier_node_ids):.1f}") - - print(f"\n LATERAL (top-20 per type):") - print(f" - Time: {lateral_time * 1000:.2f}ms") - print(f" - Rows: {lateral_count:,}") - print(f" - Rows/node: {lateral_count / len(frontier_node_ids):.1f}") - - speedup = original_time / lateral_time if lateral_time > 0 else float('inf') - reduction = (1 - lateral_count / original_count) * 100 if original_count > 0 else 0 - print(f"\n 📈 Improvement:") - print(f" - Speedup: {speedup:.2f}x faster") - print(f" - Data reduction: {reduction:.1f}% fewer rows") - - # Assert improvement (should be at least some improvement for large datasets) - if original_count > 1000: - # For large datasets, expect significant improvement - assert speedup >= 1.5, f"Expected at least 1.5x speedup, got {speedup:.2f}x" - assert reduction >= 30, f"Expected at least 30% data reduction, got {reduction:.1f}%" - print(f"\n✅ Performance test PASSED!") - else: - print(f"\n⚠️ Dataset too small ({original_count} rows) for meaningful performance comparison") - - finally: - await pool.close() - - -@requires_external_db -@pytest.mark.asyncio -async def test_mpfp_full_retrieval_performance(): - """ - Benchmark full MPFP retrieval including traversal and reranking. - - This test measures end-to-end MPFP retrieval performance. - """ - import time - - pool = await asyncpg.create_pool(EXTERNAL_DATABASE_URL, min_size=2, max_size=10) - - try: - # Get a sample query embedding from an existing memory unit - async with pool.acquire() as conn: - # Check if bank exists - bank_exists = await conn.fetchval(""" - SELECT 1 FROM banks WHERE bank_id = $1 - """, BENCHMARK_BANK_ID) - if not bank_exists: - pytest.skip(f"Bank '{BENCHMARK_BANK_ID}' not found") - - sample = await conn.fetchrow(""" - SELECT embedding::text as embedding_str - FROM memory_units - WHERE bank_id = $1 - AND embedding IS NOT NULL - LIMIT 1 - """, BENCHMARK_BANK_ID) - - if not sample: - pytest.skip("No memory units with embeddings found") - - query_embedding_str = sample['embedding_str'] - - # Run MPFP retrieval - retriever = MPFPGraphRetriever() - - print(f"\n🔍 Running MPFP retrieval benchmark on bank '{BENCHMARK_BANK_ID}'...") - - # Warm-up run - await retriever.retrieve( - pool=pool, - query_embedding_str=query_embedding_str, - bank_id=BENCHMARK_BANK_ID, - fact_type="world", - budget=100, - query_text="test query", - ) - - # Timed runs - timings_list = [] - for i in range(3): - start = time.time() - results, timings = await retriever.retrieve( - pool=pool, - query_embedding_str=query_embedding_str, - bank_id=BENCHMARK_BANK_ID, - fact_type="opinion", - budget=100, - query_text="What did I say about training models?", - ) - elapsed = time.time() - start - timings_list.append((elapsed, timings, len(results))) - - # Print results - print(f"\n⏱️ MPFP Retrieval Results (3 runs):") - for i, (elapsed, timings, count) in enumerate(timings_list): - print(f"\n Run {i + 1}:") - print(f" - Total: {elapsed * 1000:.2f}ms") - print(f" - Results: {count}") - if timings: - print(f" - Seeds: {timings.seeds_time * 1000:.2f}ms") - print(f" - Patterns: {timings.pattern_count}") - print(f" - Traverse: {timings.traverse * 1000:.2f}ms") - print(f" - Edge load: {timings.edge_load_time * 1000:.2f}ms") - print(f" - Edges: {timings.edge_count:,}") - print(f" - DB queries: {timings.db_queries}") - print(f" - Fusion: {timings.fusion * 1000:.2f}ms") - print(f" - Fetch: {timings.fetch * 1000:.2f}ms") - - avg_time = sum(t[0] for t in timings_list) / len(timings_list) - print(f"\n 📊 Average: {avg_time * 1000:.2f}ms") - print(f"\n✅ MPFP retrieval benchmark complete!") - - finally: - await pool.close() diff --git a/hindsight-docs/docs/developer/configuration.md b/hindsight-docs/docs/developer/configuration.md index 477d68ad..2cd2e466 100644 --- a/hindsight-docs/docs/developer/configuration.md +++ b/hindsight-docs/docs/developer/configuration.md @@ -587,20 +587,17 @@ For advanced authentication (JWT, OAuth, multi-tenant schemas), implement a cust | Variable | Description | Default | |----------|-------------|---------| -| `HINDSIGHT_API_GRAPH_RETRIEVER` | Graph retrieval algorithm: `link_expansion`, `mpfp`, or `bfs` | `link_expansion` | +| `HINDSIGHT_API_GRAPH_RETRIEVER` | Graph retrieval algorithm | `link_expansion` | | `HINDSIGHT_API_RECALL_MAX_CONCURRENT` | Max concurrent recall operations per worker (backpressure) | `32` | | `HINDSIGHT_API_RECALL_CONNECTION_BUDGET` | Max concurrent DB connections per recall operation | `4` | | `HINDSIGHT_API_RECALL_MAX_QUERY_TOKENS` | Maximum token length of a recall query; requests exceeding this limit are rejected with HTTP 400 | `500` | | `HINDSIGHT_API_RERANKER_MAX_CANDIDATES` | Max candidates to rerank per recall (RRF pre-filters the rest) | `300` | -| `HINDSIGHT_API_MPFP_TOP_K_NEIGHBORS` | Fan-out limit per node in MPFP graph traversal | `20` | | `HINDSIGHT_API_MENTAL_MODEL_REFRESH_CONCURRENCY` | Max concurrent mental model refreshes | `8` | | `HINDSIGHT_API_ENABLE_MENTAL_MODEL_HISTORY` | Track history of content changes to each mental model (previous content + timestamp). Disable to reduce storage if audit trails are not needed. | `true` | -#### Graph Retrieval Algorithms +#### Graph Retrieval Algorithm -- **`link_expansion`** (default): Fast, simple graph expansion from semantic seeds via entity co-occurrence and causal links. Target latency under 100ms. Recommended for most use cases. -- **`mpfp`**: Multi-Path Fact Propagation - iterative graph traversal with activation spreading. More thorough but slower. -- **`bfs`**: Breadth-first search from seed facts. Simple but less effective for large graphs. +- **`link_expansion`** (default): Fast graph expansion from semantic seeds via entity co-occurrence, semantic kNN, and causal links. Target latency under 100ms. ### Retain diff --git a/hindsight-docs/versioned_docs/version-0.3/developer/configuration.md b/hindsight-docs/versioned_docs/version-0.3/developer/configuration.md index 23ac70af..92423a61 100644 --- a/hindsight-docs/versioned_docs/version-0.3/developer/configuration.md +++ b/hindsight-docs/versioned_docs/version-0.3/developer/configuration.md @@ -291,15 +291,13 @@ For advanced authentication (JWT, OAuth, multi-tenant schemas), implement a cust | Variable | Description | Default | |----------|-------------|---------| -| `HINDSIGHT_API_GRAPH_RETRIEVER` | Graph retrieval algorithm: `link_expansion`, `mpfp`, or `bfs` | `link_expansion` | +| `HINDSIGHT_API_GRAPH_RETRIEVER` | Graph retrieval algorithm | `link_expansion` | | `HINDSIGHT_API_RECALL_MAX_CONCURRENT` | Max concurrent recall operations per worker (backpressure) | `32` | | `HINDSIGHT_API_RERANKER_MAX_CANDIDATES` | Max candidates to rerank per recall (RRF pre-filters the rest) | `300` | -#### Graph Retrieval Algorithms +#### Graph Retrieval Algorithm -- **`link_expansion`** (default): Fast, simple graph expansion from semantic seeds via entity co-occurrence and causal links. Target latency under 100ms. Recommended for most use cases. -- **`mpfp`**: Multi-Path Fact Propagation - iterative graph traversal with activation spreading. More thorough but slower. -- **`bfs`**: Breadth-first search from seed facts. Simple but less effective for large graphs. +- **`link_expansion`** (default): Fast graph expansion from semantic seeds via entity co-occurrence, semantic kNN, and causal links. Target latency under 100ms. ### Entity Observations diff --git a/hindsight-docs/versioned_docs/version-0.4/developer/configuration.md b/hindsight-docs/versioned_docs/version-0.4/developer/configuration.md index 477d68ad..2cd2e466 100644 --- a/hindsight-docs/versioned_docs/version-0.4/developer/configuration.md +++ b/hindsight-docs/versioned_docs/version-0.4/developer/configuration.md @@ -587,20 +587,17 @@ For advanced authentication (JWT, OAuth, multi-tenant schemas), implement a cust | Variable | Description | Default | |----------|-------------|---------| -| `HINDSIGHT_API_GRAPH_RETRIEVER` | Graph retrieval algorithm: `link_expansion`, `mpfp`, or `bfs` | `link_expansion` | +| `HINDSIGHT_API_GRAPH_RETRIEVER` | Graph retrieval algorithm | `link_expansion` | | `HINDSIGHT_API_RECALL_MAX_CONCURRENT` | Max concurrent recall operations per worker (backpressure) | `32` | | `HINDSIGHT_API_RECALL_CONNECTION_BUDGET` | Max concurrent DB connections per recall operation | `4` | | `HINDSIGHT_API_RECALL_MAX_QUERY_TOKENS` | Maximum token length of a recall query; requests exceeding this limit are rejected with HTTP 400 | `500` | | `HINDSIGHT_API_RERANKER_MAX_CANDIDATES` | Max candidates to rerank per recall (RRF pre-filters the rest) | `300` | -| `HINDSIGHT_API_MPFP_TOP_K_NEIGHBORS` | Fan-out limit per node in MPFP graph traversal | `20` | | `HINDSIGHT_API_MENTAL_MODEL_REFRESH_CONCURRENCY` | Max concurrent mental model refreshes | `8` | | `HINDSIGHT_API_ENABLE_MENTAL_MODEL_HISTORY` | Track history of content changes to each mental model (previous content + timestamp). Disable to reduce storage if audit trails are not needed. | `true` | -#### Graph Retrieval Algorithms +#### Graph Retrieval Algorithm -- **`link_expansion`** (default): Fast, simple graph expansion from semantic seeds via entity co-occurrence and causal links. Target latency under 100ms. Recommended for most use cases. -- **`mpfp`**: Multi-Path Fact Propagation - iterative graph traversal with activation spreading. More thorough but slower. -- **`bfs`**: Breadth-first search from seed facts. Simple but less effective for large graphs. +- **`link_expansion`** (default): Fast graph expansion from semantic seeds via entity co-occurrence, semantic kNN, and causal links. Target latency under 100ms. ### Retain diff --git a/skills/hindsight-docs/references/developer/configuration.md b/skills/hindsight-docs/references/developer/configuration.md index bda951df..24da2d29 100644 --- a/skills/hindsight-docs/references/developer/configuration.md +++ b/skills/hindsight-docs/references/developer/configuration.md @@ -587,20 +587,17 @@ For advanced authentication (JWT, OAuth, multi-tenant schemas), implement a cust | Variable | Description | Default | |----------|-------------|---------| -| `HINDSIGHT_API_GRAPH_RETRIEVER` | Graph retrieval algorithm: `link_expansion`, `mpfp`, or `bfs` | `link_expansion` | +| `HINDSIGHT_API_GRAPH_RETRIEVER` | Graph retrieval algorithm | `link_expansion` | | `HINDSIGHT_API_RECALL_MAX_CONCURRENT` | Max concurrent recall operations per worker (backpressure) | `32` | | `HINDSIGHT_API_RECALL_CONNECTION_BUDGET` | Max concurrent DB connections per recall operation | `4` | | `HINDSIGHT_API_RECALL_MAX_QUERY_TOKENS` | Maximum token length of a recall query; requests exceeding this limit are rejected with HTTP 400 | `500` | | `HINDSIGHT_API_RERANKER_MAX_CANDIDATES` | Max candidates to rerank per recall (RRF pre-filters the rest) | `300` | -| `HINDSIGHT_API_MPFP_TOP_K_NEIGHBORS` | Fan-out limit per node in MPFP graph traversal | `20` | | `HINDSIGHT_API_MENTAL_MODEL_REFRESH_CONCURRENCY` | Max concurrent mental model refreshes | `8` | | `HINDSIGHT_API_ENABLE_MENTAL_MODEL_HISTORY` | Track history of content changes to each mental model (previous content + timestamp). Disable to reduce storage if audit trails are not needed. | `true` | -#### Graph Retrieval Algorithms +#### Graph Retrieval Algorithm -- **`link_expansion`** (default): Fast, simple graph expansion from semantic seeds via entity co-occurrence and causal links. Target latency under 100ms. Recommended for most use cases. -- **`mpfp`**: Multi-Path Fact Propagation - iterative graph traversal with activation spreading. More thorough but slower. -- **`bfs`**: Breadth-first search from seed facts. Simple but less effective for large graphs. +- **`link_expansion`** (default): Fast graph expansion from semantic seeds via entity co-occurrence, semantic kNN, and causal links. Target latency under 100ms. ### Retain