Rag Skills

RAG Skills for LlamaFarm

Framework-specific patterns and code review checklists for the RAG component.

Extends: python-skills - All Python best practices apply here.

Component Overview

Directory Structure

rag/
├── api.py                 # Search and database APIs
├── celery_app.py          # Celery configuration
├── main.py                # Entry point
├── core/
│   ├── base.py            # Document, Component, Pipeline ABCs
│   ├── factories.py       # Component factories
│   ├── ingest_handler.py  # File ingestion with safety checks
│   ├── blob_processor.py  # Binary file processing
│   ├── settings.py        # Pydantic settings
│   └── logging.py         # RAGStructLogger
├── components/
│   ├── embedders/         # Embedding providers
│   ├── extractors/        # Metadata extractors
│   ├── parsers/           # Document parsers (LlamaIndex)
│   ├── retrievers/        # Retrieval strategies
│   └── stores/            # Vector stores (ChromaDB, FAISS)
├── tasks/                 # Celery tasks
│   ├── ingest_tasks.py    # File ingestion
│   ├── search_tasks.py    # Database search
│   ├── query_tasks.py     # Complex queries
│   ├── health_tasks.py    # Health checks
│   └── stats_tasks.py     # Statistics
└── utils/
    └── embedding_safety.py  # Circuit breaker, validation

Quick Reference

Core Patterns

Document Dataclass

from dataclasses import dataclass, field
from typing import Any

@dataclass
class Document:
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    source: str | None = None
    embeddings: list[float] | None = None

Component Abstract Base Class

from abc import ABC, abstractmethod

class Component(ABC):
    def __init__(
        self,
        name: str | None = None,
        config: dict[str, Any] | None = None,
        project_dir: Path | None = None,
    ):
        self.name = name or self.__class__.__name__
        self.config = config or {}
        self.logger = RAGStructLogger(__name__).bind(name=self.name)
        self.project_dir = project_dir

    @abstractmethod
    def process(self, documents: list[Document]) -> ProcessingResult:
        pass

Retrieval Strategy Pattern

class RetrievalStrategy(Component, ABC):
    @abstractmethod
    def retrieve(
        self,
        query_embedding: list[float],
        vector_store,
        top_k: int = 5,
        **kwargs
    ) -> RetrievalResult:
        pass

    @abstractmethod
    def supports_vector_store(self, vector_store_type: str) -> bool:
        pass

Embedder with Circuit Breaker

class Embedder(Component):
    DEFAULT_FAILURE_THRESHOLD = 5
    DEFAULT_RESET_TIMEOUT = 60.0

    def __init__(self, ...):
        super().__init__(...)
        self._circuit_breaker = CircuitBreaker(
            failure_threshold=config.get("failure_threshold", 5),
            reset_timeout=config.get("reset_timeout", 60.0),
        )
        self._fail_fast = config.get("fail_fast", True)

    def embed_text(self, text: str) -> list[float]:
        self.check_circuit_breaker()
        try:
            embedding = self._call_embedding_api(text)
            self.record_success()
            return embedding
        except Exception as e:
            self.record_failure(e)
            if self._fail_fast:
                raise EmbedderUnavailableError(str(e)) from e
            return [0.0] * self.get_embedding_dimension()

Review Checklist Summary

When reviewing RAG code:

1. LlamaIndex (Medium priority)

• Proper chunking configuration
• Metadata preservation during parsing
• Error handling for unsupported formats

2. ChromaDB (High priority)

• Thread-safe client access
• Proper distance metric selection
• Metadata type compatibility

3. Celery (High priority)

• Task routing to correct queue
• Error logging with context
• Proper serialization

4. Performance (Medium priority)

• Batch processing for embeddings
• Deduplication enabled
• Appropriate caching

See individual topic files for detailed checklists with grep patterns.

Celery Patterns for RAG Workers

Best practices for Celery task definition, error handling, and worker configuration.

Architecture Overview

Server (FastAPI) -> Celery Broker -> RAG Worker -> Task Execution
                         |                              |
                    filesystem/Redis              ChromaDB/Embedders

Key Files

Celery App Configuration

Location: rag/celery_app.py

from celery import Celery, signals

app = Celery("LlamaFarm-RAG-Worker")

app.conf.update(
    # Broker configuration
    broker_url=settings.CELERY_BROKER_URL or "filesystem://",
    result_backend=settings.CELERY_RESULT_BACKEND,

    # Serialization
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",

    # Timezone
    timezone="UTC",
    enable_utc=True,

    # Task routing - CRITICAL: only handle rag.* tasks
    task_routes={
        "rag.*": {"queue": "rag"},
    },

    # Import task modules
    imports=(
        "tasks.search_tasks",
        "tasks.ingest_tasks",
        "tasks.query_tasks",
        "tasks.health_tasks",
        "tasks.stats_tasks",
    ),
)

# Prevent Celery from overriding logging
@signals.setup_logging.connect
def setup_celery_logging(**kwargs):
    pass

Task Definition Pattern

Base Task Class

from celery import Task

class IngestTask(Task):
    """Base task with error logging."""

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.error(
            "RAG ingest task failed",
            extra={
                "task_id": task_id,
                "task_name": self.name,
                "error": str(exc),
                "task_args": args,
                "task_kwargs": kwargs,
            },
        )

Task Definition

@app.task(bind=True, base=IngestTask, name="rag.ingest_file")
def ingest_file_task(
    self,
    project_dir: str,
    strategy_name: str,
    database_name: str,
    source_path: str,
    filename: str | None = None,
) -> tuple[bool, dict[str, Any]]:
    """
    Ingest a file using RAG system.

    Args:
        project_dir: Project directory path
        strategy_name: Data processing strategy
        database_name: Target database
        source_path: File path to ingest
        filename: Optional display name

    Returns:
        Tuple of (success, details_dict)
    """
    logger.info(
        "Starting RAG file ingestion",
        extra={
            "task_id": self.request.id,
            "project_dir": project_dir,
            "source_path": source_path,
        },
    )

    try:
        # Task implementation
        handler = IngestHandler(...)
        result = handler.ingest_file(...)
        return True, result
    except Exception as e:
        logger.error(
            "Ingestion failed",
            extra={"task_id": self.request.id, "error": str(e)},
        )
        return False, {"error": str(e)}

Worker Configuration

Thread Pool (Required for ChromaDB)

def run_worker():
    # MUST use thread pool to avoid SQLite locking with ChromaDB
    worker_args = ["worker", "-Q", "rag", "--pool=threads"]

    concurrency = os.getenv("LF_CELERY_CONCURRENCY")
    if concurrency:
        worker_args.extend(["--concurrency", concurrency])

    app.worker_main(argv=worker_args)

macOS Fork Safety

import multiprocessing
import sys

if sys.platform == "darwin":
    multiprocessing.set_start_method("spawn", force=True)

Code Review Checklist

1. Task Naming Convention

Description: Tasks must follow the rag.* naming pattern for routing.

Search Pattern:

grep -rn '@app.task.*name="rag\.' rag/tasks/

Pass Criteria:

• All tasks named rag.<action>
• Names are descriptive
• Consistent naming pattern

Fail Criteria:

• Missing name parameter
• Name doesn't start with rag.
• Inconsistent naming

Severity: Critical

Recommendation: Always use name="rag.<module>.<action>" format.

---

2. Task Binding

Description: Tasks should be bound to access request context.

Search Pattern:

grep -rn '@app.task.*bind=True' rag/tasks/

Pass Criteria:

• bind=True in decorator
• self.request.id used for logging
• Task has access to context

Fail Criteria:

• bind=False or missing
• No task ID in logs
• Missing request context

Severity: Medium

Recommendation: Use bind=True for all tasks that need logging.

---

3. Base Task Class Usage

Description: Tasks should use base class for consistent error handling.

Search Pattern:

grep -rn 'base=.*Task' rag/tasks/

Pass Criteria:

• Custom base task defined
• on_failure logs errors
• Base task used in decorator

Fail Criteria:

• No base task
• Missing on_failure handler
• Inconsistent base usage

Severity: Medium

Recommendation: Define base task with on_failure logging.

---

4. Error Logging Context

Description: Task errors must include sufficient context for debugging.

Search Pattern:

grep -rn 'logger.error.*extra=\|on_failure' rag/tasks/

Pass Criteria:

• task_id included in logs
• Error message captured
• Input parameters logged

Fail Criteria:

• Missing task_id
• Only exception logged
• No input context

Severity: High

Recommendation: Include task_id, error, and key parameters in error logs.

---

5. JSON Serialization Compatibility

Description: Task arguments and returns must be JSON-serializable.

Search Pattern:

grep -rn 'def .*_task\(' rag/tasks/ -A 20

Pass Criteria:

• Arguments are primitive types
• Return values are dicts/lists
• No Path or custom objects

Fail Criteria:

• Path objects as arguments
• Custom class returns
• Datetime without conversion

Severity: High

Recommendation: Convert Path to str, datetime to ISO string.

---

6. Thread Pool Requirement

Description: Worker must use thread pool for ChromaDB compatibility.

Search Pattern:

grep -rn "pool=threads\|--pool" rag/

Pass Criteria:

• --pool=threads in worker args
• No prefork pool
• Documentation mentions requirement

Fail Criteria:

• prefork or default pool
• Missing pool specification
• No documentation

Severity: Critical

Recommendation: Always use --pool=threads for RAG workers.

---

7. Queue Routing

Description: RAG tasks must route to the rag queue.

Search Pattern:

grep -rn 'task_routes\|"-Q".*rag' rag/

Pass Criteria:

• task_routes configured for rag.*
• Worker consumes only rag queue
• No celery queue consumption

Fail Criteria:

• Missing task_routes
• Worker on default queue
• Mixed queue consumption

Severity: High

Recommendation: Configure task_routes and use -Q rag for worker.

Anti-Patterns

Missing Task Name

# BAD: Auto-generated name
@app.task
def ingest_file(project_dir, source_path):
    ...

# GOOD: Explicit name with rag prefix
@app.task(bind=True, name="rag.ingest_file")
def ingest_file_task(self, project_dir, source_path):
    ...

Non-Serializable Arguments

# BAD: Path object
@app.task(name="rag.process")
def process_task(file_path: Path):
    ...

# GOOD: String path
@app.task(name="rag.process")
def process_task(file_path: str):
    path = Path(file_path)
    ...

Silent Error Handling

# BAD: Swallowed exception
@app.task(name="rag.search")
def search_task(query):
    try:
        return search(query)
    except Exception:
        return []  # Silent failure!

# GOOD: Logged and re-raised
@app.task(bind=True, base=SearchTask, name="rag.search")
def search_task(self, query):
    try:
        return search(query)
    except Exception as e:
        logger.error(
            "Search failed",
            extra={"task_id": self.request.id, "error": str(e)},
        )
        raise  # Let Celery handle retry/failure

Wrong Pool Type

# BAD: Prefork causes ChromaDB SQLite errors
app.worker_main(argv=["worker", "-Q", "rag", "--pool=prefork"])

# GOOD: Thread pool for ChromaDB
app.worker_main(argv=["worker", "-Q", "rag", "--pool=threads"])

Task Return Patterns

Success with Details

return {
    "status": "success",
    "filename": filename,
    "document_count": len(documents),
    "stored_count": stored,
    "skipped_count": skipped,
}

Error with Context

return {
    "status": "error",
    "message": str(e),
    "filename": filename,
    "reason": "embedder_unavailable",
}

Tuple Pattern (Boolean + Details)

# For tasks that need simple success/fail check
return success, details_dict

ChromaDB Patterns for RAG

Best practices for vector storage, collection management, and similarity search using ChromaDB.

Architecture Overview

Documents with Embeddings -> ChromaStore -> Collection -> HNSW Index
                                              |
                                              v
                                    Persistent Storage

Key Classes

ChromaStore Implementation

Location: rag/components/stores/chroma_store/chroma_store.py

class ChromaStore(VectorStore):
    # Class-level client cache for thread safety
    _client_cache: dict[str, chromadb.ClientAPI] = {}
    _client_cache_lock = threading.Lock()
    _collection_setup_lock = threading.Lock()

    def __init__(
        self,
        name: str = "ChromaStore",
        config: dict[str, Any] | None = None,
        project_dir: Path | None = None,
    ):
        super().__init__(name, config, project_dir)
        config = config or {}

        self.collection_name = config.get("collection_name", "documents")
        self.host = config.get("host") or os.getenv("CHROMADB_HOST")
        self.port = config.get("port") or os.getenv("CHROMADB_PORT")
        self.distance_metric = config.get("distance_metric", "cosine")

        # Initialize client with caching
        if self.host and self.port:
            self.client = self._get_or_create_client(
                f"http://{self.host}:{self.port}",
                lambda: chromadb.HttpClient(host=self.host, port=self.port)
            )
        else:
            self.client = self._get_or_create_client(
                f"persistent://{self.persist_directory}",
                lambda: chromadb.PersistentClient(path=self.persist_directory)
            )

        self._setup_collection()

Thread-Safe Client Caching

@classmethod
def _get_or_create_client(
    cls, client_key: str, client_factory
) -> chromadb.ClientAPI:
    with cls._client_cache_lock:
        if client_key not in cls._client_cache:
            logger.info(f"Creating new ChromaDB client for: {client_key}")
            cls._client_cache[client_key] = client_factory()
        return cls._client_cache[client_key]

Distance Metrics

Metadata Constraints

ChromaDB only accepts these metadata types:

• str
• int
• float
• bool

NOT supported: list, dict, None

Metadata Cleaning Pattern

def _clean_metadata(self, metadata: dict[str, Any]) -> dict[str, Any]:
    cleaned = {}
    for key, value in metadata.items():
        if value is None:
            continue  # Skip None
        elif isinstance(value, (str, int, float, bool)):
            cleaned[key] = value
        elif isinstance(value, list):
            # Convert to comma-separated string
            cleaned[key] = ",".join(str(v) for v in value if v is not None)
        elif isinstance(value, dict):
            # Convert to JSON string
            cleaned[key] = json.dumps(value)
        else:
            cleaned[key] = str(value)
    return cleaned

Code Review Checklist

1. Thread-Safe Client Access

Description: ChromaDB clients must be shared safely across threads.

Search Pattern:

grep -rn "_client_cache\|_client_cache_lock\|threading.Lock" rag/components/stores/

Pass Criteria:

• Client cache uses class-level storage
• Lock protects cache access
• Single client per database path

Fail Criteria:

• New client created per request
• No locking on cache
• Multiple clients for same path

Severity: Critical

Recommendation: Use class-level client cache with threading.Lock.

---

2. Collection Setup Atomicity

Description: Collection creation should be atomic to prevent race conditions.

Search Pattern:

grep -rn "get_or_create_collection\|_collection_setup_lock" rag/components/stores/

Pass Criteria:

• Uses get_or_create_collection()
• Lock protects setup
• Metadata set during creation

Fail Criteria:

• Separate get + create calls
• No locking on setup
• Metadata set after creation

Severity: High

Recommendation: Use get_or_create_collection() within a lock.

---

3. Distance Metric Configuration

Description: Distance metric should be configurable and validated.

Search Pattern:

grep -rn "distance_metric\|hnsw:space" rag/components/stores/

Pass Criteria:

• Metric from config with default
• Validation against valid metrics
• Correct HNSW space mapping

Fail Criteria:

• Hardcoded metric
• No validation
• Wrong metric name mapping

Severity: Medium

Recommendation: Validate metric is one of: cosine, l2, ip.

---

4. Metadata Type Safety

Description: Metadata values must be ChromaDB-compatible types.

Search Pattern:

grep -rn "cleaned_metadata\|isinstance.*str.*int.*float.*bool" rag/components/stores/

Pass Criteria:

• All values converted to supported types
• None values skipped
• Lists/dicts serialized to strings

Fail Criteria:

• Raw metadata passed to ChromaDB
• None values included
• Complex types not converted

Severity: High

Recommendation: Clean all metadata before add_documents().

---

5. Similarity Score Conversion

Description: Distance scores must be converted to similarity scores correctly.

Search Pattern:

grep -rn "similarity_score\|distance.*score" rag/components/stores/

Pass Criteria:

• Distance-to-similarity conversion per metric
• Score added to document metadata
• Consistent score range (0-1)

Fail Criteria:

• Raw distances returned as scores
• Missing metric-specific conversion
• Inconsistent score interpretation

Severity: Medium

Recommendation: Use metric-appropriate conversion formulas.

---

6. Deduplication Support

Description: Duplicate detection should prevent redundant storage.

Search Pattern:

grep -rn "DeduplicationTracker\|document_hash\|chunk_hash\|_document_exists" rag/

Pass Criteria:

• Hash-based duplicate detection
• Check before add
• Configurable enable/disable

Fail Criteria:

• No deduplication
• Check after add
• Always enabled without config

Severity: Medium

Recommendation: Use content hash for deduplication with configurable toggle.

---

7. Error Handling in Search

Description: Search errors should be handled gracefully.

Search Pattern:

grep -rn "def search.*:$" -A 30 rag/components/stores/

Pass Criteria:

• Try/except around query
• Empty list on error
• Error logged with context

Fail Criteria:

• Exceptions propagate
• None returned on error
• Silent failures

Severity: High

Recommendation: Return empty list on search errors with logging.

Anti-Patterns

Creating Client Per Request

# BAD: New client every time
def search(self, query_embedding):
    client = chromadb.PersistentClient(path=self.persist_directory)
    collection = client.get_collection(self.collection_name)
    return collection.query(...)

# GOOD: Cached client
def search(self, query_embedding):
    return self.collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
    )

Passing Raw Metadata

# BAD: Raw metadata with unsupported types
self.collection.add(
    ids=[doc.id],
    embeddings=[doc.embeddings],
    metadatas=[doc.metadata],  # May contain lists, dicts, None!
)

# GOOD: Cleaned metadata
cleaned = self._clean_metadata(doc.metadata)
self.collection.add(
    ids=[doc.id],
    embeddings=[doc.embeddings],
    metadatas=[cleaned],
)

Ignoring Distance Metric

# BAD: Hardcoded score conversion
similarity = 1 - distance

# GOOD: Metric-aware conversion
if self.distance_metric == "cosine":
    similarity = 1.0 - (distance / 2.0)
elif self.distance_metric == "l2":
    similarity = 1.0 / (1.0 + distance / 100.0)
elif self.distance_metric == "ip":
    similarity = (1.0 + distance) / 2.0

Production Recommendations

1. Use HTTP Client in Production: PersistentClient has SQLite locking issues 2. Set Appropriate Collection Size: Monitor collection count 3. Regular Backups: ChromaDB persist_directory should be backed up 4. Monitor Embedding Dimensions: Ensure consistent dimensions across adds

LlamaIndex Patterns for RAG

Best practices for document parsing, chunking, and ingestion using LlamaIndex.

Architecture Overview

File Input -> Parser (LlamaIndex Reader) -> Chunking (Node Parser) -> Document Objects

Key Classes

LlamaIndexParser Base

Location: rag/components/parsers/base/llama_parser.py

from llama_index.core import Document as LlamaDocument
from llama_index.core.node_parser import (
    SentenceSplitter,
    TokenTextSplitter,
    MarkdownNodeParser,
    SemanticSplitterNodeParser,
)

class LlamaIndexParser(BaseParser):
    def __init__(self, config: dict[str, Any] = None):
        super().__init__(config)
        self.reader = None  # Set by subclass
        self.text_splitter = self._create_text_splitter()

    def _create_text_splitter(self):
        chunk_size = self.config.get("chunk_size", None)
        if chunk_size is None:
            return None  # No chunking

        chunk_overlap = self.config.get("chunk_overlap", 0)
        chunk_strategy = self.config.get("chunk_strategy", "characters")

        if chunk_strategy == "sentences":
            return SentenceSplitter(
                chunk_size=chunk_size,
                chunk_overlap=chunk_overlap,
            )
        elif chunk_strategy == "paragraphs":
            return MarkdownNodeParser()
        else:  # characters
            return TokenTextSplitter(
                chunk_size=chunk_size,
                chunk_overlap=chunk_overlap,
            )

Document Conversion

def _llama_to_rag_documents(self, llama_docs: list) -> list[Document]:
    rag_docs = []
    for llama_doc in llama_docs:
        content = llama_doc.text if hasattr(llama_doc, "text") else str(llama_doc)
        metadata = llama_doc.metadata if hasattr(llama_doc, "metadata") else {}

        # Add parser metadata
        metadata["parser_type"] = self.__class__.__name__

        doc = self.create_document(
            content=content,
            metadata=metadata,
            doc_id=llama_doc.id_ if hasattr(llama_doc, "id_") else None,
            source=metadata.get("file_path"),
        )
        rag_docs.append(doc)
    return rag_docs

Chunking Strategies

Note: Semantic chunking requires an embedder and falls back to sentence chunking if embedding fails.

Code Review Checklist

1. Chunking Configuration

Description: Verify chunking is properly configured with appropriate size and overlap.

Search Pattern:

grep -rn "chunk_size\|chunk_overlap\|SentenceSplitter\|TokenTextSplitter" rag/

Pass Criteria:

• chunk_size is specified (typically 512-2048)
• chunk_overlap is 10-20% of chunk_size
• Appropriate splitter for content type

Fail Criteria:

• chunk_size=0 or None without intentional full-document mode
• chunk_overlap >= chunk_size
• Using TokenTextSplitter for natural language

Severity: Medium

Recommendation: Use SentenceSplitter for natural language with chunk_overlap at least 50 tokens.

---

2. Metadata Preservation

Description: Ensure metadata flows from LlamaIndex documents to RAG documents.

Search Pattern:

grep -rn "_llama_to_rag_documents\|llama_doc.metadata" rag/components/parsers/

Pass Criteria:

• Metadata is copied from LlamaIndex document
• Parser type is added to metadata
• Source path is preserved

Fail Criteria:

• Metadata is discarded or overwritten
• No parser_type tracking
• Source lost during conversion

Severity: Medium

Recommendation: Always copy and extend metadata, never replace.

---

3. Chunk Metadata

Description: Each chunk should have metadata linking it to the parent document.

Search Pattern:

grep -rn "chunk_num\|chunk_index\|total_chunks" rag/

Pass Criteria:

• chunk_index (0-based) is set
• total_chunks is set
• Original document ID is preserved

Fail Criteria:

• No chunk indexing
• Missing total_chunks
• Lost parent reference

Severity: Medium

Recommendation: Include chunk_index, total_chunks, and parent_doc_id in chunk metadata.

---

4. LlamaIndex Import Guards

Description: LlamaIndex imports should be guarded for optional dependency.

Search Pattern:

grep -rn "LLAMA_INDEX_AVAILABLE\|ImportError.*llama" rag/

Pass Criteria:

• LlamaIndex imports wrapped in try/except
• LLAMA_INDEX_AVAILABLE flag checked before use
• Clear error message if not available

Fail Criteria:

• Bare imports without guards
• No fallback behavior
• Unclear import errors

Severity: Low

Recommendation: Use lazy imports with availability flags.

---

5. Parser Error Handling

Description: Parser errors should be captured and reported, not silently ignored.

Search Pattern:

grep -rn "ProcessingResult.*errors\|errors.append" rag/components/parsers/

Pass Criteria:

• Errors captured in ProcessingResult.errors
• Error includes source file path
• Error includes parser name

Fail Criteria:

• Exceptions swallowed silently
• Errors not included in result
• Missing context in error

Severity: High

Recommendation: Always return errors in ProcessingResult with full context.

---

6. File Type Detection

Description: Parsers should properly detect supported file types.

Search Pattern:

grep -rn "can_parse\|supported_extensions\|mime_types" rag/components/parsers/

Pass Criteria:

• can_parse() checks extension and/or mime type
• Supported extensions defined in metadata
• Magic library used for content-based detection

Fail Criteria:

• Only extension-based detection
• Hardcoded type checking
• No mime type support

Severity: Low

Recommendation: Use python-magic for content-based detection with extension fallback.

Anti-Patterns

Ignoring Chunking Errors

# BAD: Silent failure
try:
    chunks = self.text_splitter.split([llama_doc])
except:
    pass  # Chunks lost!

# GOOD: Fallback with logging
try:
    chunks = self.text_splitter.split([llama_doc])
except Exception as e:
    logger.warning(f"Chunking failed: {e}, using original document")
    chunked_docs.append(doc)

Losing Source Information

# BAD: No source tracking
doc = Document(content=text, metadata={})

# GOOD: Preserve source
doc = Document(
    content=text,
    metadata={"file_path": source_path, "parser": self.__class__.__name__},
    source=source_path,
)

Hardcoded Chunk Sizes

# BAD: Hardcoded values
splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)

# GOOD: Configurable
splitter = SentenceSplitter(
    chunk_size=self.config.get("chunk_size", 512),
    chunk_overlap=self.config.get("chunk_overlap", 50),
)

RAG Performance Patterns

Best practices for batching, caching, deduplication, and optimization in RAG systems.

Performance Overview

Batch Processing

Embedding Batching

Location: rag/core/base.py, rag/core/ingest_handler.py

class Embedder(Component):
    def process(self, documents: list[Document]) -> ProcessingResult:
        """Batch embed documents for efficiency."""
        # Extract all texts at once
        texts = [doc.content for doc in documents]

        # Single batch call to embedding API
        embeddings = self.embed(texts)

        # Assign embeddings back
        for doc, embedding in zip(documents, embeddings, strict=False):
            doc.embeddings = embedding

        return ProcessingResult(
            documents=documents,
            metrics={"embedded_count": len(documents)},
        )

Optimal Batch Sizes

Deduplication

Hash-Based Deduplication

Location: rag/utils/hash_utils.py, rag/components/stores/chroma_store/

class DeduplicationTracker:
    """Track document and chunk hashes to prevent duplicates."""

    def __init__(self):
        self._document_hashes: set[str] = set()
        self._chunk_hashes: set[str] = set()
        self._source_hashes: set[str] = set()

    def is_duplicate_document(self, doc_hash: str) -> bool:
        return doc_hash in self._document_hashes

    def is_duplicate_chunk(self, chunk_hash: str) -> bool:
        return chunk_hash in self._chunk_hashes

    def register_document(self, doc_hash: str, doc_id: str, source_hash: str):
        self._document_hashes.add(doc_hash)
        if source_hash:
            self._source_hashes.add(source_hash)

    def register_chunk(self, chunk_hash: str, doc_id: str):
        self._chunk_hashes.add(chunk_hash)

Document ID Generation

import hashlib

def generate_document_id(file_hash: str, chunk_index: int) -> str:
    """Generate deterministic ID from file hash and chunk index."""
    return f"{file_hash[:16]}_{chunk_index:04d}"

# In ingest_handler.py
file_hash = hashlib.sha256(file_data).hexdigest()
for i, doc in enumerate(documents):
    doc.id = generate_document_id(file_hash, i)
    doc.metadata["file_hash"] = file_hash
    doc.metadata["chunk_index"] = i

Circuit Breaker Pattern

Embedder Safety

Location: rag/utils/embedding_safety.py

class CircuitBreaker:
    """Prevent cascading failures from unavailable embedders."""

    def __init__(self, failure_threshold: int = 5, reset_timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failure_count = 0
        self.last_failure_time: float | None = None
        self.state = "closed"  # closed, open, half-open

    def can_execute(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open allows one attempt

    def record_success(self):
        self.failure_count = 0
        self.state = "closed"

    def record_failure(self, error: Exception | None = None):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"

Embedding Validation

def is_valid_embedding(
    embedding: list[float],
    expected_dimension: int | None = None,
    allow_zero: bool = False,
) -> tuple[bool, str | None]:
    """Validate embedding vector."""
    if not embedding:
        return False, "Empty embedding"

    if expected_dimension and len(embedding) != expected_dimension:
        return False, f"Wrong dimension: {len(embedding)} != {expected_dimension}"

    if not allow_zero and is_zero_vector(embedding):
        return False, "Zero vector"

    return True, None

def is_zero_vector(embedding: list[float], threshold: float = 1e-10) -> bool:
    """Check if embedding is effectively zero."""
    return all(abs(v) < threshold for v in embedding)

Code Review Checklist

1. Batch Embedding Processing

Description: Embeddings should be generated in batches, not one at a time.

Search Pattern:

grep -rn "for.*doc.*in.*documents.*embed\|embed\(\[doc" rag/

Pass Criteria:

• Texts extracted as list
• Single embed() call for batch
• Batch size is configurable

Fail Criteria:

• embed() called per document
• No batching logic
• Hardcoded batch size

Severity: High

Recommendation: Batch texts before calling embed().

---

2. Deduplication Enabled

Description: Duplicate detection should prevent redundant storage.

Search Pattern:

grep -rn "enable_deduplication\|DeduplicationTracker\|file_hash" rag/

Pass Criteria:

• Deduplication tracker initialized
• Hash checked before storage
• Configurable via settings

Fail Criteria:

• No deduplication
• Check after storage
• Always disabled

Severity: Medium

Recommendation: Enable deduplication by default with content hashing.

---

3. Circuit Breaker Configuration

Description: Embedders should have circuit breaker protection.

Search Pattern:

grep -rn "CircuitBreaker\|failure_threshold\|circuit_breaker" rag/

Pass Criteria:

• CircuitBreaker in embedder base
• Configurable threshold
• Proper state transitions

Fail Criteria:

• No circuit breaker
• Hardcoded thresholds
• Missing state management

Severity: High

Recommendation: Configure circuit breaker with 5 failures, 60s reset.

---

4. Embedding Validation

Description: Embeddings must be validated before storage.

Search Pattern:

grep -rn "is_valid_embedding\|is_zero_vector\|embedding.*validation" rag/

Pass Criteria:

• Validation before storage
• Zero vector rejection
• Dimension check

Fail Criteria:

• No validation
• Zero vectors accepted
• Wrong dimensions stored

Severity: High

Recommendation: Validate all embeddings with is_valid_embedding().

---

5. Deterministic Document IDs

Description: Document IDs should be deterministic for deduplication.

Search Pattern:

grep -rn "doc.id.*=.*hash\|file_hash.*chunk" rag/

Pass Criteria:

• ID derived from content hash
• Chunk index included
• Consistent generation

Fail Criteria:

• Random UUIDs always
• Missing chunk index
• Inconsistent format

Severity: Medium

Recommendation: Use {file_hash[:16]}_{chunk_index:04d} format.

---

6. Fail-Fast Configuration

Description: Embedder failures should stop processing early.

Search Pattern:

grep -rn "fail_fast\|EmbedderUnavailableError" rag/

Pass Criteria:

• fail_fast configurable
• Exception raised on failure
• Partial results returned

Fail Criteria:

• Silent failures
• Zero vectors substituted
• No error propagation

Severity: High

Recommendation: Enable fail_fast=True for production.

---

7. Search Result Limiting

Description: Search results should be properly limited.

Search Pattern:

grep -rn "top_k\|max_results\|n_results" rag/

Pass Criteria:

• top_k parameter respected
• Max limit configured
• Pagination supported

Fail Criteria:

• Unlimited results
• top_k ignored
• No pagination

Severity: Medium

Recommendation: Enforce max_results limit (e.g., 100).

Anti-Patterns

Per-Document Embedding

# BAD: N API calls for N documents
for doc in documents:
    embedding = embedder.embed([doc.content])[0]
    doc.embeddings = embedding

# GOOD: Single batch API call
texts = [doc.content for doc in documents]
embeddings = embedder.embed(texts)
for doc, emb in zip(documents, embeddings, strict=False):
    doc.embeddings = emb

Random Document IDs

# BAD: Always new UUID - no deduplication possible
doc.id = str(uuid.uuid4())

# GOOD: Deterministic ID from content
file_hash = hashlib.sha256(file_data).hexdigest()
doc.id = f"{file_hash[:16]}_{chunk_index:04d}"

Ignoring Circuit Breaker

# BAD: Keep trying despite failures
def embed(self, texts):
    for text in texts:
        try:
            return self._call_api(text)
        except:
            continue  # Never stops!

# GOOD: Circuit breaker stops cascade
def embed_text(self, text):
    self.check_circuit_breaker()  # May raise
    try:
        result = self._call_api(text)
        self.record_success()
        return result
    except Exception as e:
        self.record_failure(e)
        raise

Storing Invalid Embeddings

# BAD: Store whatever comes back
doc.embeddings = embedder.embed([doc.content])[0]
store.add_documents([doc])

# GOOD: Validate before storage
embedding = embedder.embed([doc.content])[0]
is_valid, error = is_valid_embedding(embedding, expected_dim)
if not is_valid:
    raise ValueError(f"Invalid embedding: {error}")
doc.embeddings = embedding
store.add_documents([doc])

Performance Metrics

Track these metrics for RAG performance:

Configuration Recommendations

# Production settings
EMBEDDER_CONFIG = {
    "batch_size": 32,
    "fail_fast": True,
    "circuit_breaker": {
        "failure_threshold": 5,
        "reset_timeout": 60.0,
    },
}

VECTOR_STORE_CONFIG = {
    "enable_deduplication": True,
    "distance_metric": "cosine",
    "max_results": 100,
}

RETRIEVAL_CONFIG = {
    "similarity_threshold": 0.5,
    "max_results": 100,
}