Skip to content

INTERVIEW_ARCHITECTURE_GUIDE.md

Latest commit

 

History

History
317 lines (257 loc) · 13.1 KB

INTERVIEW_ARCHITECTURE_GUIDE.md

File metadata and controls

317 lines (257 loc) · 13.1 KB
Raw

🏛️ IntelliQuery: Architecture & Interview Explanation Guide

Consolidated Master Flow & End-to-End System Pseudocode

Note

30-Second Interview Elevator Pitch
"IntelliQuery is an enterprise-grade agentic Text-to-SQL system and real-time ETL engine. It converts natural language prompts into optimized SQL queries using a 3-stage CrewAI multi-agent pipeline backed by local LLMs (Mistral for schema-aware query generation, CodeLlama for syntax and logic validation). It features an automated Pandas/SQLAlchemy CSV ETL pipeline that dynamically infers datatypes and creates PostgreSQL/SQLite tables on the fly, exposed via a high-performance FastAPI backend and a modern responsive dashboard."

📊 1. Master Architectural Flow Diagram

flowchart TD
    subgraph Layer1 ["1. Presentation Layer (Frontend Dashboard)"]
        UI1["Agent Studio (NL Prompt & Live Agent Visualizer)"]
        UI2["Drag & Drop CSV ETL Ingestion Hub"]
        UI3["Schema Explorer & Interactive SQL Playground"]
    end

    subgraph Layer2 ["2. Application Layer (FastAPI Engine)"]
        API1["POST /api/query"]
        API2["POST /api/upload-csv"]
        API3["GET /api/tables & POST /api/execute-sql"]
    end

    subgraph Layer3 ["3. Agentic AI Pipeline Layer (CrewAI + Ollama)"]
        A1["Agent 1: Mistral (Query Architect)"]
        A2["Agent 2: CodeLlama (SQL Verifier & Validator)"]
        A3["Agent 3: SQL Executor (Result Formatter)"]
        
        T1["Schema Introspection Tool (describe_database)"]
        T2["SQL Runner Tool (execute_sql_query)"]
        
        A1 -->|Introspect Schema| T1
        A2 -->|Re-verify Schema| T1
        A3 -->|Execute Query| T2
        A1 -->|Raw SQL Query| A2
        A2 -->|Verified SQL Query| A3
    end

    subgraph Layer4 ["4. Data & ETL Engine Layer"]
        ETL["ETL Engine (Extract -> Clean/Infer -> Load)"]
        DBM["Unified DB Manager (SQLAlchemy + Engine Failover)"]
        PG[("Primary: PostgreSQL 17")]
        SQLITE[("Fallback: SQLite user_data.db")]
    end

    UI1 -->|HTTP POST| API1
    UI2 -->|Multipart Form| API2
    UI3 -->|HTTP GET/POST| API3

    API1 --> Layer3
    API2 --> ETL
    API3 --> DBM

    ETL --> DBM
    T2 --> DBM
    DBM -->|Connection Pool| PG
    DBM -->|Local Disk Fallback| SQLITE
Loading

🧱 2. Architectural Component Breakdown

Layer Component Key Technologies Core Responsibilities
Presentation Agent Studio UI HTML5, CSS3, ES6 JS Renders natural language input, 3-step live agent progress visualizer, SQL code cards, and dynamic paginated data tables.
API Gateway REST API Engine FastAPI, Pydantic, Uvicorn Exposes clean REST endpoints (/api/query, /api/upload-csv, /api/tables), handles CORS, input validation, and asynchronous execution.
Agentic AI Multi-Agent Crew CrewAI, Ollama, Mistral, CodeLlama Executes a 3-stage LLM workflow: Generation $\rightarrow$ Logic Verification $\rightarrow$ Execution & Markdown Formatting.
ETL Engine Dynamic Ingestion Pandas, RegEx, BytesIO Sanitizes column headers/table names, infers SQL data types, cleans string values, and auto-loads datasets into the active DB.
Database Unified DB Manager SQLAlchemy, PostgreSQL, SQLite Implements a Singleton pattern with automatic failover from PostgreSQL 17 to local SQLite with zero operational downtime.

🔄 3. End-to-End Execution Lifecycle

sequenceDiagram
    autonumber
    actor User as User / Web UI
    participant API as FastAPI Gateway
    participant A1 as Agent 1: Mistral
    participant A2 as Agent 2: CodeLlama
    participant A3 as Agent 3: SQL Executor
    participant DB as Database Engine

    User->>API: POST /api/query { prompt: "Find top 5 highest earners" }
    API->>A1: Initiate CrewAI Pipeline (Task 1)
    A1->>DB: Call `describe_database` tool
    DB-->>A1: Returns database schema & column types
    A1-->>A2: Passes generated SQL query (Task 1 Output)
    A2->>DB: Call `describe_database` tool (Verification)
    A2-->>A3: Passes corrected & verified SQL (Task 2 Output)
    A3->>DB: Call `execute_sql_query` tool
    DB-->>A3: Returns raw query result set
    A3-->>API: Formatted Markdown Table & Structured JSON
    API-->>User: Returns JSON Payload (SQL + Execution Results)
Loading

💻 4. Master Architecture Flow (Pseudocode)

# =====================================================================
# INTELLIQUERY CONSOLIDATED MASTER ARCHITECTURE FLOW (PSEUDOCODE)
# =====================================================================

import re
import io
import pandas as pd
from typing import Dict, Any, List, Optional
from sqlalchemy import create_engine, inspect, text
from fastapi import FastAPI, UploadFile, File, HTTPException
from crewai import Agent, Task, Crew, LLM

# ---------------------------------------------------------------------
# COMPONENT 1: UNIFIED DATABASE MANAGER (SINGLETON WITH FAILOVER)
# ---------------------------------------------------------------------
class DatabaseManager:
    """Manages database connection pooling, schema introspection, and execution."""
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(DatabaseManager, cls).__new__(cls)
            cls._instance._init_engine()
        return cls._instance

    def _init_engine(self):
        postgres_url = "postgresql://postgres:pass@localhost:5432/intelliquery"
        sqlite_url = "sqlite:///user_data.db"
        
        # 1. Attempt primary PostgreSQL connection
        try:
            self.engine = create_engine(postgres_url, pool_pre_ping=True)
            with self.engine.connect() as conn:
                conn.execute(text("SELECT 1"))
            self.db_type = "postgresql"
            print("Connected to PostgreSQL database successfully.")
        except Exception:
            # 2. Fallback to local SQLite instance
            self.engine = create_engine(sqlite_url)
            self.db_type = "sqlite"
            print("Fallback to local SQLite database.")

    def get_schema_description(self) -> str:
        """Introspects all user tables and column types for LLM context."""
        inspector = inspect(self.engine)
        lines = [f"Database Type: {self.db_type.upper()}", "Available Schemas:"]
        
        for table_name in inspector.get_table_names():
            columns = inspector.get_columns(table_name)
            lines.append(f"\nTable: {table_name}")
            for col in columns:
                lines.append(f"  - {col['name']} ({col['type']})")
        return "\n".join(lines)

    def execute_query(self, sql_query: str) -> Dict[str, Any]:
        """Executes SQL statements safely and formats the result."""
        with self.engine.connect() as conn:
            df = pd.read_sql_query(sql_query, conn)
            return {
                "success": True,
                "row_count": len(df),
                "columns": list(df.columns),
                "rows": df.to_dict(orient="records"),
                "markdown": df.to_markdown(index=False)
            }

db_manager = DatabaseManager()


# ---------------------------------------------------------------------
# COMPONENT 2: AUTOMATED REAL-TIME CSV ETL PIPELINE
# ---------------------------------------------------------------------
class ETLPipeline:
    """Extracts CSV bytes, cleans headers, infers types, and loads into DB."""

    def sanitize_identifier(self, name: str) -> str:
        """Converts strings to valid SQL identifiers (lowercase, alphanumeric)."""
        clean = re.sub(r'[^a-zA-Z0-9_]', '_', name.lower()).strip('_')
        return clean or "tbl_data"

    def process_and_ingest_csv(self, file_bytes: bytes, filename: str) -> Dict[str, Any]:
        # Step 1: EXTRACT (Read raw CSV stream into DataFrame)
        df = pd.read_csv(io.BytesIO(file_bytes))
        if df.empty:
            return {"success": False, "error": "CSV file is empty."}

        # Step 2: TRANSFORM (Sanitize Table & Column Names)
        table_name = self.sanitize_identifier(filename.rsplit('.', 1)[0])
        df.columns = [self.sanitize_identifier(col) for col in df.columns]

        # Strip whitespace in string columns
        for col in df.select_dtypes(include=['object']).columns:
            df[col] = df[col].apply(lambda x: x.strip() if isinstance(x, str) else x)

        # Step 3: LOAD (Dynamically create table & batch write to DB)
        df.to_sql(
            name=table_name,
            con=db_manager.engine,
            if_exists="replace",
            index=False,
            chunksize=1000
        )

        return {
            "success": True,
            "table_name": table_name,
            "rows_ingested": len(df),
            "columns": list(df.columns)
        }

etl_pipeline = ETLPipeline()


# ---------------------------------------------------------------------
# COMPONENT 3: CREWAI MULTI-AGENT PIPELINE
# ---------------------------------------------------------------------
def run_crew_pipeline(user_prompt: str) -> Dict[str, Any]:
    """Orchestrates 3 agents sequentially to generate, verify, and execute SQL."""
    
    # LLM configuration (Ollama Local Runtime)
    mistral_llm = LLM(model="ollama/mistral", base_url="http://localhost:11434")
    codellama_llm = LLM(model="ollama/codellama", base_url="http://localhost:11434")

    # Agent 1: Mistral (Query Architect)
    agent1 = Agent(
        role="SQL Architect (Mistral)",
        goal="Inspect schema and write valid SQL query.",
        tools=[db_manager.get_schema_description],
        llm=mistral_llm
    )

    # Agent 2: CodeLlama (SQL Verifier & Validator)
    agent2 = Agent(
        role="SQL Verifier (CodeLlama)",
        goal="Validate syntax, verify column existence, and fix logic errors.",
        tools=[db_manager.get_schema_description],
        llm=codellama_llm
    )

    # Agent 3: SQL Executor
    agent3 = Agent(
        role="Database Executor",
        goal="Execute verified SQL and format results into markdown.",
        tools=[db_manager.execute_query],
        llm=mistral_llm
    )

    # Task Pipeline
    t1 = Task(description=f"Generate SQL for question: '{user_prompt}'", agent=agent1)
    t2 = Task(description="Verify and correct SQL from Task 1", agent=agent2, context=[t1])
    t3 = Task(description="Execute verified SQL from Task 2", agent=agent3, context=[t2])

    # Kickoff CrewAI Multi-Agent Execution
    crew = Crew(agents=[agent1, agent2, agent3], tasks=[t1, t2, t3], verbose=True)
    result = crew.kickoff()

    # Extract intermediate task outputs
    generated_sql = extract_sql(t1.output.raw)
    verified_sql = extract_sql(t2.output.raw)
    structured_data = db_manager.execute_query(verified_sql)

    return {
        "success": True,
        "natural_prompt": user_prompt,
        "generated_sql": generated_sql,
        "verified_sql": verified_sql,
        "execution_output": str(result),
        "structured_data": structured_data
    }


# ---------------------------------------------------------------------
# COMPONENT 4: FASTAPI REST API CONTROLLER
# ---------------------------------------------------------------------
app = FastAPI(title="IntelliQuery API Engine")

@app.post("/api/query")
def process_query(prompt: str):
    """Processes natural language query via CrewAI pipeline."""
    if not prompt.strip():
        raise HTTPException(status_code=400, detail="Prompt cannot be empty.")
    return run_crew_pipeline(prompt)

@app.post("/api/upload-csv")
async def process_csv(file: UploadFile = File(...)):
    """Triggers dynamic ETL pipeline for CSV upload."""
    content = await file.read()
    return etl_pipeline.process_and_ingest_csv(content, file.filename)

@app.get("/api/tables")
def list_tables():
    """Lists all available database schemas and row counts."""
    return {"tables": db_manager.get_schema_description()}

🎯 5. Core Interview Talking Points

Tip

Key Architecture Highlights to Emphasize in Interviews:

  1. Multi-Agent Decoupling over Single Prompting:
    • Interview Answer: Single-prompt LLMs frequently hallucinate table names or create bad join conditions. By separating concerns—Mistral focuses on initial schema-aware SQL construction while CodeLlama strictly reviews and validates syntax/constraints—query accuracy increases dramatically.
  2. Dynamic Real-Time ETL Pipeline:
    • Interview Answer: The system accepts arbitrary CSV datasets, cleans and normalizes headers into standard SQL identifiers using regex, infers native datatypes via Pandas, and creates corresponding tables dynamically using SQLAlchemy chunked writes.
  3. Database High Availability (Singleton + Failover):
    • Interview Answer: The DB Manager implements a Singleton pattern with connection pooling. If the PostgreSQL server is unreachable, it automatically degrades to a local SQLite database without breaking API functionality.
  4. Privacy & Zero Cost via Local LLM Orchestration:
    • Interview Answer: By leveraging Ollama to host lightweight open-source models (mistral and codellama), the solution guarantees enterprise data privacy with zero external API latency or token usage costs.