A natural language data analysis platform. Upload a CSV, JSON, Parquet, or Excel file, ask questions in plain English, and get SQL-powered answers — generated by Gemini or Groq (configurable), validated by an AST-level SQL safety layer, and executed on Apache Spark.
Live Demo: https://agentic-dataanalyst.streamlit.app
API Backend: https://agentic-dataanalyst-1.onrender.com
Answering a question about a dataset usually means one of two things: write SQL yourself, or wait on whoever can. Most people who need an answer from a spreadsheet or export — "how many customers are in Germany," "what's the top-selling product last quarter" — don't know SQL and shouldn't need to. Spinning up a full BI tool or a database just to ask one question against a CSV someone emailed you is disproportionate to the task.
At the same time, "just ask an LLM to write SQL" has its own failure modes if built naively: a model can hallucinate a column that doesn't exist, or generate SQL that's subtly wrong, or write something destructive. This project exists to close that gap: natural-language access to tabular data, with the LLM's mistakes checked against a real SQL parser and self-corrected rather than silently returned as an answer.
Upload one or more data files (CSV, JSON, Parquet, or Excel), then ask questions conversationally — including follow-ups that build on a previous answer ("now just show me the top one"). Typical users are analysts or anyone doing ad-hoc exploration of a dataset who wants an answer faster than writing the SQL by hand, without needing to trust a single LLM response blindly: every answer shows its generated SQL and the model's own reasoning for it, so it's checkable, not just a black box.
Agentic Data Analyst lets users query structured data files using natural language. The system translates questions into Spark SQL via an agentic loop (Gemini or Groq), validates the generated SQL before it ever reaches Spark, executes it, and returns results with conversational context maintained across multiple questions.
Example queries:
- "What is the ratio of male to female customers?"
- "Show me the top 5 artists by total sales"
- "Which employees have the most invoices assigned to them?"
- "What is the average invoice total by country?"
flowchart TD
User(["User (browser)"]) -->|"HTTPS, configurable timeout"| UI
UI["Streamlit UI<br/>streamlit_app.py<br/>upload · query · logs · history"] -->|"REST, async"| API
subgraph Backend["FastAPI Backend — async"]
direction TB
API["api/server.py<br/>/analyze /upload_csv /schema /session /health"]
RateLimit["core/rate_limiter.py<br/>sliding-window, shared budget"]
Planner["agent/planner.py<br/>agentic SQL loop (async)"]
Validator["core/sql_validator.py<br/>sqlglot: reject writes, qualify<br/>known tables, cap row count"]
SparkEngine["core/spark_engine.py<br/>CSV/JSON/Parquet/Excel → Spark SQL"]
Obs["core/observability.py<br/>structured JSON logs, request-id"]
API --> RateLimit
API --> Planner
Planner --> Validator
Validator --> SparkEngine
API -.-> Obs
end
Planner <-->|"structured JSON output: sql + reasoning"| LLM
subgraph LLM["LLM Provider — LLM_PROVIDER selects one"]
direction LR
Gemini["Gemini 2.5 Pro<br/>responseSchema"]
Groq["Groq openai/gpt-oss-20b<br/>response_format json_schema"]
end
subgraph Storage["core/storage.py — SQLite"]
direction LR
Uploads[("uploaded tables")]
History[("session history")]
Cache[("query cache")]
end
API <--> Storage
Planner -.->|"cache hit skips the LLM call entirely<br/>(standalone questions only)"| Cache
SparkEngine --> Files[("persistent_uploads/<br/>original files, one per table")]
There's no login and no per-user data separation — it's a single shared workspace, which is what keeps the app simple to run and simple to read. See core/sql_validator.py for how every generated query is parsed and checked against the actual set of uploaded tables before it ever reaches Spark, so the LLM can't reference a table it hallucinated or one that was never uploaded — that's a correctness/safety guarantee, independent of who's asking.
Dependency Injection (Spark Singleton) + Size-Aware Caching
Spark initializes once at server startup and is reused for every query, eliminating the 1–2 second JVM boot overhead per request. Each uploaded file under SPARK_CACHE_MAX_MB (default 200MB) is registered and cached (spark.catalog.cacheTable) at upload time, so queries against it are served from memory instead of re-reading the file off disk on every request; larger files are registered as an on-demand view instead, since forcing an arbitrarily large file fully into memory can cost more than it saves. Local-mode shuffle partitions are tuned down from Spark's 200-partition cluster default (spark_shuffle_partitions, default 8) to cut per-query scheduling overhead, and Arrow-based serialization speeds up the Spark→pandas conversion on every result.
Multi-Format Ingestion
CSV, JSON, Parquet, and Excel are all supported through the same upload endpoint, dispatched by file extension in core/spark_engine.py — CSV/JSON/Parquet stay fully Spark-native; Excel is read via pandas (Spark has no native reader) and handed to Spark as a DataFrame.
Streaming, Size-Capped Uploads
/upload_csv streams the request body to disk in 1MB chunks and rejects (413) anything over MAX_UPLOAD_MB (default 500MB) mid-stream, instead of reading the entire file into memory first. Re-uploading a table under the same name deletes the previous physical file after the new one is registered, so repeated re-uploads don't silently leak disk space.
Agentic Loop with Conversational Memory and Self-Correction
The LLM maintains full conversation history across SQL attempts (up to MAX_SQL_ATTEMPTS, default 3). When a query fails validation or execution, the LLM sees the exact error and its own previous attempt, enabling context-aware self-correction without starting over — this is what the "agentic" in the name refers to, not just a single LLM call.
Structured SQL Generation, Two Providers
The LLM is called with a JSON schema ({"sql": ..., "reasoning": ...}) instead of being asked to wrap SQL in a fence and extracted via regex, so extraction is a guaranteed-valid JSON parse rather than a pattern match that silently breaks if the model formats its answer slightly differently. LLM_PROVIDER selects between GeminiClient and GroqClient in core/llm_client.py, both built on httpx.AsyncClient — agent/planner.py builds conversation history in a neutral {"role", "content"} shape that each client translates into its own wire format, so the agentic loop itself doesn't know or care which provider is configured.
Async End-to-End, With Blocking Work Explicitly Offloaded
/analyze and /upload_csv are async def, and the LLM HTTP calls are real non-blocking I/O via httpx.AsyncClient — verified with a concurrency test that two simulated LLM calls run in ~1x latency together, not 2x. Spark's execution (which has no native async API) is explicitly wrapped in asyncio.to_thread rather than called directly, since making a route async def without offloading its blocking calls would block the event loop worse than the original synchronous version.
Exact-Text Query Cache
Normalized-text matching in core/storage.py — a repeated question skips the LLM call (and its cost/latency) entirely. Scoped narrowly on purpose: only applies to standalone questions with no conversation history (a contextual follow-up like "now just the top one" would be wrong to serve from a cache keyed on text alone), and is invalidated automatically on every upload/re-upload so it can never serve an answer against data that's since changed.
Rate Limiting
A sliding-window limiter (core/rate_limiter.py) guards /analyze and /upload_csv — the two resource-heavy endpoints — with a single shared budget for the whole deployment.
Persistent State (SQLite)
Uploaded tables, session history, and the query cache are stored in SQLite (core/storage.py), not in-process dicts. At startup, the API re-registers every previously uploaded table into the fresh Spark session automatically (_rehydrate_from_storage) — as long as STORAGE_DB_PATH and persistent_uploads/ live on a volume that survives restarts, a redeploy or crash no longer wipes out uploaded tables or query history.
- Natural language to SQL, with the model's own reasoning shown alongside the query — not just a black-box answer
- Multi-format ingestion: CSV, JSON, Parquet, and Excel, joinable across tables in a single query
- Conversational memory: follow-up questions maintain context from previous answers
- Agentic error recovery: the LLM retries with context-aware corrections when SQL fails validation or execution (up to 3 attempts)
- AST-level SQL safety: every query is parsed with sqlglot and checked against the real set of uploaded tables, not just prompted to behave
- Rate limiting and an exact-text query cache (skips the LLM entirely on a repeated standalone question)
- Configurable LLM provider: Gemini or Groq, switchable via one env var
- Async backend: verified with a real concurrency test, not just
async defkeywords - Session history and a client-side activity log, both queryable in the UI
- Result export: download results as CSV or JSON
- Error-type-aware UI: a timeout, a rate limit, and a server error each show a distinct, actionable message instead of a raw error dump
- Persistent storage: uploads and history survive a restart
| Layer | Technology | Notes |
|---|---|---|
| Frontend | Streamlit | Activity log, real backend health check, configurable request timeout |
| Backend | FastAPI + Uvicorn, async | async def routes, blocking Spark calls explicitly offloaded via asyncio.to_thread |
| Data Engine | Apache PySpark | Multi-format registration, size-aware in-memory caching |
| LLM | Google Gemini or Groq (configurable) | gemini-2.5-pro / openai/gpt-oss-20b, both via httpx.AsyncClient |
| SQL Safety | sqlglot | AST-level validation, table qualification, row-limit enforcement |
| Storage | SQLite | Uploaded tables, session history, query cache — survives restarts |
| Rate Limiting | Custom sliding-window limiter | Shared budget, in-process |
| Config | Pydantic Settings | Env-var driven, .env supported |
| Observability | Structured JSON logging | Request-id correlation across every layer |
| Testing | pytest | Stubs Spark for fast CI; real-Spark behavior verified separately during development |
| CI | GitHub Actions | Runs on every push/PR, Python 3.10 + 3.11 |
| UI Hosting | Streamlit Cloud | — |
| API Hosting | Render.com (Docker) | — |
| Runtime | Python 3.10 | — |
| Java | OpenJDK 17 | For Spark |
Agentic DB/
├── streamlit_app.py # Frontend UI (Streamlit)
├── config.py # Pydantic settings (env vars)
├── requirements.txt # Python dependencies (production)
├── requirements-dev.txt # + pytest, minus pyspark/pyarrow
├── pytest.ini # pytest config
├── Dockerfile # UI service Docker image (Render/Cloud)
├── Dockerfile.api # API service Docker image (Render)
├── docker-compose.yml # Local development orchestration
├── packages.txt # System packages (OpenJDK for Spark)
├── runtime.txt # Python version pin (3.10.13)
│
├── api/
│ └── server.py # FastAPI app + endpoints (async)
│
├── agent/
│ └── planner.py # DataAnalystAgent - async agentic loop orchestration
│
├── core/
│ ├── spark_engine.py # Spark session + multi-format registration + SQL execution
│ ├── llm_client.py # Gemini/Groq clients (httpx.AsyncClient) with structured JSON SQL output
│ ├── sql_validator.py # sqlglot-based validation, table qualification, row caps
│ ├── storage.py # SQLite-backed uploads, session history, query cache (survives restarts)
│ ├── rate_limiter.py # Sliding-window rate limiting
│ └── observability.py # Structured JSON logging + request-id correlation
│
├── tests/ # pytest suite
│ └── conftest.py # Shared fixtures, incl. the pyspark stub
│
├── .github/workflows/
│ └── tests.yml # CI: runs pytest on every push/PR
│
└── .streamlit/
└── config.toml # Streamlit theme configuration
- Python 3.10+
- Java (OpenJDK 11 or 17)
- Docker + Docker Compose (optional)
git clone https://github.com/Sonith-Bingi/Agentic-DataAnalyst.git
cd "Agentic-DataAnalyst"
export GEMINI_API_KEY="your-key-here"
docker-compose up --buildStart the API backend:
export GEMINI_API_KEY="your-key-here"
export JAVA_HOME="/opt/homebrew/opt/openjdk@11"
export PATH="$JAVA_HOME/bin:$PATH"
python3 -m uvicorn api.server:app --host 0.0.0.0 --port 8000 --reloadStart the Streamlit UI (in a new terminal):
export API_URL="http://localhost:8000"
python3 -m streamlit run streamlit_app.pyThe API logs single-line JSON to stdout (core/observability.py) instead of free-form text, so logs can be filtered/parsed by Render's log viewer, docker logs, or any log drain.
Request correlation: every inbound request gets a short request_id, attached automatically to every log line emitted while handling it — including calls several layers down in agent/planner.py and core/spark_engine.py — via a contextvars-based filter, not by threading an id through every function signature.
What gets logged, among other events: http_request (method, path, status, duration), sql_generation_attempt (SQL-validation retry count/duration), sql_validation_rejected (with reason), sql_execution_success/sql_execution_failed (duration, row count / error), csv_register_success (upload size, duration), csv_view_cached vs csv_view_registered_uncached (which caching path a given upload took), csv_old_version_removed, rehydrate_complete (tables restored at startup).
LLM call detail (core/llm_client.py): every Gemini call logs a gemini_generation event with the full prompt, the full completion, the model name, and Gemini's own token usage (input_tokens/output_tokens/total_tokens from usageMetadata) — the same information a dedicated LLM tracing SaaS would show in a dashboard, just filterable via the same JSON log stream as everything else instead of a separate tool/account. A single request correlates across every layer via request_id, so grepping one request_id shows the full story: HTTP request in, each SQL-generation retry with its prompt/completion, Spark execution, HTTP response out.
Example lines:
{"timestamp": "2026-07-07T12:00:01", "level": "INFO", "logger": "core.llm_client", "message": "gemini_generation", "request_id": "a1b2c3d4e5f6", "model": "gemini-2.5-pro", "duration_ms": 891.2, "prompt": [...], "completion": "{\"sql\": \"SELECT ...\"}", "input_tokens": 512, "output_tokens": 34, "total_tokens": 546}
{"timestamp": "2026-07-07T12:00:02", "level": "INFO", "logger": "agent.planner", "message": "sql_execution_success", "request_id": "a1b2c3d4e5f6", "attempt": 1, "duration_ms": 184.3, "row_count": 42}GET /health — checks the Spark session responds, for uptime monitors and Render's health check.
User types question
│
▼
Streamlit sends POST /analyze
│
▼
FastAPI receives query + session_id
│
▼
DataAnalystAgent.run_agentic_loop()
│
├─► Get schema from SparkEngine
│
├─► Build prompt with schema + conversation history
│
├─► Gemini 2.5 Pro generates SQL
│
├─► SQLValidator checks syntax
│
├─► SparkEngine executes SQL
│
├─► Success? Return result as DataFrame
│
└─► Failure? Retry with error context (up to 3 times)
│
└─► LLM sees what went wrong + fixes it
When SQL execution fails, the agent appends the error to the conversation history and asks Gemini to fix it. The LLM sees:
- The original question
- The SQL it previously generated
- The exact Spark error message
- A request to correct the SQL
This context-aware retry achieves ~80%+ success even on complex multi-table joins.
# api/server.py - Spark boots ONCE when the server starts
agent = DataAnalystAgent(settings) # SparkSession initialized here
# Every /analyze request reuses the same SparkSession and the same
# cached, already-registered tables — /analyze never re-reads a CSV
# off disk; only /upload_csv touches the filesystem.