Airflow Meets LangChain and LlamaIndex: Production AI Pipelines Without the Black Box

Apache Airflow’s common.ai provider now integrates LangChain and LlamaIndex natively, turning opaque agent calls into visible, retryable, auditable production workflows.

At the recent Airflow Summit, roughly one-third of the use case sessions were about using Airflow for AI workloads. That number caught my attention. Data engineers, ML engineers, and now AI engineers are converging on Airflow as their orchestration layer, and they are bringing their AI frameworks with them.

The question we kept hearing was practical: “I have LangChain agents and LlamaIndex RAG pipelines. How do I run them in production with the visibility, retry behavior, and governance I get for my data pipelines?”

With the upcoming apache-airflow-providers-common-ai release, we have an answer. I built the initial LlamaIndex and LangChain integration layer, the DocumentLoaderOperator that bridges Airflow's connector ecosystem to RAG pipelines, and the end-to-end example DAGs that demonstrate the architecture on real financial data. These framework integrations sit alongside the PydanticAI-backed operators that Kaxil Naik and Pavan Kumar designed in AIP-99, and the advanced features Kaxil shipped this cycle: a toolset bridge between common.ai and LangChain, OpenTelemetry tracing, Agent Skills, and capabilities passthrough.

Why This Matters: The Visibility Problem

Every AI framework, whether LangChain, LlamaIndex, or CrewAI, runs multi-step processes internally. An agent that researches a question might make 15 LLM calls, invoke 8 tools, retrieve from 3 indexes, and hit 2 rate limits before producing an answer. From the outside, all you see is: “agent ran.”

That opacity is fine in a notebook. It is a liability in production.

When step 5 of 8 fails, the entire task retries from scratch. When token costs spike, there is no attribution. When compliance asks “who approved this output?”, there is no artifact. When the agent calls a database, credentials are embedded in code instead of governed through RBAC.

The AI frameworks solve the reasoning problem brilliantly. They do not solve the production operations problem, because that is not what they are designed to do.

The Architecture: Outer Loop + Inner Loop

Airflow’s approach is to be the outer orchestration loop around the AI framework’s inner reasoning loop:

ConcernInner loop (AI framework)Outer loop (Airflow)LLM calls, tool use, retrievalLangChain / LlamaIndex / CrewAIScheduling and triggersCron, asset-driven, API-triggeredPer-step visibilityHidden inside agentEach step is a named task with logsRetry on failureEntire agent restartsOnly the failed step retriesCredentialsEnv vars or hardcodedAirflow Connections with RBAC and auditHuman review gatesNot built-inHITL operators (approval, input)Cost controlsManualUsageLimits per taskStructured output validationFramework-specificPydantic models on LLMOperator

The key insight: Airflow does not replace the agent. It decomposes the pipeline around the agent into observable, independently retryable steps, and lets the agent do what it does best within each step.

Why Multi-Framework?

This multi-framework approach mirrors what Airflow has always done for databases. The average production Airflow deployment connects to 3 to 5 data stores. No team standardizes on one database, and nobody asks “Snowflake or BigQuery?” as if they are mutually exclusive. They use both, for different workloads, with the same scheduling, credentials, and retry infrastructure around them. AI frameworks are heading the same direction: LlamaIndex for precision retrieval, LangChain for agent orchestration, direct LLM calls for structured extraction. The question is not which framework to pick. The question is how to run all of them in one governed pipeline.

Datadog’s 2026 State of AI Engineering report confirms this pattern: more than 70% of organizations now use three or more models, and framework adoption nearly doubled in the past year. As the report puts it, teams are “building model portfolios in order to use the optimal model for each workload’s latency, cost, operational risk, and task requirements.” Different models, different frameworks, one pipeline.

What’s Shipping soon

LlamaIndex: Embedding and Retrieval Operators

Two new operators bring LlamaIndex’s RAG capabilities into the Airflow task graph:

• **LlamaIndexEmbeddingOperator** chunks documents, generates embeddings, and persists a vector index. It supports cloud storage (S3, GCS) via ObjectStoragePath, configurable chunk size and overlap, and any embedding model (OpenAI by default, with BYO BaseEmbedding for other vendors).
• **LlamaIndexRetrievalOperator** loads a persisted index and performs similarity search. It returns ranked chunks with text, score, and metadata, ready for downstream synthesis.

Both operators use a new LlamaIndexHook that bridges Airflow connections to LlamaIndex model constructors. Importantly, the hook passes models directly to LlamaIndex constructors rather than mutating the global Settings singleton, so concurrent tasks in the same worker do not race on shared state.

# Weekly indexing: chunk and embed documents into a persistent vector index
embed = LlamaIndexEmbeddingOperator(
    task_id="embed_docs",
    documents=load.output,
    embed_model="text-embedding-3-small",
    llm_conn_id="llamaindex_default",
    chunk_size=512,
    persist_dir="/opt/airflow/data/indexes/kb_index",
)

# On-demand retrieval: similarity search against the persisted index
retrieve = LlamaIndexRetrievalOperator(
    task_id="retrieve",
    query="{{ params.question }}",
    index_persist_dir="/opt/airflow/data/indexes/kb_index",
    embed_model="text-embedding-3-small",
    llm_conn_id="llamaindex_default",
    top_k=5,
)

LangChain: Hook-Based Integration

LangChain’s integration takes a different shape, reflecting its architecture. LangChain chains and agents are self-contained: they compose their own multi-step logic internally. The integration surface is a LangChainHook that bridges Airflow connections to LangChain's model constructors via constructor injection:

from airflow.providers.common.ai.hooks.langchain import LangChainHook

# Model identifiers set on the hook; credentials resolved from the Airflow connection.
hook = LangChainHook(
    llm_conn_id="langchain_default",
    llm_model="openai:gpt-4o",
    embed_model="openai:text-embedding-3-small",
)
chat_model = hook.get_chat_model()       # vendor-agnostic dispatch via init_chat_model()
embeddings = hook.get_embedding_model()  # vendor-agnostic dispatch via init_embeddings()

This means LangChain code runs inside @task functions with Airflow-managed credentials. There is no env var leakage between concurrent tasks, no credentials in code, and a full audit trail on which connection was used.

DocumentLoaderOperator: The RAG Bridge

Both integrations build on DocumentLoaderOperator, a framework-agnostic file parser that converts raw files (PDF, DOCX, CSV, Markdown, plain text) into list[dict(text, metadata)], the universal input format for any embedding pipeline. This is the adapter that turns Airflow's 1,000+ provider connectors into RAG data sources:

S3Hook → DocumentLoaderOperator → LlamaIndexEmbeddingOperator
GCSHook → DocumentLoaderOperator → LangChain embeddings in @task
SFTPHook → DocumentLoaderOperator → any embedding pipeline

Seeing It in Action: SEC 10-K Financial Analysis

To prove this architecture works on real data, I authored end-to-end example DAGs that analyze SEC 10-K filings from the EDGAR API, the same public filings that financial analysts review quarterly. We wanted to demonstrate the architecture on real financial data, not toy prompts. Both a LlamaIndex version and a LangChain version exist, analyzing companies including Apple, Microsoft, Uber, Lyft, and Amazon.

Each example has two DAGs:

Indexing DAG (scheduled weekly): Fetches the latest 10-K filing for each company from SEC EDGAR, extracts key sections (Risk Factors, Management Discussion & Analysis), chunks and embeds them into a per-company vector index.

Analysis DAG (triggered on demand): An analyst submits tickers and a question through a Human-in-the-Loop input task. The LLM decomposes the question into sub-questions via Dynamic Task Mapping, creating one retrieval task per sub-question, each independently retryable. Results are synthesized into a structured report (Pydantic model), formatted into readable markdown, and presented to the analyst for approval before delivery.

What the DAG Graph Shows That a Notebook Hides

In the Airflow UI, the 10-K analysis DAG renders as a sequence of named tasks, each one a production concern made visible:

1. **analyst_input** is the HITL entry point. An analyst submits tickers and a research question. The DAG pauses until input arrives.
2. **fetch_filing [N]** uses Dynamic Task Mapping to create one instance per ticker at runtime. If Apple's EDGAR fetch hits a rate limit, only that instance retries. Microsoft's and Uber's filings are already cached in XCom.
3. **embed_filing [N]** chunks and embeds each company's 10-K sections into a per-company vector index. If one embedding call fails, the others are preserved.
4. **decompose_question** is where the inner reasoning loop runs. The LLM breaks the analyst's question into targeted sub-questions. Structured Pydantic output enforces the schema.
5. **retrieve_evidence [M]** retrieves from the right company's index for each sub-question. M retrieval tasks, each independently retryable, each with its own similarity scores in XCom.
6. **synthesize_report** performs cross-company synthesis with UsageLimits capping the token spend. If this fails, all individual retrievals are preserved. No re-embedding needed.
7. **format_report followed by `review_report`** transforms the structured output into readable markdown, then presents it to the analyst for approval. There is an audit trail of who approved what, and when.

None of this is visible when you call agent.run() in a single task.

The LlamaIndex Version: Sub-Question Decomposition

The LlamaIndex 10-K example showcases a pattern that is hard to replicate without a retrieval-specialized framework. When an analyst asks “Compare the risk profiles of Uber and Lyft regarding regulatory challenges,” the LLM decomposes this into targeted sub-questions:

"What regulatory risks does Uber disclose in its 10-K?"
"What regulatory risks does Lyft disclose in its 10-K?"
"How do Uber and Lyft differ in their regulatory risk mitigation strategies?"

Each sub-question becomes a separate retrieval task via Dynamic Task Mapping, hitting the right company’s vector index with a focused query. This is LlamaIndex’s strength: precision retrieval with sub-question decomposition, where each retrieval step is visible and retryable.

# LLM decomposes the analyst's question into sub-questions
sub_questions = LLMOperator.partial(
    task_id="decompose_question",
    output_type=SubQuestionList,  # Pydantic model enforces structure
    llm_conn_id=LLM_CONN_ID,
).expand(prompt=formatted_prompts)

# Each sub-question retrieves from the appropriate company index
retrieval_results = LlamaIndexRetrievalOperator.partial(
    task_id="retrieve_evidence",
    llm_conn_id="llamaindex_default",
    embed_model="text-embedding-3-small",
    top_k=8,
).expand_kwargs(retrieval_params)

The LangChain Version: FAISS Vector Store

The LangChain counterpart uses FAISS for the vector store and LangChain’s embedding and retrieval abstractions. Same architecture, same live EDGAR data, same HITL flow, different inner loop. This demonstrates that the outer-loop pattern (Airflow’s scheduling, Dynamic Task Mapping, HITL, structured output) is framework-agnostic. Teams can choose LangChain or LlamaIndex based on their retrieval needs, not their orchestration needs.

Toolset Bridge: common.ai Tools in LangChain Agents

Airflow’s common.ai ships curated toolsets (SQLToolset, HookToolset, MCPToolset) that work natively with AgentOperator. Kaxil added a bridge in the other direction: airflow_toolset_to_langchain_tools() converts any of these toolsets into LangChain StructuredTool objects, so a LangChain agent running inside an Airflow task can call Airflow's connection-managed, validated tools. The forward direction (LangChain tools into AgentOperator) is already covered by pydantic-ai's upstream LangChainToolset. Both bridge directions now live in common.ai.

Agent Skills

AgentSkillsToolset loads agentskills.io SKILL.md bundles from a local directory or a Git repository. Git credentials come from an Airflow connection (HTTPS token or SSH key) resolved through GitHook, with cleartext HTTP and credential-bearing URLs rejected. Sources are resolved on the worker when the agent enters the toolset, so tokens are never baked into the serialized DAG. Pass it via AgentOperator(toolsets=...) or use it with a raw pydantic-ai Agent. A framework-agnostic resolve_skills() helper returns local SKILL.md directories for other Agent Skills loaders (LangChain DeepAgents, Strands).

Capabilities Passthrough

With the pydantic-ai floor bumped to 1.71.0, users can pass pydantic-ai capabilities (Thinking, WebSearch, ImageGeneration, MCP) through AgentOperator(agent_params=...) without waiting for first-class operator-level support. An example DAG demonstrates the pattern with Thinking and WebSearch composed alongside SQLToolset.

OpenTelemetry Tracing

PydanticAIHook.create_agent() now attaches pydantic-ai's native OpenTelemetry instrumentation so agent, model, and tool spans (with token usage) export through Airflow's existing OTLP exporter and nest under the task span. Gated by [common.ai] otel_export_enabled (off by default). Content capture is separately gated by [common.ai] capture_content.

You Don’t Always Need a Framework

Not every AI pipeline needs LangChain or LlamaIndex. If your workload is “call an LLM with structured output,” Airflow’s LLMOperator handles it directly with 9 named model providers (OpenAI, Anthropic, Google Gemini, AWS Bedrock, Azure OpenAI, Groq, Mistral, Ollama, and vLLM) plus any OpenAI-compatible endpoint. The framework integrations are there for when you need specialized capabilities like retrieval, embedding, tool-calling agents, or multi-agent collaboration.

Two Connection Types, One Pipeline

A practical detail worth highlighting: the example DAGs use two different connections in the same pipeline:

• **pydanticai_default** for the LLM (analysis, synthesis, sub-question decomposition). Backed by Anthropic, OpenAI, or any supported provider.
• **llamaindex_default or `langchain_default`** for embeddings. Backed by OpenAI's embedding API (or any compatible endpoint).

This is not a limitation. It is a feature. The analysis LLM and the embedding model often come from different providers (for example, Anthropic Claude for reasoning and OpenAI for embeddings). This is the AI equivalent of using one connection for your Postgres transactional database and another for your Snowflake analytics warehouse, in the same DAG. Airflow Connections make this explicit, governed, and auditable rather than buried in env var combinations.

What’s Next

With hooks, operators, toolset bridges, observability, and Agent Skills now shipped, the common.ai provider is a bidirectional adapter layer between Airflow and the AI framework ecosystem. The roadmap continues:

• **@task.langchain and @task.llamaindex decorators** for a zero-change on-ramp for existing framework code (blocked on the BaseAIHook abstraction that unifies the hook layer across frameworks)
• CrewAI integration enabling multi-agent role-based collaboration with per-agent visibility
• OpenAI and Claude Code Agent Integrations enabling direct invocation of those agents

The outer loop / inner loop architecture means each framework plugs into the same production infrastructure. Teams choose their AI framework based on what it is good at, not based on which orchestrator they are locked into.

Try It

The example DAGs are in providers/common/ai/example_dags/:

• [example_llamaindex_rag.py](https://github.com/apache/airflow/blob/main/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llamaindex_rag.py) covers LlamaIndex RAG pipeline patterns (full pipeline, split index/query, multi-source)
• [example_llamaindex_10k.py](https://github.com/apache/airflow/blob/main/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llamaindex_10k.py) is the SEC 10-K financial analysis with LlamaIndex
• [example_langchain_10k.py](https://github.com/apache/airflow/blob/main/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_langchain_10k.py) is the SEC 10-K financial analysis with LangChain

To get started:

uv pip install -U 'apache-airflow-providers-common-ai[openai]'

Create an LLM connection (pydanticai_default) in the Airflow UI, trigger a DAG, and watch each step execute as a visible, retryable, auditable task.

We would love to hear how you are using these integrations. Join the conversation on the Apache Airflow Slack in #airflow-ai or open an issue on GitHub.

Apache Airflow 3.3 with apache-airflow-providers-common-ai v0.3.0. The LlamaIndex operators, LangChain hook, toolset bridge, Agent Skills, OTel tracing, and 10-K example DAGs are all available now.