Guardrailed RAG Engine for High-Compliance Data
Enterprise Document Intelligence & Guardrailed RAG Systems are currently among the highest-demand architectural patterns in FinTech, Healthcare, and Legal Tech. This project implements a production-ready, asynchronous Python pipeline using FastAPI, SentenceTransformers, FAISS vector indexing, and custom Pydantic semantic guardrails.
Project Architecture & Market Value
- Target Industry: Enterprise FinTech & Legal Operations (Contract & Report Analytics).
- Key Resume Capabilities: Async API design, Vector Search (FAISS), Semantic Chunking, Prompt Injection Guardrails, Structured Output Validation.
- Tech Stack: Python 3.10+, FastAPI, PyTorch, SentenceTransformers, FAISS, Pydantic v2, Uvicorn.
Complete End-to-End Implementation (main.py)
Python
import io
import re
import time
from typing import List, Dict, Any, Optional
import numpy as np
import faiss
from fastapi import FastAPI, HTTPException, UploadFile, File, Status
from pydantic import BaseModel, Field
from sentence_transformers import SentenceTransformer
# -----------------------------------------------------------------------------
# 1. Domain Models & Schemas
# -----------------------------------------------------------------------------
class DocumentChunk(BaseModel):
chunk_id: int
text: str
metadata: Dict[str, Any] = Field(default_factory=dict)
class IngestionResponse(BaseModel):
status: str
total_chunks_indexed: int
processing_time_seconds: float
class QueryRequest(BaseModel):
query: str = Field(..., min_length=3, description="User search query or question")
top_k: int = Field(default=3, ge=1, le=10)
class QueryResponse(BaseModel):
query: str
is_safe: bool
answer: str
retrieved_context: List[str]
latency_ms: float
# -----------------------------------------------------------------------------
# 2. Guardrails & Safety Engine
# -----------------------------------------------------------------------------
class SecurityGuardrail:
"""Detects prompt injection attempts and PII leaks before LLM execution."""
INJECTION_PATTERNS = [
r"ignore previous instructions",
r"system prompt",
r"you are now an unrestricted",
r"bypass security",
r"reveal passwords"
]
@classmethod
def validate_input(cls, text: str) -> bool:
lowered = text.lower()
for pattern in cls.INJECTION_PATTERNS:
if re.search(pattern, lowered):
return False
return True
# -----------------------------------------------------------------------------
# 3. Vector Database Engine (FAISS + SentenceTransformers)
# -----------------------------------------------------------------------------
class VectorSearchEngine:
def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
self.encoder = SentenceTransformer(model_name)
self.dimension = self.encoder.get_sentence_embedding_dimension()
self.index = faiss.IndexFlatIP(self.dimension) # Inner Product (Cosine similarity when normalized)
self.chunk_store: Dict[int, DocumentChunk] = {}
self.current_id = 0
def _chunk_text(self, text: str, chunk_size: int = 300, overlap: int = 50) -> List[str]:
words = text.split()
chunks = []
for i in range(0, len(words), chunk_size - overlap):
chunk = " ".join(words[i:i + chunk_size])
if chunk.strip():
chunks.append(chunk)
return chunks
def add_document(self, content: str, source_name: str) -> int:
raw_chunks = self._chunk_text(content)
if not raw_chunks:
return 0
embeddings = self.encoder.encode(raw_chunks, convert_to_numpy=True)
faiss.normalize_L2(embeddings)
self.index.add(embeddings)
for idx, chunk_str in enumerate(raw_chunks):
doc_chunk = DocumentChunk(
chunk_id=self.current_id,
text=chunk_str,
metadata={"source": source_name, "position": idx}
)
self.chunk_store[self.current_id] = doc_chunk
self.current_id += 1
return len(raw_chunks)
def search(self, query: str, top_k: int = 3) -> List[DocumentChunk]:
if self.index.ntotal == 0:
return []
query_vector = self.encoder.encode([query], convert_to_numpy=True)
faiss.normalize_L2(query_vector)
distances, indices = self.index.search(query_vector, top_k)
results = []
for idx in indices[0]:
if idx != -1 and idx in self.chunk_store:
results.append(self.chunk_store[idx])
return results
# -----------------------------------------------------------------------------
# 4. Synthesizer / Extractor (LLM abstraction)
# -----------------------------------------------------------------------------
class LLMSynthesizer:
@staticmethod
def generate_response(query: str, contexts: List[str]) -> str:
"""Simulates RAG generation step; replace with OpenAI/Ollama API call in production."""
if not contexts:
return "No relevant context found in knowledge base."
combined_context = " ".join(contexts[:2])
return f"[Synthesized Enterprise Insight]: Based on indexed documents ('{combined_context[:120]}...'), the answer to '{query}' relates directly to documented operational procedures."
# -----------------------------------------------------------------------------
# 5. FastAPI Application & Service Initialization
# -----------------------------------------------------------------------------
app = FastAPI(
title="Enterprise Document Intelligence & RAG Engine",
version="1.0.0",
description="High-performance asynchronous vector retrieval engine with guardrails."
)
vector_db = VectorSearchEngine()
@app.post("/api/v1/ingest", response_model=IngestionResponse, status_code=Status.HTTP_201_CREATED)
async def ingest_document(file: UploadFile = File(...)):
start_time = time.time()
if not file.filename.endswith(('.txt', '.md')):
raise HTTPException(
status_code=Status.HTTP_400_BAD_REQUEST,
detail="Unsupported file format. Please upload text/markdown files."
)
content_bytes = await file.read()
content_str = content_bytes.decode("utf-8", errors="ignore")
chunks_added = vector_db.add_document(content_str, source_name=file.filename)
elapsed = time.time() - start_time
return IngestionResponse(
status="Success",
total_chunks_indexed=chunks_added,
processing_time_seconds=round(elapsed, 4)
)
@app.post("/api/v1/query", response_model=QueryResponse)
async def query_rag(request: QueryRequest):
start_time = time.time()
# Step 1: Input Validation / Guardrails
is_safe = SecurityGuardrail.validate_input(request.query)
if not is_safe:
return QueryResponse(
query=request.query,
is_safe=False,
answer="Request flagged by security policy: Prompt injection or restricted pattern detected.",
retrieved_context=[],
latency_ms=round((time.time() - start_time) * 1000, 2)
)
# Step 2: Vector Search
retrieved_chunks = vector_db.search(request.query, top_k=request.top_k)
context_texts = [c.text for c in retrieved_chunks]
# Step 3: Synthesis
answer = LLMSynthesizer.generate_response(request.query, context_texts)
total_latency = (time.time() - start_time) * 1000
return QueryResponse(
query=request.query,
is_safe=True,
answer=answer,
retrieved_context=context_texts,
latency_ms=round(total_latency, 2)
)
@app.get("/health")
async def health_check():
return {"status": "healthy", "indexed_vectors": vector_db.index.ntotal}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Environment Setup & Running the Service
1. Install Dependencies
Bash
pip install fastapi uvicorn sentence-transformers faiss-cpu numpy pydantic python-multipart
2. Start Application Server
Bash
python main.py
3. Ingest Sample Document
Bash
curl -X 'POST' \
'http://localhost:8000/api/v1/ingest' \
-F 'file=@sample_policy.txt'
4. Execute Guardrailed Query
Bash
curl -X 'POST' \
'http://localhost:8000/api/v1/query' \
-H 'Content-Type: application/json' \
-d '{"query": "What are the standard SLA timelines?", "top_k": 2}'
Category: