변환에서 Palantir 제공 언어 모델 사용하기

palantir_models의 모델 입력을 선언하면 Python 변환에서 언어 모델이나 임베딩 모델을 사용할 수 있습니다. Spark와 단일 노드 lightweight 변환을 모두 지원합니다. 플랫폼 인스턴스에서 AIP를 활성화하고 해당 모델에 접근할 수 있어야 합니다.

저장소와 모델 선택

Libraries에서 palantir_models와 language-model-service-api를 추가하여 설치합니다. 의존성을 해석하고 checks가 끝난 뒤 모델 입력을 선언합니다. RID를 입력하면 드롭다운에 사용 가능한 모델이 표시됩니다.

모델 제공 여부는 플랫폼 인스턴스, 지역과 권한 및 정책을 따릅니다. 지원 모델과 AIP 활성화를 확인합니다. 아래 MODEL_RID는 예시이며 저장소에서 실제 선택한 RID로 바꿉니다. 데이터셋 경로도 가상의 예시입니다.

모델 입력과 응답 인터페이스

공통 텍스트 대화에는 GenericChatCompletionLanguageModelInput을 사용할 수 있습니다. Generic 요청은 DialogChatMessage의 문자열 content와 SYSTEM/USER/ASSISTANT 역할을 사용하며, 선택적으로 temperature, stop_sequences, max_tokens를 지정합니다. 생성 텍스트는 completion에 있고 token_usage는 입력/출력 토큰 정보를 담습니다. GPT 방식의 choices 목록과 구분합니다. 공통 채팅 완료 API에 전체 요청과 응답 계약이 있습니다.

모델 핸들이 주입되는 transform 계열을 사용합니다. 기존 Spark @transform과 현재 transform.spark.using(), lightweight transform.using() 선언을 엔진에 맞춰 선택합니다.

Spark 변환에서 리뷰 감정 분석

입력에는 NULL이 아닌 문자열을 담은 review_content 열이 있어야 하며 sentiment 열은 새로 추가합니다. 다음 구현은 원래 열의 스키마를 유지하면서 실제 모델 응답을 붙입니다. 생성 결과는 모델의 텍스트이며 업무 분류의 품질은 별도로 평가해야 합니다.

from language_model_service_api.languagemodelservice_api import ChatMessage, ChatMessageRole
from language_model_service_api.languagemodelservice_api_completion_v3 import GptChatCompletionRequest
from palantir_models.models import OpenAiGptChatLanguageModel
from palantir_models.transforms import OpenAiGptChatLanguageModelInput
from pyspark.sql import types as T
from transforms.api import Input, Output, transform

MODEL_RID = "ri.language-model-service..language-model.gpt-4-1-nano"


@transform.spark.using(
    reviews=Input("/Example Project/Inputs/reviews"),
    model=OpenAiGptChatLanguageModelInput(MODEL_RID),
    output=Output("/Example Project/Outputs/review_sentiment"),
)
def compute(ctx, reviews, model: OpenAiGptChatLanguageModel, output):
    source = reviews.dataframe()
    schema = T.StructType([
        *source.schema.fields,
        T.StructField("sentiment", T.StringType(), True),
    ])
    rows = []
    for row in source.toLocalIterator():
        request = GptChatCompletionRequest([
            ChatMessage(ChatMessageRole.SYSTEM, "Determine the sentiment of the following review."),
            ChatMessage(ChatMessageRole.USER, row["review_content"]),
        ])
        response = model.create_chat_completion(request)
        rows.append(tuple(row) + (response.choices[0].message.content,))
    output.write_dataframe(ctx.spark_session.createDataFrame(rows, schema=schema))

이 예제의 모델 호출은 driver에서 차례로 수행합니다. Spark 선언을 사용했다는 이유만으로 모델 요청까지 executor에 분산되지는 않습니다. 입력은 iterator로 읽지만 출력 행 목록은 메모리에 유지하므로 전체 결과가 driver 메모리에 들어가는 규모에 사용합니다. 스키마를 명시하여 입력이 비어도 출력 열을 구성할 수 있습니다.

임베딩 만들기

같은 review_content에 대해 임베딩을 요청할 수 있습니다. GenericEmbeddingsRequest의 inputs는 문자열 목록이며 아래 예제는 한 번에 한 리뷰를 전달합니다. 온톨로지의 벡터 속성에서 사용할 배열은 FloatType으로 명시적으로 변환합니다.

from language_model_service_api.languagemodelservice_api_embeddings_v3 import GenericEmbeddingsRequest
from palantir_models.models import GenericEmbeddingModel
from palantir_models.transforms import GenericEmbeddingModelInput
from pyspark.sql import functions as F, types as T
from transforms.api import Input, Output, transform

MODEL_RID = "ri.language-model-service..language-model.text-embedding-ada-002_azure"


@transform.spark.using(
    reviews=Input("/Example Project/Inputs/reviews"),
    model=GenericEmbeddingModelInput(MODEL_RID),
    output=Output("/Example Project/Outputs/review_embeddings"),
)
def compute(ctx, reviews, model: GenericEmbeddingModel, output):
    source = reviews.dataframe()
    schema = T.StructType([
        *source.schema.fields,
        T.StructField("embedding", T.ArrayType(T.DoubleType()), True),
    ])
    rows = []
    for row in source.toLocalIterator():
        request = GenericEmbeddingsRequest(inputs=[row["review_content"]])
        embedding = model.create_embeddings(request).embeddings[0]
        rows.append(tuple(row) + (embedding,))
    result = ctx.spark_session.createDataFrame(rows, schema=schema)
    output.write_dataframe(result.withColumn(
        "embedding", F.col("embedding").cast(T.ArrayType(T.FloatType()))
    ))

입력에는 embedding 열이 없다고 가정합니다. 감정 분석 예제와 같은 driver 메모리 및 순차 호출 조건을 따릅니다.

Lightweight의 감정 분석

분산 표 계산이 필요하지 않은 작업은 Spark 없이 Pandas DataFrame을 직접 쓰는 lightweight 형태를 사용할 수 있습니다. 다음 세 문장은 설명용 입력이며 응답은 실행 시 실제 모델이 생성합니다.

import pandas as pd
from language_model_service_api.languagemodelservice_api import ChatMessage, ChatMessageRole
from language_model_service_api.languagemodelservice_api_completion_v3 import GptChatCompletionRequest
from palantir_models.models import OpenAiGptChatLanguageModel
from palantir_models.transforms import OpenAiGptChatLanguageModelInput
from transforms.api import Output, transform

MODEL_RID = "ri.language-model-service..language-model.gpt-4-1-nano"


@transform.using(
    model=OpenAiGptChatLanguageModelInput(MODEL_RID),
    output=Output("/Example Project/Outputs/lightweight_sentiment"),
)
def compute(model: OpenAiGptChatLanguageModel, output):
    items = ["I love this product!", "This is terrible.", "It was okay, nothing special."]

    def sentiment(text):
        request = GptChatCompletionRequest([
            ChatMessage(ChatMessageRole.SYSTEM, "Respond with one word: positive, negative, or neutral."),
            ChatMessage(ChatMessageRole.USER, text),
        ])
        return model.create_chat_completion(request).choices[0].message.content

    frame = pd.DataFrame({"text": items})
    frame["sentiment"] = frame["text"].apply(sentiment)
    output.write_table(frame)

한 단어를 요구하는 프롬프트와 출력이 그 범주에 반드시 속한다는 보장은 구분합니다. OpenAiGptChatLanguageModelInput뿐 아니라 GenericChatCompletionLanguageModelInput, GenericEmbeddingModelInput, AnthropicClaudeLanguageModelInput 등 언어 모델 입력 타입도 lightweight에서 지원합니다.

시각 언어 모델로 PDF 추출하기

PDF의 본문, 표와 구조를 시각 언어 모델로 추출할 때는 AIP Document Intelligence에서 여러 추출 전략을 시험하고 결과와 품질을 비교한 뒤 운영 구성으로 배포하는 경로를 권장합니다.

설정을 배포하면 선택한 추출 방법, 모델과 사용자 프롬프트를 반영한 Python 코드 저장소가 자동으로 생성됩니다. PDF 추출용 변환을 처음부터 직접 작성할 필요는 없습니다. Python 변환으로 배포에 해당 절차가 있습니다. 전통 OCR과 layout-aware media 변환은 미디어 텍스트 추출에서 설명합니다.