domains / data / rag-system-design / rag-architect

rag-architect

Skill card

combined

The two copies overlap on the RAG lifecycle (chunking, embeddings, vector stores, retrieval, evaluation) but each contains material the other lacks. Jeffallan/claude-skills provides a concrete, executable implementation path with Python examples, checkpoints, hybrid search, reranking, and evaluation code. LeoYeAI/openclaw-master-skills provides a broader reference catalog of chunking strategies, embedding model trade-offs, and vector database comparisons, but lacks the end-to-end implementation steps and validation checkpoints. Crowning one would hide either the practical pipeline or the selection guidance.

Repositories

Other copies

Combined skill

---
name: rag-architect
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.
---

# 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.

## Chunking Strategies

- **Fixed-Size**: Character or token-based splitting with 10-20% overlap. Best for uniform documents.
- **Sentence-Based**: Split at sentence boundaries, group until size threshold. Best for narrative text.
- **Paragraph-Based**: Respect document structure, merge small paragraphs, split large ones. Best for structured documents.
- **Semantic**: Use topic modeling or embedding similarity to detect topic shifts. Best for long-form content.
- **Recursive**: Try larger chunks first, recursively split if needed. Best for mixed content types.
- **Document-Aware**: Handle file-type-specific structures, preserve metadata, process tables and images. Best for multi-format collections.

## Embedding Model Selection

- **128-256 dimensions**: Fast retrieval, lower memory, simple domains.
- **512-768 dimensions**: Balanced performance, most applications.
- **1024-1536 dimensions**: High quality, complex domains, higher cost.
- **2048+ dimensions**: Maximum quality, specialized use cases.

Model categories: general purpose, code, scientific, multilingual.

## Vector Database Selection

- **Pinecone**: Managed service, auto-scaling, metadata filtering, hybrid search. Best for production managed deployments.
- **Weaviate**: Open source, GraphQL API, multi-modal search, HNSW indexing. Best for complex data types.
- **Qdrant**: Rust-based, high performance, payload filtering, distributed deployment. Best for high-performance requirements.
- **Chroma**: Embedded, SQLite-based, easy local development. Best for prototyping and small deployments.
- **pgvector**: SQL integration, ACID compliance, joins with relational data. Best when using PostgreSQL.

## Implementation Examples

### 1. Chunking Documents

```python
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
    separators=["\n\n", "\n", ". ", " "],
)

chunks = splitter.create_documents(
    texts=[doc.page_content for doc in raw_docs],
    metadatas=[{"source": doc.metadata["source"], "timestamp": doc.metadata.get("timestamp")} for doc in raw_docs],
)
```

**Checkpoint:** `assert all(c.metadata.get("source") for c in chunks), "Missing source metadata"`

### 2. Generating Embeddings & Indexing

```python
from openai import OpenAI
import qdrant_client
from qdrant_client.models import VectorParams, Distance, PointStruct

client = OpenAI()
qdrant = qdrant_client.QdrantClient("localhost", port=6333)

qdrant.recreate_collection(
    collection_name="knowledge_base",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

def embed_chunks(chunks: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    response = client.embeddings.create(input=chunks, model=model)
    return [r.embedding for r in response.data]

import hashlib, uuid

points = []
for i, chunk in enumerate(chunks):
    doc_id = str(uuid.UUID(hashlib.md5(chunk.page_content.encode()).hexdigest()))
    embedding = embed_chunks([chunk.page_content])[0]
    points.append(PointStruct(id=doc_id, vector=embedding, payload=chunk.metadata))

qdrant.upsert(collection_name="knowledge_base", points=points)
```

**Checkpoint:** `assert qdrant.count("knowledge_base").count == len(set(p.id for p in points)), "Deduplication failed"`

### 3. Hybrid Search (Vector + BM25)

```python
from qdrant_client.models import Filter, FieldCondition, MatchValue
from rank_bm25 import BM25Okapi

def hybrid_search(query: str, tenant_id: str, top_k: int = 20) -> list:
    query_embedding = embed_chunks([query])[0]
    tenant_filter = Filter(must=[FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id))])
    dense_results = qdrant.search(
        collection_name="knowledge_base",
        query_vector=query_embedding,
        query_filter=tenant_filter,
        limit=top_k,
    )

    corpus = [r.payload.get("text", "") for r in dense_results]
    bm25 = BM25Okapi([doc.split() for doc in corpus])
    bm25_scores = bm25.get_scores(query.split())

    ranked = sorted(
        zip(dense_results, bm25_scores),
        key=lambda x: 0.6 * x[0].score + 0.4 * x[1],
        reverse=True,
    )
    return [r for r, _ in ranked[:top_k]]
```

**Checkpoint:** `assert len(hybrid_search("test query", tenant_id="demo")) > 0, "Hybrid search returned no results"`

### 4. Reranking Top-K Results

Load provider API keys from environment variables or a secrets manager; never commit them to source code.

```python
import os
import cohere

co = cohere.Client(os.environ["COHERE_API_KEY"])

def rerank(query: str, results: list, top_n: int = 5) -> list:
    docs = [r.payload.get("text", "") for r in results]
    reranked = co.rerank(query=query, documents=docs, top_n=top_n, model="rerank-english-v3.0")
    return [results[r.index] for r in reranked.results]
```

### 5. Retrieval Evaluation

Run precision@k, recall@k, and MRR against a labeled evaluation set.

```python
from ragas import evaluate
from ragas.metrics import context_precision, context_recall, faithfulness, answer_relevancy
from datasets import Dataset

eval_dataset = Dataset.from_dict({
    "question": ["..."],
    "ground_truth": ["..."],
    "contexts": [["..."]],
    "answer": ["..."],
})

results = evaluate(
    dataset=eval_dataset,
    metrics=[context_precision, context_recall, faithfulness, answer_relevancy],
)
```

**Checkpoint:** Review evaluation results and iterate on chunking, embedding, or retrieval parameters as needed.