poimen-memory/docs/HYBRID_SEARCH_DESIGN.md

Hybrid Search Design: Retrieval Pipeline + Index Optimization

Goal

Maximize retrieval accuracy and relevance by combining:

• Semantic search (pgvector): Understanding query intent
• Lexical search (OpenSearch BM25): Exact term matching
• Ranking fusion: Intelligent combination for best results

---

1. Retrieval Pipeline Architecture

Stage 1: Query Normalization (Entry Point)

User Query: "fix kubernetes port 8080 conflict"
     ↓
├─ Tokenize & clean
├─ Expand abbreviations (k8s → kubernetes)
├─ Extract entities (port:8080, service:kubernetes)
└─ Generate embedding (for semantic)

Implementation:

pub struct QueryContext {
    raw_query: String,
    normalized: String,           // Lowercase, trimmed
    tokens: Vec<String>,          // ["fix", "kubernetes", ...]
    entities: HashMap<String, String>, // {port: "8080"}
    embedding: Vec<f32>,          // 384-dim or 1536-dim
    timestamp: Instant,
}

---

Stage 2: Parallel Retrieval (Both Engines)

Query Context
  │
  ├─ SEMANTIC PATH (pgvector)
  │  ├─ Query PostgreSQL with embedding
  │  ├─ SELECT chunks WHERE embedding <-> query_vec < distance_threshold
  │  ├─ ORDER BY cosine_similarity DESC LIMIT 50
  │  └─ Return: [(chunk_id, score_0_to_1, chunk_text)]
  │
  └─ LEXICAL PATH (OpenSearch + JWT)
     ├─ Tokenize query
     ├─ POST vault-*/_search with BM25
     ├─ Query: multi_match on [content, breadcrumb, source]
     ├─ ORDER BY BM25 score DESC LIMIT 50
     └─ Return: [(doc_id, bm25_score_raw, chunk_text)]

Key: Execute both in parallel (tokio::join! or similar)

pub async fn hybrid_retrieve(
    query_ctx: &QueryContext,
    pg: &PgClient,
    opensearch: &OpenSearchClient,
    jwt_token: &str,
) -> Result<HybridResults> {
    let semantic_fut = pg.semantic_search(&query_ctx.embedding, 50);
    let lexical_fut = opensearch.lexical_search(&query_ctx.normalized, 50, jwt_token);
    
    let (semantic_results, lexical_results) = tokio::try_join!(semantic_fut, lexical_fut)?;
    
    // Stage 3: Normalize & Rank
    let ranked = rank_and_fuse(&semantic_results, &lexical_results)?;
    
    Ok(ranked)
}

---

Stage 3: Score Normalization & Ranking Fusion

Problem: Scores are incompatible

• pgvector: cosine similarity (0.0 to 1.0)
• BM25: raw TF-IDF scores (unbounded, typically 0-10+)

Solution: Min-Max Normalization

pub fn normalize_scores(results: &[(String, f32)]) -> Vec<(String, f32)> {
    let min_score = results.iter().map(|(_, s)| s).fold(f32::INFINITY, f32::min);
    let max_score = results.iter().map(|(_, s)| s).fold(f32::NEG_INFINITY, f32::max);
    
    let range = max_score - min_score;
    if range < 0.001 {
        // All scores identical → uniform
        return results.iter().map(|(id, _)| (id.clone(), 0.5)).collect();
    }
    
    results
        .iter()
        .map(|(id, score)| {
            let normalized = (score - min_score) / range;
            (id.clone(), normalized)
        })
        .collect()
}

---

Stage 4: Fusion Strategy

Option A: Weighted Linear Combination (Recommended for Now)

pub fn weighted_fusion(
    semantic: Vec<(String, f32)>,
    lexical: Vec<(String, f32)>,
    semantic_weight: f32,  // 0.6
    lexical_weight: f32,   // 0.4
) -> Vec<SearchResult> {
    // Normalize both
    let sem_norm = normalize_scores(&semantic);
    let lex_norm = normalize_scores(&lexical);
    
    // Create maps for O(1) lookup
    let sem_map: HashMap<String, f32> = sem_norm.into_iter().collect();
    let lex_map: HashMap<String, f32> = lex_norm.into_iter().collect();
    
    // Merge all document IDs
    let mut all_ids: HashSet<String> = sem_map.keys().cloned().collect();
    all_ids.extend(lex_map.keys().cloned());
    
    // Compute fusion scores
    let mut results: Vec<SearchResult> = all_ids
        .into_iter()
        .map(|id| {
            let sem_score = sem_map.get(&id).copied().unwrap_or(0.0);
            let lex_score = lex_map.get(&id).copied().unwrap_or(0.0);
            
            let fused_score = 
                semantic_weight * sem_score + 
                lexical_weight * lex_score;
            
            SearchResult {
                id,
                score: fused_score,
                sem_component: sem_score,
                lex_component: lex_score,
                // ... other fields
            }
        })
        .collect();
    
    // Sort by fused score
    results.sort_by(|a, b| b.score.partial_cmp(&a.score).unwrap());
    results.truncate(10);  // Top-k
    
    results
}

Pros:

• ✅ Simple, interpretable
• ✅ Easy to tune weights
• ✅ Transparent scoring

Cons:

• ❌ Assumes linear relationship
• ❌ Sensitive to weight tuning

---

Option B: Reciprocal Rank Fusion (RRF) - Alternative

pub fn reciprocal_rank_fusion(
    semantic: Vec<(String, f32)>,
    lexical: Vec<(String, f32)>,
) -> Vec<SearchResult> {
    // Convert to ranks (position in sorted list)
    let k = 60;  // Constant (typically 60)
    
    let mut fused_scores: HashMap<String, f32> = HashMap::new();
    
    // Add semantic ranks
    for (rank, (id, _)) in semantic.into_iter().enumerate() {
        let rrf_score = 1.0 / (k as f32 + (rank as f32 + 1.0));
        fused_scores.insert(id, rrf_score);
    }
    
    // Add lexical ranks (combine if already present)
    for (rank, (id, _)) in lexical.into_iter().enumerate() {
        let rrf_score = 1.0 / (k as f32 + (rank as f32 + 1.0));
        *fused_scores.entry(id).or_insert(0.0) += rrf_score;
    }
    
    // Sort by combined RRF score
    let mut results: Vec<_> = fused_scores.into_iter().collect();
    results.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap());
    results.truncate(10);
    
    results
        .into_iter()
        .map(|(id, score)| SearchResult { id, score, ..Default::default() })
        .collect()
}

Pros:

• ✅ No parameter tuning needed
• ✅ Robust to score distribution differences
• ✅ Academic consensus (best for diverse rankers)

Cons:

• ❌ Less transparent (harder to debug)
• ❌ Loses score magnitudes

---

2. Index Optimization

2.1 PostgreSQL (pgvector) Index Schema

Chunk Storage Table

CREATE TABLE chunks (
    id UUID PRIMARY KEY,
    
    -- Content
    text TEXT NOT NULL,                    -- Full chunk text
    section_id UUID,                       -- Which section of doc
    position_in_doc INT,                   -- Order for context
    
    -- Metadata for retrieval
    document_id UUID NOT NULL,             -- Source document
    source VARCHAR(255) NOT NULL,          -- File path: "runbooks/port-forward.md"
    project_id UUID NOT NULL,              -- Project filter
    
    -- Hierarchy for breadcrumb
    level VARCHAR(10),                     -- "L0", "L1", "L2"
    breadcrumb JSONB,                      -- ["runbooks", "kubernetes", "networking"]
    
    -- Embedding
    embedding vector(384),                 -- 384-dim (all-MiniLM-L6-v2)
    --   OR vector(1536) for OpenAI
    
    -- Metadata
    chunk_hash VARCHAR(64),                -- SHA256 for dedup
    created_at TIMESTAMP,
    updated_at TIMESTAMP,
    
    -- Index hints
    is_indexed BOOLEAN DEFAULT FALSE,
    INDEX_score FLOAT,                     -- For quality ranking
    
    CONSTRAINT fk_document FOREIGN KEY (document_id) REFERENCES documents(id)
);

-- PRIMARY INDEX: Vector similarity search
CREATE INDEX chunks_embedding_idx 
    ON chunks 
    USING ivfflat (embedding vector_cosine_ops) 
    WITH (lists = 100);  -- Adjust based on data size

-- SECONDARY INDEXES: Filtering/metadata
CREATE INDEX chunks_document_id_idx ON chunks(document_id);
CREATE INDEX chunks_project_id_idx ON chunks(project_id);
CREATE INDEX chunks_source_idx ON chunks(source);
CREATE INDEX chunks_level_idx ON chunks(level);

-- For deduplication during indexing
CREATE INDEX chunks_hash_idx ON chunks(chunk_hash);

Tuning Notes:

• lists parameter:
  • Small dataset (<10k): 50-100
  • Medium (10k-100k): 100-200
  • Large (>100k): 200-500
• Index Type: ivfflat (fast, approximate) vs hnsw (more accurate, slower)

Retrieval Query

-- Semantic search with filtering
SELECT 
    id,
    text,
    source,
    breadcrumb,
    level,
    1 - (embedding <-> $1::vector) as similarity_score,
    position_in_doc
FROM chunks
WHERE 
    project_id = $2  -- Filter by project first
    AND level IN ('L0', 'L1', 'L2')  -- Exclude deep sections
    AND created_at > NOW() - INTERVAL '1 year'  -- Recency
ORDER BY 
    embedding <-> $1::vector  -- Cosine distance
LIMIT 50;

-- Add context: retrieve adjacent chunks
WITH target_chunk AS (
    SELECT section_id, position_in_doc FROM chunks WHERE id = $1
)
SELECT * FROM chunks
WHERE 
    section_id = (SELECT section_id FROM target_chunk)
    AND position_in_doc BETWEEN 
        (SELECT position_in_doc FROM target_chunk) - 2 
        AND (SELECT position_in_doc FROM target_chunk) + 2
ORDER BY position_in_doc;

---

2.2 OpenSearch (Lexical) Index Schema

Index Mapping (vault-* indices)

{
  "settings": {
    "number_of_shards": 2,
    "number_of_replicas": 1,
    "index.codec": "best_compression",
    "analysis": {
      "analyzer": {
        "standard_analyzer": {
          "type": "standard",
          "stopwords": "_english_"
        },
        "ngram_analyzer": {
          "type": "custom",
          "tokenizer": "ngram_tokenizer",
          "filter": ["lowercase"]
        },
        "ngram_tokenizer": {
          "type": "ngram",
          "min_gram": 3,
          "max_gram": 4,
          "token_chars": ["letter", "digit"]
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "content": {
        "type": "text",
        "analyzer": "standard_analyzer",
        "fields": {
          "raw": { "type": "keyword" },
          "ngram": { "type": "text", "analyzer": "ngram_analyzer" }
        },
        "boost": 2.0  -- Content gets higher weight
      },
      "source": {
        "type": "keyword",
        "boost": 1.5
      },
      "breadcrumb": {
        "type": "keyword",
        "boost": 1.2
      },
      "level": {
        "type": "keyword"
      },
      "section_title": {
        "type": "text",
        "analyzer": "standard_analyzer",
        "boost": 1.8
      },
      "document_id": {
        "type": "keyword"
      },
      "project_id": {
        "type": "keyword"
      },
      "indexed_at": {
        "type": "date"
      }
    }
  }
}

Retrieval Query

{
  "size": 50,
  "query": {
    "bool": {
      "must": [
        {
          "multi_match": {
            "query": "fix kubernetes port 8080",
            "fields": [
              "content^2",           -- Content gets 2x weight
              "section_title^1.5",
              "breadcrumb",
              "source"
            ],
            "type": "best_fields",   -- Match best field, not sum
            "operator": "or",
            "fuzziness": "AUTO",
            "max_expansions": 50
          }
        }
      ],
      "filter": [
        { "term": { "project_id": "poimen" } },
        { "terms": { "level": ["L0", "L1", "L2"] } },
        { "range": { "indexed_at": { "gte": "now-1y" } } }
      ]
    }
  },
  "_source": ["content", "source", "breadcrumb", "level", "document_id"]
}

Analyzer Choices:

• standard: Good for most cases
• ngram: Better for typos/misspellings
• edge_ngram: Better for autocomplete

---

3. Retrieval Accuracy Optimization

3.1 Score Calculation Breakdown

pub struct ScoreBreakdown {
    pub doc_id: String,
    pub semantic_score: f32,       // 0.88 (cosine similarity)
    pub lexical_score: f32,        // 0.96 (BM25 normalized)
    pub semantic_weight: f32,      // 0.6
    pub lexical_weight: f32,       // 0.4
    pub final_score: f32,          // 0.92
    pub rank: usize,               // Position in results
    pub retrieval_path: String,    // "hybrid" | "semantic_only" | "lexical_only"
}

pub struct SearchResult {
    pub id: String,
    pub chunk: String,
    pub source: String,
    pub breadcrumb: Vec<String>,
    pub level: String,
    pub breakdown: ScoreBreakdown,
}

3.2 Quality Metrics

Metric 1: Mean Reciprocal Rank (MRR)

// How early is the correct answer ranked?
fn mean_reciprocal_rank(results: &[SearchResult], ground_truth_id: &str) -> f32 {
    results
        .iter()
        .position(|r| r.id == ground_truth_id)
        .map(|pos| 1.0 / (pos + 1) as f32)
        .unwrap_or(0.0)
}

// MRR@10 = average of top 10 positions across queries

Metric 2: Normalized Discounted Cumulative Gain (NDCG)

// How good are rankings, accounting for position?
fn ndcg(results: &[SearchResult], relevance_scores: &[u32]) -> f32 {
    let dcg: f32 = results
        .iter()
        .enumerate()
        .zip(relevance_scores)
        .map(|((pos, _), rel)| (*rel as f32) / (pos as f32 + 2.0).log2())
        .sum();
    
    let idcg: f32 = {
        let mut sorted = relevance_scores.to_vec();
        sorted.sort_by(|a, b| b.cmp(a));
        sorted
            .iter()
            .enumerate()
            .map(|(pos, rel)| (*rel as f32) / (pos as f32 + 2.0).log2())
            .sum()
    };
    
    if idcg == 0.0 { 0.0 } else { dcg / idcg }
}

// Score: 0.0 (worst) to 1.0 (perfect ranking)

Metric 3: Precision@K and Recall@K

fn precision_at_k(results: &[SearchResult], ground_truth: &HashSet<String>, k: usize) -> f32 {
    let retrieved_truth: HashSet<_> = results
        .iter()
        .take(k)
        .filter(|r| ground_truth.contains(&r.id))
        .map(|r| r.id.clone())
        .collect();
    
    retrieved_truth.len() as f32 / k as f32
}

fn recall_at_k(results: &[SearchResult], ground_truth: &HashSet<String>, k: usize) -> f32 {
    let retrieved_truth: HashSet<_> = results
        .iter()
        .take(k)
        .filter(|r| ground_truth.contains(&r.id))
        .map(|r| r.id.clone())
        .collect();
    
    retrieved_truth.len() as f32 / ground_truth.len() as f32
}

---

3.3 Weight Tuning Strategy

Start with defaults: semantic=0.6, lexical=0.4

Then A/B test:

pub async fn evaluate_weights(
    test_queries: &[(String, Vec<String>)],  // (query, ground_truth_ids)
    pg: &PgClient,
    opensearch: &OpenSearchClient,
) -> Result<BestWeights> {
    let weight_combinations = vec![
        (0.5, 0.5),  // Equal
        (0.6, 0.4),  // Semantic bias (default)
        (0.7, 0.3),  // Heavy semantic
        (0.4, 0.6),  // Lexical bias
    ];
    
    for (sem_w, lex_w) in weight_combinations {
        let mut ndcg_scores = Vec::new();
        
        for (query, ground_truth) in test_queries {
            let results = hybrid_retrieve(query, sem_w, lex_w, pg, opensearch).await?;
            let relevance = ground_truth.iter()
                .map(|id| if results.iter().any(|r| &r.id == id) { 1 } else { 0 })
                .collect::<Vec<_>>();
            
            let score = ndcg(&results, &relevance);
            ndcg_scores.push(score);
        }
        
        let avg_ndcg = ndcg_scores.iter().sum::<f32>() / ndcg_scores.len() as f32;
        println!("Weights ({}, {}): NDCG = {:.3}", sem_w, lex_w, avg_ndcg);
    }
    
    Ok(BestWeights { semantic: 0.6, lexical: 0.4 })
}

---

4. Query Routing & Fallback

Decision Tree

User Query
    │
    ├─ Is query very short (<3 tokens)?
    │  ├─ YES → Use LEXICAL only (BM25 better for keywords)
    │  │        "fix port" → Exact term match
    │  │
    │  └─ NO → Continue...
    │
    ├─ Does query contain special syntax (#hashtag, @mention)?
    │  ├─ YES → Use LEXICAL + filter
    │  │
    │  └─ NO → Continue...
    │
    ├─ Can we embed the query? (check LLM availability)
    │  ├─ YES → Use HYBRID (both engines)
    │  │
    │  └─ NO → Fallback to LEXICAL only
    │
    └─ Execute chosen strategy

Implementation:

pub async fn route_query(query: &str, openai: &LLMClient) -> QueryStrategy {
    let token_count = query.split_whitespace().count();
    
    // Very short queries: lexical is better
    if token_count < 3 {
        return QueryStrategy::LexicalOnly;
    }
    
    // Special syntax: use lexical to preserve exact matches
    if query.contains('#') || query.contains('@') {
        return QueryStrategy::LexicalWithFilters;
    }
    
    // Try to embed
    match openai.embed(query).await {
        Ok(_embedding) => QueryStrategy::Hybrid,
        Err(_) => {
            // LLM unavailable: fallback to lexical
            QueryStrategy::LexicalOnly
        }
    }
}

---

5. Indexing Pipeline (Write Side)

When documents change (via git merge):

Git Merge Event
    │
    ├─ Parse new/changed document
    ├─ Split into chunks (by heading)
    │
    ├─ For each chunk:
    │  ├─ Compute embedding (send to LLM)
    │  │  └─ Cache: avoid re-embedding identical chunks
    │  │
    │  ├─ Write to PostgreSQL (chunks table)
    │  │  └─ INSERT with embedding vector
    │  │
    │  └─ Index to OpenSearch
    │     ├─ POST vault-*/_doc/{id}
    │     └─ With JWT token (Memory Service → OpenSearch)
    │
    ├─ Update chunk_hash (for dedup)
    └─ Mark is_indexed = TRUE

Deduplication: If chunk_hash exists and is_indexed=TRUE, skip.

---

6. Testing Strategy

Test Fixture: Query + Expected Results

test_queries:
  - query: "fix kubernetes port 8080 conflict"
    expected_docs: ["runbooks/port-forward.md", "docs/troubleshooting.md"]
    min_mrr: 0.5  # Top 2 expected
    min_ndcg: 0.7
  
  - query: "how to debug deployment issues"
    expected_docs: ["runbooks/deployment-debug.md"]
    min_mrr: 0.8  # Top 1 expected
    min_ndcg: 0.85

Test Execution

#[tokio::test]
async fn test_hybrid_search_accuracy() {
    let test_queries = load_test_fixtures("tests/fixtures/search_queries.yaml");
    let pg = setup_pg_for_test().await;
    let opensearch = setup_opensearch_for_test().await;
    
    for test in test_queries {
        let results = hybrid_retrieve(
            &test.query,
            &pg,
            &opensearch,
            0.6,  // semantic weight
            0.4,  // lexical weight
        ).await.unwrap();
        
        let ndcg = calculate_ndcg(&results, &test.expected_docs);
        assert!(ndcg >= test.min_ndcg, "NDCG {:.3} < {:.3}", ndcg, test.min_ndcg);
    }
}

---

7. Deployment Phases

Phase 1: Lexical-Only (Week 1)

• Deploy OpenSearch + JWT
• Use LEXICAL strategy only
• Benchmark: Precision, Recall
• Goal: Ensure BM25 works reliably

Phase 2: Hybrid with Fallback (Week 2-3)

• Deploy hybrid retrieval code
• Weight tuning: 50/50, 60/40, 70/30
• A/B test: 10% traffic hybrid, 90% semantic
• Metrics: Compare NDCG, MRR

Phase 3: Gradual Rollout (Week 4+)

• 10% → 25% → 50% → 100%
• Monitor latency (parallel = slightly slower)
• Monitor accuracy (should be better)

Phase 4: Optimization (Week 5+)

• Tune index parameters (lists, refresh_interval)
• Optimize query routing
• Feature flag: weights, strategy

---

Summary: Configuration

# k8s/app/memory-deployment.yaml
env:
  # Search strategy
  - name: SEARCH_STRATEGY
    value: "hybrid"  # hybrid | semantic | lexical
  
  # Hybrid weights
  - name: HYBRID_SEMANTIC_WEIGHT
    value: "0.6"
  - name: HYBRID_LEXICAL_WEIGHT
    value: "0.4"
  
  # Ranking algorithm
  - name: RANKING_ALGORITHM
    value: "weighted_linear"  # weighted_linear | rrf
  
  # Retrieval limits
  - name: SEMANTIC_RETRIEVE_K
    value: "50"  # Retrieve top-50 from pgvector
  - name: LEXICAL_RETRIEVE_K
    value: "50"  # Retrieve top-50 from OpenSearch
  - name: FINAL_RESULT_K
    value: "10"  # Return top-10 to user
  
  # Indexing
  - name: PGVECTOR_INDEX_LISTS
    value: "100"
  - name: OPENSEARCH_REFRESH_INTERVAL
    value: "30s"
  - name: CHUNK_DEDUP_ENABLED
    value: "true"