AI-Powered Analytics Assistant: A Multi-Agent System for Self-Service Business Intelligence

Abstract

The AI-Powered Analytics Assistant is a production-oriented, multi-agent system that enables true self-service business intelligence (BI). It transforms natural language questions into validated SQL, retrieves datasets from a semantic warehouse, and produces interactive visualizations and narrative insights. Built with LangGraph for orchestration, OpenAI LLMs for reasoning, and Plotly/Streamlit for visualization, the system demonstrates agent collaboration, tool integration, and robust stateful workflows suitable for business analytics.

---

1. Introduction

Self-service BI has long been a strategic goal for organizations seeking to empower business users to independently explore data and generate insights. In practice, however, traditional BI platforms often make self-service difficult:

• Rigid, complex interfaces that require specialized training
• A gap between natural language questions and query languages (SQL)
• Manual back-and-forth between business users and data teams
• Slow iteration cycles for charts, metrics, and validation

The AI-Powered Analytics Assistant addresses these challenges with a conversational, agentic approach. Users ask plain-English questions; specialized agents parse intent, generate and validate SQL, fetch data, and render charts with narrative context. The system provides real-time progress updates, enforces validation checkpoints, and is easily extensible to new domains via a semantic layer that is currently YAML-based and can be federated to enterprise catalogs (e.g., Microsoft Purview) over time.

---

2. System overview

• Orchestration: LangGraph-based graphs coordinate agent execution and state transitions.
• Agents (specialized roles):
  • Parser Agent: interprets natural language into structured DataQuestions.
  • Data Extraction Agent: generates SQL, validates it, and executes against the warehouse.
  • Charting & Analysis Agent: produces Plotly figures and narrative analysis; validates outputs.
• UI: Streamlit app with live progress messages and interactive charts.
• Configuration: YAML-driven semantic model (warehouse joins and filters), dimensions and metrics catalogs.

2.0 Self-service BI via a semantic layer

The foundation of true self-service BI lies in eliminating the guesswork that typically plagues LLM-driven analytics systems. Rather than forcing the language model to infer business rules, calculate KPI definitions, or guess at data relationships, this system employs a curated semantic configuration that explicitly encodes all necessary business logic. The current implementation expresses these semantics through YAML configuration files (detailed in Section 6) that comprehensively define metrics, dimensions, time grains, and filtering rules.

This architectural approach is deliberately designed for enterprise evolution. The same semantic contract that currently reads from YAML files can seamlessly transition to being backed by an enterprise data catalog such as Microsoft Purview, where business rules and data lineage live in a centrally governed system maintained by designated data owners. This separation of concerns creates a powerful advantage: teams can evolve business logic, add new metrics, or refine calculations without the overhead of retraining or fine-tuning machine learning models.

By leveraging this semantic layer for all data calculation and extraction operations, the application achieves maximum determinism in its responses. The LLM can focus on understanding user intent and selecting appropriate metrics rather than deriving complex business calculations, resulting in consistent, reliable answers that align with organizational standards and definitions.

2.1 Agent Workflow

3. Architecture

The system's architecture leverages LangGraph, a state machine framework designed for complex AI agent interactions, to coordinate specialized agents while maintaining clean separation of concerns and enabling seamless extensibility.

While this project focuses on self‑service BI, the architecture is meant to work well for other multi‑agent apps too. The goals are simple: keep the code easy to change, use clear boundaries, and let multiple people work in parallel without conflicts. We aim for predictable data behavior and easy iteration. In practice, this means:

• Contract‑first structure: Nodes control flow, Services hold business logic, Tools/MCP handle external work, and Pydantic models define inputs and outputs.
• Easy to extend: You can add new agents or swap strategies without changing the main graph.
• Predictable data paths: The semantic layer (YAML) limits metrics, joins, and time grains; SQL is checked and runs in a sandbox.
• Isolated and testable: Each part can run on its own with mock data, so changes don’t break other parts.
• Built‑in observability: Structured logs with correlation at node and service boundaries make behavior easy to trace.

These map directly to the Core Architectural Principles below.

Core Architectural Principles:

• Modularity: Complex analytics tasks are decomposed into discrete operations, each implemented as a node with clear Pydantic-defined contracts
• Composability: Operations connect through well-defined state transitions, making the system easier to reason about and debug
• Extensibility: New agents integrate seamlessly without disrupting existing workflows—new chart renderers, data quality agents, or explanation generators can be added by defining state contracts and connection points

Three-Tier Structure:

• Nodes (e.g. sql_generate_node.py, chart_render_node.py): Handle workflow coordination and state transitions
• Services (e.g. sql_generation_service.py, charting_service.py): Contain business logic for respective domains
• Tools/MCP: Provide concrete implementations, from direct database connections to sandboxed MCP servers

Key Advantages:

Incremental Updates with Minimal Impact
The architectural philosophy prioritizes operational flexibility and development velocity through deliberate design choices that minimize coupling between components. When a new SQL generation algorithm emerges or a chart rendering improvement becomes available, developers can update individual agents or nodes with surgical precision — i.e. there is no cascading changes across the system and no disruption to other workflows. This isolation means that enhancements roll out safely and quickly, whether it's swapping to a more sophisticated LLM for parsing or introducing an entirely new SQL generation and validation approach.

Comprehensive Testing and Standalone Development
Equally important is the architecture's support for comprehensive testing strategies. Each agent can be developed, tested, and validated in complete isolation before integration. The clear contracts defined by Pydantic models mean that developers can unit test individual nodes with mock inputs, integration test services independently, and even run agents standalone for debugging or experimentation. This standalone capability proves invaluable during development:

• a charting agent can generate visualizations independently of the full orchestration pipeline, or
• a SQL generation service can be benchmarked against various query patterns without involving the UI or database layers.

Confident System Evolution
The design also enables confident evolution of the system. Multiple implementation strategies can coexist, allowing teams to gradually migrate from one approach to another or run A/B tests comparing different agent behaviors. Combined with the clear layer boundaries, this creates an environment where improvements can be introduced incrementally, validated thoroughly, and scaled according to specific performance requirements.

3.1 Orchestration graphs

__start__ → run_parsing turns the user question into structured DataQuestions; if none are valid, the flow ends. Otherwise init_loop seeds the queue and pick_next selects the next item. run_extractor generates a single Postgres SQL (from the YAML semantics), validates it (MCP preferred; asyncpg fallback), and executes it (MCP or SQLAlchemy) to produce a dataset. run_render_chart uses chart_hint + data to emit Plotly JSON and a short narrative, then validates the figure; any errors are fed back for a quick retry. accumulate stores results and progress; if more questions remain it loops to pick_next, else it ends (__end__).

State streams to the UI while session‑scoped JSON logs capture each step for traceability.

Underlying graphs invoked

• Orchestration: graphs/orchestrator_graph.py coordinates the whole flow and streams state.
• Parsing step (run_parsing): calls graphs/parser_graph.py to produce DataQuestions.
• Data step (run_extractor): calls graphs/data_extractor_graph.py for SQL → validate → execute.
• Chart step (run_render_chart): calls graphs/charting_graph.py to render + validate the figure.

3.2 Agents and nodes

• Parser:
  • nodes/parser_node.py, nodes/parser_validation_node.py
• Data extraction:
  • nodes/sql_generate_node.py, nodes/sql_validate_node.py, nodes/sql_extract_node.py, nodes/run_extractor_node.py
• Charting:
  • nodes/chart_render_node.py, nodes/chart_validate_node.py, nodes/run_render_chart_node.py
• Orchestration helpers:
  • nodes/accumulate_and_advance_node.py, nodes/init_loop_node.py, nodes/pick_next_question_node.py

3.3 State management

• Orchestrator state: states/agentic_orchestrator_state.py
  • Tracks user_query, semantic (YAML), processed_questions, progress_messages, validation flags, etc.
• Parser state: states/parser_state.py
• Charting state: states/charting_state.py
• Data extraction state: states/data_extractor_state.py

Design notes:

• All nodes return updated state via immutable-style merging (e.g., {**state, "field": value}) to preserve fields like progress_messages across nodes.
• Progress streaming uses LangGraph app.stream(initial, stream_mode="values") so Streamlit can update the UI as nodes complete.

3.4 Observability and Logging Infrastructure

Structured JSON logging (session-aware)
The system now emits logs as newline-delimited JSON to support downstream analysis and shipping to external tools (e.g., ELK/OpenSearch, Datadog, Splunk). Each record is machine-readable and includes a session_id so runs can be analyzed per session or aggregated across sessions. The logger can write a single combined log file or one file per session, depending on configuration.

Typical fields include:

• timestamp, level, logger
• session_id and request_id for correlation across agents and nodes
• event (e.g., run_started, progress, dq_chart_rendered) and/or message
• Optional metadata (counts, lengths, shapes) to support quick diagnostics

Example entry (abbreviated):

{"timestamp":"2025-10-31T20:32:15.896Z","level":"INFO","session_id":"…","logger":"agentic_data_assistant","event":"run_started","request_id":"…","user_query":"Show monthly revenue by product in 2025."}

Multi-agent instrumentation
Agents and nodes log key steps end‑to‑end: parsing, SQL generation and validation, data extraction, chart generation and validation, and final rendering. These events capture the flow of work with consistent correlation IDs, so a single session can be traced from input to chart.

Designed for analysis and improvement
Because logs are structured, they can be queried to track:

• User request patterns (e.g., common metrics/dimensions, query complexity)
• Semantic coverage and gaps (e.g., unknown metrics, missing joins)
• Validation outcomes and frequent errors
• Latency and throughput across agents and nodes

This creates a practical feedback loop for tuning prompts, extending the semantic layer, improving validation rules, and optimizing performance. Session‑level logging also makes it easy to reproduce, debug, or audit specific user runs.

---

4. Tool integration

4.1 Model Context Protocol (MCP) for Database functionality

MCP is used for safe database access through a dedicated TCP server. The server lives at code/mcp_server/sql_postgres_tcp_server.py and runs as a separate process that uses newline‑delimited JSON (NDJSON) over TCP.
An MCP client is used to integrate the MCP tools with the application

Execute the following command in code/mcp_server before running the agents.

python -m mcp_server.sql_postgres_tcp_server

What the SQL MCP server provides:

• Tools: sql.validate, sql.query, and schema.introspect
• Read‑only enforcement: rejects insert|update|delete|alter|drop|truncate|create|grant|revoke
• Limit and timeout safety: wraps queries to enforce a row limit and sets statement_timeout
• JSON‑serializable results: dates (ISO‑8601), decimals converted safely

Protocol (TCP, NDJSON):

• Request: { "tool": "sql.validate"|"sql.query"|"schema.introspect", "arguments": {…} }
• Response: { "ok": true, "result": {…} } or { "ok": false, "error": "…" }

How the app uses it:

• SQLValidationService and DataExtractionService prefer MCP when MCP_ENABLED=1 (set in config.SETTINGS) and a client is available via utils.mcp_client_tcp.get_tcp_mcp_sql_client_from_settings(); otherwise they fall back to local validators (asyncpg/SQLAlchemy).

Example service usage (MCP enabled):

# config.settings.py 
# ...
MCP_ENABLED = os.getenv("MCP_ENABLED", 0) # "1" to enable MCP, "0" to disable
MCP_SQL_MAX_ROWS = int(os.getenv("MCP_SQL_MAX_ROWS", "5000"))
MCP_SQL_TIMEOUT_MS = int(os.getenv("MCP_SQL_TIMEOUT_MS", "20000"))
MCP_TCP_HOST = os.getenv("MCP_TCP_HOST", "127.0.0.1")
MCP_TCP_PORT = int(os.getenv("MCP_TCP_PORT", "8765"))

EXAMPLE USAGE:

from code.services.sql_validation_service import SQLValidationService
from code.services.data_extraction_service import DataExtractionService

sql = "SELECT date, revenue FROM fact_sales WHERE date >= '2025-01-01'"

# Validate- will run via MCP if MCP_ENABLED = 1
validator = SQLValidationService()
ok, err = validator.validate(sql)
if not ok:
    raise ValueError(f"SQL invalid: {err}")

# extract (Run SQL) - will run via MCP if MCP_ENABLED = 1
extractor = DataExtractionService()
df = extractor.run_query(sql)

Development fallback (non‑MCP): see 4.2 for the direct SQLAlchemy adapter used mainly during local development. The SQL validator also supports a local asyncpg path when MCP is disabled.

4.2 SQLAlchemy integration (legacy)

For environments without MCP support, tools/sqldb_sqlalchemy.py provides direct SQLAlchemy connections. This fallback ensures compatibility while teams transition to MCP-based workflows and/or during development if mcp is not available.

4.3 Other Tools

All other tools (parsing helpers, chart validation, rendering utilities) run in‑process as plain Python functions that you can optionally register with an agent—no separate MCP servers are used for them.

Non‑MCP tools used in this app (run in‑process):

• tools/user_parser_tools.py: alias_to_canonical, try_map_template — used by the user‑query parser/agent
• tools/chart_validation_tools.py: validate_plotly_fig_json — validates Plotly figure JSON produced by the charting agent

You can call these directly or register them with an agent. For example, the chart validator:

from langchain.tools import Tool
from tools.chart_validation_tools import (
  validate_plotly_fig_json,      # tool object for agent use
  validate_plotly_fig_json_fn,   # plain function for direct calls
)

# Agent registration
tools = [validate_plotly_fig_json]

# Direct usage in services/nodes
res = validate_plotly_fig_json_fn(fig_json)
if not res["valid"]:
  # feed error back to the LLM or retry renderer
  handle_validation_error(res["error"])

---

5. Semantic layer

5.1 Current YAML-based approach

The system's semantic understanding is driven by YAML configuration files that define business rules and metrics:

• config/ag_data_extractor_config/warehouse.yaml: Database schema, table joins, column mappings
• config/ag_user_query_parser_config/metrics.yaml: Valid metrics and dimension registry with canonical names and aliases

Benefits:

• Deterministic interpretation: Eliminates LLM guesswork for business logic
• Business user friendly: Data teams can modify metrics without code changes
• Version controlled: Business rules tracked alongside code

Example metrics definition:

metrics:
  - actual_revenue
  - budget_revenue
  - units_sold
  - gross_margin

dimensions:
  - product
  - product_category
  - customer
  - region

aliases:
  actual_revenue: [revenue, turnover, sales_revenue, income, total revenue]

5.2 Enterprise evolution path

The current YAML semantics are a bridge to an enterprise catalog (e.g., Microsoft Purview) without changing the public contract. We keep the same interface and swap YAML readers for catalog-backed adapters, gaining centralized governance, lineage-aware impact analysis, glossary-aligned terminology, and federated domain models under a unified API. Migration is incremental: maintain the contract, introduce adapters, migrate domains gradually, and move change control to catalog approval workflows—avoiding a disruptive rewrite.

---

6. Data model and schemas

• Parser output models in models/user_request_parser_model.py
  • DataQuestion includes: metrics, dimensions, optional time_range, filters, sort, top_k, chart_hint
  • dataset stored in a JSON-serializable form (e.g., list of dicts) for portability
  • chart_figure_json stored as a string; narrative text optional
• SQL generation input SQLGenerationInput in services/sql_generation_service.py

LLM structured output considerations:

• Newer model families may require strict JSON schema constraints (e.g., additionalProperties: false).
• For dynamic datasets, prefer List[Dict[str, Any]] or post-process to a defined superset model when using strict function calling.

---

7. Validation and error handling

7.1 Multi-layer validation

• Parser validation: Ensures parsed metrics/dimensions exist in the semantic registry
• SQL validation: MCP validator confirms query syntax and checks for read-only compliance
• Chart validation: Verifies Plotly JSON structure and data compatibility

7.2 Retry logic

Nodes increment validation_attempts counters and retry with corrected inputs. Example pattern:

if not is_valid and state["validation_attempts"] < max_attempts:
    # Regenerate with error feedback
    return generate_corrected_output(error_message)

7.3 Graceful degradation

When validation fails repeatedly, the system provides partial results with clear error explanations rather than complete failure.

---

8. How to run

8.1 Setup

Prerequisites

• Python 3.10+ (recommended)
• A PostgreSQL database you can read from
  • Use the provided Docker script and database dump in setup/
• An OpenAI API key (for LLM steps)
• Docker installed and running.

Configuration

1. Clone the repository

   git clone https://github.com/pkasseran/ai-powered-analytics-assistant.git

2. Install dependencies:

   # Navigate to project
   cd ai-powered-analytics-assistant
   
   # Install the python dependencies
   pip install -r requirements.txt

3. Environment Variables:

• Make a copy of .env.example and rename to .env
• Setup the environment variables

# For using OpenAI
 OPENAI_API_KEY=sk-your-openai-api-key-here
 DEFAULT_LLM_MODEL="gpt-4-mini"

 # FOR SQLALCHEMY (direct Database connection)
 # Change as needed to match your local Postgres setup
 POSTGRES_URI=postgresql+psycopg2://postgres:postgres@localhost:5435/dwdb

 # FOR MCP CONFIGURATION
 # Change as needed to match your local Postgres setup
 MCP_PG_DSN=postgresql://postgres:postgres@localhost:5435/dwdb 
 MCP_PG_MAX_ROWS=5000
 MCP_PG_TIMEOUT_MS=20000
 MCP_TCP_HOST=127.0.0.1
 MCP_TCP_PORT=8765
 MCP_ENABLED=1

4. Setup Database

   • Prerequisites

     • Docker Desktop installed and running
     • setup/dwdb.dump for restoring postgres database data

   • Defaults used by scripts/examples

     • Host: localhost, Port: 5435, User: postgres, Password: postgres, Database: dwdb

   • macOS/Linux (Docker + restore)

     cd setup
     chmod +x setup_docker_postgres_db.sh
     ./setup_docker_postgres_db.sh

     What it does: pulls postgres:latest, starts postgres_dwdb (5435->5432), creates dwdb, and restores dwdb.dump if present.

     Verify:

     PGPASSWORD=postgres psql -h localhost -p 5435 -U postgres -d dwdb -c "\\dt"

   • Windows (PowerShell)

     cd setup
     ./setup_docker_postgres_db.ps1

     The script starts or reuses postgres_dwdb on port 5435, creates dwdb (if missing), restores dwdb.dump when available, and prints the connection string.

     Verify:

     docker exec -e PGPASSWORD=postgres postgres_dwdb psql -U postgres -d dwdb -c "\\dt"

   • Connection strings (align with env above)

     • SQLAlchemy: postgresql+psycopg2://postgres:postgres@localhost:5435/dwdb
     • MCP DSN: postgresql://postgres:postgres@localhost:5435/dwdb

   • Common operations

     • Stop: docker stop postgres_dwdb
     • Start: docker start postgres_dwdb
     • Remove: docker rm -f postgres_dwdb (irreversible)

5. Configure semantic layer (NO CHNAGE):

   • config/ag_data_extractor_config/warehouse.yaml for your database schema.
   • config/ag_user_query_parser_config/metrics.yaml for business metrics and dimensions.

8.2 Running the application

Streamlit UI:

streamlit run code/app_streamlit.py

8.3 MCP server setup (TCP)

Make sure the following are set in .env

# FOR MCP CONFIGURATION
# Change as needed to match your local Postgres setup
MCP_PG_DSN=postgresql://postgres:postgres@localhost:5435/dwdb # Required
MCP_PG_MAX_ROWS=5000
MCP_PG_TIMEOUT_MS=20000
MCP_TCP_HOST=127.0.0.1
MCP_TCP_PORT=8765
MCP_ENABLED=1

Run the PostgreSQL MCP server as a standalone TCP process:

# Start the server
python -m code.mcp_server.sql_postgres_tcp_server

The app’s MCP client (utils.mcp_client) connects to this TCP server. SQLValidationService and DataExtractionService will automatically use it when constructed with use_mcp=True and when the client is configured in settings.

---

9. Future Enhancement

9.1 Current limitations

• Single database support: Currently designed for PostgreSQL; expanding to multi-database requires adapter pattern
• YAML-based semantics: Manual configuration management; enterprise catalog integration pending

9.2 Planned enhancements

• Microsoft Purview integration: Replace YAML with enterprise data catalog
• Multi-database support: Extend MCP validators for MySQL, BigQuery, Snowflake
• Advanced SQL generation Agent: Implement determistic function to generate sql script using templates mapping.
• Performance optimization: Caching layer for frequently requested metrics

---

10. References

• LangGraph: https://langchain-ai.github.io/langgraph/
• Model Context Protocol: https://modelcontextprotocol.io/
• Plotly: https://plotly.com/python/
• Streamlit: https://streamlit.io/