Skip to content

Latest commit

 

History

History
286 lines (273 loc) · 18.1 KB

File metadata and controls

286 lines (273 loc) · 18.1 KB

NLP-to-SQL Agentic Pipeline - Complete Architecture

System Overview

┌─────────────────────────────────────────────────────────────────────────────┐
│                          USER LAYER (Streamlit)                             │
│  ┌─────────────────────────────────────────────────────────────────────────┐ │
│  │ • Database Configuration & Selection                                    │ │
│  │ • Schema Introspection Display                                          │ │
│  │ • Natural Language Query Input                                          │ │
│  │ • Orchestration Mode Selection (Direct / LangChain / Pipeline)         │ │
│  │ • Results Display (SQL, Data, Summary, Narrative)                      │ │
│  │ • Execution Trace Visualization                                         │ │
│  └─────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
                                    ↓
┌─────────────────────────────────────────────────────────────────────────────┐
│                      ORCHESTRATION LAYER (app.py)                           │
│  ┌──────────────────┐  ┌──────────────────┐  ┌──────────────────────────┐  │
│  │ Direct Mode      │  │ LangChain Agent  │  │ Full Agentic Pipeline   │  │
│  │ (Simple)         │  │ (Multi-tool)     │  │ (Recommended)           │  │
│  ├──────────────────┤  ├──────────────────┤  ├──────────────────────────┤  │
│  │ • Schema info    │  │ • Tool chain     │  │ • 8-step orchestration  │  │
│  │ • SQL gen        │  │ • Agent decides  │  │ • Full step tracing     │  │
│  │ • Validation     │  │ • Tool calls     │  │ • Reasoning logs        │  │
│  │ • Execution      │  │ • Results        │  │ • Performance metrics   │  │
│  │ • Insights       │  │                  │  │ • Error recovery        │  │
│  └──────────────────┘  └──────────────────┘  └──────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────────┘
                                    ↓
┌─────────────────────────────────────────────────────────────────────────────┐
│                    CORE ORCHESTRATION (agentic_pipeline.py)                │
│  ┌─────────────────────────────────────────────────────────────────────────┐│
│  │ Step 1: Introspect Schema        Step 2: Build Semantic Model          ││
│  │ ├─ introspect_schema()           ├─ build_semantic_model()             ││
│  │ ├─ Get tables, columns, FKs      ├─ Enrich with samples                ││
│  │ └─ Trace: input, output, error   └─ Trace: input, output, error       ││
│  │                                                                          ││
│  │ Step 3: Generate SQL             Step 4: Validate Syntax               ││
│  │ ├─ generate_sql_from_nl()        ├─ validate_sql()                     ││
│  │ ├─ OpenAI + retry logic (max 5)  ├─ sqlglot parsing                    ││
│  │ └─ Trace: attempts, error        └─ Trace: error                       ││
│  │                                                                          ││
│  │ Step 5: Verify Policy            Step 6: Execute SQL                   ││
│  │ ├─ verify_sql_query()            ├─ execute_read_query()               ││
│  │ ├─ Read-only check, reject CRUD  ├─ Row limit enforcement              ││
│  │ └─ Trace: approval status        └─ Trace: rows, columns               ││
│  │                                                                          ││
│  │ Step 7: Summarize Results        Step 8: Narrative Insights            ││
│  │ ├─ summarize_dataframe()         ├─ narrative_insights_from_summary()  ││
│  │ ├─ describe() + percentiles      ├─ OpenAI summarization               ││
│  │ └─ Trace: summary_keys           └─ Trace: narrative_length            ││
│  └─────────────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────────────┘
                                    ↓
┌─────────────────────────────────────────────────────────────────────────────┐
│                      SPECIALIZED MODULES LAYER                              │
│  ┌──────────────────┐ ┌────────────────┐ ┌───────────┐ ┌─────────────────┐ │
│  │  db.py           │ │  semantic.py   │ │ agent.py  │ │  nlp_sql.py     │ │
│  │ ──────────────── │ │ ────────────── │ │ ────────  │ │ ──────────────  │ │
│  │ • load_db_urls   │ │ • build_model  │ │ • verify  │ │ • call_openai   │ │
│  │ • create_engines │ │   (with        │ │   _sql_   │ │ • validate_sql  │ │
│  │ • introspect     │ │   samples)     │ │   query   │ │ • is_select_only│ │
│  │   _schema        │ │                │ │ • normalize│ │ • generate_sql  │ │
│  │ • sample_table   │ │                │ │   (strip  │ │   _from_nl      │ │
│  │ • execute_read   │ │                │ │   comments)│ │ • clean_response│ │
│  │   _query         │ │                │ │          │ │   _sql          │ │
│  └──────────────────┘ └────────────────┘ └───────────┘ └─────────────────┘ │
│                                                                              │
│  ┌──────────────────────────┐  ┌────────────────────────────────────────┐  │
│  │  insights.py             │  │  agent_chain.py (LangChain)            │  │
│  │ ──────────────────────── │  │ ──────────────────────────────────────  │  │
│  │ • summarize_dataframe    │  │ • _make_tools() [Tool.from_function]   │  │
│  │   (describe + stats)     │  │ • run_agent() [LangChain orchestration]│  │
│  │ • narrative_insights_    │  │ • Tool: generate_sql                   │  │
│  │   from_summary           │  │ • Tool: validate_sql                   │  │
│  │   (OpenAI narration)     │  │ • Tool: execute_sql                    │  │
│  │                          │  │ • Tool: narrative_insights             │  │
│  └──────────────────────────┘  └────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────────┘
                                    ↓
┌─────────────────────────────────────────────────────────────────────────────┐
│                        EXTERNAL SERVICES LAYER                              │
│  ┌──────────────────────────────┐  ┌─────────────────────────────────────┐ │
│  │  Database Backends           │  │  OpenAI API                         │ │
│  │ ─────────────────────────────│  │ ───────────────────────────────────  │ │
│  │ • PostgreSQL                 │  │ • ChatCompletion (SQL generation)   │ │
│  │ • MySQL                      │  │ • ChatCompletion (Insights)         │ │
│  │ • SQLite                     │  │ • Model: gpt-4 or gpt-3.5-turbo    │ │
│  │ • Any SQLAlchemy-supported DB│  │ • Token usage tracking              │ │
│  └──────────────────────────────┘  └─────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘

Data Flow Example: "How many orders were completed?"

User Input (Streamlit)
    ↓
Query: "How many orders were completed?"
Dialect: "sqlite"
Orchestration Mode: "Full Agentic Pipeline"
    ↓
[app.py] ─→ run_agentic_pipeline()
    ↓
[agentic_pipeline.py::AgenticPipeline.run()]
    ├─ _trace_step("introspect_schema")
    │  ├─ db.introspect_schema(engine)
    │  │  ├─ SQLAlchemy Inspector
    │  │  ├─ Extract tables: [customers, orders, products, order_items]
    │  │  ├─ Extract columns: id, customer_id, order_date, total_amount, status
    │  │  └─ Extract FKs: customer_id → customers.id
    │  └─ _end_trace() ✓ 45ms
    │
    ├─ _trace_step("build_semantic_model")
    │  ├─ semantic.build_semantic_model(engine)
    │  │  ├─ Use introspected schema
    │  │  ├─ Sample each table (3 rows)
    │  │  │  ├─ orders: [(1, 1500.0, "completed"), (2, 299.99, "completed"), ...]
    │  │  └─ Enrich schema with samples
    │  └─ _end_trace() ✓ 120ms
    │
    ├─ _trace_step("generate_sql")
    │  ├─ nlp_sql.generate_sql_from_nl(
    │  │  query="How many orders were completed?",
    │  │  semantic={schema with samples},
    │  │  dialect="sqlite",
    │  │  max_retries=5
    │  │ )
    │  ├─ Attempt 1 (Failed)
    │  │  ├─ openai.ChatCompletion.create()
    │  │  ├─ Response: "SELCT COUNT(*) FROM orders WHERE status='completed'"
    │  │  ├─ validate_sql() ✗ Parse error: "SELCT" not recognized
    │  │  └─ Retry with error feedback
    │  ├─ Attempt 2 (Success)
    │  │  ├─ openai.ChatCompletion.create()
    │  │  ├─ Response: "SELECT COUNT(*) FROM orders WHERE status='completed'"
    │  │  ├─ validate_sql() ✓ Parsed successfully
    │  │  ├─ is_select_only() ✓ SELECT node detected
    │  │  └─ Return sql + attempts
    │  └─ _end_trace() ✓ 850ms
    │
    ├─ _trace_step("validate_sql_syntax")
    │  ├─ nlp_sql.validate_sql(sql, dialect="sqlite")
    │  ├─ sqlglot.parse_one(sql)
    │  ├─ AST check ✓ Valid structure
    │  └─ _end_trace() ✓ 12ms
    │
    ├─ _trace_step("verify_sql_policy")
    │  ├─ agent.verify_sql_query(sql, dialect="sqlite")
    │  ├─ Check: first word "SELECT" ✓ (allowed)
    │  ├─ Check: keywords [INSERT, UPDATE, DELETE, ...] ✗ (none found)
    │  ├─ Check: multiple statements ✗ (only one)
    │  ├─ Check: AST node type "select" ✓ (allowed)
    │  └─ _end_trace() ✓ 8ms
    │
    ├─ _trace_step("execute_sql")
    │  ├─ db.execute_read_query(engine, sql, limit=1000)
    │  ├─ pd.read_sql(sql, engine)
    │  ├─ DataFrame:
    │  │  COUNT(*)
    │  │       42
    │  └─ _end_trace() ✓ 34ms
    │
    ├─ _trace_step("summarize_results")
    │  ├─ insights.summarize_dataframe(df)
    │  ├─ n_rows = 1 (small dataset)
    │  ├─ Use full describe()
    │  ├─ numeric columns: ["COUNT(*)"]
    │  ├─ Result:
    │  │  {
    │  │    "n_rows": 1,
    │  │    "detailed_describe": {
    │  │      "COUNT(*)": {
    │  │        "count": 1, "mean": 42, "std": NaN,
    │  │        "min": 42, "25%": 42, "50%": 42, "75%": 42, "max": 42
    │  │      }
    │  │    }
    │  │  }
    │  └─ _end_trace() ✓ 28ms
    │
    └─ _trace_step("narrative_insights")
       ├─ insights.narrative_insights_from_summary(summary, query)
       ├─ openai.ChatCompletion.create()
       ├─ Response: "Based on the database analysis, 42 orders have been 
       │             completed. This represents a solid completion rate..."
       └─ _end_trace() ✓ 680ms
    ↓
Return: {
  "user_query": "How many orders were completed?",
  "final_result": {
    "sql": "SELECT COUNT(*) FROM orders WHERE status='completed'",
    "rows_returned": 1,
    "summary": {...},
    "narrative": "Based on the database analysis, ..."
  },
  "traces": [
    {step_name, input_data, output_data, error, duration_ms, reasoning},
    ...
  ]
}
    ↓
[app.py] Display Results:
├─ Show Generated SQL
├─ Show Row Count
├─ Display Summary (describe)
├─ Show Narrative Insights
└─ Expandable Trace Details (all steps with timing)
    ↓
Streamlit UI Update ✓

Component Responsibilities

Component Responsibility Inputs Outputs
app.py UI orchestration, mode selection User input, config Display result, traces
db.py Database access layer DB URL, SQL Schema, data
semantic.py Context generation for LLM Schema, engine Enriched schema with samples
nlp_sql.py NL→SQL translation NL, semantic, dialect SQL with metadata
agent.py SQL safety verification SQL, dialect Approved/rejected status
insights.py Result summarization DataFrame, query Stats + narrative
agent_chain.py LangChain orchestration Tools, LLM Agent result
agentic_pipeline.py Full pipeline + tracing Query, engine Complete result with traces

Execution Timeline (Typical Query)

0ms     ├─ Start
45ms    ├─ Introspect schema ✓
165ms   ├─ Build semantic model ✓
1015ms  ├─ Generate SQL (2 attempts) ✓
1027ms  ├─ Validate syntax ✓
1035ms  ├─ Verify policy ✓
1069ms  ├─ Execute SQL ✓
1097ms  ├─ Summarize results ✓
1777ms  ├─ Narrative insights ✓
1777ms  └─ Complete ✓

Total: 1.8 seconds (typical for gpt-4)

Security Layers

User Input
    ↓
[1] Streamlit Session Isolation
    ↓
[2] SQL Syntax Validation (sqlglot)
    ├─ Detects malformed SQL
    ├─ Parses AST structure
    └─ Rejects complex injections
    ↓
[3] SQL Policy Verification (agent.py)
    ├─ Keyword allowlist/blocklist
    ├─ First-word enforcement (SELECT only)
    ├─ Comment injection prevention
    ├─ Multiple-statement rejection
    └─ AST node type checking
    ↓
[4] Database Access Control
    ├─ Read-only user account
    ├─ Row limit enforcement (1000)
    ├─ Query timeout (recommended: 30s)
    └─ Connection pooling
    ↓
[5] Logging & Audit Trail
    ├─ All queries logged
    ├─ OpenAI API calls logged
    ├─ Errors tracked
    └─ Cost monitoring
    ↓
Execution (Safe)

Scaling Considerations

  • Caching: Cache semantic models & query results
  • Batching: Group queries to OpenAI API
  • Async: Use Celery/RQ for long-running queries
  • Monitoring: Track API costs, query times, errors
  • Rate Limiting: Implement token budgets per user
  • Connection Pooling: Reuse DB connections
  • Query Optimization: Analyze slow traces, add indexes