Inspiration

Data pipelines fail constantly. In real-world enterprise environments, upstream APIs and data sources change without warning—converting flat addresses into nested JSON objects, formatting currency amounts as messy strings ("$149.99 USD"), or mixing Unix epoch timestamps with ISO strings. Worse, dirty data dumps frequently leak sensitive customer PII (SSNs, credit card numbers, personal emails) directly into analytics warehouses. Data engineers and developers spend over 40% of their working hours manually triaging broken schema contracts, writing one-off Pandas/SQL glue scripts, and fixing broken dbt pipelines. We asked ourselves: What if an autonomous AI agent fleet could take complete ownership of this messy, multi-step chore—from raw dirty ingestion and zero-trust PII sanitization to sandboxed self-healing code execution and live Cloud database loading with zero human intervention? That question led us to build DataForge AI.

What it does

DataForge AI is an action-taking Taskmaster agent that replaces the manual chore of data engineering through an end-to-end autonomous pipeline: Gemma 2 Zero-Trust Guardrail: Before sensitive payloads touch downstream reasoning models, Gemma 2 scans and redacts sensitive PII (SSNs, credit cards, emails, phone numbers) with deterministic tokens, producing an OpenTelemetry security audit log. Deep Schema Drift Diagnosis (Gemini 2.5/3.5 Pro): Analyzes incoming schema drift, structural mutations, nested dictionaries, dirty data types, and null spikes to produce a root-cause diagnosis. Dynamic ETL & DDL Synthesis (Gemini 2.5/3.5 Flash): Dynamically synthesizes executable Python ETL transformers (transform(raw_records)), PostgreSQL SQL DDL schemas with indexes, and production dbt models. Self-Healing Execution Sandbox: Runs the synthesized Python code in an isolated scope. If a runtime exception occurs, the agent automatically captures the traceback and line numbers, feeds them back into Gemini, and iteratively repairs the code until execution succeeds with 100% data integrity. Google Cloud Target Provisioning & Write Stream: Dynamically provisions target warehouse tables and streams clean records into Google Cloud SQL / BigQuery, while persisting job execution state and memory in Cloud Firestore. Mission Control Dashboard: A Next.js 14 interface featuring real-time WebSocket agent thought streaming, an interactive React Flow DAG lineage graph, side-by-side Before/After diff tables, and synthesized code exports.

🛠️ How We Built It Core Agent Engine: Python 3.11/3.14, FastAPI, google-genai SDK, google-cloud-firestore, google-cloud-pubsub, duckdb, pandas, opentelemetry. AI Models: Gemini 2.5/3.5 Pro (drift reasoning), Gemini 2.5/3.5 Flash (code synthesis), and Gemma 2 (27B-IT inline guardrail). Google Cloud Infrastructure: Google Cloud Run (containerized serverless agent runtime), Google Cloud Firestore (persistent Memory Bank & run history), Google Cloud SQL / BigQuery (target warehouse), and Google Cloud Build. Frontend: Next.js 14 (App Router), Tailwind CSS, Lucide Icons, WebSockets, and custom dark-mode glassmorphism UI. Observability: OpenTelemetry spans recording millisecond latency and audit traces for every agent phase.

Challenges we ran into

Zero-Trust PII Masking Without Data Corruption: Early regex-based masking methods corrupted numeric JSON formats (e.g. replacing epoch timestamp numbers with unquoted text tokens, causing invalid JSON). We solved this by building a recursive tree-traversing guardrail in gemma_guard.py that preserves schema types while securely masking sensitive strings. Autonomous Self-Healing Feedback Loop: Designing an execution engine that captures raw Python runtime exceptions (such as KeyError or TypeError) and feeds the exact line numbers and tracebacks back into Gemini to fix the code without getting stuck in infinite loops. Real-Time Multi-Agent Streaming: Synchronizing asynchronous agent execution steps in Python across WebSockets to update live DAG node pulse animations and streaming thought logs on the frontend with sub-second latency.

Accomplishments that we're proud of

Built a true Taskmaster Agent that takes end-to-end action in the physical environment sandboxed code execution, database provisioning, and live Cloud record inserts rather than just generating advisory text. Successfully integrated Gemma 2 as a zero-latency inline privacy guardrail that redacts SSNs, credit cards, phone numbers, and emails before sensitive payloads ever reach downstream reasoning models, with full OpenTelemetry compliance audit traces. Engineered a self-healing code execution sandbox that captures runtime tracebacks, feeds them autonomously back to Gemini, and repairs broken ETL scripts until 100% of records load cleanly without any human intervention. Achieved 100% automated test coverage across the full multi-agent pipeline, from PII guardrail to Cloud SQL warehouse load. Delivered a real-time Mission Control dashboard with live WebSocket agent thought streaming, an interactive React Flow DAG lineage graph, and side-by-side Before/After diff tables making every agent decision fully observable.

What we learned

How to leverage the Google GenAI SDK for high-concurrency structured JSON outputs across multiple Gemini model tiers (Pro for reasoning, Flash for code synthesis). How to combine lightweight models (Gemma 2) for low-latency zero-trust security guardrails with deep reasoning models (Gemini 2.5/3.5 Pro) for root-cause schema drift diagnosis in a single sequential pipeline without sacrificing speed. Techniques for designing self-healing sandboxed code execution environments where the agent autonomously reads its own exception tracebacks and modifies only the failing transformation logic without hallucinating missing dependencies. How to architect recursive tree-traversing PII masking that preserves JSON structure and data types (a critical challenge naive regex approaches corrupted numeric field serialization). How to synchronize asynchronous Python agent execution across WebSockets with millisecond latency to drive live DAG node state transitions in the frontend.

What's next for VoidMain

Native Google Cloud Pub/Sub integration for continuous real-time streaming ingestion from Kafka topics and Stripe webhooks — turning DataForge AI into an always-on, event-driven pipeline guardian. Automated GitHub Pull Request generation for auto-synthesized dbt models and Terraform schema migrations, so the agent contributes directly to your version-controlled data infrastructure. Google Veo video generation for automated visual executive pipeline health summaries — turning audit logs into shareable video reports for non-technical stakeholders. A multi-tenant team workspace so multiple data engineering teams can share a fleet, monitor each other's pipeline health, and collaborate on remediation strategies in real time.