抽出戦略を Python トランスフォームにデプロイする

AIP Document Intelligence で抽出戦略を検証した後、Python トランスフォームとしてデプロイし、メディアセット内のすべてのドキュメントのすべてのページに対してバッチ抽出を実行できます。デプロイされたテンプレートは、AIP Document Intelligence の対応する設定と同じ結果を生成します。

デプロイされたテンプレートを使用する

Code Repositories でテンプレートを作成したら、次の手順を実行します。

1. src/myproject/document_extraction/my_extraction.py ファイルの @transform.using デコレーターで出力データセットを指定します。
2. ビルドをトリガーします。

このテンプレートは、最適なパフォーマンスを実現するために軽量トランスフォームを使用します。レガシーバージョンでは Spark ベースのトランスフォームを使用していましたが、Spark のオーバーヘッドにより処理が大幅に遅くなります。まだ移行していない場合は、軽量トランスフォームへの移行を推奨します。

インクリメンタル（差分処理）

デフォルトでは、ドキュメント抽出トランスフォームは非インクリメンタルであり、実行のたびにすべてのドキュメントを処理します。@incremental(...) デコレーターの行のコメントを解除すると、トランスフォームをインクリメンタルに実行するよう設定できます。インクリメンタルトランスフォームでは、入力メディアセットに新しいドキュメントが追加されたときにトランスフォームを再実行すると、新しいドキュメントのみを処理し、結果を出力データセットに追加します。

プロンプトをカスタマイズする

生成 AI の設定では、テンプレートは AIP Document Intelligence で指定したプロンプトを引き継ぎます。プロンプトは src/myproject/document_extraction/prompts.py で確認できます。

テンプレート内でプロンプトを直接編集することは推奨しません。Document Intelligence の結果とバッチジョブの結果に不一致が生じるためです。代わりに、Document Intelligence でプロンプトを調整し、そこで結果を確認してから、再デプロイして新しいテンプレートを作成します。

レイアウト認識型抽出の設定では、ユーザープロンプトにレイアウト構造情報を保持する特別な JSON スキーマが含まれているため、ユーザープロンプトを固定したままにする必要があります。このプロンプトを変更すると、抽出の成功率が大幅に低下します。

カスタム画像前処理

回転したテキストコンテンツなど、抽出前に画像変換が必要なドキュメントについては、次の手順を実行してください。

1. 画像変換を適用するための独立したトランスフォームパイプラインを作成します。
2. 処理結果を新しいメディアセットに保存します。
3. 処理済みのメディアセットに対して Document Intelligence を使用し、抽出を行います。

メディアアイテムの一部に対して実行する

レイアウト認識型の生成 AI 設定では、ビジョン LLM は特定のスキーマに準拠した有効な JSON を生成する必要があります。レスポンスが無効な JSON であるか、スキーマに準拠していない場合、抽出は ERROR_RESPONSE_JSON_PARSING エラーで失敗します。

実際には、最上位のモデルでも抽出の約5%が失敗する可能性があります。失敗した行についても、レイアウトモデルのみからの抽出結果を含む有効な layoutInfo が得られます。

失敗した行に対して抽出を再実行するには、次の手順を実行します。

1. filter_on_media_items 引数にメディアアイテムIDのリストを指定して、特定のアイテムのみを処理します。
2. これらの行が完了済みと判定されずに再処理されるように、@incremental デコレーターを削除します。

実行時のパフォーマンスを改善する

THREAD_NUMBER パラメーターは同時実行スレッド数を制御します。各スレッドは、一度に1ページのドキュメントからデータを抽出します。値を大きくすると、ジョブがより速く完了します。

ログを探す

ビルドログを表示するには、ビルド詳細ページでテレメトリを選択します。ドキュメント抽出ログに絞り込むには、メッセージ列を aip_workflows で始まる値に絞り込みます。

行単位とドキュメント単位のチャンク分割

抽出出力には、ページごとに1行が含まれます。デフォルトでは、DocumentChunker.create_chunks_per_document は同じドキュメントのすべてのページを1つの Markdown 文字列に結合してから、チャンクに分割します。

ページを結合せずに各行を個別にチャンク分割するには、代わりに DocumentChunker.create_chunks_per_row を使用します。

Copied!
1
2
3
4
5
6
7
8
9
10
chunking_result = chunker.create_chunks_per_row(
    extraction_df,
    chunk_mode="markdown",       # プレーンテキストの場合は "recursive"、Markdown テキストの場合は "markdown"
    content_column="extractionResult",
    id_column="media_item_rid",  # chunk_id の接頭辞として使用します
    chunk_size=8192,
    chunk_overlap=0,
    thread_number=20,
    strip_markdown=False,        # チャンク分割前に ```markdown と ``` のラッパーを削除するには True に設定します
)

チャンク分割せずに埋め込みを作成する

埋め込みモデルにはコンテキストの制限があるため、埋め込みを作成する前にチャンク分割することを推奨します。パイプラインの構造を維持したままチャンク分割をスキップするには、create_chunks_per_row の chunk_size に sys.maxsize などの非常に大きな値を設定します。

テンプレートの例

以下の例は、抽出設定ごとに生成されるトランスフォームコードを示しています。これらは参照用です。ドキュメント抽出のトランスフォームコードを手動で記述するのではなく、Document Intelligence のデプロイツールを使用してトランスフォームを作成してください。

従来型抽出：生テキスト

ドキュメントのメタデータを読み取ってテキストを抽出します。電子的に生成された PDF でのみ利用可能です。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import polars as pl
from concurrent.futures import ThreadPoolExecutor
from transforms.api import Output, incremental, transform
from transforms.mediasets import MediaSetInput
from transforms.mediasets.utils._constants import MEDIA_ITEM_RID, MEDIA_REFERENCE, PATH


THREAD_NUMBER = 20


# @incremental(v2_semantics=True)  # インクリメンタル処理が必要な場合は、この行のコメントを解除します
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
)
def extract(media_input, output):
    """
    生テキスト抽出を使用して PDF ドキュメントからコンテンツを抽出します
    """
    media_refs = pl.from_pandas(
        media_input.list_media_items_by_path_with_media_reference().pandas(),
        schema_overrides={MEDIA_ITEM_RID: pl.String, MEDIA_REFERENCE: pl.String, PATH: pl.String},
    )

    def process_batch(batch_df: pl.DataFrame) -> pl.DataFrame:
        def create_page_tasks(row):
            media_item_rid = row[MEDIA_ITEM_RID]
            metadata = media_input.get_media_item_metadata(media_item_rid).document
            if metadata is None:
                raise ValueError(f"Media item {media_item_rid} is not a document")
            if metadata.pages is None:
                raise ValueError(f"Media item {media_item_rid} has no page count")
            return [(row, page_num) for page_num in range(metadata.pages)]

        def process_single_page(task):
            row, page_num = task
            media_item_rid = row[MEDIA_ITEM_RID]
            media_reference = row[MEDIA_REFERENCE]
            extraction_result = media_input.transform_document_to_text_raw(
                media_item_rid, page_num
            ).read().decode("utf-8")
            return {
                "media_item_rid": media_item_rid,
                "media_reference": media_reference,
                "page_num": page_num,
                "extraction_result": extraction_result
            }

        all_tasks = []
        for row in batch_df.iter_rows(named=True):
            all_tasks.extend(create_page_tasks(row))

        with ThreadPoolExecutor(max_workers=THREAD_NUMBER) as executor:
            results = list(executor.map(process_single_page, all_tasks))

        return pl.DataFrame(results)

    extracted_data = media_refs.lazy().map_batches(
        process_batch,
        schema={
            "media_item_rid": pl.String,
            "media_reference": pl.String,
            "page_num": pl.Int64,
            "extraction_result": pl.String,
        },
        streamable=True,
    )

    output.write_dataframe(extracted_data)

従来型抽出: OCR

従来の光学文字認識（OCR）を使用して、レイアウト情報を保持せずにテキストを抽出します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import polars as pl
from concurrent.futures import ThreadPoolExecutor
from transforms.api import Output, incremental, transform
from transforms.mediasets import MediaSetInput
from transforms.mediasets.utils._constants import MEDIA_ITEM_RID, MEDIA_REFERENCE, PATH


THREAD_NUMBER = 20


# @incremental(v2_semantics=True)  # インクリメンタル処理が必要な場合は、この行のコメントを解除します
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
)
def extract(media_input, output):
    """
    OCR テキスト抽出を使用して PDF ドキュメントからコンテンツを抽出します
    """
    media_refs = pl.from_pandas(
        media_input.list_media_items_by_path_with_media_reference().pandas(),
        schema_overrides={MEDIA_ITEM_RID: pl.String, MEDIA_REFERENCE: pl.String, PATH: pl.String},
    )

    def process_batch(batch_df: pl.DataFrame) -> pl.DataFrame:
        def create_page_tasks(row):
            media_item_rid = row[MEDIA_ITEM_RID]
            metadata = media_input.get_media_item_metadata(media_item_rid).document
            if metadata is None:
                raise ValueError(f"Media item {media_item_rid} is not a document")
            if metadata.pages is None:
                raise ValueError(f"Media item {media_item_rid} has no page count")
            return [(row, page_num) for page_num in range(metadata.pages)]

        def process_single_page(task):
            row, page_num = task
            media_item_rid = row[MEDIA_ITEM_RID]
            media_reference = row[MEDIA_REFERENCE]
            extraction_result = media_input.transform_document_to_text_ocr_output_text(
                media_item_rid, page_num
            ).read().decode("utf-8")
            return {
                "media_item_rid": media_item_rid,
                "media_reference": media_reference,
                "page_num": page_num,
                "extraction_result": extraction_result
            }

        all_tasks = []
        for row in batch_df.iter_rows(named=True):
            all_tasks.extend(create_page_tasks(row))

        with ThreadPoolExecutor(max_workers=THREAD_NUMBER) as executor:
            results = list(executor.map(process_single_page, all_tasks))

        return pl.DataFrame(results)

    extracted_data = media_refs.lazy().map_batches(
        process_batch,
        schema={
            "media_item_rid": pl.String,
            "media_reference": pl.String,
            "page_num": pl.Int64,
            "extraction_result": pl.String,
        },
        streamable=True,
    )

    output.write_dataframe(extracted_data)

従来型抽出：レイアウト認識型 OCR

レイアウト認識型 OCR を使用して、ドキュメントの構造と境界ボックスを保持します。以下の例では、V2 レスポンスを JSON として extraction_result に保存し、ページごとに1行を生成します。各レスポンスには、そのページのブロックを含む pages 配列があります。format を TEXT に設定すると、各ブロックの text は、type フィールドと text フィールドを含むタグ付きの値になります。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
import json
import polars as pl
from concurrent.futures import ThreadPoolExecutor
from transforms.api import Output, incremental, transform
from transforms.mediasets import MediaSetInput
from transforms.mediasets.utils._constants import MEDIA_ITEM_RID, MEDIA_REFERENCE, PATH


THREAD_NUMBER = 20


# @incremental(v2_semantics=True)  # インクリメンタル処理が必要な場合は、この行のコメントを解除します
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
)
def extract(media_input, output):
    """
    レイアウトを考慮した OCR 抽出を使用して PDF ドキュメントからコンテンツを抽出します
    """
    media_refs = pl.from_pandas(
        media_input.list_media_items_by_path_with_media_reference().pandas(),
        schema_overrides={MEDIA_ITEM_RID: pl.String, MEDIA_REFERENCE: pl.String, PATH: pl.String},
    )

    def process_batch(batch_df: pl.DataFrame) -> pl.DataFrame:
        def create_page_tasks(row):
            media_item_rid = row[MEDIA_ITEM_RID]
            metadata = media_input.get_media_item_metadata(media_item_rid).document
            if metadata is None:
                raise ValueError(f"Media item {media_item_rid} is not a document")
            if metadata.pages is None:
                raise ValueError(f"Media item {media_item_rid} has no page count")
            return [(row, page_num) for page_num in range(metadata.pages)]

        def process_single_page(task):
            row, page_num = task
            media_item_rid = row[MEDIA_ITEM_RID]
            media_reference = row[MEDIA_REFERENCE]
            extraction_result = media_input.transform_media_item(media_item_rid, str(page_num), {
                "type": "documentToText",
                "documentToText": {
                    "operation": {
                        "type": "extractLayoutAwareTextV2",
                        "extractLayoutAwareTextV2": {
                            "pageRange": {
                                "startPageInclusive": page_num,
                                "endPageExclusive": page_num + 1
                            },
                            "config": {
                                "mode": "SCAN",
                                "format": "TEXT",
                                "languages": [{"type": "language", "language": "ENG"}]
                            }
                        }
                    }
                }
            })
            extraction_result = json.dumps(extraction_result.json())
            return {
                "media_item_rid": media_item_rid,
                "media_reference": media_reference,
                "page_num": page_num,
                "extraction_result": extraction_result
            }

        all_tasks = []
        for row in batch_df.iter_rows(named=True):
            all_tasks.extend(create_page_tasks(row))

        with ThreadPoolExecutor(max_workers=THREAD_NUMBER) as executor:
            results = list(executor.map(process_single_page, all_tasks))

        return pl.DataFrame(results)

    extracted_data = media_refs.lazy().map_batches(
        process_batch,
        schema={
            "media_item_rid": pl.String,
            "media_reference": pl.String,
            "page_num": pl.Int64,
            "extraction_result": pl.String,
        },
        streamable=True,
    )

    output.write_dataframe(extracted_data)

生成 AI 抽出：基本

ビジョン言語モデルを使用して、前処理なしでコンテンツを Markdown として抽出します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from transforms.api import Output, incremental, transform
from transforms.mediasets import MediaSetInput
from aip_workflows.document_intelligence.transforms import VisionLLMDocumentsExtractorInput
from .prompts import USER_PROMPT


THREAD_NUMBER = 20


# @incremental(v2_semantics=True, snapshot_inputs=["extractor"])  # インクリメンタル処理が必要な場合は、この行のコメントを解除します
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
    extractor=VisionLLMDocumentsExtractorInput(
        "ri.language-model-service..language-model.anthropic-claude-xxx-sonnet"
    ),
)
def extract(media_input, output, extractor):
    """
    PDF ドキュメントからコンテンツを Markdown として抽出します。
    """
    extracted_data = extractor.create_extraction(
        media_input, with_ocr=False, prompt=USER_PROMPT, thread_number=THREAD_NUMBER
    )
    output.write_dataframe(extracted_data)

生成 AI 抽出: OCR 前処理あり

OCR 前処理を組み合わせたビジョン言語モデルを使用して、複雑なドキュメントの抽出を改善します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from transforms.api import Output, incremental, transform
from transforms.mediasets import MediaSetInput
from aip_workflows.document_intelligence.transforms import VisionLLMDocumentsExtractorInput
from .prompts import USER_PROMPT


THREAD_NUMBER = 20


# @incremental(v2_semantics=True, snapshot_inputs=["extractor"])  # インクリメンタル処理が必要な場合は、この行のコメントを解除します
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
    extractor=VisionLLMDocumentsExtractorInput(
        "ri.language-model-service..language-model.anthropic-claude-xxx-sonnet"
    ),
)
def extract(media_input, output, extractor):
    """
    PDF ドキュメントからコンテンツを Markdown として抽出します。
    """
    extracted_data = extractor.create_extraction(
        media_input, with_ocr=True, prompt=USER_PROMPT, thread_number=THREAD_NUMBER
    )
    output.write_dataframe(extracted_data)

生成 AI 抽出：レイアウト認識型

レイアウト認識型 OCR 前処理を組み合わせたビジョン言語モデルを使用し、抽出したコンテンツとともにレイアウト情報を返します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from transforms.api import Output, incremental, transform
from transforms.mediasets import MediaSetInput
from aip_workflows.document_intelligence.transforms import VisionLLMLayoutDocumentsExtractorInput
from .prompts import SYSTEM_PROMPT


THREAD_NUMBER = 20


# @incremental(v2_semantics=True, snapshot_inputs=["extractor"])  # インクリメンタル処理が必要な場合は、この行のコメントを解除します
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
    extractor=VisionLLMLayoutDocumentsExtractorInput(
        "ri.language-model-service..language-model.anthropic-claude-xxx-sonnet"
    ),
)
def extract(media_input, output, extractor):
    """
    PDF ドキュメントからコンテンツを Markdown として抽出します。
    """
    extracted_data = extractor.create_extraction(
        media_input,
        include_layout_info="no_overlay",
        system_prompt=SYSTEM_PROMPT,
        thread_number=THREAD_NUMBER
    )
    output.write_dataframe(extracted_data)

生成 AI 抽出：テーブルの切り抜きを伴うレイアウト認識型

レイアウト認識型 OCR 前処理とテーブルの切り抜きを組み合わせたビジョン言語モデルを使用して、テーブル抽出の精度を向上させます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from transforms.api import Output, incremental, transform
from transforms.mediasets import MediaSetInput
from aip_workflows.document_intelligence.transforms import VisionLLMLayoutDocumentsExtractorInput
from .prompts import SYSTEM_PROMPT


THREAD_NUMBER = 20


# @incremental(v2_semantics=True, snapshot_inputs=["extractor"])  # インクリメンタル処理が必要な場合は、この行のコメントを解除します
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
    extractor=VisionLLMLayoutDocumentsExtractorInput(
        "ri.language-model-service..language-model.anthropic-claude-xxx-sonnet"
    ),
)
def extract(media_input, output, extractor):
    """
    PDF ドキュメントからコンテンツを Markdown として抽出します。
    """
    extracted_data = extractor.create_extraction(
        media_input,
        include_layout_info="crop_tables",
        system_prompt=SYSTEM_PROMPT,
        thread_number=THREAD_NUMBER
    )
    output.write_dataframe(extracted_data)

抽出したテキストをチャンク分割して埋め込みを生成する

埋め込みが不要な場合は、トランスフォームデコレーターから embedder を削除し、embedding_result の行を削除します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from transforms.api import Input, Output, incremental, transform
from aip_workflows.document_intelligence.transforms import DocumentChunker, DocumentEmbedderInput

THREAD_NUMBER = 20


# @incremental(v2_semantics=True, snapshot_inputs=["embedder"])  # インクリメンタル処理が必要な場合は、この行のコメントを解除します
@transform.using(
    extraction_input=Input("ri.foundry.main.dataset.abc"),  # 通常は抽出トランスフォームの出力データセットです
    output=Output("ri.foundry.main.dataset.xyz"),
    embedder=DocumentEmbedderInput("ri.language-model-service..language-model.text-embedding-3-large"),
)
def chunk_and_embed(extraction_input, output, embedder):
    extraction_df = extraction_input.polars(lazy=True)
    chunker = DocumentChunker()
    chunking_result = chunker.create_chunks_per_document(
        extraction_df,
        chunk_mode="markdown",  # 生テキストの場合は "recursive"、Markdown テキストの場合は "markdown"
        content_column="extractionResult",  # コンテンツの列名
        id_column="media_item_rid",  # ドキュメントの ID。同じドキュメントのコンテンツ（たとえば、異なるページのコンテンツ）を結合するために使用します
        page_column="page_num",  # ページ番号の列名
        chunk_size=8192,
        chunk_overlap=0,
        thread_number=THREAD_NUMBER,
        strip_markdown=True,  # True の場合、チャンク分割前にコンテンツから ```markdown の接頭辞と ``` の接尾辞を削除します
    )
    embedding_result = embedder.create_embeddings(
        chunking_result,
        content_column="chunk_content",
        thread_number=THREAD_NUMBER,
    )
    output.write_dataframe(embedding_result)