Dataflow Pipeline Execution#

Dataflow pipeline execution is the runtime path that takes a document through a user-defined pipeline graph (stored as a DSL in MySQL), produces embedded chunks, and writes them to the vector/document store. It is implemented in the refactored task executor under rag/svr/task_executor_refactor/dataflow_service.py and is distinct from both the Pipeline Canvas Architecture (which covers graph structure and component lifecycle) and the Chunker Pipeline (which covers inter-stage data formats).

Task Queueing and Entry Point#

Dataflow tasks are enqueued by queue_dataflow() in api/db/services/task_service.py. It:

1. Creates a task row with task_type="dataflow" (or "dataflow_rerun") 1
2. Attaches dataflow_id (the UserCanvas pipeline ID), kb_id, tenant_id, and an optional raw file dict 2
3. Pushes the task to a priority-routed Redis queue via REDIS_CONN.queue_product(settings.get_svr_queue_name(priority, "common"), message=task) 3

Background worker processes pick tasks off the Redis queue and dispatch them based on task_type. For "dataflow" tasks, the worker instantiates DataflowService with a TaskContext that carries all task fields (ids, limiters, callbacks, optional write interceptor).

DataflowService.run_dataflow()#

run_dataflow() is the central orchestrator. Its steps in order:

1. DSL Loading#

_load_dsl() resolves the pipeline definition:

• If task_type == "dataflow": loads UserCanvas.dsl directly by dataflow_id 4
• Otherwise (e.g., rerun): looks up a PipelineOperationLog by ID and reads its snapshot dsl and pipeline_id 5 — enabling exact-state reruns from a previously saved DSL

2. Pipeline Instantiation and Execution#

A Pipeline object is constructed from the DSL with tenant_id, doc_id, task_id, flow_id, and language 6. pipeline.run() is then awaited, traversing the File → Parser → Chunker → Indexer chain and returning a chunks dict 7.

Debug mode: if doc_id == CANVAS_DEBUG_DOC_ID, the chunks are recorded for inspection and the function returns early without writing to the store 8.

3. Chunk Normalization#

_normalize_chunks() converts the pipeline's multi-format output (chunks, json, markdown, text, html) into a flat List[Dict] for downstream processing 9.

4. Chunk Embedding#

If no vector fields matching q_[0-9]+_vec exist on the chunks, _embed_chunks() is called 10. It:

• Resolves the KB's embedding model config, preferring kb.tenant_embd_id over the default kb.embd_id 11
• Uses EmbeddingUtils.prepare_texts_for_dataflow_embedding() to extract text from questions, summary, or text fields in priority order
• Encodes in batches (size from settings.EMBEDDING_BATCH_SIZE), throttled by ctx.embed_limiter 12
• Stacks and attaches vectors back onto chunks via EmbeddingUtils.stack_vectors() / EmbeddingUtils.attach_vectors() 13
• Has a 60-second timeout enforced by @timeout(60) 14

5. Chunk Metadata Processing#

_process_chunks() stamps each chunk with doc_id, kb_id, docnm_kwd, timestamps, and a stable ID (xxhash.xxh64 of text + doc_id if missing). It also normalizes special fields: questions → question_kwd/question_tks, keywords → important_kwd/important_tks, summary → content_ltks, and metadata (extracted and returned separately for document-level update). PDF position data is finalized via add_positions() 15.

6. Chunk Insertion#

Chunks are inserted via ChunkService.insert_chunks(), which handles the vector store and document store bulk write 16. Progress is reported at 0.82 (start) and 1.0 (done) 17.

7. Document Statistics Update#

On success, DocumentService.increment_chunk_num() records chunk count, token consumption, and elapsed time on the document row 18.

Operation Log Persistence (PipelineOperationLog)#

After every run attempt (success or failure, including empty-chunk cases), _record_pipeline_log() calls PipelineOperationLogService.create().

create() persists a snapshot to the PipelineOperationLog MySQL table with:

19

Log rotation: after insert, if the total log count for a KB exceeds PIPELINE_OPERATION_LOG_LIMIT (default 1000, via env var), the oldest rows are deleted atomically 20.

Rerun from log: because the DSL snapshot is stored on the log, a rerun task can pass the log's id as dataflow_id and _load_dsl() will restore the exact pipeline state at completion time 5.

KB-level fan-out tasks (GraphRAG, RAPTOR, MindMap, etc.) use a different code path — they are identified by _PIPELINE_TASK_TYPE_TO_FINISH_FIELD membership and stamp the KB's <type>_task_finish_at column rather than updating per-document progress 21.

DSL Serialization#

The pipeline DSL is a JSON document stored in UserCanvas.dsl (a JSONField backed by LONGTEXT in MySQL) 22. After pipeline.run(), str(pipeline) serializes the live Pipeline object — which calls Graph.__str__() — deep-copying the DSL structure and serializing each component's parameter state 23. This snapshot is what PipelineOperationLogService.create() writes as dsl=json.loads(dsl) back into PipelineOperationLog.dsl 24.

Key Files#