Build a RAG document ingestion pipeline (Crawl4AI + Unstructured)

Introduction

When you ship a RAG system that mixes web pages, PDFs, DOCX files, and image uploads, the layer that breaks first is almost always ingestion — duplicated chunks, lost metadata, and silent parse failures poison every retrieval that follows. Teams that treat ingestion as an afterthought end up rebuilding their vector store from scratch after the first content refresh, and downstream relevance scores collapse with no obvious culprit. This lesson walks through building a unified document ingestion pipeline that accepts both web URLs and uploaded files (PDF, DOCX, images) and produces a consistent stream of embedded, indexed chunks in PostgreSQL with pgvector. By the end, you will be able to wire Crawl4AI for the web-native path, Unstructured for the file-native path, and a shared chunking + embedding stage that lands content in a chunks table carrying both embedding and tsvector columns—ready for hybrid semantic + BM25 retrieval in downstream goals.

Key Terminology

• Crawl4AI: an async web crawler (AsyncWebCrawler) that renders JavaScript-heavy pages and emits clean Markdown for the web ingestion path.
• Unstructured partitioners: format-aware parsers (partition_pdf, partition_docx, partition_image) that extract layout-preserving text from uploaded files, including OCR for images.
• ChunkRecord / IngestionResult: the dataclass contract every parser returns—ChunkRecord carries content, embedding vector, source metadata, and a deterministic chunk_id; IngestionResult aggregates the per-source outcome and non-fatal errors.
• pgvector + tsvector: the dual-column storage strategy on the chunks table—embedding (pgvector) powers semantic similarity, tsvector powers PostgreSQL BM25 full-text search, both populated at insert time.

Concepts

Why Two Ingestion Paths Matter

Enterprise knowledge bases are never homogeneous. Engineering documentation lives on Confluence and internal wikis. Policy documents arrive as PDFs. Research papers come as uploaded files. Product specs sit in Google Docs exported to DOCX. A production ingestion pipeline must handle all of these without requiring manual format conversion. Crawl4AI handles the web-native path—it renders JavaScript, extracts structured markdown from HTML, and handles pagination across multi-page documentation sites. Unstructured handles the file-native path—it uses layout-aware parsing for PDFs (detecting headers, tables, and figure captions), processes DOCX paragraph structures, and applies OCR to images. The critical design decision is that both paths must produce identical output: a list of Document chunks with text content, embedding vectors, and structured metadata. This uniformity is what allows the downstream retrieval, reranking, and agentic RAG layers to operate without knowing or caring where a chunk originated.

Key Design Decisions and Trade-offs

• Term: Chunk size 512 with 64 overlap — A 512-token chunk balances retrieval precision against context sufficiency. Smaller chunks (128-256) improve precision but lose surrounding context. Larger chunks (1024+) include more context but dilute the embedding signal. The 64-token overlap ensures that sentences split at boundaries are recoverable from adjacent chunks.

• Term: Deterministic chunk IDs — Hashing content into chunk IDs makes ingestion idempotent. Re-crawling a documentation site that changed one page only updates the affected chunks. Without deterministic IDs, every re-ingestion creates duplicates that degrade retrieval quality.

• Term: Inline tsvector generation — Computing the tsvector at INSERT time rather than via a trigger ensures the column is always populated, even during bulk loads. The 'english' dictionary applies stemming (running → run) and stop-word removal, which directly impacts BM25 recall in the hybrid retrieval step.

• Term: Error accumulation vs. exception propagation — The pipeline returns error lists in IngestionResult instead of raising exceptions. This design supports batch ingestion where failing on one document out of hundreds would be unacceptable. Callers inspect the errors list and decide whether to retry, log, or alert.

The ingestion pipeline you have built here produces the foundational chunk data that every subsequent layer depends on. The pgvector embeddings enable semantic search, the tsvector columns enable BM25 search, and the structured metadata enables source filtering—all of which converge in the hybrid retrieval implementation covered next. When you reach the LlamaIndex Workflows section, you will wire this pipeline into an async step that triggers ingestion as part of a larger orchestrated flow with OpenTelemetry tracing, and in the agentic RAG section, the quality of these chunks directly determines whether the agent's iterative retrieval loop converges or spirals into repeated reformulations.

Code Walkthrough

Ingestion Architecture

The following diagram traces a document from source to storage, showing how web URLs and uploaded files converge through format-specific parsers into a shared chunking and embedding stage before landing in pgvector.

This Mermaid flowchart maps the dual-input ingestion pipeline where ingest_url routes web content through Crawl4AI's AsyncWebCrawler while ingest_file dispatches uploads via a Content Type Router to format-specific parsers—partition_pdf, partition_docx, and partition_image. Both paths converge into raw Markdown, which RecursiveCharacterTextSplitter chunks before OpenAI's text-embedding-3-small model generates vectors stored in PostgreSQL with pgvector, populating a chunks table that carries both embedding and tsvector columns for hybrid semantic-plus-keyword retrieval.

Loading diagram...

Notice that the chunks table stores both the embedding vector column (for semantic search in another goal) and a tsvector column (for BM25 full-text search). This dual-column design is intentional—it allows the hybrid retrieval layer to query a single table using two different similarity paradigms without requiring data duplication across separate stores.

Core Data Model

Before writing any ingestion logic, you need a chunk representation that carries enough metadata for downstream filtering, citation, and deduplication. The ChunkRecord dataclass defines the schema that maps directly to your PostgreSQL table. The IngestionResult dataclass wraps the output of any ingestion call, carrying the list of created chunks along with source-level metadata like the original URL or filename. These two structures form the contract between ingestion and storage—every parser must ultimately produce a list of ChunkRecord instances regardless of input format.

1from dataclasses import dataclass, field
2from datetime import datetime
3from enum import Enum
4
5class SourceType(str, Enum):
6    WEB = "web"
7    PDF = "pdf"
8    DOCX = "docx"
9    IMAGE = "image"
10
11@dataclass
12class ChunkRecord:
13    chunk_id: str
14    content: str
15    embedding: list[float]
16    source_type: SourceType
17    source_uri: str
18    chunk_index: int
19    token_count: int
20    metadata: dict = field(default_factory=dict)
21    created_at: datetime = field(default_factory=datetime.utcnow)
22
23@dataclass
24class IngestionResult:
25    source_uri: str
26    source_type: SourceType
27    chunks_created: int
28    chunk_ids: list[str]
29    total_tokens: int
30    errors: list[str] = field(default_factory=list)

The IngestionPipeline Class — Dispatch, Chunking, Embedding, Storage

The central class that orchestrates both ingestion paths is IngestionPipeline. It exposes two public methods: ingest_url accepts a URL string and delegates to Crawl4AI's AsyncWebCrawler to render the page and extract markdown content; ingest_file accepts a FastAPI UploadFile and routes to the appropriate Unstructured partition function based on the file's content type via a MIME_DISPATCH dictionary that replaces a fragile if-elif chain. Both methods converge on the private _chunk_and_store method, where the raw text is split with LangChain's RecursiveCharacterTextSplitter, embedded in a single batched OpenAI call, assigned deterministic SHA-256-derived chunk IDs for idempotent upserts, and written to PostgreSQL with the tsv column generated inline via to_tsvector('english', $2) to power BM25 retrieval. The snippet below combines all of this — dispatch, ingestion entry points, and the shared chunk/embed/store stage — into a single class so you can read the full path from input to row insert in one place.

Code snippetpython
1import hashlib
2from fastapi import UploadFile
3from crawl4ai import AsyncWebCrawler
4from unstructured.partition.pdf import partition_pdf
5from unstructured.partition.docx import partition_docx
6from unstructured.partition.image import partition_image
7from langchain.text_splitter import RecursiveCharacterTextSplitter
8from openai import AsyncOpenAI
9
10class IngestionPipeline:
11    MIME_DISPATCH = {
12        "application/pdf": (partition_pdf, SourceType.PDF),
13        "application/vnd.openxmlformats-officedocument"
14        ".wordprocessingml.document": (partition_docx, SourceType.DOCX),
15        "image/png": (partition_image, SourceType.IMAGE),
16        "image/jpeg": (partition_image, SourceType.IMAGE),
17    }
18
19    def __init__(self, db_pool, embed_model: str = "text-embedding-3-small"):
20        self._pool = db_pool
21        self._openai = AsyncOpenAI()
22        self._embed_model = embed_model
23        self._splitter = RecursiveCharacterTextSplitter(
24            chunk_size=512, chunk_overlap=64,
25            separators=["\n\n", "\n", ". ", " "],
26        )
27
28    async def ingest_url(self, url: str) -> IngestionResult:
29        async with AsyncWebCrawler() as crawler:
30            result = await crawler.arun(url=url)
31            if not result.success:
32                return IngestionResult(
33                    source_uri=url, source_type=SourceType.WEB,
34                    chunks_created=0, chunk_ids=[], total_tokens=0,
35                    errors=[result.error_message or "Crawl failed"],
36                )
37            return await self._chunk_and_store(
38                text=result.markdown, source_uri=url,
39                source_type=SourceType.WEB,
40                metadata={"title": result.metadata.get("title", "")},
41            )
42
43    async def ingest_file(self, upload: UploadFile) -> IngestionResult:
44        content_type = upload.content_type or ""
45        dispatch_entry = self.MIME_DISPATCH.get(content_type)
46        if dispatch_entry is None:
47            return IngestionResult(
48                source_uri=upload.filename or "unknown",
49                source_type=SourceType.PDF, chunks_created=0,
50                chunk_ids=[], total_tokens=0,
51                errors=[f"Unsupported content type: {content_type}"],
52            )
53        partition_fn, source_type = dispatch_entry
54        raw_bytes = await upload.read()
55        elements = partition_fn(file=raw_bytes)
56        text = "\n\n".join(el.text for el in elements if el.text)
57        return await self._chunk_and_store(
58            text=text, source_uri=upload.filename or "upload",
59            source_type=source_type,
60            metadata={"content_type": content_type},
61        )
62
63    async def _chunk_and_store(
64        self, text: str, source_uri: str,
65        source_type: SourceType, metadata: dict,
66    ) -> IngestionResult:
67        chunks = self._splitter.split_text(text)
68        if not chunks:
69            return IngestionResult(
70                source_uri=source_uri, source_type=source_type,
71                chunks_created=0, chunk_ids=[], total_tokens=0,
72            )
73        resp = await self._openai.embeddings.create(
74            input=chunks, model=self._embed_model,
75        )
76        embeddings = [item.embedding for item in resp.data]
77        total_tokens = resp.usage.total_tokens
78        records = []
79        for idx, (chunk_text, emb) in enumerate(zip(chunks, embeddings)):
80            cid = hashlib.sha256(
81                f"{source_uri}::{idx}::{chunk_text[:128]}".encode()
82            ).hexdigest()[:16]
83            records.append(ChunkRecord(
84                chunk_id=cid, content=chunk_text, embedding=emb,
85                source_type=source_type, source_uri=source_uri,
86                chunk_index=idx,
87                token_count=len(chunk_text.split()),
88                metadata=metadata,
89            ))
90        async with self._pool.acquire() as conn:
91            await conn.executemany(
92                """INSERT INTO chunks
93                   (chunk_id, content, embedding, source_type,
94                    source_uri, chunk_index, token_count, metadata,
95                    tsv)
96                   VALUES ($1, $2, $3::vector, $4, $5, $6, $7,
97                           $8::jsonb,
98                           to_tsvector('english', $2))
99                   ON CONFLICT (chunk_id) DO UPDATE
100                   SET content = EXCLUDED.content,
101                       embedding = EXCLUDED.embedding""",
102                [(r.chunk_id, r.content, r.embedding, r.source_type.value,
103                  r.source_uri, r.chunk_index, r.token_count,
104                  r.metadata) for r in records],
105            )
106        return IngestionResult(
107            source_uri=source_uri, source_type=source_type,
108            chunks_created=len(records),
109            chunk_ids=[r.chunk_id for r in records],
110            total_tokens=total_tokens,
111        )

• MIME_DISPATCH (class body): maps MIME type strings to tuples of (partition function, source type enum). Adding a new format is a single line edit rather than a new code branch. The DOCX MIME string is wrapped for readability but concatenates at parse time.
• __init__: accepts a database connection pool and an optional embedding model name. The RecursiveCharacterTextSplitter is configured with a 512-token chunk size and 64-token overlap; the separators list defines split priority (paragraph → line → sentence → space), which preserves semantic coherence within chunks.
• ingest_url: opens an async Crawl4AI session, fetches the page, and on failure returns an IngestionResult with the error message rather than raising — so batch callers can keep going. On success, it forwards the rendered markdown and the page title into _chunk_and_store.
• ingest_file: reads the MIME type from the UploadFile, looks up the matching partition function in MIME_DISPATCH, returns an unsupported-type error if none matches, otherwise reads the bytes, partitions into Unstructured Element objects, joins their text with double newlines to preserve paragraph structure, and delegates to _chunk_and_store.
• _chunk_and_store — splitter + batched embeddings: splits the raw text into chunks and short-circuits with zero chunks if the input is empty, avoiding a pointless API call. A single batched await self._openai.embeddings.create request embeds every chunk at once — roughly 10x faster than per-chunk calls — and resp.usage.total_tokens feeds cost tracking.
• _chunk_and_store — deterministic IDs + ChunkRecord build: the loop pairs each chunk with its embedding and constructs a ChunkRecord whose ID is the first 16 hex characters of sha256(source_uri::idx::chunk_text[:128]). The same input produces the same ID on every re-ingest, which is what makes the upsert below idempotent.
• _chunk_and_store — batched INSERT with inline tsvector: executemany writes all records in one round trip. The SQL casts the embedding parameter with $3::vector, generates the tsv column inline via to_tsvector('english', $2) (so BM25 search is ready immediately on insert), and the ON CONFLICT (chunk_id) DO UPDATE clause updates content and embedding rather than failing or duplicating when a re-ingest hits the same ID.

Do's and Don'ts

Do's

1. ✓Do store both an embedding vector column and a tsvector column in the same chunks table — this dual-column design lets the hybrid retrieval layer run semantic and BM25 queries against a single table without duplicating content across two separate stores, so downstream reranking sees a unified result set.
2. ✓Do route uploaded files through a MIME_DISPATCH dictionary that maps content-type strings to partition_pdf, partition_docx, and partition_image — centralizing dispatch in a dictionary makes adding new Unstructured parsers a one-line change and eliminates the brittle if-elif chain that silently falls through on unrecognized MIME types.
3. ✓Do assign deterministic SHA-256-derived chunk_id values and preserve chunk_index in every ChunkRecord — deterministic IDs make repeated ingestion calls idempotent upserts rather than duplicate insertions, and chunk_index lets the retrieval layer fetch adjacent chunks for context expansion without reordering.

Don'ts

1. ✗Don't split the embedding store and the full-text keyword index across two separate tables — the chunks table's dual embedding + tsvector columns are designed to serve both semantic and BM25 queries from one place; splitting them forces data duplication and means a document update must stay synchronized across two tables on every content refresh.
2. ✗Don't generate chunk_id values with random UUIDs or timestamps — non-deterministic IDs mean that re-running ingest_url or ingest_file on the same source appends duplicate rows instead of upserting, and the vector store accumulates stale embeddings that poison relevance scores with no visible error.
3. ✗Don't call the OpenAI text-embedding-3-small API once per chunk in a loop — _chunk_and_store batches all chunks from a single document in one API call; per-chunk calls multiply embedding latency and token cost linearly with document length and blow past rate limits on any document longer than a few pages.