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

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 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 invokes the factory function make_arxiv_client() from 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 coordinates the download phase. It calls download_pdf (also located in 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, 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 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. 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 through the OpenSearchClient class. The system uses a hybrid index configuration defined in 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 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:

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:

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:

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 fetches metadata and handles API pagination for the arXiv source.
• MetadataFetcher and 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 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 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, 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 and 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}