Docling 数据管道：构建企业级 RAG 数据管道
docling-pipelines如何用可组合的操作符与JSON流程取代脆弱的脚本：包括数据摄取、提取、质量检测、嵌入以及向量存储。
来源: Introducing Docling Pipelines: Building Enterprise-Grade RAG Data Pipelines
胶水代码的缺陷所在
想象一下：有一个存储着10,000份PDF文件的共享驱动器，一个塞满Word文件的SharePoint站点，以及一堆未处理的客户通话录音。上游的产品团队希望将所有这些数据都存入向量存储中，实现可搜索功能并对其进行分块处理，同时清除个人数据并去除重复项。
通常的工程解决方案是先写一个脚本，然后再写第二个脚本。接着引入一些库，用Pandas笔记本将它们整合起来，设置定时任务每晚自动运行整套流程，最终成品就诞生了。这种方法在一段时间内确实有效。
三个月后，有人询问为何新的文档不见了。问题根源在于下游某个列的架构发生了变化，而脚本悄悄地丢弃了那些行。因为没有任何可见的异常，所以没人注意到这一点。修复问题后需要重新运行脚本，这时才意识到没有记录显示哪些文档已经处理过，于是全部10,000份文档又都被重新处理了。
这就是文档处理流程中连接组件的典型命运：起初很简单，然后逐渐恶化，最终突然崩溃。人们只能不断添加重试机制，质量过滤也是事后才考虑的，各步骤之间也没有依赖关系跟踪，因此流程能否成功完全取决于脚本能否按正确顺序运行。如果没有稳定的文档标识，进行增量更新会极其困难。最终剩下的只是一团乱麻，只有最初的创建者才能调试，而就连他也不再确定问题所在。
什么是docling-pipelines
Docling-pipelines是一个基于操作符的模块化文档处理框架，用于构建完整的RAG数据管道。它的应用范围涵盖整个生命周期，从本地或云端源数据的导入开始，接着进行文本和实体提取、质量过滤、分块与嵌入处理，最终将结果写入向量数据库。这是标准流程，大多数实际应用中的数据管道都是对此流程的变体。
它可作为普通的Python包进行安装：
bash
pip install docling-pipelines
它在任何支持 Python 的环境中都能运行。对于基本使用而言，无需任何外部基础设施：不需要 Kafka，也不需要 Spark 集群，无需进行任何设置。流程定义在 JSON 文件中，通过 CLI 指定该文件即可开始运行。
有三种使用方式。CLI适用于一次性运行和本地开发。DocpipeFlowManager为核心的 Python 库则适合从应用程序代码或笔记本中触发流程。REST API则用于作为共享服务运行流程，供多个团队使用。下文将详细介绍这三种方式。
该项目是开源的，其首个版本 v0.1.0 于 2025 年 7 月发布。
核心理念：操作符与流程
对于初次接触该框架的用户来说，这里是最需要花时间理解的部分，因为一旦掌握了这一点，其余内容就会变得简单。
操作符是基本单元。操作符是一种 Python 类，它接收文档表并返回可能经过处理的表。该框架中的所有功能，无论是数据导入、提取、语言检测、分块还是嵌入，都属于操作符。
需要实现的接口很少：
short_name：JSON 中用于引用该操作符的字符串transform(table)：接收一个pa.Table，并返回表格列表及元数据字典get_metadata()：一个@staticmethod，用于返回操作符的元数据以便被识别get_required_features()：说明操作符在输入时需要哪些列validate()：一个可选的钩子函数，用于在启动时检查配置
最简的框架结构如下所示：
from docpipe.core.operators.abstract_operator import AbstractOperator, OperatorCategory
import pyarrow as pa
class MyOperator(AbstractOperator):
short_name = "my_operator"
category = OperatorCategory.Quality
def __init__(self, config: dict) -> None:
super().__init__(config)
def transform(self, table: pa.Table) -> tuple[list[pa.Table], dict]:
metadata = self.create_base_metadata(total_docs_count=len(table))
# … process table …
return [table], metadata
@staticmethod
def get_metadata() -> dict:
return {"short_name": "my_operator", "description": "…"}
采用这些方法可以构建出一流的组件，能够与生态系统中的其他所有操作符协同工作。
PyArrow 表负责存储数据。 每个操作符都会接收一个 pa.Table 对象，并返回另一个 pa.Table 对象。这是经过深思熟虑的选择，因为 PyArrow 的列式格式高效且具有类型约束，能够在不同处理阶段之间直接传递数据而无需复制。原始字符串和 Python 字典则永远不会在各个处理步骤之间传递。
由此带来的好处是，随着数据的处理推进，各列会逐步累积。数据摄取阶段会添加 id、name、path 和 source 字段；数据提取阶段会添加 content 字段；分块处理阶段会添加 chunked_content 字段；嵌入处理阶段则会添加 embeddings 字段。当一行数据到达向量存储时，它已经包含了完整的记录信息，包括所有列数据及元数据，可直接用于建立索引。
操作符仅处理其关心的列，其他所有内容都会原封不动地流过。因此，新操作符可以插入到任何位置，而不会影响其相邻节点。
数据流即有向无环图。整个处理流程存储在 JSON 文件中。每个节点都会指定一个type（操作符的简写名）、在图中唯一的name、一个config配置块以及一个depends_on依赖列表。在启动时，引擎会解析依赖关系图，并按拓扑顺序执行各个节点，同时执行的并行分支也会同步处理。
因此，该处理流程的结构属于数据而非代码。它可以进行版本控制、差异对比、静态验证，且在无需修改任何 Python 代码的情况下即可被替换。--validate选项能够在处理任何文档之前，检测出依赖关系错误、缺失的列以及配置问题。
模型为何有效。 实际应用中主要有两大优势。
其一是可组合性。只要上游操作已生成 get_required_features() 中列出的字段，当前操作即可随之执行。引擎会在启动时进行检查，并在运行开始前明确指出缺失的字段以及本应提供该字段的节点。
其二是可测试性。每个操作都表现为从一张表到另一张表的纯函数，因此可以单独进行测试：构建包含所需字段的 pa.Table，调用 transform()，然后对结果进行验证。无需模拟对象、无需副作用，也不需要完整的处理流程。
操作符概览
共有22个内置操作符，分为五类。以下为简要介绍。
摄取。 IngestSource 是唯一的摄取操作符，可处理本地及云端数据源。本地路径使用 filesystem 提供程序，而同一节点无需更改流程中的其他设置即可切换到八种云端提供程序（S3、IBM COS、SharePoint、OneDrive、Google Drive、Box、CSV文件及网页）。它为每个文件输出一行仅包含元数据的内容，暂不包含实际内容。后续的提取步骤会获取文件内容，这样即便在处理大规模数据时也能保持摄取速度快速且内存占用较低。
提取。 Extract 功能可将原始文件转换为结构化文本。PDF、DOCX、HTML、图片、音频及视频文件都会通过 Docling 转换为 Markdown 格式，既可以在本地模式（无需网络调用）下完成，也可以通过 API 模式的 Docling Serve 实现，后者适合将繁重的提取任务交由专用服务处理。
除此之外，实体提取是可选功能，目前有三种提供方式：LiteLLM（可接入100多家模型提供商）、IBM watsonx.ai，以及Docling基于模板的提取方法。具体选择取决于模型访问权限和成本。
ACLOperator属于提取阶段的一部分。它会将文档中的访问控制元数据提取出来并单独存放在一列中，这样后续系统无需再次读取源文件即可实施文档级安全控制。
功能型。这一类别包含流程逻辑：即那些用于转换、路由和重塑数据而非丰富数据的操作符。
ChunkerOperator 通过三种策略之一将提取的文本分割成多个片段：简单策略（固定字符长度，可配置重叠比例）、语义策略（基于内容理解）或混合策略（保留文档结构的 Docling 原生分割方式）。它还可以利用大语言模型对每个片段进行总结，从而为检索提供更丰富的信息表示。
EmbeddingsOperator 会为每个片段生成向量嵌入。该工具支持在本地运行的 HuggingFace 模型、LiteLLM（100 多家云服务提供商提供）以及 IBM watsonx.ai，而更换嵌入模型只需在流程 JSON 中进行一行修改即可。
BranchingOperator 是让流程变得有趣的关键所在。它根据列值、SQL 风格的条件或分类标签将文档拆分成多个独立的子流程。每个分支独立运行，并通过点号表示法在 depends_on 中以名称形式被引用，例如 "depends_on": ["branch_node.branch_name"]。
MergeOperator 根据下游操作符的需求，通过堆叠行或合并列的方式将各个分支的流重新整合在一起。
DocIdHashOperator 会从每份文档的内容中生成一个确定且稳定的哈希 ID。该 ID 是去重和增量处理的基础：无论在何时何地被导入，同一份文档始终会获得相同的 ID。
EntityCurationOperator和NOOPOperator构成了完整的组合。前者负责在实体提取后进行清理，后者则是一个传递节点，有助于构建有向无环图。
质量。该组中的操作符体现了仅仅拥有处理流程与拥有可靠处理流程之间的差距。
LanguageDetect利用fastText识别文档的语言，并添加一个lang_name列，可用于实现过滤、路由或标记功能。
ReadabilityOperator从五个方面对每份文档的易读性进行评分：弗莱什易读度、弗莱什-金凯德等级、冈宁模糊度、戴尔-查尔指数以及难词数量。这些评分可作为有用的路由依据，例如在选择分块策略之前区分技术文档与普通内容。
PIIAndHAPAnnotator利用基于大语言模型的检测技术来识别并遮蔽个人信息（电子邮件地址、电话号码、社会安全号码、卡号、银行账户、IP地址），以及仇恨言论、辱骂内容与脏话，不同类别设有相应的阈值。
RedactionOperator用于处理格式已知的模式化遮蔽任务，例如内部编号、产品代码或任何可通过正则表达式匹配的内容。
EdedupOperator通过doc_id_hash列以确定性的方式删除文档集中的完全重复内容。
SQLFilterOperator 是最容易被低估的组件之一。它通过基于列值的 SQL WHERE 条件来过滤行，因此 "docq_total_words > 100 AND lang_name = 'en'" 就能在无需任何 Python 代码的情况下构成一个完整的质量检测机制。结合可读性评分与语言检测功能，它还能实现无需编写代码的复杂路由处理。
DocumentClassifierOperator 能将文档分类到可配置的类别中。DocQuality 用于计算整个语料库的质量指标，而 MLEnrichmentOperator 则会将机器学习预测结果作为额外列添加进去。其他相关文档也指出，数据质量往往比模型选择对 RAG 的结果影响更大。
VectorDB与存储。 VectorDBOperator会将分块后的嵌入文档写入向量存储库，支持的开源后端包括OpenSearch和Milvus。字段映射、索引设置以及元数据规范化均在流程JSON中配置。
DocumentSetOperator会将处理后的数据表保存到基于DuckDB的文档集中。这在无需向量存储库即可查询处理后的文档，或需要创建持久化的流程中间检查点时非常有用。
StorageOutputOperator会将文档以处理后或原始形式写回外部存储位置（如S3、SharePoint、Google Drive或本地文件系统），适用于“处理并归档”的工作模式。
只要满足上游依赖条件，这些组件可以以任意顺序组合使用，因此流程并不局限于固定的结构。
实际流程示例
以下是标准的快速入门流程，即一个完整的从数据导入到向量存储的管道，该流程使用 Ollama 生成嵌入向量，再通过 OpenSearch 进行存储：
{
"flow_name": "complete-document-pipeline",
"global_config": {
"doc_column": "content",
"storage": "in-memory"
},
"flow": [
{
"type": "ingest_source",
"name": "ingest_local_folder",
"config": {
"provider": "filesystem",
"connection_params": {
"paths": [
"./sample_documents"
]
},
"include_filter": "pdf,txt,docx"
}
},
{
"type": "extract_operator",
"name": "extract_with_docling",
"config": {
"text_extraction": {
"provider": "docling_library"
},
"entity_extraction": {
"provider": "none"
}
},
"depends_on": [
"ingest_local_folder"
]
},
{
"type": "chunker",
"name": "simple_chunker",
"config": {
"chunk_type": "simple",
"chunk_size": 512,
"chunk_overlap": 50
},
"depends_on": [
"extract_with_docling"
]
},
{
"type": "embeddings",
"name": "ollama_embeddings",
"config": {
"provider": "litellm",
"provider_config": {
"model_id": "openai/nomic-embed-text",
"api_base": "http://localhost:11434/v1"
},
"embeddings_column": "embeddings"
},
"depends_on": [
"simple_chunker"
]
},
{
"type": "vectordb",
"name": "opensearch_vector_store",
"config": {
"provider": "opensearch",
"doc_id_column": "doc_id_hash",
"embeddings_column": "embeddings",
"provider_config": {
"index_name": "sample-documents-index",
"host": "localhost",
"port": 9200
}
},
"depends_on": [
"ollama_embeddings"
]
}
]
}
逐节点处理：
- ingest_local_folder 通过
ingest_source及其文件系统提供程序扫描./sample_documents目录中的 PDF、文本和 DOCX 文件。每个文件都会生成一行数据，包含编号、名称、路径和来源信息。此时不会读取文件内容，因此操作很快——只是定位文件而非加载它们。 - extract_with_docling 会获取这些元数据行，使用 Docling 读取每个文件，并添加一个存储提取出的 Markdown 内容的
content列。至此，PDF 文件便转化为结构化且可查询的文本。docling_library提供程序完全在进程内运行，因此无需外部服务。
chunked_content的结构化列表，其中每个列表项都包含对应片段的文本以及其在源文档中的位置信息。embeddings列中。它实际上就是专为Ollama设计的LiteLLM版本，只需更改model_id和api_base参数，即可将其指向LiteLLM支持的100多个服务提供商中的任意一个，其余部分无需改动。doc_id_hash作为主键。因此重新运行该工具也不会产生重复数据。有几点结构上的要点值得注意。depends_on是唯一需要的连接方式：无需显式的数据传递、无需回调注册，也无需共享状态，因为引擎会自动解析图结构并处理数据传输。global_config块设置了所有操作符都会继承的默认值。doc_column指定了存放正文内容的列，而storage: "in-memory"则让中间表格在各个处理步骤之间保留在RAM中，这种方式适用于大多数场景；对于大型数据集，也可更换为其他存储方式。
整个流程只需一条命令即可运行：
docling-pipelines - flow-file pipeline.json
高级模式
流程不必是线性的链条。在实际部署中会出现三种重复出现的模式。
分支与合并。分支功能会根据可配置的标准将文档发送到不同的子流程中。每个分支都是一个有名称的输出，下游节点会在depends_on中用点号形式引用它。基于可读性得分来决定的路由分支如下所示：
{
"name": "quality_branching",
"type": "branching",
"config": {
"branches": {
"high_quality_branch": {
"link_name": "High Quality Documents",
"criteria_json": {
"criteria_list": [
{
"variable": "flesch_reading_ease",
"operator": ">",
"value": 60
}
],
"logical_operator": "AND"
}
},
"low_quality_branch": {
"link_name": "Low Quality Documents",
"criteria_json": {
"criteria_list": [
{
"variable": "flesch_reading_ease",
"operator": "<=",
"value": 60
}
],
"logical_operator": "AND"
}
}
}
},
"depends_on": [
"readability"
]
}
高质量数据流在嵌入之前会先经过机器学习增强处理，而低质量数据流则会再通过另一个SQLFilterOperator，在嵌入之前剔除那些词数为零的文档。两者最终会被分别存储到不同的OpenSearch索引中，之后当后续逻辑需要统一视图时，MergeOperator可以重新合并这些数据流。
基于质量阈值的路由机制。其思路是先运行成本较低的质控操作，然后再让SQLFilterOperator在执行任何成本较高的大语言模型处理步骤之前剔除不符合条件的数据行。完整的质控流程可能如下所示：
lang_detect → ededup → doc_quality → readability → sql_filter → embeddings
sql_filter 条件其实就是以数据形式表达的 SQL WHERE 子句：
{
"criteria_list": [
{
"variable": "docq_total_words",
"operator": ">",
"value": 100
},
{
"variable": "lang_name",
"operator": "=",
"value": "en"
}
],
"logical_operator": "AND"
}
没有 Python 代码，也没有自定义运算符，仅在 JSON 中存在条件判断，任何不满足该条件的行都不会进入嵌入调用阶段。
多阶段增强处理。在实际应用中，常常会采用“分类、路由、提取、增强、存储”这样的处理流程。文档可能首先被分类，技术文档会被路由到专为技术内容设计的实体提取器，而政策文档则会被路由到另一个提取器。每个分支都会添加与其类型相关的增强列，这些数据流会在分块和嵌入之前合并，从而使后续的处理流程能够共享数据。
每个阶段都会添加新的列，因此下游的运算符会看到越来越丰富的表格结构。这一切都不依赖特殊的模式或标志，只是由普通运算符组合而成的，这正是运算符模型的优势所在。
三种运行方式
该框架不仅仅是一个命令行工具，可应用于多种工程场景。
命令行界面。这是启动流水线的最快方式：
docling-pipelines - flow-file pipeline.json
docling-pipelines - flow-file pipeline.json - validate # validate without running
docling-pipelines - list-operators - verbose # discover all registered operators
--validate在流程开发过程中发挥着重要作用：在读取或写入任何数据之前，它就能检测出依赖关系错误、缺失的必要列以及配置类型错误等问题。在部署流程变更之前于CI环境中运行该工具是个良好的习惯。
Python库。当管道需要通过应用程序代码触发时：
from docpipe.lib.docpipe_flow_manager import DocpipeFlowManager
manager = DocpipeFlowManager(flow_file="pipeline.json")
manager.execute()
这适用于嵌入在更大应用程序、笔记本或需要控制自身运行生命周期的编排脚本中的代码。
REST API。 用于作为长期运行的服务使用：
uvicorn docpipe.api.main:app - host 0.0.0.0 - port 8000
FastAPI服务器提供了三组路由：用于管理流程的/api/v1/flows，用于控制任务运行的/api/v1/job_runs，以及用于查找操作器的/api/v1/operators。用户可通过LDAP或OAuth2/OIDC并使用JWT令牌进行身份验证，同时/docs路径会自动生成OpenAPI/Swagger文档。这是多个团队共同使用的共享服务接口，外部事件可触发该服务，其他团队也可通过它查询状态与结果。
通过自定义操作器进行扩展
若要使该方案在真实企业中得到长期应用，这无疑是设计中最重要的一部分。
该契约结构十分简洁，操作符的描述大约需要30行即可完成。一个Python类若继承自AbstractOperator并实现了transform()、get_metadata()以及get_required_features()方法，即符合要求。此类可与所有内置操作符结合使用，会出现在--list-operators的输出结果中，且流程JSON会像对待内置操作符一样通过short_name来引用它。
注册方式共有三种，可根据开发阶段选择合适的办法。
文件系统路径适用于开发阶段。只需将环境变量指向本地目录即可：
DOCPIPE_CUSTOM_OPERATORS=./my_operators docling-pipelines - flow-file pipeline.json
启动时该目录中的所有内容都会被导入。无需进行任何安装，因此迭代速度很快。
Python包用于分发。将操作符作为带有正确入口点的pip包发布，在相同环境中安装后，它们便会自动注册。这样，其他团队无需接触核心代码库即可使用这些操作符。
S3 URI是企业级方案。将DOCPIPE_CUSTOM_OPERATORS指向远程操作符包，框架会在运行时自动获取并注册它。这种方式适用于多环境下的集中式操作符管理：只需在一个地方进行更新，所有环境在下次启动时就会应用这些变更。
因此，企业特定的逻辑不会影响核心功能。合规处理程序、领域特定的实体提取器以及专有的评分模型都可以被封装为操作符，与流程 JSON 中的内置组件结合使用，且无需修改框架即可部署。
运行方面的考量
对于这类工具，一个合理的疑问是它究竟只是个不错的原型，还是真正具备生产级性能。具体分析如下。
可观测性。该工具内置了对 OpenTelemetry 的原生支持，每个操作符的执行都会生成一个 span。日志为带有追踪关联功能的结构化 JSON，每行都包含 trace_id 和 span_id。无需在应用程序端进行任何配置，即可与 Jaeger、Grafana Tempo、Datadog 和 AWS X-Ray 集成；关闭遥测功能时也不会产生任何额外开销。
增量处理。该处理流程利用DocIdHashOperator生成的确定性ID来记录已处理的内容。重新运行时仅处理新的或已更改的文档，同一文档若被提交两次则会得到相同的哈希值而被跳过。在global_config中设置force_ingest: true可忽略这一规则，对所有文档重新进行处理。增量处理的元数据可以存储在DuckDB（默认）、PostgreSQL或文件系统中。
安全性。 ACLOperator会将权限元数据以列的形式呈现，以便后续实施管控，而PIIAndHAPAnnotator则会在敏感内容进入向量存储之前将其识别并屏蔽。流程配置中的机密信息可来自HashiCorp Vault：在JSON中写入的如vault://path/to/secret这样的路径会在运行时由引擎解析。对REST API的访问则通过LDAP以及OAuth2/OIDC进行保护。
扩展性。该工具提供三种执行模式，适用范围从笔记本电脑到分布式集群。默认模式为进程内线程池，具有速度快、实现简单且无依赖的特点。对于需要在单个节点上处理CPU密集型任务的场景，可使用进程池在独立进程中运行操作程序。对于分布式生产工作负载，则会使用Prefect工作池来运行Docker容器，从而实现水平扩展以及对任务运行的集中管理。微批处理设置可控制每种模式下的内存使用情况，这在处理大型文档的LLM提取任务时尤为重要。
下一步该做什么
该项目托管在 GitHub 上的 IBM 组织下，其快速入门指南是启动流水线的最快捷方式。值得深入研究的方面包括提取层（Docling 集成、适用于图像密集型文档的 VLM 流水线、音频和视频的 ASR 功能）、从第一个类声明到测试、注册和部署的全流程中编写具有复杂逻辑的自定义操作符、使用 Docker 运行 Prefect 以处理大规模工作负载，以及 OpenSearch 的 VectorDBOperator 所提供的字段映射和模式模板选项。
相关阅读
- 实用指南：构建生产级本地RAG管道——100%免费，无需付费 — 该指南提供了详细的操作步骤，包括合同条款、检查项以及供采用此模式的团队直接使用的代码片段。
- 在Postgres中实现混合搜索以构建健康助手：架构设计、分块处理、RRF算法与评估方法 — 讲解如何在Postgres中构建健康助手的检索层：基于结构感知的分块处理、结合RRF的pgvector混合搜索、二进制量化技术、阈值设定以及性能评估方法。