Orchestrating Enterprise RAG ETL with Airflow Multi-Agent Integration

1. The Use Case: "PolicySync" - Feeding CompliAgent

In the previous article we built CompliAgent, an enterprise multi-agent RAG system. But RAG is only as good as its corpus. In a Fortune 500, the corpus is a moving target:

25,000 policy documents across SharePoint, Confluence, email archives, and legal databases.
~400 documents change daily (new policies, amendments, revocations).
Chunking logic gets upgraded (e.g., moving from fixed-size to semantic chunking).
Embedding models get swapped (e.g., text-embedding-3-small → a domain-fine-tuned model).
Regulators occasionally demand "re-process everything from 2023-01-01 with the new retention tags."

A naive "run the ETL every night" approach fails on three counts:

It is not idempotent. Running it twice doubles the vectors.
It cannot recover. A failure at chunk #14,302 means restarting from scratch or writing brittle resume logic.
It cannot backfill intelligently. Re-embedding 25,000 docs because one chunker changed costs $900 in API fees and takes 6 hours.

We need an orchestrated, idempotent, backfillable ETL — and we need it to talk to the RAG agents so the system can heal itself.

Enter Apache Airflow + the CompliAgent Auditor agent.

2. Architecture

The critical insight: the RAG system is not just a consumer of the ETL - it is a feedback loop. When the Auditor agent detects a stale or missing citation, it can trigger a targeted Airflow backfill for just those documents.

3. The Three Pillars of Production ETL

3.1 Idempotency

A task is idempotent if running it N times produces the same result as running it once. In our pipeline:

Stage	Idempotency Mechanism
Extract	Watermark-based: only pull docs with last_modified > last_run_watermark
Transform	Deterministic: same input → same chunk IDs (content-hash-derived)
Embed	Conditional: skip if embedding already exists for this content hash
Load	INSERT ... ON CONFLICT (doc_id, version) DO UPDATE — never blind INSERT

The golden rule: every task must be safely re-runnable. Airflow will retry failed tasks; your code must welcome that.

3.2 Failure Handling

Failure Type	Response
Transient (network blip, 429 from OpenAI)	Exponential backoff retries inside the task
Deterministic (bad JSON, schema mismatch)	Route to Dead Letter Queue (DLQ), alert, continue
Catastrophic (DB down, S3 outage)	Fail the DAG, page on-call, do not retry
Silent data quality (empty chunk, hallucinated metadata)	Data quality assertions via Great Expectations or custom checks

3.3 Backfills

Airflow's logical_date (formerly execution_date) is the key. Each DAG run is parameterized by a date, and tasks read/write data partitioned by that date. This lets us:

Full backfill:

airflow dags backfill policy_sync --start-date 2023-01-01 --end-date 2026-07-14

Targeted backfill:

trigger a run with conf={"doc_ids": ["pol_123", "pol_456"]}

Partial rerun:

clear specific task instances and let the scheduler re-execute them

4. The Airflow DAG - Full Implementation

# dags/policy_sync.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.decorators import task
from airflow.models import Variable
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.openai.hooks.openai import OpenAIHook
from airflow.exceptions import AirflowSkipException, AirflowException
from airflow.utils.state import State
import hashlib, json

default_args = {
    "owner": "data-eng",
    "depends_on_past": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(minutes=60),
    "on_failure_callback": _notify_slack,   # defined below
    "sla": timedelta(hours=2),
}

dag = DAG(
    dag_id="policy_sync",
    default_args=default_args,
    start_date=datetime(2023, 1, 1),
    schedule="0 2 * * *",           # daily at 02:00 UTC
    catchup=False,
    max_active_runs=1,              # serialize runs to avoid watermark races
    tags=["rag", "compliagent", "etl"],
    params={
        "doc_ids": [],              # empty = full incremental; populated = targeted
        "force_reembed": False,
    },
)

4.1 Extract - Watermark-Based, Idempotent

@task(dag=dag)
def extract_policies(logical_date: datetime, **ctx) -> list[str]:
    """Pull only docs modified since the last successful run."""
    params = ctx["params"]
    pg = PostgresHook(postgres_conn_id="metadata_db")

    # Targeted backfill: ignore watermark, use explicit doc_ids
    if params.get("doc_ids"):
        placeholders = ",".join(["%s"] * len(params["doc_ids"]))
        rows = pg.get_records(
            f"SELECT doc_id, source, uri FROM source_registry "
            f"WHERE doc_id IN ({placeholders})",
            params["doc_ids"],
        )
    else:
        # Incremental: watermark = max(watermark) from last successful run
        last_watermark = Variable.get(
            "policy_sync_last_watermark",
            default_var="1970-01-01T00:00:00Z",
        )
        rows = pg.get_records(
            "SELECT doc_id, source, uri FROM source_registry "
            "WHERE last_modified > %s",
            (last_watermark,),
        )

    if not rows:
        raise AirflowSkipException("No new or targeted documents.")

    s3 = S3Hook(aws_conn_id="aws_default")
    keys = []
    for doc_id, source, uri in rows:
        raw = fetch_from_source(source, uri)   # SharePoint / Confluence / etc.
        key = f"raw/policies/{logical_date:%Y-%m-%d}/{doc_id}.json"
        s3.load_string(
            json.dumps({"doc_id": doc_id, "content": raw, "fetched_at": datetime.utcnow().isoformat()}),
            key=key,
            replace=True,                       # idempotent: overwrite if exists
        )
        keys.append(key)

    # Update watermark ONLY if this was an incremental run
    if not params.get("doc_ids"):
        new_watermark = pg.get_first(
            "SELECT MAX(last_modified) FROM source_registry"
        )[0]
        Variable.set("policy_sync_last_watermark", new_watermark)

    return keys

4.2 Transform - Deterministic Chunking with Content Hashes

@task(dag=dag)
def chunk_documents(raw_keys: list[str], logical_date: datetime) -> list[str]:
    """Chunk documents deterministically. Chunk ID = hash(doc_id + chunk_index + content)."""
    s3 = S3Hook(aws_conn_id="aws_default")
    out_keys = []

    for key in raw_keys:
        raw = json.loads(s3.read_key(key))
        doc_id = raw["doc_id"]
        chunks = semantic_chunk(raw["content"])   # your chunking library

        chunk_records = []
        for i, chunk in enumerate(chunks):
            content_hash = hashlib.sha256(chunk.encode()).hexdigest()[:16]
            chunk_id = f"{doc_id}#{i}#{content_hash}"   # deterministic!
            chunk_records.append({
                "chunk_id": chunk_id,
                "doc_id": doc_id,
                "index": i,
                "content": chunk,
                "content_hash": content_hash,
                "logical_date": logical_date.isoformat(),
            })

        out_key = f"processed/policies/{logical_date:%Y-%m-%d}/{doc_id}.jsonl"
        s3.load_string(
            "\n".join(json.dumps(r) for r in chunk_records),
            key=out_key,
            replace=True,
        )
        out_keys.append(out_key)

    return out_keys

4.3 Embed - Conditional, Cost-Aware

@task(dag=dag)
def embed_chunks(processed_keys: list[str], logical_date: datetime, **ctx) -> list[str]:
    """Embed chunks, but skip if the content hash already has an embedding."""
    force = ctx["params"].get("force_reembed", False)
    pg = PostgresHook(postgres_conn_id="metadata_db")
    s3 = S3Hook(aws_conn_id="aws_default")

    # Build set of already-embedded content hashes
    existing = {
        row[0] for row in pg.get_records(
            "SELECT content_hash FROM embeddings_registry"
        )
    }

    embedded_keys = []
    for key in processed_keys:
        records = [json.loads(line) for line in s3.read_key(key).splitlines()]
        to_embed = [r for r in records if force or r["content_hash"] not in existing]

        if not to_embed:
            continue

        # Batch embed — OpenAI limit is 2048 inputs per call
        texts = [r["content"] for r in to_embed]
        vectors = batch_embed(texts)   # wraps OpenAI with retry + rate-limit

        for record, vec in zip(to_embed, vectors):
            record["embedding"] = vec

        out_key = f"embedded/policies/{logical_date:%Y-%m-%d}/{key.split('/')[-1]}"
        s3.load_string(
            "\n".join(json.dumps(r) for r in records),
            key=out_key,
            replace=True,
        )
        embedded_keys.append(out_key)

    return embedded_keys

This saves enormous cost during backfills: if only 200 of 25,000 docs changed, we embed only those 200.

4.4 Load - UPSERT, Never INSERT

@task(dag=dag)
def load_to_vectorstore(embedded_keys: list[str]) -> dict:
    """UPSERT into pgvector. Idempotent by (chunk_id)."""
    pg = PostgresHook(postgres_conn_id="vector_db")
    s3 = S3Hook(aws_conn_id="aws_default")

    loaded = 0
    dead_letters = []

    for key in embedded_keys:
        records = [json.loads(line) for line in s3.read_key(key).splitlines()]
        for r in records:
            try:
                pg.run(
                    """
                    INSERT INTO policy_chunks (
                        chunk_id, doc_id, index, content, content_hash,
                        embedding, logical_date, ingested_at
                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, NOW())
                    ON CONFLICT (chunk_id) DO UPDATE SET
                        content = EXCLUDED.content,
                        content_hash = EXCLUDED.content_hash,
                        embedding = EXCLUDED.embedding,
                        logical_date = EXCLUDED.logical_date,
                        ingested_at = NOW();
                    """,
                    parameters=(
                        r["chunk_id"], r["doc_id"], r["index"], r["content"],
                        r["content_hash"], r["embedding"], r["logical_date"],
                    ),
                )
                loaded += 1
            except Exception as e:
                dead_letters.append({
                    "chunk_id": r["chunk_id"],
                    "error": str(e),
                    "logged_at": datetime.utcnow().isoformat(),
                })

    # Persist dead letters — do NOT fail the DAG for poison pills
    if dead_letters:
        pg.insert_rows("etl_dead_letter_queue", dead_letters, target_fields=list(dead_letters[0].keys()))

    return {"loaded": loaded, "dead_letters": len(dead_letters)}

4.5 Failure Callback + SLA Miss

def _notify_slack(context):
    """Called on task failure or SLA miss."""
    from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
    hook = SlackWebhookHook(slack_webhook_conn_id="slack_etl_alerts")
    ti = context["task_instance"]
    msg = (
        f":rotating_light: *{ti.dag_id}.{ti.task_id}* failed\n"
        f"Run: {ti.run_id}\n"
        f"Logical date: {ti.logical_date}\n"
        f"Exception: {context.get('exception')}"
    )
    hook.send(text=msg)

def _sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
    from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
    hook = SlackWebhookHook(slack_webhook_conn_id="slack_etl_alerts")
    hook.send(text=f":snail: SLA miss on {dag.dag_id}: {[s.task_id for s in slas]}")

5. Backfill Strategies

5.1 Full Backfill (Model Upgrade)

# Swap embedding model, re-embed everything from 2024-01-01 onward
airflow dags backfill policy_sync \
  --start-date 2024-01-01 \
  --end-date 2026-07-14 \
  --conf '{"force_reembed": true}' \
  --reset-dagruns

Because of the conditional embedding logic, this still only pays for the API calls it needs.

5.2 Targeted Backfill (Specific Documents)

airflow dags trigger policy_sync \
  --conf '{"doc_ids": ["pol_1234", "pol_5678"]}'

This bypasses the watermark and processes only the listed docs — useful when the RAG auditor detects stale answers.

5.3 Partial Rerun (Fix a Bug in One Task)

# Chunker had a bug; fix it, then re-run just the chunk→embed→load tasks
airflow tasks clear policy_sync \
  --task-regex "chunk_documents|embed_chunks|load_to_vectorstore" \
  --start-date 2026-07-01 \
  --end-date 2026-07-14

6. The LangGraph Integration - Auditor Triggers Backfill

Now the elegant part: the Auditor agent in CompliAgent can detect that a retrieved document is stale or missing, and trigger a targeted Airflow backfill.

6.1 The Tool

# agents/tools/airflow_trigger.py
import httpx
from langchain_core.tools import tool
from auth.oauth2 import get_current_user   # propagates user context

AIRFLOW_URL = "https://airflow.company.com/api/v1"

@tool
def trigger_targeted_backfill(doc_ids: list[str], reason: str) -> str:
    """Trigger a targeted PolicySync backfill for specific documents.

    Use this when retrieved documents appear stale, missing, or outdated
    relative to the user's question.

    Args:
        doc_ids: List of document IDs to reprocess.
        reason: Human-readable reason for the backfill (for audit).
    """
    user = get_current_user()
    if "admin:etl_trigger" not in user.scopes:
        return "Permission denied: requires admin:etl_trigger scope."

    # Airflow REST API with service-account OAuth
    resp = httpx.post(
        f"{AIRFLOW_URL}/dags/policy_sync/dagRuns",
        json={
            "conf": {"doc_ids": doc_ids},
            "note": f"Triggered by CompliAgent Auditor for user {user.sub}: {reason}",
        },
        headers={"Authorization": f"Bearer {_get_airflow_token()}"},
        timeout=10,
    )
    resp.raise_for_status()
    run_id = resp.json()["dag_run_id"]
    return f"Backfill triggered: run_id={run_id} for {len(doc_ids)} docs."

6.2 Wiring It Into the Auditor Agent

# agents/nodes.py (addition to Auditor)
def auditor(state: CompliState) -> CompliState:
    response = state["response"]
    scrubbed = scrub_pii(response)

    # Detect staleness: citations older than 90 days, or user pushback
    stale_docs = detect_stale_citations(state["retrieved_docs"], state["messages"])
    stale_doc_ids = [d["doc_id"] for d in stale_docs]

    audit_entry = {
        "node": "auditor",
        "pii_scrubbed": scrubbed != response,
        "stale_docs_detected": len(stale_docs),
    }

    # If stale docs found AND user has permission, trigger backfill
    if stale_doc_ids and "admin:etl_trigger" in state["user"].get("scopes", []):
        result = trigger_targeted_backfill.invoke({
            "doc_ids": stale_doc_ids[:20],   # cap to prevent runaway
            "reason": f"Auditor detected staleness in turn {state['messages'][-1].content[:80]}",
        })
        audit_entry["backfill_triggered"] = result
        # Append a note to the response
        scrubbed += "\n\n_Note: some source documents appear outdated. A refresh has been scheduled; please re-ask in ~10 minutes for the latest policy._"

    persist_audit_log(
        user_id=state["user"]["sub"],
        query=state["messages"][-1].content,
        response=scrubbed,
        trail=state["audit_trail"] + [audit_entry],
    )
    return {**state, "response": scrubbed, "audit_trail": state["audit_trail"] + [audit_entry]}

This creates a self-healing RAG system: the more users interact with it, the more it identifies and repairs its own blind spots.

7. End-to-End Walkthrough

Monday 02:00 UTC — Airflow scheduler fires policy_sync.

Extract reads watermark 2026-07-13T02:00:00Z, pulls 47 modified docs from SharePoint, uploads to s3://lake/raw/policies/2026-07-14/.
Chunk produces 312 chunks with deterministic IDs.
Embed checks embeddings_registry, finds 289 already embedded (content hash match), embeds only the 23 new ones. Cost: $0.004.
Load UPSERTs 312 rows into policy_chunks. 2 fail (malformed PDFs) → DLQ.
DAG succeeds; watermark advances to 2026-07-14T02:00:00Z.

Monday 14:32 — a compliance officer asks CompliAgent:

"What's the new travel reimbursement cap for APAC?"

Supervisor → policy_qa.
Retriever returns 5 docs, but the APAC travel policy's last_modified is 2025-03-01 — 16 months old.
Analyzer produces an answer but flags it: "Policy appears outdated; APAC region had a Q4 2025 revision not reflected."
Writer emits the answer with a staleness caveat.
Auditor detects the stale citation, calls trigger_targeted_backfill with doc_ids=["pol_apac_travel_2025"].
Airflow runs a targeted policy_sync — re-extracts, re-chunks, re-embeds, UPSERTs.
Auditor appends to the response: "A refresh has been scheduled; please re-ask in ~10 minutes."
The officer re-asks at 14:45 — gets the fresh answer.

8. Production Hardening Checklist

Concern	Mitigation
Watermark race	max_active_runs=1 + advisory lock on watermark update
Poison pills	DLQ table + weekly review dashboard
Cost blowout	Conditional embedding + force_reembed flag
Silent data rot	Great Expectations checks on chunk count, embedding dimension
Schema drift	Schema registry + contract tests between Airflow and pgvector
Airflow ↔ RAG auth	mTLS + service-account JWT with admin:etl_trigger scope
Backfill storms	Pool etl_pool with 4 slots to throttle parallelism
Observability	OpenTelemetry in both Airflow tasks and LangGraph nodes → single trace
Secrets	Airflow Connections backed by Vault (via vault-cluster secrets backend)
Compliance	Every DAG run's conf, logical_date, and task logs archived to S3 Glacier

9. Key Takeaways

Idempotency is a design constraint, not a feature. Every task must be safely re-runnable. Watermarks, deterministic IDs, and UPSERTs are non-negotiable.
Failures are typed. Transient → retry. Deterministic → DLQ. Catastrophic → page. Never conflate them.
Backfills are first-class. The logical_date partitioning model lets you reprocess any time window without corrupting current state.
The RAG system should heal itself. The Auditor agent detecting staleness and triggering targeted backfills closes the feedback loop — this is what separates an enterprise system from a demo.
Airflow and LangGraph are complementary, not competing. Airflow owns the data plane (ETL, batch, orchestration). LangGraph owns the reasoning plane (agents, state, memory). The REST API is the seam.
Cost-awareness is architectural. Conditional embedding based on content hashes can save 90%+ of API spend during routine operations.

Summary

This architecture combines Apache Airflow's strengths in orchestration, idempotent ETL, failure recovery, and backfills with LangGraph's agent-based reasoning capabilities to create a self-healing RAG system. By leveraging deterministic processing, conditional embedding, UPSERT-based loading, targeted backfills, and Auditor-driven refresh workflows, the system continuously improves data freshness while minimizing operational costs, ensuring every step remains auditable, recoverable, and production-ready.