Document Processing Pipeline

Relevant source files

• deepdoc/parser/excel_parser.py
• deepdoc/parser/pdf_parser.py
• deepdoc/vision/layout_recognizer.py
• deepdoc/vision/ocr.py
• deepdoc/vision/recognizer.py
• deepdoc/vision/t_ocr.py
• deepdoc/vision/t_recognizer.py
• deepdoc/vision/table_structure_recognizer.py
• rag/app/book.py
• rag/app/laws.py
• rag/app/manual.py
• rag/app/naive.py
• rag/app/one.py
• rag/app/paper.py
• rag/app/presentation.py
• rag/app/qa.py
• rag/app/table.py
• rag/nlp/__init__.py
• rag/settings.py
• test/unit_test/deepdoc/parser/test_excel_parser.py
• test/unit_test/rag/app/test_qa_csv.py

Purpose and Scope

The Document Processing Pipeline is the core system that transforms raw uploaded documents into searchable, semantically-indexed chunks within RAGFlow. This page documents the complete workflow from document upload through parsing, chunking, enhancement, embedding, and final indexing in the document store.

For details on parsing strategies and format support, see Document Parsing Strategies. For details on chunking logic, see Chunking Methods. For details on vision and OCR, see Vision Processing: OCR and Layout Recognition. For advanced features, see Advanced Features: GraphRAG, RAPTOR, and Agentic RAG. For external integrations, see Data Source Connectors.

---

Pipeline Architecture Overview

High-Level Pipeline Flow

The document processing pipeline follows a multi-stage asynchronous workflow. Documents are uploaded via API, tasks are queued in Redis, and background workers (task executors) process them through parsing, chunking, enhancement, embedding, and final indexing stages.

Pipeline Flow Diagram

Sources: rag/svr/task_executor.py114-131 rag/svr/task_executor.py133-147 api/db/services/task_service.py163-205 api/db/services/document_service.py42-73 rag/nlp/search.py35-36

---

Task Executor and Queue System

Task Collection and Distribution

The task executor service (task_executor.py) runs as a background worker process that continuously polls Redis Streams for new document processing tasks via the TaskManager rag/svr/task_executor.py18-19 Multiple executor instances can run concurrently, each identified by a consumer index TE_IDX rag/svr/task_executor.py168

Task Distribution Architecture

Sources: rag/svr/task_executor.py95-101 rag/svr/task_executor.py167-168 common/constants.py103

Task Types and Routing

The task executor handles multiple task types, each corresponding to different processing modes. The mapping is defined in TASK_TYPE_TO_PIPELINE_TASK_TYPE rag/svr/task_executor.py133-147

Task Type	Pipeline Task Type	Logic / Handler	Description
dataflow	PARSE	Standard Parsing	Standard document parsing through configured parsers
raptor	RAPTOR	RAPTOR_TREE_BUILDER	Hierarchical clustering for improved retrieval rag/svr/task_executor.py87-89
graphrag	GRAPH_RAG	KG Extraction	Knowledge graph entity/relationship extraction
mindmap	MINDMAP	MindMapExtractor	Mind map generation from documents api/db/services/dialog_service.py46
memory	MEMORY	handle_save_to_memory_task	Agent memory persistence rag/svr/task_executor.py44

Sources: rag/svr/task_executor.py133-147 rag/svr/task_executor.py87-89 rag/svr/task_executor.py44 api/db/services/dialog_service.py46

---

Parser Selection and Factory Pattern

FACTORY Dictionary Mapping

The task executor uses a factory pattern to select the appropriate parser based on the document's parser_id. The FACTORY dictionary maps parser types to their corresponding parser modules rag/svr/task_executor.py114-131

Parser Factory Mapping

Sources: rag/svr/task_executor.py114-131 rag/app/naive.py1-63 rag/app/manual.py33-134 rag/app/table.py65-153

Parser Selection Logic

Each parser module provides a method (typically chunk()) that takes the document binary and configuration parameters. The parser is selected at runtime using the parser_id from the task metadata.

The chunk() function is typically invoked with:

• filename: Document name
• binary: Document binary content
• from_page/to_page: Page range for processing rag/app/naive.py115-119
• lang: Document language rag/app/naive.py135
• callback: Progress callback function (set_progress) rag/app/naive.py116
• parser_config: Parser-specific configuration (e.g., layout_recognize, chunk_token_num) rag/app/manual.py141

For details on format-specific parsing, see Document Parsing Strategies.

Sources: rag/svr/task_executor.py189-210 rag/app/naive.py115-128 rag/app/manual.py137-174

---

Chunking Process

Chunk Building Workflow

The processing orchestrates the entire chunking process with timeout protection and error handling. It fetches the document binary from storage and executes the selected chunker using chunk_limiter to control concurrency rag/svr/task_executor.py97

Chunk Building Pipeline

For details on the specific strategies used for different document types, see Chunking Methods.

Sources: rag/svr/task_executor.py97 api/db/services/task_service.py146-161

Chunk Data Structure

Each chunk returned by the parser is transformed into a document structure with fields such as docnm_kwd, content_ltks, and kb_id. The indexing process uses search.index_name(uid) to target the correct document store index rag/nlp/search.py35-36

Field	Type	Description
docnm_kwd	string	Document name keyword rag/app/presentation.py137
content_ltks	string	Tokenized content for full-text search rag/nlp/search.py146
kb_id	string	Knowledge base ID rag/nlp/search.py147
img_id	string	MinIO object ID for associated images
page_num_int	int	Source page number rag/app/presentation.py147
chunk_order_int	int	Order of chunk within the document

Sources: rag/nlp/search.py146-172 rag/nlp/search.py35-36 rag/app/presentation.py137-151

---

Content Enhancement and Embedding

LLM-Based Enhancement

If enabled in parser configuration, the system generates metadata and semantic improvements for each chunk using LLMs. This includes:

• Keyword Extraction: keyword_extraction() extracts significant terms rag/svr/task_executor.py60
• Question Generation: question_proposal() generates hypothetical questions the chunk can answer rag/svr/task_executor.py60
• Metadata Extraction: gen_metadata() extracts structured fields based on a JSON schema rag/svr/task_executor.py60
• Content Tagging: content_tagging() labels content for improved classification rag/svr/task_executor.py60

For details on these features, see Content Enhancement and Embedding.

Sources: rag/svr/task_executor.py60 rag/prompts/generator.py50

Embedding Process

After chunks are built and enhanced, they are embedded into vector representations. The LLMBundle handles communication with the embedding model provider rag/svr/task_executor.py79

The system ensures context preservation by prepending document titles or specific metadata to the chunk content before embedding rag/app/presentation.py137-138

Sources: rag/svr/task_executor.py98 api/db/services/llm_service.py37 rag/app/presentation.py137-151

---

Progress Tracking and Error Handling

Progress Updates

Throughout the pipeline, progress is reported to the database via set_progress() rag/svr/task_executor.py189-210 This allows the UI to display real-time processing status.

Progress values typically follow:

• 0.0 - 0.6: OCR and layout analysis rag/app/manual.py42-49
• 0.6 - 0.7: Chunking rag/app/manual.py61
• 0.7 - 0.9: Enhancement (keywords, questions)
• 0.9 - 1.0: Embedding and indexing
• -1: Error occurred or Task Canceled rag/svr/task_executor.py197

Sources: rag/svr/task_executor.py189-210 api/db/services/task_service.py187-205 rag/app/manual.py42-61

Error Handling and Cancellation

The task executor implements multiple error handling mechanisms:

1. Task Cancellation: Checks has_canceled(task_id) before expensive operations rag/svr/task_executor.py193
2. Retry Logic: Tasks track retry_count and are re-queued if transient failures occur api/db/services/task_service.py189
3. Task Status: Documents and tasks track status using TaskStatus (UNSTART, RUNNING, FAIL, DONE) api/db/services/document_service.py33

Sources: rag/svr/task_executor.py193 api/db/services/task_service.py189 api/db/services/document_service.py33