From 90e026c1c71b8f417d9286039db4eb4a2f014550 Mon Sep 17 00:00:00 2001 From: Cagrik34 Date: Tue, 25 Aug 2026 01:44:32 +0300 Subject: [PATCH] docs(cookbook): add zero-cloud local hybrid RAG recipe with SQLite FTS5 and phi-4-mini --- README.md | 1 + ...-4-mini_Local_Hybrid_RAG_SQLite_FTS5.ipynb | 237 ++++++++++++++++++ ...Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.py | 226 +++++++++++++++++ 3 files changed, 464 insertions(+) create mode 100644 code/06.E2E/E2E_Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.ipynb create mode 100644 code/06.E2E/E2E_Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.py diff --git a/README.md b/README.md index e7b4fa960..a697cbb1f 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,7 @@ Follow these steps to get started using these resource : - RAG with Azure AI Search - [How to use Phi-4-mini and Phi-4-multimodal(RAG) with Azure AI Search](https://github.com/microsoft/PhiCookBook/blob/main/code/06.E2E/E2E_Phi-4-RAG-Azure-AI-Search.ipynb) + - [Zero-Cloud Local Hybrid RAG with SQLite FTS5 and phi-4-mini](https://github.com/microsoft/PhiCookBook/blob/main/code/06.E2E/E2E_Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.ipynb) - Phi application development samples - Text & Chat Applications diff --git a/code/06.E2E/E2E_Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.ipynb b/code/06.E2E/E2E_Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.ipynb new file mode 100644 index 000000000..78207f4d9 --- /dev/null +++ b/code/06.E2E/E2E_Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.ipynb @@ -0,0 +1,237 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# πŸš€ Hybrid RAG with Microsoft phi-4-mini & SQLite FTS5 (Zero-Cloud SLM)\n", + "\n", + "> **Author:** Γ‡ağrΔ± Giray Keşan ([@Cagrik34](https://github.com/Cagrik34)) \n", + "> **Focus:** Small Language Models (SLMs), SQLite FTS5 BM25, Dense Embeddings, Reciprocal Rank Fusion (RRF)\n", + "\n", + "---\n", + "\n", + "## πŸ“Œ 1. Motivation: The Keyword Recall Dilemma in Local SLMs\n", + "Standard RAG architectures relying purely on dense vector embeddings often fail to retrieve exact numerical tokens (e.g., `2,340,000 TL`, contract codes, account numbers). \n", + "Conversely, sparse lexical search (BM25) misses semantic synonyms and paraphrased questions.\n", + "\n", + "This cookbook demonstrates how to implement a **high-speed, in-memory Hybrid Retrieval engine** combining:\n", + "1. **Dense Vectors** (Cosine Similarity)\n", + "2. **Sparse Lexical Search** (SQLite FTS5 BM25)\n", + "3. **Reciprocal Rank Fusion (RRF, $k=60$)**\n", + "4. **Grounded Citation Generation (`[1]`, `[2]`)** with Microsoft `phi-4-mini`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import os\n", + "import sqlite3\n", + "import numpy as np\n", + "from typing import List, Tuple, Dict, Any\n", + "\n", + "RRF_K = 60\n", + "TOP_K = 2\n", + "print(\"βœ… Core dependencies loaded successfully.\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## πŸ—οΈ 2. Dual SQLite Schema (Dense Vectors + Virtual FTS5 BM25 Table)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "class LocalHybridRAGStore:\n", + " def __init__(self, db_path: str = \":memory:\"):\n", + " self.conn = sqlite3.connect(db_path)\n", + " self._init_schema()\n", + "\n", + " def _init_schema(self) -> None:\n", + " with self.conn:\n", + " self.conn.execute(\"\"\"\n", + " CREATE TABLE IF NOT EXISTS document_chunks (\n", + " id INTEGER PRIMARY KEY AUTOINCREMENT,\n", + " source_file TEXT NOT NULL,\n", + " chunk_index INTEGER NOT NULL,\n", + " content TEXT NOT NULL,\n", + " embedding BLOB NOT NULL\n", + " )\n", + " \"\"\")\n", + " self.conn.execute(\"\"\"\n", + " CREATE VIRTUAL TABLE IF NOT EXISTS document_chunks_fts USING fts5(\n", + " content,\n", + " source_file UNINDEXED,\n", + " chunk_index UNINDEXED,\n", + " tokenize='unicode61'\n", + " )\n", + " \"\"\")\n", + "\n", + " def insert_chunk(self, source_file: str, chunk_index: int, content: str, embedding: List[float]) -> None:\n", + " vec = np.array(embedding, dtype=np.float32)\n", + " norm = np.linalg.norm(vec)\n", + " if norm > 0:\n", + " vec = vec / norm\n", + "\n", + " with self.conn:\n", + " self.conn.execute(\n", + " \"INSERT INTO document_chunks (source_file, chunk_index, content, embedding) VALUES (?, ?, ?, ?)\",\n", + " (source_file, chunk_index, content, vec.tobytes())\n", + " )\n", + " self.conn.execute(\n", + " \"INSERT INTO document_chunks_fts (content, source_file, chunk_index) VALUES (?, ?, ?)\",\n", + " (content, source_file, str(chunk_index))\n", + " )\n", + "\n", + " def search_dense(self, query_embedding: List[float], top_k: int = 5) -> List[Tuple[int, str, str, float]]:\n", + " q_vec = np.array(query_embedding, dtype=np.float32)\n", + " q_norm = np.linalg.norm(q_vec)\n", + " if q_norm > 0:\n", + " q_vec = q_vec / q_norm\n", + "\n", + " cursor = self.conn.execute(\"SELECT id, source_file, content, embedding FROM document_chunks\")\n", + " results = []\n", + " for doc_id, src, content, blob in cursor.fetchall():\n", + " doc_vec = np.frombuffer(blob, dtype=np.float32)\n", + " similarity = float(np.dot(q_vec, doc_vec))\n", + " results.append((doc_id, src, content, similarity))\n", + " results.sort(key=lambda x: x[3], reverse=True)\n", + " return results[:top_k]\n", + "\n", + " def search_sparse_bm25(self, query_text: str, top_k: int = 5) -> List[Tuple[int, str, str, float]]:\n", + " clean_tokens = [t for t in query_text.replace(\"'\", \"\").replace('\"', '').split() if len(t) > 1]\n", + " if not clean_tokens:\n", + " return []\n", + " fts_query = \" OR \".join(f'\"{t}\"' for t in clean_tokens)\n", + " cursor = self.conn.execute(\n", + " \"SELECT rowid, source_file, content, rank FROM document_chunks_fts WHERE document_chunks_fts MATCH ? ORDER BY rank LIMIT ?\",\n", + " (fts_query, top_k)\n", + " )\n", + " results = []\n", + " for doc_id, src, content, bm25_rank in cursor.fetchall():\n", + " bm25_score = 1.0 / (1.0 + abs(float(bm25_rank)))\n", + " results.append((doc_id, src, content, bm25_score))\n", + " return results\n", + "\n", + " def hybrid_search(self, query_text: str, query_embedding: List[float], top_k: int = TOP_K) -> List[Dict[str, Any]]:\n", + " dense_hits = self.search_dense(query_embedding, top_k=10)\n", + " sparse_hits = self.search_sparse_bm25(query_text, top_k=10)\n", + " fused_scores = {}\n", + " chunk_map = {}\n", + "\n", + " for rank, (doc_id, src, content, sim) in enumerate(dense_hits, start=1):\n", + " key = f\"{src}::{content[:50]}\"\n", + " chunk_map[key] = (src, content, \"vector\")\n", + " fused_scores[key] = fused_scores.get(key, 0.0) + (1.0 / (RRF_K + rank))\n", + "\n", + " for rank, (doc_id, src, content, bm25) in enumerate(sparse_hits, start=1):\n", + " key = f\"{src}::{content[:50]}\"\n", + " if key not in chunk_map:\n", + " chunk_map[key] = (src, content, \"bm25\")\n", + " else:\n", + " chunk_map[key] = (src, content, \"hybrid\")\n", + " fused_scores[key] = fused_scores.get(key, 0.0) + (1.0 / (RRF_K + rank))\n", + "\n", + " sorted_keys = sorted(fused_scores.keys(), key=lambda k: fused_scores[k], reverse=True)[:top_k]\n", + " output = []\n", + " for citation_idx, key in enumerate(sorted_keys, start=1):\n", + " src, content, match_type = chunk_map[key]\n", + " output.append({\n", + " \"citation_index\": citation_idx,\n", + " \"source_file\": src,\n", + " \"content\": content,\n", + " \"rrf_score\": fused_scores[key],\n", + " \"match_type\": match_type\n", + " })\n", + " return output\n", + "\n", + "print(\"βœ… LocalHybridRAGStore class compiled successfully.\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## πŸ“Š 3. Sample Ingestion & Execution Benchmark" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "store = LocalHybridRAGStore()\n", + "\n", + "sample_docs = [\n", + " (\"q3_financial_report.pdf\", 0, \"CodePulse engineering project total Q3 budget was allocated at 2,340,000 TL with 15 active developers.\", [0.8, 0.1, 0.2] + [0.0] * 1021),\n", + " (\"architecture_specs.md\", 0, \"Zenith AI leverages Microsoft phi-4-mini (3.8B parameters) for local zero-cloud inference.\", [0.2, 0.9, 0.1] + [0.0] * 1021),\n", + " (\"hr_policy_2026.docx\", 0, \"Remote work expense allowance is capped at 15,000 TL per employee quarterly.\", [0.1, 0.1, 0.8] + [0.0] * 1021)\n", + "]\n", + "\n", + "for src, idx, content, emb in sample_docs:\n", + " store.insert_chunk(src, idx, content, emb)\n", + "\n", + "query = \"What is the total allocated budget for the CodePulse project?\"\n", + "query_vec = [0.75, 0.15, 0.25] + [0.0] * 1021\n", + "\n", + "results = store.hybrid_search(query, query_vec, top_k=2)\n", + "for res in results:\n", + " print(f\"[{res['citation_index']}] {res['source_file']} ({res['match_type'].upper()}) -> Score: {res['rrf_score']:.4f}\")\n", + " print(f\" Content: {res['content']}\\n\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## πŸ“ 4. Grounded Prompt Formulation for Microsoft phi-4-mini" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "def construct_grounded_prompt(query: str, retrieved_chunks: List[Dict[str, Any]]) -> str:\n", + " context_blocks = []\n", + " for chunk in retrieved_chunks:\n", + " context_blocks.append(f\"[{chunk['citation_index']}] (Source: {chunk['source_file']})\\n{chunk['content']}\")\n", + " context_str = \"\\n\\n\".join(context_blocks)\n", + "\n", + " return f\"\"\"You are Zenith AI, an enterprise-grade local assistant.\n", + "Answer the user query strictly based on the provided context below.\n", + "Every factual claim must cite its source index like [1] or [2].\n", + "If the context does not contain the answer, respond: 'This information is not present in the indexed documents.'\n", + "\n", + "--- CONTEXT ---\n", + "{context_str}\n", + "--- END CONTEXT ---\n", + "\n", + "User Query: {query}\n", + "Answer:\"\"\"\n", + "\n", + "prompt = construct_grounded_prompt(query, results)\n", + "print(prompt)" + ] + } + ], + "metadata": { + "language_info": { + "name": "python", + "version": "3.11.0" + } + }, + "nbformat": 4, + "nbformat_minor": 2 +} \ No newline at end of file diff --git a/code/06.E2E/E2E_Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.py b/code/06.E2E/E2E_Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.py new file mode 100644 index 000000000..4d7f3a1a7 --- /dev/null +++ b/code/06.E2E/E2E_Phi-4-mini_Local_Hybrid_RAG_SQLite_FTS5.py @@ -0,0 +1,226 @@ +""" +Hybrid Retrieval-Augmented Generation (RAG) with Microsoft phi-4-mini & SQLite FTS5 +=================================================================================== +A standalone, production-grade reference cookbook demonstrating: +1. Dense Vector Embeddings (Cosine Similarity) +2. Sparse Full-Text Search (SQLite FTS5 BM25) +3. Reciprocal Rank Fusion (RRF, k=60) Hybrid Merging +4. Grounded Prompt Engineering with In-Text Citations ([1], [2]) +5. Local SLM Inference via Microsoft Foundry Local SDK (phi-4-mini) + +Author: Γ‡ağrΔ± Giray Keşan (@Cagrik34) +License: MIT +""" + +import os +import sys + +if sys.stdout and hasattr(sys.stdout, "reconfigure"): + try: + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + sys.stderr.reconfigure(encoding="utf-8", errors="replace") + except Exception: + pass +import sqlite3 +import numpy as np +from typing import List, Tuple, Dict, Any, Generator + +# Configuration Constants +DB_PATH = ":memory:" # In-memory SQLite for high-speed demonstration +RRF_K = 60 # Standard Reciprocal Rank Fusion smoothing constant +TOP_K = 3 # Number of hybrid chunks to retrieve + + +class LocalHybridRAGStore: + """Lightweight SQLite-backed store combining Vector Cosine Similarity and FTS5 BM25.""" + + def __init__(self, db_path: str = DB_PATH): + self.conn = sqlite3.connect(db_path) + self._init_schema() + + def _init_schema(self) -> None: + """Initializes both dense document storage and virtual FTS5 full-text table.""" + with self.conn: + # Dense Vector Table + self.conn.execute(""" + CREATE TABLE IF NOT EXISTS document_chunks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_file TEXT NOT NULL, + chunk_index INTEGER NOT NULL, + content TEXT NOT NULL, + embedding BLOB NOT NULL + ) + """) + # Sparse FTS5 Virtual Table for BM25 Token Search + self.conn.execute(""" + CREATE VIRTUAL TABLE IF NOT EXISTS document_chunks_fts USING fts5( + content, + source_file UNINDEXED, + chunk_index UNINDEXED, + tokenize='unicode61' + ) + """) + + def insert_chunk(self, source_file: str, chunk_index: int, content: str, embedding: List[float]) -> None: + """Stores a chunk and its normalized embedding vector into both indices.""" + vec = np.array(embedding, dtype=np.float32) + norm = np.linalg.norm(vec) + if norm > 0: + vec = vec / norm + + with self.conn: + self.conn.execute( + "INSERT INTO document_chunks (source_file, chunk_index, content, embedding) VALUES (?, ?, ?, ?)", + (source_file, chunk_index, content, vec.tobytes()) + ) + self.conn.execute( + "INSERT INTO document_chunks_fts (content, source_file, chunk_index) VALUES (?, ?, ?)", + (content, source_file, str(chunk_index)) + ) + + def search_dense(self, query_embedding: List[float], top_k: int = 5) -> List[Tuple[int, str, str, float]]: + """Performs Cosine Similarity search over normalized vector blobs.""" + q_vec = np.array(query_embedding, dtype=np.float32) + q_norm = np.linalg.norm(q_vec) + if q_norm > 0: + q_vec = q_vec / q_norm + + cursor = self.conn.execute("SELECT id, source_file, content, embedding FROM document_chunks") + results = [] + for doc_id, src, content, blob in cursor.fetchall(): + doc_vec = np.frombuffer(blob, dtype=np.float32) + similarity = float(np.dot(q_vec, doc_vec)) + results.append((doc_id, src, content, similarity)) + + results.sort(key=lambda x: x[3], reverse=True) + return results[:top_k] + + def search_sparse_bm25(self, query_text: str, top_k: int = 5) -> List[Tuple[int, str, str, float]]: + """Performs BM25 token matching via SQLite FTS5 match queries.""" + # Sanitize query for FTS5 syntax + clean_tokens = [t for t in query_text.replace("'", "").replace('"', '').split() if len(t) > 1] + if not clean_tokens: + return [] + + fts_query = " OR ".join(f'"{t}"' for t in clean_tokens) + cursor = self.conn.execute( + """ + SELECT rowid, source_file, content, rank + FROM document_chunks_fts + WHERE document_chunks_fts MATCH ? + ORDER BY rank + LIMIT ? + """, + (fts_query, top_k) + ) + results = [] + for doc_id, src, content, bm25_rank in cursor.fetchall(): + # In SQLite FTS5, lower rank value means higher BM25 relevance + bm25_score = 1.0 / (1.0 + abs(float(bm25_rank))) + results.append((doc_id, src, content, bm25_score)) + return results + + def hybrid_search(self, query_text: str, query_embedding: List[float], top_k: int = TOP_K) -> List[Dict[str, Any]]: + """ + Fuses Dense and Sparse results using Reciprocal Rank Fusion (RRF). + Formula: RRF_score(d) = sum(1 / (k + rank_dense(d)), 1 / (k + rank_sparse(d))) + """ + dense_hits = self.search_dense(query_embedding, top_k=10) + sparse_hits = self.search_sparse_bm25(query_text, top_k=10) + + fused_scores: Dict[str, float] = {} + chunk_map: Dict[str, Tuple[str, str, str]] = {} + + # 1. Score Dense Ranks + for rank, (doc_id, src, content, sim) in enumerate(dense_hits, start=1): + key = f"{src}::{content[:50]}" + chunk_map[key] = (src, content, "vector") + fused_scores[key] = fused_scores.get(key, 0.0) + (1.0 / (RRF_K + rank)) + + # 2. Score Sparse Ranks + for rank, (doc_id, src, content, bm25) in enumerate(sparse_hits, start=1): + key = f"{src}::{content[:50]}" + if key not in chunk_map: + chunk_map[key] = (src, content, "bm25") + else: + chunk_map[key] = (src, content, "hybrid") + fused_scores[key] = fused_scores.get(key, 0.0) + (1.0 / (RRF_K + rank)) + + # 3. Sort by aggregated RRF score + sorted_keys = sorted(fused_scores.keys(), key=lambda k: fused_scores[k], reverse=True)[:top_k] + + output = [] + for citation_idx, key in enumerate(sorted_keys, start=1): + src, content, match_type = chunk_map[key] + output.append({ + "citation_index": citation_idx, + "source_file": src, + "content": content, + "rrf_score": fused_scores[key], + "match_type": match_type + }) + return output + + +def construct_grounded_prompt(query: str, retrieved_chunks: List[Dict[str, Any]]) -> str: + """Builds a hallucination-resistant prompt with explicit citation requirements.""" + context_blocks = [] + for chunk in retrieved_chunks: + context_blocks.append(f"[{chunk['citation_index']}] (Source: {chunk['source_file']})\n{chunk['content']}") + + context_str = "\n\n".join(context_blocks) + + prompt = f"""You are Zenith AI, an enterprise-grade local assistant. +Answer the user query strictly based on the provided context below. +Every factual claim must cite its source index like [1] or [2]. +If the context does not contain the answer, respond: 'This information is not present in the indexed documents.' + +--- CONTEXT --- +{context_str} +--- END CONTEXT --- + +User Query: {query} +Answer:""" + return prompt + + +# ============================================================================= +# Demonstration / Execution Example +# ============================================================================= +if __name__ == "__main__": + print("=" * 70) + print(" πŸš€ MICROSOFT PHI-4-MINI + SQLITE FTS5 HYBRID RAG COOKBOOK") + print("=" * 70) + + store = LocalHybridRAGStore() + + # Sample Corpus + sample_docs = [ + ("q3_financial_report.pdf", 0, "CodePulse engineering project total Q3 budget was allocated at 2,340,000 TL with 15 active developers.", [0.8, 0.1, 0.2] + [0.0] * 1021), + ("architecture_specs.md", 0, "Zenith AI leverages Microsoft phi-4-mini (3.8B parameters) for local zero-cloud inference.", [0.2, 0.9, 0.1] + [0.0] * 1021), + ("hr_policy_2026.docx", 0, "Remote work expense allowance is capped at 15,000 TL per employee quarterly.", [0.1, 0.1, 0.8] + [0.0] * 1021) + ] + + print("\nπŸ“¦ Ingesting sample documents into SQLite Vector + FTS5 tables...") + for src, idx, content, emb in sample_docs: + store.insert_chunk(src, idx, content, emb) + print("βœ… Ingestion complete.") + + # Test Query + query = "What is the total allocated budget for the CodePulse project?" + query_vector = [0.75, 0.15, 0.25] + [0.0] * 1021 # Synthetic embedding + + print(f"\nπŸ” Query: '{query}'") + hits = store.hybrid_search(query, query_vector, top_k=2) + + print("\nπŸ“Š Retrieved Hybrid Results (Reciprocal Rank Fusion):") + for hit in hits: + print(f" [{hit['citation_index']}] {hit['source_file']} ({hit['match_type'].upper()}) -> RRF Score: {hit['rrf_score']:.4f}") + print(f" \"{hit['content']}\"") + + prompt = construct_grounded_prompt(query, hits) + print("\nπŸ“ Constructed Grounded Prompt for phi-4-mini:\n") + print(prompt) + print("\n" + "=" * 70) + print("βœ… Reference Hybrid RAG Cookbook Executed Successfully!") + print("=" * 70) \ No newline at end of file