document-workflows

ドキュメントワークフロー — ADE パイプラインパターン

概要

このスキルは、LandingAI ADE のプリミティブ（parse、extract、split）を組み合わせて、本番環境に対応したドキュメント処理パイプラインを構築するための再利用可能な構成要素を提供します。document-extraction スキルを補完します。

項目	document-extraction	document-workflows
範囲	ADE SDK API：parse、extract、split、grounding	エンドツーエンドのパイプライン：バッチ、RAG、DB、分類と振り分け
使用場面	ADE の単一操作を呼び出す場合	複数の操作をワークフローに組み合わせる場合
コード	パラメーターを指定した SDK メソッド呼び出し	エラー処理と並列処理を含む関数全体
依存関係	landingai-ade のみ	＋ ワークフロー固有のライブラリ（pandas、chromadb など）

基本方針： ドキュメントの種類ではなく、ワークフローパターン（バッチ、RAG、DB への挿入）ごとに整理します。同じパターンは、請求書、公共料金の明細書、医療用フォームのいずれにも適用できます。

---

ステップ 0（必須）— パイプライン実装前のドキュメント調査 {#pre-flight}

このセッションですでに内部構造を調査したドキュメント以外を扱う場合は、パイプラインのコードを書く前に必ず実行してください。

ルール：サンプルドキュメントに対してツール 2（診断用パース）を実行する前に、セクション検出、見出し照合、テキスト検索のコードを書いてはいけません。 見出しの形式はドキュメントごとに異なり、タスクの説明やドキュメントの種類だけから推測できません。確実に知るには、ADE の実際の出力を確認するしかありません。

よくある意外な例：論文の「Introduction」という見出しが、1. Introduction（プレーンテキストで、# なし）、## Introduction、INTRODUCTION（すべて大文字）、または本文を含むテキストチャンク内に埋め込まれている場合があります。これを誤ると、一致するチャンクが 0 件のまま検出されない障害が発生し、原因の調査にドキュメント全体の再パースが必要になります。

コードを書く前に、代表的なサンプルドキュメント 1～3 件に対してツール 1（ビジュアルレンダリング）とツール 2（診断用パース）を実行してください。1 分もかからず、事前調査で防げたはずのデバッグを繰り返さずに済みます。

ツール 1 — ページのビジュアルレンダリング

1～2 ページを PNG としてレンダリングし、視覚的なコンテキストとして読み込みます。ADE のクレジットは消費しませんが、各 PNG はコンテキストトークンを消費します。レイアウトが曖昧な場合や、ドキュメントの出所が不明な場合（手書きか、スキャンか、フォームかなど）に使用します。

.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 が # 見出し ではなく、プレーンテキストの見出しとして出力する可能性があります）
• ドキュメントは手書きか、スキャンされたものか？ → PyMuPDF ではなく Tesseract OCR が必要です
• 1 段組みか、2 段組みか？
• ノイズはあるか：柱見出し、ページ番号、透かし、スタンプなど

ツール 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 extract を使用
ドキュメントによって見出しの形式が異なる（ある文書では # INTRO、別の文書では 1. Intro）	一部の文書で正規表現が機能しません → 堅牢性のため ADE extract を使用
すべての ch.markdown が <a id='...'></a> で始まる	文字列の照合や表示の前にアンカーを削除
2 段組み：同じページ上のチャンクで l=0.07 と l=0.50 がある	テキストは左段、次に右段の順です。セクションが両方の段にまたがる場合があります
ページ区切りでチャンクのテキストが単語の途中で切れている	セクションが複数ページにまたがっています。複数ページからチャンクを収集してください
t<0.08 または t>0.90 の位置に marginalia チャンクがある	柱見出し／ページ番号です → コンテンツの抽出対象から除外
ページ画像にスキャン画像または手書きの内容が見える	PyMuPDF のテキスト抽出は機能しません → Tesseract OCR を使用

ツール 3 — クロップ後の目視確認（バウンディングボックスを使うワークフローでは必須） {#post-crop-verification}

バウンディングボックスでクロップまたはオーバーレイを作成した後（図の抽出、チャンクのクロップ、表のセルの抽出、単語レベルのグラウンディングなど）、**少なくとも1つの出力 PNG を画像として読み取り、見えている内容を説明してください。**説明とユーザーの依頼内容を照らし合わせます。これにより、次の問題を検出できます。

• ページ違いのバグ — ADE のページ番号は 0 始まりです。1つずれると、隣接するページのまったく異なる内容をクロップしてしまいます
• 領域違いのバグ — 座標系の不一致により、空白や無関係なセクションをクロップしてしまいます

ルール：少なくとも1つの出力 PNG を目視で読み取り、その内容がユーザーの依頼と一致することを確認するまでは、クロップのワークフローを完了したと宣言しないでください。

確認手順

1. 最初のクロップを PNG として保存します（ワークフローですでに保存されています）
2. PNG ファイルを画像として読み取ります（PNG のパスに対して read_file ツールを使用します）
3. 見えている内容を説明します。どのような内容、表、図、テキストが表示されていますか？
4. ユーザーの依頼内容と照らし合わせます。
   • ユーザーが「Events テーブル」を依頼した場合 → クロップに Events テーブルが表示されていますか？
   • ユーザーが「Figure 3」を依頼した場合 → クロップにグラフや図が表示されていますか？
   • ユーザーが「Introduction セクション」を依頼した場合 → クロップに導入部分のテキストが表示されていますか？
5. 説明が一致しない場合 → 続行する前にページのインデックスとバウンディングボックスの座標を調べます
6. 最初のクロップを確認してから、残りのクロップに進みます

ヒューリスティックではなく LLM の視覚能力を使う理由

空白判定のヒューリスティック（例：「平均輝度が 250 超なら空白」）で検出できるのは、最も明白な失敗だけです。エージェント自身の視覚能力を使えば、「このクロップには棒グラフが表示されている」一方で「ユーザーが依頼したのはデータテーブルである」と意味に基づいて確認できます。これにより、クロップに別のセクションの有効な内容が含まれている場合でも、ページ違いのエラーを検出できます。

---

クイックリファレンス — 構成要素

#	構成要素	パターン	参照
0	事前確認（必須）	ページをレンダリングし、構築前に診断用の解析を実行	上記
1	解析 + 保存	単一ドキュメント → JSON + Markdown	以下
2	解析 + 抽出 + 保存	単一ドキュメント → 構造化データ	以下
3	バッチ（同期）	ThreadPoolExecutor + tqdm	batch-processing.md
4	バッチ（非同期）	AsyncLandingAIADE + aiolimiter	batch-processing.md
5	大容量ファイル	Parse Jobs API（非同期ポーリング）	batch-processing.md
6	分類 → 抽出	Enum 分類 + スキーマルーティング	以下
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 の基本となる 2 ステップのパターンです。ほかのすべてのワークフローはこのパターンを基にしています。

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,
    )

スキーマ： すぐに使える Pydantic モデル（請求書、公共料金の請求書、銀行取引明細書、給与明細、食品ラベル、CME 証明書、ドキュメント分類器）については、schema-catalog.md を参照してください。 スキーマ設計のルールについては、document-extraction スキルを参照してください。

---

分類してから抽出

最初に分類し、その後、適切なスキーマを適用することで、さまざまな種類のドキュメントを処理します。アプローチは 2 つあります。

アプローチ 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：分割 API（複数ドキュメントを含む PDF）

1 つの PDF に複数のドキュメント種別（例：請求書と領収書を含む書類一式）が含まれている場合は、まず分割 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

分割 API のパラメーター： name、description、identifier キーを持つ辞書のリストである split_class を使用します。 分割 API の詳細については、document-extraction スキルを参照してください。

分割と分類の使い分け：

• 分割 API：1 つの PDF に複数の独立したドキュメントが含まれている場合
• 分類抽出：各ファイルに含まれるドキュメントは 1 つですが、種別が異なる場合

---

セクション抽出

解析済みドキュメントの Markdown から、「Introduction」や「Abstract」などの名前付きセクションを抽出します。2 つの方法があり、ドキュメントの多様性と、追加の API コストに見合うかどうかに基づいて選択します。

判断基準： 診断用の解析（ツール 2）で、すべてのドキュメントに一貫して ATX 見出し（## Introduction、## 2. Methods）が使われていれば、アプローチ A を使用します。プレーンテキストの番号付き見出し（1. Introduction）や、ドキュメント間の書式のばらつきが見られる場合は、アプローチ A を完全にスキップし、アプローチ B に進みます。

アプローチ	使用する状況
A — 正規表現	均一で、構造が整ったドキュメント（学術論文、レポート）。無料で高速です。
B — ADE 抽出	書式が混在または予測困難なドキュメント（スライド、スキャンされた論文、さまざまなテンプレート）。ドキュメントごとに抽出クレジットが 1 件追加で必要です。

アプローチ 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 はページごとに分割された表チャンクとして出力することがあります。また、一部のページを表チャンクではなくプレーンテキストとして出力する場合もあります。この不整合は、最終ページに限らずどのページでも発生する可能性があります。

この問題には 3 つの方法があり、コスト、精度、脆弱性のトレードオフがそれぞれ異なります。

アプローチ	ADE 呼び出し回数	表以外のチャンクへの対応	脆弱性
A — 解析 + 抽出	2	✓ LLM が Markdown 全体を読み取ります	低 — 独自の解析は不要です
B — HTML テーブル解析	1	✓ フォールバック正規表現で対応	高 — 行構造が均一である必要があります
C — pandas read_html	1	✗ 表以外のチャンクを取りこぼします	中

判断ガイド：

• 精度を最優先し、コストは二次的な要素である場合は、アプローチ A を使用します
• 行の均一性が非常に高く、ドキュメント構造が予測可能で、コスト削減が正規表現ベースの解析の脆弱性に見合う場合は、アプローチ B を使用します
• 迅速なプロトタイピングや、一部の行を取りこぼしても問題ない場合は、アプローチ C を使用します

表の結合に関する事前確認項目

アプローチを選ぶ前に、診断用の解析（ツール 2）を実行して、以下を確認します。

確認項目	方法	理由
ページごとのチャンク種別	ページごとに type == "table" と "text" の数を数えます	どのページでも種別が不整合になる可能性があります
列数の一貫性	表チャンク間で列数を比較します	列数が一致しない場合、異なる表である可能性があります
ヘッダー行の有無	各表チャンクの最初の行を確認します	検出と行のフィルタリングに必要です
対象外の表	同じ列数の要約表やメタデータ表がないか確認します	対象の表と区別する必要があります
行の均一性	ページ間で行構造を比較します	均一性が低いと、アプローチ B は脆弱になります

ドメイン固有の意味チェック

結合後に、ドメイン知識を活用した検証チェックを追加します。

• 財務: 残高の推移、列の合計 = 行の合計
• 在庫: 行間で数量が保持されていること
• 時系列: 時系列順であること、シーケンスに欠落がないこと
• 科学: 単位に一貫性があること、ID が単調に増加すること

これらのチェックは、検証（正しさの確認）と 曖昧さの解消（解析結果における構造上の曖昧さの解消）の両方に役立ちます。

3 つすべてのアプローチの全コードと再利用可能なパターンについては、 table-stitching.md を参照してください。

---

バッチ処理

規模に応じて 2 つのパターンがあります。どちらもドキュメント単位のエラー処理を含みます。

クイック: 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 システムをすばやく構築します。埋め込みには 2 つの選択肢があります。ローカル（無料、オフライン）または 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 — 請求書、光熱費明細書、銀行取引明細書、給与明細書、食品ラベル、CME 証明書、文書分類に対応する、すぐに使える Pydantic スキーマ
• 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 の各種アプローチ