🗺 Overview

🏗 Architecture

⚙️ Build It

📋 All Prompts

🧠 Knowledge Check

🚀 Deploy

🤖 Claude Skill Engineering Bootcamp · Season 02

BUILD YOUR
RAG SQL AGENT

From zero to a production AI agent that speaks natural language, generates SQL, self-corrects errors, and answers questions over your data — all inside VS Code.

Build Phases

Files Built

1000

XP to Earn

60s

To First Query

WHAT YOU'LL BUILD

A fully operational RAG SQL Agent system — the same architecture used in enterprise data intelligence tools.

⚙️

Phase 1 — Project Setup

Create the folder structure, install dependencies, configure your OpenRouter API key. The foundation everything else rests on.

1 Prompt · 50 XP

📊

Phase 2 — Data Layer

Build the multi-format file loader that reads CSV, Excel, JSON, Parquet and converts them into DuckDB-queryable DataFrames.

3 Prompts · 100 XP

🧠

Phase 3 — RAG Layer

Build schema intelligence: extract column profiles, embed them into ChromaDB vector store, and retrieve context for SQL generation.

4 Prompts · 150 XP

🤖

Phase 4 — Agent Core

Build the SQL agent: intent classification, LLM-powered SQL generation, self-correcting execution loop, and result interpretation.

6 Prompts · 250 XP

💻

Phase 5 — Streamlit UI

Build the professional web interface: sidebar controls, chat window, data preview panel, SQL display, and results visualization.

1 Prompt · 150 XP

🚀

Phase 6 — QA & Deploy

Run the 12-point quality gate, test all code paths, verify self-repair loops, and launch the agent in your browser.

2 Prompts · 100 XP

THE TECH STACK

Every tool is pinned to an exact version. No "latest" — that breaks production systems.

AI & API

openai==1.51.0
chromadb==0.5.15
sentence-transformers==3.2.1

Data & SQL

duckdb==1.1.1
pandas==2.2.3
sqlglot==25.23.2

UI & Utils

streamlit==1.40.0
plotly==5.24.1
python-dotenv==1.0.1

Reliability

pydantic==2.9.2
tenacity==9.0.0
rich==13.9.4

⚠️ Prerequisites: You need Python 3.10+, VS Code (or any terminal), and an OpenRouter API key (free at openrouter.ai). That's it. No databases, no servers, no cloud accounts required.

SYSTEM ARCHITECTURE

Understand the blueprint before you build. Every prompt in Phase 3 maps to one of these layers.

┌─────────────────────────────────────────────────────────────────┐ │ STREAMLIT UI (app.py) │ │ ┌────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │ │ Sidebar │ │ Chat Interface │ │ Data Preview │ │ │ │ · API Key │ │ · User Input │ │ · DataFrame │ │ │ │ · Model │ │ · AI Response │ │ · Schema Info │ │ │ │ · File Upload │ │ · SQL Display │ │ · Statistics │ │ │ └────────────────┘ └─────────────────┘ └─────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ SQL AGENT CORE │ │ (agent/sql_agent.py) │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │ │ │ INTENT │ │ RAG │ │ SQL GENERATION │ │ │ │ PARSER │─▶│ RETRIEVER │─▶│ (OpenRouter LLM) │ │ │ │ │ │ (ChromaDB) │ │ │ │ │ └──────────────┘ └──────────────┘ └──────────────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ SQL EXECUTION LOOP │ │ │ │ ┌────────────┐ error ┌───────────┐ 3x retries │ │ │ │ │ DuckDB │────────▶│ SQL REPAIR│──────────────▶ │ │ │ │ │ Execute │◀────────│(OpenRouter│ graceful fail │ │ │ │ └────────────┘ fixed └───────────┘ │ │ │ └──────────────────────────────────────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ RESULT INTERPRETER (OpenRouter LLM) │ │ │ │ DataFrame → Natural Language Analysis + Insights │ │ │ └──────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ │ │ │ ▼ ▼ ▼ ┌──────────────┐ ┌──────────────────┐ ┌───────────────────┐ │ DATA LOADER │ │ SCHEMA EXTRACTOR │ │ VECTOR STORE │ │ (pandas + │ │ (auto-profile │ │ (ChromaDB + │ │ multi-format│ │ all columns) │ │ sentence- │ │ ingestion) │ │ │ │ transformers) │ └──────────────┘ └──────────────────┘ └───────────────────┘ │ │ │ └────────────────────┴────────────────────┘ │ ▼ ┌──────────────────┐ │ DuckDB Memory │ │ (in-process, │ │ no server) │ └──────────────────┘

DATA FLOW

How a question becomes an answer — 7 steps every query travels.

💬

1. User asks

Natural language

🎯

2. Intent

Classify query type

🔍

3. RAG Retrieve

Get schema context

⚡

4. Generate SQL

LLM + schema

🦆

5. Execute

DuckDB in-memory

🔄

6. Self-repair

Auto-fix errors

✨

7. Interpret

Natural language

FILE STRUCTURE

22 files across 5 modules. Each prompt in the Build phase creates one module completely.

📁 rag_sql_agent/22 files · 5 modules

📄 app.pyStreamlit entry point — only file the user runs

📄 requirements.txtExact pinned versions

📄 .env.exampleAPI key template

📄 README.mdProject overview

🤖agent/6 files

__init__.pyExports: SQLAgent

sql_agent.pyCore orchestration

sql_generator.pyLLM SQL generation

sql_executor.pyDuckDB + self-correction

result_interpreter.pyLLM result analysis

intent_classifier.pyQuery type classifier

🧠rag/4 files

__init__.pyExports: SchemaRAG

schema_extractor.pyColumn profiler

vector_store.pyChromaDB wrapper

context_builder.pyLLM context assembler

📁data/2 files

__init__.pyExports: DataLoader

loader.pyMulti-format file loader

⚙️config/2 files

__init__.pyExports: Settings

settings.pyPydantic config

🔧utils/4 files

__init__.py

sql_validator.pysqlglot syntax check

formatters.pyDataFrame → markdown

session_state.pyStreamlit state helpers

BUILD YOUR AGENT

Copy each prompt, paste it into VS Code (Cursor, Windsurf, or Claude Code), and let AI build each layer. Complete each phase to unlock the next.

⚙️Setup

📊Config

📁Data

🧠RAG

🤖Agent

💻UI

🚀Deploy

📍 Where to paste: Open VS Code → open your terminal (Ctrl+` ) → open a new Claude or Cursor chat window → paste the prompt below. The AI will generate commands you run in the terminal.

📁 Create Project & Install Dependencies

Bootstrap the entire project skeleton and install all pinned packages

+50 XP

PHASE 1 PROMPT — Paste into Claude / Cursor / Windsurf Chat

You are a senior Python project scaffolding engineer. TASK: Create a complete project setup for a RAG SQL Agent system. EXECUTE these shell commands in sequence: 1. Create the project structure: mkdir rag_sql_agent cd rag_sql_agent mkdir -p agent rag data config utils 2. Create requirements.txt with EXACTLY these pinned versions: openai==1.51.0 chromadb==0.5.15 sentence-transformers==3.2.1 duckdb==1.1.1 pandas==2.2.3 openpyxl==3.1.5 xlrd==2.0.1 pyarrow==17.0.0 sqlglot==25.23.2 streamlit==1.40.0 plotly==5.24.1 python-dotenv==1.0.1 pydantic==2.9.2 rich==13.9.4 tenacity==9.0.0 pytest==8.3.3 3. Create .env.example file with content: OPENROUTER_API_KEY=sk-or-v1-your-key-here DEFAULT_MODEL=openai/gpt-4o-mini MAX_RETRIES=3 4. Install all packages: pip install -r requirements.txt 5. Create empty __init__.py files for all modules: touch agent/__init__.py rag/__init__.py data/__init__.py config/__init__.py utils/__init__.py 6. Confirm setup by listing the directory structure. OUTPUT: Show every command, its output, and confirm all packages installed successfully.

✅ Expected result: A rag_sql_agent/ folder with 5 subfolders, requirements.txt, .env.example, and all packages installed. The folder structure should match the architecture diagram exactly.

🔧 Build config/settings.py + utils/ (3 files)

Pydantic settings, SQL validator, formatters, and session state helpers

+100 XP

Build order matters. Config is imported by everything. Utils is imported by agent and RAG. Always build bottom-up: config → utils → data → rag → agent → app.py

PHASE 2 PROMPT — Config & Utils Layer

You are a Principal AI Systems Engineer building a RAG SQL Agent. TASK: Build the config and utils layer. Create these 5 files COMPLETELY — no stubs, no TODOs. FILE 1: config/settings.py """Pydantic settings loaded from .env file""" Requirements: - Use pydantic-settings BaseSettings - Load from .env using python-dotenv - Fields: openrouter_api_key (str), default_model (str, default "openai/gpt-4o-mini"), max_retries (int, default 3), max_sql_tokens (int, default 2000), embedding_model (str, default "all-MiniLM-L6-v2"), chroma_collection (str, default "schema_rag"), app_title (str, default "RAG SQL Agent"), debug_mode (bool, default False) - Export: Settings class + get_settings() cached function FILE 2: config/__init__.py Requirements: - Export: Settings, get_settings FILE 3: utils/sql_validator.py """SQL syntax validation using sqlglot before execution""" Requirements: - validate_sql(sql: str) → tuple[bool, str]: parse with sqlglot dialect="duckdb", return (True,"") or (False, error_message) - extract_table_names(sql: str) → list[str]: return all table names referenced in the SQL - sanitize_sql(sql: str) → str: strip markdown code fences, strip leading/trailing whitespace FILE 4: utils/formatters.py """DataFrame to display-ready formats""" Requirements: - df_to_markdown(df: pd.DataFrame, max_rows: int = 20) → str: truncate to max_rows, format as markdown table - format_result_summary(df: pd.DataFrame) → str: "X rows × Y columns. Columns: col1, col2..." - truncate_for_llm(text: str, max_chars: int = 3000) → str: trim to max_chars with "...[truncated]" suffix FILE 5: utils/session_state.py """Streamlit session state management — all state reads/writes go here""" Requirements: - SessionState class with classmethods: - init_defaults(): set all default keys (messages=[], dataframes={}, agent=None, etc.) - get(key, default=None): safe getter - set(key, value): setter - add_message(role, content, sql=None, df=None): append to messages list - clear_chat(): reset messages to [] FILE 6: utils/__init__.py - Export: validate_sql, sanitize_sql, extract_table_names, df_to_markdown, format_result_summary, truncate_for_llm, SessionState RULE: Use exact library versions from requirements.txt. No external imports beyond what is listed there. Write complete implementations — not just function signatures.

📊 Build data/loader.py — Multi-Format Ingestion

CSV, Excel, JSON, Parquet, TSV → pandas DataFrame → DuckDB registration

+100 XP

🦆 DuckDB trick: DuckDB doesn't have a persistent database here — it's in-process. You register a pandas DataFrame as a named view using conn.register("table_name", df). Always drop the view first: conn.execute("DROP VIEW IF EXISTS table_name")

PHASE 3 PROMPT — Data Layer

You are a Principal AI Systems Engineer building a RAG SQL Agent. TASK: Build the data ingestion layer. Create 2 files COMPLETELY. FILE 1: data/loader.py """Multi-format data loader → pandas DataFrame + DuckDB registration""" Create a DataLoader class with: __init__(self): - self.conn = duckdb.connect() # in-memory, no file path - self.loaded_tables: dict[str, pd.DataFrame] = {} load_file(self, file_obj, filename: str) → tuple[str, pd.DataFrame]: - Detect format from filename extension (.csv, .tsv, .xlsx, .xls, .json, .parquet) - Use correct pandas reader for each format: - .csv → pd.read_csv(file_obj) - .tsv → pd.read_csv(file_obj, sep='\t') - .xlsx → pd.read_excel(file_obj, engine='openpyxl') - .xls → pd.read_excel(file_obj, engine='xlrd') - .json → pd.read_json(file_obj) - .parquet → pd.read_parquet(file_obj) - Sanitize table name: filename without extension, replace non-alphanumeric with _, lowercase - Register in DuckDB: self.conn.execute(f"DROP VIEW IF EXISTS {table_name}") self.conn.register(table_name, df) - Store in self.loaded_tables[table_name] = df - Return (table_name, df) get_table_names(self) → list[str]: - Return list(self.loaded_tables.keys()) execute_query(self, sql: str) → pd.DataFrame: - Execute via self.conn.execute(sql).df() - Raise descriptive ValueError on DuckDB error (include original error message) get_dataframe(self, table_name: str) → pd.DataFrame | None: - Return self.loaded_tables.get(table_name) FILE 2: data/__init__.py - Export: DataLoader CRITICAL RULES: - NEVER use duckdb.connect(':memory:') — use duckdb.connect() with no args - Always DROP VIEW before registering to avoid "already exists" error on re-upload - Wrap the entire load_file body in try/except — return descriptive error messages - Column names with spaces: wrap in double quotes in SQL. DataLoader must NOT rename columns. - For JSON: handle both records format and nested — use pd.json_normalize if needed

🧠 Build rag/ — Schema Extractor + Vector Store + Context Builder

ChromaDB embeddings + sentence-transformers for local, no-API schema retrieval

+150 XP

🧠 Why RAG for SQL? When a user asks "what's the average sale per region?", the agent needs to know your table is called sales_data and the columns are region and sale_amount. RAG retrieves this schema context without stuffing the entire schema into every LLM call — it only retrieves the relevant parts.

PHASE 4 PROMPT — RAG Layer (3 files)

You are a Principal AI Systems Engineer building a RAG SQL Agent. TASK: Build the complete RAG layer — 4 files, fully implemented, no stubs. FILE 1: rag/schema_extractor.py """Extract rich schema intelligence from pandas DataFrames""" Create SchemaExtractor class: extract(self, df: pd.DataFrame, table_name: str) → list[dict]: - For each column in df: - dtype = str(df[col].dtype) - n_unique = int(df[col].nunique()) - null_pct = round(df[col].isnull().mean() * 100, 1) - sample_vals: top 5 most frequent values as strings (for categoricals) - If numeric: also capture min, max, mean (rounded to 2 decimals) - Build a natural language description: f"Column '{col}' in table '{table_name}': type={dtype}, unique={n_unique}, nulls={null_pct}%. Sample: {sample_vals}" - Return as dict: {table_name, column_name, dtype, description, n_unique, null_pct} - Return list of these dicts (one per column) extract_table_summary(self, df: pd.DataFrame, table_name: str) → str: - Return a one-paragraph natural language summary of the whole table: f"Table '{table_name}' has {len(df)} rows and {len(df.columns)} columns: {', '.join(df.columns.tolist())}" FILE 2: rag/vector_store.py """ChromaDB wrapper with sentence-transformers embeddings — runs fully local""" Create SchemaVectorStore class: __init__(self, collection_name: str = "schema_rag"): - self.client = chromadb.Client() # in-memory, no persistence needed - self.embedding_model = SentenceTransformer("all-MiniLM-L6-v2") - Try to get or create collection. On error, recreate. - self.collection = self.client.get_or_create_collection(name=collection_name) add_schema(self, schema_docs: list[dict]) → None: - For each doc in schema_docs: - id = f"{doc['table_name']}__{doc['column_name']}" - embedding = self.embedding_model.encode(doc['description']).tolist() - Add to self.collection with that embedding, id, and doc as metadata - Use collection.upsert() not add(), to allow re-indexing on file re-upload query(self, question: str, n_results: int = 8) → list[dict]: - Embed question: self.embedding_model.encode(question).tolist() - Query collection: self.collection.query(query_embeddings=[...], n_results=n_results) - Return list of metadata dicts from results['metadatas'][0] clear(self) → None: - Delete and recreate the collection (call on new file upload) FILE 3: rag/context_builder.py """Assemble retrieved schema metadata into LLM-ready context string""" Create ContextBuilder class: build(self, retrieved_docs: list[dict], table_names: list[str], table_summaries: dict[str,str]) → str: - Build context string with sections: SECTION 1: "AVAILABLE TABLES:" — list each table name SECTION 2: "TABLE SUMMARIES:" — one paragraph per table SECTION 3: "RELEVANT SCHEMA DETAILS:" — one line per retrieved doc (the description field) - Max 2500 chars total — truncate SECTION 3 if needed - Return the complete context string FILE 4: rag/__init__.py - Export: SchemaExtractor, SchemaVectorStore, ContextBuilder CRITICAL RULES: - SentenceTransformer("all-MiniLM-L6-v2") will download ~80MB on first run — this is EXPECTED - chromadb.Client() is in-memory. Do NOT use PersistentClient or file paths. - Upsert, never add — prevents duplicate ID errors on re-upload - encode() returns numpy array — call .tolist() before passing to chroma - Wrap all chroma operations in try/except with clear error messages

🎯 Build agent/intent_classifier.py + sql_generator.py

Intent routing (data query vs meta vs greeting) + LLM SQL generation via OpenRouter

+80 XP

🎯 Why intent classification? Not every question is a SQL query. "What can you do?" is a meta question. "Hello!" is a greeting. "How many customers are in Europe?" is a data query. The intent classifier routes each message to the right handler without wasting tokens generating SQL for non-data questions.

PHASE 5A PROMPT — Intent Classifier + SQL Generator

You are a Principal AI Systems Engineer building a RAG SQL Agent. TASK: Build 2 agent layer files completely. FILE 1: agent/intent_classifier.py Create IntentClassifier class with: IntentType enum: DATA_QUERY | META_QUESTION | GREETING | UNSUPPORTED classify(self, question: str, has_data_loaded: bool) → IntentType: - If not has_data_loaded and question doesn't mention uploading: return META_QUESTION - Greeting patterns (hi, hello, hey, thanks, bye): return GREETING - Meta patterns (what can you do, how does this work, help, capabilities): return META_QUESTION - Data keywords (how many, average, total, sum, count, max, min, list, show, find, which, top, bottom, compare, trend, group by, filter, where, rank): DATA_QUERY - SQL keywords (select, from, where, join, having, order by): DATA_QUERY - Default: DATA_QUERY if data is loaded, else META_QUESTION handle_non_data(self, intent: IntentType, has_data: bool) → str: - GREETING: "Hello! I'm your RAG SQL Agent. Upload a CSV, Excel, JSON or Parquet file to get started." - META_QUESTION (no data): "I can answer questions about your data using SQL. Please upload a file first." - META_QUESTION (data loaded): "I analyze your uploaded data using SQL. Try asking 'show me the top 10 rows' or 'what is the average [column] by [group]?'" - UNSUPPORTED: "I can only answer questions about your uploaded data files." FILE 2: agent/sql_generator.py Create SQLGenerator class with: __init__(self, settings: Settings): - self.client = OpenAI(api_key=settings.openrouter_api_key, base_url="https://openrouter.ai/api/v1") - self.model = settings.default_model - self.extra_headers = {"HTTP-Referer": "https://asjprompts.in", "X-Title": "RAG SQL Agent"} generate(self, question: str, schema_context: str, table_names: list[str]) → str: System prompt: "You are an expert DuckDB SQL generator. Given schema context and a question, generate ONLY a syntactically correct DuckDB SQL query. Return ONLY the SQL — no explanation, no markdown, no code fences. Use exact column and table names from the schema." User prompt: f"Schema context:\n{schema_context}\n\nAvailable tables: {table_names}\n\nQuestion: {question}\n\nSQL:" - Call client.chat.completions.create(model=self.model, messages=[...], max_tokens=500, temperature=0, extra_headers=self.extra_headers) - Return response.choices[0].message.content.strip() - Wrap in try/except — raise descriptive error with "SQL generation failed: {e}" repair(self, broken_sql: str, error_message: str, schema_context: str) → str: System prompt: "You are a DuckDB SQL repair specialist. You receive broken SQL and its error. Return ONLY the corrected SQL — no explanation, no markdown." User prompt: f"Broken SQL:\n{broken_sql}\n\nError: {error_message}\n\nSchema context:\n{schema_context}\n\nCorrected SQL:" - Same API call pattern as generate() - Return repaired SQL string RULES: - Every client call must include extra_headers - Temperature=0 for SQL generation (deterministic output) - All exceptions wrapped — never let raw OpenAI errors surface to calling code

🔄 Build agent/sql_executor.py + result_interpreter.py

Self-correcting execution loop (3 retries) + natural language result narration

+100 XP

⚠️ The self-repair loop is critical. DuckDB errors happen when: column names have spaces, functions differ from standard SQL, or type casting fails. The executor catches the error message, sends it back to the LLM repair function, and retries up to 3 times. This is what separates a toy from a production agent.

PHASE 5B PROMPT — SQL Executor + Result Interpreter

You are a Principal AI Systems Engineer building a RAG SQL Agent. TASK: Build 2 more agent layer files completely. FILE 1: agent/sql_executor.py Create ExecutionResult dataclass: - success: bool - df: pd.DataFrame | None - sql_used: str - error: str | None - was_repaired: bool = False - attempts: int = 1 Create SQLExecutor class with: __init__(self, data_loader: DataLoader, sql_generator: SQLGenerator, max_retries: int = 3): - Store all three params execute_with_retry(self, sql: str, schema_context: str) → ExecutionResult: - First: validate SQL using validate_sql() from utils — if invalid, attempt repair immediately - Attempt 1: sanitize_sql(sql) then data_loader.execute_query(clean_sql) - On SUCCESS: return ExecutionResult(success=True, df=result_df, sql_used=clean_sql, ...) - On DuckDB ERROR: - If attempts < max_retries: - Call sql_generator.repair(broken_sql=current_sql, error_message=str(e), schema_context=schema_context) - Sanitize the repaired SQL - Retry execution with repaired SQL - Set was_repaired=True - If max_retries exhausted: return ExecutionResult(success=False, error=f"Failed after {max_retries} attempts. Last error: {str(e)}", ...) FILE 2: agent/result_interpreter.py Create ResultInterpreter class with: __init__(self, settings: Settings): - Same OpenAI client setup as SQLGenerator (openrouter base_url + extra_headers) interpret(self, question: str, df: pd.DataFrame, sql_used: str) → str: System prompt: "You are a data analyst. Given a question, the SQL that was run, and the results, provide a clear natural-language interpretation. Be specific — reference actual values from the results. If the result is empty, say so. Keep your interpretation under 150 words." User prompt: f"Question: {question}\n\nSQL executed:\n{sql_used}\n\nResults:\n{df_to_markdown(df, max_rows=15)}\n\nInterpretation:" - Call API (temperature=0.3 for slight creativity in narration) - Return interpretation string - On error: return f"Query returned {len(df)} rows. Review the results table above." RULES: - ExecutionResult must be a dataclass (from dataclasses import dataclass) - execute_with_retry must track attempt count in the return value - Repair loop must log each attempt via rich.console print (not visible in UI, only terminal) - Result interpreter gracefully degrades — NEVER raise an exception, always return a string

🤖 Build agent/sql_agent.py + agent/__init__.py

The main orchestrator that ties every component together into a single .query() call

+70 XP

🤖 The orchestrator pattern: sql_agent.py is a thin coordination layer. It doesn't do the heavy lifting — it delegates. Intent → IntentClassifier. Context → SchemaRAG. SQL → SQLGenerator. Execution → SQLExecutor. Narration → ResultInterpreter. The orchestrator's only job is sequencing.

PHASE 5C PROMPT — SQL Agent Orchestrator

You are a Principal AI Systems Engineer building a RAG SQL Agent. TASK: Build the main agent orchestrator — 2 files. FILE 1: agent/sql_agent.py Create AgentResponse dataclass: - success: bool - answer: str # natural language interpretation - sql: str | None = None # the SQL that was run - df: pd.DataFrame | None = None # raw results - was_repaired: bool = False - error: str | None = None Create SQLAgent class: __init__(self, settings: Settings, data_loader: DataLoader): - self.settings = settings - self.data_loader = data_loader - self.schema_extractor = SchemaExtractor() - self.vector_store = SchemaVectorStore(settings.chroma_collection) - self.context_builder = ContextBuilder() - self.intent_classifier = IntentClassifier() - self.sql_generator = SQLGenerator(settings) - self.sql_executor = SQLExecutor(data_loader, self.sql_generator, settings.max_retries) - self.result_interpreter = ResultInterpreter(settings) - self.table_summaries: dict[str, str] = {} index_table(self, df: pd.DataFrame, table_name: str) → None: - Call schema_extractor.extract(df, table_name) - Call schema_extractor.extract_table_summary(df, table_name) → store in self.table_summaries - Call vector_store.add_schema(schema_docs) - Log success with rich query(self, question: str) → AgentResponse: STEP 1 — Intent: classifier.classify(question, has_data_loaded=bool(data_loader.get_table_names())) - If not DATA_QUERY: return AgentResponse(success=True, answer=classifier.handle_non_data(...)) STEP 2 — RAG: vector_store.query(question, n_results=8) STEP 3 — Context: context_builder.build(retrieved_docs, table_names, table_summaries) STEP 4 — SQL Generation: sql_generator.generate(question, schema_context, table_names) STEP 5 — Execution: sql_executor.execute_with_retry(sql, schema_context) - If execution.success is False: return AgentResponse(success=False, error=execution.error, sql=execution.sql_used) STEP 6 — Interpretation: result_interpreter.interpret(question, execution.df, execution.sql_used) STEP 7 — Return: AgentResponse(success=True, answer=interpretation, sql=execution.sql_used, df=execution.df, was_repaired=execution.was_repaired) Wrap entire query() in try/except → return AgentResponse(success=False, error=str(e)) on any unexpected error. FILE 2: agent/__init__.py - Export: SQLAgent, AgentResponse CRITICAL: - agent/ must NEVER import from app.py - All exceptions caught within query() — nothing propagates raw to the UI - Log every step using rich.console print for debugging

💻 Build app.py — Complete Streamlit Interface

Sidebar controls, chat interface, SQL expander, results table, and data preview

+150 XP

⚠️ Streamlit session state: ALL reads/writes go through SessionState class. No direct st.session_state["key"] in app.py. This prevents state corruption across reruns.

PHASE 6 PROMPT — Streamlit UI (app.py)

You are a Principal AI Systems Engineer building a RAG SQL Agent. TASK: Build the complete Streamlit UI in app.py. This is the ONLY file the user runs. STRUCTURE OF app.py: SECTION 1 — Page Config: st.set_page_config(title="RAG SQL Agent", page_icon="🤖", layout="wide", initial_sidebar_state="expanded") SECTION 2 — Session Init: SessionState.init_defaults() called at top of every run SECTION 3 — Sidebar (st.sidebar): - Title: "⚙️ Configuration" - API Key input: st.text_input("OpenRouter API Key", type="password", key="api_key_input") - Model selector: st.selectbox with options: ["openai/gpt-4o-mini","openai/gpt-4o","anthropic/claude-3.5-sonnet","anthropic/claude-3-haiku","google/gemini-flash-1.5","meta-llama/llama-3.1-70b-instruct"] - File uploader: st.file_uploader("Upload Data File", type=["csv","xlsx","xls","json","parquet","tsv"], key="uploaded_file") - On new file upload: instantiate DataLoader, load file, instantiate SQLAgent from Settings, call agent.index_table(), store both in SessionState - Show loaded tables as expandable with row/column counts - "Clear Chat" button: SessionState.clear_chat() SECTION 4 — Main Area (2 columns: ratio 3:2): LEFT COLUMN — Chat Interface: - Header: "💬 Ask Your Data" - Display message history from SessionState.get("messages") - user messages: st.chat_message("user") with st.write(content) - assistant messages: st.chat_message("assistant") - st.write(msg.answer) - If msg.sql: st.expander("🔍 View SQL") → st.code(msg.sql, language="sql") - If msg.was_repaired: st.info("⚠️ SQL was auto-repaired on first attempt") - If msg.df is not None: st.dataframe(msg.df, use_container_width=True) - Chat input: prompt = st.chat_input("Ask a question about your data...") - On prompt submit: - SessionState.add_message("user", prompt) - Call SessionState.get("agent").query(prompt) → AgentResponse - SessionState.add_message("assistant", response) - st.rerun() RIGHT COLUMN — Data Preview: - Header: "📊 Data Overview" - For each loaded table: - st.subheader(table_name) - Show df.head(5) in st.dataframe - Show df.describe() in st.expander("📈 Statistics") - Show column info (name, dtype, nulls%) in st.expander("🗂 Schema") SECTION 5 — Guard Rails: - If no API key set: show st.warning and disable chat input - If no file loaded: show st.info("Upload a file to begin") - If agent query fails: show st.error(response.error) CRITICAL RULES: - type="password" on API key input — NEVER log full key, use key[:8]+"..." if needed - NO direct st.session_state["x"] access — only SessionState class methods - The API key from sidebar MUST update Settings before creating SQLAgent - st.rerun() after every message to refresh chat - Use st.spinner("Thinking...") wrapping the agent.query() call - Columns ratio: st.columns([3,2]) for chat vs data preview

✅ After this step: You have a complete, runnable RAG SQL Agent. The next phase is quality gates and launch.

🚀 Quality Gate Audit + Launch

Verify all 12 checkpoints before going live — then fire it up

+100 XP

PHASE 7 PROMPT — Quality Gate Audit

You are a Principal AI Systems Engineer auditing a RAG SQL Agent before launch. TASK: Run this 12-point quality gate audit across ALL files in the rag_sql_agent/ project. Report each item as [PASS], [FAIL — fix: ...], or [WARN — review: ...]. QUALITY GATE CHECKLIST: QG-01 ZERO STUBS: Search all .py files for "TODO", "FIXME", "pass #", "raise NotImplementedError", "...". None may exist as placeholders. Every function must be fully implemented. QG-02 IMPORT GRAPH: Verify no circular imports: - agent/ must NOT import from app.py - utils/ must NOT import from agent/ or rag/ - data/ must NOT import from agent/ QG-03 OPENROUTER CALLS: Every client.chat.completions.create() must have: - base_url = "https://openrouter.ai/api/v1" - extra_headers with HTTP-Referer and X-Title - try/except wrapping the call QG-04 DUCKDB REGISTRATION: Every conn.register() must be preceded by a matching DROP VIEW IF EXISTS. QG-05 ERROR PROPAGATION: Every exception in agent/ must be caught and converted to AgentResponse(success=False, ...). Nothing raw should reach app.py. QG-06 SESSION STATE: All session_state access in app.py goes through SessionState class. No bare st.session_state["key"] = value in app.py. QG-07 API KEY PRIVACY: API key uses type="password" in st.text_input. Key never appears in any log, print, or st.write call. QG-08 STARTUP TEST: Can the app start with: streamlit run app.py ? (Look for any code that runs at import time that could crash on startup.) QG-09 UPSERT NOT ADD: vector_store.py uses collection.upsert() not collection.add() for all schema documents. QG-10 REPAIR LOOP: sql_executor.py execute_with_retry(): - Calls sql_generator.repair() on error - Tracks attempt count - Returns was_repaired=True after successful repair QG-11 GRACEFUL EMPTY: If df is empty after a query, the interpreter returns a helpful message, not an error. QG-12 REQUIREMENTS COMPLETE: requirements.txt contains all imports used across all .py files. No undeclared dependency. OUTPUT FORMAT: For each item: [PASS/FAIL/WARN] QG-XX — short description If FAIL: provide the exact fix (code snippet or file + line guidance) End with: LAUNCH READINESS: READY / NOT READY + count of issues found

LAUNCH COMMAND — Run this in your VS Code terminal

# Step 1: Navigate to your project cd rag_sql_agent # Step 2: Copy your API key into .env cp .env.example .env # Then edit .env and paste your real OpenRouter API key # Step 3: Launch the agent streamlit run app.py # Expected output: You can now view your Streamlit app in your browser. Local URL: http://localhost:8501 Network URL: http://192.168.x.x:8501 # Step 4: In the browser — # 1. Paste your OpenRouter API key in the sidebar # 2. Upload a CSV or Excel file # 3. Ask: "How many rows are in this dataset?" # 4. Watch the agent generate SQL, run it, and explain the result

🚀 YOUR AGENT IS LIVE

If the browser opens and you can upload a file and ask a question — you built a production RAG SQL Agent from scratch using AI-powered prompts.

ALL PROMPTS

Every prompt in one place — organized by build layer. Use these to rebuild any component or extend the system.

All Prompts

Bonus Prompts

Debug Prompts

Quick Reference — All 8 Build Prompts

P1 Project Setup — scaffold structure, install packages 50 XP

PHASE 1 PROMPT — Paste into Claude / Cursor / Windsurf Chat

P2 Config + Utils — settings, SQL validator, formatters, session state 100 XP

PHASE 2 PROMPT — Config & Utils Layer

P3 Data Layer — multi-format loader + DuckDB registration 100 XP

PHASE 3 PROMPT — Data Layer

P4 RAG Layer — schema extractor, ChromaDB vector store, context builder 150 XP

PHASE 4 PROMPT — RAG Layer (3 files)

P5A Agent — Intent Classifier + SQL Generator (OpenRouter) 80 XP

PHASE 5A PROMPT — Intent Classifier + SQL Generator

P5B Agent — SQL Executor (self-repair loop) + Result Interpreter 100 XP

PHASE 5B PROMPT — SQL Executor + Result Interpreter

P5C Agent — SQLAgent Orchestrator + agent/__init__.py 70 XP

PHASE 5C PROMPT — SQL Agent Orchestrator

P6 Streamlit UI — complete app.py with chat + sidebar + data preview 150 XP

PHASE 6 PROMPT — Streamlit UI (app.py)

P7 QA Gate Audit — 12-point quality check + launch command 100 XP

PHASE 7 PROMPT — Quality Gate Audit

BONUS EXTENSIONS

After the agent is running, use these prompts to add advanced features.

BONUS 1 — Add Plotly Chart Auto-Generation

You are extending an existing RAG SQL Agent. TASK: Add automatic chart generation to agent/result_interpreter.py and app.py. In result_interpreter.py — add method: suggest_chart(self, df: pd.DataFrame, question: str) → dict | None: - Call LLM with prompt: "Given this question and DataFrame columns, suggest ONE Plotly chart. Respond with JSON only: {chart_type: 'bar'|'line'|'pie'|'scatter', x_col: str, y_col: str, title: str} If no chart is appropriate, respond with null." - Parse response JSON. Return dict or None on parse failure. In AgentResponse — add field: chart_suggestion: dict | None = None In sql_agent.py query() — after interpretation: - Call result_interpreter.suggest_chart(execution.df, question) - Add to AgentResponse In app.py — after showing dataframe: if msg.chart_suggestion: chart = msg.chart_suggestion fig = px.{chart['chart_type']}(msg.df, x=chart['x_col'], y=chart['y_col'], title=chart['title']) st.plotly_chart(fig, use_container_width=True) Make all changes non-breaking — the agent still works without chart suggestions.

BONUS 2 — Add Multi-Table JOIN Support

You are extending the RAG SQL Agent to support multi-table joins. TASK: Enhance rag/context_builder.py and agent/sql_generator.py for multi-table queries. In context_builder.py — add method: detect_join_opportunity(self, table_summaries: dict) → str: - Compare column names across tables - If shared column names exist (e.g., "customer_id" in both tables): return f"JOIN HINT: {table1}.{col} may link to {table2}.{col}" - Add this hint to the context string under "JOIN OPPORTUNITIES:" section In sql_generator.py — update generate() system prompt: Add: "When multiple tables are available and the question references data from more than one, use DuckDB JOIN syntax. Prefer INNER JOIN unless the question implies all records. Always qualify column names with table prefix in JOIN queries." Add to the user prompt: f"\nJoin hints from schema analysis:\n{join_hints}" The generator must now be passed join_hints from context_builder. Update sql_agent.py query() to compute and pass join hints.

BONUS 3 — Add Query History & Export

You are adding query history and CSV export to the RAG SQL Agent Streamlit UI. TASK: Add to app.py sidebar (below existing controls): QUERY HISTORY section: - Track last 10 successful queries in SessionState: list of {question, sql, row_count, timestamp} - Show as numbered list in st.expander("📜 Query History") - Each item: question text + "Re-run" button that inserts the SQL directly to executor EXPORT section: - For each message with a df result, add a download button: st.download_button("⬇ Export CSV", data=df.to_csv(index=False), file_name=f"result_{timestamp}.csv", mime="text/csv") In SessionState — add: add_to_history(question, sql, row_count): append to "query_history" list, cap at 10 items Make all additions non-breaking and use existing SessionState class pattern.

DEBUG PROMPTS

When things break — and they will — use these targeted prompts to diagnose and fix.

DEBUG 1 — ChromaDB Collection Error

I am getting this ChromaDB error in rag/vector_store.py: [PASTE THE EXACT ERROR HERE] Context: - chromadb version: 0.5.15 - I am using chromadb.Client() (in-memory) - Error occurs when: [on startup / after file upload / on query] TASK: 1. Diagnose the exact cause of this error 2. Show the corrected vector_store.py code for the failing method only 3. Explain why the fix works in the context of chromadb 0.5.15 API Do not rewrite the entire file — only the broken section.

DEBUG 2 — DuckDB SQL Error Not Auto-Repairing

My RAG SQL Agent SQL executor is failing to self-repair. Here is the situation: Error message: [PASTE DuckDB ERROR] SQL that failed: [PASTE SQL] Schema context available: [PASTE FIRST 500 CHARS OF SCHEMA] TASK: 1. Explain WHY DuckDB is throwing this specific error 2. Show what the repaired SQL should look like 3. If the repair_sql() function in sql_generator.py is likely failing, show the fix 4. Provide a DuckDB-specific SQL cheat sheet for this error pattern (e.g., if it's a date function error, show correct DuckDB date functions) Focus on the DuckDB dialect — not standard SQL, not PostgreSQL.

DEBUG 3 — Streamlit Session State Issue

My Streamlit app is losing state between reruns. Symptom: [DESCRIBE: e.g., "uploaded file disappears after sending a chat message"] Current session state management: [PASTE YOUR SessionState class or relevant app.py code] TASK: 1. Identify which state key is not being persisted correctly 2. Explain how Streamlit's rerun model causes this issue 3. Show the fix — only the affected SessionState methods and app.py calls 4. Add a one-line comment explaining the fix so I understand the pattern for future debugging Context: using streamlit==1.40.0, SessionState class wraps st.session_state.

KNOWLEDGE CHECK

Test your understanding of RAG SQL Agent architecture. 5 questions, 50 XP available.

QUESTION 1 OF 5

In the RAG SQL Agent, why does ChromaDB only store schema metadata — not the actual data rows?

A) Because ChromaDB can't handle tabular data formats

B) To give the LLM the context it needs to write SQL, without blowing up the context window with millions of rows

C) Because pandas DataFrames can't be serialized to vectors

D) Because vector databases are too slow for real data

QUESTION 2 OF 5

The SQL self-repair loop calls sql_generator.repair() with the broken SQL and the DuckDB error message. What is the maximum number of repair attempts before the agent gives up?

A) 1

B) 2

C) 3 (configurable via max_retries in Settings)

D) Unlimited — it keeps trying until success

QUESTION 3 OF 5

Why does the DataLoader use conn.execute("DROP VIEW IF EXISTS table_name") BEFORE every conn.register()?

A) To prevent "view already exists" errors when a user re-uploads the same file

B) To clear cached query results

C) Because DuckDB doesn't support named views

D) To free memory from the previous DataFrame

QUESTION 4 OF 5

The sentence-transformers library is used for embeddings instead of the OpenRouter API. What is the main reason for this design choice?

A) sentence-transformers produce better embeddings than OpenAI

B) It runs completely locally — no API costs, no internet needed for schema indexing, and no data leaves your machine

C) ChromaDB only works with sentence-transformers

D) OpenRouter doesn't offer an embeddings endpoint

QUESTION 5 OF 5

The IntentClassifier returns META_QUESTION instead of DATA_QUERY for "hello" and "what can you do?". Why is intent classification important for production agents?

A) To filter out offensive questions

B) To avoid wasting LLM tokens generating SQL for questions that don't need SQL — and to give better UX responses for greetings and meta questions

C) Because DuckDB can't handle natural language questions directly

D) To limit the number of API calls per session

DEPLOY & EXTEND

Your agent is live. Here's how to use it, test it, and take it further.

🎯 TEST THESE QUERIES ON YOUR DATA

"How many rows are in this dataset?"

"Show me the top 10 rows by [numeric column]"

"What is the average [column] grouped by [category column]?"

"Find all rows where [column] is greater than [value]"

"What are the unique values in [column]?"

"Give me a summary of this dataset"

DuckDB CHEAT SHEET

Common DuckDB SQL patterns — share this with the LLM in your prompts if it generates wrong dialect.

DuckDB SQL Reference — Paste this into repair prompts when needed

-- Date functions (DuckDB dialect) SELECT strftime('%Y-%m', order_date) as month FROM orders; SELECT date_trunc('month', order_date) FROM orders; SELECT year(order_date), month(order_date) FROM orders; SELECT date_diff('day', start_date, end_date) FROM events; -- String functions SELECT regexp_replace(col, '[^a-zA-Z0-9]', '') FROM table; SELECT string_split(col, ',')[1] FROM table; -- 1-indexed! -- Safe type casting (handles NULLs gracefully) SELECT TRY_CAST(col AS FLOAT) FROM table; -- returns NULL on fail SELECT TRY_CAST(col AS INTEGER) FROM table; -- Aggregation SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY amount) as median FROM sales; SELECT mode() WITHIN GROUP (ORDER BY category) FROM products; -- Window functions SELECT ROW_NUMBER() OVER (PARTITION BY dept ORDER BY salary DESC) as rank FROM employees; SELECT LAG(revenue, 1) OVER (ORDER BY date) as prev_revenue FROM monthly; -- Native PIVOT (DuckDB only) PIVOT sales ON year USING sum(amount) GROUP BY region; -- Sampling for large tables SELECT * FROM big_table USING SAMPLE 10%; -- Common errors and fixes: -- ERROR: "Referenced column not found" -- FIX: Use exact column name from schema. Wrap with spaces in double quotes: "column name" -- ERROR: "Could not convert string to FLOAT" -- FIX: Use TRY_CAST(col AS FLOAT) instead of CAST -- ERROR: "No function matches DATEDIFF" -- FIX: DuckDB uses date_diff('day', start, end) not DATEDIFF

MODEL REFERENCE

Model	Model ID	Best For	Cost
GPT-4o Mini	openai/gpt-4o-mini	Fast SQL, high volume queries	Very Low
GPT-4o	openai/gpt-4o	Complex multi-table SQL, joins	Medium
Claude 3.5 Sonnet	anthropic/claude-3.5-sonnet	Best reasoning, complex data analysis	Medium
Claude 3 Haiku	anthropic/claude-3-haiku	Ultra-fast responses, simple queries	Very Low
Gemini Flash 1.5	google/gemini-flash-1.5	Fast, large context window	Low
Llama 3.1 70B	meta-llama/llama-3.1-70b-instruct	Open weights, private/sensitive data	Low

🎓 YOU SHIPPED A PRODUCTION AI AGENT

You understand RAG, SQL agents, vector embeddings, self-repair loops, and Streamlit interfaces — from first principles, by building each layer yourself.

BUILD YOURRAG SQL AGENT

🚀 YOUR AGENT IS LIVE

🧠 QUIZ COMPLETE

🎓 YOU SHIPPED A PRODUCTION AI AGENT

BUILD YOUR
RAG SQL AGENT