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:
airflow dags backfill policy_sync --start-date 2023-01-01 --end-date 2026-07-14
trigger a run with conf={"doc_ids": ["pol_123", "pol_456"]}
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.