Task Execution and Queue System

Relevant source files

• admin/server/admin_server.py
• api/apps/restful_apis/chunk_api.py
• api/apps/restful_apis/dataset_api.py
• api/apps/services/dataset_api_service.py
• api/db/__init__.py
• api/db/db_models.py
• api/db/services/dialog_service.py
• api/db/services/document_service.py
• api/db/services/knowledgebase_service.py
• api/db/services/task_service.py
• api/db/services/user_service.py
• api/ragflow_server.py
• common/doc_store/infinity_conn_base.py
• conf/infinity_mapping.json
• rag/advanced_rag/knowlege_compile/_common.py
• rag/advanced_rag/knowlege_compile/dataset_nav.py
• rag/advanced_rag/knowlege_compile/raptor.py
• rag/advanced_rag/knowlege_compile/runner.py
• rag/advanced_rag/knowlege_compile/structure.py
• rag/advanced_rag/knowlege_compile/wiki.py
• rag/flow/compiler/compiler.py
• rag/nlp/query.py
• rag/nlp/rag_tokenizer.py
• rag/nlp/search.py
• rag/nlp/term_weight.py
• rag/svr/task_executor.py
• rag/svr/task_executor_refactor/chunk_post_processor.py
• rag/svr/task_executor_refactor/dataflow_service.py
• rag/svr/task_executor_refactor/dataset_structure_merger.py
• rag/svr/task_executor_refactor/dataset_wiki_generator.py
• rag/svr/task_executor_refactor/task_handler.py
• rag/utils/es_conn.py
• rag/utils/infinity_conn.py
• rag/utils/redis_conn.py
• test/testcases/test_web_api/test_chunk_app/test_chunk_routes_unit.py
• test/unit_test/rag/test_raptor_psi_tree_builder.py
• web/src/components/ui/tree-view.tsx

Purpose and Scope

This document details RAGFlow's background task processing architecture. The system offloads computationally intensive operations—such as document parsing, embedding, and Knowledge Graph construction—to asynchronous workers. It utilizes Redis Streams for reliable message delivery and a sophisticated task lifecycle management system to handle progress tracking, cancellation, and retries. The system includes a refactored task executor architecture located in rag/svr/task_executor_refactor/, utilizing a TaskManager and RecordingContext for enhanced observability.

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

---

Architecture Overview

RAGFlow implements a producer-consumer pattern. The API layer (producers) creates task records in MySQL and pushes task IDs into Redis Streams. Dedicated worker processes (consumers) monitor these streams, retrieve task metadata, and execute specialized pipelines using the refactored TaskManager.

Request to Execution Flow

The following diagram bridges the "Natural Language Space" (user actions) to the "Code Entity Space" (specific functions and classes).

Diagram: Task Dispatch and Processing

Sources: rag/svr/task_executor.py18-20 api/db/services/document_service.py127-146 api/db/services/task_service.py165-178 rag/utils/redis_conn.py34-40

---

Redis Streams Queue System

RAGFlow leverages Redis Streams to manage task distribution. Unlike simple lists, streams provide consumer groups and message acknowledgement (ACK), ensuring tasks are not lost if a worker crashes.

Queue Definitions and Task Types

The system maps internal task types to specific Redis Streams. The SVR_CONSUMER_GROUP_NAME is used to coordinate multiple workers. common/constants.py103

Task Type	Pipeline Mapping	Stream Name	Purpose
dataflow	PipelineTaskType.PARSE	ragflow_dataflow	Standard document parsing, OCR, and chunking.
raptor	PipelineTaskType.RAPTOR	ragflow_raptor	Recursive summarization for hierarchical retrieval.
graphrag	PipelineTaskType.GRAPH_RAG	ragflow_graphrag	Entity extraction and Knowledge Graph construction.
mindmap	PipelineTaskType.MINDMAP	ragflow_mindmap	Generating visual mind maps from document context.
memory	PipelineTaskType.MEMORY	ragflow_memory	Summarizing and indexing conversation history.
artifact	PipelineTaskType.ARTIFACT	ragflow_artifact	Processing artifacts from agent runs.
skill	PipelineTaskType.SKILL	ragflow_skill	Executing agent skills.
structure_graph	PipelineTaskType.STRUCTURE_GRAPH	ragflow_structure_graph	Building graph structures from documents.
structure_mindmap	PipelineTaskType.STRUCTURE_MINDMAP	ragflow_structure_mindmap	Building mindmap structures from documents.
timeline	PipelineTaskType.TIMELINE	ragflow_timeline	Extracting and organizing timeline events.
session_graph	PipelineTaskType.SESSION_GRAPH	ragflow_session_graph	Constructing session graphs.
session_essence	PipelineTaskType.SESSION_ESSENCE	ragflow_session_essence	Extracting session essences.
structure	PipelineTaskType.STRUCTURE	ragflow_structure	General document structuring.

Sources: rag/svr/task_executor.py133-147 common/constants.py103

Message Lifecycle

1. Consumption: The TaskManager in the refactored executor blocks on Redis Stream reads using the SVR_CONSUMER_GROUP_NAME. rag/svr/task_executor.py18 common/constants.py103
2. Task Loading: The worker extracts the task_id and calls TaskService.get_task(task_id) to fetch parameters from MySQL, including document location and parser configuration. api/db/services/task_service.py165-178
3. Acknowledgement: Once a task is processed or fails definitively, it is acknowledged in Redis to prevent re-delivery. rag/utils/redis_conn.py34-40

---

Worker Process Architecture

Startup and Initialization

Workers are initialized with a specific index (TE_IDX) and a heartbeat mechanism to report status.

• Heartbeat: Controlled by WORKER_HEARTBEAT_TIMEOUT (default 120s). rag/svr/task_executor.py178
• Environment: Workers set LITELLM_LOCAL_MODEL_COST_MAP="True" to avoid network-blocked imports of model costs, saving approximately 10s during startup. rag/svr/task_executor.py27
• Recording Context: Uses RecordingContext (and NullRecordingContext as fallback) to track execution metrics like time and token usage via the timed_with_recording decorator. rag/svr/task_executor.py19-20

Concurrency and Rate Limiting

To prevent resource exhaustion (OOM or API rate limits), workers use several specialized limiters defined in rag/svr/task_executor_limiter.py.

Diagram: Resource Limiter Mapping

Sources: rag/svr/task_executor.py95-101

---

Task Execution Pipelines

The Parser Factory

For dataflow tasks, the system uses a FACTORY dictionary to route documents to specific parsing logic based on the ParserType.

Parser Type	Code Module	Use Case
NAIVE	rag.app.naive	General text/markdown parsing.
PAPER	rag.app.paper	Academic papers with citation/reference handling.
LAWS	rag.app.laws	Structured legal documents.
MANUAL	rag.app.manual	Technical manuals with hierarchical sections.
TABLE	rag.app.table	Excel/CSV data extraction.
QA	rag.app.qa	Pre-formatted question-answer pairs.
TAG	rag.app.tag	Document tagging and classification.
BOOK	rag.app.book	Book-like documents with chapters and sections.
PRESENTATION	rag.app.presentation	Presentation slides.
RESUME	rag.app.resume	Resumes/CVs.
PICTURE	rag.app.picture	Image processing (OCR).
ONE	rag.app.one	Single-page documents.
AUDIO	rag.app.audio	Audio transcription.
EMAIL	rag.app.email	Email parsing.
KG	rag.app.naive	Knowledge Graph processing (uses naive parser as base).

Sources: rag/svr/task_executor.py114-131

RAPTOR Processing

The raptor pipeline handles hierarchical summarization logic.

• Tree Building: It uses strategies retrieved via get_raptor_tree_builder. rag/svr/task_executor.py52
• Clustering: Supports methods retrieved via get_raptor_clustering_method. rag/svr/task_executor.py51
• Summary Generation: Summarizes clusters to create higher-level abstraction layers. rag/svr/task_executor.py54

Sources: rag/svr/task_executor.py48-56 rag/svr/task_executor.py87-89

---

Lifecycle Management and Error Handling

Progress Tracking

Workers update progress via set_progress(task_id, from_page, to_page, prog, msg). This function:

1. Checks for cancellation via has_canceled(task_id). rag/svr/task_executor.py193
2. Formats a timestamped message and updates the progress percentage. rag/svr/task_executor.py200-201
3. Updates the Task table via TaskService.update_progress. rag/svr/task_executor.py206

Cancellation Mechanism

The TaskService.has_canceled function checks the task status in the database. If a task is marked as canceled, set_progress raises a TaskCanceledException, which terminates the current execution pipeline. rag/svr/task_executor.py190-192 api/db/services/task_service.py80

Redis Counter Synchronization

To coordinate document-level status across multiple chunking tasks, the system uses Redis-based counters:

• seed_doc_chunking_counter: Initializes the pending count for a document. api/db/services/task_service.py56-69
• credit_doc_chunking_task: Atomically decrements the pending count when a sub-task completes. api/db/services/task_service.py104-127
• abort_doc_chunking_counter: Marks a document's processing as aborted in Redis. api/db/services/task_service.py80-92

Retry Strategy

Tasks are resilient to transient failures:

• Retry Count: TaskService.get_task returns the retry_count for the task, allowing executors to track attempt history. api/db/services/task_service.py189
• Connection Management: Database connections are closed via close_connection() after progress updates to prevent connection leaks in long-running workers. rag/svr/task_executor.py84

Sources: rag/svr/task_executor.py167-198 api/db/services/task_service.py165-207