컨테이너 HTTP 서비스에 연결하는 어댑터

이 예제는 컨테이너 안의 실제 HTTP 서비스가 POST /mirror에서 text 필드 하나를 받고 returnedText 필드에 같은 문자열을 반환하는 통신 계약을 사용합니다. 어댑터가 결과를 만들어 내는 대신 서비스에 요청하고 응답을 읽는 연결 예제입니다. 이미지에는 이 계약을 구현한 서버가 있어야 합니다.

단일 컨테이너 이름과 서비스 주소 하나를 사용하는 예제이며 여러 컨테이너라면 사용할 이름과 주소를 명시적으로 선택합니다. 컨테이너의 수신 포트는 모델 버전에 설정해야 합니다.

초기화와 API

ContainerModelAdapter의 init_container는 공유 볼륨 경로와 서비스 주소를 받아 인스턴스를 초기화합니다. container_context.services는 컨테이너 이름에서 주소 목록으로 가는 맵이며 shared_empty_dir_mount_path는 컨테이너와 어댑터가 함께 사용하는 경로입니다.

입력 input_df에는 text 문자열이 있어야 합니다. 출력 output_df는 입력 text와 서비스가 반환한 returnedText를 포함합니다. 필요한 Python 의존성은 palantir_models, requests와 pandas입니다.

import requests
import json
import pandas as pd

import palantir_models as pm


class ExampleIdentityFunctionModelAdapter(pm.ContainerModelAdapter):
    """
        :display-name: Example Identity Function Model Adapter
        :description: Reference example of a model adapter for container-backed model
    """
    def __init__(self, shared_volume_path, model_host_and_port):
        self.shared_volume_path = shared_volume_path
        self.model_host_and_port = model_host_and_port

    @classmethod
    def init_container(cls, container_context):
        shared_volume_path = container_context.shared_empty_dir_mount_path
        # Note this adapter only expects one container name with one provided service URI.
        model_host_and_port = list(container_context.services.values())[0][0]
        return cls(shared_volume_path, model_host_and_port)

    @classmethod
    def api(cls):
        inputs = {"input_df": pm.Pandas(columns=[("text", str)])}
        outputs = {"output_df": pm.Pandas(columns=[("text", str), ("returnedText", str)])}
        return inputs, outputs

    def predict(self, input_df):
        def run_inference_on_row(row):
            request = {"text": row.text}
            response = requests.post("http://" + self.model_host_and_port + "/mirror", json=request)
            json_res = json.loads(response.content.decode("utf-8"))
            return (row.text, json_res["returnedText"])

        results = [run_inference_on_row(row) for row in input_df.itertuples()]
        columns = ["text", "returnedText"]
        return pd.DataFrame(results, columns=columns)

각 행을 JSON 요청으로 전달하고 응답 JSON에서 returnedText를 읽어 결과 표를 만듭니다. 실제 서비스가 다른 URL, 인증이나 요청 및 응답 형식을 사용한다면 그 계약에 맞춰 클라이언트 호출과 모델 API를 작성합니다. 컨텍스트 전체는 ModelAdapter 참조에 있습니다.