# 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:** ```rust pub struct QueryContext { raw_query: String, normalized: String, // Lowercase, trimmed tokens: Vec, // ["fix", "kubernetes", ...] entities: HashMap, // {port: "8080"} embedding: Vec, // 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)** ```rust pub async fn hybrid_retrieve( query_ctx: &QueryContext, pg: &PgClient, opensearch: &OpenSearchClient, jwt_token: &str, ) -> Result { 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** ```rust 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) ```rust pub fn weighted_fusion( semantic: Vec<(String, f32)>, lexical: Vec<(String, f32)>, semantic_weight: f32, // 0.6 lexical_weight: f32, // 0.4 ) -> Vec { // Normalize both let sem_norm = normalize_scores(&semantic); let lex_norm = normalize_scores(&lexical); // Create maps for O(1) lookup let sem_map: HashMap = sem_norm.into_iter().collect(); let lex_map: HashMap = lex_norm.into_iter().collect(); // Merge all document IDs let mut all_ids: HashSet = sem_map.keys().cloned().collect(); all_ids.extend(lex_map.keys().cloned()); // Compute fusion scores let mut results: Vec = 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 ```rust pub fn reciprocal_rank_fusion( semantic: Vec<(String, f32)>, lexical: Vec<(String, f32)>, ) -> Vec { // Convert to ranks (position in sorted list) let k = 60; // Constant (typically 60) let mut fused_scores: HashMap = 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 ```sql 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 ```sql -- 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) ```json { "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 ```json { "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 ```rust 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, pub level: String, pub breakdown: ScoreBreakdown, } ``` ### 3.2 Quality Metrics #### Metric 1: Mean Reciprocal Rank (MRR) ```rust // 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) ```rust // 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 ```rust fn precision_at_k(results: &[SearchResult], ground_truth: &HashSet, 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, 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:** ```rust pub async fn evaluate_weights( test_queries: &[(String, Vec)], // (query, ground_truth_ids) pg: &PgClient, opensearch: &OpenSearchClient, ) -> Result { 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::>(); let score = ndcg(&results, &relevance); ndcg_scores.push(score); } let avg_ndcg = ndcg_scores.iter().sum::() / 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:** ```rust 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 ```yaml 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 ```rust #[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 ```yaml # 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" ```