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

把抽取策略部署到 Python transform

验证好的策略可以部署成 Python transform，对媒体集里所有文档的所有页面跑批量抽取 —— 这是从"试验"到"规模化"的关键一步。

全部目录 AIP 首页 ← 上一篇把抽取策略部署到 Python transform 下一篇 →

本文来源 · Source 内容整理自 Palantir Foundry 官方文档：
https://www.palantir.com/docs/foundry/document-intelligence/deploy-to-python-transforms/
原始标题：AIP Document Intelligence • Deploy extraction strategies to Python transforms • Palantir · 所属：AIP Document Intelligence（读懂文档）

★

先记住这几条

① 部署后即可批量跑

覆盖整个媒体集与全部页面。

② 产出是一套模板

生成与交互式验证一致的结果。

③ 适合定期批处理

接到数据管道里，随新文档自动跑。

写在前面

在 AIP Document Intelligence 中验证抽取策略之后，你可以将其部署为一个 Python transform，以便对媒体集中所有文档的所有页面运行批量抽取。部署出的模板会产生与 AIP Document Intelligence 中对应配置相同的结果。

使用部署出的模板

要点：生成产物怎么用。

模板在 Code Repositories 中创建完成后：

在 src/myproject/document_extraction/my_extraction.py 文件中的 @transform.using 装饰器里指定你的输出数据集。
触发构建。

该模板使用轻量级 transforms（lightweight transforms）以获得最佳性能。旧版本使用的是基于 Spark 的 transforms，由于 Spark 开销，其速度要慢得多。如果你还没有迁移，我们建议迁移到轻量级 transforms。

该模板目前尚不支持预览模式（Preview mode）。使用预览时预计会出现错误，但实际构建会正常工作。

Incremental processing

默认情况下，文档抽取 transform 是非增量的，这意味着每次运行都会处理所有文档。你可以通过取消注释 @incremental(...) 装饰器行，将 transform 配置为增量运行。对于增量 transform，当新文档被添加到输入媒体集时，重新运行该 transform 将只处理新文档，并将结果追加到输出数据集。

Customizing the prompt

对于生成式 AI 配置，模板会继承你在 AIP Document Intelligence 中指定的提示词。你可以在 src/myproject/document_extraction/prompts.py 中查看提示词。

我们不建议直接在模板中编辑提示词，因为这会导致 Document Intelligence 结果与批处理作业结果之间出现差异。相反，应在 Document Intelligence 中调整提示词，在那里验证结果，然后重新部署以创建新模板。

Transform input type	Customizable prompt
`VisionLLMDocumentsExtractorInput`	User prompt only (system prompt is fixed)
`VisionLLMLayoutDocumentsExtractorInput` (layout-aware extraction)	System prompt only (user prompt is fixed)

对于布局感知抽取配置，用户提示词必须保持固定，因为它包含一个特殊的 JSON schema，用于保留布局结构信息。修改此提示词会显著降低抽取成功率。

Custom image preprocessing

对于在抽取前需要进行图像变换的文档，例如文本内容有旋转的文档，你应该：

创建一个单独的 transform 管道来应用图像变换。
将处理后的结果保存到一个新的媒体集。
在处理后的媒体集上使用 Document Intelligence 进行抽取。

Run on a subset of media items

对于布局感知的生成式 AI 配置，视觉 LLM 必须生成符合特定 schema 的有效 JSON。如果响应是无效 JSON 或不符合该 schema，抽取会失败并报 ERROR_RESPONSE_JSON_PARSING 错误。

在实践中，使用顶级模型时大约 5% 的抽取可能会失败。对于失败的行，你仍然可以获得有效的 layoutInfo，其中只包含来自布局模型的抽取结果。

要对失败的行重新运行抽取：

使用 filter_on_media_items 参数并传入媒体项 ID 列表，以只处理特定的项。
移除 @incremental 装饰器，这样这些行会被重新处理，而不会被判定为已完成。

Improve runtime performance

THREAD_NUMBER 参数控制并发线程数，其中每个线程一次从一页文档抽取数据。值越高，作业完成得越快。

Setting	Value	Notes
Default	20	Conservative setting suitable for most environments
Maximum tested	300	Achievable in development environments with abundant Vision LLM capacity

在容量受限的环境中把 THREAD_NUMBER 的值设得太高会导致速率限制错误。随后重试循环会消耗大量容量，影响使用同一模型的其他作业。调整此参数时，你应该监控用量。

Find logs

要查看构建日志，请在构建详情页面上选择 Telemetry。要筛选文档抽取日志，请在 message 列中筛选以 aip_workflows 开头的值。

Row-level vs. document-level chunking

抽取输出为每页一行。默认情况下，DocumentChunker.create_chunks_per_document 会在分块前将同一文档的所有页面合并为一个 Markdown 字符串。

要独立地对每一行进行分块而不合并页面，请改用 DocumentChunker.create_chunks_per_row：

````python

chunking_result = chunker.create_chunks_per_row( extraction_df, chunk_mode="markdown", # "recursive" for plain text, "markdown" for markdown text content_column="extractionResult", id_column="media_item_rid", # used as prefix for chunk_id chunk_size=8192, chunk_overlap=0, thread_number=20, strip_markdown=False, # set True to remove ``markdown and `` wrappers before chunking )

````

Create embeddings without chunking

在创建嵌入之前建议先进行分块，因为嵌入模型有上下文限制。要在保留管道结构的同时跳过分块，请在 create_chunks_per_row 中把 chunk_size 设为一个非常大的值，例如 sys.maxsize。

模板示例

要点：可参考的代码。

以下示例展示了为每种抽取配置生成的 transform 代码。这些仅供参考。你应该使用 Document Intelligence 中的部署工具来创建 transform，而不是为文档抽取手写 transform 代码。

Traditional extraction: Raw text

通过读取文档元数据来抽取文本。仅适用于电子生成的 PDF。

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)  # uncomment this line if incremental is needed
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
)
def extract(media_input, output):
    """
    Extracts content from pdf documents with raw text extraction
    """
    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)

Traditional extraction: OCR

使用传统光学字符识别（OCR）来抽取文本，但不保留布局信息。

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)  # uncomment this line if incremental is needed
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
)
def extract(media_input, output):
    """
    Extracts content from pdf documents with OCR text extraction
    """
    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)

Traditional extraction: Layout-aware OCR

使用带边界框的高级 OCR，以保留文档布局和结构。

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)  # uncomment this line if incremental is needed
@transform.using(
    output=Output("ri.foundry.main.dataset.abc"),
    media_input=MediaSetInput("ri.mio.main.media-set.abc"),
)
def extract(media_input, output):
    """
    Extracts content from pdf documents with layout-aware OCR extraction
    """
    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": "extractLayoutAwareContent",
                        "extractLayoutAwareContent": {
                            "parameters": {
                                "languages": ["ENG"]
                            }
                        }
                    }
                }
            })
            extraction_result = str(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)

Generative AI extraction: Basic

使用视觉语言模型将内容抽取为 Markdown，不进行预处理。

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"])  # uncomment this line if incremental is needed
@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):
    """
    Extracts content from pdf documents as markdown.
    """
    extracted_data = extractor.create_extraction(
        media_input, with_ocr=False, prompt=USER_PROMPT, thread_number=THREAD_NUMBER
    )
    output.write_dataframe(extracted_data)

Generative AI extraction: With OCR preprocessing

使用带 OCR 预处理的视觉语言模型，以改善对复杂文档的抽取效果。

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"])  # uncomment this line if incremental is needed
@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):
    """
    Extracts content from pdf documents as markdown.
    """
    extracted_data = extractor.create_extraction(
        media_input, with_ocr=True, prompt=USER_PROMPT, thread_number=THREAD_NUMBER
    )
    output.write_dataframe(extracted_data)

Generative AI extraction: Layout-aware

使用带布局感知 OCR 预处理的视觉语言模型，在返回抽取内容的同时返回布局信息。

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"])  # uncomment this line if incremental is needed
@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):
    """
    Extracts content from pdf documents as 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)

Generative AI extraction: Layout-aware with table cropping

使用带布局感知 OCR 预处理和表格裁剪（table cropping）的视觉语言模型，以提高表格抽取的准确性。

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"])  # uncomment this line if incremental is needed
@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):
    """
    Extracts content from pdf documents as 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)

Chunk extracted text and generate embeddings

如果不需要嵌入，请从 transform 装饰器中移除 embedder，并移除 embedding_result 那一行。

````python

from transforms.api import Input, Output, incremental, transform from aip_workflows.document_intelligence.transforms import DocumentChunker, DocumentEmbedderInput

THREAD_NUMBER = 20

@transform.using( extraction_input=Input("ri.foundry.main.dataset.abc"), # typically the output dataset from the extraction transform 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" for raw text, "markdown" for markdown text content_column="extractionResult", # content column name id_column="media_item_rid", # id of the document, used to combine content (e.g. from different pages) of the single document page_column="page_num", # page number column name chunk_size=8192, chunk_overlap=0, thread_number=THREAD_NUMBER, strip_markdown=True, # when True, removes ``markdown prefix and `` suffix from the content before chunking ) embedding_result = embedder.create_embeddings( chunking_result, content_column="chunk_content", thread_number=THREAD_NUMBER, ) output.write_dataframe(embedding_result)

````

延伸阅读 · 相关页面

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

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

同一主题下的相邻内容。

常见问题速答 · FAQ

关于「把抽取策略部署到 Python transform」，读者最常问的几个问题。

使用部署出的模板是什么？

生成产物怎么用。模板在 Code Repositories 中创建完成后。

模板示例是什么？

可参考的代码。以下示例展示了为每种抽取配置生成的 transform 代码。这些仅供参考。你应该使用 Document Intelligence 中的部署工具来创建 transform，而不是为文档抽取手写 transform 代码。

全部目录 AIP 首页 ← 上一篇：文档智能核心概念下一篇：把抽取策略部署到 Python 函数 →