多模态科研内容理解 06：理解流水线集成与 M3 里程碑验收

系列四收官。前五课各自做出了理解能力的零件：PDF 结构化、语义向量、深度分类、图表抽取、融合策略。这一课把它们组装成一条从 PDF 到统一表示的端到端流水线，完成 M3 里程碑验收。集成本身也是工程课——零件都能跑不等于系统能跑。

前置阅读：系列四全部前五课。管道设计思想参考 AI 自动化（幂等、落盘、失败出口三原则）。

流水线架构：阶段、产物、契约

PDF 文件
  → S1 解析（GROBID）        产物：结构化 JSON（章节/参考文献）
  → S2 图表抽取（版面检测）   产物：图片文件 + 图注配对记录
  → S3 文本向量化            产物：多路 Embedding（标题摘要/图注）
  → S4 融合表示              产物：统一论文表示记录
  → S5 入库                  产物：features 层更新

每个阶段之间靠落盘的产物文件 + 明确的 Schema 通信，而不是函数调用。这就是大数据管理分层思想和 AI 自动化幂等原则的组合：任何阶段挂了，从它的输入产物重跑即可，不用从头再来。

各阶段间的契约示例（S1 → S2/S3 的输入约定）：

# 每个阶段输入输出都是 paper_id 键控的记录
{
  "paper_id": "arxiv:1706.03762",
  "stage": "s1_parsed",
  "payload": {"sections": [...], "references": [...]},
  "parser_version": "grobid-0.8.0",
  "processed_at": "2026-08-29T01:00:00"
}

parser_version 字段关键：解析器升级后，能精确找出”哪些记录是旧版本产的”，选择性重跑。

编排代码：朴素但可靠的第一版

系列五会用 Prefect 正式编排，这里先用一个调度循环把设计跑通——顺序执行、断点续传、失败隔离：

STAGES = [s1_parse, s2_figures, s3_embed, s4_fuse, s5_load]

def run_pipeline(paper_ids: list[str], state_path: Path):
    state = load_json(state_path, default={})

    for pid in paper_ids:
        for stage in STAGES:
            key = f"{pid}:{stage.__name__}"
            if state.get(key) == "done":
                continue                      # 幂等：做过的跳过
            try:
                stage(pid)
                state[key] = "done"
            except Exception as e:
                state[key] = f"failed: {e}"
                log_failed(pid, stage.__name__, e)
                break                         # 该论文后续阶段没意义，跳下一篇
        save_json(state_path, state)

这是展示控制流的伪代码，load_json、save_json 和五个阶段都需另行实现；跳过 done 不是幂等性的充分条件。状态应至少包含论文 ID、输入摘要、阶段实现版本、产物校验和。解析器升级后，S1 及依赖其结果的下游阶段都要失效；只清 S1 会混用新解析和旧向量。产物应先写临时文件、校验完成后原子替换，再提交状态。进程在二者之间中断时，由恢复逻辑检查产物，而不是仅相信状态字符串。

端到端质量门

每个阶段输出应同时记录输入版本、处理器版本和成功状态。文件存在但写到一半，不应被当作已完成产物。恢复执行时要检查完整性与依赖版本，再决定复用还是重算；否则断点续传可能一直沿用旧错误。

每个阶段有自己的质量检查，集成层再加一道端到端的：

def pipeline_quality_gate(paper_id):
    rec = load_record(paper_id)
    checks = {
        "解析完整": len(rec["sections"]) >= 3,
        "向量存在": rec["fused_embedding"] is not None,
        "图表有归属": all(f["caption"] for f in rec["figures"]),
        "检索可用": smoke_search_recall(paper_id) > 0,   # 用它自己的标题搜它
    }
    return all(checks.values()), checks

标题自检是低成本冒烟线索，不应保证一定排第一：同名论文、重复版本和近似检索都会影响排名。先验证 ID—向量映射，再在规定 K 内检查目标是否存在，并单独测试真实查询的相关性。此处 all([]) 会返回真，图表列表为空不代表抽取成功；需要结合论文是否有图、阶段成功状态及抽样标注判断。sections >= 3 也只是教学阈值，不能直接代表解析完整。

M3 验收清单

流水线五个阶段全部幂等可重跑，状态按论文×阶段粒度持久化。
在 500 篇论文上端到端跑通，端到端质量门通过率 ≥ 95%，失败样本有画像分类。
检索质量复测：第 2 课评测集上融合表示的 Recall@10 不低于单独文本向量（融合不能开倒车）。
每篇论文的处理成本（耗时、token/算力）有台账，为规模化做预算依据。
已知局限文档化：扫描版 PDF 未覆盖、公式只做定位未识别、参考文献未做实体消歧。
git tag m3-understanding-pipeline。

与 M1/M2 的联动检查

里程碑不是孤立的：M3 的理解流水线输出要回流验证前两个里程碑——解析出的正文章节能不能提升 M2 分类器（留作系列六的优化实验）；融合检索的语料版本是不是钉在 M1 的 m1.0 数据集上。里程碑之间的依赖关系画成图，是项目叙事（系列六）的主线。

踩坑排查

症状	原因	处理
重跑流水线结果翻倍	阶段产物追加写不幂等	产物按 paper_id 覆盖写
某篇失败导致全批中断	异常没隔离到论文粒度	try/except 按论文隔离 + 失败台账
升级解析器后新旧结果混杂	无 parser_version	产物带版本，按版本选择性重跑
流水线越跑越慢	每阶段重复读全量数据	阶段间传路径，数据按需加载
检索冒烟测试失败	向量/id 错位或漏入库	按第 2 课的对齐校验排查
端到端通过率低	坏 PDF 没分流	扫描版/损坏件提前标记跳过

作品集证据

M3 是”系统集成能力”的证明：五个异构组件（含外部服务、深度学习模型、向量计算）编排成一条可靠流水线，有状态管理、质量门、成本台账和失败画像。这是从”会写模型”到”会做系统”的分水岭证据。

练习

实现五阶段流水线并在 100 篇论文上验证断点续传（中断后重跑只补未完成项）。
实现端到端质量门，构造一个”向量错位”故障验证冒烟测试能抓住。
统计每篇论文的分阶段耗时，找出瓶颈阶段并给出一个优化方案。
画 M1→M2→M3 的里程碑依赖图，标注每个里程碑的验收标准。

面试常问

Q：多阶段流水线的设计原则？ 阶段间落盘通信（而非内存传递）、每阶段幂等、状态细粒度持久化（论文×阶段）、失败隔离不中断批次、产物带版本支持选择性重跑、端到端冒烟测试。

Q：怎么验收一条 AI 流水线？ 三层：各阶段质量指标（解析完整率、检测准确率）、端到端指标（质量门通过率、检索 Recall 不回退）、工程指标（吞吐、单篇成本、失败率）。只有单点指标没有端到端验证，集成等于没验收。

Q：已有模型/服务升级时怎么管理？ 产物带版本号（parser_version/model_version），升级后按版本标记选择性重跑；新旧版本指标对比通过后切换；保留回滚路径（旧产物不立即删）。

完成上述验收后，才能将你的 M3 标记为达成。500 篇与 95% 是本课练习目标，不是已有运行结果。 下一课进入系列五：数据管理与 AI 自动化 01：PostgreSQL + pgvector 科研知识库。