# Data Flow from arXiv to OpenSearch Indexing: A Production RAG Pipeline Deep Dive

> Understand the data flow from arXiv to OpenSearch indexing in a production RAG pipeline. Learn about PDF parsing, chunking, embedding generation with Jina AI, and OpenSearch hybrid indexing.

- Repository: [jamwithai/production-agentic-rag-course](https://github.com/jamwithai/production-agentic-rag-course)
- Tags: deep-dive
- Published: 2026-03-23

---

**The pipeline fetches papers from the arXiv API, downloads and parses PDFs into text, splits content into metadata-rich chunks, generates 768-dimensional embeddings via Jina AI, and bulk-indexes everything into an OpenSearch hybrid index supporting both BM25 lexical search and KNN vector similarity.**

The `jamwithai/production-agentic-rag-course` repository implements a fully automated ingestion system that transforms raw arXiv submissions into a searchable hybrid knowledge base. Understanding the complete data flow from arXiv to OpenSearch indexing is critical for debugging latency issues and optimizing retrieval performance in production RAG applications. The architecture follows a four-stage ETL pattern: ingestion, enrichment, transformation, and indexing.

## Stage 1: Ingesting Papers from the arXiv API

The pipeline begins in [`src/services/arxiv/client.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/arxiv/client.py) with the **`ArxivClient`** class. This component queries the arXiv REST API at `https://export.arxiv.org/api/query`, handling pagination, rate-limiting, and optional date filters to retrieve new submissions.

The client returns structured **`ArxivPaper`** objects containing essential metadata: title, authors, abstract, publication date, and the direct PDF URL. For scheduled production workloads, the Airflow DAG defined in [`airflow/dags/arxiv_ingestion/fetching.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/airflow/dags/arxiv_ingestion/fetching.py) invokes the factory function `make_arxiv_client()` from [`src/services/arxiv/factory.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/arxiv/factory.py) to instantiate the client and trigger the fetch operation.

## Stage 2: PDF Retrieval and Metadata Enrichment

Once metadata is retrieved, the **`MetadataFetcher`** in [`src/services/metadata_fetcher.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/metadata_fetcher.py) coordinates the download phase. It calls **`download_pdf`** (also located in [`src/services/arxiv/client.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/arxiv/client.py)) to retrieve the binary PDF content, caching files locally to avoid redundant network requests.

Raw PDF bytes are then passed to [`src/services/pdf_parser.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/pdf_parser.py), which uses `pdfminer` to extract plain text. The enriched record—containing both the structured metadata and the full text content—is then passed downstream, ready for segmentation.

## Stage 3: Text Chunking and Vector Embedding

With the full text extracted, the pipeline moves to [`src/services/indexing/text_chunker.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/indexing/text_chunker.py) where the **`TextChunker`** performs section-aware splitting. This preserves document structure while creating searchable units, retaining critical provenance fields like `arxiv_id`, `paper_id`, and `chunk_index` in every segment.

Each `TextChunk` is then sent to **`JinaEmbeddingClient`** located in [`src/services/embeddings/jina_client.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/embeddings/jina_client.py). This service calls the Jina AI embedding API to generate **768-dimensional dense vectors** representing the semantic meaning of each chunk. These vectors enable semantic similarity search alongside traditional keyword matching.

## Stage 4: Hybrid Indexing in OpenSearch

The final stage occurs in [`src/services/opensearch/client.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/opensearch/client.py) through the **`OpenSearchClient`** class. The system uses a hybrid index configuration defined in [`src/services/opensearch/index_config_hybrid.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/opensearch/index_config_hybrid.py), creating the `arxiv-papers-chunks` index with two distinct field types:

- **BM25 fields** for inverted-index lexical search on titles, abstracts, and chunk text
- **KNN vector fields** optimized for efficient nearest-neighbor search on the 768-dim embeddings

The **`HybridIndexingService`** in [`src/services/indexing/hybrid_indexer.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/indexing/hybrid_indexer.py) orchestrates the bulk write operation via **`bulk_index_chunks()`**, persisting chunks alongside their embeddings. For query-time fusion, the client supports the **`HYBRID_RRF_PIPELINE`** (Reciprocal Rank Fusion), which automatically combines BM25 and vector similarity scores into a single relevance ranking.

## Orchestrating the End-to-End Pipeline

While the Airflow DAG manages scheduling, you can trigger the complete data flow programmatically using the factory functions and service classes. Below are runnable examples that demonstrate the end-to-end orchestration.

First, initialize the core services and ensure the hybrid index exists:

```python
from src.services.arxiv.factory import make_arxiv_client
from src.services.opensearch.client import OpenSearchClient
from src.services.indexing.factory import make_hybrid_indexing_service

# Initialise OpenSearch client (reads settings from `src/config.py`)

opensearch = OpenSearchClient()

# Ensure the hybrid index exists

opensearch.setup_hybrid_index(force=False)

# Build the indexing service that stitches together the components

indexer = make_hybrid_indexing_service(opensearch_client=opensearch)

```

To ingest a specific paper by its arXiv ID:

```python
import asyncio
from src.services.arxiv.client import ArxivClient

async def ingest_one(arxiv_id: str):
    arxiv_client: ArxivClient = make_arxiv_client()
    # Fetch the paper metadata (single-paper endpoint)

    paper = await arxiv_client.fetch_paper_by_id(arxiv_id)

    # Download PDF, extract text, and prepare chunks + embeddings

    stats = await indexer.index_paper(paper.dict())
    print("Indexing stats:", stats)

# Example usage

asyncio.run(ingest_one("2301.00001"))

```

For batch ingestion of recent submissions:

```python
import asyncio

async def bulk_ingest(max_results: int = 10):
    arxiv_client = make_arxiv_client()
    # Pull the most recent N papers from arXiv

    papers = await arxiv_client.fetch_papers(max_results=max_results)

    # Convert the Pydantic models to plain dicts for the indexer

    paper_dicts = [p.dict() for p in papers]

    # Run batch indexing (creates chunks, embeddings, and bulk-indexes)

    batch_stats = await indexer.index_papers_batch(paper_dicts, replace_existing=False)
    print("Batch stats:", batch_stats)

asyncio.run(bulk_ingest(20))

```

## Summary

- The **`ArxivClient`** in [`src/services/arxiv/client.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/arxiv/client.py) fetches metadata and handles API pagination for the arXiv source.
- **`MetadataFetcher`** and [`pdf_parser.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/pdf_parser.py) manage PDF download and text extraction, caching content locally.
- **`TextChunker`** splits papers into logical, metadata-rich chunks while preserving identifiers like `arxiv_id` and `chunk_index`.
- **`JinaEmbeddingClient`** generates 768-dimensional dense vectors for semantic search capabilities.
- **`HybridIndexingService`** bulk-indexes chunks into the `arxiv-papers-chunks` OpenSearch index, configured with both BM25 and KNN fields for hybrid retrieval.
- An Airflow DAG in `airflow/dags/arxiv_ingestion/` orchestrates the entire flow, though components can be invoked directly via factory functions.

## Frequently Asked Questions

### How does the pipeline handle arXiv API rate limits?

The `ArxivClient` implementation in [`src/services/arxiv/client.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/arxiv/client.py) includes built-in pagination logic and rate-limiting mechanisms to respect the `export.arxiv.org` API constraints. This ensures reliable metadata fetching for large batches without triggering upstream throttling or IP blocks.

### What chunking strategy is used for the paper text?

The `TextChunker` class in [`src/services/indexing/text_chunker.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/indexing/text_chunker.py) employs a section-aware splitting strategy that maintains the logical structure of academic papers. It preserves critical metadata fields including `arxiv_id`, `paper_id`, and `chunk_index` in every chunk, ensuring traceability from any retrieved segment back to its original source document.

### How does the hybrid search combine BM25 and vector similarity?

The system stores data in a hybrid index defined by [`index_config_hybrid.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/index_config_hybrid.py), which contains both inverted-index text fields for BM25 scoring and dense vector fields for KNN search. During querying, the `OpenSearchClient` can utilize the `HYBRID_RRF_PIPELINE` to apply Reciprocal Rank Fusion, automatically combining lexical relevance scores from BM25 with semantic similarity scores from vector search into a unified ranking.

### Can the indexing process be triggered outside of the Airflow scheduler?

Yes. The factory functions `make_arxiv_client()` and `make_hybrid_indexing_service()` in [`src/services/arxiv/factory.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/arxiv/factory.py) and [`src/services/indexing/factory.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/indexing/factory.py) allow full programmatic access. You can instantiate these services in custom scripts, Jupyter notebooks, or FastAPI endpoints and call `indexer.index_paper()` or `indexer.index_papers_batch()` directly without depending on the Airflow DAG infrastructure.