in my weblog publish<\/a>, I had explored the best way to construct a Multi-agent SQL assistant utilizing CrewAI & Streamlit. The person may question a SQLite database in pure language. The AI brokers would generate a SQL question primarily based on person enter, overview it, and verify for compliance earlier than operating it towards the database to get outcomes. I additionally applied a human-in-the-loop checkpoint to take care of management and displayed the LLM prices related to each question for transparency and price management. Whereas the prototype was nice and generated good outcomes for my small demo database, I knew this is able to not be sufficient for real-life databases. Within the earlier setup, I used to be sending the entire database schema as context together with the person enter. As database schemas develop bigger, passing the complete schema to the LLM will increase token utilization, slows down response occasions, and makes hallucination extra seemingly. I wanted a strategy to feed solely related schema snippets to the LLM. That is the place RAG (Retrieval Augmented Era)<\/strong> is available in.<\/p>\n

On this weblog publish, I construct a RAG supervisor and add a number of<\/strong> RAG methods<\/strong> to my SQL assistant to match their efficiency on metrics like response time and token utilization. The assistant now helps 4 RAG methods:<\/p>\n

\n
No RAG:<\/strong> Passes the complete schema (baseline for comparability)<\/li>\n
Key phrase RAG:<\/strong> Makes use of domain-specific key phrase matching to pick out related tables<\/li>\n
FAISS RAG:<\/strong> Leverages semantic vector similarity by way of FAISS with all-MiniLM-L6-v2<\/code> embeddings<\/li>\n

Chroma RAG:<\/strong> A persistent vector retailer resolution with ChromaDB for scalable production-grade search<\/li>\n<\/ul>\nFor this challenge, I solely centered on RAG methods which are sensible, light-weight, and cost-efficient (free). You possibly can add any variety of implementations on prime and select one of the best one on your case. To facilitate experimentation and evaluation, I constructed an interactive efficiency comparability device that evaluates token discount, desk depend, response time, and question accuracy throughout all 4 methods.<\/p>\n
$\"\"$
Screenshot of App, by Writer<\/figcaption><\/figure>\n
Constructing a RAG Supervisor<\/h2>\n
The rag_manager.py<\/code> file incorporates the complete implementation for the RAG supervisor. First, I created a BaseRAG<\/code> class \u2013 a template that I take advantage of for all my completely different RAG methods. It makes certain each RAG strategy follows the identical construction. Any new technique could have two issues: a way to fetch the related schema primarily based on the person question, and one other technique that explains what the strategy is about. By utilizing the summary base class<\/strong> (ABC), I hold the code clear, modular, and straightforward to increase later.<\/p>\n
from typing import Dict, Listing, Any, Non-compulsory\nfrom abc import ABC, abstractmethod\n\nclass BaseRAG(ABC):\n \"\"\"Base class for all RAG implementations.\"\"\"\n \n def __init__(self, db_path: str = DB_PATH):\n self.db_path = db_path\n self.identify = self.__class__.__name__\n \n @abstractmethod\n def get_relevant_schema(self, user_query: str, max_tables: int = 5) -> str:\n \"\"\"Get related schema for the person question.\"\"\"\n go\n \n @abstractmethod\n def get_approach_info(self) -> Dict[str, Any]:\n \"\"\"Get details about this RAG strategy.\"\"\"\n go\n<\/code><\/pre>\n
No RAG Technique<\/h2>\n
That is principally the identical strategy that I used beforehand, the place I despatched in the complete database schema as context to the LLM with none filtering or optimization. This strategy is greatest for very small schemas (ideally lower than 10 tables).<\/p>\n
class NoRAG(BaseRAG):\n \"\"\"No RAG - returns full schema.\"\"\"\n \n def get_relevant_schema(self, user_query: str, max_tables: int = 5) -> str:\n return get_structured_schema(self.db_path)\n \n def get_approach_info(self) -> Dict[str, Any]:\n return {\n \"identify\": \"No RAG (Full Schema)\",\n \"description\": \"Makes use of full database schema\",\n \"professionals\": [\"Simple\", \"Always complete\", \"No setup required\"],\n \"cons\": [\"High token usage\", \"Slower for large schemas\"],\n \"best_for\": \"Small schemas (< 10 tables)\"\n }\n<\/code><\/pre>\n
Key phrase RAG Technique<\/h2>\n
Within the key phrase RAG strategy, I take advantage of a bunch of predefined key phrases mapped to every desk within the schema. When a person asks one thing, the system checks for key phrase matches within the question and picks out solely probably the most related tables. This fashion, I don\u2019t ship the complete schema to the LLM \u2013 saving tokens and dashing issues up. It really works properly when your schema is acquainted and your queries are business-related or comply with widespread patterns.<\/p>\n
The _build_table_keywords(self)<\/code> technique is the core of how the key phrase matching logic works. It incorporates a hardcoded key phrase mapping for every desk within the schema. It helps to affiliate person question phrases (like \u201cgross sales\u201d, \u201cmodel\u201d, \u201cbuyer\u201d) with the more than likely related tables.<\/p>\n
class KeywordRAG(BaseRAG):\n \"\"\"Key phrase-based RAG utilizing enterprise context matching.\"\"\"\n \n def __init__(self, db_path: str = DB_PATH):\n tremendous().__init__(db_path)\n self.table_keywords = self._build_table_keywords()\n \n def _build_table_keywords(self) -> Dict[str, List[str]]:\n \"\"\"Construct key phrase mappings for every desk.\"\"\"\n return {\n 'merchandise': ['product', 'item', 'catalog', 'price', 'category', 'brand', 'sales', 'sold'],\n 'product_variants': ['variant', 'product', 'sku', 'color', 'size', 'brand', 'sales', 'sold'],\n 'clients': ['customer', 'user', 'client', 'buyer', 'person', 'email', 'name'],\n 'orders': ['order', 'purchase', 'transaction', 'sale', 'buy', 'total', 'amount', 'sales'],\n 'order_items': ['item', 'product', 'quantity', 'line', 'detail', 'sales', 'sold', 'brand'],\n 'funds': ['payment', 'pay', 'money', 'revenue', 'amount'],\n 'stock': ['inventory', 'stock', 'quantity', 'warehouse', 'available'],\n 'evaluations': ['review', 'rating', 'feedback', 'comment', 'opinion'],\n 'suppliers': ['supplier', 'vendor', 'procurement', 'purchase'],\n 'classes': ['category', 'type', 'classification', 'group'],\n 'manufacturers': ['brand', 'manufacturer', 'company', 'sales', 'sold', 'quantity', 'total'],\n 'addresses': ['address', 'location', 'shipping', 'billing'],\n 'shipments': ['shipment', 'delivery', 'shipping', 'tracking'],\n 'reductions': ['discount', 'coupon', 'promotion', 'offer'],\n 'warehouses': ['warehouse', 'facility', 'location', 'storage'],\n 'workers': ['employee', 'staff', 'worker', 'person'],\n 'departments': ['department', 'division', 'team'],\n 'product_images': ['image', 'photo', 'picture', 'media'],\n 'purchase_orders': ['purchase', 'procurement', 'supplier', 'order'],\n 'purchase_order_items': ['purchase', 'procurement', 'supplier', 'item'],\n 'order_discounts': ['discount', 'coupon', 'promotion', 'order'],\n 'shipment_items': ['shipment', 'delivery', 'item', 'tracking']\n }\n \n def get_relevant_schema(self, user_query: str, max_tables: int = 5) -> str:\n import re\n \n # Rating tables by key phrase relevance\n query_words = set(re.findall(r'bw+b', user_query.decrease()))\n table_scores = {}\n \n for table_name, key phrases in self.table_keywords.objects():\n rating = 0\n \n # Depend key phrase matches\n for key phrase in key phrases:\n if key phrase in query_words:\n rating += 3\n # Partial matches\n for query_word in query_words:\n if key phrase in query_word or query_word in key phrase:\n rating += 1\n \n # Bonus for actual desk identify match\n if table_name.decrease() in query_words:\n rating += 10\n \n table_scores[table_name] = rating\n \n # Get prime scoring tables\n sorted_tables = sorted(table_scores.objects(), key=lambda x: x[1], reverse=True)\n relevant_tables = [table for table, score in sorted_tables[:max_tables] if rating > 0]\n \n # Fallback to default tables if no matches\n if not relevant_tables:\n relevant_tables = self._get_default_tables(user_query)[:max_tables]\n \n # Construct schema for chosen tables\n return self._build_schema(relevant_tables)\n \n def _get_default_tables(self, user_query: str) -> Listing[str]:\n \"\"\"Get default tables primarily based on question patterns.\"\"\"\n query_lower = user_query.decrease()\n \n # Gross sales\/income queries\n if any(phrase in query_lower for phrase in ['revenue', 'sales', 'total', 'amount', 'brand']):\n return ['orders', 'order_items', 'product_variants', 'products', 'brands']\n \n # Product queries\n if any(phrase in query_lower for phrase in ['product', 'item', 'catalog']):\n return ['products', 'product_variants', 'categories', 'brands']\n \n # Buyer queries\n if any(phrase in query_lower for phrase in ['customer', 'user', 'buyer']):\n return ['customers', 'orders', 'addresses']\n \n # Default\n return ['products', 'customers', 'orders', 'order_items']\n \n def _build_schema(self, table_names: Listing[str]) -> str:\n \"\"\"Construct schema string for specified tables.\"\"\"\n if not table_names:\n return get_structured_schema(self.db_path)\n \n conn = sqlite3.join(self.db_path)\n cursor = conn.cursor()\n schema_lines = [\"Available tables and columns:\"]\n \n strive:\n for table_name in table_names:\n cursor.execute(f\"PRAGMA table_info({table_name});\")\n columns = cursor.fetchall()\n if columns:\n col_names = [col[1] for col in columns]\n schema_lines.append(f\"- {table_name}: {', '.be a part of(col_names)}\")\n lastly:\n conn.shut()\n \n return 'n'.be a part of(schema_lines)\n \n def get_approach_info(self) -> Dict[str, Any]:\n return {\n \"identify\": \"Key phrase RAG\",\n \"description\": \"Makes use of enterprise context key phrases to match related tables\",\n \"professionals\": [\"Fast\", \"No external dependencies\", \"Good for business queries\"],\n \"cons\": [\"Limited by predefined keywords\", \"May miss complex relationships\"],\n \"best_for\": \"Enterprise queries with clear area phrases\"\n }\n<\/code><\/pre>\nFAISS RAG Method<\/h2>\nThe FAISS RAG<\/strong> technique is the place issues begin getting smarter. As a substitute of dumping the entire schema, I embed every desk\u2019s metadata (columns, relationships, enterprise context) into vectors utilizing a sentence transformer. When the person asks a query, it’s embedded into that question too and makes use of FAISS to do a semantic search matching on the \u201cwhich means\u201d as an alternative of simply key phrases. It\u2019s good for queries the place customers aren\u2019t being very particular or when tables have associated phrases. I like FAISS as a result of it\u2019s free, runs regionally, and provides fairly correct outcomes whereas saving tokens.<\/p>\n The one catch is that setting it up takes some additional steps, and it makes use of extra reminiscence than fundamental approaches. LLMs and embedding fashions don\u2019t know what your tables imply except you clarify it to them. Within the _get_business_context()<\/code> technique, we have to manually write a brief description of what every desk represents within the enterprise.<\/p>\n Within the _extract_table_info()<\/code> technique, I pull in desk names, column names, and international key relationships from SQLite\u2019s PRAGMA queries to construct a dictionary with structured data about every desk. Lastly, within the _create_table_description()<\/code> technique, complete descriptions for every desk are constructed to be embedded by a SentenceTransformer.<\/p>\n class FAISSVectorRAG(BaseRAG):\n \"\"\"FAISS-based vector RAG utilizing sentence transformers.\"\"\"\n \n def __init__(self, db_path: str = DB_PATH):\n tremendous().__init__(db_path)\n self.mannequin = None\n self.index = None\n self.table_info = {}\n self.table_names = []\n self._initialize()\n \n def _initialize(self):\n \"\"\"Initialize the FAISS vector retailer and embeddings.\"\"\"\n strive:\n from sentence_transformers import SentenceTransformer\n import faiss\n import numpy as np\n \n print(\"\ud83d\udd04 Initializing FAISS Vector RAG...\")\n \n # Load embedding mannequin\n self.mannequin = SentenceTransformer('all-MiniLM-L6-v2')\n print(\"\u2705 Loaded embedding mannequin: all-MiniLM-L6-v2\")\n \n # Extract desk data and create embeddings\n self.table_info = self._extract_table_info()\n \n # Create embeddings for every desk\n table_descriptions = []\n self.table_names = []\n \n for table_name, data in self.table_info.objects():\n description = self._create_table_description(table_name, data)\n table_descriptions.append(description)\n self.table_names.append(table_name)\n \n # Generate embeddings\n print(f\"\ud83d\udd04 Producing embeddings for {len(table_descriptions)} tables...\")\n embeddings = self.mannequin.encode(table_descriptions)\n \n # Create FAISS index\n dimension = embeddings.form[1]\n self.index = faiss.IndexFlatIP(dimension) # Interior product for cosine similarity\n \n # Normalize embeddings for cosine similarity\n faiss.normalize_L2(embeddings)\n self.index.add(embeddings.astype('float32'))\n \n print(f\"\u2705 FAISS Vector RAG initialized with {len(table_descriptions)} tables\")\n \n besides Exception as e:\n print(f\"\u274c Error initializing FAISS Vector RAG: {e}\")\n self.mannequin = None\n self.index = None\n \n def _extract_table_info(self) -> Dict[str, Dict]:\n \"\"\"Extract detailed details about every desk.\"\"\"\n conn = sqlite3.join(self.db_path)\n cursor = conn.cursor()\n table_info = {}\n \n strive:\n # Get all desk names\n cursor.execute(\"SELECT identify FROM sqlite_master WHERE sort='desk';\")\n tables = cursor.fetchall()\n \n for (table_name,) in tables:\n data = {\n 'columns': [],\n 'foreign_keys': [],\n 'business_context': self._get_business_context(table_name)\n }\n \n # Get column data\n cursor.execute(f\"PRAGMA table_info({table_name});\")\n columns = cursor.fetchall()\n for col in columns:\n data['columns'].append({\n 'identify': col[1],\n 'sort': col[2],\n 'primary_key': bool(col[5])\n })\n \n # Get international key data\n cursor.execute(f\"PRAGMA foreign_key_list({table_name});\")\n fks = cursor.fetchall()\n for fk in fks:\n data['foreign_keys'].append({\n 'column': fk[3],\n 'references_table': fk[2],\n 'references_column': fk[4]\n })\n \n table_info[table_name] = data\n \n lastly:\n conn.shut()\n \n return table_info\n \n def _get_business_context(self, table_name: str) -> str:\n \"\"\"Get enterprise context description for tables.\"\"\"\n contexts = {\n 'merchandise': 'Product catalog with objects, costs, classes, and model data. Core stock knowledge.',\n 'product_variants': 'Product variations like colours, sizes, SKUs. Hyperlinks merchandise to particular sellable objects.',\n 'clients': 'Buyer profiles with private data, contact particulars, and account standing.',\n 'orders': 'Buy transactions with totals, dates, standing, and buyer relationships.',\n 'order_items': 'Particular person line objects inside orders. Comprises portions, costs, and product references.',\n 'funds': 'Fee processing data with strategies, quantities, and transaction standing.',\n 'stock': 'Inventory ranges and warehouse portions for product variants.',\n 'evaluations': 'Buyer suggestions, scores, and product evaluations.',\n 'suppliers': 'Vendor data for procurement and provide chain administration.',\n 'classes': 'Product categorization hierarchy for organizing catalog.',\n 'manufacturers': 'Model data for merchandise and advertising functions.',\n 'addresses': 'Buyer delivery and billing tackle data.',\n 'shipments': 'Supply monitoring and delivery standing data.',\n 'reductions': 'Promotional codes, coupons, and low cost campaigns.',\n 'warehouses': 'Storage facility places and warehouse administration.',\n 'workers': 'Employees data and organizational construction.',\n 'departments': 'Organizational divisions and crew construction.',\n 'product_images': 'Product pictures and media property.',\n 'purchase_orders': 'Procurement orders from suppliers.',\n 'purchase_order_items': 'Line objects for provider buy orders.',\n 'order_discounts': 'Utilized reductions and promotions on orders.',\n 'shipment_items': 'Particular person objects inside cargo packages.'\n }\n \n return contexts.get(table_name, f'Database desk for {table_name} associated operations.')\n \n def _create_table_description(self, table_name: str, data: Dict) -> str:\n \"\"\"Create a complete description for embedding.\"\"\"\n description = f\"Desk: {table_name}n\"\n description += f\"Objective: {data['business_context']}n\"\n \n # Add column data\n description += \"Columns: \"\n col_names = [col['name'] for col in data['columns']]\n description += \", \".be a part of(col_names) + \"n\"\n \n # Add relationship data\n if data['foreign_keys']:\n description += \"Relationships: \"\n relationships = []\n for fk in data['foreign_keys']:\n relationships.append(f\"hyperlinks to {fk['references_table']} by way of {fk['column']}\")\n description += \"; \".be a part of(relationships) + \"n\"\n \n # Add widespread use instances primarily based on desk sort\n use_cases = self._get_use_cases(table_name)\n if use_cases:\n description += f\"Widespread queries: {use_cases}\"\n \n return description\n \n def _get_use_cases(self, table_name: str) -> str:\n \"\"\"Get widespread use instances for every desk.\"\"\"\n use_cases = {\n 'merchandise': 'product searches, catalog listings, value queries, stock checks',\n 'clients': 'buyer lookup, registration evaluation, geographic distribution',\n 'orders': 'gross sales evaluation, income monitoring, order historical past, standing monitoring',\n 'order_items': 'product gross sales efficiency, income by product, order composition',\n 'funds': 'cost processing, income reconciliation, cost technique evaluation',\n 'manufacturers': 'model efficiency, gross sales by model, model comparability',\n 'classes': 'class evaluation, product group, catalog construction'\n }\n \n return use_cases.get(table_name, 'common knowledge queries and evaluation')\n \n def get_relevant_schema(self, user_query: str, max_tables: int = 5) -> str:\n \"\"\"Get related schema utilizing vector similarity search.\"\"\"\n if self.mannequin is None or self.index is None:\n print(\"\u26a0\ufe0f FAISS not initialized, falling again to full schema\")\n return get_structured_schema(self.db_path)\n \n strive:\n import faiss\n import numpy as np\n \n # Generate question embedding\n query_embedding = self.mannequin.encode([user_query])\n faiss.normalize_L2(query_embedding)\n \n # Seek for related tables\n scores, indices = self.index.search(query_embedding.astype('float32'), max_tables)\n \n # Get related desk names\n relevant_tables = []\n for i, (rating, idx) in enumerate(zip(scores[0], indices[0])):\n if idx < len(self.table_names) and rating > 0.1: # Minimal similarity threshold\n relevant_tables.append(self.table_names[idx])\n \n # Fallback if no related tables discovered\n if not relevant_tables:\n print(\"\u26a0\ufe0f No related tables discovered, utilizing defaults\")\n relevant_tables = self._get_default_tables(user_query)[:max_tables]\n \n # Construct schema for chosen tables\n return self._build_schema(relevant_tables)\n \n besides Exception as e:\n print(f\"\u26a0\ufe0f Vector search failed: {e}, falling again to full schema\")\n return get_structured_schema(self.db_path)\n \n def _get_default_tables(self, user_query: str) -> Listing[str]:\n \"\"\"Get default tables primarily based on question patterns.\"\"\"\n query_lower = user_query.decrease()\n \n if any(phrase in query_lower for phrase in ['revenue', 'sales', 'total', 'amount', 'brand']):\n return ['orders', 'order_items', 'product_variants', 'products', 'brands']\n elif any(phrase in query_lower for phrase in ['product', 'item', 'catalog']):\n return ['products', 'product_variants', 'categories', 'brands']\n elif any(phrase in query_lower for phrase in ['customer', 'user', 'buyer']):\n return ['customers', 'orders', 'addresses']\n else:\n return ['products', 'customers', 'orders', 'order_items']\n \n def _build_schema(self, table_names: Listing[str]) -> str:\n \"\"\"Construct schema string for specified tables.\"\"\"\n if not table_names:\n return get_structured_schema(self.db_path)\n \n conn = sqlite3.join(self.db_path)\n cursor = conn.cursor()\n schema_lines = [\"Available tables and columns:\"]\n \n strive:\n for table_name in table_names:\n cursor.execute(f\"PRAGMA table_info({table_name});\")\n columns = cursor.fetchall()\n if columns:\n col_names = [col[1] for col in columns]\n schema_lines.append(f\"- {table_name}: {', '.be a part of(col_names)}\")\n lastly:\n conn.shut()\n \n return 'n'.be a part of(schema_lines)\n \n def get_approach_info(self) -> Dict[str, Any]:\n return {\n \"identify\": \"FAISS Vector RAG\",\n \"description\": \"Makes use of semantic embeddings and vector similarity search\",\n \"professionals\": [\"Semantic understanding\", \"Handles complex queries\", \"No API costs\"],\n \"cons\": [\"Requires model download\", \"Higher memory usage\", \"Setup complexity\"],\n \"best_for\": \"Complicated queries, giant schemas, semantic relationships\"\n }\n<\/code><\/pre>\nChroma RAG Technique<\/h2>\nChroma RAG is a extra production-friendly model of FAISS as a result of it presents persistent storage. As a substitute of conserving embeddings in reminiscence, Chroma shops them regionally, so even when I restart the app, the vector index remains to be there. Similar to in FAISS, I nonetheless have to manually describe what every desk does in enterprise phrases (in _get_business_context()<\/code>). I embed my schema descriptions and retailer them in ChromaDB. Upon initialization, sentence-transformer (MiniLM) is loaded. If the vector already exists, it’s loaded. If not, I extract data + descriptions and name _populate_collection()<\/code> to generate and retailer vectors. This course of solely must be executed as soon as or when the schema adjustments.<\/p>\n It\u2019s quick, constant throughout classes, and straightforward to arrange. I selected it as a result of it\u2019s free, doesn\u2019t want exterior companies, and works properly for real-world use instances the place you wish to scale with out worrying about dropping the vector index or reprocessing every little thing each time.<\/p>\n class ChromaVectorRAG(BaseRAG):\n \"\"\"Chroma-based vector RAG utilizing sentence transformers with persistent storage.\"\"\"\n \n def __init__(self, db_path: str = DB_PATH):\n tremendous().__init__(db_path)\n self.mannequin = None\n self.chroma_client = None\n self.assortment = None\n self.table_info = {}\n self.table_names = []\n self._initialize()\n \n def _initialize(self):\n \"\"\"Initialize the Chroma vector retailer and embeddings.\"\"\"\n strive:\n import chromadb\n from sentence_transformers import SentenceTransformer\n \n print(\"\ud83d\udd04 Initializing Chroma Vector RAG...\")\n \n # Load embedding mannequin\n self.mannequin = SentenceTransformer('all-MiniLM-L6-v2')\n print(\"\u2705 Loaded embedding mannequin: all-MiniLM-L6-v2\")\n \n # Initialize Chroma shopper (persistent storage)\n self.chroma_client = chromadb.PersistentClient(path=\".\/knowledge\/chroma_db\")\n \n # Get or create assortment\n collection_name = \"schema_tables\"\n strive:\n self.assortment = self.chroma_client.get_collection(collection_name)\n print(\"\u2705 Loaded current Chroma assortment\")\n besides:\n # Create new assortment if it would not exist\n self.assortment = self.chroma_client.create_collection(\n identify=collection_name,\n metadata={\"description\": \"Database schema desk embeddings\"}\n )\n print(\"\u2705 Created new Chroma assortment\")\n \n # Extract desk data and create embeddings\n self.table_info = self._extract_table_info()\n self._populate_collection()\n \n # Load desk names for reference\n self._load_table_names()\n \n print(f\"\u2705 Chroma Vector RAG initialized with {len(self.table_names)} tables\")\n \n besides Exception as e:\n print(f\"\u274c Error initializing Chroma Vector RAG: {e}\")\n self.mannequin = None\n self.chroma_client = None\n self.assortment = None\n \n def _extract_table_info(self) -> Dict[str, Dict]:\n \"\"\"Extract detailed details about every desk.\"\"\"\n conn = sqlite3.join(self.db_path)\n cursor = conn.cursor()\n table_info = {}\n \n strive:\n # Get all desk names\n cursor.execute(\"SELECT identify FROM sqlite_master WHERE sort='desk';\")\n tables = cursor.fetchall()\n \n for (table_name,) in tables:\n data = {\n 'columns': [],\n 'foreign_keys': [],\n 'business_context': self._get_business_context(table_name)\n }\n \n # Get column data\n cursor.execute(f\"PRAGMA table_info({table_name});\")\n columns = cursor.fetchall()\n for col in columns:\n data['columns'].append({\n 'identify': col[1],\n 'sort': col[2],\n 'primary_key': bool(col[5])\n })\n \n # Get international key data\n cursor.execute(f\"PRAGMA foreign_key_list({table_name});\")\n fks = cursor.fetchall()\n for fk in fks:\n data['foreign_keys'].append({\n 'column': fk[3],\n 'references_table': fk[2],\n 'references_column': fk[4]\n })\n \n table_info[table_name] = data\n \n lastly:\n conn.shut()\n \n return table_info\n \n def _get_business_context(self, table_name: str) -> str:\n \"\"\"Get enterprise context description for tables.\"\"\"\n contexts = {\n 'merchandise': 'Product catalog with objects, costs, classes, and model data. Core stock knowledge.',\n 'product_variants': 'Product variations like colours, sizes, SKUs. Hyperlinks merchandise to particular sellable objects.',\n 'clients': 'Buyer profiles with private data, contact particulars, and account standing.',\n 'orders': 'Buy transactions with totals, dates, standing, and buyer relationships.',\n 'order_items': 'Particular person line objects inside orders. Comprises portions, costs, and product references.',\n 'funds': 'Fee processing data with strategies, quantities, and transaction standing.',\n 'stock': 'Inventory ranges and warehouse portions for product variants.',\n 'evaluations': 'Buyer suggestions, scores, and product evaluations.',\n 'suppliers': 'Vendor data for procurement and provide chain administration.',\n 'classes': 'Product categorization hierarchy for organizing catalog.',\n 'manufacturers': 'Model data for merchandise and advertising functions.',\n 'addresses': 'Buyer delivery and billing tackle data.',\n 'shipments': 'Supply monitoring and delivery standing data.',\n 'reductions': 'Promotional codes, coupons, and low cost campaigns.',\n 'warehouses': 'Storage facility places and warehouse administration.',\n 'workers': 'Employees data and organizational construction.',\n 'departments': 'Organizational divisions and crew construction.',\n 'product_images': 'Product pictures and media property.',\n 'purchase_orders': 'Procurement orders from suppliers.',\n 'purchase_order_items': 'Line objects for provider buy orders.',\n 'order_discounts': 'Utilized reductions and promotions on orders.',\n 'shipment_items': 'Particular person objects inside cargo packages.'\n }\n \n return contexts.get(table_name, f'Database desk for {table_name} associated operations.')\n \n def _populate_collection(self):\n \"\"\"Populate Chroma assortment with desk embeddings.\"\"\"\n if not self.assortment or not self.table_info:\n return\n \n paperwork = []\n metadatas = []\n ids = []\n \n for table_name, data in self.table_info.objects():\n # Create complete description\n description = self._create_table_description(table_name, data)\n \n paperwork.append(description)\n metadatas.append({\n 'table_name': table_name,\n 'column_count': len(data['columns']),\n 'has_foreign_keys': len(data['foreign_keys']) > 0,\n 'business_context': data['business_context']\n })\n ids.append(f\"table_{table_name}\")\n \n # Add to assortment\n self.assortment.add(\n paperwork=paperwork,\n metadatas=metadatas,\n ids=ids\n )\n \n print(f\"\u2705 Added {len(paperwork)} desk embeddings to Chroma assortment\")\n \n def _create_table_description(self, table_name: str, data: Dict) -> str:\n \"\"\"Create a complete description for embedding.\"\"\"\n description = f\"Desk: {table_name}n\"\n description += f\"Objective: {data['business_context']}n\"\n \n # Add column data\n description += \"Columns: \"\n col_names = [col['name'] for col in data['columns']]\n description += \", \".be a part of(col_names) + \"n\"\n \n # Add relationship data\n if data['foreign_keys']:\n description += \"Relationships: \"\n relationships = []\n for fk in data['foreign_keys']:\n relationships.append(f\"hyperlinks to {fk['references_table']} by way of {fk['column']}\")\n description += \"; \".be a part of(relationships) + \"n\"\n \n # Add widespread use instances\n use_cases = self._get_use_cases(table_name)\n if use_cases:\n description += f\"Widespread queries: {use_cases}\"\n \n return description\n \n def _get_use_cases(self, table_name: str) -> str:\n \"\"\"Get widespread use instances for every desk.\"\"\"\n use_cases = {\n 'merchandise': 'product searches, catalog listings, value queries, stock checks',\n 'clients': 'buyer lookup, registration evaluation, geographic distribution',\n 'orders': 'gross sales evaluation, income monitoring, order historical past, standing monitoring',\n 'order_items': 'product gross sales efficiency, income by product, order composition',\n 'funds': 'cost processing, income reconciliation, cost technique evaluation',\n 'manufacturers': 'model efficiency, gross sales by model, model comparability',\n 'classes': 'class evaluation, product group, catalog construction'\n }\n \n return use_cases.get(table_name, 'common knowledge queries and evaluation')\n \n def _load_table_names(self):\n \"\"\"Load desk names from the gathering.\"\"\"\n if not self.assortment:\n return\n \n strive:\n # Get all objects from assortment\n outcomes = self.assortment.get()\n self.table_names = [metadata['table_name'] for metadata in outcomes['metadatas']]\n besides Exception as e:\n print(f\"\u26a0\ufe0f Couldn't load desk names from Chroma: {e}\")\n self.table_names = []\n \n def get_relevant_schema(self, user_query: str, max_tables: int = 5) -> str:\n \"\"\"Get related schema utilizing Chroma vector similarity search.\"\"\"\n if not self.assortment:\n print(\"\u26a0\ufe0f Chroma not initialized, falling again to full schema\")\n return get_structured_schema(self.db_path)\n \n strive:\n # Seek for related tables\n outcomes = self.assortment.question(\n query_texts=[user_query],\n n_results=max_tables\n )\n \n # Extract related desk names\n relevant_tables = []\n if outcomes['metadatas'] and len(outcomes['metadatas']) > 0:\n for metadata in outcomes['metadatas'][0]:\n relevant_tables.append(metadata['table_name'])\n \n # Fallback if no related tables discovered\n if not relevant_tables:\n print(\"\u26a0\ufe0f No related tables discovered, utilizing defaults\")\n relevant_tables = self._get_default_tables(user_query)[:max_tables]\n \n # Construct schema for chosen tables\n return self._build_schema(relevant_tables)\n \n besides Exception as e:\n print(f\"\u26a0\ufe0f Chroma search failed: {e}, falling again to full schema\")\n return get_structured_schema(self.db_path)\n \n def _get_default_tables(self, user_query: str) -> Listing[str]:\n \"\"\"Get default tables primarily based on question patterns.\"\"\"\n query_lower = user_query.decrease()\n \n if any(phrase in query_lower for phrase in ['revenue', 'sales', 'total', 'amount', 'brand']):\n return ['orders', 'order_items', 'product_variants', 'products', 'brands']\n elif any(phrase in query_lower for phrase in ['product', 'item', 'catalog']):\n return ['products', 'product_variants', 'categories', 'brands']\n elif any(phrase in query_lower for phrase in ['customer', 'user', 'buyer']):\n return ['customers', 'orders', 'addresses']\n else:\n return ['products', 'customers', 'orders', 'order_items']\n \n def _build_schema(self, table_names: Listing[str]) -> str:\n \"\"\"Construct schema string for specified tables.\"\"\"\n if not table_names:\n return get_structured_schema(self.db_path)\n \n conn = sqlite3.join(self.db_path)\n cursor = conn.cursor()\n schema_lines = [\"Available tables and columns:\"]\n \n strive:\n for table_name in table_names:\n cursor.execute(f\"PRAGMA table_info({table_name});\")\n columns = cursor.fetchall()\n if columns:\n col_names = [col[1] for col in columns]\n schema_lines.append(f\"- {table_name}: {', '.be a part of(col_names)}\")\n lastly:\n conn.shut()\n \n return 'n'.be a part of(schema_lines)\n \n def get_approach_info(self) -> Dict[str, Any]:\n return {\n \"identify\": \"Chroma Vector RAG\",\n \"description\": \"Makes use of Chroma DB for persistent vector storage with semantic search\",\n \"professionals\": [\"Persistent storage\", \"Fast queries\", \"Scalable\", \"Easy management\"],\n \"cons\": [\"Requires disk space\", \"Initial setup time\", \"Additional dependency\"],\n \"best_for\": \"Manufacturing environments, persistent workflows, crew collaboration\"\n }\n<\/code><\/pre>\nEvaluating the completely different RAG Methods<\/h2>\nThis RAGManager<\/code> class is the management heart for switching between completely different RAG methods. Primarily based on the person question, it picks the proper strategy, fetches probably the most related a part of the schema, and tracks efficiency like response time, token financial savings, and desk depend. It additionally has a examine perform to benchmark all RAGs side-by-side, and shops historic metrics so you possibly can analyze how each is doing over time. Tremendous helpful for testing what works greatest and conserving issues optimized.<\/p>\n All of the completely different RAG technique courses are initialized and saved in self.approaches<\/code>. Every RAG strategy is a category that inherits from BaseRAG<\/code>, so all of them have a constant interface (get_relevant_schema()<\/code> and get_approach_info()<\/code>). This implies you possibly can simply plug in a brand new technique (say Pinecone or Weaviate) so long as it extends BaseRAG<\/code>.<\/p>\n The tactic get_relevant_schema()<\/code> returns the schema related to that question primarily based on the chosen technique. If an invalid technique is handed or there\u2019s a failure for some motive, it neatly falls again to the 'Key phrase RAG'<\/code> technique.<\/p>\n The tactic compare_approaches()<\/code> runs the identical question via all of the RAG methods. It measures:<\/p>\n \nSize of ensuing schema<\/li>\n % Token discount vs full schema<\/li>\n Response time<\/li>\n Variety of tables returned<\/li>\n<\/ul>\nThat is actually helpful to benchmark methods<\/strong> side-by-side and choose the one greatest suited on your use case.<\/p>\n class RAGManager:\n \"\"\"Supervisor for a number of RAG approaches.\"\"\"\n \n def __init__(self, db_path: str = DB_PATH):\n self.db_path = db_path\n self.approaches = {\n 'no_rag': NoRAG(db_path),\n 'key phrase': KeywordRAG(db_path),\n 'faiss': FAISSVectorRAG(db_path),\n 'chroma': ChromaVectorRAG(db_path)\n }\n self.performance_metrics = {}\n \n def get_available_approaches(self) -> Dict[str, Dict[str, Any]]:\n \"\"\"Get details about all accessible RAG approaches.\"\"\"\n return {\n approach_id: strategy.get_approach_info() \n for approach_id, strategy in self.approaches.objects()\n }\n \n def get_relevant_schema(self, user_query: str, strategy: str = 'key phrase', max_tables: int = 5) -> str:\n \"\"\"Get related schema utilizing specified strategy.\"\"\"\n if strategy not in self.approaches:\n print(f\"\u26a0\ufe0f Unknown strategy '{strategy}', falling again to key phrase\")\n strategy = 'key phrase'\n \n start_time = time.time()\n \n strive:\n schema = self.approaches[approach].get_relevant_schema(user_query, max_tables)\n \n # Report efficiency metrics\n end_time = time.time()\n self._record_performance(strategy, user_query, schema, end_time - start_time)\n \n return schema\n \n besides Exception as e:\n print(f\"\u26a0\ufe0f Error with {strategy} strategy: {e}\")\n # Fallback to key phrase strategy\n if strategy != 'key phrase':\n return self.get_relevant_schema(user_query, 'key phrase', max_tables)\n else:\n return get_structured_schema(self.db_path)\n \n def compare_approaches(self, user_query: str, max_tables: int = 5) -> Dict[str, Any]:\n \"\"\"Examine all approaches for a given question.\"\"\"\n outcomes = {}\n full_schema = get_structured_schema(self.db_path)\n full_schema_length = len(full_schema)\n \n for approach_id, strategy in self.approaches.objects():\n start_time = time.time()\n \n strive:\n schema = strategy.get_relevant_schema(user_query, max_tables)\n end_time = time.time()\n \n outcomes[approach_id] = {\n 'schema': schema,\n 'schema_length': len(schema),\n 'token_reduction': ((full_schema_length - len(schema)) \/ full_schema_length) * 100,\n 'response_time': end_time - start_time,\n 'table_count': len([line for line in schema.split('n') if line.startswith('- ')]),\n 'success': True\n }\n \n besides Exception as e:\n outcomes[approach_id] = {\n 'schema': '',\n 'schema_length': 0,\n 'token_reduction': 0,\n 'response_time': 0,\n 'table_count': 0,\n 'success': False,\n 'error': str(e)\n }\n \n return outcomes\n \n def _record_performance(self, strategy: str, question: str, schema: str, response_time: float):\n \"\"\"Report efficiency metrics for evaluation.\"\"\"\n if strategy not in self.performance_metrics:\n self.performance_metrics[approach] = []\n \n full_schema_length = len(get_structured_schema(self.db_path))\n schema_length = len(schema)\n \n metrics = {\n 'question': question,\n 'schema_length': schema_length,\n 'token_reduction': ((full_schema_length - schema_length) \/ full_schema_length) * 100,\n 'response_time': response_time,\n 'table_count': len([line for line in schema.split('n') if line.startswith('- ')]),\n 'timestamp': time.time()\n }\n \n self.performance_metrics[approach].append(metrics)\n \n def get_performance_summary(self) -> Dict[str, Any]:\n \"\"\"Get efficiency abstract for all approaches.\"\"\"\n abstract = {}\n \n for strategy, metrics_list in self.performance_metrics.objects():\n if not metrics_list:\n proceed\n \n avg_token_reduction = sum(m['token_reduction'] for m in metrics_list) \/ len(metrics_list)\n avg_response_time = sum(m['response_time'] for m in metrics_list) \/ len(metrics_list)\n avg_table_count = sum(m['table_count'] for m in metrics_list) \/ len(metrics_list)\n \n abstract[approach] = {\n 'queries_processed': len(metrics_list),\n 'avg_token_reduction': spherical(avg_token_reduction, 1),\n 'avg_response_time': spherical(avg_response_time, 3),\n 'avg_table_count': spherical(avg_table_count, 1)\n }\n \n return abstract\n\n\n# Comfort capabilities for backward compatibility\ndef get_rag_enhanced_schema(user_query: str, db_path: str = DB_PATH, strategy: str = 'key phrase') -> str:\n \"\"\"Get RAG-enhanced schema utilizing specified strategy.\"\"\"\n supervisor = RAGManager(db_path)\n return supervisor.get_relevant_schema(user_query, strategy)\n\n\n# International cached occasion\n_rag_manager_instance = None\n\ndef get_cached_rag_manager(db_path: str = DB_PATH) -> RAGManager:\n \"\"\"Get cached RAG supervisor occasion.\"\"\"\n international _rag_manager_instance\n if _rag_manager_instance is None:\n _rag_manager_instance = RAGManager(db_path)\n return _rag_manager_instance<\/code><\/pre>\nThe Streamlit app is totally built-in with this supervisor, so customers can select the technique they need and see real-time outcomes. You possibly can try the entire code on GitHub right here<\/a>. Right here\u2019s a working demo of the brand new App in motion:<\/p>\n RAG Implementation In Motion, by Writer<\/figcaption><\/figure>\nUltimate Ideas<\/h2>\nThis isn’t the top; there’s nonetheless lots to enhance. I have to stress take a look at towards quite a lot of assaults and reinforce guardrails to scale back hallucinations and guarantee knowledge security. Will probably be good to construct a role-based entry system for knowledge governance. Possibly changing Streamlit with a frontend framework like React may make the app extra scalable for real-world deployments. All this, for subsequent time.<\/p>\n \nEarlier than you go\u2026<\/strong><\/p>\nComply with me so that you don\u2019t miss any new posts I write in future; you will see extra of my articles on\u00a0my\u00a0profile web page<\/a>.\u00a0You may also join with me on\u00a0<\/em>LinkedIn<\/em><\/a>\u00a0or\u00a0<\/em>X<\/em><\/a>!<\/em><\/p>\n<\/div>\n\n","protected":false},"excerpt":{"rendered":" in my weblog publish, I had explored the best way to construct a Multi-agent SQL assistant utilizing CrewAI & Streamlit. The person may question a SQLite database in pure language. The AI brokers would generate a SQL question primarily based on person enter, overview it, and verify for compliance earlier than operating it towards the […]<\/p>\n","protected":false},"author":2,"featured_media":8453,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[55],"tags":[122,475,1075,1287,668,1729,1808],"class_list":["post-8451","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-machine-learning","tag-assistant","tag-building","tag-manager","tag-multiagent","tag-part","tag-rag","tag-sql"],"_links":{"self":[{"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=\/wp\/v2\/posts\/8451","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=8451"}],"version-history":[{"count":1,"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=\/wp\/v2\/posts\/8451\/revisions"}],"predecessor-version":[{"id":8452,"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=\/wp\/v2\/posts\/8451\/revisions\/8452"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=\/wp\/v2\/media\/8453"}],"wp:attachment":[{"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=8451"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=8451"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/techtrendfeed.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=8451"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}

No RAG Technique<\/h2>\nThat is principally the identical strategy that I used beforehand, the place I despatched in the complete database schema as context to the LLM with none filtering or optimization. This strategy is greatest for very small schemas (ideally lower than 10 tables).<\/p>\n