docs: add hybrid search documentation (#671)

docs/hybrid-search.md

@@ -0,0 +1,54 @@
+# Hybrid Search Router
+
+Combines three search methods with query-type routing and Reciprocal Rank Fusion (RRF).
+
+## Architecture
+
+```
+Query → analyze_query() → QueryType
+                              │
+        ┌─────────────────────┼─────────────────────┐
+        ▼                     ▼                     ▼
+    FTS5 (keyword)      Qdrant (semantic)     HRR (compositional)
+        │                     │                     │
+        └─────────────────────┼─────────────────────┘
+                              ▼
+                    Reciprocal Rank Fusion
+                              ▼
+                      Merged Results
+```
+
+## Query Types
+
+| Type | Detection | Backend | Example |
+|------|-----------|---------|---------|
+| `keyword` | Identifiers, quoted terms, short queries | FTS5 | `function_name`, `"exact match"` |
+| `semantic` | Questions, "how/why/what" patterns | Qdrant | `What did we discuss about X?` |
+| `compositional` | Contradiction, related, entity queries | HRR | `Are there contradictions?` |
+| `hybrid` | No strong signals or mixed signals | All three | `deployment process` |
+
+## Usage
+
+```python
+# Automatic routing
+results = hybrid_engine.search("What did we decide about deploy?")
+# → Routes to semantic (Qdrant) + HRR, merges with RRF
+
+results = hybrid_engine.search("function_name")
+# → Routes to keyword (FTS5)
+
+# Manual query type override (future)
+results = hybrid_engine.search("deploy", force_type=QueryType.KEYWORD)
+```
+
+## RRF Parameters
+
+- **k=60**: Standard RRF constant (Cormack et al., 2009)
+- **Weights**: Qdrant gets 1.2x boost (semantic results tend to be more relevant)
+- **Fetch limit**: Each backend returns 3x the requested limit for merge headroom
+
+## Graceful Degradation
+
+- **Qdrant unavailable**: Falls back to FTS5 + HRR only
+- **HRR unavailable** (no numpy): Falls back to FTS5 + Qdrant
+- **All backends fail**: Falls back to existing `retriever.search()`

plugins/memory/holographic/hybrid_search.py

@@ -0,0 +1,277 @@
+"""Hybrid search engine with Reciprocal Rank Fusion.
+
+Combines results from multiple search backends:
+  - FTS5 (keyword search via SQLite full-text index)
+  - Qdrant (semantic search via vector similarity)
+  - HRR (compositional search via holographic reduced representations)
+
+Uses Reciprocal Rank Fusion (RRF) to merge ranked lists into a single
+result set. RRF is simple, parameter-free, and consistently outperforms
+individual rankers.
+
+RRF formula: score(d) = sum over rankers r of 1/(k + rank_r(d))
+where k=60 (standard constant from Cormack et al., 2009).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional
+
+from .query_router import QueryType, QueryAnalysis, analyze_query
+
+logger = logging.getLogger(__name__)
+
+# RRF constant — standard value from the literature
+_RRF_K = 60
+
+
+@dataclass
+class SearchResult:
+    """A single search result with source tracking."""
+    fact_id: int
+    content: str
+    score: float
+    source: str  # "fts5", "qdrant", "hrr"
+    rank: int    # rank in source's list
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+def reciprocal_rank_fusion(
+    ranked_lists: List[List[SearchResult]],
+    k: int = _RRF_K,
+    weights: Optional[Dict[str, float]] = None,
+) -> List[SearchResult]:
+    """Merge multiple ranked lists using Reciprocal Rank Fusion.
+
+    Args:
+        ranked_lists: List of ranked result lists from different sources.
+        k: RRF constant (default 60).
+        weights: Optional per-source weights. Default: all 1.0.
+
+    Returns:
+        Merged and re-ranked list of SearchResults.
+    """
+    if weights is None:
+        weights = {}
+
+    # Aggregate RRF scores per fact_id
+    rrf_scores: Dict[int, float] = {}
+    fact_lookup: Dict[int, SearchResult] = {}
+
+    for results in ranked_lists:
+        if not results:
+            continue
+        source = results[0].source if results else "unknown"
+        w = weights.get(source, 1.0)
+
+        for rank, result in enumerate(results, 1):
+            fid = result.fact_id
+            contribution = w / (k + rank)
+            rrf_scores[fid] = rrf_scores.get(fid, 0.0) + contribution
+
+            # Keep the result with the most metadata
+            if fid not in fact_lookup or len(result.metadata) > len(fact_lookup[fid].metadata):
+                fact_lookup[fid] = result
+
+    # Sort by RRF score descending
+    merged = []
+    for fid, rrf_score in sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True):
+        result = fact_lookup[fid]
+        result.score = rrf_score
+        merged.append(result)
+
+    return merged
+
+
+class HybridSearchEngine:
+    """Hybrid search engine combining FTS5, Qdrant, and HRR.
+
+    Routes queries through the query analyzer, dispatches to appropriate
+    backends, and merges results with RRF.
+    """
+
+    def __init__(self, store, retriever, qdrant_client=None):
+        self._store = store
+        self._retriever = retriever
+        self._qdrant = qdrant_client
+
+    def search(
+        self,
+        query: str,
+        category: str | None = None,
+        min_trust: float = 0.3,
+        limit: int = 10,
+    ) -> List[dict]:
+        """Hybrid search with query routing and RRF merge.
+
+        Analyzes the query, dispatches to appropriate backends,
+        merges results, and returns the top `limit` results.
+        """
+        # Step 1: Analyze query type
+        analysis = analyze_query(query)
+        logger.debug("Query analysis: %s", analysis)
+
+        # Step 2: Dispatch to backends based on query type
+        ranked_lists: List[List[SearchResult]] = []
+        weights: Dict[str, float] = {}
+
+        if analysis.query_type in (QueryType.KEYWORD, QueryType.HYBRID):
+            fts_results = self._search_fts5(query, category, min_trust, limit * 3)
+            if fts_results:
+                ranked_lists.append(fts_results)
+                weights["fts5"] = 1.0
+
+        if analysis.query_type in (QueryType.SEMANTIC, QueryType.HYBRID):
+            qdrant_results = self._search_qdrant(query, category, min_trust, limit * 3)
+            if qdrant_results:
+                ranked_lists.append(qdrant_results)
+                weights["qdrant"] = 1.2  # Slight boost for semantic search
+
+        if analysis.query_type in (QueryType.COMPOSITIONAL, QueryType.HYBRID):
+            hrr_results = self._search_hrr(query, category, min_trust, limit * 3)
+            if hrr_results:
+                ranked_lists.append(hrr_results)
+                weights["hrr"] = 1.0
+
+        # Step 3: Merge with RRF
+        if not ranked_lists:
+            # Fallback to existing search if no backends returned results
+            return self._retriever.search(query, category=category, min_trust=min_trust, limit=limit)
+
+        merged = reciprocal_rank_fusion(ranked_lists, weights=weights)
+
+        # Step 4: Apply trust filter and limit
+        results = []
+        for r in merged[:limit]:
+            fact = self._store.get_fact(r.fact_id)
+            if fact and fact.get("trust_score", 0) >= min_trust:
+                fact["score"] = r.score
+                fact["search_source"] = r.source
+                fact.pop("hrr_vector", None)
+                results.append(fact)
+
+        return results
+
+    def _search_fts5(
+        self, query: str, category: str | None, min_trust: float, limit: int
+    ) -> List[SearchResult]:
+        """Search using SQLite FTS5 full-text index."""
+        try:
+            raw = self._retriever._fts_candidates(query, category, min_trust, limit)
+            return [
+                SearchResult(
+                    fact_id=f["fact_id"],
+                    content=f.get("content", ""),
+                    score=f.get("fts_rank", 0.0),
+                    source="fts5",
+                    rank=i + 1,
+                    metadata={"category": f.get("category", "")},
+                )
+                for i, f in enumerate(raw)
+            ]
+        except Exception as e:
+            logger.debug("FTS5 search failed: %s", e)
+            return []
+
+    def _search_qdrant(
+        self, query: str, category: str | None, min_trust: float, limit: int
+    ) -> List[SearchResult]:
+        """Search using Qdrant vector similarity.
+
+        If Qdrant is not available, returns empty list (graceful degradation).
+        """
+        if not self._qdrant:
+            return []
+
+        try:
+            from qdrant_client import models
+
+            # Build filter
+            filters = []
+            if category:
+                filters.append(
+                    models.FieldCondition(
+                        key="category",
+                        match=models.MatchValue(value=category),
+                    )
+                )
+            if min_trust > 0:
+                filters.append(
+                    models.FieldCondition(
+                        key="trust_score",
+                        range=models.Range(gte=min_trust),
+                    )
+                )
+
+            query_filter = models.Filter(must=filters) if filters else None
+
+            results = self._qdrant.query_points(
+                collection_name="hermes_facts",
+                query=query,  # Qdrant handles embedding
+                limit=limit,
+                query_filter=query_filter,
+            )
+
+            return [
+                SearchResult(
+                    fact_id=int(r.id),
+                    content=r.payload.get("content", ""),
+                    score=r.score,
+                    source="qdrant",
+                    rank=i + 1,
+                    metadata=r.payload,
+                )
+                for i, r in enumerate(results.points)
+            ]
+        except Exception as e:
+            logger.debug("Qdrant search failed: %s", e)
+            return []
+
+    def _search_hrr(
+        self, query: str, category: str | None, min_trust: float, limit: int
+    ) -> List[SearchResult]:
+        """Search using HRR compositional vectors."""
+        try:
+            import plugins.memory.holographic.holographic as hrr
+            if not hrr._HAS_NUMPY:
+                return []
+
+            conn = self._store._conn
+            query_vec = hrr.encode_text(query, dim=1024)
+
+            where = "WHERE hrr_vector IS NOT NULL"
+            params: list = []
+            if category:
+                where += " AND category = ?"
+                params.append(category)
+
+            rows = conn.execute(
+                f"SELECT fact_id, content, trust_score, hrr_vector FROM facts {where}",
+                params,
+            ).fetchall()
+
+            scored = []
+            for row in rows:
+                if row["trust_score"] < min_trust:
+                    continue
+                fact_vec = hrr.bytes_to_phases(row["hrr_vector"])
+                sim = hrr.similarity(query_vec, fact_vec)
+                scored.append((row["fact_id"], row["content"], sim))
+
+            scored.sort(key=lambda x: x[2], reverse=True)
+
+            return [
+                SearchResult(
+                    fact_id=fid,
+                    content=content,
+                    score=sim,
+                    source="hrr",
+                    rank=i + 1,
+                )
+                for i, (fid, content, sim) in enumerate(scored[:limit])
+            ]
+        except Exception as e:
+            logger.debug("HRR search failed: %s", e)
+            return []

plugins/memory/holographic/query_router.py

@@ -0,0 +1,168 @@
+"""Query type detection and routing for hybrid search.
+
+Analyzes the incoming query to determine which search methods should be used,
+then dispatches to the appropriate backends (FTS5, Qdrant, HRR).
+
+Query types:
+  - keyword: Exact term matching → FTS5
+  - semantic: Natural language concepts → Qdrant
+  - compositional: Entity relationships, contradictions → HRR
+  - hybrid: Multiple types → all methods + RRF merge
+"""
+
+from __future__ import annotations
+
+import re
+import logging
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import List, Optional, Set
+
+logger = logging.getLogger(__name__)
+
+
+class QueryType(Enum):
+    """Detected query type determines which search methods to use."""
+    KEYWORD = "keyword"         # Exact terms → FTS5
+    SEMANTIC = "semantic"       # Natural language → Qdrant
+    COMPOSITIONAL = "compositional"  # Entity relationships → HRR
+    HYBRID = "hybrid"           # Multiple types → all methods
+
+
+@dataclass
+class QueryAnalysis:
+    """Result of query analysis."""
+    query_type: QueryType
+    confidence: float
+    signals: List[str] = field(default_factory=list)
+    entities: List[str] = field(default_factory=list)
+    keywords: List[str] = field(default_factory=list)
+
+    def __repr__(self) -> str:
+        return f"QueryAnalysis(type={self.query_type.value}, conf={self.confidence:.2f}, signals={self.signals})"
+
+
+# Patterns that indicate compositional queries
+_COMPOSITIONAL_PATTERNS = [
+    re.compile(r"\b(contradiction|contradict|conflicting|conflicts)\b", re.I),
+    re.compile(r"\b(related to|connects to|links to|associated with)\b", re.I),
+    re.compile(r"\b(what does .* know about|tell me about .* entity|facts about .*)\b", re.I),
+    re.compile(r"\b(shared|common|overlap)\b.*\b(entities|concepts|topics)\b", re.I),
+    re.compile(r"\b(probe|entity|entities)\b", re.I),
+]
+
+# Patterns that indicate keyword queries
+_KEYWORD_SIGNALS = [
+    re.compile(r"^[a-z_][a-z0-9_.]+$", re.I),  # Single identifier: function_name, Class.method
+    re.compile(r"\b(find|search|locate|grep|where)\b.*\b(exact|specific|literal)\b", re.I),
+    re.compile(r"["\']([^"\']+)["\']"),  # Quoted exact terms
+    re.compile(r"^[A-Z_]{2,}$"),  # ALL_CAPS constants
+    re.compile(r"\b\w+\.\w+\.\w+\b"),  # Dotted paths: module.sub.func
+]
+
+# Patterns that indicate semantic queries
+_SEMANTIC_SIGNALS = [
+    re.compile(r"\b(what did|how does|why is|explain|describe|summarize|discuss)\b", re.I),
+    re.compile(r"\b(remember|recall|think|know|understand)\b.*\b(about|regarding)\b", re.I),
+    re.compile(r"\?$"),  # Questions
+    re.compile(r"\b(the best way to|how to|what\'s the|approach to)\b", re.I),
+]
+
+
+def analyze_query(query: str) -> QueryAnalysis:
+    """Analyze a query to determine which search methods to use.
+
+    Returns QueryAnalysis with detected type, confidence, and extracted signals.
+    """
+    if not query or not query.strip():
+        return QueryAnalysis(
+            query_type=QueryType.HYBRID,
+            confidence=0.5,
+            signals=["empty_query"],
+        )
+
+    query = query.strip()
+
+    # Score each query type
+    comp_score = 0.0
+    kw_score = 0.0
+    sem_score = 0.0
+    signals = []
+    entities = []
+    keywords = []
+
+    # Check compositional patterns
+    for pattern in _COMPOSITIONAL_PATTERNS:
+        if pattern.search(query):
+            comp_score += 0.3
+            signals.append(f"compositional:{pattern.pattern[:30]}")
+
+    # Check keyword patterns
+    for pattern in _KEYWORD_SIGNALS:
+        if pattern.search(query):
+            kw_score += 0.25
+            match = pattern.search(query)
+            if match:
+                keywords.append(match.group(0))
+            signals.append(f"keyword:{pattern.pattern[:30]}")
+
+    # Check semantic patterns
+    for pattern in _SEMANTIC_SIGNALS:
+        if pattern.search(query):
+            sem_score += 0.25
+            signals.append(f"semantic:{pattern.pattern[:30]}")
+
+    # Extract entities (capitalized multi-word phrases, quoted terms)
+    entity_patterns = [
+        re.compile(r"\b([A-Z][a-z]+(?:\s+[A-Z][a-z]+)+)\b"),
+        re.compile(r"["\']([^"\']+)["\']"),
+    ]
+    for ep in entity_patterns:
+        for m in ep.finditer(query):
+            entities.append(m.group(1))
+
+    # Short queries (< 5 words) with no semantic signals → keyword
+    word_count = len(query.split())
+    if word_count <= 4 and sem_score == 0 and comp_score == 0:
+        kw_score += 0.3
+        signals.append("short_query_keyword_boost")
+
+    # Normalize scores
+    max_score = max(comp_score, kw_score, sem_score, 0.1)
+
+    # Determine query type
+    if max_score < 0.15:
+        # No strong signals → use hybrid (all methods)
+        return QueryAnalysis(
+            query_type=QueryType.HYBRID,
+            confidence=0.5,
+            signals=["no_strong_signals"],
+            entities=entities,
+            keywords=keywords,
+        )
+
+    if comp_score == max_score and comp_score >= 0.3:
+        return QueryAnalysis(
+            query_type=QueryType.COMPOSITIONAL,
+            confidence=min(comp_score, 1.0),
+            signals=signals,
+            entities=entities,
+            keywords=keywords,
+        )
+
+    if kw_score > sem_score:
+        return QueryAnalysis(
+            query_type=QueryType.KEYWORD,
+            confidence=min(kw_score, 1.0),
+            signals=signals,
+            entities=entities,
+            keywords=keywords,
+        )
+
+    return QueryAnalysis(
+        query_type=QueryType.SEMANTIC,
+        confidence=min(sem_score, 1.0),
+        signals=signals,
+        entities=entities,
+        keywords=keywords,
+    )

tests/test_hybrid_search.py

@@ -0,0 +1,97 @@
+"""Tests for hybrid search router — query analysis and RRF merge."""
+
+import pytest
+import sys, os
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "plugins", "memory", "holographic"))
+
+from query_router import QueryType, analyze_query
+from hybrid_search import SearchResult, reciprocal_rank_fusion
+
+
+class TestQueryAnalysis:
+    def test_keyword_single_identifier(self):
+        result = analyze_query("function_name")
+        assert result.query_type == QueryType.KEYWORD
+
+    def test_keyword_quoted_term(self):
+        result = analyze_query('Find "exact phrase" in code')
+        assert result.query_type in (QueryType.KEYWORD, QueryType.HYBRID)
+
+    def test_keyword_dotted_path(self):
+        result = analyze_query("module.sub.function")
+        assert result.query_type == QueryType.KEYWORD
+
+    def test_semantic_question(self):
+        result = analyze_query("What did we discuss about deployment?")
+        assert result.query_type == QueryType.SEMANTIC
+
+    def test_semantic_how_to(self):
+        result = analyze_query("How to configure the gateway?")
+        assert result.query_type == QueryType.SEMANTIC
+
+    def test_compositional_contradiction(self):
+        result = analyze_query("Are there any contradictions in the facts?")
+        assert result.query_type == QueryType.COMPOSITIONAL
+
+    def test_compositional_related(self):
+        result = analyze_query("What facts are related to Alexander?")
+        assert result.query_type == QueryType.COMPOSITIONAL
+
+    def test_empty_query(self):
+        result = analyze_query("")
+        assert result.query_type == QueryType.HYBRID
+
+    def test_complex_query(self):
+        result = analyze_query("What did we decide about the deploy script?")
+        assert result.query_type in (QueryType.SEMANTIC, QueryType.HYBRID)
+
+
+class TestReciprocalRankFusion:
+    def test_single_list(self):
+        results = [
+            SearchResult(fact_id=1, content="A", score=0.9, source="fts5", rank=1),
+            SearchResult(fact_id=2, content="B", score=0.8, source="fts5", rank=2),
+        ]
+        merged = reciprocal_rank_fusion([results])
+        assert len(merged) == 2
+        assert merged[0].fact_id == 1  # Rank 1 should be first
+
+    def test_two_lists_merge(self):
+        list1 = [
+            SearchResult(fact_id=1, content="A", score=0.9, source="fts5", rank=1),
+            SearchResult(fact_id=2, content="B", score=0.8, source="fts5", rank=2),
+        ]
+        list2 = [
+            SearchResult(fact_id=2, content="B", score=0.95, source="qdrant", rank=1),
+            SearchResult(fact_id=3, content="C", score=0.7, source="qdrant", rank=2),
+        ]
+        merged = reciprocal_rank_fusion([list1, list2])
+        # Fact 2 appears in both lists → should rank highest
+        assert merged[0].fact_id == 2
+        assert len(merged) == 3
+
+    def test_empty_lists(self):
+        merged = reciprocal_rank_fusion([[], []])
+        assert len(merged) == 0
+
+    def test_weighted_merge(self):
+        list1 = [
+            SearchResult(fact_id=1, content="A", score=0.9, source="fts5", rank=1),
+        ]
+        list2 = [
+            SearchResult(fact_id=2, content="B", score=0.9, source="qdrant", rank=1),
+        ]
+        merged = reciprocal_rank_fusion(
+            [list1, list2],
+            weights={"fts5": 1.0, "qdrant": 2.0},
+        )
+        # Qdrant has higher weight → fact 2 should win
+        assert merged[0].fact_id == 2
+
+    def test_rrf_score_formula(self):
+        list1 = [
+            SearchResult(fact_id=1, content="A", score=0.9, source="fts5", rank=1),
+        ]
+        merged = reciprocal_rank_fusion([list1], k=60)
+        # RRF score = 1/(60+1) = 0.01639...
+        assert abs(merged[0].score - 1.0/61.0) < 0.001