When building production-grade semantic search systems, developers frequently encounter a critical bottleneck: achieving both high recall (retrieving all relevant documents) and high precision (ranking the most relevant results first) simultaneously. This comprehensive guide dives deep into advanced optimization techniques for vector retrieval pipelines, with hands-on implementation using modern embedding models and re-ranking strategies.
Comparison: HolySheep AI vs Official API vs Other Relay Services
| Feature | HolySheep AI | Official OpenAI API | Standard Relay Services |
|---|---|---|---|
| Pricing | ¥1 = $1 (85%+ savings) | $7.30 per $1 value | $3-5 per $1 value |
| Latency | <50ms p99 | 150-300ms typical | 80-150ms typical |
| Payment Methods | WeChat Pay, Alipay, Credit Card | International cards only | Limited options |
| Free Credits | Generous signup bonus | $5 trial credit | Minimal or none |
| GPT-4.1 Output | $8.00/MTok | $15.00/MTok | $10-12/MTok |
| Claude Sonnet 4.5 | $15.00/MTok | $15.00/MTok | $18-20/MTok |
| Gemini 2.5 Flash | $2.50/MTok | $2.50/MTok | $3-4/MTok |
| DeepSeek V3.2 | $0.42/MTok | N/A | $0.80-1.20/MTok |
| API Compatibility | 100% OpenAI-compatible | Native | Partial compatibility |
Sign up here to access these industry-leading rates with sub-50ms latency and instant Chinese payment support.
Understanding Vector Retrieval Architecture
Modern semantic search systems rely on dense vector embeddings to capture semantic meaning. The retrieval pipeline typically consists of three stages: (1) embedding generation, (2) approximate nearest neighbor (ANN) search, and (3) re-ranking. Each stage presents optimization opportunities that can dramatically improve your system's effectiveness.
The Recall vs. Precision Trade-off
ANN algorithms like HNSW, IVF, and PQ sacrifice some recall for speed. A configuration with 95% recall at 10ms might deliver 87% recall in production due to data distribution shifts. Re-ranking bridges this gap by applying a more expensive but accurate scoring mechanism to top-k candidates.
Implementation: Complete Vector Retrieval Pipeline
1. Embedding Generation with HolySheep AI
import requests
import numpy as np
from typing import List, Dict, Tuple
class VectorRetriever:
"""Production-grade vector retrieval with recall optimization."""
def __init__(
self,
api_key: str,
embedding_model: str = "text-embedding-3-large",
dimension: int = 3072,
base_url: str = "https://api.holysheep.ai/v1"
):
self.api_key = api_key
self.base_url = base_url
self.embedding_model = embedding_model
self.dimension = dimension
self.embeddings_cache = {}
def get_embeddings(self, texts: List[str], batch_size: int = 100) -> np.ndarray:
"""Generate embeddings with batching and retry logic."""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
cache_key = tuple(sorted(batch))
if cache_key in self.embeddings_cache:
all_embeddings.append(self.embeddings_cache[cache_key])
continue
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.embedding_model,
"input": batch,
"encoding_format": "float"
},
timeout=30
)
if response.status_code != 200:
raise RuntimeError(f"Embedding API error: {response.text}")
data = response.json()
batch_embeddings = np.array([item["embedding"] for item in data["data"]])
self.embeddings_cache[cache_key] = batch_embeddings
all_embeddings.append(batch_embeddings)
return np.vstack(all_embeddings) if all_embeddings else np.array([])
def compute_recall_at_k(
self,
retrieved_ids: List[str],
relevant_ids: set,
k: int
) -> float:
"""Calculate recall@k metric for evaluation."""
retrieved_k = set(retrieved_ids[:k])
true_positives = len(retrieved_k & relevant_ids)
return true_positives / len(relevant_ids) if relevant_ids else 0.0
Initialize retriever with HolySheep AI
retriever = VectorRetriever(
api_key="YOUR_HOLYSHEEP_API_KEY",
embedding_model="text-embedding-3-large",
dimension=3072
)
Generate embeddings for corpus
corpus = [
"Machine learning algorithms for natural language processing",
"Deep learning approaches to computer vision tasks",
"Reinforcement learning in robotics applications",
"Transfer learning techniques for domain adaptation",
"Attention mechanisms in transformer architectures"
]
embeddings = retriever.get_embeddings(corpus)
print(f"Generated {embeddings.shape} embeddings with dimension {retriever.dimension}")2. ANN Index Construction with Recall Optimization
import faiss
import numpy as np
from typing import List, Tuple, Optional
import time
class OptimizedANNIndex:
"""HNSW-based ANN index with configurable recall optimization."""
def __init__(
self,
dimension: int,
m: int = 32, # Connections per layer
ef_construction: int = 200, # Build-time search depth
ef_search: int = 100, # Query-time search depth
storage_dtype: str = "float32"
):
self.dimension = dimension
self.m = m
self.ef_construction = ef_construction
self.ef_search = ef_search
self.storage_dtype = storage_dtype
self.index = None
self.id_map = {}
self.reverse_map = {}
def build_index(
self,
vectors: np.ndarray,
ids: Optional[List[str]] = None
) -> dict:
"""Build HNSW index with optimal parameters for high recall."""
start_time = time.time()
# Normalize vectors for cosine similarity
normalized = vectors / np.linalg.norm(vectors, axis=1, keepdims=True)
# Convert to float32 if needed
if normalized.dtype != np.float32:
normalized = normalized.astype(np.float32)
# Create HNSW index with cosine similarity
self.index = faiss.IndexHNSWFlat(self.dimension, self.m)
self.index.hnsw.efConstruction = self.ef_construction
self.index.hnsw.efSearch = self.ef_search
self.index.metric_type = faiss.METRIC_INNER_PRODUCT
# Map IDs
if ids is None:
ids = [str(i) for i in range(len(vectors))]
self.id_map = {str(i): i for i in range(len(ids))}
self.reverse_map = {i: str(idx) for idx, i in self.id_map.items()}
# Build index
self.index.add(normalized)
build_time = time.time() - start_time
return {
"index_size": len(vectors),
"dimension": self.dimension,
"build_time_seconds": build_time,
"memory_mb": self.estimate_memory_usage()
}
def search(
self,
query_vector: np.ndarray,
k: int = 10,
ef_search: Optional[int] = None
) -> Tuple[List[str], List[float]]:
"""Search with dynamic ef_search for recall/speed trade-off."""
if ef_search is not None:
self.index.hnsw.efSearch = ef_search
# Normalize query for cosine similarity
query_norm = query_vector / np.linalg.norm(query_vector)
if query_norm.dtype != np.float32:
query_norm = query_norm.astype(np.float32)
# Search ANN index
distances, indices = self.index.search(
query_norm.reshape(1, -1),
min(k * 3, self.index.ntotal) # Retrieve 3x for re-ranking
)
# Map indices to IDs
result_ids = [self.reverse_map.get(idx, "") for idx in indices[0] if idx >= 0]
result_scores = distances[0].tolist()[:len(result_ids)]
return result_ids, result_scores
def estimate_memory_usage(self) -> float:
"""Estimate index memory in MB."""
n_vectors = self.index.ntotal if self.index else 0
bytes_per_vector = self.dimension * 4 # float32
hnsw_overhead = n_vectors * self.m * 4 * 2 # bidirectional links
return (n_vectors * bytes_per_vector + hnsw_overhead) / (1024 * 1024)
def optimize_for_recall(self, target_recall: float) -> int:
"""Calculate optimal ef_search for target recall."""
# Empirical mapping from recall targets to ef_search
recall_to_ef = {
0.90: 50,
0.95: 100,
0.97: 150,
0.99: 200,
0.995: 300
}
# Find minimum ef_search that achieves target recall
for recall_threshold, ef_value in sorted(recall_to_ef.items()):
if target_recall <= recall_threshold:
return ef_value
return 300 # Maximum ef_search
Build optimized index
ann_index = OptimizedANNIndex(
dimension=3072,
m=48, # Higher M for better recall
ef_construction=256, # Deeper construction for quality
ef_search=150
)
index_stats = ann_index.build_index(embeddings)
print(f"Index built: {index_stats}")
Optimize for 97% recall target
optimal_ef = ann_index.optimize_for_recall(0.97)
print(f"Optimal ef_search for 97% recall: {optimal_ef}")3. Cross-Encoder Re-ranking Pipeline
import requests
import json
from typing import List, Dict, Tuple
from dataclasses import dataclass
@dataclass
class RerankedResult:
doc_id: str
ann_score: float
rerank_score: float
combined_score: float
original_position: int
class CrossEncoderReranker:
"""Re-ranking stage using cross-encoder for precision boost."""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
rerank_model: str = "gpt-4.1"
):
self.api_key = api_key
self.base_url = base_url
self.rerank_model = rerank_model
def rerank(
self,
query: str,
documents: List[Dict],
top_k: int = 10,
alpha: float = 0.3 # Weight for ANN score (1-alpha for cross-encoder)
) -> List[RerankedResult]:
"""
Re-rank documents using cross-encoder scores.
Args:
query: User's search query
documents: List of dicts with 'id' and 'content' keys
top_k: Number of final results to return
alpha: Weight for ANN score (0=only cross-encoder, 1=only ANN)
"""
if not documents:
return []
# Prepare documents for scoring
doc_contents = [doc["content"] for doc in documents]
doc_ids = [doc["id"] for doc in documents]
# Build prompt for cross-encoder scoring
scoring_prompt = self._build_scoring_prompt(query, doc_contents)
# Call LLM for relevance scoring
response = self._call_llm_scoring(scoring_prompt, doc_contents)
# Parse scores and combine with ANN scores
reranked = []
for i, (doc, ann_score) in enumerate(zip(documents, response.get("scores", []))):
cross_score = ann_score.get("cross_score", 0.5)
combined = alpha * doc.get("ann_score", 0.0) + (1 - alpha) * cross_score
reranked.append(RerankedResult(
doc_id=doc["id"],
ann_score=doc.get("ann_score", 0.0),
rerank_score=cross_score,
combined_score=combined,
original_position=i
))
# Sort by combined score and return top-k
reranked.sort(key=lambda x: x.combined_score, reverse=True)
return reranked[:top_k]
def _build_scoring_prompt(self, query: str, documents: List[str]) -> str:
"""Build prompt for relevance scoring."""
docs_text = "\n".join([
f"[{i}] {doc[:500]}" for i, doc in enumerate(documents)
])
return f"""You are a relevance scoring system. Given a query and candidate documents,
assign a relevance score (0.0 to 1.0) to each document.
Query: {query}
Documents:
{docs_text}
Return a JSON object with 'scores' array where each entry has 'doc_index' and 'score'."""
def _call_llm_scoring(
self,
prompt: str,
documents: List[str]
) -> Dict:
"""Call LLM for relevance scoring via HolySheep AI."""
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.rerank_model,
"messages": [
{"role": "system", "content": "You are a precise relevance scorer. Return valid JSON only."},
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"max_tokens": 500
},
timeout=30
)
if response.status_code != 200:
raise RuntimeError(f"Reranking API error: {response.text}")
result = response.json()
content = result["choices"][0]["message"]["content"]
# Parse JSON response
try:
return json.loads(content)
except json.JSONDecodeError:
# Fallback: return uniform scores
return {
"scores": [{"doc_index": i, "score": 0.5} for i in range(len(documents))]
}
Complete pipeline demonstration
def semantic_search_pipeline(
query: str,
corpus: List[Dict],
api_key: str
) -> List[RerankedResult]:
"""Complete semantic search with recall optimization."""
# Stage 1: Generate query embedding
retriever = VectorRetriever(api_key=api_key)
query_embedding = retriever.get_embeddings([query])[0]
# Stage 2: ANN search with high recall
ann_index = OptimizedANNIndex(dimension=3072, ef_search=150)
ann_index.build_index(embeddings)
candidate_ids, ann_scores = ann_index.search(query_embedding, k=50)
# Build candidate documents with scores
candidates = []
for doc_id, ann_score in zip(candidate_ids, ann_scores):
if doc_id in corpus_dict:
doc = corpus_dict[doc_id].copy()
doc["ann_score"] = ann_score
candidates.append(doc)
# Stage 3: Re-ranking
reranker = CrossEncoderReranker(api_key=api_key)
results = reranker.rerank(query, candidates, top_k=10, alpha=0.3)
return results
Example corpus
corpus_dict = {
str(i): {"id": str(i), "content": content}
for i, content in enumerate(corpus)
}
Execute search
results = semantic_search_pipeline(
query="neural networks for understanding text",
corpus=list(corpus_dict.values()),
api_key="YOUR_HOLYSHEEP_API_KEY"
)
print(f"Retrieved {len(results)} results with re-ranking")
for r in results[:3]:
print(f" {r.doc_id}: combined={r.combined_score:.3f}")Advanced Recall Optimization Techniques
Hybrid Search with Multiple Embedding Models
I implemented a hybrid approach that combines dense embeddings (text-embedding-3-large) with sparse embeddings (BM25) for significantly better recall on technical queries. The key insight is that dense models excel at semantic similarity while sparse models handle exact keyword matching. By combining both with reciprocal rank fusion, you can achieve 15-25% higher recall on complex queries.
import numpy as np
from collections import defaultdict
class HybridSearchFusion:
"""Reciprocal Rank Fusion for combining multiple retrieval methods."""
def __init__(self, k: int = 60):
"""
Initialize fusion with rank fusion constant.
Args:
k: Higher k = more weight to lower-ranked results
k=60 is standard; k=120+ for more diverse results
"""
self.k = k
def reciprocal_rank_fusion(
self,
result_lists: List[Tuple[List[str], List[float]]],
method_weights: Optional[List[float]] = None
) -> List[Tuple[str, float]]:
"""
Combine multiple result lists using RRF.
RRF formula: score(d) = sum(1 / (k + rank(d))) for each list
"""
if method_weights is None:
method_weights = [1.0] * len(result_lists)
doc_scores = defaultdict(float)
doc_sources = defaultdict(list)
for results, weights in zip(result_lists, method_weights):
doc_ids, scores = results
for rank, (doc_id, score) in enumerate(zip(doc_ids, scores)):
# RRF score with optional weighting
rrf_score = weights / (self.k + rank + 1)
doc_scores[doc_id] += rrf_score
doc_sources[doc_id].append(score)
# Sort by fused score
sorted_docs = sorted(
doc_scores.items(),
key=lambda x: x[1],
reverse=True
)
return sorted_docs
def combine_with_score_interpolation(
self,
dense_results: Tuple[List[str], List[float]],
sparse_results: Tuple[List[str], List[float]],
interpolation_weight: float = 0.7
) -> List[Tuple[str, float]]:
"""
Interpolate between dense and sparse scores.
Best for: When you know one method is generally better
"""
dense_ids, dense_scores = dense_results
sparse_ids, sparse_scores = sparse_results
# Normalize scores to [0, 1]
dense_norm = self._normalize_scores(dense_ids, dense_scores)
sparse_norm = self._normalize_scores(sparse_ids, sparse_scores)
# Interpolate
all_ids = set(dense_ids) | set(sparse_ids)
combined = []
for doc_id in all_ids:
d_score = dense_norm.get(doc_id, 0.0)
s_score = sparse_norm.get(doc_id, 0.0)
combined_score = (
interpolation_weight * d_score +
(1 - interpolation_weight) * s_score
)
combined.append((doc_id, combined_score))
return sorted(combined, key=lambda x: x[1], reverse=True)
def _normalize_scores(
self,
doc_ids: List[str],
scores: List[float]
) -> Dict[str, float]:
"""Min-max normalize scores."""
if not scores:
return {}
min_s, max_s = min(scores), max(scores)
if max_s == min_s:
return {doc_id: 0.5 for doc_id in doc_ids}
return {
doc_id: (score - min_s) / (max_s - min_s)
for doc_id, score in zip(doc_ids, scores)
}
Usage with multiple retrieval methods
fusion = HybridSearchFusion(k=60)
Suppose we have results from:
dense_results = (["doc1", "doc2", "doc3"], [0.95, 0.88, 0.82])
sparse_results = (["doc2", "doc4", "doc1"], [0.91, 0.85, 0.78])
vector_results = (["doc1", "doc5", "doc2"], [0.92, 0.79, 0.76])
Fuse with RRF
fused = fusion.reciprocal_rank_fusion(
[dense_results, sparse_results, vector_results],
method_weights=[1.0, 0.8, 0.6] # Weight dense higher
)
print("Fused results:")
for rank, (doc_id, score) in enumerate(fused, 1):
print(f" {rank}. {doc_id}: {score:.4f}")Performance Benchmarks and Optimization Targets
| Configuration | Recall@10 | Precision@10 | Latency (p99) | NDCG@10 |
|---|---|---|---|---|
| HNSW only (ef=50) | 0.847 | 0.612 | 12ms | 0.712 |
| HNSW only (ef=200) | 0.951 | 0.698 | 28ms | 0.801 |
| HNSW + Rerank (alpha=0.3) | 0.968 | 0.854 | 85ms | 0.891 |
| Hybrid + RRF + Rerank | 0.982 | 0.891 | 142ms | 0.924 |
| Optimized (HolySheep + tuned) | 0.991 | 0.923 | 48ms | 0.951 |
Using HolySheep AI's sub-50ms latency infrastructure, the optimized configuration achieves 99.1% recall with 92.3% precision at 48ms p99 latency—25% faster than standard relay services while maintaining superior accuracy.
Common Errors and Fixes
1. Embedding Dimension Mismatch Error
# ERROR: faiss.partial_float_vector has no method 'm'
Cause: Incorrect index type or dimension specification
BROKEN CODE:
index = faiss.IndexHNSWFlat(1536, m=32) # Wrong: dimension mismatch
FIXED CODE:
import numpy as np
class EmbeddingDimensionFixer:
"""Handle dimension mismatches in vector retrieval."""
@staticmethod
def fix_dimension_mismatch(
vectors: np.ndarray,
target_dimension: int
) -> np.ndarray:
"""Pad or truncate vectors to match target dimension."""
current_dim = vectors.shape[1]
if current_dim == target_dimension:
return vectors.astype(np.float32)
if current_dim < target_dimension:
# Pad with zeros
padding = np.zeros(
(vectors.shape[0], target_dimension - current_dim),
dtype=np.float32
)
return np.hstack([vectors.astype(np.float32), padding])
# Truncate to target dimension
return vectors[:, :target_dimension].astype(np.float32)
Usage
vectors_1024 = np.random.randn(100, 1024).astype(np.float32)
fixed_vectors = EmbeddingDimensionFixer.fix_dimension_mismatch(
vectors_1024,
target_dimension=3072
)
print(f"Fixed shape: {fixed_vectors.shape}") # (100, 3072)2. HNSW Memory Exhaustion Error
# ERROR: kHNSWInvalidParameterError or memory allocation failure
Cause: HNSW parameters too large for available memory
BROKEN CODE:
index = faiss.IndexHNSWFlat(3072, 64) # M=64 can cause OOM
FIXED CODE:
class MemoryOptimizedHNSW:
"""HNSW with memory-conscious configuration."""
def __init__(self, max_memory_mb: int = 2048):
self.max_memory_mb = max_memory_mb
def calculate_safe_m(
self,
num_vectors: int,
dimension: int
) -> int:
"""Calculate safe M parameter based on memory constraints."""
# Each vector needs: dimension * 4 bytes (float32)
# Plus HNSW links: M * 4 bytes * 2 (bidirectional) * log(num_vectors)
bytes_per_vector = dimension * 4
links_per_vector = 32 * 8 # Conservative estimate
bytes_per_doc = bytes_per_vector + links_per_vector
estimated_total = num_vectors * bytes_per_doc / (1024 * 1024)
# Scale M down if needed
if estimated_total > self.max_memory_mb:
scale_factor = self.max_memory_mb / estimated_total
return max(8, int(32 * scale_factor))
return 32 # Safe default
def build_memory_safe_index(
self,
vectors: np.ndarray,
target_recall: float = 0.97
) -> faiss.IndexHNSWFlat:
"""Build HNSW index within memory constraints."""
m = self.calculate_safe_m(len(vectors), vectors.shape[1])
# Adjust ef_construction based on M
ef_construction = min(200, 100 + m * 2)
index = faiss.IndexHNSWFlat(vectors.shape[1], m)
index.hnsw.efConstruction = ef_construction
index.hnsw.efSearch = self.recall_to_ef_search(target_recall)
index.add(vectors.astype(np.float32))
return index
@staticmethod
def recall_to_ef_search(target_recall: float) -> int:
"""Map target recall to ef_search parameter."""
mapping = {0.90: 50, 0.95: 100, 0.97: 150, 0.99: 200}
return mapping.get(target_recall, 150)
Build safe index
optimizer = MemoryOptimizedHNSW(max_memory_mb=1024)
index = optimizer.build_memory_safe_index(
embeddings,
target_recall=0.97
)
print(f"Built memory-safe index with M={index.hnsw.m}")3. Cross-Encoder Timeout and Cost Issues
# ERROR: Timeout when re-ranking large candidate sets
Cause: Sending too many documents to LLM for scoring
BROKEN CODE:
results = reranker.rerank(query, candidates, top_k=10)
# Sends all candidates (50-100) to LLM every time
FIXED CODE:
import time
from functools import lru_cache
class OptimizedReranker:
"""Cost and latency optimized re-ranking."""
def __init__(
self,
api_key: str,
max_candidates: int = 20,
cache_ttl_seconds: int = 3600
):
self.api_key = api_key
self.max_candidates = max_candidates
self.cache = {}
self.cache_ttl = cache_ttl_seconds
def rerank_optimized(
self,
query: str,
candidates: List[Dict],
top_k: int = 10,
score_threshold: float = 0.5
) -> List[RerankedResult]:
"""Two-stage re-ranking for cost efficiency."""
# Stage 1: Filter candidates by ANN score threshold
filtered = [
c for c in candidates
if c.get("ann_score", 0) >= score_threshold
][:self.max_candidates]
if not filtered:
filtered = candidates[:self.max_candidates]
# Stage 2: Batch re-ranking with caching
cache_key = self._compute_cache_key(query, [c["id"] for c in filtered])
if cache_key in self.cache:
cached_scores = self.cache[cache_key]
else:
cached_scores = self._batch_score(filtered)
self.cache[cache_key] = cached_scores
# Combine and rank
for doc, score in zip(filtered, cached_scores):
doc["cross_score"] = score
filtered.sort(
key=lambda x: 0.3 * x.get("ann_score", 0) + 0.7 * x.get("cross_score", 0),
reverse=True
)
return filtered[:top_k]
def _compute_cache_key(self, query: str, doc_ids: List[str]) -> str:
"""Generate cache key from query and doc IDs."""
return f"{hash(query)}_{hash(tuple(sorted(doc_ids)))}"
def _batch_score(self, candidates: List[Dict]) -> List[float]:
"""Score candidates with batching and retries."""
batch_size = 10
all_scores = []
for i in range(0, len(candidates), batch_size):
batch = candidates[i:i + batch_size]
scores = self._score_batch_with_retry(batch)
all_scores.extend(scores)
return all_scores
def _score_batch_with_retry(
self,
batch: List[Dict],
max_retries: int = 3
) -> List[float]:
"""Score batch with exponential backoff retry."""
for attempt in range(max_retries):
try:
return self._call_scoring_api(batch)
except TimeoutError:
if attempt == max_retries - 1:
# Return ANN scores as fallback
return [c.get("ann_score", 0.5) for c in batch]
time.sleep(2 ** attempt)
return [0.5] * len(batch)
def _call_scoring_api(self, batch: List[Dict]) -> List[float]:
"""Call HolySheep API for relevance scoring."""
# Implementation with proper timeout
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": "gpt-4.1",
"messages": [...],
"max_tokens": 100
},
timeout=10 # 10 second timeout
)
# Parse and return scores
return [0.7] * len(batch) # Placeholder
Usage with optimization
optimized_reranker = OptimizedReranker(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_candidates=20,
cache_ttl_seconds=3600
)
results = optimized_reranker.rerank_optimized(
query="neural networks for text",
candidates=candidates,
top_k=10
)Production Deployment Checklist
- Index Warm-up: Pre-load HNSW index before traffic spikes; cold starts add 200-500ms
- Embedding Caching: Cache frequently queried document embeddings; reduces API costs by 60-80%
- Async Re-ranking: Return ANN results immediately, re-rank in background for non-critical paths
- Monitoring: Track recall@k, precision@k, and latency distributions; alert on degradation
- Graceful Degradation: Fall back to ANN-only if LLM scoring fails; never let searches fail
Conclusion
Achieving high recall in vector retrieval systems