Building a "Hello World" RAG (Retrieval-Augmented Generation) system is trivial: you vectorize a few documents, store them, and attach the retrieved context to an LLM prompt. However, moving this to an enterprise environment introduces a set of complex challenges. How do you handle multi-gigabyte datasets without hitting context limits? More importantly, how do you ensure that a user from Marketing cannot "accidentally" query sensitive payroll data from HR?
In this article, I will dive into the architectural mindset required to build a production-grade RAG system that is high-performing, scalable, and secure by design.
The Two Pillars of Enterprise RAG
To move beyond experimental scripts, we must focus on two fundamental architectural pillars.
Pillar 1: Data Granularity & Retrieval Strategy
Sending entire documents to an LLM is a recipe for failure. Large inputs lead to the "Lost in the Middle" phenomenon, where models struggle to extract information from the center of a long prompt. They also increase costs and latency.
Data Granularity is about breaking down documents into precise, manageable units. By using intelligent chunking and metadata tagging, the system can perform a "surgical" retrieval. Instead of a 50-page PDF, we feed the LLM exactly 3–4 paragraphs containing the answer. This maximizes the Signal-to-Noise Ratio and ensures the model stays focused and factual.
Pillar 2: Data Encapsulation & Hybrid Access Control
In an enterprise, data access is never "all or nothing." We need a robust security layer. While many refer to this as RBAC (Role-Based Access Control), in a RAG system, we often implement ABAC (Attribute-Based Access Control).
We encapsulate data by tagging every vector with ownership attributes (e.g., department_id). Access control must be deterministic and enforced at the database level. The goal is simple: the LLM must never "see" data that the user is not explicitly cleared to access.
The Tech Stack: Performance & Precision
My implementation follows a modern, asynchronous architecture:
- FastAPI & SQLModel: Provides a high-performance web layer. I use SQLModel to bridge the gap between Python classes and our relational "Source of Truth" (PostgreSQL).
- Qdrant: A high-performance vector database. Its ability to handle complex payload filtering is essential for implementing security at the retrieval stage.
- LangChain: Used specifically for its robust document loaders and text-splitting utilities.
- PyJWT: Handles secure, stateless authentication, though critical authorization data is verified server-side.
Architectural Note: Maintain a strict separation between the relational database (PostgreSQL) and the vector store (Qdrant). Postgres manages user identities, access rights, and file metadata, while Qdrant handles high-dimensional search and context filtering.
Implementing Pillar 1: Precision Ingestion
The first step to granularity is a refined ingestion pipeline. Using RecursiveCharacterTextSplitter, we ensure that chunks respect paragraph boundaries, preventing loss of context.
loader = PyPDFLoader(location)
pages = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=150)
chunks = splitter.split_documents(pages)
When upserting to Qdrant, we don't just store vectors; we store a rich Payload. This includes a postgres_id to maintain referential integrity between our databases.
points = []
for i, chunk in enumerate(chunks):
vector = self.embeddings.embed_query(chunk.page_content)
points.append(rest.PointStruct(
id=str(uuid.uuid4()),
vector=vector,
payload={
"text": chunk.page_content,
"postgres_id": file_id,
"chunk_index": i,
**extra_data # This is where security attributes live (e.g., department_id)
}
))
self.client.upsert(collection_name="company_wiki", points=points)
Implementing Pillar 2: Securing the Context
Security in RAG isn't just about hiding UI elements; it's about deterministic Vector Filtering.
1. Verification at the Source
We extract the identity from the authenticated JWT token, but resolve the current department_id directly from the database. Relying on the relational store instead of stale claims inside the token allows for immediate revocation of access rights—a critical requirement for enterprise security.
def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
token_service: TokenService = Depends(get_token_service),
session: Session = Depends(db_connection)
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = token_service.decode_jwt(token)
user_id: str = payload.get("sub")
if user_id is None:
raise credentials_exception
except Exception:
raise credentials_exception
user = session.get(User, int(user_id))
if user is None:
raise credentials_exception
return user
CurrentUserDep = Annotated[User, Depends(get_current_user)]
2. Payload Indexing (Crucial for Scaling)
To ensure that filtering doesn't degrade retrieval latency, Qdrant must index security attributes. Without a payload index, the engine falls back to an expensive full-scan across stored payloads, causing query latency to spike under high vector volume:
client.create_payload_index(
collection_name="company_wiki",
field_name="department_id",
field_schema="keyword",
)
3. The Heart of the System: get_context
The final retrieval logic uses a Hybrid Filter. It computes semantic similarity over vector embeddings while strictly constraining the search space to documents matching the user's verified department_id or public documents (department_id: 0).
def get_context(self, query: str, user_department_id: int, limit: int = 3) -> str:
query_vector = self.model_embeddings.embed_query(query)
rbac_filter = Filter(
should=[
FieldCondition(key="department_id", match=MatchValue(value=user_department_id)),
FieldCondition(key="department_id", match=MatchValue(value=0)) # Public sources
]
)
vector_result = self.qdrant_client.query_points(
collection_name="company_wiki",
query=query_vector,
query_filter=rbac_filter,
limit=limit,
with_payload=True
)
context = ""
for hit in vector_result.points:
text = hit.payload.get("text", "No text")
context += f"\n---\n{text}\n"
return context
Summary
Building a production-grade RAG system is a transition from prompt engineering to data architecture.
By focusing on Data Granularity, we resolve performance bottlenecks and hallucination risks inherent in bloated context windows. Through Data Encapsulation and Payload Filtering, we establish a Zero-Trust architecture where authorization boundaries are enforced at the storage engine level—not delegated to the application prompt.
Moving AI into production is rarely a modeling challenge—it is an access control, data pipeline, and infrastructure challenge. If security isn't enforced deterministically at the storage engine level, your enterprise RAG is a liability waiting to happen.
The full reference implementation and deployment setup are available on GitHub: company-wiki-ai.
Top comments (0)