基于质量意识的RAG数据导入：按可读性路由文档
一种声明式的文档处理流程，可对语言、重复内容及弗莱什可读性进行评分，随后将文档发送到两个OpenSearch索引中，以便客户和开发者检索。
来源: Architecting a Multi-Tier RAG Pipeline: Quality-Aware Document Routing with docling-pipelines
将客户支持知识库转换为人工智能搜索系统时，通常会在同一个环节出问题：所有文件都被一视同仁地处理。一个真实的语料库其实是混杂的集合，其中既有条理清晰的常见问题解答、结构化的产品指南、内容繁多的开发者手册、重复出现的法律文本，还有一些完全无法阅读的损坏文件。
如果把所有这些内容都放入同一个对话式RAG系统中，检索质量就会下降。那些只想获取简单操作指南的客户反而会得到工程手册中的片段内容。将无用的文本或系统日志交给昂贵的模型进行总结或翻译，纯粹是浪费金钱。
解决办法是根据文件质量对它们进行分类处理。采用两级设计，先检查每个文件，删除空白页面，评估可读性，然后再对语料库进行划分：
- 第一层级：消费者搜索。可读性高的文档会获得诸如字数之类的结构化元数据，以支持面向用户的对话式搜索。
- 第二层级：开发者和参考资料搜索。内容密集的技术材料会被放入独立的索引中，用于处理高级技术问题。
docling-pipelines使这一方案得以实现。它是一个基于MIT许可的开源框架，由IBM提供支持，能够直接通过JSON配置构建并运行文档处理流程。无需编写自定义的Python代码，因为各个处理环节会以内存中的Apache Arrow表格形式传递文档，因此中间转换和路由检查几乎不会占用额外的CPU和内存资源。
流程的结构
所有内容都存储在一个描述有向无环图的JSON流程文件中：
ingest能从磁盘读取 PDF、TXT、DOCX 和 HTML 格式的文件。extract使用 Docling 库提取结构化文本。detect_language可识别每份文档的语言类型。dedup通过内容哈希值来删除重复的记录。doc_quality会计算诸如总字数之类的结构化指标。readability负责计算文本的复杂度得分。quality_branching根据 Flesch 阅读易度得分进行分类：得分高于 60 的归入高质量分支，60 及以下则归入低质量分支。
ml_enrichment_high_quality、embeddings_high_quality和vectordb_high_quality。它会添加基于机器学习的语言特征，生成向量嵌入，并将结果写入high_quality_docs索引中。sql_filter_low_quality、embeddings_low_quality和vectordb_low_quality。它会删除空行，对剩余内容进行向量嵌入，然后将结果写入low_quality_docs中。分割前的性能分析
每份文档在被路由处理之前都需要先生成对应的配置文件。在ingest读取数据内容且extract将布局转换为文本之后，后续的所有处理操作都相当于转换器：它们会检查表格结构，并在同一内存中的Arrow表格中添加新列。首先会进行一些仅需CPU即可完成的简单检测：语言类型、重复内容以及文本复杂度。
语言检测
后续的机器学习模型和可读性计算公式都需要知道文档的语言类型，因此语言检测是第一步。
{
"name": "detect_language",
"type": "lang_detect",
"config": {
"doc_column": "content"
},
"depends_on": [
"extract"
]
}
操作员会读取content列，并追加lang_name（ISO 639-1编码）和lang_score。将其直接放在提取结果之后，可使后续所有对语言敏感的处理步骤直接使用这些数据。
去重
重复的文档会浪费嵌入计算资源，并导致搜索结果出现偏差。
{
"name": "dedup",
"type": "ededup",
"config": {
"doc_column": "content",
"doc_id_column": "doc_id_hash"
},
"depends_on": [
"detect_language"
]
}
它指的是存储在doc_id_hash中的内容哈希值以及content中的原始文本，然后从表格中删除重复的行。提前执行此操作可避免后续步骤做无用功。
文档质量
基本的结构检查能够发现空白页面和异常布局。
{
"name": "doc_quality",
"type": "doc_quality",
"config": {
"doc_column": "content",
"doc_content_column": "content",
"text_lang": "en"
},
"depends_on": [
"dedup"
]
}
这会新增11个结构列，所有列的前缀均为docq_。其中重要的是docq_total_words，即每份文档的总字数。
可读性
可读性计算公式会根据单词长度、音节数以及句子结构来判断文本的难易程度。
{
"name": "readability",
"type": "readability",
"config": {
"doc_column": "content",
"readability_score_list": [
"flesch_reading_ease",
"flesch_kincaid_grade",
"gunning_fog",
"difficult_words",
"dale_chall_readability_score"
]
},
"depends_on": [
"doc_quality"
]
}
readability_score_list 数组用于告知操作员应计算哪些指标并将其作为列添加进去。路由功能依赖于 Flesch 阅读易度分数，该分数的取值范围为 0 到 100：
- 高于 60 表示语言通俗易懂，适合普通帮助中心用户阅读。
- 60 或更低则意味着文本内容复杂专业，需要相关领域的专业知识才能理解。
分支步骤
分支操作符会将 Arrow 数据集划分成多个独立的处理路径。
{
"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"
]
}
路由规则由一个包含四个字段的声明式 criteria_json 块控制：
variable指定要检测的列，例如flesch_reading_ease。operator规定比较运算符，如>或<=。value为阈值，此处为60。logical_operator用AND或OR连接多个条件。
以 flesch_reading_ease > 60 作为规则时，引擎会直接评估这些条件，划分内存中的表格，并执行各自的后续分支。
高质量路径：丰富处理、嵌入与索引
通过 high_quality_branch 发送的文档会首先经过语言学机器学习处理以进行丰富化。
{
"name": "ml_enrichment_high_quality",
"type": "ml_enrichment",
"config": {
"doc_column": "content",
"lang_column": "lang_name",
"output_column_prefix": "ml_",
"error_column_name": "enrichment_error"
},
"depends_on": [
"quality_branching.high_quality_branch"
]
}
ml_enrichment会使用具备语言识别能力的模型来解析文本，并生成诸如句子数量、音节特征以及名词列表等元数据，这些都有助于提升检索效果。该功能依赖于检测步骤中得到的lang_name字段。由于这是整个流程中计算负荷较大的部分，因此仅能在可读取的文本上运行。output_column_prefix设置会将新生成的列前缀加上ml_，其中包括ml_num_words这一列。
随后，文本会被转换为2560维的嵌入向量。所使用的模型为openai/granite4:latest，通过LiteLLM由本地的Ollama服务器提供支持：
{
"name": "embeddings_high_quality",
"type": "embeddings",
"config": {
"embeddings_column": "embeddings",
"overlap_ratio": 0.2,
"doc_column": "content",
"provider_config": {
"api_base": "http://localhost:11434/v1",
"api_key": "${OLLAMA_API_KEY}",
"model_id": "openai/granite4:latest"
},
"provider": "litellm"
},
"depends_on": [
"ml_enrichment_high_quality"
]
}
最后，文档会被索引到 OpenSearch 中，feature_mappings 用于将 Arrow 列与 OpenSearch 字段关联起来：
{
"name": "vectordb_high_quality",
"type": "vectordb",
"config": {
"provider": "opensearch",
"doc_id_column": "doc_id_hash",
"embeddings_column": "embeddings",
"vector_dimension": 2560,
"create_index": "true",
"provider_config": {
"index_name": "high_quality_docs",
"host": "localhost",
"port": 9200,
"username": "${OPENSEARCH_USERNAME}",
"password": "${OPENSEARCH_PASSWORD}",
"use_ssl": false,
"verify_certs": false,
"batch_size": 100,
"engine": "faiss",
"algorithm": "hnsw",
"space_type": "l2"
},
"feature_mappings": [
{"feature_name": "doc_id_hash", "mapped_column_name": "pk"},
{"feature_name": "name", "mapped_column_name": "doc_name"},
{"feature_name": "content", "mapped_column_name": "text"},
{"feature_name": "embeddings", "mapped_column_name": "vector_embeddings"},
{"feature_name": "lang_name", "mapped_column_name": "language"},
{"feature_name": "flesch_reading_ease", "mapped_column_name": "readability_score"},
{"feature_name": "ml_num_words", "mapped_column_name": "word_count"}
]
},
"depends_on": [
"embeddings_high_quality"
]
}
低质量路径：过滤、嵌入、索引
位于low_quality_branch上的文档会跳过成本较高的ml_enrichment步骤。这些文档在提取后可能根本不含可识别的文字，比如损坏的文件或空白扫描件，而将空字符串嵌入向量空间只会增加噪声。过滤器可用于剔除这类文档：
{
"name": "sql_filter_low_quality",
"type": "sql_filter",
"config": {
"criteria_json": {
"criteria_list": [
{
"variable": "docq_total_words",
"operator": ">",
"value": 0
}
],
"logical_operator": "AND"
}
},
"depends_on": [
"quality_branching.low_quality_branch"
]
}
该过滤器会检查doc_quality中的docq_total_words列，剔除词数为零的文档。
剩余的文档多为技术类资料，它们会使用相同的granite4模型进行处理，并被写入独立的OpenSearch索引中：
{
"name": "vectordb_low_quality",
"type": "vectordb",
"config": {
"provider": "opensearch",
"doc_id_column": "doc_id_hash",
"embeddings_column": "embeddings",
"vector_dimension": 2560,
"create_index": "true",
"provider_config": {
"index_name": "low_quality_docs",
"host": "localhost",
"port": 9200,
"username": "${OPENSEARCH_USERNAME}",
"password": "${OPENSEARCH_PASSWORD}",
"use_ssl": false,
"verify_certs": false,
"batch_size": 100,
"engine": "faiss",
"algorithm": "hnsw",
"space_type": "l2"
},
"feature_mappings": [
{"feature_name": "doc_id_hash", "mapped_column_name": "pk"},
{"feature_name": "name", "mapped_column_name": "doc_name"},
{"feature_name": "content", "mapped_column_name": "text"},
{"feature_name": "embeddings", "mapped_column_name": "vector_embeddings"},
{"feature_name": "lang_name", "mapped_column_name": "language"},
{"feature_name": "flesch_reading_ease", "mapped_column_name": "readability_score"},
{"feature_name": "difficult_words", "mapped_column_name": "difficult_words"}
]
},
"depends_on": [
"embeddings_low_quality"
]
}
由于该分支跳过了内容丰富处理，这些文档不存在word_count。映射功能改用计算出的difficult_words数值，这有助于处理技术类查询。
两种索引，两种查询方式
按可读性等级划分的独立索引让不同受众能够以不同的方式进行搜索。
帮助中心查询
对于最终用户，查询会指向high_quality_docs。该索引仅保留那些易于阅读且内容充实（readability_score > 60，word_count > 100）的指南：
curl -XGET "http://localhost:9200/high_quality_docs/_search" \
-H 'Content-Type: application/json' -d'
{
"query": {
"bool": {
"must": [
{
"knn": {
"vector_embeddings": {
"vector": [0.015, -0.023, 0.004],
"k": 10
}
}
}
],
"filter": [
{
"range": {
"readability_score": {
"gt": 60
}
}
},
{
"range": {
"word_count": {
"gt": 100
}
}
}
]
}
}
}'
开发者参考资料搜索
对于正在查找技术规格的专家用户，查询目标为low_quality_docs。该功能会分析复杂的资料内容，保留那些包含大量高难度技术术语的片段（difficult_words > 5）：
curl -XGET "http://localhost:9200/low_quality_docs/_search" \
-H 'Content-Type: application/json' -d'
{
"query": {
"bool": {
"must": [
{
"knn": {
"vector_embeddings": {
"vector": [0.015, -0.023, 0.004],
"k": 10
}
}
}
],
"filter": [
{
"range": {
"difficult_words": {
"gt": 5
}
}
}
]
}
}
}'
运行它
该仓库提供了一个设置脚本，用于创建虚拟环境、安装依赖项、启动本地的 OpenSearch 容器以及注册 Ollama 模型。
步骤 1：设置环境
克隆该仓库并运行脚本：
git clone https://github.com/IBM/docling-pipelines.git
cd docling-pipelines
./scripts/setup_docling_pipelines_environment.sh
或者直接从 PyPI 安装：
pip install docling-pipelines
步骤2：设置环境变量
export OLLAMA_API_KEY=ollama
export OPENSEARCH_USERNAME=admin
export OPENSEARCH_PASSWORD=<your-password>
步骤 3：运行流程
通过命令行启动流水线：
docling-pipelines --flow-file sample_flows/advanced/branching_quality_routing.json
运行时输出的内容
输出会记录整个流程，包括每一项转换操作和路由决策：
$ docling-pipelines --flow-file sample_flows/advanced/branching_quality_routing.json
[INFO] [ingest] Loading files from tests/fixtures/customer_support_docs/
[INFO] [ingest] Ingested 12 files.
[INFO] [extract] Running extract_operator using docling_library...
[INFO] [extract] Extracted text from 12 documents.
[INFO] [detect_language] Running lang_detect...
[INFO] [detect_language] Complete. lang_name assigned.
[INFO] [dedup] Running ededup...
[INFO] [dedup] Removed 2 duplicate documents. 10 documents remaining.
[INFO] [doc_quality] Running doc_quality...
[INFO] [doc_quality] Computed docq_* features for 10 documents.
[INFO] [readability] Running readability...
[INFO] [readability] Computed readability features for 10 documents.
[INFO] [quality_branching] Branching 10 documents...
[INFO] [quality_branching] Routed 6 documents to link 'High Quality Documents' (flesch_reading_ease > 60).
[INFO] [quality_branching] Routed 4 documents to link 'Low Quality Documents' (flesch_reading_ease <= 60).
[INFO] [ml_enrichment_high_quality] Running ml_enrichment on 6 documents...
[INFO] [embeddings_high_quality] Generating embeddings with model 'openai/granite4:latest'...
[INFO] [vectordb_high_quality] Uploaded 6 documents to OpenSearch index 'high_quality_docs'.
[INFO] [sql_filter_low_quality] Running sql_filter on 4 documents...
[INFO] [sql_filter_low_quality] Filtered out 0 empty documents. 4 documents remaining.
[INFO] [embeddings_low_quality] Generating embeddings with model 'openai/granite4:latest'...
[INFO] [vectordb_low_quality] Uploaded 4 documents to OpenSearch index 'low_quality_docs'.
[INFO] Flow execution finished successfully in 4.23s.
将质量指标和路由规则从自定义的 Python 代码中移出，放入声明式配置中，可大幅提升数据接入速度并便于维护。处理成本取决于文档的价值，RAG 应用程序能为不同受众生成整洁且定制化的索引。
这种分支机制适用于流程中提前计算的任何指标。可根据语言、个人信息风险评分或自定义分类标签来调整路由规则。由于管道逻辑与调度代码相互独立，团队在扩展管道时无需增加维护负担。