문서 추출을 Python 함수로 배포하기

앱의 Deploy to functions 안내에서 생성된 추출 코드를 사용하여 Python 함수 저장소를 구성합니다. 함수는 입력 미디어의 지정 페이지를 처리하므로 많은 문서나 페이지를 처리할 때는 별도의 실행 조정 전략이 필요합니다.

저장소 준비

새 Python 함수 저장소를 만들거나 기존 저장소를 사용합니다. 왼쪽 Libraries에서 foundry-platform-sdk 버전 1.78 이상을 설치합니다. 청크 분할에는 aip-workflows 0.40.0 이상, 임베딩에는 openai가 필요합니다. LLM 기반 추출이면 선택한 모델을 저장소에 가져옵니다.

함수 제한 시간을 늘리는 것이 권장되며 플랫폼 인스턴스에서 허용하는 최대값을 고려합니다. 제한 시간만 늘려 여러 PDF나 페이지 수가 많은 PDF를 한 호출에 모두 처리하려고 하지 말고 페이지별 실행을 조정합니다.

공통 실행 및 오류 처리

보조 함수는 모든 추출 전략에서 같습니다. 추출 함수와 같은 파일이나 공통 유틸리티 파일에 둡니다. 변환 작업을 시작하고 0.5초마다 상태를 확인하여 SUCCESSFUL 또는 FAILED가 되면 결과를 UTF-8로 읽습니다.

TransformResult는 성공 시 result에 추출 Markdown 또는 JSON을, error와 retryable에는 None을 반환합니다. 실패 시 result=None이고 오류와 재시도 가능 여부를 반환하므로 호출하는 조정 로직에서 분기할 수 있습니다. PalantirQoSException은 힌트가 DO_NOT_RETRY가 아니면 재시도 가능으로 분류합니다. PalantirRPCException과 나머지 PalantirException은 재시도 불가로 분류합니다. 이 보조 코드 자체가 실패 작업을 자동 재실행하지는 않습니다.

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:
    """The outcome of a transform. Functions return this instead of raising on failure so your
    orchestration can branch on the result.

    result: The extracted content (Markdown or JSON, depending on your configuration) on success,
        otherwise None.
    error: The specific error name on failure (or a human-readable reason for rate-limit and
        availability errors), otherwise None.
    retryable: True when the failure is transient (rate limits or service availability) and worth
        retrying, False when it is not, and None on success.
    """

    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 문자열 조합 또는 객체 속성을 고릅니다. 객체 속성 템플릿은 해당 객체를 저장소로 가져와 Python Ontology SDK를 생성해야 합니다.

청크와 임베딩

chunk_text는 문자열 배열을 받아 청크 문자열의 중첩 배열을 반환합니다. 기본 chunk_size=8192, chunk_overlap=0, concat_before_chunk=False, chunk_mode="markdown"입니다. 크기와 겹침은 문자 수이며 recursive는 형식 없는 일반 텍스트에 적합합니다. concat_before_chunk=True이면 입력 전체를 합쳐 분할하고 바깥 배열 길이가 1이 됩니다. 문서의 페이지별 텍스트를 합칠 때 사용할 수 있습니다. False는 입력 문자열별로 하나의 내부 목록을 반환합니다.

embed_text는 입력 문자열마다 벡터 하나를 반환합니다. 앱에서 선택한 임베딩 모델 RID를 기본값에 넣어야 하며 아래의 <YOUR_EMBEDDING_MODEL_RID>는 사용 환경에서 치환할 설정 자리입니다. 코드의 Foundry 인증 토큰, OpenAI 호환 기본 URL과 HTTP 클라이언트는 런타임 유틸리티에서 가져옵니다. 생성 함수의 기본 매개변수는 앱에서 선택한 청크 및 임베딩 설정에 맞춥니다.

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]]:
    """
    Chunks a list of text strings.
    Args:
        texts: A list of text strings to chunk.
        chunk_size: Maximum number of characters per chunk.
        chunk_overlap: Number of overlapping characters between consecutive chunks.
        concat_before_chunk: If True, concatenates all texts into a single string before chunking,
            returning a single inner list. Recommended when input is per-page text of a document.
            If False, chunks each text independently, returning one inner list per input text.
        chunk_mode: The chunking strategy to use. Options:
            - "markdown": Recommended for Markdown text.
            - "recursive": Recommended for raw text without any format.
    Returns:
        A list of lists of chunk strings. Each inner list contains the chunks for one input text.
        When concat_before_chunk is True, input texts are concatenated into a single string so the
        output list has length of 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]]:
    """
    Generates embeddings for a list of text strings using the configured embedding model.
    Args:
        texts: A list of text strings to embed.
        embedding_model_rid: The RID of the embedding model to use.
    Returns:
        A list of vector embeddings, one per input text.
    """
    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))]