循序渐进 · AIP 教学 · 文档智能（五）

文档转文本（Document-to-text）变换

这是底层能力：把多种格式的文档转成文本，并且保留版面结构（段落、标题、表格）。抽取质量的上限由它决定。

全部目录 AIP 首页 ← 上一篇 文档转文本（Document-to-text）变换 下一篇 →
本文来源 · Source 内容整理自 Palantir Foundry 官方文档：
https://www.palantir.com/docs/foundry/document-intelligence/document-to-text/
原始标题：AIP Document Intelligence • Document-to-text media transformations • Palantir · 所属：AIP Document Intelligence（读懂文档）
★

先记住这几条

① 多格式支持
不只 PDF，覆盖多种常见文档格式。
② 保留版面结构
段落、标题、表格都能识别出来。
③ 兼顾灵活与性能
既能精细抽取，又能跑得动。
④ 是上层抽取的基础
抽不准往往是这一步没做好。
0

写在前面

Document-to-text 变换提供了一种灵活且高性能的方式来抽取多种格式的文档内容，包括抽取文档布局结构的能力，例如段落、标题和表格。

AIP Document Intelligence 提供两种 document-to-text 媒体变换操作：

  • extractTextV2：返回一个字符串列表，包含抽取出的文档文本。
  • extractLayoutAwareTextV2：返回跨页面的布局感知（layout-aware）文本块列表。

我们建议使用上面列出的操作，它们取代了 extractLayoutAwareContent 和 ocrOnPage 操作。

你可以参考下面的示例操作签名：

{
  "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"
            }
          ]
        }
      }
    }
  }
}
1

操作签名参数

要点：这个变换接收哪些参数。
  • 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 页。使用页面范围可以提升较大文档的构建性能。不必为每一页发起一次请求，你可以将多个页面批量合并处理。根据文档大小、模型限制和速率限制的不同，大约五到十页的页面范围是一个合适的起点。

Parameter considerations

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

Python 示例

要点：可直接运行的代码。

参考以下示例，帮助你使用 Python transform 或 function 执行 document-to-text 变换。

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

延伸阅读 · 相关页面

按主题横向跳转，不必顺着目录一篇篇读。

本组其他页面 · AIP Document Intelligence（读懂文档）

同一主题下的相邻内容。

全部目录 AIP 首页 ← 上一篇：把抽取策略部署到 Python 函数 下一篇：AIP Analyst 总览：对话式分析 →