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

AIP Document Intelligence の関数にデプロイガイドに沿って進め、生成されたコードスニペットを使って、抽出戦略を組み込んだ Python 関数リポジトリをセットアップします。

セットアップ

関数でのドキュメント抽出をはじめるには、まず Python 関数リポジトリを作成するか、既存のリポジトリを使用します。

ドキュメント抽出を実行するために、適切なインポートと権限をリポジトリに設定します。

1. プラットフォーム SDK をインストールします（まだインストールされていない場合）。

• 左側のパネルでライブラリを選択し、ライブラリを追加します。
  • foundry-platform-sdk のバージョン 1.78 以上をインストールします。
  • チャンク分割が必要な場合は、aip-workflows のバージョン 0.40.0 以上をインストールします。
  • 埋め込みを生成する場合は、openai をインストールします。

2. LLM ベースの抽出では、選択したモデルをリポジトリにインポートします。
3. 関数のタイムアウトを長く設定します。エンロールメントで許可されている最大値を選択することを推奨します。多数の PDF やページ数の多い PDF を処理するには、これらの関数を抽出戦略の一部として使用する必要があります。

以降のセクションでは、関数にデプロイガイドに含まれる各種コードスニペットについて説明します。

ヘルパー関数

ヘルパー関数は、すべての抽出戦略で共通です。関数と同じ 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
from dataclasses import dataclass
from time import sleep
from typing import Optional

from foundry_sdk import FoundryClient
from foundry_sdk._errors import PalantirException, PalantirQoSException, PalantirRPCException
from foundry_sdk._errors.palantir_qos_exception import QoSRetryHint
from foundry_sdk.v2.media_sets import models
from functions.api import function


@dataclass
class TransformResult:
    """変換の結果です。関数は失敗時に例外を送出せず、これを返すため、
    オーケストレーションで結果に応じて分岐できます。

    result: 成功時は抽出されたコンテンツ（設定に応じて Markdown または JSON）、
        それ以外は None です。
    error: 失敗時は具体的なエラー名（レート制限や可用性のエラーの場合は、
        人が読める理由）、それ以外は None です。
    retryable: 失敗が一時的（レート制限やサービスの可用性）で再試行する価値がある場合は
        True、そうでない場合は False、成功時は None です。
    """

    result: Optional[str]
    error: Optional[str]
    retryable: Optional[bool]


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)


def _run_transform_error_handled(
    media_set_rid: str, media_item_rid: str, transformation: models.DocumentToTextTransformation
) -> TransformResult:
    try:
        result_text = _run_transform_blocking(media_set_rid, media_item_rid, transformation)
        return TransformResult(result=result_text, error=None, retryable=None)
    except PalantirQoSException as e:
        retryable = e.retry_hint != QoSRetryHint.DO_NOT_RETRY
        return TransformResult(result=None, error=e.reason, retryable=retryable)
    except PalantirRPCException as e:
        return TransformResult(result=None, error=e.name or type(e).__name__, retryable=False)
    except PalantirException as e:
        return TransformResult(result=None, error=type(e).__name__, retryable=False)

抽出戦略

この関数は抽出戦略に対応しており、メディア入力と指定のページを必要とします。関数は動的に生成されるため、アプリ内からコードをコピーします。公開時に付けた名前で登録されるため、必要に応じて名前を変更できます。メディア入力は、メディアセットと項目の RID 文字列の組み合わせ、またはオブジェクトプロパティとして指定できます。左上のセレクターでテンプレートを選択します。オブジェクトプロパティのテンプレートでは、オブジェクトをインポートする必要があります。

チャンク分割と埋め込みの関数

必要に応じて、抽出したテキストをチャンクに分割し、下流の検索や取得のワークフローで使用する埋め込みを生成します。以下の関数は静的です。チャンクサイズ、チャンクのオーバーラップ、埋め込みモデルはアプリで設定されるため、デフォルトのパラメーター値を選択した値に置き換えます。

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
from functions.api import function, Array, Boolean, Float, Integer, String
from aip_workflows.document_intelligence.transforms import DocumentChunker

@function(beta=True)
def chunk_text(
    texts: Array[String],
    chunk_size: Integer = 8192,
    chunk_overlap: Integer = 0,
    concat_before_chunk: Boolean = False,
    chunk_mode: String = "markdown",
) -> Array[Array[String]]:
    """
    テキスト文字列のリストをチャンクに分割します。
    Args:
        texts: チャンクに分割するテキスト文字列のリストです。
        chunk_size: チャンクごとの最大文字数です。
        chunk_overlap: 連続するチャンク間で重複する文字数です。
        concat_before_chunk: True の場合、チャンクに分割する前にすべてのテキストを1つの文字列に連結し、
            内側のリストを1つ返します。入力がドキュメントのページごとのテキストである場合に推奨します。
            False の場合、各テキストを個別にチャンクに分割し、入力テキストごとに内側のリストを1つ返します。
        chunk_mode: 使用するチャンク分割の方式です。オプション：
            - "markdown": Markdown テキストに推奨します。
            - "recursive": 形式のない生のテキストに推奨します。
    Returns:
        チャンク文字列のリストのリストです。内側の各リストには、1つの入力テキストのチャンクが含まれます。
        concat_before_chunk が True の場合、入力テキストは1つの文字列に連結されるため、
        出力リストの長さは1になります。
    """
    chunks = DocumentChunker.create_chunks(
        texts,
        chunk_mode=chunk_mode,
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        strip_markdown=True,
        concat_before_chunk=concat_before_chunk,
    )
    if concat_before_chunk:
        return [chunks]
    return chunks

from foundry_sdk.v2.language_models.utils import (
    get_foundry_token,
    get_openai_base_url,
    get_http_client,
)
from openai import OpenAI

@function(beta=True)
def embed_text(
    texts: Array[String],
    embedding_model_rid: String = "<YOUR_EMBEDDING_MODEL_RID>",
) -> Array[Array[Float]]:
    """
    設定された埋め込みモデルを使用して、テキスト文字列のリストの埋め込みを生成します。
    Args:
        texts: 埋め込みを生成するテキスト文字列のリストです。
        embedding_model_rid: 使用する埋め込みモデルの RID です。
    Returns:
        入力テキストごとに1つのベクトル埋め込みを含むリストです。
    """
    client = OpenAI(
        api_key=get_foundry_token(preview=True),
        base_url=get_openai_base_url(preview=True),
        http_client=get_http_client(preview=True),
    )

    response = client.embeddings.create(
        input=texts,
        model=embedding_model_rid,
    )

    return [response.data[i].embedding for i in range(len(texts))]