跳到主要内容

流水线层:声明式直出 vs 多线程分阶段

30 秒导读: 后端(02)把字节变成"可处理的形态"后,流水线(pipeline) 负责把它真正加工成 DoclingDocument。所有流水线都走同一条骨架——搭建 → 装配 → 富化 → 定状态; 真正拉开差距的是"搭建"这一步:能直出的格式一行调用就完事,PDF 则要拆成 OCR/版面/表格等多个阶段、 用多线程流水并行跑。本章讲这条骨架和三种主流水线(Simple / StandardPdf / Vlm)的取舍。

本章位置:后端把原始输入变成 backend 对象后,DocumentConverter 按格式挑一条流水线来跑 (挑选逻辑见 01 入口与分派)。各阶段内部的模型(版面、表格、OCR 具体怎么算) 留给 04;DoclingDocument 怎么拼、怎么导出给 RAG 留给 05。

1. 这是什么(零基础也能懂)​

一句话定义: 流水线是"把一份输入文档加工成统一 DoclingDocument 的装配车间"—— 它规定了先做什么、后做什么、出错怎么办。

为什么需要它。 不同格式的加工难度天差地别:

  • 一个 Markdown / HTML 文件,后端本身就能直接吐出结构化文档——几乎不用"加工"。
  • 一个扫描版 PDF,得先渲染每一页、跑 OCR 认字、跑版面模型分块、跑表格模型还原表格、 再按阅读顺序拼起来——十几个步骤,还要顶得住上百页的量。

如果每种格式各写一套流程,代码会散架。Docling 的做法是:用一条统一骨架框住所有流水线, 把"每种格式的差异"收敛到骨架里的少数几个可重写方法。于是:

  • 简单格式 → 重写"搭建"这一步,一行直出。
  • PDF → 重写"搭建"这一步,塞进一整套多线程分阶段流水。

一句话直觉: 把流水线想成工厂流水线的总控。传送带的走向(骨架)是固定的; 不同产品(格式)只是换掉传送带上的加工机器。

用起来什么样。 使用者几乎感觉不到流水线的存在——DocumentConverter 会替你选好:

# 示意,非源码
from docling.document_converter import DocumentConverter

conv = DocumentConverter()
res = conv.convert("scan.pdf")      # 背后跑 StandardPdfPipeline(多线程分阶段)
res = conv.convert("notes.md")      # 背后跑 SimplePipeline(后端直出)
print(res.document.export_to_markdown())

重点看:同一个 API,背后是两条完全不同的流水线;这一层的价值就是"把差异藏起来"。

2. 顶层全景(它大概怎么转)​

2.1 统一骨架:每条流水线都跑这四步​

不管哪种流水线,对外只有一个入口 execute(),它固定按四步走。真源码在 docling/pipeline/base_pipeline.py:69(BasePipeline.execute):

              BasePipeline.execute(in_doc)
                        │
    ┌───────────────────┼────────────────────┐
    │  ① _build_document     搭建:原始 → 页/文档骨架    │
    │  ② _assemble_document  装配:页拼成 DoclingDocument │
    │  ③ _enrich_document    富化:对文档元素做二次加工    │
    │  ④ _determine_status   定状态:SUCCESS / PARTIAL…  │
    └───────────────────┬────────────────────┘
                        │  (无论成败)
                   finally: _unload   卸载后端、释放资源

怎么读这张图: 从上到下是严格顺序;①②只管"搭出文档结构",从③起只依赖 conv_res.document (源码在 base_pipeline.py:81 明确注释了这个边界)。整段包在 try/except/finally 里—— 异常→状态判 FAILURE,finally 里必定 _unload 释放资源。

四步各自的职责:

步骤方法干什么谁必须实现
① 搭建_build_document把后端产物变成页列表或直接的文档@abstractmethod,每条流水线必写
② 装配_assemble_document把逐页结果拼成一个 DoclingDocument默认空实现,PDF/VLM 重写
③ 富化_enrich_document遍历文档元素跑 enrichment_pipe基类通用实现,一般不重写
④ 定状态_determine_status决定 SUCCESS / PARTIAL / FAILURE@abstractmethod,每条流水线必写

2.2 一个关键收尾:PARTIAL_SUCCESS 与"有错不算成功"​

_determine_status 定完状态后,execute 还有一道兜底(base_pipeline.py:86): 只要 conv_res.errors 非空,就绝不报 SUCCESS,自动降级为 PARTIAL_SUCCESS。

# 摘自 base_pipeline.py:83-87(execute 内)
conv_res.status = self._determine_status(conv_res)
if conv_res.status == ConversionStatus.SUCCESS and conv_res.errors:
    conv_res.status = ConversionStatus.PARTIAL_SUCCESS

这条规则贯穿全书:某一页 OCR 挂了、某页 VLM 输出被截断,文档整体还能出——但状态诚实地标成 "部分成功",错误明细进 conv_res.errors。

2.3 三种主流水线一览​

流水线文件适用输入搭建方式
SimplePipelinesimple_pipeline.pyMD/HTML/DOCX 等声明式后端后端一步 convert() 直出
StandardPdfPipelinestandard_pdf_pipeline.pyPDF / 图片多线程分阶段流水(本章重点)
VlmPipelinevlm_pipeline.pyPDF(视觉大模型路线)逐页喂给 VLM,整页转 DocTags/MD
AsrPipelineasr_pipeline.py音频/视频Whisper 转写成带时间戳的文本

3. 类层次:骨架如何一层层被特化​

理解流水线的关键,是看清继承链上每一层新增了什么。

BasePipeline (ABC)                       ← execute 骨架 + enrichment 遍历
   │   base_pipeline.py:50
   ├── ConvertPipeline                   ← 装配"通用富化模型"(图片分类/描述/图表)
   │      base_pipeline.py:153
   │      ├── SimplePipeline             ← _build_document = backend.convert()
   │      │      simple_pipeline.py:16
   │      ├── StandardPdfPipeline        ← 自带多线程 _build_document(本章重点)
   │      │      standard_pdf_pipeline.py:567
   │      └── PaginatedPipeline          ← 按 page_batch 逐页跑 build_pipe
   │             base_pipeline.py:231
   │             └── VlmPipeline         ← build_pipe = 整页 VLM
   │                    vlm_pipeline.py:83
   └── AsrPipeline                       ← 直接继承 BasePipeline(不需要富化/分页)
          asr_pipeline.py:611

注意:StandardPdfPipeline 直接继承 ConvertPipeline 而不走 PaginatedPipeline—— 它用自己的线程化 _build_document 取代了逐页循环。PaginatedPipeline 现在主要服务 VlmPipeline(和已弃用的 legacy PDF 流水)。

3.1 BasePipeline——骨架 + 富化遍历​

除了 §2 的 execute 骨架,基类还实现了通用的 _enrich_document (base_pipeline.py:111)。它做的事很固定:

  1. 遍历 conv_res.document 的所有元素(iterate_items);
  2. 对 enrichment_pipe 里的每个模型,用 model.prepare_element 挑出它关心的元素;
  3. 按 elements_batch_size 分批喂给模型。

关键一行是那句"必须耗尽"的注释(base_pipeline.py:128):模型返回的是生成器, for element in model(...) 必须走完,否则副作用不落地。

3.2 ConvertPipeline——预置"通用富化模型"​

ConvertPipeline(base_pipeline.py:153)在构造时就把三类跨后端通用的富化模型装进 enrichment_pipe:

富化模型干什么源码符号
图片分类判断插图是照片/图标/图表…DocumentPictureClassifier
图片描述给插图生成文字描述(可接远程模型)_get_picture_description_model
图表抽取把图表还原成结构化数据ChartExtractionModelGraniteVision / …V4

一个细节:图表抽取依赖图片分类,所以构造函数里用局部变量把 do_picture_classification 或上 do_chart_extraction(base_pipeline.py:160), 避免直接改动共享的 pipeline_options(改了会影响其哈希、破坏缓存)。

3.3 PaginatedPipeline——逐页跑 build_pipe​

PaginatedPipeline(base_pipeline.py:231)是"顺序分页"的流水线基类。它的 _build_document(base_pipeline.py:244)骨架是:

按 page_batch_size 把页切成一批批(默认 4,settings.py:32)
  每批:
    1. initialize_page      给每页挂上 page backend、量尺寸
    2. _apply_on_pages      让 build_pipe 里每个 model 依次处理这一批页
    3. 逐页清理             清 image_cache、unload page backend
  每批结束检查 document_timeout;超时→PARTIAL_SUCCESS,break

两个要点:

  • _apply_on_pages(base_pipeline.py:236)就是 build_pipe 的执行器——把 page_batch 依次穿过每个 build 模型:page_batch = model(conv_res, page_batch)。
  • 超时是"批粒度"的:累计耗时超过 document_timeout 就停,已处理的页保留, 未处理的页最后被过滤掉(base_pipeline.py:333 剔除 size is None 的页)。

_determine_status(base_pipeline.py:353)则扫描每一页的 backend:任何一页 backend 失效, 就追加一条 BACKEND_FAILURE 错误并降级 PARTIAL_SUCCESS。

4. build_pipe vs enrichment_pipe(一定要分清)​

这是本层最容易混的两个"管道"。它们跑在不同的对象、不同的阶段上。

build_pipeenrichment_pipe
作用对象Page 对象(逐页)DoclingDocument 的元素(NodeItem)
运行阶段① 搭建期(_build_document)③ 富化期(_enrich_document)
目的把一页"看懂":OCR/版面/表格/整页 VLM对成品文档做二次加工:图片描述、图表抽取、代码/公式识别
谁填充它各子类构造函数ConvertPipeline 预置 + 子类追加

一句话:build_pipe 负责"从像素到结构";enrichment_pipe 负责"结构好了再补料"。

时序上,富化永远在装配之后——因为富化操作的是已经拼好的 conv_res.document, 而不是零散的页(见 §2.1 的边界注释)。

5. SimplePipeline——声明式后端直出​

这是最简单的一条,适合本身就能吐结构化文档的格式(Markdown、HTML、DOCX…)。

它对"搭建"的实现几乎没有内容——直接调用后端的 convert() (simple_pipeline.py:26,SimplePipeline._build_document):

# 摘自 simple_pipeline.py:39-40
with TimeRecorder(conv_res, "doc_build", scope=ProfilingScope.DOCUMENT):
    conv_res.document = conv_res.input._backend.convert()

前提是后端必须是 DeclarativeDocumentBackend(能直出文档的后端,见 02); 否则直接抛 RuntimeError。_determine_status 也简单——没有别的可评估,直接 SUCCESS (simple_pipeline.py:43)。

为什么它没有 build_pipe。 因为"看懂一页"的活儿全在后端里做完了,流水线无须再逐页加工—— 它只借用基类的 execute 骨架和 ConvertPipeline 的富化能力(仍会跑图片分类/描述等)。

6. StandardPdfPipeline——多线程分阶段(本章重点)​

PDF 是最重的路径,也是这个文件里工程密度最高的地方。它没有用 PaginatedPipeline 的 顺序分页,而是自己实现了一套多线程流水(standard_pdf_pipeline.py:567)。

6.1 为什么要线程化​

一份 PDF 的每页要顺序经过 OCR → 版面 → 表格 → 装配。如果一页一页串行走,慢阶段会拖死整体。 更好的做法是让各阶段像工厂流水线一样并行:第 1 页在跑表格时,第 3 页可以同时在跑 OCR。

Docling 的方案:每个阶段一个专属线程,阶段之间用有界队列连起来,页像零件一样在传送带上流动。

producer 线程                     每个阶段 = 1 个线程,阶段间 = 有界队列
逐页 load_page                    (队列满则上游阻塞 = 背压)
     │
     ▼
 [preprocess] ─q─▶ [ocr] ─q─▶ [layout] ─q─▶ [table] ─q─▶ [assemble] ─q─▶ output_q
                                                                              │
                                                                     主线程 drain 汇总

怎么读:横向是数据流向,q 是 ThreadedQueue。每个方框独立跑在自己的线程里, 谁先干完谁就把结果丢进下游队列,不必等别人。

6.2 三块积木​

积木 A:ThreadedQueue——有界队列 + 背压 + close 传播(standard_pdf_pipeline.py:149)

一个手写的线程安全队列,三个特性:

  • 有界:put 在队列满时阻塞上游(standard_pdf_pipeline.py:169 的 while len >= max)—— 这就是"背压",防止快阶段把慢阶段淹没、内存爆掉。
  • 批量取:get_batch(size)(:184)一次取最多 size 个,凑批喂模型更划算。
  • close 传播:close()(:204)把 _closed 置真并唤醒所有等待者;下游读到"空且已关闭" 就知道该收工了——不需要哨兵值(sentinel),关闭本身就是终止信号。

积木 B:ThreadedPipelineStage——每阶段一线程,按 run_id 分组批处理(standard_pdf_pipeline.py:216)

每个阶段自己起一个线程(_run,:273):循环 get_batch → _process_batch → _emit 到下游。 核心巧思在 _process_batch(:288):它先按 run_id 把批分组再喂模型。

# 示意,非源码:同一次 execute 的页才拼在一起批处理
groups = defaultdict(list)
for item in batch:
    groups[item.run_id].append(item)   # run_id = 一次 execute 调用的唯一号
for run_id, items in groups.items():
    processed = model(items[0].conv_res, [i.payload for i in items])

重点看 run_id:并发调用 execute 时,不同文档的页可能挤在同一个阶段队列里; 按 run_id 分组保证"只有同一次转换的页被一起批处理",互不串味。run_id 由 itertools.count(1) 单调发号(:573),比用 id() 稳(对象回收后 id 会复用)。

阶段出错不会炸掉整条流水:某个 run 的模型抛异常时,把该组所有 item 标 is_failed=True 并附上 ErrorItem,照样往下游传(:368)——失败的页会一路流到出口被统计,而不是卡死。

积木 C:PreprocessThreadedStage——带前置校验的首阶段(standard_pdf_pipeline.py:397)

首阶段特殊:它在跑预处理前先逐页校验 backend(为空?没挂上?is_valid() 失败?), 把坏页标记成对应类别的失败并恰好发射一次(:449 的注释),避免模型later再抛导致重复计错。

6.3 wiring:五个阶段怎么接起来​

_create_run_ctx(standard_pdf_pipeline.py:682)为每一次 execute 新建一套阶段和队列 (per-run 隔离,并发不共享可变状态),然后线性接线:

preprocess.out → ocr.in
ocr.out        → layout.in
layout.out     → table.in
table.out      → assemble.in
assemble.out   → output_q     ← assemble 阶段还挂了 postprocess=_release_page_resources

各阶段的批大小、轮询间隔、队列上限都来自 ThreadedPdfPipelineOptions (默认见下,pipeline_options.py:1810 起):

参数默认含义
ocr_batch_size / layout_batch_size / table_batch_size4各阶段凑批页数
batch_polling_interval_seconds0.5阶段等待凑批的最长时间
queue_max_size100阶段间队列上限(背压阈值)
page_batch_size(全局)4见 settings.py:32,分页流水用

6.4 producer + drain:页怎么进、结果怎么出​

_build_document(standard_pdf_pipeline.py:764)是总调度:

  1. 起阶段线程,再起一个独立的 producer 线程(_produce_pages,:811):逐页 load_page 挂上 backend、量尺寸,把 ThreadedItem 塞进首阶段队列;喂完 close() 首队列。
  2. 主线程 drain(:846 起循环):不断从 output_q.get_batch 拉成品页,成功进 proc.pages,失败进 proc.failed_pages,直到"成功+失败 == 总页数"。
  3. 早退保护(:878):如果下游队列空了且已关闭但还有页没到齐,把缺失的页补记为失败, 避免主线程死等。

6.5 document_timeout 与 timed_out_run_ids​

超时机制和分页流水不同——这里是"打标记 + 让在途工作自己识别":

  • 主线程 drain 时检查累计耗时(:849),一旦超过 document_timeout:把该 run_id 加进 共享集合 timed_out_run_ids、关掉首阶段队列、立刻 break(不等在途页)。
  • 各阶段 _process_batch 每次会看这个集合(:298):属于超时 run 的页跳过模型、直接标 TIMEOUT 失败往下传——已完成的工作能流出,新工作被中止。
  • 收尾时把没跑完的页补记成 TIMEOUT 失败(:907)。

妙处:timed_out_run_ids 是一个所有阶段共享引用的 set(RunContext.timed_out_run_ids,:559), 主线程往里加、工作线程读——用最轻的方式实现了"跨线程的软取消"。

6.6 _integrate_results:把散落的结果汇总成状态​

drain 完成后,_integrate_results(standard_pdf_pipeline.py:941)负责收口:

  • 只保留成功完成的页(conv_res.pages 用成功页重建);
  • 把每个失败页的 ErrorItem 追加进 conv_res.errors;
  • 定状态:超时→PARTIAL_SUCCESS;全挂→FAILURE;部分成功→PARTIAL_SUCCESS;否则 SUCCESS (借助 ProcessingResult 的 is_partial_success / is_complete_failure 属性,:141)。

装配阶段 _assemble_document(:997)再把逐页结果跑 ReadingOrderModel(阅读顺序)、 HeadingHierarchyModel(标题层级),并把失败/跳过的页也补进 document.pages (_add_failed_pages_to_document,:1098)以保证页码和分页符正确。这些模型细节归 04。

一个已知边界(文件顶部与 :764 注释都写明了):如果某工作线程卡在阻塞调用(模型推理或 PDF backend 迭代),清理时只等 15 秒(:264、:930)就放弃该线程,资源可能泄漏。

7. VlmPipeline——整页交给视觉大模型​

VlmPipeline(vlm_pipeline.py:83)走的是完全不同的哲学:不再拆成 OCR/版面/表格, 而是把整页图像丢给一个视觉语言模型(VLM),让它一次性输出结构化标记。

它是 PaginatedPipeline 的子类,所以复用 §3.3 的逐页循环;差别只在 build_pipe 里放的是 一个 VLM 模型:

  • 新运行时:VlmConvertModel(vlm_pipeline.py:122,build_pipe)。
  • initialize_page(:207)给每页挂 backend、量尺寸;需要时抓底层文本(force_backend_text)。

它的产物是文本标记,_assemble_document(:276)按响应格式把它翻成 DoclingDocument:

响应格式转换方法
DOCTAGS_turn_dt_into_doc(SmolDocling 的 DocTags)
DOCLANG_turn_doclang_into_doc
MARKDOWN / HTML复用对应后端再解析
DEEPSEEKOCR_MARKDOWN / DOTS_JSON / CHANDRA_HTML各自的解析器

_determine_status(:235)在基类之上额外识别 VLM 特有的失败:某页没有 VLM 输出, 或 stop_reason 是 LENGTH(被截断)/ CONTENT_FILTERED(被过滤),都降级 PARTIAL_SUCCESS。

StandardPdf vs Vlm 的取舍:

StandardPdfPipelineVlmPipeline
思路多个专用模型分阶段单个大模型整页理解
并行阶段间多线程流水顺序逐页(靠模型自身批处理)
强项可控、可调、资源省复杂版面/表格更鲁棒,少调参
代价组件多、要接多个模型VLM 重、可能截断/被过滤

8. AsrPipeline——音频转写(一句带过)​

AsrPipeline(asr_pipeline.py:611)直接继承 BasePipeline(不需要富化,也不分页)。 它的 _build_document(:681)把音频交给一个 Whisper 系模型转写 (原生 Whisper / MLX Whisper / WhisperS2T 三选一,构造时按 options 类型挑), 产出带时间戳、带说话人的文本段落塞进文档。_determine_status(:665)在"文档没有任何 非空文本"时报 PARTIAL_SUCCESS。它只支持 NoOpBackend(音频没有可解析的字节结构)。

9. 巧妙之处(可借鉴的技术)​

  • 关闭即终止,免哨兵。 ThreadedQueue.close() 唤醒所有等待者,下游读到"空且关闭"即收工 (standard_pdf_pipeline.py:204)——比在队列里塞 None 哨兵更干净,不会因分支多而漏发。
  • run_id 而非 id() 做批分组键。 itertools.count 单调发号(:573),规避了对象回收后 id() 复用导致的串号风险;还顺带让并发 execute 的页在共享阶段里天然隔离(:290)。
  • 软取消用一个共享 set。 timed_out_run_ids(:559)让主线程和工作线程用最轻的方式协作取消—— 没有锁、没有事件对象,读写一个集合即可(:298)。
  • "有错就不 SUCCESS"收在骨架里。 降级逻辑放在 execute(base_pipeline.py:86)而非各子类, 保证所有流水线状态语义一致,子类不会各写各的。
  • 改 options 前先 copy,保住缓存哈希。 _init_models 里 code_formula_options.model_copy (standard_pdf_pipeline.py:624)避免就地改动 pipeline_options 破坏 pipeline 缓存(见 issue #3109)。

10. 边界与局限(诚实)​

  • 卡住的线程会被放弃。 工作线程/producer 若卡在阻塞的模型或 backend 调用,清理只等 15 秒 就 abandon,资源(如 pypdfium2_lock)可能泄漏(standard_pdf_pipeline.py:264、:764 注释)。
  • 两种超时语义不同。 分页流水是"批粒度"累计计时(base_pipeline.py:294); 线程 PDF 流水是"打标记 + 在途识别"(standard_pdf_pipeline.py:849)。用户看到的都是 PARTIAL_SUCCESS + TIMEOUT,但触发时机不一样。
  • VLM 可能悄悄截断。 输出超长会 stop_reason=LENGTH,靠 _determine_status(vlm_pipeline.py:258) 兜住降级;不看状态只看文档的话,可能拿到不完整结果。
  • VlmPipeline 不支持 ThreadedDoclingParseDocumentBackend——它需要有序/随机页访问 (vlm_pipeline.py:72),遇到会直接抛错让你改用 StandardPdfPipeline。

11. 代码地图(导航索引)​

主题文件路径符号名
统一骨架 executedocling/pipeline/base_pipeline.py:69BasePipeline.execute
富化遍历docling/pipeline/base_pipeline.py:111BasePipeline._enrich_document
通用富化模型装配docling/pipeline/base_pipeline.py:153ConvertPipeline
逐页分批循环docling/pipeline/base_pipeline.py:244PaginatedPipeline._build_document
build_pipe 执行器docling/pipeline/base_pipeline.py:236PaginatedPipeline._apply_on_pages
分页流水定状态docling/pipeline/base_pipeline.py:353PaginatedPipeline._determine_status
声明式直出docling/pipeline/simple_pipeline.py:26SimplePipeline._build_document
有界队列/背压/closedocling/pipeline/standard_pdf_pipeline.py:149ThreadedQueue
阶段线程 + run_id 分组docling/pipeline/standard_pdf_pipeline.py:288ThreadedPipelineStage._process_batch
首阶段前置校验docling/pipeline/standard_pdf_pipeline.py:397PreprocessThreadedStage
阶段接线docling/pipeline/standard_pdf_pipeline.py:682StandardPdfPipeline._create_run_ctx
producer + drain 调度docling/pipeline/standard_pdf_pipeline.py:764StandardPdfPipeline._build_document
结果汇总定状态docling/pipeline/standard_pdf_pipeline.py:941StandardPdfPipeline._integrate_results
线程配置项docling/datamodel/pipeline_options.py:1884ThreadedPdfPipelineOptions
整页 VLM 流水docling/pipeline/vlm_pipeline.py:83VlmPipeline
VLM 状态判定docling/pipeline/vlm_pipeline.py:235VlmPipeline._determine_status
音频转写流水docling/pipeline/asr_pipeline.py:611AsrPipeline

相关章节:上游怎么选到这条流水线见 01 入口与三级分派;流水线消费的 backend 形态见 02 后端层;各阶段模型的内部实现见 04 PDF 识别流水线; 产物 DoclingDocument 的结构与 RAG 导出见 05 数据模型与输出。