RAG Pipeline Architecture with Claude for Enterprise Knowledge

Retrieval-augmented generation solves the fundamental problem of LLMs: they can't know what's in your proprietary documents. RAG gives Claude access to your organization's knowledge without fine-tuning. But the difference between a demo RAG system and a production one is vast — most failures happen in retrieval, not generation. Here's the architecture that actually works at scale.

Why naive RAG fails

The typical RAG demo: chunk documents, embed them, retrieve top-K by cosine similarity, stuff them into the prompt. It works on clean, well-structured documents. It falls apart on:

• Documents with tables, images, and complex formatting
• Queries that require synthesizing information across multiple sections
• Ambiguous queries where the "right" chunk isn't obvious
• Large knowledge bases where top-K retrieval has low precision

In our enterprise deployment, naive RAG achieved 62% answer accuracy. After implementing the architecture below, we hit 91%.

Architecture: production RAG with Claude

Stage 1: Intelligent document processing

Chunking is where most RAG systems go wrong. Fixed-size chunks (512 tokens) are simple but destroy context boundaries. A better approach:

from dataclasses import dataclass
from typing import Optional
import re

@dataclass
class DocumentChunk:
    content: str
    metadata: dict
    chunk_id: str
    parent_id: Optional[str]  # For hierarchical retrieval
    token_count: int
    source_page: Optional[int]
    section_title: Optional[str]

class SemanticChunker:
    """Chunk documents by semantic boundaries, not arbitrary token counts."""

    def __init__(self, max_chunk_tokens: int = 800, overlap_tokens: int = 100):
        self.max_chunk_tokens = max_chunk_tokens
        self.overlap_tokens = overlap_tokens

    def chunk_document(self, text: str, metadata: dict) -> list[DocumentChunk]:
        """Split document at semantic boundaries."""
        # First pass: split by section headers
        sections = self._split_by_headers(text)

        chunks = []
        for section_idx, section in enumerate(sections):
            section_title = section.get("title", "")
            section_content = section["content"]

            # Second pass: split large sections by paragraph boundaries
            if self._estimate_tokens(section_content) > self.max_chunk_tokens:
                sub_chunks = self._split_by_paragraphs(section_content)
            else:
                sub_chunks = [section_content]

            for chunk_idx, chunk_text in enumerate(sub_chunks):
                chunk_id = f"{metadata.get('doc_id', 'unknown')}_{section_idx:03d}_{chunk_idx:03d}"
                chunks.append(DocumentChunk(
                    content=chunk_text,
                    metadata={
                        **metadata,
                        "section_title": section_title,
                        "chunk_position": f"{section_idx}.{chunk_idx}",
                    },
                    chunk_id=chunk_id,
                    parent_id=metadata.get("doc_id"),
                    token_count=self._estimate_tokens(chunk_text),
                    source_page=section.get("page"),
                    section_title=section_title,
                ))

        return chunks

    def _split_by_headers(self, text: str) -> list[dict]:
        """Split text by markdown-style headers."""
        header_pattern = r'^(#{1,4})\s+(.+)$'
        sections = []
        current_section = {"title": "", "content": ""}

        for line in text.split("\n"):
            match = re.match(header_pattern, line)
            if match:
                if current_section["content"].strip():
                    sections.append(current_section)
                current_section = {"title": match.group(2), "content": ""}
            else:
                current_section["content"] += line + "\n"

        if current_section["content"].strip():
            sections.append(current_section)

        return sections

    def _split_by_paragraphs(self, text: str) -> list[str]:
        """Split by paragraph boundaries, respecting max chunk size."""
        paragraphs = text.split("\n\n")
        chunks = []
        current_chunk = ""

        for para in paragraphs:
            if self._estimate_tokens(current_chunk + para) > self.max_chunk_tokens:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = para
            else:
                current_chunk += "\n\n" + para

        if current_chunk.strip():
            chunks.append(current_chunk.strip())

        return chunks

    def _estimate_tokens(self, text: str) -> int:
        return len(text) // 4  # Rough approximation

Stage 2: Hybrid retrieval

Vector similarity alone isn't enough. Combine it with keyword search for robust retrieval:

interface RetrievalResult {
  chunkId: string;
  content: string;
  score: number;
  source: "vector" | "keyword" | "both";
  metadata: Record<string, unknown>;
}

class HybridRetriever {
  private vectorStore: VectorStore;
  private keywordIndex: KeywordIndex;

  constructor(vectorStore: VectorStore, keywordIndex: KeywordIndex) {
    this.vectorStore = vectorStore;
    this.keywordIndex = keywordIndex;
  }

  async retrieve(
    query: string,
    topK: number = 10,
    vectorWeight: number = 0.7
  ): Promise<RetrievalResult[]> {
    // Run both retrieval strategies in parallel
    const [vectorResults, keywordResults] = await Promise.all([
      this.vectorStore.search(query, topK * 2),
      this.keywordIndex.search(query, topK * 2),
    ]);

    // Reciprocal Rank Fusion to merge results
    const merged = this.reciprocalRankFusion(
      vectorResults,
      keywordResults,
      vectorWeight
    );

    // Rerank with cross-encoder for final ordering
    const reranked = await this.rerank(query, merged.slice(0, topK * 2));

    return reranked.slice(0, topK);
  }

  private reciprocalRankFusion(
    vectorResults: SearchResult[],
    keywordResults: SearchResult[],
    vectorWeight: number
  ): RetrievalResult[] {
    const scores: Map<string, { score: number; source: string }> = new Map();
    const k = 60; // RRF constant

    // Score vector results
    vectorResults.forEach((result, rank) => {
      const rrfScore = vectorWeight / (k + rank + 1);
      const existing = scores.get(result.id) || { score: 0, source: "" };
      scores.set(result.id, {
        score: existing.score + rrfScore,
        source: existing.source ? "both" : "vector",
      });
    });

    // Score keyword results
    keywordResults.forEach((result, rank) => {
      const rrfScore = (1 - vectorWeight) / (k + rank + 1);
      const existing = scores.get(result.id) || { score: 0, source: "" };
      scores.set(result.id, {
        score: existing.score + rrfScore,
        source: existing.source ? "both" : "keyword",
      });
    });

    // Sort by combined score
    return Array.from(scores.entries())
      .sort(([, a], [, b]) => b.score - a.score)
      .map(([id, { score, source }]) => ({
        chunkId: id,
        content: this.getContent(id),
        score,
        source: source as "vector" | "keyword" | "both",
        metadata: this.getMetadata(id),
      }));
  }

  private async rerank(
    query: string,
    candidates: RetrievalResult[]
  ): Promise<RetrievalResult[]> {
    // Use a cross-encoder model for precise reranking
    // This catches cases where embedding similarity misses semantic relevance
    // Implementation depends on your reranking service
    return candidates; // Placeholder
  }

  private getContent(id: string): string { return ""; }
  private getMetadata(id: string): Record<string, unknown> { return {}; }
}

Stage 3: Answer generation with grounding

The generation prompt must enforce grounding — Claude should only use information from the retrieved context:

Answer the user's question based ONLY on the provided context documents.

Rules:
1. Only use information explicitly stated in the context documents
2. If the context doesn't contain enough information to answer fully, say so
3. Cite your sources using [Doc X] notation
4. Never invent information not present in the documents
5. If documents contain conflicting information, acknowledge the conflict

Context documents:
{retrieved_chunks}

User question: {query}

Production quality metrics

Metric	Naive RAG	Production RAG	Improvement
Answer accuracy	62%	91%	+47%
Hallucination rate	18%	2.3%	-87%
Retrieval precision@10	0.45	0.78	+73%
Source attribution accuracy	71%	96%	+35%
User satisfaction (CSAT)	3.2/5	4.4/5	+38%

The biggest gains came from: semantic chunking (+12% accuracy), hybrid retrieval (+9%), reranking (+6%), and grounded generation prompting (+2%).

Failure modes and mitigations

1. No relevant documents found — return "I don't have information about that" rather than hallucinating. Set a minimum relevance threshold (we use 0.65).
2. Retrieved context contradicts itself — instruct Claude to surface the contradiction to the user with citations to both sources.
3. Query is ambiguous — use a classification step to determine if the query needs clarification before retrieval.
4. Stale documents — implement document freshness scoring and bias retrieval toward recently-updated content.

Key takeaways

1. Retrieval quality determines RAG quality. Invest 70% of your effort in chunking and retrieval, 30% in generation.
2. Semantic chunking beats fixed-size. Respect document structure — headers, paragraphs, sections are natural boundaries.
3. Hybrid retrieval is non-negotiable. Vector search misses keyword-heavy queries; BM25 misses semantic similarity. Use both.
4. Enforce grounding explicitly. Tell Claude to only use the provided context and cite sources. Measure hallucination rate continuously.
5. Build feedback loops. Track which queries get low-confidence answers and use them to improve your chunking and retrieval.

RAG is an information retrieval problem more than a generation problem. Get retrieval right, and Claude handles the rest.