リポジトリの原文
見出し、例、コード、表、リンク、参照画像を含む原文を表示しています。
RAG Architect
Core Workflow
- Requirements Analysis — Identify retrieval needs, latency constraints, accuracy requirements, and scale
- Vector Store Design — Select database, schema design, indexing strategy, sharding approach
- Chunking Strategy — Document splitting, overlap, semantic boundaries, metadata enrichment
- Retrieval Pipeline — Embedding selection, query transformation, hybrid search, reranking
- Evaluation & Iteration — Metrics tracking, retrieval debugging, continuous optimization
For each step, validate before moving on (see checkpoints below).
Reference Guide
Load detailed guidance based on context:
| Topic | Reference | Load When |
|---|---|---|
| Vector Databases | references/vector-databases.md | Comparing Pinecone, Weaviate, Chroma, pgvector, Qdrant |
| Embedding Models | references/embedding-models.md | Selecting embeddings, fine-tuning, dimension trade-offs |
| Chunking Strategies | references/chunking-strategies.md | Document splitting, overlap, semantic chunking |
| Retrieval Optimization | references/retrieval-optimization.md | Hybrid search, reranking, query expansion, filtering |
| RAG Evaluation | references/rag-evaluation.md | Metrics, evaluation frameworks, debugging retrieval |
Implementation Examples
1. Chunking Documents
python
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Evaluate chunk_size on your domain data — never use 512 blindly
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)
# Create collection
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]
# Idempotent upsert with deduplication via deterministic IDs
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, SparseVector
from rank_bm25 import BM25Okapi
def hybrid_search(query: str, tenant_id: str, top_k: int = 20) -> list:
# Dense retrieval
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,
)
# Sparse retrieval (BM25)
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())
# Reciprocal Rank Fusion
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
python
# Run precision@k and recall@k against a labeled evaluation set
# python evaluate.py --metrics precision@10 recall@10 mrr --collection knowledge_base
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": questions,
"contexts": retrieved_contexts,
"answer": generated_answers,
"ground_truth": ground_truth_answers,
})
results = evaluate(eval_dataset, metrics=[context_precision, context_recall, faithfulness, answer_relevancy])
print(results)Checkpoint: Target context_precision >= 0.7 and context_recall >= 0.6 before moving to LLM integration.
Constraints
MUST DO
- Evaluate multiple embedding models on your domain data before committing
- Implement hybrid search (vector + keyword) for production systems
- Add metadata filters for multi-tenant or domain-specific retrieval
- Measure retrieval metrics (precision@k, recall@k, MRR, NDCG)
- Use reranking for top-k results before passing context to LLM
- Implement idempotent ingestion with deduplication (deterministic IDs)
- Monitor retrieval latency and quality over time
- Version embeddings and plan for model migration
MUST NOT DO
- Use default chunk size (512) without evaluation on your domain data
- Skip metadata enrichment (source, timestamp, section)
- Ignore retrieval quality metrics in favor of only LLM output quality
- Store raw documents without preprocessing/cleaning
- Use cosine similarity alone for complex multi-domain retrieval
- Deploy without testing on production-like data volumes
- Forget to handle edge cases (empty results, malformed docs)
- Couple the embedding model tightly to application code
Output Templates
When designing RAG architecture, deliver:
- System architecture diagram (ingestion + retrieval pipelines)
- Vector database selection with trade-off analysis
- Chunking strategy with examples and rationale
- Retrieval pipeline design (query → results flow)
- Evaluation plan with metrics, benchmarks, and pass/fail thresholds
Maintained by @jeffallan, Principal Consultant at Synergetic Solutions
