---
title: "Rag Retrieval"
description: "Retrieval-Augmented Generation patterns for grounded LLM responses. Use when building RAG pipelines, embedding documents, implementing hybrid search, contextual retrieval, HyDE, agentic RAG, multimodal RAG, query decomposition, reranking, or pgvector search."
canonical: "https://orchestkit.yonyon.ai/docs/reference/skills/rag-retrieval"
---
# Rag Retrieval
Retrieval-Augmented Generation patterns for grounded LLM responses. Use when building RAG pipelines, embedding documents, implementing hybrid search, contextual retrieval, HyDE, agentic RAG, multimodal RAG, query decomposition, reranking, or pgvector search.
Reference high
> **Not directly invocable** — no slash command and no model auto-selection. An agent loads it explicitly via `Read()`.
> **Rag Retrieval** Retrieval-Augmented Generation patterns for grounded LLM responses. Use when building RAG pipelines, embedding documents, implementing hybrid search, contextual retrieval, HyDE, agentic RAG, multimodal RAG, query decomposition, reranking, or pgvector search.
# RAG Retrieval
Comprehensive patterns for building production RAG systems. Each category has individual rule files in `rules/` loaded on-demand.
House thresholds, fusion ordering, and the latency and quality budgets we assert live in
[House delta](#house-delta) below. Vendor documentation is linked, not restated (see
[Upstream coverage](#upstream-coverage-do-not-restate)).
## Quick Reference
| Category | Rules | Impact | When to Use |
|----------|-------|--------|-------------|
| [Core RAG](#core-rag) | 4 | CRITICAL | Basic RAG, citations, hybrid search, context management |
| [Embeddings](#embeddings) | 3 | HIGH | Model selection, chunking, batch/cache optimization |
| [Contextual Retrieval](#contextual-retrieval) | 3 | HIGH | Context-prepending, hybrid BM25+vector, pipeline |
| [HyDE](#hyde) | 3 | HIGH | Vocabulary mismatch, hypothetical document generation |
| [Agentic RAG](#agentic-rag) | 4 | HIGH | Self-RAG, CRAG, knowledge graphs, adaptive routing |
| [Multimodal RAG](#multimodal-rag) | 3 | MEDIUM | Image+text retrieval, PDF chunking, cross-modal search |
| [Query Decomposition](#query-decomposition) | 3 | MEDIUM | Multi-concept queries, parallel retrieval, RRF fusion |
| [Reranking](#reranking) | 3 | MEDIUM | Cross-encoder, LLM scoring, combined signals |
| [PGVector](#pgvector) | 4 | HIGH | PostgreSQL hybrid search, HNSW indexes, schema design |
**Total: 30 rules across 9 categories**
## Core RAG
Fundamental patterns for retrieval, generation, and pipeline composition.
| Rule | File | Key Pattern |
|------|------|-------------|
| Basic RAG | `rules/core-basic-rag.md` | Retrieve + context + generate with citations |
| Hybrid Search | `rules/core-hybrid-search.md` | RRF fusion (k=60) for semantic + keyword |
| Context Management | `rules/core-context-management.md` | Token budgeting + sufficiency check |
| Pipeline Composition | `rules/core-pipeline-composition.md` | Composable Decompose → HyDE → Retrieve → Rerank |
## Embeddings
Embedding models, chunking strategies, and production optimization.
| Rule | File | Key Pattern |
|------|------|-------------|
| Models & API | `rules/embeddings-models.md` | Model selection, batch API, similarity |
| Chunking | `rules/embeddings-chunking.md` | Semantic boundary splitting, 512 token sweet spot |
| Advanced | `rules/embeddings-advanced.md` | Redis cache, Matryoshka dims, batch processing |
## Contextual Retrieval
Anthropic's context-prepending technique — 67% fewer retrieval failures.
| Rule | File | Key Pattern |
|------|------|-------------|
| Context Prepending | `rules/contextual-prepend.md` | LLM-generated context + prompt caching |
| Hybrid Search | `rules/contextual-hybrid.md` | 40% BM25 / 60% vector weight split |
| Complete Pipeline | `rules/contextual-pipeline.md` | End-to-end indexing + hybrid retrieval |
## HyDE
Hypothetical Document Embeddings for bridging vocabulary gaps.
| Rule | File | Key Pattern |
|------|------|-------------|
| Generation | `rules/hyde-generation.md` | Embed hypothetical doc, not query |
| Per-Concept | `rules/hyde-per-concept.md` | Parallel HyDE for multi-topic queries |
| Fallback | `rules/hyde-fallback.md` | 2-3s timeout → direct embedding fallback |
## Agentic RAG
Self-correcting retrieval with LLM-driven decision making.
| Rule | File | Key Pattern |
|------|------|-------------|
| Self-RAG | `rules/agentic-self-rag.md` | Binary document grading for relevance |
| Corrective RAG | `rules/agentic-corrective-rag.md` | CRAG workflow with web fallback |
| Knowledge Graph | `rules/agentic-knowledge-graph.md` | KG + vector hybrid for entity-rich domains |
| Adaptive Retrieval | `rules/agentic-adaptive-retrieval.md` | Query routing to optimal strategy |
## Multimodal RAG
Image + text retrieval with cross-modal search.
| Rule | File | Key Pattern |
|------|------|-------------|
| Embeddings | `rules/multimodal-embeddings.md` | CLIP, SigLIP 2, Voyage multimodal-3 |
| Chunking | `rules/multimodal-chunking.md` | PDF extraction preserving images |
| Pipeline | `rules/multimodal-pipeline.md` | Dedup + hybrid retrieval + generation |
## Query Decomposition
Breaking complex queries into concepts for parallel retrieval.
| Rule | File | Key Pattern |
|------|------|-------------|
| Detection | `rules/query-detection.md` | Heuristic indicators (<1ms fast path) |
| Decompose + RRF | `rules/query-decompose.md` | LLM concept extraction + parallel retrieval |
| HyDE Combo | `rules/query-hyde-combo.md` | Decompose + HyDE for maximum coverage |
## Reranking
Post-retrieval re-scoring for higher precision.
| Rule | File | Key Pattern |
|------|------|-------------|
| Cross-Encoder | `rules/reranking-cross-encoder.md` | ms-marco-MiniLM (~50ms, free) |
| LLM Reranking | `rules/reranking-llm.md` | Batch scoring + Cohere API |
| Combined | `rules/reranking-combined.md` | Multi-signal weighted scoring |
## PGVector
Production hybrid search with PostgreSQL.
| Rule | File | Key Pattern |
|------|------|-------------|
| Schema | `rules/pgvector-schema.md` | HNSW index + pre-computed tsvector |
| Hybrid Search | `rules/pgvector-hybrid-search.md` | SQLAlchemy RRF with FULL OUTER JOIN |
| Indexing | `rules/pgvector-indexing.md` | HNSW (17x faster) vs IVFFlat |
| Metadata | `rules/pgvector-metadata.md` | Filtering, boosting, Redis 8 comparison |
## Quick Start Example
```python
from openai import OpenAI
client = OpenAI()
async def rag_query(question: str, top_k: int = 5) -> dict:
"""Basic RAG with citations."""
docs = await vector_db.search(question, limit=top_k)
context = "\n\n".join([f"[{i+1}] {doc.text}" for i, doc in enumerate(docs)])
response = await llm.chat([
{"role": "system", "content": "Answer with inline citations [1], [2]. Use ONLY provided context."},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
])
return {"answer": response.content, "sources": [d.metadata['source'] for d in docs]}
```
## Key Decisions
| Decision | Recommendation |
|----------|----------------|
| Embedding model | `text-embedding-3-small` (general), `voyage-3.5` (production) |
| Chunk size | 256-1024 tokens (512 typical) |
| Hybrid weight | 40% BM25 / 60% vector |
| Top-k | 3-10 documents |
| Temperature | 0.1-0.3 (factual) |
| Context budget | 4K-8K tokens |
| Reranking | Retrieve 50, rerank to 10 |
| Vector index | HNSW (production), IVFFlat (high-volume) |
| HyDE timeout | 2-3 seconds with fallback |
| Query decomposition | Heuristic first, LLM only if multi-concept |
## Upstream coverage (do not restate)
Fetch these from the source instead of restating them here. Where a row says the house
subset stays, that named file carries only the tuned values and the reason, not a tutorial.
| Topic | Source |
|-------|--------|
| pgvector install, operators, index build syntax, "why isn't my index used" | https://github.com/pgvector/pgvector; house subset (HNSW `m=16`, `ef_construction=64`, halfvec, binary quantize) stays in `rules/pgvector-indexing.md` and `rules/pgvector-schema.md` |
| Postgres full-text search: tsvector, tsquery, GIN, `ts_rank_cd` | https://www.postgresql.org/docs/current/textsearch.html; house subset (generated STORED column) stays in `rules/pgvector-schema.md` |
| Reading query plans, confirming an index is used, VACUUM/ANALYZE | https://www.postgresql.org/docs/current/using-explain.html |
| RRF mechanics: rank constant, rank window size, tie handling | https://www.elastic.co/docs/reference/elasticsearch/rest-apis/reciprocal-rank-fusion; house subset (`k=60`, 3x fetch) stays in `rules/core-hybrid-search.md` and `rules/pgvector-hybrid-search.md` |
| Field-weighted boosting and scoring profiles as a concept | https://learn.microsoft.com/en-us/azure/search/index-add-scoring-profiles; our boost factors and their ordering are in [House delta](#house-delta) and `rules/pgvector-metadata.md` |
| Embedding API mechanics: batch limits, `input_type`, dimensions, pricing | https://docs.voyageai.com/docs/embeddings and https://platform.openai.com/docs/guides/embeddings; model choice stays in `rules/embeddings-models.md` |
| Retrieval metric definitions (precision@k, recall@k, MRR, nDCG) and building the query set | ork skill `golden-dataset` |
| Search endpoint shape, pagination, error bodies | ork skill `api-design` |
| Tracing and dashboarding search calls | ork skill `monitoring-observability` |
| Writing the integration test that runs these assertions | ork skill `testing-integration` |
## Common Mistakes
1. No citation tracking (unverifiable answers)
2. Context too large (dilutes relevance)
3. Single retrieval method (misses keyword matches)
4. Not chunking long documents (context gets lost)
5. Embedding queries differently than documents
6. No fallback path in agentic RAG (workflow hangs)
7. Infinite rewrite loops (no retry limit)
8. Using wrong similarity metric (cosine vs euclidean)
9. Not caching embeddings (recomputing unchanged content)
10. Missing image captions in multimodal RAG (limits text search)
## Evaluations
See `test-cases.json` for 30 test cases across all categories.
There is **no** pass-rate, precision, recall, or MRR floor for this skill. Earlier
wording here promised those numbers; they were never measured and are written down
nowhere. They are left unset rather than filled in with plausible-looking values,
because a threshold nobody measured is worse than an absent one: it gets cited in
review as though it means something, and passes or fails changes for no defensible
reason. `test-cases.json` is the right substrate for establishing real ones, and a
baseline run over those 30 cases would produce numbers worth asserting.
The one budget this skill does assert is latency, in [House delta](#house-delta).
## Related Skills
- `ork:langgraph` - LangGraph workflow patterns (for agentic RAG workflows)
- `ork:golden-dataset` - Evaluate retrieval quality
- `ork:llm-integration` - Local embeddings with nomic-embed-text
- `ork:multimodal-llm` - Image analysis for multimodal RAG
- `ork:database-patterns` - Schema design for vector search
- `ork:performance` - Caching repeated RAG responses
## Capability Details
### retrieval-patterns
**Keywords:** retrieval, context, chunks, relevance, rag
**Solves:**
- Retrieve relevant context for LLM
- Implement RAG pipeline with citations
- Optimize retrieval quality
### hybrid-search
**Keywords:** hybrid, bm25, vector, fusion, rrf
**Solves:**
- Combine keyword and semantic search
- Implement reciprocal rank fusion
- Balance precision and recall
### embeddings
**Keywords:** embedding, text to vector, vectorize, chunk, similarity
**Solves:**
- Convert text to vector embeddings
- Choose embedding models and dimensions
- Implement chunking strategies
### contextual-retrieval
**Keywords:** contextual, anthropic, context-prepend, bm25
**Solves:**
- Prepend context to chunks for better retrieval
- Reduce retrieval failures by 67%
- Implement hybrid BM25+vector search
### hyde
**Keywords:** hyde, hypothetical, vocabulary mismatch
**Solves:**
- Bridge vocabulary gaps in semantic search
- Generate hypothetical documents for embedding
- Handle abstract or conceptual queries
### agentic-rag
**Keywords:** self-rag, crag, corrective, adaptive, grading
**Solves:**
- Build self-correcting RAG workflows
- Grade document relevance
- Implement web search fallback
### multimodal-rag
**Keywords:** multimodal, image, clip, vision, pdf
**Solves:**
- Build RAG with images and text
- Cross-modal search (text → image)
- Process PDFs with mixed content
### query-decomposition
**Keywords:** decompose, multi-concept, complex query
**Solves:**
- Break complex queries into concepts
- Parallel retrieval per concept
- Improve coverage for compound questions
### reranking
**Keywords:** rerank, cross-encoder, precision, scoring
**Solves:**
- Improve search precision post-retrieval
- Score relevance with cross-encoder or LLM
- Combine multiple scoring signals
### pgvector-search
**Keywords:** pgvector, postgresql, hnsw, tsvector, hybrid
**Solves:**
- Production hybrid search with PostgreSQL
- HNSW vs IVFFlat index selection
- SQL-based RRF fusion
---
## House delta
Inlined rather than placed in `references/`: this skill carries its house knowledge
in `rules/` (32 files) and has never had a `references/` directory, so a lone
delta file there would be the only occupant.
Embedding-model choice, chunking algorithms, vector-index tuning, reranker APIs and
the query-rewriting literature are vendor and upstream territory; see
[Upstream coverage](#upstream-coverage-do-not-restate) for where each lives. What
follows is only what OrchestKit adds, contradicts, or has to warn about.
### Retrieval latency budget: p95 under 500ms end to end
Source: `checklists/rag-quality.md:57`, the only retrieval budget this skill has ever
actually asserted. It covers the whole path a user waits on (embed the query, search,
rerank, assemble context), not one stage in isolation. A pipeline that clears 500ms
per stage and blows it in aggregate has failed this budget.
Measure at p95, not mean. Retrieval latency is dominated by tail behaviour (cold index
shards, reranker queueing), and a mean hides exactly the requests users complain about.
### Corpus-specific tuning does not transfer
Fusion weights, rerank depth and hybrid alpha are properties of a corpus, not of the
technique. A weighting that lifts recall on prose documentation regularly hurts it on
code or tabular data, so values copied between projects are noise rather than a
starting point. Re-derive them per corpus against that corpus's own eval set.
## Apply metadata boosting after RRF fusion, never before
Why: boost factors multiply onto the RRF score of an already-fused result set and the list is then re-sorted, using section-title match 1.5x, section-path match 1.15x, and code block on a technical query 1.2x; boosting per-method scores before fusion destroys the property RRF depends on (it fuses by rank, not by score) and lets a lone strong keyword hit outrank a document both methods agreed on, and this exact order is what produced the +6% MRR credited to boosting in `rules/pgvector-metadata.md` (distilled from the retired checklists/search-implementation-checklist.md and examples/examples/orchestkit-retrieval.md; no traced incident).
Upstream: https://learn.microsoft.com/en-us/azure/search/index-add-scoring-profiles
## Stop raising the RRF fetch multiplier past 3x
Why: measured pass rate on the reference corpus went 87.2% at 1x, 89.5% at 2x, 91.1% at 3x, 91.3% at 4x, so the fourth multiple bought 0.2 points for a third more rows scanned per method, which is why 3x is the house default hard-coded in `rules/core-hybrid-search.md` and `rules/pgvector-hybrid-search.md`; raise it only with a golden-set number that beats this curve (distilled from the retired examples/examples/orchestkit-retrieval.md; no traced incident).
Upstream: https://www.elastic.co/docs/reference/elasticsearch/rest-apis/reciprocal-rank-fusion
## Gate every retrieval change on the golden-set thresholds
Why: the house bar before a retrieval change ships is pass rate at least 90% on the golden query set, precision@10 at least 0.70, recall@10 at least 0.85, and MRR at least 0.65, floors set just under the reference pipeline's observed 91.6% pass rate and 0.777 overall MRR (0.695 on the hard slice) so that a real regression trips them instead of a noisy run (distilled from the retired checklists/search-implementation-checklist.md and examples/examples/orchestkit-retrieval.md; no traced incident).
Upstream: ork skill `golden-dataset`
## Assert search latency per stage, not just end to end
Why: the integration test asserts vector search under 100ms, keyword search under 50ms, and fused hybrid under 150ms separately, against an observed 415-chunk baseline of P50 15ms / P95 32ms / P99 62ms, because a single total-latency assertion stays green while one stage silently doubles, which is precisely the shape of an index that quietly stopped being used (distilled from the retired checklists/search-implementation-checklist.md and examples/examples/orchestkit-retrieval.md; no traced incident).
Upstream: https://www.postgresql.org/docs/current/using-explain.html
## Budget the embedding call before tuning the index
Why: in the reference measurement the query embedding was P50 8ms of a 15ms end-to-end search against 2ms for vector search and 3ms for keyword search, and it was the throughput ceiling at roughly 120 requests per second, so cache and batch embeddings first because index tuning below that ceiling moves the total by single-digit percent (distilled from the retired examples/examples/orchestkit-retrieval.md; no traced incident).
Upstream: https://docs.voyageai.com/docs/embeddings
---
## Rules (30)
### Route queries to the best retrieval strategy using adaptive selection per query type — MEDIUM
## Adaptive Retrieval
Route queries to optimal retrieval strategies based on query characteristics.
**Query Router:**
```python
from pydantic import BaseModel, Field
from typing import Literal
class QueryRoute(BaseModel):
strategy: Literal["direct", "hyde", "decompose", "web"] = Field(
description="Best retrieval strategy for this query"
)
reasoning: str
async def route_query(question: str) -> str:
route = await llm.with_structured_output(QueryRoute).ainvoke(
f"Choose the best retrieval strategy for: {question}\n"
"- direct: Simple factual queries with clear keywords\n"
"- hyde: Abstract/conceptual queries with vocabulary mismatch\n"
"- decompose: Multi-concept queries spanning multiple topics\n"
"- web: Recent events or data not in knowledge base"
)
return route.strategy
```
**Multi-Source Orchestration:**
```python
async def adaptive_search(question: str, top_k: int = 10) -> list[dict]:
strategy = await route_query(question)
if strategy == "direct":
return await retriever.search(question, top_k=top_k)
elif strategy == "hyde":
hyde_result = await hyde_service.generate(question)
return await retriever.search_by_embedding(hyde_result.embedding, top_k=top_k)
elif strategy == "decompose":
return await decomposed_search(question, top_k=top_k)
elif strategy == "web":
return await web_search(question)
```
**Incorrect — hardcoded single retrieval strategy:**
```python
async def search(question: str) -> list[dict]:
# Always uses HyDE regardless of query type
hyde_result = await hyde_service.generate(question)
return await retriever.search_by_embedding(hyde_result.embedding, top_k=10)
```
**Correct — adaptive routing based on query characteristics:**
```python
async def adaptive_search(question: str) -> list[dict]:
strategy = await route_query(question) # Choose best approach
if strategy == "direct":
return await retriever.search(question, top_k=10) # Fast path
elif strategy == "hyde":
hyde_result = await hyde_service.generate(question)
return await retriever.search_by_embedding(hyde_result.embedding, top_k=10)
```
**Key rules:**
- Route queries to optimal sources based on query type
- Direct search for simple factual queries (fastest)
- HyDE for abstract/conceptual queries (vocabulary bridging)
- Decomposition for multi-concept queries (comprehensive coverage)
- Web search for recent events or out-of-knowledge-base queries
- Routing adds ~200ms overhead — use heuristics for fast-path decisions
### Apply corrective RAG with quality assurance and web fallback for self-correction — HIGH
## Corrective RAG (CRAG)
Self-correcting retrieval with query rewriting and web search fallback.
**CRAG Workflow:**
```python
def build_crag_workflow() -> StateGraph:
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("web_search", web_search)
workflow.add_node("transform_query", transform_query)
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", route_after_grading, {
"generate": "generate",
"transform_query": "transform_query",
"web_search": "web_search"
})
workflow.add_edge("transform_query", "retrieve") # Retry
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)
return workflow.compile()
def route_after_grading(state: RAGState) -> str:
if state["web_search_needed"]:
if state.get("retry_count", 0) < 2:
return "transform_query"
return "web_search"
return "generate"
```
**Web Search Fallback:**
```python
def web_search(state: RAGState) -> dict:
web_results = tavily_client.search(state["question"], max_results=5, search_depth="advanced")
web_docs = [
Document(page_content=r["content"], metadata={"source": r["url"], "type": "web"})
for r in web_results
]
return {"documents": web_docs, "web_search_needed": False}
```
**Incorrect — no fallback path or retry limits:**
```python
def route_after_grading(state: RAGState) -> str:
if state["web_search_needed"]:
return "transform_query" # Infinite loop possible!
return "generate"
```
**Correct — bounded retries with web fallback:**
```python
def route_after_grading(state: RAGState) -> str:
if state["web_search_needed"]:
if state.get("retry_count", 0) < 2: # Max 2 retries
return "transform_query"
return "web_search" # Fallback to web
return "generate"
```
**Key rules:**
- Fallback order: Rewrite query (2x max) -> Web search -> Abstain
- Max 2-3 retries for query rewriting to prevent infinite loops
- Web search as last resort (latency + cost)
- Always include retry_count to prevent infinite loops
- No fallback path = workflow hangs on bad queries
### Build knowledge graph RAG for multi-hop reasoning over entity-rich domains — MEDIUM
## Knowledge Graph RAG (GraphRAG)
Combine knowledge graphs with vector search for entity-rich domains.
**Architecture:**
```
Query → [Entity Extraction] → [KG Lookup] → [Vector Search] → [Merge] → [Generate]
```
**Pattern Comparison:**
| Pattern | When to Use | Key Feature |
|---------|-------------|-------------|
| Self-RAG | Need adaptive retrieval | LLM decides when to retrieve |
| CRAG | Need quality assurance | Document grading + web fallback |
| GraphRAG | Entity-rich domains | Knowledge graph + vector hybrid |
| Agentic | Complex multi-step | Full plan-route-act-verify loop |
**Incorrect — vector-only search missing entity relationships:**
```python
async def search(query: str) -> list[dict]:
# Misses relationships between entities
return await vector_db.search(query, limit=10)
```
**Correct — hybrid KG + vector search:**
```python
async def graph_rag_search(query: str) -> list[dict]:
entities = await extract_entities(query) # Extract entities from query
kg_results = await neo4j.lookup_entities(entities) # KG lookup
vector_results = await vector_db.search(query, limit=10) # Vector search
return merge_results(kg_results, vector_results) # Combine both
```
**Key rules:**
- Use GraphRAG when domain has rich entity relationships (people, organizations, products)
- Combine KG entity lookup with vector similarity for hybrid results
- Entity extraction should use structured output (Pydantic) for reliability
- Multi-hop reasoning: follow graph edges to find connected information
- Neo4j or similar graph DB for production knowledge graphs
### Grade document relevance with Self-RAG to prevent context contamination from irrelevant results — HIGH
## Self-RAG — Document Grading
LLM grades retrieved documents for relevance before generation.
**State Definition:**
```python
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List, Annotated
from langchain_core.documents import Document
import operator
class RAGState(TypedDict):
question: str
documents: Annotated[List[Document], operator.add]
generation: str
web_search_needed: bool
retry_count: int
relevance_scores: dict[str, float]
```
**Document Grading:**
```python
from pydantic import BaseModel, Field
class GradeDocuments(BaseModel):
binary_score: str = Field(description="Relevance score 'yes' or 'no'")
def grade_documents(state: RAGState) -> dict:
"""Grade documents for relevance — core Self-RAG pattern."""
question = state["question"]
documents = state["documents"]
filtered_docs, relevance_scores = [], {}
for doc in documents:
score = retrieval_grader.invoke({
"question": question, "document": doc.page_content
})
doc_id = doc.metadata.get("id", hash(doc.page_content))
relevance_scores[doc_id] = 1.0 if score.binary_score == "yes" else 0.0
if score.binary_score == "yes":
filtered_docs.append(doc)
web_search_needed = len(filtered_docs) < len(documents) // 2
return {
"documents": filtered_docs,
"web_search_needed": web_search_needed,
"relevance_scores": relevance_scores
}
```
**Incorrect — no document grading, all docs used:**
```python
def generate(state: RAGState) -> dict:
# Uses all retrieved docs without quality check
context = "\n\n".join([d.page_content for d in state["documents"]])
return {"generation": llm.invoke(context)}
```
**Correct — grade documents before generation:**
```python
def grade_documents(state: RAGState) -> dict:
filtered_docs = []
for doc in state["documents"]:
score = grader.invoke({"question": state["question"], "document": doc.page_content})
if score.binary_score == "yes": # Only keep relevant docs
filtered_docs.append(doc)
web_search_needed = len(filtered_docs) < len(state["documents"]) // 2
return {"documents": filtered_docs, "web_search_needed": web_search_needed}
```
**Key rules:**
- Binary grading (yes/no) is simpler and more reliable than numeric scores
- Trigger web search fallback when >50% of docs are filtered out
- Track relevance scores for debugging and quality monitoring
- Self-RAG lets the LLM decide when to retrieve — adaptive by design
### Combine contextual embeddings with BM25 hybrid search for maximum retrieval coverage — HIGH
## Contextual Retrieval — Hybrid BM25+Vector
Combine contextual embeddings with BM25 for maximum retrieval quality.
**Hybrid Retriever:**
```python
from rank_bm25 import BM25Okapi
import numpy as np
class HybridRetriever:
def __init__(self, chunks: list[str], embeddings: np.ndarray):
self.chunks = chunks
self.embeddings = embeddings
tokenized = [c.lower().split() for c in chunks]
self.bm25 = BM25Okapi(tokenized)
def search(
self, query: str, query_embedding: np.ndarray,
top_k: int = 20, bm25_weight: float = 0.4, vector_weight: float = 0.6
) -> list[tuple[int, float]]:
bm25_scores = self.bm25.get_scores(query.lower().split())
bm25_scores = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min() + 1e-6)
vector_scores = np.dot(self.embeddings, query_embedding)
vector_scores = (vector_scores - vector_scores.min()) / (vector_scores.max() - vector_scores.min() + 1e-6)
combined = bm25_weight * bm25_scores + vector_weight * vector_scores
top_indices = np.argsort(combined)[::-1][:top_k]
return [(i, combined[i]) for i in top_indices]
```
**Results (Anthropic Research):**
| Method | Retrieval Failure Rate |
|--------|----------------------|
| Traditional embeddings | 5.7% |
| + Contextual embeddings | 3.5% |
| + Contextual + BM25 hybrid | 1.9% |
| + Contextual + BM25 + reranking | 1.3% |
**Incorrect — vector-only search without BM25:**
```python
def search(query: str, query_embedding: np.ndarray, top_k: int = 20) -> list[int]:
# Misses exact-match queries
vector_scores = np.dot(self.embeddings, query_embedding)
return np.argsort(vector_scores)[::-1][:top_k]
```
**Correct — hybrid BM25 + vector with proper weighting:**
```python
def search(query: str, query_embedding: np.ndarray, top_k: int = 20) -> list[tuple[int, float]]:
bm25_scores = self.bm25.get_scores(query.lower().split())
bm25_norm = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min() + 1e-6)
vector_scores = np.dot(self.embeddings, query_embedding)
vector_norm = (vector_scores - vector_scores.min()) / (vector_scores.max() - vector_scores.min() + 1e-6)
combined = 0.4 * bm25_norm + 0.6 * vector_norm # Research-backed weights
return [(i, combined[i]) for i in np.argsort(combined)[::-1][:top_k]]
```
**Key rules:**
- 67% reduction in retrieval failures with full contextual retrieval pipeline
- Default weight split: 40% BM25 / 60% vector (from Anthropic research)
- BM25 catches exact-match queries that vector search misses
- Normalize scores before weighted combination (min-max normalization)
- Adding reranking on top takes failure rate from 1.9% to 1.3%
### Build complete contextual retrieval pipeline integrating context generation, embedding, and hybrid search — MEDIUM
## Contextual Retrieval — Complete Pipeline
End-to-end pipeline with context generation, hybrid indexing, and retrieval.
**Complete Pipeline:**
```python
from dataclasses import dataclass
@dataclass
class ContextualChunk:
original: str
contextualized: str
embedding: list[float]
doc_id: str
chunk_index: int
class ContextualRetriever:
def __init__(self, embed_model, llm_client):
self.embed_model = embed_model
self.llm = llm_client
self.chunks: list[ContextualChunk] = []
def add_document(self, doc_id: str, text: str, chunk_size: int = 512):
raw_chunks = self._chunk_text(text, chunk_size)
contextualized = self._contextualize_batch(text, raw_chunks)
embeddings = self.embed_model.embed(contextualized)
for i, (raw, ctx, emb) in enumerate(zip(raw_chunks, contextualized, embeddings)):
self.chunks.append(ContextualChunk(
original=raw, contextualized=ctx, embedding=emb,
doc_id=doc_id, chunk_index=i
))
self._rebuild_bm25()
def search(self, query: str, top_k: int = 10) -> list[ContextualChunk]:
query_emb = self.embed_model.embed([query])[0]
bm25_scores = self.bm25.get_scores(query.lower().split())
embeddings = np.array([c.embedding for c in self.chunks])
vector_scores = np.dot(embeddings, query_emb)
combined = 0.4 * self._normalize(bm25_scores) + 0.6 * self._normalize(vector_scores)
top_indices = np.argsort(combined)[::-1][:top_k]
return [self.chunks[i] for i in top_indices]
```
**Parallel Processing:**
```python
async def contextualize_parallel(document: str, chunks: list[str]) -> list[str]:
semaphore = asyncio.Semaphore(10) # Max 10 concurrent
async def process_chunk(chunk: str) -> str:
async with semaphore:
context = await async_generate_context(document, chunk)
return f"{context}\n\n{chunk}"
return await asyncio.gather(*[process_chunk(c) for c in chunks])
```
**Incorrect — missing context generation and hybrid indexing:**
```python
def add_document(self, doc_id: str, text: str):
# No contextualization, no hybrid indexing
raw_chunks = self._chunk_text(text, 512)
embeddings = self.embed_model.embed(raw_chunks)
self.chunks.extend(embeddings)
```
**Correct — complete pipeline with contextualization and hybrid search:**
```python
def add_document(self, doc_id: str, text: str, chunk_size: int = 512):
raw_chunks = self._chunk_text(text, chunk_size)
contextualized = self._contextualize_batch(text, raw_chunks) # Add context
embeddings = self.embed_model.embed(contextualized) # Embed with context
for i, (raw, ctx, emb) in enumerate(zip(raw_chunks, contextualized, embeddings)):
self.chunks.append(ContextualChunk(
original=raw, contextualized=ctx, embedding=emb,
doc_id=doc_id, chunk_index=i
))
self._rebuild_bm25() # Hybrid BM25 + vector
```
**Key rules:**
- Use contextual retrieval when: documents have important metadata, chunks lose context, quality is critical
- Skip if: chunks are self-contained (Q&A pairs), low-latency indexing required, cost-sensitive with many small docs
- Parallel processing with semaphore (10 concurrent) for batch contextualization
- Prompt caching reduces cost by ~90% when processing many chunks from same document
### Prepend situational context to chunks before embedding to reduce retrieval failures — HIGH
## Contextual Retrieval — Context Prepending
Prepend situational context to chunks before embedding to preserve document-level meaning.
**The Problem:**
```
Original: "ACME Q3 2024 Earnings Report..."
Chunk: "Revenue increased 15% compared to the previous quarter."
Query: "What was ACME's Q3 2024 revenue growth?"
Result: Chunk doesn't mention "ACME" or "Q3 2024" — retrieval fails
```
**Context Generation:**
```python
import anthropic
client = anthropic.Anthropic()
CONTEXT_PROMPT = """
{document}
Here is the chunk we want to situate within the document:
{chunk}
Please give a short, succinct context (1-2 sentences) to situate this chunk
within the overall document. Focus on information that would help retrieval.
Answer only with the context, nothing else.
"""
def contextualize_chunk(document: str, chunk: str) -> str:
response = client.messages.create(
model="claude-sonnet-5",
max_tokens=150,
messages=[{"role": "user",
"content": CONTEXT_PROMPT.format(document=document, chunk=chunk)}]
)
return f"{response.content[0].text}\n\n{chunk}"
```
**With Prompt Caching (90% cost reduction):**
```python
def contextualize_cached(document: str, chunk: str) -> str:
response = client.messages.create(
model="claude-sonnet-5",
max_tokens=150,
messages=[{"role": "user", "content": [
{"type": "text", "text": f"\n{document}\n",
"cache_control": {"type": "ephemeral"}},
{"type": "text", "text": f"Situate this chunk (1-2 sentences):\n\n{chunk}\n"}
]}]
)
return f"{response.content[0].text}\n\n{chunk}"
```
**Incorrect — chunk without document context:**
```python
def index_chunk(chunk: str) -> str:
# Missing document context — retrieval will fail
embedding = embed_model.embed([chunk])[0]
return embedding
```
**Correct — prepend situational context before embedding:**
```python
def contextualize_chunk(document: str, chunk: str) -> str:
context = client.messages.create(
model="claude-sonnet-5",
max_tokens=150,
messages=[{"role": "user", "content": [
{"type": "text", "text": f"\n{document}\n",
"cache_control": {"type": "ephemeral"}}, # Cache for 90% cost reduction
{"type": "text", "text": f"Situate this chunk (1-2 sentences):\n\n{chunk}\n"}
]}]
)
return f"{context.content[0].text}\n\n{chunk}" # Prepend context
```
**Key rules:**
- Good context: "This chunk is from ACME Corp's Q3 2024 earnings report, specifically the revenue section."
- Bad context: "This is a chunk from the document." (too generic)
- Context length: 1-2 sentences — enough without excessive token overhead
- Use prompt caching (ephemeral) for 90% cost reduction when processing many chunks from same doc
### Construct basic RAG pipeline with proper context assembly and citation tracking — CRITICAL
## Basic RAG Pattern
Retrieve relevant documents, construct context, and generate grounded responses with citations.
**Basic RAG:**
```python
async def rag_query(question: str, top_k: int = 5) -> str:
"""Basic RAG: retrieve then generate."""
docs = await vector_db.search(question, limit=top_k)
context = "\n\n".join([
f"[{i+1}] {doc.text}"
for i, doc in enumerate(docs)
])
response = await llm.chat([
{"role": "system", "content":
"Answer using ONLY the provided context. "
"If not in context, say 'I don't have that information.'"},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
])
return response.content
```
**RAG with Citations:**
```python
async def rag_with_citations(question: str) -> dict:
"""RAG with inline citations [1], [2], etc."""
docs = await vector_db.search(question, limit=5)
context = "\n\n".join([
f"[{i+1}] {doc.text}\nSource: {doc.metadata['source']}"
for i, doc in enumerate(docs)
])
response = await llm.chat([
{"role": "system", "content":
"Answer with inline citations like [1], [2]. "
"End with a Sources section."},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
])
return {
"answer": response.content,
"sources": [doc.metadata['source'] for doc in docs]
}
```
**Incorrect — no citations, no grounding constraint:**
```python
async def rag_query(question: str) -> str:
docs = await vector_db.search(question, limit=5)
context = "\n\n".join([doc.text for doc in docs]) # No citations
response = await llm.chat([
{"role": "user", "content": f"{context}\n\n{question}"} # No grounding instruction
])
return response.content # No source tracking
```
**Correct — citations with grounding constraint:**
```python
async def rag_with_citations(question: str) -> dict:
docs = await vector_db.search(question, limit=5)
context = "\n\n".join([
f"[{i+1}] {doc.text}\nSource: {doc.metadata['source']}" # Numbered citations
for i, doc in enumerate(docs)
])
response = await llm.chat([
{"role": "system", "content": "Answer with inline citations like [1], [2]. Use ONLY the provided context."},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
])
return {"answer": response.content, "sources": [doc.metadata['source'] for doc in docs]}
```
**Key rules:**
- Always include citation tracking (`[1]`, `[2]`) for verifiable answers
- Set system prompt to constrain answers to retrieved context only
- Use top-k of 3-10 documents, temperature 0.1-0.3 for factual tasks
- Return sources alongside answers for transparency
### Manage context window size with sufficiency checks to balance relevance and cost — HIGH
## Context Window Management
Budget tokens for context and validate sufficiency before generation.
**Token Budget Fitting:**
```python
def fit_context(docs: list, max_tokens: int = 6000) -> list:
"""Truncate context to fit token budget."""
total_tokens = 0
selected = []
for doc in docs:
doc_tokens = count_tokens(doc.text)
if total_tokens + doc_tokens > max_tokens:
break
selected.append(doc)
total_tokens += doc_tokens
return selected
```
**Sufficiency Check (Google Research 2025):**
```python
from pydantic import BaseModel
class SufficiencyCheck(BaseModel):
is_sufficient: bool
confidence: float # 0.0-1.0
missing_info: str | None = None
async def rag_with_sufficiency(question: str, top_k: int = 5) -> str:
"""RAG with hallucination prevention via sufficiency check."""
docs = await vector_db.search(question, limit=top_k)
context = "\n\n".join([f"[{i+1}] {doc.text}" for i, doc in enumerate(docs)])
check = await llm.with_structured_output(SufficiencyCheck).ainvoke(
f"Does this context contain sufficient information to answer?\n"
f"Question: {question}\nContext:\n{context}"
)
if not check.is_sufficient and check.confidence > 0.7:
return f"I don't have enough information. Missing: {check.missing_info}"
if not check.is_sufficient and check.confidence <= 0.7:
more_docs = await vector_db.search(question, limit=top_k * 2)
context = "\n\n".join([f"[{i+1}] {doc.text}" for i, doc in enumerate(more_docs)])
return await generate_with_context(question, context)
```
**Incorrect — no token budget or sufficiency check:**
```python
async def rag_query(question: str) -> str:
docs = await vector_db.search(question, limit=100) # No limit!
context = "\n\n".join([doc.text for doc in docs]) # Could exceed context window
return await generate_with_context(question, context) # No sufficiency check
```
**Correct — token budget with sufficiency validation:**
```python
async def rag_with_sufficiency(question: str, top_k: int = 5) -> str:
docs = await vector_db.search(question, limit=top_k)
fitted = fit_context(docs, max_tokens=6000) # Budget enforcement
context = "\n\n".join([f"[{i+1}] {doc.text}" for i, doc in enumerate(fitted)])
check = await llm.with_structured_output(SufficiencyCheck).ainvoke(
f"Does this context contain sufficient information?\nQuestion: {question}\nContext:\n{context}"
)
if not check.is_sufficient and check.confidence > 0.7:
return f"I don't have enough information. Missing: {check.missing_info}"
return await generate_with_context(question, context)
```
**Key rules:**
- Keep context under 75% of model limit, reserve for system prompt + response
- Prioritize highest-relevance documents first
- Context budget: 4K-8K tokens typical for factual tasks
- RAG paradoxically increases hallucinations when context is insufficient — use sufficiency check
- Abstain when confidence > 0.7 and context is insufficient
### Combine semantic and keyword search with reciprocal rank fusion for best coverage — HIGH
## Hybrid Search (Semantic + Keyword)
Combine vector similarity and keyword matching using Reciprocal Rank Fusion for best coverage.
**Reciprocal Rank Fusion:**
```python
def reciprocal_rank_fusion(
semantic_results: list,
keyword_results: list,
k: int = 60
) -> list:
"""Combine semantic and keyword search with RRF."""
scores = {}
for rank, doc in enumerate(semantic_results):
scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + rank + 1)
for rank, doc in enumerate(keyword_results):
scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + rank + 1)
ranked_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
return [get_doc(id) for id in ranked_ids]
```
**Multi-list RRF (for query decomposition):**
```python
from collections import defaultdict
def multi_rrf(result_lists: list[list[dict]], k: int = 60) -> list[dict]:
"""Combine multiple ranked lists using RRF."""
scores: defaultdict[str, float] = defaultdict(float)
docs: dict[str, dict] = {}
for results in result_lists:
for rank, doc in enumerate(results, start=1):
doc_id = doc["id"]
scores[doc_id] += 1.0 / (k + rank)
docs[doc_id] = doc
ranked_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
return [docs[doc_id] for doc_id in ranked_ids]
```
**Incorrect — no reciprocal rank fusion, just simple averaging:**
```python
def hybrid_search(query: str, top_k: int = 10) -> list:
semantic = vector_search(query, top_k)
keyword = bm25_search(query, top_k)
# Naive merge without RRF
return semantic[:5] + keyword[:5]
```
**Correct — proper RRF combination:**
```python
def reciprocal_rank_fusion(semantic_results: list, keyword_results: list, k: int = 60) -> list:
scores = {}
for rank, doc in enumerate(semantic_results):
scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + rank + 1)
for rank, doc in enumerate(keyword_results):
scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + rank + 1)
ranked_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
return [get_doc(id) for id in ranked_ids]
```
**Key rules:**
- Default weight split: 40% BM25 / 60% vector (Anthropic research optimal)
- RRF smoothing constant k=60 is the standard — robust and parameter-free
- Retrieve 3x the final top-k for better RRF coverage (e.g., top-30 for final top-10)
- Normalize scores before combining if not using RRF
### Compose retrieval pipeline stages in correct order to avoid redundant processing — MEDIUM
## Pipeline Composition
Compose retrieval techniques in the right order for optimal results.
**Standard Pipeline:**
```
Query → [Decompose?] → [HyDE?] → [Retrieve] → [Rerank] → [Context Fit] → [Generate]
```
**Composition Pattern:**
```python
class RAGPipeline:
"""Composable RAG pipeline with optional stages."""
def __init__(self, retriever, reranker=None, hyde_service=None, decomposer=None):
self.retriever = retriever
self.reranker = reranker
self.hyde = hyde_service
self.decomposer = decomposer
async def query(self, question: str, top_k: int = 10) -> list[dict]:
# Stage 1: Query enhancement (optional)
queries = [question]
if self.decomposer:
concepts = await self.decomposer.decompose(question)
if len(concepts) > 1:
queries = concepts
# Stage 2: Retrieve (with optional HyDE)
all_results = []
for q in queries:
if self.hyde:
hyde_result = await self.hyde.generate(q)
results = await self.retriever.search_by_embedding(hyde_result.embedding, top_k=top_k * 3)
else:
results = await self.retriever.search(q, top_k=top_k * 3)
all_results.append(results)
# Stage 3: Fuse if multiple queries
if len(all_results) > 1:
merged = reciprocal_rank_fusion(all_results)
else:
merged = all_results[0]
# Stage 4: Rerank (optional)
if self.reranker:
merged = await self.reranker.rerank(question, merged, top_k=top_k)
return merged[:top_k]
```
**Incorrect — monolithic retrieval without composition:**
```python
async def query(question: str) -> list[dict]:
# No optional stages, fixed pipeline
docs = await retriever.search(question, top_k=10)
return docs
```
**Correct — composable pipeline with optional stages:**
```python
async def query(self, question: str, top_k: int = 10) -> list[dict]:
queries = [question]
if self.decomposer: # Optional decomposition
concepts = await self.decomposer.decompose(question)
if len(concepts) > 1:
queries = concepts
all_results = []
for q in queries:
if self.hyde: # Optional HyDE
hyde_result = await self.hyde.generate(q)
results = await self.retriever.search_by_embedding(hyde_result.embedding, top_k * 3)
else:
results = await self.retriever.search(q, top_k * 3)
all_results.append(results)
merged = reciprocal_rank_fusion(all_results) if len(all_results) > 1 else all_results[0]
if self.reranker: # Optional reranking
merged = await self.reranker.rerank(question, merged, top_k)
return merged[:top_k]
```
**Key rules:**
- Compose: Decompose → HyDE → Retrieve → Rerank → Context Fit → Generate
- HyDE adds ~500ms latency; use with fallback timeout (2-3s)
- Reranking adds ~50-500ms; retrieve more (3x), rerank to final top-k
- Query decomposition only when heuristic detects multi-concept query
- Each stage is optional — start simple, add stages as needed
### Implement production embedding pipelines with batching, caching, and cost optimization — MEDIUM
## Advanced Embedding Patterns
Production patterns for embedding at scale.
**Embedding Cache (Redis):**
```python
import hashlib
import json
import redis
class EmbeddingCache:
def __init__(self, redis_client: redis.Redis, ttl: int = 86400):
self.redis = redis_client
self.ttl = ttl
def _key(self, text: str, model: str) -> str:
h = hashlib.md5(f"{model}:{text}".encode()).hexdigest()
return f"emb:{h}"
async def get_or_embed(self, text: str, model: str, embed_fn) -> list[float]:
key = self._key(text, model)
cached = self.redis.get(key)
if cached:
return json.loads(cached)
embedding = await embed_fn(text)
self.redis.setex(key, self.ttl, json.dumps(embedding))
return embedding
```
**Batch Processing with Rate Limiting:**
```python
import asyncio
async def batch_embed(texts: list[str], embed_fn, batch_size: int = 100) -> list[list[float]]:
"""Embed texts in batches with rate limiting."""
results = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
embeddings = await embed_fn(batch)
results.extend(embeddings)
if i + batch_size < len(texts):
await asyncio.sleep(0.1) # Rate limit courtesy
return results
```
**Matryoshka Dimension Reduction:**
```python
# text-embedding-3 models support Matryoshka embeddings
# Truncate to fewer dimensions with minimal quality loss
response = client.embeddings.create(
model="text-embedding-3-large",
input="Your text",
dimensions=1536 # Reduce from 3072 to 1536 (saves 50% storage)
)
```
**Incorrect — no caching or batching, wasteful API calls:**
```python
async def embed_texts(texts: list[str]) -> list[list[float]]:
results = []
for text in texts: # One API call per text!
embedding = await client.embeddings.create(
model="text-embedding-3-large",
input=text
)
results.append(embedding.data[0].embedding)
return results
```
**Correct — cached batching with rate limiting:**
```python
async def batch_embed(texts: list[str], batch_size: int = 100) -> list[list[float]]:
results = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
# Check cache first
cached_keys = [cache.get(text) for text in batch]
uncached = [t for t, c in zip(batch, cached_keys) if not c]
if uncached:
embeddings = await client.embeddings.create(
model="text-embedding-3-large",
input=uncached,
dimensions=1536 # Matryoshka reduction
)
for text, emb in zip(uncached, embeddings.data):
cache.set(text, emb.embedding) # Cache for reuse
results.extend([cached or cache.get(t) for t, cached in zip(batch, cached_keys)])
await asyncio.sleep(0.1) # Rate limiting
return results
```
**Key rules:**
- Late Chunking: Embed full document, extract chunk vectors from contextualized tokens
- Cache aggressively — same text + model = same embedding, no need to recompute
- Batch size 100-500 per API call for optimal throughput
- Matryoshka: Truncate `text-embedding-3-large` from 3072 to 1536 dims with ~2% quality loss
- Rate limit: 0.1s delay between batches as courtesy to API providers
### Choose chunking strategies carefully since chunk boundaries determine retrieval quality — HIGH
## Chunking Strategies
Split documents into optimal chunks that preserve semantic meaning.
**Basic Overlapping Chunks:**
```python
def chunk_text(text: str, chunk_size: int = 512, overlap: int = 50) -> list[str]:
"""Split text into overlapping chunks for embedding."""
words = text.split()
chunks = []
for i in range(0, len(words), chunk_size - overlap):
chunk = " ".join(words[i:i + chunk_size])
if chunk:
chunks.append(chunk)
return chunks
```
**Semantic Boundary Chunking (OrchestKit Standard):**
```python
CHUNK_CONFIG = {
"target_tokens": 500, # ~400-600 tokens per chunk
"max_tokens": 800, # Hard limit
"overlap_tokens": 75, # ~15% overlap
"boundary_markers": [ # Prefer splitting at:
"\n## ", # H2 headers
"\n### ", # H3 headers
"\n\n", # Paragraphs
". ", # Sentences (last resort)
]
}
```
**Sentence-Aware Chunking:**
```python
def chunk_by_sentences(text: str, chunk_size: int = 512) -> list[str]:
sentences = text.split('. ')
chunks, current, current_len = [], [], 0
for sent in sentences:
if current_len + len(sent) > chunk_size and current:
chunks.append('. '.join(current) + '.')
current, current_len = [sent], len(sent)
else:
current.append(sent)
current_len += len(sent)
if current:
chunks.append('. '.join(current))
return chunks
```
**Incorrect — fixed-size splits without overlap or semantic boundaries:**
```python
def chunk_text(text: str) -> list[str]:
# Arbitrary splits, no overlap, breaks mid-sentence
return [text[i:i+500] for i in range(0, len(text), 500)]
```
**Correct — semantic boundary chunking with overlap:**
```python
def chunk_by_sentences(text: str, chunk_size: int = 512, overlap: int = 75) -> list[str]:
sentences = text.split('. ')
chunks, current, current_len = [], [], 0
for sent in sentences:
if current_len + len(sent) > chunk_size and current:
chunk_text = '. '.join(current) + '.'
chunks.append(chunk_text)
# Keep last few sentences for overlap
overlap_sents = current[-2:] if len(current) > 2 else current
current, current_len = overlap_sents, sum(len(s) for s in overlap_sents)
else:
current.append(sent)
current_len += len(sent)
if current:
chunks.append('. '.join(current))
return chunks
```
**Key rules:**
- Chunk size: 256-1024 tokens (512 typical sweet spot)
- Overlap: 10-20% for context continuity between chunks
- Include metadata (title, source, section) with each chunk
- Prefer semantic boundaries (headers, paragraphs) over fixed-size splits
- Not chunking long documents is a common mistake — context gets lost in embeddings
### Select embedding models and dimensions correctly to ensure index compatibility and quality — HIGH
## Embedding Models & API
Choose the right embedding model and use the API correctly.
**Quick Start:**
```python
from openai import OpenAI
client = OpenAI()
# Single text embedding
response = client.embeddings.create(
model="text-embedding-3-small",
input="Your text here"
)
vector = response.data[0].embedding # 1536 dimensions
# Batch embedding (efficient)
texts = ["text1", "text2", "text3"]
response = client.embeddings.create(
model="text-embedding-3-small",
input=texts
)
vectors = [item.embedding for item in response.data]
```
**Model Selection (2026-Q2 — verified current):**
| Model | Dims | Cost | Use Case |
|-------|------|------|----------|
| `text-embedding-3-small` | 1536 | $0.02/1M | General purpose (OpenAI) |
| `text-embedding-3-large` | 3072 | $0.13/1M | High accuracy (OpenAI) — Matryoshka-truncatable |
| `voyage-3.5` | 1024 | $0.06/1M | Production (OrchestKit default, launched 2025-05-20) |
| `voyage-3.5-lite` | 512 | $0.02/1M | Cost-tier, shorter queries, Matryoshka base |
| `voyage-3.5-large` | 2048 | $0.18/1M | Highest recall, multilingual + code |
| `gemini-embedding-001` (Gemini) | 3072 (MRL: 768/1536/3072) | $0.025/1M | Google stack / multilingual (GA 2025-07-14) |
| `nomic-embed-text` (Ollama) | 768 | Free | Local/CI |
Guidance:
- Default to `voyage-3.5` unless you have a reason not to. The `-lite` variant is the cost tier for short queries; `-large` wins on recall and multilingual content.
- `text-embedding-3-large` is still the strongest OpenAI option when you need Matryoshka dim reduction (3072 → 1536) without retraining.
- `gemini-embedding-001` is the Gemini option; use when you're already on Google's stack or need multilingual beyond Voyage's coverage. It defaults to 3072 dims but supports MRL output truncation (e.g. 768/1536) — set the output dimensionality explicitly if you need cost savings.
**Similarity Calculation:**
```python
import numpy as np
def cosine_similarity(a: list[float], b: list[float]) -> float:
a, b = np.array(a), np.array(b)
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
# 1.0 = identical, 0.0 = orthogonal
```
**Incorrect — mixing different embedding models:**
```python
# Index with one model
docs_embeddings = client.embeddings.create(
model="text-embedding-3-large", # 3072 dims
input=documents
)
# Query with different model
query_embedding = client.embeddings.create(
model="text-embedding-3-small", # 1536 dims - MISMATCH!
input=query
)
# Results will be nonsensical due to dimension mismatch
```
**Correct — consistent model for queries and documents:**
```python
MODEL = "text-embedding-3-small" # Use same model everywhere
# Index
docs_embeddings = client.embeddings.create(model=MODEL, input=documents)
# Query
query_embedding = client.embeddings.create(model=MODEL, input=query)
# Now cosine similarity is meaningful
similarity = cosine_similarity(query_embedding, docs_embeddings[0])
```
**Key rules:**
- Embed queries and documents with the SAME model — never mix
- Dimension reduction: Can truncate `text-embedding-3-large` to 1536 dims (Matryoshka)
- Batch size: 100-500 texts per API call for efficiency
- Cache embeddings — never re-embed unchanged content
- Most models return normalized vectors (cosine = dot product)
### Configure HyDE fallback strategy to avoid latency degradation from slow generation — MEDIUM
## HyDE Fallback Strategy
Implement graceful degradation when HyDE generation is too slow.
**Timeout with Fallback:**
```python
async def hyde_with_fallback(
query: str,
hyde_service: HyDEService,
embed_fn: callable,
timeout: float = 3.0,
) -> list[float]:
"""HyDE with fallback to direct embedding on timeout."""
try:
async with asyncio.timeout(timeout):
result = await hyde_service.generate(query)
return result.embedding
except TimeoutError:
# Fallback to direct query embedding
return await embed_fn(query)
```
**Performance Tips:**
- Use fast model (gpt-5-mini, claude-haiku-4-5) for generation
- Cache aggressively (queries often repeat)
- Set tight timeouts (2-3s) with fallback
- Keep hypothetical docs concise (100-200 tokens)
- Combine with query decomposition for best results
**Incorrect — no timeout or fallback, blocking forever:**
```python
async def hyde_search(query: str) -> list[float]:
# No timeout! May hang indefinitely
result = await hyde_service.generate(query)
return result.embedding
```
**Correct — timeout with graceful fallback:**
```python
async def hyde_with_fallback(query: str, timeout: float = 3.0) -> list[float]:
try:
async with asyncio.timeout(timeout):
result = await hyde_service.generate(query)
return result.embedding
except TimeoutError:
# Fallback to direct query embedding
return await embed_fn(query)
```
**Key rules:**
- Always implement timeout fallback — HyDE generation model may be slow or unavailable
- Default timeout: 2-3 seconds is the sweet spot (balances quality vs latency)
- Fallback to direct query embedding maintains functionality (just lower quality)
- Log fallback events to monitor HyDE generation reliability
### Bridge query-document vocabulary mismatch with hypothetical document embeddings via HyDE — HIGH
## HyDE Generation
Generate hypothetical answer documents to bridge vocabulary gaps in semantic search.
**The Problem:**
```
Query: "scaling async data pipelines"
Docs use: "event-driven messaging", "Apache Kafka", "message brokers"
-> Low similarity scores despite high relevance
```
**The Solution:**
```python
from openai import AsyncOpenAI
from pydantic import BaseModel
class HyDEResult(BaseModel):
original_query: str
hypothetical_doc: str
embedding: list[float]
async def generate_hyde(
query: str, llm: AsyncOpenAI, embed_fn: callable, max_tokens: int = 150
) -> HyDEResult:
"""Generate hypothetical document and embed it."""
response = await llm.chat.completions.create(
model="gpt-5-mini",
messages=[
{"role": "system", "content":
"Write a short paragraph that would answer this query. "
"Use technical terminology that documentation would use."},
{"role": "user", "content": query}
],
max_tokens=max_tokens,
temperature=0.3,
)
hypothetical_doc = response.choices[0].message.content
embedding = await embed_fn(hypothetical_doc) # Embed the hypothetical doc, not the query
return HyDEResult(
original_query=query,
hypothetical_doc=hypothetical_doc,
embedding=embedding,
)
```
**When to use HyDE:**
| Scenario | Use HyDE? |
|----------|-----------|
| Abstract/conceptual queries | Yes |
| Exact term searches | No (use keyword) |
| Code snippet searches | No |
| Natural language questions | Yes |
| Vocabulary mismatch suspected | Yes |
**Incorrect — embedding the query instead of hypothetical document:**
```python
async def generate_hyde(query: str) -> HyDEResult:
response = await llm.chat.completions.create(
model="gpt-5-mini",
messages=[{"role": "user", "content": query}],
max_tokens=150
)
hypothetical_doc = response.choices[0].message.content
embedding = await embed_fn(query) # WRONG: Embeds query, not hypothetical doc!
return HyDEResult(query, hypothetical_doc, embedding)
```
**Correct — embed the hypothetical document:**
```python
async def generate_hyde(query: str) -> HyDEResult:
response = await llm.chat.completions.create(
model="gpt-5-mini",
messages=[
{"role": "system", "content": "Write a short paragraph that would answer this query."},
{"role": "user", "content": query}
],
max_tokens=150,
temperature=0.3
)
hypothetical_doc = response.choices[0].message.content
embedding = await embed_fn(hypothetical_doc) # Embed the hypothetical doc!
return HyDEResult(query, hypothetical_doc, embedding)
```
**Key rules:**
- Embed the hypothetical document, NOT the original query
- Use fast/cheap model (gpt-5-mini, claude-haiku-4-5) for generation
- Temperature 0.3 for consistent, factual hypothetical docs
- Keep hypothetical docs concise: 100-200 tokens
- Adds ~500ms latency — always implement with timeout fallback
### Generate separate HyDE documents per concept for multi-topic vocabulary bridging — MEDIUM
## Per-Concept HyDE
Generate HyDE embeddings for each concept in multi-concept queries.
**Per-Concept Pattern:**
```python
async def batch_hyde(
concepts: list[str], hyde_service: HyDEService
) -> list[HyDEResult]:
"""Generate HyDE embeddings for multiple concepts in parallel."""
tasks = [hyde_service.generate(concept) for concept in concepts]
return await asyncio.gather(*tasks)
```
**With Caching:**
```python
class HyDEService:
def __init__(self, llm, embed_fn):
self.llm = llm
self.embed_fn = embed_fn
self._cache: dict[str, HyDEResult] = {}
def _cache_key(self, query: str) -> str:
return hashlib.md5(query.lower().strip().encode()).hexdigest()
async def generate(self, query: str) -> HyDEResult:
key = self._cache_key(query)
if key in self._cache:
return self._cache[key]
result = await generate_hyde(query, self.llm, self.embed_fn)
self._cache[key] = result
return result
```
**Incorrect — sequential HyDE generation, slow:**
```python
async def batch_hyde(concepts: list[str]) -> list[HyDEResult]:
results = []
for concept in concepts: # Sequential! Slow for many concepts
result = await hyde_service.generate(concept)
results.append(result)
return results
```
**Correct — parallel HyDE generation:**
```python
async def batch_hyde(concepts: list[str]) -> list[HyDEResult]:
# Parallel generation for all concepts simultaneously
tasks = [hyde_service.generate(concept) for concept in concepts]
return await asyncio.gather(*tasks)
```
**Key rules:**
- For multi-concept queries, decompose first then generate HyDE per concept
- Cache aggressively — queries often repeat
- Parallel generation with asyncio.gather for all concepts simultaneously
- Combine with query decomposition for best results on complex queries
### Chunk multimodal documents to preserve relationships between text, images, and tables — MEDIUM
## Multimodal Document Chunking
Chunk PDFs preserving images, tables, and text relationships.
**Multimodal Chunks:**
```python
from dataclasses import dataclass
from typing import Literal, Optional
@dataclass
class Chunk:
content: str
chunk_type: Literal["text", "image", "table", "chart"]
page: int
image_path: Optional[str] = None
embedding: Optional[list[float]] = None
def chunk_multimodal_document(pdf_path: str) -> list[Chunk]:
import fitz # PyMuPDF
doc = fitz.open(pdf_path)
chunks = []
for page_num, page in enumerate(doc):
text_blocks = page.get_text("blocks")
current_text = ""
for block in text_blocks:
if block[6] == 0: # Text block
current_text += block[4] + "\n"
else: # Image block
if current_text.strip():
chunks.append(Chunk(content=current_text.strip(), chunk_type="text", page=page_num))
current_text = ""
xref = block[7]
img = doc.extract_image(xref)
img_path = f"/tmp/page{page_num}_img{xref}.{img['ext']}"
with open(img_path, "wb") as f:
f.write(img["image"])
caption = generate_image_caption(img_path)
chunks.append(Chunk(content=caption, chunk_type="image", page=page_num, image_path=img_path))
if current_text.strip():
chunks.append(Chunk(content=current_text.strip(), chunk_type="text", page=page_num))
return chunks
```
**Incorrect — text-only chunking, loses images and tables:**
```python
def chunk_pdf(pdf_path: str) -> list[str]:
import fitz
doc = fitz.open(pdf_path)
chunks = []
for page in doc:
chunks.append(page.get_text()) # Text only, images lost!
return chunks
```
**Correct — multimodal chunking with images and captions:**
```python
def chunk_multimodal_document(pdf_path: str) -> list[Chunk]:
import fitz
doc = fitz.open(pdf_path)
chunks = []
for page_num, page in enumerate(doc):
text_blocks = page.get_text("blocks")
for block in text_blocks:
if block[6] == 0: # Text block
chunks.append(Chunk(content=block[4], chunk_type="text", page=page_num))
else: # Image block
xref = block[7]
img = doc.extract_image(xref)
img_path = f"/tmp/page{page_num}_img{xref}.{img['ext']}"
with open(img_path, "wb") as f:
f.write(img["image"])
caption = generate_image_caption(img_path)
chunks.append(Chunk(content=caption, chunk_type="image", page=page_num, image_path=img_path))
return chunks
```
**Key rules:**
- Extract images separately and generate captions for text-based search
- Preserve page numbers for citation and navigation
- Use PyMuPDF (fitz) for reliable PDF extraction
- Process large PDFs in page-range batches (CC 2.1.30: max 20 pages per Read)
- Always store image paths alongside embeddings for result display
### Use multimodal embedding models for cross-modal search across text and images — HIGH
## Multimodal Embeddings
Embed images and text in the same vector space for cross-modal retrieval.
**Model Selection:**
| Model | Context | Modalities | Best For |
|-------|---------|------------|----------|
| **Voyage multimodal-3.5** | 32K tokens | Text + Image | Long documents |
| **SigLIP 2** | Standard | Text + Image | Large-scale retrieval |
| **CLIP ViT-L/14** | 77 tokens | Text + Image | General purpose |
| **ImageBind** | Standard | 6 modalities | Audio/video included |
| **ColPali** | Document | Text + Image | PDF/document RAG |
**CLIP Embeddings:**
```python
from transformers import CLIPProcessor, CLIPModel
import torch
model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")
def embed_image(image_path: str) -> list[float]:
image = Image.open(image_path)
inputs = processor(images=image, return_tensors="pt")
with torch.no_grad():
embeddings = model.get_image_features(**inputs)
embeddings = embeddings / embeddings.norm(dim=-1, keepdim=True)
return embeddings[0].tolist()
def embed_text(text: str) -> list[float]:
inputs = processor(text=[text], return_tensors="pt", padding=True)
with torch.no_grad():
embeddings = model.get_text_features(**inputs)
embeddings = embeddings / embeddings.norm(dim=-1, keepdim=True)
return embeddings[0].tolist()
```
**Voyage Multimodal-3 (Long Context):**
```python
import voyageai
client = voyageai.Client()
def embed_multimodal(texts=None, images=None) -> list[list[float]]:
inputs = []
if texts:
inputs.extend([{"type": "text", "content": t} for t in texts])
if images:
for path in images:
with open(path, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
inputs.append({"type": "image", "content": f"data:image/png;base64,{b64}"})
return client.multimodal_embed(inputs=inputs, model="voyage-multimodal-3.5").embeddings
```
**Incorrect — using text-only embeddings for images:**
```python
def embed_image(image_path: str) -> list[float]:
# Using text embedding model for images - wrong modality!
caption = generate_caption(image_path)
return text_embed_model.embed([caption])[0] # Loses visual features
```
**Correct — multimodal embeddings for cross-modal search:**
```python
def embed_image(image_path: str) -> list[float]:
image = Image.open(image_path)
inputs = processor(images=image, return_tensors="pt")
with torch.no_grad():
embeddings = model.get_image_features(**inputs)
embeddings = embeddings / embeddings.norm(dim=-1, keepdim=True) # Normalize
return embeddings[0].tolist()
```
**Key rules:**
- Normalize embeddings for cosine similarity (CLIP already normalized)
- Voyage multimodal-3.5 for long documents (32K context, released 2025-05-20)
- SigLIP 2 for large-scale production retrieval
- Always embed both images AND captions for maximum coverage
### Build unified multimodal RAG pipeline that merges cross-modal results with deduplication — MEDIUM
## Multimodal RAG Pipeline
Build end-to-end multimodal retrieval and generation pipeline.
**Hybrid Retrieval:**
```python
class MultimodalRAG:
def __init__(self, vector_db, vision_model, text_model):
self.vector_db = vector_db
self.vision_model = vision_model
self.text_model = text_model
async def retrieve(self, query: str, query_image: str = None, top_k: int = 10) -> list[dict]:
results = []
text_emb = embed_text(query)
text_results = await self.vector_db.search(embedding=text_emb, top_k=top_k)
results.extend(text_results)
if query_image:
img_emb = embed_image(query_image)
img_results = await self.vector_db.search(embedding=img_emb, top_k=top_k)
results.extend(img_results)
# Dedupe by doc_id, keep highest score
seen = {}
for r in results:
doc_id = r["metadata"]["doc_id"]
if doc_id not in seen or r["score"] > seen[doc_id]["score"]:
seen[doc_id] = r
return sorted(seen.values(), key=lambda x: x["score"], reverse=True)[:top_k]
```
**Multimodal Generation:**
```python
async def generate_with_context(query: str, chunks: list[Chunk], model: str = "claude-opus-4-8") -> str:
content = []
# Add images first (attention positioning)
for chunk in chunks:
if chunk.chunk_type == "image" and chunk.image_path:
b64, media_type = encode_image_base64(chunk.image_path)
content.append({"type": "image", "source": {"type": "base64", "media_type": media_type, "data": b64}})
# Add text context
text_context = "\n\n".join([f"[Page {c.page}]: {c.content}" for c in chunks if c.chunk_type == "text"])
content.append({"type": "text", "text": f"Context:\n{text_context}\n\nQuestion: {query}"})
response = client.messages.create(model=model, max_tokens=4096, messages=[{"role": "user", "content": content}])
return response.content[0].text
```
**Incorrect — no deduplication, fragmented results:**
```python
async def retrieve(self, query: str, top_k: int = 10) -> list[dict]:
text_emb = embed_text(query)
text_results = await self.vector_db.search(embedding=text_emb, top_k=top_k)
img_emb = embed_image(query_image)
img_results = await self.vector_db.search(embedding=img_emb, top_k=top_k)
# No deduplication! Same doc may appear twice
return text_results + img_results
```
**Correct — deduplicated cross-modal results:**
```python
async def retrieve(self, query: str, query_image: str = None, top_k: int = 10) -> list[dict]:
results = []
text_emb = embed_text(query)
text_results = await self.vector_db.search(embedding=text_emb, top_k=top_k)
results.extend(text_results)
if query_image:
img_emb = embed_image(query_image)
img_results = await self.vector_db.search(embedding=img_emb, top_k=top_k)
results.extend(img_results)
# Dedupe by doc_id, keep highest score
seen = {}
for r in results:
doc_id = r["metadata"]["doc_id"]
if doc_id not in seen or r["score"] > seen[doc_id]["score"]:
seen[doc_id] = r
return sorted(seen.values(), key=lambda x: x["score"], reverse=True)[:top_k]
```
**Key rules:**
- Deduplicate by document ID — keep highest scoring result per document
- Place images before text in generation prompt (attention positioning)
- Always embed both image features AND text captions for maximum coverage
- Use hybrid approach: CLIP + text embeddings for best accuracy
- Missing image URL storage is a common mistake — always store paths for display
### Implement PGVector hybrid search with FULL OUTER JOIN and RRF fusion ranking — HIGH
## PGVector Hybrid Search (SQL)
Hybrid vector+keyword search with RRF fusion in SQLAlchemy.
**Hybrid Search Query:**
```python
async def hybrid_search(query: str, query_embedding: list[float], top_k: int = 10) -> list[Chunk]:
FETCH_MULTIPLIER = 3
K = 60 # RRF smoothing constant
# Vector search subquery
vector_subq = (
select(Chunk.id,
func.row_number().over(
order_by=Chunk.embedding.cosine_distance(query_embedding)
).label("vector_rank"))
.limit(top_k * FETCH_MULTIPLIER)
.subquery()
)
# Keyword search subquery
ts_query = func.plainto_tsquery("english", query)
keyword_subq = (
select(Chunk.id,
func.row_number().over(
order_by=func.ts_rank_cd(Chunk.content_tsvector, ts_query).desc()
).label("keyword_rank"))
.where(Chunk.content_tsvector.op("@@")(ts_query))
.limit(top_k * FETCH_MULTIPLIER)
.subquery()
)
# RRF fusion with FULL OUTER JOIN
rrf_subq = (
select(
func.coalesce(vector_subq.c.id, keyword_subq.c.id).label("chunk_id"),
(func.coalesce(1.0 / (K + vector_subq.c.vector_rank), 0.0) +
func.coalesce(1.0 / (K + keyword_subq.c.keyword_rank), 0.0)
).label("rrf_score"))
.select_from(vector_subq.outerjoin(keyword_subq, ..., full=True))
.order_by("rrf_score DESC")
.limit(top_k)
.subquery()
)
return await session.execute(
select(Chunk).join(rrf_subq, Chunk.id == rrf_subq.c.chunk_id)
)
```
**RRF Formula:**
```python
rrf_score = 1/(k + vector_rank) + 1/(k + keyword_rank) # k=60 (standard)
```
**Incorrect — separate queries without RRF fusion:**
```python
async def hybrid_search(query: str, embedding: list[float], top_k: int = 10):
# Separate queries, no fusion!
vector_results = await session.execute(
select(Chunk).order_by(Chunk.embedding.cosine_distance(embedding)).limit(top_k)
)
keyword_results = await session.execute(
select(Chunk).where(Chunk.content_tsvector.op("@@")(plainto_tsquery(query))).limit(top_k)
)
# Naive merge, no RRF
return list(vector_results) + list(keyword_results)
```
**Correct — RRF fusion with FULL OUTER JOIN:**
```python
async def hybrid_search(query: str, query_embedding: list[float], top_k: int = 10):
K = 60 # RRF smoothing constant
# Vector search subquery
vector_subq = select(Chunk.id, func.row_number().over(
order_by=Chunk.embedding.cosine_distance(query_embedding)
).label("vector_rank")).limit(top_k * 3).subquery()
# Keyword search subquery
keyword_subq = select(Chunk.id, func.row_number().over(
order_by=func.ts_rank_cd(Chunk.content_tsvector, plainto_tsquery(query)).desc()
).label("keyword_rank")).limit(top_k * 3).subquery()
# RRF fusion with FULL OUTER JOIN
rrf_subq = select(
func.coalesce(vector_subq.c.id, keyword_subq.c.id).label("chunk_id"),
(func.coalesce(1.0 / (K + vector_subq.c.vector_rank), 0.0) +
func.coalesce(1.0 / (K + keyword_subq.c.keyword_rank), 0.0)).label("rrf_score")
).select_from(vector_subq.outerjoin(keyword_subq, ..., full=True)).order_by("rrf_score DESC").limit(top_k)
return await session.execute(select(Chunk).join(rrf_subq, Chunk.id == rrf_subq.c.chunk_id))
```
**Key rules:**
- Use FULL OUTER JOIN to catch docs found by only one search method
- 3x fetch multiplier for better RRF coverage (30 per method for top 10 final)
- RRF smoothing constant k=60 is the standard
- Use `func.coalesce(..., 0.0)` for documents found by only one method
- `plainto_tsquery` for user queries (handles multi-word safely)
### Choose correct PGVector index type to avoid 17x slower queries in production — HIGH
## PGVector Index Strategies
Choose and configure the right vector index for your workload.
**Index Comparison:**
| Metric | IVFFlat | HNSW |
|--------|---------|------|
| **Query speed** | 50ms | 3ms (17x faster) |
| **Index time** | 2 min | 20 min |
| **Best for** | < 100k vectors | 100k+ vectors |
| **Recall@10** | 0.85-0.95 | 0.95-0.99 |
**HNSW Configuration (Recommended):**
```sql
-- Create HNSW index
CREATE INDEX idx_chunks_embedding_hnsw ON chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- Query-time tuning
SET hnsw.ef_search = 40; -- Higher = better recall, slower
-- Iterative scan for filtered queries (pgvector 0.8+)
SET hnsw.iterative_scan = 'relaxed_order';
```
**Search Type Comparison:**
| Aspect | Semantic (Vector) | Keyword (BM25) |
|--------|-------------------|----------------|
| **Query** | Embedding similarity | Exact word matches |
| **Strengths** | Synonyms, concepts | Exact phrases, rare terms |
| **Weaknesses** | Exact matches, technical terms | No semantic understanding |
| **Index** | HNSW (pgvector) | GIN (tsvector) |
**Incorrect — no index, sequential scan on every query:**
```sql
-- No index! Sequential scan is 17x slower
SELECT * FROM chunks
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;
```
**Correct — HNSW index for fast queries:**
```sql
-- Create HNSW index
CREATE INDEX idx_chunks_embedding_hnsw ON chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- Query-time tuning
SET hnsw.ef_search = 40; -- Higher = better recall
-- Now queries are 17x faster
SELECT * FROM chunks
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;
```
**Key rules:**
- Use HNSW for production (scales to millions, 17x faster queries)
- IVFFlat only for >1000 queries/sec where index build time matters
- m=16, ef_construction=64 are good defaults for most workloads
- Set `hnsw.ef_search = 40` at query time for production recall
- Use `iterative_scan = 'relaxed_order'` for filtered vector queries
## Storage Reduction — halfvec & Binary Quantization (pgvector 0.7+)
For high-dimensional embeddings (e.g. 3072-dim `gemini-embedding-001` / `text-embedding-3-large`), full `vector` storage and HNSW graphs get expensive. pgvector 0.7+ offers two reductions:
**halfvec — 16-bit floats (~50% storage, negligible recall loss):**
```sql
ALTER TABLE chunks ALTER COLUMN embedding TYPE halfvec(3072);
CREATE INDEX ON chunks USING hnsw (embedding halfvec_cosine_ops)
WITH (m = 16, ef_construction = 64);
```
**Binary quantization — bit vectors for a coarse first pass (~32x smaller), then exact rerank:**
```sql
CREATE INDEX ON chunks
USING hnsw ((binary_quantize(embedding)::bit(3072)) bit_hamming_ops);
-- two-phase: cheap Hamming shortlist, then exact cosine rerank
SELECT * FROM (
SELECT * FROM chunks
ORDER BY binary_quantize(embedding)::bit(3072) <~> binary_quantize($1)::bit(3072)
LIMIT 200
) shortlist
ORDER BY embedding <=> $1
LIMIT 10;
```
**When to use:**
- `halfvec`: sensible default for any 1536+ dim embedding — ~50% smaller index, recall within ~0.5%
- `binary_quantize` + rerank: 3072-dim at very large scale where even halfvec HNSW is too big; expect a recall hit unless you rerank the shortlist exactly
### Filter PGVector results by metadata and boost scores for improved retrieval relevance — MEDIUM
## PGVector Metadata Filtering & Patterns
Filter and boost search results using metadata.
**Filtered Search:**
```python
results = await hybrid_search(
query="binary search",
query_embedding=embedding,
content_type_filter=["code_block"]
)
```
**Similarity Threshold:**
```python
results = await hybrid_search(query, embedding, top_k=50)
filtered = [r for r in results if (1 - r.vector_distance) >= 0.75][:10]
```
**Multi-Query Retrieval:**
```python
queries = ["machine learning", "ML algorithms", "neural networks"]
all_results = [await hybrid_search(q, embed(q)) for q in queries]
final = deduplicate_and_rerank(all_results)
```
**Redis 8 FT.HYBRID Alternative:**
| Aspect | pgvector | Redis 8 FT.HYBRID |
|--------|----------|-------------------|
| **Setup** | Medium | Low |
| **RRF** | Manual SQL | Native `COMBINE RRF` |
| **Latency** | 5-20ms | 2-5ms |
| **Persistence** | ACID | AOF/RDB |
| **Max dataset** | Billions | Memory-bound (~100M) |
**Incorrect — no metadata filtering, irrelevant results:**
```python
# Returns all content types mixed together
results = await session.execute(
select(Chunk).order_by(Chunk.embedding.cosine_distance(embedding)).limit(10)
)
```
**Correct — filtered search with similarity threshold:**
```python
# Filter by content_type and similarity threshold
results = await session.execute(
select(Chunk)
.where(Chunk.content_type == "code_block") # Pre-filter
.order_by(Chunk.embedding.cosine_distance(embedding))
.limit(50)
)
# Apply similarity threshold
filtered = [r for r in results if (1 - r.vector_distance) >= 0.75][:10]
```
**Key rules:**
- Metadata boosting (title/path matching) adds +6% MRR
- Pre-filter by content_type for targeted search
- Similarity threshold 0.75 is a good default for filtering low-relevance results
- Choose pgvector for: ACID, complex joins, large datasets, existing PostgreSQL
- Choose Redis 8 for: sub-5ms latency, caching layer, simpler deployment
### Design PGVector schema with pre-computed tsvector columns and proper index configuration — HIGH
## PGVector Database Schema
Production schema with pre-computed tsvector and HNSW index.
**Schema:**
```sql
CREATE TABLE chunks (
id UUID PRIMARY KEY,
document_id UUID REFERENCES documents(id),
content TEXT NOT NULL,
embedding vector(1024), -- PGVector
content_tsvector tsvector GENERATED ALWAYS AS (
to_tsvector('english', content)
) STORED,
section_title TEXT,
content_type TEXT,
created_at TIMESTAMP DEFAULT NOW()
);
-- HNSW index for vector search
CREATE INDEX idx_chunks_embedding ON chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- GIN index for keyword search
CREATE INDEX idx_chunks_content_tsvector ON chunks
USING gin (content_tsvector);
```
**Incorrect — computing tsvector at query time, slow:**
```sql
CREATE TABLE chunks (
id UUID PRIMARY KEY,
content TEXT NOT NULL,
embedding vector(1024)
);
-- Slow query: computes tsvector every time!
SELECT * FROM chunks
WHERE to_tsvector('english', content) @@ plainto_tsquery('search query');
```
**Correct — pre-computed tsvector as GENERATED column:**
```sql
CREATE TABLE chunks (
id UUID PRIMARY KEY,
content TEXT NOT NULL,
embedding vector(1024),
content_tsvector tsvector GENERATED ALWAYS AS (
to_tsvector('english', content)
) STORED -- Pre-computed, 5-10x faster
);
-- GIN index for fast keyword search
CREATE INDEX idx_chunks_content_tsvector ON chunks USING gin (content_tsvector);
-- HNSW index for fast vector search
CREATE INDEX idx_chunks_embedding ON chunks
USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);
-- Fast query using pre-computed tsvector
SELECT * FROM chunks WHERE content_tsvector @@ plainto_tsquery('search query');
```
**Key rules:**
- Pre-compute tsvector as GENERATED column — 5-10x faster than `to_tsvector()` at query time
- Use `vector(1024)` for Voyage-3 embeddings (match your model dimension)
- HNSW index with m=16, ef_construction=64 for production workloads
- Always include document_id FK for document-level operations
- Include content_type for filtered search (code, text, table)
### Decompose complex multi-topic queries with parallel retrieval and RRF fusion — HIGH
## Query Decomposition + RRF Fusion
Break complex queries into concepts, retrieve separately, fuse with RRF.
**LLM Decomposition:**
```python
from pydantic import BaseModel, Field
class ConceptExtraction(BaseModel):
concepts: list[str] = Field(..., min_length=1, max_length=5)
reasoning: str | None = None
async def decompose_query(query: str, llm: AsyncOpenAI) -> list[str]:
response = await llm.chat.completions.create(
model="gpt-5-mini",
messages=[
{"role": "system", "content":
"Extract 2-4 independent concepts from this query. "
"Each concept should be searchable on its own."},
{"role": "user", "content": query}
],
response_format={"type": "json_object"},
temperature=0,
)
result = ConceptExtraction.model_validate_json(response.choices[0].message.content)
return result.concepts
```
**Decomposed Search with RRF:**
```python
async def decomposed_search(query: str, search_fn, llm, top_k: int = 10) -> list[dict]:
if not is_multi_concept_heuristic(query):
return await search_fn(query, limit=top_k)
concepts = await decompose_query(query, llm)
if len(concepts) <= 1:
return await search_fn(query, limit=top_k)
# Parallel retrieval per concept
tasks = [search_fn(concept, limit=top_k) for concept in concepts]
results_per_concept = await asyncio.gather(*tasks)
return reciprocal_rank_fusion(results_per_concept, k=60)[:top_k]
```
**Incorrect — no decomposition, single query for complex topics:**
```python
async def search(query: str, top_k: int = 10) -> list[dict]:
# "How does authentication affect database performance?"
# Single query misses one of the two concepts
return await vector_search(query, limit=top_k)
```
**Correct — decompose and fuse with RRF:**
```python
async def decomposed_search(query: str, top_k: int = 10) -> list[dict]:
if not is_multi_concept_heuristic(query):
return await search_fn(query, limit=top_k) # Fast path
# Decompose: "authentication" + "database performance"
concepts = await decompose_query(query, llm)
if len(concepts) <= 1:
return await search_fn(query, limit=top_k)
# Parallel retrieval per concept
tasks = [search_fn(concept, limit=top_k) for concept in concepts]
results_per_concept = await asyncio.gather(*tasks)
# Fuse with RRF
return reciprocal_rank_fusion(results_per_concept, k=60)[:top_k]
```
**Key rules:**
- Max 2-4 concepts per query (more increases latency without proportional benefit)
- Use gpt-5-mini for decomposition (fast, cheap, good at concept extraction)
- RRF fusion is robust and parameter-free for combining per-concept results
- Cache decomposition results — same query often asked repeatedly
- Set timeout with fallback to original query if decomposition fails
### Detect multi-concept queries with heuristic fast-path to avoid unnecessary LLM decomposition — MEDIUM
## Multi-Concept Query Detection
Fast heuristic to determine if query decomposition is needed.
**Heuristic Detection (Fast Path):**
```python
MULTI_CONCEPT_INDICATORS = [
" vs ", " versus ", " compared to ", " or ",
" and ", " with ", " affect ", " impact ",
"difference between", "relationship between",
]
def is_multi_concept_heuristic(query: str) -> bool:
"""Fast check for multi-concept indicators (<1ms)."""
query_lower = query.lower()
return any(ind in query_lower for ind in MULTI_CONCEPT_INDICATORS)
```
**When to Decompose:**
| Query Type | Decompose? |
|------------|------------|
| "What is X?" | No |
| "X vs Y" | Yes |
| "How does X affect Y?" | Yes |
| "Best practices for X" | No |
| "X and Y in Z" | Yes |
| "Difference between X, Y, Z" | Yes |
**Incorrect — always decomposing, even for simple queries:**
```python
async def search(query: str, top_k: int = 10) -> list[dict]:
# Always calls LLM for decomposition, even for "What is React?"
concepts = await decompose_query(query, llm) # Wasteful!
tasks = [search_fn(concept, limit=top_k) for concept in concepts]
return await asyncio.gather(*tasks)
```
**Correct — heuristic fast path before LLM decomposition:**
```python
async def search(query: str, top_k: int = 10) -> list[dict]:
if not is_multi_concept_heuristic(query): # Sub-millisecond check
return await search_fn(query, limit=top_k) # Fast path
# Only call LLM if heuristic detects multi-concept
concepts = await decompose_query(query, llm)
tasks = [search_fn(concept, limit=top_k) for concept in concepts]
return await asyncio.gather(*tasks)
def is_multi_concept_heuristic(query: str) -> bool:
query_lower = query.lower()
return any(ind in query_lower for ind in [" vs ", " and ", "difference between"])
```
**Key rules:**
- Heuristic first (sub-millisecond), LLM decomposition only if heuristic triggers
- Single-concept queries should skip decomposition entirely (no LLM cost)
- Keywords: "vs", "compared to", "affect", "difference between" indicate multi-concept
- This is the fast path — always check before calling the LLM decomposer
### Combine query decomposition with HyDE for comprehensive vocabulary-bridged retrieval coverage — MEDIUM
## Decomposition + HyDE Combo
Best of both: decompose into concepts, then generate HyDE for each concept.
**Combined Pattern:**
```python
async def decomposed_hyde_search(
query: str,
decomposer: QueryDecomposer,
hyde_service: HyDEService,
vector_search: callable,
top_k: int = 10,
) -> list[dict]:
"""Decomposition + HyDE for maximum coverage."""
# Decompose query into concepts
concepts = await decomposer.get_concepts(query)
# Generate HyDE for each concept in parallel
hyde_results = await asyncio.gather(*[
hyde_service.generate(concept) for concept in concepts
])
# Search with HyDE embeddings
search_tasks = [
vector_search(embedding=hr.embedding, limit=top_k)
for hr in hyde_results
]
results_per_concept = await asyncio.gather(*search_tasks)
# Fuse results with RRF
return reciprocal_rank_fusion(results_per_concept)[:top_k]
```
**Incorrect — sequential decomposition and HyDE, slow:**
```python
async def search(query: str, top_k: int = 10) -> list[dict]:
concepts = await decompose_query(query, llm)
all_results = []
for concept in concepts: # Sequential! Slow
hyde_result = await hyde_service.generate(concept)
results = await vector_search(embedding=hyde_result.embedding, limit=top_k)
all_results.append(results)
return reciprocal_rank_fusion(all_results)[:top_k]
```
**Correct — parallel HyDE generation and search:**
```python
async def decomposed_hyde_search(query: str, top_k: int = 10) -> list[dict]:
concepts = await decomposer.get_concepts(query)
# Generate HyDE for each concept in parallel
hyde_results = await asyncio.gather(*[
hyde_service.generate(concept) for concept in concepts
])
# Search with HyDE embeddings in parallel
search_tasks = [
vector_search(embedding=hr.embedding, limit=top_k) for hr in hyde_results
]
results_per_concept = await asyncio.gather(*search_tasks)
# Fuse with RRF
return reciprocal_rank_fusion(results_per_concept)[:top_k]
```
**Key rules:**
- Use this combo for complex queries with both multi-concept AND vocabulary mismatch
- Decompose first, then HyDE per concept, then parallel search, then RRF fuse
- Total latency: ~1-2s (decomposition + HyDE generation + parallel search)
- Cache both decomposition and HyDE results for efficiency
- This is the most expensive retrieval path — use only when simpler methods fail
### Combine base, LLM, and recency scores for robust multi-signal reranking — MEDIUM
## Combined Scoring & Reranking Service
Combine multiple scoring signals with weighted average and timeout fallback.
**Combined Scoring:**
```python
def combined_rerank(
documents: list[dict], llm_scores: dict[str, float],
alpha: float = 0.3, beta: float = 0.5, gamma: float = 0.2
) -> list[dict]:
scored = []
for doc in documents:
base = doc.get("score", 0.5)
llm = llm_scores.get(doc["id"], 0.5)
recency = calculate_recency_score(doc.get("created_at"))
final = (alpha * base) + (beta * llm) + (gamma * recency)
scored.append({**doc, "score": final,
"score_components": {"base": base, "llm": llm, "recency": recency}})
scored.sort(key=lambda x: x["score"], reverse=True)
return scored
```
**Service with Timeout Fallback:**
```python
class ReRankingService:
def __init__(self, llm: AsyncOpenAI, timeout_seconds: float = 5.0):
self.llm = llm
self.timeout = timeout_seconds
async def rerank(self, query: str, documents: list[dict], top_k: int = 10) -> list[dict]:
if len(documents) <= top_k:
return documents
try:
async with asyncio.timeout(self.timeout):
return await llm_rerank(query, documents, self.llm, top_k)
except TimeoutError:
return sorted(documents, key=lambda x: x.get("score", 0), reverse=True)[:top_k]
```
**Incorrect — single scoring signal without timeout:**
```python
async def rerank(query: str, documents: list[dict], top_k: int = 10) -> list[dict]:
# Only uses LLM score, no timeout, no fallback
llm_scores = await llm_rerank(query, documents, llm) # May hang!
return sorted(documents, key=lambda x: llm_scores.get(x["id"], 0), reverse=True)[:top_k]
```
**Correct — combined scoring with timeout fallback:**
```python
async def rerank(query: str, documents: list[dict], top_k: int = 10) -> list[dict]:
if len(documents) <= top_k:
return documents # Skip if no benefit
try:
async with asyncio.timeout(5.0): # 5s timeout
llm_scores = await llm_rerank(query, documents, llm)
# Combined scoring: 30% base + 50% LLM + 20% recency
scored = []
for doc in documents:
base = doc.get("score", 0.5)
llm = llm_scores.get(doc["id"], 0.5)
recency = calculate_recency_score(doc.get("created_at"))
final = 0.3 * base + 0.5 * llm + 0.2 * recency
scored.append({**doc, "score": final})
return sorted(scored, key=lambda x: x["score"], reverse=True)[:top_k]
except TimeoutError:
# Fallback to base ranking
return sorted(documents, key=lambda x: x.get("score", 0), reverse=True)[:top_k]
```
**Key rules:**
- Default weights: 30% base retrieval + 50% LLM score + 20% recency
- Always set timeout (5s) with fallback to base ranking
- Skip reranking if document count <= top_k (no benefit)
- Cache scores: same query+doc pair = same score
- Store score components for debugging and tuning
### Rerank results with cross-encoder models for accurate query-document relevance scoring — HIGH
## Cross-Encoder Reranking
Re-score retrieved documents with cross-encoder for higher precision.
**Cross-Encoder Pattern:**
```python
from sentence_transformers import CrossEncoder
class CrossEncoderReranker:
def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L6-v2"):
self.model = CrossEncoder(model_name)
def rerank(self, query: str, documents: list[dict], top_k: int = 10) -> list[dict]:
pairs = [(query, doc["content"]) for doc in documents]
scores = self.model.predict(pairs)
scored_docs = list(zip(documents, scores))
scored_docs.sort(key=lambda x: x[1], reverse=True)
return [{**doc, "score": float(score)} for doc, score in scored_docs[:top_k]]
```
**Model Selection:**
| Model | Latency | Cost | Quality |
|-------|---------|------|---------|
| `cross-encoder/ms-marco-MiniLM-L6-v2` | ~50ms | Free | Good |
| `BAAI/bge-reranker-large` | ~100ms | Free | Better |
| `cohere rerank-v4.0-pro` | ~200ms | $1/1K | Best |
**Incorrect — retrieving few, no reranking:**
```python
async def search(query: str) -> list[dict]:
# Retrieve only 10, no reranking - misses good results
return await vector_search(query, limit=10)
```
**Correct — retrieve many, rerank to few:**
```python
async def search_with_reranking(query: str) -> list[dict]:
# Retrieve many candidates
candidates = await vector_search(query, limit=50)
# Rerank with cross-encoder
pairs = [(query, doc["content"][:400]) for doc in candidates]
scores = cross_encoder.predict(pairs)
scored_docs = list(zip(candidates, scores))
scored_docs.sort(key=lambda x: x[1], reverse=True)
# Return top 10 after reranking
return [{**doc, "score": float(score)} for doc, score in scored_docs[:10]]
```
**Key rules:**
- Retrieve many (50-100), rerank to few (10) — "retrieve more, rerank less"
- Cross-encoder processes query+doc pair together (slow but accurate)
- Default model: `ms-marco-MiniLM-L6-v2` (good quality, free, ~50ms)
- Truncate document content to 200-400 chars for reranking efficiency
### Use LLM reranking for domain-adaptive scoring without deploying a dedicated model — MEDIUM
## LLM Reranking
Score document relevance using LLM in a single batch call.
**LLM Batch Reranking:**
```python
async def llm_rerank(query: str, documents: list[dict], llm: AsyncOpenAI, top_k: int = 10) -> list[dict]:
docs_text = "\n\n".join([f"[Doc {i+1}]\n{doc['content'][:300]}..." for i, doc in enumerate(documents)])
response = await llm.chat.completions.create(
model="gpt-5-mini",
messages=[
{"role": "system", "content": "Rate each document's relevance to the query (0.0-1.0).\nOutput one score per line."},
{"role": "user", "content": f"Query: {query}\n\nDocuments:\n{docs_text}"}
],
temperature=0,
)
scores = parse_scores(response.choices[0].message.content, len(documents))
scored_docs = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [{**doc, "score": score} for doc, score in scored_docs[:top_k]]
def parse_scores(response: str, expected_count: int) -> list[float]:
scores = []
for line in response.strip().split("\n"):
try:
scores.append(max(0.0, min(1.0, float(line.strip()))))
except ValueError:
scores.append(0.5)
while len(scores) < expected_count:
scores.append(0.5)
return scores[:expected_count]
```
**Cohere Rerank API:**
```python
import cohere
class CohereReranker:
def __init__(self, api_key: str):
self.client = cohere.ClientV2(api_key=api_key)
def rerank(self, query: str, documents: list[dict], top_k: int = 10) -> list[dict]:
results = self.client.rerank(
model="rerank-v4.0-pro", query=query,
documents=[{"type": "text", "text": doc["content"]} for doc in documents],
top_n=top_k, max_tokens_per_doc=4096
)
return [{**documents[r.index], "score": r.relevance_score} for r in results.results]
```
**Incorrect — one LLM call per document, extremely slow:**
```python
async def llm_rerank(query: str, documents: list[dict]) -> list[dict]:
scores = []
for doc in documents: # Sequential LLM calls!
response = await llm.chat.completions.create(
model="gpt-5-mini",
messages=[{"role": "user", "content": f"Rate relevance (0-1):\nQuery: {query}\nDoc: {doc['content']}"}]
)
scores.append(float(response.choices[0].message.content))
return sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
```
**Correct — batch all docs in one LLM call:**
```python
async def llm_rerank(query: str, documents: list[dict], top_k: int = 10) -> list[dict]:
# Batch all docs in ONE LLM call
docs_text = "\n\n".join([
f"[Doc {i+1}]\n{doc['content'][:300]}..." # Truncate
for i, doc in enumerate(documents)
])
response = await llm.chat.completions.create(
model="gpt-5-mini",
messages=[
{"role": "system", "content": "Rate each document's relevance (0.0-1.0). One score per line."},
{"role": "user", "content": f"Query: {query}\n\nDocuments:\n{docs_text}"}
],
temperature=0
)
scores = parse_scores(response.choices[0].message.content, len(documents))
scored_docs = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [{**doc, "score": score} for doc, score in scored_docs[:top_k]]
```
**Key rules:**
- Batch all docs in one LLM call (reduces latency vs per-doc calls)
- Truncate to 200-400 chars per doc for LLM reranking
- Parse scores defensively (default 0.5 on parse error)
- LLM reranking at ~500ms, Cohere ClientV2/rerank-v4.0-pro at ~200ms
- Set timeout with fallback to base ranking
- Use ClientV2 (deprecated Client) and rerank-v4.0-pro (deprecated v3.0) models
---
## Checklists (2)
### Rag Quality
# RAG Quality Checklist
Quality assurance for agentic RAG implementations.
## Retrieval Quality
- [ ] Semantic search configured with appropriate embedding model
- [ ] Chunk size optimized (512-1024 tokens typical)
- [ ] Chunk overlap configured (10-20% of chunk size)
- [ ] Metadata filtering implemented for scoping
- [ ] Top-k tuned for precision/recall balance
## Document Grading
- [ ] Relevance grading implemented (binary or scored)
- [ ] Grading prompt tested with diverse queries
- [ ] Threshold tuned for false positive/negative balance
- [ ] Fallback behavior defined for low-relevance results
## Query Transformation
- [ ] Query rewriting enabled for failed retrievals
- [ ] Maximum retry count configured (2-3 typical)
- [ ] Query decomposition for multi-concept queries
- [ ] HyDE integration for vocabulary mismatch
## Web Fallback (CRAG)
- [ ] Web search integration configured
- [ ] Rate limiting for web search API
- [ ] Result filtering and quality check
- [ ] Source attribution for web results
## Self-RAG Patterns
- [ ] Adaptive retrieval decision logic implemented
- [ ] Reflection tokens for quality assessment
- [ ] Skip retrieval path for simple queries
- [ ] Confidence thresholds calibrated
## Generation Quality
- [ ] Context formatting optimized
- [ ] Citation/source attribution enforced
- [ ] Hallucination detection enabled
- [ ] Output length appropriate
## Error Handling
- [ ] Graceful degradation on retrieval failure
- [ ] Fallback responses configured
- [ ] Retry logic with exponential backoff
- [ ] Error logging and alerting
## Performance
- [ ] Retrieval latency acceptable (<500ms)
- [ ] Caching for repeated queries
- [ ] Batch embedding for efficiency
- [ ] Async execution where possible
## Monitoring
- [ ] Retrieval metrics tracked (precision, recall)
- [ ] Query success/failure rates logged
- [ ] Web fallback frequency monitored
- [ ] User feedback integration
### Search Implementation Checklist
# PGVector Hybrid Search Implementation Checklist
Use this checklist when implementing semantic + keyword search with PGVector.
## Pre-Implementation
### Index Strategy Planning
- [ ] **Choose vector algorithm** - HNSW (recommended) or IVFFlat
- [ ] **Select embedding model** - OpenAI (1536), Voyage AI (1024), etc.
- [ ] **Determine dimensions** - Match model output dimensions
- [ ] **Plan distance metric** - Cosine (most common) or L2/Inner Product
- [ ] **Set HNSW parameters** - m=16, ef_construction=64 (good defaults)
### Embedding Model Selection
- [ ] **Test embedding quality** - Validate on sample queries
- [ ] **Measure embedding latency** - API call time
- [ ] **Budget embedding costs** - Track usage for bulk ingestion
- [ ] **Plan batch embedding** - Batch API calls for efficiency
- [ ] **Cache embeddings** - Store in database, don't re-compute
### RRF Configuration
- [ ] **Set fetch multiplier** - 3x (retrieve 30 for top-10 results)
- [ ] **Choose RRF constant (k)** - 60 (standard value)
- [ ] **Plan score normalization** - Use rank, not raw scores
- [ ] **Define boosting factors** - Section title (1.5x), path (1.15x), code (1.2x)
- [ ] **Set similarity threshold** - Minimum cosine similarity (e.g., 0.75)
### Schema Design
- [ ] **Define chunks table** - id, content, embedding, metadata
- [ ] **Add tsvector column** - Pre-computed for keyword search
- [ ] **Plan metadata fields** - section_title, section_path, content_type
- [ ] **Add timestamps** - created_at, updated_at
- [ ] **Foreign keys** - Link to documents/artifacts
## Implementation
### Database Schema
```sql
-- 1. Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;
-- 2. Create chunks table
CREATE TABLE chunks (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
document_id UUID REFERENCES documents(id) ON DELETE CASCADE,
content TEXT NOT NULL,
-- Vector embedding (match model dimensions)
embedding vector(1024), -- Voyage AI 1024 dims
-- Pre-computed tsvector for full-text search
content_tsvector tsvector GENERATED ALWAYS AS (
to_tsvector('english', content)
) STORED,
-- Metadata
section_title TEXT,
section_path TEXT,
chunk_index INT,
content_type TEXT, -- 'code_block', 'paragraph', 'list'
-- Timestamps
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- 3. Create indexes
-- Vector search (HNSW for speed)
CREATE INDEX idx_chunks_embedding ON chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- Full-text search (GIN for tsvector)
CREATE INDEX idx_chunks_content_tsvector ON chunks
USING gin (content_tsvector);
-- Metadata indexes
CREATE INDEX idx_chunks_document_id ON chunks(document_id);
CREATE INDEX idx_chunks_content_type ON chunks(content_type);
```
- [ ] pgvector extension enabled
- [ ] Chunks table created
- [ ] Embedding column dimensions match model
- [ ] tsvector column generated and stored
- [ ] HNSW index created for vectors
- [ ] GIN index created for tsvector
- [ ] Metadata indexes created
### Vector Search Query
```python
from sqlalchemy import select, func
from pgvector.sqlalchemy import Vector
async def vector_search(
query_embedding: list[float],
top_k: int = 10,
content_type_filter: list[str] | None = None
) -> list[Chunk]:
"""Perform vector similarity search."""
# Fetch multiplier for better RRF coverage
FETCH_MULTIPLIER = 3
fetch_limit = top_k * FETCH_MULTIPLIER
# Build query
query = (
select(
Chunk.id,
(Chunk.embedding.cosine_distance(query_embedding)).label("distance"),
func.row_number().over(
order_by=Chunk.embedding.cosine_distance(query_embedding)
).label("rank")
)
.where(Chunk.embedding.isnot(None))
)
# Apply content type filter
if content_type_filter:
query = query.where(Chunk.content_type.in_(content_type_filter))
query = query.limit(fetch_limit).subquery("vector_results")
result = await session.execute(query)
return result.all()
```
- [ ] Query embedding passed as parameter
- [ ] Cosine distance calculated
- [ ] Row number (rank) computed
- [ ] Fetch multiplier applied (3x)
- [ ] Content type filter optional
- [ ] Returns top-k * 3 results
### Keyword Search Query
```python
async def keyword_search(
query: str,
top_k: int = 10,
content_type_filter: list[str] | None = None
) -> list[Chunk]:
"""Perform BM25 keyword search."""
FETCH_MULTIPLIER = 3
fetch_limit = top_k * FETCH_MULTIPLIER
# Generate tsquery from plain text
ts_query = func.plainto_tsquery("english", query)
# Build query
query = (
select(
Chunk.id,
func.ts_rank_cd(Chunk.content_tsvector, ts_query).label("score"),
func.row_number().over(
order_by=func.ts_rank_cd(Chunk.content_tsvector, ts_query).desc()
).label("rank")
)
.where(Chunk.content_tsvector.op("@@")(ts_query))
)
# Apply content type filter
if content_type_filter:
query = query.where(Chunk.content_type.in_(content_type_filter))
query = query.limit(fetch_limit).subquery("keyword_results")
result = await session.execute(query)
return result.all()
```
- [ ] Uses pre-indexed content_tsvector (not to_tsvector on query)
- [ ] plainto_tsquery handles special characters
- [ ] ts_rank_cd for BM25-like scoring
- [ ] Row number (rank) computed
- [ ] Fetch multiplier applied
- [ ] Only matches where tsvector matches query
### Reciprocal Rank Fusion (RRF)
```python
async def hybrid_search(
query: str,
query_embedding: list[float],
top_k: int = 10,
content_type_filter: list[str] | None = None
) -> list[Chunk]:
"""Combine vector + keyword search with RRF."""
FETCH_MULTIPLIER = 3
fetch_limit = top_k * FETCH_MULTIPLIER
K = 60 # RRF smoothing constant
# ===== 1. VECTOR SEARCH =====
vector_subquery = (
select(
Chunk.id,
(Chunk.embedding.cosine_distance(query_embedding)).label("vector_distance"),
func.row_number().over(
order_by=Chunk.embedding.cosine_distance(query_embedding)
).label("vector_rank")
)
.where(Chunk.embedding.isnot(None))
)
if content_type_filter:
vector_subquery = vector_subquery.where(
Chunk.content_type.in_(content_type_filter)
)
vector_subquery = vector_subquery.limit(fetch_limit).subquery("vector_results")
# ===== 2. KEYWORD SEARCH =====
ts_query = func.plainto_tsquery("english", query)
keyword_subquery = (
select(
Chunk.id,
func.ts_rank_cd(Chunk.content_tsvector, ts_query).label("bm25_score"),
func.row_number().over(
order_by=func.ts_rank_cd(Chunk.content_tsvector, ts_query).desc()
).label("keyword_rank")
)
.where(Chunk.content_tsvector.op("@@")(ts_query))
)
if content_type_filter:
keyword_subquery = keyword_subquery.where(
Chunk.content_type.in_(content_type_filter)
)
keyword_subquery = keyword_subquery.limit(fetch_limit).subquery("keyword_results")
# ===== 3. RECIPROCAL RANK FUSION =====
rrf_query = (
select(
func.coalesce(
vector_subquery.c.id,
keyword_subquery.c.id
).label("chunk_id"),
(
func.coalesce(1.0 / (K + vector_subquery.c.vector_rank), 0.0) +
func.coalesce(1.0 / (K + keyword_subquery.c.keyword_rank), 0.0)
).label("rrf_score"),
vector_subquery.c.vector_distance,
keyword_subquery.c.bm25_score
)
.select_from(
vector_subquery.outerjoin(
keyword_subquery,
vector_subquery.c.id == keyword_subquery.c.id,
full=True # FULL OUTER JOIN
)
)
.order_by(literal("rrf_score").desc())
.limit(top_k)
).subquery("rrf_results")
# ===== 4. FETCH FULL CHUNKS =====
final_query = (
select(Chunk, rrf_query.c.rrf_score)
.join(rrf_query, Chunk.id == rrf_query.c.chunk_id)
.order_by(rrf_query.c.rrf_score.desc())
)
result = await session.execute(final_query)
chunks = result.all()
return chunks
```
- [ ] Both vector and keyword searches executed
- [ ] Full outer join combines results
- [ ] RRF score = 1/(k+rank_vector) + 1/(k+rank_keyword)
- [ ] Results sorted by RRF score descending
- [ ] Top-k returned
- [ ] Full chunk objects fetched
### Metadata Boosting
```python
def apply_metadata_boosting(
chunks: list[tuple[Chunk, float]],
query: str
) -> list[tuple[Chunk, float]]:
"""Boost RRF scores based on metadata relevance."""
boosted_chunks = []
for chunk, rrf_score in chunks:
boost_factor = 1.0
# Boost section titles (1.5x)
if chunk.section_title and query_matches_section_title(chunk.section_title, query):
boost_factor *= 1.5
# Boost document path (1.15x)
if chunk.section_path and query_matches_path(chunk.section_path, query):
boost_factor *= 1.15
# Boost code blocks for technical queries (1.2x)
if is_technical_query(query) and chunk.content_type == "code_block":
boost_factor *= 1.2
boosted_chunks.append((chunk, rrf_score * boost_factor))
# Re-sort by boosted score
boosted_chunks.sort(key=lambda x: x[1], reverse=True)
return boosted_chunks
def query_matches_section_title(section_title: str, query: str) -> bool:
"""Check if query keywords appear in section title."""
query_terms = set(query.lower().split())
title_terms = set(section_title.lower().split())
return len(query_terms & title_terms) > 0
def is_technical_query(query: str) -> bool:
"""Detect technical queries (code-focused)."""
technical_keywords = {
"function", "class", "method", "code", "implement",
"algorithm", "syntax", "example", "snippet"
}
query_terms = set(query.lower().split())
return len(query_terms & technical_keywords) > 0
```
- [ ] Boosting applied after RRF
- [ ] Section title matching implemented
- [ ] Document path matching implemented
- [ ] Technical query detection implemented
- [ ] Results re-sorted after boosting
## Verification
### Golden Dataset Testing
```python
import pytest
@pytest.mark.asyncio
async def test_hybrid_search_golden_dataset():
"""Test hybrid search against golden queries."""
golden_queries = load_golden_queries() # Load test cases
results = []
for query_data in golden_queries:
query = query_data["query"]
expected_chunks = query_data["expected_chunk_ids"]
# Generate embedding
embedding = await embed_text(query)
# Perform search
retrieved = await hybrid_search(query, embedding, top_k=10)
retrieved_ids = {c.id for c in retrieved}
# Check if expected chunks are in top 10
found = len(expected_chunks & retrieved_ids)
results.append({
"query": query,
"expected": len(expected_chunks),
"found": found,
"pass": found == len(expected_chunks)
})
# Calculate metrics
pass_rate = sum(r["pass"] for r in results) / len(results)
mrr = calculate_mrr(results)
print(f"Pass Rate: {pass_rate:.1%}")
print(f"MRR: {mrr:.3f}")
assert pass_rate >= 0.90, f"Pass rate {pass_rate:.1%} below 90% threshold"
def calculate_mrr(results: list[dict]) -> float:
"""Calculate Mean Reciprocal Rank."""
reciprocal_ranks = []
for result in results:
if result["found"] > 0:
# Assume first expected chunk found at rank 1 (simplified)
reciprocal_ranks.append(1.0)
else:
reciprocal_ranks.append(0.0)
return sum(reciprocal_ranks) / len(reciprocal_ranks)
```
- [ ] **Golden dataset loaded** - 98+ test queries
- [ ] **Pass rate measured** - Target: 90%+
- [ ] **MRR calculated** - Mean Reciprocal Rank
- [ ] **Hard queries tested** - Technical, ambiguous queries
- [ ] **Failures analyzed** - Inspect failing queries
### Retrieval Quality Metrics
```python
@pytest.mark.asyncio
async def test_retrieval_quality_metrics():
"""Measure retrieval quality metrics."""
test_cases = load_golden_queries()
precision_at_k = []
recall_at_k = []
for case in test_cases:
query = case["query"]
relevant_chunks = set(case["expected_chunk_ids"])
# Perform search
embedding = await embed_text(query)
retrieved = await hybrid_search(query, embedding, top_k=10)
retrieved_ids = {c.id for c in retrieved}
# Precision@10: Relevant chunks in top-10 / 10
precision = len(relevant_chunks & retrieved_ids) / 10
precision_at_k.append(precision)
# Recall@10: Relevant chunks in top-10 / Total relevant
recall = len(relevant_chunks & retrieved_ids) / len(relevant_chunks)
recall_at_k.append(recall)
avg_precision = sum(precision_at_k) / len(precision_at_k)
avg_recall = sum(recall_at_k) / len(recall_at_k)
print(f"Precision@10: {avg_precision:.3f}")
print(f"Recall@10: {avg_recall:.3f}")
assert avg_precision >= 0.70, "Precision@10 below 70%"
assert avg_recall >= 0.85, "Recall@10 below 85%"
```
- [ ] **Precision@10** - Target: 70%+ (relevant in top-10)
- [ ] **Recall@10** - Target: 85%+ (found most relevant)
- [ ] **MRR** - Target: 0.65+ (relevant chunks ranked high)
- [ ] **nDCG** - Normalized Discounted Cumulative Gain (optional)
### Performance Benchmarks
```python
@pytest.mark.asyncio
async def test_search_latency():
"""Measure search latency."""
import time
query = "How to implement binary search in Python?"
embedding = await embed_text(query)
# Measure vector search latency
start = time.perf_counter()
vector_results = await vector_search(embedding, top_k=30)
vector_latency = (time.perf_counter() - start) * 1000
# Measure keyword search latency
start = time.perf_counter()
keyword_results = await keyword_search(query, top_k=30)
keyword_latency = (time.perf_counter() - start) * 1000
# Measure hybrid search latency
start = time.perf_counter()
hybrid_results = await hybrid_search(query, embedding, top_k=10)
hybrid_latency = (time.perf_counter() - start) * 1000
print(f"Vector search: {vector_latency:.2f}ms")
print(f"Keyword search: {keyword_latency:.2f}ms")
print(f"Hybrid search: {hybrid_latency:.2f}ms")
# Latency targets
assert vector_latency < 100, f"Vector search latency {vector_latency:.2f}ms > 100ms"
assert keyword_latency < 50, f"Keyword search latency {keyword_latency:.2f}ms > 50ms"
assert hybrid_latency < 150, f"Hybrid search latency {hybrid_latency:.2f}ms > 150ms"
```
- [ ] **Vector search** - < 100ms (HNSW index)
- [ ] **Keyword search** - < 50ms (GIN index)
- [ ] **Hybrid search** - < 150ms (combined)
- [ ] **P95 latency** - 95th percentile acceptable
- [ ] **Index scans** - Verify indexes used (EXPLAIN ANALYZE)
### Index Performance Validation
```sql
-- Check if indexes are being used
EXPLAIN ANALYZE
SELECT id, embedding <=> '[0.1, 0.2, ..., 0.9]' AS distance
FROM chunks
ORDER BY distance
LIMIT 30;
-- Should show "Index Scan using idx_chunks_embedding"
-- NOT "Seq Scan" (sequential scan = no index!)
```
- [ ] **Vector index used** - EXPLAIN shows "Index Scan using idx_chunks_embedding"
- [ ] **Keyword index used** - EXPLAIN shows "Bitmap Index Scan using idx_chunks_content_tsvector"
- [ ] **No sequential scans** - Avoid full table scans
- [ ] **Index size reasonable** - Check pg_indexes view
- [ ] **Vacuum/Analyze run** - Update statistics for query planner
## Post-Implementation
### Production Monitoring
- [ ] **Search latency dashboard** - P50, P95, P99 latency
- [ ] **Retrieval quality tracking** - Pass rate, MRR over time
- [ ] **Index bloat** - Monitor index size growth
- [ ] **Query patterns** - Log common queries, identify gaps
- [ ] **Error rate** - Track search failures
### Optimization Opportunities
- [ ] **Tune HNSW parameters** - Increase m or ef_construction for accuracy
- [ ] **Increase fetch multiplier** - 3x → 5x for better RRF coverage
- [ ] **Add more boosting** - Domain-specific metadata boosts
- [ ] **Multi-query retrieval** - Generate query variations
- [ ] **Hybrid query rewriting** - Expand acronyms, synonyms
### Index Maintenance
- [ ] **Run VACUUM ANALYZE** - Weekly or after bulk inserts
- [ ] **Rebuild indexes** - If bloated (pg_repack)
- [ ] **Monitor index usage** - Drop unused indexes
- [ ] **Update statistics** - Ensure query planner has fresh stats
- [ ] **Test on production-scale data** - Validate performance at scale
## Troubleshooting
| Issue | Check |
|-------|-------|
| Slow vector search | HNSW index exists? Dimensions match? Increase m/ef_construction? |
| Slow keyword search | GIN index on tsvector? Using content_tsvector, not to_tsvector()? |
| Low pass rate | Increase fetch multiplier, add boosting, check embeddings quality |
| No keyword matches | Check tsvector generation, query language (English?), special chars |
| Wrong results | Validate RRF logic, check boosting factors, inspect rankings |
| Index not used | Run ANALYZE, check query plan (EXPLAIN), verify index conditions |
## OrchestKit Integration
```python
# Example: Search for content in OrchestKit
from app.shared.services.search.search_service import SearchService
search_service = SearchService()
results = await search_service.search(
query="How to implement hybrid search?",
top_k=10,
filters={"content_type": ["code_block", "paragraph"]}
)
# Results include chunk content, metadata, and RRF score
for chunk, score in results:
print(f"Score: {score:.4f} | {chunk.section_title}")
print(chunk.content[:200])
```
- [ ] Search service integrated with API endpoints
- [ ] Results exposed via `/api/v1/search` endpoint
- [ ] Filters applied for content_type, document_id
- [ ] Results paginated (offset/limit)
- [ ] Searchable in frontend UI
## References
- **PGVector Docs**: https://github.com/pgvector/pgvector
- **OrchestKit Implementation**: `backend/app/db/repositories/chunk_repository.py`
- **Search Service**: `backend/app/shared/services/search/search_service.py`
- **Constants**: `backend/app/core/constants.py`
- **Related Skill**: `the `database-schema-designer` skill`