Enterprise RAG

Enterprise RAG vs Basic RAG

A demo RAG pipeline and a production enterprise RAG system are fundamentally different in scope. This comparison shows exactly where basic RAG falls short and what you need to add for real-world deployment.

Ingestion Pipeline

The ingestion pipeline is the foundation of your entire RAG system -- garbage in means garbage out at query time. Getting document processing, metadata extraction, and ACL mapping right during ingestion prevents every downstream problem.

Data Sources           Processing              Storage
┌──────────┐     ┌──────────────────┐     ┌─────────────┐
│ SharePoint│     │ Format Detection │     │ Vector DB   │
│ Confluence│────>│ Text Extraction  │────>│ (embeddings)│
│ Google Dr │     │ Chunking         │     ├─────────────┤
│ S3 / GCS  │     │ Metadata Extract │     │ Search Index│
│ Databases │     │ ACL Mapping      │     │ (BM25/FTS)  │
│ APIs      │     │ Embedding        │     ├─────────────┤
└──────────┘     │ Quality Filter   │     │ Metadata DB │
                  └──────────────────┘     │ (ACLs, etc) │
                                           └─────────────┘

Document Processing by Format

Metadata to Extract and Store

Access Control (ACL) Filtering

In enterprise environments, returning a document the user should not see is a security incident, not a UX bug. ACL filtering must be enforced at the retrieval layer, not as a post-processing step.

Pattern: Pre-filter at Query Time

def search_with_acl(query: str, user: User, top_k: int = 10):
    # Get user's access groups
    user_groups = get_user_groups(user.id)

    # Build metadata filter
    acl_filter = {
        "$or": [
            {"acl_groups": {"$in": user_groups}},
            {"acl_users": {"$in": [user.id]}},
            {"acl_groups": {"$in": ["public"]}}
        ]
    }

    # Vector search with filter
    results = vector_db.query(
        embedding=embed(query),
        filter=acl_filter,
        top_k=top_k
    )
    return results

ACL Sync Strategy

Recommendation: ACL cache with 5-15 minute sync interval for most enterprise use cases.

Hybrid Search

Enterprise queries range from exact error codes to broad conceptual questions -- no single search method handles both well. Hybrid search combines dense and sparse retrieval with metadata filtering to cover the full spectrum.

Architecture

Query
  │
  ├──> Dense Search (embedding similarity)  ──> Scores A
  │
  ├──> Sparse Search (BM25 / keyword)       ──> Scores B
  │
  └──> Metadata Filter (date, source, ACL)   ──> Scores C

  Fusion (RRF or weighted sum) ──> Reranker ──> Top K results

Reciprocal Rank Fusion

def rrf_fuse(result_lists: list[list], k: int = 60) -> list:
    """Fuse multiple ranked lists using Reciprocal Rank Fusion."""
    scores = {}
    for result_list in result_lists:
        for rank, doc in enumerate(result_list):
            scores[doc.id] = scores.get(doc.id, 0) + 1.0 / (k + rank + 1)

    sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_docs

# Usage
dense_results = vector_search(query, top_k=50)
sparse_results = bm25_search(query, top_k=50)
fused = rrf_fuse([dense_results, sparse_results])

Search Type Strengths

Reranking in Production

Production reranking is a multi-stage funnel that balances retrieval breadth against latency constraints. Over-fetch candidates cheaply, then progressively narrow with more expensive models.

Citation and Source Attribution

Enterprise users need to verify answers against source documents -- an answer without a citation is an answer nobody trusts. Enforce citation in your prompt template and validate that cited sources actually exist.

Citation Pattern

PROMPT_TEMPLATE = """Answer the question using ONLY the provided sources.
For every claim, cite the source using [Source N] format.
If the answer is not in the sources, say "I don't have information about that."

Sources:
{formatted_sources}

Question: {question}
Answer:"""

def format_sources(chunks):
    formatted = []
    for i, chunk in enumerate(chunks, 1):
        formatted.append(
            f"[Source {i}] ({chunk.title}, {chunk.source_type}, "
            f"updated {chunk.updated_at})\n{chunk.text}"
        )
    return "\n\n".join(formatted)

Citation Verification

def verify_citations(answer: str, sources: list[str]) -> dict:
    """Check that cited sources exist and claims are grounded."""
    cited = re.findall(r'\[Source (\d+)\]', answer)
    valid_range = set(str(i) for i in range(1, len(sources) + 1))

    invalid = [c for c in cited if c not in valid_range]
    uncited_sentences = [s for s in split_sentences(answer)
                         if not re.search(r'\[Source \d+\]', s)
                         and len(s.split()) > 5]

    return {
        "citations_found": len(cited),
        "invalid_citations": invalid,
        "uncited_claims": uncited_sentences,
        "all_valid": len(invalid) == 0
    }

Freshness Management

Stale data produces wrong answers -- and in enterprise contexts, outdated policy or compliance information can create real liability. Your freshness strategy must match how frequently your source data changes.

Freshness Boosting

def apply_freshness_boost(results, decay_factor=0.1):
    """Boost recent documents in search results."""
    now = datetime.utcnow()
    for result in results:
        age_days = (now - result.updated_at).days
        freshness_score = math.exp(-decay_factor * age_days)
        result.final_score = result.relevance_score * 0.8 + freshness_score * 0.2
    return sorted(results, key=lambda r: r.final_score, reverse=True)

Data Lineage

When an answer is wrong, you need to trace it back to the exact document, chunk strategy, and embedding model that produced it. Data lineage metadata makes debugging possible instead of guesswork.

Scaling Considerations

Your architecture must match your document scale -- what works for 10K documents will collapse at 10M. Plan your infrastructure tier based on where you are today and where you expect to be in 12 months.

Performance Optimization

Monitoring Metrics

Enterprise RAG systems degrade silently -- retrieval quality drops, indexes go stale, and ACL filters miss updates. These metrics are your early warning system for catching problems before users report them.

Common Pitfalls

Enterprise RAG failures tend to be more severe than basic RAG failures -- data leakage, compliance violations, and stale answers affecting business decisions. Audit your system against this list before going to production.