Hardening Internal Knowledge Systems: From Prototype RAG to Secure Production Deployment

Current Situation Analysis

Most organizations treat Retrieval-Augmented Generation (RAG) as a pure relevance problem. Teams optimize for benchmark scores, tune chunking strategies, and fine-tune embedding models, assuming that if the retrieval step returns the right documents, the system is production-ready. This approach fundamentally misunderstands how internal knowledge systems operate in regulated or multi-tenant environments.

The industry pain point is not retrieval quality; it is data sovereignty. When a RAG pipeline fetches documents before applying access controls, it creates a silent compliance vulnerability. The LLM receives restricted content in its context window, and even with system prompts instructing it to ignore sensitive data, model behavior remains probabilistic. Restricted document leakage becomes a statistical inevitability rather than an engineering guarantee.

This problem is overlooked because evaluation frameworks prioritize semantic similarity and answer relevance. Standard metrics like NDCG or MRR measure how well the system matches queries to documents, but they completely ignore whether the user should have seen those documents in the first place. Teams ship systems that score highly on relevance benchmarks while failing basic internal security audits.

Data from production deployments consistently shows that shifting access control upstream changes the entire risk profile. When role-based filtering occurs before retrieval scoring, systems can track exactly how many chunks were blocked per query, measure restricted leakage as a primary metric, and enforce citation-backed generation that only references authorized sources. The gap between a prototype and a production-ready system isn't about model size or embedding quality; it's about architectural ordering, identity resolution, and security-first evaluation.

WOW Moment: Key Findings

The transition from prototype to production RAG requires a fundamental shift in how systems are measured and architected. The following comparison highlights the structural differences between a relevance-optimized pipeline and a security-hardened deployment.

Dimension	Prototype Pipeline	Production-Ready Pipeline
Access Control Timing	Post-retrieval or LLM prompt-level	Pre-retrieval (query builder level)
Identity Source	Static API keys or manual role assignment	Dynamic OIDC/Entra ID token claims
Retrieval Strategy	Lexical or single-vector search	Hybrid (lexical + semantic) with fallback
Evaluation Focus	Relevance, MRR, NDCG	Leakage count, citation coverage, pass rate
Rate Limiting	In-memory, single-instance	Distributed (Redis/API Gateway)
Data Isolation	Single-tenant flat schema	Tenant-partitioned with PII classification

This finding matters because it redefines what "working" means for enterprise RAG. A system that returns highly relevant answers but leaks restricted HR or finance documents is a compliance liability, not a success. Pre-retrieval filtering combined with claim-based identity resolution transforms access control from a probabilistic suggestion into a deterministic guarantee. The evaluation shift from relevance to leakage measurement ensures that regression testing catches security violations before they reach users.

Core Solution

Building a production-grade internal knowledge system requires four architectural layers that operate in strict sequence: identity resolution, pre-retrieval filtering, citation-backed generation, and security-focused evaluation. Each layer must be independently testable and observable.

Step 1: Identity Resolution & Role Mapping

Production systems must derive user roles from authenticated identity tokens, not from static API key registrations. When a request arrives, the system validates the OIDC token, extracts group or role claims, and maps them to internal retrieval permissions. This prevents role elevation attacks and ensures that organizational changes (transfers, terminations) propagate immediately.

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer
from pydantic import BaseModel
from typing import Optional

class SecurityContext(BaseModel):
    user_id: str
    assigned_role: str
    tenant_id: Optional[str] = None

async def resolve_identity(token_header: str = Depends(HTTPBearer())) -> SecurityContext:
    # In production, validate JWT against Entra ID/OIDC provider
    # Extract claims and map to internal role namespace
    if not token_header.credentials:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing token")
    
    # Simulated claim extraction for architectural clarity
    claims = {"sub": "usr_8842", "groups": ["engineering", "finance_read"]}
    role_mapping = {"engineering": "eng_viewer", "finance_read": "fin_viewer"}
    
    matched_role = next((role for grp, role in role_mapping.items() if grp in claims["groups"]), "default_viewer")
    
    return SecurityContext(user_id=claims["sub"], assigned_role=matched_role)

Step 2: Pre-Retrieval Filtering Architecture

Access control must execute at the database query layer, before any vector or lexical scoring occurs. This ensures restricted chunks never enter the retrieval pipeline, eliminating context window pollution and guaranteeing deterministic filtering.

from sqlalchemy import select, and_
from sqlalchemy.ext.asyncio import AsyncSession

class RetrievalEngine:
    def __init__(self, db_session: AsyncSession):
        self.session = db_session

    async def fetch_authorized_chunks(self, query_vector: list[float], role: str, tenant: str) -> list[dict]:
        # Role constraint applied at query construction time
        base_query = select(ChunkDocument).where(
            and_(
                ChunkDocument.tenant_id == tenant,
                ChunkDocument.access_level.in_(self._resolve_allowed_levels(role))
            )
        )
        
        # Execute and apply semantic/lexical scoring only to authorized subset
        authorized_records = await self.session.execute(base_query)
        candidates = authorized_records.scalars().all()
        
        # Scoring happens exclusively on filtered set
        scored = self._rank_candidates(candidates, query_vector)
        return scored

    def _resolve_allowed_levels(self, role: str) -> list[str]:
        policy_map = {
            "eng_viewer": ["public", "engineering"],
            "fin_viewer": ["public", "finance"],
            "admin": ["public", "engineering", "finance", "restricted"]
        }
        return policy_map.get(role, ["public"])

Step 3: Citation-Backed Generation

Answers must be traceable to specific document IDs. The generation layer receives only authorized, scored chunks and returns structured output containing the response, source references, and metadata for audit trails.

class GenerationResponse(BaseModel):
    answer: str
    source_ids: list[str]
    confidence_score: float
    generation_metadata: dict

async def generate_response(prompt: str, context_chunks: list[dict]) -> GenerationResponse:
    # Format context with explicit source markers
    formatted_context = "\n".join(
        f"[DOC:{chunk['id']}] {chunk['content']}" for chunk in context_chunks
    )
    
    # Call Azure OpenAI or equivalent generation adapter
    # Production systems should enforce temperature=0 for deterministic citation mapping
    response = await llm_client.complete(
        prompt=f"Context:\n{formatted_context}\n\nQuestion: {prompt}",
        temperature=0.0,
        max_tokens=1024
    )
    
    # Extract cited document IDs from response for validation
    cited_ids = self._extract_citations(response.text)
    
    return GenerationResponse(
        answer=response.text,
        source_ids=cited_ids,
        confidence_score=response.logprobs,
        generation_metadata={"model": "azure-openai-gpt-4o", "chunk_count": len(context_chunks)}
    )

Step 4: Security-First Evaluation Pipeline

Evaluation must measure restricted document leakage, not just retrieval relevance. The runner executes live queries against the full pipeline, compares expected vs. retrieved document IDs, and calculates pass/fail rates based on access control compliance.

class EvaluationRunner:
    def __init__(self, pipeline: KnowledgePipeline, test_cases: list[TestCase]):
        self.pipeline = pipeline
        self.test_cases = test_cases

    async def execute_security_eval(self) -> EvalReport:
        results = []
        leakage_count = 0
        citation_coverage = 0
        
        for case in self.test_cases:
            response = await self.pipeline.resolve_query(case.query, case.user_role)
            
            # Check if any restricted documents leaked into response
            leaked = [doc_id for doc_id in response.source_ids if doc_id in case.restricted_docs]
            if leaked:
                leakage_count += len(leaked)
                
            # Verify all expected sources were cited
            covered = set(case.expected_sources).issubset(set(response.source_ids))
            if covered:
                citation_coverage += 1
                
            results.append(EvalResult(
                case_id=case.id,
                passed=len(leaked) == 0,
                leaked_docs=leaked,
                latency_ms=response.latency
            ))
            
        return EvalReport(
            total_cases=len(self.test_cases),
            pass_rate=(len(self.test_cases) - leakage_count) / len(self.test_cases),
            restricted_leakage_count=leakage_count,
            citation_coverage=citation_coverage / len(self.test_cases),
            avg_latency_ms=sum(r.latency_ms for r in results) / len(results)
        )

Architecture Rationale:

• Pre-retrieval filtering is non-negotiable. Moving RBAC upstream eliminates probabilistic LLM filtering and reduces token costs by preventing unauthorized chunks from entering the context window.
• SQLAlchemy abstraction allows seamless switching between SQLite (local development) and PostgreSQL/Azure Cosmos DB (production) without schema migrations.
• Citation tracking is enforced at the generation layer, not inferred post-hoc. This enables automated regression testing and audit compliance.
• Evaluation runs against the live pipeline, not mocked paths, ensuring that security guarantees hold under real query distribution and latency constraints.

Pitfall Guide

1. Post-Retrieval Access Filtering

Explanation: Applying role checks after vector search returns results. The LLM receives restricted documents in its context window, and prompt-based refusal is unreliable. Fix: Move authorization to the query builder. Filter at the database or search index level before scoring occurs. Use declarative access policies mapped to user claims.

2. Static API Key Role Binding

Explanation: Assigning roles during API key creation. When employees change teams or leave, keys remain active with outdated permissions until manually revoked. Fix: Derive roles dynamically from OIDC/Entra ID token claims. Map identity provider groups to internal retrieval roles at request time. Invalidate access immediately upon identity provider changes.

3. Lexical-Only Search in Production

Explanation: Relying solely on token overlap for retrieval. Queries without exact keyword matches return poor results, especially for internal jargon or paraphrased questions. Fix: Implement hybrid retrieval combining lexical and semantic search. Use Azure AI Search vector/hybrid modes with configurable weighting. Maintain lexical fallback for deterministic compliance queries.

4. In-Memory Rate Limiting

Explanation: Using application-level counters that don't share state across horizontally scaled instances. Leads to uneven throttling and potential DoS vulnerability. Fix: Deploy distributed rate limiting via Redis or API Gateway. Configure sliding window algorithms with tenant-aware quotas. Monitor limit exhaustion through operational metrics.

5. Relevance-Only Evaluation

Explanation: Optimizing exclusively for MRR/NDCG while ignoring security boundaries. Systems score highly on benchmarks but fail internal audits due to restricted document leakage. Fix: Implement leakage-aware evaluation. Track restricted document exposure, citation coverage, and pass rates. Integrate regression thresholds into CI pipelines to block deployments that increase leakage.

6. Ignoring PII Classification & Retention

Explanation: Ingesting raw internal documents without sensitivity labeling. Stored queries and generated answers retain PII indefinitely, violating retention policies. Fix: Add a classification step to the ingestion pipeline. Apply sensitivity labels, enforce field-level redaction, and configure explicit TTL policies for query logs and cached answers.

7. Single-Tenant Data Assumptions

Explanation: Building flat schemas without tenant partitioning. Multi-business-unit deployments risk cross-tenant data exposure through query leakage or misconfigured filters. Fix: Implement tenant isolation at the data model level. Include tenant_id in all queries, enforce partitioned indexes, and validate isolation through dedicated evaluation suites.

Production Bundle

Action Checklist

• Identity Integration: Configure AUTH_PROVIDER=entra and map OIDC group claims to internal retrieval roles
• Pre-Retrieval Filtering: Implement role-based constraints at the database/search index query layer
• Hybrid Search: Deploy vector + lexical retrieval with configurable weighting and fallback logic
• Distributed Throttling: Replace in-memory rate limits with Redis-backed or API Gateway enforcement
• PII Pipeline: Add content classification, sensitivity labeling, and explicit retention policies to ingestion
• Security Evaluation: Build leakage-aware test suites with regression thresholds integrated into CI
• Tenant Partitioning: Enforce tenant_id isolation across data models, queries, and audit logs
• Observability: Instrument Prometheus metrics, structured JSON logging, and audit trails for all administrative actions

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Single department, <10k documents	SQLite + Lexical + In-Memory Limits	Low operational overhead, deterministic behavior, fast local iteration	Minimal infrastructure cost
Multi-department, compliance required	PostgreSQL + Hybrid Search + Redis Throttling	Scalable, distributed state, audit-ready, supports tenant isolation	Moderate increase in managed service costs
Enterprise-wide, strict data sovereignty	Azure AI Search + Entra ID + Cosmos DB + Key Vault	Native cloud integration, enterprise SSO, automated secret rotation, geo-redundancy	Higher baseline cost, reduced compliance risk
High-volume query traffic	API Gateway Rate Limiting + Async Generation Queue	Prevents instance overload, decouples retrieval from LLM inference, enables backpressure	Infrastructure scaling cost offset by reduced timeout failures

Configuration Template

# docker-compose.yml (Production-Ready Base)
version: "3.9"
services:
  knowledge-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - AUTH_PROVIDER=entra
      - AZURE_TENANT_ID=${AZURE_TENANT_ID}
      - AZURE_CLIENT_ID=${AZURE_CLIENT_ID}
      - AZURE_CLIENT_SECRET=${AZURE_CLIENT_SECRET}
      - DB_DRIVER=postgresql+asyncpg
      - DB_HOST=${DB_HOST}
      - DB_PORT=5432
      - DB_NAME=knowledge_db
      - DB_USER=${DB_USER}
      - DB_PASSWORD=${DB_PASSWORD}
      - SEARCH_PROVIDER=azure_ai
      - AZURE_SEARCH_ENDPOINT=${AZURE_SEARCH_ENDPOINT}
      - AZURE_SEARCH_KEY=${AZURE_SEARCH_KEY}
      - GENERATION_PROVIDER=azure_openai
      - AZURE_OPENAI_ENDPOINT=${AZURE_OPENAI_ENDPOINT}
      - AZURE_OPENAI_KEY=${AZURE_OPENAI_KEY}
      - RATE_LIMIT_BACKEND=redis
      - REDIS_URL=redis://cache:6379/0
      - JSON_LOGS=true
      - SECURITY_HEADERS=true
    depends_on:
      - cache
      - db

  cache:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: redis-server --requirepass ${REDIS_PASSWORD}

  db:
    image: postgres:16-alpine
    environment:
      - POSTGRES_DB=knowledge_db
      - POSTGRES_USER=${DB_USER}
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

Quick Start Guide

1. Initialize Local Environment: Clone the repository, copy .env.example to .env, and set AUTH_PROVIDER=local for development. Run docker compose up -d to start the API, cache, and database containers.
2. Register Test Identity: Execute POST /auth/register with a test payload containing username, password, and role. The system returns an API key and stores a SHA-256 hash. Use this key for initial pipeline validation.
3. Ingest Sample Documents: Send markdown files with front-matter role metadata to POST /ingest. The pipeline parses chunks, applies access tags, and stores them in the local SQLite/PostgreSQL instance. Verify chunk counts via the metrics endpoint.
4. Execute Security Evaluation: Trigger POST /eval/run to run the leakage-aware test suite. Review the report for pass rate, restricted leakage count, citation coverage, and average latency. Adjust role mappings or retrieval weights if leakage exceeds threshold.
5. Switch to Production Config: Update environment variables to point to Azure services (AZURE_* endpoints, Entra ID tenant, Redis cluster). Remove AUTH_PROVIDER=local and set AUTH_PROVIDER=entra. Restart containers; the system fails fast if required cloud credentials are missing, preventing silent degradation.