Document-to-text media transformations（文档到文本的媒体转换）

Document-to-text transformations provide a flexible and performant way to extract document content in multiple formats, including the ability to extract document layout structure, such as paragraphs, headers, and tables.

AIP Document Intelligence provides two document-to-text media transformation operations:

• extractTextV2: Returns a list of strings containing extracted document text.
• extractLayoutAwareTextV2: Returns a list of layout-aware text blocks across pages.

:::callout{theme="neutral"} We recommend using the operations listed above, which replace the extractLayoutAwareContent and ocrOnPage operations. :::

You can reference an example operation signature below:

{
  "type": "documentToText",
  "documentToText": {
    "operation": {
      "type": "{operation}", // {operation}: "extractTextV2" or "extractLayoutAwareTextV2"
      "{operation}": {
        "pageRange": {
          "startPageInclusive": 0,
          "endPageExclusive": 5
        },
        "config": {
          "mode": "SCAN", // or "ELECTRONIC" or "AUTO"
          "format": "TEXT", // or "MARKDOWN" or "HTML"
          "languages": [
            {
              "type": "language",
              "language": "KOR"
            }
          ]
        }
      }
    }
  }
}

Operation signature parameters

• config.mode: Controls how the document pages are interpreted.
• ELECTRONIC: Treats all pages as electronic PDF files and extracts embedded/raw text. Use ELECTRONIC when you know your documents contain embedded text.
• SCAN: Treats all pages as scanned images and performs Optical Character Recognition (OCR). Use SCAN when you know your documents are scanned images.
• AUTO: Automatically decides per-page whether OCR is needed. Use only when you do not know ahead of time whether the PDF files are electronic, scanned, or mixed. AUTO mode runs with a small compute overhead.
• config.format: Controls the output format.
• TEXT
• MARKDOWN
• HTML
• config.languages: Controls the languages to be detected, which is used for OCR in SCAN or AUTO mode.
• language: The language code, such as KOR.
• type: Always set to "language".
• pageRange: Controls the pages to be extracted. In the operation signature above, the page range processes pages 0, 1, 2, 3, and 4. Use page ranges to improve build performance for larger documents. Instead of issuing one request per page, you can batch multiple pages together. A page range of roughly five to ten pages is an appropriate starting point, depending on document size, model limits, and rate limits.

Parameter considerations

• You do not need to specify languages for electronic PDF files.
• You do not need to specify languages if the document language is English.
• For non-English scanned documents, specify the relevant OCR language.

Python examples

Reference the examples below to help you execute a document-to-text transformation using a Python transform or function.

Transform

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)  # Uncomment this line if incremental is needed.
@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)

Function

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

---

中文翻译

文档到文本的媒体转换

文档到文本的媒体转换（Document-to-text media transformations）提供了一种灵活且高效的方式来提取多种格式的文档内容，包括提取文档布局结构（如段落、标题和表格）的能力。

AIP Document Intelligence 提供两种文档到文本的媒体转换操作：

• extractTextV2：返回包含提取文档文本的字符串列表。
• extractLayoutAwareTextV2：返回跨页面的布局感知文本块列表。

:::callout{theme="neutral"} 我们建议使用上述操作，它们取代了 extractLayoutAwareContent 和 ocrOnPage 操作。 :::

以下是一个操作签名示例：

{
  "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。使用页面范围可以提高较大文档的构建性能。您可以将多个页面批量处理，而不是每页发送一个请求。根据文档大小、模型限制和速率限制，大约五到十页的页面范围是一个合适的起点。

参数注意事项

• 对于电子 PDF 文件，您无需指定语言。
• 如果文档语言为英语，您无需指定语言。
• 对于非英语的扫描文档，请指定相关的 OCR 语言。

Python 示例

参考以下示例，帮助您使用 Python 转换或函数执行文档到文本的转换。

转换

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_rid} 不是文档")

            if metadata.pages is None:
                raise ValueError(f"媒体项 {media_item_rid} 没有页数信息")

            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)

函数

from time import sleep

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

#### 辅助函数

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)

# 如果您的输入是媒体集，请使用此函数
# 建议在五到十页的批次上运行此函数以避免超时
@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),
            )
        ),
    )

# 如果您的输入是对象，请使用此函数
# 建议在五到十页的批次上运行此函数以避免超时
@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),
            )
        ),
    )