# Enterprise RAG Pipeline & Vector Retrieval OS ==================================================================== MODULE 1: DOCUMENT PARSING, CONTEXTUAL CHUNKING & METADATA SCHEMAS ==================================================================== - Ingestion Taxonomy & Document Parsing Strategy: * Corpus: 10,000+ SEC 10-K, 10-Q filings and internal banking credit risk policies. * Parsing Pipeline: Ingestion via LlamaParse layout-aware PDF extraction. Financial balance sheets and cash flow statements are converted into clean GitHub-flavored Markdown tables to preserve row-column entity relationships. * Hierarchy: Parent Document (doc_id, filing date, ticker) -> Sub-Section (Item 1A: Risk Factors) -> Semantic Leaf Chunks. - Anthropic-Style Contextual Chunking Architecture: * Window Size: 512 tokens with 64-token sliding overlap. * Pre-Embedding Context Prompt: Prior to vectorization, an LLM call prepends a 50-word global summary to each chunk: "This chunk is from the 2025 SEC Form 10-K of Vanguard Regional Bank, Section 7 (Management Discussion), discussing liquidity risk management and commercial real estate loan provisions." - Production PostgreSQL + pgvector DDL Schema (schema.sql): ```sql CREATE EXTENSION IF NOT EXISTS vector; CREATE EXTENSION IF NOT EXISTS pg_trgm; CREATE TABLE documents ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), title VARCHAR(255) NOT NULL, ticker VARCHAR(10), filing_type VARCHAR(20), filing_date DATE NOT NULL, source_url TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE TABLE document_chunks ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE, chunk_index INTEGER NOT NULL, content TEXT NOT NULL, context_prefix TEXT NOT NULL, full_chunk_text TEXT GENERATED ALWAYS AS (context_prefix || E'\n\n' || content) STORED, embedding vector(1536), tsv_content tsvector GENERATED ALWAYS AS (to_tsvector('english', context_prefix || ' ' || content)) STORED, metadata JSONB NOT NULL DEFAULT '{}'::jsonb, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX idx_chunks_embedding_hnsw ON document_chunks USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64); CREATE INDEX idx_chunks_tsv ON document_chunks USING gin (tsv_content); CREATE INDEX idx_chunks_metadata ON document_chunks USING gin (metadata); WITH dense_search AS ( SELECT id, ROW_NUMBER() OVER (ORDER BY embedding <=> $1) AS rank FROM document_chunks WHERE metadata->>'ticker' = $2 LIMIT 50 ), sparse_search AS ( SELECT id, ROW_NUMBER() OVER (ORDER BY ts_rank_cd(tsv_content, plainto_tsquery('english', $3)) DESC) AS rank FROM document_chunks WHERE tsv_content @@ plainto_tsquery('english', $3) AND metadata->>'ticker' = $2 LIMIT 50 ) SELECT COALESCE(d.id, s.id) AS chunk_id, (COALESCE(1.0 / (60 + d.rank), 0.0) + COALESCE(1.0 / (60 + s.rank), 0.0)) AS rrf_score FROM dense_search d FULL OUTER JOIN sparse_search s ON d.id = s.id ORDER BY rrf_score DESC LIMIT 30; import os import asyncpg from typing import List, Dict, Any import cohere from openai import AsyncOpenAI openai_client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"]) cohere_client = cohere.AsyncClientV2(api_key=os.environ["COHERE_API_KEY"]) class ProductionRAGEngine: def __init__(self, db_pool: asyncpg.Pool): self.db_pool = db_pool async def generate_embedding(self, text: str) -> List[float]: response = await openai_client.embeddings.create( model="text-embedding-3-small", input=text ) return response.data[0].embedding async def retrieve_hybrid_rrf(self, query: str, ticker_filter: str, top_k: int = 30) -> List[Dict[str, Any]]: query_embedding = await self.generate_embedding(query) embedding_str = "[" + ",".join(map(str, query_embedding)) + "]" sql = """ WITH dense AS ( SELECT id, content, context_prefix, document_id, ROW_NUMBER() OVER (ORDER BY embedding <=> $1::vector) AS rank FROM document_chunks WHERE ($3::text IS NULL OR metadata->>'ticker' = $3) LIMIT 50 ), sparse AS ( SELECT id, content, context_prefix, document_id, ROW_NUMBER() OVER (ORDER BY ts_rank_cd(tsv_content, plainto_tsquery('english', $2)) DESC) AS rank FROM document_chunks WHERE tsv_content @@ plainto_tsquery('english', $2) AND ($3::text IS NULL OR metadata->>'ticker' = $3) LIMIT 50 ) SELECT COALESCE(d.id, s.id) AS chunk_id, COALESCE(d.content, s.content) AS content, COALESCE(d.context_prefix, s.context_prefix) AS context_prefix, COALESCE(d.document_id, s.document_id) AS document_id, (COALESCE(1.0 / (60 + d.rank), 0.0) + COALESCE(1.0 / (60 + s.rank), 0.0)) AS rrf_score FROM dense d FULL OUTER JOIN sparse s ON d.id = s.id ORDER BY rrf_score DESC LIMIT $4; """ async with self.db_pool.acquire() as conn: rows = await conn.fetch(sql, embedding_str, query, ticker_filter, top_k) return [dict(r) for r in rows] async def rerank_results(self, query: str, candidate_chunks: List[Dict[str, Any]], final_k: int = 5) -> List[Dict[str, Any]]: if not candidate_chunks: return [] docs_to_rerank = [f"{c['context_prefix']}\n\n{c['content']}" for c in candidate_chunks] response = await cohere_client.rerank( model="rerank-v3.5", query=query, documents=docs_to_rerank, top_n=final_k ) final_results = [] for item in response.results: if item.relevance_score >= 0.65: chunk = candidate_chunks[item.index] chunk["rerank_score"] = item.relevance_score final_results.append(chunk) return final_results You are an Elite Financial Compliance Intelligence Assistant. Your task is to answer the user's inquiry based STRICTLY on the provided retrieved context below. OPERATIONAL CITATION DIRECTIVES: 1. STRICT GROUNDING: State only facts directly supported by the context. Never infer or speculate. 2. MANDATORY CITATION SYNTAX: Every factual sentence or metric must be followed immediately by its document citation tag formatted exactly as: <Doc: Chunk: {chunk_id} {document_id} |>. 3. ABSTENTION RULE: If the retrieved documents do not contain sufficient evidence to answer the query with 100% confidence, you MUST decline by stating: "The provided internal regulatory and filing records do not contain sufficient evidence to answer this inquiry." 4. ZERO CONVERSATIONAL FILLER: Output the analysis directly without preambles like "Based on the documents provided...". === RETRIEVED CONTEXT CHUNKS === {formatted_context_blocks} ================================ import os import pytest from datasets import Dataset from ragas import evaluate from ragas.metrics import faithfulness, answer_relevance, context_precision, context_recall def test_evaluate_rag_pipeline(): sample_dataset = Dataset.from_dict({ "question": ["What was Vanguard Regional Bank's Tier 1 Capital Ratio in Q4 2025?"], "contexts": [["According to the Q4 2025 regulatory filing, Vanguard Regional Bank reported a Tier 1 Capital Ratio of 13.4%."]], "answer": ["Vanguard Regional Bank reported a Tier 1 Capital Ratio of 13.4% in Q4 2025 <Doc: Chunk: ch_99 doc_01 |>."], "ground_truth": ["The Tier 1 Capital Ratio was 13.4% in Q4 2025."] }) results = evaluate( dataset=sample_dataset, metrics=[faithfulness, answer_relevance, context_precision, context_recall] ) assert results["faithfulness"] >= 0.90, f"Faithfulness below SLA: {results['faithfulness']}" assert results["answer_relevance"] >= 0.88, f"Answer relevance below SLA: {results['answer_relevance']}"