Technical Case Study · AI & Agents
CSV QA Agent
Groq LLM writes sandboxed pandas code to answer questions on CSV data
Enterprise-grade conversational tabular data agent that eliminates LLM arithmetic hallucinations. Never exposes raw dataset rows to the LLM—instead, extracts compact schema metadata, prompts Groq (`temperature=0.0`) to synthesize executable Pandas code, inspects the Python AST for security violations, and executes deterministically in a locked-down local sandbox.
Problem Statement
Sending raw CSV rows directly to an LLM causes arithmetic hallucinations on aggregations, blows past token context windows, and leaks sensitive tabular records.
Deterministic Architecture Advantage
Extracts only column headers and dtypes for the Groq LLaMA-3.3-70B prompt, synthesizes executable Python/Pandas code, validates the syntax tree via a custom AST Security Visitor, and executes deterministically in a restricted local namespace.
Execution Pipeline & System Topology
Schema & Stats Extractor
Ingests CSV/TSV/Excel/JSON, auto-parses dates, and builds a compact metadata summary.
Deterministic Code Synthesis
Prompts Groq/Anthropic at temperature=0.0 with few-shot examples to assign Pandas code to `result`.
AST & Builtin Lockdown
Inspects AST nodes to block forbidden modules, dunders, and I/O calls before compilation.
Audit Trail & UI Receipt
Executes on df.copy() and logs question, code, result, and timestamp to outputs/qa_log.json.
Verification & Engineering Highlights
- Schema-only context window: passes column names, dtypes, null counts, min/max/mean distributions, and top categorical values without leaking raw rows.
- Hardened multi-layer AST sandbox (`code_executor.py`) blocking `os`, `sys`, `subprocess`, `socket`, `open()`, `eval()`, `exec()`, dunder escapes (`__subclasses__`), and filesystem writes (`to_csv`, `to_pickle`).
- Includes both a glassmorphic dark-mode Streamlit web app (`streamlit_app.py`) and a dual-mode REPL/headless CLI (`app.py`) backed by 66 passing unit tests.
import ast
BLOCKED_MODULES = {"os", "sys", "subprocess", "shutil", "socket", "ctypes", "requests", "builtins", "io"}
BLOCKED_CALLS = {"open", "eval", "exec", "compile", "__import__", "to_csv", "to_pickle", "read_pickle"}
def validate_ast_safety(code: str) -> ast.AST:
tree = ast.parse(code, mode="exec")
for node in ast.walk(tree):
if isinstance(node, (ast.Import, ast.ImportFrom)):
raise PermissionError("Import statements are blocked in sandbox.")
if isinstance(node, ast.Attribute) and (
node.attr.startswith("__") or node.attr in BLOCKED_CALLS
):
raise PermissionError(f"Blocked attribute access: {node.attr}")
if isinstance(node, ast.Name) and node.id in BLOCKED_MODULES | BLOCKED_CALLS:
raise PermissionError(f"Blocked identifier: {node.id}")
return tree