Pipeline Canvas Architecture#

RAGFlow's Pipeline is a DataflowCanvas-category agent for document ingestion — distinct from the conversational Agent Canvas, but sharing the same visual orchestration and DSL execution infrastructure. Both are built on a common Graph base class 1 with the same JSON DSL schema, component lifecycle, and downstream-edge wiring.

The distinction lies in their runtime context:

• Agent Canvas (Canvas, agent/canvas.py) handles conversational multi-turn workflows: it stores history, globals (sys.query, sys.user_id, etc.), and resolves inter-component references via {{cpn_id@var}} variable syntax.
• DataFlow Canvas (Pipeline, rag/flow/pipeline.py) is scoped to a single document ingestion task. It carries doc_id, kb_id, and flow_id; tracks progress in Redis; and passes the previous component's full output() dict directly as **kwargs to the next — no variable reference syntax needed.

The CanvasCategory enum separates the two: pipelines are stored and queried as DataflowCanvas, agents as Agent.

Class Hierarchy#

Graph (agent/canvas.py:49)
├── Canvas (agent/canvas.py:330) ← Agent / Chatbot canvases
└── Pipeline (rag/flow/pipeline.py:28) ← DataFlow / ingestion canvases

Graph.load() 2 is the shared bootstrap: it reads the DSL components dict, instantiates each component via component_class(name)(canvas, id, param), and restores the execution path list. Both Canvas and Pipeline call this on construction.

Pipeline-specific additions 3:

• _doc_id / _kb_id — links the run to a specific document and knowledge base
• _flow_id — identifies which DataFlow canvas definition is being executed
• callback() 4 — writes per-component trace logs to Redis ({flow_id}-{task_id}-logs) and pushes progress percentages to TaskService

All Pipeline components extend ProcessBase 5 rather than the agent's ComponentBase directly. ProcessBase.invoke() wraps _invoke() with asyncio.wait_for timeout enforcement and wires up the parent canvas's callback automatically 6.

Execution Model#

Pipeline execution is linear. Pipeline.run() 7 always starts at the "File" component and walks the downstream edges:

File → Parser → Chunker (Token or Title) → Indexer

Each step passes the entire previous component's output as **kwargs to the next 8: await cpn_obj.invoke(**last_cpn.output()). There is no variable-reference resolution or branching — the DAG is effectively a sequential chain.

Execution is fully asynchronous and task-queued. queue_dataflow() enqueues a task_type="dataflow" entry in Redis; background workers pick it up and instantiate Pipeline, then call run(). Cancellation is checked inside callback() via has_canceled(task_id) 9.

Contrast with Canvas.run() which uses {{cpn_id@param}} variable references, supports Switch/Categorize branching, loop components, and multi-turn conversation state.

Stage Components and Inter-component Handoff#

All Pipeline stage components live in rag/flow/ and are auto-discovered at import time via rag/flow/__init__.py's dynamic pkgutil.walk_packages scan. The standard ingestion chain is:

The inter-component data contract between Parser and Chunker is TokenChunkerFromUpstream (a Pydantic model). The critical field is output_format — a routing signal that tells the Chunker which field contains the actual data (json, text, markdown, html, or chunks). A mismatch between what the Parser writes and what output_format declares causes silent zero-chunk failures because the Chunker reads the empty declared field and returns {"chunks": []} with no error.

Known format-mismatch bug (fixed in v0.26.0): In v0.25.6, Parser._image() read output_format from the DSL configuration instead of hardcoding "json". If the pipeline DSL had output_format: "text" for images, the Parser populated the json field correctly but declared output_format="text" — causing the Token Chunker to read the empty text field and produce zero chunks. The fix (PR #15847) unconditionally sets output_format="json" in _image(). 10

Frontend Integration#

The DataFlowSelect component in web/src/components/data-pipeline-select/ is the UI entry point for attaching a pipeline to a dataset. It queries the /agents endpoint filtered by canvas_category: AgentCategory.DataflowCanvas 11 and exposes a searchable dropdown. It appears in the dataset settings page 12 to let users route new document ingestion through a custom pipeline instead of the default chunker.

Key Files#