ドキュメントからテキストへのメディア変換

ドキュメントからテキストへの変換は、ドキュメントの内容を複数の形式で柔軟かつ高いパフォーマンスで抽出する方法です。段落、見出し、テーブルなど、ドキュメントのレイアウト構造も抽出できます。

AIP Document Intelligence は、ドキュメントからテキストへのメディア変換操作を2つ提供します。

extractTextV2：抽出したドキュメントのテキストを含む文字列のリストを返します。
extractLayoutAwareTextV2：ページをまたぐ、レイアウトを考慮したテキストブロックのリストを返します。

extractLayoutAwareContent および ocrOnPage 操作に代わる、上記の操作の使用を推奨します。

操作シグネチャの例を以下に示します。

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
{
  "type": "documentToText",
  "documentToText": {
    "operation": {
      "type": "{operation}", // {operation}: "extractTextV2" または "extractLayoutAwareTextV2"
      "{operation}": {
        "pageRange": {
          "startPageInclusive": 0,
          "endPageExclusive": 5
        },
        "config": {
          "mode": "SCAN", // または "ELECTRONIC" または "AUTO"
          "format": "TEXT", // または "MARKDOWN" または "HTML"
          "languages": [
            {
              "type": "language",
              "language": "KOR"
            }
          ]
        }
      }
    }
  }
}

操作シグネチャのパラメーター

config.mode：ドキュメントのページをどのように解釈するかを制御します。
- ELECTRONIC：すべてのページを電子 PDF ファイルとして扱い、埋め込みテキストまたは生のテキストを抽出します。ドキュメントに埋め込みテキストが含まれていることがわかっている場合は、ELECTRONIC を使用します。
- SCAN：すべてのページをスキャン画像として扱い、光学文字認識 (OCR) を実行します。ドキュメントがスキャン画像であることがわかっている場合は、SCAN を使用します。
- AUTO：OCR が必要かどうかをページごとに自動で判断します。PDF ファイルが電子形式、スキャン画像、または両者の混在のいずれであるかを事前に把握していない場合にのみ使用します。AUTO モードの実行には、わずかな計算オーバーヘッドが伴います。
config.format：出力形式を制御します。
- TEXT
- MARKDOWN
- HTML
config.languages：検出する言語を制御します。SCAN または AUTO モードでの OCR に使用されます。
- language：KOR などの言語コードです。
- type：常に "language" に設定します。
pageRange：抽出するページを制御します。上記の操作シグネチャでは、ページ範囲によってページ 0、1、2、3、4 が処理されます。大きなドキュメントのビルドパフォーマンスを向上させるには、ページ範囲を使用します。ページごとに1件のリクエストを送信する代わりに、複数のページをまとめてバッチ処理できます。ドキュメントのサイズ、モデルの制限、レート制限によりますが、まずは5〜10ページ程度のページ範囲から始めるのが適切です。

パラメーターに関する考慮事項

電子 PDF ファイルでは、言語を指定する必要はありません。
ドキュメントの言語が英語の場合、言語を指定する必要はありません。
英語以外のスキャンドキュメントでは、該当する OCR 言語を指定します。

Python の例

Python のトランスフォームまたは関数を使用してドキュメントからテキストへの変換を実行するには、以下の例を参照してください。

トランスフォーム

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
88
89
90
91
92
93
94
import polars as pl
from concurrent.futures import ThreadPoolExecutor
from transforms.api import Output, 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(OUTPUT_DATASET_RID),
    media_input=MediaSetInput(INPUT_MEDIA_SET_RID),
)
def extract(media_input, output):
    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": "ELECTRONIC",
                                    "format": "TEXT",
                                },
                            },
                        },
                    },
                },
            )

            return {
                "media_item_rid": media_item_rid,
                "media_reference": media_reference,
                "page_num": page_num,
                "extraction_result": str(extraction_result.json()),
            }

        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)

関数

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
from time import sleep

from foundry_sdk import FoundryClient
from foundry_sdk.v2.media_sets import models
from functions.api import function

#### Helper functions

def _create_transform_job(
    media_set_rid: str, media_item_rid: str, transformation: models.DocumentToTextTransformation
) -> str:
    fc = FoundryClient()
    job_initiation_resp = fc.media_sets.MediaSet.transform(
        media_set_rid=media_set_rid,
        media_item_rid=media_item_rid,
        transformation=transformation,
        preview=True,
    )
    job_id = job_initiation_resp.job_id
    return job_id


def _is_transform_finished(media_set_rid: str, media_item_rid: str, job_id: str) -> bool:
    fc = FoundryClient()
    status = fc.media_sets.MediaSet.get_status(media_set_rid, media_item_rid, job_id, preview=True)
    return status.status in ("SUCCESSFUL", "FAILED")


def _get_transform_result(media_set_rid: str, media_item_rid: str, job_id: str) -> str:
    fc = FoundryClient()
    result = fc.media_sets.MediaSet.get_result(media_set_rid, media_item_rid, job_id, preview=True)
    return result.decode("utf-8")


def _run_transform_blocking(
    media_set_rid: str, media_item_rid: str, transformation: models.DocumentToTextTransformation
) -> str:
    job_id = _create_transform_job(media_set_rid, media_item_rid, transformation)
    while not _is_transform_finished(media_set_rid, media_item_rid, job_id):
        sleep(0.5)
    return _get_transform_result(media_set_rid, media_item_rid, job_id)

# Use this function if your input is a media set
# We suggest running the function across batches of five to ten pages to avoid timeout
@function(beta=True)
def transform_vlm(media_set_rid: str, media_item_rid: str, start_page_inclusive: int, end_page_exclusive: int) -> str:
    LANGUAGES: list[models.OcrLanguageOrScript] = [models.OcrLanguageWrapper(language="ENG")]

    return _run_transform_blocking(
        media_set_rid,
        media_item_rid,
        models.DocumentToTextTransformation(
            operation=models.ExtractDocumentLayoutAwareTextV2Operation(
                page_range=models.PageRange(start_page_inclusive=start_page_inclusive, end_page_exclusive=end_page_exclusive),
                config=models.ExtractDocumentLayoutAwareTextV2Config(languages=LANGUAGES),
            )
        ),
    )

# Use this function if your input is an object
# We suggest running the function across batches of five to ten pages to avoid timeout
@function(beta=True, edits=[<YOUR_OBJECT_TYPE>])
def transform_vlm_object(myObject: <YOUR_OBJECT_TYPE>, start_page_inclusive: int, end_page_exclusive: int) -> str:
    reference_view = myObject.media_reference.get_media_reference().reference.media_set_view_item

    media_set_rid = reference_view.media_set_rid
    media_item_rid = reference_view.media_item_rid
    LANGUAGES: list[models.OcrLanguageOrScript] = [models.OcrLanguageWrapper(language="ENG")]

    return _run_transform_blocking(
        media_set_rid,
        media_item_rid,
        models.DocumentToTextTransformation(
            operation=models.ExtractDocumentLayoutAwareTextV2Operation(
                page_range=models.PageRange(start_page_inclusive=start_page_inclusive, end_page_exclusive=end_page_exclusive),
                config=models.ExtractDocumentLayoutAwareTextV2Config(languages=LANGUAGES),
            )
        ),
    )

←

PREVIOUS抽出戦略をPython関数にデプロイ

NEXTAIP Evals / 概要

→