diff --git a/hindsight-api/hindsight_api/engine/memory_engine.py b/hindsight-api/hindsight_api/engine/memory_engine.py index 464de44b..0b88a272 100644 --- a/hindsight-api/hindsight_api/engine/memory_engine.py +++ b/hindsight-api/hindsight_api/engine/memory_engine.py @@ -1203,49 +1203,57 @@ class MemoryEngine: temporal_info = f" | temporal_range={start_dt.strftime('%Y-%m-%d')} to {end_dt.strftime('%Y-%m-%d')}" log_buffer.append(f" [2] {total_retrievals}-way retrieval ({len(fact_type)} fact_types): {', '.join(timing_parts)} in {step_duration:.3f}s{temporal_info}") - # Record retrieval results for tracer (convert typed results to old format) + # Record retrieval results for tracer - per fact type if tracer: # Convert RetrievalResult to old tuple format for tracer def to_tuple_format(results): return [(r.id, r.__dict__) for r in results] - # Add semantic retrieval results - tracer.add_retrieval_results( - method_name="semantic", - results=to_tuple_format(semantic_results), - duration_seconds=aggregated_timings["semantic"], - score_field="similarity", - metadata={"limit": thinking_budget} - ) + # Add retrieval results per fact type (to show parallel execution in UI) + for idx, (ft_semantic, ft_bm25, ft_graph, ft_temporal, ft_timings, _) in enumerate(all_retrievals): + ft_name = fact_type[idx] if idx < len(fact_type) else "unknown" - # Add BM25 retrieval results - tracer.add_retrieval_results( - method_name="bm25", - results=to_tuple_format(bm25_results), - duration_seconds=aggregated_timings["bm25"], - score_field="bm25_score", - metadata={"limit": thinking_budget} - ) - - # Add graph retrieval results - tracer.add_retrieval_results( - method_name="graph", - results=to_tuple_format(graph_results), - duration_seconds=aggregated_timings["graph"], - score_field="similarity", # Graph uses similarity for activation - metadata={"budget": thinking_budget} - ) - - # Add temporal retrieval results if present - if temporal_results: + # Add semantic retrieval results for this fact type tracer.add_retrieval_results( - method_name="temporal", - results=to_tuple_format(temporal_results), - duration_seconds=aggregated_timings["temporal"], - score_field="temporal_score", - metadata={"budget": thinking_budget} + method_name="semantic", + results=to_tuple_format(ft_semantic), + duration_seconds=ft_timings.get("semantic", 0.0), + score_field="similarity", + metadata={"limit": thinking_budget}, + fact_type=ft_name ) + # Add BM25 retrieval results for this fact type + tracer.add_retrieval_results( + method_name="bm25", + results=to_tuple_format(ft_bm25), + duration_seconds=ft_timings.get("bm25", 0.0), + score_field="bm25_score", + metadata={"limit": thinking_budget}, + fact_type=ft_name + ) + + # Add graph retrieval results for this fact type + tracer.add_retrieval_results( + method_name="graph", + results=to_tuple_format(ft_graph), + duration_seconds=ft_timings.get("graph", 0.0), + score_field="activation", + metadata={"budget": thinking_budget}, + fact_type=ft_name + ) + + # Add temporal retrieval results for this fact type (even if empty, to show it ran) + if ft_temporal is not None: + tracer.add_retrieval_results( + method_name="temporal", + results=to_tuple_format(ft_temporal), + duration_seconds=ft_timings.get("temporal", 0.0), + score_field="temporal_score", + metadata={"budget": thinking_budget}, + fact_type=ft_name + ) + # Record entry points (from semantic results) for legacy graph view for rank, retrieval in enumerate(semantic_results[:10], start=1): # Top 10 as entry points tracer.add_entry_point(retrieval.id, retrieval.text, retrieval.similarity or 0.0, rank) @@ -1287,31 +1295,24 @@ class MemoryEngine: step_duration = time.time() - step_start log_buffer.append(f" [4] Reranking: {len(scored_results)} candidates scored in {step_duration:.3f}s") - if tracer: - # Convert to old format for tracer - results_dict = [sr.to_dict() for sr in scored_results] - tracer_merged = [(mc.id, mc.retrieval.__dict__, {"rrf_score": mc.rrf_score, **mc.source_ranks}) - for mc in merged_candidates] - tracer.add_reranked(results_dict, tracer_merged) - tracer.add_phase_metric("reranking", step_duration, { - "reranker_type": "cross-encoder", - "candidates_reranked": len(scored_results) - }) - # Step 4.5: Combine cross-encoder score with retrieval signals # This preserves retrieval work (RRF, temporal, recency) instead of pure cross-encoder ranking if scored_results: - # Normalize RRF scores to [0, 1] range + # Normalize RRF scores to [0, 1] range using min-max normalization rrf_scores = [sr.candidate.rrf_score for sr in scored_results] - max_rrf = max(rrf_scores) if rrf_scores else 1.0 + max_rrf = max(rrf_scores) if rrf_scores else 0.0 min_rrf = min(rrf_scores) if rrf_scores else 0.0 - rrf_range = max_rrf - min_rrf if max_rrf > min_rrf else 1.0 + rrf_range = max_rrf - min_rrf # Don't force to 1.0, let fallback handle it # Calculate recency based on occurred_start (more recent = higher score) now = utcnow() for sr in scored_results: - # Normalize RRF score - sr.rrf_normalized = (sr.candidate.rrf_score - min_rrf) / rrf_range if rrf_range > 0 else 0.5 + # Normalize RRF score (0-1 range, 0.5 if all same) + if rrf_range > 0: + sr.rrf_normalized = (sr.candidate.rrf_score - min_rrf) / rrf_range + else: + # All RRF scores are the same, use neutral value + sr.rrf_normalized = 0.5 # Calculate recency (decay over 365 days, minimum 0.1) sr.recency = 0.5 # default for missing dates @@ -1343,6 +1344,17 @@ class MemoryEngine: scored_results.sort(key=lambda x: x.weight, reverse=True) log_buffer.append(f" [4.6] Combined scoring: cross_encoder(0.6) + rrf(0.2) + temporal(0.1) + recency(0.1)") + # Add reranked results to tracer AFTER combined scoring (so normalized values are included) + if tracer: + results_dict = [sr.to_dict() for sr in scored_results] + tracer_merged = [(mc.id, mc.retrieval.__dict__, {"rrf_score": mc.rrf_score, **mc.source_ranks}) + for mc in merged_candidates] + tracer.add_reranked(results_dict, tracer_merged) + tracer.add_phase_metric("reranking", step_duration, { + "reranker_type": "cross-encoder", + "candidates_reranked": len(scored_results) + }) + # Step 5: Truncate to thinking_budget * 2 for token filtering rerank_limit = thinking_budget * 2 top_scored = scored_results[:rerank_limit] diff --git a/hindsight-api/hindsight_api/engine/search/__init__.py b/hindsight-api/hindsight_api/engine/search/__init__.py index aef63cd5..20d3c8a0 100644 --- a/hindsight-api/hindsight_api/engine/search/__init__.py +++ b/hindsight-api/hindsight_api/engine/search/__init__.py @@ -3,13 +3,23 @@ Search module for memory retrieval. Provides modular search architecture: - Retrieval: 4-way parallel (semantic + BM25 + graph + temporal) +- Graph retrieval: Pluggable strategies (BFS, PPR) - Reranking: Pluggable strategies (heuristic, cross-encoder) """ -from .retrieval import retrieve_parallel +from .retrieval import ( + retrieve_parallel, + get_default_graph_retriever, + set_default_graph_retriever, +) +from .graph_retrieval import GraphRetriever, BFSGraphRetriever from .reranking import CrossEncoderReranker __all__ = [ "retrieve_parallel", + "get_default_graph_retriever", + "set_default_graph_retriever", + "GraphRetriever", + "BFSGraphRetriever", "CrossEncoderReranker", ] diff --git a/hindsight-api/hindsight_api/engine/search/graph_retrieval.py b/hindsight-api/hindsight_api/engine/search/graph_retrieval.py new file mode 100644 index 00000000..2ebfa3cd --- /dev/null +++ b/hindsight-api/hindsight_api/engine/search/graph_retrieval.py @@ -0,0 +1,225 @@ +""" +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. +""" + +from abc import ABC, abstractmethod +from typing import List, Optional +from datetime import datetime +import logging + +from .types import RetrievalResult +from ..db_utils import acquire_with_retry + +logger = logging.getLogger(__name__) + + +class GraphRetriever(ABC): + """ + Abstract base class for graph-based memory retrieval. + + Implementations traverse the memory graph (entity links, temporal links, + causal links) to find relevant facts that might not be found by + semantic or keyword search alone. + """ + + @property + @abstractmethod + def name(self) -> str: + """Return identifier for this retrieval strategy (e.g., 'bfs', 'ppr').""" + pass + + @abstractmethod + async def retrieve( + self, + pool, + query_embedding_str: str, + bank_id: str, + fact_type: str, + budget: int, + query_text: Optional[str] = None, + ) -> List[RetrievalResult]: + """ + Retrieve relevant facts via graph traversal. + + Args: + pool: Database connection pool + query_embedding_str: Query embedding as string (for finding entry points) + bank_id: Memory bank identifier + fact_type: Fact type to filter ('world', 'experience', 'opinion', 'observation') + budget: Maximum number of nodes to explore/return + query_text: Original query text (optional, for some strategies) + + Returns: + List of RetrievalResult objects with activation scores set + """ + 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: Optional[str] = None, + ) -> List[RetrievalResult]: + """ + 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 + """ + async with acquire_with_retry(pool) as conn: + return await self._retrieve_with_conn( + conn, query_embedding_str, bank_id, fact_type, budget + ) + + async def _retrieve_with_conn( + self, + conn, + query_embedding_str: str, + bank_id: str, + fact_type: str, + budget: int, + ) -> List[RetrievalResult]: + """Internal implementation with connection.""" + + # Step 1: Find entry points + entry_points = await conn.fetch( + """ + SELECT id, text, context, event_date, occurred_start, occurred_end, + mentioned_at, access_count, embedding, fact_type, document_id, chunk_id, + 1 - (embedding <=> $1::vector) AS similarity + FROM memory_units + WHERE bank_id = $2 + AND embedding IS NOT NULL + AND fact_type = $3 + AND (1 - (embedding <=> $1::vector)) >= $4 + ORDER BY embedding <=> $1::vector + LIMIT $5 + """, + query_embedding_str, bank_id, fact_type, + self.entry_point_threshold, self.entry_point_limit + ) + + if not entry_points: + return [] + + # 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( + """ + SELECT mu.id, mu.text, mu.context, mu.occurred_start, mu.occurred_end, + mu.mentioned_at, mu.access_count, mu.embedding, mu.fact_type, + mu.document_id, mu.chunk_id, + ml.weight, ml.link_type, ml.from_unit_id + FROM memory_links ml + JOIN 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)) + + return results diff --git a/hindsight-api/hindsight_api/engine/search/retrieval.py b/hindsight-api/hindsight_api/engine/search/retrieval.py index f2784f35..3fa8e142 100644 --- a/hindsight-api/hindsight_api/engine/search/retrieval.py +++ b/hindsight-api/hindsight_api/engine/search/retrieval.py @@ -4,7 +4,7 @@ Retrieval module for 4-way parallel search. Implements: 1. Semantic retrieval (vector similarity) 2. BM25 retrieval (keyword/full-text search) -3. Graph retrieval (spreading activation) +3. Graph retrieval (via pluggable GraphRetriever interface) 4. Temporal retrieval (time-aware search with spreading) """ @@ -13,6 +13,24 @@ from datetime import datetime import asyncio from ..db_utils import acquire_with_retry from .types import RetrievalResult +from .graph_retrieval import GraphRetriever, BFSGraphRetriever + +# Default graph retriever instance (can be overridden) +_default_graph_retriever: Optional[GraphRetriever] = None + + +def get_default_graph_retriever() -> GraphRetriever: + """Get or create the default graph retriever.""" + global _default_graph_retriever + if _default_graph_retriever is None: + _default_graph_retriever = BFSGraphRetriever() + return _default_graph_retriever + + +def set_default_graph_retriever(retriever: GraphRetriever) -> None: + """Set the default graph retriever (for configuration/testing).""" + global _default_graph_retriever + _default_graph_retriever = retriever async def retrieve_semantic( @@ -105,121 +123,6 @@ async def retrieve_bm25( return [RetrievalResult.from_db_row(dict(r)) for r in results] -async def retrieve_graph( - conn, - query_emb_str: str, - bank_id: str, - fact_type: str, - budget: int -) -> List[RetrievalResult]: - """ - Graph retrieval via spreading activation. - - Args: - conn: Database connection - query_emb_str: Query embedding as string - agent_id: bank ID - fact_type: Fact type to filter - budget: Node budget for graph traversal - - Returns: - List of RetrievalResult objects - """ - # Find entry points - entry_points = await conn.fetch( - """ - SELECT id, text, context, event_date, occurred_start, occurred_end, mentioned_at, access_count, embedding, fact_type, document_id, chunk_id, - 1 - (embedding <=> $1::vector) AS similarity - FROM memory_units - WHERE bank_id = $2 - AND embedding IS NOT NULL - AND fact_type = $3 - AND (1 - (embedding <=> $1::vector)) >= 0.5 - ORDER BY embedding <=> $1::vector - LIMIT 5 - """, - query_emb_str, bank_id, fact_type - ) - - if not entry_points: - return [] - - # BFS-style spreading activation with batched neighbor fetching - visited = set() - results = [] - queue = [(RetrievalResult.from_db_row(dict(r)), r["similarity"]) for r in entry_points] - budget_remaining = budget - - # Process nodes in batches to reduce DB roundtrips - batch_size = 20 # Fetch neighbors for up to 20 nodes at once - - while queue and budget_remaining > 0: - # Collect a batch of nodes to process - batch_nodes = [] - batch_activations = {} - - while queue and len(batch_nodes) < 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 - results.append(current) - batch_nodes.append(current.id) - batch_activations[unit_id] = activation - - # Batch fetch neighbors for all nodes in this batch - # Fetch top weighted neighbors (batch_size * 20 = ~400 for good distribution) - if batch_nodes and budget_remaining > 0: - max_neighbors = len(batch_nodes) * 20 - neighbors = await conn.fetch( - """ - SELECT mu.id, mu.text, mu.context, mu.occurred_start, mu.occurred_end, mu.mentioned_at, - mu.access_count, mu.embedding, mu.fact_type, mu.document_id, mu.chunk_id, - ml.weight, ml.link_type, ml.from_unit_id - FROM memory_links ml - JOIN memory_units mu ON ml.to_unit_id = mu.id - WHERE ml.from_unit_id = ANY($1::uuid[]) - AND ml.weight >= 0.1 - AND mu.fact_type = $2 - ORDER BY ml.weight DESC - LIMIT $3 - """, - batch_nodes, fact_type, max_neighbors - ) - - for n in neighbors: - neighbor_id = str(n["id"]) - if neighbor_id not in visited: - # Get parent activation - parent_id = str(n["from_unit_id"]) - activation = batch_activations.get(parent_id, 0.5) - - # Boost activation for causal links (they're high-value relationships) - link_type = n["link_type"] - base_weight = n["weight"] - - # Causal links get 1.5-2.0x boost depending on type - if link_type in ("causes", "caused_by"): - # Direct causation - very strong relationship - causal_boost = 2.0 - elif link_type in ("enables", "prevents"): - # Conditional causation - strong but not as direct - causal_boost = 1.5 - else: - # Temporal, semantic, entity links - standard weight - causal_boost = 1.0 - - effective_weight = base_weight * causal_boost - new_activation = activation * effective_weight * 0.8 - if new_activation > 0.1: - neighbor_result = RetrievalResult.from_db_row(dict(n)) - queue.append((neighbor_result, new_activation)) - - return results - - async def retrieve_temporal( conn, query_emb_str: str, @@ -419,7 +322,8 @@ async def retrieve_parallel( fact_type: str, thinking_budget: int, question_date: Optional[datetime] = None, - query_analyzer: Optional["QueryAnalyzer"] = None + query_analyzer: Optional["QueryAnalyzer"] = None, + graph_retriever: Optional[GraphRetriever] = None, ) -> Tuple[List[RetrievalResult], List[RetrievalResult], List[RetrievalResult], Optional[List[RetrievalResult]], Dict[str, float], Optional[Tuple[datetime, datetime]]]: """ Run 3-way or 4-way parallel retrieval (adds temporal if detected). @@ -428,11 +332,12 @@ async def retrieve_parallel( pool: Database connection pool query_text: Query text query_embedding_str: Query embedding as string - agent_id: bank ID + bank_id: Bank ID fact_type: Fact type to filter thinking_budget: Budget for graph traversal and retrieval limits question_date: Optional date when question was asked (for temporal filtering) query_analyzer: Query analyzer to use (defaults to TransformerQueryAnalyzer) + graph_retriever: Graph retrieval strategy (defaults to BFSGraphRetriever) Returns: Tuple of (semantic_results, bm25_results, graph_results, temporal_results, timings, temporal_constraint) @@ -449,6 +354,9 @@ async def retrieve_parallel( query_text, reference_date=question_date, analyzer=query_analyzer ) + # Use provided graph retriever or default + retriever = graph_retriever or get_default_graph_retriever() + # Wrapper to track timing for each retrieval method async def timed_retrieval(name: str, coro): start = time.time() @@ -465,8 +373,14 @@ async def retrieve_parallel( return await retrieve_bm25(conn, query_text, bank_id, fact_type, limit=thinking_budget) async def run_graph(): - async with acquire_with_retry(pool) as conn: - return await retrieve_graph(conn, query_embedding_str, bank_id, fact_type, budget=thinking_budget) + return await retriever.retrieve( + pool=pool, + query_embedding_str=query_embedding_str, + bank_id=bank_id, + fact_type=fact_type, + budget=thinking_budget, + query_text=query_text, + ) async def run_temporal(start_date, end_date): async with acquire_with_retry(pool) as conn: diff --git a/hindsight-api/hindsight_api/engine/search/trace.py b/hindsight-api/hindsight_api/engine/search/trace.py index f3fcef85..959ec93b 100644 --- a/hindsight-api/hindsight_api/engine/search/trace.py +++ b/hindsight-api/hindsight_api/engine/search/trace.py @@ -108,6 +108,7 @@ class RetrievalResult(BaseModel): class RetrievalMethodResults(BaseModel): """Results from a single retrieval method.""" method_name: Literal["semantic", "bm25", "graph", "temporal"] = Field(description="Name of retrieval method") + fact_type: Optional[str] = Field(default=None, description="Fact type this retrieval was for (world, experience, opinion)") results: List[RetrievalResult] = Field(description="Retrieved results with ranks") duration_seconds: float = Field(description="Time taken for this retrieval") metadata: Dict[str, Any] = Field(default_factory=dict, description="Method-specific metadata") diff --git a/hindsight-api/hindsight_api/engine/search/tracer.py b/hindsight-api/hindsight_api/engine/search/tracer.py index cb2b8bf7..8d4312a6 100644 --- a/hindsight-api/hindsight_api/engine/search/tracer.py +++ b/hindsight-api/hindsight_api/engine/search/tracer.py @@ -289,7 +289,8 @@ class SearchTracer: results: List[tuple], # List of (doc_id, data) tuples duration_seconds: float, score_field: str, # e.g., "similarity", "bm25_score" - metadata: Optional[Dict[str, Any]] = None + metadata: Optional[Dict[str, Any]] = None, + fact_type: Optional[str] = None ): """ Record results from a single retrieval method. @@ -300,6 +301,7 @@ class SearchTracer: duration_seconds: Time taken for this retrieval score_field: Field name containing the score in data dict metadata: Optional metadata about this retrieval method + fact_type: Fact type this retrieval was for (world, experience, opinion) """ retrieval_results = [] for rank, (doc_id, data) in enumerate(results, start=1): @@ -313,7 +315,7 @@ class SearchTracer: text=data.get("text", ""), context=data.get("context", ""), event_date=data.get("event_date"), - fact_type=data.get("fact_type"), + fact_type=data.get("fact_type") or fact_type, score=score, score_name=score_field, ) @@ -322,6 +324,7 @@ class SearchTracer: self.retrieval_results.append( RetrievalMethodResults( method_name=method_name, + fact_type=fact_type, results=retrieval_results, duration_seconds=duration_seconds, metadata=metadata or {}, @@ -367,8 +370,10 @@ class SearchTracer: rank_change = rrf_rank - rank # Positive = moved up # Extract score components (only include non-None values) + # Keys from ScoredResult.to_dict(): cross_encoder_score, cross_encoder_score_normalized, + # rrf_normalized, temporal, recency, combined_score, weight score_components = {} - for key in ["semantic_similarity", "bm25_score", "rrf_score", "recency_normalized", "frequency_normalized", "cross_encoder_score", "cross_encoder_score_normalized"]: + for key in ["cross_encoder_score", "cross_encoder_score_normalized", "rrf_score", "rrf_normalized", "temporal", "recency", "combined_score"]: if key in result and result[key] is not None: score_components[key] = result[key] diff --git a/hindsight-api/hindsight_api/engine/search/types.py b/hindsight-api/hindsight_api/engine/search/types.py index 431ddecf..5e6c234d 100644 --- a/hindsight-api/hindsight_api/engine/search/types.py +++ b/hindsight-api/hindsight_api/engine/search/types.py @@ -31,8 +31,9 @@ class RetrievalResult: embedding: Optional[List[float]] = None # Retrieval-specific scores (only one will be set depending on retrieval method) - similarity: Optional[float] = None # Semantic/graph retrieval + similarity: Optional[float] = None # Semantic retrieval bm25_score: Optional[float] = None # BM25 retrieval + activation: Optional[float] = None # Graph retrieval (spreading activation) temporal_score: Optional[float] = None # Temporal retrieval temporal_proximity: Optional[float] = None # Temporal retrieval @@ -54,6 +55,7 @@ class RetrievalResult: embedding=row.get("embedding"), similarity=row.get("similarity"), bm25_score=row.get("bm25_score"), + activation=row.get("activation"), temporal_score=row.get("temporal_score"), temporal_proximity=row.get("temporal_proximity"), ) @@ -152,6 +154,7 @@ class ScoredResult: result["cross_encoder_score"] = self.cross_encoder_score result["cross_encoder_score_normalized"] = self.cross_encoder_score_normalized result["rrf_normalized"] = self.rrf_normalized + result["temporal"] = self.temporal result["recency"] = self.recency result["combined_score"] = self.combined_score result["weight"] = self.weight diff --git a/hindsight-api/tests/test_combined_scoring.py b/hindsight-api/tests/test_combined_scoring.py new file mode 100644 index 00000000..82f22c9e --- /dev/null +++ b/hindsight-api/tests/test_combined_scoring.py @@ -0,0 +1,323 @@ +""" +Tests for combined scoring functionality. + +Verifies that: +1. RRF scores are properly normalized to [0, 1] range +2. Combined scoring formula is applied correctly +3. Tracer captures normalized values (not raw values) +""" +import pytest +from datetime import datetime, timezone +from hindsight_api.engine.search.types import RetrievalResult, MergedCandidate, ScoredResult +from hindsight_api.engine.memory_engine import Budget + + +class TestRRFNormalization: + """Test that RRF scores are properly normalized.""" + + def test_rrf_normalized_range(self): + """RRF normalized values should be in [0, 1] range, not raw [0.04, 0.06].""" + # Simulate RRF scores like what we get from actual retrieval + raw_rrf_scores = [0.0607, 0.0550, 0.0480, 0.0390] + + max_rrf = max(raw_rrf_scores) + min_rrf = min(raw_rrf_scores) + rrf_range = max_rrf - min_rrf + + normalized = [] + for score in raw_rrf_scores: + if rrf_range > 0: + norm = (score - min_rrf) / rrf_range + else: + norm = 0.5 + normalized.append(norm) + + # Verify normalized values are in [0, 1] + for i, norm in enumerate(normalized): + assert 0.0 <= norm <= 1.0, f"Normalized RRF {norm} not in [0, 1] for raw {raw_rrf_scores[i]}" + + # Highest raw should be 1.0 + assert normalized[0] == 1.0, f"Highest RRF should normalize to 1.0, got {normalized[0]}" + + # Lowest raw should be 0.0 + assert normalized[-1] == 0.0, f"Lowest RRF should normalize to 0.0, got {normalized[-1]}" + + def test_rrf_all_same_scores(self): + """When all RRF scores are the same, normalized should be 0.5 (neutral).""" + raw_rrf_scores = [0.0500, 0.0500, 0.0500] + + max_rrf = max(raw_rrf_scores) + min_rrf = min(raw_rrf_scores) + rrf_range = max_rrf - min_rrf + + normalized = [] + for score in raw_rrf_scores: + if rrf_range > 0: + norm = (score - min_rrf) / rrf_range + else: + norm = 0.5 # Neutral value when all same + normalized.append(norm) + + # All should be 0.5 when scores are identical + for norm in normalized: + assert norm == 0.5, f"Expected 0.5 for identical scores, got {norm}" + + +class TestCombinedScoringFormula: + """Test that the combined scoring formula is applied correctly.""" + + def test_combined_score_calculation(self): + """Verify the weighted combination: 0.6*CE + 0.2*RRF + 0.1*temporal + 0.1*recency.""" + # Test case 1: All components at 1.0 + ce_norm = 1.0 + rrf_norm = 1.0 + temporal = 1.0 + recency = 1.0 + + expected = 0.6 * ce_norm + 0.2 * rrf_norm + 0.1 * temporal + 0.1 * recency + assert expected == 1.0, f"All 1.0 should give 1.0, got {expected}" + + # Test case 2: All components at 0.0 + ce_norm = 0.0 + rrf_norm = 0.0 + temporal = 0.0 + recency = 0.0 + + expected = 0.6 * ce_norm + 0.2 * rrf_norm + 0.1 * temporal + 0.1 * recency + assert expected == 0.0, f"All 0.0 should give 0.0, got {expected}" + + # Test case 3: High CE, low RRF (cross-encoder finds something retrieval missed) + ce_norm = 0.999 + rrf_norm = 0.0 # Lowest in set + temporal = 0.5 + recency = 0.5 + + expected = 0.6 * ce_norm + 0.2 * rrf_norm + 0.1 * temporal + 0.1 * recency + # 0.5994 + 0.0 + 0.05 + 0.05 = 0.6994 + assert abs(expected - 0.6994) < 0.001, f"Expected ~0.6994, got {expected}" + + # Test case 4: Medium CE, high RRF (retrieval consensus) + ce_norm = 0.8 + rrf_norm = 1.0 # Highest in set + temporal = 0.5 + recency = 0.5 + + expected = 0.6 * ce_norm + 0.2 * rrf_norm + 0.1 * temporal + 0.1 * recency + # 0.48 + 0.2 + 0.05 + 0.05 = 0.78 + assert abs(expected - 0.78) < 0.001, f"Expected ~0.78, got {expected}" + + def test_rrf_contribution_is_significant(self): + """Verify RRF actually contributes to the final score (not negligible).""" + # Same CE, different RRF + ce_norm = 0.8 + temporal = 0.5 + recency = 0.5 + + # Low RRF + score_low_rrf = 0.6 * ce_norm + 0.2 * 0.0 + 0.1 * temporal + 0.1 * recency + + # High RRF + score_high_rrf = 0.6 * ce_norm + 0.2 * 1.0 + 0.1 * temporal + 0.1 * recency + + # Difference should be 0.2 (20% contribution) + diff = score_high_rrf - score_low_rrf + assert abs(diff - 0.2) < 0.001, f"RRF should contribute 0.2 difference, got {diff}" + + +@pytest.mark.asyncio +async def test_trace_has_normalized_rrf(memory): + """Integration test: verify trace contains normalized RRF values, not raw.""" + bank_id = f"test_scoring_{datetime.now(timezone.utc).timestamp()}" + + try: + # Store multiple memories to ensure different RRF scores + await memory.retain_async( + bank_id=bank_id, + content="Python is a programming language created by Guido van Rossum", + context="tech facts", + ) + await memory.retain_async( + bank_id=bank_id, + content="JavaScript was created by Brendan Eich at Netscape", + context="tech facts", + ) + await memory.retain_async( + bank_id=bank_id, + content="The Eiffel Tower is located in Paris, France", + context="geography facts", + ) + await memory.retain_async( + bank_id=bank_id, + content="Mount Everest is the tallest mountain on Earth", + context="geography facts", + ) + + # Search with tracing + result = await memory.recall_async( + bank_id=bank_id, + query="programming languages", + fact_type=["world"], + budget=Budget.LOW, + max_tokens=1024, + enable_trace=True, + ) + + assert result.trace is not None, "Trace should be present" + trace = result.trace + + # Check reranked results have proper score_components + assert "reranked" in trace, "Trace should have reranked results" + assert len(trace["reranked"]) > 0, "Should have reranked results" + + has_valid_rrf = False + has_valid_temporal = False + has_valid_recency = False + + for r in trace["reranked"]: + sc = r.get("score_components", {}) + + # Check RRF normalized is present and in valid range + if "rrf_normalized" in sc: + rrf_norm = sc["rrf_normalized"] + assert 0.0 <= rrf_norm <= 1.0, f"rrf_normalized {rrf_norm} should be in [0, 1]" + # Should NOT be raw RRF score (which would be ~0.04-0.06) + # A normalized value of exactly 0.0 or 1.0 is valid (min/max of set) + # But raw scores like 0.0607 should never appear as normalized + if rrf_norm > 0.1: # Any value > 0.1 is likely properly normalized + has_valid_rrf = True + + # Check temporal is present and in valid range + if "temporal" in sc: + temporal = sc["temporal"] + assert 0.0 <= temporal <= 1.0, f"temporal {temporal} should be in [0, 1]" + has_valid_temporal = True + + # Check recency is present and in valid range + if "recency" in sc: + recency = sc["recency"] + assert 0.0 <= recency <= 1.0, f"recency {recency} should be in [0, 1]" + has_valid_recency = True + + # At least some results should have these components + # (might not have rrf > 0.1 if all scores are same, which is fine) + assert has_valid_temporal, "Should have temporal scores in trace" + assert has_valid_recency, "Should have recency scores in trace" + + print("\n✓ Combined scoring trace test passed!") + print(f" - Reranked results: {len(trace['reranked'])}") + if trace["reranked"]: + sc = trace["reranked"][0].get("score_components", {}) + print(f" - First result score components: {sc}") + + finally: + await memory.delete_bank(bank_id) + + +@pytest.mark.asyncio +async def test_rrf_normalized_not_raw_in_trace(memory): + """Verify that raw RRF scores (0.04-0.06 range) don't appear as normalized values.""" + bank_id = f"test_rrf_raw_{datetime.now(timezone.utc).timestamp()}" + + try: + # Store enough memories to get varied RRF scores + for i in range(5): + await memory.retain_async( + bank_id=bank_id, + content=f"Test fact number {i} about various topics", + context="test context", + ) + + result = await memory.recall_async( + bank_id=bank_id, + query="test fact", + fact_type=["world"], + budget=Budget.LOW, + max_tokens=512, + enable_trace=True, + ) + + trace = result.trace + assert trace is not None + + # Check that rrf_normalized values are NOT in the raw range + raw_rrf_range = (0.01, 0.08) # Raw RRF scores are typically in this range + + for r in trace.get("reranked", []): + sc = r.get("score_components", {}) + + if "rrf_normalized" in sc and "rrf_score" in sc: + rrf_norm = sc["rrf_normalized"] + rrf_raw = sc["rrf_score"] + + # Raw should be in the typical range + assert raw_rrf_range[0] <= rrf_raw <= raw_rrf_range[1], \ + f"Raw RRF {rrf_raw} should be in typical range {raw_rrf_range}" + + # Normalized should either be: + # - 0.0 (min in set) + # - 1.0 (max in set) + # - 0.5 (all same) + # - Something in between (0.0 to 1.0) + # But NOT the same as raw (which would indicate no normalization) + if len(trace["reranked"]) > 1: + # If we have multiple results, normalized should differ from raw + # (unless by coincidence, which is very unlikely) + assert rrf_norm != rrf_raw, \ + f"Normalized RRF ({rrf_norm}) should differ from raw ({rrf_raw})" + + print("\n✓ RRF raw vs normalized test passed!") + + finally: + await memory.delete_bank(bank_id) + + +@pytest.mark.asyncio +async def test_combined_score_matches_components(memory): + """Verify the final score actually equals the weighted sum of components.""" + bank_id = f"test_combined_{datetime.now(timezone.utc).timestamp()}" + + try: + await memory.retain_async( + bank_id=bank_id, + content="The quick brown fox jumps over the lazy dog", + context="test", + ) + await memory.retain_async( + bank_id=bank_id, + content="A quick test of the emergency broadcast system", + context="test", + ) + + result = await memory.recall_async( + bank_id=bank_id, + query="quick test", + fact_type=["world"], + budget=Budget.LOW, + max_tokens=512, + enable_trace=True, + ) + + trace = result.trace + assert trace is not None + + for r in trace.get("reranked", []): + sc = r.get("score_components", {}) + final_score = r.get("rerank_score", 0) + + # Get components (use defaults if missing) + ce = sc.get("cross_encoder_score_normalized", 0) + rrf = sc.get("rrf_normalized", 0.5) + tmp = sc.get("temporal", 0.5) + rec = sc.get("recency", 0.5) + + # Calculate expected score + expected = 0.6 * ce + 0.2 * rrf + 0.1 * tmp + 0.1 * rec + + # Allow small floating point difference + assert abs(final_score - expected) < 0.01, \ + f"Final score {final_score} doesn't match expected {expected} from components" + + print("\n✓ Combined score verification test passed!") + + finally: + await memory.delete_bank(bank_id) diff --git a/hindsight-control-plane/src/app/layout.tsx b/hindsight-control-plane/src/app/layout.tsx index 01ae1415..ddefc242 100644 --- a/hindsight-control-plane/src/app/layout.tsx +++ b/hindsight-control-plane/src/app/layout.tsx @@ -18,7 +18,7 @@ export default function RootLayout({ }>) { return ( -
+{selectedDocument.id}
{selectedDocument.original_text}
+ {selectedDocument.original_text}