Workflow-003: RAG System Development
FreshSource: Building with the Claude API
Document Control
| Field | Value |
|---|---|
| Workflow ID | WF-003 |
| Version | 1.0 |
| Status | Active |
| Last Updated | 2024-12 |
Overview
Build production-ready Retrieval Augmented Generation systems.
RAG Architecture
Phase 1: Chunking
| Strategy | Description | Trade-offs |
|---|---|---|
| Fixed Size | Split by character/token count, Add overlap for context | Simple but may break semantics |
| Semantic | Split by paragraphs/sections, Preserves meaning | Variable chunk sizes |
| Hierarchical | Parent-child relationships, Search child retrieve parent | Best context preservation |
python
def chunk_text(text, chunk_size=500, overlap=50):
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
chunks.append(chunk)
start = end - overlap
return chunksPhase 2: Embedding
python
import voyageai
client = voyageai.Client()
def embed_texts(texts):
result = client.embed(
texts=texts,
model="voyage-3",
input_type="document"
)
return result.embeddings
def embed_query(query):
result = client.embed(
texts=[query],
model="voyage-3",
input_type="query"
)
return result.embeddings[0]Phase 3: Hybrid Search
python
from rank_bm25 import BM25Okapi
def hybrid_search(query, chunks, embeddings, k=10):
# Semantic search
query_embedding = embed_query(query)
semantic_scores = cosine_similarity([query_embedding], embeddings)[0]
# BM25 search
tokenized_chunks = [c.split() for c in chunks]
bm25 = BM25Okapi(tokenized_chunks)
bm25_scores = bm25.get_scores(query.split())
# Reciprocal rank fusion
combined = reciprocal_rank_fusion(semantic_scores, bm25_scores)
return get_top_k(combined, chunks, k)Phase 4: Reranking
python
def rerank(query, documents, top_k=5):
# Use Claude or dedicated reranker
rerank_prompt = f"""
Query: {query}
Rate each document 1-10 for relevance:
{format_documents(documents)}
"""
# Get scores and sort
scores = get_relevance_scores(rerank_prompt)
return sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)[:top_k]Phase 5: Augment & Generate
python
def generate_response(query, context_chunks):
context = "\n\n".join(context_chunks)
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
system="""Answer based on the provided context.
If the context doesn't contain the answer, say so.""",
messages=[
{
"role": "user",
"content": f"Context:\n{context}\n\nQuestion: {query}"
}
]
)
return response.content[0].textVerification Checklist
- [ ] Documents chunked appropriately
- [ ] Embeddings generated and stored
- [ ] Hybrid search implemented
- [ ] Reranking improves relevance
- [ ] Context fits within token limits
- [ ] Responses cite sources