How elidehaene01/data-observability-agent works
Data-Observability-Agent compares source ERP data with dbt warehouse models, then handles scheduled data-load checks and PR-triggered discrepancy classification as distinct workflows. Documented PR capabilities also include merge gating, per-model documentation refresh, and Confluence reconciliation reports. The graph preserves these README-described capabilities; their wiring is not fully visible in sampled source, so uncertain implementation links are omitted. Sampled source supports the reconciliation connectors, LangGraph classification branches, results history, and dashboard read path.
Main entry points: aggregate_checks.py Aggregate checks · graph.py Agent routing graph · fetch_reconciliation_results.py Fetch results · action_nodes.py Branch actions
Components 24
Reconciliation
- Aggregate checks
reconciliation/aggregate_checks.py - Sample checks
reconciliation/sample_checks.py - Reconciliation results
reconciliation/models.py
Agent workflow
- Agent routing graph
agent/graph.py - Fetch results
agent/nodes/fetch_reconciliation_results.py - Analyze SQL diff
agent/nodes/analyze_diff.py - Classify discrepancy
agent/nodes/classify_discrepancy.py - Branch actions
agent/nodes/action_nodes.py - Agent state
agent/state.py - Merge gate
scripts/run_code_change_check.py
Model docs
- Read dbt artifacts
model_docs/manifest.py - Model structures
model_docs/models.py - Render model docs
model_docs/render.py
External connectors
- Confluence
- Jira and Slack
connectors/ticketing - Postgres connector
connectors/source/postgres_source.py - DuckDB connector
connectors/target/duckdb_target.py
Results and reporting
- Run history
results_store/writer.py - Results reader
results_store/reader.py - HTML dashboard
connectors/reporting/html_dashboard.py
Other components
- Scheduled trigger
- dbt PR
- ERP source
- dbt warehouse
Connections 25
- Scheduled trigger → Aggregate checks: starts checks
- dbt PR → Aggregate checks: starts checks
- Aggregate checks → Postgres connector: queries source
- Aggregate checks → DuckDB connector: queries target
- Postgres connector → ERP source: reads data
- DuckDB connector → dbt warehouse: reads models
- Aggregate checks → Sample checks: runs checks
- Aggregate checks → Reconciliation results: creates results
- dbt PR → Agent routing graph: invokes agent
- Agent routing graph → Fetch results: dispatches
- Fetch results → Run history: reads history
- Fetch results → Analyze SQL diff: passes results
- Analyze SQL diff → Classify discrepancy: passes evidence
- Classify discrepancy → Agent state: updates state
- Agent routing graph → Branch actions: routes outcome
- Branch actions → Confluence: publishes docs
- Branch actions → Jira and Slack: creates and notifies
- dbt PR → Merge gate: checks classification
- dbt PR → Read dbt artifacts: refreshes model docs
- Read dbt artifacts → Model structures: builds structures
- Model structures → Render model docs: renders docs
- Render model docs → Confluence: publishes pages
- Aggregate checks → Run history: writes results
- Results reader → Run history: reads rows
- HTML dashboard → Results reader: queries history