architecture.md

Architecture

Components

Airflow DAGs (dags/etl_pipeline.py) Standard extract-transform-load pipeline with four failure scenarios wired up for demo purposes. Each DAG attaches on_failure_callback at the default_args level so any task failure automatically triggers remediation.

Failure Callback (dags/callbacks.py) Writes a structured JSON log entry for the failing task, then hands off to the remediation agent. Runs inside a try/except so a secondary agent error never masks the original task failure in the Airflow UI.

Remediation Agent (agent/remediation_agent.py) Orchestrates the full repair loop: calls the diagnostic agent, decides whether the suggested fix is actionable, validates it, routes to the appropriate Slack notification, and persists a log entry to logs/remediation_history.json.

Diagnostic Agent (agent/diagnostic_agent.py) Agentic loop backed by OpenAI gpt-4o. Given a DAG ID, task ID, and error string it invokes the four tools below via OpenAI function calling, then returns a structured diagnosis and fix. Tools are declared using LangChain's @tool decorator for clean input validation.

Tools (agent/tools.py)

• read_recent_logs — reads the last five entries from the per-DAG JSON log
• detect_schema_change — regex-based detection of column/table/type errors
• suggest_query_fix — patches a SQL query by replacing the offending column with a NULL AS placeholder
• classify_failure — categorises the error and flags whether it is auto-fixable

Slack Notifier (agent/slack_notifier.py) Two message types: a review-request block with a merge link when a fix is ready, and a plain failure alert when no fix could be produced.