rag-architect — Skillopedia

RAG Architect Core Workflow 1. Requirements Analysis — Identify retrieval needs, latency constraints, accuracy requirements, and scale 2. Vector Store Design — Select database, schema design, indexing strategy, sharding approach 3. Chunking Strategy — Document splitting, overlap, semantic boundaries, metadata enrichment 4. Retrieval Pipeline — Embedding selection, query transformation, hybrid search, reranking 5. Evaluation & Iteration — Metrics tracking, retrieval debugging, continuous optimization For each step, validate before moving on (see checkpoints below). Reference Guide Load detaile…

\n\n lines = text.split('\\n')\n chunks = []\n current_chunk_lines = []\n current_heading = None\n current_level = 0\n heading_stack = [] # For breadcrumb context\n\n for line in lines:\n header_match = re.match(header_pattern, line)\n\n if header_match:\n # Save current chunk if exists\n if current_chunk_lines:\n chunk_text = '\\n'.join(current_chunk_lines)\n if len(chunk_text.strip()) > 0:\n prefix = f\"# {current_heading}\\n\\n\" if include_heading_in_chunk and current_heading else \"\"\n chunks.append(MarkdownChunk(\n text=prefix + chunk_text,\n heading=current_heading,\n heading_level=current_level,\n metadata={\"breadcrumb\": \" > \".join(heading_stack)}\n ))\n\n # Update heading context\n level = len(header_match.group(1))\n heading = header_match.group(2).strip()\n\n # Maintain heading stack for breadcrumbs\n while heading_stack and current_level >= level:\n heading_stack.pop()\n current_level -= 1\n\n heading_stack.append(heading)\n current_heading = heading\n current_level = level\n current_chunk_lines = []\n\n else:\n current_chunk_lines.append(line)\n\n # Check chunk size\n current_text = '\\n'.join(current_chunk_lines)\n if len(current_text) > max_chunk_size:\n # Split at paragraph boundary\n paragraphs = current_text.split('\\n\\n')\n if len(paragraphs) > 1:\n split_point = len('\\n\\n'.join(paragraphs[:-1]))\n chunk_text = current_text[:split_point]\n prefix = f\"# {current_heading}\\n\\n\" if include_heading_in_chunk and current_heading else \"\"\n chunks.append(MarkdownChunk(\n text=prefix + chunk_text,\n heading=current_heading,\n heading_level=current_level,\n metadata={\"breadcrumb\": \" > \".join(heading_stack)}\n ))\n current_chunk_lines = [current_text[split_point:].strip()]\n\n # Don't forget the last chunk\n if current_chunk_lines:\n chunk_text = '\\n'.join(current_chunk_lines)\n if len(chunk_text.strip()) > 0:\n prefix = f\"# {current_heading}\\n\\n\" if include_heading_in_chunk and current_heading else \"\"\n chunks.append(MarkdownChunk(\n text=prefix + chunk_text,\n heading=current_heading,\n heading_level=current_level,\n metadata={\"breadcrumb\": \" > \".join(heading_stack)}\n ))\n\n return chunks\n```\n\n### Code-Aware Chunking\n\n```python\nimport re\nfrom dataclasses import dataclass\n\n@dataclass\nclass CodeChunk:\n text: str\n language: str | None\n chunk_type: str # \"code\", \"text\", \"mixed\"\n\ndef chunk_with_code_blocks(\n text: str,\n max_chunk_size: int = 1500\n) -> list[CodeChunk]:\n \"\"\"Chunk text while keeping code blocks intact.\"\"\"\n # Pattern to match code blocks\n code_block_pattern = r'```(\\w+)?\\n(.*?)```'\n\n chunks = []\n last_end = 0\n\n for match in re.finditer(code_block_pattern, text, re.DOTALL):\n # Text before code block\n text_before = text[last_end:match.start()].strip()\n if text_before:\n # Chunk the text portion\n text_chunks = recursive_chunk(text_before, max_chunk_size)\n chunks.extend([\n CodeChunk(text=t, language=None, chunk_type=\"text\")\n for t in text_chunks\n ])\n\n # Code block (keep intact if possible)\n language = match.group(1)\n code_content = match.group(2)\n full_block = match.group(0)\n\n if len(full_block) \u003c= max_chunk_size:\n chunks.append(CodeChunk(\n text=full_block,\n language=language,\n chunk_type=\"code\"\n ))\n else:\n # Split large code blocks by function/class\n code_chunks = split_code_block(code_content, language, max_chunk_size)\n chunks.extend(code_chunks)\n\n last_end = match.end()\n\n # Remaining text after last code block\n remaining = text[last_end:].strip()\n if remaining:\n text_chunks = recursive_chunk(remaining, max_chunk_size)\n chunks.extend([\n CodeChunk(text=t, language=None, chunk_type=\"text\")\n for t in text_chunks\n ])\n\n return chunks\n\ndef split_code_block(code: str, language: str, max_size: int) -> list[CodeChunk]:\n \"\"\"Split code block at logical boundaries.\"\"\"\n # Simple function/class boundary splitting for Python\n if language == \"python\":\n pattern = r'\\n(?=def |class |async def )'\n elif language in [\"javascript\", \"typescript\"]:\n pattern = r'\\n(?=function |class |const |export )'\n else:\n pattern = r'\\n\\n'\n\n parts = re.split(pattern, code)\n chunks = []\n current = \"\"\n\n for part in parts:\n if len(current) + len(part) > max_size and current:\n chunks.append(CodeChunk(\n text=f\"```{language}\\n{current}```\",\n language=language,\n chunk_type=\"code\"\n ))\n current = part\n else:\n current += part\n\n if current:\n chunks.append(CodeChunk(\n text=f\"```{language}\\n{current}```\",\n language=language,\n chunk_type=\"code\"\n ))\n\n return chunks\n```\n\n---\n\n## Contextual/Agentic Chunking\n\n```python\nfrom openai import OpenAI\n\ndef contextual_chunk(\n document: str,\n max_chunk_size: int = 1500\n) -> list[dict]:\n \"\"\"Use LLM to add context to each chunk.\"\"\"\n # First, do structural chunking\n base_chunks = recursive_chunk(document, max_chunk_size)\n\n client = OpenAI()\n contextualized_chunks = []\n\n for chunk in base_chunks:\n # Generate contextual summary\n response = client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": \"\"\"Provide a brief context for this document chunk.\n Include: what topic it covers, how it relates to the broader document,\n and key concepts mentioned. Keep it under 100 words.\"\"\"\n },\n {\n \"role\": \"user\",\n \"content\": f\"Document excerpt:\\n\\n{chunk}\"\n }\n ],\n max_tokens=150\n )\n\n context = response.choices[0].message.content\n\n contextualized_chunks.append({\n \"text\": chunk,\n \"context\": context,\n \"text_with_context\": f\"Context: {context}\\n\\nContent: {chunk}\"\n })\n\n return contextualized_chunks\n```\n\n### Propositions-Based Chunking\n\n```python\ndef extract_propositions(text: str) -> list[str]:\n \"\"\"Extract atomic propositions from text using LLM.\"\"\"\n client = OpenAI()\n\n response = client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": \"\"\"Extract atomic propositions from the text.\n Each proposition should:\n - Be a single, complete fact\n - Be self-contained (understandable without context)\n - Include necessary entity references\n\n Return as a JSON array of strings.\"\"\"\n },\n {\n \"role\": \"user\",\n \"content\": text\n }\n ],\n response_format={\"type\": \"json_object\"}\n )\n\n import json\n result = json.loads(response.choices[0].message.content)\n return result.get(\"propositions\", [])\n\n# Usage: For very fine-grained retrieval\npropositions = extract_propositions(document_text)\n# Each proposition becomes its own retrievable unit\n```\n\n---\n\n## Late Chunking (for Long-Context Embeddings)\n\n```python\nfrom transformers import AutoTokenizer, AutoModel\nimport torch\n\nclass LateChunker:\n \"\"\"\n Late chunking: embed full document, then pool token embeddings into chunks.\n Preserves full document context while creating retrievable chunks.\n \"\"\"\n\n def __init__(self, model_name: str = \"jinaai/jina-embeddings-v2-base-en\"):\n self.tokenizer = AutoTokenizer.from_pretrained(model_name)\n self.model = AutoModel.from_pretrained(model_name, trust_remote_code=True)\n self.model.eval()\n\n def chunk_and_embed(\n self,\n text: str,\n chunk_size: int = 512,\n overlap: int = 64\n ) -> list[dict]:\n \"\"\"\n Embed full document, then create chunk embeddings via mean pooling.\n \"\"\"\n # Tokenize full document\n inputs = self.tokenizer(\n text,\n return_tensors=\"pt\",\n truncation=True,\n max_length=8192 # Model's max context\n )\n\n # Get token-level embeddings\n with torch.no_grad():\n outputs = self.model(**inputs)\n token_embeddings = outputs.last_hidden_state[0] # [seq_len, hidden_dim]\n\n # Get token-to-text mapping\n tokens = self.tokenizer.convert_ids_to_tokens(inputs[\"input_ids\"][0])\n\n # Create chunks from token embeddings\n chunks = []\n seq_len = token_embeddings.shape[0]\n start = 0\n\n while start \u003c seq_len:\n end = min(start + chunk_size, seq_len)\n\n # Mean pool token embeddings for this chunk\n chunk_embedding = token_embeddings[start:end].mean(dim=0).numpy()\n\n # Reconstruct text for this chunk\n chunk_token_ids = inputs[\"input_ids\"][0][start:end]\n chunk_text = self.tokenizer.decode(chunk_token_ids, skip_special_tokens=True)\n\n chunks.append({\n \"text\": chunk_text,\n \"embedding\": chunk_embedding,\n \"start_token\": start,\n \"end_token\": end\n })\n\n start = end - overlap\n\n return chunks\n\n# Usage\nlate_chunker = LateChunker()\nchunks_with_embeddings = late_chunker.chunk_and_embed(\n long_document,\n chunk_size=512,\n overlap=64\n)\n```\n\n---\n\n## Metadata Enrichment\n\n```python\nfrom dataclasses import dataclass\nfrom datetime import datetime\nimport hashlib\n\n@dataclass\nclass EnrichedChunk:\n text: str\n embedding: list[float] | None\n metadata: dict\n\ndef enrich_chunk(\n text: str,\n source_file: str,\n chunk_index: int,\n total_chunks: int,\n additional_metadata: dict | None = None\n) -> EnrichedChunk:\n \"\"\"Add comprehensive metadata to chunk.\"\"\"\n metadata = {\n # Source tracking\n \"source\": source_file,\n \"chunk_index\": chunk_index,\n \"total_chunks\": total_chunks,\n\n # Content characteristics\n \"char_count\": len(text),\n \"word_count\": len(text.split()),\n \"content_hash\": hashlib.md5(text.encode()).hexdigest()[:12],\n\n # Temporal\n \"indexed_at\": datetime.utcnow().isoformat(),\n\n # Position context\n \"position\": \"start\" if chunk_index == 0 else (\n \"end\" if chunk_index == total_chunks - 1 else \"middle\"\n )\n }\n\n if additional_metadata:\n metadata.update(additional_metadata)\n\n return EnrichedChunk(text=text, embedding=None, metadata=metadata)\n```\n\n---\n\n## Chunk Size Selection Guide\n\n| Document Type | Recommended Size | Overlap | Rationale |\n|--------------|------------------|---------|-----------|\n| FAQ/Q&A | 200-400 tokens | 20-50 | Keep Q&A pairs together |\n| Technical docs | 400-600 tokens | 50-100 | Balance context vs precision |\n| Legal/contracts | 600-800 tokens | 100-150 | Preserve clause context |\n| Code documentation | 300-500 tokens | 50-100 | Keep function docs together |\n| Chat transcripts | 150-300 tokens | 25-50 | Natural turn boundaries |\n| Research papers | 500-800 tokens | 100-200 | Section-level coherence |\n\n---\n\n## Quick Reference\n\n| Strategy | Use Case | Code Pattern |\n|----------|----------|--------------|\n| Fixed-size | Logs, baseline | `text[i:i+chunk_size]` |\n| Recursive | General text | Split by `[\"\\n\\n\", \"\\n\", \". \"]` |\n| Sentence | Q&A content | `sent_tokenize()` + merge |\n| Semantic | Technical docs | Similarity-based breaks |\n| Markdown | Documentation | Header-aware splitting |\n| Late chunking | Long-context models | Embed full, pool chunks |\n\n## Related Skills\n\n- **RAG Architect** - Integration with vector databases\n- **Python Pro** - Preprocessing pipelines\n- **NLP Engineer** - Tokenization and text processing\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":25083,"content_sha256":"1b732310ebf7b5e64c5abef46b13a11614f1c3dc331c214389f2643504a9a809"},{"filename":"references/embedding-models.md","content":"# Embedding Models\n\n---\n\n## Model Comparison Matrix\n\n| Model | Dimensions | Max Tokens | Strengths | Provider |\n|-------|------------|------------|-----------|----------|\n| **text-embedding-3-large** | 3072 (or 256-3072) | 8191 | Best quality, flexible dims | OpenAI |\n| **text-embedding-3-small** | 1536 (or 256-1536) | 8191 | Cost-effective, good quality | OpenAI |\n| **embed-english-v3.0** | 1024 | 512 | Excellent compression, fast | Cohere |\n| **embed-multilingual-v3.0** | 1024 | 512 | 100+ languages | Cohere |\n| **voyage-large-2** | 1536 | 16000 | Long context, code-aware | Voyage AI |\n| **voyage-code-2** | 1536 | 16000 | Code retrieval specialist | Voyage AI |\n| **BGE-large-en-v1.5** | 1024 | 512 | Open source, high quality | BAAI |\n| **BGE-M3** | 1024 | 8192 | Multi-lingual, multi-granularity | BAAI |\n| **E5-large-v2** | 1024 | 512 | Strong benchmark performance | Microsoft |\n| **GTE-large** | 1024 | 512 | Good general-purpose | Alibaba |\n| **all-MiniLM-L6-v2** | 384 | 256 | Fast, lightweight | Sentence Transformers |\n| **nomic-embed-text-v1.5** | 768 | 8192 | Long context, open weights | Nomic AI |\n\n---\n\n## When to Use Each Model\n\n### OpenAI text-embedding-3-large\n```\nBest For:\n- Production RAG requiring highest accuracy\n- Enterprise applications with quality SLAs\n- Flexible dimension requirements (can reduce to save cost)\n- English and major languages\n\nWhen to Avoid:\n- Cost-sensitive high-volume applications\n- Air-gapped or offline deployments\n- Specialized domains without fine-tuning budget\n```\n\n### OpenAI text-embedding-3-small\n```\nBest For:\n- Cost-effective production deployments\n- Good quality-to-cost ratio\n- General-purpose retrieval tasks\n- Quick prototyping with API simplicity\n\nWhen to Avoid:\n- Maximum accuracy requirements\n- Specialized technical domains\n- When open-source is required\n```\n\n### Cohere embed-v3\n```\nBest For:\n- Multi-lingual applications (100+ languages)\n- Search-optimized retrieval (search_document/search_query types)\n- Compression (int8/binary quantization built-in)\n- Production with cost constraints\n\nWhen to Avoid:\n- Very long documents (512 token limit)\n- Code-heavy retrieval tasks\n```\n\n### Voyage AI\n```\nBest For:\n- Code retrieval and technical documentation\n- Long-context documents (16K tokens)\n- Domain-specific fine-tuning options\n- Legal/financial specialized models\n\nWhen to Avoid:\n- Budget-constrained projects\n- Simple general-purpose retrieval\n```\n\n### BGE / E5 (Open Source)\n```\nBest For:\n- Self-hosted deployments\n- Air-gapped environments\n- Cost elimination (no API fees)\n- Fine-tuning on custom domains\n\nWhen to Avoid:\n- Teams without GPU infrastructure\n- Need for zero maintenance\n- Maximum out-of-box quality\n```\n\n---\n\n## OpenAI Embeddings\n\n```python\nfrom openai import OpenAI\n\nclient = OpenAI(api_key=\"\u003copenai-api-key>\")\n\ndef get_embedding(\n text: str,\n model: str = \"text-embedding-3-small\",\n dimensions: int | None = None\n) -> list[float]:\n \"\"\"Get embedding with optional dimension reduction.\"\"\"\n params = {\"input\": text, \"model\": model}\n if dimensions:\n params[\"dimensions\"] = dimensions\n\n response = client.embeddings.create(**params)\n return response.data[0].embedding\n\n# Single embedding\nembedding = get_embedding(\"How do I install the software?\")\n\n# Batch embeddings (more efficient)\ndef get_embeddings_batch(\n texts: list[str],\n model: str = \"text-embedding-3-small\",\n dimensions: int | None = None\n) -> list[list[float]]:\n \"\"\"Batch embed multiple texts.\"\"\"\n params = {\"input\": texts, \"model\": model}\n if dimensions:\n params[\"dimensions\"] = dimensions\n\n response = client.embeddings.create(**params)\n # Sort by index to maintain order\n return [item.embedding for item in sorted(response.data, key=lambda x: x.index)]\n\nembeddings = get_embeddings_batch([\"text1\", \"text2\", \"text3\"])\n\n# Dimension reduction (cost/storage savings)\n# text-embedding-3-large: 3072 -> 1024 (66% storage savings)\nreduced_embedding = get_embedding(\n \"Installation guide...\",\n model=\"text-embedding-3-large\",\n dimensions=1024 # Reduce from 3072\n)\n```\n\n### Dimension Trade-offs\n\n| Original | Reduced | Quality Loss | Storage Savings |\n|----------|---------|--------------|-----------------|\n| 3072 | 1536 | ~1-2% | 50% |\n| 3072 | 1024 | ~2-4% | 67% |\n| 3072 | 512 | ~5-8% | 83% |\n| 3072 | 256 | ~10-15% | 92% |\n\n---\n\n## Cohere Embeddings\n\n```python\nimport cohere\n\nco = cohere.Client(api_key=\"\u003ccohere-api-key>\")\n\n# Document embeddings (for indexing)\ndoc_embeddings = co.embed(\n texts=[\"Installation guide content...\", \"Configuration steps...\"],\n model=\"embed-english-v3.0\",\n input_type=\"search_document\", # Use for documents being indexed\n truncate=\"END\"\n).embeddings\n\n# Query embeddings (for search)\nquery_embedding = co.embed(\n texts=[\"how to install\"],\n model=\"embed-english-v3.0\",\n input_type=\"search_query\", # Use for search queries\n).embeddings[0]\n\n# Multilingual\nmultilingual_embedding = co.embed(\n texts=[\"Comment installer le logiciel?\"], # French\n model=\"embed-multilingual-v3.0\",\n input_type=\"search_query\"\n).embeddings[0]\n\n# Compressed embeddings (int8)\ncompressed = co.embed(\n texts=[\"Document content...\"],\n model=\"embed-english-v3.0\",\n input_type=\"search_document\",\n embedding_types=[\"int8\"] # 4x smaller than float32\n).embeddings\n```\n\n### Cohere Input Types\n\n| Type | Use Case |\n|------|----------|\n| `search_document` | Documents being indexed in vector DB |\n| `search_query` | User search queries |\n| `classification` | Text classification tasks |\n| `clustering` | Document clustering |\n\n---\n\n## Voyage AI Embeddings\n\n```python\nimport voyageai\n\nvo = voyageai.Client(api_key=\"\u003cvoyage-api-key>\")\n\n# General embeddings\nresult = vo.embed(\n texts=[\"Installation guide for the software...\"],\n model=\"voyage-large-2\",\n input_type=\"document\"\n)\nembeddings = result.embeddings\n\n# Code embeddings (specialized)\ncode_result = vo.embed(\n texts=[\n \"def install_package(name):\\n subprocess.run(['pip', 'install', name])\",\n \"How do I install packages in Python?\"\n ],\n model=\"voyage-code-2\",\n input_type=\"document\" # or \"query\" for search\n)\n\n# Long context (up to 16K tokens)\nlong_doc_embedding = vo.embed(\n texts=[very_long_document], # Up to 16K tokens\n model=\"voyage-large-2\",\n input_type=\"document\"\n).embeddings[0]\n```\n\n---\n\n## Open Source Models (Sentence Transformers)\n\n```python\nfrom sentence_transformers import SentenceTransformer\n\n# Load model (downloads on first use)\nmodel = SentenceTransformer(\"BAAI/bge-large-en-v1.5\")\n\n# Single embedding\nembedding = model.encode(\"How do I install the software?\")\n\n# Batch encoding (GPU accelerated)\nembeddings = model.encode(\n [\"doc1\", \"doc2\", \"doc3\"],\n batch_size=32,\n show_progress_bar=True,\n convert_to_numpy=True,\n normalize_embeddings=True # For cosine similarity\n)\n\n# BGE requires instruction prefix for queries\nquery_embedding = model.encode(\n \"Represent this sentence for searching relevant passages: How do I install?\"\n)\n\n# GPU acceleration\nmodel = SentenceTransformer(\"BAAI/bge-large-en-v1.5\", device=\"cuda\")\n\n# Multi-GPU encoding\npool = model.start_multi_process_pool()\nembeddings = model.encode_multi_process(\n sentences=large_corpus,\n pool=pool,\n batch_size=64\n)\nmodel.stop_multi_process_pool(pool)\n```\n\n### BGE-M3 (Multi-lingual, Multi-granularity)\n\n```python\nfrom FlagEmbedding import BGEM3FlagModel\n\nmodel = BGEM3FlagModel(\"BAAI/bge-m3\", use_fp16=True)\n\n# Dense, sparse, and colbert embeddings in one call\noutput = model.encode(\n [\"Installation guide in English\", \"Guide d'installation en francais\"],\n return_dense=True,\n return_sparse=True,\n return_colbert_vecs=True\n)\n\ndense_embeddings = output[\"dense_vecs\"]\nsparse_embeddings = output[\"lexical_weights\"]\ncolbert_embeddings = output[\"colbert_vecs\"]\n```\n\n---\n\n## Embedding Fine-Tuning\n\n### When to Fine-Tune\n\n| Scenario | Recommendation |\n|----------|----------------|\n| Domain-specific jargon (legal, medical) | Fine-tune on domain corpus |\n| Low retrieval precision (\u003c80%) | Fine-tune with hard negatives |\n| Out-of-distribution queries | Fine-tune with query-doc pairs |\n| Cost optimization | Fine-tune smaller model to match larger |\n\n### Fine-Tuning with Sentence Transformers\n\n```python\nfrom sentence_transformers import SentenceTransformer, InputExample, losses\nfrom torch.utils.data import DataLoader\n\n# Prepare training data\ntrain_examples = [\n InputExample(\n texts=[\"query: how to install\", \"doc: Installation guide content...\"],\n label=1.0 # Relevance score\n ),\n InputExample(\n texts=[\"query: how to install\", \"doc: Unrelated content...\"],\n label=0.0 # Negative example\n ),\n]\n\n# Load base model\nmodel = SentenceTransformer(\"BAAI/bge-base-en-v1.5\")\n\n# Create dataloader\ntrain_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)\n\n# Contrastive loss for similarity learning\ntrain_loss = losses.CosineSimilarityLoss(model)\n\n# Fine-tune\nmodel.fit(\n train_objectives=[(train_dataloader, train_loss)],\n epochs=3,\n warmup_steps=100,\n output_path=\"./fine-tuned-model\"\n)\n\n# Or use Multiple Negatives Ranking Loss (better for retrieval)\ntrain_examples_mnrl = [\n InputExample(texts=[\"query\", \"positive_doc\", \"negative_doc1\", \"negative_doc2\"])\n]\ntrain_loss = losses.MultipleNegativesRankingLoss(model)\n```\n\n### Hard Negative Mining\n\n```python\nfrom sentence_transformers import SentenceTransformer\nfrom sentence_transformers.util import semantic_search\nimport torch\n\ndef mine_hard_negatives(\n queries: list[str],\n positives: list[str],\n corpus: list[str],\n model: SentenceTransformer,\n top_k: int = 10\n) -> list[InputExample]:\n \"\"\"Mine hard negatives from corpus for each query-positive pair.\"\"\"\n\n query_embeddings = model.encode(queries, convert_to_tensor=True)\n corpus_embeddings = model.encode(corpus, convert_to_tensor=True)\n positive_set = set(positives)\n\n examples = []\n for i, query in enumerate(queries):\n # Find similar documents that are NOT the positive\n hits = semantic_search(\n query_embeddings[i:i+1],\n corpus_embeddings,\n top_k=top_k + 1\n )[0]\n\n hard_negatives = [\n corpus[hit[\"corpus_id\"]]\n for hit in hits\n if corpus[hit[\"corpus_id\"]] not in positive_set\n ][:3] # Top 3 hard negatives\n\n examples.append(InputExample(\n texts=[query, positives[i]] + hard_negatives\n ))\n\n return examples\n```\n\n---\n\n## Embedding Pipeline Best Practices\n\n### Text Preprocessing\n\n```python\nimport re\nfrom typing import Callable\n\ndef clean_for_embedding(text: str) -> str:\n \"\"\"Clean text before embedding.\"\"\"\n # Remove excessive whitespace\n text = re.sub(r'\\s+', ' ', text)\n # Remove special characters that don't add meaning\n text = re.sub(r'[^\\w\\s\\.\\,\\!\\?\\-\\:\\;\$\$]', '', text)\n # Truncate to reasonable length (model dependent)\n text = text[:8000] # Leave room for tokenization expansion\n return text.strip()\n\ndef preprocess_for_embedding(\n text: str,\n prefix: str = \"\",\n max_length: int = 8000\n) -> str:\n \"\"\"Preprocess with optional prefix (for instruction-tuned models).\"\"\"\n cleaned = clean_for_embedding(text)\n prefixed = f\"{prefix}{cleaned}\" if prefix else cleaned\n return prefixed[:max_length]\n\n# BGE-style prefix for queries\nquery_text = preprocess_for_embedding(\n \"how to install\",\n prefix=\"Represent this sentence for searching relevant passages: \"\n)\n```\n\n### Caching Embeddings\n\n```python\nimport hashlib\nimport json\nfrom functools import lru_cache\nfrom pathlib import Path\n\nclass EmbeddingCache:\n \"\"\"Disk-based embedding cache.\"\"\"\n\n def __init__(self, cache_dir: str = \".embedding_cache\"):\n self.cache_dir = Path(cache_dir)\n self.cache_dir.mkdir(exist_ok=True)\n\n def _hash_key(self, text: str, model: str) -> str:\n content = f\"{model}:{text}\"\n return hashlib.sha256(content.encode()).hexdigest()\n\n def get(self, text: str, model: str) -> list[float] | None:\n key = self._hash_key(text, model)\n cache_file = self.cache_dir / f\"{key}.json\"\n if cache_file.exists():\n return json.loads(cache_file.read_text())\n return None\n\n def set(self, text: str, model: str, embedding: list[float]) -> None:\n key = self._hash_key(text, model)\n cache_file = self.cache_dir / f\"{key}.json\"\n cache_file.write_text(json.dumps(embedding))\n\n# Usage\ncache = EmbeddingCache()\n\ndef get_embedding_cached(text: str, model: str = \"text-embedding-3-small\") -> list[float]:\n cached = cache.get(text, model)\n if cached:\n return cached\n\n embedding = get_embedding(text, model) # Call API\n cache.set(text, model, embedding)\n return embedding\n```\n\n### Batching Strategy\n\n```python\nfrom typing import Iterator\nimport asyncio\nfrom openai import AsyncOpenAI\n\ndef batch_texts(texts: list[str], batch_size: int = 100) -> Iterator[list[str]]:\n \"\"\"Yield batches of texts.\"\"\"\n for i in range(0, len(texts), batch_size):\n yield texts[i:i + batch_size]\n\nasync def get_embeddings_async(\n texts: list[str],\n model: str = \"text-embedding-3-small\",\n batch_size: int = 100,\n max_concurrent: int = 5\n) -> list[list[float]]:\n \"\"\"Async batch embedding with concurrency control.\"\"\"\n client = AsyncOpenAI()\n semaphore = asyncio.Semaphore(max_concurrent)\n\n async def embed_batch(batch: list[str]) -> list[list[float]]:\n async with semaphore:\n response = await client.embeddings.create(\n input=batch,\n model=model\n )\n return [item.embedding for item in sorted(response.data, key=lambda x: x.index)]\n\n batches = list(batch_texts(texts, batch_size))\n results = await asyncio.gather(*[embed_batch(b) for b in batches])\n\n # Flatten results\n return [emb for batch_result in results for emb in batch_result]\n```\n\n---\n\n## Model Selection Flowchart\n\n```\nStart\n │\n ├─ Need offline/self-hosted?\n │ └─ Yes → BGE-large or E5-large (open source)\n │\n ├─ Multi-lingual requirement?\n │ └─ Yes → Cohere embed-multilingual-v3 or BGE-M3\n │\n ├─ Code/technical documentation?\n │ └─ Yes → Voyage-code-2\n │\n ├─ Long documents (>8K tokens)?\n │ └─ Yes → Voyage-large-2 or nomic-embed-text\n │\n ├─ Cost is primary concern?\n │ └─ Yes → text-embedding-3-small (reduced dims)\n │\n ├─ Maximum quality needed?\n │ └─ Yes → text-embedding-3-large\n │\n └─ Default → text-embedding-3-small (best balance)\n```\n\n---\n\n## Quick Reference\n\n| Task | Recommendation |\n|------|----------------|\n| Production RAG (English) | text-embedding-3-small/large |\n| Multi-lingual | Cohere embed-multilingual-v3 |\n| Code retrieval | Voyage-code-2 |\n| Self-hosted | BGE-large-en-v1.5 |\n| Long documents | Voyage-large-2, nomic-embed-text |\n| Prototyping | all-MiniLM-L6-v2 (fast, free) |\n| Maximum quality | text-embedding-3-large |\n| Cost optimized | text-embedding-3-small @ 512 dims |\n\n## Related Skills\n\n- **RAG Architect** - Vector database integration\n- **Python Pro** - Async embedding pipelines\n- **ML Pipeline** - Embedding model deployment\n- **Fine-Tuning Expert** - Custom embedding training\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":15342,"content_sha256":"d6916896df46601bf32eacc51e21358296195b4bec979c3ec74ede28a12e01d8"},{"filename":"references/rag-evaluation.md","content":"# RAG Evaluation\n\n---\n\n## Evaluation Framework Overview\n\n| Framework | Focus | Strengths | Use Case |\n|-----------|-------|-----------|----------|\n| **RAGAS** | RAG-specific metrics | Faithfulness, relevance | Production RAG evaluation |\n| **TruLens** | LLM app observability | Tracing, feedback functions | Debugging and monitoring |\n| **LangSmith** | LangChain ecosystem | Traces, datasets, testing | LangChain projects |\n| **Custom** | Specific requirements | Full control | Domain-specific needs |\n\n---\n\n## Core Metrics\n\n### Retrieval Metrics\n\n| Metric | Formula | What It Measures |\n|--------|---------|------------------|\n| **Precision@k** | Relevant in top-k / k | Are retrieved docs relevant? |\n| **Recall@k** | Relevant in top-k / Total relevant | Did we get all relevant docs? |\n| **MRR** | 1 / Rank of first relevant | How quickly do we find relevant? |\n| **NDCG@k** | DCG@k / IDCG@k | Is ranking order correct? |\n| **Hit Rate** | Queries with relevant in top-k / Total queries | Binary success rate |\n\n### Generation Metrics\n\n| Metric | What It Measures |\n|--------|------------------|\n| **Faithfulness** | Is answer grounded in retrieved context? |\n| **Answer Relevance** | Does answer address the question? |\n| **Context Relevance** | Is retrieved context relevant to question? |\n| **Context Utilization** | How much context was actually used? |\n\n---\n\n## Implementing Core Metrics\n\n### Precision, Recall, and Hit Rate\n\n```python\nfrom dataclasses import dataclass\nfrom typing import Set\n\n@dataclass\nclass RetrievalMetrics:\n precision_at_k: float\n recall_at_k: float\n hit_rate: float\n mrr: float\n\ndef calculate_retrieval_metrics(\n retrieved_ids: list[str],\n relevant_ids: set[str],\n k: int\n) -> RetrievalMetrics:\n \"\"\"Calculate core retrieval metrics.\"\"\"\n top_k = retrieved_ids[:k]\n top_k_set = set(top_k)\n\n # Precision@k: relevant in top-k / k\n relevant_in_top_k = len(top_k_set & relevant_ids)\n precision = relevant_in_top_k / k if k > 0 else 0\n\n # Recall@k: relevant in top-k / total relevant\n recall = relevant_in_top_k / len(relevant_ids) if relevant_ids else 0\n\n # Hit Rate: 1 if any relevant in top-k, else 0\n hit_rate = 1.0 if relevant_in_top_k > 0 else 0.0\n\n # MRR: 1 / rank of first relevant result\n mrr = 0.0\n for i, doc_id in enumerate(top_k, 1):\n if doc_id in relevant_ids:\n mrr = 1.0 / i\n break\n\n return RetrievalMetrics(\n precision_at_k=precision,\n recall_at_k=recall,\n hit_rate=hit_rate,\n mrr=mrr\n )\n\n# Usage\nretrieved = [\"doc1\", \"doc2\", \"doc3\", \"doc4\", \"doc5\"]\nrelevant = {\"doc2\", \"doc5\", \"doc7\"} # Ground truth\n\nmetrics = calculate_retrieval_metrics(retrieved, relevant, k=5)\nprint(f\"Precision@5: {metrics.precision_at_k:.2f}\") # 2/5 = 0.40\nprint(f\"Recall@5: {metrics.recall_at_k:.2f}\") # 2/3 = 0.67\nprint(f\"MRR: {metrics.mrr:.2f}\") # 1/2 = 0.50\n```\n\n### NDCG (Normalized Discounted Cumulative Gain)\n\n```python\nimport numpy as np\n\ndef dcg_at_k(relevance_scores: list[float], k: int) -> float:\n \"\"\"Calculate Discounted Cumulative Gain.\"\"\"\n relevance_scores = np.array(relevance_scores[:k])\n if len(relevance_scores) == 0:\n return 0.0\n\n # DCG = sum(rel_i / log2(i + 1)) for i in 1..k\n discounts = np.log2(np.arange(2, len(relevance_scores) + 2))\n return np.sum(relevance_scores / discounts)\n\ndef ndcg_at_k(\n retrieved_ids: list[str],\n relevance_scores: dict[str, float],\n k: int\n) -> float:\n \"\"\"\n Calculate NDCG@k.\n relevance_scores: dict mapping doc_id to relevance (e.g., 0, 1, 2, 3)\n \"\"\"\n # Get relevance scores for retrieved docs\n retrieved_relevance = [\n relevance_scores.get(doc_id, 0)\n for doc_id in retrieved_ids[:k]\n ]\n\n # Calculate DCG for retrieved order\n dcg = dcg_at_k(retrieved_relevance, k)\n\n # Calculate ideal DCG (perfect ranking)\n ideal_relevance = sorted(relevance_scores.values(), reverse=True)[:k]\n idcg = dcg_at_k(ideal_relevance, k)\n\n return dcg / idcg if idcg > 0 else 0.0\n\n# Usage with graded relevance\nretrieved = [\"doc1\", \"doc2\", \"doc3\", \"doc4\", \"doc5\"]\nrelevance = {\n \"doc1\": 0, # Not relevant\n \"doc2\": 3, # Highly relevant\n \"doc3\": 1, # Somewhat relevant\n \"doc5\": 2, # Relevant\n \"doc7\": 3, # Highly relevant (not retrieved)\n}\n\nndcg = ndcg_at_k(retrieved, relevance, k=5)\nprint(f\"NDCG@5: {ndcg:.3f}\")\n```\n\n---\n\n## RAGAS Framework\n\n### Installation and Setup\n\n```python\n# pip install ragas\n\nfrom ragas import evaluate\nfrom ragas.metrics import (\n faithfulness,\n answer_relevancy,\n context_precision,\n context_recall,\n context_utilization,\n)\nfrom datasets import Dataset\n\n# Prepare evaluation dataset\neval_data = {\n \"question\": [\n \"What is the capital of France?\",\n \"How do I install Python?\"\n ],\n \"answer\": [\n \"The capital of France is Paris.\",\n \"You can install Python by downloading it from python.org.\"\n ],\n \"contexts\": [\n [\"Paris is the capital and largest city of France.\"],\n [\"Python can be installed from the official website python.org.\",\n \"You can also use package managers like brew or apt.\"]\n ],\n \"ground_truth\": [\n \"Paris is the capital of France.\",\n \"Install Python from python.org or use a package manager.\"\n ]\n}\n\ndataset = Dataset.from_dict(eval_data)\n\n# Run evaluation\nresults = evaluate(\n dataset,\n metrics=[\n faithfulness,\n answer_relevancy,\n context_precision,\n context_recall,\n ]\n)\n\nprint(results)\n# {'faithfulness': 0.95, 'answer_relevancy': 0.88, ...}\n```\n\n### Custom RAGAS Evaluation\n\n```python\nfrom ragas.metrics import Metric\nfrom ragas.llms import LangchainLLM\nfrom langchain_openai import ChatOpenAI\n\n# Use custom LLM\ncustom_llm = LangchainLLM(llm=ChatOpenAI(model=\"gpt-4o-mini\"))\n\n# Evaluate with custom settings\nresults = evaluate(\n dataset,\n metrics=[faithfulness, answer_relevancy],\n llm=custom_llm,\n raise_exceptions=False # Continue on errors\n)\n\n# Per-sample scores\nfor i, row in enumerate(results.to_pandas().itertuples()):\n print(f\"Q{i+1}: Faithfulness={row.faithfulness:.2f}, \"\n f\"Relevancy={row.answer_relevancy:.2f}\")\n```\n\n### RAGAS Metrics Explained\n\n```python\n\"\"\"\nRAGAS Core Metrics:\n\n1. Faithfulness (0-1):\n - Measures if answer is grounded in context\n - LLM extracts claims from answer, verifies against context\n - High score = answer doesn't hallucinate\n\n2. Answer Relevancy (0-1):\n - Measures if answer addresses the question\n - Generates questions from answer, compares to original\n - High score = answer is on-topic\n\n3. Context Precision (0-1):\n - Measures if retrieved contexts are relevant\n - Ranks contexts by relevance, calculates precision at each rank\n - High score = top contexts are most relevant\n\n4. Context Recall (0-1):\n - Measures if all ground truth info is in context\n - Checks if ground truth sentences are supported by context\n - High score = context contains needed information\n\"\"\"\n\n# Debugging low scores\ndef diagnose_ragas_scores(results_df):\n \"\"\"Identify problematic samples.\"\"\"\n issues = []\n\n for idx, row in results_df.iterrows():\n if row.get('faithfulness', 1) \u003c 0.5:\n issues.append({\n \"index\": idx,\n \"issue\": \"Low faithfulness - answer may contain hallucinations\",\n \"question\": row['question'],\n \"answer\": row['answer'][:200]\n })\n\n if row.get('context_recall', 1) \u003c 0.5:\n issues.append({\n \"index\": idx,\n \"issue\": \"Low context recall - retrieval missing relevant docs\",\n \"question\": row['question']\n })\n\n return issues\n```\n\n---\n\n## TruLens Evaluation\n\n### Setup and Basic Usage\n\n```python\n# pip install trulens-eval\n\nfrom trulens_eval import Tru, TruChain, Feedback\nfrom trulens_eval.feedback import Groundedness\nfrom trulens_eval.feedback.provider import OpenAI as fOpenAI\n\n# Initialize TruLens\ntru = Tru()\n\n# Create feedback provider\nprovider = fOpenAI()\n\n# Define feedback functions\nf_groundedness = Feedback(\n provider.groundedness_measure_with_cot_reasons,\n name=\"Groundedness\"\n).on(\n TruChain.select_context().node.text # Retrieved context\n).on_output()\n\nf_relevance = Feedback(\n provider.relevance_with_cot_reasons,\n name=\"Answer Relevance\"\n).on_input().on_output()\n\nf_context_relevance = Feedback(\n provider.context_relevance_with_cot_reasons,\n name=\"Context Relevance\"\n).on_input().on(\n TruChain.select_context().node.text\n)\n\n# Wrap your RAG chain\nfrom langchain.chains import RetrievalQA\n\nrag_chain = RetrievalQA.from_chain_type(\n llm=llm,\n retriever=vector_store.as_retriever()\n)\n\ntru_recorder = TruChain(\n rag_chain,\n app_id=\"rag-v1\",\n feedbacks=[f_groundedness, f_relevance, f_context_relevance]\n)\n\n# Run with recording\nwith tru_recorder as recording:\n response = rag_chain.invoke({\"query\": \"How do I configure authentication?\"})\n\n# View results\ntru.run_dashboard() # Opens web UI\n# Or get programmatically\nrecords = tru.get_records_and_feedback(app_ids=[\"rag-v1\"])\n```\n\n### Custom Feedback Functions\n\n```python\nfrom trulens_eval import Feedback, Select\n\ndef custom_citation_check(response: str, context: str) -> float:\n \"\"\"Check if response cites sources from context.\"\"\"\n # Extract citations from response (e.g., [1], [Source: X])\n import re\n citations = re.findall(r'\\[[\\d\\w\\s:]+\\]', response)\n\n if not citations:\n return 0.0 # No citations\n\n # Verify citations reference actual context\n valid_citations = sum(1 for c in citations if c.lower() in context.lower())\n return valid_citations / len(citations)\n\nf_citation = Feedback(\n custom_citation_check,\n name=\"Citation Accuracy\"\n).on_output().on(Select.RecordCalls.retriever.get_relevant_documents.rets.page_content)\n```\n\n---\n\n## Building Custom Evaluation Pipelines\n\n### LLM-as-Judge Evaluation\n\n```python\nfrom openai import OpenAI\nfrom dataclasses import dataclass\nfrom typing import Literal\n\nclient = OpenAI()\n\n@dataclass\nclass EvalResult:\n score: float\n reasoning: str\n criteria: str\n\ndef evaluate_with_llm(\n question: str,\n answer: str,\n context: str,\n criteria: Literal[\"faithfulness\", \"relevance\", \"completeness\"]\n) -> EvalResult:\n \"\"\"Use LLM as judge for evaluation.\"\"\"\n\n criteria_prompts = {\n \"faithfulness\": \"\"\"\n Evaluate if the answer is fully supported by the provided context.\n Score 1.0 if every claim in the answer is verifiable from context.\n Score 0.5 if most claims are supported but some are not.\n Score 0.0 if the answer contains significant unsupported claims.\n \"\"\",\n \"relevance\": \"\"\"\n Evaluate if the answer directly addresses the question.\n Score 1.0 if the answer fully addresses the question.\n Score 0.5 if the answer partially addresses the question.\n Score 0.0 if the answer is off-topic or doesn't address the question.\n \"\"\",\n \"completeness\": \"\"\"\n Evaluate if the answer covers all aspects of the question.\n Score 1.0 if the answer is comprehensive and complete.\n Score 0.5 if the answer covers main points but misses details.\n Score 0.0 if the answer is significantly incomplete.\n \"\"\"\n }\n\n response = client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": f\"\"\"You are an expert evaluator for RAG systems.\n {criteria_prompts[criteria]}\n\n Respond in JSON format:\n {{\"score\": \u003c0.0-1.0>, \"reasoning\": \"\u003cexplanation>\"}}\"\"\"\n },\n {\n \"role\": \"user\",\n \"content\": f\"\"\"Question: {question}\n\nContext:\n{context}\n\nAnswer: {answer}\n\nEvaluate the answer for {criteria}:\"\"\"\n }\n ],\n response_format={\"type\": \"json_object\"}\n )\n\n import json\n result = json.loads(response.choices[0].message.content)\n\n return EvalResult(\n score=result[\"score\"],\n reasoning=result[\"reasoning\"],\n criteria=criteria\n )\n\n# Usage\neval_result = evaluate_with_llm(\n question=\"How do I configure OAuth2?\",\n answer=\"Configure OAuth2 by setting client_id and client_secret in config.yaml.\",\n context=\"OAuth2 configuration requires client_id, client_secret, and redirect_uri in config.yaml.\",\n criteria=\"faithfulness\"\n)\nprint(f\"Faithfulness: {eval_result.score:.2f}\")\nprint(f\"Reasoning: {eval_result.reasoning}\")\n```\n\n### Batch Evaluation Pipeline\n\n```python\nimport asyncio\nfrom tqdm.asyncio import tqdm_asyncio\n\nasync def evaluate_batch(\n test_cases: list[dict],\n retriever,\n generator,\n metrics: list[str] = [\"precision\", \"faithfulness\", \"relevance\"]\n) -> dict:\n \"\"\"Run batch evaluation on test cases.\"\"\"\n\n results = {\n \"per_sample\": [],\n \"aggregated\": {}\n }\n\n async def evaluate_single(case: dict) -> dict:\n # Retrieve\n retrieved = await retriever.aretrieve(case[\"question\"])\n retrieved_ids = [r.id for r in retrieved]\n\n # Generate\n answer = await generator.agenerate(\n question=case[\"question\"],\n context=[r.text for r in retrieved]\n )\n\n # Calculate metrics\n sample_result = {\n \"question\": case[\"question\"],\n \"answer\": answer,\n \"retrieved_ids\": retrieved_ids\n }\n\n if \"relevant_ids\" in case and \"precision\" in metrics:\n retrieval_metrics = calculate_retrieval_metrics(\n retrieved_ids,\n set(case[\"relevant_ids\"]),\n k=5\n )\n sample_result[\"precision@5\"] = retrieval_metrics.precision_at_k\n sample_result[\"recall@5\"] = retrieval_metrics.recall_at_k\n\n if \"faithfulness\" in metrics:\n faith_eval = evaluate_with_llm(\n case[\"question\"],\n answer,\n \"\\n\".join([r.text for r in retrieved]),\n \"faithfulness\"\n )\n sample_result[\"faithfulness\"] = faith_eval.score\n\n return sample_result\n\n # Run evaluations concurrently\n tasks = [evaluate_single(case) for case in test_cases]\n results[\"per_sample\"] = await tqdm_asyncio.gather(*tasks)\n\n # Aggregate results\n for metric in [\"precision@5\", \"recall@5\", \"faithfulness\"]:\n scores = [r.get(metric) for r in results[\"per_sample\"] if r.get(metric) is not None]\n if scores:\n results[\"aggregated\"][metric] = {\n \"mean\": sum(scores) / len(scores),\n \"min\": min(scores),\n \"max\": max(scores)\n }\n\n return results\n```\n\n---\n\n## Debugging Poor Retrieval\n\n### Retrieval Diagnostics\n\n```python\ndef diagnose_retrieval(\n query: str,\n retrieved_docs: list,\n expected_docs: list,\n embedding_model\n) -> dict:\n \"\"\"Diagnose why retrieval might be failing.\"\"\"\n\n query_embedding = embedding_model.encode(query)\n retrieved_embeddings = [embedding_model.encode(d) for d in retrieved_docs]\n expected_embeddings = [embedding_model.encode(d) for d in expected_docs]\n\n from sklearn.metrics.pairwise import cosine_similarity\n import numpy as np\n\n diagnosis = {\n \"query\": query,\n \"issues\": []\n }\n\n # Check query-document similarity\n for i, (doc, emb) in enumerate(zip(retrieved_docs, retrieved_embeddings)):\n sim = cosine_similarity([query_embedding], [emb])[0][0]\n if sim \u003c 0.5:\n diagnosis[\"issues\"].append({\n \"type\": \"low_similarity\",\n \"doc_index\": i,\n \"similarity\": float(sim),\n \"doc_preview\": doc[:100]\n })\n\n # Check if expected docs would score higher\n for i, (doc, emb) in enumerate(zip(expected_docs, expected_embeddings)):\n sim = cosine_similarity([query_embedding], [emb])[0][0]\n retrieved_max_sim = max(\n cosine_similarity([query_embedding], [e])[0][0]\n for e in retrieved_embeddings\n )\n\n if sim > retrieved_max_sim:\n diagnosis[\"issues\"].append({\n \"type\": \"missed_better_doc\",\n \"expected_doc_index\": i,\n \"expected_sim\": float(sim),\n \"best_retrieved_sim\": float(retrieved_max_sim),\n \"doc_preview\": doc[:100]\n })\n\n # Check for vocabulary mismatch\n query_terms = set(query.lower().split())\n for i, doc in enumerate(retrieved_docs):\n doc_terms = set(doc.lower().split())\n overlap = query_terms & doc_terms\n if len(overlap) \u003c len(query_terms) * 0.3:\n diagnosis[\"issues\"].append({\n \"type\": \"vocabulary_mismatch\",\n \"doc_index\": i,\n \"query_terms\": list(query_terms),\n \"overlapping_terms\": list(overlap)\n })\n\n return diagnosis\n\n# Usage\ndiagnosis = diagnose_retrieval(\n query=\"How to configure OAuth authentication\",\n retrieved_docs=retrieved_texts,\n expected_docs=expected_texts,\n embedding_model=sentence_transformer\n)\n\nfor issue in diagnosis[\"issues\"]:\n print(f\"Issue: {issue['type']}\")\n print(f\"Details: {issue}\")\n```\n\n### Query Analysis\n\n```python\ndef analyze_query_performance(\n query_logs: list[dict],\n threshold_precision: float = 0.6\n) -> dict:\n \"\"\"Analyze query patterns to find systematic issues.\"\"\"\n\n analysis = {\n \"total_queries\": len(query_logs),\n \"low_performing\": [],\n \"patterns\": {}\n }\n\n for log in query_logs:\n if log.get(\"precision@5\", 1.0) \u003c threshold_precision:\n analysis[\"low_performing\"].append(log)\n\n # Analyze low-performing queries\n if analysis[\"low_performing\"]:\n # Check for common patterns\n low_perf_queries = [l[\"query\"] for l in analysis[\"low_performing\"]]\n\n # Query length analysis\n avg_length = sum(len(q.split()) for q in low_perf_queries) / len(low_perf_queries)\n analysis[\"patterns\"][\"avg_low_perf_query_length\"] = avg_length\n\n # Common terms in failing queries\n from collections import Counter\n all_terms = []\n for q in low_perf_queries:\n all_terms.extend(q.lower().split())\n analysis[\"patterns\"][\"common_failing_terms\"] = Counter(all_terms).most_common(10)\n\n # Question type analysis\n question_words = [\"how\", \"what\", \"why\", \"when\", \"where\", \"who\"]\n question_types = Counter()\n for q in low_perf_queries:\n for qw in question_words:\n if q.lower().startswith(qw):\n question_types[qw] += 1\n break\n else:\n question_types[\"other\"] += 1\n analysis[\"patterns\"][\"failing_question_types\"] = dict(question_types)\n\n return analysis\n```\n\n---\n\n## Continuous Monitoring\n\n### Production Metrics Dashboard\n\n```python\nimport time\nfrom dataclasses import dataclass, field\nfrom collections import deque\nfrom threading import Lock\n\n@dataclass\nclass RAGMetricsCollector:\n \"\"\"Collect and track RAG metrics in production.\"\"\"\n\n window_size: int = 1000\n _latencies: deque = field(default_factory=lambda: deque(maxlen=1000))\n _retrieval_scores: deque = field(default_factory=lambda: deque(maxlen=1000))\n _generation_scores: deque = field(default_factory=lambda: deque(maxlen=1000))\n _lock: Lock = field(default_factory=Lock)\n\n def record_query(\n self,\n latency_ms: float,\n retrieval_score: float | None = None,\n generation_score: float | None = None\n ):\n \"\"\"Record metrics for a single query.\"\"\"\n with self._lock:\n self._latencies.append(latency_ms)\n if retrieval_score is not None:\n self._retrieval_scores.append(retrieval_score)\n if generation_score is not None:\n self._generation_scores.append(generation_score)\n\n def get_summary(self) -> dict:\n \"\"\"Get current metrics summary.\"\"\"\n with self._lock:\n import numpy as np\n\n summary = {\n \"queries_in_window\": len(self._latencies),\n \"latency\": {\n \"p50\": np.percentile(self._latencies, 50) if self._latencies else 0,\n \"p95\": np.percentile(self._latencies, 95) if self._latencies else 0,\n \"p99\": np.percentile(self._latencies, 99) if self._latencies else 0,\n },\n \"retrieval_score\": {\n \"mean\": np.mean(self._retrieval_scores) if self._retrieval_scores else 0,\n \"std\": np.std(self._retrieval_scores) if self._retrieval_scores else 0,\n },\n \"generation_score\": {\n \"mean\": np.mean(self._generation_scores) if self._generation_scores else 0,\n \"std\": np.std(self._generation_scores) if self._generation_scores else 0,\n }\n }\n\n return summary\n\n# Usage\nmetrics = RAGMetricsCollector()\n\n# In your RAG endpoint\nstart = time.time()\nresponse = rag_pipeline.query(question)\nlatency = (time.time() - start) * 1000\n\nmetrics.record_query(\n latency_ms=latency,\n retrieval_score=response.get(\"retrieval_score\"),\n generation_score=response.get(\"generation_score\")\n)\n\n# Periodically check\nprint(metrics.get_summary())\n```\n\n### Alerting on Quality Degradation\n\n```python\nclass RAGQualityMonitor:\n \"\"\"Monitor RAG quality and alert on degradation.\"\"\"\n\n def __init__(\n self,\n baseline_precision: float = 0.8,\n alert_threshold: float = 0.1, # Alert if drops by 10%\n window_size: int = 100\n ):\n self.baseline = baseline_precision\n self.threshold = alert_threshold\n self.window_size = window_size\n self.recent_scores = deque(maxlen=window_size)\n\n def record_score(self, precision: float) -> dict | None:\n \"\"\"Record score and return alert if quality degraded.\"\"\"\n self.recent_scores.append(precision)\n\n if len(self.recent_scores) \u003c self.window_size // 2:\n return None # Not enough data\n\n current_mean = sum(self.recent_scores) / len(self.recent_scores)\n degradation = self.baseline - current_mean\n\n if degradation > self.threshold:\n return {\n \"alert\": \"QUALITY_DEGRADATION\",\n \"baseline\": self.baseline,\n \"current\": current_mean,\n \"degradation\": degradation,\n \"window_size\": len(self.recent_scores)\n }\n\n return None\n\n# Usage\nmonitor = RAGQualityMonitor(baseline_precision=0.85)\n\nfor query_result in production_queries:\n alert = monitor.record_score(query_result[\"precision@5\"])\n if alert:\n send_alert(alert) # Slack, PagerDuty, etc.\n```\n\n---\n\n## Evaluation Best Practices\n\n| Practice | Description |\n|----------|-------------|\n| **Golden test set** | Maintain 50-200 curated Q&A pairs with ground truth |\n| **Stratified sampling** | Include diverse query types in test set |\n| **Human baselines** | Compare LLM judges against human annotators |\n| **Version control** | Track evaluation results alongside model versions |\n| **Regular re-evaluation** | Re-run golden tests on every retrieval change |\n| **A/B testing** | Compare new retrieval strategies on live traffic |\n\n---\n\n## Quick Reference\n\n| Goal | Metric | Target |\n|------|--------|--------|\n| Are docs relevant? | Precision@5 | > 0.7 |\n| Did we get all docs? | Recall@5 | > 0.8 |\n| Is ranking good? | NDCG@5 | > 0.7 |\n| Is answer grounded? | Faithfulness | > 0.9 |\n| Does answer fit question? | Answer Relevance | > 0.8 |\n| Is context useful? | Context Relevance | > 0.7 |\n\n| Framework | Best For |\n|-----------|----------|\n| RAGAS | Quick RAG-specific evaluation |\n| TruLens | Production monitoring and tracing |\n| Custom LLM-judge | Domain-specific criteria |\n| Manual annotation | Ground truth creation |\n\n## Related Skills\n\n- **RAG Architect** - System design\n- **ML Pipeline** - Evaluation automation\n- **Data Scientist** - Statistical analysis\n- **Monitoring Expert** - Production observability\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":24288,"content_sha256":"de20bdaa8180b00bdd931dd79870926f747de4dafc1c0a66a64609293a8131e0"},{"filename":"references/retrieval-optimization.md","content":"# Retrieval Optimization\n\n---\n\n## Optimization Techniques Overview\n\n| Technique | Impact | Complexity | When to Use |\n|-----------|--------|------------|-------------|\n| **Hybrid Search** | High | Medium | Always for production |\n| **Reranking** | High | Low | Top-k refinement |\n| **Query Expansion** | Medium | Medium | Ambiguous queries |\n| **HyDE** | Medium-High | Medium | Concept-heavy retrieval |\n| **Metadata Filtering** | High | Low | Multi-tenant, categorical |\n| **Query Decomposition** | Medium | High | Complex questions |\n| **Contextual Compression** | Medium | Medium | Long retrieved chunks |\n\n---\n\n## Hybrid Search (Vector + Keyword)\n\n### Reciprocal Rank Fusion (RRF)\n\n```python\nfrom dataclasses import dataclass\nfrom typing import Callable\n\n@dataclass\nclass SearchResult:\n id: str\n text: str\n score: float\n source: str # \"vector\" or \"keyword\"\n\ndef reciprocal_rank_fusion(\n vector_results: list[SearchResult],\n keyword_results: list[SearchResult],\n k: int = 60,\n vector_weight: float = 0.5\n) -> list[SearchResult]:\n \"\"\"\n Combine vector and keyword results using RRF.\n k is a constant that reduces the impact of high rankings (typically 60).\n \"\"\"\n scores: dict[str, float] = {}\n docs: dict[str, SearchResult] = {}\n\n # Score vector results\n for rank, result in enumerate(vector_results, 1):\n rrf_score = vector_weight * (1 / (k + rank))\n scores[result.id] = scores.get(result.id, 0) + rrf_score\n docs[result.id] = result\n\n # Score keyword results\n keyword_weight = 1 - vector_weight\n for rank, result in enumerate(keyword_results, 1):\n rrf_score = keyword_weight * (1 / (k + rank))\n scores[result.id] = scores.get(result.id, 0) + rrf_score\n if result.id not in docs:\n docs[result.id] = result\n\n # Sort by combined score\n sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)\n\n return [\n SearchResult(\n id=doc_id,\n text=docs[doc_id].text,\n score=scores[doc_id],\n source=\"hybrid\"\n )\n for doc_id in sorted_ids\n ]\n\n# Usage\nhybrid_results = reciprocal_rank_fusion(\n vector_results=vector_search(query_embedding, top_k=20),\n keyword_results=bm25_search(query_text, top_k=20),\n vector_weight=0.6 # Favor semantic similarity\n)\n```\n\n### BM25 + Vector with Weaviate\n\n```python\nfrom weaviate.classes.query import HybridFusion\n\ncollection = client.collections.get(\"Documents\")\n\n# Hybrid search with configurable fusion\nresults = collection.query.hybrid(\n query=\"how to configure authentication\",\n alpha=0.5, # 0 = pure BM25, 1 = pure vector\n fusion_type=HybridFusion.RELATIVE_SCORE, # or RANKED\n limit=10,\n return_metadata=[\"score\", \"explain_score\"]\n)\n\n# Iterate results\nfor obj in results.objects:\n print(f\"Score: {obj.metadata.score}\")\n print(f\"Explanation: {obj.metadata.explain_score}\")\n print(f\"Text: {obj.properties['content'][:200]}\")\n```\n\n### Pinecone Sparse-Dense\n\n```python\nfrom pinecone_text.sparse import BM25Encoder\n\n# Train BM25 encoder on your corpus\nbm25 = BM25Encoder()\nbm25.fit(corpus_documents)\n\n# Encode query for hybrid search\nsparse_vector = bm25.encode_queries(query_text)\ndense_vector = get_embedding(query_text)\n\n# Search with both vectors\nresults = index.query(\n vector=dense_vector,\n sparse_vector=sparse_vector,\n top_k=10,\n include_metadata=True\n)\n```\n\n---\n\n## Reranking\n\n### Cohere Rerank\n\n```python\nimport cohere\n\nco = cohere.Client(api_key=\"\u003ccohere-api-key>\")\n\ndef rerank_results(\n query: str,\n documents: list[str],\n top_n: int = 5,\n model: str = \"rerank-english-v3.0\"\n) -> list[dict]:\n \"\"\"Rerank documents using Cohere.\"\"\"\n response = co.rerank(\n query=query,\n documents=documents,\n top_n=top_n,\n model=model,\n return_documents=True\n )\n\n return [\n {\n \"text\": result.document.text,\n \"relevance_score\": result.relevance_score,\n \"original_index\": result.index\n }\n for result in response.results\n ]\n\n# Pipeline: retrieve more, rerank fewer\ninitial_results = vector_search(query_embedding, top_k=50)\ndocuments = [r.text for r in initial_results]\n\nreranked = rerank_results(\n query=\"how to configure OAuth2 authentication\",\n documents=documents,\n top_n=5\n)\n\n# Use top 5 reranked docs for LLM context\ncontext = \"\\n\\n\".join([r[\"text\"] for r in reranked])\n```\n\n### Cross-Encoder Reranking (Open Source)\n\n```python\nfrom sentence_transformers import CrossEncoder\n\nclass Reranker:\n \"\"\"Rerank using cross-encoder model.\"\"\"\n\n def __init__(self, model_name: str = \"cross-encoder/ms-marco-MiniLM-L-6-v2\"):\n self.model = CrossEncoder(model_name)\n\n def rerank(\n self,\n query: str,\n documents: list[str],\n top_k: int = 5\n ) -> list[tuple[str, float]]:\n \"\"\"Rerank documents by relevance to query.\"\"\"\n # Create query-document pairs\n pairs = [[query, doc] for doc in documents]\n\n # Get relevance scores\n scores = self.model.predict(pairs)\n\n # Sort by score\n doc_scores = list(zip(documents, scores))\n doc_scores.sort(key=lambda x: x[1], reverse=True)\n\n return doc_scores[:top_k]\n\n# Usage\nreranker = Reranker()\ntop_docs = reranker.rerank(\n query=\"OAuth2 setup guide\",\n documents=retrieved_documents,\n top_k=5\n)\n```\n\n### ColBERT-Style Late Interaction\n\n```python\nfrom colbert import Searcher\nfrom colbert.infra import Run, RunConfig\n\n# Setup ColBERT index (one-time)\nwith Run().context(RunConfig(nranks=1)):\n searcher = Searcher(index=\"path/to/colbert_index\")\n\n# Search with late interaction scoring\nresults = searcher.search(\n query=\"how to configure authentication\",\n k=10\n)\n\n# Results include token-level matching scores\nfor passage_id, rank, score in zip(*results):\n print(f\"Rank {rank}: Doc {passage_id}, Score: {score}\")\n```\n\n---\n\n## Query Expansion\n\n### LLM-Based Query Expansion\n\n```python\nfrom openai import OpenAI\n\nclient = OpenAI()\n\ndef expand_query(query: str, num_expansions: int = 3) -> list[str]:\n \"\"\"Generate query variations using LLM.\"\"\"\n response = client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": f\"\"\"Generate {num_expansions} alternative search queries\n that would help find relevant documents for the user's question.\n Include:\n - Synonym variations\n - More specific versions\n - More general versions\n Return as JSON array of strings.\"\"\"\n },\n {\n \"role\": \"user\",\n \"content\": query\n }\n ],\n response_format={\"type\": \"json_object\"}\n )\n\n import json\n result = json.loads(response.choices[0].message.content)\n return [query] + result.get(\"queries\", [])\n\n# Usage\noriginal_query = \"how to fix memory leak\"\nexpanded_queries = expand_query(original_query)\n# [\"how to fix memory leak\", \"debug memory issues\", \"memory leak detection\",\n# \"troubleshoot high memory usage\"]\n\n# Search with all queries and merge results\nall_results = []\nfor q in expanded_queries:\n results = vector_search(get_embedding(q), top_k=10)\n all_results.extend(results)\n\n# Deduplicate and rank by frequency\ndeduped = deduplicate_by_id(all_results)\n```\n\n### Query Rewriting\n\n```python\ndef rewrite_query_for_retrieval(\n conversational_query: str,\n chat_history: list[dict]\n) -> str:\n \"\"\"Rewrite conversational query to standalone search query.\"\"\"\n response = client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": \"\"\"Rewrite the user's question as a standalone search query.\n Include relevant context from chat history.\n Output only the rewritten query, nothing else.\"\"\"\n },\n {\n \"role\": \"user\",\n \"content\": f\"\"\"Chat history:\n{format_chat_history(chat_history)}\n\nUser's question: {conversational_query}\n\nRewritten search query:\"\"\"\n }\n ],\n max_tokens=100\n )\n\n return response.choices[0].message.content.strip()\n\n# Example\nhistory = [\n {\"role\": \"user\", \"content\": \"Tell me about Python web frameworks\"},\n {\"role\": \"assistant\", \"content\": \"Popular Python web frameworks include Django, Flask, and FastAPI...\"}\n]\nquery = \"Which one is best for APIs?\"\n\nrewritten = rewrite_query_for_retrieval(query, history)\n# Output: \"Best Python web framework for building REST APIs: Django vs Flask vs FastAPI\"\n```\n\n---\n\n## HyDE (Hypothetical Document Embeddings)\n\n```python\ndef hyde_search(\n query: str,\n vector_store,\n embedding_model,\n top_k: int = 10\n) -> list[SearchResult]:\n \"\"\"\n Generate hypothetical answer, embed it, and search.\n Aligns query embedding space with document embedding space.\n \"\"\"\n # Generate hypothetical document\n response = client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": \"\"\"Write a passage that would answer the user's question.\n Write as if you're an expert documentation author.\n Be specific and technical. About 100-200 words.\"\"\"\n },\n {\n \"role\": \"user\",\n \"content\": query\n }\n ],\n max_tokens=300\n )\n\n hypothetical_doc = response.choices[0].message.content\n\n # Embed hypothetical document\n hyde_embedding = embedding_model.encode(hypothetical_doc)\n\n # Search with hypothetical doc embedding\n results = vector_store.search(\n vector=hyde_embedding,\n top_k=top_k\n )\n\n return results\n\n# Usage\nresults = hyde_search(\n query=\"How do I handle rate limiting in my API?\",\n vector_store=qdrant_client,\n embedding_model=sentence_transformer\n)\n```\n\n### Multi-HyDE (Multiple Perspectives)\n\n```python\ndef multi_hyde_search(\n query: str,\n vector_store,\n embedding_model,\n num_hypotheticals: int = 3,\n top_k: int = 10\n) -> list[SearchResult]:\n \"\"\"Generate multiple hypothetical docs for diverse retrieval.\"\"\"\n response = client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": f\"\"\"Generate {num_hypotheticals} different passages\n that could answer the question from different angles:\n 1. Technical deep-dive\n 2. Beginner-friendly explanation\n 3. Best practices summary\n\n Return as JSON with \"passages\" array.\"\"\"\n },\n {\n \"role\": \"user\",\n \"content\": query\n }\n ],\n response_format={\"type\": \"json_object\"}\n )\n\n import json\n passages = json.loads(response.choices[0].message.content)[\"passages\"]\n\n # Embed all hypotheticals\n all_results = []\n for passage in passages:\n embedding = embedding_model.encode(passage)\n results = vector_store.search(vector=embedding, top_k=top_k)\n all_results.extend(results)\n\n # Deduplicate and combine scores\n return deduplicate_and_merge(all_results)\n```\n\n---\n\n## Metadata Filtering\n\n### Multi-Tenant Filtering\n\n```python\nclass MultiTenantRetriever:\n \"\"\"Retriever with mandatory tenant isolation.\"\"\"\n\n def __init__(self, vector_store):\n self.vector_store = vector_store\n\n def search(\n self,\n query_embedding: list[float],\n tenant_id: str,\n top_k: int = 10,\n additional_filters: dict | None = None\n ) -> list[SearchResult]:\n \"\"\"Search with mandatory tenant filter.\"\"\"\n # Build filter - tenant is always required\n filters = {\"tenant_id\": {\"$eq\": tenant_id}}\n\n if additional_filters:\n filters = {\"$and\": [filters, additional_filters]}\n\n return self.vector_store.search(\n vector=query_embedding,\n filter=filters,\n top_k=top_k\n )\n\n# Usage\nretriever = MultiTenantRetriever(pinecone_index)\nresults = retriever.search(\n query_embedding=embedding,\n tenant_id=\"acme-corp\",\n additional_filters={\n \"doc_type\": {\"$in\": [\"manual\", \"faq\"]},\n \"published\": {\"$eq\": True}\n }\n)\n```\n\n### Temporal Filtering\n\n```python\nfrom datetime import datetime, timedelta\n\ndef search_recent_documents(\n query_embedding: list[float],\n vector_store,\n days_back: int = 30,\n top_k: int = 10\n) -> list[SearchResult]:\n \"\"\"Search documents updated within time window.\"\"\"\n cutoff_date = datetime.utcnow() - timedelta(days=days_back)\n\n return vector_store.search(\n vector=query_embedding,\n filter={\n \"updated_at\": {\"$gte\": cutoff_date.isoformat()}\n },\n top_k=top_k\n )\n\ndef search_with_recency_boost(\n query_embedding: list[float],\n vector_store,\n recency_weight: float = 0.2,\n top_k: int = 10\n) -> list[SearchResult]:\n \"\"\"Boost recent documents in ranking.\"\"\"\n # Get more results to apply post-filtering\n results = vector_store.search(\n vector=query_embedding,\n top_k=top_k * 3\n )\n\n now = datetime.utcnow()\n\n def compute_boosted_score(result):\n doc_date = datetime.fromisoformat(result.metadata[\"updated_at\"])\n days_old = (now - doc_date).days\n recency_score = max(0, 1 - (days_old / 365)) # Decay over 1 year\n return result.score * (1 - recency_weight) + recency_score * recency_weight\n\n # Rerank with recency boost\n for result in results:\n result.boosted_score = compute_boosted_score(result)\n\n results.sort(key=lambda x: x.boosted_score, reverse=True)\n return results[:top_k]\n```\n\n---\n\n## Query Decomposition\n\n```python\ndef decompose_complex_query(query: str) -> list[str]:\n \"\"\"Break complex query into sub-questions.\"\"\"\n response = client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": \"\"\"Break this complex question into simpler sub-questions\n that can be answered independently. Each sub-question should be\n searchable. Return as JSON with \"questions\" array.\"\"\"\n },\n {\n \"role\": \"user\",\n \"content\": query\n }\n ],\n response_format={\"type\": \"json_object\"}\n )\n\n import json\n result = json.loads(response.choices[0].message.content)\n return result.get(\"questions\", [query])\n\ndef search_with_decomposition(\n complex_query: str,\n vector_store,\n embedding_model,\n top_k_per_subquery: int = 5\n) -> dict:\n \"\"\"Search for each sub-question and aggregate results.\"\"\"\n sub_questions = decompose_complex_query(complex_query)\n\n aggregated_results = {\n \"sub_questions\": [],\n \"all_documents\": []\n }\n\n seen_doc_ids = set()\n\n for sub_q in sub_questions:\n embedding = embedding_model.encode(sub_q)\n results = vector_store.search(vector=embedding, top_k=top_k_per_subquery)\n\n sub_q_results = []\n for r in results:\n if r.id not in seen_doc_ids:\n seen_doc_ids.add(r.id)\n sub_q_results.append(r)\n aggregated_results[\"all_documents\"].append(r)\n\n aggregated_results[\"sub_questions\"].append({\n \"question\": sub_q,\n \"results\": sub_q_results\n })\n\n return aggregated_results\n\n# Usage\ncomplex_q = \"Compare the security features of OAuth2 and API keys, and explain when to use each\"\nresults = search_with_decomposition(complex_q, vector_store, embedding_model)\n```\n\n---\n\n## Contextual Compression\n\n```python\ndef compress_retrieved_context(\n query: str,\n documents: list[str],\n max_tokens: int = 2000\n) -> str:\n \"\"\"Extract only query-relevant parts from documents.\"\"\"\n response = client.chat.completions.create(\n model=\"gpt-4o-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": f\"\"\"Extract only the parts of these documents that are\n relevant to answering the user's question.\n Remove irrelevant information.\n Keep extracted content under {max_tokens} tokens.\n Maintain source attribution.\"\"\"\n },\n {\n \"role\": \"user\",\n \"content\": f\"\"\"Question: {query}\n\nDocuments:\n{chr(10).join([f'[Doc {i+1}]: {doc}' for i, doc in enumerate(documents)])}\n\nExtracted relevant content:\"\"\"\n }\n ],\n max_tokens=max_tokens\n )\n\n return response.choices[0].message.content\n```\n\n### Extractive Compression with Cross-Encoder\n\n```python\nfrom sentence_transformers import CrossEncoder\n\ndef extractive_compress(\n query: str,\n document: str,\n cross_encoder: CrossEncoder,\n top_k_sentences: int = 5\n) -> str:\n \"\"\"Extract most relevant sentences from document.\"\"\"\n import re\n sentences = re.split(r'(?\u003c=[.!?])\\s+', document)\n\n if len(sentences) \u003c= top_k_sentences:\n return document\n\n # Score each sentence\n pairs = [[query, sent] for sent in sentences]\n scores = cross_encoder.predict(pairs)\n\n # Get top sentences in original order\n scored_sentences = list(zip(range(len(sentences)), sentences, scores))\n top_sentences = sorted(scored_sentences, key=lambda x: x[2], reverse=True)[:top_k_sentences]\n top_sentences = sorted(top_sentences, key=lambda x: x[0]) # Restore order\n\n return \" \".join([s[1] for s in top_sentences])\n```\n\n---\n\n## Complete Optimized Pipeline\n\n```python\nclass OptimizedRetriever:\n \"\"\"Production retrieval pipeline with all optimizations.\"\"\"\n\n def __init__(\n self,\n vector_store,\n embedding_model,\n reranker,\n bm25_index\n ):\n self.vector_store = vector_store\n self.embedding_model = embedding_model\n self.reranker = reranker\n self.bm25_index = bm25_index\n\n async def retrieve(\n self,\n query: str,\n tenant_id: str,\n top_k: int = 5,\n use_hyde: bool = False,\n use_query_expansion: bool = True\n ) -> list[dict]:\n \"\"\"Full optimized retrieval pipeline.\"\"\"\n # Step 1: Query preprocessing\n processed_query = self._preprocess_query(query)\n\n # Step 2: Optional HyDE\n if use_hyde:\n query_embedding = await self._hyde_embed(processed_query)\n else:\n query_embedding = self.embedding_model.encode(processed_query)\n\n # Step 3: Hybrid search (vector + BM25)\n vector_results = self.vector_store.search(\n vector=query_embedding,\n filter={\"tenant_id\": tenant_id},\n top_k=50\n )\n bm25_results = self.bm25_index.search(processed_query, top_k=50)\n\n # Step 4: Merge with RRF\n merged = reciprocal_rank_fusion(\n vector_results,\n bm25_results,\n vector_weight=0.6\n )[:30]\n\n # Step 5: Optional query expansion\n if use_query_expansion:\n expanded_queries = await self._expand_query(processed_query)\n for exp_query in expanded_queries[1:]: # Skip original\n exp_embedding = self.embedding_model.encode(exp_query)\n exp_results = self.vector_store.search(\n vector=exp_embedding,\n filter={\"tenant_id\": tenant_id},\n top_k=10\n )\n merged.extend(exp_results)\n merged = deduplicate_by_id(merged)[:30]\n\n # Step 6: Rerank\n documents = [r.text for r in merged]\n reranked = self.reranker.rerank(\n query=processed_query,\n documents=documents,\n top_k=top_k\n )\n\n return [\n {\n \"text\": doc,\n \"score\": score,\n \"metadata\": merged[i].metadata\n }\n for i, (doc, score) in enumerate(reranked)\n ]\n\n def _preprocess_query(self, query: str) -> str:\n \"\"\"Clean and normalize query.\"\"\"\n import re\n query = re.sub(r'\\s+', ' ', query).strip()\n return query\n\n async def _hyde_embed(self, query: str) -> list[float]:\n \"\"\"Generate hypothetical document and embed.\"\"\"\n # Implementation from HyDE section\n pass\n\n async def _expand_query(self, query: str) -> list[str]:\n \"\"\"Expand query with variations.\"\"\"\n # Implementation from Query Expansion section\n pass\n```\n\n---\n\n## Performance Benchmarks\n\n| Technique | Latency Impact | Quality Impact | Cost Impact |\n|-----------|----------------|----------------|-------------|\n| Vector only | Baseline | Baseline | Baseline |\n| + BM25 hybrid | +10-20ms | +5-15% precision | Minimal |\n| + Reranking | +50-100ms | +10-20% precision | +$0.001/query |\n| + Query expansion | +100-200ms | +5-10% recall | +$0.002/query |\n| + HyDE | +200-500ms | +10-25% precision | +$0.003/query |\n\n---\n\n## Quick Reference\n\n| Goal | Technique | Implementation |\n|------|-----------|----------------|\n| Improve precision | Reranking | Cross-encoder or Cohere |\n| Improve recall | Query expansion | LLM-generated variations |\n| Handle synonyms | Hybrid search | BM25 + vector with RRF |\n| Concept search | HyDE | Hypothetical doc embedding |\n| Multi-tenant | Metadata filter | Mandatory tenant_id |\n| Fresh content | Temporal filter | Date range queries |\n| Complex questions | Decomposition | Sub-question retrieval |\n\n## Related Skills\n\n- **RAG Architect** - System design and architecture\n- **NLP Engineer** - Query understanding\n- **Python Pro** - Async implementation\n- **ML Pipeline** - Model serving for rerankers\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":21926,"content_sha256":"3a5b7181bd0991be743b2761a792f5371d93b38747d78393606f5ce75247c90c"},{"filename":"references/vector-databases.md","content":"# Vector Databases\n\n---\n\n## Database Comparison Matrix\n\n| Feature | Pinecone | Weaviate | Qdrant | Chroma | pgvector |\n|---------|----------|----------|--------|--------|----------|\n| **Hosting** | Managed only | Managed + Self-hosted | Managed + Self-hosted | Self-hosted (cloud beta) | Self-hosted |\n| **Hybrid Search** | Yes (sparse-dense) | Yes (BM25 + vector) | Yes (sparse vectors) | Limited | Manual (+ pg_trgm) |\n| **Filtering** | Excellent | Excellent | Excellent | Basic | SQL-native |\n| **Max Dimensions** | 20,000 | Unlimited | 65,535 | Unlimited | 2,000 |\n| **Pricing Model** | Per-vector/query | Per-node | Per-node | Free (OSS) | Free (extension) |\n| **Multi-tenancy** | Namespaces | Multi-tenant class | Collections + payloads | Collections | Schema/RLS |\n| **Best For** | Enterprise SaaS | Semantic apps | High-performance | Prototyping | Postgres shops |\n\n## When to Use Each\n\n### Pinecone\n```\nBest For:\n- Enterprise RAG with strict SLAs\n- Teams wanting zero infrastructure management\n- Applications needing sparse-dense hybrid search\n- High-volume production with predictable costs\n\nWhen to Avoid:\n- Cost-sensitive projects (expensive at scale)\n- Need for self-hosting or data residency\n- Complex filtering requirements beyond metadata\n- Wanting to avoid vendor lock-in\n```\n\n### Weaviate\n```\nBest For:\n- Semantic search with built-in vectorization\n- Multi-modal (text, image) applications\n- GraphQL-native teams\n- Hybrid BM25 + vector search requirements\n\nWhen to Avoid:\n- Simple embedding storage only\n- Memory-constrained environments\n- Teams unfamiliar with GraphQL\n```\n\n### Qdrant\n```\nBest For:\n- High-performance, low-latency requirements\n- Complex filtering with payload indexes\n- Rust/performance-focused teams\n- Self-hosted with full control\n\nWhen to Avoid:\n- Teams wanting fully managed simplicity\n- GraphQL preference (REST/gRPC only)\n```\n\n### Chroma\n```\nBest For:\n- Local development and prototyping\n- LangChain/LlamaIndex integration\n- Simple RAG proof-of-concepts\n- Educational projects\n\nWhen to Avoid:\n- Production workloads at scale\n- Multi-tenant applications\n- High availability requirements\n```\n\n### pgvector\n```\nBest For:\n- Existing PostgreSQL infrastructure\n- Transactional + vector in same DB\n- SQL-native teams\n- Cost optimization (no new infra)\n\nWhen to Avoid:\n- Vectors > 2000 dimensions\n- Billions of vectors (scaling limits)\n- Sub-millisecond latency requirements\n```\n\n---\n\n## Pinecone Setup\n\n```python\nfrom pinecone import Pinecone, ServerlessSpec\n\n# Initialize client\npc = Pinecone(api_key=\"\u003cpinecone-api-key>\")\n\n# Create index with serverless\npc.create_index(\n name=\"rag-index\",\n dimension=1536, # OpenAI ada-002\n metric=\"cosine\",\n spec=ServerlessSpec(\n cloud=\"aws\",\n region=\"us-east-1\"\n )\n)\n\n# Get index reference\nindex = pc.Index(\"rag-index\")\n\n# Upsert vectors with metadata\nindex.upsert(\n vectors=[\n {\n \"id\": \"doc-1\",\n \"values\": embedding_vector,\n \"metadata\": {\n \"source\": \"manual.pdf\",\n \"page\": 42,\n \"section\": \"installation\",\n \"tenant_id\": \"acme-corp\"\n }\n }\n ],\n namespace=\"production\"\n)\n\n# Query with metadata filter\nresults = index.query(\n vector=query_embedding,\n top_k=10,\n include_metadata=True,\n namespace=\"production\",\n filter={\n \"tenant_id\": {\"$eq\": \"acme-corp\"},\n \"section\": {\"$in\": [\"installation\", \"setup\"]}\n }\n)\n\n# Hybrid search (sparse-dense)\nfrom pinecone_text.sparse import BM25Encoder\n\nbm25 = BM25Encoder()\nbm25.fit(corpus) # Fit on your documents\n\nresults = index.query(\n vector=dense_embedding,\n sparse_vector=bm25.encode_queries(query_text),\n top_k=10,\n alpha=0.5 # Balance dense vs sparse\n)\n```\n\n---\n\n## Weaviate Setup\n\n```python\nimport weaviate\nfrom weaviate.classes.config import Configure, Property, DataType\n\n# Connect to Weaviate Cloud\nclient = weaviate.connect_to_weaviate_cloud(\n cluster_url=\"https://your-cluster.weaviate.network\",\n auth_credentials=weaviate.auth.AuthApiKey(\"\u003cweaviate-api-key>\")\n)\n\n# Or self-hosted\nclient = weaviate.connect_to_local(\n host=\"localhost\",\n port=8080\n)\n\n# Create collection with vectorizer\nclient.collections.create(\n name=\"Document\",\n vectorizer_config=Configure.Vectorizer.text2vec_openai(\n model=\"text-embedding-3-small\"\n ),\n properties=[\n Property(name=\"content\", data_type=DataType.TEXT),\n Property(name=\"source\", data_type=DataType.TEXT),\n Property(name=\"page\", data_type=DataType.INT),\n Property(name=\"tenant_id\", data_type=DataType.TEXT, index_filterable=True)\n ]\n)\n\n# Insert with auto-vectorization\ndocuments = client.collections.get(\"Document\")\ndocuments.data.insert(\n properties={\n \"content\": \"Installation guide content...\",\n \"source\": \"manual.pdf\",\n \"page\": 42,\n \"tenant_id\": \"acme-corp\"\n }\n)\n\n# Or with pre-computed vector\ndocuments.data.insert(\n properties={\"content\": \"...\", \"source\": \"...\"},\n vector=precomputed_embedding\n)\n\n# Hybrid search (BM25 + vector)\nfrom weaviate.classes.query import MetadataQuery, Filter\n\nresults = documents.query.hybrid(\n query=\"how to install\",\n alpha=0.5, # 0=BM25 only, 1=vector only\n limit=10,\n filters=Filter.by_property(\"tenant_id\").equal(\"acme-corp\"),\n return_metadata=MetadataQuery(score=True, explain_score=True)\n)\n\nfor obj in results.objects:\n print(f\"Score: {obj.metadata.score}, Content: {obj.properties['content'][:100]}\")\n\nclient.close()\n```\n\n---\n\n## Qdrant Setup\n\n```python\nfrom qdrant_client import QdrantClient\nfrom qdrant_client.models import (\n Distance, VectorParams, PointStruct,\n Filter, FieldCondition, MatchValue,\n PayloadSchemaType\n)\n\n# Connect to Qdrant Cloud\nclient = QdrantClient(\n url=\"https://your-cluster.qdrant.io\",\n api_key=\"\u003cqdrant-api-key>\"\n)\n\n# Or local\nclient = QdrantClient(host=\"localhost\", port=6333)\n\n# Create collection\nclient.create_collection(\n collection_name=\"documents\",\n vectors_config=VectorParams(\n size=1536,\n distance=Distance.COSINE\n )\n)\n\n# Create payload index for fast filtering\nclient.create_payload_index(\n collection_name=\"documents\",\n field_name=\"tenant_id\",\n field_schema=PayloadSchemaType.KEYWORD\n)\n\n# Upsert points\nclient.upsert(\n collection_name=\"documents\",\n points=[\n PointStruct(\n id=\"doc-1\",\n vector=embedding_vector,\n payload={\n \"content\": \"Installation guide...\",\n \"source\": \"manual.pdf\",\n \"page\": 42,\n \"tenant_id\": \"acme-corp\"\n }\n )\n ]\n)\n\n# Search with filter\nresults = client.search(\n collection_name=\"documents\",\n query_vector=query_embedding,\n limit=10,\n query_filter=Filter(\n must=[\n FieldCondition(\n key=\"tenant_id\",\n match=MatchValue(value=\"acme-corp\")\n )\n ]\n ),\n with_payload=True\n)\n\n# Batch upsert for large datasets\nfrom qdrant_client.models import Batch\n\nclient.upsert(\n collection_name=\"documents\",\n points=Batch(\n ids=ids_list,\n vectors=vectors_list,\n payloads=payloads_list\n )\n)\n```\n\n---\n\n## Chroma Setup\n\n```python\nimport chromadb\nfrom chromadb.config import Settings\n\n# Persistent local storage\nclient = chromadb.PersistentClient(\n path=\"./chroma_data\",\n settings=Settings(anonymized_telemetry=False)\n)\n\n# Create collection with custom embedding function\nfrom chromadb.utils.embedding_functions import OpenAIEmbeddingFunction\n\nembedding_fn = OpenAIEmbeddingFunction(\n api_key=\"\u003copenai-api-key>\",\n model_name=\"text-embedding-3-small\"\n)\n\ncollection = client.get_or_create_collection(\n name=\"documents\",\n embedding_function=embedding_fn,\n metadata={\"hnsw:space\": \"cosine\"}\n)\n\n# Add documents (auto-embeds)\ncollection.add(\n ids=[\"doc-1\", \"doc-2\"],\n documents=[\"Installation guide...\", \"Configuration steps...\"],\n metadatas=[\n {\"source\": \"manual.pdf\", \"page\": 42},\n {\"source\": \"manual.pdf\", \"page\": 43}\n ]\n)\n\n# Or with pre-computed embeddings\ncollection.add(\n ids=[\"doc-3\"],\n embeddings=[precomputed_vector],\n metadatas=[{\"source\": \"guide.pdf\"}],\n documents=[\"Original text for reference\"]\n)\n\n# Query\nresults = collection.query(\n query_texts=[\"how to install\"],\n n_results=10,\n where={\"source\": \"manual.pdf\"},\n include=[\"documents\", \"metadatas\", \"distances\"]\n)\n\n# Update existing document\ncollection.update(\n ids=[\"doc-1\"],\n documents=[\"Updated installation guide...\"],\n metadatas=[{\"source\": \"manual_v2.pdf\", \"page\": 42}]\n)\n```\n\n---\n\n## pgvector Setup\n\n```sql\n-- Enable extension\nCREATE EXTENSION IF NOT EXISTS vector;\n\n-- Create table with vector column\nCREATE TABLE documents (\n id SERIAL PRIMARY KEY,\n content TEXT NOT NULL,\n embedding vector(1536), -- OpenAI dimensions\n source VARCHAR(255),\n page INTEGER,\n tenant_id VARCHAR(100),\n created_at TIMESTAMP DEFAULT NOW()\n);\n\n-- Create HNSW index (recommended for most cases)\nCREATE INDEX ON documents\nUSING hnsw (embedding vector_cosine_ops)\nWITH (m = 16, ef_construction = 64);\n\n-- Or IVFFlat for very large datasets\nCREATE INDEX ON documents\nUSING ivfflat (embedding vector_cosine_ops)\nWITH (lists = 100);\n\n-- Create index on filter columns\nCREATE INDEX ON documents (tenant_id);\n```\n\n```python\nimport psycopg2\nfrom pgvector.psycopg2 import register_vector\n\nconn = psycopg2.connect(\"postgresql://localhost/ragdb\")\nregister_vector(conn)\n\n# Insert with embedding\ncur = conn.cursor()\ncur.execute(\n \"\"\"\n INSERT INTO documents (content, embedding, source, page, tenant_id)\n VALUES (%s, %s, %s, %s, %s)\n RETURNING id\n \"\"\",\n (\"Installation guide...\", embedding_vector, \"manual.pdf\", 42, \"acme-corp\")\n)\n\n# Similarity search with filter\ncur.execute(\n \"\"\"\n SELECT id, content, source, page,\n 1 - (embedding \u003c=> %s) AS similarity\n FROM documents\n WHERE tenant_id = %s\n ORDER BY embedding \u003c=> %s\n LIMIT 10\n \"\"\",\n (query_embedding, \"acme-corp\", query_embedding)\n)\n\nresults = cur.fetchall()\n\n# Hybrid search with pg_trgm\ncur.execute(\n \"\"\"\n SELECT id, content,\n (0.5 * (1 - (embedding \u003c=> %s))) +\n (0.5 * similarity(content, %s)) AS hybrid_score\n FROM documents\n WHERE tenant_id = %s\n AND content %% %s -- Trigram similarity threshold\n ORDER BY hybrid_score DESC\n LIMIT 10\n \"\"\",\n (query_embedding, query_text, \"acme-corp\", query_text)\n)\n```\n\n---\n\n## Index Tuning Guide\n\n### HNSW Parameters\n\n| Parameter | Description | Trade-off |\n|-----------|-------------|-----------|\n| `m` | Connections per node | Higher = better recall, more memory |\n| `ef_construction` | Build-time search width | Higher = better index, slower build |\n| `ef_search` | Query-time search width | Higher = better recall, slower query |\n\n```python\n# Qdrant HNSW tuning\nclient.update_collection(\n collection_name=\"documents\",\n hnsw_config=HnswConfigDiff(\n m=16, # Default: 16, increase for better recall\n ef_construct=100, # Default: 100, higher for better index\n full_scan_threshold=10000 # Use brute force below this size\n )\n)\n\n# Query-time ef adjustment\nresults = client.search(\n collection_name=\"documents\",\n query_vector=query_embedding,\n limit=10,\n search_params=SearchParams(hnsw_ef=128) # Higher for better recall\n)\n```\n\n### Quantization for Scale\n\n```python\n# Qdrant scalar quantization (4x memory reduction)\nfrom qdrant_client.models import ScalarQuantization, ScalarQuantizationConfig\n\nclient.update_collection(\n collection_name=\"documents\",\n quantization_config=ScalarQuantization(\n scalar=ScalarQuantizationConfig(\n type=\"int8\",\n quantile=0.99,\n always_ram=True\n )\n )\n)\n```\n\n---\n\n## Multi-Tenancy Patterns\n\n### Namespace Isolation (Pinecone)\n```python\n# Tenant data in separate namespaces\nindex.upsert(vectors=[...], namespace=\"tenant-acme\")\nindex.upsert(vectors=[...], namespace=\"tenant-globex\")\n\n# Query within tenant namespace\nresults = index.query(\n vector=query_embedding,\n namespace=\"tenant-acme\",\n top_k=10\n)\n```\n\n### Metadata Filtering (Qdrant/Weaviate)\n```python\n# Add tenant_id to all documents\npoint = PointStruct(\n id=\"doc-1\",\n vector=embedding,\n payload={\"tenant_id\": \"acme\", \"content\": \"...\"}\n)\n\n# Always filter by tenant\nresults = client.search(\n collection_name=\"documents\",\n query_vector=query_embedding,\n query_filter=Filter(\n must=[FieldCondition(key=\"tenant_id\", match=MatchValue(value=\"acme\"))]\n )\n)\n```\n\n### Collection per Tenant (High Isolation)\n```python\n# Create tenant-specific collection\nclient.create_collection(\n collection_name=f\"docs_{tenant_id}\",\n vectors_config=VectorParams(size=1536, distance=Distance.COSINE)\n)\n```\n\n---\n\n## Decision Flowchart\n\n```\nStart\n │\n ├─ Need managed service with zero ops?\n │ └─ Yes → Pinecone\n │\n ├─ Have existing PostgreSQL?\n │ └─ Yes → pgvector (if vectors \u003c 2000 dims)\n │\n ├─ Need built-in vectorization?\n │ └─ Yes → Weaviate\n │\n ├─ Need maximum performance + self-host?\n │ └─ Yes → Qdrant\n │\n ├─ Prototyping / local development?\n │ └─ Yes → Chroma\n │\n └─ Default recommendation → Qdrant (balance of features/performance)\n```\n\n---\n\n## Quick Reference\n\n| Task | Pinecone | Weaviate | Qdrant | pgvector |\n|------|----------|----------|--------|----------|\n| Create index/collection | `create_index()` | `collections.create()` | `create_collection()` | `CREATE TABLE` |\n| Insert | `upsert()` | `data.insert()` | `upsert()` | `INSERT` |\n| Search | `query()` | `query.near_vector()` | `search()` | `ORDER BY \u003c=>` |\n| Filter | `filter={}` | `Filter.by_property()` | `query_filter=Filter()` | `WHERE` |\n| Delete | `delete()` | `data.delete_by_id()` | `delete()` | `DELETE` |\n| Hybrid | sparse_vector param | `query.hybrid()` | sparse vectors | Manual |\n\n## Related Skills\n\n- **Database Optimizer** - Index tuning and query performance\n- **Cloud Architect** - Infrastructure decisions for vector DB hosting\n- **Python Pro** - Implementation patterns with async clients\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":14274,"content_sha256":"faddfb37db3207e346c83298c8a972442501927ec73b278e6f4d1cded13ac115"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"RAG Architect","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Core Workflow","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Requirements Analysis","type":"text","marks":[{"type":"strong"}]},{"text":" — Identify retrieval needs, latency constraints, accuracy requirements, and scale","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Vector Store Design","type":"text","marks":[{"type":"strong"}]},{"text":" — Select database, schema design, indexing strategy, sharding approach","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Chunking Strategy","type":"text","marks":[{"type":"strong"}]},{"text":" — Document splitting, overlap, semantic boundaries, metadata enrichment","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Retrieval Pipeline","type":"text","marks":[{"type":"strong"}]},{"text":" — Embedding selection, query transformation, hybrid search, reranking","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Evaluation & Iteration","type":"text","marks":[{"type":"strong"}]},{"text":" — Metrics tracking, retrieval debugging, continuous optimization","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"For each step, validate before moving on (see checkpoints below).","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Reference Guide","type":"text"}]},{"type":"paragraph","content":[{"text":"Load detailed guidance based on context:","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Topic","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Reference","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Load When","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Vector Databases","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"references/vector-databases.md","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Comparing Pinecone, Weaviate, Chroma, pgvector, Qdrant","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Embedding Models","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"references/embedding-models.md","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Selecting embeddings, fine-tuning, dimension trade-offs","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Chunking Strategies","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"references/chunking-strategies.md","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Document splitting, overlap, semantic chunking","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Retrieval Optimization","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"references/retrieval-optimization.md","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Hybrid search, reranking, query expansion, filtering","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"RAG Evaluation","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"references/rag-evaluation.md","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Metrics, evaluation frameworks, debugging retrieval","type":"text"}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Implementation Examples","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"1. Chunking Documents","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"from langchain.text_splitter import RecursiveCharacterTextSplitter\n\n# Evaluate chunk_size on your domain data — never use 512 blindly\nsplitter = RecursiveCharacterTextSplitter(\n chunk_size=800,\n chunk_overlap=100,\n separators=[\"\\n\\n\", \"\\n\", \". \", \" \"],\n)\n\nchunks = splitter.create_documents(\n texts=[doc.page_content for doc in raw_docs],\n metadatas=[{\"source\": doc.metadata[\"source\"], \"timestamp\": doc.metadata.get(\"timestamp\")} for doc in raw_docs],\n)","type":"text"}]},{"type":"paragraph","content":[{"text":"Checkpoint:","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"assert all(c.metadata.get(\"source\") for c in chunks), \"Missing source metadata\"","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"2. Generating Embeddings & Indexing","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"from openai import OpenAI\nimport qdrant_client\nfrom qdrant_client.models import VectorParams, Distance, PointStruct\n\nclient = OpenAI()\nqdrant = qdrant_client.QdrantClient(\"localhost\", port=6333)\n\n# Create collection\nqdrant.recreate_collection(\n collection_name=\"knowledge_base\",\n vectors_config=VectorParams(size=1536, distance=Distance.COSINE),\n)\n\ndef embed_chunks(chunks: list[str], model: str = \"text-embedding-3-small\") -> list[list[float]]:\n response = client.embeddings.create(input=chunks, model=model)\n return [r.embedding for r in response.data]\n\n# Idempotent upsert with deduplication via deterministic IDs\nimport hashlib, uuid\n\npoints = []\nfor i, chunk in enumerate(chunks):\n doc_id = str(uuid.UUID(hashlib.md5(chunk.page_content.encode()).hexdigest()))\n embedding = embed_chunks([chunk.page_content])[0]\n points.append(PointStruct(id=doc_id, vector=embedding, payload=chunk.metadata))\n\nqdrant.upsert(collection_name=\"knowledge_base\", points=points)","type":"text"}]},{"type":"paragraph","content":[{"text":"Checkpoint:","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"assert qdrant.count(\"knowledge_base\").count == len(set(p.id for p in points)), \"Deduplication failed\"","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"3. Hybrid Search (Vector + BM25)","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"from qdrant_client.models import Filter, FieldCondition, MatchValue, SparseVector\nfrom rank_bm25 import BM25Okapi\n\ndef hybrid_search(query: str, tenant_id: str, top_k: int = 20) -> list:\n # Dense retrieval\n query_embedding = embed_chunks([query])[0]\n tenant_filter = Filter(must=[FieldCondition(key=\"tenant_id\", match=MatchValue(value=tenant_id))])\n dense_results = qdrant.search(\n collection_name=\"knowledge_base\",\n query_vector=query_embedding,\n query_filter=tenant_filter,\n limit=top_k,\n )\n\n # Sparse retrieval (BM25)\n corpus = [r.payload.get(\"text\", \"\") for r in dense_results]\n bm25 = BM25Okapi([doc.split() for doc in corpus])\n bm25_scores = bm25.get_scores(query.split())\n\n # Reciprocal Rank Fusion\n ranked = sorted(\n zip(dense_results, bm25_scores),\n key=lambda x: 0.6 * x[0].score + 0.4 * x[1],\n reverse=True,\n )\n return [r for r, _ in ranked[:top_k]]","type":"text"}]},{"type":"paragraph","content":[{"text":"Checkpoint:","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"assert len(hybrid_search(\"test query\", tenant_id=\"demo\")) > 0, \"Hybrid search returned no results\"","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"4. Reranking Top-K Results","type":"text"}]},{"type":"paragraph","content":[{"text":"Load provider API keys from environment variables or a secrets manager; never commit them to source code.","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"import os\n\nimport cohere\n\nco = cohere.Client(os.environ[\"COHERE_API_KEY\"])\n\ndef rerank(query: str, results: list, top_n: int = 5) -> list:\n docs = [r.payload.get(\"text\", \"\") for r in results]\n reranked = co.rerank(query=query, documents=docs, top_n=top_n, model=\"rerank-english-v3.0\")\n return [results[r.index] for r in reranked.results]","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"5. Retrieval Evaluation","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"# Run precision@k and recall@k against a labeled evaluation set\n# python evaluate.py --metrics precision@10 recall@10 mrr --collection knowledge_base\n\nfrom ragas import evaluate\nfrom ragas.metrics import context_precision, context_recall, faithfulness, answer_relevancy\nfrom datasets import Dataset\n\neval_dataset = Dataset.from_dict({\n \"question\": questions,\n \"contexts\": retrieved_contexts,\n \"answer\": generated_answers,\n \"ground_truth\": ground_truth_answers,\n})\n\nresults = evaluate(eval_dataset, metrics=[context_precision, context_recall, faithfulness, answer_relevancy])\nprint(results)","type":"text"}]},{"type":"paragraph","content":[{"text":"Checkpoint:","type":"text","marks":[{"type":"strong"}]},{"text":" Target ","type":"text"},{"text":"context_precision >= 0.7","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"context_recall >= 0.6","type":"text","marks":[{"type":"code_inline"}]},{"text":" before moving to LLM integration.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Constraints","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"MUST DO","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Evaluate multiple embedding models on your domain data before committing","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Implement hybrid search (vector + keyword) for production systems","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Add metadata filters for multi-tenant or domain-specific retrieval","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Measure retrieval metrics (precision@k, recall@k, MRR, NDCG)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use reranking for top-k results before passing context to LLM","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Implement idempotent ingestion with deduplication (deterministic IDs)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Monitor retrieval latency and quality over time","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Version embeddings and plan for model migration","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"MUST NOT DO","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use default chunk size (512) without evaluation on your domain data","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Skip metadata enrichment (source, timestamp, section)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Ignore retrieval quality metrics in favor of only LLM output quality","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Store raw documents without preprocessing/cleaning","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use cosine similarity alone for complex multi-domain retrieval","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Deploy without testing on production-like data volumes","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Forget to handle edge cases (empty results, malformed docs)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Couple the embedding model tightly to application code","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Output Templates","type":"text"}]},{"type":"paragraph","content":[{"text":"When designing RAG architecture, deliver:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"System architecture diagram (ingestion + retrieval pipelines)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Vector database selection with trade-off analysis","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Chunking strategy with examples and rationale","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Retrieval pipeline design (query → results flow)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Evaluation plan with metrics, benchmarks, and pass/fail thresholds","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Maintained by ","type":"text"},{"text":"@jeffallan","type":"text","marks":[{"type":"link","attrs":{"href":"https://github.com/jeffallan","title":null}}]},{"text":", Principal Consultant at ","type":"text"},{"text":"Synergetic Solutions","type":"text","marks":[{"type":"link","attrs":{"href":"https://synergetic.solutions","title":null}}]}]},{"type":"paragraph","content":[{"text":"Documentation","type":"text","marks":[{"type":"link","attrs":{"href":"https://jeffallan.github.io/claude-skills/skills/data-ml/rag-architect/","title":null}}]}]}]},"metadata":{"date":"2026-11-02","name":"rag-architect","author":"@skillopedia","source":{"stars":12181,"repo_name":"claude-skills","origin_url":"https://github.com/Jeffallan/claude-skills/blob/HEAD/skills/rag-architect/SKILL.md","repo_owner":"Jeffallan","body_sha256":"36e21290e5df36ed3eaf007d0e91781c7453dc46c6dedd6d31c1c382b28cf137","cluster_key":"e913c47f1d49e4786c2cdad39cece761e0f88e4f76953f582f8bcce00b4398c8","clean_bundle":{"format":"clean-skill-bundle-v1","source":"Jeffallan/claude-skills/skills/rag-architect/SKILL.md","text_bytes":100913,"source_commit":"078a25cc5bfd8011459be7b91d004e900c1d02d3","skills_sh_rank":4504,"attachment_count":5,"text_attachments":5,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/rag-architect/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"ai-agent-development","category_label":"AI"},"exact_dupes_collapsed_into_this":0},"license":"MIT","version":"v1","category":"ai-agent-development","metadata":{"role":"architect","scope":"system-design","author":"https://github.com/Jeffallan","domain":"data-ml","company":"https://synergetic.solutions","version":"1.1.0","triggers":"RAG, retrieval-augmented generation, vector search, embeddings, semantic search, vector database, document retrieval, knowledge base, context retrieval, similarity search","output-format":"architecture","related-skills":"python-pro, database-optimizer, monitoring-expert, api-designer"},"import_tag":"skills-sh-top20k-2026-11","description":"Designs and implements production-grade RAG systems by chunking documents, generating embeddings, configuring vector stores, building hybrid search pipelines, applying reranking, and evaluating retrieval quality. Use when building RAG systems, vector databases, or knowledge-grounded AI applications requiring semantic search, document retrieval, context augmentation, similarity search, or embedding-based indexing."}},"renderedAt":1793666562685}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.