流水线阶段¶

Cartex 管道按顺序运行两个阶段：提取和富化。入口点为 main.py::run()，它先调用提取器，再将结果连同用户定义的模式一起传递给富化器。

def run(file_path: str, page_numbers: list[int], schema: UserTableSchema) -> list[EnrichedRow]:
    extracted = extractor.extract(file_path, page_numbers[0])
    results = enricher.enrich(extracted, schema)
    return results

阶段一：提取¶

src/pipeline/extractor.py 中的 Extractor 类将 PDF 页面转换为包含表格和上下文信息的 ExtractionResult。

页面渲染¶

提取器使用 PyMuPDF 以 src/config.py 中配置的 DPI（默认 210）将 PDF 页面渲染为 PNG 图像，产生 Gemini 处理所需的图像字节。

def _pdf_to_image(self, file_path: str, page_number: int) -> bytes:
    doc = pymupdf.open(file_path)
    page = doc.load_page(page_number)
    pix = page.get_pixmap(dpi=config.dpi)
    return pix.tobytes()

Gemini 视觉调用¶

针对渲染图像独立执行两次 Gemini 调用：

TABLE_EXTRACTION — 检测页面上所有表格（主清单及辅助参考表），提取表头和行数据，分配角色（MAIN、AUXILIARY、OTHER），并识别主键列。
CONTEXT_EXTRACTION — 检测所有非表格上下文信息：通用备注、性能规格、规范要求、图例图示、条目卡片及尺寸图。

两次调用均使用高级 Gemini 模型（默认为 gemini-3.1-pro-preview），并返回经 Pydantic 模式（GeminiTableResult 和 GeminiContextResult）验证的结构化 JSON。

多页提取¶

提供多个页码时，extract_pages() 通过 asyncio.gather() 并发处理所有页面。每页并行运行表格提取和上下文提取两次调用。结果合并时，对上下文条目进行基于内容的去重，以避免跨页重复。

异步页面管道¶

对于单页，异步路径（extract_async）并发运行表格提取和上下文提取：

tables_result, contexts_result = await asyncio.gather(
    self._extract_tables_async(image_bytes),
    self._extract_context_async(image_bytes),
)

阶段二：富化¶

src/pipeline/enricher.py 中的 Enricher 类接收 ExtractionResult 和 UserTableSchema，为每个主清单行填充所有目标列。

路由¶

富化器首先调用 Router 确定适用的专业策略。路由器将提取数据的紧凑摘要（表格元数据、上下文片段、模式列）发送至快速 Gemini 模型，接收列出适用 StrategyType 值的 GeminiRoutingResult。

若路由器返回空列表，富化器将回退至单体 ENRICHMENT 提示词——单次 Gemini 调用处理所有富化。

专业策略执行¶

选定策略后，富化器通过 asyncio.gather() 并发运行每个专业策略：

tasks = [
    self._run_specialist_async(extraction_result, schema, strategy)
    for strategy in strategies
]
specialist_results = await asyncio.gather(*tasks)

每个专业策略接收相同的输入（注入了 __row_id__ 字段的表格、上下文条目及目标模式），但使用专注于单一富化方式的专业提示词。

合并¶

所有专业策略完成后，_merge_specialist_results() 将其输出合并为单个 list[EnrichedRow]。详见合并算法。

端到端流程¶

下图展示了两个阶段及其内部并发结构。

flowchart TB
    subgraph stage1["阶段一：提取"]
        PDF["PDF 页面"] --> RENDER["PyMuPDF 渲染<br/>(210 DPI)"]
        RENDER --> IMG["图像字节"]
        IMG --> TABLE_CALL["TABLE_EXTRACTION<br/>(Gemini Pro)"]
        IMG --> CTX_CALL["CONTEXT_EXTRACTION<br/>(Gemini Pro)"]
        TABLE_CALL --> TABLES["list[TableModel]"]
        CTX_CALL --> CONTEXTS["list[ContextModel]"]
        TABLES --> ER["ExtractionResult"]
        CONTEXTS --> ER
    end

    subgraph stage2["阶段二：富化"]
        ER --> ROUTER["路由器<br/>(Gemini Flash)"]
        ROUTER --> STRATS["list[StrategyType]"]
        STRATS --> S1["专业策略 1"]
        STRATS --> S2["专业策略 2"]
        STRATS --> SN["专业策略 N"]
        S1 --> MERGE["_merge_specialist_results()"]
        S2 --> MERGE
        SN --> MERGE
        MERGE --> OUT["list[EnrichedRow]"]
    end

    stage1 --> stage2

行 ID 分配¶

在将表格传递给专业策略之前，富化器通过 _assign_row_ids() 向每条主清单行注入 __row_id__ 字段。该 ID 来源于提取阶段检测到的 primary_key_column（如 Type Mark）。主键值重复时附加数字后缀（W1_2、W1_3）。若未检测到主键列，行将回退至基于索引的 ID（table_0_0_row_0）。

专业策略必须将 __row_id__ 值原样复制至输出的 row_id 字段。合并步骤使用该值跨策略聚合同一行的输出。