DataGuard: Autonomous PII Agent

An autonomous agent using DataHub metadata context to detect untagged PII, generate masked dbt transformation models, and write governance tags & assertions back to the Context Graph.

Comment

posted an update —

Here is the update post, evolution log, code snippet, and screenshot preview, completely revised without icons or emojis.

DataGuard: Autonomous PII Governance Agent — Development Update & Changelog

Public Update Post

Introducing DataGuard Agent: Autonomous PII Governance & dbt Remediation Engine

Data governance and compliance shouldn't mean endless manual tagging or writing tedious dbt masking queries. Meet DataGuard—an autonomous agent built for modern data stacks that detects untagged Personally Identifiable Information (PII), generates production-ready anonymized dbt models, and emits Metadata Change Proposals (MCPs) directly back to your DataHub catalog.

What's New in This Release?

  • Multi-Platform Target Selection: Choose target dataset URNs dynamically across PostgreSQL, Snowflake, and BigQuery directly from the Agent Control Panel.
  • Live Subprocess Execution Logging: Stream real-time terminal output, execution stages, and graph emission logs live in the Streamlit UI.
  • Anonymized dbt SQL Auto-Generation: Instantly compile dbt SQL transformation models with built-in MD5 hashing and regex masking for sensitive fields.
  • DataHub Graph Remediation: Emit automated MCP payloads to attach PII-Sensitive and GDPR-Restricted tags to target assets.
  • Adaptive Light/Dark Mode: Dynamic theme styling ensuring seamless usability across light and dark system settings.

Check out the full changelog below, grab the code snippet, and let us know what features you'd like to see next in the comments.

Product Evolution Log

Version 1.2.0 (Current Version)

  • Cross-Platform UTF-8 Subprocess Stream: Fixed Windows character encoding issues (cp1252 console crash) by enforcing UTF-8 streams for real-time subprocess logging across OS environments.
  • Full DataHub Client Mocking Fallback: Added defensive import fallbacks (DataHubGraphWriter, attach_pii_tags, and emit_pii_assertion) to ensure smooth demo execution in offline or cloud sandbox environments like Streamlit Community Cloud.
  • Theme-Native Visual System: Updated metric cards and navigation tabs to utilize Streamlit CSS variables (var(--secondary-background-color)), maintaining dark and light mode styling without visual breakage.

Version 1.1.0

  • Live Log Terminal Integration: Replaced static execution displays with dynamic subprocess.Popen log streaming.
  • DataHub Graph Writer Module: Integrated src/datahub_client/mcp_emitter.py for direct metadata assertions and tag emission.
  • Streamlit Web Interface: Built an interactive dashboard with audit tables, code previewers, and JSON payload inspection windows.

Version 1.0.0

  • Core Detection Engine: Initial release of src/agent/pii_detector.py for identifying sensitive schema patterns (email, phone_number, ssn).
  • dbt Template Engine: Auto-compilation of SQL transformation models inside examples/generated_dbt_model.sql.

Code Snippet Highlights

Here is the exact implementation used inside app.py to spawn the background agent process, set the PYTHONPATH, handle Windows UTF-8 execution safely, and stream terminal logs directly to the user interface:

import os
import subprocess
import streamlit as st

if st.sidebar.button("Trigger Agent Execution", use_container_width=True):
  with st.sidebar.spinner("Running agent..."):
    try:
      # Pass current environment with PYTHONPATH set to the root folder
      env = os.environ.copy()
      env["PYTHONPATH"] = "."
      env["PYTHONIOENCODING"] = "utf-8"  # Enforces UTF-8 output on Windows

      process = subprocess.Popen(
          ["python", "src/agent/runner.py"],
          stdout=subprocess.PIPE,
          stderr=subprocess.STDOUT,
          text=True,
          encoding="utf-8",
          errors="replace",
          bufsize=1,
          env=env,
      )

      log_box = st.sidebar.empty()
      full_logs = ""

      # Stream the output live to the UI
      for line in process.stdout:
        full_logs += line
        log_box.code(full_logs, language="bash")

      process.wait()
      if process.returncode == 0:
        st.sidebar.success("Pipeline executed successfully!")
      else:
        st.sidebar.error("Execution failed. Check log output.")
    except Exception as e:
      st.sidebar.error(f"Error: {e}")

Application Screenshot Preview

+---------------------------------------------------------------------------------------------+
| Agent Control Panel        | DataGuard: Autonomous PII Governance Agent                     |
| Target Dataset URN:        | Metadata-Driven Compliance, dbt Code Generation, & Graph    |
| [ analytics.users    v ]   |                                                             |
|                            | +-------------------+ +-------------------+ +-----------------+ |
| [ Trigger Execution ]      | | Target Dataset    | | Detected PII      | | Compliance      | |
|                            | | analytics.users   | | 2 (High Risk)     | | GDPR Enforced   | |
| Live Execution Logs:       | +-------------------+ +-------------------+ +-----------------+ |
| [1/4] Fetching URN...      |                                                             |
| [2/4] Detected: Email/Phone| [ Generated dbt Model ] [ DataHub MCP ] [ Audit Report ]     |
| [3/4] dbt Model Saved      | ----------------------------------------------------------- |
| [4/4] DataHub Tags Emitted | SELECT                                                      |
|                            |     MD5(user_id) AS user_id,                                |
| Agent Status: Online       |     REGEXP_REPLACE(email, '(?i)(?<=^.).*?(?=@)', '***') ... |
| Mode: Offline Fallback     | FROM {{ source('production', 'users') }}                    |
+---------------------------------------------------------------------------------------------+

Log in or sign up for Devpost to join the conversation.