document-workflows

文档工作流 — ADE 流水线模式

概述

此技能提供了可复用的构建模块，用于将 LandingAI ADE 基础功能（解析、提取、拆分）组合成可用于生产环境的文档处理流水线。它是对 document-extraction 技能的补充：

关注点	document-extraction	document-workflows
范围	ADE SDK API：解析、提取、拆分、依据定位	端到端工作流：批处理、RAG、数据库、分类路由
适用场景	需要调用单个 ADE 操作	需要将多个操作组合成工作流
代码	带参数的 SDK 方法调用	包含错误处理和并行处理的完整函数
依赖项	仅 landingai-ade	+ 工作流专用库（pandas、chromadb 等）

理念： 按工作流模式（批处理、RAG、数据库插入）组织，而不是按文档类型组织。无论文档是发票、公用事业账单还是医疗表单，都适用同一种模式。

---

步骤 0（必做）— 预检文档探索 {#pre-flight}

**在编写任何流水线代码前，**只要当前会话中尚未检查过文档的内部结构，就先执行此步骤。

规则：在先对样本文档运行工具 2（诊断解析）之前，绝不编写章节检测、标题匹配或文本搜索代码。 标题格式因文档而异，无法仅根据任务描述或文档类型推断；要确定标题格式，唯一可靠的方法就是查看实际的 ADE 输出。

常见意外情况：论文中的“Introduction”标题可能显示为 1. Introduction（纯文本，没有 #）、## Introduction、INTRODUCTION（全大写），或嵌入正文文本块中。如果判断有误，就会无声地匹配失败（匹配到 0 个文本块），而排查问题需要重新完整解析。

编写任何代码之前，先对 1–3 份有代表性的样本文档运行工具 1（视觉渲染）和工具 2（诊断解析）。这不到一分钟，就能避免原本通过预检即可避免的调试迭代。

工具 1 — 页面视觉渲染

将 1–2 页渲染为 PNG，并将其作为视觉上下文阅读。不会消耗 ADE 额度，但每张 PNG 都会占用上下文 token。布局不明确或文档来源不明（手写？扫描件？表单？）时使用。

.venv/bin/python - << 'EOF'
import pymupdf
from pathlib import Path
from PIL import Image

pdf = Path('path/to/sample.pdf')
out_dir = Path('/tmp/ade_preflight'); out_dir.mkdir(exist_ok=True)
doc = pymupdf.open(pdf)
for pg in range(min(2, len(doc))):   # first 2 pages only
    pix = doc[pg].get_pixmap(matrix=pymupdf.Matrix(1.5, 1.5))   # 108 DPI
    img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
    out = out_dir / f"{pdf.stem}_page{pg + 1}.png"
    img.save(out)
    print(out)
doc.close()
EOF

然后阅读保存的 PNG。可立即解答以下问题：

• 标题是否为粗体文本（→ ADE 可能输出纯文本标题，而非 # Heading）
• 文档是手写的还是扫描件？→ 需要 Tesseract OCR，而不是 PyMuPDF
• 是单栏还是双栏布局？
• 是否存在噪声：页眉、页码、水印、印章？

工具 2 — ADE 诊断解析

解析 1 份样本并打印 Markdown 结构和文本块清单。此操作会消耗 ADE 额度，最多只解析 1–3 份样本，绝不要解析整个语料库。

.venv/bin/python - << 'EOF'
import os
from pathlib import Path
from collections import Counter
from dotenv import load_dotenv

# Load API key: prefer existing env var, then .env file lookup
load_dotenv()  # Load API key from .env. Add a path to the .env if needed.

from landingai_ade import LandingAIADE
client = LandingAIADE()
pr = client.parse(document=Path('path/to/sample.pdf'))

print("=== MARKDOWN (first 80 lines) ===")
for i, ln in enumerate(pr.markdown.splitlines()[:80], 1):
    print(f"{i:3}: {ln}")

print("\n=== CHUNKS ===")
for ch in pr.chunks:
    txt = (ch.markdown or '').replace('\n', ' ')[:70]
    b = ch.grounding.box
    print(f"p{ch.grounding.page} {ch.type:12} "
          f"l={b.left:.2f} t={b.top:.2f} r={b.right:.2f} b={b.bottom:.2f} | {txt}")

print(f"\nPages: {pr.metadata.page_count}  "
      f"Chunks: {len(pr.chunks)}  "
      f"Types: {dict(Counter(ch.type for ch in pr.chunks))}")
EOF

**费用说明：**首次运行后，用 pr.model_dump() 将解析结果保存为 JSON 文件。后续开发时加载该文件，而不要再次调用 client.parse()。只有文档集发生变化时才重新解析。

需要检查的内容

观察结果	影响
标题为 1. Introduction（纯文本，没有 #）	ADE Markdown 不会使用 ATX 标题 → 使用 ADE 提取，而不是正则表达式
不同文档的标题格式不一致（一份中是 # INTRO，另一份中是 1. Intro）	正则表达式会在某些文档中失效 → 使用 ADE 提取以提高稳健性
每个 ch.markdown 都以 <a id='...'></a> 开头	字符串匹配或显示前先移除锚点
双栏：同一页上的文本块分别有 l=0.07 和 l=0.50	文本顺序是先左栏后右栏；章节可能横跨两栏
文本块在分页处被截断，单词只剩一半	章节跨页；需要收集多个页面上的文本块
marginalia 文本块位于 t<0.08 或 t>0.90	页眉 / 页码 → 从内容提取中排除
页面图像中可见扫描或手写内容	PyMuPDF 文本提取无法奏效 → 使用 Tesseract OCR

工具 3 — 裁剪后视觉验证（边界框工作流必需） {#post-crop-verification}

生成任何边界框裁剪图或叠加图（提取图形、裁剪分块、提取表格单元格、单词级定位）后，至少将一张输出 PNG 作为图像读回，并描述你看到的内容。将描述与用户的请求进行比较。这有助于发现：

• 页面错误 — ADE 页码从 0 开始；索引偏差 1 会使裁剪落在相邻页面上，而该页内容可能完全不同
• 区域错误 — 坐标系不匹配，导致裁剪出空白区域或无关部分

规则：在目视读取至少一张输出 PNG 并确认其内容与用户请求相符之前，绝不要宣布裁剪工作流已完成。

验证步骤

1. 将第一张裁剪图保存为 PNG（工作流本身已完成此操作）
2. 将 PNG 文件作为图像读取（对 PNG 路径使用 read_file 工具）
3. 描述你看到的内容：出现了什么内容、表格、图形或文本？
4. 与用户的请求进行比较：
   • 用户要求“Events 表格” → 裁剪图是否显示了 Events 表格？
   • 用户要求“Figure 3” → 裁剪图是否显示了图表／示意图？
   • 用户要求“Introduction 部分” → 裁剪图是否显示了引言文本？
5. 如果描述不符 → 在继续之前检查页面索引和边界框坐标
6. 验证第一张裁剪图后，再继续处理其余裁剪图

为什么使用 LLM 视觉能力，而不是启发式方法

空白检查启发式方法（例如“平均亮度 > 250 → 空白”）只能捕捉最明显的问题。Agent 自身的视觉能力可以进行语义验证：“这张裁剪图显示的是柱状图”，而“用户要求的是数据表格”。即使裁剪图包含了错误部分中的有效内容，这也能发现页面错误。

---

快速参考 — 构建模块

#	模块	模式	参考
0	前置检查（必需）	构建前渲染页面 + 诊断解析	上文
1	解析 + 保存	单个文档 → JSON + Markdown	下文
2	解析 + 提取 + 保存	单个文档 → 结构化数据	下文
3	批处理（同步）	ThreadPoolExecutor + tqdm	batch-processing.md
4	批处理（异步）	AsyncLandingAIADE + aiolimiter	batch-processing.md
5	大文件	解析任务 API（异步轮询）	batch-processing.md
6	分类 → 提取	枚举分类 + 架构路由	下文
7	结果 → DataFrame	将嵌套提取结果展平为表格	database-integration.md
8	结果 → CSV	摘要 + 按文档导出	database-integration.md
9	结果 → Snowflake	4 张规范化表 + COPY 上传	database-integration.md
10	分块 → RAG CSV	19 列分块数据集	rag-pipelines.md
11	分块 → ChromaDB	OpenAI 嵌入 + 持久化存储	rag-pipelines.md
12	分块 → FAISS	LangChain Documents + FAISS 索引	rag-pipelines.md
13	RAG 查询	带来源的 RetrievalQA 链	rag-pipelines.md
14	分块图像	将页面中的分块裁剪为 PNG	visualization.md
15	定位叠加层	在页面上显示彩色边界框	visualization.md
16	单词级定位	OCR + 模糊匹配高亮	visualization.md
17	章节提取	从 Markdown 中提取命名章节（正则表达式或 ADE 提取）	下文
18	嵌入计算	本地（FastEmbed）或 API（OpenAI），包含最佳实践	rag-pipelines.md
19	层次分块	将 ADE 分块归组为语义单元以生成嵌入	rag-pipelines.md
20	多粒度 RAG	分块、层次或文档级策略	rag-pipelines.md
21	表格拼接	仅解析或解析 + 提取，合并跨页表格	table-stitching.md
—	架构目录	可直接使用的 Pydantic 模型	schema-catalog.md

---

核心工作流：解析 + 提取 + 保存

ADE 的基本两步模式。其他所有工作流都基于此构建。

import io
from pathlib import Path
from typing import Any, Tuple, Type

from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema


def parse_extract_save(
    doc_path: Path,
    client: LandingAIADE,
    schema_cls: Type[Any],
    output_dir: str = "./ade_results",
) -> Tuple[Any, Any]:
    """Parse a document, extract structured data, save both
    as JSON via save_to. Returns (parse_result, extract_result)."""
    # Step 1 — Parse (auto-saves {stem}_parse_output.json)
    parse_result = client.parse(
        document=doc_path, save_to=output_dir,
    )

    # Step 2 — Extract (auto-saves {stem}_extract_output.json)
    extract_result = client.extract(
        schema=pydantic_to_json_schema(schema_cls),
        markdown=io.BytesIO(
            parse_result.markdown.encode("utf-8")
        ),
        save_to=output_dir,
    )
    return parse_result, extract_result

save_to 参数： 可用于 parse()、extract() 和 split()。 如果文件夹不存在则创建，并写入 {input_filename}_{method}_output.json。 这是客户端侧的便捷功能 — API 调用后会将完整响应保存在本地。

仅解析（不提取）

def parse_and_save(
    doc_path: Path,
    client: LandingAIADE,
    output_dir: str = "./ade_results",
) -> Any:
    return client.parse(
        document=doc_path, save_to=output_dir,
    )

架构： 请参阅 schema-catalog.md，获取 可直接使用的 Pydantic 模型（发票、水电费账单、银行对账单、 工资单、食品标签、CME 证书、文档分类器）。 有关架构设计规则，请参阅 document-extraction 技能。

---

先分类再提取

先分类，再应用相应架构，以处理混合类型的文档。有两种方法：

方法 1：分类提取（适用于任意文档组合）

from typing import Literal
from pydantic import BaseModel, Field


class DocType(BaseModel):
    type: Literal[
        "invoice", "bank_statement", "pay_stub",
        "utility_bill",
    ] = Field(description="The type of the document.")


# Map types to schemas (from schema-catalog.md)
SCHEMA_MAP: dict[str, type] = {
    "invoice": InvoiceSchema,
    "bank_statement": BankStatementSchema,
    "pay_stub": PayStubSchema,
    "utility_bill": UtilityBillSchema,
}


def classify_and_extract(
    doc_path: Path,
    client: LandingAIADE,
) -> dict:
    """Classify a document then extract with the matching
    schema."""
    pr = client.parse(document=doc_path)

    # Classify using first page
    cls = client.extract(
        schema=pydantic_to_json_schema(DocType),
        markdown=pr.markdown,
    )
    doc_type: str = cls.extraction["type"]

    # Extract with type-specific schema
    schema_cls = SCHEMA_MAP[doc_type]
    er = client.extract(
        schema=pydantic_to_json_schema(schema_cls),
        markdown=pr.markdown,
    )
    return {
        "type": doc_type,
        "extraction": er.extraction,
        "parse_result": pr,
        "extract_result": er,
    }

方法 2：Split API（多文档 PDF）

当单个 PDF 中包含多种文档类型（例如，包含发票和收据的文件包）时，请先使用 Split API：

def split_classify_extract(
    pdf_path: Path,
    client: LandingAIADE,
    split_classes: list[dict],
) -> list[dict]:
    """Split a multi-doc PDF, classify each split, extract."""
    pr = client.parse(document=pdf_path, split="page")

    # Split into sub-documents
    split_result = client.split(
        markdown=pr.markdown,
        split_class=split_classes,
    )

    results = []
    for split_doc in split_result.splits:
        # Classify
        cls = client.extract(
            schema=pydantic_to_json_schema(DocType),
            markdown=split_doc.markdowns[0],
        )
        doc_type = cls.extraction["type"]

        # Extract
        schema_cls = SCHEMA_MAP[doc_type]
        er = client.extract(
            schema=pydantic_to_json_schema(schema_cls),
            markdown=split_doc.markdowns[0],
        )
        results.append({
            "type": doc_type,
            "extraction": er.extraction,
            "pages": split_doc.pages,
        })
    return results

Split API 参数： 使用 split_class（包含 name、description、identifier 键的字典列表）。 完整的 Split API 参考请参阅 document-extraction 技能。

何时使用 Split API，何时使用分类：

• Split API：一个 PDF 包含多个独立文档
• 分类提取：每个文件都是一个文档，但文档类型各不相同

---

章节提取

从已解析文档的 Markdown 中提取指定章节（例如“引言”“摘要”）。有两种方法——根据文档的多样性以及额外的 API 成本是否值得来选择。

**决策：**如果诊断性解析（工具 2）显示所有文档都使用一致的 ATX 标题（## Introduction、## 2. Methods），请使用方法 A。如果发现任何纯文本编号标题（1. Introduction）或文档间格式有所不同，请完全跳过方法 A，直接使用方法 B。

方法	适用场景
A — 正则表达式	结构统一、格式规范的文档（学术论文、报告）。免费、快速。
B — ADE 提取	格式混杂或难以预测的文档（幻灯片、扫描件、模板各异的文档）。每份文档需额外消耗一次提取额度。

方法 A — 基于规则的正则表达式（免费、快速、脆弱）

即使在同一文档中，ADE 也可能将标题输出为 ATX Markdown（## 2. Related Work）或纯文本（1. Introduction）。请同时处理这两种模式：

import re

def find_section(markdown: str, name: str) -> str | None:
    """Extract a named section from ADE markdown, handling both ATX
    headers (## Introduction) and plain-text numbered headings
    (1. Introduction) which ADE may emit inconsistently."""

    # Pattern 1: ATX header  (# Introduction, ## 1. Introduction …)
    m = re.search(
        r"^(#{1,6})\s+(?:\d+\.?\s+)?" + re.escape(name) + r"\b.*$",
        markdown, re.IGNORECASE | re.MULTILINE,
    )
    if m:
        level = len(m.group(1))
        end = re.search(r"^#{1," + str(level) + r"}\s",
                        markdown[m.end():], re.MULTILINE)
        end_pos = m.end() + (end.start() if end else len(markdown[m.end():]))
        return markdown[m.start():m.end() + end_pos].strip()

    # Pattern 2: plain-text numbered heading  (1. Introduction)
    m2 = re.search(r"^(?:\d+\.?\s+)?" + re.escape(name) + r"\s*$",
                   markdown, re.IGNORECASE | re.MULTILINE)
    if m2:
        end2 = re.search(
            r"^#{1,6}\s|^(?:\d+\.?\s+)[A-Z][a-zA-Z ]{3,}\s*$",
            markdown[m2.end():], re.MULTILINE,
        )
        end_pos = m2.end() + (end2.start() if end2 else len(markdown[m2.end():]))
        return markdown[m2.start():end_pos].strip()
    return None

方法 B — ADE 提取（稳健，可处理多样的文档）

使用 ADE 自带的提取功能，通过语义定位章节——无需正则表达式。 即使格式不一致，LLM 也能理解章节含义：

from pydantic import BaseModel, Field
from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema
from pathlib import Path


class PaperSections(BaseModel):
    abstract: str = Field(
        description="The abstract section, plain text only, "
                    "no markdown formatting or anchor tags."
    )
    introduction: str = Field(
        description="The introduction section, plain text only, "
                    "no markdown formatting or anchor tags."
    )


client = LandingAIADE()
pr = client.parse(document=Path("paper.pdf"))
er = client.extract(
    schema=pydantic_to_json_schema(PaperSections),
    markdown=pr.markdown,
)
intro_text = er.extraction["introduction"]

**成本说明：**每次 extract() 调用都会在 parse() 的基础上额外消耗额度。对于文档类型统一的高吞吐量管道，方法 A 可以避免这项成本。对于多样或难以预测的文档，准确性的提升值得额外消耗额度。

---

多页表格拼接 {#table-stitching}

当表格跨越多页时，ADE 可能会将其按页输出为多个独立的表格块——也可能将某些页面输出为纯文本，而不是表格块。这种不一致可能出现在任何页面，而不仅仅是最后一页。

有三种方法可以处理这种情况，各自在成本、准确性和脆弱性方面有所取舍：

方法	ADE 调用次数	可处理非表格块	脆弱性
A — Parse + Extract	2	✓ LLM 读取完整 Markdown	低——无需自定义解析
B — HTML 表格解析	1	✓ 可使用备用正则表达式	高——要求行结构一致
C — pandas read_html	1	✗ 会漏掉非表格块	中

选择指南：

• 当准确性至关重要、成本次之时，使用方法 A
• 当行结构高度统一、文档结构可预测，且节省成本值得承担基于正则表达式解析的脆弱性时，使用方法 B
• 对于快速原型开发，或可以接受漏掉部分行的情况，使用方法 C

表格拼接的预检补充项

选择方法之前，运行诊断性解析（工具 2）并检查：

检查项	方法	原因
每页的块类型	按页统计 type == "table" 与 "text" 的数量	任何页面的块类型都可能不一致
列数一致性	比较各表格块的列数	列数不一致可能表示这些表格并非同一张表
是否存在表头行	检查每个表格块的首行	检测表格和筛选行时需要
非目标表格	查找列数相同的汇总表或元数据表	必须与目标表格区分开
行结构一致性	比较各页的行结构	一致性低会使方法 B 更脆弱

领域专用语义检查

拼接后，添加利用领域知识的验证检查：

• **金融：**余额递增，列合计 = 各行之和
• **库存：**各行之间的数量守恒
• **时间序列：**按时间顺序排列，序列无间隔
• **科学：**单位一致，ID 单调递增

这些检查既可用于验证（确认正确性），也可用于消歧（解决解析输出中的结构歧义）。

所有三种方法的完整代码及可复用模式： 请参阅 table-stitching.md。

---

批量处理

根据规模可选择两种模式。两者都包含逐文档错误处理。

快速：ThreadPoolExecutor（同步）

from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm


def batch_process(
    files: list[Path],
    schema_cls: type,
    max_workers: int = 4,
) -> list[tuple[Path, Any, Any]]:
    client = LandingAIADE()
    results: list[tuple[Path, Any, Any]] = []
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(
                parse_extract_save, fp, client, schema_cls
            ): fp
            for fp in files
        }
        for fut in tqdm(
            as_completed(futures), total=len(futures)
        ):
            fp = futures[fut]
            try:
                results.append((fp, *fut.result()))
            except Exception as e:
                print(f"FAILED {fp.name}: {e}")
    return results

可扩展：AsyncLandingAIADE（异步）

import asyncio
from aiolimiter import AsyncLimiter
from landingai_ade import AsyncLandingAIADE


async def batch_parse_async(
    files: list[Path],
    rate_limit: int = 30,
) -> list[dict]:
    client = AsyncLandingAIADE()
    limiter = AsyncLimiter(rate_limit, 60)

    async def _process(fp: Path) -> dict | None:
        try:
            async with limiter:
                return {
                    "path": fp,
                    "result": await client.parse(document=fp),
                }
        except Exception as e:
            print(f"FAILED {fp.name}: {e}")
            return None

    raw = await asyncio.gather(*[_process(fp) for fp in files])
    return [r for r in raw if r]

完整代码，包含输出目录组织、CSV 导出和分块图像保存： 请参阅 batch-processing.md。

---

将结果转换为 DataFrame 和 CSV

将嵌套的 ADE 提取结果展平为 4 个规范化表：

import uuid
from datetime import datetime, timezone


def rows_from_doc(
    file_path: str,
    parse_result: Any,
    extract_result: Any,
    run_id: str = "",
) -> tuple[dict, list[dict], list[dict], dict]:
    """Returns (main_row, line_rows, chunk_rows, md_record).

    - main_row: flattened top-level fields (nested__field)
    - line_rows: one per list item (line items, transactions)
    - chunk_rows: one per parsed chunk with bounding boxes
    - md_record: full markdown for traceability
    """
    doc_uuid = str(uuid.uuid4())
    f = extract_result.extraction

    # Flatten top-level fields
    main_row = {"doc_uuid": doc_uuid, "document_name": Path(file_path).name}
    for k, v in f.items():
        if isinstance(v, dict):
            for sk, sv in v.items():
                main_row[f"{k}__{sk}"] = sv
        elif not isinstance(v, list):
            main_row[k] = v

    # Extract list fields as line rows
    line_rows = [
        {"doc_uuid": doc_uuid, "list_field": k, "line_index": i, **item}
        for k, v in f.items() if isinstance(v, list)
        for i, item in enumerate(v) if isinstance(item, dict)
    ]

    # Chunk rows from parse result
    chunk_rows = [
        {
            "doc_uuid": doc_uuid,
            "chunk_id": getattr(ch, "id", None),
            "chunk_type": getattr(ch, "type", None),
            "page": ch.grounding.page if hasattr(ch, "grounding") else None,
        }
        for ch in (parse_result.chunks or [])
    ]

    md_record = {
        "doc_uuid": doc_uuid,
        "markdown": parse_result.markdown,
    }
    return main_row, line_rows, chunk_rows, md_record

完整代码，包含 Snowflake 上传、UUID 可追溯性和边界框列： 请参阅 database-integration.md。

---

RAG 准备

将解析后的文档快速构建为可查询的 RAG 系统。提供两种嵌入选项：本地（免费、离线）或 API（质量更高）。

选项 A — 使用 FastEmbed 进行本地嵌入（免费）

import re
from fastembed import TextEmbedding


def ade_to_embeddings_local(
    parse_results: list[dict],
    model: str = "BAAI/bge-small-en-v1.5",
) -> list[dict]:
    """Embed ADE chunks locally. Returns list of dicts with
    text, vector, and grounding metadata."""
    embedder = TextEmbedding(model_name=model)
    items: list[dict] = []
    for pr in parse_results:
        for ch in (pr["parse_result"].chunks or []):
            text = re.sub(
                r"<a id='[^']*'>\s*</a>", "", ch.markdown,
            ).strip()
            if not text:
                continue
            items.append({
                "text": text,
                "source": pr["name"],
                "page": ch.grounding.page,
                "box": {
                    "l": ch.grounding.box.left,
                    "t": ch.grounding.box.top,
                    "r": ch.grounding.box.right,
                    "b": ch.grounding.box.bottom,
                },
            })
    vecs = list(embedder.embed([i["text"] for i in items]))
    for item, vec in zip(items, vecs):
        item["vector"] = vec.tolist()
    return items

选项 B — 使用 OpenAI 进行 API 嵌入

from langchain.docstore.document import Document
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings


def ade_to_rag(
    parse_results: list[dict],
    embedding_model: str = "text-embedding-3-small",
) -> FAISS:
    """Convert ADE parse results to a FAISS vector store.

    Args:
        parse_results: list of {"name": str, "parse_result": ParseResponse}
    """
    docs = [
        Document(
            page_content=ch.markdown,
            metadata={
                "source": item["name"],
                "chunk_type": getattr(ch, "type", ""),
                "page": ch.grounding.page if hasattr(ch, "grounding") else -1,
            },
        )
        for item in parse_results
        for ch in (item["parse_result"].chunks or [])
        if ch.markdown.strip()
    ]
    return FAISS.from_documents(
        docs, OpenAIEmbeddings(model=embedding_model)
    )

完整代码，包含嵌入最佳实践、分层分块、多粒度策略、ChromaDB、LangChain RetrievalQA 和 CSV 导出： 请参阅 rag-pipelines.md。

rag-pipelines.md 中的高级 RAG 模式：

• 嵌入计算（区块 18–19）——在本地嵌入（FastEmbed，免费）和 API 嵌入（OpenAI，质量更高）之间进行选择，包括批次大小和速率限制
• 分层分块（区块 20）——在多个粒度级别（分块、章节、文档）进行嵌入，以支持混合检索
• 多粒度 RAG（区块 21）——结合分块级精度和文档级上下文，根据查询范围将其路由到合适的嵌入级别

---

可视化

用于在解析后的页面上叠加边界框的快速代码片段：

from PIL import Image, ImageDraw
import pymupdf

CHUNK_COLORS = {
    "text": (40, 167, 69),
    "table": (0, 123, 255),
    "figure": (255, 0, 255),
    "marginalia": (111, 66, 193),
}

def annotate_page(
    img: Image.Image, chunks: list, page: int,
) -> Image.Image:
    annotated = img.copy()
    draw = ImageDraw.Draw(annotated)
    w, h = img.size
    for ch in chunks:
        if not hasattr(ch, "grounding") or ch.grounding.page != page:
            continue
        box = ch.grounding.box
        color = CHUNK_COLORS.get(getattr(ch, "type", ""), (200, 200, 200))
        draw.rectangle(
            [int(box.left * w), int(box.top * h),
             int(box.right * w), int(box.bottom * h)],
            outline=color, width=3,
        )
    return annotated

完整代码，包含分块图像裁剪、仅提取内容的叠加层和词级 OCR 对齐： 请参阅 visualization.md。

---

Streamlit UI 模式

用于交互式文档处理的快速 Streamlit 应用：

import streamlit as st
from pathlib import Path
from landingai_ade import LandingAIADE
from landingai_ade.lib import pydantic_to_json_schema

st.title("Document Processor")

uploaded = st.file_uploader(
    "Upload document", type=["pdf", "png", "jpg"]
)
if uploaded:
    # Save temp file
    tmp = Path(f"/tmp/{uploaded.name}")
    tmp.write_bytes(uploaded.read())

    client = LandingAIADE()

    with st.spinner("Parsing..."):
        pr = client.parse(document=tmp)

    st.subheader("Markdown Preview")
    st.markdown(pr.markdown[:2000])

    st.subheader("Chunks")
    for ch in pr.chunks:
        with st.expander(
            f"{ch.type} (page {ch.grounding.page})"
        ):
            st.text(ch.markdown[:500])

完整的 Streamlit 应用，包含批量上传、提取结果显示和可视化选项卡：根据 batch-processing.md 和 visualization.md 中的模式进行调整。

---

依赖项指南

工作流	安装
核心（解析 + 提取）	pip install landingai-ade
批量同步	pip install landingai-ade tqdm
批量异步	pip install landingai-ade aiolimiter
DataFrame / CSV	pip install landingai-ade pandas
Snowflake	pip install landingai-ade pandas snowflake-connector-python[pandas]
RAG（本地嵌入）	pip install landingai-ade fastembed
RAG（ChromaDB）	pip install landingai-ade chromadb openai
RAG（FAISS + LangChain）	pip install landingai-ade langchain langchain-openai langchain-community faiss-cpu
可视化	pip install landingai-ade Pillow pymupdf
词级对齐	pip install landingai-ade Pillow pymupdf pytesseract fuzzywuzzy + tesseract 二进制文件
Streamlit UI	pip install landingai-ade streamlit
架构转换	from landingai_ade.lib import pydantic_to_json_schema（已包含在 landingai-ade 中）

---

参考文件

针对特定工作流构建时，请阅读以下文件以了解完整实现：

• schema-catalog.md — 可直接使用的 Pydantic 架构，适用于发票、水电费账单、银行对账单、工资单、食品标签、CME 证书和文档分类
• batch-processing.md — ThreadPoolExecutor、AsyncLandingAIADE 和 Parse Jobs API 模式，包含完整的错误处理
• rag-pipelines.md — 将分块转换为 CSV、ChromaDB 导入、FAISS + LangChain，以及 RAG 查询链
• database-integration.md — DataFrame 规范化、Snowflake 上传和 CSV 导出模式
• visualization.md — 分块图像裁剪、边界框叠加和词级 OCR 定位
• table-stitching.md — Parse+Extract（稳健）、HTML 解析（脆弱）和 pandas 方法，用于将跨页表格合并为单一输出